Merge PR #1200 to update heartbeat tests to match two-phase tool-call architecture

fix: update heartbeat tests to match two-phase tool-call architecture

tests/test_heartbeat_service.py

@@ -2,34 +2,28 @@ import asyncio
 
 import pytest
 
-from nanobot.heartbeat.service import (
+from nanobot.heartbeat.service import HeartbeatService
-    HEARTBEAT_OK_TOKEN,
+from nanobot.providers.base import LLMResponse, ToolCallRequest
-    HeartbeatService,
-)
 
 
-def test_heartbeat_ok_detection() -> None:
+class DummyProvider:
-    def is_ok(response: str) -> bool:
+    def __init__(self, responses: list[LLMResponse]):
-        return HEARTBEAT_OK_TOKEN in response.upper()
+        self._responses = list(responses)
 
-    assert is_ok("HEARTBEAT_OK")
+    async def chat(self, *args, **kwargs) -> LLMResponse:
-    assert is_ok("`HEARTBEAT_OK`")
+        if self._responses:
-    assert is_ok("**HEARTBEAT_OK**")
+            return self._responses.pop(0)
-    assert is_ok("heartbeat_ok")
+        return LLMResponse(content="", tool_calls=[])
-    assert is_ok("HEARTBEAT_OK.")
-
-    assert not is_ok("HEARTBEAT_NOT_OK")
-    assert not is_ok("all good")
 
 
 @pytest.mark.asyncio
 async def test_start_is_idempotent(tmp_path) -> None:
-    async def _on_heartbeat(_: str) -> str:
+    provider = DummyProvider([])
-        return "HEARTBEAT_OK"
 
     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