src/metrics.py

@@ -1761,7 +1761,7 @@ def classifier_degradation_warning(conn: sqlite3.Connection, cfg: Any) -> list[s
             ) AS degraded
           FROM route_decisions
          WHERE classification_source IS NOT NULL
-           AND observed_at >= datetime('now', '-24 hours')
+           AND julianday(observed_at) >= julianday('now', '-24 hours')
         """
     ).fetchone()
     total = (row["total"] if row else 0) or 0
@@ -1801,7 +1801,7 @@ def _declined_for(conn: sqlite3.Connection) -> str:
            AND classification_source IN
                ('fallback', 'session_stale', 'session_history',
                 'classifier_cloud')
-           AND observed_at >= datetime('now', '-24 hours')
+           AND julianday(observed_at) >= julianday('now', '-24 hours')
          GROUP BY classifier_reject
          ORDER BY n DESC, classifier_reject
         """

tests/test_classifier_cascade.py

@@ -7,7 +7,7 @@ happened to produce the same answer.
 from __future__ import annotations
 
 import sqlite3
-from datetime import datetime, timedelta, timezone
+from datetime import datetime, time, timedelta, timezone
 from pathlib import Path
 from types import SimpleNamespace
 
@@ -222,7 +222,8 @@ def test_cascade_with_no_session_key_falls_straight_through(db):
 # ---------------------------------------------------------------------------
 
 
-def _seed_sources(conn, pairs):
+def _seed_sources(conn, pairs, at=None):
+    ts = at.isoformat() if at is not None else _now().isoformat()
     for source, count in pairs:
         for _ in range(count):
             conn.execute(
@@ -233,7 +234,7 @@ def _seed_sources(conn, pairs):
                      observed_at)
                 VALUES ('chat','general_chat',2,0,'m','neuralwatt',?,?)
                 """,
-                (source, _now().isoformat()),
+                (source, ts),
             )
     conn.commit()
 
@@ -272,7 +273,8 @@ def test_degradation_warning_silent_when_healthy(db):
     assert classifier_degradation_warning(db, cfg) == []
 
 
-def _seed_declined(conn, source, reason, count):
+def _seed_declined(conn, source, reason, count, at=None):
+    ts = at.isoformat() if at is not None else _now().isoformat()
     for _ in range(count):
         conn.execute(
             """
@@ -282,7 +284,7 @@ def _seed_declined(conn, source, reason, count):
                  classifier_reject, observed_at)
             VALUES ('chat','general_chat',2,0,'m','neuralwatt',?,?,?)
             """,
-            (source, reason, _now().isoformat()),
+            (source, reason, ts),
         )
     conn.commit()
 
@@ -334,3 +336,59 @@ def test_degradation_warning_survives_a_database_without_the_column(db, monkeypa
     (warning,) = metrics.classifier_degradation_warning(db, cfg)
 
     assert "60%" in warning and "Declined for" not in warning
+
+
+def test_degradation_window_excludes_rows_older_than_24h_on_the_cutoff_date(db):
+    from metrics import classifier_degradation_warning
+
+    now = _now()
+    cutoff = now - timedelta(hours=24)
+    just_outside = datetime.combine(cutoff.date(), time(0, 0, 0), tzinfo=timezone.utc)
+    inside = now - timedelta(hours=1)
+
+    cfg = SimpleNamespace(
+        classifier=SimpleNamespace(degraded_warn_min=20, degraded_warn_threshold=0.5)
+    )
+    _seed_sources(db, [("fallback", 25)], at=just_outside)
+    _seed_sources(db, [("classifier", 25)], at=inside)
+
+    out = classifier_degradation_warning(db, cfg)
+    assert out == []
+
+
+def test_degradation_window_counts_rows_inside_24h(db):
+    from metrics import classifier_degradation_warning
+
+    now = _now()
+    inside = now - timedelta(hours=1)
+
+    cfg = SimpleNamespace(
+        classifier=SimpleNamespace(degraded_warn_min=20, degraded_warn_threshold=0.5)
+    )
+    _seed_sources(db, [("fallback", 15), ("classifier", 10)], at=inside)
+
+    out = classifier_degradation_warning(db, cfg)
+    assert len(out) == 1
+    assert "60%" in out[0]
+
+
+def test_declined_for_ignores_reasons_older_than_24h(db):
+    from metrics import classifier_degradation_warning
+
+    now = _now()
+    cutoff = now - timedelta(hours=24)
+    just_outside = datetime.combine(cutoff.date(), time(0, 0, 0), tzinfo=timezone.utc)
+    inside = now - timedelta(hours=1)
+
+    cfg = SimpleNamespace(
+        classifier=SimpleNamespace(degraded_warn_min=20, degraded_warn_threshold=0.2)
+    )
+    _seed_sources(db, [("classifier", 60)], at=inside)
+    _seed_declined(db, "session_history", "below_confidence_min", 20, at=inside)
+    _seed_declined(db, "fallback", "below_coverage_min", 7, at=just_outside)
+
+    (warning,) = classifier_degradation_warning(db, cfg)
+
+    assert "Declined for: 20 below_confidence_min." in warning
+    assert "below_coverage_min" not in warning
+

tests/test_metrics.py

@@ -30,6 +30,7 @@ from starlette.testclient import TestClient
 import dispatcher
 from config import load_config
 from metrics import (
+    _billing_period_start,
     _next_reset_date,
     cache_rate_series,
     cache_rate_warnings,
@@ -155,6 +156,31 @@ def _telemetry_provider_cfg(**objective_overrides) -> SimpleNamespace:
     )
 
 
+def _pace_reset_day(today: date) -> int:
+    """A billing_reset_day that puts ``today`` at least one day into the period.
+
+    quota_accounts sets elapsed_fraction to None below 0.02, which is day 0 of a
+    period, and then skips the pace rule on purpose: a ratio over a few hours of
+    elapsed time is noise. A hardcoded reset day therefore turns the pace tests
+    red on that one calendar day every month. They did on 2026-10-06 UTC with
+    billing_reset_day=6, while their comments claimed "on any day". Valid days
+    are 1..28.
+    """
+    return today.day - 1 if 2 <= today.day <= 29 else 28
+
+
+def test_pace_reset_day_is_never_the_reset_day_itself():
+    """Every calendar day of a leap year and a common year lands past the 0.02 guard."""
+    d = date(2024, 1, 1)
+    while d < date(2026, 1, 1):
+        reset_day = _pace_reset_day(d)
+        assert 1 <= reset_day <= 28, d
+        start = date.fromisoformat(_billing_period_start(reset_day, d))
+        nxt = date.fromisoformat(_next_reset_date(reset_day, d))
+        assert (d - start).days / (nxt - start).days >= 0.02, d
+        d += timedelta(days=1)
+
+
 # --- Import / no-circular-import smoke tests -----------------------------------
 
 
@@ -466,10 +492,13 @@ def test_quota_accounts_interleaved_providers_independent(tmp_path):
 
 def test_quota_accounts_alarm_plan_pace_warning(tmp_path):
     """Usage pace > 1.25x triggers plan_pace alarm."""
-    cfg = _telemetry_provider_cfg(plan_kwh_per_period=6.25, billing_reset_day=6)
+    cfg = _telemetry_provider_cfg(
+        plan_kwh_per_period=6.25, billing_reset_day=_pace_reset_day(_now().date())
+    )
     conn = _make_db(tmp_path)
     now = _now()
-    # Seed 8.0 kWh to guarantee used_fraction > 1.25 * elapsed_fraction on any day
+    # Seed 8.0 kWh to guarantee used_fraction > 1.25 * elapsed_fraction; the reset
+    # day keeps today past the 0.02 elapsed guard on every calendar day
     conn.execute(
         "INSERT INTO energy_observations "
         "(model_id, provider, energy_kwh, completion_tokens, observed_at) "
@@ -510,10 +539,13 @@ def test_quota_accounts_alarm_stale_reading(tmp_path):
 
 def test_quota_accounts_alarm_plan_pace_critical(tmp_path):
     """Usage pace > 2.0 triggers critical severity."""
-    cfg = _telemetry_provider_cfg(plan_kwh_per_period=6.25, billing_reset_day=6)
+    cfg = _telemetry_provider_cfg(
+        plan_kwh_per_period=6.25, billing_reset_day=_pace_reset_day(_now().date())
+    )
     conn = _make_db(tmp_path)
     now = _now()
-    # Seed 15 kWh to guarantee pace > 2.0 on any day
+    # Seed 15 kWh to guarantee pace > 2.0; the reset day keeps today past the
+    # 0.02 elapsed guard on every calendar day
     conn.execute(
         "INSERT INTO energy_observations "
         "(model_id, provider, energy_kwh, completion_tokens, observed_at) "

tests/test_sql_time_windows.py

@@ -0,0 +1,136 @@
+"""Tripwire: detect ISO-T vs datetime('now', ...) comparisons in src/*.py.
+
+SQLite ``datetime()`` returns a space-separated string (``2026-08-23 00:18:04``).
+Columns in this codebase are written by ``datetime.now(timezone.utc).isoformat()``,
+so they carry a ``T`` separator (``2026-08-23T00:20:04.577131+00:00``).
+
+Comparing them with ``>=`` / ``<=`` / ``>`` / ``<`` is a silent bug: ``T > ' '``,
+so once the dates match the time-of-day never participates and the window
+becomes "everything today".
+
+Use ``julianday()`` on both sides of the comparison instead.
+"""
+
+from __future__ import annotations
+
+import re
+from pathlib import Path
+
+import pytest
+
+SRC = Path(__file__).resolve().parent.parent / "src"
+
+_COLUMNS = (
+    "observed_at|ticked_at|created_at|opened_at|"
+    "last_fired_at|resolved_at|updated_at"
+)
+
+# Column on the left:  observed_at >= datetime('now', ...)
+_RE_COL_OP_NOW = re.compile(
+    rf"\b(?:{_COLUMNS})\s*(?:>=|<=|>|<)\s*"
+    rf"datetime\(\s*'now'"
+)
+
+# Column on the right:  datetime('now', ...) <= observed_at
+_RE_NOW_OP_COL = re.compile(
+    rf"datetime\(\s*'now'[^)]*\)\s*(?:>=|<=|>|<)\s*"
+    rf"\b(?:{_COLUMNS})\b"
+)
+
+_FAIL_MSG = (
+    "compare with julianday(col) >= julianday('now', ...); "
+    "the column is ISO with a T, "
+    "datetime() returns a space, "
+    "and 'T' > ' ' makes the time of day never participate."
+)
+
+
+def _iter_source_files() -> list[Path]:
+    """Return all .py files under src/, sorted."""
+    return sorted(SRC.rglob("*.py"))
+
+
+# ---------------------------------------------------------------------------
+# Tripwire
+# ---------------------------------------------------------------------------
+
+
+def test_no_iso_t_datetime_comparison():
+    """Fail if any src/*.py file compares an ISO-T column against
+    datetime('now', ...) with an operator."""
+    hits: list[tuple[Path, int, str]] = []
+    for fpath in _iter_source_files():
+        text = fpath.read_text(encoding="utf-8")
+        for lineno, line in enumerate(text.splitlines(), 1):
+            if _RE_COL_OP_NOW.search(line) or _RE_NOW_OP_COL.search(line):
+                hits.append((fpath, lineno, line.strip()))
+
+    if hits:
+        parts = [_FAIL_MSG, ""]
+        for fpath, lineno, line in hits:
+            parts.append(f"  {fpath}:{lineno}  {line}")
+        pytest.fail("\n".join(parts))
+
+
+# ---------------------------------------------------------------------------
+# Positive controls: verify the regexes themselves are correct
+# ---------------------------------------------------------------------------
+
+
+class TestRegexPositiveControls:
+    """Regexes must match the bad pattern and reject the safe pattern."""
+
+    # --- bad pattern (column op datetime) -----------------------------------
+
+    def test_re_col_op_now_matches_forbidden_pattern(self):
+        assert _RE_COL_OP_NOW.search(
+            "AND observed_at >= datetime('now', '-24 hours')"
+        )
+
+    def test_re_now_op_col_matches_forbidden_pattern(self):
+        assert _RE_NOW_OP_COL.search(
+            "AND datetime('now', '-24 hours') <= observed_at"
+        )
+
+    def test_re_col_op_now_matches_gt(self):
+        assert _RE_COL_OP_NOW.search("WHERE observed_at > datetime('now')")
+
+    def test_re_col_op_now_matches_lt(self):
+        assert _RE_COL_OP_NOW.search("WHERE ticked_at < datetime('now')")
+
+    def test_re_now_op_col_matches_ge(self):
+        assert _RE_NOW_OP_COL.search(
+            "datetime('now') >= opened_at"
+        )
+
+    # --- safe pattern (julianday) -------------------------------------------
+
+    def test_re_col_op_now_rejects_julianday(self):
+        assert not _RE_COL_OP_NOW.search(
+            "AND julianday(observed_at) >= julianday('now', '-24 hours')"
+        )
+
+    def test_re_now_op_col_rejects_julianday(self):
+        assert not _RE_NOW_OP_COL.search(
+            "AND julianday(observed_at) >= julianday('now', '-24 hours')"
+        )
+
+    def test_both_regexes_reject_julianday_all_columns(self):
+        """Spot-check a few column names with julianday wrappers."""
+        for col in ("observed_at", "created_at", "updated_at", "last_fired_at",
+                     "resolved_at", "ticked_at", "opened_at"):
+            sql = (
+                f"AND julianday({col}) >= "
+                f"julianday('now', '-24 hours')"
+            )
+            assert not _RE_COL_OP_NOW.search(sql), f"col side failed for {col}"
+            assert not _RE_NOW_OP_COL.search(sql), f"now side failed for {col}"
+
+    # --- regression: docstring must not trip the scan -----------------------
+
+    def test_docstring_does_not_trip(self):
+        """The docstring near dispatcher.py:3187 mentions both names without
+        an operator between them."""
+        line = "``datetime('now', ...)``. ``observed_at`` is written by"
+        assert not _RE_COL_OP_NOW.search(line)
+        assert not _RE_NOW_OP_COL.search(line)