From 19a5efa89eaefb9c800f5c888a8a195c1d9fa548 Mon Sep 17 00:00:00 2001
From: Elliot Lee <github.public@intelliot.com>
Date: Wed, 25 Feb 2026 07:47:52 -0800
Subject: [PATCH] fix: update heartbeat tests to match two-phase tool-call
 architecture

HeartbeatService was refactored from free-text HEARTBEAT_OK token
matching to a structured two-phase design (LLM tool call for
skip/run decision, then execution). The tests still used the old
on_heartbeat callback constructor and HEARTBEAT_OK_TOKEN import.

- Remove obsolete test_heartbeat_ok_detection test
- Update test_start_is_idempotent to use new provider+model constructor
- Add tests for _decide() skip path, trigger_now() run/skip paths

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
---
 tests/test_heartbeat_service.py | 109 ++++++++++++++++++++++++++------
 1 file changed, 91 insertions(+), 18 deletions(-)

diff --git a/tests/test_heartbeat_service.py b/tests/test_heartbeat_service.py
index ec91c6b..c5478af 100644
--- a/tests/test_heartbeat_service.py
+++ b/tests/test_heartbeat_service.py
@@ -2,34 +2,28 @@ import asyncio
 
 import pytest
 
-from nanobot.heartbeat.service import (
-    HEARTBEAT_OK_TOKEN,
-    HeartbeatService,
-)
+from nanobot.heartbeat.service import HeartbeatService
+from nanobot.providers.base import LLMResponse, ToolCallRequest
 
 
-def test_heartbeat_ok_detection() -> None:
-    def is_ok(response: str) -> bool:
-        return HEARTBEAT_OK_TOKEN in response.upper()
+class DummyProvider:
+    def __init__(self, responses: list[LLMResponse]):
+        self._responses = list(responses)
 
-    assert is_ok("HEARTBEAT_OK")
-    assert is_ok("`HEARTBEAT_OK`")
-    assert is_ok("**HEARTBEAT_OK**")
-    assert is_ok("heartbeat_ok")
-    assert is_ok("HEARTBEAT_OK.")
-
-    assert not is_ok("HEARTBEAT_NOT_OK")
-    assert not is_ok("all good")
+    async def chat(self, *args, **kwargs) -> LLMResponse:
+        if self._responses:
+            return self._responses.pop(0)
+        return LLMResponse(content="", tool_calls=[])
 
 
 @pytest.mark.asyncio
 async def test_start_is_idempotent(tmp_path) -> None:
-    async def _on_heartbeat(_: str) -> str:
-        return "HEARTBEAT_OK"
+    provider = DummyProvider([])
 
     service = HeartbeatService(
         workspace=tmp_path,
-        on_heartbeat=_on_heartbeat,
+        provider=provider,
+        model="openai/gpt-4o-mini",
         interval_s=9999,
         enabled=True,
     )
@@ -42,3 +36,82 @@ async def test_start_is_idempotent(tmp_path) -> None:
 
     service.stop()
     await asyncio.sleep(0)
+
+
+@pytest.mark.asyncio
+async def test_decide_returns_skip_when_no_tool_call(tmp_path) -> None:
+    provider = DummyProvider([LLMResponse(content="no tool call", tool_calls=[])])
+    service = HeartbeatService(
+        workspace=tmp_path,
+        provider=provider,
+        model="openai/gpt-4o-mini",
+    )
+
+    action, tasks = await service._decide("heartbeat content")
+    assert action == "skip"
+    assert tasks == ""
+
+
+@pytest.mark.asyncio
+async def test_trigger_now_executes_when_decision_is_run(tmp_path) -> None:
+    (tmp_path / "HEARTBEAT.md").write_text("- [ ] do thing", encoding="utf-8")
+
+    provider = DummyProvider([
+        LLMResponse(
+            content="",
+            tool_calls=[
+                ToolCallRequest(
+                    id="hb_1",
+                    name="heartbeat",
+                    arguments={"action": "run", "tasks": "check open tasks"},
+                )
+            ],
+        )
+    ])
+
+    called_with: list[str] = []
+
+    async def _on_execute(tasks: str) -> str:
+        called_with.append(tasks)
+        return "done"
+
+    service = HeartbeatService(
+        workspace=tmp_path,
+        provider=provider,
+        model="openai/gpt-4o-mini",
+        on_execute=_on_execute,
+    )
+
+    result = await service.trigger_now()
+    assert result == "done"
+    assert called_with == ["check open tasks"]
+
+
+@pytest.mark.asyncio
+async def test_trigger_now_returns_none_when_decision_is_skip(tmp_path) -> None:
+    (tmp_path / "HEARTBEAT.md").write_text("- [ ] do thing", encoding="utf-8")
+
+    provider = DummyProvider([
+        LLMResponse(
+            content="",
+            tool_calls=[
+                ToolCallRequest(
+                    id="hb_1",
+                    name="heartbeat",
+                    arguments={"action": "skip"},
+                )
+            ],
+        )
+    ])
+
+    async def _on_execute(tasks: str) -> str:
+        return tasks
+
+    service = HeartbeatService(
+        workspace=tmp_path,
+        provider=provider,
+        model="openai/gpt-4o-mini",
+        on_execute=_on_execute,
+    )
+
+    assert await service.trigger_now() is None