diff --git a/frontends/conductor.py b/frontends/conductor.py index c85fc99b..ef976597 100644 --- a/frontends/conductor.py +++ b/frontends/conductor.py @@ -1,6 +1,5 @@ import os, sys, re, time, json, uuid, queue, asyncio, threading, argparse, base64, secrets from dataclasses import dataclass, field -from pathlib import Path from typing import Dict, Any, Optional, List from contextlib import asynccontextmanager, suppress @@ -11,7 +10,6 @@ from pydantic import BaseModel FRONTENDS_DIR = os.path.dirname(os.path.abspath(__file__)) if FRONTENDS_DIR not in sys.path: sys.path.insert(0, FRONTENDS_DIR) -from desktop_settings import DesktopSettingsError, update_settings def _resolve_ga_root() -> str: """Use the external core selected by the package-owned desktop bridge when valid.""" @@ -72,41 +70,6 @@ HOST = args.host PORT = args.port -SETTINGS_PATH = Path.home() / ".ga_desktop_settings.json" - - -def _settings_doc() -> dict: - try: - doc = json.loads(SETTINGS_PATH.read_text(encoding="utf-8")) - return doc if isinstance(doc, dict) else {} - except Exception: - return {} - - -def _persist_conductor_llm_no(llm_no: int) -> None: - def mutate(document: dict) -> None: - section = document.get("conductor") - if not isinstance(section, dict): - section = {} - document["conductor"] = section - section["llmNo"] = llm_no - - update_settings(SETTINGS_PATH, mutate) - - -def _conductor_llm_no() -> Optional[int]: - """Read the model index bound to the conductor session. - Falls back to the legacy desktop default ui.llmNo for existing installs.""" - doc = _settings_doc() - for section in (doc.get("conductor"), doc.get("ui")): - if isinstance(section, dict) and section.get("llmNo") is not None: - try: - return int(section.get("llmNo")) - except (TypeError, ValueError): - pass - return None - - def _client_usable(agent: "GenericAgent") -> bool: return hasattr(getattr(agent, "llmclient", None), "backend") @@ -131,89 +94,12 @@ def _activate_model(agent: "GenericAgent", no: int) -> None: agent.next_llm(no) -def _runtime_model_state(agent: "GenericAgent", configured: Optional[int], reason: Optional[str]) -> dict: - effective = getattr(agent, "llm_no", None) if _client_usable(agent) else None - current = None - if _client_usable(agent): - with suppress(Exception): - current = str(agent.llmclient.backend.name) - return {"configured": configured, "effective": effective, - "fallbackReason": reason, "current": current} - - -def _apply_desktop_model(agent: "GenericAgent") -> dict: - """Make the conductor's session reflect the current desktop config before a task: - switch to its bound model if one is set, otherwise still refresh sessions from - mykey so live key/model edits (e.g. importing keys) take effect without a restart. - next_llm() already reloads internally; the no-bound-model branch must reload too, - or a conductor started on an empty/stale mykey would never pick up imported keys.""" - doc = _settings_doc() - conductor_cfg = doc.get("conductor") if isinstance(doc.get("conductor"), dict) else {} - ui_cfg = doc.get("ui") if isinstance(doc.get("ui"), dict) else {} - raw_configured = conductor_cfg.get("llmNo") - configured = _parse_model_no(raw_configured) - ui_default = _parse_model_no(ui_cfg.get("llmNo")) - - try: - agent.load_llm_sessions() # mtime-guarded; rebuilds only when mykey changed - except Exception as e: - print(f"[conductor] failed to refresh model sessions: {e}", file=sys.stderr) - - clients = getattr(agent, "llmclients", []) or [] - - configured_failed = False - if _usable_model(agent, configured): - try: - _activate_model(agent, configured) - return _runtime_model_state(agent, configured, None) - except Exception as e: - configured_failed = True - print(f"[conductor] configured model #{configured} is unavailable: {e}", file=sys.stderr) - - if _usable_model(agent, ui_default): - if raw_configured is None: - reason = "ui_default" - elif configured_failed: - reason = "configured_unavailable" - elif configured is not None: - reason = "configured_unavailable" if 0 <= configured < len(clients) else "invalid_configured" - else: - reason = "invalid_configured" - try: - _activate_model(agent, ui_default) - return _runtime_model_state(agent, configured, reason) - except Exception as e: - print(f"[conductor] UI default model #{ui_default} is unavailable: {e}", file=sys.stderr) - - for i, client in enumerate(clients): - if hasattr(client, "backend"): - try: - _activate_model(agent, i) - return _runtime_model_state(agent, configured, "first_available") - except Exception: - continue - - print("[conductor] no usable model is available", file=sys.stderr) - return {"configured": configured, "effective": None, - "fallbackReason": "no_models", "current": None} - - -def _selected_conductor_llm_no(agent: "GenericAgent") -> int: - configured = _conductor_llm_no() - return configured if _usable_model(agent, configured) else agent.llm_no - - -def _set_conductor_llm_no(agent: "GenericAgent", value: Any) -> int: - """Persist the standalone UI binding without mutating an in-flight client. - - The Conductor loop applies this binding at its existing idle task boundary, so - a selection made immediately before sending a message controls that next turn. - """ - no = _parse_model_no(value) - if no is None or not _usable_model(agent, no): - raise ValueError(f"llm index out of range or unavailable: {value}") - _persist_conductor_llm_no(no) - return no +def _runtime_model_state(agent: "GenericAgent") -> dict: + usable = _client_usable(agent) + return {"configured": None, + "effective": agent.llm_no if usable else None, + "fallbackReason": None if usable else "no_models", + "current": str(agent.llmclient.backend.name) if usable else None} def _select_llm(agent: "GenericAgent", llm: Any) -> bool: @@ -447,7 +333,7 @@ class SubagentPool: agent.inc_out = True agent.verbose = False agent.no_print = True - if not _select_llm(agent, llm): _apply_desktop_model(agent) + _select_llm(agent, llm) th = start_agent_runner(agent, f"subagent-{sid}") state = SubAgentState(id=sid, agent=agent, prompt=prompt, status="running", thread=th) with self.lock: self.subagents[sid] = state @@ -532,24 +418,30 @@ class Conductor: self._runner_thread: Optional[threading.Thread] = None self.log: list = [] self._model_lock = threading.RLock() - self._model_state = { - "configured": None, - "effective": None, - "fallbackReason": "no_models", - "current": None, - "running": False, - } + self._model_running = False def model_snapshot(self) -> dict: with self._model_lock: - return dict(self._model_state) + return {**_runtime_model_state(self.agent), "running": self._model_running} def _publish_model_state(self, state: dict, running: bool) -> None: snapshot = {**state, "running": bool(running)} with self._model_lock: - self._model_state = snapshot + self._model_running = bool(running) schedule_broadcast({"type": "model", "model": snapshot}) + def select_model(self, value: Any) -> int: + with self._model_lock: + if self._model_running or self.agent.task_queue.unfinished_tasks: + raise ValueError("conductor is busy; switch models after the task finishes") + no = _parse_model_no(value) + if no is None: + raise ValueError(f"invalid model index: {value}") + self.agent.load_llm_sessions() + _activate_model(self.agent, no) + self._publish_model_state(_runtime_model_state(self.agent), running=False) + return self.agent.llm_no + def notify(self, event: dict): self.inbox.put(event) def _wait_for_agent_idle(self, timeout: float = 10.0) -> None: @@ -657,24 +549,23 @@ API: {base}；requests，GET /readme查用法，GET /chat读未读对话，GET / # mutate the live client until that previous task has reached # ``task_done``; the queue join is the authoritative idle barrier. self._wait_for_agent_idle() - # Settings can change while this long-lived process remains healthy. - # Refresh only at the task boundary so an in-flight task keeps its - # client, while the next task observes the newly persisted binding. - model_state = _apply_desktop_model(self.agent) - if model_state.get("effective") is None: - self._publish_model_state(model_state, running=False) - self._record_unavailable_model(events) - deferred_events = events - continue - kind = events[0].get("type") - other_events = [event for event in events if event.get("type") != kind] - events = [event for event in events if event.get("type") == kind] - for event in other_events: - self.inbox.put(event) - prompt = self._build_prompt(events) - self._publish_model_state(model_state, running=True) - try: + with self._model_lock: + self.agent.load_llm_sessions() + model_state = _runtime_model_state(self.agent) + if model_state.get("effective") is None: + self._publish_model_state(model_state, running=False) + self._record_unavailable_model(events) + deferred_events = events + continue + kind = events[0].get("type") + other_events = [event for event in events if event.get("type") != kind] + events = [event for event in events if event.get("type") == kind] + for event in other_events: + self.inbox.put(event) + prompt = self._build_prompt(events) dq = self.agent.put_task(prompt, source="conductor") + self._publish_model_state(model_state, running=True) + try: self._drain(dq, events) finally: self._publish_model_state(model_state, running=False) @@ -682,7 +573,7 @@ API: {base}；requests，GET /readme查用法，GET /chat读未读对话，GET / def start(self): self.agent = GenericAgent() - self._publish_model_state(_apply_desktop_model(self.agent), running=False) + self._publish_model_state(_runtime_model_state(self.agent), running=False) threading.Thread(target=self._run, name="conductor-loop", daemon=True).start() @@ -838,17 +729,17 @@ async def websocket(ws: WebSocket): "log": conductor.log, "running": running, "model": conductor.model_snapshot(), "llms": conductor.agent.list_llms(), - "llm": _selected_conductor_llm_no(conductor.agent)}) + "llm": conductor.agent.llm_no}) while True: data = await ws.receive_json() if "llm" in data: try: - llm_no = _set_conductor_llm_no(conductor.agent, data["llm"]) - except (TypeError, ValueError, OSError, DesktopSettingsError) as error: + llm_no = conductor.select_model(data["llm"]) + except (TypeError, ValueError, RuntimeError) as error: await ws.send_json({ "type": "model_error", "error": str(error), - "llm": _selected_conductor_llm_no(conductor.agent), + "llm": conductor.agent.llm_no, }) else: await broadcast({"type": "model_selected", "llm": llm_no}) diff --git a/frontends/tests/test_conductor_model_routing.py b/frontends/tests/test_conductor_model_routing.py index 3daf9e34..ca98bb11 100644 --- a/frontends/tests/test_conductor_model_routing.py +++ b/frontends/tests/test_conductor_model_routing.py @@ -61,6 +61,7 @@ class FakeAgent: self.next_llm_calls: list[int] = [] self.reloads = 0 self.failing = failing or set() + self.task_queue = queue.Queue() def load_llm_sessions(self): self.reloads += 1 @@ -79,20 +80,6 @@ class FakeAgent: return client.backend.model if model else client.backend.name -def test_out_of_range_config_falls_back_without_modulo(monkeypatch): - agent = FakeAgent(["zero", "one", "two", "three"]) - monkeypatch.setattr( - conductor, - "_settings_doc", - lambda: {"conductor": {"llmNo": 99}, "ui": {"llmNo": 1}}, - ) - - state = conductor._apply_desktop_model(agent) - - assert agent.next_llm_calls == [1] - assert state["effective"] == 1 - assert state["fallbackReason"] == "invalid_configured" - def test_conductor_default_port_override_is_e2e_scoped(monkeypatch): monkeypatch.delenv(conductor.E2E_REPORT_DIR_ENV, raising=False) @@ -114,45 +101,7 @@ def test_conductor_port_parser_rejects_invalid_values(value): conductor._parse_conductor_port(value) -def test_unusable_configured_client_falls_back_to_ui_default(monkeypatch): - agent = FakeAgent(["zero", "one", None, "three"]) - monkeypatch.setattr( - conductor, - "_settings_doc", - lambda: {"conductor": {"llmNo": 2}, "ui": {"llmNo": 1}}, - ) - state = conductor._apply_desktop_model(agent) - - assert agent.next_llm_calls == [1] - assert state["effective"] == 1 - assert state["fallbackReason"] == "configured_unavailable" - - -def test_configured_activation_failure_falls_back_to_ui_default(monkeypatch): - agent = FakeAgent(["zero", "one", "two"], failing={2}) - monkeypatch.setattr( - conductor, - "_settings_doc", - lambda: {"conductor": {"llmNo": 2}, "ui": {"llmNo": 1}}, - ) - - state = conductor._apply_desktop_model(agent) - - assert agent.next_llm_calls == [2, 1] - assert state["effective"] == 1 - assert state["fallbackReason"] == "configured_unavailable" - - -def test_missing_config_uses_first_usable_when_ui_is_invalid(monkeypatch): - agent = FakeAgent([None, None, "two", "three"]) - monkeypatch.setattr(conductor, "_settings_doc", lambda: {"ui": {"llmNo": 99}}) - - state = conductor._apply_desktop_model(agent) - - assert agent.next_llm_calls == [2] - assert state["effective"] == 2 - assert state["fallbackReason"] == "first_available" def test_explicit_numeric_worker_model_rejects_out_of_range(): @@ -164,52 +113,12 @@ def test_explicit_numeric_worker_model_rejects_out_of_range(): assert agent.next_llm_calls == [] -def test_standalone_model_selection_persists_without_mutating_live_client(tmp_path, monkeypatch): - settings_path = tmp_path / "settings.json" - settings_path.write_text('{"theme":"dark","ui":{"llmNo":0}}', encoding="utf-8") - monkeypatch.setattr(conductor, "SETTINGS_PATH", settings_path) - agent = FakeAgent(["zero", "one", "two"]) - selected = conductor._set_conductor_llm_no(agent, 2) - - assert selected == 2 - assert agent.llm_no == 0 - assert agent.next_llm_calls == [] - assert conductor._settings_doc() == { - "theme": "dark", - "ui": {"llmNo": 0}, - "conductor": {"llmNo": 2}, - } - - -def test_standalone_model_selection_rejects_unavailable_model(tmp_path, monkeypatch): - settings_path = tmp_path / "settings.json" - settings_path.write_text('{"theme":"dark"}', encoding="utf-8") - monkeypatch.setattr(conductor, "SETTINGS_PATH", settings_path) - agent = FakeAgent(["zero", None]) - - with pytest.raises(ValueError, match="unavailable"): - conductor._set_conductor_llm_no(agent, 1) - - assert conductor._settings_doc() == {"theme": "dark"} - - -def test_standalone_model_selection_applies_at_next_task_boundary(tmp_path, monkeypatch): - settings_path = tmp_path / "settings.json" - monkeypatch.setattr(conductor, "SETTINGS_PATH", settings_path) - agent = FakeAgent(["zero", "one", "two"]) - - conductor._set_conductor_llm_no(agent, 2) - state = conductor._apply_desktop_model(agent) - - assert agent.next_llm_calls == [2] - assert state["configured"] == 2 - assert state["effective"] == 2 - assert state["fallbackReason"] is None def test_runtime_model_snapshot_is_broadcast_with_running_state(monkeypatch): instance = conductor.Conductor() + instance.agent = FakeAgent(["model-one"]) payloads: list[dict] = [] monkeypatch.setattr(conductor, "schedule_broadcast", payloads.append) state = { @@ -221,12 +130,12 @@ def test_runtime_model_snapshot_is_broadcast_with_running_state(monkeypatch): instance._publish_model_state(state, running=True) - assert instance.model_snapshot() == {**state, "running": True} + assert instance.model_snapshot() == {**conductor._runtime_model_state(instance.agent), "running": True} assert payloads == [{"type": "model", "model": {**state, "running": True}}] def test_long_lived_conductor_refreshes_model_between_tasks(monkeypatch): - """A saved binding waits for task cleanup, then affects the next task.""" + """Display completion does not permit switching before task cleanup.""" class EndOfInbox(Exception): pass @@ -254,7 +163,6 @@ def test_long_lived_conductor_refreshes_model_between_tasks(monkeypatch): # Keep each event as a separate task boundary for this regression. return True - selected = {"llmNo": 1} timeline: list[tuple[str, int]] = [] allow_first_cleanup = threading.Event() first_cleanup_done = threading.Event() @@ -276,10 +184,7 @@ def test_long_lived_conductor_refreshes_model_between_tasks(monkeypatch): self.task_models.append(self.llm_no) timeline.append(("submit", self.llm_no)) if len(self.task_models) == 1: - # Simulate save_config while task one is already submitted. The - # display queue reports done before the agent runner persists its - # final state and calls task_done(). - selected["llmNo"] = 2 + # Display completion precedes task cleanup. assert self.llm_no == 1 self.task_queue.put("first") @@ -302,13 +207,9 @@ def test_long_lived_conductor_refreshes_model_between_tasks(monkeypatch): instance = conductor.Conductor() instance.agent = TaskAgent() + instance.agent.next_llm(1) instance.inbox = BoundaryInbox() payloads: list[dict] = [] - monkeypatch.setattr( - conductor, - "_settings_doc", - lambda: {"conductor": {"llmNo": selected["llmNo"]}}, - ) monkeypatch.setattr(conductor, "schedule_broadcast", payloads.append) monkeypatch.setattr(conductor, "start_agent_runner", lambda *_args: None) monkeypatch.setattr(conductor.time, "sleep", lambda _seconds: None) @@ -333,6 +234,8 @@ def test_long_lived_conductor_refreshes_model_between_tasks(monkeypatch): assert instance.agent.task_models == [1] assert timeline == [("select", 1), ("submit", 1)] + with pytest.raises(ValueError, match="busy"): + instance.select_model(2) allow_first_cleanup.set() assert first_cleanup_done.wait(timeout=2) runner.join(timeout=2) @@ -342,21 +245,20 @@ def test_long_lived_conductor_refreshes_model_between_tasks(monkeypatch): assert not runner.is_alive() assert run_errors == [] - assert instance.agent.task_models == [1, 2] + assert instance.agent.task_models == [1, 1] assert timeline == [ ("select", 1), ("submit", 1), - ("select", 2), - ("submit", 2), + ("submit", 1), ] assert instance.model_snapshot() == { - "configured": 2, - "effective": 2, + "configured": None, + "effective": 1, "fallbackReason": None, - "current": "two", + "current": "one", "running": False, } - assert [item["model"]["effective"] for item in payloads] == [1, 1, 2, 2] + assert [item["model"]["effective"] for item in payloads] == [1, 1, 1, 1] assert [item["model"]["running"] for item in payloads] == [True, False, True, False] @@ -397,6 +299,7 @@ def test_conductor_defers_events_without_models_and_recovers(monkeypatch): def load_llm_sessions(self): if ready["value"] and not self.llmclients: self.llmclients = [Client("recovered")] + self.llmclient = self.llmclients[0] def next_llm(self, no: int): self.llm_no = no @@ -417,7 +320,6 @@ def test_conductor_defers_events_without_models_and_recovers(monkeypatch): instance.agent = RecoveringAgent() instance.inbox = BoundaryInbox() payloads: list[dict] = [] - monkeypatch.setattr(conductor, "_settings_doc", lambda: {}) monkeypatch.setattr(conductor, "schedule_broadcast", payloads.append) monkeypatch.setattr(conductor, "start_agent_runner", lambda *_args: None) monkeypatch.setattr(conductor.time, "sleep", lambda _seconds: None) @@ -453,3 +355,112 @@ def test_parallel_worker_model_selection_keeps_agents_isolated(): assert selected == [True] * 10 assert [agent.llm_no for agent in agents] == requested assert [agent.next_llm_calls for agent in agents] == [[no] for no in requested] + + +def test_model_selection_is_memory_only_and_survives_refresh(monkeypatch): + instance = conductor.Conductor() + instance.agent = FakeAgent(["zero", "one", "two"]) + monkeypatch.setattr(conductor, "schedule_broadcast", lambda _: None) + def forbidden(*args, **kwargs): + raise AssertionError("model selection must not access settings files") + monkeypatch.setattr("builtins.open", forbidden) + monkeypatch.setattr(Path, "open", forbidden) + assert instance.select_model(2) == 2 + instance.agent.load_llm_sessions() + assert instance.model_snapshot()["effective"] == 2 + assert instance.agent.next_llm_calls == [2] + # A fresh agent uses its own default, not a previously saved selection. + other = conductor.Conductor() + other.agent = FakeAgent(["zero", "one", "two"]) + assert other.model_snapshot()["effective"] == 0 + + +@pytest.mark.parametrize("value", [99, -1, True, None, "invalid", 1]) +def test_invalid_model_selection_keeps_current_agent(value, monkeypatch): + instance = conductor.Conductor() + instance.agent = FakeAgent(["zero", None]) + monkeypatch.setattr(conductor, "schedule_broadcast", lambda _: None) + with pytest.raises(ValueError): + instance.select_model(value) + assert instance.agent.llm_no == 0 + assert instance.agent.next_llm_calls == [] + + +def test_running_model_selection_is_rejected(monkeypatch): + instance = conductor.Conductor() + instance.agent = FakeAgent(["zero", "one"]) + instance._model_running = True + with pytest.raises(ValueError, match="busy"): + instance.select_model(1) + assert instance.agent.next_llm_calls == [] + + +def test_actual_generic_agent_next_llm_preserves_history(monkeypatch): + # Execute the real method with inert clients; no credentials or LLM calls. + import ast + tree = ast.parse((ROOT / "agentmain.py").read_text(encoding="utf-8")) + cls = next(n for n in tree.body if isinstance(n, ast.ClassDef) and n.name == "GenericAgent") + method = next(n for n in cls.body if isinstance(n, ast.FunctionDef) and n.name == "next_llm") + namespace = {"load_tool_schema": lambda: None} + exec(compile(ast.Module(body=[method], type_ignores=[]), "agentmain.py", "exec"), namespace) + instance = conductor.Conductor() + instance.agent = FakeAgent(["zero", "one"]) + instance.agent.llmclient.backend.history = [{"role": "user", "content": "history"}] + history = instance.agent.llmclient.backend.history + instance.agent.next_llm = types.MethodType(namespace["next_llm"], instance.agent) + monkeypatch.setattr(conductor, "schedule_broadcast", lambda _: None) + assert instance.select_model(1) == 1 + assert instance.agent.llmclient is instance.agent.llmclients[1] + assert instance.agent.llmclient.backend.history is history + assert instance.agent.llmclient.last_tools == "" + + +def test_websocket_switch_reconnect_and_busy_error(monkeypatch): + from fastapi.testclient import TestClient + instance = conductor.Conductor() + instance.agent = FakeAgent(["zero", "one"]) + instance.agent.list_llms = lambda: [(i, c.backend.name, i == instance.agent.llm_no) + for i, c in enumerate(instance.agent.llmclients)] + monkeypatch.setattr(conductor, "conductor", instance) + monkeypatch.setattr(conductor, "schedule_broadcast", lambda _: None) + # Do not enter TestClient lifespan: no real runner or IM poller is started. + client = TestClient(conductor.app) + with client.websocket_connect("/ws") as ws: + assert ws.receive_json()["llm"] == 0 + ws.send_json({"llm": 1}) + assert ws.receive_json() == {"type": "model_selected", "llm": 1} + instance.agent.task_queue.put("running") + ws.send_json({"llm": 0}) + error = ws.receive_json() + assert error["type"] == "model_error" + assert error["llm"] == 1 + instance.agent.task_queue.get_nowait() + instance.agent.task_queue.task_done() + ws.send_json({"llm": 99}) + assert ws.receive_json()["type"] == "model_error" + with client.websocket_connect("/ws") as ws: + hello = ws.receive_json() + assert hello["llm"] == hello["model"]["effective"] == 1 + + +def test_conductor_has_no_desktop_settings_dependency(): + source = CONDUCTOR_PATH.read_text(encoding="utf-8") + for removed in ["desktop_settings", "SETTINGS_PATH", "_apply_desktop_model", "ui_default"]: + assert removed not in source + + +def test_start_uses_generic_agent_default_without_files(monkeypatch): + agent = FakeAgent(["zero", "one"]) + agent.next_llm(1) + monkeypatch.setattr(conductor, "GenericAgent", lambda: agent) + monkeypatch.setattr(conductor, "schedule_broadcast", lambda _: None) + monkeypatch.setattr(conductor.threading.Thread, "start", lambda _: None) + def forbidden(*args, **kwargs): + raise AssertionError("startup must not access desktop settings") + monkeypatch.setattr("builtins.open", forbidden) + monkeypatch.setattr(Path, "open", forbidden) + instance = conductor.Conductor() + instance.start() + assert instance.agent is agent + assert instance.model_snapshot()["effective"] == 1 + assert agent.next_llm_calls == [1]