jinghui/hermes-webui
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #7816 from nesquena/stage/0924-m

Browse Source 
Keep project-assigned CLI sessions past the recent cap (#6659)
This commit is contained in:
nesquena-hermes authored and GitHub committed 2026-09-24 12:00:33 -07:00
2 parents da8f6f5989 8622bdfb31
commit 56e08ba064
7 files changed
+3998 -121
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

CHANGELOG.md
+10
View File
@@ -44,6 +44,16 @@
### Fixed
- **CLI sessions you moved into a project stay in that project.** CLI sessions were cut to the
recent-session limit before project assignment was checked, so a CLI conversation you'd moved
into a project vanished from its project chip once enough newer sessions existed. The
assignment was still saved; only the row was gone. The recent limit now applies to unassigned
conversations only. Assigned ones stay in the payload (hidden from the main list once the recent
window is full) so the project chips can show them. Assigned rows have their own bound
(`CLI_PROJECT_ASSIGNED_CAP`, 200 across all projects, shared fairly so one busy project can't
take every slot), and a compressed CLI conversation keeps its project through its whole
continuation chain. Thanks @rodrigogs. (#6659)
- **Two open WebUI windows no longer overwrite each other's unread state.** Two clients on the
same origin (for example the desktop PWA and a browser tab) share one `localStorage`, but each
kept its own in-memory copy of which sessions had been viewed. When one wrote, it replaced the
api/agent_sessions.py
+321 -71
View File
@@ -113,6 +113,16 @@ 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
# come up short on a lineage-heavy profile. The second factor is only paid when
# the first window was fully consumed AND still under-delivered.
CANDIDATE_WINDOW_MULTIPLIERS = (8, 32)
SOURCE_LABELS = {
'acp': 'ACP',
'api_server': 'API',
@@ -540,7 +550,7 @@ def _project_agent_session_rows(rows: list[dict]) -> list[dict]:
for children in children_by_parent.values():
children.sort(key=lambda row: row.get('started_at') or 0, reverse=True)
def compression_tip(row: dict) -> tuple[dict | None, int]:
def compression_tip(row: dict) -> tuple[dict | None, int, str | None]:
"""Return the freshest importable continuation descendant for ``row``.
Compression parents can have multiple continuation-looking children when
@@ -552,6 +562,9 @@ def _project_agent_session_rows(rows: list[dict]) -> list[dict]:
latest_importable = row if (row.get('actual_message_count') or 0) > 0 else None
segment_count = 0
best_depth = 1
latest_project_id: str | None = None
project_depth = 0
project_score = -1.0
best_score = (
_as_score(latest_importable.get('last_activity'), latest_importable.get('started_at'))
if latest_importable
@@ -569,6 +582,17 @@ def _project_agent_session_rows(rows: list[dict]) -> list[dict]:
segment_count += 1
current_score = _as_score(current.get('last_activity'), current.get('started_at'))
current_project_id = str(current.get('project_id') or '').strip()
if (
current_project_id
and (
current_score > project_score
or (current_score == project_score and depth >= project_depth)
)
):
latest_project_id = current_project_id
project_depth = depth
project_score = current_score
if (
(current.get('actual_message_count') or 0) > 0
and (current_score > best_score or (current_score == best_score and depth >= best_depth))
@@ -584,7 +608,7 @@ def _project_agent_session_rows(rows: list[dict]) -> list[dict]:
continue
stack.append((child, depth + 1))
return latest_importable, max(segment_count, 1)
return latest_importable, max(segment_count, 1), latest_project_id
projected = []
for row in rows:
@@ -593,13 +617,21 @@ def _project_agent_session_rows(rows: list[dict]) -> list[dict]:
segment_count = 1
tip = row
lineage_project_id = str(row.get('project_id') or '').strip() or None
if row.get('end_reason') in {'compression', 'cli_close'}:
tip, segment_count = compression_tip(row)
tip, segment_count, lineage_project_id = compression_tip(row)
if not tip or (tip.get('actual_message_count') or 0) <= 0:
continue
if tip is row:
projected.append(dict(row))
# The root can still be the freshest *importable* segment while a
# newer EMPTY continuation carries the project assignment. Applying
# the resolved lineage id here too keeps that assignment instead of
# silently dropping it on the early-return path (#6659).
root_only = dict(row)
if lineage_project_id:
root_only['project_id'] = lineage_project_id
projected.append(root_only)
continue
merged = dict(row)
@@ -617,6 +649,8 @@ def _project_agent_session_rows(rows: list[dict]) -> list[dict]:
):
if key in tip:
merged[key] = tip[key]
if lineage_project_id:
merged['project_id'] = lineage_project_id
if str(tip.get('source') or '').strip().lower() == 'tui':
# TUI continuation rows are user-visible session segments (#6, #17,
# ...), not opaque compression snapshots. Keep navigation pointed at
@@ -649,7 +683,10 @@ def read_importable_agent_session_rows(
log=None,
exclude_sources: tuple[str, ...] | None = ("cron", "webui"),
include_sources: tuple[str, ...] | None = None,
) -> list[dict]:
project_assignment: str | None = None,
project_ids: tuple[str, ...] | None = None,
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
@@ -666,22 +703,67 @@ 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).
``project_ids`` narrows the assigned set further, to conversations whose
LINEAGE is assigned to one of the given projects. It exists so a caller can
give each project its own ``limit`` instead of sharing one global window: a
single ``project_assignment='assigned'`` query orders by recency, so one busy
project can consume the whole window and leave quieter projects unreachable
(greptile P1 on #6659). Only meaningful together with
``project_assignment='assigned'``; any other combination raises ValueError
rather than silently returning an unnarrowed (and therefore starvable) set.
``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
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
(``limit * 8`` newest sessions): it re-uses rows the projection already
fetched and never issues an extra query, so an ancestor older than the
oversample stays unresolved and its children render top-level, exactly as
they did before. Widening that window is a ``candidate_limit`` change, not
a change to this walk.
(the ``CANDIDATE_WINDOW_MULTIPLIERS`` window that actually ran — ``limit *
8``, or ``limit * 32`` when the first window was consumed and re-widened):
it re-uses rows the projection already fetched and never issues an extra
query, so an ancestor older than the oversample stays unresolved and its
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':
raise ValueError(
"project_ids requires project_assignment='assigned'"
)
wanted_project_ids = tuple(
cleaned for value in project_ids if (cleaned := str(value or '').strip())
)
if not wanted_project_ids:
# An empty narrowing selects nothing. Returning [] keeps that exact
# instead of degrading to "every assigned conversation".
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)
@@ -705,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)
@@ -721,6 +803,7 @@ def read_importable_agent_session_rows(
origin_user_id_expr = _optional_col('origin_user_id', session_cols)
platform_expr = _optional_col('platform', session_cols)
archived_expr = _optional_col('archived', session_cols)
project_id_expr = _optional_col('project_id', session_cols)
# Older/minimal state.db schemas can have NO ``messages`` table at all,
# or a ``messages`` table without a ``session_id`` / ``timestamp`` column.
# The projection SQL below joins ``messages`` and aggregates
@@ -782,6 +865,90 @@ def read_importable_agent_session_rows(
placeholders = ", ".join("?" for _ in excluded)
where_clauses.append(f"s.source NOT IN ({placeholders})")
params.extend(excluded)
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:
# 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 _result([])
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 "
"OR LOWER(TRIM(parent.source)) = LOWER(TRIM(child.source)))",
]
if 'ended_at' in session_cols:
continuation_checks.append(
"(parent.ended_at IS NULL OR child.started_at >= parent.ended_at)"
)
if 'session_source' in session_cols:
continuation_checks.append(
"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"
# ``project_ids`` narrows only the SEED: the lineage walk still
# pulls in the whole compression chain around a seeded row, so a
# per-project query returns the same logical conversations the
# global one would have, minus the other projects'.
seed_project_filter = ""
if wanted_project_ids:
placeholders = ", ".join("?" for _ in wanted_project_ids)
seed_project_filter = (
f"\n AND TRIM(seed.project_id) IN ({placeholders})"
)
where_clauses.append(
f"""
s.id {membership} (
WITH RECURSIVE project_lineage(id) AS (
SELECT seed.id
FROM sessions seed
WHERE seed.project_id IS NOT NULL
AND TRIM(seed.project_id) != ''{seed_project_filter}
UNION
SELECT child.id
FROM sessions child
JOIN project_lineage lineage ON child.parent_session_id = lineage.id
JOIN sessions parent ON parent.id = lineage.id
WHERE {continuation_where}
UNION
SELECT parent.id
FROM sessions child
JOIN project_lineage lineage ON child.id = lineage.id
JOIN sessions parent ON parent.id = child.parent_session_id
WHERE {continuation_where}
)
SELECT id FROM project_lineage
)
"""
)
params.extend(wanted_project_ids)
elif wanted == 'assigned':
if wanted_project_ids:
placeholders = ", ".join("?" for _ in wanted_project_ids)
where_clauses.append(
f"TRIM(COALESCE(s.project_id, '')) IN ({placeholders})"
)
params.extend(wanted_project_ids)
else:
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
@@ -828,6 +995,7 @@ def read_importable_agent_session_rows(
{origin_chat_id_expr},
{origin_user_id_expr},
{platform_expr},
{project_id_expr},
{parent_expr},
{ended_expr},
{end_reason_expr},
@@ -836,10 +1004,21 @@ def read_importable_agent_session_rows(
{user_message_count_expr} AS actual_user_message_count,
{last_activity_expr} AS last_activity
"""
def _project(raw_rows: list[dict]) -> list[dict]:
projected = _project_agent_session_rows(raw_rows)
# model_config is selected only so the continuation classifier can
# read the real _branched_from/_delegate_from markers; it is internal
# agent state and must not leak into the /api/sessions response.
for row in projected:
row.pop('model_config', None)
projected = [_with_normalized_source(row) for row in projected]
return [row for row in projected if is_cli_session_row_visible(row)]
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
@@ -848,7 +1027,6 @@ def read_importable_agent_session_rows(
# can be resumed days later and should still surface at the top.
# Oversampling preserves room for hidden compression segments or
# other rows filtered after projection.
candidate_limit = max(result_limit * 8, result_limit)
if latest_messages_cte:
candidate_cte = (
"WITH {latest_messages_cte}, candidates AS (\n"
@@ -878,8 +1056,7 @@ def read_importable_agent_session_rows(
candidate_order_clause=candidate_order_clause,
)
cur.execute(
f"""
candidate_sql = f"""
{candidate_cte}
{select_sql}
FROM sessions s
@@ -887,69 +1064,142 @@ def read_importable_agent_session_rows(
{join_clause}
{group_by_clause}
{order_by_clause}
""",
[*params, candidate_limit],
"""
# Compression segments and the post-projection visibility filters are
# spent from the RAW candidate window before the logical slice below,
# so a lineage-heavy profile can return far fewer conversations than
# asked for (greptile P1 on #6659). Widen the window once when it was
# 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)
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]
# The recency slice is per-row, but subagent rows are only renderable as
# children: the sidebar nests a child under its parent solely when that
# parent row is present in the same payload. A frozen orchestrator stops
# writing while its leaves keep streaming, so the leaves win the recency
# race and the parent falls outside the window — leaving the leaves to be
# promoted to top-level sidebar rows. Re-add subagent parents that the
# oversampled candidate set already projected (no extra query); webui
# ancestors are left out because that sidebar bucket already has them.
#
# Bounded by construction: ``by_id`` only holds the candidates of the
# window that was actually executed (``limit * 8``, or ``limit * 32``
# when the re-widening pass above had to run), so an ancestor older
# than that oversample is not recovered and its children stay
# top-level — the pre-existing behaviour, narrowed rather than fixed.
# Resolving those would need an unbounded per-row ancestor query on
# the hot sidebar path; widen ``CANDIDATE_WINDOW_MULTIPLIERS`` instead
# if the window proves too tight.
# NOTE: this can return more than ``limit`` rows (see docstring).
have = {row.get('id') for row in selected}
by_id = {row.get('id'): row for row in projected if row.get('id')}
pending = list(selected)
while pending:
row = pending.pop()
if str(row.get('raw_source') or row.get('source') or '').strip().lower() != 'subagent':
continue
parent_id = row.get('parent_session_id')
if not parent_id or parent_id in have:
continue
parent = by_id.get(parent_id)
if parent is None:
continue
if str(parent.get('raw_source') or parent.get('source') or '').strip().lower() != 'subagent':
continue
selected.append(parent)
have.add(parent_id)
pending.append(parent)
return _result(selected, window_exhausted)
cur.execute(
f"""
{select_sql}
FROM sessions s
{join_clause}
WHERE {' AND '.join(where_clauses)}
{group_by_clause}
{order_by_clause}
""",
params,
)
return _result(_project([dict(row) for row in cur.fetchall()]))
def read_assigned_project_row_counts(
db_path: Path,
log=None,
exclude_sources: tuple[str, ...] | None = None,
) -> dict[str, int]:
"""Return ``{project_id: assigned raw row count}`` from ``state.db``.
A single GROUP BY over ``sessions.project_id``: no messages join, no lineage
recursion, and an answer bounded by the number of distinct project ids. It
exists so the assigned-recovery pass can tell WHICH projects still have rows
it has not delivered, and pay a per-project follow-up query only for those,
instead of one query per registered project (greptile P1 on #6659).
The counts are RAW rows, so a compression lineage counts once per segment.
That makes the signal deliberately conservative — it can overestimate what a
project still owes, never underestimate it, so a starved project is always
detected while at worst one follow-up query comes back empty.
"""
db_path = Path(db_path)
log = log or logger
if not db_path.exists():
return {}
try:
with closing(open_state_db_readonly(db_path, log)) as conn:
cur = conn.cursor()
cur.execute("PRAGMA table_info(sessions)")
session_cols = {row[1] for row in cur.fetchall()}
if not {'project_id', 'source'} <= session_cols:
return {}
where_clauses = [
"s.source IS NOT NULL",
"s.project_id IS NOT NULL",
"TRIM(s.project_id) != ''",
]
params: list[object] = []
excluded = tuple(
str(source) for source in (exclude_sources or ()) if source
)
else:
if excluded:
placeholders = ", ".join("?" for _ in excluded)
where_clauses.append(f"s.source NOT IN ({placeholders})")
params.extend(excluded)
cur.execute(
f"""
{select_sql}
SELECT TRIM(s.project_id) AS project_id, COUNT(*) AS row_count
FROM sessions s
{join_clause}
WHERE {' AND '.join(where_clauses)}
{group_by_clause}
{order_by_clause}
GROUP BY TRIM(s.project_id)
""",
params,
)
projected = _project_agent_session_rows([dict(row) for row in cur.fetchall()])
# model_config is selected only so the continuation classifier can
# read the real _branched_from/_delegate_from markers; it is internal
# agent state and must not leak into the /api/sessions response.
for row in projected:
row.pop('model_config', None)
projected = [_with_normalized_source(row) for row in projected]
projected = [row for row in projected if is_cli_session_row_visible(row)]
if limit is None:
return projected
selected = projected[:max(0, int(limit))]
# The recency slice is per-row, but subagent rows are only renderable as
# children: the sidebar nests a child under its parent solely when that
# parent row is present in the same payload. A frozen orchestrator stops
# writing while its leaves keep streaming, so the leaves win the recency
# race and the parent falls outside the window — leaving the leaves to be
# promoted to top-level sidebar rows. Re-add subagent parents that the
# oversampled candidate set already projected (no extra query); webui
# ancestors are left out because that sidebar bucket already has them.
#
# Bounded by construction: ``by_id`` only holds the ``limit * 8``
# newest candidates, so an ancestor older than that oversample is not
# recovered and its children stay top-level — the pre-existing
# behaviour, narrowed rather than fixed. Resolving those would need an
# unbounded per-row ancestor query on the hot sidebar path; widen
# ``candidate_limit`` instead if the window proves too tight.
# NOTE: this can return more than ``limit`` rows (see docstring).
have = {row.get('id') for row in selected}
by_id = {row.get('id'): row for row in projected if row.get('id')}
pending = list(selected)
while pending:
row = pending.pop()
if str(row.get('raw_source') or row.get('source') or '').strip().lower() != 'subagent':
continue
parent_id = row.get('parent_session_id')
if not parent_id or parent_id in have:
continue
parent = by_id.get(parent_id)
if parent is None:
continue
if str(parent.get('raw_source') or parent.get('source') or '').strip().lower() != 'subagent':
continue
selected.append(parent)
have.add(parent_id)
pending.append(parent)
return selected
return {
str(row[0]): int(row[1] or 0)
for row in cur.fetchall()
if str(row[0] or '').strip()
}
except Exception:
log.debug("assigned project row-count probe failed", exc_info=True)
return {}
def _lineage_report_row(row: dict, role: str) -> dict:
api/models.py
+756 -10
View File
@@ -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
@@ -42,6 +42,7 @@ from api.agent_sessions import (
is_cli_session_row,
normalize_agent_session_source,
open_state_db_readonly,
read_assigned_project_row_counts,
read_importable_agent_session_rows,
read_session_lineage_metadata,
)
@@ -53,6 +54,63 @@ logger = logging.getLogger(__name__)
# nests when it wins a slot in this window. Resolved in api.config before
# 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 (#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, every project's share of the recovery pass shrinks to
# PROJECT_ASSIGNED_CLI_SCAN_CEILING // len(projects) instead of any one project
# eating the whole window; project-filtered server pagination is still the real
# fix for profiles that large.
PROJECT_ASSIGNED_CLI_SCAN_CEILING = 2000
# The project-scoped follow-up below is deliberately NOT capped by project
# 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 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
# probe + one small project-scoped query per starved project, plus the pass's
# single widening budget (query_limit, see `widening_budget`) shared by the
# compression-heavy retries — a retry is NOT granted the global allowance per
# project, which would make a build scale with the number of starved projects
# (greptile P1 on #6659).
#
# --- Bounds for the UNASSIGNED refill pass (see _load_cli_sessions_uncached).
# The SQL 'unassigned' filter reads state.db.project_id, but a WebUI-side move
# records the assignment on the session's sidecar ONLY, so that filter hands back
# conversations this projection classifies as ASSIGNED. How many is not knowable
# before the query runs, so the refill re-classifies its own result and widens
# again while the window is still short. These two constants are what keep that
# loop finite. Both bound ONE profile context: get_cli_sessions(
# all_profiles=1) runs the whole loader once per context, so an N-profile sidebar
# pays N times the numbers below.
#
# At most this many unassigned queries per profile context (the first included).
# The widening below is geometric and proportional, not one row at a time, so 4
# covers both moves spread through the history and moves clustered just under the
# window edge; the last of the four abandons estimating and reads the ceiling in
# one go, which is the widest read the loop is ever allowed to pay. It is also
# what paces the geometric growth: each widening adds at least width // queries
# still left, so the steps get bolder as the forced ceiling read gets closer.
UNASSIGNED_CLI_REFILL_MAX_QUERIES = 4
# ...and no single refill query reads deeper than this many logical
# conversations. It bounds both the I/O and the payload: a build cannot be made
# to scan the whole of a large state.db by moving conversations into projects.
# The unassigned window is filled whenever at least CLI_VISIBLE_SESSION_LIMIT of
# the newest UNASSIGNED_CLI_REFILL_SCAN_CEILING state.db-unassigned conversations
# are still unassigned once sidecars are read; past that (>180 of the newest 200
# moved from the WebUI) the sidebar is honestly short rather than unbounded.
UNASSIGNED_CLI_REFILL_SCAN_CEILING = 200
# 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
@@ -63,6 +121,11 @@ WEBHOOK_PROJECT_CHIP_LIMIT = 200
# higher project-chip cap so project-assigned kanban rows stay addressable when
# the toggle is on, without letting them dominate the default sidebar window.
KANBAN_PROJECT_CHIP_LIMIT = 200
# Sources with their own bounded project-chip pass. They are kept out of the
# interactive CLI window and out of its recovery passes, and they keep the
# system-chip answer wherever they are projected, so the same background row can
# never report two different project chips.
BACKGROUND_CLI_SOURCES = ("cron", "webhook", "kanban")
_CLI_SESSIONS_CACHE_TTL_SECONDS = 5.0
# While a turn is actively streaming, hold the CLI/cron projection longer than
# one poll interval (mirrors the route-level #4808 hold-down). The frontend
@@ -7350,6 +7413,48 @@ def _profile_has_user_projects(profile: str | None = None) -> bool:
return False
# Sentinel for profile_scoped_project_ids(): "no profile given, use the active
# one". A bare None cannot mean that, because None is a legitimate profile value
# (_profiles_match reads it as the root profile).
_ACTIVE_PROFILE = object()
def profile_scoped_project_ids(profile=_ACTIVE_PROFILE) -> frozenset[str]:
"""Project IDs that currently exist for ``profile`` (default: the active one).
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`` must be the SESSION's profile, not the process-wide active one,
wherever the two can differ. The all-profiles sidebar view scans one state.db
per profile in a single request, so resolving every context against the active
profile made another profile's LIVE assignment look unresolvable and dropped
the row from the payload entirely — while its chip stayed selectable in that
same view (#6659 review finding 3).
Profile/alias matching is delegated to ``api.profiles._profiles_match``,
the canonical helper every other profile-scoped read already uses, so this
set cannot disagree with the rest of the app about which profile owns a
project (renamed root, legacy ``'default'`` tag, and a missing ``profile``
key are all its business, not ours). Read-only.
"""
from api.profiles import get_active_profile_name, _profiles_match
active = get_active_profile_name() if profile is _ACTIVE_PROFILE else profile
resolved: set[str] = set()
for p in load_projects():
project_id = str(p.get('project_id') or '').strip()
if not project_id:
continue
if _profiles_match(p.get('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':
@@ -8171,7 +8276,14 @@ def clear_sidecar_metadata_cache() -> None:
def _state_projection_sidecar_metadata(sid: str) -> dict:
"""Return UI-owned metadata (title + archived) for a state.db-projected row.
"""Return UI-owned metadata (title + archived + project_id) for a state row.
``project_id`` is UI-owned in exactly the same way as the other two:
``/api/session/move`` writes it onto the sidecar and NOTHING writes
``state.db.sessions.project_id``. Reading it here is what lets the sidebar
projection agree with the assignment ``all_sessions()`` re-surfaces, instead
of counting a moved conversation against the unassigned window (#6659 review
finding 2).
Memoized by the sidecar file's (path, mtime_ns, size, ctime_ns) stat
signature so the sidebar projection — which calls this once per row in both
@@ -8182,11 +8294,11 @@ def _state_projection_sidecar_metadata(sid: str) -> dict:
Returns a COPY so callers can't mutate the cached dict.
NOTE: this stat-gates on ``SESSION_DIR / f'{sid}.json'`` because that file is
``Session.load_metadata_only``'s sole source for title+archived. If that ever
stops being true (metadata moves to another store), this gate would short-
circuit before the real source — update both together.
``Session.load_metadata_only``'s sole source for title+archived+project_id. If
that ever stops being true (metadata moves to another store), this gate would
short-circuit before the real source — update both together.
"""
default = {"title": None, "archived": None}
default = {"title": None, "archived": None, "project_id": None}
if not is_safe_session_id(sid):
return dict(default)
p = SESSION_DIR / f'{sid}.json'
@@ -8214,6 +8326,8 @@ def _state_projection_sidecar_metadata(sid: str) -> dict:
if title:
metadata["title"] = title
metadata["archived"] = bool(getattr(webui_meta, 'archived', False))
project_id = str(getattr(webui_meta, 'project_id', None) or '').strip()
metadata["project_id"] = project_id or None
with _SIDECAR_METADATA_CACHE_LOCK:
# Re-check under lock in case a concurrent build populated it; either
@@ -8226,6 +8340,16 @@ def _state_projection_sidecar_metadata(sid: str) -> dict:
return dict(metadata)
def _state_db_supports_project_ids(db_path: Path) -> bool:
"""Return whether the agent session schema can persist project assignments."""
try:
with closing(open_state_db_readonly(db_path)) as conn:
rows = conn.execute("PRAGMA table_info(sessions)").fetchall()
except Exception:
return False
return any(str(row[1]) == "project_id" for row in rows)
@profile_home_resolve_cache_scope()
def _load_cli_sessions_uncached(
hermes_home: Path,
@@ -8234,6 +8358,7 @@ def _load_cli_sessions_uncached(
source_filter=None,
*,
visible_session_limit: int | None = None,
project_assigned_limit: int | None | bool = PROJECT_ASSIGNED_CLI_LIMIT,
cron_project_limit: int | None | bool = CRON_PROJECT_CHIP_LIMIT,
webhook_project_limit: int | None | bool = WEBHOOK_PROJECT_CHIP_LIMIT,
kanban_project_limit: int | None | bool = KANBAN_PROJECT_CHIP_LIMIT,
@@ -8321,13 +8446,98 @@ def _load_cli_sessions_uncached(
_webhook_pid_cache[0] = ensure_webhook_project(profile=_cli_profile)
return _webhook_pid_cache[0]
# The live project catalog for THIS SCAN'S 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).
#
# Scoped to ``_cli_profile`` — the profile whose state.db this scan is
# reading — not to the process-wide active profile. The all-profiles view
# calls this loader once per profile context in a single request, so the
# active-profile catalog would misread another profile's LIVE assignment as
# unresolvable and drop the row from the payload (#6659 review finding 3).
_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(_cli_profile)
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.
cron and webhook own an auto-provisioned project; kanban deliberately
does not, so it falls through to None (upstream behaviour) and is
reachable through the normal sidebar list instead of a chip.
"""
if is_cron_session(sid, source):
return _cron_pid()
if is_webhook_session(sid, source):
return _webhook_pid()
return None
# Sidecar-carried assignments, memoized per sid for this scan.
# _state_projection_sidecar_metadata is already stat-cached, but
# _interactive_row_project_id is called several times per row (budget
# seeding, the unassigned count, the projection), and one os.stat per call
# per row is the kind of cold-sidebar I/O #4842 removed.
_sidecar_pid_cache: dict[str, str | None] = {}
def _sidecar_row_project_id(sid: str) -> str | None:
if sid not in _sidecar_pid_cache:
_sidecar_pid_cache[sid] = _state_projection_sidecar_metadata(sid).get('project_id')
return _sidecar_pid_cache[sid]
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.
The WebUI sidecar is consulted when state.db carries no assignment,
because ``/api/session/move`` writes ``project_id`` onto the sidecar ONLY
— nothing writes ``state.db.sessions.project_id``. Without this the
"``CLI_VISIBLE_SESSION_LIMIT`` unassigned conversations" guarantee counted
a WebUI-side assignment as unassigned while the route (which sees the
sidecar's value through ``all_sessions()``) classified the same
conversation as ASSIGNED, so three moved sessions silently shortened
everyone's sidebar to 17 rows (#6659 review finding 2). state.db wins when
it has an assignment of its own: it is the agent's own record.
Background sources keep the system-chip answer even when a
``source_filter`` routes them through this loop, so one kanban row cannot
report a resolved ``project_id`` here and ``None`` from its own bounded
second pass further down.
"""
sid = row['id']
source = row.get('source')
if (
str(source or '').strip().lower() in BACKGROUND_CLI_SOURCES
or is_cron_session(sid, source)
or is_webhook_session(sid, source)
):
return _state_row_project_id(sid, source)
resolved = _resolved_project_id(row.get('project_id'))
if resolved is None:
resolved = _resolved_project_id(_sidecar_row_project_id(sid))
return resolved
profile_value = _cli_profile or 'default'
# A deleted WebUI session is tombstoned (see _record_webui_deleted_session_tombstone)
# so recovery/audit/claim treat it as gone. The sidebar's own state.db projection
@@ -8339,7 +8549,7 @@ def _load_cli_sessions_uncached(
_deleted_webui_tombstone = _load_webui_deleted_session_tombstone()
except Exception:
_deleted_webui_tombstone = frozenset()
for row in read_importable_agent_session_rows(
state_rows = read_importable_agent_session_rows(
db_path,
limit=visible_session_limit if visible_session_limit is not None else (
CRON_PROJECT_CHIP_LIMIT if source_filter == 'cron'
@@ -8351,9 +8561,536 @@ def _load_cli_sessions_uncached(
# Background sources have independent bounded passes below. Keeping them
# out of this 20-row interactive window prevents a busy worker source
# (especially kanban) from evicting every CLI/TUI/ACP conversation.
# Spelled as a literal on purpose: tests/test_issue2841_show_cron_sessions_toggle.py
# reads this line as source text. Must stay equal to BACKGROUND_CLI_SOURCES
# (pinned by test_background_source_exclusion_literal_matches_the_constant).
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:
# 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 = BACKGROUND_CLI_SOURCES
first_pass_count = len(state_rows)
represented_rows: dict[str, dict] = {}
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, which is
what lets recovery pass 1 spend its per-project budget on
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. Pass 2 needs no budget of its own (its
query limit is the bound) and ignores the result.
"Upgrade" is one-way. Overwriting a row that already carries a
RESOLVED project_id with a projection that has none would delete the
chip this whole function exists to restore — and swap the row's
identity from the lineage tip back to its root. Pass 2 projects the
same conversations as pass 1 without the assignment context, so it
must never be able to win that race.
"""
lineage_ids = _lineage_ids(row)
existing = next(
(represented_rows[value] for value in lineage_ids if value in represented_rows),
None,
)
if existing is not None:
if (
_resolved_project_id(row.get('project_id')) is not None
or _resolved_project_id(existing.get('project_id')) is None
):
# Either the incoming row adds an assignment, or the row on
# file has none to lose: overwriting cannot drop a chip.
existing.clear()
existing.update(row)
for value in lineage_ids:
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.
#
# Two tiers, because a per-project BUDGET on top of one global recency
# window is not a per-project bound: a project that owns the whole window
# leaves every other project unrepresented. So the global query runs
# first (the only cost most profiles pay), and only if it came back
# saturated does a project-scoped follow-up top up the projects it
# starved. Worst case per build: 1 global query + 1 GROUP BY probe +
# one project-scoped query per starved project, each bounded to that
# project's remaining budget (whose sum the scan ceiling bounds).
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)
):
project_count = max(len(known_project_ids), 1)
per_project_limit = (
None if project_assigned_limit is None
else max(0, int(project_assigned_limit))
)
# Every project gets an EQUAL share of the global scan ceiling. The
# share only bites past PROJECT_ASSIGNED_CLI_SCAN_CEILING /
# PROJECT_ASSIGNED_CLI_LIMIT projects, and it is what keeps the
# recovered payload bounded (<= the ceiling) while making a project's
# allowance independent of how loud its neighbours are.
effective_limit = (
None if per_project_limit is None
else min(
per_project_limit,
max(1, PROJECT_ASSIGNED_CLI_SCAN_CEILING // project_count),
)
)
query_limit = (
None if effective_limit is None
else min(
effective_limit * project_count,
PROJECT_ASSIGNED_CLI_SCAN_CEILING,
)
)
try:
# An assigned conversation ALREADY inside the recent window
# occupies one of its project's slots, so this pass tops each
# project UP TO the bound instead of stacking a second bound on
# top of the window.
kept_per_project: dict[str, int] = {}
for row in state_rows:
seeded_project_id = _interactive_row_project_id(row)
if seeded_project_id is not None:
kept_per_project[seeded_project_id] = (
kept_per_project.get(seeded_project_id, 0) + 1
)
def _spend_assigned_rows(rows: list[dict]) -> None:
"""Merge ``rows`` into the window under the per-project budget."""
for row in rows:
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 finding 3).
continue
if (
effective_limit is not None
and kept_per_project.get(project_id, 0) >= effective_limit
):
continue
if _merge_state_row(row):
# Spend the budget on LOGICAL conversations only: a
# row that merely upgrades an already-represented
# lineage (a compression segment, or a window row
# learning its assignment) costs its project nothing.
kept_per_project[project_id] = (
kept_per_project.get(project_id, 0) + 1
)
# 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, 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)
# 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
# for projects that can actually be starved — not one per
# registered project.
owed_rows = read_assigned_project_row_counts(
db_path,
log=logger,
exclude_sources=interactive_excluded,
)
starved = sorted(
(kept_per_project.get(project_id, 0), project_id)
for project_id, row_count in owed_rows.items()
if project_id in known_project_ids
and kept_per_project.get(project_id, 0) < effective_limit
and row_count > kept_per_project.get(project_id, 0)
)
# Neediest first (fewest conversations delivered, then id for
# a deterministic order). EVERY starved project is served: a
# fixed per-build project cap would leave every project past
# it unreachable on each rebuild — the very starvation this
# pass exists to repair (greptile P1 on #6659). The per-build
# cost stays bounded because each query is limited to one
# project's remaining budget (sum <= the scan ceiling) and
# only fires when the global window was saturated.
#
# The WIDENING below is rationed per PASS instead (greptile
# P1 on #6659): `remaining` bounds each project's FIRST
# query, but a retry that jumps to `query_limit` is not
# bounded by the project's own share at all, so granting it
# to every starved project made a build scan
# len(starved) * query_limit — an order of magnitude past
# the ceiling on a profile with ten compression-heavy
# projects. One aggregate pool caps the scoped reading of
# a pass at a constant multiple of `query_limit` however
# many projects are starved.
# Only the widening is rationed; the first query of every
# starved project is never skipped, which is what keeps the
# completeness guarantee above intact.
#
# (greptile P1, 2026-09-22 re-review) The pool is debited
# only for what a retry reads BEYOND the project's own
# first-query width — `widened - scoped_limit`, not the
# whole `widened` grant. Debiting the whole grant spent the
# entire pool on the first starved project whose window
# merely touched its own 8x oversample: it read
# `grant - share` extra rows but was billed the full grant,
# so every later starved project lost its retry and its
# older conversations stayed absent from its sidebar on
# every rebuild. With incremental debits a later project
# retries whenever real reads have left it pool; if an
# earlier project genuinely read the pool down to zero,
# that was real scan work — the aggregate bound, not a
# reservation, is what caps the pass. The bound is
# `3 * query_limit`: every starved project's first query
# sums to at most `query_limit`, a retrying project re-reads
# at most its own first width again, and the funded
# increments sum to at most the pool — still constant in
# the number of starved projects, which is what retires the
# original `len(starved) * query_limit` blow-up.
widening_budget = query_limit
for starved_index, (_kept, project_id) in enumerate(starved):
remaining = effective_limit - kept_per_project.get(project_id, 0)
if remaining <= 0:
continue
# A scoped query carries a raw candidate window of its
# own, and compression segments plus the post-projection
# visibility filters are spent from that window BEFORE
# the logical slice. A lineage-heavy project can
# therefore come back short with its window fully
# consumed and older assigned conversations of its own
# still waiting behind it. Without the exhaustion signal
# the loop would advance to the next project and leave
# this one under-delivered on every rebuild — the same
# failure the global query already guards against
# (greptile P1 on #6659). Re-query once, widened to the
# project's own width plus whatever the pass's widening
# pool still has, which is what bounds the retry at the
# PASS level instead of per project (greptile P1 on
# #6659).
scoped_limit = remaining
while True:
scoped_rows, scoped_window_exhausted = cast(
tuple[list[dict], bool],
read_importable_agent_session_rows(
db_path,
limit=scoped_limit,
log=logger,
exclude_sources=interactive_excluded,
project_assignment='assigned',
project_ids=(project_id,),
return_window_exhaustion=True,
),
)
_spend_assigned_rows(scoped_rows)
if (
not scoped_window_exhausted
# A binding window on a project that is now
# full needs no second query, and neither does
# one that cannot be widened any further.
or kept_per_project.get(project_id, 0)
>= effective_limit
or scoped_limit >= query_limit
):
break
# (greptile P1, 2026-09-22 re-review) Each
# retry is funded by a FAIR SHARE of what the
# pool still holds — one slice per starved
# project not yet tried, so later projects
# always keep pool for their own first retry.
# The retry goes to `scoped_limit + share`:
# debiting the whole grant spent the entire
# pool on the first starved project whose
# window merely touched its own 8x oversample
# and left every later one unfunded, but
# granting `min(query_limit, pool)` whole also
# let the first project READ the pool down to
# zero — every project whose raw window is
# consumed re-reads its own width once before
# delivering, so an honest retry must buy
# roughly one more width. The share converges
# there in a few bounded steps; anything still
# short after the pool is spent stays for the
# next build — the alternative is the
# per-project grant that made the pass scan
# `len(starved) * query_limit` (the original
# greptile P1 on #6659).
pool_share = (
widening_budget
// max(len(starved) - starved_index, 1)
)
if pool_share <= 0:
# The pool is spent: no further starved
# project's retry can be funded. Whatever a
# project could not reach this pass stays
# for the next build — the alternative is
# buying reads nobody budgeted for.
break
widened = min(
query_limit, scoped_limit + pool_share
)
widening_budget -= widened - scoped_limit
scoped_limit = widened
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 came up short AND something other than an empty database
# can explain it: either it was saturated, or assigned conversations
# inside it displaced unassigned candidates. A profile with no
# assignments and a small state.db still pays nothing.
unassigned_target = (
visible_session_limit if visible_session_limit is not None
else CLI_VISIBLE_SESSION_LIMIT
)
def _count_unassigned() -> int:
"""Conversations in the window this projection calls unassigned.
Recomputed after every refill query instead of trusted from before
it: the classification depends on the sidecar, so only the rows in
hand can answer it. Cheap to repeat — both inputs
(``_sidecar_row_project_id``, ``_known_project_ids``) are memoized
for the scan, so no pass re-reads a file (#4842).
"""
return sum(
1 for row in state_rows if _interactive_row_project_id(row) is None
)
unassigned_seen = _count_unassigned()
# A sidecar-carried assignment is invisible to the SQL 'unassigned'
# filter (state.db still says NULL), so the refill query would spend
# that many of its own slots re-fetching conversations this pass already
# classified as assigned and already represents. Widen the FIRST query by
# that count so it can actually reach the older unassigned conversations
# the moved rows displaced (#6659 review finding 2). It is only the first
# estimate — the moved conversations this window cannot see yet are what
# the re-examination loop below is for.
sidecar_only_assigned = sum(
1 for row in state_rows
if _interactive_row_project_id(row) is not None
and _resolved_project_id(row.get('project_id')) is None
)
if (
unassigned_target
and unassigned_seen < unassigned_target
and (
first_pass_count >= unassigned_target
# A short MIXED first pass can still be under-delivering: the
# narrower unassigned-only query reaches conversations the mixed
# candidate window spent on assigned rows and segments.
or unassigned_seen < first_pass_count
)
):
try:
# ONE pre-computed allowance is the wrong shape. It is derived
# from the rows fetched BEFORE the refill, but the refill reaches
# OLDER conversations, and those can carry sidecar-only
# assignments of their own — each of which spends an unassigned
# slot again, the exact failure the widening exists to remove.
# (Three moves at the bottom edge of a 30-conversation database
# delivered 19 rows and never fetched the 20th conversation;
# eleven straddling the boundary delivered 15.)
#
# So the refill re-classifies ITS OWN result and widens again
# while the window is short. Bounds, per PROFILE CONTEXT (not per
# sidebar build: all_profiles=1 runs this loader once per context,
# so an N-profile view pays N times everything below):
# * at most UNASSIGNED_CLI_REFILL_MAX_QUERIES queries;
# * none of them reading deeper than the scan ceiling;
# * and it stops early the moment the window is full or the
# filter has demonstrably nothing deeper to give.
# Guarantee: the window is filled whenever at least
# ``unassigned_target`` of the newest ceiling
# state.db-unassigned conversations are still unassigned once
# their sidecars are read — the last query reads exactly that
# band. Past that the sidebar is honestly short (a shortfall the
# database itself imposes) instead of unbounded work.
#
# What that costs, stated at its WORST rather than its best. The
# window this loop must deliver is 20 rows; the best case is a
# single query carrying 24 of them, and quoting that as the
# trade-off understates it by an order of magnitude:
# * PAYLOAD. Every refill row is merged into the projection and
# travels on, so the surplus is trimmed by the sidebar cap
# only AFTER it has been read, sidecar-stat()ed and (on
# /api/sessions/gateway/stream, which snapshots the uncapped
# list) serialised. A shape that keeps the window short
# through all three estimates reaches the forced ceiling
# query, and the model then carries the whole band — up to
# scan_ceiling (200) conversations, nearly all of them
# unassigned, to hand the sidebar 20. The widening below is
# what keeps ordinary shapes off that path; it does not, and
# cannot, remove it, because the ceiling read is exactly what
# the guarantee above is made of.
# * I/O. "No query deeper than 200 conversations" is a LOGICAL
# depth. read_importable_agent_session_rows oversamples raw
# rows by CANDIDATE_WINDOW_MULTIPLIERS = (8, 32) to survive
# compression segments, so a ceiling-width query touches
# 200 * 8 = 1600 raw rows, and 200 * 32 = 6400 when the first
# window is consumed and re-widened. Per profile context.
# Cheaper than any of it: project-filtered server pagination, or
# recording WebUI-side moves in state.db so the SQL filter stops
# lying to this pass.
query_limit = unassigned_target + sidecar_only_assigned
# A caller-supplied window wider than the ceiling still gets its
# first query at full width — the ceiling bounds the WIDENING,
# it does not shrink the requested window.
scan_ceiling = max(UNASSIGNED_CLI_REFILL_SCAN_CEILING, query_limit)
queries_left = UNASSIGNED_CLI_REFILL_MAX_QUERIES
while queries_left > 0:
queries_left -= 1
if queries_left == 0:
# Out of estimates: pay one query at the ceiling rather
# than hand back a short window after guessing three
# times. This is the only query allowed to over-read, and
# the ceiling is what bounds it.
query_limit = scan_ceiling
refill_rows, refill_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='unassigned',
return_window_exhaustion=True,
),
)
for row in refill_rows:
_merge_state_row(row)
unassigned_seen = _count_unassigned()
shortfall = unassigned_target - unassigned_seen
if (
shortfall <= 0
or query_limit >= scan_ceiling
# A short result from an UNFILLED raw candidate window has
# already seen every conversation this filter can reach,
# so a wider query would re-read the same rows for
# nothing. A short result whose raw window WAS consumed
# (by compression segments) is the opposite: widening the
# limit re-widens that window, so it is worth another go.
or (
len(refill_rows) < query_limit
and not refill_window_exhausted
)
):
break
# Widen GEOMETRICALLY, not by one row at a time. Up to three
# candidate widths; the widest of the ones that apply wins:
# * query_limit + shortfall — the exact width needed if
# every conversation deeper than the current one is
# unassigned. It is the FLOOR: always strictly wider, so
# the loop cannot stall, and it is exact (no over-read)
# when the moved conversations are clustered.
# * query_limit + query_limit // queries_left — GEOMETRIC
# growth, paced by the budget that is left. The floor alone
# advances by only the shortfall, so a handful of moves
# clustered just below the window edge crawls (21, 23, 25)
# until every estimate is spent and the forced ceiling read
# does the job anyway: 200 rows fetched to deliver 20. The
# step grows as the budget shrinks (width/3, then width/2)
# because a too-narrow LAST estimate costs the whole ceiling
# read, while a too-wide one costs only its own surplus.
# Those eight clustered moves now finish at (21, 28).
# * ceil(target * query_limit / unassigned_seen) — the width
# implied by the moved SHARE observed so far, for moves
# spread through the history. Without it, a database where
# every other conversation was moved would close the gap
# by halves and need ~log2(target) queries.
# That last term applies ONLY when this window found at least
# one still-unassigned conversation. With zero, the share is
# undefined, and guarding the divisor with max(seen, 1)
# evaluated it as target * query_limit (20 * 40 = 800, clamped
# to the ceiling): every profile that had moved >= 21 of its
# newest conversations jumped straight to a full 200-row read
# where the 60-row floor would have filled the window. Zero
# unassigned rows is not evidence about density, so the
# geometric step — not a degenerate ratio — is what carries it.
# The price of escalating smoothly is that a profile whose
# newest ceiling band is ENTIRELY moved now spends its whole
# budget (40, 60, 90, 200) climbing to the ceiling it was
# always going to need, instead of two queries. That is the
# deliberate side of the trade: an extra narrow query costs SQL
# only, while an early-and-wrong wide query costs payload — and
# payload is read, sidecar-stat()ed and serialised. And a
# cluster deeper than the last estimate can reach — measured at
# 23+ moves sitting just under the window edge, where the three
# estimates are 21, 28, 42 — still ends on the ceiling read, as
# it did before: that is the guarantee doing its job, not the
# estimate failing.
next_query_limit = max(
query_limit + shortfall,
query_limit + query_limit // max(queries_left, 1),
)
if unassigned_seen > 0:
next_query_limit = max(
next_query_limit,
-( # ceil(target * width / unassigned)
-unassigned_target * query_limit // unassigned_seen
),
)
query_limit = min(scan_ceiling, next_query_limit)
except Exception:
logger.debug("Unassigned CLI refill pass failed", exc_info=True)
for row in state_rows:
sid = row['id']
raw_ts = row['last_activity'] or row['started_at']
# Prefer the CLI session's own profile from the DB; fall back to
@@ -8398,7 +9135,7 @@ def _load_cli_sessions_uncached(
'updated_at': raw_ts,
'pinned': False,
'archived': _archived,
'project_id': _state_row_project_id(sid, _source),
'project_id': _interactive_row_project_id(row),
'profile': profile,
'source_tag': _source,
'raw_source': row.get('raw_source') or _source_meta.get('raw_source'),
@@ -8696,8 +9433,17 @@ def get_cli_sessions(
merged: list[dict] = []
for idx, (ctx_home, ctx_db_path, ctx_profile) in enumerate(contexts):
load_kwargs = {
# NOTE: visible_session_limit=None is NOT "unbounded" for the
# interactive pass — it resolves to CLI_VISIBLE_SESSION_LIMIT
# above, so this view truncates assigned conversations by
# recency exactly like the single-profile one and needs the
# same recovery passes. project_assigned_limit therefore keeps
# its default per-project bound here. Only the three limits
# below are handed straight to the reader as ``limit=``, where
# None really does mean unbounded.
'source_filter': source_filter,
'visible_session_limit': None,
'project_assigned_limit': PROJECT_ASSIGNED_CLI_LIMIT,
'cron_project_limit': None,
'webhook_project_limit': None,
'kanban_project_limit': None,
@@ -8713,7 +9459,7 @@ def get_cli_sessions(
)
)
return merged
load_kwargs = {'source_filter': source_filter}
load_kwargs: dict = {'source_filter': source_filter}
if loader_supports_include_claude_code:
load_kwargs['include_claude_code'] = include_claude_code
return _load_cli_sessions_uncached(
api/routes.py
+153 -8
View File
@@ -10488,19 +10488,164 @@ 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."""
# Bound on project-assigned CLI rows in the FINAL MERGED payload, across EVERY
# project. 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).
#
# It has to bound the MERGED set, not one project: 200 rows x N projects grows
# with the project count, and 1,000 assigned conversations spread over 5 projects
# still returned all 1,000 — the exact reproduction from that finding. Pinned
# equal to models.PROJECT_ASSIGNED_CLI_LIMIT (an independent literal on the
# model side; the test is what keeps the two in lockstep) by
# test_route_merged_assigned_cap_is_the_existing_model_row_cap.
CLI_PROJECT_ASSIGNED_CAP = 200
def _draw_assigned_cli_rows_fairly(
rows_by_project: dict[str, list[int]], budget: int
) -> set[int]:
"""Pick ``budget`` assigned row indices, spread fairly across the projects.
``rows_by_project`` maps project id -> that project's row indices, newest
first, keyed in order of each project's most recent assigned conversation
(``sessions`` is newest-first, so insertion order already is that order).
Each round hands one slot to every project that still has history left, so:
* the drawn set never exceeds ``budget`` — that is the whole point, a
per-project bound does not bound the payload (#6659 review finding 1);
* no single busy project can eat every slot, which a flat ``sessions[:200]``
truncation would do to whichever project sorts first — the starvation
greptile rejected as P1 on #6659;
* every project keeps at least one row whenever
``budget >= len(rows_by_project)``. Past that the bound wins: with more
assigned projects than slots, the ``budget`` most recently active projects
get one row each, because the review's number is the hard constraint.
Within a project the draw is newest-first, so what a chip loses is always the
oldest end of its own history.
"""
drawn: set[int] = set()
if budget <= 0 or not rows_by_project:
return drawn
queues = list(rows_by_project.values())
offsets = [0] * len(queues)
remaining = budget
while remaining > 0:
progressed = False
for position, project_rows in enumerate(queues):
offset = offsets[position]
if offset >= len(project_rows):
continue
drawn.add(project_rows[offset])
offsets[position] = offset + 1
remaining -= 1
progressed = True
if remaining <= 0:
break
if not progressed:
# Every project is exhausted — the whole assigned set fits.
break
return drawn
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 IN TOTAL, across every project, stay
in the payload so the project chips 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.
That assigned budget is spent by a fair round-robin draw across the projects
(see ``_draw_assigned_cli_rows_fairly``) instead of by truncating the merged
list, so bounding the payload cannot starve a quiet project (greptile P1 on
#6659). ``project_cap <= 0`` disables the assigned bound entirely.
"""
if cli_cap is None:
cli_cap = _cli_visible_session_cap()
if cli_cap <= 0:
return sessions
kept = []
cli_seen = 0
for session in sessions:
if _is_cli_session_for_settings(session):
cli_seen += 1
if cli_seen > cli_cap:
# Group the assigned rows per project first: the draw has to weigh the
# projects against each other, which a single forward pass cannot do.
rows_by_project: dict[str, list[int]] = {}
for index, session in enumerate(sessions):
if not _is_cli_session_for_settings(session):
continue
project_id = str(session.get("project_id") or "").strip()
if project_id:
rows_by_project.setdefault(project_id, []).append(index)
if project_cap > 0 and rows_by_project:
# Reserve the CLI rows the recent window already shows — the first
# ``cli_cap`` CLI rows in sort order, assigned or not — before the fair
# draw spreads the REST of the assigned budget across projects. Without
# this, a project holding all the newest sessions can lose its newest
# rows in the draw and the payload drops sessions the base displays
# (2026-09-24 re-gate reproduction: 11 projects x 20 sessions, all 20
# newest in one project, ``p0-19`` vanished from the payload).
reserved: set[int] = set()
seen_cli = 0
for index, session in enumerate(sessions):
if not _is_cli_session_for_settings(session):
continue
if seen_cli >= cli_cap:
break
seen_cli += 1
reserved.add(index)
# The reserved rows have already been paid for by the recent window;
# draw the remaining budget over the queues MINUS those rows, so the
# reservation cannot double-spend slots the draw would have granted.
remaining_rows_by_project: dict[str, list[int]] = {
project: [index for index in indices if index not in reserved]
for project, indices in rows_by_project.items()
}
assigned_reserved = sum(
1 for index in reserved
if str(sessions[index].get("project_id") or "").strip()
)
drawn = reserved | _draw_assigned_cli_rows_fairly(
remaining_rows_by_project, max(project_cap - assigned_reserved, 0)
)
else:
drawn = None if project_cap <= 0 else set()
kept = []
recent_seen = 0
unassigned_seen = 0
for index, session in enumerate(sessions):
if _is_cli_session_for_settings(session):
project_id = str(session.get("project_id") or "").strip()
if not project_id:
unassigned_seen += 1
if unassigned_seen > cli_cap:
continue
recent_seen += 1
else:
if drawn is not None and index not in drawn:
continue
if recent_seen >= cli_cap:
session = dict(session)
session["default_hidden"] = True
else:
recent_seen += 1
kept.append(session)
return kept
tests/test_issue2628_cli_sessions_perf.py
+116 -31
View File
@@ -239,37 +239,6 @@ def test_importable_agent_rows_push_sidebar_limit_into_sql(tmp_path):
assert cap == 160, candidates_sql
def test_importable_agent_rows_candidate_window_overfilters_then_fills(tmp_path):
"""Rows dropped after projection must not shrink the returned page.
The SQL candidate window is oversampled (limit * 8) precisely so sessions
filtered client-side of the SQL (invisible CLI rows here) only consume
candidates, not result slots: a page that would be short if the cap were
the plain ``limit`` still fills to ``limit`` visible rows.
"""
db = tmp_path / "state.db"
_make_state_db(db, sessions=120, messages_per_session=5)
# 30 newest sessions are CLI rows that the projection hides (ended, default
# title, no user turn visible): they must fall out of the window without
# starving the visible page below them.
conn = sqlite3.connect(str(db))
for i in range(90, 120):
conn.execute(
"UPDATE sessions SET title = 'cli session', ended_at = 1.0, end_reason = 'timeout' WHERE id = ?",
(f"cli_perf_{i:04d}",),
)
conn.commit()
conn.close()
rows = agent_sessions.read_importable_agent_session_rows(db, limit=20, exclude_sources=("webui",))
assert len(rows) == 20
assert rows[0]["id"] == "cli_perf_0089"
assert rows[-1]["id"] == "cli_perf_0070"
assert {row["actual_message_count"] for row in rows} == {5}
def test_importable_agent_rows_candidate_ordering_stays_under_progress_budget(tmp_path, monkeypatch):
"""Cron-only missing-index scans should fail under the old shape budget, then pass after pre-aggregation."""
db = tmp_path / "state.db"
@@ -373,3 +342,119 @@ def test_importable_agent_rows_zero_limit_skips_query_work(tmp_path):
_make_state_db(db, sessions=5, messages_per_session=1)
assert agent_sessions.read_importable_agent_session_rows(db, limit=0, exclude_sources=("webui",)) == []
def _lineage_and_subagent_db(path):
"""Two 12-segment compression lineages own the 24 newest raw rows.
They collapse to 2 logical conversations, so a ``limit=3`` request consumes
the whole ``limit * 8`` window and still under-delivers. A hot subagent leaf
and its quiet subagent parent sit BELOW that window, reachable only once
``CANDIDATE_WINDOW_MULTIPLIERS`` re-widens it.
"""
conn = sqlite3.connect(str(path))
conn.executescript(
"""
CREATE TABLE sessions (
id TEXT PRIMARY KEY,
source TEXT,
title TEXT,
model TEXT,
started_at REAL NOT NULL,
message_count INTEGER DEFAULT 0,
parent_session_id TEXT,
ended_at REAL,
end_reason TEXT
);
CREATE TABLE messages (
id TEXT PRIMARY KEY,
session_id TEXT,
role TEXT,
content TEXT,
timestamp REAL
);
"""
)
def add(sid, title, started, source, parent=None, ended_at=None, end_reason=None):
conn.execute(
"INSERT INTO sessions (id, source, title, model, started_at, message_count, "
"parent_session_id, ended_at, end_reason) VALUES (?, ?, ?, 'openai/gpt-5', ?, 2, ?, ?, ?)",
(sid, source, title, started, parent, ended_at, end_reason),
)
for index, role in enumerate(("user", "assistant")):
conn.execute(
"INSERT INTO messages (id, session_id, role, content, timestamp) VALUES (?, ?, ?, 'hi', ?)",
(f"{sid}_m{index}", sid, role, started + (index + 4) / 10),
)
for name, base in (("a", 9000.0), ("b", 8000.0)):
for index in range(12):
last = index == 11
started = base + index
add(
f"lin{name}_seg{index}",
f"Lineage {name} part {index}",
started,
"cli",
parent=None if index == 0 else f"lin{name}_seg{index - 1}",
ended_at=None if last else started + 0.9,
end_reason=None if last else "compression",
)
add("hot_leaf", "Hot leaf", 7000.0, "subagent", parent="quiet_orch")
add("quiet_orch", "Quiet orchestrator", 100.0, "subagent")
conn.commit()
conn.close()
def test_rewidened_candidate_window_still_recovers_subagent_parents(tmp_path):
"""The parent-recovery walk must run on the window that ACTUALLY executed.
``read_importable_agent_session_rows`` re-adds subagent parents that the
oversampled candidate set already projected, so a frozen orchestrator is not
evicted while its streaming leaves stay (upstream #7089 / supersedes #7031).
That walk reads the projection of the candidate window, and #6659 turned the
single window into a re-widening loop. This pins that the walk consumes the
LAST window rather than a stale first pass: on the 8x window neither the leaf
nor its parent is even fetched, and only the widened window can anchor them.
"""
db = tmp_path / "state.db"
_lineage_and_subagent_db(db)
rows = agent_sessions.read_importable_agent_session_rows(db, limit=3, exclude_sources=None)
ids = [row["id"] for row in rows]
# The 8x window (24 raw rows) is exactly the two lineages: under-delivered.
assert ids[:2] == ["lina_seg11", "linb_seg11"]
# The widened window reaches the leaf...
assert "hot_leaf" in ids
# ...and the walk re-adds its quiet parent even though the recency slice
# (limit=3) had already dropped it, so the leaf still renders as a child.
assert "quiet_orch" in ids
assert next(row for row in rows if row["id"] == "hot_leaf")["parent_session_id"] == "quiet_orch"
# Contract from the docstring: the anchor pushes the result past ``limit``.
assert len(rows) == 4
def test_first_candidate_window_alone_cannot_anchor_the_subagent_parent(tmp_path):
"""Negative control: without the re-widening pass the same db under-delivers.
Pins that the previous test is not vacuous — the 8x window really does stop
at the two lineages, so the parent recovery above is attributable to the
widened window and not to the first one.
"""
db = tmp_path / "state.db"
_lineage_and_subagent_db(db)
single_window = agent_sessions.CANDIDATE_WINDOW_MULTIPLIERS[:1]
original = agent_sessions.CANDIDATE_WINDOW_MULTIPLIERS
agent_sessions.CANDIDATE_WINDOW_MULTIPLIERS = single_window
try:
rows = agent_sessions.read_importable_agent_session_rows(db, limit=3, exclude_sources=None)
finally:
agent_sessions.CANDIDATE_WINDOW_MULTIPLIERS = original
ids = [row["id"] for row in rows]
assert ids == ["lina_seg11", "linb_seg11"]
assert "hot_leaf" not in ids
assert "quiet_orch" not in ids
tests/test_issue4842_cron_projection_perf.py
+1 -1
View File
@@ -201,7 +201,7 @@ def test_missing_sidecar_returns_default_without_caching_growth(tmp_path):
meta = models._state_projection_sidecar_metadata("cron_nope_999")
# No sidecar means no opinion on archived (None), so the projection falls back
# to the state.db row's archived flag (#7548) instead of forcing it to False.
assert meta == {"title": None, "archived": None}
assert meta == {"title": None, "archived": None, "project_id": None}
# No file → nothing cached (so the cache can't be poisoned by absent files).
assert len(models._SIDECAR_METADATA_CACHE) == 0
tests/test_project_assigned_cli_session_limit.py
Executable
+2641
View File
File diff suppressed because it is too large. Load diff