fix(codex-app): refuse automation install without a resolved target thread

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 <host-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 <host-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():

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 == "<host-thread-id>" 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")