diff --git a/.gitignore b/.gitignore index 19b91d92..8d622c95 100644 --- a/.gitignore +++ b/.gitignore @@ -156,3 +156,6 @@ frontends/conductor_im_plugins/* # Beta 渠道：订阅门户插件（仅 beta 分发携带，不入库） plugins/subscription_portal.py + +# Local-only model reload regression tests +/frontends/tests/test_mykeys_reload.py diff --git a/agentmain.py b/agentmain.py index dfa01e6c..3eb288c8 100644 --- a/agentmain.py +++ b/agentmain.py @@ -46,6 +46,7 @@ class GenericAgent: def __init__(self): os.makedirs(os.path.join(script_dir, 'temp'), exist_ok=True) self.lock = threading.Lock() + self._model_lock = threading.RLock() self.task_dir = None self.history = []; self.handler = None; self.all_outputs = [] self.task_queue = queue.Queue() @@ -64,48 +65,57 @@ class GenericAgent: self.intervene = self.extrakeyinfo = None def load_llm_sessions(self): - mykeys, updated_at = mykeys_store.snapshot() - if hasattr(self, 'llmclients') and getattr(self, '_mykeys_updated_at', None) == updated_at: return - try: oldhistory, oldname = self.llmclient.backend.history, self.llmclient.backend.name - except: oldhistory = oldname = None - llm_sessions = [] - for k, cfg in mykeys.items(): - if not any(x in k for x in ['api', 'config', 'cookie']): continue - try: - if 'mixin' in k: llm_sessions += [{'mixin_cfg': cfg}] - elif c := resolve_client(k, cfg): llm_sessions += [c] - except: pass - for i, s in enumerate(llm_sessions): - if isinstance(s, dict) and 'mixin_cfg' in s: + with self._model_lock: + if self.is_running: return + mykeys, updated_at = mykeys_store.snapshot() + if hasattr(self, 'llmclients') and getattr(self, '_mykeys_updated_at', None) == updated_at: return + try: oldhistory = self.llmclient.backend.history + except: oldhistory = None + try: oldkey = self._llm_config_keys[self.llm_no] + except (AttributeError, IndexError): oldkey = None + llm_sessions, config_keys = [], [] + for k, cfg in mykeys.items(): + if not any(x in k for x in ['api', 'config', 'cookie']): continue try: - mixin = MixinSession(llm_sessions, s['mixin_cfg']) - if isinstance(mixin._sessions[0], (NativeClaudeSession, NativeOAISession)): llm_sessions[i] = NativeToolClient(mixin) - else: llm_sessions[i] = ToolClient(mixin) - except Exception as e: print(f'\n\n\n[ERROR] Failed to init MixinSession with cfg {s["mixin_cfg"]}: {e}!!!\n\n') - self.llmclients = llm_sessions - if not self.llmclients: - self.llmclient = None + if 'mixin' in k: llm_sessions += [{'mixin_cfg': cfg}] + elif c := resolve_client(k, cfg): llm_sessions += [c] + else: continue + config_keys.append(k) + except: pass + for i, s in enumerate(llm_sessions): + if isinstance(s, dict) and 'mixin_cfg' in s: + try: + mixin = MixinSession(llm_sessions, s['mixin_cfg']) + if isinstance(mixin._sessions[0], (NativeClaudeSession, NativeOAISession)): llm_sessions[i] = NativeToolClient(mixin) + else: llm_sessions[i] = ToolClient(mixin) + except Exception as e: print(f'\n\n\n[ERROR] Failed to init MixinSession with cfg {s["mixin_cfg"]}: {e}!!!\n\n') + self.llmclients, self._llm_config_keys = llm_sessions, config_keys + if not self.llmclients: + self.llmclient = None + self._mykeys_updated_at = updated_at + return + if oldkey in config_keys: self.llm_no = config_keys.index(oldkey) + self.llm_no %= len(self.llmclients) + self.llmclient = self.llmclients[self.llm_no] + if oldhistory: self.llmclient.backend.history = oldhistory self._mykeys_updated_at = updated_at - return - names = [c.backend.name if not isinstance(c, dict) else f'BADMIXIN_{i}' for i, c in enumerate(self.llmclients)] - if oldname in names: self.llm_no = names.index(oldname) - self.llmclient = self.llmclients[self.llm_no%len(self.llmclients)] - if oldhistory: self.llmclient.backend.history = oldhistory - self._mykeys_updated_at = updated_at def next_llm(self, n=-1): - self.load_llm_sessions() - if not self.llmclients: return - self.llm_no = ((self.llm_no + 1) if n < 0 else n) % len(self.llmclients) - lastc = self.llmclient - self.llmclient = self.llmclients[self.llm_no] - try: self.llmclient.backend.history = lastc.backend.history - except: raise Exception('[ERROR] BAD Mixin config: Check your mykey.py') - self.llmclient.last_tools = '' - load_tool_schema() + with self._model_lock: + if self.is_running: return + self.load_llm_sessions() + if not self.llmclients: return + self.llm_no = ((self.llm_no + 1) if n < 0 else n) % len(self.llmclients) + lastc = self.llmclient + self.llmclient = self.llmclients[self.llm_no] + try: self.llmclient.backend.history = lastc.backend.history + except: raise Exception('[ERROR] BAD Mixin config: Check your mykey.py') + self.llmclient.last_tools = '' + load_tool_schema() def list_llms(self): - self.load_llm_sessions() - return [(i, self.get_llm_name(b), i == self.llm_no) for i, b in enumerate(self.llmclients)] + with self._model_lock: + self.load_llm_sessions() + return [(i, self.get_llm_name(b), i == self.llm_no) for i, b in enumerate(self.llmclients)] def get_llm_name(self, b=None, model=False): b = self.llmclient if b is None else b if isinstance(b, dict): return 'BADCONFIG_MIXIN' @@ -158,36 +168,37 @@ class GenericAgent: task = self.task_queue.get() if isinstance(task, str): break raw_query, source, display_queue = task["query"], task["source"], task["output"] - raw_query = self._handle_slash_cmd(raw_query, display_queue) - if raw_query is None: - self.task_queue.task_done(); continue - self.is_running = True; self._current_queue = display_queue - if len(raw_query) > 2000: - task_file = os.path.join(script_dir, 'temp', f'user_prompt_{os.getpid()}_{time.time_ns()}.md') - with open(task_file, 'w', encoding='utf-8') as f: f.write(raw_query) - raw_query = f'Long user prompt saved to {task_file}. Read and execute.' - self.all_outputs.append({"input": raw_query, "outputs": []}) - if len(self.all_outputs) > 10000: self.all_outputs = self.all_outputs[-5000:] - rquery = smart_format(raw_query.replace('\n', ' '), max_str_len=200) - self.history.append(f"[USER]: {rquery}") - sys_prompt = get_system_prompt() + '\n'.join(self.extra_sys_prompts) + getattr(self.llmclient.backend, 'extra_sys_prompt', '') - if self.peer_hint: sys_prompt += f"\n[Peer] 用户提及其他会话/后台任务状态时: temp/model_responses/ (只找近期修改的文件尾部)\n" - handler = GenericAgentHandler(self, self.history, os.path.join(script_dir, 'temp')) - if getattr(self, 'no_print', False): handler.print = lambda *a, **k: None - if self.handler and 'key_info' in self.handler.working: - ki = re.sub(r'\n\[SYSTEM\] 此为.*?工作记忆[。\n]*', '', self.handler.working['key_info']) # 去旧 - handler.working['key_info'] = ki - handler.working['passed_sessions'] = ps = self.handler.working.get('passed_sessions', 0) + 1 - if ps > 0: handler.working['key_info'] += f'\n[SYSTEM] 此为 {ps} 个对话前设置的key_info，若已在新任务，先更新或清除工作记忆。\n' - self.handler = handler # although new handler, the **full** history is in llmclient, so it is full history! - self.llmclient.log_path = self.log_path - if self.force_non_stream: - self.llmclient.backend.stream = False - self.llmclient.backend.read_timeout = max(self.llmclient.backend.read_timeout, 1200) - gen = agent_runner_loop(self.llmclient, sys_prompt, raw_query, handler, TOOLS_SCHEMA, - max_turns=180, verbose=self.verbose, yield_info=True) + full_resp = ""; last_pos = 0; curr_turn = 0; turn_resps = [] try: - full_resp = ""; last_pos = 0; curr_turn = 0; turn_resps = self.all_outputs[-1]["outputs"] + with self._model_lock: + self.load_llm_sessions() + self.is_running = True; self._current_queue = display_queue + raw_query = self._handle_slash_cmd(raw_query, display_queue) + if raw_query is None: continue + if len(raw_query) > 2000: + task_file = os.path.join(script_dir, 'temp', f'user_prompt_{os.getpid()}_{time.time_ns()}.md') + with open(task_file, 'w', encoding='utf-8') as f: f.write(raw_query) + raw_query = f'Long user prompt saved to {task_file}. Read and execute.' + self.all_outputs.append({"input": raw_query, "outputs": turn_resps}) + if len(self.all_outputs) > 10000: self.all_outputs = self.all_outputs[-5000:] + rquery = smart_format(raw_query.replace('\n', ' '), max_str_len=200) + self.history.append(f"[USER]: {rquery}") + sys_prompt = get_system_prompt() + '\n'.join(self.extra_sys_prompts) + getattr(self.llmclient.backend, 'extra_sys_prompt', '') + if self.peer_hint: sys_prompt += f"\n[Peer] 用户提及其他会话/后台任务状态时: temp/model_responses/ (只找近期修改的文件尾部)\n" + handler = GenericAgentHandler(self, self.history, os.path.join(script_dir, 'temp')) + if getattr(self, 'no_print', False): handler.print = lambda *a, **k: None + if self.handler and 'key_info' in self.handler.working: + ki = re.sub(r'\n\[SYSTEM\] 此为.*?工作记忆[。\n]*', '', self.handler.working['key_info']) # 去旧 + handler.working['key_info'] = ki + handler.working['passed_sessions'] = ps = self.handler.working.get('passed_sessions', 0) + 1 + if ps > 0: handler.working['key_info'] += f'\n[SYSTEM] 此为 {ps} 个对话前设置的key_info，若已在新任务，先更新或清除工作记忆。\n' + self.handler = handler # although new handler, the **full** history is in llmclient, so it is full history! + self.llmclient.log_path = self.log_path + if self.force_non_stream: + self.llmclient.backend.stream = False + self.llmclient.backend.read_timeout = max(self.llmclient.backend.read_timeout, 1200) + gen = agent_runner_loop(self.llmclient, sys_prompt, raw_query, handler, TOOLS_SCHEMA, + max_turns=180, verbose=self.verbose, yield_info=True) for chunk in gen: if consume_file(self.task_dir, '_stop'): self.abort() if self.stop_sig: break @@ -208,9 +219,12 @@ class GenericAgent: display_queue.put({'done': full_resp + f'\n```\n{format_error(e)}\n```', 'source': source, 'turn': curr_turn, 'outputs': turn_resps.copy()}) finally: if self.stop_sig: print('User aborted the task.') - self.is_running = self.stop_sig = False # keep _current_queue: its final 'done' may still be unclaimed (refreshed UI salvages it); next task overwrites it - self.task_queue.task_done() if self.handler is not None: self.handler.code_stop_signal.append(1) + with self._model_lock: + self.is_running = self.stop_sig = False # keep _current_queue for final output retrieval + try: self.load_llm_sessions() + except Exception as e: print(f'[ERROR] Failed to reload models: {e}') + self.task_queue.task_done() GeneraticAgent = GenericAgent diff --git a/frontends/desktop_bridge.py b/frontends/desktop_bridge.py index 9f3644fb..d3c96748 100644 --- a/frontends/desktop_bridge.py +++ b/frontends/desktop_bridge.py @@ -672,6 +672,14 @@ class AgentManager: def _save_mykey_text(self, text: str) -> list: self._mykey_file().write_text(text, encoding="utf-8") + with self.lock: + for sess in self.sessions.values(): + if sess.agent is not None: + try: + sess.agent.list_llms() + emit_session_state(sess, sess.status) + except Exception as e: + bridge_print(f"[model reload] {sess.id}: {e}") return self.list_model_profiles() def add_model_profile(self, data: dict) -> dict: @@ -741,6 +749,9 @@ class AgentManager: agentmain = importlib.import_module("agentmain") GA = getattr(agentmain, "GenericAgent") agent = GA() + no = sess.llm_no if sess.llm_no is not None else _global_default_llm_no() + if no is not None: + agent.next_llm(int(no)) agent.inc_out = True agent.verbose = True threading.Thread(target=agent.run, daemon=True, name=f"GA-{sess.id}").start() @@ -796,7 +807,7 @@ class AgentManager: cfg = mk.get(k) if isinstance(mk.get(k), dict) else {} if "mixin" in k: members = [str(m) for m in (cfg.get("llm_nos") or [])] - out.append({"id": i, "varName": k, "kind": "mixin", "name": "", + out.append({"id": i, "varName": k, "kind": "mixin", "name": str(cfg.get("name") or k), "members": members, "active": i == active}) else: name = self._base_display_name(k, cfg) @@ -871,14 +882,14 @@ class AgentManager: def _live_model(sess: Session) -> Optional[dict]: """该会话 agent 当前真正在用的模型(渠道组会随故障转移变化)。 agent 还没建(没跑过 turn)时返回静态绑定信息,前端据 llmNo 回显选择器。 - llmNo: 始终以 sess.llm_no 为权威(用户选择),agent.llm_no 在初始化窗口可能滞后。""" + 已创建的 agent 是模型选择的唯一来源。""" ag = getattr(sess, "agent", None) if ag is None: return {"current": None, "isMixin": False, "llmNo": sess.llm_no, "runningLlmNo": sess.running_llm_no, "runningModel": sess.running_model} try: back = ag.llmclient.backend - live_no = sess.llm_no if sess.llm_no is not None else getattr(ag, "llm_no", 0) + live_no = sess.llm_no = ag.llm_no if "Mixin" in type(back).__name__: return {"current": back.current_name, "isMixin": True, "llmNo": live_no, "runningLlmNo": sess.running_llm_no, "runningModel": sess.running_model} @@ -1080,13 +1091,8 @@ class AgentManager: except Exception: pass agent = sess.agent - # 模型取会话绑定 sess.llm_no,未绑定回退全局默认。切换走 set_session_model。 - no = sess.llm_no if sess.llm_no is not None else _global_default_llm_no() - if no is not None and hasattr(agent, "next_llm"): - with contextlib.suppress(Exception): - agent.next_llm(int(no)) with self.lock: - sess.running_llm_no = getattr(agent, "llm_no", no) + sess.running_llm_no = agent.llm_no try: running_backend = agent.llmclient.backend sess.running_model = str( @@ -1311,18 +1317,18 @@ class AgentManager: def set_session_model(self, sid: str, llm_no: int) -> dict: """Bind a model to a session. - An idle agent switches immediately. A running turn keeps its captured client; - the new binding is applied at the start of the next turn. + Existing agents own selection; running turns ignore switch requests. """ with self.mutation(): sess = self.sessions.get(sid) if not sess: raise web.HTTPNotFound(text=json.dumps({"error": f"session not found: {sid}"}, ensure_ascii=False), content_type="application/json") - sess.llm_no = int(llm_no) - if (sess.status != "running" and sess.agent is not None - and hasattr(sess.agent, "next_llm")): - with contextlib.suppress(Exception): + if sess.status != "running": + if sess.agent is None: + sess.llm_no = int(llm_no) + else: sess.agent.next_llm(int(llm_no)) + sess.llm_no = sess.agent.llm_no sess.updated_at = time.time() self._persist_session(sess) return {"ok": True, "sessionId": sid, "llmNo": sess.llm_no, "model": self._live_model(sess)} diff --git a/frontends/tests/test_mykeys_reload.py b/frontends/tests/test_mykeys_reload.py deleted file mode 100644 index f18121ea..00000000 --- a/frontends/tests/test_mykeys_reload.py +++ /dev/null @@ -1,203 +0,0 @@ -"""Isolate configuration loading from real keys, network and agent startup.""" -import ast -import importlib.util -import json -import os -import sys -import threading -import time -from pathlib import Path -from types import SimpleNamespace -from concurrent.futures import ThreadPoolExecutor - -import pytest - -ROOT = Path(__file__).resolve().parents[2] - - -def extract(path, names, namespace): - tree = ast.parse(path.read_text(encoding="utf-8")) - nodes = [n for n in tree.body if getattr(n, "name", None) in names] - exec(compile(ast.Module(body=nodes, type_ignores=[]), str(path), "exec"), namespace) - - -@pytest.fixture -def core(tmp_path, monkeypatch): - # Never resolve or import the user's actual mykey module. - source = tmp_path / "mykey.py" - def find_spec(name): - assert name == "mykey" - return importlib.util.spec_from_file_location(name, source) if source.exists() else None - monkeypatch.setattr(importlib.util, "find_spec", find_spec) - monkeypatch.delitem(sys.modules, "mykey", raising=False) - ns = dict(os=os, sys=sys, json=json, time=time, threading=threading, - importlib=importlib, __file__=str(tmp_path / "llmcore.py")) - extract(ROOT / "llmcore.py", {"MyKeys", "reload_mykeys", "_load_mykeys", "__getattr__", "resolve_session", "resolve_client"}, ns) - ns["mykeys_store"] = ns["MyKeys"]() - yield ns, source - sys.modules.pop("mykey", None) - - -def write(source, value, stamp=1700000000000000100): - source.write_text("oai_config = " + repr({"name": "test", "model": value}) + "\n", encoding="utf-8") - os.utime(source, ns=(stamp, stamp)) - - -def agent_type(ns): - 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") - cls.body = [n for n in cls.body if getattr(n, "name", None) == "load_llm_sessions"] - exec(compile(ast.Module(body=[cls], type_ignores=[]), "agentmain.py", "exec"), ns) - return ns["GenericAgent"] - - -def test_two_agents_update_even_if_bridge_read_first(core): - ns, source = core - write(source, "A") - built = [] - def resolve(name, cfg): - built.append(cfg["model"]) - return SimpleNamespace(backend=SimpleNamespace(name=cfg["name"], model=cfg["model"], history=[])) - ns["resolve_client"] = resolve - Agent = agent_type(ns) - a, b = Agent(), Agent() - for ga in (a, b): - ga.llm_no = 0 - ga.load_llm_sessions() - history = [{"role": "user", "content": "preserve"}] - a.llmclient.backend.history = history - old = a.llmclient - a.load_llm_sessions() - assert a.llmclient is old - write(source, "B", 1700000000000000200) - assert ns["reload_mykeys"]()[1] is True # bridge consumes old changed signal - assert ns["reload_mykeys"]()[1] is False - for ga in (a, b): - ga.load_llm_sessions() - assert ga.llmclient.backend.model == "B" - assert a.llmclient.backend.history is history - assert built == ["A", "A", "B", "B"] - - -def test_same_second_same_size_and_backwards_mtime(core): - ns, source = core - write(source, "A") - store = ns["mykeys_store"] - first = store.snapshot()[1] - write(source, "B", 1700000000000000200) - data, second = store.snapshot() - assert data["oai_config"]["model"] == "B" and second > first - write(source, "C", 1600000000000000000) - data, third = store.snapshot() - assert data["oai_config"]["model"] == "C" and third > second - - -def test_json_fallback_and_legacy_interfaces(core): - ns, source = core - source.with_suffix(".json").write_text('{"oai_config": {"model": "A"}}', encoding="utf-8") - data, changed = ns["reload_mykeys"]() - assert changed is True - assert ns["__getattr__"]("mykeys") is data - assert ns["reload_mykeys"]() == (data, False) - assert ns["_load_mykeys"]() == data - with pytest.raises(AttributeError): ns["__getattr__"]("unknown") - - -def test_failure_keeps_last_good_snapshot_and_recovers(core): - ns, source = core - write(source, "A") - store = ns["mykeys_store"] - before = store.snapshot() - source.write_text("oai_config = {", encoding="utf-8") - assert store.snapshot() == before - source.unlink() - assert store.snapshot() == before - write(source, "B", 1700000000000000300) - assert store.snapshot()[0]["oai_config"]["model"] == "B" - - -def test_force_refresh_publishes_once_for_all_readers(core): - ns, source = core - write(source, "A") - store = ns["mykeys_store"] - before = store.snapshot()[1] - store.reload(force=True) - after = store.snapshot()[1] - assert after > before - with ThreadPoolExecutor(max_workers=8) as pool: - results = list(pool.map(lambda _: store.snapshot()[1], range(32))) - assert results == [after] * 32 - - -def test_resolvers_accept_snapshot_without_reading_global_or_mutating_it(core): - ns, source = core - class Session: - def __init__(self, cfg): self.cfg = cfg - ns.update(LLMSession=Session, ClaudeSession=Session, NativeClaudeSession=type("NC", (), {}), - NativeOAISession=type("NO", (), {}), ToolClient=lambda s: s) - cfg = {"model": "A"} - ns["reload_mykeys"] = lambda: pytest.fail("unexpected global config read") - session = ns["resolve_client"]("oai_config", cfg) - assert session.cfg == {"model": "A", "_mykey_name": "oai_config"} - assert cfg == {"model": "A"} - ns["reload_mykeys"] = lambda: ({"oai_config": cfg}, False) - assert ns["resolve_session"]("oai_config").cfg == session.cfg - - -def test_bridge_save_only_writes_config_and_reads_profiles(tmp_path): - tree = ast.parse((ROOT / "frontends/desktop_bridge.py").read_text(encoding="utf-8")) - cls = next(n for n in tree.body if isinstance(n, ast.ClassDef) and n.name == "AgentManager") - method = next(n for n in cls.body if getattr(n, "name", None) == "_save_mykey_text") - ns = {} - exec(compile(ast.Module(body=[method], type_ignores=[]), "desktop_bridge.py", "exec"), ns) - file = tmp_path / "synthetic_config.py" - manager = SimpleNamespace(_mykey_file=lambda: file, list_model_profiles=lambda: ["profile"]) - assert ns["_save_mykey_text"](manager, "synthetic = {}") == ["profile"] - assert file.read_text(encoding="utf-8") == "synthetic = {}" - - -def test_ga_preserves_selected_model_when_config_order_changes(core): - ns, source = core - ns["resolve_client"] = lambda name, cfg: SimpleNamespace(backend=SimpleNamespace(name=cfg["name"], history=[])) - Agent = agent_type(ns) - ga = Agent() - ga.llm_no = 1 - source.write_text("oai_config_a = {'name': 'A'}\noai_config_b = {'name': 'B'}\n", encoding="utf-8") - ga.load_llm_sessions() - assert ga.llmclient.backend.name == "B" - source.write_text("oai_config_b = {'name': 'B'}\noai_config_a = {'name': 'A'}\n", encoding="utf-8") - os.utime(source, ns=(1700000000000000000, 1700000000000000000)) - ga.load_llm_sessions() - assert ga.llmclient.backend.name == "B" and ga.llm_no == 0 - - -def test_ga_drops_stale_client_when_models_removed(core): - ns, source = core - ns["resolve_client"] = lambda name, cfg: SimpleNamespace(backend=SimpleNamespace(name=cfg["name"], history=[])) - Agent = agent_type(ns) - ga = Agent() - ga.llm_no = 0 - write(source, "A") - ga.load_llm_sessions() - source.write_text("", encoding="utf-8") - os.utime(source, ns=(1700000000000000500, 1700000000000000500)) - ga.load_llm_sessions() - assert ga.llmclients == [] and ga.llmclient is None - - -def test_loader_does_not_reuse_same_second_bytecode(core): - import py_compile - ns, source = core - write(source, "A") - py_compile.compile(str(source), doraise=True) - write(source, "B", 1700000000000000200) - assert ns["mykeys_store"].snapshot()[0]["oai_config"]["model"] == "B" - - -def test_legacy_direct_loader_raises_and_cleans_failed_module(core): - ns, source = core - source.write_text("partial = 1\nraise RuntimeError('synthetic failure')\n", encoding="utf-8") - with pytest.raises(RuntimeError, match="synthetic failure"): - ns["_load_mykeys"]() - assert "mykey" not in sys.modules - assert ns["reload_mykeys"]() == ({}, False)