diff --git a/docs/reference/shell-worker.md b/docs/reference/shell-worker.md index a44ceb25ab..b65eaf064f 100644 --- a/docs/reference/shell-worker.md +++ b/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 diff --git a/scripts/external_scheduler_worker.py b/scripts/external_scheduler_worker.py index 5b833fd6ff..5f6822e25c 100644 --- a/scripts/external_scheduler_worker.py +++ b/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,13 +886,13 @@ 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']} " @@ -857,8 +900,9 @@ def save(fields: dict[str, Any]) -> None: "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( diff --git a/tests/control_plane/test_external_worker_replan_cli.py b/tests/control_plane/test_external_worker_replan_cli.py index 064a8e90fb..9d106b5048 100644 --- a/tests/control_plane/test_external_worker_replan_cli.py +++ b/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 diff --git a/tests/test_external_scheduler_replan_retry.py b/tests/test_external_scheduler_replan_retry.py new file mode 100644 index 0000000000..b6ae5fbe27 --- /dev/null +++ b/tests/test_external_scheduler_replan_retry.py @@ -0,0 +1,87 @@ +"""Transport retry must continue one replan, never authorize a replacement.""" +from copy import deepcopy +import json +from pathlib import Path + +import pytest + +from tests.test_external_scheduler_worker import _guarded_host, _hint_payload, worker + + +def _replan(tmp_path): + return {**_hint_payload(should_run=True), + "effective_action": "autonomous_replan_required", "goal_id": "goal", + "agent_identity": {"agent_id": "agent"}, "runtime_root": str(tmp_path), + "autonomous_replan_obligation": {"required": True, "obligation_id": "replan-1111111111111111"}} + + +@pytest.mark.parametrize("recover_at,expected_calls,rc", [(2, 2, 0), (None, 3, 2)]) +def test_bound_replan_retries_original_host_with_one_semantic_attempt(tmp_path, monkeypatch, recover_at, expected_calls, rc): + args, count = _guarded_host(tmp_path, kind="provider_capacity", recover_at=recover_at) + now, delays = [1000.0], [] + def sleep(seconds): + delays.append(seconds) + now[0] += seconds + def quota(*a, **kw): + if recover_at and count.exists() and int(count.read_text()) >= recover_at: + return _hint_payload(should_run=False, local_scheduler_directive="stop") + return _replan(tmp_path) + monkeypatch.setattr(worker, "run_quota_should_run", quota) + assert worker.run_worker(args, sleep=sleep, clock=lambda: now[0]) == rc + assert count.read_text() == str(expected_calls) + assert delays == ([30.0] if recover_at else [30.0, 60.0]) + state = json.loads(Path(args.state_file).read_text()) + assert len(state["replan_attempts"]) == 1 + attempt = state["replan_attempts"][0] + assert attempt["obligation_id"] == "replan-1111111111111111" + assert attempt["status"] == ("returned" if recover_at else "unknown") + if recover_at: + assert "host_attempt" not in state + # Transport success is not semantic acknowledgement. The unchanged + # obligation must still pause instead of starting a third conversation. + monkeypatch.setattr(worker, "run_quota_should_run", lambda *a, **kw: _replan(tmp_path)) + assert worker.run_worker(args, sleep=sleep, clock=lambda: now[0]) == 0 + else: + assert state["host_attempt"]["attempt"] == 3 + assert worker.run_worker(args, sleep=sleep, clock=lambda: now[0]) == 2 + assert count.read_text() == str(expected_calls) + + +@pytest.mark.parametrize("change", ["unchanged", "legacy", "other_obligation", "normal_work", "other_lifetime", + "ambiguous", "started", "disabled", "stopped"]) +def test_restart_keeps_replan_binding_and_existing_holds(tmp_path, monkeypatch, change): + args, count = _guarded_host(tmp_path, kind="provider_capacity", recover_at=2) + args.once = True + payload = _replan(tmp_path) + monkeypatch.setattr(worker, "run_quota_should_run", lambda *a, **kw: payload) + assert worker.run_worker(args, clock=lambda: 1000.0) == 1 + state_path = Path(args.state_file) + state = json.loads(state_path.read_text()) + assert state["host_attempt"]["replan_obligation_id"] == "replan-1111111111111111" + if change == "legacy": + state["host_attempt"].pop("replan_obligation_id") + elif change == "other_obligation": + payload["autonomous_replan_obligation"]["obligation_id"] = "replan-2222222222222222" + elif change == "normal_work": + payload["effective_action"] = "normal_run" + elif change == "other_lifetime": + payload["goal_ref"] = {"goal_id": "goal", "goal_instance_id": "new-lifetime"} + elif change == "ambiguous": + another = deepcopy(state["replan_attempts"][0]) + another["obligation_id"] = "replan-3333333333333333" + state["replan_attempts"].append(another) + elif change == "started": + state["replan_attempts"][0]["status"] = "started" + elif change == "disabled": + args.wake_host_failure = False + elif change == "stopped": + payload = _hint_payload(should_run=False, local_scheduler_directive="stop") + state_path.write_text(json.dumps(state)) + rc = worker.run_worker(args, clock=lambda: 1031.0) + assert rc == (0 if change in {"unchanged", "stopped"} else 2) + assert count.read_text() == ("2" if change == "unchanged" else "1") + after = json.loads(state_path.read_text()) + assert len(after["replan_attempts"]) == len(state["replan_attempts"]) + if change != "unchanged": + assert after["host_attempt"]["attempt"] == 1 + assert after["replan_attempts"] == state["replan_attempts"]