diff --git a/online/etl/db/connection.py b/online/etl/db/connection.py index aab1cfd..d39135b 100644 --- a/online/etl/db/connection.py +++ b/online/etl/db/connection.py @@ -119,19 +119,24 @@ def _translate_params(self, sql: str, args: tuple | list | None) -> tuple[str, t Handles re-used parameters (e.g. $2 appearing twice in ON CONFLICT) by expanding the args list so each ? gets the right positional value. + + Also strips PostgreSQL-style `::type` casts (e.g. `$2::timestamptz`) for + SQLite. The casts are needed in PG to disambiguate NULL parameter types + in expressions like `$2::timestamptz IS NULL OR col >= $2`; SQLite is + dynamically typed and handles NULL comparisons natively. """ if self.is_postgres: return sql, self._coerce_args(args) if args is None: - return sql, args - # Find all $N references in order of appearance - refs = re.findall(r"\$(\d+)", sql) + # Still strip casts so DDL/queries with no params parse cleanly under SQLite. + return re.sub(r"::\w+", "", sql), args + sql_no_casts = re.sub(r"::\w+", "", sql) + refs = re.findall(r"\$(\d+)", sql_no_casts) if not refs: - return sql, args - # Build expanded args list matching each ? to the referenced $N + return sql_no_casts, args args_tuple = tuple(args) expanded = tuple(args_tuple[int(r) - 1] for r in refs) - translated = re.sub(r"\$\d+", "?", sql) + translated = re.sub(r"\$\d+", "?", sql_no_casts) return translated, expanded # -- Query execution ------------------------------------------------------- diff --git a/online/etl/db/queries.py b/online/etl/db/queries.py index 6b63bc0..b807255 100644 --- a/online/etl/db/queries.py +++ b/online/etl/db/queries.py @@ -99,6 +99,59 @@ LIMIT $2 """ +# Sweep-mode variants: same filter as the default queries but ordered by +# assembled_at DESC. Used by --sort sweep to prioritize PRs that recently became +# ready, catching late-discovered PRs that bot_reviewed_at ordering misses. +GET_ASSEMBLED_PRS_NOT_ANALYZED_SWEEP = """ + SELECT p.* FROM prs p + LEFT JOIN llm_analyses la ON la.pr_id = p.id AND la.chatbot_id = p.chatbot_id + WHERE p.chatbot_id = $1 + AND p.status = 'assembled' + AND la.id IS NULL + AND p.pr_merged = TRUE + ORDER BY p.assembled_at DESC NULLS LAST + LIMIT $2 +""" + +GET_ALL_ASSEMBLED_NOT_ANALYZED_SWEEP = """ + SELECT p.* FROM prs p + LEFT JOIN llm_analyses la ON la.pr_id = p.id AND la.chatbot_id = p.chatbot_id + WHERE p.status = 'assembled' + AND la.id IS NULL + AND p.pr_merged = TRUE + ORDER BY p.assembled_at DESC NULLS LAST + LIMIT $1 +""" + +# Bounded variants accept both `since` (inclusive lower bound) and `until` (exclusive +# upper bound). Either may be NULL — the WHERE clauses become no-ops, which Postgres' +# planner folds away. Use these whenever `until` is specified; the *_SINCE variants +# above remain the fast path for the common since-only / no-bound case. +GET_ASSEMBLED_PRS_NOT_ANALYZED_BOUNDED = """ + SELECT p.* FROM prs p + LEFT JOIN llm_analyses la ON la.pr_id = p.id AND la.chatbot_id = p.chatbot_id + WHERE p.chatbot_id = $1 + AND p.status = 'assembled' + AND la.id IS NULL + AND p.pr_merged = TRUE + AND ($2::timestamptz IS NULL OR p.bot_reviewed_at >= $2) + AND ($3::timestamptz IS NULL OR p.bot_reviewed_at < $3) + ORDER BY p.bot_reviewed_at DESC NULLS LAST + LIMIT $4 +""" + +GET_ALL_ASSEMBLED_NOT_ANALYZED_BOUNDED = """ + SELECT p.* FROM prs p + LEFT JOIN llm_analyses la ON la.pr_id = p.id AND la.chatbot_id = p.chatbot_id + WHERE p.status = 'assembled' + AND la.id IS NULL + AND p.pr_merged = TRUE + AND ($1::timestamptz IS NULL OR p.bot_reviewed_at >= $1) + AND ($2::timestamptz IS NULL OR p.bot_reviewed_at < $2) + ORDER BY p.bot_reviewed_at DESC NULLS LAST + LIMIT $3 +""" + # -- PR locking ---------------------------------------------------------------- LOCK_PR = """ @@ -263,6 +316,59 @@ LIMIT $2 """ +# Analyzed-sorted variants for the labeling stage: sort by analyzed_at DESC +# to prioritize PRs that most recently became ready for labeling. +GET_ANALYZED_NOT_LABELED_BY_ASSEMBLED = """ + SELECT p.*, la.bot_suggestions, la.matching_results + FROM prs p + JOIN llm_analyses la ON la.pr_id = p.id AND la.chatbot_id = p.chatbot_id + LEFT JOIN pr_labels pl ON pl.pr_id = p.id AND pl.chatbot_id = p.chatbot_id + WHERE p.chatbot_id = $1 + AND p.status = 'analyzed' + AND pl.id IS NULL + ORDER BY p.analyzed_at DESC NULLS LAST + LIMIT $2 +""" + +GET_ALL_ANALYZED_NOT_LABELED_SWEEP = """ + SELECT p.*, la.bot_suggestions, la.matching_results + FROM prs p + JOIN llm_analyses la ON la.pr_id = p.id AND la.chatbot_id = p.chatbot_id + LEFT JOIN pr_labels pl ON pl.pr_id = p.id AND pl.chatbot_id = p.chatbot_id + WHERE p.status = 'analyzed' + AND pl.id IS NULL + ORDER BY p.analyzed_at DESC NULLS LAST + LIMIT $1 +""" + +# See note above on bounded queries — same pattern, applied to the labeling stage. +GET_ANALYZED_NOT_LABELED_BOUNDED = """ + SELECT p.*, la.bot_suggestions, la.matching_results + FROM prs p + JOIN llm_analyses la ON la.pr_id = p.id AND la.chatbot_id = p.chatbot_id + LEFT JOIN pr_labels pl ON pl.pr_id = p.id AND pl.chatbot_id = p.chatbot_id + WHERE p.chatbot_id = $1 + AND p.status = 'analyzed' + AND pl.id IS NULL + AND ($2::timestamptz IS NULL OR p.bot_reviewed_at >= $2) + AND ($3::timestamptz IS NULL OR p.bot_reviewed_at < $3) + ORDER BY p.bot_reviewed_at DESC NULLS LAST + LIMIT $4 +""" + +GET_ALL_ANALYZED_NOT_LABELED_BOUNDED = """ + SELECT p.*, la.bot_suggestions, la.matching_results + FROM prs p + JOIN llm_analyses la ON la.pr_id = p.id AND la.chatbot_id = p.chatbot_id + LEFT JOIN pr_labels pl ON pl.pr_id = p.id AND pl.chatbot_id = p.chatbot_id + WHERE p.status = 'analyzed' + AND pl.id IS NULL + AND ($1::timestamptz IS NULL OR p.bot_reviewed_at >= $1) + AND ($2::timestamptz IS NULL OR p.bot_reviewed_at < $2) + ORDER BY p.bot_reviewed_at DESC NULLS LAST + LIMIT $3 +""" + # -- PR volumes ---------------------------------------------------------------- UPSERT_PR_VOLUME = """ diff --git a/online/etl/db/repository.py b/online/etl/db/repository.py index cf656a1..62b15b2 100644 --- a/online/etl/db/repository.py +++ b/online/etl/db/repository.py @@ -143,8 +143,33 @@ async def get_pending_prs(self, chatbot_id: int, limit: int = 100) -> list[dict[ return await self.db.fetchall(query, (chatbot_id, limit)) async def get_assembled_not_analyzed( - self, chatbot_id: int | None = None, limit: int = 100, since: str | None = None + self, + chatbot_id: int | None = None, + limit: int = 100, + since: str | None = None, + until: str | None = None, + sort_by: str = "reviewed", ) -> list[dict[str, Any]]: + # `until` is exclusive: --since 2026-04-18 --until 2026-04-19 yields just 4/18. + # When `until` is set we use the bounded variant (which also handles since=None); + # otherwise we keep the existing fast paths to preserve query plans. + # `sort_by` == "sweep" uses assembled_at DESC (for catching late-discovered PRs). + if sort_by == "sweep": + if chatbot_id is not None: + return await self.db.fetchall( + q.GET_ASSEMBLED_PRS_NOT_ANALYZED_SWEEP, (chatbot_id, limit) + ) + return await self.db.fetchall( + q.GET_ALL_ASSEMBLED_NOT_ANALYZED_SWEEP, (limit,) + ) + if until is not None: + if chatbot_id is not None: + return await self.db.fetchall( + q.GET_ASSEMBLED_PRS_NOT_ANALYZED_BOUNDED, (chatbot_id, since, until, limit) + ) + return await self.db.fetchall( + q.GET_ALL_ASSEMBLED_NOT_ANALYZED_BOUNDED, (since, until, limit) + ) if since: if chatbot_id is not None: return await self.db.fetchall(q.GET_ASSEMBLED_PRS_NOT_ANALYZED_SINCE, (chatbot_id, since, limit)) @@ -289,8 +314,31 @@ async def insert_labels( ) async def get_analyzed_not_labeled( - self, chatbot_id: int | None = None, limit: int = 100, since: str | None = None + self, + chatbot_id: int | None = None, + limit: int = 100, + since: str | None = None, + until: str | None = None, + sort_by: str = "reviewed", ) -> list[dict[str, Any]]: + # See note on get_assembled_not_analyzed: `until` is exclusive. + # `sort_by` == "sweep" uses analyzed_at DESC (for catching stragglers). + if sort_by == "sweep": + if chatbot_id is not None: + return await self.db.fetchall( + q.GET_ANALYZED_NOT_LABELED_SWEEP, (chatbot_id, limit) + ) + return await self.db.fetchall( + q.GET_ALL_ANALYZED_NOT_LABELED_SWEEP, (limit,) + ) + if until is not None: + if chatbot_id is not None: + return await self.db.fetchall( + q.GET_ANALYZED_NOT_LABELED_BOUNDED, (chatbot_id, since, until, limit) + ) + return await self.db.fetchall( + q.GET_ALL_ANALYZED_NOT_LABELED_BOUNDED, (since, until, limit) + ) if since: if chatbot_id is not None: return await self.db.fetchall(q.GET_ANALYZED_NOT_LABELED_SINCE, (chatbot_id, since, limit)) diff --git a/online/etl/main.py b/online/etl/main.py index 5aae312..4346aad 100644 --- a/online/etl/main.py +++ b/online/etl/main.py @@ -179,7 +179,20 @@ def build_parser() -> argparse.ArgumentParser: p_ana.add_argument("--chatbot", help="Specific chatbot, or use --all") p_ana.add_argument("--all", action="store_true", dest="all_chatbots") p_ana.add_argument("--limit", type=int, default=100) - p_ana.add_argument("--since", help="Only analyze PRs reviewed since this date (e.g. '7d', '2026-02-05')") + p_ana.add_argument("--since", help="Inclusive lower bound on bot_reviewed_at (e.g. '7d', '2026-02-05')") + p_ana.add_argument( + "--until", + help=( + "Exclusive upper bound on bot_reviewed_at (e.g. '2d', '2026-04-19'). " + "With --since 2026-04-18 --until 2026-04-19 you get just 2026-04-18." + ), + ) + p_ana.add_argument( + "--sort", + choices=["reviewed", "sweep"], + default="reviewed", + help="Sort order: 'reviewed' (default, bot_reviewed_at DESC) or 'sweep' (assembled_at DESC, catches late-merged PRs).", + ) p_ana.add_argument("--database-url") p_ana.add_argument("--verbose", action="store_true") @@ -195,7 +208,17 @@ def build_parser() -> argparse.ArgumentParser: p_lbl.add_argument("--chatbot", help="Specific chatbot, or use --all") p_lbl.add_argument("--all", action="store_true", dest="all_chatbots") p_lbl.add_argument("--limit", type=int, default=100) - p_lbl.add_argument("--since", help="Only label PRs reviewed since this date (e.g. '7d', '2026-02-05')") + p_lbl.add_argument("--since", help="Inclusive lower bound on bot_reviewed_at (e.g. '7d', '2026-02-05')") + p_lbl.add_argument( + "--until", + help="Exclusive upper bound on bot_reviewed_at (e.g. '2d', '2026-04-19')", + ) + p_lbl.add_argument( + "--sort", + choices=["reviewed", "sweep"], + default="reviewed", + help="Sort order: 'reviewed' (bot_reviewed_at DESC, default) or 'sweep' (analyzed_at DESC, for catching stragglers).", + ) p_lbl.add_argument("--database-url") p_lbl.add_argument("--verbose", action="store_true") @@ -407,6 +430,28 @@ async def cmd_enrich(args: argparse.Namespace) -> None: await db.close() +def _parse_time_bound(value: str | None) -> str | None: + """Parse a CLI time bound: relative ("7d") or absolute ("2026-02-05") -> ISO timestamp. + + Returns None when value is falsy. Relative values are anchored to "now" (UTC). + Bare dates ("YYYY-MM-DD") are normalized to midnight UTC so asyncpg can bind + them to a timestamptz column — without this the bare-date form is forwarded + as a raw string and asyncpg raises DataError. + """ + if not value: + return None + from datetime import datetime + from datetime import timedelta + import re + + m = re.match(r"^(\d+)d$", value) + if m: + return (datetime.now(UTC) - timedelta(days=int(m.group(1)))).isoformat() + if re.match(r"^\d{4}-\d{2}-\d{2}$", value): + return f"{value}T00:00:00+00:00" + return value + + async def cmd_analyze(args: argparse.Namespace) -> None: from db.connection import DBAdapter from db.repository import PRRepository @@ -420,16 +465,18 @@ async def cmd_analyze(args: argparse.Namespace) -> None: logger.error("MARTIAN_API_KEY required") return - # Parse --since: supports relative ("7d") or absolute ("2026-02-05") - since = None - if args.since: - from datetime import datetime - from datetime import timedelta - import re - - m = re.match(r"^(\d+)d$", args.since) - since = (datetime.now(UTC) - timedelta(days=int(m.group(1)))).isoformat() if m else args.since - logger.info(f"Filtering PRs reviewed since {since}") + since = _parse_time_bound(args.since) + until = _parse_time_bound(args.until) + sort_by = args.sort + if sort_by == "sweep": + if since or until: + logger.warning("--since/--until are ignored in sweep mode (sweep processes all unanalyzed PRs by assembled_at)") + logger.info("Sweep mode: sorting by assembled_at DESC") + else: + if since: + logger.info(f"Filtering PRs reviewed since {since}") + if until: + logger.info(f"Filtering PRs reviewed before {until} (exclusive)") db = DBAdapter(cfg.database_url) await db.connect() @@ -440,13 +487,21 @@ async def cmd_analyze(args: argparse.Namespace) -> None: if args.all_chatbots: chatbots = await repo.get_all_chatbots() for bot in chatbots: - await analyze_prs(cfg, db, bot["id"], bot["github_username"], limit=args.limit, since=since) + await analyze_prs( + cfg, db, bot["id"], bot["github_username"], + limit=args.limit, since=since, until=until, + sort_by=sort_by, + ) elif args.chatbot: bot = await repo.get_chatbot(args.chatbot) if not bot: logger.error(f"Chatbot '{args.chatbot}' not found.") return - await analyze_prs(cfg, db, bot["id"], bot["github_username"], limit=args.limit, since=since) + await analyze_prs( + cfg, db, bot["id"], bot["github_username"], + limit=args.limit, since=since, until=until, + sort_by=sort_by, + ) else: logger.error("Specify --chatbot or --all") finally: @@ -466,16 +521,18 @@ async def cmd_label(args: argparse.Namespace) -> None: logger.error("MARTIAN_API_KEY required") return - # Parse --since - since = None - if args.since: - from datetime import datetime - from datetime import timedelta - import re - - m = re.match(r"^(\d+)d$", args.since) - since = (datetime.now(UTC) - timedelta(days=int(m.group(1)))).isoformat() if m else args.since - logger.info(f"Filtering PRs reviewed since {since}") + since = _parse_time_bound(args.since) + until = _parse_time_bound(args.until) + sort_by = args.sort + if sort_by == "sweep": + if since or until: + logger.warning("--since/--until are ignored in sweep mode (sweep processes all unlabeled PRs by analyzed_at)") + logger.info("Sweep mode: sorting by analyzed_at DESC") + else: + if since: + logger.info(f"Filtering PRs reviewed since {since}") + if until: + logger.info(f"Filtering PRs reviewed before {until} (exclusive)") db = DBAdapter(cfg.database_url) await db.connect() @@ -486,13 +543,21 @@ async def cmd_label(args: argparse.Namespace) -> None: if args.all_chatbots: chatbots = await repo.get_all_chatbots() for bot in chatbots: - await label_prs(cfg, db, bot["id"], bot["github_username"], limit=args.limit, since=since) + await label_prs( + cfg, db, bot["id"], bot["github_username"], + limit=args.limit, since=since, until=until, + sort_by=sort_by, + ) elif args.chatbot: bot = await repo.get_chatbot(args.chatbot) if not bot: logger.error(f"Chatbot '{args.chatbot}' not found.") return - await label_prs(cfg, db, bot["id"], bot["github_username"], limit=args.limit, since=since) + await label_prs( + cfg, db, bot["id"], bot["github_username"], + limit=args.limit, since=since, until=until, + sort_by=sort_by, + ) else: logger.error("Specify --chatbot or --all") finally: diff --git a/online/etl/pipeline/analyze.py b/online/etl/pipeline/analyze.py index 6a9a942..cc1b30d 100644 --- a/online/etl/pipeline/analyze.py +++ b/online/etl/pipeline/analyze.py @@ -421,13 +421,21 @@ async def analyze_prs( chatbot_username: str, limit: int = 100, since: str | None = None, + until: str | None = None, + sort_by: str = "reviewed", ) -> int: """Run LLM analysis on all assembled, unanalyzed PRs for a chatbot. + `since` is an inclusive lower bound on bot_reviewed_at; `until` is an exclusive + upper bound. `sort_by` controls priority: "reviewed" (bot_reviewed_at DESC) or + "sweep" (assembled_at DESC, for catching late-discovered PRs). Returns the number of PRs analyzed. """ repo = PRRepository(db) - prs = await repo.get_assembled_not_analyzed(chatbot_id=chatbot_id, limit=limit, since=since) + prs = await repo.get_assembled_not_analyzed( + chatbot_id=chatbot_id, limit=limit, since=since, until=until, + sort_by=sort_by, + ) if not prs: logger.info(f"No unanalyzed PRs for {chatbot_username}") diff --git a/online/etl/pipeline/label.py b/online/etl/pipeline/label.py index ecebb66..1fe7d36 100644 --- a/online/etl/pipeline/label.py +++ b/online/etl/pipeline/label.py @@ -111,10 +111,19 @@ async def label_prs( chatbot_username: str, limit: int = 100, since: str | None = None, + until: str | None = None, + sort_by: str = "reviewed", ) -> int: - """Label all analyzed, unlabeled PRs for a chatbot. Returns count labeled.""" + """Label all analyzed, unlabeled PRs for a chatbot. Returns count labeled. + + `since` is an inclusive lower bound on bot_reviewed_at; `until` is an exclusive + upper bound. `sort_by` controls priority: "reviewed" or "sweep". + """ repo = PRRepository(db) - prs = await repo.get_analyzed_not_labeled(chatbot_id=chatbot_id, limit=limit, since=since) + prs = await repo.get_analyzed_not_labeled( + chatbot_id=chatbot_id, limit=limit, since=since, until=until, + sort_by=sort_by, + ) if not prs: logger.info(f"No unlabeled PRs for {chatbot_username}") diff --git a/online/etl/tests/test_main.py b/online/etl/tests/test_main.py new file mode 100644 index 0000000..b813ab7 --- /dev/null +++ b/online/etl/tests/test_main.py @@ -0,0 +1,46 @@ +"""Tests for CLI helpers in main.py.""" + +from __future__ import annotations + +from datetime import UTC +from datetime import datetime +from datetime import timedelta + +from main import _parse_time_bound + + +class TestParseTimeBound: + def test_none_passes_through(self) -> None: + assert _parse_time_bound(None) is None + + def test_empty_string_passes_through(self) -> None: + assert _parse_time_bound("") is None + + def test_relative_form(self) -> None: + result = _parse_time_bound("7d") + assert result is not None + # Should round-trip through fromisoformat and land roughly 7 days back + parsed = datetime.fromisoformat(result) + delta = datetime.now(UTC) - parsed + # Allow up to a few seconds of drift between call sites + assert timedelta(days=7) - timedelta(seconds=5) <= delta <= timedelta(days=7) + timedelta(seconds=5) + + def test_bare_date_is_normalized_to_midnight_utc(self) -> None: + # The bug fix: asyncpg rejects bare-date strings when binding to + # a timestamptz column, so we expand them here. + result = _parse_time_bound("2026-04-18") + assert result == "2026-04-18T00:00:00+00:00" + # And the result must be parseable as a tz-aware datetime + dt = datetime.fromisoformat(result) + assert dt.tzinfo is not None + assert dt.year == 2026 and dt.month == 4 and dt.day == 18 + + def test_full_iso_timestamp_passes_through(self) -> None: + # Already a full ISO timestamp — leave it alone, _coerce_args will + # convert it to a datetime when binding. + value = "2026-04-18T12:34:56+00:00" + assert _parse_time_bound(value) == value + + def test_iso_with_space_separator_passes_through(self) -> None: + value = "2026-04-18 12:34:56+00:00" + assert _parse_time_bound(value) == value diff --git a/online/etl/tests/test_repository.py b/online/etl/tests/test_repository.py index aa4d835..812ca98 100644 --- a/online/etl/tests/test_repository.py +++ b/online/etl/tests/test_repository.py @@ -314,6 +314,130 @@ async def test_mark_error(self, repo: PRRepository) -> None: assert row["status"] == "error" assert row["error_message"] == "Something broke" + @pytest.mark.asyncio + async def test_get_assembled_not_analyzed_with_until( + self, db: DBAdapter, repo: PRRepository # noqa: ARG002 + ) -> None: + """`until` is exclusive: --since 4/18 --until 4/19 yields just 4/18.""" + cid = await repo.upsert_chatbot("rangetest[bot]") + + # Three PRs at distinct timestamps, all assembled+merged + timestamps = { + 100: "2026-04-17T10:00:00+00:00", + 101: "2026-04-18T12:00:00+00:00", + 102: "2026-04-19T08:00:00+00:00", + } + for pr_num, ts in timestamps.items(): + await repo.insert_pr( + chatbot_id=cid, repo_name="org/repo", pr_number=pr_num, + pr_url=f"https://x/{pr_num}", pr_merged=True, bot_reviewed_at=ts, + bq_events=[{"event_id": str(pr_num), "type": "PullRequestReviewEvent", + "actor": "rangetest[bot]", "created_at": ts, + "payload": {"pull_request": {"title": "t", "user": {"login": "a"}}}}], + ) + pr = await repo.get_pr(cid, "org/repo", pr_num) + await repo.mark_assembled(pr["id"], {"pr_merged": True}) + + # since-only: includes 4/18 and 4/19 + rows = await repo.get_assembled_not_analyzed( + chatbot_id=cid, limit=10, since="2026-04-18T00:00:00+00:00", + ) + assert {r["pr_number"] for r in rows} == {101, 102} + + # since + until: just 4/18 (until is exclusive) + rows = await repo.get_assembled_not_analyzed( + chatbot_id=cid, limit=10, + since="2026-04-18T00:00:00+00:00", + until="2026-04-19T00:00:00+00:00", + ) + assert {r["pr_number"] for r in rows} == {101} + + # until-only: everything strictly before 4/19 + rows = await repo.get_assembled_not_analyzed( + chatbot_id=cid, limit=10, until="2026-04-19T00:00:00+00:00", + ) + assert {r["pr_number"] for r in rows} == {100, 101} + + # No bounds: all three + rows = await repo.get_assembled_not_analyzed(chatbot_id=cid, limit=10) + assert {r["pr_number"] for r in rows} == {100, 101, 102} + + @pytest.mark.asyncio + async def test_get_assembled_not_analyzed_until_all_chatbots( + self, db: DBAdapter, repo: PRRepository # noqa: ARG002 + ) -> None: + """until param works for the all-chatbots variant too.""" + c1 = await repo.upsert_chatbot("rangetest1[bot]") + c2 = await repo.upsert_chatbot("rangetest2[bot]") + for cid, pr_num, ts in [ + (c1, 200, "2026-04-17T10:00:00+00:00"), + (c2, 201, "2026-04-18T12:00:00+00:00"), + (c2, 202, "2026-04-19T08:00:00+00:00"), + ]: + await repo.insert_pr( + chatbot_id=cid, repo_name="org/repo", pr_number=pr_num, + pr_url=f"https://x/{pr_num}", pr_merged=True, bot_reviewed_at=ts, + bq_events=[{"event_id": str(pr_num), "type": "PullRequestReviewEvent", + "actor": "x", "created_at": ts, + "payload": {"pull_request": {"title": "t", "user": {"login": "a"}}}}], + ) + pr = await repo.get_pr(cid, "org/repo", pr_num) + await repo.mark_assembled(pr["id"], {"pr_merged": True}) + + rows = await repo.get_assembled_not_analyzed( + chatbot_id=None, limit=10, + since="2026-04-18T00:00:00+00:00", + until="2026-04-19T00:00:00+00:00", + ) + assert {r["pr_number"] for r in rows} == {201} + + @pytest.mark.asyncio + async def test_get_assembled_not_analyzed_sort_by_sweep( + self, db: DBAdapter, repo: PRRepository # noqa: ARG002 + ) -> None: + """sort_by='sweep' orders by assembled_at DESC, catching late-discovered PRs.""" + cid = await repo.upsert_chatbot("sorttest[bot]") + + # PR 300: reviewed long ago, assembled recently (late-discovered) + # PR 301: reviewed recently, assembled earlier + await repo.insert_pr( + chatbot_id=cid, repo_name="org/repo", pr_number=300, + pr_url="https://x/300", pr_merged=True, + bot_reviewed_at="2026-03-01T10:00:00+00:00", + bq_events=[{"event_id": "300", "type": "PullRequestReviewEvent", + "actor": "sorttest[bot]", "created_at": "2026-03-01T10:00:00+00:00", + "payload": {"pull_request": {"title": "t", "user": {"login": "a"}}}}], + ) + await repo.insert_pr( + chatbot_id=cid, repo_name="org/repo", pr_number=301, + pr_url="https://x/301", pr_merged=True, + bot_reviewed_at="2026-04-20T10:00:00+00:00", + bq_events=[{"event_id": "301", "type": "PullRequestReviewEvent", + "actor": "sorttest[bot]", "created_at": "2026-04-20T10:00:00+00:00", + "payload": {"pull_request": {"title": "t", "user": {"login": "a"}}}}], + ) + + # Assemble PR 301 first (earlier assembled_at), then PR 300 (later assembled_at) + pr301 = await repo.get_pr(cid, "org/repo", 301) + await repo.mark_assembled(pr301["id"], {"pr_merged": True}) + + # Small delay to ensure distinct assembled_at timestamps + import asyncio + await asyncio.sleep(0.05) + + pr300 = await repo.get_pr(cid, "org/repo", 300) + await repo.mark_assembled(pr300["id"], {"pr_merged": True}) + + # Default sort (bot_reviewed_at DESC): PR 301 first (reviewed 2026-04-20) + rows = await repo.get_assembled_not_analyzed(chatbot_id=cid, limit=10) + assert rows[0]["pr_number"] == 301 + + # Sweep sort: PR 300 first (assembled more recently) + rows = await repo.get_assembled_not_analyzed( + chatbot_id=cid, limit=10, sort_by="sweep" + ) + assert rows[0]["pr_number"] == 300 + @pytest.mark.asyncio async def test_mark_skipped(self, repo: PRRepository) -> None: cid = await repo.upsert_chatbot("testbot[bot]")