From 4cccd593d1347798b8456e29878e8f9dd8bb1fd4 Mon Sep 17 00:00:00 2001
From: Ashley Zhang <69987606+ashleyzhang01@users.noreply.github.com>
Date: Tue, 21 Apr 2026 21:30:48 -0400
Subject: [PATCH 01/11] Add --until flag to analyze and label commands

Allows targeting a specific date range when re-running analyze/label,
which is needed for incident-recovery backfills (e.g. analyzing only
2026-04-18 without first burning budget on newer days that the DESC-
ordered query would walk first).

- New BOUNDED query variants in queries.py with nullable since/until
  bounds; existing SINCE / no-bound queries kept as fast paths.
- repository.get_assembled_not_analyzed and get_analyzed_not_labeled
  gain an `until` param and route to the bounded queries when set.
- pipeline.analyze.analyze_prs and pipeline.label.label_prs thread
  `until` through to the repository.
- main.py exposes --until on analyze + label, parsed identically to
  --since (relative "Nd" or absolute date). Refactored the parsing
  into a shared _parse_time_bound helper.
- connection._translate_params strips PG-only ::type casts when
  running against SQLite, so the bounded queries work in both
  backends without duplication.
- Tests cover since-only, until-only, since+until, and the
  all-chatbots variant.

Semantics: --since is inclusive, --until is exclusive, so
  --since 2026-04-18 --until 2026-04-19  ->  just 2026-04-18.
Made-with: Cursor
---
 online/etl/db/connection.py         | 17 +++---
 online/etl/db/queries.py            | 57 ++++++++++++++++++++
 online/etl/db/repository.py         | 32 +++++++++++-
 online/etl/main.py                  | 80 ++++++++++++++++++++---------
 online/etl/pipeline/analyze.py      |  8 ++-
 online/etl/pipeline/label.py        | 11 +++-
 online/etl/tests/test_repository.py | 77 +++++++++++++++++++++++++++
 7 files changed, 246 insertions(+), 36 deletions(-)

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..94b2d13 100644
--- a/online/etl/db/queries.py
+++ b/online/etl/db/queries.py
@@ -99,6 +99,35 @@
     LIMIT $2
 """
 
+# 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 +292,34 @@
     LIMIT $2
 """
 
+# 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..7c8fd9d 100644
--- a/online/etl/db/repository.py
+++ b/online/etl/db/repository.py
@@ -143,8 +143,23 @@ 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,
     ) -> 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.
+        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 +304,21 @@ 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,
     ) -> list[dict[str, Any]]:
+        # See note on get_assembled_not_analyzed: `until` is exclusive.
+        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 8424a75..3afc498 100644
--- a/online/etl/main.py
+++ b/online/etl/main.py
@@ -178,7 +178,14 @@ 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("--database-url")
     p_ana.add_argument("--verbose", action="store_true")
 
@@ -194,7 +201,11 @@ 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("--database-url")
     p_lbl.add_argument("--verbose", action="store_true")
 
@@ -406,6 +417,23 @@ 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).
+    """
+    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()
+    return value
+
+
 async def cmd_analyze(args: argparse.Namespace) -> None:
     from db.connection import DBAdapter
     from db.repository import PRRepository
@@ -419,16 +447,12 @@ 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
+    since = _parse_time_bound(args.since)
+    until = _parse_time_bound(args.until)
+    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()
@@ -439,13 +463,19 @@ 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,
+                )
         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,
+            )
         else:
             logger.error("Specify --chatbot or --all")
     finally:
@@ -465,16 +495,12 @@ 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
+    since = _parse_time_bound(args.since)
+    until = _parse_time_bound(args.until)
+    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()
@@ -485,13 +511,19 @@ 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,
+                )
         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,
+            )
         else:
             logger.error("Specify --chatbot or --all")
     finally:
diff --git a/online/etl/pipeline/analyze.py b/online/etl/pipeline/analyze.py
index 6a9a942..c2e4633 100644
--- a/online/etl/pipeline/analyze.py
+++ b/online/etl/pipeline/analyze.py
@@ -421,13 +421,17 @@ async def analyze_prs(
     chatbot_username: str,
     limit: int = 100,
     since: str | None = None,
+    until: str | None = None,
 ) -> int:
     """Run LLM analysis on all assembled, unanalyzed PRs for a chatbot.
 
-    Returns the number of PRs analyzed.
+    `since` is an inclusive lower bound on bot_reviewed_at; `until` is an exclusive
+    upper bound. 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
+    )
 
     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..06c7a7c 100644
--- a/online/etl/pipeline/label.py
+++ b/online/etl/pipeline/label.py
@@ -111,10 +111,17 @@ async def label_prs(
     chatbot_username: str,
     limit: int = 100,
     since: str | None = None,
+    until: str | None = None,
 ) -> 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.
+    """
     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
+    )
 
     if not prs:
         logger.info(f"No unlabeled PRs for {chatbot_username}")
diff --git a/online/etl/tests/test_repository.py b/online/etl/tests/test_repository.py
index 97809fc..1b6214d 100644
--- a/online/etl/tests/test_repository.py
+++ b/online/etl/tests/test_repository.py
@@ -314,6 +314,83 @@ 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
+    ) -> 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
+    ) -> 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_mark_skipped(self, repo: PRRepository) -> None:
         cid = await repo.upsert_chatbot("testbot[bot]")

From 19598744711e6f3f1b813ecaf8d7098b2c000356 Mon Sep 17 00:00:00 2001
From: Ashley Zhang <69987606+ashleyzhang01@users.noreply.github.com>
Date: Tue, 21 Apr 2026 21:35:37 -0400
Subject: [PATCH 02/11] Normalize bare-date --since/--until to midnight UTC ISO

asyncpg requires datetime objects for timestamptz parameters, and
_coerce_args only converts strings that match the full ISO regex
(YYYY-MM-DDThh:mm:ss). Bare dates like "2026-04-18" passed straight
through and triggered:

  asyncpg.exceptions.DataError: invalid input for query argument $2:
    '2026-04-18' (expected a datetime.date or datetime.datetime
    instance, got 'str')

Fix in _parse_time_bound: detect a bare YYYY-MM-DD and expand to
midnight UTC. Affects both --since and --until on analyze and label.

This also resolves a latent bug: the original --since handler had
the same problem, but it was only ever invoked with the relative
"Nd" form (which produces a full ISO timestamp), so nobody hit it.

Made-with: Cursor
---
 online/etl/main.py            |  5 ++++
 online/etl/tests/test_main.py | 46 +++++++++++++++++++++++++++++++++++
 2 files changed, 51 insertions(+)
 create mode 100644 online/etl/tests/test_main.py

diff --git a/online/etl/main.py b/online/etl/main.py
index 3afc498..be8045a 100644
--- a/online/etl/main.py
+++ b/online/etl/main.py
@@ -421,6 +421,9 @@ 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
@@ -431,6 +434,8 @@ def _parse_time_bound(value: str | None) -> str | None:
     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
 
 
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

From 42dfe4df9ee2fc069dd1f4e70bab590025563ad4 Mon Sep 17 00:00:00 2001
From: Ashley Zhang <69987606+ashleyzhang01@users.noreply.github.com>
Date: Thu, 21 May 2026 15:37:09 -0700
Subject: [PATCH 03/11] allow sort unanalyzed PRs by assembled time

---
 online/etl/db/queries.py    | 24 ++++++++++++++++++++++++
 online/etl/db/repository.py | 10 ++++++++++
 2 files changed, 34 insertions(+)

diff --git a/online/etl/db/queries.py b/online/etl/db/queries.py
index 94b2d13..b4a00b4 100644
--- a/online/etl/db/queries.py
+++ b/online/etl/db/queries.py
@@ -99,6 +99,30 @@
     LIMIT $2
 """
 
+# Assembled-sorted variants: same filter as the default queries but ordered by
+# assembled_at DESC. Used by --sort assembled (sweep mode) to prioritize PRs that
+# recently became ready, catching late-discovered PRs that bot_reviewed_at ordering misses.
+GET_ASSEMBLED_PRS_NOT_ANALYZED_BY_ASSEMBLED = """
+    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_BY_ASSEMBLED = """
+    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
diff --git a/online/etl/db/repository.py b/online/etl/db/repository.py
index 7c8fd9d..281e721 100644
--- a/online/etl/db/repository.py
+++ b/online/etl/db/repository.py
@@ -148,10 +148,20 @@ async def get_assembled_not_analyzed(
         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` == "assembled" uses assembled_at DESC (for sweep/catch-up mode).
+        if sort_by == "assembled":
+            if chatbot_id is not None:
+                return await self.db.fetchall(
+                    q.GET_ASSEMBLED_PRS_NOT_ANALYZED_BY_ASSEMBLED, (chatbot_id, limit)
+                )
+            return await self.db.fetchall(
+                q.GET_ALL_ASSEMBLED_NOT_ANALYZED_BY_ASSEMBLED, (limit,)
+            )
         if until is not None:
             if chatbot_id is not None:
                 return await self.db.fetchall(

From 516f87b39b93fb9e761df4b2eae370dcb4fb0ecb Mon Sep 17 00:00:00 2001
From: Ashley Zhang <69987606+ashleyzhang01@users.noreply.github.com>
Date: Thu, 21 May 2026 15:37:29 -0700
Subject: [PATCH 04/11] add sort by flag to analyze pipeline

---
 online/etl/main.py             | 11 +++++++++++
 online/etl/pipeline/analyze.py |  8 ++++++--
 2 files changed, 17 insertions(+), 2 deletions(-)

diff --git a/online/etl/main.py b/online/etl/main.py
index be8045a..f889665 100644
--- a/online/etl/main.py
+++ b/online/etl/main.py
@@ -186,6 +186,12 @@ def build_parser() -> argparse.ArgumentParser:
             "With --since 2026-04-18 --until 2026-04-19 you get just 2026-04-18."
         ),
     )
+    p_ana.add_argument(
+        "--sort",
+        choices=["reviewed", "assembled"],
+        default="reviewed",
+        help="Sort order: 'reviewed' (bot_reviewed_at DESC, default) or 'assembled' (assembled_at DESC, for catching late-discovered PRs).",
+    )
     p_ana.add_argument("--database-url")
     p_ana.add_argument("--verbose", action="store_true")
 
@@ -454,10 +460,13 @@ async def cmd_analyze(args: argparse.Namespace) -> None:
 
     since = _parse_time_bound(args.since)
     until = _parse_time_bound(args.until)
+    sort_by = args.sort
     if since:
         logger.info(f"Filtering PRs reviewed since {since}")
     if until:
         logger.info(f"Filtering PRs reviewed before {until} (exclusive)")
+    if sort_by == "assembled":
+        logger.info("Sorting by assembled_at DESC (sweep mode)")
 
     db = DBAdapter(cfg.database_url)
     await db.connect()
@@ -471,6 +480,7 @@ async def cmd_analyze(args: argparse.Namespace) -> None:
                 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)
@@ -480,6 +490,7 @@ async def cmd_analyze(args: argparse.Namespace) -> None:
             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")
diff --git a/online/etl/pipeline/analyze.py b/online/etl/pipeline/analyze.py
index c2e4633..d4236c7 100644
--- a/online/etl/pipeline/analyze.py
+++ b/online/etl/pipeline/analyze.py
@@ -422,15 +422,19 @@ async def analyze_prs(
     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. Returns the number of PRs analyzed.
+    upper bound. `sort_by` controls priority: "reviewed" (bot_reviewed_at DESC) or
+    "assembled" (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, until=until
+        chatbot_id=chatbot_id, limit=limit, since=since, until=until,
+        sort_by=sort_by,
     )
 
     if not prs:

From 4e599a8bb90ec60b19e4c515b696ea91b2f60bd8 Mon Sep 17 00:00:00 2001
From: Ashley Zhang <69987606+ashleyzhang01@users.noreply.github.com>
Date: Thu, 21 May 2026 15:37:38 -0700
Subject: [PATCH 05/11] add tests

---
 online/etl/tests/test_repository.py | 47 +++++++++++++++++++++++++++++
 1 file changed, 47 insertions(+)

diff --git a/online/etl/tests/test_repository.py b/online/etl/tests/test_repository.py
index 1b6214d..c732e13 100644
--- a/online/etl/tests/test_repository.py
+++ b/online/etl/tests/test_repository.py
@@ -391,6 +391,53 @@ async def test_get_assembled_not_analyzed_until_all_chatbots(
         )
         assert {r["pr_number"] for r in rows} == {201}
 
+    @pytest.mark.asyncio
+    async def test_get_assembled_not_analyzed_sort_by_assembled(
+        self, db: DBAdapter, repo: PRRepository
+    ) -> None:
+        """sort_by='assembled' 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
+
+        # Assembled sort: PR 300 first (assembled more recently)
+        rows = await repo.get_assembled_not_analyzed(
+            chatbot_id=cid, limit=10, sort_by="assembled"
+        )
+        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]")

From e531b88362f43153bdd85bca03c88bdaa546767d Mon Sep 17 00:00:00 2001
From: Ashley Zhang <69987606+ashleyzhang01@users.noreply.github.com>
Date: Thu, 21 May 2026 15:53:37 -0700
Subject: [PATCH 06/11] allow sort label by assembled time

---
 online/etl/db/queries.py     | 24 ++++++++++++++++++++++++
 online/etl/db/repository.py  | 10 ++++++++++
 online/etl/main.py           | 11 +++++++++++
 online/etl/pipeline/label.py |  6 ++++--
 4 files changed, 49 insertions(+), 2 deletions(-)

diff --git a/online/etl/db/queries.py b/online/etl/db/queries.py
index b4a00b4..dfc1ddd 100644
--- a/online/etl/db/queries.py
+++ b/online/etl/db/queries.py
@@ -316,6 +316,30 @@
     LIMIT $2
 """
 
+# Assembled-sorted variants for the labeling stage.
+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.assembled_at DESC NULLS LAST
+    LIMIT $2
+"""
+
+GET_ALL_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.status = 'analyzed'
+      AND pl.id IS NULL
+    ORDER BY p.assembled_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
diff --git a/online/etl/db/repository.py b/online/etl/db/repository.py
index 281e721..4f9a05e 100644
--- a/online/etl/db/repository.py
+++ b/online/etl/db/repository.py
@@ -319,8 +319,18 @@ async def get_analyzed_not_labeled(
         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` == "assembled" uses assembled_at DESC (sweep mode).
+        if sort_by == "assembled":
+            if chatbot_id is not None:
+                return await self.db.fetchall(
+                    q.GET_ANALYZED_NOT_LABELED_BY_ASSEMBLED, (chatbot_id, limit)
+                )
+            return await self.db.fetchall(
+                q.GET_ALL_ANALYZED_NOT_LABELED_BY_ASSEMBLED, (limit,)
+            )
         if until is not None:
             if chatbot_id is not None:
                 return await self.db.fetchall(
diff --git a/online/etl/main.py b/online/etl/main.py
index f889665..c91d61b 100644
--- a/online/etl/main.py
+++ b/online/etl/main.py
@@ -212,6 +212,12 @@ def build_parser() -> argparse.ArgumentParser:
         "--until",
         help="Exclusive upper bound on bot_reviewed_at (e.g. '2d', '2026-04-19')",
     )
+    p_lbl.add_argument(
+        "--sort",
+        choices=["reviewed", "assembled"],
+        default="reviewed",
+        help="Sort order: 'reviewed' (bot_reviewed_at DESC, default) or 'assembled' (assembled_at DESC).",
+    )
     p_lbl.add_argument("--database-url")
     p_lbl.add_argument("--verbose", action="store_true")
 
@@ -513,10 +519,13 @@ async def cmd_label(args: argparse.Namespace) -> None:
 
     since = _parse_time_bound(args.since)
     until = _parse_time_bound(args.until)
+    sort_by = args.sort
     if since:
         logger.info(f"Filtering PRs reviewed since {since}")
     if until:
         logger.info(f"Filtering PRs reviewed before {until} (exclusive)")
+    if sort_by == "assembled":
+        logger.info("Sorting by assembled_at DESC (sweep mode)")
 
     db = DBAdapter(cfg.database_url)
     await db.connect()
@@ -530,6 +539,7 @@ async def cmd_label(args: argparse.Namespace) -> None:
                 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)
@@ -539,6 +549,7 @@ async def cmd_label(args: argparse.Namespace) -> None:
             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")
diff --git a/online/etl/pipeline/label.py b/online/etl/pipeline/label.py
index 06c7a7c..78f3515 100644
--- a/online/etl/pipeline/label.py
+++ b/online/etl/pipeline/label.py
@@ -112,15 +112,17 @@ async def label_prs(
     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.
 
     `since` is an inclusive lower bound on bot_reviewed_at; `until` is an exclusive
-    upper bound.
+    upper bound. `sort_by` controls priority: "reviewed" or "assembled".
     """
     repo = PRRepository(db)
     prs = await repo.get_analyzed_not_labeled(
-        chatbot_id=chatbot_id, limit=limit, since=since, until=until
+        chatbot_id=chatbot_id, limit=limit, since=since, until=until,
+        sort_by=sort_by,
     )
 
     if not prs:

From da1982db29d1f872117624a7b835eae7691b5877 Mon Sep 17 00:00:00 2001
From: Ashley Zhang <69987606+ashleyzhang01@users.noreply.github.com>
Date: Thu, 21 May 2026 16:10:28 -0700
Subject: [PATCH 07/11] label sorts by analyzed for sweep

---
 online/etl/db/queries.py | 7 ++++---
 1 file changed, 4 insertions(+), 3 deletions(-)

diff --git a/online/etl/db/queries.py b/online/etl/db/queries.py
index dfc1ddd..cd3be52 100644
--- a/online/etl/db/queries.py
+++ b/online/etl/db/queries.py
@@ -316,7 +316,8 @@
     LIMIT $2
 """
 
-# Assembled-sorted variants for the labeling stage.
+# 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
@@ -325,7 +326,7 @@
     WHERE p.chatbot_id = $1
       AND p.status = 'analyzed'
       AND pl.id IS NULL
-    ORDER BY p.assembled_at DESC NULLS LAST
+    ORDER BY p.analyzed_at DESC NULLS LAST
     LIMIT $2
 """
 
@@ -336,7 +337,7 @@
     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.assembled_at DESC NULLS LAST
+    ORDER BY p.analyzed_at DESC NULLS LAST
     LIMIT $1
 """
 

From 37aefcd52248eb42dcfa9851384bdf926a7d7305 Mon Sep 17 00:00:00 2001
From: Ashley Zhang <69987606+ashleyzhang01@users.noreply.github.com>
Date: Thu, 21 May 2026 16:12:06 -0700
Subject: [PATCH 08/11] change sort key from assembled to sweep

default is `--sort reviewed` by when bot reviewed at desc. new `--sort sweep` sorts by assembled at desc for analyze and analyzed at desc for label. this catches straggler PRs that were discovered/processed late
---
 online/etl/pipeline/analyze.py      | 2 +-
 online/etl/pipeline/label.py        | 2 +-
 online/etl/tests/test_repository.py | 8 ++++----
 3 files changed, 6 insertions(+), 6 deletions(-)

diff --git a/online/etl/pipeline/analyze.py b/online/etl/pipeline/analyze.py
index d4236c7..cc1b30d 100644
--- a/online/etl/pipeline/analyze.py
+++ b/online/etl/pipeline/analyze.py
@@ -428,7 +428,7 @@ async def analyze_prs(
 
     `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
-    "assembled" (assembled_at DESC, for catching late-discovered PRs).
+    "sweep" (assembled_at DESC, for catching late-discovered PRs).
     Returns the number of PRs analyzed.
     """
     repo = PRRepository(db)
diff --git a/online/etl/pipeline/label.py b/online/etl/pipeline/label.py
index 78f3515..1fe7d36 100644
--- a/online/etl/pipeline/label.py
+++ b/online/etl/pipeline/label.py
@@ -117,7 +117,7 @@ async def label_prs(
     """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 "assembled".
+    upper bound. `sort_by` controls priority: "reviewed" or "sweep".
     """
     repo = PRRepository(db)
     prs = await repo.get_analyzed_not_labeled(
diff --git a/online/etl/tests/test_repository.py b/online/etl/tests/test_repository.py
index c732e13..6eda444 100644
--- a/online/etl/tests/test_repository.py
+++ b/online/etl/tests/test_repository.py
@@ -392,10 +392,10 @@ async def test_get_assembled_not_analyzed_until_all_chatbots(
         assert {r["pr_number"] for r in rows} == {201}
 
     @pytest.mark.asyncio
-    async def test_get_assembled_not_analyzed_sort_by_assembled(
+    async def test_get_assembled_not_analyzed_sort_by_sweep(
         self, db: DBAdapter, repo: PRRepository
     ) -> None:
-        """sort_by='assembled' orders by assembled_at DESC, catching late-discovered PRs."""
+        """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)
@@ -432,9 +432,9 @@ async def test_get_assembled_not_analyzed_sort_by_assembled(
         rows = await repo.get_assembled_not_analyzed(chatbot_id=cid, limit=10)
         assert rows[0]["pr_number"] == 301
 
-        # Assembled sort: PR 300 first (assembled more recently)
+        # Sweep sort: PR 300 first (assembled more recently)
         rows = await repo.get_assembled_not_analyzed(
-            chatbot_id=cid, limit=10, sort_by="assembled"
+            chatbot_id=cid, limit=10, sort_by="sweep"
         )
         assert rows[0]["pr_number"] == 300
 

From 7546b9e30699976d4ae0c8f1e129fc252ce9a0e9 Mon Sep 17 00:00:00 2001
From: Ashley Zhang <69987606+ashleyzhang01@users.noreply.github.com>
Date: Thu, 21 May 2026 16:14:41 -0700
Subject: [PATCH 09/11] change new sort key from assembled to sweep

default is `--sort reviewed` by when bot reviewed at desc. new `--sort sweep` sorts by assembled at desc for analyze and analyzed at desc for label. this catches straggler PRs that were discovered/processed late
---
 online/etl/db/repository.py |  8 ++++----
 online/etl/main.py          | 16 ++++++++--------
 2 files changed, 12 insertions(+), 12 deletions(-)

diff --git a/online/etl/db/repository.py b/online/etl/db/repository.py
index 4f9a05e..108106c 100644
--- a/online/etl/db/repository.py
+++ b/online/etl/db/repository.py
@@ -153,8 +153,8 @@ async def get_assembled_not_analyzed(
         # `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` == "assembled" uses assembled_at DESC (for sweep/catch-up mode).
-        if sort_by == "assembled":
+        # `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_BY_ASSEMBLED, (chatbot_id, limit)
@@ -322,8 +322,8 @@ async def get_analyzed_not_labeled(
         sort_by: str = "reviewed",
     ) -> list[dict[str, Any]]:
         # See note on get_assembled_not_analyzed: `until` is exclusive.
-        # `sort_by` == "assembled" uses assembled_at DESC (sweep mode).
-        if sort_by == "assembled":
+        # `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_BY_ASSEMBLED, (chatbot_id, limit)
diff --git a/online/etl/main.py b/online/etl/main.py
index c91d61b..8f7e8f8 100644
--- a/online/etl/main.py
+++ b/online/etl/main.py
@@ -188,9 +188,9 @@ def build_parser() -> argparse.ArgumentParser:
     )
     p_ana.add_argument(
         "--sort",
-        choices=["reviewed", "assembled"],
+        choices=["reviewed", "sweep"],
         default="reviewed",
-        help="Sort order: 'reviewed' (bot_reviewed_at DESC, default) or 'assembled' (assembled_at DESC, for catching late-discovered PRs).",
+        help="Sort order: 'reviewed' (bot_reviewed_at DESC, default) or 'sweep' (assembled_at DESC, for catching late-discovered PRs).",
     )
     p_ana.add_argument("--database-url")
     p_ana.add_argument("--verbose", action="store_true")
@@ -214,9 +214,9 @@ def build_parser() -> argparse.ArgumentParser:
     )
     p_lbl.add_argument(
         "--sort",
-        choices=["reviewed", "assembled"],
+        choices=["reviewed", "sweep"],
         default="reviewed",
-        help="Sort order: 'reviewed' (bot_reviewed_at DESC, default) or 'assembled' (assembled_at DESC).",
+        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")
@@ -471,8 +471,8 @@ async def cmd_analyze(args: argparse.Namespace) -> None:
         logger.info(f"Filtering PRs reviewed since {since}")
     if until:
         logger.info(f"Filtering PRs reviewed before {until} (exclusive)")
-    if sort_by == "assembled":
-        logger.info("Sorting by assembled_at DESC (sweep mode)")
+    if sort_by == "sweep":
+        logger.info("Sweep mode: sorting by assembled_at DESC")
 
     db = DBAdapter(cfg.database_url)
     await db.connect()
@@ -524,8 +524,8 @@ async def cmd_label(args: argparse.Namespace) -> None:
         logger.info(f"Filtering PRs reviewed since {since}")
     if until:
         logger.info(f"Filtering PRs reviewed before {until} (exclusive)")
-    if sort_by == "assembled":
-        logger.info("Sorting by assembled_at DESC (sweep mode)")
+    if sort_by == "sweep":
+        logger.info("Sweep mode: sorting by analyzed_at DESC")
 
     db = DBAdapter(cfg.database_url)
     await db.connect()

From e40ee5e9576b2c8e099a0415dbedf159d7fabcfa Mon Sep 17 00:00:00 2001
From: Ashley Zhang <69987606+ashleyzhang01@users.noreply.github.com>
Date: Thu, 21 May 2026 17:03:17 -0700
Subject: [PATCH 10/11] rename comments for clarity

---
 online/etl/db/queries.py    | 12 ++++++------
 online/etl/db/repository.py |  8 ++++----
 online/etl/main.py          | 24 +++++++++++++++---------
 3 files changed, 25 insertions(+), 19 deletions(-)

diff --git a/online/etl/db/queries.py b/online/etl/db/queries.py
index cd3be52..b807255 100644
--- a/online/etl/db/queries.py
+++ b/online/etl/db/queries.py
@@ -99,10 +99,10 @@
     LIMIT $2
 """
 
-# Assembled-sorted variants: same filter as the default queries but ordered by
-# assembled_at DESC. Used by --sort assembled (sweep mode) to prioritize PRs that
-# recently became ready, catching late-discovered PRs that bot_reviewed_at ordering misses.
-GET_ASSEMBLED_PRS_NOT_ANALYZED_BY_ASSEMBLED = """
+# 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
@@ -113,7 +113,7 @@
     LIMIT $2
 """
 
-GET_ALL_ASSEMBLED_NOT_ANALYZED_BY_ASSEMBLED = """
+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'
@@ -330,7 +330,7 @@
     LIMIT $2
 """
 
-GET_ALL_ANALYZED_NOT_LABELED_BY_ASSEMBLED = """
+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
diff --git a/online/etl/db/repository.py b/online/etl/db/repository.py
index 108106c..62b15b2 100644
--- a/online/etl/db/repository.py
+++ b/online/etl/db/repository.py
@@ -157,10 +157,10 @@ async def get_assembled_not_analyzed(
         if sort_by == "sweep":
             if chatbot_id is not None:
                 return await self.db.fetchall(
-                    q.GET_ASSEMBLED_PRS_NOT_ANALYZED_BY_ASSEMBLED, (chatbot_id, limit)
+                    q.GET_ASSEMBLED_PRS_NOT_ANALYZED_SWEEP, (chatbot_id, limit)
                 )
             return await self.db.fetchall(
-                q.GET_ALL_ASSEMBLED_NOT_ANALYZED_BY_ASSEMBLED, (limit,)
+                q.GET_ALL_ASSEMBLED_NOT_ANALYZED_SWEEP, (limit,)
             )
         if until is not None:
             if chatbot_id is not None:
@@ -326,10 +326,10 @@ async def get_analyzed_not_labeled(
         if sort_by == "sweep":
             if chatbot_id is not None:
                 return await self.db.fetchall(
-                    q.GET_ANALYZED_NOT_LABELED_BY_ASSEMBLED, (chatbot_id, limit)
+                    q.GET_ANALYZED_NOT_LABELED_SWEEP, (chatbot_id, limit)
                 )
             return await self.db.fetchall(
-                q.GET_ALL_ANALYZED_NOT_LABELED_BY_ASSEMBLED, (limit,)
+                q.GET_ALL_ANALYZED_NOT_LABELED_SWEEP, (limit,)
             )
         if until is not None:
             if chatbot_id is not None:
diff --git a/online/etl/main.py b/online/etl/main.py
index 8f7e8f8..912ebd5 100644
--- a/online/etl/main.py
+++ b/online/etl/main.py
@@ -190,7 +190,7 @@ def build_parser() -> argparse.ArgumentParser:
         "--sort",
         choices=["reviewed", "sweep"],
         default="reviewed",
-        help="Sort order: 'reviewed' (bot_reviewed_at DESC, default) or 'sweep' (assembled_at DESC, for catching late-discovered PRs).",
+        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")
@@ -467,12 +467,15 @@ async def cmd_analyze(args: argparse.Namespace) -> None:
     since = _parse_time_bound(args.since)
     until = _parse_time_bound(args.until)
     sort_by = args.sort
-    if since:
-        logger.info(f"Filtering PRs reviewed since {since}")
-    if until:
-        logger.info(f"Filtering PRs reviewed before {until} (exclusive)")
     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()
@@ -520,12 +523,15 @@ async def cmd_label(args: argparse.Namespace) -> None:
     since = _parse_time_bound(args.since)
     until = _parse_time_bound(args.until)
     sort_by = args.sort
-    if since:
-        logger.info(f"Filtering PRs reviewed since {since}")
-    if until:
-        logger.info(f"Filtering PRs reviewed before {until} (exclusive)")
     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()

From 742b23ee5e715200488adf938af8a43be4ff49de Mon Sep 17 00:00:00 2001
From: Ashley Zhang <69987606+ashleyzhang01@users.noreply.github.com>
Date: Thu, 21 May 2026 17:08:03 -0700
Subject: [PATCH 11/11] fix ruff errors

---
 online/etl/tests/test_repository.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/online/etl/tests/test_repository.py b/online/etl/tests/test_repository.py
index 6eda444..b83aa62 100644
--- a/online/etl/tests/test_repository.py
+++ b/online/etl/tests/test_repository.py
@@ -316,7 +316,7 @@ async def test_mark_error(self, repo: PRRepository) -> None:
 
     @pytest.mark.asyncio
     async def test_get_assembled_not_analyzed_with_until(
-        self, db: DBAdapter, repo: PRRepository
+        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]")
@@ -364,7 +364,7 @@ async def test_get_assembled_not_analyzed_with_until(
 
     @pytest.mark.asyncio
     async def test_get_assembled_not_analyzed_until_all_chatbots(
-        self, db: DBAdapter, repo: PRRepository
+        self, db: DBAdapter, repo: PRRepository  # noqa: ARG002
     ) -> None:
         """until param works for the all-chatbots variant too."""
         c1 = await repo.upsert_chatbot("rangetest1[bot]")
@@ -393,7 +393,7 @@ async def test_get_assembled_not_analyzed_until_all_chatbots(
 
     @pytest.mark.asyncio
     async def test_get_assembled_not_analyzed_sort_by_sweep(
-        self, db: DBAdapter, repo: PRRepository
+        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]")