fix(sessions): bound project-assigned CLI sessions per project and budget the unassigned window separately

Letting assigned sessions past the *recent* cap was right; letting them past
*all* bounds was not. Three linked defects from review of #6659:

1. The assigned list was effectively unbounded. models.py's 200 cap only applied
   to state.db's own pass; imported CLI sidecars from all_sessions() never pass
   through it and _cap_recent_cli_sessions() retained every assigned row (1,000
   assigned rows -> 1,000 returned). The FINAL MERGED payload is the only place
   that sees every source of assigned rows, so bound it there - per project, so
   one busy project cannot evict another's history (the >200-assigned case
   greptile flagged), and counting logical conversations, not raw rows.

2. Assigned rows spent the unassigned quota: 3 assigned + 20 unassigned yielded
   17 unassigned, and the model fetched 20 rows total up front so bumping only
   the route counter could not refill. Count the two separately in the route cap
   and add a bounded unassigned refill pass, keyed on the lineage, so 20 unique
   unassigned logical conversations survive sidecar/lineage dedup. The refill
   only runs when the window was actually saturated AND came up short.

3. A deleted or cross-profile project_id hid the session outright: it was
   surfaced unchecked, so the row left "Unassigned" and its overflow became
   default_hidden with no chip left to reveal it. Resolve non-system ids against
   profile-scoped load_projects() and coerce a miss to None BEFORE either cap -
   the mirror of the bug this PR fixes.

read_importable_agent_session_rows() grows a `project_assignment` filter
('assigned'/'unassigned') replacing the require_project_id boolean. Both arms
key on the whole compression lineage and are exact complements, so neither
budget can double-count a lineage.

_state_row_project_id() keeps its (sid, source) signature: master's kanban pass
calls it that way, and giving it a row-dict signature turned that whole pass
into a swallowed TypeError that dropped every kanban row.

api/agent_sessions.py

@@ -113,6 +113,9 @@ CLI_MIN_UNTITLED_USER_MESSAGE_COUNT = 2
 # window cannot collapse genuinely concurrently started children.
 CONTINUATION_STARTED_AT_TOLERANCE_SECONDS = 2.0
 
+# Accepted ``project_assignment`` values for read_importable_agent_session_rows.
+PROJECT_ASSIGNMENT_FILTERS = frozenset({'assigned', 'unassigned'})
+
 # Raw-row oversample factors for the bounded candidate window. ``limit`` counts
 # LOGICAL conversations, but compression segments and the post-projection
 # visibility filters are spent from the raw window first, so a single 8x pass can
@@ -680,7 +683,7 @@ def read_importable_agent_session_rows(
     log=None,
     exclude_sources: tuple[str, ...] | None = ("cron", "webui"),
     include_sources: tuple[str, ...] | None = None,
-    require_project_id: bool = False,
+    project_assignment: str | None = None,
 ) -> list[dict]:
     """Return agent sessions projected as importable conversations.
 
@@ -698,6 +701,17 @@ def read_importable_agent_session_rows(
     filter; callers that want an include-only query should explicitly pass
     ``exclude_sources=None`` so the default exclusions do not also apply.
 
+    ``project_assignment`` narrows to project-assigned (``'assigned'``) or
+    project-free (``'unassigned'``) conversations; ``None`` returns both. The
+    two are exact complements and key on the whole compression lineage, so a
+    caller can budget assigned and unassigned conversations independently
+    without either filter double-counting a lineage (#6659).
+
+    ``limit`` bounds LOGICAL conversations, not raw rows: the slice is applied
+    after lineage projection, and the raw candidate window is re-widened once
+    if compression segments consumed it before the projection could fill the
+    request.
+
     ``limit`` bounds the *recency slice*, not the returned row count. Subagent
     rows only render as children when their parent row is in the same payload,
     so subagent ancestors of selected rows are re-added afterwards and the
@@ -816,10 +830,20 @@ def read_importable_agent_session_rows(
                 placeholders = ", ".join("?" for _ in excluded)
                 where_clauses.append(f"s.source NOT IN ({placeholders})")
                 params.extend(excluded)
-        if require_project_id:
+        if project_assignment is not None:
+            wanted = str(project_assignment).strip().lower()
+            if wanted not in PROJECT_ASSIGNMENT_FILTERS:
+                raise ValueError(
+                    "project_assignment must be one of "
+                    f"{sorted(PROJECT_ASSIGNMENT_FILTERS)} or None"
+                )
             if 'project_id' not in session_cols:
-                return []
-            if {'parent_session_id', 'end_reason'} <= session_cols:
+                # A schema that cannot persist assignments has no assigned 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 []
+            elif {'parent_session_id', 'end_reason'} <= session_cols:
                 continuation_checks = [
                     "parent.end_reason IN ('compression', 'cli_close')",
                     "(parent.source IS NULL OR child.source IS NULL "
@@ -834,9 +858,14 @@ def read_importable_agent_session_rows(
                         "LOWER(TRIM(COALESCE(child.session_source, ''))) != 'fork'"
                     )
                 continuation_where = " AND ".join(continuation_checks)
+                # An assignment anywhere in a compression lineage assigns the
+                # whole logical conversation, so both filters must key on the
+                # lineage, not the individual row: 'unassigned' is the exact
+                # complement of 'assigned' (#6659).
+                membership = "IN" if wanted == 'assigned' else "NOT IN"
                 where_clauses.append(
                     f"""
-                    s.id IN (
+                    s.id {membership} (
                         WITH RECURSIVE project_lineage(id) AS (
                             SELECT seed.id
                             FROM sessions seed
@@ -859,10 +888,14 @@ def read_importable_agent_session_rows(
                     )
                     """
                 )
-            else:
+            elif wanted == 'assigned':
                 where_clauses.append(
                     "s.project_id IS NOT NULL AND TRIM(s.project_id) != ''"
                 )
+            else:
+                where_clauses.append(
+                    "(s.project_id IS NULL OR TRIM(s.project_id) = '')"
+                )
 
         # Without ``idx_messages_session`` the correlated ``MAX(mx.timestamp)``
         # candidate ordering rescans ``messages`` once per session (seconds on

api/models.py

@@ -54,8 +54,19 @@ logger = logging.getLogger(__name__)
 # profile init; override with HERMES_WEBUI_VISIBLE_SESSION_LIMIT (clamped 1–200).
 CLI_VISIBLE_SESSION_LIMIT = _cfg.CLI_VISIBLE_SESSION_LIMIT
 # Project chips must remain able to reveal older assigned CLI/TUI sessions even
-# after newer unassigned rows fill the normal sidebar window.
+# after newer unassigned rows fill the normal sidebar window (#6659).
+#
+# The bound is PER PROJECT and counts LOGICAL conversations (after lineage and
+# sidecar dedup), not raw state.db rows. Per project, because a single global
+# budget re-creates the bug this fixes — one busy project would evict another
+# project's whole history. Logical conversations, because compression segments
+# are not separately addressable and must not spend a project's budget.
 PROJECT_ASSIGNED_CLI_LIMIT = 200
+# Hard ceiling on the assigned-recovery QUERY so the per-project budget cannot
+# multiply into an unbounded scan on a profile with dozens of projects. Past
+# this, the recovery pass degrades to "newest N assigned conversations overall";
+# project-filtered server pagination is the fix for profiles that large.
+PROJECT_ASSIGNED_CLI_SCAN_CEILING = 2000
 # How many messageful cron sessions to surface in the project-chip layer.
 # Needs to exceed CLI_VISIBLE_SESSION_LIMIT so older cron runs stay
 # addressable even when many newer non-cron sessions dominate the default
@@ -7343,6 +7354,35 @@ def _profile_has_user_projects(profile: str | None = None) -> bool:
     return False
 
 
+def profile_scoped_project_ids() -> frozenset[str]:
+    """Project IDs that currently exist for the active profile.
+
+    A state.db ``project_id`` is an opaque string the agent wrote; the project
+    it names can since have been deleted, or can belong to a different profile.
+    Callers resolve against this set and treat a miss as "unassigned" so a stale
+    assignment can never HIDE a session: an unresolved id would otherwise drop
+    the row out of "Unassigned" and turn it into a ``default_hidden`` row with no
+    project chip left to reveal it (#6659 review finding 3).
+
+    Profile/alias matching mirrors ``ensure_cron_project`` so the two never
+    disagree about which profile owns a project. Read-only.
+    """
+    from api.profiles import get_active_profile_name, _is_root_profile
+
+    active = get_active_profile_name() or 'default'
+    resolved: set[str] = set()
+    for p in load_projects():
+        project_id = str(p.get('project_id') or '').strip()
+        if not project_id:
+            continue
+        row_profile = p.get('profile')
+        if row_profile == active or (
+            _is_root_profile(row_profile or 'default') and _is_root_profile(active)
+        ):
+            resolved.add(project_id)
+    return frozenset(resolved)
+
+
 def is_cron_session(session_id: str, source_tag: str | None = None) -> bool:
     """Return True if a session originates from a cron job."""
     if source_tag == 'cron':
@@ -8325,14 +8365,52 @@ def _load_cli_sessions_uncached(
             _webhook_pid_cache[0] = ensure_webhook_project(profile=_cli_profile)
         return _webhook_pid_cache[0]
 
-    def _state_row_project_id(row: dict) -> str | None:
-        sid = row['id']
-        source = row.get('source')
+    # The live project catalog for this profile, read lazily and at most once per
+    # scan (same [resolved, value] cell pattern as _cron_pid above, for the same
+    # #4842 reason: a per-row load_projects() is a cold-sidebar I/O blowup).
+    _known_project_ids_cache: list = [False, frozenset()]
+    def _known_project_ids() -> frozenset[str]:
+        if not _known_project_ids_cache[0]:
+            _known_project_ids_cache[0] = True
+            try:
+                _known_project_ids_cache[1] = profile_scoped_project_ids()
+            except Exception:
+                logger.debug("Project catalog read failed for CLI projection", exc_info=True)
+        return _known_project_ids_cache[1]
+
+    def _resolved_project_id(raw) -> str | None:
+        """A state.db assignment, or None when its project no longer resolves.
+
+        Coercing an unresolved (deleted or cross-profile) id to None BEFORE
+        either cap runs is what keeps the row in the default "Unassigned" window
+        instead of exiling it to a ``default_hidden`` row whose project chip does
+        not exist any more — the mirror of the bug this PR fixes (#6659).
+        """
+        project_id = str(raw or '').strip()
+        if not project_id:
+            return None
+        return project_id if project_id in _known_project_ids() else None
+
+    def _state_row_project_id(sid: str, source: str | None) -> str | None:
+        """Dedicated system-source chip for a background row, else None."""
         if is_cron_session(sid, source):
             return _cron_pid()
         if is_webhook_session(sid, source):
             return _webhook_pid()
-        return row.get('project_id')
+        return None
+
+    def _interactive_row_project_id(row: dict) -> str | None:
+        """Project chip for an interactive (CLI/TUI/ACP) state.db row.
+
+        Single resolved value for both the cap decisions below and the projected
+        payload, so the code that budgets a row and the code that renders it can
+        never disagree about which project it belongs to.
+        """
+        sid = row['id']
+        source = row.get('source')
+        if is_cron_session(sid, source) or is_webhook_session(sid, source):
+            return _state_row_project_id(sid, source)
+        return _resolved_project_id(row.get('project_id'))
 
     profile_value = _cli_profile or 'default'
     # A deleted WebUI session is tombstoned (see _record_webui_deleted_session_tombstone)
@@ -8360,46 +8438,129 @@ def _load_cli_sessions_uncached(
         exclude_sources=("cron", "webhook", "kanban") if source_filter is None else None,
         include_sources=None if source_filter is None else (source_filter,),
     )
-    if (
-        source_filter is None
-        and project_assigned_limit is not False
-        and _state_db_supports_project_ids(db_path)
-    ):
+    if source_filter is None:
+        # The interactive window above is a MIXED budget of assigned and
+        # unassigned conversations, and it truncates by recency. Two bounded
+        # recovery passes repair the two ways that loses a conversation:
+        #   1. an assigned conversation older than the window is unreachable even
+        #      though its project chip still claims it;
+        #   2. every assigned row inside the window spends a slot the sidebar
+        #      owes to an unassigned conversation (#6659 review findings 1-2).
+        # Both passes are keyed on the LOGICAL conversation (lineage), never the
+        # raw row, so compression segments cannot consume either budget.
+        interactive_excluded = ("cron", "webhook", "kanban")
+        first_pass_count = len(state_rows)
         represented_rows: dict[str, dict] = {}
-        for row in state_rows:
-            for key in ("id", "_lineage_root_id", "_lineage_tip_id"):
-                value = row.get(key)
-                if value:
-                    represented_rows.setdefault(str(value), row)
-        try:
-            assigned_rows = read_importable_agent_session_rows(
-                db_path,
-                limit=project_assigned_limit,
-                log=logger,
-                exclude_sources=("cron", "webhook"),
-                require_project_id=True,
+
+        def _lineage_ids(row: dict) -> list[str]:
+            return [
+                str(value)
+                for key in ("id", "_lineage_root_id", "_lineage_tip_id")
+                if (value := row.get(key))
+            ]
+
+        def _merge_state_row(row: dict) -> bool:
+            """Add ``row`` unless its lineage is already represented.
+
+            Returns True only when a NEW logical conversation was added, so the
+            callers below budget conversations rather than rows. An already
+            represented lineage is upgraded in place: the recovery pass sees the
+            same conversation carrying its resolved project assignment, and the
+            projection loop must read that richer row.
+            """
+            lineage_ids = _lineage_ids(row)
+            existing = next(
+                (represented_rows[value] for value in lineage_ids if value in represented_rows),
+                None,
             )
-            for row in assigned_rows:
-                lineage_ids = [
-                    str(value)
-                    for key in ("id", "_lineage_root_id", "_lineage_tip_id")
-                    if (value := row.get(key))
-                ]
-                existing = next(
-                    (represented_rows[value] for value in lineage_ids if value in represented_rows),
-                    None,
-                )
-                if existing is not None:
-                    existing.clear()
-                    existing.update(row)
-                    for value in lineage_ids:
-                        represented_rows[value] = existing
-                    continue
-                state_rows.append(row)
+            if existing is not None:
+                existing.clear()
+                existing.update(row)
                 for value in lineage_ids:
-                    represented_rows[value] = row
-        except Exception:
-            logger.debug("Project-assigned CLI second pass failed", exc_info=True)
+                    represented_rows[value] = existing
+                return False
+            state_rows.append(row)
+            for value in lineage_ids:
+                represented_rows[value] = row
+            return True
+
+        for row in state_rows:
+            for value in _lineage_ids(row):
+                represented_rows.setdefault(value, row)
+
+        # --- Recovery pass 1: project-assigned conversations the recent window
+        # truncated. Bounded PER PROJECT (see PROJECT_ASSIGNED_CLI_LIMIT) so a
+        # chip can reveal its project's history without any one project growing
+        # the payload without limit. Skipped entirely when the profile has no
+        # projects at all — nothing could resolve, so the query would be waste.
+        known_project_ids = _known_project_ids() if project_assigned_limit is not False else frozenset()
+        if (
+            project_assigned_limit is not False
+            and known_project_ids
+            and _state_db_supports_project_ids(db_path)
+        ):
+            per_project_limit = (
+                None if project_assigned_limit is None
+                else max(0, int(project_assigned_limit))
+            )
+            query_limit = (
+                None if per_project_limit is None
+                else min(
+                    per_project_limit * max(len(known_project_ids), 1),
+                    PROJECT_ASSIGNED_CLI_SCAN_CEILING,
+                )
+            )
+            try:
+                kept_per_project: dict[str, int] = {}
+                for row in read_importable_agent_session_rows(
+                    db_path,
+                    limit=query_limit,
+                    log=logger,
+                    exclude_sources=interactive_excluded,
+                    project_assignment='assigned',
+                ):
+                    project_id = _resolved_project_id(row.get('project_id'))
+                    if project_id is None:
+                        # A deleted/cross-profile id is not an assignment. Leave
+                        # the row to the normal window so it stays reachable
+                        # under "Unassigned" (#6659 review finding 3).
+                        continue
+                    if per_project_limit is not None:
+                        kept = kept_per_project.get(project_id, 0)
+                        if kept >= per_project_limit:
+                            continue
+                        kept_per_project[project_id] = kept + 1
+                    _merge_state_row(row)
+            except Exception:
+                logger.debug("Project-assigned CLI recovery pass failed", exc_info=True)
+
+        # --- Recovery pass 2: top the interactive window back up to
+        # CLI_VISIBLE_SESSION_LIMIT *unassigned* conversations. Only runs when
+        # the window was actually saturated and came up short, so a schema with
+        # no assignments (and a small state.db) pays nothing.
+        unassigned_target = (
+            visible_session_limit if visible_session_limit is not None
+            else CLI_VISIBLE_SESSION_LIMIT
+        )
+        unassigned_seen = sum(
+            1 for row in state_rows if _interactive_row_project_id(row) is None
+        )
+        if (
+            unassigned_target
+            and first_pass_count >= unassigned_target
+            and unassigned_seen < unassigned_target
+        ):
+            try:
+                for row in read_importable_agent_session_rows(
+                    db_path,
+                    limit=unassigned_target,
+                    log=logger,
+                    exclude_sources=interactive_excluded,
+                    project_assignment='unassigned',
+                ):
+                    _merge_state_row(row)
+            except Exception:
+                logger.debug("Unassigned CLI refill pass failed", exc_info=True)
 
     for row in state_rows:
         sid = row['id']
@@ -8446,7 +8607,7 @@ def _load_cli_sessions_uncached(
             'updated_at': raw_ts,
             'pinned': False,
             'archived': _archived,
-            'project_id': _state_row_project_id(row),
+            'project_id': _interactive_row_project_id(row),
             'profile': profile,
             'source_tag': _source,
             'raw_source': row.get('raw_source') or _source_meta.get('raw_source'),
@@ -8746,6 +8907,10 @@ def get_cli_sessions(
                 load_kwargs = {
                     'source_filter': source_filter,
                     'visible_session_limit': None,
+                    # An unbounded first pass already returns every assigned and
+                    # unassigned conversation, so the bounded recovery passes
+                    # have nothing to recover — skip their queries.
+                    'project_assigned_limit': False,
                     'cron_project_limit': None,
                     'webhook_project_limit': None,
                     'kanban_project_limit': None,

api/routes.py

@@ -10488,23 +10488,64 @@ def _cli_visible_session_cap() -> int:
     return CLI_VISIBLE_SESSION_LIMIT
 
 
-def _cap_recent_cli_sessions(sessions: list[dict], cli_cap: int | None = None) -> list[dict]:
-    """Keep only the most recent CLI-visible sessions after filtering."""
+# Per-project bound on project-assigned CLI rows in the FINAL MERGED payload.
+# This is the only place that sees every source of assigned rows at once —
+# state.db's own bounded passes plus imported WebUI sidecars from all_sessions(),
+# which no model-side cap applies to — so it is the only place that can actually
+# bound the assigned set (#6659 review finding 1).
+CLI_PROJECT_ASSIGNED_CAP = 200
+
+
+def _cap_recent_cli_sessions(
+    sessions: list[dict],
+    cli_cap: int | None = None,
+    project_cap: int = CLI_PROJECT_ASSIGNED_CAP,
+) -> list[dict]:
+    """Cap the default CLI list while retaining project-addressable rows.
+
+    ``sessions`` is newest-first and already deduplicated (WebUI sidecars merged,
+    lineages collapsed, messaging sources folded), so every row counted here is
+    one logical conversation.
+
+    Two independent budgets, because they answer to different users (#6659):
+
+    * ``cli_cap`` unassigned conversations own the default sidebar window. An
+      assigned row must not spend one of those slots, or assigning three sessions
+      to a project silently shortens everyone's sidebar to 17 rows. Resolved
+      lazily from the shared configurable window (HERMES_WEBUI_VISIBLE_SESSION_LIMIT),
+      never a second hard-coded 20.
+    * ``project_cap`` assigned conversations PER PROJECT stay in the payload so
+      the project chip can reveal them, marked ``default_hidden`` once the recent
+      window is full. Past that bound they are dropped: keeping assigned rows past
+      the *recent* cap is the fix, keeping them past *all* bounds just trades a
+      vanishing session for a stalled sidebar.
+    """
     if cli_cap is None:
         cli_cap = _cli_visible_session_cap()
     if cli_cap <= 0:
         return sessions
     kept = []
-    cli_seen = 0
+    recent_seen = 0
+    unassigned_seen = 0
+    assigned_seen: dict[str, int] = {}
     for session in sessions:
         if _is_cli_session_for_settings(session):
-            cli_seen += 1
-            if cli_seen > cli_cap:
-                project_id = str(session.get("project_id") or "").strip()
-                if not project_id:
+            project_id = str(session.get("project_id") or "").strip()
+            if not project_id:
+                unassigned_seen += 1
+                if unassigned_seen > cli_cap:
                     continue
-                session = dict(session)
-                session["default_hidden"] = True
+                recent_seen += 1
+            else:
+                seen = assigned_seen.get(project_id, 0) + 1
+                assigned_seen[project_id] = seen
+                if project_cap > 0 and seen > project_cap:
+                    continue
+                if recent_seen >= cli_cap:
+                    session = dict(session)
+                    session["default_hidden"] = True
+                else:
+                    recent_seen += 1
         kept.append(session)
     return kept