Restore desktop sessions lazily into GA; use disposable output projection

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

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

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 == []

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