feat(runtime): heartbeat immediately after claim

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a06ba0-10aa-7ea0-b20a-4f3fac39efe9

tests/test_claim_loop.py

@@ -8,6 +8,8 @@ import time
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
+import pytest
+
 from rein_aharness.approaches import ApproachResult, APPROACH_FI_RESEARCH_BRIEF
 from rein_aharness.claim_loop import (
     _cancel_active_run,
@@ -118,6 +120,24 @@ def _commit_file(repo: Path, relative: str, value: str = "result\n") -> str:
     ).stdout.strip()
 
 
+def _accept_initial_heartbeat_then_reject(run: OpsRun):
+    calls = 0
+
+    def heartbeat(*_args, **_kwargs):
+        nonlocal calls
+        calls += 1
+        if calls == 1:
+            return run
+        raise OpsRunError(
+            "lease rejected",
+            action="heartbeat",
+            status_code=409,
+            code="expired_lease",
+        )
+
+    return heartbeat
+
+
 def test_process_one_empty() -> None:
     client = MagicMock(spec=ActivityCoreOpsClient)
     client.config = OpsRunConfig(worker_id="w", lease_seconds=90)
@@ -166,6 +186,67 @@ def test_process_one_success_completes() -> None:
     client.fail.assert_not_called()
 
 
+def test_process_one_sends_initial_heartbeat_before_dispatch() -> None:
+    client = MagicMock(spec=ActivityCoreOpsClient)
+    client.config = OpsRunConfig(worker_id="w", lease_seconds=90)
+    run = _claimed_run()
+    events: list[str] = []
+    client.claim.side_effect = lambda **_kwargs: events.append("claim") or [run]
+    client.heartbeat.side_effect = (
+        lambda *_args, **_kwargs: events.append("heartbeat") or run
+    )
+    client.complete.side_effect = lambda *_args, **_kwargs: (
+        events.append("complete") or run
+    )
+    result = ApproachResult(
+        ok=True,
+        approach=APPROACH_FI_RESEARCH_BRIEF,
+        reason="ok",
+    )
+
+    def execute(*_args, **_kwargs):
+        events.append("execute")
+        return result
+
+    with patch("rein_aharness.claim_loop.execute_approach", side_effect=execute):
+        processed = process_one(client)
+
+    assert processed.ok is True
+    assert events == ["claim", "heartbeat", "execute", "complete"]
+    client.heartbeat.assert_called_once_with(run.id, lease_seconds=90)
+
+
+def test_process_one_initial_heartbeat_rejection_refuses_dispatch() -> None:
+    client = MagicMock(spec=ActivityCoreOpsClient)
+    client.config = OpsRunConfig(worker_id="w", lease_seconds=90)
+    run = _claimed_run()
+    client.claim.return_value = [run]
+    client.heartbeat.side_effect = OpsRunError(
+        "lease rejected",
+        action="heartbeat",
+        status_code=409,
+        code="expired_lease",
+    )
+
+    with patch("rein_aharness.claim_loop.execute_approach") as execute:
+        result = process_one(client)
+
+    assert result.claimed is True
+    assert result.ok is False
+    assert result.retry_full_interval is True
+    assert result.reason == "initial heartbeat rejected: lease rejected"
+    assert result.detail == {
+        "lease_loss": {
+            "stage": "pre-dispatch-heartbeat",
+            "status_code": 409,
+            "code": "expired_lease",
+        }
+    }
+    execute.assert_not_called()
+    client.complete.assert_not_called()
+    client.fail.assert_not_called()
+
+
 def test_process_one_failure_reopens() -> None:
     client = MagicMock(spec=ActivityCoreOpsClient)
     client.config = OpsRunConfig(worker_id="w", lease_seconds=90)
@@ -222,7 +303,9 @@ def test_process_one_refuses_close_after_lease_loss() -> None:
     client = MagicMock(spec=ActivityCoreOpsClient)
     client.config = OpsRunConfig(worker_id="w", lease_seconds=90)
     client.claim.return_value = [_claimed_run()]
-    client.heartbeat.side_effect = OpsRunError("lease rejected", status_code=409)
+    client.heartbeat.side_effect = _accept_initial_heartbeat_then_reject(
+        client.claim.return_value[0]
+    )
     ar = ApproachResult(
         ok=True,
         approach=APPROACH_FI_RESEARCH_BRIEF,
@@ -251,7 +334,9 @@ def test_process_one_lease_loss_cancels_registered_adapter_process() -> None:
     client = MagicMock(spec=ActivityCoreOpsClient)
     client.config = OpsRunConfig(worker_id="w", lease_seconds=90)
     client.claim.return_value = [_claimed_run()]
-    client.heartbeat.side_effect = OpsRunError("lease rejected", status_code=409)
+    client.heartbeat.side_effect = _accept_initial_heartbeat_then_reject(
+        client.claim.return_value[0]
+    )
     proc = MagicMock()
     proc.poll.return_value = None
 
@@ -339,7 +424,9 @@ def test_process_one_records_adapter_exception_without_unbound_result() -> None:
 
 def test_profiled_exception_after_lease_loss_skips_close(tmp_path: Path) -> None:
     repo, _run, client = _profiled_case(tmp_path)
-    client.heartbeat.side_effect = OpsRunError("lease rejected", status_code=409)
+    client.heartbeat.side_effect = _accept_initial_heartbeat_then_reject(
+        client.claim.return_value[0]
+    )
 
     def slow_profile(*_args, **_kwargs):
         time.sleep(0.05)
@@ -693,14 +780,28 @@ def test_pending_close_failure_blocks_new_claim(tmp_path: Path) -> None:
     client.claim.assert_not_called()
 
 
-def test_permanent_close_conflict_is_quarantined_before_claim(tmp_path: Path) -> None:
+@pytest.mark.parametrize(
+    "code",
+    (
+        "not_found",
+        "wrong_owner",
+        "expired_lease",
+        "state_conflict",
+        "terminal_conflict",
+        "evidence_conflict",
+    ),
+)
+def test_permanent_close_refusal_is_quarantined_before_claim(
+    tmp_path: Path,
+    code: str,
+) -> None:
     client = MagicMock(spec=ActivityCoreOpsClient)
     client.config = OpsRunConfig(worker_id="w", lease_seconds=90)
     client.complete.side_effect = OpsRunError(
         "conflict",
         action="complete",
-        status_code=409,
-        code="terminal_conflict",
+        status_code=404 if code == "not_found" else 409,
+        code=code,
     )
     client.claim.return_value = []
     outbox = CloseOutbox(state_dir=tmp_path / "state")