From a7f610e9f5b5b4fbf9d7e5c0ddd6c7497eddbd44 Mon Sep 17 00:00:00 2001 From: Ace Data Cloud Dev Date: Sat, 4 Jul 2026 15:45:14 +0800 Subject: [PATCH] feat(portal): live daemon status + recent-turns panel MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The portal was a pure config editor — you couldn't tell from it whether 'channels start' was actually running or what it had done. Now the daemon writes a heartbeat + a bounded ring of content-free turn metrics under /status/, and the portal shows a live 'Daemon running/stopped + recent turns' card (polls every 5s). - channels/status.py: StatusStore (atomic heartbeat write/read + is_running via 60s-stale window; record_turn ring capped at 50; read_turns newest-first). - observability.py: optional, failure-isolated set_turn_sink() called after log_turn (no sink = turn path unchanged). - channels start: writes run.json, mirrors each TurnEvent into the ring, 20s heartbeat task, clears run.json on clean exit. - portal: status_overview() + GET /api/status; #status card (running dot, uptime, recent turns as adapter/instance/outcome/latency/chars — NEVER message content). 8 new tests; full suite 498 pass/1 skip. Playwright-verified card renders running + 4 turns from a seeded heartbeat. Adversarial (Explore subagent): VERDICT CLEAN (heartbeat lifecycle, atomic writes, single-writer race, privacy content-free, sink failure-isolated). --- coding_bridge/channels/observability.py | 21 ++++ coding_bridge/channels/portal.py | 25 +++++ coding_bridge/channels/portal_html.py | 58 +++++++++- coding_bridge/channels/status.py | 135 ++++++++++++++++++++++ coding_bridge/channels_cli.py | 46 ++++++++ tests/test_channels_portal.py | 143 ++++++++++++++++++++++++ 6 files changed, 426 insertions(+), 2 deletions(-) create mode 100644 coding_bridge/channels/status.py diff --git a/coding_bridge/channels/observability.py b/coding_bridge/channels/observability.py index 5532dad..9bebcf3 100644 --- a/coding_bridge/channels/observability.py +++ b/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) diff --git a/coding_bridge/channels/portal.py b/coding_bridge/channels/portal.py index d2ed6ab..3318276 100644 --- a/coding_bridge/channels/portal.py +++ b/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": diff --git a/coding_bridge/channels/portal_html.py b/coding_bridge/channels/portal_html.py index 4dd0fbe..e02c18a 100644 --- a/coding_bridge/channels/portal_html.py +++ b/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); diff --git a/coding_bridge/channels/status.py b/coding_bridge/channels/status.py new file mode 100644 index 0000000..fa60ff3 --- /dev/null +++ b/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 +``/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"] diff --git a/coding_bridge/channels_cli.py b/coding_bridge/channels_cli.py index 3956cfd..d1bf3c5 100644 --- a/coding_bridge/channels_cli.py +++ b/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() diff --git a/tests/test_channels_portal.py b/tests/test_channels_portal.py index e12339c..8bfbde1 100644 --- a/tests/test_channels_portal.py +++ b/tests/test_channels_portal.py @@ -4,6 +4,7 @@ import socket import threading +import time from http.server import ThreadingHTTPServer import httpx @@ -941,3 +942,145 @@ def test_index_has_recent_senders_control(): ): assert token in INDEX_HTML + +# --------------------------------------------------------------------------- # +# Run-status panel (StatusStore heartbeat + turn ring, read by the portal) +# --------------------------------------------------------------------------- # + + +def test_status_store_run_and_turns(tmp_path): + from coding_bridge.channels.status import StatusStore + + store = StatusStore(tmp_path / "status") + assert store.is_running() is False + store.write_run([{"adapter": "wechat", "instance_id": "wx"}], started_at=time.time()) + assert store.is_running() is True + assert store.read_run()["channels"][0]["instance_id"] == "wx" + store.record_turn({"ts": time.time(), "adapter": "telegram", "outcome": "ok"}) + store.record_turn({"ts": time.time(), "adapter": "wechat", "outcome": "timeout"}) + turns = store.read_turns() + assert [t["outcome"] for t in turns] == ["timeout", "ok"] # newest first + store.clear_run() + assert store.is_running() is False + + +def test_status_store_ring_bounded(tmp_path): + from coding_bridge.channels.status import StatusStore + + store = StatusStore(tmp_path / "status", max_turns=3) + for i in range(6): + store.record_turn({"ts": i, "adapter": "telegram", "outcome": "ok"}) + turns = store.read_turns(limit=100) + assert [t["ts"] for t in turns] == [5, 4, 3] # bounded to 3, newest first + + +def test_status_store_stale_heartbeat_not_running(tmp_path): + import json as _json + + from coding_bridge.channels.status import StatusStore + + (tmp_path / "status").mkdir(parents=True) + (tmp_path / "status" / "run.json").write_text( + _json.dumps({"pid": 1, "started_at": 0, "updated_at": 0, "channels": []}), + encoding="utf-8", + ) + assert StatusStore(tmp_path / "status").is_running() is False # ancient → stopped + + +def test_status_overview_running_and_stopped(tmp_path, monkeypatch): + from coding_bridge.channels.status import StatusStore + + s = _settings(tmp_path) + svc = PortalService(s, client=httpx.Client(transport=_tg_getme_transport())) + ov = svc.status_overview() + assert ov["running"] is False and ov["channels"] == [] and ov["recent_turns"] == [] + store = StatusStore(tmp_path / "status") + store.write_run([{"adapter": "telegram", "instance_id": "tg"}], started_at=time.time()) + store.record_turn( + {"ts": time.time(), "adapter": "telegram", "instance_id": "tg", "outcome": "ok"} + ) + ov = svc.status_overview() + assert ov["running"] is True + assert ov["channels"][0]["instance_id"] == "tg" + assert ov["recent_turns"][0]["outcome"] == "ok" + svc.close() + + +def test_http_status_route(tmp_path, monkeypatch): + from coding_bridge.channels.status import StatusStore + + s = _settings(tmp_path) + StatusStore(tmp_path / "status").write_run( + [{"adapter": "wechat", "instance_id": "wx"}], started_at=time.time() + ) + svc = PortalService(s, client=httpx.Client(transport=_tg_getme_transport())) + token = "tk" + port = _free_port() + httpd = ThreadingHTTPServer(("127.0.0.1", port), _make_handler(svc, token, port)) + thread = threading.Thread(target=httpd.serve_forever, daemon=True) + thread.start() + try: + r = httpx.get( + f"http://127.0.0.1:{port}/api/status", headers={"X-Portal-Token": token}, timeout=5 + ) + assert r.status_code == 200 and r.json()["running"] is True + finally: + httpd.shutdown() + httpd.server_close() + svc.close() + + +def test_observability_turn_sink_and_isolation(): + from coding_bridge.channels.observability import TurnEvent, log_turn, set_turn_sink + + seen: list[TurnEvent] = [] + set_turn_sink(seen.append) + try: + log_turn( + TurnEvent( + adapter="telegram", instance_id="tg", provider="claude", outcome="ok", + latency_ms=10, prompt_chars=1, reply_chars=2, session_id="s1", + ) + ) + finally: + set_turn_sink(None) + assert len(seen) == 1 and seen[0].instance_id == "tg" + # cleared → no more + log_turn( + TurnEvent( + adapter="a", instance_id="b", provider="c", outcome="ok", + latency_ms=0, prompt_chars=0, reply_chars=0, session_id="s2", + ) + ) + assert len(seen) == 1 + + +def test_observability_sink_failure_never_raises(): + from coding_bridge.channels.observability import TurnEvent, log_turn, set_turn_sink + + def boom(_ev): + raise RuntimeError("sink down") + + set_turn_sink(boom) + try: + log_turn( # must not raise despite the failing sink + TurnEvent( + adapter="a", instance_id="b", provider="c", outcome="ok", + latency_ms=0, prompt_chars=0, reply_chars=0, session_id="s", + ) + ) + finally: + set_turn_sink(None) + + +def test_index_has_status_panel(): + from coding_bridge.channels.portal_html import INDEX_HTML + + for token in ( + "function renderStatus", + "function startStatusPoll", + "/api/status", + "Recent turns", + ): + assert token in INDEX_HTML +