Add heartbeat system test coverage

Tests for heartbeat CLI/PID/log behavior, dream.daily_complete emission,
supervisor trigger handler, and scheduler fallback registration.

tests/test_heartbeat.py

···
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright (c) 2026 sol pbc
+
+import argparse
+import os
+
+import pytest
+
+
[e-mail]
+def journal_path(tmp_path, monkeypatch):
+    monkeypatch.setenv("_SOLSTONE_JOURNAL_OVERRIDE", str(tmp_path))
+    (tmp_path / "health").mkdir()
+    return tmp_path
+
+
[e-mail]
+def heartbeat_mocks(monkeypatch):
+    monkeypatch.setattr("think.heartbeat.setup_cli", lambda parser: argparse.Namespace())
+    monkeypatch.setattr("think.heartbeat.ensure_sol_directory", lambda *args, **kwargs: None)
+    monkeypatch.setattr("think.heartbeat.cortex_request", lambda *args, **kwargs: "agent-123")
+    monkeypatch.setattr(
+        "think.heartbeat.wait_for_agents",
+        lambda *args, **kwargs: ({"agent-123": "finish"}, []),
+    )
+
+
+def test_heartbeat_command_mapping():
+    """heartbeat key in COMMANDS maps to think.heartbeat module."""
+    from sol import COMMANDS
+
+    assert COMMANDS["heartbeat"] == "think.heartbeat"
+
+
+def test_heartbeat_main_is_callable():
+    """think.heartbeat.main is a callable function."""
+    from think.heartbeat import main
+
+    assert callable(main)
+
+
+def test_pid_guard_live_process_exits_zero(journal_path, heartbeat_mocks):
+    """When PID file contains current process PID, main() exits 0 without cortex."""
+    import think.heartbeat as mod
+
+    pid_file = journal_path / "health" / "heartbeat.pid"
+    pid_file.write_text(str(os.getpid()))
+
+    mod.cortex_request = lambda *a, **kw: pytest.fail("cortex_request should not be called")
+
+    with pytest.raises(SystemExit) as exc_info:
+        mod.main()
+    assert exc_info.value.code == 0
+
+
+def test_pid_guard_dead_process_removes_stale_pid(journal_path, heartbeat_mocks):
+    """When PID file contains a dead PID, main() removes it and proceeds to cortex."""
+    import think.heartbeat as mod
+
+    pid_file = journal_path / "health" / "heartbeat.pid"
+    dead_pid = 99999999
+    try:
+        os.kill(dead_pid, 0)
+        pytest.skip("PID 99999999 is unexpectedly alive")
+    except ProcessLookupError:
+        pass
+
+    pid_file.write_text(str(dead_pid))
+
+    cortex_called = []
+
+    def fake_cortex(*args, **kwargs):
+        cortex_called.append(True)
+        return "agent-123"
+
+    mod.cortex_request = fake_cortex
+
+    with pytest.raises(SystemExit) as exc_info:
+        mod.main()
+    assert exc_info.value.code == 0
+    assert len(cortex_called) == 1
+
+
+def test_pid_file_created_and_removed_on_success(journal_path, heartbeat_mocks):
+    """PID file exists during execution and is removed after main() completes."""
+    import think.heartbeat as mod
+
+    pid_file = journal_path / "health" / "heartbeat.pid"
+    pid_during_run = []
+
+    def capture_pid_cortex(*args, **kwargs):
+        pid_during_run.append(pid_file.exists())
+        if pid_file.exists():
+            pid_during_run.append(pid_file.read_text().strip())
+        return "agent-123"
+
+    mod.cortex_request = capture_pid_cortex
+
+    with pytest.raises(SystemExit):
+        mod.main()
+
+    assert pid_during_run[0] is True
+    assert pid_during_run[1] == str(os.getpid())
+    assert not pid_file.exists()
+
+
+def test_pid_file_removed_on_error(journal_path, heartbeat_mocks):
+    """PID file is removed even when cortex_request returns None (error path)."""
+    import think.heartbeat as mod
+
+    pid_file = journal_path / "health" / "heartbeat.pid"
+    mod.cortex_request = lambda *a, **kw: None
+
+    with pytest.raises(SystemExit) as exc_info:
+        mod.main()
+    assert exc_info.value.code == 1
+    assert not pid_file.exists()
+
+
+def test_pid_file_removed_on_timeout(journal_path, heartbeat_mocks):
+    """PID file is removed on timeout path."""
+    import think.heartbeat as mod
+
+    pid_file = journal_path / "health" / "heartbeat.pid"
+    mod.wait_for_agents = lambda *a, **kw: ({}, ["agent-123"])
+
+    with pytest.raises(SystemExit) as exc_info:
+        mod.main()
+    assert exc_info.value.code == 2
+    assert not pid_file.exists()
+
+
+def test_log_run_appends_line(journal_path):
+    """_log_run appends a correctly formatted line to heartbeat.log."""
+    import time
+
+    from think.heartbeat import _log_run
+
+    health_dir = journal_path / "health"
+    start_time = time.monotonic() - 5
+
+    _log_run(health_dir, start_time, "success")
+
+    log_file = health_dir / "heartbeat.log"
+    assert log_file.exists()
+    content = log_file.read_text()
+    assert content.endswith("\n")
+    line = content.strip()
+    assert "duration=" in line
+    assert "outcome=success" in line
+
+
+def test_log_written_after_successful_run(journal_path, heartbeat_mocks):
+    """After a successful main() run, heartbeat.log has a success entry."""
+    import think.heartbeat as mod
+
+    with pytest.raises(SystemExit) as exc_info:
+        mod.main()
+    assert exc_info.value.code == 0
+
+    log_file = journal_path / "health" / "heartbeat.log"
+    assert log_file.exists()
+    content = log_file.read_text()
+    assert "outcome=success" in content
+
+
+def test_dream_emit_daily_complete_shape(monkeypatch):
+    """dream.emit('daily_complete', ...) calls _callosum.emit with correct tract and fields."""
+    from unittest.mock import Mock
+
+    import think.dream as dream_mod
+
+    mock_conn = Mock()
+    monkeypatch.setattr(dream_mod, "_callosum", mock_conn)
+
+    dream_mod.emit("daily_complete", day="20260318", success=3, failed=0, duration_ms=5000)
+
+    mock_conn.emit.assert_called_once_with(
+        "dream",
+        "daily_complete",
+        day="20260318",
+        success=3,
+        failed=0,
+        duration_ms=5000,
+    )
+
+
+def test_dream_emit_noop_without_callosum(monkeypatch):
+    """dream.emit() does nothing when _callosum is None."""
+    import think.dream as dream_mod
+
+    monkeypatch.setattr(dream_mod, "_callosum", None)
+    dream_mod.emit("daily_complete", day="20260318")

tests/test_scheduler.py

···
         assert "due" in status[0]
 
 
+class TestHeartbeatSchedule:
+    """Tests for heartbeat schedule registration and daily firing."""
+
+    def test_register_defaults_creates_heartbeat(self, journal_path):
+        """register_defaults() creates a heartbeat entry in the config file."""
+        import think.scheduler as mod
+
+        mock_cal = Mock()
+        mod.init(mock_cal)
+        mod.register_defaults()
+
+        assert "heartbeat" in mod._entries
+        assert mod._entries["heartbeat"]["cmd"] == ["sol", "heartbeat"]
+        assert mod._entries["heartbeat"]["every"] == "daily"
+
+        config_path = journal_path / "config" / "schedules.json"
+        assert config_path.exists()
+        with open(config_path) as f:
+            raw = json.load(f)
+        assert "heartbeat" in raw
+        assert raw["heartbeat"]["cmd"] == ["sol", "heartbeat"]
+
+    def test_register_defaults_idempotent(self, journal_path):
+        """register_defaults() does not overwrite existing heartbeat config."""
+        import think.scheduler as mod
+
+        _write_config(
+            journal_path,
+            {
+                "heartbeat": {
+                    "cmd": ["sol", "heartbeat", "--custom"],
+                    "every": "daily",
+                    "enabled": True,
+                }
+            },
+        )
+
+        mock_cal = Mock()
+        mod.init(mock_cal)
+        mod.register_defaults()
+
+        assert mod._entries["heartbeat"]["cmd"] == ["sol", "heartbeat", "--custom"]
+
+    def test_heartbeat_is_due_when_never_run(self, journal_path):
+        """_is_due returns True for heartbeat entry with no prior run."""
+        import think.scheduler as mod
+
+        entry = {"cmd": ["sol", "heartbeat"], "every": "daily", "enabled": True}
+        now = datetime(2026, 3, 19, 10, 0, 0)
+        assert mod._is_due(entry, None, now) is True
+
+    def test_heartbeat_not_due_when_recently_run(self, journal_path):
+        """_is_due returns False for heartbeat entry that ran after the daily mark."""
+        import think.scheduler as mod
+
+        entry = {"cmd": ["sol", "heartbeat"], "every": "daily", "enabled": True}
+        now = datetime(2026, 3, 19, 10, 0, 0)
+        last_run_ts = datetime(2026, 3, 19, 1, 0, 0).timestamp()
+        state_entry = {"last_run": last_run_ts}
+        assert mod._is_due(entry, state_entry, now) is False
+
+
 # ---------------------------------------------------------------------------
 # CLI main()
 # ---------------------------------------------------------------------------

tests/test_supervisor_schedule.py

···
 
 """Test supervisor daily scheduling functionality."""
 
+import os
 from datetime import date
 from unittest.mock import patch
 
···
         handle_daily_tasks()
 
     assert captured_exclude["value"] == {"20250102"}
+
+
+def test_handle_dream_daily_complete_submits_heartbeat(
+    mock_callosum, tmp_path, monkeypatch
+):
+    """_handle_dream_daily_complete submits heartbeat when no PID file exists."""
+    import think.supervisor as mod
+
+    monkeypatch.setenv("_SOLSTONE_JOURNAL_OVERRIDE", str(tmp_path))
+    (tmp_path / "health").mkdir(exist_ok=True)
+
+    submitted = []
+    original_submit = mod._task_queue.submit
+
+    def capture_submit(cmd, *args, **kwargs):
+        submitted.append(cmd)
+        return original_submit(cmd, *args, **kwargs)
+
+    mod._task_queue.submit = capture_submit
+
+    message = {
+        "tract": "dream",
+        "event": "daily_complete",
+        "day": "20260318",
+        "success": 3,
+        "failed": 0,
+        "duration_ms": 5000,
+    }
+    mod._handle_dream_daily_complete(message)
+
+    assert len(submitted) == 1
+    assert submitted[0] == ["sol", "heartbeat"]
+
+
+def test_handle_dream_daily_complete_ignores_wrong_event(
+    mock_callosum, tmp_path, monkeypatch
+):
+    """_handle_dream_daily_complete ignores messages with wrong tract or event."""
+    import think.supervisor as mod
+
+    monkeypatch.setenv("_SOLSTONE_JOURNAL_OVERRIDE", str(tmp_path))
+    (tmp_path / "health").mkdir(exist_ok=True)
+
+    submitted = []
+    original_submit = mod._task_queue.submit
+
+    def capture_submit(cmd, *args, **kwargs):
+        submitted.append(cmd)
+        return original_submit(cmd, *args, **kwargs)
+
+    mod._task_queue.submit = capture_submit
+
+    mod._handle_dream_daily_complete({"tract": "supervisor", "event": "daily_complete"})
+    mod._handle_dream_daily_complete({"tract": "dream", "event": "started"})
+    mod._handle_dream_daily_complete({})
+
+    assert len(submitted) == 0
+
+
+def test_handle_dream_daily_complete_skips_when_pid_alive(
+    mock_callosum, tmp_path, monkeypatch
+):
+    """_handle_dream_daily_complete does not submit when PID file shows running process."""
+    import think.supervisor as mod
+
+    monkeypatch.setenv("_SOLSTONE_JOURNAL_OVERRIDE", str(tmp_path))
+    health = tmp_path / "health"
+    health.mkdir(exist_ok=True)
+
+    (health / "heartbeat.pid").write_text(str(os.getpid()))
+
+    submitted = []
+    original_submit = mod._task_queue.submit
+
+    def capture_submit(cmd, *args, **kwargs):
+        submitted.append(cmd)
+        return original_submit(cmd, *args, **kwargs)
+
+    mod._task_queue.submit = capture_submit
+
+    message = {"tract": "dream", "event": "daily_complete", "day": "20260318"}
+    mod._handle_dream_daily_complete(message)
+
+    assert len(submitted) == 0
+
+
+def test_handle_dream_daily_complete_proceeds_on_dead_pid(
+    mock_callosum, tmp_path, monkeypatch
+):
+    """_handle_dream_daily_complete submits heartbeat when PID file has dead process."""
+    import think.supervisor as mod
+
+    monkeypatch.setenv("_SOLSTONE_JOURNAL_OVERRIDE", str(tmp_path))
+    health = tmp_path / "health"
+    health.mkdir(exist_ok=True)
+    (health / "heartbeat.pid").write_text("99999999")
+
+    submitted = []
+    original_submit = mod._task_queue.submit
+
+    def capture_submit(cmd, *args, **kwargs):
+        submitted.append(cmd)
+        return original_submit(cmd, *args, **kwargs)
+
+    mod._task_queue.submit = capture_submit
+
+    message = {"tract": "dream", "event": "daily_complete", "day": "20260318"}
+    mod._handle_dream_daily_complete(message)
+
+    assert len(submitted) == 1
+    assert submitted[0] == ["sol", "heartbeat"]