From ecaaefc41a8a681e7bf7226852ecd115a154c2d2 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 20 Oct 2026 19:32:09 +0800 Subject: [PATCH] fix(host): preserve pending attempts across damaged worker state Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- docs/reference/shell-worker.md | 17 ++ scripts/external_scheduler_worker.py | 86 +++++++- tests/test_external_scheduler_worker.py | 251 ++++++++++++++++++++++++ 3 files changed, 349 insertions(+), 5 deletions(-) diff --git a/docs/reference/shell-worker.md b/docs/reference/shell-worker.md index 15c029792e..78615c2759 100644 --- a/docs/reference/shell-worker.md +++ b/docs/reference/shell-worker.md @@ -22,6 +22,23 @@ authentication and unknown failures require the original Host's recovery; restarting, changing a Todo/reset token or removing the opt-in flag does not erase that hold. A configuration mismatch also stays held. Core Stop still wins, without declaring that the original effect was settled. +The same lock also protects a content-free `.host-attempt` witness. +It is marked pending before the main started record and Host invocation, and +resolved only after confirmed success is committed to the main state. A pending +or unreadable witness keeps the guarded route even when the main JSON is damaged +or missing and the flag is removed. Conflicting or interrupted writes hold +without another Host call; Core Stop remains readable and preserves the evidence. +Retain both files during recovery, upgrade and rollback. This is local process +evidence, not a new Core state owner or a generic unlock. Once success is confirmed, +disabling the flag retains ordinary command behavior; never-guarded paths are +unchanged, including their existing treatment of damaged backoff state. +For a predecessor without the witness, the matching Host-adapter lock record +also retains the guarded route when its primary state is unreadable or missing. +An unrelated ordinary lock does not enable the adapter. This compatibility +readback never treats PID liveness as permission to dispatch or bypass the lock. +The retained adapter lock also fences its initial Core read before intent is +written; after confirmed success, flag-off commands keep their ordinary result +handling under that fence. An explicitly configured `--wake-stop-exit-code` also remains a worker stop after restart, without another dispatch or a Goal-completion claim. It cannot clear a prior failed/unknown attempt, which is checked before invoking the Host. diff --git a/scripts/external_scheduler_worker.py b/scripts/external_scheduler_worker.py index 1b72a65394..372ef72df3 100644 --- a/scripts/external_scheduler_worker.py +++ b/scripts/external_scheduler_worker.py @@ -47,6 +47,7 @@ SCHEDULER_DETAIL_KEY = "local_scheduler" LOCAL_SCHEDULER_STOP_DIRECTIVE = "stop" PROCESS_OUTPUT_LIMIT_BYTES = 1_000_000 +HOST_LOCK_OPERATION = "external_worker_host" @dataclass(frozen=True) @@ -261,6 +262,46 @@ def _save_state(path: Path | None, state: dict[str, Any], *, durable: bool = Fal os.replace(tmp, path) +def _host_pending_path(state_path: Path) -> Path: + return state_path.with_name(f"{state_path.name}.host-attempt") + + +def _host_pending(state_path: Path) -> bool | None: + """Independent local IO witness: absent/resolved, pending, or unreadable.""" + path = _host_pending_path(state_path) + if path.is_symlink(): + return None + try: + value = json.loads(path.read_text(encoding="utf-8")) + except FileNotFoundError: + return False + except (OSError, UnicodeError, json.JSONDecodeError): + return None + pending = value.get("pending") if isinstance(value, dict) else None + return pending if isinstance(pending, bool) else None + + +def _save_host_pending(state_path: Path, pending: bool) -> None: + _save_state(_host_pending_path(state_path), {"pending": pending}, durable=True) + + +def _legacy_guarded_state(state_path: Path) -> tuple[bool, bool]: + from loopx.file_lock import lock_holder_liveness + + # The predecessor's adapter marker can retain an unknown attempt. PID + # liveness never grants dispatch; the current kernel fence is still required. + _, holder = lock_holder_liveness(state_path) + if holder.get("operation") != HOST_LOCK_OPERATION: + return False, False + try: + if not state_path.exists(): + return True, True + _load_state(state_path, strict=True) + except (OSError, ValueError): + return True, True + return True, False + + def select_interval( decision: TickDecision, *, @@ -427,18 +468,30 @@ def run_worker( clock: Callable[[], float] = time.time, ) -> int: opted_in = bool(getattr(args, "wake_host_failure", False)) - retained = "host_attempt" in _load_state(resolve_state_path(args)) - if not opted_in and not retained: + state_path = resolve_state_path(args).resolve() + retained = (_host_pending(state_path) is not False + or "host_attempt" in _load_state(state_path)) + legacy_lock, legacy_unreadable = ( + _legacy_guarded_state(state_path) if not opted_in and not retained else (False, False) + ) + if not opted_in and not retained and not legacy_lock: return _run_worker(args, sleep=sleep, clock=clock, host_guard=False) from loopx.file_lock import try_exclusive_file_lock try: with try_exclusive_file_lock(resolve_state_path(args).resolve(), agent_id=args.agent_id, - operation="external_worker_host") as held: + operation=HOST_LOCK_OPERATION) as held: if held is None: _log("status=host_failure_hold kind=unknown reason=worker_already_running") return 2 - return _run_worker(args, sleep=sleep, clock=clock, host_guard=True) + # A retained adapter lock also fences its initial Core read, before + # intent exists. After a confirmed success, keep opaque flag-off + # commands ordinary; recheck evidence under the kernel fence. + host_guard = (opted_in or retained or legacy_unreadable + or _host_pending(state_path) is not False + or "host_attempt" in _load_state(state_path)) + return _run_worker(args, sleep=sleep, clock=clock, host_guard=host_guard, + unresolved_state=legacy_unreadable) except (OSError, ValueError): _log("status=host_failure_hold kind=unknown reason=state_or_original_result_unresolved") return 2 @@ -446,7 +499,7 @@ def run_worker( def _run_worker( args: argparse.Namespace, *, sleep: Callable[[float], None], clock: Callable[[], float], - host_guard: bool, + host_guard: bool, unresolved_state: bool = False, ) -> int: should_run_command = build_should_run_command(args) state_path = resolve_state_path(args) @@ -457,11 +510,19 @@ def _run_worker( while True: state_unreadable = False + pending = _host_pending(state_path) if host_guard else False try: + if unresolved_state: + raise ValueError("legacy Host state is unreadable") state = _load_state(state_path, strict=host_guard) previous_token = str(state.get("reset_token") or "") unchanged_count = int(state.get("unchanged_count") or 0) prior = _host_attempt(state, scope) if host_guard else {} + # A validated original worker-stop permits readback, never dispatch, + # even when the secondary witness is unreadable. + if ((pending is None and prior.get("status") != "stopped") + or (pending is True and not prior)): + raise ValueError("original Host attempt is unresolved") except (ValueError, TypeError, OverflowError): if not host_guard: raise @@ -469,6 +530,14 @@ def _run_worker( state, previous_token, unchanged_count, prior = {}, "", 0, {} state_unreadable = True + if host_guard and pending is False and (prior or state_unreadable): + # Retain an older valid attempt, or an opted-in unreadable state, + # before a later flag-off restart can lose that evidence. + try: + _save_host_pending(state_path, True) + except OSError: + state_unreadable = True + def save(fields: dict[str, Any]) -> None: _save_state(state_path, {**state, **fields} if host_guard else fields, durable=host_guard) @@ -544,6 +613,9 @@ def save(fields: dict[str, Any]) -> None: sleep(min(delay, 60.0)) retry_after_error = True break + # Persist intent separately before either the main + # state write or the Host invocation can be interrupted. + _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")} @@ -580,6 +652,10 @@ def save(fields: dict[str, Any]) -> None: last_wake_status="wake_failed" if wake_failed else "wake_ok", last_wake_failure_kind=wake_failure_kind) save({}) + if failure is None: + # Only confirmed success, after its main-state + # commit, resolves the independent witness. + _save_host_pending(state_path, False) wake_status = "wake_failed" if wake_failed else "wake_ok" _log( f"status=should_run {wake_status} " diff --git a/tests/test_external_scheduler_worker.py b/tests/test_external_scheduler_worker.py index 7f52bcec0b..1b73b51ab7 100644 --- a/tests/test_external_scheduler_worker.py +++ b/tests/test_external_scheduler_worker.py @@ -670,6 +670,257 @@ def test_guarded_cli_restart_keeps_hard_failure_and_public_safe_liveness(tmp_pat assert "PRIVATE-SENTINEL" not in first.stdout + second.stdout + Path(args.state_file).read_text() +@pytest.mark.parametrize("corrupt", [False, True]) +@pytest.mark.parametrize("enabled", [False, True]) +@pytest.mark.parametrize("terminal", [False, True]) +def test_cli_guard_state_switch_matrix(tmp_path, corrupt, enabled, terminal): + args, count = _guarded_host(tmp_path) + cli = Path(args.cli_bin) + payload = json.dumps(_hint_payload(should_run=True)) + _write_executable(cli, f"#!{sys.executable}\nprint({payload!r})\n") + command = [sys.executable, str(SCRIPTS / "external_scheduler_worker.py"), + "--cli-bin", args.cli_bin, "--registry", args.registry, + "--goal-id", args.goal_id, "--agent-id", args.agent_id, + "--state-file", args.state_file, "--wake-cmd", args.wake_cmd, "--once"] + first = subprocess.run([*command, "--wake-host-failure"], capture_output=True, + text=True, timeout=15) + assert first.returncode == 1 and count.read_text() == "1", first.stderr + if corrupt: + Path(args.state_file).write_text("broken-state") + if terminal: + payload = json.dumps(_hint_payload(should_run=False, local_scheduler_directive="stop")) + _write_executable(cli, f"#!{sys.executable}\nprint({payload!r})\n") + second = subprocess.run([*command, *(["--wake-host-failure"] if enabled else [])], + capture_output=True, text=True, timeout=15) + assert count.read_text() == "1", second.stdout + second.stderr + assert second.returncode == (0 if terminal else 2), second.stderr + assert ("status=terminal" if terminal else "host_failure_hold") in second.stdout + if corrupt: + assert Path(args.state_file).read_text() == "broken-state" + assert "PRIVATE-SENTINEL" not in first.stdout + second.stdout + + +@pytest.mark.parametrize("state_text", [None, "{}", "broken-state"]) +@pytest.mark.parametrize("ordinary_lock", [False, True]) +def test_never_guarded_cli_retains_ordinary_command_behavior(tmp_path, state_text, ordinary_lock): + count = tmp_path / "ordinary-calls" + cli = tmp_path / "quota" + payload = json.dumps(_hint_payload(should_run=True)) + _write_executable(cli, f"#!{sys.executable}\nprint({payload!r})\n") + state = tmp_path / "state.json" + if state_text is not None: + state.write_text(state_text) + if ordinary_lock: + from loopx.file_lock import try_exclusive_file_lock + with try_exclusive_file_lock(state, operation="ordinary_fixture") as held: + assert held is not None + wake = shlex.join([sys.executable, "-c", + f"from pathlib import Path; Path({str(count)!r}).write_text('called'); print('opaque success')"]) + completed = subprocess.run( + [sys.executable, str(SCRIPTS / "external_scheduler_worker.py"), + "--cli-bin", str(cli), "--registry", str(tmp_path / "registry"), + "--goal-id", "goal", "--agent-id", "agent", "--state-file", str(state), + "--wake-cmd", wake, "--once"], + capture_output=True, text=True, timeout=15, + ) + assert completed.returncode == 0 and count.read_text() == "called", completed.stderr + assert "host_failure_hold" not in completed.stdout + + +@pytest.mark.parametrize("replacement", [None, "broken-state"]) +def test_legacy_guarded_hold_cannot_downgrade_without_new_witness(tmp_path, monkeypatch, replacement): + args, count = _guarded_host(tmp_path) + args.once = True + monkeypatch.setattr(worker, "run_quota_should_run", lambda *a, **kw: _hint_payload(should_run=True)) + assert worker.run_worker(args) == 1 + # The predecessor retained only its primary state and original lock record. + Path(args.state_file + ".host-attempt").unlink() + state = Path(args.state_file) + if replacement is None: + state.unlink() + else: + state.write_text(replacement) + args.wake_host_failure = False + assert worker.run_worker(args) == 2 + assert count.read_text() == "1" + monkeypatch.setattr(worker, "run_quota_should_run", lambda *a, **kw: _hint_payload( + should_run=False, local_scheduler_directive="stop")) + assert worker.run_worker(args) == 0 + assert (state.read_text() if state.exists() else None) == replacement + + +@pytest.mark.parametrize("replacement", [None, "{}", "[]"]) +def test_pending_host_survives_missing_or_lost_primary_state(tmp_path, monkeypatch, replacement): + args, count = _guarded_host(tmp_path) + args.once = True + monkeypatch.setattr(worker, "run_quota_should_run", lambda *a, **kw: _hint_payload(should_run=True)) + assert worker.run_worker(args) == 1 + state = Path(args.state_file) + if replacement is None: + state.unlink() + else: + state.write_text(replacement) + args.wake_host_failure = False + assert worker.run_worker(args) == 2 + monkeypatch.setattr(worker, "run_quota_should_run", lambda *a, **kw: _hint_payload( + should_run=False, local_scheduler_directive="stop")) + assert worker.run_worker(args) == 0 + assert count.read_text() == "1" + assert (state.read_text() if state.exists() else None) == replacement + + +def test_pre_invocation_write_crash_is_held_after_disabling_adapter(tmp_path, monkeypatch): + args, count = _guarded_host(tmp_path) + state = Path(args.state_file) + monkeypatch.setattr(worker, "run_quota_should_run", lambda *a, **kw: _hint_payload(should_run=True)) + save = worker._save_state + def interrupted(path, fields, **kwargs): + if path == state and fields.get("host_attempt", {}).get("status") == "started": + raise KeyboardInterrupt + return save(path, fields, **kwargs) + monkeypatch.setattr(worker, "_save_state", interrupted) + with pytest.raises(KeyboardInterrupt): + worker.run_worker(args) + monkeypatch.setattr(worker, "_save_state", save) + args.wake_host_failure = False + assert worker.run_worker(args) == 2 + assert not count.exists() + monkeypatch.setattr(worker, "run_quota_should_run", lambda *a, **kw: _hint_payload( + should_run=False, local_scheduler_directive="stop")) + assert worker.run_worker(args) == 0 and not count.exists() + + +def test_interrupted_success_commit_cannot_replay_confirmed_host(tmp_path, monkeypatch): + args, count = _guarded_host(tmp_path, recover_at=1) + args.once = True + state = Path(args.state_file) + monkeypatch.setattr(worker, "run_quota_should_run", lambda *a, **kw: _hint_payload(should_run=True)) + save = worker._save_state + def interrupted(path, fields, **kwargs): + if path != state and fields.get("pending") is False: + raise OSError("synthetic result commit failure") + return save(path, fields, **kwargs) + monkeypatch.setattr(worker, "_save_state", interrupted) + assert worker.run_worker(args) == 2 + assert json.loads(state.read_text())["last_wake_status"] == "wake_ok" + monkeypatch.setattr(worker, "_save_state", save) + args.wake_host_failure = False + assert worker.run_worker(args) == 2 + assert count.read_text() == "1" + + +@pytest.mark.parametrize("witness", ["broken", "{}", "[]", '{"pending":"false"}']) +def test_unreadable_pending_witness_cannot_authorize_transient_retry(tmp_path, monkeypatch, witness): + args, count = _guarded_host(tmp_path, kind="provider_capacity") + args.once = True + monkeypatch.setattr(worker, "run_quota_should_run", lambda *a, **kw: _hint_payload(should_run=True)) + assert worker.run_worker(args, clock=lambda: 1000) == 1 + Path(args.state_file + ".host-attempt").write_text(witness) + assert worker.run_worker(args, clock=lambda: 2000) == 2 + assert count.read_text() == "1" + + +def test_confirmed_success_allows_disabling_adapter_for_ordinary_command(tmp_path, monkeypatch): + args, count = _guarded_host(tmp_path, recover_at=1) + args.once = True + monkeypatch.setattr(worker, "run_quota_should_run", lambda *a, **kw: _hint_payload(should_run=True)) + assert worker.run_worker(args) == 0 + args.wake_host_failure = False + ordinary = tmp_path / "ordinary-invoked" + args.wake_cmd = shlex.join([sys.executable, "-c", + f"from pathlib import Path; Path({str(ordinary)!r}).touch(); print('opaque success')"]) + assert worker.run_worker(args) == 0 + assert ordinary.exists() and count.read_text() == "1" + + +def test_worker_stop_readback_survives_unreadable_pending_witness(tmp_path, monkeypatch): + from loopx.extensions.process_runtime import CappedProcessResult + args, _ = _guarded_host(tmp_path) + args.wake_stop_exit_code = 75 + calls = [] + def stopped(*a, **kw): + calls.append(True) + return CappedProcessResult(75, b'{}', None) + monkeypatch.setattr(worker, "run_quota_should_run", lambda *a, **kw: _hint_payload(should_run=True)) + monkeypatch.setattr(worker, "_run_wake", stopped) + assert worker.run_worker(args) == 0 + Path(args.state_file + ".host-attempt").write_text("broken witness") + args.wake_host_failure = False + assert worker.run_worker(args) == 0 + assert calls == [True] + + +@pytest.mark.parametrize("state_text", [None, "{}"]) +def test_cli_flag_off_contender_cannot_dispatch_during_original_quota_read(tmp_path, state_text): + args, count = _guarded_host(tmp_path) + if state_text is not None: + Path(args.state_file).write_text(state_text) + ready, release = tmp_path / "ready", tmp_path / "release" + payload = json.dumps(_hint_payload(should_run=True)) + _write_executable(Path(args.cli_bin), f"#!{sys.executable}\n" + f""" +import os,pathlib,time +if os.environ.get('FIXTURE_FIRST') == '1': + pathlib.Path({str(ready)!r}).touch() + while not pathlib.Path({str(release)!r}).exists(): + time.sleep(.01) +print({payload!r}) +""") + command = [sys.executable, str(SCRIPTS / "external_scheduler_worker.py"), + "--cli-bin", args.cli_bin, "--registry", args.registry, + "--goal-id", args.goal_id, "--agent-id", args.agent_id, + "--state-file", args.state_file, "--wake-cmd", args.wake_cmd, "--once"] + first = subprocess.Popen([*command, "--wake-host-failure"], stdout=subprocess.PIPE, + stderr=subprocess.PIPE, text=True, + env={**os.environ, "FIXTURE_FIRST": "1"}) + try: + deadline = time.monotonic() + 10 + while not ready.exists() and first.poll() is None and time.monotonic() < deadline: + time.sleep(.01) + assert ready.exists() + second = subprocess.run(command, capture_output=True, text=True, timeout=15) + assert second.returncode == 2 and "worker_already_running" in second.stdout + assert not count.exists() + finally: + release.touch() + stdout, stderr = first.communicate(timeout=15) + assert first.returncode == 1 and count.read_text() == "1", stdout + stderr + + +def test_cli_flag_off_contender_cannot_bypass_inflight_guard_lock(tmp_path): + args, count = _guarded_host(tmp_path) + ready, release = tmp_path / "ready", tmp_path / "release" + host = tmp_path / "host.py" + source = host.read_text().replace( + "counter.write_text(str(n))", + "counter.write_text(str(n))\n" + f"pathlib.Path({str(ready)!r}).touch()\n" + f"while not pathlib.Path({str(release)!r}).exists():\n" + " __import__('time').sleep(.01)", + ) + host.write_text(source) + payload = json.dumps(_hint_payload(should_run=True)) + _write_executable(Path(args.cli_bin), f"#!{sys.executable}\nprint({payload!r})\n") + command = [sys.executable, str(SCRIPTS / "external_scheduler_worker.py"), + "--cli-bin", args.cli_bin, "--registry", args.registry, + "--goal-id", args.goal_id, "--agent-id", args.agent_id, + "--state-file", args.state_file, "--wake-cmd", args.wake_cmd, "--once"] + first = subprocess.Popen([*command, "--wake-host-failure"], stdout=subprocess.PIPE, + stderr=subprocess.PIPE, text=True) + try: + deadline = time.monotonic() + 10 + while not ready.exists() and first.poll() is None and time.monotonic() < deadline: + time.sleep(.01) + assert ready.exists() + Path(args.state_file).write_text("broken-state") + second = subprocess.run(command, capture_output=True, text=True, timeout=15) + assert second.returncode == 2 and "worker_already_running" in second.stdout + assert count.read_text() == "1" + finally: + release.touch() + stdout, stderr = first.communicate(timeout=15) + assert first.returncode == 1, stdout + stderr + + def test_explicit_host_stop_is_readable_on_restart_without_new_dispatch(tmp_path, monkeypatch, capsys): from loopx.extensions.process_runtime import CappedProcessResult args, _ = _guarded_host(tmp_path)