fix(hosts): audit only liveness transitions, not every heartbeat

apps/control_plane/lib/workharbor/host_connections.ex

@@ -6,8 +6,8 @@ defmodule WorkHarbor.HostConnections do
   alias WorkHarbor.{Repo, Hosts, Accounts}
   alias WorkHarbor.HostConnections.Protocol
   alias WorkHarbor.Hosts.Peer
-  # Contract cadence is 10 s. Each heartbeat costs a transaction and an undeletable audit
-  # row, so arrivals are limited by a token bucket (capacity 3, one token per 5 s) measured
+  # Contract cadence is 10 s. Each heartbeat costs a transaction (no audit row; only liveness
+  # transitions are audited), so arrivals are limited by a token bucket (capacity 3, one token per 5 s) measured
   # on receive time, not on the previous commit; the first heartbeat after hello is free.
   @heartbeat_refill_ms 5000
   @heartbeat_capacity 3
@@ -91,11 +91,33 @@ defmodule WorkHarbor.HostConnections do
       case Hosts.transaction(fn ->
              if not Hosts.current_peer?(peer), do: Repo.rollback(:unauthorized_host)
 
+             # A cleanly closed slot has no connection_id, so a remaining one means a live or
+             # unreleased connection is being superseded, even if its close write failed.
+             replacing? =
+               not is_nil(
+                 Hosts.one(
+                   "SELECT host_id FROM host_connection_slots WHERE host_id=? AND connection_id IS NOT NULL",
+                   [peer.host_id]
+                 )
+               )
+
              Hosts.sql(
                "INSERT INTO host_connection_slots(host_id,owner_id,connection_id,state) VALUES(?,?,?,'unknown') ON CONFLICT(host_id) DO UPDATE SET connection_id=excluded.connection_id,state='unknown'",
                [peer.host_id, peer.owner_id, id]
              )
 
+             # The old connection's close no longer matches the slot, so the replacement itself
+             # is the one audited transition for it.
+             if replacing?,
+               do:
+                 Hosts.audit(
+                   peer.owner_id,
+                   peer.host_id,
+                   "host_connection_replaced",
+                   "host_replace",
+                   "unknown"
+                 )
+
              Hosts.audit(peer.owner_id, peer.host_id, "host_connected", "host_connect", "unknown")
            end) do
         {:ok, _} ->
@@ -350,7 +372,7 @@ defmodule WorkHarbor.HostConnections do
           Hosts.transaction(fn ->
             changed =
               Hosts.sql(
-                "UPDATE host_connection_slots SET state='unknown' WHERE host_id=? AND connection_id=?",
+                "UPDATE host_connection_slots SET state='unknown',connection_id=NULL WHERE host_id=? AND connection_id=?",
                 [entry.peer.host_id, entry.id]
               )
 
@@ -359,7 +381,10 @@ defmodule WorkHarbor.HostConnections do
                 Hosts.audit(
                   entry.peer.owner_id,
                   entry.peer.host_id,
-                  "host_connection_unknown",
+                  if(reason == :heartbeat_too_fast,
+                    do: "host_heartbeat_refused",
+                    else: "host_connection_unknown"
+                  ),
                   "host_disconnect",
                   "unknown"
                 )
@@ -423,14 +448,6 @@ defmodule WorkHarbor.HostConnections do
                "UPDATE host_connection_slots SET state='connected' WHERE host_id=? AND connection_id=?",
                [entry.peer.host_id, entry.id]
              )
-
-             Hosts.audit(
-               entry.peer.owner_id,
-               entry.peer.host_id,
-               "host_heartbeat",
-               "host_heartbeat",
-               "authenticated"
-             )
            end) do
         {:ok, _} ->
           entry = %{entry | seen: monotonic(), lease: make_ref()}

apps/control_plane/test/workharbor/host_connections_test.exs

@@ -108,21 +108,111 @@ defmodule WorkHarbor.HostConnectionsTest do
     refute_received {:host_connection_close, _}
   end
 
-  test "sustained heartbeat burst beyond capacity closes without a heartbeat write", c do
-    ref = connect(c)
+  defp audits(c) do
+    Repo.query!(
+      "SELECT event FROM audit_events WHERE host_id=? AND (event LIKE 'host_conn%' OR event LIKE 'host_heartbeat%') ORDER BY rowid",
+      [c.peer.host_id]
+    ).rows
+    |> List.flatten()
+  end
 
-    count = fn ->
-      Repo.query!("SELECT count(*) FROM audit_events WHERE event='host_heartbeat'").rows
-    end
+  test "sustained heartbeat burst beyond capacity closes with one refusal audit", c do
+    ref = connect(c)
 
     for _ <- 1..3 do
       assert {:ok, _} = HostConnections.heartbeat(ref, message(c, "heartbeat"))
     end
 
-    before = count.()
+    before = audits(c)
     assert {:error, :heartbeat_too_fast} = HostConnections.heartbeat(ref, message(c, "heartbeat"))
     assert_receive {:host_connection_close, :heartbeat_too_fast}
-    assert count.() == before
+    assert audits(c) -- before == ["host_heartbeat_refused"]
+    refute "host_heartbeat" in audits(c)
+  end
+
+  test "steady heartbeats add no audit rows and still renew host last_seen and lease", c do
+    ref = connect(c)
+    before = audits(c)
+    assert before == ["host_connected"]
+
+    hb = fn ->
+      Hosts.one("SELECT last_seen,lease_expires_at FROM hosts WHERE id=?", [c.peer.host_id])
+    end
+
+    Hosts.sql(
+      "UPDATE hosts SET last_seen='2000-01-01T00:00:00Z',lease_expires_at='2000-01-01T00:00:00Z' WHERE id=?",
+      [c.peer.host_id]
+    )
+
+    Hosts.sql("UPDATE host_connection_slots SET state='unknown' WHERE host_id=?", [c.peer.host_id])
+
+    stale = hb.()
+
+    for i <- 1..5 do
+      Agent.update(c.mono, fn _ -> i * 10_000 end)
+
+      assert {:ok, %{"kind" => "heartbeat_ack"}} =
+               HostConnections.heartbeat(ref, message(c, "heartbeat"))
+    end
+
+    assert audits(c) == before
+    assert {:ok, "connected"} = HostConnections.host_state(c.peer.host_id)
+
+    assert %{"state" => "connected"} =
+             Hosts.one("SELECT state FROM host_connection_slots WHERE host_id=?", [c.peer.host_id])
+
+    now = hb.()
+    assert now["last_seen"] > stale["last_seen"]
+    assert now["lease_expires_at"] > stale["lease_expires_at"]
+  end
+
+  test "clean close then reopen or restart audits no replacement or reset", c do
+    ref = connect(c)
+    assert :ok = HostConnections.disconnect(ref)
+    assert {:ok, _} = HostConnections.open(c.peer, self())
+
+    assert audits(c) == ["host_connected", "host_connection_unknown", "host_connected"]
+
+    ref2 = connect(c)
+    assert :ok = HostConnections.disconnect(ref2)
+    before = audits(c)
+    :ok = stop_supervised(HostConnections)
+    start_supervised!({HostConnections, repo: c.repo})
+    assert audits(c) == before
+  end
+
+  test "reopen after a failed close write audits the superseded connection once", c do
+    ref = connect(c)
+
+    Hosts.sql(
+      "CREATE TRIGGER fail_close BEFORE UPDATE ON host_connection_slots BEGIN SELECT RAISE(ABORT,'denied'); END"
+    )
+
+    assert {:error, :persistence_failed} = HostConnections.disconnect(ref)
+    Hosts.sql("DROP TRIGGER fail_close")
+    before = audits(c)
+    assert {:ok, _} = HostConnections.open(c.peer, self())
+    assert audits(c) -- before == ["host_connection_replaced", "host_connected"]
+  end
+
+  test "each liveness transition adds exactly one audit row", c do
+    ref = connect(c)
+    assert audits(c) == ["host_connected"]
+
+    assert {:ok, ref2} = HostConnections.open(c.peer, self())
+    assert audits(c) == ["host_connected", "host_connection_replaced", "host_connected"]
+    assert_receive {:host_connection_close, :replaced}
+    assert {:error, _} = HostConnections.heartbeat(ref, message(c, "heartbeat"))
+    assert audits(c) == ["host_connected", "host_connection_replaced", "host_connected"]
+
+    assert :ok = HostConnections.disconnect(ref2)
+
+    assert audits(c) == [
+             "host_connected",
+             "host_connection_replaced",
+             "host_connected",
+             "host_connection_unknown"
+           ]
   end
 
   test "tokens refill at one per five seconds measured from arrival", c do

docs/development/ci-acceptance.json

@@ -1142,7 +1142,11 @@
       "WorkHarbor.HostIdentityTest/test native leaf and issuer profiles reject extra purposes critical and duplicate extensions",
       "WorkHarbor.HostIdentityTest/test issuer startup rejects signed critical path length and duration profile variants",
       "WorkHarbor.HostConnectionsTest/test first heartbeat after hello passes and a catch-up burst within capacity stays connected",
-      "WorkHarbor.HostConnectionsTest/test sustained heartbeat burst beyond capacity closes without a heartbeat write",
+      "WorkHarbor.HostConnectionsTest/test sustained heartbeat burst beyond capacity closes with one refusal audit",
+      "WorkHarbor.HostConnectionsTest/test steady heartbeats add no audit rows and still renew host last_seen and lease",
+      "WorkHarbor.HostConnectionsTest/test clean close then reopen or restart audits no replacement or reset",
+      "WorkHarbor.HostConnectionsTest/test reopen after a failed close write audits the superseded connection once",
+      "WorkHarbor.HostConnectionsTest/test each liveness transition adds exactly one audit row",
       "WorkHarbor.HostConnectionsTest/test tokens refill at one per five seconds measured from arrival",
       "WorkHarbor.HostConnectionsTest/test authenticated heartbeat becomes stale exactly30seconds without changing history",
       "WorkHarbor.HostIdentityTest/test issuing a code revokes the host's earlier unconsumed codes",

specs/006-whr-cli/contracts/host-enrollment.md

@@ -188,8 +188,14 @@ watchdog close. Agent mirrors expiry and stops expired active/pending use.
 Heartbeat immediately after hello then every10s; acknowledge each heartbeat.
 Rate limit: token bucket, capacity3, one token per5s, measured on monotonic receive time
 (not previous commit); first heartbeat after hello free. Heartbeat arriving with no token
-is refused without a heartbeat write and closes the connection (slot/unknown audit as for
-any close).
+is refused without a heartbeat write and closes the connection (audited once as
+`host_heartbeat_refused`).
+Connection audit rows, each exactly once per transition: `host_connected` (authenticated socket open),
+`host_connection_replaced` (new connection supersedes a live one, written in the
+replacement transaction; the superseded close writes none), `host_connection_unknown`
+(stale, disconnect, identity invalid or other close), `host_heartbeat_refused` (rate
+limit), `host_connection_reset` (control-plane restart). Accepted heartbeats update
+hosts last_seen/lease, set the slot to connected and write no audit row.
 Without authenticated heartbeat for>=30s mark unknown and close stale socket.
 Only transport reconnect uses bounded1/2/4/8/16/30s backoff then30s maximum; no blind
 provider resends or command-history resets. Health requires authenticated hello
@@ -248,7 +254,8 @@ never agent identity state/environment. Without confirmation keep agent stopped,
 retain recoverable credentials/data and report incomplete/nonzero; absence is not
 successful offboarding. Default project data retention remains unchanged.
 
-Audit every persisted connection/heartbeat/identity/freshness mutation with fixed
+Audit every persisted connection-liveness transition (not each heartbeat), identity and
+freshness mutation with fixed
 server-known IDs/action/outcome/code only; respect accepted scope constraints and
 one-time operation audit uniqueness. Never raw codes, keys, CSR/certs, requests,
 URLs, names, provider payloads, SQL or exceptions in logs/reports/audit. Secret input