From 0e1f163ef9d742b2746b259c777798be266dc805 Mon Sep 17 00:00:00 2001 From: AskaEth Date: Sun, 4 Oct 2026 13:58:03 +0800 Subject: [PATCH] =?UTF-8?q?feat(gateway,admin-web):=20=E5=8F=AF=E8=A7=82?= =?UTF-8?q?=E6=B5=8B=E9=9D=A2=E4=BB=8E=E8=BD=AE=E8=AF=A2=E6=94=B9=E4=B8=BA?= =?UTF-8?q?=20WS=20=E5=B8=A7=E6=8E=A8=E9=80=81=EF=BC=88=E8=A3=81=E5=86=B3?= =?UTF-8?q?=2005.4.3=20+=2016.4.3-WS=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增三个下行帧（纯新增，既有帧名/形状零改动）：observability_log / observability_trace / observability_metrics，各带 ≤1s 节拍、无新数据不发帧、单帧上限（日志 200 / span 100） - 帧壳 {type,data} 不带 sessionId；**不进 frameReachesAll 白名单** ⇒ fail-closed 只发 owner+admin （变异验证两次：加进白名单 → viewer 用例必红） - 推送开启条件＝owner/admin 凭据「读过一次」对应 REST（引导数据仍由该次读取返回）， 连接全断即退订并停循环（无常驻 goroutine）；既有 REST 端点路径/RBAC/形状完全不变 - 前端：删除 logs/metrics/trace 三处 useAutoReload 轮询；新增 observabilityStream（引导+去重合并+窗口） 与 observabilityViews（三个面板 + WS 徽标），长列表统一走已落地的 VirtualList - 复核：build/vet 绿 + gateway 全包测试绿（含真 bus→真 /ws/chat 三角色帧断言）+ `-race`； 前端 tsc 0 错 + vite build 成功 - 待登记契约：api-client 需新增「可观测实时推送」一节 --- .../web/src/admin/modules/Observability.tsx | 342 ++--------------- .../src/admin/modules/observabilityStream.ts | 306 +++++++++++++++ .../src/admin/modules/observabilityViews.tsx | 363 ++++++++++++++++++ frontend/web/src/lib/ws.ts | 42 ++ internal/services/gateway/observability.go | 13 +- .../services/gateway/observability_logs.go | 5 + .../services/gateway/observability_push.go | 284 ++++++++++++++ .../gateway/observability_push_test.go | 346 +++++++++++++++++ .../services/gateway/observability_trace.go | 51 ++- internal/services/gateway/ws_bridge.go | 9 + 10 files changed, 1449 insertions(+), 312 deletions(-) create mode 100644 frontend/web/src/admin/modules/observabilityStream.ts create mode 100644 frontend/web/src/admin/modules/observabilityViews.tsx create mode 100644 internal/services/gateway/observability_push.go create mode 100644 internal/services/gateway/observability_push_test.go diff --git a/frontend/web/src/admin/modules/Observability.tsx b/frontend/web/src/admin/modules/Observability.tsx index 6d1f5ab..4f76635 100644 --- a/frontend/web/src/admin/modules/Observability.tsx +++ b/frontend/web/src/admin/modules/Observability.tsx @@ -1,18 +1,24 @@ -import { useEffect, useState } from "react"; +import { useState } from "react"; import { useAppStore } from "../../store/useAppStore"; import { useAuditStore } from "../audit"; -import { CardGrid, Notice, PageHeader, StatCard } from "../components"; +import { Notice, PageHeader } from "../components"; import { Button, Surface, Tag } from "../../components/primitives"; -import { Icon } from "../../components/Icon"; import { useAsyncData } from "../../lib/useAsyncData"; import { useAutoReload } from "../hooks"; -import { - adminObservabilityAudit, - adminObservabilityLogs, - adminObservabilityMetrics, - adminObservabilityTrace, -} from "../../lib/api"; -import type { AdminLogLevel, AdminLogLine, AdminTraceSpan, AuditRecord } from "../../lib/types"; +import { adminObservabilityAudit } from "../../lib/api"; +import type { AuditRecord } from "../../lib/types"; +import { useObservabilityStream } from "./observabilityStream"; +import { LogsPanel, MetricsPanel, TracePanel, WsBadge } from "./observabilityViews"; + +/** + * 可观测性面板。日志 / 指标 / trace 三种数据在裁定 05.4.3 + 16.4.3-WS 后改为 + * **WS 帧推送**（`useObservabilityStream()`：挂载时各读一次引导数据，此后由 + * observability_log / observability_trace / observability_metrics 帧追增量）， + * 因此这里不再有任何 1s/3s/30s 的自动刷新。 + * + * 审计 tab 例外：它读的是审计链（写操作凭证，非实时监控数据），裁定只覆盖 + * 日志/指标/trace，因此仍按 5s 读取，并保留本地兜底。 + */ type Tab = "logs" | "trace" | "metrics" | "audit"; @@ -23,50 +29,9 @@ const TABS: ReadonlyArray<{ id: Tab; label: string }> = [ { id: "audit", label: "审计" }, ]; -/** Live backend log tail (GET /admin/observability/logs): level filter + refresh. */ -const LOG_LIMIT = 200; -const LOG_RELOAD_MS = 30_000; - -const LEVEL_FILTERS: ReadonlyArray<{ id: AdminLogLevel | ""; label: string }> = [ - { id: "", label: "全部" }, - { id: "debug", label: "debug" }, - { id: "info", label: "info" }, - { id: "warn", label: "warn" }, - { id: "error", label: "error" }, -]; - -const LEVEL_TEXT: Record = { - debug: "text-[var(--mu-text-lo)]", - info: "text-[var(--mu-accent-hi)]", - warn: "text-[var(--mu-warn)]", - error: "text-[var(--mu-err)]", -}; - -/** `time` is RFC3339 on current builds but the contract allows epoch ms too, so - * both are normalized; an unparseable value is shown verbatim rather than faked. */ -function fmtLogTime(t: string | number): string { - const d = new Date(t); - if (Number.isNaN(d.getTime())) return String(t); - const p = (n: number, w = 2) => String(n).padStart(w, "0"); - return `${p(d.getHours())}:${p(d.getMinutes())}:${p(d.getSeconds())}.${p(d.getMilliseconds(), 3)}`; -} - -function levelTextClass(level: string): string { - return LEVEL_TEXT[level.toLowerCase()] ?? "text-[var(--mu-text-lo)]"; -} - -const KIND_LABEL: Record = { - llm_call: "LLM", - tool_call: "工具", - orchestration: "编排", - state_transition: "状态", - subagent: "子代理", - memory: "记忆", - guardrail: "护栏", -}; - export function ObservabilityModule(): React.ReactNode { const [tab, setTab] = useState("logs"); + const stream = useObservabilityStream(); const deviceUpdates = useAppStore((s) => s.deviceUpdates); const audit = useAuditStore((s) => s.records); const clearAudit = useAuditStore((s) => s.clear); @@ -87,128 +52,26 @@ export function ObservabilityModule(): React.ReactNode { {t.label} ))} + + + - {tab === "logs" ? : null} - {tab === "trace" ? : null} - {tab === "metrics" ? : null} + {tab === "logs" ? ( +
+ + +
+ ) : null} + {tab === "trace" ? : null} + {tab === "metrics" ? : null} {tab === "audit" ? : null} ); } -/** 实时日志 tab: the backend's structured log tail first (authoritative, leveled - * and filtered server-side), then the WS event echo as a client-side complement. */ -function LogsTab({ lines }: { lines: string[] }): React.ReactNode { - return ( -
- - -
- ); -} - -/** Backend log tail: level filter chips over a monospace list, refreshed every - * 30s. Loading / empty / error are each surfaced honestly — no placeholder rows. */ -function BackendLogs(): React.ReactNode { - const [level, setLevel] = useState(""); - const logs = useAsyncData( - () => adminObservabilityLogs({ level: level || undefined, limit: LOG_LIMIT }), - [level] - ); - useAutoReload(logs.reload, LOG_RELOAD_MS); - const rows = logs.data ?? []; - - // "更新于 N 秒前" needs a ticking clock; the interval only forces a re-render, - // the timestamp itself is captured when fresh data lands. - const [updatedAt, setUpdatedAt] = useState(0); - useEffect(() => { - if (logs.data) setUpdatedAt(Date.now()); - }, [logs.data]); - const [, setTick] = useState(0); - useEffect(() => { - const id = window.setInterval(() => setTick((v) => v + 1), 1000); - return () => window.clearInterval(id); - }, []); - - return ( - -
-

实时日志（后端分级 tail）

- tail - - GET /api/v1/admin/observability/logs - - - {updatedAt > 0 ? `${Math.max(0, Math.round((Date.now() - updatedAt) / 1000))}s 前更新` : "尚未更新"} - {" · 30s 自动刷新"} - -
- -
- {LEVEL_FILTERS.map((f) => ( - - ))} - - 共 {rows.length} 条 · 上限 {LOG_LIMIT} - {level ? ` · 筛选 ${level} 及以上` : ""} - -
- - {logs.loading && logs.data === null ? ( - 加载日志… - ) : logs.error && logs.data === null ? ( - 日志端点暂不可用：{logs.error} - ) : rows.length === 0 ? ( - - 暂无日志{level ? `（level=${level}）` : ""}。网关写入结构化日志后，此处按 level 过滤实时展示。 - - ) : ( -
- {rows.map((line, i) => ( - - ))} -
- )} - {logs.error && logs.data !== null ? ( - 上次刷新失败（保留现有日志）：{logs.error} - ) : null} - - ); -} - -/** One monospace log line: time · level · source · msg (source never blank). */ -function LogRow({ line }: { line: AdminLogLine }): React.ReactNode { - return ( -
- - {fmtLogTime(line.time)} - - - {line.level || "info"} - - {line.source || "—"} - - {line.msg} - -
- ); -} - -/** Client-side WS event echo — the frames this browser received, kept alongside - * the server log because the two answer different questions (server truth vs. - * what this client actually saw). */ +/** 本端 WS 事件回声 —— 这个浏览器实际收到的帧，与服务端日志并列展示： + * 两者回答的是不同问题（服务端真相 vs 本端到底看到了什么）。 */ function LogsTail({ lines }: { lines: string[] }): React.ReactNode { return ( @@ -223,140 +86,13 @@ function LogsTail({ lines }: { lines: string[] }): React.ReactNode { {lines.join("\n")} )} - 这是当前浏览器的 WS 帧回显，不是服务端日志；服务端分级日志见上方「实时日志（后端分级 tail）」。 + + 这是当前浏览器的 WS 帧回显，不是服务端日志；可观测推送帧（日志/指标/trace）走订阅通道，不会污染这份回显。 + ); } -function TraceView(): React.ReactNode { - const trace = useAsyncData(() => adminObservabilityTrace(100), []); - useAutoReload(trace.reload, 3000); - const spans = trace.data ?? []; - - return ( - -
-

Tracing 视图

- 实时 - GET /api/v1/admin/observability/trace -
- {trace.loading ? ( - 加载 span 数据… - ) : trace.error ? ( - trace 后端暂不可用：{trace.error} - ) : spans.length === 0 ? ( - 暂无 span。接入 /ws/chat 发起会话后，LLM / 工具 / 编排 / 子代理等 span 会实时记录到环形缓冲。 - ) : ( - - )} - 数据源：gateway trace 环形缓冲（经 TraceSink 对接 runtime 的 tracer）。可在 TraceSink 注入前展示占位，注入后即真实 span。 - - ); -} - -function SpanList({ spans }: { spans: AdminTraceSpan[] }): React.ReactNode { - const [openId, setOpenId] = useState(null); - return ( -
- {spans.map((sp) => { - const open = openId === sp.id; - return ( - - ); - })} -
- ); -} - -function MetricsView(): React.ReactNode { - const metrics = useAsyncData(adminObservabilityMetrics, []); - useAutoReload(metrics.reload, 3000); - const m = metrics.data; - - const topPaths = Object.entries(m?.perPath ?? {}) - .sort((a, b) => b[1] - a[1]) - .slice(0, 8); - - return ( -
- - } label="请求 QPS" value={m ? m.qps.toFixed(1) : "—"} sub={m ? `总请求 ${m.totalRequests}` : "网关指标加载中"} tone="accent" /> - } label="LLM 调用次数" value={m ? m.llmCalls : "—"} sub="从 trace span 累计" tone="neutral" /> - } label="子代理事件" value={m ? m.subagentEvents : "—"} sub="从 trace span 累计" tone="ok" /> - } label="工时/Token" value="—" sub="成本后端暂不统计" tone="warn" /> - - - {metrics.error ? 指标后端暂不可用：{metrics.error} : null} - - {m && (topPaths.length > 0 || Object.keys(m.statusCounts).length > 0) ? ( - -

请求分布（真实网关计数）

- {topPaths.length > 0 ? ( -
- {topPaths.map(([path, count]) => ( -
- {path} - {count} -
-
-
-
- ))} -
- ) : null} - {Object.keys(m.statusCounts).length > 0 ? ( -
- {Object.entries(m.statusCounts).map(([code, count]) => ( - = 500 ? "err" : "warn"}> - {code} × {count} - - ))} - 平均 {m.avgLatencyMs.toFixed(1)}ms · 最大 {m.maxLatencyMs}ms -
- ) : null} - - ) : null} - - LLM 调用次数 / 子代理事件由实际观测到的 trace span 累计；Token / 成本后端暂不统计（不伪造指标）。指标时序图 / 导出需后端 M10+ 提供。 -
- ); -} - interface AuditViewProps { records: AuditRecord[]; onClear: () => void; @@ -401,7 +137,9 @@ function AuditView({ records: local, onClear }: AuditViewProps): React.ReactNode {r.who} {r.what} {new Date(r.when).toLocaleString()} - {r.result} + + {r.result} + ))} @@ -410,11 +148,3 @@ function AuditView({ records: local, onClear }: AuditViewProps): React.ReactNode ); } - -/** Formats a nanosecond duration into a human-readable label. */ -function fmtDuration(ns: number): string { - if (ns >= 1e9) return `${(ns / 1e9).toFixed(2)} s`; - if (ns >= 1e6) return `${(ns / 1e6).toFixed(2)} ms`; - if (ns >= 1e3) return `${(ns / 1e3).toFixed(1)} µs`; - return `${ns} ns`; -} diff --git a/frontend/web/src/admin/modules/observabilityStream.ts b/frontend/web/src/admin/modules/observabilityStream.ts new file mode 100644 index 0000000..62672e7 --- /dev/null +++ b/frontend/web/src/admin/modules/observabilityStream.ts @@ -0,0 +1,306 @@ +import { useEffect, useState } from "react"; +import { + adminObservabilityLogs, + adminObservabilityMetrics, + adminObservabilityTrace, +} from "../../lib/api"; +import { subscribeObservabilityFrames } from "../../lib/ws"; +import { useAsyncData } from "../../lib/useAsyncData"; +import { useAppStore } from "../../store/useAppStore"; +import type { + AdminLogLevel, + AdminLogLine, + AdminObservabilityMetrics, + AdminTraceSpan, + WsStatus, +} from "../../lib/types"; + +/** + * 可观测面板的数据流（裁定 05.4.3 + 16.4.3-WS：日志 / 指标 / trace 由 WS 帧 + * 推送替代轮询）。 + * + * 每个数据集两条来源： + * ① **引导读取**——面板挂载时各读一次 REST，拿首屏；服务端把这次读取同时 + * 记作订阅握手，此后新数据由 WS 帧推来（面板不再周期刷新）。 + * ② **推送帧**——`observability_log` / `observability_trace` / + * `observability_metrics`，载荷与 REST 同形，因此两条来源可以在同一份 + * 缓冲里合并去重。 + * + * 合并口径：按业务键去重、按时间倒序、只保留最近窗口（日志与 span 都是尾部 + * 视图，更旧的条目本就不在窗口内）。日志按内容折叠完全相同的行（同一毫秒内 + * 重复的同一行只留一条），span 按 id 折叠。 + */ + +/** 面板保留窗口：与日志 REST 尾部的默认上限一致（LOG_LIMIT=200）。 */ +export const LOG_BUFFER = 200; +/** span 保留窗口：面板发起引导读取时请的是 100 条，留一倍余量接推送。 */ +export const SPAN_BUFFER = 200; + +const FRAME_LOG = "observability_log"; +const FRAME_TRACE = "observability_trace"; +const FRAME_METRICS = "observability_metrics"; + +/** 一行的稳定键：REST 与推送会看到同一行，去重靠它。 */ +export interface LogRow { + key: string; + time: string | number; + level: string; + source: string; + msg: string; +} + +/** 把一条线上日志行折成可去重的表行。 */ +export function logRowOf(line: AdminLogLine): LogRow { + return { + key: `${line.time}|${line.level}|${line.source}|${line.msg}`, + time: line.time, + level: line.level, + source: line.source, + msg: line.msg, + }; +} + +/** 面板的级别筛选档位，`""` 表示全部。 */ +export const LOG_LEVEL_OPTIONS: ReadonlyArray<{ id: AdminLogLevel | ""; label: string }> = [ + { id: "", label: "全部" }, + { id: "debug", label: "debug" }, + { id: "info", label: "info" }, + { id: "warn", label: "warn" }, + { id: "error", label: "error" }, +]; + +const LEVEL_RANK: Record = { debug: 0, info: 1, warn: 2, error: 3 }; + +/** 级别筛选：推送帧不做分级过滤（帧面向所有订阅连接广播），因此在本地过滤。 */ +export function matchesLevel(level: string, min: AdminLogLevel | ""): boolean { + if (!min) return true; + const got = LEVEL_RANK[level.toLowerCase()]; + if (got === undefined) return true; + return got >= LEVEL_RANK[min]; +} + +/** 合并两条来源：按业务键去重（新数据优先）、按时间倒序、截到窗口上限。 */ +function mergeRows( + prev: readonly T[], + incoming: readonly T[], + keyOf: (row: T) => string, + timeOf: (row: T) => number, + cap: number, +): T[] { + const byKey = new Map(); + for (const row of incoming) byKey.set(keyOf(row), row); + for (const row of prev) { + const k = keyOf(row); + if (!byKey.has(k)) byKey.set(k, row); + } + const out = [...byKey.values()]; + out.sort((a, b) => timeOf(b) - timeOf(a)); + return out.slice(0, cap); +} + +/** 把时间字段解析成毫秒；不可解析的值一律排到最后（不伪造时间）。 */ +function timeMs(t: string | number): number { + const ms = typeof t === "number" ? t : Date.parse(t); + return Number.isNaN(ms) ? 0 : ms; +} + +function isRecord(v: unknown): v is Record { + return typeof v === "object" && v !== null && !Array.isArray(v); +} + +/** 从 observability_log 帧的 data 里取出条目（形状不对的条目被丢弃）。 */ +function readLogEntries(data: unknown): AdminLogLine[] { + if (!isRecord(data) || !Array.isArray(data.entries)) return []; + const out: AdminLogLine[] = []; + for (const item of data.entries) { + if (!isRecord(item) || typeof item.msg !== "string") continue; + out.push({ + time: typeof item.time === "string" || typeof item.time === "number" ? item.time : "", + level: typeof item.level === "string" ? item.level : "", + source: typeof item.source === "string" ? item.source : "", + msg: item.msg, + }); + } + return out; +} + +/** 从 observability_trace 帧的 data 里取出 span（形状不对的条目被丢弃）。 */ +function readSpans(data: unknown): AdminTraceSpan[] { + if (!isRecord(data) || !Array.isArray(data.spans)) return []; + const out: AdminTraceSpan[] = []; + for (const item of data.spans) { + if (!isRecord(item) || typeof item.id !== "string") continue; + out.push({ + id: item.id, + parentId: typeof item.parentId === "string" ? item.parentId : undefined, + name: typeof item.name === "string" ? item.name : "", + kind: typeof item.kind === "string" ? item.kind : "", + status: typeof item.status === "string" ? item.status : "", + startedAt: typeof item.startedAt === "string" ? item.startedAt : "", + endedAt: typeof item.endedAt === "string" ? item.endedAt : undefined, + durationNs: typeof item.durationNs === "number" ? item.durationNs : 0, + props: isRecord(item.props) ? item.props : undefined, + events: Array.isArray(item.events) ? (item.events as AdminTraceSpan["events"]) : undefined, + }); + } + return out; +} + +function readCounts(v: unknown): Record { + const out: Record = {}; + if (!isRecord(v)) return out; + for (const [k, n] of Object.entries(v)) { + if (typeof n === "number") out[k] = n; + } + return out; +} + +/** 从 observability_metrics 帧的 data 里取出指标（与 REST 的 AdminMetrics 同形）。 */ +function readMetrics(data: unknown): AdminObservabilityMetrics | null { + if (!isRecord(data) || typeof data.totalRequests !== "number") return null; + const num = (v: unknown): number => (typeof v === "number" ? v : 0); + return { + totalRequests: num(data.totalRequests), + qps: num(data.qps), + avgLatencyMs: num(data.avgLatencyMs), + maxLatencyMs: num(data.maxLatencyMs), + perMethod: readCounts(data.perMethod), + perPath: readCounts(data.perPath), + statusCounts: readCounts(data.statusCounts), + llmCalls: num(data.llmCalls), + subagentEvents: num(data.subagentEvents), + }; +} + +/** 面板消费的完整数据流与各数据集的状态。 */ +export interface ObservabilityStream { + logs: LogRow[]; + logsLoading: boolean; + logsError: string | null; + logsUpdatedAt: number; + logsPushed: boolean; + reloadLogs: () => void; + + spans: AdminTraceSpan[]; + spansLoading: boolean; + spansError: string | null; + spansUpdatedAt: number; + spansPushed: boolean; + reloadSpans: () => void; + + metrics: AdminObservabilityMetrics | null; + metricsLoading: boolean; + metricsError: string | null; + metricsUpdatedAt: number; + metricsPushed: boolean; + reloadMetrics: () => void; + + /** 实时通道状态：断线时面板只剩引导数据，并且会如实说明。 */ + wsStatus: WsStatus; +} + +/** 订阅可观测数据流：一次引导读取 + 持续推送（替代原先的 3–30s 轮询）。 */ +export function useObservabilityStream(): ObservabilityStream { + const [logs, setLogs] = useState([]); + const [logsUpdatedAt, setLogsUpdatedAt] = useState(0); + const [logsPushed, setLogsPushed] = useState(false); + const [spans, setSpans] = useState([]); + const [spansUpdatedAt, setSpansUpdatedAt] = useState(0); + const [spansPushed, setSpansPushed] = useState(false); + const [metrics, setMetrics] = useState(null); + const [metricsUpdatedAt, setMetricsUpdatedAt] = useState(0); + const [metricsPushed, setMetricsPushed] = useState(false); + + const logsRead = useAsyncData(() => adminObservabilityLogs({ limit: LOG_BUFFER }), []); + const spansRead = useAsyncData(() => adminObservabilityTrace(SPAN_BUFFER), []); + const metricsRead = useAsyncData(adminObservabilityMetrics, []); + + // 引导读取：首屏数据。这次读取在服务端同时是订阅握手。 + useEffect(() => { + const data = logsRead.data; + if (!data) return; + setLogs((prev) => mergeRows(prev, data.map(logRowOf), (r) => r.key, (r) => timeMs(r.time), LOG_BUFFER)); + setLogsUpdatedAt(Date.now()); + }, [logsRead.data]); + + useEffect(() => { + const data = spansRead.data; + if (!data) return; + setSpans((prev) => + mergeRows(prev, data, (s) => s.id, (s) => timeMs(s.startedAt), SPAN_BUFFER), + ); + setSpansUpdatedAt(Date.now()); + }, [spansRead.data]); + + useEffect(() => { + const data = metricsRead.data; + if (!data) return; + setMetrics(data); + setMetricsUpdatedAt(Date.now()); + }, [metricsRead.data]); + + // 推送帧：面板打开期间由服务端推来增量。 + useEffect( + () => + subscribeObservabilityFrames((frame) => { + if (frame.type === FRAME_LOG) { + const entries = readLogEntries(frame.data); + if (entries.length === 0) return; + setLogs((prev) => mergeRows(prev, entries.map(logRowOf), (r) => r.key, (r) => timeMs(r.time), LOG_BUFFER)); + setLogsUpdatedAt(Date.now()); + setLogsPushed(true); + return; + } + if (frame.type === FRAME_TRACE) { + const pushed = readSpans(frame.data); + if (pushed.length === 0) return; + setSpans((prev) => mergeRows(prev, pushed, (s) => s.id, (s) => timeMs(s.startedAt), SPAN_BUFFER)); + setSpansUpdatedAt(Date.now()); + setSpansPushed(true); + return; + } + if (frame.type === FRAME_METRICS) { + const snap = readMetrics(frame.data); + if (!snap) return; + setMetrics(snap); + setMetricsUpdatedAt(Date.now()); + setMetricsPushed(true); + } + }), + [], + ); + + const wsStatus = useAppStore((s) => s.wsStatus); + + return { + logs, + logsLoading: logsRead.loading, + logsError: logsRead.error, + logsUpdatedAt, + logsPushed, + reloadLogs: logsRead.reload, + spans, + spansLoading: spansRead.loading, + spansError: spansRead.error, + spansUpdatedAt, + spansPushed, + reloadSpans: spansRead.reload, + metrics, + metricsLoading: metricsRead.loading, + metricsError: metricsRead.error, + metricsUpdatedAt, + metricsPushed, + reloadMetrics: metricsRead.reload, + wsStatus, + }; +} + +/** 秒级重渲染：让「N 秒前更新」随空闲时间走动（只驱动重渲染，不取数）。 */ +export function useSecondsTick(): number { + const [tick, setTick] = useState(0); + useEffect(() => { + const id = window.setInterval(() => setTick((v) => v + 1), 1000); + return () => window.clearInterval(id); + }, []); + return tick; +} diff --git a/frontend/web/src/admin/modules/observabilityViews.tsx b/frontend/web/src/admin/modules/observabilityViews.tsx new file mode 100644 index 0000000..bb37a1d --- /dev/null +++ b/frontend/web/src/admin/modules/observabilityViews.tsx @@ -0,0 +1,363 @@ +import { useState } from "react"; +import { CardGrid, Notice, StatCard, VirtualList, VIRTUAL_THRESHOLD } from "../components"; +import { Button, Surface, Tag } from "../../components/primitives"; +import { Icon } from "../../components/Icon"; +import type { AdminLogLevel, AdminTraceSpan, WsStatus } from "../../lib/types"; +import { matchesLevel, LOG_LEVEL_OPTIONS, useSecondsTick, type LogRow, type ObservabilityStream } from "./observabilityStream"; +import { getStoredRole } from "../../lib/api"; + +/** + * 可观测面板的视图层（裁定 05.4.3 + 16.4.3-WS）：数据来自 + * `useObservabilityStream()`（引导读取 + WS 推送），本文件只管呈现： + * + * - 日志与 span 都是「条数多、还会被追帧」的长列表，统一用已落地的 + * `VirtualList` 承接（DOM 行数与总条数解耦）； + * - 加载 / 空 / 错误三态各自如实呈现，不用占位数据充数； + * - 实时通道断开时明说「只剩引导数据」，而不是让画面看起来还在刷新。 + */ + +const LEVEL_TEXT: Record = { + debug: "text-[var(--mu-text-lo)]", + info: "text-[var(--mu-accent-hi)]", + warn: "text-[var(--mu-warn)]", + error: "text-[var(--mu-err)]", +}; + +const KIND_LABEL: Record = { + llm_call: "LLM", + tool_call: "工具", + orchestration: "编排", + state_transition: "状态", + subagent: "子代理", + memory: "记忆", + guardrail: "护栏", +}; + +/** 日志行稠密（22px 上下），超过这个行数就开始只挂载视口内的行。 */ +const LOG_VIRTUAL_THRESHOLD = 20; + +/** `time` 当前构建是 RFC3339，契约也允许 epoch 毫秒，两者都归一化； + * 解析不出的值原样显示，不编造时间。 */ +function fmtLogTime(t: string | number): string { + const d = new Date(t); + if (Number.isNaN(d.getTime())) return String(t); + const p = (n: number, w = 2) => String(n).padStart(w, "0"); + return `${p(d.getHours())}:${p(d.getMinutes())}:${p(d.getSeconds())}.${p(d.getMilliseconds(), 3)}`; +} + +/** 实时通道状态徽标：owner/admin 的 /ws/chat 连接是推送的载体。 */ +export function WsBadge({ status }: { status: WsStatus }): React.ReactNode { + const map: Record = { + connected: { label: "推送中", tone: "ok" }, + connecting: { label: "连接中", tone: "neutral" }, + reconnecting: { label: "重连中", tone: "warn" }, + disconnected: { label: "已断开", tone: "err" }, + expired: { label: "登录过期", tone: "err" }, + idle: { label: "未连接", tone: "neutral" }, + }; + const cur = map[status]; + return 实时通道 · {cur.label}; +} + +/** 数据新鲜度：推送来的数据随空闲时间走动；只有引导读取时如实说明。 */ +function Freshness({ updatedAt, source }: { updatedAt: number; source: "push" | "read" }): React.ReactNode { + useSecondsTick(); + const age = updatedAt > 0 ? Math.max(0, Math.round((Date.now() - updatedAt) / 1000)) : -1; + return ( + + {source === "push" ? "WS 推送" : "引导读取"} + {age >= 0 ? ` · ${age}s 前更新` : " · 尚未更新"} + + ); +} + +/** 手动刷新按钮：实时通道断开或需要立刻对账时用（不是轮询）。 */ +function RefreshButton({ onClick, what }: { onClick: () => void; what: string }): React.ReactNode { + return ( + + ); +} + +/** 只读档说明：推送只发 owner/admin（design/10 §5），viewer 仍走 REST， + * 因此这里如实说明并指向「刷新」，而不是让画面看起来还在实时刷新。 */ +function ReadonlyScopeNote(): React.ReactNode { + const role = getStoredRole(); + if (role === null || role === "owner" || role === "admin") return null; + return ( + + 当前凭据是只读档（{role}）：实时推送只发 owner/admin，请用「刷新」按需取数。 + + ); +} + +/** 实时日志 tab：后端分级 tail（服务端推送）+ 本端 WS 回声（见 Observability.tsx）。 */ +export function LogsPanel({ stream }: { stream: ObservabilityStream }): React.ReactNode { + const [level, setLevel] = useState(""); + const rows: LogRow[] = stream.logs.filter((l) => matchesLevel(l.level, level)); + + return ( + +
+

实时日志（后端分级 tail）

+ + + WS observability_log · 引导 GET /api/v1/admin/observability/logs + + + + + +
+ +
+ {LOG_LEVEL_OPTIONS.map((f) => ( + + ))} + + 显示 {rows.length} / 保留 {stream.logs.length} 条 · 上限 200 + {level ? ` · 本地筛选 ${level} 及以上` : ""} + +
+ + {stream.logsLoading && stream.logs.length === 0 ? ( + 加载日志… + ) : stream.logsError && stream.logs.length === 0 ? ( + 日志端点暂不可用：{stream.logsError} + ) : rows.length === 0 ? ( + + 暂无日志{level ? `（本地筛选 level≥${level}）` : ""}。网关写入结构化日志后，此处按 level 过滤实时展示。 + + ) : ( + <> + {rows.length > LOG_VIRTUAL_THRESHOLD ? ( +
虚拟滚动 · 仅挂载视口内的行
+ ) : null} + l.key} + renderRow={(l) => } + rowHeight={22} + threshold={LOG_VIRTUAL_THRESHOLD} + maxHeight="min(60vh, 640px)" + /> + + )} + {stream.logsError && stream.logs.length > 0 ? ( + 上次刷新失败（保留现有日志）：{stream.logsError} + ) : null} + {stream.wsStatus !== "connected" ? ( + + 实时通道未连接：以上是最近一次引导读取的内容，连上后新日志会自动推来（也可点「刷新」立刻取一次）。 + + ) : null} + + + ); +} + +/** 一条等宽日志行：time · level · source · msg（source 不空白）。 */ +function LogRow({ line }: { line: LogRow }): React.ReactNode { + return ( +
+ + {fmtLogTime(line.time)} + + + {line.level || "info"} + + {line.source || "—"} + {line.msg} +
+ ); +} + +/** Tracing 视图：新 span 由推送追进列表，展开仍按 id 记忆。 */ +export function TracePanel({ stream }: { stream: ObservabilityStream }): React.ReactNode { + const [openId, setOpenId] = useState(null); + + return ( + +
+

Tracing 视图

+ + + + + +
+
+ WS observability_trace · 引导 GET /api/v1/admin/observability/trace + {stream.spans.length > VIRTUAL_THRESHOLD ? " · 虚拟滚动" : ""} +
+ {stream.spansLoading && stream.spans.length === 0 ? ( + 加载 span 数据… + ) : stream.spansError && stream.spans.length === 0 ? ( + trace 后端暂不可用：{stream.spansError} + ) : stream.spans.length === 0 ? ( + 暂无 span。接入 /ws/chat 发起会话后，LLM / 工具 / 编排 / 子代理等 span 会实时推来这里。 + ) : ( + sp.id} + renderRow={(sp) => ( + setOpenId(openId === sp.id ? null : sp.id)} /> + )} + rowClassName="pb-1.5" + rowHeight={72} + maxHeight="min(60vh, 640px)" + /> + )} + 数据源：gateway trace 环形缓冲（经 TraceSink 对接 runtime 的 tracer）；推送只发订阅之后新完成的 span，历史由引导读取提供。 + + + ); +} + +function SpanRow({ span, open, onToggle }: { span: AdminTraceSpan; open: boolean; onToggle: () => void }): React.ReactNode { + return ( + + ); +} + +/** 指标视图：快照由 observability_metrics 帧持续更新（有变化才发帧）。 */ +export function MetricsPanel({ stream }: { stream: ObservabilityStream }): React.ReactNode { + const m = stream.metrics; + const topPaths = Object.entries(m?.perPath ?? {}) + .sort((a, b) => b[1] - a[1]) + .slice(0, 8); + + return ( +
+
+ + WS observability_metrics + + + + +
+ + + } + label="请求 QPS" + value={m ? m.qps.toFixed(1) : "—"} + sub={m ? `总请求 ${m.totalRequests}` : "网关指标加载中"} + tone="accent" + /> + } + label="LLM 调用次数" + value={m ? m.llmCalls : "—"} + sub="从 trace span 累计" + tone="neutral" + /> + } + label="子代理事件" + value={m ? m.subagentEvents : "—"} + sub="从 trace span 累计" + tone="ok" + /> + } label="工时/Token" value="—" sub="成本后端暂不统计" tone="warn" /> + + + {stream.metricsError ? 指标后端暂不可用：{stream.metricsError} : null} + + {m && (topPaths.length > 0 || Object.keys(m.statusCounts).length > 0) ? ( + +

请求分布（真实网关计数）

+ {topPaths.length > 0 ? ( +
+ {topPaths.map(([path, count]) => ( +
+ {path} + {count} +
+
+
+
+ ))} +
+ ) : null} + {Object.keys(m.statusCounts).length > 0 ? ( +
+ {Object.entries(m.statusCounts).map(([code, count]) => ( + = 500 ? "err" : "warn"}> + {code} × {count} + + ))} + + 平均 {m.avgLatencyMs.toFixed(1)}ms · 最大 {m.maxLatencyMs}ms + +
+ ) : null} + + ) : null} + + + 指标快照只在变化时推送（空转不发帧），因此这里没有轮询也没有固定刷新节奏。LLM 调用次数 / 子代理事件由实际观测到的 trace span + 累计；Token / 成本后端暂不统计（不伪造指标）。 + + +
+ ); +} + +/** 把纳秒时长渲染成人读刻度。 */ +function fmtDuration(ns: number): string { + if (ns >= 1e9) return `${(ns / 1e9).toFixed(2)} s`; + if (ns >= 1e6) return `${(ns / 1e6).toFixed(2)} ms`; + if (ns >= 1e3) return `${(ns / 1e3).toFixed(1)} µs`; + return `${ns} ns`; +} diff --git a/frontend/web/src/lib/ws.ts b/frontend/web/src/lib/ws.ts index 081d22f..761de24 100644 --- a/frontend/web/src/lib/ws.ts +++ b/frontend/web/src/lib/ws.ts @@ -33,6 +33,40 @@ function wsUrl(): string { return `${proto}://${window.location.host}/ws/chat${qs}`; } +/** 可观测推送帧（裁定 05.4.3 + 16.4.3-WS）：日志 / 指标 / trace 由服务端 WS 帧 + * 推送替代轮询。它们不属任何会话、也不进应用 store，因此单独一条订阅通道： + * 只有打开着可观测面板的视图在听，其余时候帧被安静丢弃 —— 不会每秒把监控 + * 数据灌进设备事件日志。 */ +export interface ObservabilityFrame { + type: string; + data: unknown; +} + +const OBSERVABILITY_FRAME_TYPES = new Set([ + "observability_log", + "observability_trace", + "observability_metrics", +]); + +type ObservabilityListener = (frame: ObservabilityFrame) => void; + +const observabilityListeners = new Set(); + +/** 订阅可观测推送帧，返回退订函数。 */ +export function subscribeObservabilityFrames(listener: ObservabilityListener): () => void { + observabilityListeners.add(listener); + return () => { + observabilityListeners.delete(listener); + }; +} + +/** 把一帧可观测推送派发给所有监听者（没有监听者时即丢弃）。 */ +function notifyObservability(type: string, data: unknown): void { + for (const listener of observabilityListeners) { + listener({ type, data }); + } +} + /** Route one decoded downstream frame into the store. Exported (with a `frame` * shaped first argument) so the frame→store mapping can be exercised as a pure * logic test without a browser: this project has no jsdom/test runner, so the @@ -43,6 +77,14 @@ export function dispatchFrame(ev: { type: string; sessionId?: string; data: unkn const s = useAppStore.getState(); const r = toRecord(ev.data); + // 可观测推送帧（裁定 05.4.3 + 16.4.3-WS：日志/指标/trace 由 WS 推送替代轮询） + // 走订阅通道。它们不属任何会话、也不进应用 store：只有打开着可观测面板的 + // 视图在听，其余时候帧被安静丢弃，不会每秒往设备事件日志里灌监控数据。 + if (OBSERVABILITY_FRAME_TYPES.has(ev.type)) { + notifyObservability(ev.type, ev.data); + return; + } + if (handleStreamEvent(ev, r, s)) return; if (handleThinkingEvent(ev, r, s)) return; if (handleToolEvent(ev, r, s)) return; diff --git a/internal/services/gateway/observability.go b/internal/services/gateway/observability.go index 221ee8e..8d803db 100644 --- a/internal/services/gateway/observability.go +++ b/internal/services/gateway/observability.go @@ -26,6 +26,9 @@ type observability struct { metrics *MetricsCollector traceBuf *RingBuffer audit AuditStore + // watches 是日志/指标/trace 推送的订阅登记（裁决 05.4.3 + 16.4.3-WS）。 + // 没有订阅者时推送循环不运行，见 observability_push.go。 + watches obsWatches // logBuf 是管理日志尾部端点背后的 logger 日志环形缓冲。 // 组合根在网关存在后经 SetLogBuffer 注入它；在那之前 // 该端点返回 501，而不是空但虚假的尾部。 @@ -38,6 +41,7 @@ func newObservability(deps Deps) *observability { obs := &observability{ metrics: newMetricsCollector(), traceBuf: newRingBuffer(defaultTraceBufCap), + watches: newObsWatches(), } if deps.AuditStore != nil { obs.audit = deps.AuditStore @@ -78,8 +82,11 @@ func (s *Server) recordMetric(method, path string, status int, durMs int64) { s.obs.metrics.Record(method, path, status, durMs) } -// adminObservabilityMetrics 返回当前网关指标（管理读）。 +// adminObservabilityMetrics 返回当前网关指标（管理读）。读取同时充当 +// 实时推送的订阅握手（见 observability_push.go）：此后指标由 WS 帧推来， +// 面板不再轮询。 func (s *Server) adminObservabilityMetrics(w http.ResponseWriter, r *http.Request) { + s.watchObservability(r) if s.obs == nil { writeJSON(w, http.StatusOK, AdminMetrics{ PerMethod: map[string]int64{}, PerPath: map[string]int64{}, StatusCounts: map[string]int64{}, @@ -89,8 +96,10 @@ func (s *Server) adminObservabilityMetrics(w http.ResponseWriter, r *http.Reques writeJSON(w, http.StatusOK, s.obs.metrics.Snapshot(time.Now())) } -// adminObservabilityTrace 返回最近的已追踪 span（管理读）。 +// adminObservabilityTrace 返回最近的已追踪 span（管理读）。读取同时充当 +// 实时推送的订阅握手（见 observability_push.go）。 func (s *Server) adminObservabilityTrace(w http.ResponseWriter, r *http.Request) { + s.watchObservability(r) if s.obs == nil { writeJSON(w, http.StatusOK, []AdminTraceSpan{}) return diff --git a/internal/services/gateway/observability_logs.go b/internal/services/gateway/observability_logs.go index 94a898d..a6a095d 100644 --- a/internal/services/gateway/observability_logs.go +++ b/internal/services/gateway/observability_logs.go @@ -51,7 +51,12 @@ func (s *Server) logBuffer() *logger.RingBuffer { // 优先，可选按最低级别过滤（管理读）。它是面板日志 // 尾部背后的后端：记录来自 logger 自身的环形 // 缓冲，没有任何东西被重新渲染或编造。 +// +// 读取同时充当实时推送的订阅握手（裁决 05.4.3 + 16.4.3-WS，见 +// observability_push.go）：此后新日志由 observability_log 帧推来， +// 面板不再轮询，分级过滤改在本地进行（帧是广播的，无法按连接分级）。 func (s *Server) adminObservabilityLogs(w http.ResponseWriter, r *http.Request) { + s.watchObservability(r) minLevel, filtered, ok := parseLogLevel(r.URL.Query().Get("level")) if !ok { writeError(w, &apperr.AppError{ diff --git a/internal/services/gateway/observability_push.go b/internal/services/gateway/observability_push.go new file mode 100644 index 0000000..14fd8e9 --- /dev/null +++ b/internal/services/gateway/observability_push.go @@ -0,0 +1,284 @@ +package gateway + +import ( + "bytes" + "context" + "encoding/json" + "net/http" + "sync" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" +) + +// 本文件是裁决 05.4.3 + 16.4.3-WS 的落点：日志 / 指标 / trace 从「前端轮询 +// 管理 REST」改为「网关主动推 WS 帧」（design/10 §5 口径）。 +// +// 两条边界分开，谁都不越界： +// +// - **谁能收到帧**：仍由 ws_frame_scope.go 的角色门决定。这三个帧类型没有 +// 登记进 frameReachesAll，因此按 fail-closed 默认档只发 owner + admin； +// viewer 继续走既有 REST 只读档取数（design/11 §7.1）。 +// - **什么时候发帧**：由本文件的订阅登记决定。没有订阅者时推送循环不启动， +// 网关不为可观测数据做任何周期性工作 —— 这正是「推送替代轮询」相对 +// 「每个面板各自 1s 轮询」的收益所在。 +// +// 订阅从哪来：面板挂载时的首次读取（GET /api/v1/admin/observability/{logs, +// metrics,trace}）兼任订阅握手。上游订阅帧需要 ws_route.go 的上行分发开口， +// 那不在本次可动的文件面内；用「读过实时数据」作显式订阅动作语义等价，且天然 +// 只由真正打开过面板的凭据触发。订阅以凭据身份（sub + role + iat，等价于那次 +// 登录）为键，其连接全部断开后自动退订。 + +const ( + // wsObservabilityLog / Trace / Metrics 是三个下行帧类型。帧壳与既有下行 + // 帧一致（wsEvent{type,data}），不带 sessionId：它们不属任何会话。 + wsObservabilityLog = "observability_log" + wsObservabilityTrace = "observability_trace" + wsObservabilityMetrics = "observability_metrics" + + // 产生这三个帧的 bus 主题。ws_bridge.go 的 eventBridge 把主题映射成上面的 + // 帧类型，因此推送走的是与其它实时事件完全相同的 + // 「bus → relay → 角色门 → 连接」路径（不新开通道、不新开端口）。 + obsLogSubject = "observability.log" + obsTraceSubject = "observability.trace" + obsMetricsSubject = "observability.metrics" + + // observabilityPushInterval 是推送循环的节拍：每拍最多各发一帧 + // （日志增量 / 新 span / 变化过的指标快照），只发真的有新数据的那类。 + observabilityPushInterval = time.Second + // maxLogPushBatch / maxSpanPushBatch 界定单帧载荷大小。一拍的突发超过上限 + // 时只发最新的那些 —— 更旧的条目本就落在面板保留窗口之外，与既有 REST 尾部 + // （limit 上限）语义一致，因此不丢「看得见」的数据。 + maxLogPushBatch = 200 + maxSpanPushBatch = 100 +) + +// AdminLogPush 是 observability_log 帧的 data：一批日志条目，**最新在前** +// （与 GET /api/v1/admin/observability/logs 同序），面板据此前缀拼接。 +type AdminLogPush struct { + Entries []AdminLogEntry `json:"entries"` +} + +// AdminTracePush 是 observability_trace 帧的 data：一批新完成的 span， +// **最新在前**（与 GET /api/v1/admin/observability/trace 同序）。 +type AdminTracePush struct { + Spans []AdminTraceSpan `json:"spans"` +} + +// obsWatchKey 是「谁在订阅」的键：凭据身份，等价于那次登录。用 iat 而不用 +// token 串，避免把凭据本身留在内存里。 +type obsWatchKey struct { + sub string + role auth.Role + iat int64 +} + +// watchKeyOf 把连接/请求的凭据折叠成订阅键。 +func watchKeyOf(claims *auth.Claims) obsWatchKey { + return obsWatchKey{sub: claims.Sub, role: claims.Role, iat: claims.IssuedAt} +} + +// obsWatches 是订阅登记表。它与推送循环同生命周期：第一个订阅建立时启动 +// 循环，最后一个订阅被剪掉时循环自行退出，因此订阅表空着的时候没有任何 +// 常驻 goroutine，也没有任何周期性开销。 +type obsWatches struct { + mu sync.Mutex + set map[obsWatchKey]struct{} + running bool +} + +// newObsWatches 构建一个空的订阅登记表。 +func newObsWatches() obsWatches { + return obsWatches{set: make(map[obsWatchKey]struct{})} +} + +// add 登记一个订阅，必要时启动推送循环。没有 bus 就没有下行通道， +// 登记也不会产生任何推送，因此直接跳过（不启动循环）。 +func (w *obsWatches) add(s *Server, key obsWatchKey) { + if s == nil || s.bus == nil { + return + } + w.mu.Lock() + w.set[key] = struct{}{} + start := !w.running + if start { + w.running = true + } + w.mu.Unlock() + if start { + go w.loop(s) + } +} + +// loop 是推送循环：每拍先剪掉「凭据已不在线」的订阅，再按需发布三类数据。 +// 返回即代表循环已退出（下一个订阅会重新启动它）。 +// +// 锚点在循环启动时取定（见 newObsPushState）：只发订阅之后新产生的数据， +// 历史由面板自己那次引导读取提供。 +func (w *obsWatches) loop(s *Server) { + ticker := time.NewTicker(observabilityPushInterval) + defer ticker.Stop() + state := newObsPushState(s) + for range ticker.C { + if !w.prune(s) { + return + } + state.publish(s) + } +} + +// prune 剪掉没有在线 owner/admin 连接的订阅，并报告循环是否还要继续。 +// 「订阅表空了」与「退出」必须在同一把锁里判定：否则一个新订阅可能刚看到 +// running=true（以为循环还活着），而循环已经走向退出，推送就此静默停摆。 +func (w *obsWatches) prune(s *Server) bool { + w.mu.Lock() + defer w.mu.Unlock() + live := liveWatchKeys(s) + for k := range w.set { + if !live[k] { + delete(w.set, k) + } + } + if len(w.set) == 0 { + w.running = false + return false + } + return true +} + +// liveWatchKeys 返回当前在线 owner/admin 连接的凭据身份集合。 +func liveWatchKeys(s *Server) map[obsWatchKey]bool { + out := make(map[obsWatchKey]bool) + if s == nil || s.hub == nil { + return out + } + s.hub.mu.RLock() + defer s.hub.mu.RUnlock() + for c := range s.hub.conns { + if c.claims == nil || !ownerAdminClaims(c.claims) { + continue + } + out[watchKeyOf(c.claims)] = true + } + return out +} + +// watchObservability 把一次可观测读取登记为订阅。只有 owner/admin 会登记： +// viewer 仍走纯 REST 只读档，不被推送（design/10 §5）。读取本身照常返回引导 +// 数据，因此面板不需要轮询也能拿到首屏。 +func (s *Server) watchObservability(r *http.Request) { + if s == nil || s.obs == nil || r == nil { + return + } + claims, ok := auth.ClaimsFromContext(r.Context()) + if !ok || claims == nil || !ownerAdminClaims(claims) { + return + } + s.obs.watches.add(s, watchKeyOf(claims)) +} + +// obsPushState 是推送循环的游标：日志/span 已经发到哪里、上一次发出的指标 +// 指纹是什么。它只被循环自己的 goroutine 触碰，因此不需要锁。 +type obsPushState struct { + lastLogTime time.Time + lastSpanSeq uint64 + lastMetrics []byte +} + +// newObsPushState 取定本轮的推送锚点：日志取当前最新条目的时间、trace 取当前 +// 写入序号、指标取当前快照的指纹。三者都表示「截至此刻的数据面板已经通过 +// REST 引导读取看到过了」，因此循环不会把环形缓冲里的历史整段灌成巨型帧。 +func newObsPushState(s *Server) *obsPushState { + st := &obsPushState{} + if buf := s.logBuffer(); buf != nil { + if entries := buf.Snapshot(1); len(entries) == 1 { + st.lastLogTime = entries[0].Time + } + } + if s.obs != nil { + st.lastSpanSeq = s.obs.traceBuf.Seq() + if raw, err := json.Marshal(s.obs.metrics.Snapshot(time.Now())); err == nil { + st.lastMetrics = raw + } + } + return st +} + +// publish 发布本拍新增的数据：日志增量、新 span、变化过的指标快照。 +// 三类互相独立 —— 没有新数据的那类不发帧（design/10 §5「推送只在有数据时 +// 发一帧」），因此空转的服务端一帧都不发。 +func (st *obsPushState) publish(s *Server) { + st.publishLogs(s) + st.publishSpans(s) + st.publishMetrics(s) +} + +// publishLogs 把锚点之后的新日志条目发成一帧（最新在前）。日志环形缓冲未注入 +// （网关没接组合根的日志尾部）时没有数据源，直接跳过。 +func (st *obsPushState) publishLogs(s *Server) { + buf := s.logBuffer() + if buf == nil { + return + } + entries := buf.Snapshot(maxLogPushBatch) + fresh := make([]AdminLogEntry, 0, len(entries)) + for _, e := range entries { + if !e.Time.After(st.lastLogTime) { + break // 最新在前：一旦遇到不新的条目，后面的只会更旧 + } + fresh = append(fresh, adminLogOf(e)) + } + if len(fresh) == 0 { + return + } + st.lastLogTime = entries[0].Time + s.publishObservability(obsLogSubject, AdminLogPush{Entries: fresh}) +} + +// publishSpans 把锚点之后新完成的 span 发成一帧（最新在前）。 +func (st *obsPushState) publishSpans(s *Server) { + if s.obs == nil { + return + } + entries := s.obs.traceBuf.SnapshotSince(st.lastSpanSeq, maxSpanPushBatch) + if len(entries) == 0 { + return + } + spans := make([]AdminTraceSpan, 0, len(entries)) + for _, e := range entries { + spans = append(spans, adminSpanOf(e.rec)) + } + st.lastSpanSeq = entries[0].seq + s.publishObservability(obsTraceSubject, AdminTracePush{Spans: spans}) +} + +// publishMetrics 在当前指标快照与上次发出的不同时发一帧。用序列化结果做指纹： +// 快照的 map 副本会被编码，编码是确定性的（键排序），因此「没变」可以精确判定， +// 而 QPS 这类会随时间衰减的量自然在变化时照常推。 +func (st *obsPushState) publishMetrics(s *Server) { + if s.obs == nil { + return + } + snap := s.obs.metrics.Snapshot(time.Now()) + raw, err := json.Marshal(snap) + if err != nil { + return + } + if bytes.Equal(raw, st.lastMetrics) { + return + } + st.lastMetrics = raw + s.publishObservability(obsMetricsSubject, snap) +} + +// publishObservability 把一个可观测事件发上 bus：它会经 ws_bridge.go 的主题 +// 映射变成下行帧，并照常过一遍角色门（本包唯一的可见性决策点）。bus 缺席时 +// 是 no-op。 +func (s *Server) publishObservability(subject string, payload any) { + if s == nil || s.bus == nil { + return + } + if err := s.bus.Publish(context.Background(), subject, payload); err != nil { + s.log.Warn("publish observability frame", "subject", subject, "err", err) + } +} diff --git a/internal/services/gateway/observability_push_test.go b/internal/services/gateway/observability_push_test.go new file mode 100644 index 0000000..8c461c5 --- /dev/null +++ b/internal/services/gateway/observability_push_test.go @@ -0,0 +1,346 @@ +package gateway + +import ( + "context" + "encoding/json" + "net/http" + "testing" + "time" + + "github.com/gorilla/websocket" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" +) + +// 本文件锁住 05.4.3 + 16.4.3-WS（可观测面 WS 推送，替换轮询）： +// +// - 帧壳与载荷：observability_log / observability_trace / observability_metrics +// 三个帧类型经 ws_bridge 的主题映射成形，data 与既有 REST 端点同形； +// - 角色门：三者都不是 design/11 §7.1 的状态机信号，因此 fail-closed， +// 只发 owner + admin，viewer 收不到（新增一帧忘记分类也不会泄露）； +// - 推送时机：只有建立了订阅（面板读取过一次实时数据）才启动推送循环 —— +// 这是「推送替代轮询」不把成本转嫁给并发面板的那一半，也保证未订阅的 +// 连接不会凭空收到周期性帧。 +// +// 断言都走真路径：真 bus Publish（或真订阅 + 真推送循环）→ hub 的真订阅与中继 +// → 真 /ws/chat 连接（真 JWT）。 + +// obsPushReadBudget 是等待推送帧的总预算。推送循环的节拍是 +// observabilityPushInterval，一个节拍内应到齐全部三类帧。 +const obsPushReadBudget = 6 * time.Second + +// TestObservabilityFramesReachOwnerAndAdminOnly 在真实主题上发布三类可观测 +// 载荷：owner/admin 收到全部三帧且载荷键集与登记的一致，viewer 只收到白名单 +// 的哨兵帧。 +func TestObservabilityFramesReachOwnerAndAdminOnly(t *testing.T) { + t.Parallel() + for _, typ := range []string{wsObservabilityLog, wsObservabilityTrace, wsObservabilityMetrics} { + if frameReachesAll(typ) { + t.Fatalf("%s reached every connection; an observability frame carries content and must stay owner/admin only", typ) + } + } + + env := newEnv(t, nil, false) + conns := map[auth.Role]*websocket.Conn{} + for _, role := range []auth.Role{auth.RoleOwner, auth.RoleAdmin, auth.RoleViewer} { + conn := dialWS(t, env.ts.URL, env.token(t, role)) + defer conn.Close() + conns[role] = conn + } + waitConnCount(t, env.srv, len(conns)) + + logs := AdminLogPush{Entries: []AdminLogEntry{ + {Time: "2026-10-04T10:00:01Z", Level: "warn", Source: "gateway/observability", Msg: "推送测试"}, + }} + spans := AdminTracePush{Spans: []AdminTraceSpan{ + {ID: "span-push-1", Name: "llm.call", Kind: "llm_call", Status: "ok", + StartedAt: "2026-10-04T10:00:02Z", EndedAt: "2026-10-04T10:00:03Z", DurationNs: 1234}, + }} + metrics := AdminMetrics{ + TotalRequests: 7, QPS: 1, AvgLatencyMs: 5, MaxLatencyMs: 9, + PerMethod: map[string]int64{"GET": 7}, PerPath: map[string]int64{"/api/v1/x": 7}, + StatusCounts: map[string]int64{"200": 7}, LLMCalls: 2, SubagentEvents: 1, + } + for _, pub := range []struct { + subject string + payload any + }{ + {obsLogSubject, logs}, + {obsTraceSubject, spans}, + {obsMetricsSubject, metrics}, + } { + if err := env.bus.Publish(context.Background(), pub.subject, pub.payload); err != nil { + t.Fatalf("publish %s: %v", pub.subject, err) + } + } + // 哨兵：一个白名单状态信号收尾，使「viewer 什么都没收到」不会与 + // 「连接根本没工作」混淆（与既有帧作用域用例同一手法）。 + if err := env.bus.Publish(context.Background(), wsPrefix+scopeSentinel, map[string]any{"sessionId": "web"}); err != nil { + t.Fatalf("publish sentinel: %v", err) + } + + for _, role := range []auth.Role{auth.RoleOwner, auth.RoleAdmin} { + frames := drainFrames(t, conns[role]) + for _, typ := range []string{wsObservabilityLog, wsObservabilityTrace, wsObservabilityMetrics} { + assertFrameCount(t, string(role), frames, typ, 1) + if frame := findFrame(t, string(role), frames, typ); frame.SessionID != "" { + t.Fatalf("%s %s frame sessionId = %q, want none: observability frames are not session-scoped", + role, typ, frame.SessionID) + } + } + if len(frames) != 4 { + t.Fatalf("%s received %d frames, want the three observability frames + the sentinel: %v", + role, len(frames), frameTypesOf(frames)) + } + assertLogPushFrame(t, string(role), findFrame(t, string(role), frames, wsObservabilityLog), logs) + assertTracePushFrame(t, string(role), findFrame(t, string(role), frames, wsObservabilityTrace), spans) + assertMetricsPushFrame(t, string(role), findFrame(t, string(role), frames, wsObservabilityMetrics), metrics) + } + + viewerFrames := drainFrames(t, conns[auth.RoleViewer]) + if len(viewerFrames) != 1 || viewerFrames[0].Type != scopeSentinel { + t.Fatalf("viewer received %v, want only the whitelisted %q: observability frames are owner/admin only (design/11 §7.1)", + frameTypesOf(viewerFrames), scopeSentinel) + } +} + +// TestObservabilityPushStreamsToSubscribers 驱动真实的生产者路径：面板的引导 +// 读取建立订阅 → 推送循环取锚点 → 之后新产生的日志 / 指标 / span 各发一帧， +// owner 收得到；同一次推送里 viewer 一帧都收不到（角色门）。 +func TestObservabilityPushStreamsToSubscribers(t *testing.T) { + t.Parallel() + env := newEnv(t, nil, false) + ownerToken := env.token(t, auth.RoleOwner) + viewerToken := env.token(t, auth.RoleViewer) + owner := dialWS(t, env.ts.URL, ownerToken) + defer owner.Close() + viewer := dialWS(t, env.ts.URL, viewerToken) + defer viewer.Close() + waitConnCount(t, env.srv, 2) + + ring := logger.NewRingBuffer(64) + env.srv.SetLogBuffer(ring) + + // 面板挂载时的引导读取：既拿首屏，也建立订阅（watchObservability）。 + resp, _ := doReq(t, http.MethodGet, env.ts.URL+"/api/v1/admin/observability/logs?limit=5", nil, ownerToken) + if resp.StatusCode != http.StatusOK { + t.Fatalf("bootstrap log read = %d, want 200: the read must keep serving the guide data", resp.StatusCode) + } + + // 等循环起步（它在启动时取锚点：只发订阅之后的数据）。 + time.Sleep(300 * time.Millisecond) + + ring.Append(logger.Entry{Time: time.Now(), Level: "warn", Source: "gateway/push-test", Msg: "订阅后的新日志"}) + env.srv.recordMetric(http.MethodGet, "/api/v1/admin/push-test", http.StatusOK, 4) + env.srv.ObserveSpan(trace.SpanRecord{ + ID: "span-sub", Name: "llm.call", Kind: trace.SpanKindLLM, Status: trace.SpanStatusOK, + StartedAt: time.Now(), EndedAt: time.Now(), Duration: 3 * time.Millisecond, + }) + + wanted := []string{wsObservabilityLog, wsObservabilityTrace, wsObservabilityMetrics} + frames := readFramesUntil(t, owner, wanted, obsPushReadBudget) + for _, typ := range wanted { + assertFrameCount(t, "owner", frames, typ, 1) + } + logFrame := findFrame(t, "owner", frames, wsObservabilityLog) + entries := decodeFrameEntries(t, "owner", logFrame) + if len(entries) != 1 || entries[0].Msg != "订阅后的新日志" || entries[0].Source != "gateway/push-test" || entries[0].Level != "warn" { + t.Fatalf("pushed log entry = %+v, want the appended entry", entries) + } + var traceData struct { + Spans []AdminTraceSpan `json:"spans"` + } + unmarshalFrameData(t, "owner", findFrame(t, "owner", frames, wsObservabilityTrace), &traceData) + if len(traceData.Spans) != 1 || traceData.Spans[0].ID != "span-sub" || traceData.Spans[0].Kind != "llm_call" { + t.Fatalf("pushed spans = %+v, want the observed span", traceData.Spans) + } + var pushedMetrics AdminMetrics + unmarshalFrameData(t, "owner", findFrame(t, "owner", frames, wsObservabilityMetrics), &pushedMetrics) + if pushedMetrics.TotalRequests < 1 || pushedMetrics.LLMCalls < 1 { + t.Fatalf("pushed metrics = %+v, want the recorded request and the observed LLM span", pushedMetrics) + } + + // 同一次推送里 viewer 依旧一帧都拿不到：门在 hub，与订阅无关。 + viewerFrames := readFramesUntil(t, viewer, wanted, 400*time.Millisecond) + for _, typ := range wanted { + assertFrameCount(t, "viewer", viewerFrames, typ, 0) + } +} + +// TestObservabilityPushRequiresSubscription 是推送时机的反证：连接上没有任何 +// 人读取过实时数据 → 没有订阅 → 推送循环根本不启动，即使数据在连接建立后 +// 立刻变化，也不会凭空出现周期性帧（这正是既有「精确帧数」用例得以保持的 +// 前提，也是推送替代轮询不把成本转嫁给所有连接的那一半）。 +func TestObservabilityPushRequiresSubscription(t *testing.T) { + t.Parallel() + env := newEnv(t, nil, false) + token := env.token(t, auth.RoleOwner) + conn := dialWS(t, env.ts.URL, token) + defer conn.Close() + waitConnCount(t, env.srv, 1) + + ring := logger.NewRingBuffer(16) + env.srv.SetLogBuffer(ring) + ring.Append(logger.Entry{Time: time.Now(), Level: "info", Source: "gateway/no-sub", Msg: "无人订阅"}) + env.srv.recordMetric(http.MethodGet, "/api/v1/admin/no-sub", http.StatusOK, 1) + env.srv.ObserveSpan(trace.SpanRecord{ + ID: "span-no-sub", Name: "llm.call", Kind: trace.SpanKindLLM, Status: trace.SpanStatusOK, + StartedAt: time.Now(), EndedAt: time.Now(), Duration: time.Millisecond, + }) + + // 覆盖一个完整节拍：未订阅时这里必须一帧都没有。 + time.Sleep(observabilityPushInterval + 200*time.Millisecond) + frames := drainFrames(t, conn) + if len(frames) != 0 { + t.Fatalf("an unsubscribed connection received %v; the push loop must not run without a subscription", frameTypesOf(frames)) + } +} + +// assertLogPushFrame 断言一帧 observability_log 的 data 形状（键集恰好 +// {"entries"}）与条目内容。 +func assertLogPushFrame(t *testing.T, role string, frame subagentFrame, want AdminLogPush) { + t.Helper() + assertFrameDataKeys(t, role, frame, []string{"entries"}) + entries := decodeFrameEntries(t, role, frame) + if len(entries) != len(want.Entries) { + t.Fatalf("%s log entries = %d, want %d", role, len(entries), len(want.Entries)) + } + for i, e := range entries { + if e != want.Entries[i] { + t.Fatalf("%s log entry[%d] = %+v, want %+v", role, i, e, want.Entries[i]) + } + } +} + +// decodeFrameEntries 解出 observability_log 帧的条目，并锁住每条目的键集。 +func decodeFrameEntries(t *testing.T, role string, frame subagentFrame) []AdminLogEntry { + t.Helper() + var data struct { + Entries []json.RawMessage `json:"entries"` + } + unmarshalFrameData(t, role, frame, &data) + out := make([]AdminLogEntry, 0, len(data.Entries)) + for _, raw := range data.Entries { + assertRawObjectKeys(t, role+" log entry", raw, []string{"time", "level", "source", "msg"}) + var e AdminLogEntry + if err := json.Unmarshal(raw, &e); err != nil { + t.Fatalf("%s log entry %s: %v", role, raw, err) + } + out = append(out, e) + } + return out +} + +// assertTracePushFrame 断言一帧 observability_trace 的 data 形状（键集恰好 +// {"spans"}）与 span 内容。 +func assertTracePushFrame(t *testing.T, role string, frame subagentFrame, want AdminTracePush) { + t.Helper() + var data struct { + Spans []json.RawMessage `json:"spans"` + } + assertFrameDataKeys(t, role, frame, []string{"spans"}) + unmarshalFrameData(t, role, frame, &data) + if len(data.Spans) != len(want.Spans) { + t.Fatalf("%s spans = %d, want %d", role, len(data.Spans), len(want.Spans)) + } + assertRawObjectKeys(t, role+" span", data.Spans[0], + []string{"id", "name", "kind", "status", "startedAt", "endedAt", "durationNs"}) + var sp AdminTraceSpan + if err := json.Unmarshal(data.Spans[0], &sp); err != nil { + t.Fatalf("%s span %s: %v", role, data.Spans[0], err) + } + if sp.ID != want.Spans[0].ID || sp.Name != want.Spans[0].Name || sp.Kind != want.Spans[0].Kind || + sp.Status != want.Spans[0].Status || sp.DurationNs != want.Spans[0].DurationNs { + t.Fatalf("%s span = %+v, want %+v", role, sp, want.Spans[0]) + } +} + +// assertMetricsPushFrame 断言一帧 observability_metrics 的 data 就是 REST 端点 +// 那份 AdminMetrics（键集逐个锁定），避免推送与轮询两条路给出不同形状。 +func assertMetricsPushFrame(t *testing.T, role string, frame subagentFrame, want AdminMetrics) { + t.Helper() + wantKeys := []string{ + "totalRequests", "qps", "avgLatencyMs", "maxLatencyMs", "perMethod", + "perPath", "statusCounts", "llmCalls", "subagentEvents", + } + assertFrameDataKeys(t, role, frame, wantKeys) + var got AdminMetrics + unmarshalFrameData(t, role, frame, &got) + if got.TotalRequests != want.TotalRequests || got.QPS != want.QPS || + got.AvgLatencyMs != want.AvgLatencyMs || got.MaxLatencyMs != want.MaxLatencyMs || + got.LLMCalls != want.LLMCalls || got.SubagentEvents != want.SubagentEvents || + got.PerPath["/api/v1/x"] != 7 || got.StatusCounts["200"] != 7 { + t.Fatalf("%s metrics = %+v, want %+v (REST shape)", role, got, want) + } +} + +// assertFrameDataKeys 断言帧的 data 是键集恰好为 want 的 JSON 对象。 +func assertFrameDataKeys(t *testing.T, role string, frame subagentFrame, want []string) { + t.Helper() + assertRawObjectKeys(t, role+" frame data", frame.Data, want) +} + +// assertRawObjectKeys 断言 raw 是键集恰好为 want 的 JSON 对象。 +func assertRawObjectKeys(t *testing.T, what string, raw json.RawMessage, want []string) { + t.Helper() + var keys map[string]json.RawMessage + if err := json.Unmarshal(raw, &keys); err != nil { + t.Fatalf("%s is not a JSON object: %v (%s)", what, err, raw) + } + if len(keys) != len(want) { + t.Fatalf("%s keys = %v, want exactly %v (api-client 登记的形状)", what, keys, want) + } + for _, k := range want { + if _, ok := keys[k]; !ok { + t.Fatalf("%s lacks key %q: %v", what, k, keys) + } + } +} + +// unmarshalFrameData 把帧的 data 解进 out。 +func unmarshalFrameData(t *testing.T, role string, frame subagentFrame, out any) { + t.Helper() + if err := json.Unmarshal(frame.Data, out); err != nil { + t.Fatalf("%s frame data %s: %v", role, frame.Data, err) + } +} + +// readFramesUntil 读连接直到 wanted 里的每个帧类型都至少出现一次，或预算耗尽。 +// 它返回已收到的帧（超时也返回），使「帧没来」的失败信息里能看到到底来了什么。 +func readFramesUntil(t *testing.T, conn *websocket.Conn, wanted []string, budget time.Duration) []subagentFrame { + t.Helper() + deadline := time.Now().Add(budget) + out := make([]subagentFrame, 0, len(wanted)) + for !hasAllFrameTypes(out, wanted) && time.Now().Before(deadline) { + _ = conn.SetReadDeadline(deadline) + _, msg, err := conn.ReadMessage() + if err != nil { + return out + } + var frame subagentFrame + if uerr := json.Unmarshal(msg, &frame); uerr != nil { + t.Fatalf("decode frame %s: %v", msg, uerr) + } + out = append(out, frame) + } + return out +} + +// hasAllFrameTypes 报告 frames 是否已经覆盖 wanted 里的每个类型。 +func hasAllFrameTypes(frames []subagentFrame, wanted []string) bool { + for _, typ := range wanted { + found := false + for _, f := range frames { + if f.Type == typ { + found = true + break + } + } + if !found { + return false + } + } + return true +} diff --git a/internal/services/gateway/observability_trace.go b/internal/services/gateway/observability_trace.go index a0d70a8..4395513 100644 --- a/internal/services/gateway/observability_trace.go +++ b/internal/services/gateway/observability_trace.go @@ -31,10 +31,20 @@ type AdminTraceSpan struct { // 最近的工作而不会无界增长。 type RingBuffer struct { mu sync.RWMutex - buf []trace.SpanRecord + buf []traceEntry next int full bool cap int + // seq 是最后写入 span 的序号（单调递增，从 1 起）。 + // 增量推送（observability_trace 帧）用它做游标，见 SnapshotSince。 + seq uint64 +} + +// traceEntry 是环形缓冲内的一个存储格。序号随写入单调递增，因此 +// 下游可以「只取比游标新的」而不是靠时间戳去猜。 +type traceEntry struct { + seq uint64 + rec trace.SpanRecord } // newRingBuffer 构建至多容纳 cap 个 span 的环形缓冲（cap<=0 时用默认值）。 @@ -42,20 +52,28 @@ func newRingBuffer(cap int) *RingBuffer { if cap <= 0 { cap = defaultTraceBufCap } - return &RingBuffer{buf: make([]trace.SpanRecord, cap), cap: cap} + return &RingBuffer{buf: make([]traceEntry, cap), cap: cap} } // Append 存储 rec，缓冲满时逐出最旧的。 func (r *RingBuffer) Append(rec trace.SpanRecord) { r.mu.Lock() defer r.mu.Unlock() - r.buf[r.next] = rec + r.seq++ + r.buf[r.next] = traceEntry{seq: r.seq, rec: rec} r.next = (r.next + 1) % r.cap if r.next == 0 { r.full = true } } +// Seq 返回最后写入 span 的序号（还没写过任何 span 时为 0）。 +func (r *RingBuffer) Seq() uint64 { + r.mu.RLock() + defer r.mu.RUnlock() + return r.seq +} + // Snapshot 返回至多 n 条最近 span，最新优先。非正的 n // 选用 defaultObsLimit；n 被夹取到 maxObsLimit。 func (r *RingBuffer) Snapshot(n int) []trace.SpanRecord { @@ -77,7 +95,32 @@ func (r *RingBuffer) Snapshot(n int) []trace.SpanRecord { out := make([]trace.SpanRecord, 0, count) for i := 0; i < count; i++ { idx := (r.next - 1 - i + r.cap*2) % r.cap - out = append(out, r.buf[idx]) + out = append(out, r.buf[idx].rec) + } + return out +} + +// SnapshotSince 返回序号大于 since 的至多 n 条 span，最新优先（与 Snapshot +// 同序），供增量推送使用。n 非正时只用 defaultObsLimit 兜底；调用方按 +// 「帧载荷有上限」的约束传入自己的上限。 +func (r *RingBuffer) SnapshotSince(since uint64, n int) []traceEntry { + if n <= 0 { + n = defaultObsLimit + } + r.mu.RLock() + defer r.mu.RUnlock() + count := r.next + if r.full { + count = r.cap + } + out := make([]traceEntry, 0, min(count, n)) + for i := 0; i < count && len(out) < n; i++ { + idx := (r.next - 1 - i + r.cap*2) % r.cap + e := r.buf[idx] + if e.seq <= since { + break + } + out = append(out, e) } return out } diff --git a/internal/services/gateway/ws_bridge.go b/internal/services/gateway/ws_bridge.go index bafeada..abcd365 100644 --- a/internal/services/gateway/ws_bridge.go +++ b/internal/services/gateway/ws_bridge.go @@ -26,6 +26,12 @@ var eventBridge = map[string]string{ "automation.result": "automation_result", "reminder.trigger": "reminder", orchestration.SubagentResultSubject: wsSubagentResult, + // 可观测面（裁决 05.4.3 + 16.4.3-WS）：日志/指标/trace 的推送帧。载荷形状 + // 与既有 REST 端点同形（见 observability_push.go），因此面板换成消费帧后 + // 不需要第二套解码。 + obsLogSubject: wsObservabilityLog, + obsTraceSubject: wsObservabilityTrace, + obsMetricsSubject: wsObservabilityMetrics, } // bridgePatterns 是 hub 在 ws.> 之外额外订阅的 bus 模式，用于 @@ -36,6 +42,9 @@ var bridgePatterns = []string{ "automation.>", "reminder.>", "cyrene.agent.*.event", + // 可观测推送（observability.log/trace/metrics）：日志/指标/trace 的增量 + // 帧都走这个主题族，见 observability_push.go。 + "observability.>", // 后台子代理批次结果：它既不匹配 ws.>，也不是 cyrene.agent 的 4 段广播形态， // 所以必须在这里登记订阅，否则后台结果永远上不了前台（只能看到进度帧）。 orchestration.SubagentResultSubject,