fix(kanban): bind lifecycle handoff to worker claim

agent/conversation_loop.py

@@ -6801,7 +6801,9 @@ def _perform_api_call(next_api_kwargs):
 
                 agent._execute_tool_calls(assistant_message, messages, effective_task_id, api_call_count)
 
-                if getattr(agent, "_kanban_lifecycle_handoff", None):
+                from agent.tool_executor import _kanban_handoff_matches_current_claim
+
+                if _kanban_handoff_matches_current_claim(agent):
                     _turn_exit_reason = "kanban_lifecycle_handoff"
                     # The lifecycle tool transferred custody and closed this
                     # run. Do not ask the model for another turn: a stale

agent/tool_executor.py

@@ -71,29 +71,72 @@ def _runtime_identity(agent) -> dict[str, str]:
     }
 
 
-def _record_successful_kanban_handoff(agent, function_name: str, result: Any) -> bool:
-    """Latch a successful lifecycle transfer so this worker stops immediately."""
+def _kanban_worker_claim() -> tuple[str, int] | None:
+    """Return the exact dispatcher-owned task/run claim for this process."""
     from agent.delegation_context import is_dispatcher_owned_worker_context
 
-    if not os.environ.get("HERMES_KANBAN_TASK") or not is_dispatcher_owned_worker_context():
-        return False
+    if not is_dispatcher_owned_worker_context():
+        return None
+    task_id = str(os.environ.get("HERMES_KANBAN_TASK") or "").strip()
+    raw_run_id = os.environ.get("HERMES_KANBAN_RUN_ID")
+    if not task_id or isinstance(raw_run_id, bool):
+        return None
+    try:
+        run_id = int(raw_run_id)
+    except (TypeError, ValueError):
+        return None
+    if run_id <= 0:
+        return None
+    return task_id, run_id
+
+
+def _record_successful_kanban_handoff(agent, function_name: str, result: Any) -> bool:
+    """Latch a successful lifecycle transfer so this worker stops immediately."""
     if function_name not in _KANBAN_LIFECYCLE_HANDOFF_TOOLS:
         return False
+    claim = _kanban_worker_claim()
+    if claim is None:
+        return False
     try:
         payload = json.loads(result) if isinstance(result, str) else result
     except (TypeError, json.JSONDecodeError):
         return False
     if not isinstance(payload, dict) or payload.get("ok") is not True:
         return False
+    payload_run_id = payload.get("run_id")
+    if isinstance(payload_run_id, bool):
+        return False
+    try:
+        payload_run_id = int(payload_run_id)
+    except (TypeError, ValueError):
+        return False
+    task_id, run_id = claim
+    if str(payload.get("task_id") or "").strip() != task_id or payload_run_id != run_id:
+        return False
     agent._kanban_lifecycle_handoff = {
         "tool": function_name,
-        "task_id": payload.get("task_id"),
-        "run_id": payload.get("run_id"),
+        "task_id": task_id,
+        "run_id": run_id,
         "status": payload.get("status"),
     }
     return True
 
 
+def _kanban_handoff_matches_current_claim(agent) -> bool:
+    """Consume a latch only while its originating task/run claim is current."""
+    handoff = getattr(agent, "_kanban_lifecycle_handoff", None)
+    claim = _kanban_worker_claim()
+    if not isinstance(handoff, dict) or claim is None:
+        if handoff is not None:
+            agent._kanban_lifecycle_handoff = None
+        return False
+    task_id, run_id = claim
+    if handoff.get("task_id") != task_id or handoff.get("run_id") != run_id:
+        agent._kanban_lifecycle_handoff = None
+        return False
+    return True
+
+
 def _ensure_file_checkpoint(
     agent,
     function_name: str,
@@ -2496,7 +2539,7 @@ def execute_tool_calls_segmented(agent, assistant_message, messages: list, effec
         if getattr(agent, "_incremental_persistence_failed", False):
             return
 
-        if getattr(agent, "_kanban_lifecycle_handoff", None):
+        if _kanban_handoff_matches_current_claim(agent):
             remaining_calls = [
                 call
                 for _kind, later_calls in segments[segment_index + 1:]

tests/run_agent/test_tool_batch_segmentation.py

@@ -400,6 +400,7 @@ def test_kanban_handoff_cancels_later_parallel_segment(self, agent, monkeypatch)
         messages = []
         executed = []
         monkeypatch.setenv("HERMES_KANBAN_TASK", "t_1")
+        monkeypatch.setenv("HERMES_KANBAN_RUN_ID", "1")
 
         def fake_handle(name, args, task_id, **kwargs):
             executed.append(kwargs["tool_call_id"])
@@ -412,6 +413,41 @@ def fake_handle(name, args, task_id, **kwargs):
         assert [m["tool_call_id"] for m in messages] == ["k1", "s1", "s2"]
         assert all("successful Kanban lifecycle handoff" in m["content"] for m in messages[1:])
 
+    def test_stale_kanban_handoff_does_not_stop_successor(self, agent, monkeypatch):
+        from agent.tool_executor import execute_tool_calls_segmented
+
+        calls = [
+            _tc("web_search", '{"query":"successor"}', call_id="s1"),
+            _tc("terminal", '{"command":"true"}', call_id="t1"),
+        ]
+        msg = SimpleNamespace(content="", tool_calls=calls)
+        messages = []
+        executed = []
+        agent._kanban_lifecycle_handoff = {
+            "tool": "kanban_request_changes",
+            "task_id": "t_1",
+            "run_id": 1,
+            "status": "ready",
+        }
+        monkeypatch.setenv("HERMES_KANBAN_TASK", "t_1")
+        monkeypatch.setenv("HERMES_KANBAN_RUN_ID", "2")
+
+        def fake_handle(name, args, task_id, **kwargs):
+            executed.append(kwargs["tool_call_id"])
+            return json.dumps({"ok": True})
+
+        with patch("run_agent.handle_function_call", side_effect=fake_handle):
+            execute_tool_calls_segmented(
+                agent,
+                msg,
+                messages,
+                "task-1",
+                segments=[("parallel", calls[:1]), ("sequential", calls[1:])],
+            )
+
+        assert executed == ["s1", "t1"]
+        assert agent._kanban_lifecycle_handoff is None
+
     def test_mixed_batch_runs_safe_prefix_concurrently_and_barrier_after(self, agent):
         """Two web_search calls must overlap in time; terminal must start only
         after both finish; results land in the model's emission order."""

tests/tools/test_kanban_runtime_receipts.py

@@ -149,6 +149,7 @@ def test_successful_lifecycle_handoff_latches_worker_stop(monkeypatch):
 
     agent = SimpleNamespace()
     monkeypatch.setenv("HERMES_KANBAN_TASK", "t_1")
+    monkeypatch.setenv("HERMES_KANBAN_RUN_ID", "7")
     assert _record_successful_kanban_handoff(
         agent,
         "kanban_request_review",
@@ -162,6 +163,110 @@ def test_successful_lifecycle_handoff_latches_worker_stop(monkeypatch):
     }
 
 
+@pytest.mark.parametrize(
+    "payload",
+    [
+        {"ok": True, "task_id": "t_successor", "run_id": 7},
+        {"ok": True, "task_id": "t_1", "run_id": 8},
+        {"ok": True, "task_id": "t_1"},
+    ],
+)
+def test_lifecycle_handoff_requires_exact_dispatcher_claim(monkeypatch, payload):
+    from agent.tool_executor import _record_successful_kanban_handoff
+
+    agent = SimpleNamespace()
+    monkeypatch.setenv("HERMES_KANBAN_TASK", "t_1")
+    monkeypatch.setenv("HERMES_KANBAN_RUN_ID", "7")
+
+    assert not _record_successful_kanban_handoff(
+        agent,
+        "kanban_request_review",
+        json.dumps(payload),
+    )
+    assert not hasattr(agent, "_kanban_lifecycle_handoff")
+
+
+def test_lifecycle_handoff_requires_dispatcher_run_id(monkeypatch):
+    from agent.tool_executor import _record_successful_kanban_handoff
+
+    agent = SimpleNamespace()
+    monkeypatch.setenv("HERMES_KANBAN_TASK", "t_1")
+    monkeypatch.delenv("HERMES_KANBAN_RUN_ID", raising=False)
+
+    assert not _record_successful_kanban_handoff(
+        agent,
+        "kanban_complete",
+        json.dumps({"ok": True, "task_id": "t_1", "run_id": 7}),
+    )
+    assert not hasattr(agent, "_kanban_lifecycle_handoff")
+
+
+def test_stale_handoff_latch_cannot_stop_successor_claim(monkeypatch):
+    from agent.tool_executor import _kanban_handoff_matches_current_claim
+
+    agent = SimpleNamespace(
+        _kanban_lifecycle_handoff={
+            "tool": "kanban_request_changes",
+            "task_id": "t_1",
+            "run_id": 7,
+            "status": "ready",
+        }
+    )
+    monkeypatch.setenv("HERMES_KANBAN_TASK", "t_1")
+    monkeypatch.setenv("HERMES_KANBAN_RUN_ID", "8")
+    assert not _kanban_handoff_matches_current_claim(agent)
+    assert agent._kanban_lifecycle_handoff is None
+
+
+def test_lifecycle_result_uses_closed_worker_run_not_later_latest_run(
+    monkeypatch,
+    worker_env,
+):
+    from hermes_cli import kanban_db as kb
+    from tools import kanban_tools as kt
+
+    worker_run_id = int(__import__("os").environ["HERMES_KANBAN_RUN_ID"])
+    monkeypatch.setattr(
+        kb,
+        "latest_run",
+        lambda _conn, _task_id: SimpleNamespace(id=worker_run_id + 1),
+    )
+
+    out = json.loads(
+        kt._handle_block(
+            {"reason": "external input required", "kind": "needs_input"},
+            runtime_identity=_identity(),
+        )
+    )
+
+    assert out["ok"] is True
+    assert out["run_id"] == worker_run_id
+
+
+def test_runtime_identity_uses_effective_route_not_initial_config():
+    from agent.tool_executor import _runtime_identity
+
+    agent = SimpleNamespace(
+        provider="fallback-provider",
+        model="effective-model",
+        api_mode="responses",
+        session_id="effective-session",
+        _session_init_model_config={
+            "provider": "requested-provider",
+            "model": "requested-model",
+            "api_mode": "chat_completions",
+        },
+    )
+
+    assert _runtime_identity(agent) == {
+        "provider": "fallback-provider",
+        "model": "effective-model",
+        "api_mode": "responses",
+        "session_id": "effective-session",
+        "source": "agent_runtime_after_provider_response",
+    }
+
+
 def test_failed_lifecycle_call_does_not_latch_worker_stop(monkeypatch):
     from agent.tool_executor import _record_successful_kanban_handoff
 

tools/kanban_tools.py

@@ -161,9 +161,10 @@ def _worker_run_id(task_id: str) -> Optional[int]:
     if not raw:
         return None
     try:
-        return int(raw)
+        run_id = int(raw)
     except ValueError:
         return None
+    return run_id if run_id > 0 else None
 
 
 def _stamp_worker_session_metadata(
@@ -774,12 +775,13 @@ def _handle_complete(args: dict, **kw) -> str:
                     f"and keep this task alive."
                 )
 
+            expected_run_id = _worker_run_id(tid)
             try:
                 ok = kb.complete_task(
                     conn, tid,
                     result=result, summary=summary, metadata=metadata,
                     created_cards=created_cards,
-                    expected_run_id=_worker_run_id(tid),
+                    expected_run_id=expected_run_id,
                 )
             except kb.ArtifactPreservationError as artifact_err:
                 return tool_error(
@@ -813,7 +815,14 @@ def _handle_complete(args: dict, **kw) -> str:
                     f"could not complete {tid} (unknown id or already terminal)"
                 )
             run = kb.latest_run(conn, tid)
-            return _ok(task_id=tid, run_id=run.id if run else None)
+            return _ok(
+                task_id=tid,
+                run_id=(
+                    expected_run_id
+                    if expected_run_id is not None
+                    else (run.id if run else None)
+                ),
+            )
         finally:
             conn.close()
     except ValueError as e:
@@ -876,12 +885,13 @@ def _handle_block(args: dict, **kw) -> str:
                 f"another reason, call kanban_complete instead — the "
                 f"completion judge will evaluate it."
             )
+        expected_run_id = _worker_run_id(tid)
         try:
             ok = kb.block_task(
                 conn, tid,
                 reason=reason,
                 kind=kind,
-                expected_run_id=_worker_run_id(tid),
+                expected_run_id=expected_run_id,
                 metadata=metadata,
             )
             if not ok:
@@ -895,7 +905,11 @@ def _handle_block(args: dict, **kw) -> str:
             landed = kb.get_task(conn, tid)
             return _ok(
                 task_id=tid,
-                run_id=run.id if run else None,
+                run_id=(
+                    expected_run_id
+                    if expected_run_id is not None
+                    else (run.id if run else None)
+                ),
                 status=landed.status if landed else "blocked",
                 block_kind=kind,
             )
@@ -959,12 +973,13 @@ def _handle_request_review(args: dict, **kw) -> str:
                     "Provide acceptance evidence matching the card before "
                     "requesting review."
                 )
+            expected_run_id = _worker_run_id(tid)
             ok, fail_reason = kb.request_review(
                 conn, tid,
                 summary=summary,
                 metadata=metadata,
                 reviewer=reviewer,
-                expected_run_id=_worker_run_id(tid),
+                expected_run_id=expected_run_id,
                 with_reason=True,
             )
             if not ok:
@@ -976,7 +991,11 @@ def _handle_request_review(args: dict, **kw) -> str:
             landed = kb.get_task(conn, tid)
             return _ok(
                 task_id=tid,
-                run_id=run.id if run else None,
+                run_id=(
+                    expected_run_id
+                    if expected_run_id is not None
+                    else (run.id if run else None)
+                ),
                 status=landed.status if landed else "review",
             )
         finally:
@@ -1012,11 +1031,12 @@ def _handle_request_changes(args: dict, **kw) -> str:
     try:
         kb, conn = _connect(board=board)
         try:
+            expected_run_id = _worker_run_id(tid)
             ok, detail = kb.request_changes(
                 conn,
                 tid,
                 reason=reason,
-                expected_run_id=_worker_run_id(tid),
+                expected_run_id=expected_run_id,
                 metadata=metadata,
             )
             if not ok:
@@ -1027,7 +1047,11 @@ def _handle_request_changes(args: dict, **kw) -> str:
             run = kb.latest_run(conn, tid)
             return _ok(
                 task_id=tid,
-                run_id=run.id if run else None,
+                run_id=(
+                    expected_run_id
+                    if expected_run_id is not None
+                    else (run.id if run else None)
+                ),
                 status=landed.status if landed else "ready",
                 implementer=detail,
             )