diff --git a/api/agent_sessions.py b/api/agent_sessions.py index a9016e85..72fa8340 100644 --- a/api/agent_sessions.py +++ b/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 diff --git a/api/models.py b/api/models.py index cb5aeb12..56265252 100644 --- a/api/models.py +++ b/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, diff --git a/api/routes.py b/api/routes.py index 6c0bfb0d..81727497 100644 --- a/api/routes.py +++ b/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