feat(admin): 接上可观测性下行帧，概览页实时刷新指标 / trace / 日志
契约 v0.55 §3.12 的三帧（`observability_metrics` / `observability_trace` / `observability_log`）
此前在本端没有 handler，落到未知帧被静默丢弃，概览页只能「进页面 / 手动刷新各拉一次 REST」。
- 解析（data/ws）：三个事件类 + `dispatchObservability` 分支，载荷按契约逐字对齐
（metrics = 扁平本体、trace = `{spans}`、log = `{entries}`），帧壳 `sessionId` 有就透传。
载荷缺席时**整帧安全忽略**——metrics 回落成默认值就是一份全零快照，会把界面上的真实指标
擦成 0；未知帧、坏帧仍安静丢弃（回归用例钉住，绝不抛）。
- 消费（ui/admin）：AdminViewModel 自持一条 socket（照 SessionsViewModel 的做法：登录且
角色够档时连、登出 / 角色降档 / onCleared 关），把三帧合并进概览状态。REST 仍是兜底真源：
进页面与手动刷新照旧各拉一次全量；每次「连上」（含首次连接与断线重连）后再重读一次——
断线窗口里的帧永久丢失，而服务端只为「有在线 owner/admin 连接」的凭据保留订阅，
重读同时也重建订阅（契约 §3.12：读取即订阅握手）。
- 角色门在客户端也判（domain/admin 新增纯函数 `canReceiveObservabilityFrames`）：这三帧服务端
fail-closed 只发 owner + admin，与概览页自身的 Monitor 档是两条门，viewer 既不开连接也不合帧。
- 日志只有推送这一个来源（本端没有 logs REST 端点）：一帧都没到过时整段不渲染——不写
「暂无日志」，那等于把「没有推送」说成「没有数据」；等级过滤按契约在客户端本地做。
- 合并与过滤是 ui/admin 下的纯函数（新帧在前、按业务键去重、窗口 100/200），单测覆盖。
测试 485 → 514（+29），0 失败 0 错误；`assembleDebug` 通过。
This commit is contained in:
AskaEth committed
1 parent
a6b4bae2a0
commit
b5e6f78910
9 files changed
+831
-11
No files matched your search
@@ -0,0 +1,22 @@
|
||||
package top.yeij.cyrene.data.ws
|
||||
|
||||
import kotlinx.serialization.Serializable
|
||||
|
||||
/**
|
||||
* `observability_log` 帧的一条日志（契约 v0.55 §3.12，与 `GET /api/v1/admin/observability/logs`
|
||||
* 同形，后端生产者 `gateway.AdminLogEntry`）：`{time,level,source,msg}`。
|
||||
*
|
||||
* - [time] 是 **RFC3339（UTC，秒精度）**——与 `trace.startedAt` / `audit.when` 同源同形状，
|
||||
* 而**不是** backthink 那种 Unix 毫秒，两者不可共用解析器。
|
||||
* - [level] ∈ `debug|info|warn|error`，是**开放集合**：本层原样保留，未知等级既不丢弃也不归一。
|
||||
* - [source] 是 logger 的来源标签，记录没带来源时是空串（不是 null）。
|
||||
*
|
||||
* 放在 `data.ws` 而不是 `data.api.dto`：本端**没有**日志 REST 端点，这个类型唯一的来源就是推送帧。
|
||||
*/
|
||||
@Serializable
|
||||
data class ObservabilityLogEntryDto(
|
||||
val time: String = "",
|
||||
val level: String = "",
|
||||
val source: String = "",
|
||||
val msg: String = "",
|
||||
)
|
||||
@@ -1,6 +1,8 @@
|
||||
package top.yeij.cyrene.data.ws
|
||||
|
||||
import top.yeij.cyrene.data.api.ApiClientFactory
|
||||
import top.yeij.cyrene.data.api.dto.AdminMetricsDto
|
||||
import top.yeij.cyrene.data.api.dto.AdminTraceSpanDto
|
||||
import top.yeij.cyrene.data.api.dto.HistoryMessageDto
|
||||
import kotlinx.serialization.Serializable
|
||||
import kotlinx.serialization.json.JsonElement
|
||||
@@ -349,6 +351,60 @@ sealed class WsEvent {
|
||||
override val type: String get() = "channel_persisted"
|
||||
}
|
||||
|
||||
/**
|
||||
* 指标快照（契约 v0.55 §3.12「可观测实时推送」）。
|
||||
*
|
||||
* 载荷就是 `AdminMetrics` **本体、扁平**，与 `GET /admin/observability/metrics` 同形，
|
||||
* 因此消费方应当**整份替换**而不是累加：服务端只在快照与「上次发出的」不同时才发帧
|
||||
* （序列化指纹比较，键排序确定）。
|
||||
*
|
||||
* [metrics] 为 `null` = 帧里没有可用的载荷对象。这时**必须整帧跳过**：
|
||||
* 回落成默认值会得到一份「全零快照」，把界面上的真实指标擦成 0。
|
||||
*
|
||||
* 帧壳**不带** `sessionId` / `turnId` / `seq`——它不属任何会话（订阅只跟凭据绑定）。
|
||||
*/
|
||||
data class ObservabilityMetrics(
|
||||
val metrics: AdminMetricsDto?,
|
||||
override val turnId: String? = null,
|
||||
override val seq: Long? = null,
|
||||
override val sessionId: String? = null,
|
||||
) : WsEvent() {
|
||||
override val type: String get() = "observability_metrics"
|
||||
}
|
||||
|
||||
/**
|
||||
* 新完成的 trace span（契约 v0.55 §3.12）。
|
||||
*
|
||||
* 载荷 `{spans:[…]}` **最新在前**，与 `GET /admin/observability/trace` 同形，
|
||||
* 单帧上限 100 条；触发是「订阅之后**新完成**的 span」（订阅锚点 = 当时的 trace 写入序号），
|
||||
* 所以它是**增量**——历史由页面那次 REST 引导读取提供，这里只负责往后接。
|
||||
*/
|
||||
data class ObservabilityTrace(
|
||||
val spans: List<AdminTraceSpanDto>,
|
||||
override val turnId: String? = null,
|
||||
override val seq: Long? = null,
|
||||
override val sessionId: String? = null,
|
||||
) : WsEvent() {
|
||||
override val type: String get() = "observability_trace"
|
||||
}
|
||||
|
||||
/**
|
||||
* 新增日志（契约 v0.55 §3.12）。
|
||||
*
|
||||
* 载荷 `{entries:[…]}` **最新在前**（与 `GET /admin/observability/logs` 同形），
|
||||
* 单帧上限 200 条；触发是「订阅之后**新增**的日志」（订阅锚点 = 当时最新的条目时间），
|
||||
* 所以它同样是**增量**。服务端的 `?level=` 分级过滤对推送不适用（帧是广播的，hub 不按连接分级），
|
||||
* 分级选择由客户端本地做。
|
||||
*/
|
||||
data class ObservabilityLog(
|
||||
val entries: List<ObservabilityLogEntryDto>,
|
||||
override val turnId: String? = null,
|
||||
override val seq: Long? = null,
|
||||
override val sessionId: String? = null,
|
||||
) : WsEvent() {
|
||||
override val type: String get() = "observability_log"
|
||||
}
|
||||
|
||||
/** 语音合成回复（定向回请求连接）。 */
|
||||
data class VoiceOutput(
|
||||
val audio: String?,
|
||||
@@ -413,8 +469,41 @@ sealed class WsEvent {
|
||||
sessionId: String?,
|
||||
data: JsonElement?,
|
||||
): WsEvent? = dispatchStreaming(type, turnId, seq, sessionId, data)
|
||||
?: dispatchObservability(type, turnId, seq, sessionId, data)
|
||||
?: dispatchSession(type, turnId, seq, sessionId, data)
|
||||
|
||||
/**
|
||||
* 可观测实时推送三帧（契约 v0.55 §3.12）。
|
||||
*
|
||||
* `data` **必须是 JSON 对象**才认：这三帧的载荷是它们的全部内容，缺席时没有任何可用的东西。
|
||||
* 尤其 metrics——若回落成默认值，消费方会拿一份全零快照覆盖界面上的真实指标。返回 `null`
|
||||
* 让整帧按坏帧被安全忽略（与未知 `type` 同一条出口，绝不抛）。
|
||||
*/
|
||||
private fun dispatchObservability(
|
||||
type: String,
|
||||
turnId: String?,
|
||||
seq: Long?,
|
||||
sessionId: String?,
|
||||
data: JsonElement?,
|
||||
): WsEvent? {
|
||||
val payload = data as? JsonObject ?: return null
|
||||
return when (type) {
|
||||
"observability_metrics" -> ObservabilityMetrics(
|
||||
decode<AdminMetricsDto>(payload), turnId, seq, sessionId,
|
||||
)
|
||||
|
||||
"observability_trace" -> ObservabilityTrace(
|
||||
decode<TracePushData>(payload).spans, turnId, seq, sessionId,
|
||||
)
|
||||
|
||||
"observability_log" -> ObservabilityLog(
|
||||
decode<LogPushData>(payload).entries, turnId, seq, sessionId,
|
||||
)
|
||||
|
||||
else -> null
|
||||
}
|
||||
}
|
||||
|
||||
private fun dispatchStreaming(
|
||||
type: String,
|
||||
turnId: String?,
|
||||
@@ -597,6 +686,14 @@ private data class SessionReadData(val readAt: Long = 0L)
|
||||
@Serializable
|
||||
private data class PersistedData(val message: HistoryMessageDto? = null)
|
||||
|
||||
/** `observability_trace` 载荷（后端 `gateway.AdminTracePush`）：`{spans:[…]}`，最新在前。 */
|
||||
@Serializable
|
||||
private data class TracePushData(val spans: List<AdminTraceSpanDto> = emptyList())
|
||||
|
||||
/** `observability_log` 载荷（后端 `gateway.AdminLogPush`）：`{entries:[…]}`，最新在前。 */
|
||||
@Serializable
|
||||
private data class LogPushData(val entries: List<ObservabilityLogEntryDto> = emptyList())
|
||||
|
||||
@Serializable
|
||||
private data class VoiceOutputData(
|
||||
val audio: String? = null,
|
||||
|
||||
Reference in new issue
Block a user