Add --until and --sort sweep flags to analyze/label

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 -------------------------------------------------------

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 = """

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))

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: