diff --git a/scripts/codex_app_apply_rrule.py b/scripts/codex_app_apply_rrule.py index 6b13e41a93..0abe639543 100644 --- a/scripts/codex_app_apply_rrule.py +++ b/scripts/codex_app_apply_rrule.py @@ -271,6 +271,48 @@ def _ensure_sqlite_row( connection.close() +def _resolve_bound_thread_id( + goal: dict[str, Any], + *, + goal_id: str, + agent_id: str, + automation_id: str, +) -> str: + """Resolve the one bound host thread this automation must target. + + The installed manifest carries its target thread, and host-side readback + resolves a goal's threads through the goal's ``thread_agent_bindings``. + Installing with an unresolved thread therefore writes an automation that no + readback can observe, so an unbound or ambiguous binding is refused here + instead of being persisted as an empty value. + """ + bindings = (goal.get("coordination") or {}).get("thread_agent_bindings") + matches: list[str] = [] + for binding in bindings if isinstance(bindings, list) else []: + if not isinstance(binding, dict): + continue + if str(binding.get("agent_id") or "") != agent_id: + continue + thread_id = str(binding.get("thread_id") or "").strip() + if thread_id: + matches.append(thread_id) + if len(matches) != 1: + # The recovery command must be runnable as printed: ``bind-agent-thread`` + # is registered at the root, and without ``--execute`` it only previews + # the binding it is asked to write. + raise SystemExit( + f"{automation_id}: expected exactly one bound host thread for agent " + f"{agent_id!r}, found {len(matches)}; bind one with " + f"`loopx bind-agent-thread --goal-id {goal_id} --agent-id {agent_id} " + "--host-surface codex-app --thread-id --execute`, or " + "remove an extra binding with " + f"`loopx unbind-agent-thread --goal-id {goal_id} --agent-id {agent_id} " + "--host-surface codex-app --thread-id --execute`, " + "before installing the Codex App automation" + ) + return matches[0] + + def _ensure_automation( *, loopx: str, @@ -287,17 +329,12 @@ def _ensure_automation( if toml_path.exists() and _sqlite_row_exists(db_path, automation_id): return False goal = _load_registry_goal(registry, goal_id) - thread_id = "" - bindings = (goal.get("coordination") or {}).get("thread_agent_bindings") or [] - if isinstance(bindings, list): - matches = [ - binding - for binding in bindings - if isinstance(binding, dict) - and str(binding.get("agent_id") or "") == agent_id - ] - if len(matches) == 1: - thread_id = str(matches[0].get("thread_id") or "") + thread_id = _resolve_bound_thread_id( + goal, + goal_id=goal_id, + agent_id=agent_id, + automation_id=automation_id, + ) cwd = str(goal.get("repo") or "") name = f"{goal_id} LoopX" if toml_path.exists(): diff --git a/tests/test_codex_app_apply_rrule.py b/tests/test_codex_app_apply_rrule.py index 6cdf923b2b..70246340f9 100644 --- a/tests/test_codex_app_apply_rrule.py +++ b/tests/test_codex_app_apply_rrule.py @@ -2,16 +2,19 @@ import json import re +import shlex import sqlite3 import tomllib from pathlib import Path import pytest +from loopx.cli import build_parser from scripts.codex_app_apply_rrule import ( _now_ms, _parse_args, _scheduler_hint_turn_instance_id, + _sqlite_row_exists, _update_sqlite, _update_toml, main, @@ -465,3 +468,200 @@ def fake_run(command, **kwargs): assert "/tmp/workspace" in row[3] assert any("scheduler-ack-current" in call for call in calls) assert list(tmp_path.glob("codex-dev.db.bak-*")) + + +def _write_binding_registry(path: Path, *, bindings: list[dict]) -> None: + path.write_text( + json.dumps( + { + "goals": [ + { + "id": "goal", + "repo": "/tmp/workspace", + "coordination": {"thread_agent_bindings": bindings}, + } + ] + } + ), + encoding="utf-8", + ) + + +def _install_argv( + automations_root: Path, db_path: Path, registry_path: Path +) -> list[str]: + return [ + "--automations-root", + str(automations_root), + "--db-path", + str(db_path), + "--registry", + str(registry_path), + "--goal-id", + "goal", + "--agent-id", + "agent", + "--automation-id", + "loopx-goal-agent", + "--loopx", + "loopx", + ] + + +def _stub_loopx(monkeypatch) -> None: + def fake_run(command, **kwargs): + if "should-run" in command: + return _FakeCompleted(_hint_payload(apply_needed=True)) + if "heartbeat-prompt" in command: + return _FakeCompleted({"ok": True, "task_body": "Advance `goal`."}) + return _FakeCompleted({"ok": True}) + + monkeypatch.setattr("scripts.codex_app_apply_rrule.subprocess.run", fake_run) + + +def _install_against_bindings( + tmp_path: Path, + monkeypatch, + *, + bindings: list[dict], +) -> str: + """First-install the automation for one binding set and return the refusal.""" + + automations_root = tmp_path / "automations" + db_path = tmp_path / "codex-dev.db" + registry_path = tmp_path / "registry.json" + sqlite3.connect(str(db_path)).close() + _write_binding_registry(registry_path, bindings=bindings) + _stub_loopx(monkeypatch) + + with pytest.raises(SystemExit) as excinfo: + main(_install_argv(automations_root, db_path, registry_path)) + + # A refusal leaves no partial automation behind: no manifest, no row. + assert not (automations_root / "loopx-goal-agent" / "automation.toml").exists() + assert not _sqlite_row_exists(db_path, "loopx-goal-agent") + return str(excinfo.value) + + +def test_apply_refuses_automation_without_a_bound_thread( + tmp_path: Path, + monkeypatch, +) -> None: + """An unbound agent has no target thread, so nothing may be installed.""" + + message = _install_against_bindings(tmp_path, monkeypatch, bindings=[]) + + assert "expected exactly one bound host thread" in message + assert "found 0" in message + + +def test_apply_refuses_automation_with_an_ambiguous_bound_thread( + tmp_path: Path, + monkeypatch, +) -> None: + """Two threads for one agent cannot be resolved to the automation's target.""" + + message = _install_against_bindings( + tmp_path, + monkeypatch, + bindings=[ + {"agent_id": "agent", "thread_id": "thread-app"}, + {"agent_id": "agent", "thread_id": "thread-tui"}, + ], + ) + + assert "expected exactly one bound host thread" in message + assert "found 2" in message + + +_REFUSAL_COMMAND = re.compile(r"`(loopx [^`]+)`") + + +def _advertised_commands(message: str) -> list[list[str]]: + """The recovery commands the refusal hands to the operator.""" + + return [shlex.split(command) for command in _REFUSAL_COMMAND.findall(message)] + + +@pytest.mark.parametrize( + "bindings", + [ + [], + [ + {"agent_id": "agent", "thread_id": "thread-app"}, + {"agent_id": "agent", "thread_id": "thread-tui"}, + ], + ], + ids=["unbound", "ambiguous"], +) +def test_the_refusal_advertises_a_command_this_cli_can_run( + tmp_path: Path, + monkeypatch, + bindings: list[dict], +) -> None: + """A repair the parser rejects would leave the operator with a dead end.""" + + message = _install_against_bindings(tmp_path, monkeypatch, bindings=bindings) + commands = _advertised_commands(message) + + assert {command[1] for command in commands} == { + "bind-agent-thread", + "unbind-agent-thread", + } + for command in commands: + assert command[0] == "loopx" + # Substitute the operator's own thread for the placeholder, then run the + # argv through the registered parser: an unregistered path exits here. + argv = [ + "thread-app" if part == "" else part + for part in command[1:] + ] + parsed = build_parser().parse_args(argv) + + assert parsed.command == command[1] + # Without --execute the advertised command only previews the binding. + assert parsed.execute is True + assert ( + parsed.goal_id, + parsed.agent_id, + parsed.host_surface, + parsed.thread_id, + ) == ("goal", "agent", "codex-app", "thread-app") + + +def test_a_refused_install_continues_once_the_operator_binds_the_thread( + tmp_path: Path, + monkeypatch, +) -> None: + """The advertised repair has to actually unblock the install.""" + + automations_root = tmp_path / "automations" + db_path = tmp_path / "codex-dev.db" + registry_path = tmp_path / "registry.json" + sqlite3.connect(str(db_path)).close() + _write_binding_registry(registry_path, bindings=[]) + _stub_loopx(monkeypatch) + argv = _install_argv(automations_root, db_path, registry_path) + + with pytest.raises(SystemExit): + main(argv) + + # The operator follows the advertised binding, keeping every other argument. + _write_binding_registry( + registry_path, + bindings=[ + { + "agent_id": "agent", + "host_surface": "codex-app", + "thread_id": "thread-app", + } + ], + ) + + assert main(argv) == 0 + + manifest = automations_root / "loopx-goal-agent" / "automation.toml" + assert tomllib.loads(manifest.read_text(encoding="utf-8"))["target_thread_id"] == ( + "thread-app" + ) + assert _sqlite_row_exists(db_path, "loopx-goal-agent")