feat(portal): live daemon status + recent-turns panel

coding_bridge/channels/observability.py

@@ -18,6 +18,7 @@
 from __future__ import annotations
 
 import logging
+from collections.abc import Callable
 from dataclasses import dataclass
 from typing import Literal
 
@@ -44,6 +45,20 @@ class TurnEvent:
     session_id: str
 
 
+_turn_sink: Callable[[TurnEvent], None] | None = None
+
+
+def set_turn_sink(sink: Callable[[TurnEvent], None] | None) -> None:
+    """Register (or clear) an extra per-turn sink, invoked after the log record.
+
+    ``channels start`` uses this to mirror turn metrics into the status ring the
+    portal reads. Optional + failure-isolated: with no sink registered the turn
+    path is byte-for-byte unchanged.
+    """
+    global _turn_sink
+    _turn_sink = sink
+
+
 def log_turn(event: TurnEvent) -> None:
     """Emit exactly one structured INFO record for a completed turn.
 
@@ -72,3 +87,9 @@ def log_turn(event: TurnEvent) -> None:
             "channel_reply_chars": event.reply_chars,
         },
     )
+    sink = _turn_sink
+    if sink is not None:
+        try:
+            sink(event)
+        except Exception:  # observability must never break a turn
+            logger.debug("channel turn sink failed", exc_info=True)

coding_bridge/channels/portal.py

@@ -353,6 +353,29 @@ def list_approvals(self) -> list[dict[str, Any]]:
     def decide_approval(self, request_id: str, decision: str) -> bool:
         return self._approvals.decide(request_id, decision)
 
+    # ---- run status (read-only view of the `channels start` daemon) ---- #
+
+    def status_overview(self) -> dict[str, Any]:
+        """Whether the daemon is running + a content-free recent-turn ring."""
+        from .status import StatusStore
+
+        store = StatusStore(self._settings.config_dir / "status")
+        run = store.read_run()
+        running = store.is_running()
+        started_at: float | None = None
+        channels: list[dict[str, Any]] = []
+        if running and isinstance(run, dict):
+            sa = run.get("started_at")
+            started_at = float(sa) if isinstance(sa, (int, float)) else None
+            ch = run.get("channels")
+            channels = ch if isinstance(ch, list) else []
+        return {
+            "running": running,
+            "started_at": started_at,
+            "channels": channels,
+            "recent_turns": store.read_turns(limit=15),
+        }
+
     # ---- WeChat gateway proxy ------------------------------------------- #
 
     def _gateway_request(
@@ -836,6 +859,8 @@ def _handle_api_get(self, path: str, query: dict[str, list[str]]) -> None:
                     self._send_json(200, service.public_config())
                 elif path == "/api/approvals":
                     self._send_json(200, {"approvals": service.list_approvals()})
+                elif path == "/api/status":
+                    self._send_json(200, service.status_overview())
                 elif path == "/api/wechat/account":
                     self._send_json(200, service.account(instance))
                 elif path == "/api/wechat/status":

coding_bridge/channels/portal_html.py

@@ -100,6 +100,15 @@
 .appr .btn{padding:7px 14px}
 .btn.approve{background:var(--ok);color:#fff}
 .btn.deny{background:transparent;color:var(--danger);border:1px solid var(--danger)}
+.bigdot{width:11px;height:11px;border-radius:50%;background:var(--ok);box-shadow:0 0 0 4px color-mix(in srgb,var(--ok) 20%,transparent);flex:0 0 auto}
+.bigdot.off{background:var(--muted);box-shadow:0 0 0 4px color-mix(in srgb,var(--muted) 16%,transparent)}
+.turnrow{display:flex;align-items:center;gap:8px;padding:5px 0;font-size:12.5px;border-top:1px solid var(--border)}
+.turnrow:first-of-type{border-top:0}
+.tdot{width:7px;height:7px;border-radius:50%;flex:0 0 auto}
+.tdot.ok{background:var(--ok)}
+.tdot.bad{background:var(--danger)}
+.tadapter{font-weight:600}
+.tmeta{color:var(--muted);margin-left:auto;font-family:ui-monospace,Menlo,Consolas,monospace;font-size:11px}
 .chanlist{display:flex;flex-direction:column;gap:6px;margin-top:6px}
 .chan{display:flex;align-items:center;gap:12px;padding:11px 12px;border:1px solid var(--border);border-radius:12px;cursor:pointer;background:var(--bg)}
 .chan:hover{border-color:var(--accent)}
@@ -165,19 +174,22 @@
 let STATE={instances:[],idx:0,contacts:new Map()};
 let qrPollTimer=null;
 let approvalPollTimer=null;
+let statusPollTimer=null;
+let lastStatus=null;
 function stopQrPoll(){ if(qrPollTimer){ clearInterval(qrPollTimer); qrPollTimer=null; } }
 function stopApprovalPoll(){ if(approvalPollTimer){ clearInterval(approvalPollTimer); approvalPollTimer=null; } }
+function stopStatusPoll(){ if(statusPollTimer){ clearInterval(statusPollTimer); statusPollTimer=null; } }
 
 function current(){return STATE.instances[STATE.idx];}
 
 async function load(){
-  stopQrPoll(); stopApprovalPoll();
+  stopQrPoll(); stopApprovalPoll(); stopStatusPoll();
   const cfg=await api("/api/config"); STATE.instances=cfg.instances; STATE.idx=0;
   $("#cfgpath").textContent=cfg.config_path;
   $("#instwrap").innerHTML="";
   if(!STATE.instances.length){ $("#app").innerHTML=""; $("#app").append(noInstances()); $("#bar").style.display="none"; return; }
   $("#bar").style.display="block"; render();
-  loadForCurrent(); startApprovalPoll();
+  loadForCurrent(); startApprovalPoll(); startStatusPoll();
 }
 
 function loadForCurrent(){
@@ -203,6 +215,46 @@
   approvalPollTimer=setInterval(tick, 2500);
 }
 
+function startStatusPoll(){
+  stopStatusPoll();
+  const tick=async ()=>{ let d; try{ d=await api("/api/status"); }catch(e){ return; } renderStatus(d); };
+  tick();
+  statusPollTimer=setInterval(tick, 5000);
+}
+
+function fmtDur(sec){ sec=Math.max(0,Math.floor(sec)); const h=Math.floor(sec/3600),m=Math.floor(sec%3600/60),s=sec%60; return h?(h+"h "+m+"m"):(m?(m+"m "+s+"s"):(s+"s")); }
+
+function renderStatus(d){
+  lastStatus=d;
+  const box=$("#status"); if(!box) return;
+  const running=!!(d&&d.running);
+  const chans=(d&&d.channels)||[];
+  const card=el("div",{class:"card",id:"statuscard",style:running?"border-color:var(--ok)":""});
+  const up=(running&&d.started_at)?(" · up "+fmtDur(Date.now()/1000 - d.started_at)):"";
+  card.append(el("div",{class:"row",style:"align-items:center;gap:11px"},
+    el("span",{class:"bigdot"+(running?"":" off")}),
+    el("div",{style:"flex:1;min-width:0"},
+      el("div",{style:"font-weight:650"}, running?("Daemon running — "+chans.length+" channel"+(chans.length===1?"":"s")):"Daemon stopped"),
+      el("div",{class:"pill"}, running?("channels start is serving"+up):"run `coding-bridge channels start` to go live")),
+    el("button",{class:"btn ghost",style:"border:1px solid var(--border)",onclick:()=>startStatusPoll()},"Refresh")));
+  const turns=(d&&d.recent_turns)||[];
+  if(turns.length){
+    const t=el("div",{style:"margin-top:12px"});
+    t.append(el("label",{class:"fld",style:"margin:0 0 4px"},"Recent turns (no message content)"));
+    turns.slice(0,8).forEach(tr=>{
+      const ok=tr.outcome==="ok";
+      t.append(el("div",{class:"turnrow"},
+        el("span",{class:"tdot "+(ok?"ok":"bad")}),
+        el("span",{class:"tadapter"}, (tr.adapter||"?")+"/"+(tr.instance_id||"?")),
+        el("span",{class:"tmeta"}, (tr.provider||"")+" · "+(tr.outcome||"")+" · "+(tr.latency_ms||0)+"ms · "+(tr.reply_chars||0)+"c")));
+    });
+    card.append(t);
+  } else if(running){
+    card.append(el("div",{class:"empty",style:"margin-top:8px"},"no turns yet — message a channel to see activity"));
+  }
+  box.replaceChildren(card);
+}
+
 function renderApprovals(list){
   const box=$("#approvals"); if(!box) return;
   if(!list.length){ box.innerHTML=""; return; }
@@ -355,6 +407,8 @@
 function render(){
   const it=current(); const app=$("#app"); app.innerHTML="";
   app.append(el("div",{id:"approvals"}));
+  app.append(el("div",{id:"status"}));
+  if(lastStatus) renderStatus(lastStatus);
   app.append(renderOverview());
   app.append(el("div",{id:"onboarding"}));
   if(it.type==="telegram") renderTelegramEditor(app,it);

coding_bridge/channels/status.py

@@ -0,0 +1,135 @@
+"""Run-status heartbeat + recent-turn ring for the channels daemon.
+
+``channels start`` (the daemon) and ``channels portal`` (the UI) are separate
+processes. The daemon writes a small **heartbeat** file describing what's
+running and appends a bounded ring of recent turn *metrics* (adapter, instance,
+provider, outcome, latency, sizes) — **never** message content. The portal reads
+both to render a live status panel. Everything lives under
+``<config_dir>/status/`` and never leaves the box — same local trust boundary as
+the rest of the agent.
+
+Opt-in by process: only ``channels start`` writes here; if the daemon isn't
+running the files simply go stale / absent and the portal shows "stopped".
+"""
+
+from __future__ import annotations
+
+import contextlib
+import json
+import os
+import tempfile
+import time
+from pathlib import Path
+from typing import Any
+
+_RUN_FILE = "run.json"
+_TURNS_FILE = "turns.jsonl"
+_MAX_TURNS = 50
+
+#: A heartbeat older than this (seconds) is treated as "not running" — covers a
+#: daemon that was hard-killed (SIGKILL / power loss) without clearing run.json.
+RUN_STALE_SECONDS = 60.0
+
+
+def _write_atomic(path: Path, body: str) -> None:
+    fd, tmp_name = tempfile.mkstemp(dir=str(path.parent), prefix=".status-", suffix=".tmp")
+    tmp = Path(tmp_name)
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as f:
+            f.write(body)
+        if os.name == "posix":
+            with contextlib.suppress(OSError):
+                os.chmod(tmp, 0o600)
+        os.replace(tmp, path)
+    except BaseException:
+        with contextlib.suppress(OSError):
+            tmp.unlink()
+        raise
+
+
+class StatusStore:
+    """Cross-process run heartbeat + bounded recent-turn ring."""
+
+    def __init__(self, root: Path | str, *, max_turns: int = _MAX_TURNS) -> None:
+        self._root = Path(root)
+        self._max_turns = max(1, int(max_turns))
+
+    @property
+    def root(self) -> Path:
+        return self._root
+
+    def _ensure(self) -> None:
+        self._root.mkdir(parents=True, exist_ok=True)
+        if os.name == "posix":
+            with contextlib.suppress(OSError):
+                os.chmod(self._root, 0o700)
+
+    # ---- run heartbeat (daemon writes, portal reads) -------------------- #
+
+    def write_run(self, channels: list[dict[str, Any]], *, started_at: float) -> None:
+        """Refresh the heartbeat. ``channels`` is a content-free descriptor list."""
+        self._ensure()
+        payload = {
+            "pid": os.getpid(),
+            "started_at": float(started_at),
+            "updated_at": time.time(),
+            "channels": list(channels),
+        }
+        with contextlib.suppress(OSError):
+            _write_atomic(self._root / _RUN_FILE, json.dumps(payload))
+
+    def clear_run(self) -> None:
+        """Remove the heartbeat on a clean shutdown (best-effort)."""
+        with contextlib.suppress(OSError):
+            (self._root / _RUN_FILE).unlink()
+
+    def read_run(self) -> dict[str, Any] | None:
+        try:
+            data = json.loads((self._root / _RUN_FILE).read_text(encoding="utf-8"))
+        except (OSError, ValueError):
+            return None
+        return data if isinstance(data, dict) else None
+
+    def is_running(self, *, now: float | None = None) -> bool:
+        run = self.read_run()
+        if not isinstance(run, dict):
+            return False
+        updated = run.get("updated_at")
+        if not isinstance(updated, (int, float)):
+            return False
+        return (now or time.time()) - float(updated) < RUN_STALE_SECONDS
+
+    # ---- recent-turn ring (daemon appends, portal reads) --------------- #
+
+    def record_turn(self, event: dict[str, Any]) -> None:
+        """Append one content-free turn metric, keeping only the last N."""
+        self._ensure()
+        path = self._root / _TURNS_FILE
+        try:
+            lines = path.read_text(encoding="utf-8").splitlines() if path.exists() else []
+        except OSError:
+            lines = []
+        lines.append(json.dumps(event))
+        lines = lines[-self._max_turns :]
+        with contextlib.suppress(OSError):
+            _write_atomic(path, "\n".join(lines) + "\n")
+
+    def read_turns(self, limit: int = 20) -> list[dict[str, Any]]:
+        """Return the most recent turns, newest first."""
+        try:
+            lines = (self._root / _TURNS_FILE).read_text(encoding="utf-8").splitlines()
+        except (OSError, ValueError):
+            return []
+        out: list[dict[str, Any]] = []
+        for ln in lines[-max(0, limit) :]:
+            try:
+                d = json.loads(ln)
+            except ValueError:
+                continue
+            if isinstance(d, dict):
+                out.append(d)
+        out.reverse()
+        return out
+
+
+__all__ = ["StatusStore", "RUN_STALE_SECONDS"]

coding_bridge/channels_cli.py

@@ -23,6 +23,7 @@
 import logging
 import stat
 import sys
+import time
 from pathlib import Path
 from typing import TYPE_CHECKING
 
@@ -342,6 +343,41 @@ async def _go() -> None:
         # Shared across instances; only used by instances with require_approval.
         approval_store = ApprovalStore(settings.config_dir / "approvals")
 
+        # Run status + recent-turn ring the portal reads (content-free).
+        from .channels.observability import TurnEvent, set_turn_sink
+        from .channels.status import StatusStore
+
+        status_store = StatusStore(settings.config_dir / "status")
+        started_at = time.time()
+        channel_descs = [
+            {"adapter": "wechat", "instance_id": i.instance_id, "endpoint": i.base_url}
+            for i, _ in wechat_resolved
+        ] + [
+            {"adapter": "telegram", "instance_id": i.instance_id, "endpoint": i.api_base}
+            for i, _ in telegram_resolved
+        ]
+
+        def _record_turn(ev: TurnEvent) -> None:
+            status_store.record_turn(
+                {
+                    "ts": time.time(),
+                    "adapter": ev.adapter,
+                    "instance_id": ev.instance_id,
+                    "provider": ev.provider,
+                    "outcome": ev.outcome,
+                    "latency_ms": ev.latency_ms,
+                    "prompt_chars": ev.prompt_chars,
+                    "reply_chars": ev.reply_chars,
+                }
+            )
+
+        async def _heartbeat() -> None:
+            while not stop.is_set():
+                with contextlib.suppress(Exception):
+                    status_store.write_run(channel_descs, started_at=started_at)
+                with contextlib.suppress(asyncio.TimeoutError):
+                    await asyncio.wait_for(stop.wait(), timeout=20)
+
         def _make_dispatcher(inst: object) -> SessionDispatcher:
             return SessionDispatcher(
                 settings,
@@ -381,6 +417,10 @@ def _make_dispatcher(inst: object) -> SessionDispatcher:
             runners.append(asyncio.create_task(tg_adapter.run()))
             print(f"channel started: telegram/{inst.instance_id} → {inst.api_base}")
 
+        set_turn_sink(_record_turn)
+        status_store.write_run(channel_descs, started_at=started_at)
+        hb_task = asyncio.create_task(_heartbeat())
+
         try:
             # Wait for any runner to exit (auth failure, etc.) or Ctrl-C.
             done, pending = await asyncio.wait(
@@ -391,6 +431,12 @@ def _make_dispatcher(inst: object) -> SessionDispatcher:
                     logger.error("adapter exited: %s", t.exception().__class__.__name__)
         finally:
             stop.set()
+            set_turn_sink(None)
+            hb_task.cancel()
+            with contextlib.suppress(BaseException):
+                await hb_task
+            with contextlib.suppress(Exception):
+                status_store.clear_run()
             for a in adapters:
                 with contextlib.suppress(Exception):
                     await a.aclose()