From a657397769ab69b3bc72afca38161e04ee36aff7 Mon Sep 17 00:00:00 2001
From: Ben <ben@nousresearch.com>
Date: Thu, 18 Jun 2026 13:08:21 +1000
Subject: [PATCH] test(cron): characterize in-process + desktop ticker contract
 before provider refactor

---
 tests/cron/test_scheduler_provider.py | 83 +++++++++++++++++++++++++++
 1 file changed, 83 insertions(+)
 create mode 100644 tests/cron/test_scheduler_provider.py

diff --git a/tests/cron/test_scheduler_provider.py b/tests/cron/test_scheduler_provider.py
new file mode 100644
index 00000000000..1e94347dfa8
--- /dev/null
+++ b/tests/cron/test_scheduler_provider.py
@@ -0,0 +1,83 @@
+"""Characterization tests for the cron trigger before/after the provider refactor.
+
+These lock the CURRENT in-process-ticker contract (Phase 0 of the pluggable
+CronScheduler plan, .hermes/plans/cron-scheduler-provider-interface.md). They
+must pass unchanged on `main` now, and after every subsequent phase of the
+refactor — they are the regression harness that proves the built-in firing
+behavior is byte-for-byte preserved when the ticker is moved behind the
+CronScheduler provider interface.
+
+No production code is exercised beyond the two ticker entry points:
+  - gateway/run.py::_start_cron_ticker        (production gateway ticker)
+  - hermes_cli/web_server.py::_start_desktop_cron_ticker  (desktop fallback)
+
+Both call `cron.scheduler.tick(...)` on a loop and exit when their stop_event
+is set. We patch `cron.scheduler.tick` (both tickers import it locally as
+`cron_tick`, so the module-attribute patch is observed) and assert the loop
+drives it and stops promptly.
+"""
+import threading
+import time
+from unittest.mock import patch
+
+
+def test_ticker_calls_tick_at_least_once_then_stops():
+    """The gateway in-process ticker loop calls cron.scheduler.tick repeatedly
+    and exits promptly once the stop_event is set."""
+    from gateway.run import _start_cron_ticker
+
+    calls = []
+    stop = threading.Event()
+
+    def fake_tick(*args, **kwargs):
+        calls.append(kwargs)
+        return 0
+
+    with patch("cron.scheduler.tick", side_effect=fake_tick):
+        # interval=0 keeps the loop tight; stop after a brief beat.
+        t = threading.Thread(
+            target=_start_cron_ticker,
+            args=(stop,),
+            kwargs={"interval": 0},
+            daemon=True,
+        )
+        t.start()
+        time.sleep(0.2)
+        stop.set()
+        t.join(timeout=5)
+
+    assert not t.is_alive(), "ticker did not exit after stop_event was set"
+    assert len(calls) >= 1, "ticker never called tick()"
+    # Contract: the ticker invokes tick with sync=False (fire-and-forget from
+    # the background thread, never the synchronous CLI path).
+    assert calls[0].get("sync") is False
+
+
+def test_desktop_ticker_calls_tick_then_stops():
+    """The desktop dashboard ticker loop calls cron.scheduler.tick and exits
+    once the stop_event is set. Desktop has no live adapters, so it ticks with
+    no adapters/loop."""
+    from hermes_cli.web_server import _start_desktop_cron_ticker
+
+    calls = []
+    stop = threading.Event()
+
+    def fake_tick(*args, **kwargs):
+        calls.append(kwargs)
+        return 0
+
+    with patch("cron.scheduler.tick", side_effect=fake_tick):
+        t = threading.Thread(
+            target=_start_desktop_cron_ticker,
+            args=(stop,),
+            kwargs={"interval": 0},
+            daemon=True,
+        )
+        t.start()
+        time.sleep(0.2)
+        stop.set()
+        t.join(timeout=5)
+
+    assert not t.is_alive(), "desktop ticker did not exit after stop_event was set"
+    assert len(calls) >= 1, "desktop ticker never called tick()"
+    assert calls[0].get("sync") is False