fix(worker): preserve original replan through bounded Host retries

docs/reference/shell-worker.md

@@ -56,9 +56,22 @@ model invocation. A retry reads that original binding even when normal wakes
 use fresh context. A confirmed successful response ends the failure streak;
 an exhausted policy remains held. Provider hold/backoff is checked before
 recording a new semantic replan attempt; a rejected dispatch cannot consume one.
-An unresolved semantic replan still requires its original settlement recovery,
-even when the provider would otherwise permit a transport retry. Host state is
-local transport evidence, not Goal/Todo/quota authority or proof of completion.
+For a replan, dispatch also binds the Host attempt to its original obligation.
+A permitted transport retry may continue that one unknown semantic attempt only
+while current admission retains the exact Goal/agent scope and obligation. It
+reuses the recorded attempt and session, preserves its evidence, and does not
+acknowledge or settle the obligation. Changed admission, multiple unresolved
+attempts, interrupted dispatch, or legacy state without this association still
+requires original recovery. After transport success, an unchanged unresolved
+obligation remains paused; it cannot authorize another semantic attempt. Host
+state is local transport evidence, not Goal/Todo/quota authority or completion.
+If the original canonical settlement is instead verified complete, the worker
+may continue newly admitted work under the same bounded Host retry policy.
+It requires one exact dispatch-bound original attempt and rechecks its original
+settlement identity, including after a saved recovery marker survives restart.
+A changed admission or local marker alone cannot supply that proof. Missing,
+conflicting or unreadable proof retains the hold; session, backoff, retry limits
+and Core Stop still apply. Recovery never repeats the original refresh or spend.
 
 Inspect the existing state file, the original wake's `receipt.json`, and the
 content-free `host_failure_hold` / `host_failure_backoff` liveness output. Use

scripts/external_scheduler_worker.py

@@ -380,19 +380,51 @@ def _pending_replan_attempt(state: dict[str, Any], payload: dict[str, Any], args
     return any(_same_replan_scope(attempt, scope) for attempt in unresolved)
 
 
-def _read_resolved_replan_attempts(state: dict[str, Any], payload: dict[str, Any], args: argparse.Namespace) -> bool:
+def _host_retry_replan(state: dict[str, Any], payload: dict[str, Any],
+                       args: argparse.Namespace, prior: dict[str, Any]) -> dict[str, Any] | None:
+    """Reuse a dispatch-bound replan only after the Host retry gate admits it.
+
+    A native session alone cannot associate a transport failure with an older
+    semantic attempt. Legacy/unbound or ambiguous attempts retain their hold.
+    This does not acknowledge the obligation or settle its original effects.
+    """
+    obligation = prior.get("replan_obligation_id")
+    current = _replan_attempt(payload, args)
+    if not obligation or current is None or current["obligation_id"] != obligation:
+        return None
+    pending = [attempt for attempt in state.get("replan_attempts", [])
+               if attempt["status"] in {"started", "unknown"}
+               and _same_replan_scope(attempt, current)]
+    if (len(pending) != 1 or pending[0]["status"] != "unknown"
+            or pending[0]["obligation_id"] != obligation
+            or _attempt_scope(pending[0]) != _attempt_scope(current)):
+        return None
+    return pending[0]
+
+
+def _read_resolved_replan_attempts(
+    state: dict[str, Any], payload: dict[str, Any], args: argparse.Namespace,
+    prior: dict[str, Any] | None = None,
+) -> list[dict[str, Any]]:
     """Only the original settlement owner can resolve a known lost response.
 
     An opaque unbound shell command has no such proof. Leave it visibly unknown
     instead of inferring a Turn from the latest row or dispatching a new command.
+    Return only attempts verified by this read, not historical local markers.
     """
-    unresolved = [attempt for attempt in state.get("replan_attempts", [])
-                  if attempt["status"] in {"started", "unknown"} and attempt.get("settlement_identity")]
-    if not unresolved:
-        return False
+    retry_obligation = (prior or {}).get("replan_obligation_id")
+    # Recheck a saved recovery marker after restart: local transport state is
+    # not a substitute for the original canonical settlement proof.
+    candidates = [attempt for attempt in state.get("replan_attempts", [])
+                  if attempt.get("settlement_identity") and (
+                      attempt["status"] in {"started", "unknown"}
+                      or (retry_obligation == attempt["obligation_id"]
+                          and attempt.get("recovery") == "original_settlement_verified"))]
+    if not candidates:
+        return []
     scope = _admission_scope(payload, args)
-    changed = False
-    for attempt in unresolved:
+    resolved: list[dict[str, Any]] = []
+    for attempt in candidates:
         if not _same_replan_scope(attempt, scope):
             continue
         original = _attempt_settlement_identity(attempt)
@@ -408,8 +440,8 @@ def _read_resolved_replan_attempts(state: dict[str, Any], payload: dict[str, Any
                 and readback.replay_phase == "settled" and readback.settlement.failure is None
                 and readback.settlement.value is not None):
             attempt.update(status="returned", recovery="original_settlement_verified")
-            changed = True
-    return changed
+            resolved.append(attempt)
+    return resolved
 
 
 def _observe_worker_replan(
@@ -832,9 +864,20 @@ def save(fields: dict[str, Any]) -> None:
                             sleep(min(delay, 60.0))
                             retry_after_error = True
                             break
-                    if _read_resolved_replan_attempts(state, payload, args):
+                    resolved_replans = _read_resolved_replan_attempts(state, payload, args, prior)
+                    if resolved_replans:
                         _save_state(state_path, state)
-                    if _pending_replan_attempt(state, payload, args):
+                    retry_replan = _host_retry_replan(state, payload, args, prior) if host_guard and prior else None
+                    if prior.get("replan_obligation_id") and retry_replan is None:
+                        scope_now = _admission_scope(payload, args)
+                        bound = [attempt for attempt in state.get("replan_attempts", [])
+                                 if attempt["obligation_id"] == prior["replan_obligation_id"]
+                                 and _attempt_scope(attempt) == scope_now]
+                        if len(bound) != 1 or not any(attempt is bound[0] for attempt in resolved_replans):
+                            _log("status=host_failure_hold reason=original_replan_not_retryable "
+                                 "next=original_host_recovery goal_complete=false new_wake=false")
+                            return 2
+                    if _pending_replan_attempt(state, payload, args) and retry_replan is None:
                         replan_scope = _admission_scope(payload, args)
                         original = next(previous for previous in state["replan_attempts"]
                                         if previous["status"] in {"started", "unknown"}
@@ -843,22 +886,23 @@ def save(fields: dict[str, Any]) -> None:
                         _log("status=replan_outcome_unknown next=use_original_host_recovery "
                              "goal_complete=false new_wake=false")
                         return 2
-                    attempt = _replan_attempt(payload, args)
+                    attempt = retry_replan if retry_replan is not None else _replan_attempt(payload, args)
                     if attempt is not None:
                         attempts = state.setdefault("replan_attempts", [])
                         original = next((previous for previous in attempts
                                          if _same_replan_scope(previous, attempt)
                                          and previous["obligation_id"] == attempt["obligation_id"]), None)
-                        if original is not None:
+                        if original is not None and original is not retry_replan:
                             _observe_worker_replan(payload, args, original, "replan_paused")
                             _log("status=replan_paused reason=unresolved_original_obligation "
                                  f"obligation={attempt['obligation_id']} "
                                  "next=inspect_original_work_or_supply_concrete_direction "
                                  "goal_complete=false new_wake=false")
                             return 0
                         attempt["status"] = "started"
-                        attempt.update(_worker_goal_observation_digests(args))
-                        attempts.append(attempt)
+                        if retry_replan is None:
+                            attempt.update(_worker_goal_observation_digests(args))
+                            attempts.append(attempt)
                         _save_state(state_path, {**state, "reset_token": decision.reset_token,
                                                 "unchanged_count": unchanged_count})
                     _observe_worker_replan(payload, args, None, None)
@@ -868,7 +912,9 @@ def save(fields: dict[str, Any]) -> None:
                         _save_host_pending(state_path, True)
                         state["host_attempt"] = {"scope": scope, "status": "started",
                             "attempt": prior.get("attempt", 0) + 1,
-                            "session_id": prior.get("session_id")}
+                            "session_id": prior.get("session_id"),
+                            **({"replan_obligation_id": attempt["obligation_id"]}
+                               if attempt is not None else {})}
                         save({})
                         environment = {**os.environ, "LOOPX_HOST_RETRY_SESSION_ID": prior.get("session_id") or ""}
                     wake_result = _run_wake(

tests/control_plane/test_external_worker_replan_cli.py

@@ -11,7 +11,7 @@
 import pytest
 
 from test_replan_successor_guard_reentry import AGENT, GOAL, TURN, _add, _fixture, _guard
-from test_external_scheduler_worker import _args, worker
+from test_external_scheduler_worker import _args, _guarded_host, worker
 from loopx.extensions.process_runtime import CappedProcessResult
 from loopx.history import collect_history
 
@@ -306,3 +306,120 @@ def read(registry):
     assert Path(args.state_file).read_bytes() == original_attempt
     assert requests == [1]
     assert index.read_bytes() == before
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_real_quota_retry_preserves_original_obligation_without_settlement(tmp_path, monkeypatch, provider):
+    _, runtime, history = _fixture(tmp_path, monkeypatch, provider, future_monitor=False)
+    args, count = _guarded_host(tmp_path, kind="provider_capacity", recover_at=2)
+    args.cli_bin = str(Path(sys.executable).parent / "loopx")
+    args.registry, args.runtime_root = str(tmp_path / "registry.json"), str(runtime)
+    args.goal_id, args.agent_id, args.quota_timeout_seconds = GOAL, AGENT, 90
+    now, delays = [1000.0], []
+    def sleep(seconds):
+        delays.append(seconds)
+        now[0] += seconds
+    before = history.read_bytes()
+    # Real CLI admission and authority stores; only the model process is synthetic.
+    # Its second response succeeds without semantically settling the replan.
+    assert worker.run_worker(args, sleep=sleep, clock=lambda: now[0]) == 0
+    assert count.read_text() == "2"
+    assert delays == [30.0]
+    attempts = json.loads(Path(args.state_file).read_text())["replan_attempts"]
+    assert len(attempts) == 1 and attempts[0]["status"] == "returned"
+    assert history.read_bytes() == before
+    args.once = True
+    assert worker.run_worker(args, clock=lambda: now[0]) == 0
+    assert count.read_text() == "2"
+    assert history.read_bytes() == before
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+@pytest.mark.parametrize("recovery", ["fresh", "restart", "missing_proof", "unreadable_proof", "wrong_identity"])
+def test_settled_replan_host_failure_allows_only_verified_next_work(tmp_path, monkeypatch, provider, recovery):
+    call, runtime, history = _fixture(tmp_path, monkeypatch, provider, future_monitor=False)
+    args, _ = _guarded_host(tmp_path, kind="provider_capacity", recover_at=2)
+    args.cli_bin = str(Path(sys.executable).parent / "loopx")
+    args.registry, args.runtime_root = str(tmp_path / "registry.json"), str(runtime)
+    args.goal_id, args.agent_id, args.quota_timeout_seconds, args.once = GOAL, AGENT, 90, True
+    real_quota = worker.run_quota_should_run
+    turn, decisions, requests, original = [TURN], [], [], {}
+
+    def quota(command, **kwargs):
+        packet = real_quota([*command, "--turn-instance-id", turn[0]], **kwargs)
+        decisions.append(packet)
+        return packet
+
+    session = {"binding_scope": "agent", "session_id": "session-a"}
+
+    def host(*command, **kwargs):
+        requests.append(kwargs.get("environment", {}).get("LOOPX_HOST_RETRY_SESSION_ID"))
+        if len(requests) == 1:
+            identity = decisions[-1]["heartbeat_receipt"]["settlement_identity"]
+            successor = _add(call, identity["replan_obligation_id"])
+            actions = _guard(call)["interaction_contract"]["cli_channel"]["next_cli_actions"]
+            refreshed = call(*shlex.split(actions[0])[1:], "--no-global-sync", "--suppress-external-sinks")
+            assert refreshed["settlement_progress"]["state"] == "spend_required"
+            spent = call(*shlex.split(actions[1])[1:])
+            assert spent["appended"] is True and spent["settlement_progress"]["state"] == "settled"
+            original.update(identity=identity, successor=successor["todo_id"], spend=actions[1])
+            receipt = {"ok": False, "session": session, "host_failure": {
+                "schema_version": "loopx_turn_host_failure_v0", "kind": "provider_capacity",
+                "attempt": 1, "retryable": True, "retry": {
+                    "strategy": "same_configuration", "max_attempts": 3, "backoff_seconds": 30}}}
+            return CappedProcessResult(returncode=1, stdout=json.dumps(receipt).encode())
+        assert decisions[-1]["selected_todo"]["todo_id"] == original["successor"]
+        return CappedProcessResult(returncode=0, stdout=json.dumps({"ok": True, "session": session}).encode())
+
+    monkeypatch.setattr(worker, "run_quota_should_run", quota)
+    monkeypatch.setattr(worker, "_run_wake", host)
+    assert worker.run_worker(args, clock=lambda: 1000.0) == 1
+    turn[0] = "next-work-after-settled-replan"
+    # The transport retry keeps its original backoff even though Core settled.
+    assert worker.run_worker(args, clock=lambda: 1001.0) == 2
+    assert requests == [""]
+
+    if recovery != "fresh":
+        real_save = worker._save_state
+
+        def interrupt_after_proof(path, state, **kwargs):
+            real_save(path, state, **kwargs)
+            if any(a.get("recovery") == "original_settlement_verified" for a in state.get("replan_attempts", [])):
+                raise OSError("synthetic interruption after durable recovery proof")
+
+        with monkeypatch.context() as interrupted:
+            interrupted.setattr(worker, "_save_state", interrupt_after_proof)
+            assert worker.run_worker(args, clock=lambda: 1031.0) == 2
+        assert requests == [""]
+        if recovery == "missing_proof":
+            monkeypatch.setattr(worker, "read_heartbeat_settlement", lambda *a, **kw: None)
+        elif recovery == "unreadable_proof":
+            def unavailable(*a, **kw):
+                raise OSError("synthetic original authority unavailable")
+            monkeypatch.setattr(worker, "read_heartbeat_settlement", unavailable)
+        elif recovery == "wrong_identity":
+            from dataclasses import replace
+            real_read = worker.read_heartbeat_settlement
+
+            def conflicting(*a, **kw):
+                proof = real_read(*a, **kw)
+                return replace(proof, identity=replace(proof.identity,
+                    value=replace(proof.identity.value, turn_instance_id="different-original-turn")))
+
+            monkeypatch.setattr(worker, "read_heartbeat_settlement", conflicting)
+
+    allowed = recovery in {"fresh", "restart"}
+    assert worker.run_worker(args, clock=lambda: 1031.0) == (0 if allowed else 2)
+    assert requests == (["", "session-a"] if allowed else [""])
+    state = json.loads(Path(args.state_file).read_text())
+    assert len(state["replan_attempts"]) == 1
+    attempt = state["replan_attempts"][0]
+    assert attempt["status"] == "returned" and attempt["recovery"] == "original_settlement_verified"
+    assert attempt["settlement_identity"] == original["identity"]
+    if allowed:
+        assert "host_attempt" not in state
+    else:
+        assert state["host_attempt"]["attempt"] == 1
+    rows = [json.loads(line) for line in history.read_text().splitlines()]
+    assert sum(row.get("classification") == "quota_slot_spent" for row in rows) == 1
+    assert call(*shlex.split(original["spend"])[1:])["appended"] is False