Reuse independent conductor and scheduler without owning their lifecycle

frontends/desktop_bridge.py

@@ -1507,12 +1507,9 @@ def discover_extra_services(ga_root: Path) -> List[dict]:
         out.append({
             "id": "reflect/scheduler.py",
             "cmd": [sys.executable, "agentmain.py", "--reflect", "reflect/scheduler.py"],
+            "shared": True, "singleton_port": 45762,
         })
-    # conductor 跟 scheduler 一样,bridge 启动时自动拉起。--no-browser 是关键:
-    # conductor.py 默认会用 webbrowser.open 在用户浏览器弹一个 8900 端口 UI,
-    # 桌面版自启时不需要这个独立 UI(用户从「指挥家」页直接访问)。
-    # The desktop conductor evolves with this bridge. Keep both package-owned and inject the
-    # external core through GA_ROOT instead of executing arbitrary UI glue from that checkout.
+    # Shared services may be launched, but never owned by the desktop.
     conductor = APP_DIR / "conductor.py"
     if conductor.is_file():
         conductor_port = _configured_conductor_port()
@@ -1525,7 +1522,7 @@ def discover_extra_services(ga_root: Path) -> List[dict]:
                 "--port",
                 str(conductor_port),
             ],
-            "port": conductor_port,
+            "port": conductor_port, "shared": True,
         })
     return out
 
@@ -1540,6 +1537,23 @@ def _port_alive(port: Optional[int]) -> bool:
         return s.connect_ex(("127.0.0.1", port)) == 0
 
 
+def _shared_service_alive(svc: dict) -> bool:
+    port = svc.get("singleton_port")
+    if not port:
+        return _port_alive(svc.get("port"))
+    # Scheduler's lock socket never accepts: probing with connect fills its backlog.
+    import errno
+    import socket
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe:
+        try:
+            probe.bind(("127.0.0.1", port))
+        except OSError as exc:
+            if exc.errno == errno.EADDRINUSE or getattr(exc, "winerror", None) == 10048:
+                return True
+            raise
+    return False
+
+
 def _mem_mb(pid: Optional[int]) -> Optional[int]:
     if not pid:
         return None
@@ -1655,6 +1669,18 @@ class ServiceManager:
 
     def _state(self, sid: str, *, err: str = "") -> dict:
         proc = self.procs.get(sid)
+        svc = self._catalog.get(sid, {})
+        if svc.get("shared"):
+            running = _shared_service_alive(svc)
+            failed = proc is not None and proc.poll() is not None and not running
+            error = "" if running else err or (f"exit code {proc.returncode}" if failed else "")
+            return {
+                "id": sid, "status": "running" if running else "error" if error else "offline",
+                "running": running, "owned": False, "external": running,
+                "portConflict": False, "port": svc.get("port") or svc.get("singleton_port"),
+                "pid": None, "lastError": error, "errorKey": "",
+                "lastWarning": "", "warningKey": "",
+            }
         owned = proc is not None and proc.poll() is None
         running = owned
         status = "running" if owned else "offline"
@@ -1741,7 +1767,8 @@ class ServiceManager:
         item["name"] = sid
         item["memMb"] = _mem_mb(item.get("pid"))
         item["cpuPct"] = _cpu_pct(item.get("pid"))
-        item["managed"] = True
+        item["managed"] = not self._catalog[sid].get("shared", False)
+        item["canStart"] = True
         return item
 
     def list_panel_state(self) -> List[dict]:
@@ -1776,6 +1803,8 @@ class ServiceManager:
             svc = self._catalog.get(sid)
             if not svc:
                 raise KeyError(sid)
+            if svc.get("shared"):
+                return self._ensure_shared(sid, svc)
             proc = self.procs.get(sid)
             if proc is not None and proc.poll() is None:
                 return {"ok": True, "service": self._managed_state(sid)}
@@ -1817,6 +1846,31 @@ class ServiceManager:
                 return {"ok": False, "error": item["lastError"] or "start_failed", "service": item}
             return {"ok": True, "service": item}
 
+    def _ensure_shared(self, sid: str, svc: dict) -> dict:
+        if not _shared_service_alive(svc):
+            proc = self.procs.get(sid)
+            if proc is None or proc.poll() is not None:
+                kw = dict(cwd=str(self.ga_root), stdin=subprocess.DEVNULL,
+                          stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+                          env={**os.environ, "GA_ROOT": str(self.ga_root)})
+                if sys.platform == "win32":
+                    kw["creationflags"] = (subprocess.DETACHED_PROCESS |
+                                           subprocess.CREATE_NEW_PROCESS_GROUP)
+                else:
+                    kw["start_new_session"] = True
+                proc = subprocess.Popen(svc["cmd"], **kw)
+                self.procs[sid] = proc  # Observe this launch only; never terminate it.
+            deadline = time.monotonic() + 10
+            while not _shared_service_alive(svc) and time.monotonic() < deadline:
+                if proc.poll() is not None:
+                    break
+                time.sleep(.1)
+        item = self._managed_state(sid)
+        if not item["running"] and not item["lastError"]:
+            item = self._managed_state(sid, err="service not ready")
+        self._emit({"type": "service.changed", "service": item})
+        return {"ok": item["running"], "error": item["lastError"], "service": item}
+
     def autostart_extras(self) -> None:
         """Auto-start non-IM services on bridge boot. Currently:
           - reflect/scheduler.py (drives L4 archive cron every 12h).
@@ -1831,6 +1885,8 @@ class ServiceManager:
 
     def stop_all_extras(self) -> None:
         for sid in sorted(set(self._catalog) - set(self._im_catalog)):
+            if self._catalog[sid].get("shared"):
+                continue
             with contextlib.suppress(Exception):
                 self.stop_service(sid)
 
@@ -1850,11 +1906,9 @@ class ServiceManager:
         return bool(buf) and any("Exception in thread conductor-agent" in ln for ln in buf)
 
     def restart_broken_extras(self) -> None:
-        """mykey 被整体重写(导入密钥/编辑渠道配置)后,只重启「已经坏掉」的
-        conductor/scheduler。健康运行中的进程不动——它们会在下个任务靠自身 mtime
-        热重载新 key,强行重启反而会打断正在跑的任务。"""
+        """Retry failed owned extras after configuration changes; never restart shared services."""
         for sid in sorted(set(self._catalog) - set(self._im_catalog)):
-            if not self._extra_is_broken(sid):
+            if self._catalog[sid].get("shared") or not self._extra_is_broken(sid):
                 continue
             with contextlib.suppress(Exception):
                 self.stop_service(sid)
@@ -1869,6 +1923,8 @@ class ServiceManager:
         with self.lock:
             if sid not in self._catalog:
                 raise KeyError(sid)
+            if self._catalog[sid].get("shared"):
+                return {"ok": False, "error": "not_owned", "service": self._managed_state(sid)}
             self._stopping.add(sid)
             proc = self.procs.get(sid)
             stop_error = ""

frontends/tests/test_bridge_sessions.py

@@ -43,6 +43,7 @@ _plan_state_stub = _make_stub("plan_state", {
     "is_session_scoped_plan_path": lambda p, sid: True,
     "is_plan_preset_prompt": lambda prompt: False,
     "PLAN_PRESETS": {},
+    "desktop_plan_payload_from_session": lambda *args: None,
 })
 
 for _name in ("agentmain", "llmcore", "agent_loop", "plugins", "reflect",
@@ -1179,3 +1180,67 @@ def test_legacy_ui_model_is_neither_read_nor_written(manager, monkeypatch, tmp_p
     assert response.status == 200
     assert "llmNo" not in manager.config
     assert json.loads(settings.read_text(encoding="utf-8"))["ui"]["llmNo"] == 4
+
+
+class TestSharedServices:
+    @pytest.mark.parametrize('sid', ['frontends/conductor.py', 'reflect/scheduler.py'])
+    def test_online_service_is_reused_even_after_duplicate_exit(self, manager, tmp_path, monkeypatch, sid):
+        from types import SimpleNamespace
+        sm = TestManagedServiceOwnership.service_manager(tmp_path)
+        sm._catalog = {sid: {'id': sid, 'shared': True, 'port': 8900, 'cmd': ['unused']}}
+        sm.procs[sid] = SimpleNamespace(poll=lambda: 1, returncode=1)
+        monkeypatch.setattr(_mod, 'manager', manager)
+        monkeypatch.setattr(_mod, '_shared_service_alive', lambda svc: True)
+        def unexpected(*args, **kwargs):
+            raise AssertionError('must not spawn, stop, or restart a shared service')
+        monkeypatch.setattr(_mod.subprocess, 'Popen', unexpected)
+        state = sm.start_service(sid)['service']
+        assert state['running'] and state['status'] == 'running'
+        assert not state['owned'] and not state['managed'] and not state['portConflict']
+        assert not state['lastError'] and state['canStart']
+        assert sm.stop_service(sid)['error'] == 'not_owned'
+        assert sm.running_managed_ids() == []
+        monkeypatch.setattr(sm, 'stop_service', unexpected)
+        monkeypatch.setattr(sm, 'start_service', unexpected)
+        sm.stop_all_extras()
+        sm.restart_broken_extras()
+
+    def test_spawned_service_is_detached_and_never_owned(self, manager, tmp_path, monkeypatch):
+        from types import SimpleNamespace
+        sm = TestManagedServiceOwnership.service_manager(tmp_path)
+        sid = 'frontends/conductor.py'
+        sm._catalog[sid]['shared'] = True
+        calls = []
+        def spawn(cmd, **kw):
+            calls.append(kw)
+            return SimpleNamespace(poll=lambda: None)
+        monkeypatch.setattr(_mod, 'manager', manager)
+        monkeypatch.setattr(_mod, '_shared_service_alive', lambda svc: bool(calls))
+        monkeypatch.setattr(_mod.subprocess, 'Popen', spawn)
+        assert sm.start_service(sid)['ok']
+        assert sm.start_service(sid)['ok']
+        assert len(calls) == 1
+        kw = calls[0]
+        assert kw['stdout'] == _mod.subprocess.DEVNULL
+        assert kw['stdin'] == _mod.subprocess.DEVNULL
+        assert kw['env']['GA_ROOT'] == str(tmp_path)
+        if sys.platform == 'win32':
+            assert kw['creationflags'] & _mod.subprocess.DETACHED_PROCESS
+        else:
+            assert kw['start_new_session']
+        assert sm.stop_service(sid)['error'] == 'not_owned'
+        sm.stop_all_extras()
+        assert not sm.running_managed_ids()
+
+    def test_singleton_probe_does_not_fill_accept_backlog(self):
+        import socket
+        with socket.socket() as listener:
+            listener.bind(('127.0.0.1', 0))
+            listener.listen(1)
+            port = listener.getsockname()[1]
+            for _ in range(20):
+                assert _mod._shared_service_alive({'singleton_port': port})
+            listener.settimeout(.05)
+            with pytest.raises(socket.timeout):
+                listener.accept()
+        assert not _mod._shared_service_alive({'singleton_port': port})