diff --git a/Docs/api-client.md b/Docs/api-client.md index 39e5b7a..878dff0 100644 --- a/Docs/api-client.md +++ b/Docs/api-client.md @@ -45,14 +45,22 @@ | 12 | `/api/v1/admin/rag` | GET | ✅ | 知识库文档列表（分块/嵌入状态）；带 `?query=` 时并入检索结果 | | 13 | `/api/v1/admin/rag/query` | POST | ✅ | 知识库检索（真实检索；配置向量增强走语义路径） | | 14 | `/api/v1/admin/rag/delete` | POST | 🕐(写) | 删除知识库文档；写操作 M10，当前返回 `501 not_implemented` | +| 15 | `/api/v1/admin/observability/metrics` | GET | ✅ | 网关请求指标（总数/QPS/延迟/方法·路径·状态计数/LLM·子代理计数） | +| 16 | `/api/v1/admin/observability/trace` | GET | ✅ | 最近 trace span（环形缓冲，默认 512 条） | +| 17 | `/api/v1/admin/observability/audit` | GET | ✅ | 审计记录（who/what/when/result，SQLite 持久化） | -> **M10 前端契约补充（管理面板）**：`GET /api/v1/admin/users` · `GET /api/v1/admin/llm` · `GET /api/v1/admin/automations` · `GET /api/v1/admin/backthink` · `GET /api/v1/admin/rag` · `POST /api/v1/admin/rag/query` 已实现，返回真实数据；`GET /api/v1/admin/logs` · `GET /api/v1/admin/trace` · `GET /api/v1/admin/audit` · `GET /api/v1/admin/system` 仍🕐未实现（前端按契约封装、后端 404 时显示"待实现"壳，不报错）。写操作（策略/档案/密钥/系统设置/用户与自动化创建、知识库上传/删除）经前端审计登记，待后端 M10+ 落地真实持久化与热重载。 +> **M10 前端契约补充（管理面板）**：`GET /api/v1/admin/users` · `GET /api/v1/admin/llm` · `GET /api/v1/admin/automations` · `GET /api/v1/admin/backthink` · `GET /api/v1/admin/rag` · `POST /api/v1/admin/rag/query` · `GET /api/v1/admin/observability/metrics` · `GET /api/v1/admin/observability/trace` · `GET /api/v1/admin/observability/audit` 已实现，返回真实数据；`GET /api/v1/admin/logs` · `GET /api/v1/admin/system` 仍🕐未实现（前端按契约封装、后端 404 时显示"待实现"壳，不报错）。写操作（策略/档案/密钥/系统设置/用户与自动化创建、知识库上传/删除）经前端审计登记，待后端 M10+ 落地真实持久化与热重载。 **知识库 RAG 契约**： - `GET /admin/rag` → `{"data":{"documents":[{id,name,desc?,chunks,embedded}],"hits"?:[{docId,docName,chapter?,text,score,semantic}]}}`。`hits` 仅在携带 `?query=` 时出现；`embedded` 反映部署是否配置向量增强（未配置为待嵌入）。 - `POST /admin/rag/query` body `{"query":"..."}` → `{"data":{"query":"...","hits":[{docId,docName,chapter?,text,score,semantic}]}}`。检索走项目索引关键词+行上下文；配置 pgvector/向量增强时并入语义/向量命中（`semantic:true`）。 - `POST /admin/rag/delete` body `{"id":"..."}` → 写操作 M10，返回 `501` `{"error":{"code":"not_implemented",...}}`。 +**可观测性契约**（`GET /api/v1/admin/observability/*`，均需 owner/admin，`?limit=N` 缺省 50、上限 500）： +- `GET /metrics` → `{"data":{"totalRequests","qps","avgLatencyMs","maxLatencyMs","perMethod{}","perPath{}","statusCounts{}","llmCalls","subagentEvents"}}`。`qps` 为近 1 秒窗口计数；`llmCalls`/`subagentEvents` 由实际观测到的 trace span 按 `kind` 累计（Tracer 未注入时为 0）。**Token/成本后端不统计、不返回**（不伪造指标）。 +- `GET /trace?limit=100` → `{"data":[{id,parentId?,name,kind,status,startedAt,endedAt?,durationNs,props?,events?]}`。`kind` ∈ `llm_call|tool_call|orchestration|state_transition|subagent|memory|guardrail`；`durationNs` 为纳秒；`events` 为 `[{time,type,props?}]`。 +- `GET /audit?limit=100` → `{"data":[{id,who,what,when,result}]}`。`id` 为整数自增；`when` 为 RFC3339；来源为 SQLite `audit_logs` 表（未注入 AuditStore 时退化为进程内存）。 + ## 3. WebSocket `/ws/chat`（实时） - **握手**: `ws://:8080/ws/chat?token=` 或 `Authorization: Bearer`；未认证在握手前回 `401` JSON。 @@ -78,6 +86,7 @@ - **离线**: PWA / 原生端建议缓存 `sessions`/`personas` 列表。 ## 5. 变更记录 +- `v0.6`（可观测性）：新增 `GET /api/v1/admin/observability/{metrics,trace,audit}`（owner/admin）——metrics 返回网关实时请求计数/QPS/延迟/方法·路径·状态分布 + LLM/子代理 span 计数；trace 返回 gateway 环形缓冲里的最近 span（经 TraceSink 对接 runtime tracer，映射为 `AdminTraceSpan`）；audit 返回持久化审计（SQLite `audit_logs`，未注入时退化为内存）。`llmCalls`/`subagentEvents` 由实际观测 span 累计；Token/成本后端不统计、不返回（不伪造指标）。 - `v0.5`（批次B·视觉真推理 + 语音端）：`POST /api/v1/chat` 增 `image` 字段（base64 data: URI 或 URL），入 loop 视觉路由（vision/OCR 工具或主模型多模态，design/06 §0「先看图再对话」）；WS 上行 `voice_input` 接 STT→聊天→TTS 管道，下行新增 `voice_output`（定向回请求连接，含合成音频 + 识别文本 + 回复文本）。 - `v0.4`（结构化流）：WS 下行新增 `message.*` / `thinking.*` / `tool_call.*` / `tool_execution.update` 结构化流事件（event shell 补 `turnId`/`seq`，`data` 类型化），对齐 OpenClaw AssistantMessageEvent；保留 `stream_chunk`/`stream_end` 向后兼容。 - `v0.3`（M10）：管理面板 `/admin` 落地——按 §2 契约封装 admin 只读端点（personas/sessions/memories/plugins/channels/iot）+ owner-only `plugins/policy` 写；其余 M10 端点以契约路径封装、后端未实现时前端显式"待实现"壳。 diff --git a/cmd/cyrene/wire.go b/cmd/cyrene/wire.go index 5b9f429..f3155ef 100644 --- a/cmd/cyrene/wire.go +++ b/cmd/cyrene/wire.go @@ -451,6 +451,11 @@ func (a *app) wireFrontend() error { VectorOn: a.vindex != nil, Log: a.log, }) + auditStore, aerr := gateway.NewSQLiteAuditStore(context.Background(), a.db, a.log) + if aerr != nil { + a.log.Warn("audit store unavailable, falling back to in-memory", "error", aerr) + auditStore = nil + } gw, err := gateway.New(a.cfg, gateway.Deps{ Log: logger.WithSource(a.log, "gateway"), Auth: jwt, @@ -467,6 +472,7 @@ func (a *app) wireFrontend() error { Users: apiKeys, Backthink: a.backthink, RAG: ragStore, + AuditStore: auditStore, Streaming: true, Frontend: resolveFrontendFS(a.log), Webhooks: a.channelBridge.WebhookHandlers(), diff --git a/frontend/web/src/admin/modules/Observability.tsx b/frontend/web/src/admin/modules/Observability.tsx index f136a2e..c8ebdea 100644 --- a/frontend/web/src/admin/modules/Observability.tsx +++ b/frontend/web/src/admin/modules/Observability.tsx @@ -4,6 +4,10 @@ import { useAuditStore } from "../audit"; import { CardGrid, Notice, PageHeader, StatCard } 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, adminObservabilityMetrics, adminObservabilityTrace } from "../../lib/api"; +import type { AdminTraceSpan, AuditRecord } from "../../lib/types"; type Tab = "logs" | "trace" | "metrics" | "audit"; @@ -14,6 +18,16 @@ const TABS: ReadonlyArray<{ id: Tab; label: string }> = [ { id: "audit", label: "审计" }, ]; +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 deviceUpdates = useAppStore((s) => s.deviceUpdates); @@ -66,45 +80,152 @@ function LogsTail({ lines }: { lines: string[] }): React.ReactNode { } function TraceView(): React.ReactNode { + const trace = useAsyncData(() => adminObservabilityTrace(100), []); + useAutoReload(trace.reload, 3000); + const spans = trace.data ?? []; + return ( -

Tracing 视图

-
- {["LLM 调用链", "工具链", "编排链"].map((x) => ( -
-
{x}
- {x} 占位 —— 后端 M10+ 提供 span/trace 数据 -
- ))} +
+

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="—" sub="网关指标占位" tone="accent" /> - } label="LLM 调用次数" value="—" sub="用量/费用占位" tone="neutral" /> - } label="子代理事件" value="—" sub="编排指标占位" tone="ok" /> - } label="工时/Token" value="—" sub="成本占位" tone="warn" /> + } 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" /> - 指标时序图 / 导出需后端 M10+ 提供。 + + {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: ReturnType["records"]; + records: AuditRecord[]; onClear: () => void; } -function AuditView({ records, onClear }: AuditViewProps): React.ReactNode { +function AuditView({ records: local, onClear }: AuditViewProps): React.ReactNode { + const audit = useAsyncData(() => adminObservabilityAudit(100), []); + useAutoReload(audit.reload, 5000); + // Backend is the authoritative source; the local store is a fallback while the + // endpoint is unavailable. + const records = audit.data ?? local; + const source = audit.data ? "后端" : "本地（后端未接入）"; + return (

审计日志（who / what / when / result）

+ {source} 管理 + 敏感操作全程落审计
+ {audit.error && !audit.data ? 审计端点暂不可用，回退本地记录：{audit.error} : null} {records.length === 0 ? ( 暂无审计记录。本面板的管理/敏感操作（策略写入、主人档案编辑、API key 签发等）都会在此登记。 ) : ( @@ -139,3 +261,11 @@ function AuditView({ records, 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/lib/api.ts b/frontend/web/src/lib/api.ts index d78133c..f34cdd3 100644 --- a/frontend/web/src/lib/api.ts +++ b/frontend/web/src/lib/api.ts @@ -3,11 +3,14 @@ import type { AdminChannelRecord, AdminIoTDevice, AdminLlmConfig, + AdminObservabilityMetrics, AdminPluginRecord, AdminRagHit, AdminRagResponse, + AdminTraceSpan, AdminUserRecord, ApiErrorShape, + AuditRecord, ChatSyncResult, LoginResult, MemoryRecord, @@ -269,3 +272,21 @@ export function adminDeleteRag(id: string): Promise<{ id: string }> { body: JSON.stringify({ id }), }); } + +/** GET /api/v1/admin/observability/metrics — gateway request metrics. */ +export function adminObservabilityMetrics(): Promise { + return request("/admin/observability/metrics"); +} +/** GET /api/v1/admin/observability/trace?limit=N — most recent traced spans. */ +export function adminObservabilityTrace(limit?: number): Promise { + return request(`/admin/observability/trace${limitQuery(limit)}`); +} +/** GET /api/v1/admin/observability/audit?limit=N — most recent audit entries. */ +export function adminObservabilityAudit(limit?: number): Promise { + return request(`/admin/observability/audit${limitQuery(limit)}`); +} + +/** Builds an optional ?limit=N query suffix (empty when no positive limit). */ +function limitQuery(limit?: number): string { + return limit && limit > 0 ? `?limit=${limit}` : ""; +} diff --git a/frontend/web/src/lib/types.ts b/frontend/web/src/lib/types.ts index c68b8e7..74baabf 100644 --- a/frontend/web/src/lib/types.ts +++ b/frontend/web/src/lib/types.ts @@ -274,15 +274,52 @@ export interface AdminRagResponse { hits?: AdminRagHit[]; } -/** One audit entry: who / what / when / result. */ +/** One audit entry: who / what / when / result. `id` is a string for the local + * (client) trail but a number for backend-persisted entries. */ export interface AuditRecord { - id: string; + id: string | number; who: string; what: string; when: string; result: string; } +/** Gateway request metrics (GET /api/v1/admin/observability/metrics). Token/cost + * is intentionally absent: the gateway does not track spend, and a fabricated + * number would violate the "不伪造指标" red line. */ +export interface AdminObservabilityMetrics { + totalRequests: number; + qps: number; + avgLatencyMs: number; + maxLatencyMs: number; + perMethod: Record; + perPath: Record; + statusCounts: Record; + llmCalls: number; + subagentEvents: number; +} + +/** One traced span (GET /api/v1/admin/observability/trace). */ +export interface AdminTraceSpan { + id: string; + parentId?: string; + name: string; + kind: string; + status: string; + startedAt: string; + endedAt?: string; + durationNs: number; + props?: Record; + events?: TraceEvent[]; +} + +/** A fine-grained event attached to a span. */ +export interface TraceEvent { + time: string; + type: string; + props?: Record; +} + /** The owner profile (mirrors configs/users/owner.yaml). */ export interface OwnerProfile { id: string; diff --git a/internal/services/gateway/audit.go b/internal/services/gateway/audit.go index b5bd446..859fcc9 100644 --- a/internal/services/gateway/audit.go +++ b/internal/services/gateway/audit.go @@ -1,6 +1,7 @@ package gateway import ( + "context" "net/http" "time" @@ -16,15 +17,23 @@ func (s *Server) auditClaims(r *http.Request, what, result string) { if claims, ok := auth.ClaimsFromContext(r.Context()); ok { who = claims.Sub + ":" + string(claims.Role) } - s.audit(who, what, result) + s.audit(r.Context(), who, what, result) } -// audit emits one structured audit record on the gateway logger. -func (s *Server) audit(who, what, result string) { +// audit emits one structured audit record on the gateway logger and persists it to +// the configured AuditStore (falling back to an in-memory store when none is +// injected). A persist failure is logged but never blocks the admin write, so a +// broken audit store cannot take down the audit trail's primary job. +func (s *Server) audit(ctx context.Context, who, what, result string) { s.log.Info("admin audit", "who", who, "what", what, "when", time.Now().UTC().Format(time.RFC3339), "result", result, ) + if s.obs != nil && s.obs.audit != nil { + if err := s.obs.audit.Append(ctx, who, what, result, time.Now()); err != nil { + s.log.Error("admin audit persist", "err", err, "what", what) + } + } } diff --git a/internal/services/gateway/middleware.go b/internal/services/gateway/middleware.go index 41c228b..91b9eda 100644 --- a/internal/services/gateway/middleware.go +++ b/internal/services/gateway/middleware.go @@ -74,11 +74,13 @@ func (s *Server) logging(next http.Handler) http.Handler { start := time.Now() rec := &statusRecorder{ResponseWriter: w, status: http.StatusOK} next.ServeHTTP(rec, r) + durMs := time.Since(start).Milliseconds() + s.recordMetric(r.Method, r.URL.Path, rec.status, durMs) s.log.Info("request", "method", r.Method, "path", r.URL.Path, "status", rec.status, - "duration_ms", time.Since(start).Milliseconds(), + "duration_ms", durMs, ) }) } diff --git a/internal/services/gateway/observability.go b/internal/services/gateway/observability.go new file mode 100644 index 0000000..c7993c2 --- /dev/null +++ b/internal/services/gateway/observability.go @@ -0,0 +1,125 @@ +package gateway + +import ( + "net/http" + "strconv" + "time" + + apperr "git.yeij.top/AskaEth/Cyrene_v2/internal/errors" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" +) + +// Default list limits shared by the trace/audit endpoints. Compile-time constants +// so no magic number appears in the run path (project rule §10). +const ( + defaultObsLimit = 50 + maxObsLimit = 500 +) + +// observability bundles the gateway's live admin instrumentation: the request +// metrics counter, the trace span ring buffer and the audit store. It is built in +// New and is safe for concurrent use. +type observability struct { + metrics *MetricsCollector + traceBuf *RingBuffer + audit AuditStore +} + +// newObservability builds the instrumentation, falling back to an in-memory audit +// store when no store is injected. +func newObservability(deps Deps) *observability { + obs := &observability{ + metrics: newMetricsCollector(), + traceBuf: newRingBuffer(defaultTraceBufCap), + } + if deps.AuditStore != nil { + obs.audit = deps.AuditStore + } else { + obs.audit = newMemoryAuditStore() + } + return obs +} + +// ObserveSpan feeds a completed trace span into the gateway's ring buffer and the +// LLM/subagent metric counters. It is the gateway's trace injection point: the +// composition root wires the loop tracer here (e.g. through TraceSink) so the +// admin panel can query spans without the gateway touching wire.go. +func (s *Server) ObserveSpan(rec trace.SpanRecord) { + if s.obs == nil { + return + } + s.obs.traceBuf.Append(rec) + s.obs.metrics.Observe(rec) +} + +// TraceSink returns a trace.Writer that appends completed spans to the gateway's +// trace ring buffer. Combine it with the existing JSONRecorder in the composition +// root (e.g. trace.New(MultiTraceWriter{...})) so spans both hit stderr and the +// admin panel. +func (s *Server) TraceSink() trace.Writer { + if s.obs == nil { + return nil + } + return traceSink{s.obs.traceBuf} +} + +// recordMetric counts one request for the observability panel. +func (s *Server) recordMetric(method, path string, status int, durMs int64) { + if s.obs == nil { + return + } + s.obs.metrics.Record(method, path, status, durMs) +} + +// adminObservabilityMetrics returns the current gateway metrics (admin read). +func (s *Server) adminObservabilityMetrics(w http.ResponseWriter, r *http.Request) { + if s.obs == nil { + writeJSON(w, http.StatusOK, AdminMetrics{ + PerMethod: map[string]int64{}, PerPath: map[string]int64{}, StatusCounts: map[string]int64{}, + }) + return + } + writeJSON(w, http.StatusOK, s.obs.metrics.Snapshot(time.Now())) +} + +// adminObservabilityTrace returns the most recent traced spans (admin read). +func (s *Server) adminObservabilityTrace(w http.ResponseWriter, r *http.Request) { + if s.obs == nil { + writeJSON(w, http.StatusOK, []AdminTraceSpan{}) + return + } + spans := s.obs.traceBuf.Snapshot(parseLimit(r)) + out := make([]AdminTraceSpan, 0, len(spans)) + for _, sp := range spans { + out = append(out, adminSpanOf(sp)) + } + writeJSON(w, http.StatusOK, out) +} + +// adminObservabilityAudit returns the most recent audit entries (admin read). +func (s *Server) adminObservabilityAudit(w http.ResponseWriter, r *http.Request) { + if s.obs == nil || s.obs.audit == nil { + writeJSON(w, http.StatusOK, []AuditRecord{}) + return + } + records, err := s.obs.audit.List(r.Context(), parseLimit(r)) + if err != nil { + s.log.Error("list audit", "err", err) + writeError(w, apperr.Wrap(apperr.ErrInternal, "audit_failed", "failed to list audit records")) + return + } + writeJSON(w, http.StatusOK, records) +} + +// parseLimit parses the ?limit= query param, clamping to [1, maxObsLimit], and +// selects defaultObsLimit when absent or invalid. +func parseLimit(r *http.Request) int { + n, err := strconv.Atoi(r.URL.Query().Get("limit")) + if err != nil || n <= 0 { + return defaultObsLimit + } + if n > maxObsLimit { + return maxObsLimit + } + return n +} diff --git a/internal/services/gateway/observability_audit.go b/internal/services/gateway/observability_audit.go new file mode 100644 index 0000000..248195d --- /dev/null +++ b/internal/services/gateway/observability_audit.go @@ -0,0 +1,164 @@ +package gateway + +import ( + "context" + "database/sql" + "fmt" + "log/slog" + "sync" + "time" +) + +// AuditRecord is one persisted admin audit entry (who/what/when/result). +type AuditRecord struct { + ID int64 `json:"id"` + Who string `json:"who"` + What string `json:"what"` + When string `json:"when"` + Result string `json:"result"` +} + +// AuditStore persists the admin audit trail. Implementations must be safe for +// concurrent use. *SQLiteAuditStore is the production implementation (wired by the +// composition root over a.db); when none is injected the gateway falls back to an +// in-memory store so the endpoint always returns data. +type AuditStore interface { + // Append persists one audit entry at ts. + Append(ctx context.Context, who, what, result string, ts time.Time) error + // List returns at most limit most-recent entries, newest first. A non-positive + // limit selects the default. + List(ctx context.Context, limit int) ([]AuditRecord, error) +} + +// memoryAuditStore is the nil-injection fallback: it holds the audit trail in +// process memory so the admin endpoint works before the composition root injects a +// SQLite store. It is concurrency-safe. +type memoryAuditStore struct { + mu sync.Mutex + records []AuditRecord + seq int64 +} + +// newMemoryAuditStore builds an empty in-memory audit store. +func newMemoryAuditStore() *memoryAuditStore { + return &memoryAuditStore{records: make([]AuditRecord, 0, 64)} +} + +// Append stores one entry with a monotonically increasing id. +func (m *memoryAuditStore) Append(ctx context.Context, who, what, result string, ts time.Time) error { + if err := ctx.Err(); err != nil { + return err + } + m.mu.Lock() + defer m.mu.Unlock() + m.seq++ + m.records = append(m.records, AuditRecord{ + ID: m.seq, Who: who, What: what, When: ts.UTC().Format(time.RFC3339), Result: result, + }) + return nil +} + +// List returns the most recent entries, newest first. +func (m *memoryAuditStore) List(ctx context.Context, limit int) ([]AuditRecord, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + if limit <= 0 { + limit = defaultObsLimit + } + if limit > maxObsLimit { + limit = maxObsLimit + } + m.mu.Lock() + defer m.mu.Unlock() + n := len(m.records) + start := n - limit + if start < 0 { + start = 0 + } + out := make([]AuditRecord, n-start) + copy(out, m.records[start:]) + for i, j := 0, len(out)-1; i < j; i, j = i+1, j-1 { + out[i], out[j] = out[j], out[i] + } + return out, nil +} + +// audit DDL for a self-contained SQLite store (no migration bookkeeping), the +// same pattern channel/session and backthink use. recorded_at is the UTC RFC3339 +// wall-clock of the audited action. +const ( + createAuditSQL = ` +CREATE TABLE IF NOT EXISTS audit_logs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + who TEXT NOT NULL, + what TEXT NOT NULL, + result TEXT NOT NULL, + recorded_at TEXT NOT NULL, + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')) +)` + createAuditIndexSQL = `CREATE INDEX IF NOT EXISTS idx_audit_logs_recorded_at ON audit_logs(recorded_at)` + insertAuditSQL = `INSERT INTO audit_logs (who, what, result, recorded_at) VALUES (?, ?, ?, ?)` + selectAuditSQL = `SELECT id, who, what, result, recorded_at FROM audit_logs ORDER BY id DESC LIMIT ?` +) + +// SQLiteAuditStore persists the admin audit trail in SQLite. It is safe for +// concurrent use by virtue of the underlying *sql.DB. +type SQLiteAuditStore struct { + db *sql.DB + log *slog.Logger +} + +// NewSQLiteAuditStore builds a SQLite-backed audit store over db, guaranteeing the +// audit_logs table and index exist. It returns an error when db is nil or the DDL +// fails. The composition root wires this over the primary *sql.DB (a.db). +func NewSQLiteAuditStore(ctx context.Context, db *sql.DB, log *slog.Logger) (*SQLiteAuditStore, error) { + if db == nil { + return nil, fmt.Errorf("gateway: audit store: nil database") + } + if _, err := db.ExecContext(ctx, createAuditSQL); err != nil { + return nil, fmt.Errorf("gateway: ensure audit_logs: %w", err) + } + if _, err := db.ExecContext(ctx, createAuditIndexSQL); err != nil { + return nil, fmt.Errorf("gateway: ensure audit_logs index: %w", err) + } + return &SQLiteAuditStore{db: db, log: sourcedLog(log)}, nil +} + +// Append persists one audit entry at ts. +func (s *SQLiteAuditStore) Append(ctx context.Context, who, what, result string, ts time.Time) error { + if _, err := s.db.ExecContext(ctx, insertAuditSQL, who, what, result, ts.UTC().Format(time.RFC3339Nano)); err != nil { + return fmt.Errorf("gateway: insert audit: %w", err) + } + return nil +} + +// List returns the most recent entries, newest first. +func (s *SQLiteAuditStore) List(ctx context.Context, limit int) ([]AuditRecord, error) { + if limit <= 0 { + limit = defaultObsLimit + } + if limit > maxObsLimit { + limit = maxObsLimit + } + if err := ctx.Err(); err != nil { + return nil, err + } + rows, err := s.db.QueryContext(ctx, selectAuditSQL, limit) + if err != nil { + return nil, fmt.Errorf("gateway: list audit: %w", err) + } + defer rows.Close() + records := make([]AuditRecord, 0, limit) + for rows.Next() { + var r AuditRecord + if err := rows.Scan(&r.ID, &r.Who, &r.What, &r.Result, &r.When); err != nil { + return nil, fmt.Errorf("gateway: scan audit: %w", err) + } + records = append(records, r) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("gateway: iterate audit: %w", err) + } + return records, nil +} diff --git a/internal/services/gateway/observability_metrics.go b/internal/services/gateway/observability_metrics.go new file mode 100644 index 0000000..100d3a5 --- /dev/null +++ b/internal/services/gateway/observability_metrics.go @@ -0,0 +1,131 @@ +package gateway + +import ( + "strconv" + "sync" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" +) + +// Metrics collector tuning. Compile-time constants. +const ( + // metricsWindow bounds how long request timestamps are retained for QPS. + metricsWindow = 2 * time.Second + // metricsQPSSpan is the sliding window over which QPS is measured. + metricsQPSSpan = time.Second +) + +// AdminMetrics is the wire shape of GET /api/v1/admin/observability/metrics. +// LLM/Token cost is deliberately absent: the gateway does not track token usage or +// spend, and a fabricated number would violate the "不伪造指标" red line. +type AdminMetrics struct { + TotalRequests int64 `json:"totalRequests"` + QPS float64 `json:"qps"` + AvgLatencyMs float64 `json:"avgLatencyMs"` + MaxLatencyMs int64 `json:"maxLatencyMs"` + PerMethod map[string]int64 `json:"perMethod"` + PerPath map[string]int64 `json:"perPath"` + StatusCounts map[string]int64 `json:"statusCounts"` + LLMCalls int64 `json:"llmCalls"` + SubagentEvents int64 `json:"subagentEvents"` +} + +// MetricsCollector accumulates request metrics and derived counters for the admin +// observability panel. It is concurrency-safe. +type MetricsCollector struct { + mu sync.Mutex + total int64 + perMethod map[string]int64 + perPath map[string]int64 + statuses map[int]int64 + sumDur int64 + maxDur int64 + recent []time.Time + llm int64 + subagent int64 +} + +// newMetricsCollector builds an empty metrics collector. +func newMetricsCollector() *MetricsCollector { + return &MetricsCollector{ + perMethod: make(map[string]int64), + perPath: make(map[string]int64), + statuses: make(map[int]int64), + } +} + +// Record counts one request with the given method, path, status and duration. +func (m *MetricsCollector) Record(method, path string, status int, durMs int64) { + m.mu.Lock() + defer m.mu.Unlock() + now := time.Now() + m.total++ + m.perMethod[method]++ + m.perPath[path]++ + m.statuses[status]++ + m.sumDur += durMs + if durMs > m.maxDur { + m.maxDur = durMs + } + m.recent = append(m.recent, now) + m.prune(now) +} + +// Observe increments the LLM/subagent counters when a span of that kind is +// recorded. It is called by ObserveSpan so the counters reflect real work. +func (m *MetricsCollector) Observe(rec trace.SpanRecord) { + m.mu.Lock() + defer m.mu.Unlock() + switch rec.Kind { + case trace.SpanKindLLM: + m.llm++ + case trace.SpanKindSubagent: + m.subagent++ + } +} + +// prune drops request times older than metricsWindow so recent stays small. The +// caller must hold m.mu. +func (m *MetricsCollector) prune(now time.Time) { + min := now.Add(-metricsWindow) + i := 0 + for i < len(m.recent) && m.recent[i].Before(min) { + i++ + } + if i > 0 { + m.recent = append([]time.Time(nil), m.recent[i:]...) + } +} + +// Snapshot returns a point-in-time copy of the metrics. qps counts the requests +// started within metricsQPSSpan of now. +func (m *MetricsCollector) Snapshot(now time.Time) AdminMetrics { + m.mu.Lock() + defer m.mu.Unlock() + statuses := make(map[string]int64, len(m.statuses)) + for k, v := range m.statuses { + statuses[strconv.Itoa(k)] = v + } + var qps int64 + for _, t := range m.recent { + if now.Sub(t) <= metricsQPSSpan { + qps++ + } + } + avg := int64(0) + if m.total > 0 { + avg = m.sumDur / m.total + } + return AdminMetrics{ + TotalRequests: m.total, + QPS: float64(qps), + AvgLatencyMs: float64(avg), + MaxLatencyMs: m.maxDur, + PerMethod: m.perMethod, + PerPath: m.perPath, + StatusCounts: statuses, + LLMCalls: m.llm, + SubagentEvents: m.subagent, + } +} diff --git a/internal/services/gateway/observability_trace.go b/internal/services/gateway/observability_trace.go new file mode 100644 index 0000000..0bb3e8d --- /dev/null +++ b/internal/services/gateway/observability_trace.go @@ -0,0 +1,137 @@ +package gateway + +import ( + "sync" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" +) + +// defaultTraceBufCap is how many completed spans the ring buffer holds before the +// oldest is evicted. Compile-time constant. +const defaultTraceBufCap = 512 + +// AdminTraceSpan is the camelCase wire shape of one traced span. It decouples the +// frontend from the trace package's snake_case SpanRecord. +type AdminTraceSpan struct { + ID string `json:"id"` + ParentID string `json:"parentId,omitempty"` + Name string `json:"name"` + Kind string `json:"kind"` + Status string `json:"status"` + StartedAt string `json:"startedAt"` + EndedAt string `json:"endedAt,omitempty"` + DurationNs int64 `json:"durationNs"` + Props map[string]any `json:"props,omitempty"` + Events []trace.Event `json:"events,omitempty"` +} + +// RingBuffer is a concurrency-safe bounded buffer of completed trace spans. The +// oldest span is evicted when the buffer is full, so the admin panel always sees +// the most recent work without unbounded growth. +type RingBuffer struct { + mu sync.RWMutex + buf []trace.SpanRecord + next int + full bool + cap int +} + +// newRingBuffer builds a ring buffer holding up to cap spans (default when cap<=0). +func newRingBuffer(cap int) *RingBuffer { + if cap <= 0 { + cap = defaultTraceBufCap + } + return &RingBuffer{buf: make([]trace.SpanRecord, cap), cap: cap} +} + +// Append stores rec, evicting the oldest when the buffer is full. +func (r *RingBuffer) Append(rec trace.SpanRecord) { + r.mu.Lock() + defer r.mu.Unlock() + r.buf[r.next] = rec + r.next = (r.next + 1) % r.cap + if r.next == 0 { + r.full = true + } +} + +// Snapshot returns up to n most-recent spans, newest first. A non-positive n +// selects defaultObsLimit; n is clamped to maxObsLimit. +func (r *RingBuffer) Snapshot(n int) []trace.SpanRecord { + if n <= 0 { + n = defaultObsLimit + } + if n > maxObsLimit { + n = maxObsLimit + } + r.mu.RLock() + defer r.mu.RUnlock() + count := r.next + if r.full { + count = r.cap + } + if count > n { + count = n + } + 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]) + } + return out +} + +// traceSink adapts a *RingBuffer to trace.Writer so a completed span is appended +// to the gateway's trace buffer. +type traceSink struct{ b *RingBuffer } + +// WriteSpan appends rec to the ring buffer. +func (s traceSink) WriteSpan(rec trace.SpanRecord) error { + if s.b == nil { + return nil + } + s.b.Append(rec) + return nil +} + +// MultiTraceWriter fans a completed span out to several trace.Writers, so a tracer +// can record to both the gateway's ring buffer and a JSONL file in one pass. A nil +// Writer is skipped. It is the composition-root helper the main session uses to +// route the loop tracer's output into the gateway without dropping stderr. +type MultiTraceWriter struct { + // Writers are the sinks a span is written to. A nil Writer is skipped. + Writers []trace.Writer +} + +// WriteSpan writes rec to every configured Writer. +func (m MultiTraceWriter) WriteSpan(rec trace.SpanRecord) error { + for _, w := range m.Writers { + if w == nil { + continue + } + if err := w.WriteSpan(rec); err != nil { + return err + } + } + return nil +} + +// adminSpanOf maps a trace.SpanRecord to the camelCase admin wire shape. +func adminSpanOf(rec trace.SpanRecord) AdminTraceSpan { + sp := AdminTraceSpan{ + ID: rec.ID, + ParentID: rec.ParentID, + Name: rec.Name, + Kind: string(rec.Kind), + Status: string(rec.Status), + StartedAt: rec.StartedAt.UTC().Format(time.RFC3339), + DurationNs: rec.Duration.Nanoseconds(), + Props: rec.Props, + Events: rec.Events, + } + if !rec.EndedAt.IsZero() { + sp.EndedAt = rec.EndedAt.UTC().Format(time.RFC3339) + } + return sp +} diff --git a/internal/services/gateway/server.go b/internal/services/gateway/server.go index 4c892f2..c595504 100644 --- a/internal/services/gateway/server.go +++ b/internal/services/gateway/server.go @@ -191,6 +191,11 @@ type Deps struct { // RAG backs the admin knowledge-base RAG module. Optional; list/search // return empty when unset, and write endpoints report a clear 501. RAG RagStore + // AuditStore persists the admin audit trail (observability → audit). When nil + // the gateway degrades to an in-memory store so the endpoint still returns + // data; the composition root injects NewSQLiteAuditStore over a.db for real + // persistence. Optional. + AuditStore AuditStore // Streaming marks the injected Loop as streaming-capable (its OnStream is // wired to publish ws.stream_chunk). When set the hub skips its own single // final stream_chunk emission, since deltas already stream through OnStream. @@ -225,6 +230,7 @@ type Server struct { users UserStore backthink BackthinkStore rag RagStore + obs *observability voice VoiceService streaming bool frontend fs.FS @@ -259,6 +265,7 @@ func New(cfg *config.Config, deps Deps) (*Server, error) { users: deps.Users, backthink: deps.Backthink, rag: deps.RAG, + obs: newObservability(deps), streaming: deps.Streaming, frontend: deps.Frontend, webhooks: deps.Webhooks, @@ -303,6 +310,9 @@ func (s *Server) Handler() http.Handler { mux.Handle("GET /api/v1/admin/rag", adminAuth(http.HandlerFunc(s.adminListRag))) mux.Handle("POST /api/v1/admin/rag/query", adminAuth(http.HandlerFunc(s.adminSearchRag))) mux.Handle("POST /api/v1/admin/rag/delete", adminAuth(http.HandlerFunc(s.adminDeleteRag))) + mux.Handle("GET /api/v1/admin/observability/metrics", adminAuth(http.HandlerFunc(s.adminObservabilityMetrics))) + mux.Handle("GET /api/v1/admin/observability/trace", adminAuth(http.HandlerFunc(s.adminObservabilityTrace))) + mux.Handle("GET /api/v1/admin/observability/audit", adminAuth(http.HandlerFunc(s.adminObservabilityAudit))) mux.Handle("POST /api/v1/admin/plugins/policy", s.auth.Authorize(auth.RoleOwner)(http.HandlerFunc(s.adminSetPolicy))) // External webhook callbacks (e.g. the webhook channel) are mounted without