From e92ef58488c6c69d73389ed13aaa5f2bf19c2ebc Mon Sep 17 00:00:00 2001 From: Liang Jiaqing Date: Mon, 19 Oct 2026 10:18:15 +0800 Subject: [PATCH] Restore desktop sessions lazily into GA; use disposable output projection --- frontends/desktop_bridge.py | 95 ++++++++---- frontends/desktop_history.py | 38 +++++ frontends/tests/test_bridge_hub.py | 70 +++++++++ frontends/tests/test_bridge_sessions.py | 196 +----------------------- 4 files changed, 178 insertions(+), 221 deletions(-) diff --git a/frontends/desktop_bridge.py b/frontends/desktop_bridge.py index a4a8cd7c..9ad82f1c 100644 --- a/frontends/desktop_bridge.py +++ b/frontends/desktop_bridge.py @@ -239,6 +239,10 @@ class Session: updated_at: float = field(default_factory=time.time) messages: List[dict] = field(default_factory=list) msg_seq: int = 0 + output_cache_source: Any = None + output_cache_count: int = 0 + output_cache_tail: int = 0 + output_cache_signature: Any = None partial: Optional[dict] = None status: str = "idle" # idle|running|error|cancelled agent: Any = None @@ -250,7 +254,6 @@ class Session: untitled: bool = True plan_scan_baseline: int = 0 plan_path: str = "" - llm_history: Optional[List[dict]] = None # memory only; reconstructed from MR/L4 first_message: str = "" first_message_at: float = 0 log_refs: List[dict] = field(default_factory=list) @@ -450,20 +453,15 @@ class AgentManager: self._persist_session(s, strict=strict) def _session_from_item(self, item: dict) -> Optional["Session"]: - from desktop_history import legacy_anchors, load - anchors = legacy_anchors(item) + from desktop_history import metadata + meta = metadata(item) item = canonical_session_record(item, default_cwd=self.ga_root) - restored = load(self.ga_root, item, anchors) - if restored is None: - return None - msgs, history, meta = restored return Session(id=item["id"], title=item["title"], cwd=item["cwd"], created_at=item["created_at"], updated_at=item["updated_at"], - messages=msgs, msg_seq=max(item["msg_seq"], len(msgs)), pinned=item["pinned"], untitled=item["untitled"], - plan_scan_baseline=min(item["plan_scan_baseline"], len(msgs)), + plan_scan_baseline=item["plan_scan_baseline"], plan_path=_sanitize_desktop_plan_path(item["id"], item["plan_path"]), - status="idle", agent=None, llm_history=history, llm_no=item["llm_no"], + llm_no=item["llm_no"], first_message=meta["first_message"], first_message_at=meta["first_message_at"], log_refs=meta["log_refs"], has_content=meta["has_content"]) @@ -510,8 +508,6 @@ class AgentManager: except Exception as e: bridge_print(f"[bridge] migrate sessions failed: {e}") - if self.sessions: - self.active_session_id = max(self.sessions.values(), key=lambda s: s.updated_at).id def import_sessions(self, source_dir: str) -> dict: """把 source 的桌面会话合并进当前列表(按 id 去重)。 @@ -758,13 +754,15 @@ class AgentManager: if sess.llm_no is not None and sess.llm_no != agent.llm_no: agent.next_llm(int(sess.llm_no)) sess.llm_no = agent.llm_no - if sess.llm_history: - from continue_cmd import _derive_hist_info, _load_history_into - _load_history_into(agent, copy.deepcopy(sess.llm_history), _derive_hist_info(sess.llm_history)) + if sess.has_content: + from desktop_history import restore_agent + meta = restore_agent(agent, self.ga_root, self._session_dict(sess)) + sess.log_refs = meta["log_refs"] agent.inc_out = True agent.verbose = True threading.Thread(target=agent.run, daemon=True, name=f"GA-{sess.id}").start() sess.agent = agent + self._sync_display_cache(sess) from frontends import hub hub.connect(agent, f"desktop-{sess.id}", put_task=lambda text: self._hub_command(sess, "put", text), @@ -928,13 +926,46 @@ class AgentManager: return {"current": None, "isMixin": False, "llmNo": sess.llm_no, "runningLlmNo": sess.running_llm_no, "runningModel": sess.running_model} + def _sync_display_cache(self, sess: Session): + """Disposable projection. Completed prefixes are reused; GA owns all text.""" + agent = sess.agent + if agent is None or sess.status == "running" or getattr(agent, "is_running", False): + return # stream queue supplies the temporary in-progress overlay + tasks = getattr(agent, "all_outputs", None) + if tasks is None: + return + signature = (len(tasks), str(tasks[-1].get("input", "")), + tuple(tasks[-1].get("outputs") or [])) if tasks else (0,) + reset = sess.output_cache_source is not tasks or len(tasks) < sess.output_cache_count + if not reset and signature == sess.output_cache_signature and (sess.messages or not tasks): + return + from desktop_history import display_messages + start = 0 if reset or not sess.messages else max(0, sess.output_cache_count - 1) + offset = 0 if start == 0 else sess.output_cache_tail + projected = display_messages(tasks[start:]) + for i, message in enumerate(projected, offset): + # Retain display-only metadata (attachments/timestamps), never cached text. + old = sess.messages[i] if i < len(sess.messages) else {} + projected[i - offset] = {**old, **message, "id": i + 1, + "ts": message.get("ts") or old.get("ts")} + sess.messages[offset:] = projected + sess.msg_seq = len(sess.messages) + sess.output_cache_source = tasks + sess.output_cache_count = len(tasks) + last_size = len(display_messages(tasks[-1:])) if tasks else 0 + sess.output_cache_tail = len(sess.messages) - last_size + sess.output_cache_signature = signature + def snapshot(self, sess: Session, include_messages: bool = True) -> dict: + if include_messages: + self._sync_display_cache(sess) out = { "sessionId": sess.id, "id": sess.id, "title": sess.title, "cwd": sess.cwd, "status": sess.status, + "loaded": sess.agent is not None, "createdAt": sess.created_at, "updatedAt": sess.updated_at, "lastError": sess.last_error, @@ -1022,6 +1053,8 @@ class AgentManager: raise web.HTTPNotFound(text=json.dumps({"error": f"session not found: {sid}"}, ensure_ascii=False), content_type="application/json") if sess.status == "running": raise web.HTTPConflict(text=json.dumps({"error": "session is already running"}, ensure_ascii=False), content_type="application/json") + if sess.agent is None and _e2e_control_token() is None: + self.restore_context(sid) extra = {} if image_ids: extra["image_ids"] = image_ids @@ -1122,7 +1155,7 @@ class AgentManager: emit_session_state(sess, "idle") return if sess.agent is None: - sess.agent = self.make_agent(sess) + self.restore_context(sess.id) agent = sess.agent from desktop_history import preview log_path = getattr(agent, "log_path", None) @@ -1238,8 +1271,6 @@ class AgentManager: if turn_started: msg_extra["executionMs"] = round((time.time() - turn_started) * 1000) self.add_message(sess, "assistant", full, **msg_extra) - try: sess.llm_history = json.loads(json.dumps(agent.llmclient.backend.history, ensure_ascii=False, default=str)) - except Exception: pass sess.running_llm_no = None sess.running_model = None sess.status = "idle" @@ -1267,6 +1298,7 @@ class AgentManager: 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") + self._sync_display_cache(sess) msgs = [m for m in sess.messages if int(m.get("id", 0)) > after] if limit > 0: msgs = msgs[-limit:] @@ -1317,17 +1349,27 @@ class AgentManager: return {"ok": True, "sessionId": sid} def restore_context(self, sid: str) -> dict: - with self.lock: + # Serialize publication with mutations: concurrent clicks cannot create duplicate GAs. + 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") + raise web.HTTPNotFound(text=json.dumps({"error": f"session not found: {sid}"}), content_type="application/json") if sess.agent is not None: - return {"ok": True, "sessionId": sid, "restored": False, "reason": "agent already alive"} - agent = self.make_agent(sess) - with self.lock: - sess.agent = agent + return {"ok": True, "sessionId": sid, "restored": False, "loaded": True} + try: + self.make_agent(sess) + except FileNotFoundError: + # No source remains: omit the record, without deleting its metadata file. + self.sessions.pop(sid, None) + raise web.HTTPNotFound(text=json.dumps({"error": "history source no longer exists"}), content_type="application/json") + except Exception as exc: + sess.last_error = str(exc) + raise web.HTTPInternalServerError(text=json.dumps({"error": str(exc)}), content_type="application/json") sess.status = "idle" - return {"ok": True, "sessionId": sid, "restored": True, "messageCount": len(sess.llm_history or sess.messages)} + sess.last_error = "" + self._persist_session(sess) + return {"ok": True, "sessionId": sid, "restored": True, "loaded": True, + "messageCount": len(sess.messages)} def set_session_model(self, sid: str, llm_no: int) -> dict: """Bind a model to a session. @@ -2268,8 +2310,7 @@ async def cancel_handler(request): async def restore_handler(request): sid = request.match_info["sid"] - with manager.mutation(): - return json_ok(manager.restore_context(sid)) + return json_ok(await asyncio.to_thread(manager.restore_context, sid)) async def session_model_handler(request): diff --git a/frontends/desktop_history.py b/frontends/desktop_history.py index 0ceb65fc..0bc73915 100644 --- a/frontends/desktop_history.py +++ b/frontends/desktop_history.py @@ -183,3 +183,41 @@ def load(root, item, anchors=None): message['id'] = i meta['log_refs'] = found or meta['log_refs'] return messages, history or None, meta + + +def restore_agent(agent, root, item): + """Restore execution state and outputs together; never publish a half-restored GA.""" + restored = load(root, item) + if restored is None: + raise FileNotFoundError("history source no longer exists") + messages, history, meta = restored + try: + from .continue_cmd import _derive_hist_info + except ImportError: + from continue_cmd import _derive_hist_info + backend = agent.llmclient.backend + if not hasattr(backend, 'history'): + raise RuntimeError("model backend does not support history restoration") + outputs = [] + for message in messages: + if message['role'] == 'user': + outputs.append({'input': message['content'], 'outputs': [], 'ts': message.get('ts')}) + elif message['role'] == 'assistant' and outputs: + outputs[-1]['outputs'].extend(message.get('turn_segs') or [message['content']]) + backend.history = list(history or []) + agent.history = _derive_hist_info(history or []) + agent.all_outputs = outputs + return meta + + +def display_messages(outputs): + """Disposable desktop projection; the GA output list is authoritative.""" + messages = [] + for task in outputs: + messages.append({'id': len(messages) + 1, 'role': 'user', + 'content': task['input'], 'ts': task.get('ts')}) + steps = task.get('outputs') or [] + if steps: + messages.append({'id': len(messages) + 1, 'role': 'assistant', + 'content': ''.join(steps), 'turn_segs': list(steps), 'ts': task.get('ts')}) + return messages diff --git a/frontends/tests/test_bridge_hub.py b/frontends/tests/test_bridge_hub.py index a7278d20..8d2787ba 100644 --- a/frontends/tests/test_bridge_hub.py +++ b/frontends/tests/test_bridge_hub.py @@ -128,3 +128,73 @@ def test_closed_hub_never_connects(monkeypatch): client = hub.HubClient("closed") client.close() asyncio.run(client._loop()) + + +def test_lazy_restore_populates_ga_once(manager, monkeypatch): + from test_desktop_history import seed_mr, record + import desktop_history + seed_mr(Path(manager.ga_root), 'model_responses_lazy.txt') + meta = record('model_responses_lazy.txt') + meta.update(id='sess-lazy', title='lazy', cwd=manager.ga_root, llm_no=1) + sess = manager._session_from_item(meta) + manager.sessions[sess.id] = sess + assert sess.agent is None and sess.messages == [] + assert not hasattr(sess, 'llm_history') + assert manager.snapshot(sess)['loaded'] is False + connections = [] + monkeypatch.setattr(hub, 'connect', lambda agent, *args, **kw: connections.append(agent)) + monkeypatch.setattr(manager, 'ensure_ga_import_path', lambda: Path(manager.ga_root)) + monkeypatch.setitem(sys.modules, 'agentmain', _make_stub('agentmain', {'GenericAgent': FakeAgent})) + assert manager.restore_context(sess.id)['restored'] + agent = sess.agent + assert agent.llm_no == 1 + assert agent.llmclient.backend.history + assert agent.history + assert agent.all_outputs[0]['input'] == 'hello' + assert 'world' in ''.join(agent.all_outputs[0]['outputs']) + assert sess.messages == desktop_history.display_messages(agent.all_outputs) + assert connections == [agent] + assert not manager.restore_context(sess.id)['restored'] + assert connections == [agent] and sess.agent is agent + assert manager.snapshot(sess)['loaded'] is True + + +def test_missing_history_never_connects(manager, monkeypatch): + from test_desktop_history import record + meta = record('model_responses_missing.txt') + meta.update(id='sess-missing', title='missing', cwd=manager.ga_root) + sess = manager._session_from_item(meta) + manager.sessions[sess.id] = sess + monkeypatch.setattr(manager, 'ensure_ga_import_path', lambda: Path(manager.ga_root)) + monkeypatch.setitem(sys.modules, 'agentmain', _make_stub('agentmain', {'GenericAgent': FakeAgent})) + connections = [] + monkeypatch.setattr(hub, 'connect', lambda *a, **kw: connections.append(a)) + import pytest + with pytest.raises(_mod.web.HTTPNotFound): + manager.restore_context(sess.id) + assert sess.agent is None and not connections + assert sess.id not in manager.sessions + + +def test_display_cache_is_rebuildable_and_follows_ga(manager): + sess = manager.create_session() + sess.agent = FakeAgent() + sess.agent.all_outputs = [{'input': 'one', 'outputs': ['answer1']}] + manager._sync_display_cache(sess) + prefix = sess.messages[0] + assert [m['content'] for m in sess.messages] == ['one', 'answer1'] + sess.messages[0]['ts'] = 12345 + sess.agent.all_outputs.append({'input': 'two', 'outputs': ['answer2']}) + manager._sync_display_cache(sess) + assert [m['content'] for m in sess.messages] == ['one', 'answer1', 'two', 'answer2'] + assert sess.messages[0]['ts'] == 12345 + sess.agent.all_outputs.append({'input': 'three', 'outputs': ['answer3']}) + prefix = sess.messages[0] + manager._sync_display_cache(sess) + assert sess.messages[0] is prefix + sess.messages.clear() + manager._sync_display_cache(sess) + assert len(sess.messages) == 6 + sess.agent.all_outputs.clear() + manager._sync_display_cache(sess) + assert sess.messages == [] diff --git a/frontends/tests/test_bridge_sessions.py b/frontends/tests/test_bridge_sessions.py index 34330605..a8f901c6 100644 --- a/frontends/tests/test_bridge_sessions.py +++ b/frontends/tests/test_bridge_sessions.py @@ -106,7 +106,6 @@ def _make_session(sid: str = "sess-test-1", messages: list | None = None) -> Ses plan_path="", status="idle", agent=None, - llm_history=[{"role": "user", "content": "hello"}], llm_no=None, ) @@ -150,34 +149,6 @@ class TestPersistSessionConcurrentMutation: assert "messages" not in data assert data["storage_version"] == 2 - def test_concurrent_mutation_llm_history(self, manager: AgentManager): - """Mutating llm_history from another thread during persist must not corrupt.""" - sess = _make_session() - sess.llm_history = [{"role": "user", "content": "hi"}] - manager.sessions[sess.id] = sess - - stop = threading.Event() - - def mutator(): - i = 0 - while not stop.is_set(): - sess.llm_history.append({"role": "assistant", "content": f"turn-{i}"}) - i += 1 - time.sleep(0.001) - - t = threading.Thread(target=mutator, daemon=True) - t.start() - try: - for _ in range(15): - manager._persist_session(sess) - time.sleep(0.005) - finally: - stop.set() - t.join(timeout=2) - - f = manager._session_file(sess.id) - data = json.loads(f.read_text(encoding="utf-8")) - assert "llm_history" not in data class TestTransactionalDataImport: @@ -396,170 +367,9 @@ class TestSessionContinuityAfterRestart: assert sess.status == "idle" assert sess.messages == [] - def test_run_agent_turn_injects_history(self, manager: AgentManager): - """After bridge restart (agent=None, llm_history populated), run_agent_turn - should inject persisted llm_history into the newly created agent.""" - history = [ - {"role": "user", "content": [{"type": "text", "text": "hello"}]}, - {"role": "assistant", "content": [{"type": "text", "text": "hi there"}]}, - {"role": "user", "content": [{"type": "text", "text": "what did I just say?"}]}, - ] - sess = _make_session(sid="sess-continuity-1", messages=[ - {"role": "user", "content": "hello"}, - {"role": "assistant", "content": "hi there"}, - ]) - sess.llm_history = history - sess.agent = None - manager.sessions[sess.id] = sess - class FakeBackend: - def __init__(self): - self.history = [] - self.name = "test-backend" - class FakeLLMClient: - def __init__(self): - self.backend = FakeBackend() - class FakeAgent: - def __init__(self): - self.llmclient = FakeLLMClient() - self.llm_no = 0 - self.inc_out = True - self.verbose = True - - def next_llm(self, n): - self.llm_no = n - - fake_agent = FakeAgent() - with patch.object(manager, "make_agent", return_value=fake_agent): - if sess.agent is None: - sess.agent = manager.make_agent(sess) - if sess.llm_history: - try: - sess.agent.llmclient.backend.history = sess.llm_history - except Exception: - pass - - assert sess.agent.llmclient.backend.history == history - assert len(sess.agent.llmclient.backend.history) == 3 - - def test_no_history_no_crash(self, manager: AgentManager): - """New session with no llm_history should not crash on agent creation.""" - sess = _make_session(sid="sess-continuity-2") - sess.llm_history = None - sess.agent = None - manager.sessions[sess.id] = sess - - class FakeBackend: - def __init__(self): - self.history = [] - self.name = "test" - - class FakeLLMClient: - def __init__(self): - self.backend = FakeBackend() - - class FakeAgent: - def __init__(self): - self.llmclient = FakeLLMClient() - self.llm_no = 0 - - def next_llm(self, n): - self.llm_no = n - - fake_agent = FakeAgent() - with patch.object(manager, "make_agent", return_value=fake_agent): - if sess.agent is None: - sess.agent = manager.make_agent(sess) - if sess.llm_history: - try: - sess.agent.llmclient.backend.history = sess.llm_history - except Exception: - pass - - assert sess.agent.llmclient.backend.history == [] - - def test_empty_history_list_no_inject(self, manager: AgentManager): - """Empty llm_history list should not overwrite agent's default state.""" - sess = _make_session(sid="sess-continuity-3") - sess.llm_history = [] - sess.agent = None - manager.sessions[sess.id] = sess - - class FakeBackend: - def __init__(self): - self.history = [{"role": "system", "content": "default"}] - self.name = "test" - - class FakeLLMClient: - def __init__(self): - self.backend = FakeBackend() - - class FakeAgent: - def __init__(self): - self.llmclient = FakeLLMClient() - self.llm_no = 0 - - def next_llm(self, n): - self.llm_no = n - - fake_agent = FakeAgent() - with patch.object(manager, "make_agent", return_value=fake_agent): - if sess.agent is None: - sess.agent = manager.make_agent(sess) - if sess.llm_history: - try: - sess.agent.llmclient.backend.history = sess.llm_history - except Exception: - pass - - assert sess.agent.llmclient.backend.history == [{"role": "system", "content": "default"}] - - def test_model_preserved_after_restart(self, manager: AgentManager): - """sess.llm_no should be applied to recreated agent via next_llm.""" - sess = _make_session(sid="sess-continuity-4") - sess.llm_no = 3 - sess.llm_history = [{"role": "user", "content": [{"type": "text", "text": "test"}]}] - sess.agent = None - manager.sessions[sess.id] = sess - - class FakeBackend: - def __init__(self): - self.history = [] - self.name = "test" - - class FakeLLMClient: - def __init__(self): - self.backend = FakeBackend() - - class FakeAgent: - def __init__(self): - self.llmclient = FakeLLMClient() - self.llm_no = 0 - self.next_llm_calls = [] - - def next_llm(self, n): - self.llm_no = n - self.next_llm_calls.append(n) - - fake_agent = FakeAgent() - with patch.object(manager, "make_agent", return_value=fake_agent): - if sess.agent is None: - sess.agent = manager.make_agent(sess) - if sess.llm_history: - try: - sess.agent.llmclient.backend.history = sess.llm_history - except Exception: - pass - agent = sess.agent - no = sess.llm_no - if no is not None and hasattr(agent, "next_llm"): - agent.next_llm(int(no)) - - assert fake_agent.llm_no == 3 - assert fake_agent.next_llm_calls == [3] - assert fake_agent.llmclient.backend.history == sess.llm_history def test_persist_and_reload_preserves_llm_no(self, manager: AgentManager): """Full cycle: persist session with llm_no, reload, verify llm_no survives.""" @@ -571,7 +381,6 @@ class TestSessionContinuityAfterRestart: sess.first_message = meta["first_message"] sess.first_message_at = meta["first_message_at"] sess.llm_no = 5 - sess.llm_history = [{"role": "user", "content": [{"type": "text", "text": "hi"}]}] manager.sessions[sess.id] = sess manager._persist_session(sess) @@ -581,8 +390,7 @@ class TestSessionContinuityAfterRestart: reloaded = manager.sessions.get("sess-roundtrip") assert reloaded is not None assert reloaded.llm_no == 5 - assert reloaded.llm_history[0]["content"][0]["text"] == "hello" - assert len(reloaded.llm_history) == 2 + assert reloaded.agent is None and reloaded.messages == [] assert reloaded.agent is None @@ -1261,7 +1069,7 @@ class TestCanonicalSessionRestart: manager._load_sessions() assert set(manager.sessions) == {"sess-canonical"} - assert manager.active_session_id == "sess-canonical" + assert manager.active_session_id is None assert manager.sessions["sess-canonical"].updated_at == 11