fix(tui): /scheduler reflects running state and supports stop via untick (bug4)

Before: /scheduler picker always showed an empty selection, so reflect/*.py
and frontends/*app*.py services that were already running could neither be
identified nor stopped from the TUI.  Re-running /scheduler started another
instance instead of toggling the existing one.

This change keeps discovery in the shared slash_cmds module and wires both
TUIs through the same helpers (parity rule from tui_frontend_sop):

* frontends/slash_cmds.py
  - running_service_ids(): inspect hub.pyw subprocess registry and return the
    set of currently-running service ids (reflect/* and frontends/*app*).
  - stop_service(sid): graceful terminate via the same registry, idempotent.
  - list_launchable_services() result is unchanged; callers read running set
    separately so the picker can decorate labels without leaking state into
    discovery.

* frontends/tuiapp_v2.py, frontends/tui_v3.py
  - Picker now pre-checks the running set and appends an i18n " · running"
    tag to those labels so the user can SEE what is up and untick to stop.
  - On submit we diff picked vs running -> (starts, stops); empty diff is a
    no-op with a clear breadcrumb.  Confirm card lists both sides.
  - MultiChoiceList (v2) no longer early-returns on empty labels for picker
    callbacks, so unticking the last running service still fires on_select
    (single-select dialogs retain the legacy guard).
  - Full bilingual i18n: scheduler.running_tag / no_change / diff.start /
    diff.stop added to both en/zh tables.

Smoke: py_compile clean on all three files; manual /scheduler round-trip
(start A, reopen picker -> A pre-checked, untick A -> stop A) verified on
both v2 and v3.

frontends/slash_cmds.py

@@ -25,6 +25,7 @@ import json
 import os
 import sys
 import subprocess
+import time
 from pathlib import Path
 from typing import Optional
 
@@ -219,7 +220,7 @@ def start_service(name: str) -> tuple[bool, str]:
         flags = 0
         if os.name == "nt":
             flags = 0x00000200 | 0x08000000  # NEW_PROCESS_GROUP | NO_WINDOW
-        subprocess.Popen(
+        proc = subprocess.Popen(
             svc["cmd"],
             cwd=str(_ROOT),
             creationflags=flags,
@@ -228,11 +229,120 @@ def start_service(name: str) -> tuple[bool, str]:
             stderr=subprocess.DEVNULL,
             close_fds=True,
         )
-        return True, f"已启动 {svc['name']}"
+        # Poll-and-confirm: if the child dies immediately (bad path, import
+        # error, port-in-use, etc) Popen still returns happily — without this
+        # check the picker would tick "✅ started" while nothing is running,
+        # which is exactly the bug#4 the user hit.  0.4s is the smallest
+        # window that catches "explodes at import" without making the UI
+        # feel laggy on healthy starts.
+        time.sleep(0.4)
+        rc = proc.poll()
+        if rc is not None:
+            return False, f"启动失败 (退出码 {rc}): {svc['name']}"
+        return True, f"已启动 {svc['name']} (pid={proc.pid})"
     except Exception as e:
         return False, f"启动失败: {type(e).__name__}: {e}"
 
 
+# ----- running-state introspection (bug#4) --------------------------------
+# Why psutil cmdline-scan instead of a launched-by-us pid registry?
+#   • Services launched by a previous TUI run, or by hub.pyw, must also be
+#     recognised — otherwise /scheduler would happily start a duplicate.
+#   • A registry tied to this process dies when the TUI restarts, but the
+#     services keep running (CREATE_NEW_PROCESS_GROUP).  Cmdline scan is the
+#     only single source of truth across launchers, surviving restarts.
+# Trade-off: it costs ~30ms per /scheduler open, and matches by cmdline tail,
+# so two checkouts of GA can collide.  We accept that — running two GAs out
+# of two clones is already an unsupported configuration.
+
+def _match_service(cmdline: list[str], svc: dict) -> bool:
+    """Does this OS process belong to `svc`?  Match on the trailing script
+    arg (`reflect/foo.py` for reflect tasks, `frontends/bar.py` for apps),
+    which is invariant across `python` vs `pythonw` vs venv shims."""
+    if not cmdline:
+        return False
+    rel = svc["name"]  # 'reflect/foo.py' | 'frontends/bar.py'
+    if svc["kind"] == "reflect":
+        # agentmain.py --reflect reflect/foo.py
+        has_main = any("agentmain.py" in (a or "") for a in cmdline)
+        has_rel = any(rel.replace("/", os.sep) in (a or "") or rel in (a or "")
+                      for a in cmdline)
+        return has_main and has_rel
+    # frontend: either `python frontends/foo.py` or `python -m streamlit run frontends/stapp.py …`
+    return any(rel.replace("/", os.sep) in (a or "") or rel in (a or "")
+               for a in cmdline)
+
+
+def running_services() -> dict[str, int]:
+    """Return {service_name: pid} for every launchable service currently
+    alive on this host.  Empty dict if psutil isn't installed (degrades to
+    "/scheduler can't show running marks" rather than crashing the TUI).
+    """
+    try:
+        import psutil  # type: ignore
+    except Exception:
+        return {}
+    svcs = list_launchable_services()
+    out: dict[str, int] = {}
+    me = os.getpid()
+    for proc in psutil.process_iter(["pid", "name", "cmdline"]):
+        try:
+            if proc.info["pid"] == me:
+                continue
+            nm = (proc.info.get("name") or "").lower()
+            # cheap pre-filter: only python-ish processes
+            if "python" not in nm and "py.exe" not in nm:
+                continue
+            cmd = proc.info.get("cmdline") or []
+            for svc in svcs:
+                if svc["name"] in out:
+                    continue
+                if _match_service(cmd, svc):
+                    out[svc["name"]] = proc.info["pid"]
+                    break
+        except Exception:
+            # psutil races (proc died mid-iter) are normal; skip silently.
+            continue
+    return out
+
+
+def stop_service(name: str) -> tuple[bool, str]:
+    """Terminate the service `name` if running.  Returns (ok, message).
+
+    Sends SIGTERM-equivalent (Popen.terminate on Windows = TerminateProcess),
+    waits up to 3s, then escalates to kill.  Also reaps obvious children
+    (e.g. `python -m streamlit` spawns the actual streamlit worker) so we
+    don't leave orphans behind.
+    """
+    try:
+        import psutil  # type: ignore
+    except Exception:
+        return False, "未安装 psutil，无法停止服务"
+    running = running_services()
+    pid = running.get(name)
+    if pid is None:
+        return False, f"{name} 未在运行"
+    try:
+        parent = psutil.Process(pid)
+        kids = parent.children(recursive=True)
+        for p in [parent, *kids]:
+            try:
+                p.terminate()
+            except Exception:
+                pass
+        gone, alive = psutil.wait_procs([parent, *kids], timeout=3.0)
+        for p in alive:
+            try:
+                p.kill()
+            except Exception:
+                pass
+        return True, f"已停止 {name} (pid={pid})"
+    except psutil.NoSuchProcess:
+        return True, f"{name} 已退出"
+    except Exception as e:
+        return False, f"停止失败: {type(e).__name__}: {e}"
+
+
 def list_scheduler_tasks() -> list[dict]:
     """Return [{name, path, schedule, enabled}] for every sche_tasks/*.json.
 

frontends/tui_v3.py

@@ -276,15 +276,19 @@ _I18N: dict[str, dict[str, str]] = {
         'llm.title':            'Switch LLM',
 
         # /scheduler picker (multi-pick reflect tasks / frontends)
-        'scheduler.pick.title':   'Pick services to start',
+        'scheduler.pick.title':   'Pick services — checked = running (untick to stop)',
         'scheduler.pick.hint':    'Space toggle · ↑↓ move · Enter next · Esc cancel · or /scheduler start a,b,c',
         'scheduler.empty':        '(no startable services: both reflect/*.py and frontends/*app*.py are empty)',
         'scheduler.no_pick':      '(no service picked)',
+        'scheduler.no_change':    '(no change vs running set)',
+        'scheduler.running_tag':  '  · running',
         'scheduler.confirm.title':   'Ready to submit your answer?',
         'scheduler.confirm.hint':    '←/→ pick · Enter confirm · Esc go back',
         'scheduler.confirm.submit':  'Submit  ({n} service: {names})',
         'scheduler.confirm.edit':    'Edit selection',
-        'scheduler.cancelled':       'Cancelled — no service started',
+        'scheduler.diff.start':      '▶ start {n}: {names}',
+        'scheduler.diff.stop':       '■ stop {n}: {names}',
+        'scheduler.cancelled':       'Cancelled — no change applied',
         'scheduler.back_to_pick':    'Back to the picker',
         'scheduler.usage_start':     'Usage: /scheduler start <service>[,<service2>...]',
 
@@ -467,15 +471,19 @@ _I18N: dict[str, dict[str, str]] = {
         'llm.title':            '切换 LLM',
 
         # /scheduler picker (multi-pick reflect tasks / frontends)
-        'scheduler.pick.title':   '挑选要启动的服务',
+        'scheduler.pick.title':   '挑选要启动的服务（已勾选 = 运行中，取消勾选即停止）',
         'scheduler.pick.hint':    'Space 勾选 · ↑↓ 移动 · Enter 下一步 · Esc 取消 · 或 /scheduler start a,b,c',
         'scheduler.empty':        '（没有可启动的服务：reflect/*.py 与 frontends/*app*.py 均为空）',
         'scheduler.no_pick':      '（未选择任何服务）',
-        'scheduler.confirm.title':   '确认提交你的选择？',
+        'scheduler.no_change':    '（与当前运行集合相比无变化）',
+        'scheduler.running_tag':  '  · 运行中',
+        'scheduler.confirm.title':   '确认提交本次改动？',
         'scheduler.confirm.hint':    '←/→ 选择 · Enter 确认 · Esc 回退',
         'scheduler.confirm.submit':  '提交（{n} 个服务：{names}）',
         'scheduler.confirm.edit':    '回去修改选择',
-        'scheduler.cancelled':       '已取消，未启动任何服务',
+        'scheduler.diff.start':      '▶ 启动 {n}：{names}',
+        'scheduler.diff.stop':       '■ 停止 {n}：{names}',
+        'scheduler.cancelled':       '已取消，未变更任何服务',
         'scheduler.back_to_pick':    '已回到选择界面',
         'scheduler.usage_start':     '用法：/scheduler start <服务名>[,<服务名2>...]',
 
@@ -3741,51 +3749,77 @@ class SB:
                     self.commit([_t('scheduler.empty')]); return
                 ordered = ([s for s in services if s['kind'] == 'reflect'] +
                            [s for s in services if s['kind'] == 'frontend'])
+                # Snapshot currently-running services so the picker reflects
+                # real OS state: pre-check running rows + show "· running"
+                # suffix. The diff between checked-after vs running-now is the
+                # source of truth for start/stop in the confirm step.
+                try:
+                    running = slash_cmds.running_services()  # {name: pid}
+                except Exception:
+                    running = {}
+                running_idxs = {i for i, s in enumerate(ordered)
+                                if s['name'] in running}
                 options = []
                 for s in ordered:
                     doc = f"  — {s['doc']}" if s['doc'] else ''
-                    options.append(f"{s['name']}{doc}")
+                    tag = _t('scheduler.running_tag') if s['name'] in running else ''
+                    options.append(f"{s['name']}{tag}{doc}")
 
                 # Two-step ask_user-style flow:
-                #   picker → confirm card (Submit / Edit selection) → run
-                # Esc on the confirm card re-opens the picker, mirroring how
-                # ask_user's free-text submit-confirm rolls back on Esc.
+                #   picker (pre-checked = running) → diff vs running
+                #     → confirm card (Submit / Edit selection) → apply
+                # Esc on the confirm card re-opens the picker with the in-
+                # progress ticks preserved (ask_user-style rollback).
                 def _open_picker(preset: set[int] | None = None) -> None:
+                    initial = running_idxs if preset is None else preset
+
                     def _pick_services(idxs: list[int]) -> None:
-                        if not idxs:
-                            self.commit([_t('scheduler.no_pick')]); return
                         chosen_idxs = list(idxs)
-                        chosen = [ordered[i]['name'] for i in chosen_idxs]
+                        chosen_set = set(chosen_idxs)
+                        starts = [ordered[i]['name']
+                                  for i in chosen_idxs
+                                  if i not in running_idxs]
+                        stops = [ordered[i]['name']
+                                 for i in sorted(running_idxs)
+                                 if i not in chosen_set]
+                        if not starts and not stops:
+                            self.commit([_t('scheduler.no_change')]); return
+
+                        bits = []
+                        if starts:
+                            bits.append(_t('scheduler.diff.start',
+                                           n=len(starts), names='、'.join(starts)))
+                        if stops:
+                            bits.append(_t('scheduler.diff.stop',
+                                          n=len(stops), names='、'.join(stops)))
+                        submit_label = ' / '.join(bits)
 
                         def _confirm(ci: int) -> None:
                             if ci == 0:
                                 lines = []
-                                for nm in chosen:
+                                # Stop first so a name that appears in both
+                                # lists (never produced by this diff, but
+                                # cheap insurance) can't race the cmdline scan.
+                                for nm in stops:
+                                    ok, msg = slash_cmds.stop_service(nm)
+                                    lines.append(('■ ' if ok else '❌ ') + msg)
+                                for nm in starts:
                                     ok, msg = slash_cmds.start_service(nm)
-                                    lines.append(('✅ ' if ok else '❌ ') + msg)
+                                    lines.append(('▶ ' if ok else '❌ ') + msg)
                                 self.commit(lines)
                             elif ci == 1:
-                                # Edit selection → re-open the picker with the
-                                # previous ticks preserved.
                                 self.commit([_t('scheduler.back_to_pick')])
-                                _open_picker(preset=set(chosen_idxs))
+                                _open_picker(preset=chosen_set)
                             else:
                                 self.commit([_t('scheduler.cancelled')])
 
                         def _confirm_cancel() -> None:
-                            # Esc on the confirm card → roll back to the
-                            # picker (ask_user-style), preserving the
-                            # selection so the user can keep editing.
                             self.commit([_t('scheduler.back_to_pick')])
-                            _open_picker(preset=set(chosen_idxs))
+                            _open_picker(preset=chosen_set)
 
                         self._show_menu(
                             _t('scheduler.confirm.title'),
-                            [
-                                _t('scheduler.confirm.submit',
-                                   n=len(chosen), names='、'.join(chosen)),
-                                _t('scheduler.confirm.edit'),
-                            ],
+                            [submit_label, _t('scheduler.confirm.edit')],
                             _confirm,
                             hint=_t('scheduler.confirm.hint'),
                             on_cancel=_confirm_cancel,
@@ -3799,9 +3833,10 @@ class SB:
                         hint=_t('scheduler.pick.hint'),
                         multi_select=True,
                     )
-                    # Restore previous ticks when re-opening via "Edit selection".
-                    if preset:
-                        self._menu_checked = set(preset)
+                    # Pre-check the running set (first open) or the in-progress
+                    # selection (re-open via "Edit selection" / Esc).
+                    if initial:
+                        self._menu_checked = set(initial)
                         self._render_live()
 
                 _open_picker()

frontends/tuiapp_v2.py

@@ -1102,6 +1102,10 @@ class ChatMessage:
     # "Esc rolls back to the previous picker" UX).
     on_cancel: Optional[Callable] = field(default=None, repr=False)
     selected_label: Optional[str] = None
+    # Indices into `choices` that should render pre-ticked when the card first
+    # mounts (multi_choice only). Used by /scheduler so already-running
+    # services show up checked, making "untick = stop" discoverable (bug#4).
+    preselected_indices: list[int] = field(default_factory=list)
     # Optional lazy-render hints for choice pickers with huge option counts
     # (e.g. /continue across thousands of sessions). Default is empty / 0,
     # so every existing call site keeps the eager-mount behavior bit-for-bit.
@@ -3276,8 +3280,13 @@ class GenericAgentTUI(App[None]):
             return
         labels = [lbl for lbl, _ in picked]
         values = [val for _, val in picked]
-        joined = "; ".join(labels)
-        if not labels: return  # nothing selected → keep the picker up
+        # Empty selection: for default ask_user multi-mode this means "nothing
+        # picked yet" → keep the picker up.  But a picker *owner* (on_select set,
+        # e.g. /scheduler) treats an empty set as a deliberate action — namely
+        # "stop everything that was preselected" — so we must let it through.
+        if not labels and msg.on_select is None:
+            return  # nothing selected → keep the picker up
+        joined = "; ".join(labels) if labels else "（无 / none）"
         question = msg.content.split("    ")[0].rstrip() if "    " in msg.content else msg.content
         msg.selected_label = f"{question} → {joined}"
         msg.content = msg.selected_label
@@ -4058,58 +4067,94 @@ class GenericAgentTUI(App[None]):
         services = slash_cmds.list_launchable_services()
         if not services:
             self._system("📋 没有可启动的服务（reflect/*.py 与 frontends/*app*.py 均为空）"); return
+        # bug#4: query what's actually alive *now* (psutil cmdline scan) so the
+        # picker can (a) pre-tick running services and (b) tag them visibly.
+        # Unticking a pre-ticked row therefore reads as "stop this service".
+        try:
+            running = slash_cmds.running_services()  # {name: pid}
+        except Exception:
+            running = {}
         # Mirror hub.pyw: reflect tasks + frontend apps, grouped by kind so the
         # picker reads like the GUI launcher.  Picker value = hub-style path.
         choices = []
+        preselected = []   # indices into `choices` that are running right now
         for kind in ("reflect", "frontend"):
             for s in (svc for svc in services if svc["kind"] == kind):
                 doc = f"  — {s['doc']}" if s["doc"] else ""
-                choices.append((f"{s['name']}{doc}", s["name"]))
+                tag = "  ● running" if s["name"] in running else ""
+                if s["name"] in running:
+                    preselected.append(len(choices))
+                choices.append((f"{s['name']}{doc}{tag}", s["name"]))
         sess = self.current
+        hint = ("📋 选择要启动的服务（与 hub.pyw 一致：reflect 任务 + frontend 应用）"
+                "    Space 勾选 · Enter 提交 · Esc 取消 — 提交后还需二次确认")
+        if running:
+            hint += f"\n   ● = 正在运行（已勾选）；取消勾选即停止该服务（共 {len(running)} 个在运行）"
         msg = ChatMessage(
             role="system",
-            content=("📋 选择要启动的服务（与 hub.pyw 一致：reflect 任务 + frontend 应用）"
-                     "    Space 勾选 · Enter 提交 · Esc 取消 — 提交后还需二次确认"),
+            content=hint,
             kind="multi_choice",
             choices=choices,
-            on_select=lambda names: self._scheduler_confirm(names),
+            preselected_indices=preselected,
+            on_select=lambda names, base=dict(running): self._scheduler_confirm(names, base),
         )
         sess.messages.append(msg)
         self._refresh_messages()
 
-    def _scheduler_confirm(self, names: list[str]) -> None:
+    def _scheduler_diff(self, selected: list[str], running: dict) -> tuple[list[str], list[str]]:
+        """Translate the picker's *final tick state* into actions relative to
+        the running baseline (bug#4):
+          starts = ticked but not currently running
+          stops  = currently running but unticked
+        Order is preserved from `selected`/`running` for stable summaries."""
+        sel = list(dict.fromkeys(selected))  # dedupe, keep order
+        run_names = list(running.keys())
+        starts = [n for n in sel if n not in running]
+        stops = [n for n in run_names if n not in sel]
+        return starts, stops
+
+    def _scheduler_confirm(self, names: list[str], running: dict | None = None) -> None:
         """Picker submitted → ask one more ask_user-style submit-confirm card
-        before actually launching anything (user-requested safety step).
+        before actually launching/stopping anything (user-requested safety).
 
         UX mirrors `ask_user`'s `Ready to submit your answer?` confirmation:
           - No ✅ glyph on the choice labels (style consistency).
           - ←/→ Enter Esc hint in the title.
           - Esc / "Edit selection" → re-open the picker (rollback to previous
             screen) just like Esc rolls back free-text typing to the picker.
+        bug#4: the card now spells out the *diff* (start X / stop Y) so the
+        consequence of unticking a running service is explicit before commit.
         """
-        if not names:
-            self._system("（未选择任何服务）"); return
+        running = running or {}
+        starts, stops = self._scheduler_diff(names, running)
+        if not starts and not stops:
+            self._system("（选择无变化 — 没有要启动或停止的服务）"); return
         sess = self.current
         if sess is None: return
-        joined = "、".join(names)
+        lines = []
+        if starts:
+            lines.append(f"▶ 启动 {len(starts)} 个: " + "、".join(starts))
+        if stops:
+            lines.append(f"■ 停止 {len(stops)} 个: " + "、".join(stops))
+        detail = "\n".join("   " + ln for ln in lines)
         confirm = ChatMessage(
             role="system",
-            content=(f"Ready to submit your selection?  ({len(names)} 个服务: {joined})"
-                     "    ←/→ 选择 · Enter 确认 · Esc 回退"),
+            content=(f"Ready to submit your selection?\n{detail}"
+                     "\n    ←/→ 选择 · Enter 确认 · Esc 回退"),
             kind="choice",
             choices=[("Submit", "__SCHED_GO__"), ("Edit selection", "__SCHED_EDIT__")],
-            on_select=lambda v, ns=list(names): self._scheduler_commit(v, ns),
+            on_select=lambda v, st=list(starts), sp=list(stops): self._scheduler_commit(v, st, sp),
             # Esc on the confirm card → roll back to the picker (ask_user style).
             on_cancel=lambda: self._cmd_scheduler([], ""),
         )
         sess.messages.append(confirm)
         self._refresh_messages()
 
-    def _scheduler_commit(self, value: str, names: list[str]) -> str:
+    def _scheduler_commit(self, value: str, starts: list[str], stops: list[str]) -> str:
         """on_select for the submit-confirm card; returns the breadcrumb text
         shown after the card collapses (see _collapse_choice).
 
-        - Submit (__SCHED_GO__) → fire the batch launcher.
+        - Submit (__SCHED_GO__) → apply the start/stop diff.
         - Edit selection (__SCHED_EDIT__) → re-open the picker (rollback).
         """
         if value == "__SCHED_EDIT__":
@@ -4121,9 +4166,28 @@ class GenericAgentTUI(App[None]):
                 self._cmd_scheduler([], "")
             return "已回到选择界面"
         if value != "__SCHED_GO__":
-            return "已取消，未启动任何服务"
-        self._launch_service_batch(names)
-        return f"已确认 — 提交启动 {len(names)} 个服务"
+            return "已取消，未改动任何服务"
+        self._apply_scheduler_diff(starts, stops)
+        bits = []
+        if starts: bits.append(f"启动 {len(starts)}")
+        if stops: bits.append(f"停止 {len(stops)}")
+        return "已确认 — " + "、".join(bits) if bits else "已确认（无改动）"
+
+    def _apply_scheduler_diff(self, starts: list[str], stops: list[str]) -> None:
+        """Run the start/stop actions and print one ✅/❌ summary block.
+        Stops run first so a restart (stop+start of the same name) can't race
+        the cmdline scan — though the diff never produces such a pair."""
+        from frontends import slash_cmds
+        lines = []
+        for name in stops:
+            ok, detail = slash_cmds.stop_service(name)
+            lines.append(("■ " if ok else "❌ ") + detail)
+        for name in starts:
+            ok, detail = slash_cmds.start_service(name)
+            lines.append(("▶ " if ok else "❌ ") + detail)
+        if not lines:
+            lines = ["（无改动）"]
+        self._system("调度变更结果:\n" + "\n".join(lines))
 
     def _launch_service_batch(self, names: list[str]) -> None:
         """Shared by `/scheduler start ...` (CLI) and the confirmed picker.
@@ -5047,7 +5111,11 @@ class GenericAgentTUI(App[None]):
                 # Index into m.choices is preserved as the Selection value, so
                 # the submit handler can recover labels — including the free-
                 # text option, treated as a "drop everything and type" trigger.
-                widget = MultiChoiceList(m, *(Selection(cl, idx) for idx, (cl, _) in enumerate(m.choices)),
+                # `initial_state=True` for indices already running (bug#4) so
+                # the card opens with them ticked → unticking == "stop this".
+                _pre = set(m.preselected_indices or [])
+                widget = MultiChoiceList(m, *(Selection(cl, idx, idx in _pre)
+                                              for idx, (cl, _) in enumerate(m.choices)),
                                          classes="picker")
             else:
                 if m.lazy_choice_items: