Beat on startup and on a timer, so an idle watcher is not reported dead

s16code/main.py

@@ -7,9 +7,10 @@
 """
 from __future__ import annotations
 
+import asyncio
 import os
 import time
-from contextlib import asynccontextmanager
+from contextlib import asynccontextmanager, suppress
 from pathlib import Path
 
 import httpx
@@ -46,6 +47,27 @@ async def lifespan(app: FastAPI):
     data_dir = app.state.runtime.root
     app.state.event_store = EventStore(data_dir / "events")
     app.state.event_engine = AutonomousEventEngine(app.state.event_store, app.state.runtime)
+
+    # The pulse has to mean "I am alive", not "an event arrived". Beating only on
+    # ingest made a correctly idle watcher report itself dead after
+    # STALE_AFTER_SECONDS, so the alarm fired on the very behaviour Section 8 calls
+    # right, and a fresh process was dead on arrival until something happened to it.
+    app.state.event_store.beat(note="process started")
+    heartbeat_seconds = max(1.0, float(os.getenv("S16_HEARTBEAT_SECONDS", "60")))
+
+    async def _heartbeat() -> None:
+        while True:
+            try:
+                await asyncio.sleep(heartbeat_seconds)
+                app.state.event_store.beat(note="idle heartbeat")
+            except asyncio.CancelledError:
+                raise
+            except Exception:
+                # A failed beat must not kill the pulse: the next tick tries again,
+                # and a genuinely stuck store still goes stale and raises the alarm.
+                continue
+
+    app.state.heartbeat_task = asyncio.create_task(_heartbeat())
     bearers, api_keys = _secrets("S16_A2A_BEARER_TOKENS"), _secrets("S16_A2A_API_KEYS")
     base_url = os.getenv("S16_BASE_URL", f"http://127.0.0.1:{PORT}").rstrip("/")
     card = {
@@ -119,6 +141,9 @@ async def handle_a2a_task(text: str) -> str:
         await app.state.a2a_grpc_server.start()
     app.state.started_at = time.time()
     yield
+    app.state.heartbeat_task.cancel()
+    with suppress(asyncio.CancelledError):
+        await app.state.heartbeat_task
     if app.state.a2a_grpc_server:
         await app.state.a2a_grpc_server.stop()
     await app.state.a2a_server.close()

tests/test_control_plane_auth.py

@@ -70,18 +70,29 @@ def test_the_completion_callback_has_its_own_token_and_also_fails_closed(
 
 
 def test_read_only_observability_is_available_to_an_operator(app_client) -> None:
-    """Liveness answers 503 when the watcher is silent, which is the alarm."""
+    """Liveness answers 200 while beating, and the report is readable unauthenticated.
+
+    This previously asserted 503 for a freshly started process, because the
+    heartbeat only advanced when an event arrived, so a process that had just come
+    up had never beaten. That made a healthy idle watcher indistinguishable from a
+    dead one — the exact ambiguity §9 exists to remove. The process now beats on
+    startup and on a timer, so a live agent says so with no events at all.
+
+    The alarm itself is asserted separately, against a genuinely stale beat, in
+    `test_liveness_does_not_require_events.py`.
+    """
     liveness = app_client.get("/v1/agent/liveness")
-    assert liveness.status_code == 503
-    assert liveness.json()["alive"] is False
+    assert liveness.status_code == 200
+    assert liveness.json()["alive"] is True
 
     report = app_client.get("/v1/agent/report")
     assert report.status_code == 200
     assert report.json()["totals"]["events_seen"] == 0
 
     markdown = app_client.get("/v1/agent/report", params={"fmt": "markdown"})
     assert "Overnight report" in markdown.text
-    assert "NOT ALIVE" in markdown.text
+    assert "**Watcher:** alive" in markdown.text
+    assert "NOT ALIVE" not in markdown.text
 
 
 def test_the_operator_console_is_served_and_is_read_only(app_client) -> None:

tests/test_liveness_does_not_require_events.py

@@ -0,0 +1,127 @@
+"""A living watcher must say so, even on a night when nothing happens.
+
+`EventStore.beat()` has exactly one caller: `AutonomousEventEngine.process`. So the
+heartbeat advances only when an event arrives. There is no beat at startup and none
+on a timer.
+
+Section 9 is entirely about the ambiguity this creates:
+
+    nothing happened, because nothing needed to
+                        or
+    it died at 11pm and has not noticed anything since
+
+and its remedy: "The agent taps a counter every time it *wakes up* and every time
+it handles an event. Answers 200 while it is beating. Answers 503 once the last
+beat is too old." The wake-up beat is missing, so the counter only ever means "an
+event arrived", never "I am alive".
+
+## What breaks
+
+A healthy agent that correctly receives nothing for `stale_after_seconds` (900)
+reports itself dead:
+
+    $ curl http://127.0.0.1:8113/v1/agent/liveness      # process up, serving
+    {"alive": false,
+     "reason": "no heartbeat for 16235s; the watcher may have died",
+     "beats": 15,
+     "last_beat": "2026-08-10T13:54:30Z"}               # the last EVENT
+
+Observed live on 2026-08-10, minutes after a restart, while the process answered
+every other request normally.
+
+Three consequences:
+
+1. The alarm fires on correct behaviour. Section 8 says "a quiet night is often the
+   right answer"; here a quiet night returns 503. Anyone paging on that endpoint is
+   woken by every idle night, learns to ignore it, and then misses a real death.
+2. The morning report's header - the one line that tells you whether the watcher
+   was up all night - reads "NOT ALIVE ... the watcher may have died" for a process
+   that never stopped.
+3. A freshly started agent is dead on arrival: with no events yet, liveness reports
+   "no heartbeat has ever been recorded" until something happens to it.
+
+The direction of the signal is what Section 9 cares about: "a dead process cannot
+send you an error, but it also cannot fake a pulse." A pulse that only exists when
+work arrives cannot distinguish idleness from death, which is the one thing it is
+for.
+"""
+
+from __future__ import annotations
+
+import os
+
+
+def test_a_freshly_started_process_reports_itself_alive(app_client) -> None:
+    """No events have been ingested. The process is up, so it must say so."""
+    response = app_client.get("/v1/agent/liveness")
+    body = response.json()
+
+    assert body["alive"] is True, (
+        f"a running process with no events reported alive={body['alive']} "
+        f"({body.get('reason')!r}); an idle watcher is indistinguishable from a dead one"
+    )
+    assert response.status_code == 200, (
+        f"liveness answered {response.status_code} for a healthy process, so an "
+        "uptime monitor pages on a quiet night"
+    )
+
+
+def test_the_pulse_exists_without_any_event_arriving(app_client) -> None:
+    """The beat must mean "I am alive", not "an event arrived"."""
+    body = app_client.get("/v1/agent/liveness").json()
+
+    assert body.get("beats", 0) >= 1, (
+        "no beat was ever recorded, so the counter only advances when work arrives"
+    )
+    assert body.get("last_beat") is not None
+    assert body["reason"] == "beating"
+
+
+def test_a_periodic_beat_is_scheduled(app_client) -> None:
+    """Startup alone is not enough: the pulse has to keep going while idle."""
+    from s16code.main import app
+
+    task = getattr(app.state, "heartbeat_task", None)
+    assert task is not None, (
+        "no background heartbeat is scheduled, so a process that starts and then "
+        "sits idle goes stale after stale_after_seconds and reports itself dead"
+    )
+    assert not task.done(), "the heartbeat task is not running"
+
+
+def test_the_beat_interval_stays_well_inside_the_staleness_window(app_client) -> None:
+    """A beat slower than the window would make the alarm fire on a live agent."""
+    from s16code.events.report import STALE_AFTER_SECONDS
+
+    interval = float(os.getenv("S16_HEARTBEAT_SECONDS", "60"))
+    assert 0 < interval < STALE_AFTER_SECONDS / 2, (
+        f"beat interval {interval}s against a {STALE_AFTER_SECONDS}s staleness "
+        "window leaves no margin for a slow write or a paused container"
+    )
+
+
+class _FakeStore:
+    def __init__(self, raw: dict) -> None:
+        self._raw = raw
+
+    def liveness(self) -> dict:
+        return dict(self._raw)
+
+
+def test_liveness_still_reports_death_when_the_beat_really_stops() -> None:
+    """The fix must not make the endpoint answer 200 unconditionally."""
+    from s16code.events.report import liveness_status
+
+    stale = liveness_status(_FakeStore({"last_beat": "2020-01-01T00:00:00+00:00", "beats": 3}),
+                            stale_after=900)
+    assert stale["alive"] is False
+    assert "may have died" in stale["reason"]
+
+
+def test_a_store_that_never_beat_is_not_reported_alive() -> None:
+    """Absence of any pulse remains the alarm, as Section 9 requires."""
+    from s16code.events.report import liveness_status
+
+    never = liveness_status(_FakeStore({}), stale_after=900)
+    assert never["alive"] is False
+    assert never["reason"] == "no heartbeat has ever been recorded"