chore(consistency): sync task status from DB [auto]

Updated by fix-consistency on 2026-09-04:
  - update .custodian-brief.md for rein-aharness

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

tests/test_adapter.py

@@ -111,11 +111,58 @@ def test_execute_streaming_raises_timeout(tmp_path) -> None:
 def test_execute_blocking_path_unchanged_without_callback(tmp_path) -> None:
     adapter = AgenticClaudeCodeAdapter(workdir=tmp_path)
     config = RunConfig(timeout_seconds=30)
-    fake_result = MagicMock(returncode=0, stdout="plain output", stderr="")
+    proc = MagicMock()
+    proc.communicate.return_value = ("plain output", "")
+    proc.returncode = 0
+    proc.poll.return_value = 0
 
-    with patch("rein_aharness.adapter.subprocess.run", return_value=fake_result) as run:
+    with patch("rein_aharness.adapter.subprocess.Popen", return_value=proc) as popen:
         response = adapter.execute_prompt("do it", config)
 
     assert response.content == "plain output"
-    argv = run.call_args.args[0]
+    argv = popen.call_args.args[0]
     assert "--output-format" not in argv
+    proc.communicate.assert_called_once()
+
+
+def test_blocking_cancel_kills_process_and_raises(tmp_path) -> None:
+    from rein_aharness.execution_cancel import ExecutionCancel, ExecutionCancelled
+
+    cancel = ExecutionCancel()
+    adapter = AgenticClaudeCodeAdapter(workdir=tmp_path, cancel=cancel)
+    config = RunConfig(timeout_seconds=30)
+    proc = MagicMock()
+    proc.poll.return_value = None
+
+    def _communicate(input=None, timeout=None):  # noqa: A002
+        cancel.cancel("lease-loss")
+        raise RuntimeError("subprocess interrupted")
+
+    proc.communicate.side_effect = _communicate
+
+    with patch("rein_aharness.adapter.subprocess.Popen", return_value=proc):
+        with pytest.raises(ExecutionCancelled, match="lease-loss"):
+            adapter.execute_prompt("do it", config)
+    proc.kill.assert_called_once()
+
+
+def test_streaming_cancel_refuses_nonzero_exit(tmp_path) -> None:
+    from rein_aharness.execution_cancel import ExecutionCancel, ExecutionCancelled
+
+    cancel = ExecutionCancel()
+    adapter = AgenticClaudeCodeAdapter(
+        workdir=tmp_path, on_tool_event=lambda e: None, cancel=cancel
+    )
+    config = RunConfig(timeout_seconds=30)
+    proc = _fake_proc([])
+    proc.poll.return_value = None
+
+    def _wait(timeout=None):
+        cancel.cancel("signal")
+        return 1
+
+    proc.wait.side_effect = _wait
+
+    with patch("rein_aharness.adapter.subprocess.Popen", return_value=proc):
+        with pytest.raises(ExecutionCancelled, match="signal"):
+            adapter.execute_prompt("do it", config)

tests/test_claim_loop.py

@@ -6,7 +6,13 @@ from unittest.mock import MagicMock, patch
 import time
 
 from rein_aharness.approaches import ApproachResult, APPROACH_FI_RESEARCH_BRIEF
-from rein_aharness.claim_loop import process_one, poll_peek
+from rein_aharness.claim_loop import (
+    _cancel_active_run,
+    _set_active_run_cancel,
+    process_one,
+    poll_peek,
+)
+from rein_aharness.execution_cancel import ExecutionCancel, active_cancel
 from rein_aharness.glas_execution import GLAS_APPROACH, GlasExecutionError
 from rein_aharness.ops_run_client import (
     ActivityCoreOpsClient,
@@ -135,6 +141,78 @@ def test_process_one_refuses_close_after_lease_loss() -> None:
     client.fail.assert_not_called()
 
 
+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)
+    proc = MagicMock()
+    proc.poll.return_value = None
+
+    def slow_execute(*_args, **_kwargs):
+        cancel = active_cancel()
+        assert cancel is not None
+        cancel.register_process(proc)
+        deadline = time.monotonic() + 1.0
+        while not cancel.cancelled and time.monotonic() < deadline:
+            time.sleep(0.005)
+        cancel.check()
+
+    with (
+        patch("rein_aharness.claim_loop._heartbeat_interval", return_value=0.01),
+        patch("rein_aharness.claim_loop.execute_approach", side_effect=slow_execute),
+    ):
+        result = process_one(client)
+
+    assert result.ok is False
+    assert result.reason.startswith("lease lost")
+    proc.kill.assert_called()
+    client.complete.assert_not_called()
+    client.fail.assert_not_called()
+
+
+def test_process_one_skips_close_after_signal_cancel() -> None:
+    client = MagicMock(spec=ActivityCoreOpsClient)
+    client.config = OpsRunConfig(worker_id="w", lease_seconds=90)
+    client.claim.return_value = [_claimed_run()]
+    ar = ApproachResult(
+        ok=True,
+        approach=APPROACH_FI_RESEARCH_BRIEF,
+        result={"path": "briefs/x.md"},
+        reason="ok",
+    )
+
+    def execute_then_signal(*_args, **_kwargs):
+        cancel = active_cancel()
+        assert cancel is not None
+        cancel.cancel("signal")
+        return ar
+
+    with patch("rein_aharness.claim_loop.execute_approach", side_effect=execute_then_signal):
+        result = process_one(client)
+
+    assert result.ok is False
+    assert result.reason == "execution cancelled (signal)"
+    assert result.detail == {"cancellation": {"cancelled": "true", "reason": "signal"}}
+    client.complete.assert_not_called()
+    client.fail.assert_not_called()
+
+
+def test_shutdown_signal_cancels_active_run() -> None:
+    cancel = ExecutionCancel()
+    proc = MagicMock()
+    proc.poll.return_value = None
+    cancel.register_process(proc)
+    _set_active_run_cancel(cancel)
+    try:
+        _cancel_active_run("signal")
+    finally:
+        _set_active_run_cancel(None)
+
+    assert cancel.reason == "signal"
+    proc.kill.assert_called_once()
+
+
 def test_process_one_records_adapter_exception_without_unbound_result() -> None:
     client = MagicMock(spec=ActivityCoreOpsClient)
     client.config = OpsRunConfig(worker_id="w", lease_seconds=90)
@@ -220,8 +298,16 @@ def test_profiled_run_uses_glas_and_completes_with_full_result() -> None:
 
     assert result.ok is True
     assert result.approach == GLAS_APPROACH
-    assert result.detail == {"execution_evidence": gateway_result["evidence"]}
-    client.complete.assert_called_once_with(run.id, result=gateway_result)
+    assert result.detail["execution_evidence"] == gateway_result["evidence"]
+    transaction = result.detail["repository_transaction"]
+    assert transaction["correlation_id"] == run.id
+    assert transaction["baseline"]["clean"] is True
+    client.complete.assert_called_once()
+    complete_run_id = client.complete.call_args.args[0]
+    complete_result = client.complete.call_args.kwargs["result"]
+    assert complete_run_id == run.id
+    assert complete_result["evidence"] == gateway_result["evidence"]
+    assert complete_result["repository_transaction"] == transaction
     client.fail.assert_not_called()
     select.assert_not_called()
     execute.assert_not_called()

tests/test_execution_cancel.py

@@ -0,0 +1,113 @@
+from __future__ import annotations
+
+import threading
+import time
+from unittest.mock import MagicMock
+
+import pytest
+
+from rein_aharness.execution_cancel import (
+    ExecutionCancel,
+    ExecutionCancelled,
+    active_cancel,
+    using_cancel,
+)
+
+
+def test_cancel_is_one_shot_and_bounded() -> None:
+    cancel = ExecutionCancel()
+    first = cancel.cancel("lease-loss")
+    second = cancel.cancel("timeout")
+
+    assert first == second == "lease-loss"
+    assert cancel.cancelled is True
+    assert cancel.reason == "lease-loss"
+    assert cancel.evidence() == {"cancelled": "true", "reason": "lease-loss"}
+
+
+def test_unknown_reason_is_normalized() -> None:
+    cancel = ExecutionCancel()
+    cancel.cancel("provider exploded with secrets")
+    assert cancel.reason == "cancelled"
+
+
+def test_check_raises_only_after_cancel() -> None:
+    cancel = ExecutionCancel()
+    cancel.check()
+    cancel.cancel("timeout")
+    with pytest.raises(ExecutionCancelled, match="timeout") as excinfo:
+        cancel.check()
+    assert excinfo.value.reason == "timeout"
+
+
+def test_register_process_is_killed_on_cancel() -> None:
+    proc = MagicMock()
+    proc.poll.return_value = None
+    cancel = ExecutionCancel()
+    cancel.register_process(proc)
+    cancel.cancel("lease-loss")
+    proc.kill.assert_called_once()
+
+
+def test_already_cancelled_register_kills_immediately() -> None:
+    proc = MagicMock()
+    proc.poll.return_value = None
+    cancel = ExecutionCancel()
+    cancel.cancel("signal")
+    cancel.register_process(proc)
+    proc.kill.assert_called_once()
+
+
+def test_exited_process_is_not_killed() -> None:
+    proc = MagicMock()
+    proc.poll.return_value = 0
+    cancel = ExecutionCancel()
+    cancel.register_process(proc)
+    cancel.cancel("timeout")
+    proc.kill.assert_not_called()
+
+
+def test_stop_callback_failure_does_not_block_cancel() -> None:
+    cancel = ExecutionCancel()
+    cancel.register_stop(lambda: (_ for _ in ()).throw(RuntimeError("stop failed")))
+    assert cancel.cancel("signal") == "signal"
+
+
+def test_concurrent_cancel_invokes_stop_once() -> None:
+    seen = []
+    cancel = ExecutionCancel()
+    cancel.register_stop(lambda: seen.append("stop"))
+    threads = [
+        threading.Thread(target=cancel.cancel, args=("lease-loss",)) for _ in range(8)
+    ]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join()
+    assert seen == ["stop"]
+
+
+def test_using_cancel_exposes_active_cancel() -> None:
+    cancel = ExecutionCancel()
+    assert active_cancel() is None
+    with using_cancel(cancel):
+        assert active_cancel() is cancel
+        cancel.cancel("timeout")
+        assert active_cancel() is cancel
+    assert active_cancel() is None
+
+
+def test_wait_returns_after_cancel() -> None:
+    cancel = ExecutionCancel()
+    started = threading.Event()
+
+    def _cancel_soon() -> None:
+        started.wait(timeout=1)
+        time.sleep(0.01)
+        cancel.cancel("signal")
+
+    thread = threading.Thread(target=_cancel_soon)
+    thread.start()
+    started.set()
+    assert cancel.wait(timeout=1) is True
+    thread.join()

tests/test_glas_execution.py

@@ -127,3 +127,48 @@ def test_profiled_actor_validates_against_real_glas_and_sandboxer(tmp_path: Path
     assert captured["request"].actor == "agt"
     assert captured["sandbox_request"].consumer.actor == sandbox_models.ActorType.AGT
     assert captured["sandbox_request"].consumer.run_id == "run-1"
+
+
+def test_profiled_run_does_not_invoke_gateway_when_already_cancelled(
+    tmp_path: Path,
+) -> None:
+    from rein_aharness.execution_cancel import ExecutionCancel, ExecutionCancelled
+
+    repo = _repo(tmp_path)
+    cancel = ExecutionCancel()
+    cancel.cancel("lease-loss")
+    called = []
+    with pytest.raises(ExecutionCancelled, match="lease-loss"):
+        execute_profiled_run(
+            _run(repo),
+            OpsRunConfig(repo_roots=(str(tmp_path),)),
+            request_factory=lambda **kwargs: kwargs,
+            gateway=lambda request: called.append(request) or {
+                "ok": True,
+                "evidence": {"outcome": "succeeded"},
+            },
+            cancel=cancel,
+        )
+    assert called == []
+
+
+def test_profiled_run_refuses_success_if_cancelled_during_gateway(
+    tmp_path: Path,
+) -> None:
+    from rein_aharness.execution_cancel import ExecutionCancel, ExecutionCancelled
+
+    repo = _repo(tmp_path)
+    cancel = ExecutionCancel()
+
+    def gateway(request):
+        cancel.cancel("timeout")
+        return {"ok": True, "evidence": {"outcome": "succeeded"}}
+
+    with pytest.raises(ExecutionCancelled, match="timeout"):
+        execute_profiled_run(
+            _run(repo),
+            OpsRunConfig(repo_roots=(str(tmp_path),)),
+            request_factory=lambda **kwargs: kwargs,
+            gateway=gateway,
+            cancel=cancel,
+        )

tests/test_mail_triage.py

@@ -257,3 +257,31 @@ def test_get_client_requires_env(monkeypatch) -> None:
         from rein_aharness.llm_connect_client import get_llm_connect_client
 
         get_llm_connect_client()
+
+
+def test_llm_connect_cancel_closes_client_and_raises(monkeypatch) -> None:
+    import rein_aharness.llm_connect_client as mod
+    from rein_aharness.execution_cancel import ExecutionCancel, ExecutionCancelled
+
+    closed = []
+    cancel = ExecutionCancel()
+
+    class FakeClient:
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *args):
+            return False
+
+        def close(self):
+            closed.append("closed")
+
+        def post(self, url, json=None):  # noqa: A002
+            cancel.cancel("timeout")
+            raise mod.httpx.ConnectError("connection reset")
+
+    monkeypatch.setattr(mod.httpx, "Client", lambda timeout=None: FakeClient())
+    client = LLMConnectClient("http://llm.test", timeout_seconds=5)
+    with pytest.raises(ExecutionCancelled, match="timeout"):
+        client.complete("hi", cancel=cancel)
+    assert closed == ["closed"]

tests/test_runner.py

@@ -126,6 +126,28 @@ def test_run_task_fails_without_commit(tmp_path) -> None:
     assert result.reason == "session completed without committing"
 
 
+def test_run_task_records_bounded_cancellation_without_session_output(tmp_path) -> None:
+    from rein_aharness.execution_cancel import ExecutionCancelled
+
+    repo = _make_repo(tmp_path)
+
+    class CancellingAdapter:
+        def execute_prompt(self, prompt, config):
+            raise ExecutionCancelled("lease-loss")
+
+    result = run_task(
+        _spec(repo),
+        adapter=CancellingAdapter(),
+        report_to_hub=False,
+        write_metrics=False,
+    )
+
+    assert result.ok is False
+    assert result.committed is False
+    assert result.session_output == ""
+    assert result.reason == "execution cancelled (lease-loss)"
+
+
 def test_run_task_refuses_unknown_tool_profile(tmp_path) -> None:
     repo = _make_repo(tmp_path)
     _write_manifest(
@@ -160,6 +182,21 @@ def test_run_task_resolves_manifest_profile_and_budget(tmp_path) -> None:
             }
         },
     )
+    subprocess.run(["git", "add", ".kaizen"], cwd=repo, check=True)
+    subprocess.run(
+        [
+            "git",
+            "-c",
+            "user.email=t@t",
+            "-c",
+            "user.name=t",
+            "commit",
+            "-qm",
+            "configure rein",
+        ],
+        cwd=repo,
+        check=True,
+    )
     adapter = CommittingAdapter(repo)
 
     result = run_task(
@@ -186,6 +223,21 @@ def test_run_task_budget_exhaustion_fails(tmp_path) -> None:
             }
         },
     )
+    subprocess.run(["git", "add", ".kaizen"], cwd=repo, check=True)
+    subprocess.run(
+        [
+            "git",
+            "-c",
+            "user.email=t@t",
+            "-c",
+            "user.name=t",
+            "commit",
+            "-qm",
+            "configure rein",
+        ],
+        cwd=repo,
+        check=True,
+    )
 
     result = run_task(
         _spec(repo),