feat(evolution): process every 4h (fix backlog growth: ~25 gen/day vs ~1-2 proc/night)

cron/evolution/analysis.yaml

@@ -1,5 +1,5 @@
 name: evolution-analysis
-schedule: "0 21 * * *"  # Daily at 9 PM
+schedule: "0 1,5,9,13,17,21 * * *"  # Every 4h (was daily 21:00). Raises processing throughput vs ~25 issues/day generation. 21:00 slot still follows introspection (20:00). Watchdog STAGES mirrors the FIRST slot (1).
 enabled: true
 mode: PRIVATE
 

cron/evolution/implementation.yaml

@@ -1,5 +1,5 @@
 name: evolution-implementation
-schedule: "0 22 * * *"  # Daily at 10 PM
+schedule: "0 2,6,10,14,18,22 * * *"  # Every 4h, +1h after analysis (1,5,9,...). Watchdog STAGES mirrors the FIRST slot (2).
 enabled: true
 mode: PRIVATE
 

cron/evolution/integration.yaml

@@ -1,5 +1,5 @@
 name: evolution-integration
-schedule: "0 23 * * *"  # Daily 23:00 — after implementation (22:00), CI has settled
+schedule: "0 3,7,11,15,19,23 * * *"  # Every 4h, +1h after implementation (2,6,10,...) so CI settles. Watchdog STAGES mirrors the FIRST slot (3).
 enabled: true
 mode: PRIVATE
 

scripts/evolution_watchdog.py

@@ -41,9 +41,13 @@
 STAGES: Dict[str, Tuple[int, str]] = {
     "research": (9, "md"),
     "introspection": (20, "json"),
-    "analysis": (21, "json"),
-    "implementation": (22, "md"),
-    "integration": (23, "json"),
+    # analysis/implementation/integration run every 4h (processing throughput);
+    # the slot here is the FIRST daily slot — the watchdog only needs "a report
+    # for today exists by then", and reports are date-keyed (overwritten each
+    # run). Mirrors cron/evolution/*.yaml first hour (locked by the mirror test).
+    "analysis": (1, "json"),
+    "implementation": (2, "md"),
+    "integration": (3, "json"),
 }
 
 GRACE_HOURS = 2

tests/scripts/test_evolution_watchdog.py

@@ -39,14 +39,17 @@ def test_within_grace_still_expects_yesterday(self):
 
 
 class TestStageReports:
-    def _make_reports(self, tmp_path, date="2026-06-10", skip=(), tiny=()):
+    def _make_reports(self, tmp_path, date=None, skip=(), tiny=()):
+        # Each stage's report is dated at ITS OWN expected slot date (slot-aware),
+        # so the helper stays correct regardless of per-stage schedules.
         for stage, (slot, ext) in STAGES.items():
             if stage in skip:
                 continue
             d = tmp_path / stage
             d.mkdir(exist_ok=True)
+            dt = date or expected_report_date(NOW, slot)
             content = "x" * 10 if stage in tiny else "x" * 500
-            (d / f"{date}.{ext}").write_text(content)
+            (d / f"{dt}.{ext}").write_text(content)
 
     def test_all_present_no_alerts(self, tmp_path):
         self._make_reports(tmp_path)
@@ -57,7 +60,8 @@ def test_missing_report_alerts(self, tmp_path):
         alerts = check_stage_reports(tmp_path, NOW)
         assert len(alerts) == 1
         assert "implementation" in alerts[0]
-        assert "2026-06-10" in alerts[0]
+        exp = expected_report_date(NOW, STAGES["implementation"][0])
+        assert exp in alerts[0]
 
     def test_trivially_small_report_alerts(self, tmp_path):
         self._make_reports(tmp_path, tiny=("analysis",))
@@ -80,9 +84,14 @@ def _jobs_file(self, tmp_path, name, *, status="ok", last_run="2026-06-10T22:01:
 
     def test_missing_report_quiet_when_job_ran_clean(self, tmp_path):
         # implementation report missing, but its cron job ran ok at/after the
-        # 22:00 slot for the expected date (2026-06-10) → idle clean cycle, no alert.
+        # slot for the expected date → idle clean cycle, no alert. Slot-aware.
+        slot = STAGES["implementation"][0]
+        exp = expected_report_date(NOW, slot)
         self._make_reports(tmp_path, skip=("implementation",))
-        jf = self._jobs_file(tmp_path, "evolution-implementation")
+        jf = self._jobs_file(
+            tmp_path, "evolution-implementation",
+            last_run=f"{exp}T{slot:02d}:01:00",
+        )
         assert check_stage_reports(tmp_path, NOW, jf) == []
 
     def test_missing_report_alerts_when_job_errored(self, tmp_path):
@@ -301,11 +310,14 @@ def test_extension_matches_output_file(self):
     def test_slot_hour_matches_schedule(self):
         for stage, (slot, _ext) in STAGES.items():
             spec = (self.CRON_DIR / f"{stage}.yaml").read_text()
-            m = re.search(r'^schedule:\s*"(\d+)\s+(\d+)\s', spec, re.M)
-            assert m, f"{stage}.yaml has no parsable daily schedule"
-            assert int(m.group(2)) == slot, (
-                f"watchdog STAGES says '{stage}' runs at {slot:02d}:00, "
-                f"but {stage}.yaml schedules hour {m.group(2)}"
+            # Hour field may be a single hour ("21") or a multi-slot list
+            # ("1,5,9,13,17,21"); STAGES mirrors the FIRST slot.
+            m = re.search(r'^schedule:\s*"(\d+)\s+([\d,]+)\s', spec, re.M)
+            assert m, f"{stage}.yaml has no parsable schedule"
+            first_hour = int(m.group(2).split(",")[0])
+            assert first_hour == slot, (
+                f"watchdog STAGES says '{stage}' first slot is {slot:02d}:00, "
+                f"but {stage}.yaml's first scheduled hour is {first_hour}"
             )