From 7a596f9114e3f067c9bc840a40c309591875c5dc Mon Sep 17 00:00:00 2001 From: Rodrigo Gomes da Silva Date: Sat, 15 Aug 2026 15:22:09 -0300 Subject: [PATCH] fix(sessions): stop a busy project starving quieter ones in the assigned pass The per-project budget was applied to the rows of ONE global recency-ordered query, so it was not a per-project bound: when a single project owns the whole `limit * len(projects)` window, every other project's assigned conversations are never considered and their chips show nothing (greptile P1 on #6659, reproduced at the real 200 as well as scaled down). Two tiers now: * the global query still runs first and is the only cost a profile whose assigned history fits in the window pays; * when it comes back SATURATED, one GROUP BY over sessions.project_id says which projects still own undelivered rows, and each of those gets its own project-scoped query with its own remaining budget. `project_ids` narrows the lineage CTE's seed, so a per-project query is cheaper than the global one and still returns whole compression lineages. Cost stays bounded: worst case per build is 1 global query + 1 GROUP BY probe + PROJECT_ASSIGNED_CLI_REFILL_PROJECTS (32) project-scoped queries, each bounded to one project's remaining budget. The scan ceiling is now shared equally between projects (ceiling // len(projects)) rather than first-come, so the recovered payload stays under the ceiling without letting the loudest project take it all. --- api/agent_sessions.py | 109 +++++- api/models.py | 131 +++++-- ...test_project_assigned_cli_session_limit.py | 352 ++++++++++++++++++ 3 files changed, 563 insertions(+), 29 deletions(-) diff --git a/api/agent_sessions.py b/api/agent_sessions.py index 72fa8340..f560f392 100644 --- a/api/agent_sessions.py +++ b/api/agent_sessions.py @@ -684,6 +684,7 @@ def read_importable_agent_session_rows( exclude_sources: tuple[str, ...] | None = ("cron", "webui"), include_sources: tuple[str, ...] | None = None, project_assignment: str | None = None, + project_ids: tuple[str, ...] | None = None, ) -> list[dict]: """Return agent sessions projected as importable conversations. @@ -707,6 +708,15 @@ def read_importable_agent_session_rows( 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 @@ -726,6 +736,19 @@ def read_importable_agent_session_rows( children render top-level, exactly as they did before. Widening that window is a ``CANDIDATE_WINDOW_MULTIPLIERS`` change, not a change to this walk. """ + 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 [] db_path = Path(db_path) if not db_path.exists(): return [] @@ -863,6 +886,16 @@ def read_importable_agent_session_rows( # 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} ( @@ -870,7 +903,7 @@ def read_importable_agent_session_rows( SELECT seed.id FROM sessions seed WHERE seed.project_id IS NOT NULL - AND TRIM(seed.project_id) != '' + AND TRIM(seed.project_id) != ''{seed_project_filter} UNION SELECT child.id FROM sessions child @@ -888,10 +921,18 @@ def read_importable_agent_session_rows( ) """ ) + params.extend(wanted_project_ids) elif wanted == 'assigned': - where_clauses.append( - "s.project_id IS NOT NULL AND TRIM(s.project_id) != ''" - ) + 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) = '')" @@ -1082,6 +1123,66 @@ def read_importable_agent_session_rows( return _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 + ) + if excluded: + placeholders = ", ".join("?" for _ in excluded) + where_clauses.append(f"s.source NOT IN ({placeholders})") + params.extend(excluded) + cur.execute( + f""" + SELECT TRIM(s.project_id) AS project_id, COUNT(*) AS row_count + FROM sessions s + WHERE {' AND '.join(where_clauses)} + GROUP BY TRIM(s.project_id) + """, + params, + ) + 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: updated_at = row.get('ended_at') if row.get('ended_at') is not None else row.get('started_at') diff --git a/api/models.py b/api/models.py index 985caaca..f5bfb397 100644 --- a/api/models.py +++ b/api/models.py @@ -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, ) @@ -64,9 +65,17 @@ CLI_VISIBLE_SESSION_LIMIT = _cfg.CLI_VISIBLE_SESSION_LIMIT 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. +# 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 +# How many project-scoped follow-up queries one sidebar build may pay after the +# global assigned window came back saturated. The follow-up is what keeps a busy +# project from starving quieter ones (greptile P1 on #6659); the cap is what +# keeps the worst case bounded at 1 global query + 1 GROUP BY probe + this many +# project-scoped queries, each itself bounded to one project's remaining budget. +PROJECT_ASSIGNED_CLI_REFILL_PROJECTS = 32 # 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 @@ -8530,20 +8539,42 @@ def _load_cli_sessions_uncached( # 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 + + # PROJECT_ASSIGNED_CLI_REFILL_PROJECTS project-scoped queries, each + # bounded to one project's remaining budget. 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)) ) - query_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(len(known_project_ids), 1), + 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, ) ) @@ -8559,32 +8590,82 @@ def _load_cli_sessions_uncached( kept_per_project[seeded_project_id] = ( kept_per_project.get(seeded_project_id, 0) + 1 ) - for row in read_importable_agent_session_rows( + + 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 = 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 - and kept_per_project.get(project_id, 0) >= per_project_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 - ) + ) + _spend_assigned_rows(global_rows) + + # When that window came back FULL it truncated by recency, so a + # single busy project can have consumed all of it and left + # quieter projects with nothing — their chips would show an empty + # project even though the rows exist (greptile P1 on #6659). A + # short window cannot hide anything: it already saw every + # assigned conversation the db has, so no follow-up is paid. + if effective_limit and query_limit and len(global_rows) >= query_limit: + # 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), and capped: a profile with hundreds + # of simultaneously starved projects must not turn one sidebar + # build into hundreds of scans. + for _kept, project_id in starved[:PROJECT_ASSIGNED_CLI_REFILL_PROJECTS]: + remaining = effective_limit - kept_per_project.get(project_id, 0) + if remaining <= 0: + continue + _spend_assigned_rows(read_importable_agent_session_rows( + db_path, + limit=remaining, + log=logger, + exclude_sources=interactive_excluded, + project_assignment='assigned', + project_ids=(project_id,), + )) except Exception: logger.debug("Project-assigned CLI recovery pass failed", exc_info=True) diff --git a/tests/test_project_assigned_cli_session_limit.py b/tests/test_project_assigned_cli_session_limit.py index 2edc0fab..c99cba68 100755 --- a/tests/test_project_assigned_cli_session_limit.py +++ b/tests/test_project_assigned_cli_session_limit.py @@ -7,6 +7,9 @@ swing to the opposite failure, so these tests pin BOTH edges of the contract: * an assigned conversation older than the recent window is still reachable (the original bug), including past the 200-conversation mark when the assignments are spread over several projects; +* one busy project cannot starve the others: a per-project budget applied to a + single global recency window is not a per-project bound, so a saturated window + is followed by project-scoped, per-project-budgeted queries (greptile P1); * the recovered set is BOUNDED — per project, counting logical conversations — including assigned rows that arrive as imported WebUI sidecars and therefore never pass through any state.db cap; @@ -390,6 +393,355 @@ def test_model_bounds_a_skewed_project_inside_a_wider_scan_window( } +# ── greptile P1: a busy project must not starve the quieter ones ────────────── + + +def _assigned_counts(sessions): + counts: dict[str, int] = {} + for session in sessions: + if session["project_id"]: + counts[session["project_id"]] = counts.get(session["project_id"], 0) + 1 + return counts + + +def test_busy_project_does_not_starve_quieter_projects( + fake_hermes_home, tmp_path, monkeypatch +): + """The per-project BUDGET is not enough while the QUERY is one global window. + + The recovery pass asked for ``limit * len(projects)`` newest assigned + conversations in a single recency-ordered query and only then applied the + per-project budget. When one project owns that whole window, every other + project's assigned conversations are never even considered: their chips show + nothing at all. Reproduced with a scaled budget (3) so the fixture stays + small; ``test_busy_project_starvation_at_production_limits`` pins the same + behaviour at the real 200. + """ + monkeypatch.setattr(models, "CLI_VISIBLE_SESSION_LIMIT", 5) + _register_projects(tmp_path, "project-busy", "project-quiet-a", "project-quiet-b") + + rows = [ + _session("quiet-a-old", BASE_TS, project_id="project-quiet-a"), + _session("quiet-b-old", BASE_TS + 1, project_id="project-quiet-b"), + ] + # 12 newer assigned conversations in one project fill the 3 * 3 = 9 window. + rows.extend( + _session(f"busy-{index:03d}", BASE_TS + 100 + index, project_id="project-busy") + for index in range(12) + ) + rows.extend(_session(f"plain-{index:02d}", BASE_TS + 500 + index) for index in range(25)) + _write_state_db(fake_hermes_home / "state.db", rows) + + sessions = models._load_cli_sessions_uncached( + fake_hermes_home, + fake_hermes_home / "state.db", + "default", + project_assigned_limit=3, + ) + + # Was {'project-busy': 3}: both quiet projects were completely absent. + assert _assigned_counts(sessions) == { + "project-busy": 3, + "project-quiet-a": 1, + "project-quiet-b": 1, + } + # The busy project still gets its newest, and is still bounded. + assert {s["session_id"] for s in sessions if s["project_id"] == "project-busy"} == { + "busy-011", "busy-010", "busy-009", + } + + +def test_busy_project_starvation_at_production_limits(fake_hermes_home, tmp_path): + """The same starvation with the real PROJECT_ASSIGNED_CLI_LIMIT (200). + + Two registered projects give the old global query a 400-conversation window; + 400 newer conversations in one project consumed all of it, so the quiet + project's single assigned conversation was unreachable. + """ + _register_projects(tmp_path, "project-busy", "project-quiet") + + rows = [_session("quiet-old", BASE_TS, project_id="project-quiet")] + rows.extend( + _session(f"busy-{index:04d}", BASE_TS + 100 + index, project_id="project-busy") + for index in range(models.PROJECT_ASSIGNED_CLI_LIMIT * 2) + ) + rows.extend(_session(f"plain-{index:02d}", BASE_TS + 5000 + index) for index in range(25)) + _write_state_db(fake_hermes_home / "state.db", rows) + + counts = _assigned_counts(models.get_cli_sessions()) + + # Was {'project-busy': 200}. + assert counts == { + "project-busy": models.PROJECT_ASSIGNED_CLI_LIMIT, + "project-quiet": 1, + } + + +def test_starved_project_recovers_its_whole_compression_lineage( + fake_hermes_home, tmp_path, monkeypatch +): + """The project-scoped follow-up query is still lineage-keyed. + + ``project_ids`` narrows only the recursive CTE's SEED, so a starved project's + conversation still arrives as ONE row even when its assignment sits on the + root of a compression chain. + """ + monkeypatch.setattr(models, "CLI_VISIBLE_SESSION_LIMIT", 5) + _register_projects(tmp_path, "project-busy", "project-quiet") + + rows = _lineage("quiet-chain", BASE_TS, 3, project_id="project-quiet") + rows.extend( + _session(f"busy-{index:03d}", BASE_TS + 100 + index, project_id="project-busy") + for index in range(12) + ) + rows.extend(_session(f"plain-{index:02d}", BASE_TS + 500 + index) for index in range(25)) + _write_state_db(fake_hermes_home / "state.db", rows) + + sessions = models._load_cli_sessions_uncached( + fake_hermes_home, + fake_hermes_home / "state.db", + "default", + project_assigned_limit=2, + ) + + quiet = [s for s in sessions if s["project_id"] == "project-quiet"] + assert len(quiet) == 1, [s["session_id"] for s in quiet] + assert quiet[0]["session_id"] in {f"quiet-chain-seg{index}" for index in range(3)} + + +def test_no_followup_queries_when_the_global_window_is_not_saturated( + fake_hermes_home, tmp_path, monkeypatch +): + """A short global window already saw everything, so it must cost 1 query. + + The starvation follow-up is gated on saturation. Without that gate every + sidebar build on every profile with projects would pay a GROUP BY probe plus + a query per project. + """ + _register_projects(tmp_path, "project-a", "project-b", "project-c") + + rows = [ + _session("a-1", BASE_TS, project_id="project-a"), + _session("b-1", BASE_TS + 1, project_id="project-b"), + ] + rows.extend(_session(f"plain-{index:02d}", BASE_TS + 500 + index) for index in range(25)) + _write_state_db(fake_hermes_home / "state.db", rows) + + assigned_queries = [] + real_reader = models.read_importable_agent_session_rows + + def _counting_reader(*args, **kwargs): + if kwargs.get("project_assignment") == "assigned": + assigned_queries.append(kwargs.get("project_ids")) + return real_reader(*args, **kwargs) + + probes = [] + real_probe = models.read_assigned_project_row_counts + + def _counting_probe(*args, **kwargs): + probes.append(args) + return real_probe(*args, **kwargs) + + monkeypatch.setattr(models, "read_importable_agent_session_rows", _counting_reader) + monkeypatch.setattr(models, "read_assigned_project_row_counts", _counting_probe) + + sessions = models._load_cli_sessions_uncached( + fake_hermes_home, + fake_hermes_home / "state.db", + "default", + project_assigned_limit=3, + ) + + assert _assigned_counts(sessions) == {"project-a": 1, "project-b": 1} + assert assigned_queries == [None], "one global assigned query, no per-project ones" + assert probes == [], "the GROUP BY probe is only paid on a saturated window" + + +def test_followup_queries_are_capped_and_serve_the_neediest_first( + fake_hermes_home, tmp_path, monkeypatch +): + """The follow-up is bounded: at most PROJECT_ASSIGNED_CLI_REFILL_PROJECTS. + + With the cap forced to 1 and two equally starved projects, the single + follow-up goes to the lower-numbered project id (the deterministic + neediest-first order) and the other keeps the pre-fix behaviour. This pins + the documented residual limitation instead of pretending it is absent. + """ + monkeypatch.setattr(models, "CLI_VISIBLE_SESSION_LIMIT", 5) + monkeypatch.setattr(models, "PROJECT_ASSIGNED_CLI_REFILL_PROJECTS", 1) + _register_projects(tmp_path, "project-busy", "project-quiet-a", "project-quiet-b") + + rows = [ + _session("quiet-a-old", BASE_TS, project_id="project-quiet-a"), + _session("quiet-b-old", BASE_TS + 1, project_id="project-quiet-b"), + ] + rows.extend( + _session(f"busy-{index:03d}", BASE_TS + 100 + index, project_id="project-busy") + for index in range(12) + ) + rows.extend(_session(f"plain-{index:02d}", BASE_TS + 500 + index) for index in range(25)) + _write_state_db(fake_hermes_home / "state.db", rows) + + sessions = models._load_cli_sessions_uncached( + fake_hermes_home, + fake_hermes_home / "state.db", + "default", + project_assigned_limit=3, + ) + + assert _assigned_counts(sessions) == { + "project-busy": 3, + "project-quiet-a": 1, + } + + +def test_scan_ceiling_is_shared_equally_between_projects( + fake_hermes_home, tmp_path, monkeypatch +): + """Past the ceiling every project's share shrinks; none is left with zero. + + Ceiling 6 over 3 projects is a share of 2 even though the requested budget is + 5, so the recovered payload stays under the ceiling AND every project is + represented — the old code let the newest project take 5 of the 6. + """ + monkeypatch.setattr(models, "CLI_VISIBLE_SESSION_LIMIT", 5) + monkeypatch.setattr(models, "PROJECT_ASSIGNED_CLI_SCAN_CEILING", 6) + _register_projects(tmp_path, "project-a", "project-b", "project-c") + + rows = [] + for offset, project in enumerate(("project-a", "project-b", "project-c")): + rows.extend( + _session( + f"{project}-{index}", + BASE_TS + offset * 100 + index, + project_id=project, + ) + for index in range(4) + ) + rows.extend(_session(f"plain-{index:02d}", BASE_TS + 900 + index) for index in range(25)) + _write_state_db(fake_hermes_home / "state.db", rows) + + sessions = models._load_cli_sessions_uncached( + fake_hermes_home, + fake_hermes_home / "state.db", + "default", + project_assigned_limit=5, + ) + + counts = _assigned_counts(sessions) + assert counts == {"project-a": 2, "project-b": 2, "project-c": 2} + assert sum(counts.values()) <= models.PROJECT_ASSIGNED_CLI_SCAN_CEILING + + +# ── The reader-level contract the per-project follow-up relies on ───────────── + + +def test_project_ids_narrows_the_assigned_query_to_one_project(tmp_path): + """``project_ids`` seeds the lineage CTE with one project's rows only.""" + db_path = tmp_path / "state.db" + rows = [ + _session("a-1", BASE_TS, project_id="project-a"), + _session("b-1", BASE_TS + 1, project_id="project-b"), + _session("plain", BASE_TS + 2), + ] + rows.extend(_lineage("b-chain", BASE_TS + 10, 3, project_id="project-b")) + _write_state_db(db_path, rows) + + def _ids(**kwargs): + return sorted( + row["id"] + for row in agent_sessions.read_importable_agent_session_rows( + db_path, limit=50, exclude_sources=None, **kwargs + ) + ) + + assert _ids(project_assignment="assigned") == ["a-1", "b-1", "b-chain-seg2"] + assert _ids(project_assignment="assigned", project_ids=("project-b",)) == [ + "b-1", "b-chain-seg2", + ] + assert _ids(project_assignment="assigned", project_ids=("project-a",)) == ["a-1"] + # Unknown ids are not an error, they simply select nothing. + assert _ids(project_assignment="assigned", project_ids=("nope",)) == [] + + +def test_project_ids_requires_the_assigned_filter(tmp_path): + """A silently unnarrowed query is exactly the starvable one — refuse it.""" + db_path = tmp_path / "state.db" + _write_state_db(db_path, [_session("a-1", BASE_TS, project_id="project-a")]) + + for assignment in (None, "unassigned"): + with pytest.raises(ValueError, match="project_ids requires"): + agent_sessions.read_importable_agent_session_rows( + db_path, project_assignment=assignment, project_ids=("project-a",) + ) + + +def test_empty_project_ids_selects_nothing(tmp_path): + """``project_ids=()`` must not degrade to "every assigned conversation".""" + db_path = tmp_path / "state.db" + _write_state_db(db_path, [_session("a-1", BASE_TS, project_id="project-a")]) + + assert agent_sessions.read_importable_agent_session_rows( + db_path, exclude_sources=None, project_assignment="assigned", project_ids=() + ) == [] + + +def test_project_ids_narrows_on_a_schema_without_lineage_columns(tmp_path): + """The no-parent_session_id schema takes the plain ``project_id IN`` path.""" + db_path = tmp_path / "state.db" + _write_state_db( + db_path, + [ + _session("a-1", BASE_TS, project_id="project-a"), + _session("b-1", BASE_TS + 1, project_id="project-b"), + ], + lineage_columns=False, + ) + + rows = agent_sessions.read_importable_agent_session_rows( + db_path, + limit=50, + exclude_sources=None, + project_assignment="assigned", + project_ids=("project-b",), + ) + + assert [row["id"] for row in rows] == ["b-1"] + + +def test_assigned_project_row_counts_probe(tmp_path): + """The probe counts assigned RAW rows per project and honours exclusions.""" + db_path = tmp_path / "state.db" + rows = [ + _session("a-1", BASE_TS, project_id="project-a"), + _session("a-2", BASE_TS + 1, project_id="project-a"), + _session("b-1", BASE_TS + 2, project_id="project-b"), + _session("b-cron", BASE_TS + 3, project_id="project-b", source="cron"), + _session("plain", BASE_TS + 4), + _session("blank", BASE_TS + 5, project_id=" "), + ] + _write_state_db(db_path, rows) + + assert agent_sessions.read_assigned_project_row_counts(db_path) == { + "project-a": 2, "project-b": 2, + } + assert agent_sessions.read_assigned_project_row_counts( + db_path, exclude_sources=("cron", "webhook", "kanban") + ) == {"project-a": 2, "project-b": 1} + # Missing db and a schema without project_id are answered, not raised. + assert agent_sessions.read_assigned_project_row_counts(tmp_path / "nope.db") == {} + + +def test_assigned_project_row_counts_probe_on_schema_without_project_id(tmp_path): + db_path = tmp_path / "legacy.db" + conn = sqlite3.connect(str(db_path)) + conn.execute("CREATE TABLE sessions (id TEXT PRIMARY KEY, source TEXT)") + conn.commit() + conn.close() + + assert agent_sessions.read_assigned_project_row_counts(db_path) == {} + + # ── Finding 2: assigned rows must not spend the unassigned quota ──────────────