From b5e6f7891039bb60e01d6a99bfc7f567f048114d Mon Sep 17 00:00:00 2001 From: AskaEth Date: Sun, 4 Oct 2026 18:40:45 +0800 Subject: [PATCH] =?UTF-8?q?feat(admin):=20=E6=8E=A5=E4=B8=8A=E5=8F=AF?= =?UTF-8?q?=E8=A7=82=E6=B5=8B=E6=80=A7=E4=B8=8B=E8=A1=8C=E5=B8=A7=EF=BC=8C?= =?UTF-8?q?=E6=A6=82=E8=A7=88=E9=A1=B5=E5=AE=9E=E6=97=B6=E5=88=B7=E6=96=B0?= =?UTF-8?q?=E6=8C=87=E6=A0=87=20/=20trace=20/=20=E6=97=A5=E5=BF=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 契约 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` 通过。 --- .../cyrene/data/ws/ObservabilityLogEntry.kt | 22 ++ .../kotlin/top/yeij/cyrene/data/ws/WsEvent.kt | 97 +++++++++ .../yeij/cyrene/domain/admin/AdminAccess.kt | 14 ++ .../cyrene/ui/admin/AdminOverviewMerge.kt | 95 +++++++++ .../cyrene/ui/admin/AdminOverviewScreen.kt | 82 +++++++- .../yeij/cyrene/ui/admin/AdminViewModel.kt | 177 +++++++++++++++- .../data/ws/WsEventObservabilityTest.kt | 198 ++++++++++++++++++ .../cyrene/domain/admin/AdminAccessTest.kt | 22 ++ .../cyrene/ui/admin/AdminOverviewMergeTest.kt | 135 ++++++++++++ 9 files changed, 831 insertions(+), 11 deletions(-) create mode 100644 app/src/main/kotlin/top/yeij/cyrene/data/ws/ObservabilityLogEntry.kt create mode 100644 app/src/main/kotlin/top/yeij/cyrene/ui/admin/AdminOverviewMerge.kt create mode 100644 app/src/test/kotlin/top/yeij/cyrene/data/ws/WsEventObservabilityTest.kt create mode 100644 app/src/test/kotlin/top/yeij/cyrene/ui/admin/AdminOverviewMergeTest.kt diff --git a/app/src/main/kotlin/top/yeij/cyrene/data/ws/ObservabilityLogEntry.kt b/app/src/main/kotlin/top/yeij/cyrene/data/ws/ObservabilityLogEntry.kt new file mode 100644 index 0000000..489845b --- /dev/null +++ b/app/src/main/kotlin/top/yeij/cyrene/data/ws/ObservabilityLogEntry.kt @@ -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 = "", +) diff --git a/app/src/main/kotlin/top/yeij/cyrene/data/ws/WsEvent.kt b/app/src/main/kotlin/top/yeij/cyrene/data/ws/WsEvent.kt index 80eca0f..4b6050b 100644 --- a/app/src/main/kotlin/top/yeij/cyrene/data/ws/WsEvent.kt +++ b/app/src/main/kotlin/top/yeij/cyrene/data/ws/WsEvent.kt @@ -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, + 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, + 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(payload), turnId, seq, sessionId, + ) + + "observability_trace" -> ObservabilityTrace( + decode(payload).spans, turnId, seq, sessionId, + ) + + "observability_log" -> ObservabilityLog( + decode(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 = emptyList()) + +/** `observability_log` 载荷（后端 `gateway.AdminLogPush`）：`{entries:[…]}`，最新在前。 */ +@Serializable +private data class LogPushData(val entries: List = emptyList()) + @Serializable private data class VoiceOutputData( val audio: String? = null, diff --git a/app/src/main/kotlin/top/yeij/cyrene/domain/admin/AdminAccess.kt b/app/src/main/kotlin/top/yeij/cyrene/domain/admin/AdminAccess.kt index ce16194..b7b65bd 100644 --- a/app/src/main/kotlin/top/yeij/cyrene/domain/admin/AdminAccess.kt +++ b/app/src/main/kotlin/top/yeij/cyrene/domain/admin/AdminAccess.kt @@ -56,5 +56,19 @@ enum class AdminPage( /** 当前角色能否使用该页；不能则 UI 禁用入口、ViewModel 直接给「无权限」。 */ fun canAccess(role: AdminRole, page: AdminPage): Boolean = role.rank >= page.required.rank +/** + * 能否接收可观测实时推送帧（`observability_metrics` / `observability_trace` / + * `observability_log`，契约 §3.12）。 + * + * 这三帧**不在**那 7 个「不含内容的状态机信号」白名单里，服务端按 fail-closed 只发 + * **owner + admin**。注意这与 [AdminPage.Overview] 的可见性**是两回事**：概览页本身是 + * [AdminTier.Monitor]，`viewer` 也能进、也能读那两个 REST 端点（v0.40 只读监控档）， + * 但它**一帧推送都收不到**（服务端不推，客户端也就不该开连接去等）。 + * + * 客户端按同一档自行降级：不够档就**不开连接、不合并帧**，而不是等「怎么一直没帧」再猜。 + */ +fun canReceiveObservabilityFrames(role: AdminRole): Boolean = + role.rank >= AdminTier.OwnerAdmin.rank + /** 当前角色可用的页，顺序与 [AdminPage] 声明一致。 */ fun availablePages(role: AdminRole): List = AdminPage.entries.filter { canAccess(role, it) } diff --git a/app/src/main/kotlin/top/yeij/cyrene/ui/admin/AdminOverviewMerge.kt b/app/src/main/kotlin/top/yeij/cyrene/ui/admin/AdminOverviewMerge.kt new file mode 100644 index 0000000..1bb9585 --- /dev/null +++ b/app/src/main/kotlin/top/yeij/cyrene/ui/admin/AdminOverviewMerge.kt @@ -0,0 +1,95 @@ +package top.yeij.cyrene.ui.admin + +import top.yeij.cyrene.data.api.dto.AdminTraceSpanDto +import top.yeij.cyrene.data.ws.ObservabilityLogEntryDto + +/** + * 可观测实时推送帧的**合并**与**本地过滤**（纯函数，契约 v0.55 §3.12）。 + * + * 两个来源各自负责一半，谁都不替代谁： + * - **REST 引导读取**（进页面 / 手动刷新 / 断线重连）给一份**全量**快照（最新在前）； + * - **推送帧**是**增量**：`observability_trace` 只带订阅之后新完成的 span， + * `observability_log` 只带订阅之后新增的日志（订阅时以「当前最新」为锚，服务端**不会** + * 把环形缓冲里的历史灌成巨型帧）；`observability_metrics` 是**整份快照**，见 + * `AdminViewModel`（它是替换而不是合并，所以不在这里）。 + * + * 因此这里的合并语义是「**往前接**」：新到的在前，与已有按业务键去重，再按窗口截断。 + * 两端本来就是最新在前，所以**不做排序**——排序需要解析时间戳（日志是 RFC3339 秒精度、 + * span 是纳秒精度、还可能带不同偏移），一旦解析口径与后端不一致，反而会把正确的顺序排乱。 + */ + +/** trace 保留窗口，与 REST 引导读取的 `limit`（`AdminRepository.DEFAULT_TRACE_LIMIT`）一致。 */ +internal const val TRACE_WINDOW = 100 + +/** 日志保留窗口，与契约 §3.12 的单帧上限一致（更旧的条目本就落在保留窗口之外）。 */ +internal const val LOG_WINDOW = 200 + +/** + * 合并 trace 帧：新 span 在前，按 `id` 去重（同一条 span 不会因为「帧与 REST 竞态」出现两次）， + * 保留最近 [window] 条。[incoming] 为空时**原样返回** [existing]（空增量不改状态， + * 也让 REST 那份快照保持原样）。 + */ +internal fun mergeTraceSpans( + existing: List, + incoming: List, + window: Int = TRACE_WINDOW, +): List = mergeNewestFirst(existing, incoming, window) { it.id } + +/** + * 合并日志帧：新条目在前，按业务键（时间 + 等级 + 来源 + 消息）去重，保留最近 [window] 条。 + * + * 去重键取整条内容，是因为日志**没有 id**（REST `logs` 的条目形状就是这四个字段）。 + * 代价如实说：同一秒内、来源与消息完全相同的两条日志会被收成一条——与 Web 面板 + * （`observabilityStream.ts` 按业务键合并）同一口径。 + */ +internal fun mergeLogEntries( + existing: List, + incoming: List, + window: Int = LOG_WINDOW, +): List = mergeNewestFirst(existing, incoming, window, ::logEntryKey) + +/** 日志条目的业务键；合并去重与列表 `key` 都用它，两处必须一致。 */ +internal fun logEntryKey(entry: ObservabilityLogEntryDto): String = + "${entry.time}|${entry.level}|${entry.source}|${entry.msg}" + +/** + * 按**最低等级**过滤日志，纯客户端行为：服务端的 `?level=` 对推送不适用 + * （帧广播给所有 owner/admin 连接，订阅只决定推送循环跑不跑）。 + * + * [minLevel] 为空 / 未知 → 不过滤。**等级未知的条目一律保留**：`level` 是开放集合， + * 判不出高低的条目不能因为「可能低于阈值」被静默藏掉（与原样保留未知值的约定一致）。 + */ +internal fun filterLogEntries( + entries: List, + minLevel: String?, +): List { + val threshold = levelRank(minLevel) ?: return entries + return entries.filter { (levelRank(it.level) ?: Int.MAX_VALUE) >= threshold } +} + +/** 日志等级序：契约里的 `debug|info|warn|error` 由低到高；未知等级返回 null。 */ +private fun levelRank(level: String?): Int? = when (level?.trim()?.lowercase()) { + "debug" -> 0 + "info" -> 1 + "warn" -> 2 + "error" -> 3 + else -> null +} + +/** + * 「最新在前」列表的合并：先把 [incoming] 按原序接在前面（内部同样按 [key] 去重）， + * 再补上 [existing] 里没出现过的那些；最后截断到 [window]。 + */ +private fun mergeNewestFirst( + existing: List, + incoming: List, + window: Int, + key: (T) -> String, +): List { + if (incoming.isEmpty()) return existing + val seen = HashSet(incoming.size + existing.size) + val merged = ArrayList(incoming.size + existing.size) + for (item in incoming) if (seen.add(key(item))) merged += item + for (item in existing) if (seen.add(key(item))) merged += item + return if (merged.size <= window) merged else ArrayList(merged.subList(0, window)) +} diff --git a/app/src/main/kotlin/top/yeij/cyrene/ui/admin/AdminOverviewScreen.kt b/app/src/main/kotlin/top/yeij/cyrene/ui/admin/AdminOverviewScreen.kt index e8140b0..c598af6 100644 --- a/app/src/main/kotlin/top/yeij/cyrene/ui/admin/AdminOverviewScreen.kt +++ b/app/src/main/kotlin/top/yeij/cyrene/ui/admin/AdminOverviewScreen.kt @@ -14,7 +14,9 @@ import androidx.compose.runtime.LaunchedEffect import androidx.compose.runtime.collectAsState import androidx.compose.runtime.getValue import androidx.compose.runtime.mutableStateMapOf +import androidx.compose.runtime.mutableStateOf import androidx.compose.runtime.remember +import androidx.compose.runtime.setValue import androidx.compose.ui.Modifier import androidx.compose.ui.draw.rotate import androidx.compose.ui.text.font.FontFamily @@ -23,6 +25,7 @@ import androidx.compose.ui.text.style.TextOverflow import androidx.compose.ui.unit.dp import top.yeij.cyrene.data.api.dto.AdminMetricsDto import top.yeij.cyrene.data.api.dto.AdminTraceSpanDto +import top.yeij.cyrene.data.ws.ObservabilityLogEntryDto import top.yeij.cyrene.domain.admin.AdminFailure import top.yeij.cyrene.domain.admin.AdminPage import top.yeij.cyrene.domain.admin.AdminRole @@ -42,14 +45,19 @@ import top.yeij.cyrene.ui.common.PreferenceCard import top.yukonga.miuix.kmp.basic.BasicComponent import top.yukonga.miuix.kmp.basic.HorizontalDivider import top.yukonga.miuix.kmp.basic.Icon +import top.yukonga.miuix.kmp.basic.TabRow import top.yukonga.miuix.kmp.basic.Text import top.yukonga.miuix.kmp.icon.MiuixIcons import top.yukonga.miuix.kmp.icon.extended.ChevronForward import top.yukonga.miuix.kmp.theme.MiuixTheme /** - * 概览页：`metrics` 卡片 + 「按路径」分组列表 + 其下的 `trace` 列表 - * （两次请求，见 api-client §2「可观测性契约」）。 + * 概览页：`metrics` 卡片 + 「按路径」分组列表 + 其下的 `trace` 列表 + 实时推送的日志 + * （api-client §2「可观测性契约」与 §3.12「可观测实时推送」）。 + * + * 首屏仍是**两次 REST**（metrics / trace）；之后 `observability_*` 三帧（仅 owner/admin 收得到） + * 由 [AdminViewModel] 合并进同一份状态。日志只有推送这一个来源，因此**一帧都没到过时整段不渲染** + * （详见 `AdminOverviewData.logs`）。 */ @Composable fun AdminOverviewScreen(viewModel: AdminViewModel, role: AdminRole, onBack: () -> Unit) { @@ -60,11 +68,19 @@ fun AdminOverviewScreen(viewModel: AdminViewModel, role: AdminRole, onBack: () - // 只记**用户动过的组**，其余按分组默认档（见 PathStats.expanded），刷新不会把大组锁在打开状态。 val toggledGroups = remember { mutableStateMapOf() } + // 日志等级选择（`null` = 全部）。**在本地过滤**：推送帧是广播给所有 owner/admin 连接的， + // 服务端的 `?level=` 对帧不适用（契约 §3.12）。放这里而不是 LazyColumn 的 item 里， + // 是为了让「读状态 → 重算列表」发生在正常的 compose 作用域里。 + var logLevel by remember { mutableStateOf(null) } + LaunchedEffect(Unit) { reload() } AdminPageScaffold(title = AdminPage.Overview.title, onBack = onBack, onRefresh = reload) { padding -> val data = state.data val failure = state.failure + val visibleLogs = remember(data?.logs, logLevel) { + filterLogEntries(data?.logs.orEmpty(), logLevel) + } Box( modifier = Modifier .fillMaxSize() @@ -116,6 +132,20 @@ fun AdminOverviewScreen(viewModel: AdminViewModel, role: AdminRole, onBack: () - TraceSpanCard(span) } } + // 日志：**只有推送一个来源**（本端没有 logs REST 端点）。 + // 一帧都没到过就整段不渲染——这里既没有 REST 引导，也没有「确实没有日志」的依据， + // 写「暂无日志」等于把「没有推送」说成「没有数据」。 + if (data.logs.isNotEmpty()) { + item(key = "__logs_title__") { AdminSectionTitle("最近日志") } + item(key = "__logs_level__") { LogLevelTabs(logLevel) { logLevel = it } } + if (visibleLogs.isEmpty()) { + item(key = "__logs_empty__") { AdminInlineHint("当前等级下没有日志") } + } else { + items(items = visibleLogs, key = { logEntryKey(it) }) { entry -> + LogEntryCard(entry) + } + } + } item(key = "__bottom__") { Spacer(Modifier.height(32.dp)) } } } @@ -248,6 +278,54 @@ private fun TraceSpanCard(span: AdminTraceSpanDto) { ) } +/** + * 日志一行：消息是主体，来源 / 时间作副文本，等级作尾注。 + * + * 等级只给 `error` 上错误色——与 trace 卡按 `status` 着色同一口径（Miuix 配色里没有「警告」档， + * 不猜一个近似色）。消息**不截断**：这是诊断视图，被截掉的往往正是要看的那半句。 + */ +@Composable +private fun LogEntryCard(entry: ObservabilityLogEntryDto) { + AdminTextCard( + title = entry.msg.ifBlank { "（无消息）" }, + body = logMeta(entry), + trailing = entry.level.ifBlank { null }, + trailingColor = if (entry.level.equals("error", ignoreCase = true)) { + MiuixTheme.colorScheme.error + } else { + null + }, + ) +} + +/** 副文本：来源 + 本地时间；两者都可能缺，缺了就不占位。 */ +private fun logMeta(entry: ObservabilityLogEntryDto): String = listOfNotNull( + entry.source.takeIf { it.isNotBlank() }, + formatRfc3339(entry.time), +).joinToString(" · ") + +/** + * 日志等级过滤档（`null` = 全部）。最低等级语义与后端 `?level=` 一致（`warn` 含 `error`）， + * 只是过滤发生在客户端——推送帧是广播的，服务端没有按连接分级。 + */ +private val LOG_LEVEL_TABS = listOf>( + "全部" to null, + "信息+" to "info", + "警告+" to "warn", + "错误" to "error", +) + +@Composable +private fun LogLevelTabs(selected: String?, onSelect: (String?) -> Unit) { + TabRow( + tabs = LOG_LEVEL_TABS.map { it.first }, + // 未知取值回落第一档：列表与选中的档位由**同一份清单**驱动，不会出现「都不选中」。 + selectedTabIndex = LOG_LEVEL_TABS.indexOfFirst { it.second == selected }.coerceAtLeast(0), + onTabSelected = { index -> onSelect(LOG_LEVEL_TABS[index].second) }, + modifier = Modifier.padding(horizontal = 20.dp), + ) +} + private fun traceBody(span: AdminTraceSpanDto): String = buildList { add("${traceKindLabel(span.kind)} · ${formatNanos(span.durationNs)}") add("开始 ${formatRfc3339(span.startedAt) ?: orDash(span.startedAt)}") diff --git a/app/src/main/kotlin/top/yeij/cyrene/ui/admin/AdminViewModel.kt b/app/src/main/kotlin/top/yeij/cyrene/ui/admin/AdminViewModel.kt index f0774ac..b4795df 100644 --- a/app/src/main/kotlin/top/yeij/cyrene/ui/admin/AdminViewModel.kt +++ b/app/src/main/kotlin/top/yeij/cyrene/ui/admin/AdminViewModel.kt @@ -7,17 +7,24 @@ import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.asStateFlow +import kotlinx.coroutines.flow.update import kotlinx.coroutines.launch import top.yeij.cyrene.app.AppContainer +import top.yeij.cyrene.core.auth.SessionState import top.yeij.cyrene.data.api.dto.AdminBackthinkItemDto import top.yeij.cyrene.data.api.dto.AdminMetricsDto import top.yeij.cyrene.data.api.dto.AdminTraceSpanDto import top.yeij.cyrene.data.api.dto.MemorySummaryDto import top.yeij.cyrene.data.api.dto.PersonaSummaryDto +import top.yeij.cyrene.data.ws.ChatSocket +import top.yeij.cyrene.data.ws.ObservabilityLogEntryDto +import top.yeij.cyrene.data.ws.WsEvent +import top.yeij.cyrene.data.ws.WsState import top.yeij.cyrene.domain.admin.AdminFailure import top.yeij.cyrene.domain.admin.AdminPage import top.yeij.cyrene.domain.admin.AdminRole import top.yeij.cyrene.domain.admin.canAccess +import top.yeij.cyrene.domain.admin.canReceiveObservabilityFrames import top.yeij.cyrene.domain.admin.toAdminFailure /** @@ -35,6 +42,12 @@ data class AdminUiState( data class AdminOverviewData( val metrics: AdminMetricsDto? = null, val trace: List = emptyList(), + /** + * 日志：本端**没有** logs REST 端点（`data/api` 不在本模块改动面内）， + * 所以它**只**来自 `observability_log` 帧累积；一帧都没到过时是空列表。 + * UI 据此**整段不渲染**——不写「暂无日志」，那等于把「没有推送」说成「确实没有日志」。 + */ + val logs: List = emptyList(), ) /** @@ -42,6 +55,23 @@ data class AdminOverviewData( * * [load] 先按 [AdminRole] 短路：角色不够的页**不发请求**，直接给 * [AdminFailure.Forbidden]，UI 因此能显示「无权限」而不是等一个 403 回来。 + * + * ## 可观测实时推送（契约 v0.55 §3.12） + * 概览页在已连接时把 `observability_metrics` / `observability_trace` / `observability_log` + * 三帧合并进 [overview]：**REST 仍是兜底真源**（进页面 / 手动刷新 / 断线重连照旧各拉一次 + * 全量快照），帧只是叠在上面的增量。三帧都只发给 `owner` / `admin`（fail-closed）， + * 客户端按 [canReceiveObservabilityFrames] 自行降级：`viewer` 既不开连接也不合帧， + * 继续走它本就有权读的那两个 REST 端点。 + * + * ## 为什么本类自己开一条 WS 连接 + * `observability_*` 三帧在 WS 上，而 `AppContainer` 只提供 `createChatSocket()` + * （**每次调用新建**，没有进程级共享连接），`ChatSocket` / 聊天侧却都在别的改动范围里。 + * 要在这里收帧，唯一不改别人的办法就是自持一条连接——与 `SessionsViewModel` 同一做法， + * 也**同一笔已知技术债**（「每 VM 一条 socket」，收敛方向是「一条应用级共享 socket + + * 按会话路由」，见 `devdocs/05`；那个重构不在本模块内）。 + * + * 生命周期：登录且角色够档时连接、登出（或角色降档）断开、VM 销毁断开（[onCleared]）。 + * 本类只关心这三个帧类型，收不到也不会让功能失效——REST 会兜住。 */ class AdminViewModel(private val container: AppContainer) : ViewModel() { @@ -57,6 +87,43 @@ class AdminViewModel(private val container: AppContainer) : ViewModel() { private val _backthink = MutableStateFlow(AdminUiState>()) val backthink: StateFlow>> = _backthink.asStateFlow() + /** 本 VM 自持的那条推送连接；见类注释。 */ + private val socket: ChatSocket = container.createChatSocket() + + init { + viewModelScope.launch { + container.authManager.state.collect { auth -> + when (auth) { + SessionState.Unknown -> Unit + + // 登出：断开连接并丢掉**帧来源**的状态（日志）。它会跨账号残留是因为 + // 它没有 REST 兜底——metrics / trace 每次进页面都会被重拉的快照整体替换。 + SessionState.LoggedOut -> { + socket.close() + _overview.update { state -> + state.copy(data = state.data?.copy(logs = emptyList())) + } + } + + // 连接只在「登录 + 角色够档」时建立：viewer 收不到这三帧（服务端 fail-closed）， + // 也就没有必要开一条连接去等。角色可能随重登变化，所以这里两个分支都要显式写。 + is SessionState.LoggedIn -> if (canReceiveObservabilityFrames(AdminRole.of(auth.role))) { + socket.connect() + } else { + socket.close() + } + } + } + } + viewModelScope.launch { socket.events.collect(::onEvent) } + viewModelScope.launch { reloadOnConnect() } + } + + override fun onCleared() { + socket.close() + super.onCleared() + } + /** 进入页面时调用；每次进入都重新取数（监控数据要新鲜）。 */ fun load(page: AdminPage, role: AdminRole) { if (!canAccess(role, page)) { @@ -73,9 +140,18 @@ class AdminViewModel(private val container: AppContainer) : ViewModel() { } } - /** 概览页两个请求并发；任一 403 即整页无权限，其它失败则保留已到的部分。 */ + /** + * 概览页两个请求并发；任一 403 即整页无权限，其它失败则保留已到的部分。 + * + * 帧累积的 [AdminOverviewData.logs] **不参与本次替换**：日志没有 REST 来源， + * 重拉清掉它就等于让它永久消失（帧只送增量，不会把历史再灌一遍）。 + * 用 `update` 原子地取「此刻的日志」，是为了不把**请求在飞期间**到达的日志帧抹掉。 + * + * metrics / trace 则反过来**以本次快照为准**：帧送的是增量，而 REST 是全量（含所有新 span）， + * 竞态下丢掉的那一拍增量会被下一次条目/指标变化补回来，不值得为它引入版本号。 + */ private suspend fun loadOverview() { - _overview.value = _overview.value.copy(loading = true, failure = null) + _overview.update { it.copy(loading = true, failure = null) } val repository = container.adminRepository val (metrics, trace) = coroutineScope { val metricsCall = async { repository.metrics() } @@ -84,13 +160,19 @@ class AdminViewModel(private val container: AppContainer) : ViewModel() { } val failures = listOfNotNull(metrics.exceptionOrNull(), trace.exceptionOrNull()) .map { it.toAdminFailure() } - _overview.value = AdminUiState( - data = AdminOverviewData( - metrics = metrics.getOrNull(), - trace = trace.getOrNull().orEmpty(), - ), - failure = failures.firstOrNull { it is AdminFailure.Forbidden } ?: failures.firstOrNull(), - ) + val freshMetrics = metrics.getOrNull() + val freshTrace = trace.getOrNull().orEmpty() + val failure = failures.firstOrNull { it is AdminFailure.Forbidden } ?: failures.firstOrNull() + _overview.update { state -> + AdminUiState( + data = AdminOverviewData( + metrics = freshMetrics, + trace = freshTrace, + logs = state.data?.logs.orEmpty(), + ), + failure = failure, + ) + } } private suspend fun loadList( @@ -111,4 +193,81 @@ class AdminViewModel(private val container: AppContainer) : ViewModel() { AdminPage.Backthink -> _backthink.value = AdminUiState(failure = AdminFailure.Forbidden) } } + + /** + * 下行帧收敛：只认可观测推送三帧，且**逐帧再判一次角色门**——帧是自己收的， + * 但「连接当时角色够档」不等于「此刻够档」（重登换角色只换了同一实例的连接）。 + * 不够档就整帧丢弃（fail-closed），而不是先渲染再解释。 + */ + private fun onEvent(event: WsEvent) { + if (!canReceiveObservabilityFrames(currentRole())) return + when (event) { + is WsEvent.ObservabilityMetrics -> onMetricsFrame(event) + is WsEvent.ObservabilityTrace -> onTraceFrame(event) + is WsEvent.ObservabilityLog -> onLogFrame(event) + else -> Unit + } + } + + private fun currentRole(): AdminRole = + AdminRole.of((container.authManager.state.value as? SessionState.LoggedIn)?.role) + + /** + * 指标帧 = **整份替换**（载荷就是与 REST 同形的扁平快照本体，服务端只在它变化时才发）。 + * + * 两道闸：帧里没有载荷对象（[WsEvent.ObservabilityMetrics.metrics] 为 null）时跳过， + * 免得把界面上的真实指标写成全零；还没有 REST 基线（[AdminUiState.data] 为 null， + * 即首屏那次引导读取尚未回来）时也不落地——增量只能叠在基线上，凭空造一份 + * 「只有指标、trace 为空」的状态会把「还没拉到」渲染成「没有 trace」。 + */ + private fun onMetricsFrame(event: WsEvent.ObservabilityMetrics) { + val metrics = event.metrics ?: return + _overview.update { state -> + val data = state.data ?: return@update state + state.copy(data = data.copy(metrics = metrics)) + } + } + + /** trace 帧是**增量**：与已有快照按 `id` 去重合并（见 [mergeTraceSpans]）。 */ + private fun onTraceFrame(event: WsEvent.ObservabilityTrace) { + if (event.spans.isEmpty()) return + _overview.update { state -> + val data = state.data ?: return@update state + state.copy(data = data.copy(trace = mergeTraceSpans(data.trace, event.spans))) + } + } + + /** 日志帧是**增量**：与已有条目按业务键去重合并（见 [mergeLogEntries]）。 */ + private fun onLogFrame(event: WsEvent.ObservabilityLog) { + if (event.entries.isEmpty()) return + _overview.update { state -> + val data = state.data ?: return@update state + state.copy(data = data.copy(logs = mergeLogEntries(data.logs, event.entries))) + } + } + + /** + * 每次「连上」（含首次连接与断线重连）都补一次引导读取（REST 兜底）。 + * + * 为什么重连后必须重读：推送是**增量**，总线也不回放——断线窗口里的帧是永久丢了， + * 那份增量流不再可信。而且服务端只为「有在线 owner/admin 连接」的凭据保留订阅 + * （`observability_push.go` 的 `prune`），订阅可能在连接还没上线的窗口里被剪掉、 + * 推送循环随之退出；重读 REST 同时把订阅重新建立起来（契约 §3.12：读取即订阅握手）。 + * 首次连接同样要补，正是为了兜住「页面先读、连接后上」这个顺序。 + * + * 只在**已经拿到过引导数据**之后触发（[AdminUiState.data] 非空）：还没读过的那一次 + * 由页面自己的 `load` 负责，不重复请求。`StateFlow` 只在值变化时发射，所以这里 + * 一次连接只会补一次。 + */ + private suspend fun reloadOnConnect() { + socket.state.collect { state -> + if (state == WsState.Connected) reloadOverview() + } + } + + private fun reloadOverview() { + if (!canReceiveObservabilityFrames(currentRole())) return + if (_overview.value.data == null) return + viewModelScope.launch { loadOverview() } + } } diff --git a/app/src/test/kotlin/top/yeij/cyrene/data/ws/WsEventObservabilityTest.kt b/app/src/test/kotlin/top/yeij/cyrene/data/ws/WsEventObservabilityTest.kt new file mode 100644 index 0000000..20249eb --- /dev/null +++ b/app/src/test/kotlin/top/yeij/cyrene/data/ws/WsEventObservabilityTest.kt @@ -0,0 +1,198 @@ +package top.yeij.cyrene.data.ws + +import org.junit.Assert.assertEquals +import org.junit.Assert.assertNull +import org.junit.Assert.assertTrue +import org.junit.Test + +/** + * 可观测实时推送三帧的解码（契约 v0.55 §3.12，api-client §3.12）。 + * + * 帧形状取自上**后端生产者逐字**（`internal/services/gateway/observability_push.go` + + * `ws_bridge.go` 的 `eventBridge`）：`wsEvent{Type: typ, Data: e.Payload}`——**帧壳不带** + * `sessionId` / `turnId` / `seq`（它们不属任何会话），载荷分别是 + * `AdminLogPush{entries}` / `AdminTracePush{spans}` / `AdminMetrics` **本体**。 + * + * 三条要点在这里钉死： + * 1. 三个帧名与载荷键逐字对齐（`entries` / `spans` / 9 个扁平指标键），任何一条对不上， + * 推送就会安静地落到「未知帧」被丢掉——这正是本模块要修的那个缺口； + * 2. 载荷缺席时**整帧安全忽略**（返回 `null`），绝不回落成默认值：metrics 的默认值是 + * 全零快照，会把界面上的真实指标擦掉； + * 3. 未知帧、坏帧仍然安静丢弃（回归：不能因为接了新帧就把这条约定弄坏）。 + */ +class WsEventObservabilityTest { + + private fun parse(raw: String): WsEvent = + requireNotNull(WsEvent.parse(raw)) { "帧没解出来：$raw" } + + /** 真实 metrics 帧（9 个键，扁平，与 `GET /admin/observability/metrics` 同形）。 */ + private val metricsFrame = """ + {"type":"observability_metrics","data":{"totalRequests":42,"qps":1.5,"avgLatencyMs":3.25, + "maxLatencyMs":88,"perMethod":{"GET":40,"POST":2}, + "perPath":{"/api/v1/x":7},"statusCounts":{"200":7},"llmCalls":3,"subagentEvents":1}} + """.trimIndent().replace("\n", "") + + private val traceFrame = """ + {"type":"observability_trace","data":{"spans":[ + {"id":"sp-2","name":"llm.chat","kind":"llm","status":"ok","startedAt":"2026-10-04T12:00:01Z", + "endedAt":"2026-10-04T12:00:02Z","durationNs":1000000000}, + {"id":"sp-1","name":"tool.run","kind":"tool","status":"error","startedAt":"2026-10-04T12:00:00Z", + "durationNs":500000}]}} + """.trimIndent().replace("\n", "") + + private val logFrame = """ + {"type":"observability_log","data":{"entries":[ + {"time":"2026-10-04T12:00:02Z","level":"warn","source":"gateway","msg":"慢请求"}, + {"time":"2026-10-04T12:00:01Z","level":"info","source":"","msg":"启动完成"}]}} + """.trimIndent().replace("\n", "") + + @Test + fun `the produced metrics frame decodes as the flat REST snapshot`() { + val event = parse(metricsFrame) as WsEvent.ObservabilityMetrics + + assertEquals("observability_metrics", event.type) + val metrics = requireNotNull(event.metrics) + assertEquals(42L, metrics.totalRequests) + assertEquals(1.5, metrics.qps, 0.0) + assertEquals(3.25, metrics.avgLatencyMs, 0.0) + assertEquals(88L, metrics.maxLatencyMs) + assertEquals(40L, metrics.perMethod["GET"]) + assertEquals(7L, metrics.perPath["/api/v1/x"]) + assertEquals(7L, metrics.statusCounts["200"]) + assertEquals(3L, metrics.llmCalls) + assertEquals(1L, metrics.subagentEvents) + // 这三帧不属任何会话：帧壳本来就没有这些字段。 + assertNull(event.sessionId) + assertNull(event.turnId) + assertNull(event.seq) + } + + @Test + fun `a metrics frame without a payload is dropped instead of becoming an all zero snapshot`() { + // 关键：默认值是一份「全零指标」，合并进界面就是把真实数字擦成 0。宁可整帧丢掉。 + assertNull(WsEvent.parse("""{"type":"observability_metrics"}""")) + assertNull(WsEvent.parse("""{"type":"observability_metrics","data":null}""")) + assertNull(WsEvent.parse("""{"type":"observability_metrics","data":"oops"}""")) + } + + @Test + fun `unknown keys inside the metrics payload do not break the frame`() { + val frame = """{"type":"observability_metrics","data":{"totalRequests":9,"tokens":123}}""" + + val event = parse(frame) as WsEvent.ObservabilityMetrics + + assertEquals(9L, requireNotNull(event.metrics).totalRequests) + } + + @Test + fun `the trace frame decodes its spans in the order the server sent them`() { + val event = parse(traceFrame) as WsEvent.ObservabilityTrace + + assertEquals("observability_trace", event.type) + assertEquals(listOf("sp-2", "sp-1"), event.spans.map { it.id }) + val first = event.spans.first() + assertEquals("llm.chat", first.name) + assertEquals("llm", first.kind) + assertEquals("ok", first.status) + assertEquals("2026-10-04T12:00:01Z", first.startedAt) + assertEquals("2026-10-04T12:00:02Z", first.endedAt) + assertEquals(1_000_000_000L, first.durationNs) + assertNull(event.sessionId) + } + + @Test + fun `a trace frame with no spans is an empty delta, not a failure`() { + val event = parse("""{"type":"observability_trace","data":{"spans":[]}}""") + as WsEvent.ObservabilityTrace + + assertTrue(event.spans.isEmpty()) + } + + @Test + fun `a trace frame whose payload is missing is safely ignored`() { + // 空增量的语义是「什么也不改」；而「没有任何载荷」根本不该被当成一帧数据。 + assertNull(WsEvent.parse("""{"type":"observability_trace"}""")) + assertNull(WsEvent.parse("""{"type":"observability_trace","data":null}""")) + } + + @Test + fun `a trace span of the wrong shape drops the frame without throwing`() { + // span 缺 id（DTO 必填）→ 整帧按坏帧忽略。契约里 id 恒在，这里只钉「绝不抛」。 + val frame = """{"type":"observability_trace","data":{"spans":[{"name":"no-id"}]}}""" + + assertNull(WsEvent.parse(frame)) + } + + @Test + fun `the log frame decodes its entries verbatim`() { + val event = parse(logFrame) as WsEvent.ObservabilityLog + + assertEquals("observability_log", event.type) + assertEquals(listOf("慢请求", "启动完成"), event.entries.map { it.msg }) + assertEquals("2026-10-04T12:00:02Z", event.entries.first().time) + assertEquals("warn", event.entries.first().level) + assertEquals("gateway", event.entries.first().source) + // 记录没带来源时是空串（不是 null）——原样保留，不做归一。 + assertEquals("", event.entries[1].source) + assertNull(event.sessionId) + } + + @Test + fun `an unknown log level is kept as is instead of being dropped`() { + // level 是开放集合：未知值原样留着（UI 侧过滤时也保留，见 filterLogEntries）。 + val frame = """ + {"type":"observability_log","data":{"entries":[ + {"time":"2026-10-04T12:00:00Z","level":"trace","source":"x","msg":"未来等级"}]}} + """.trimIndent().replace("\n", "") + + val event = parse(frame) as WsEvent.ObservabilityLog + + assertEquals("trace", event.entries.single().level) + } + + @Test + fun `a log entry with unknown keys still decodes`() { + val frame = """ + {"type":"observability_log","data":{"entries":[ + {"time":"2026-10-04T12:00:00Z","level":"error","source":"x","msg":"炸了","stack":"..."}]}} + """.trimIndent().replace("\n", "") + + val event = parse(frame) as WsEvent.ObservabilityLog + + assertEquals("炸了", event.entries.single().msg) + } + + @Test + fun `a log frame whose payload is missing is safely ignored`() { + assertNull(WsEvent.parse("""{"type":"observability_log"}""")) + assertNull(WsEvent.parse("""{"type":"observability_log","data":null}""")) + } + + @Test + fun `a session on the shell would be passed through, though the server sends none`() { + // 契约说这三帧不带 sessionId；解析层不替服务端做判断，有就原样带上。 + val event = parse("""{"type":"observability_log","sessionId":"web","data":{"entries":[]}}""") + as WsEvent.ObservabilityLog + + assertEquals("web", event.sessionId) + } + + @Test + fun `unknown frames are still ignored quietly`() { + // 回归：本次新增三个帧不能把「未知帧安静忽略」这条约定弄坏 + //（审计 tab 的帧后端至今没做，将来真做了也不该在旧客户端上抛）。 + assertNull(WsEvent.parse("""{"type":"observability_audit","data":{"x":1}}""")) + assertNull(WsEvent.parse("""{"type":"brand.new.frame","data":{}}""")) + assertNull(WsEvent.parse("不是 JSON")) + assertNull(WsEvent.parse("""{"no":"type"}""")) + } + + @Test + fun `the observability frames are not session scoped, so every session accepts them`() { + // 会话闸对它们是空操作：帧壳没有 sessionId（空 = 无法判定 = 放行）。 + // 上层因此只能按「连接身份 + 角色」决定收不收，而不是按会话。 + val event = parse(logFrame) + assertTrue(event.isForSession("web")) + assertTrue(event.isForSession("web:other")) + } +} diff --git a/app/src/test/kotlin/top/yeij/cyrene/domain/admin/AdminAccessTest.kt b/app/src/test/kotlin/top/yeij/cyrene/domain/admin/AdminAccessTest.kt index 7477899..df54df7 100644 --- a/app/src/test/kotlin/top/yeij/cyrene/domain/admin/AdminAccessTest.kt +++ b/app/src/test/kotlin/top/yeij/cyrene/domain/admin/AdminAccessTest.kt @@ -68,4 +68,26 @@ class AdminAccessTest { assertTrue(AdminRole.Viewer.rank < AdminRole.Admin.rank) assertTrue(AdminRole.Admin.rank < AdminRole.Owner.rank) } + + @Test + fun `only owner and admin receive observability frames`() { + // 契约 §3.12：这三个帧不在「状态机信号」白名单里 ⇒ 后端 fail-closed 只发 owner + admin。 + assertTrue(canReceiveObservabilityFrames(AdminRole.Owner)) + assertTrue(canReceiveObservabilityFrames(AdminRole.Admin)) + assertFalse(canReceiveObservabilityFrames(AdminRole.Viewer)) + } + + @Test + fun `being able to open the overview page is not enough for the frames`() { + // 概览页是 Monitor 档（viewer 也能进、也能读那两个 REST），但它收不到推送—— + // 两条门必须分开判，否则 viewer 会开着连接干等一份永远不来的帧。 + assertTrue(canAccess(AdminRole.Viewer, AdminPage.Overview)) + assertFalse(canReceiveObservabilityFrames(AdminRole.Viewer)) + } + + @Test + fun `an unknown role is treated as a viewer for the frames too`() { + assertFalse(canReceiveObservabilityFrames(AdminRole.of("superowner"))) + assertFalse(canReceiveObservabilityFrames(AdminRole.of(null))) + } } diff --git a/app/src/test/kotlin/top/yeij/cyrene/ui/admin/AdminOverviewMergeTest.kt b/app/src/test/kotlin/top/yeij/cyrene/ui/admin/AdminOverviewMergeTest.kt new file mode 100644 index 0000000..9659867 --- /dev/null +++ b/app/src/test/kotlin/top/yeij/cyrene/ui/admin/AdminOverviewMergeTest.kt @@ -0,0 +1,135 @@ +package top.yeij.cyrene.ui.admin + +import org.junit.Assert.assertEquals +import org.junit.Assert.assertSame +import org.junit.Assert.assertTrue +import org.junit.Test +import top.yeij.cyrene.data.api.dto.AdminTraceSpanDto +import top.yeij.cyrene.data.ws.ObservabilityLogEntryDto + +/** + * 可观测推送帧的合并与本地过滤（契约 v0.55 §3.12）的纯函数。 + * + * 断言的是**三条不会写进类型里的语义**：新帧在前（列表两端本来就最新在前，不排序）、 + * 按业务键去重（帧与 REST 竞态时不会出现两条同样的数据）、窗口截断（不许无界增长）； + * 以及日志的等级过滤保留「判不出等级」的条目。 + */ +class AdminOverviewMergeTest { + + private fun span(id: String) = AdminTraceSpanDto(id = id, name = "span-$id", startedAt = "2026-10-04T12:00:00Z") + + private fun entry(msg: String, level: String = "info") = + ObservabilityLogEntryDto(time = "2026-10-04T12:00:00Z", level = level, source = "gateway", msg = msg) + + @Test + fun `an empty delta leaves the snapshot untouched`() { + val existing = listOf(span("a")) + + // 空增量不是「清空」：REST 引导读取拿到的快照必须原样留着。 + assertSame(existing, mergeTraceSpans(existing, emptyList())) + val logs = listOf(entry("启动")) + assertSame(logs, mergeLogEntries(logs, emptyList())) + } + + @Test + fun `new spans go in front of the guide snapshot`() { + val merged = mergeTraceSpans(listOf(span("old")), listOf(span("new"))) + + assertEquals(listOf("new", "old"), merged.map { it.id }) + } + + @Test + fun `a span already present does not appear twice`() { + // 帧与那次 REST 引导读取可能覆盖同一段：按 id 去重，且**新到的那一份**留在前面。 + val merged = mergeTraceSpans(listOf(span("a"), span("b")), listOf(span("b"), span("c"))) + + assertEquals(listOf("b", "c", "a"), merged.map { it.id }) + } + + @Test + fun `the trace window keeps only the newest entries`() { + val existing = (1..5).map { span("s$it") } + + val merged = mergeTraceSpans(existing, listOf(span("n")), window = 3) + + assertEquals(listOf("n", "s1", "s2"), merged.map { it.id }) + } + + @Test + fun `log entries are deduplicated by their whole business key`() { + // 服务端两端都是**最新在前**，合并保留各自的相对顺序（不排序）：帧里新到的在前， + // 已经在列表里的那条不会被重复追加。 + val merged = mergeLogEntries(listOf(entry("启动")), listOf(entry("慢请求"), entry("启动"))) + + assertEquals(listOf("慢请求", "启动"), merged.map { it.msg }) + } + + @Test + fun `log entries differing only in level are distinct`() { + val merged = mergeLogEntries( + listOf(entry("同一句", level = "info")), + listOf(entry("同一句", level = "error")), + ) + + assertEquals(2, merged.size) + } + + @Test + fun `the log window keeps only the newest entries`() { + val existing = (1..4).map { entry("m$it") } + + val merged = mergeLogEntries(existing, listOf(entry("fresh")), window = 2) + + assertEquals(listOf("fresh", "m1"), merged.map { it.msg }) + } + + @Test + fun `no level filter keeps every entry`() { + val entries = listOf(entry("a", "debug"), entry("b", "info"), entry("c", "error")) + + assertSame(entries, filterLogEntries(entries, null)) + assertSame(entries, filterLogEntries(entries, "")) + // 未知的过滤档不猜：退回「不过滤」，而不是把列表过滤空。 + assertSame(entries, filterLogEntries(entries, "verbose")) + } + + @Test + fun `a minimum level keeps that level and above`() { + val entries = listOf(entry("d", "debug"), entry("i", "info"), entry("w", "warn"), entry("e", "error")) + + // 与后端 `?level=` 同口径：是最低等级，不是精确匹配。 + assertEquals(listOf("w", "e"), filterLogEntries(entries, "warn").map { it.msg }) + assertEquals(listOf("e"), filterLogEntries(entries, "error").map { it.msg }) + assertEquals(4, filterLogEntries(entries, "debug").size) + } + + @Test + fun `the level filter tolerates case and blanks and keeps unknown levels`() { + val entries = listOf(entry("w", " WARN "), entry("u", "trace"), entry("i", "info")) + + // 等级是开放集合：判不出高低的条目一律保留，绝不因为「可能低于阈值」被静默藏掉。 + assertEquals(listOf("w", "u"), filterLogEntries(entries, "warn").map { it.msg }) + } + + @Test + fun `log keys distinguish the four fields of an entry`() { + val base = entry("同一句") + val keys = setOf( + logEntryKey(base), + logEntryKey(base.copy(time = "2026-10-04T12:00:01Z")), + logEntryKey(base.copy(level = "warn")), + logEntryKey(base.copy(source = "other")), + logEntryKey(base.copy(msg = "另一句")), + ) + + assertEquals(5, keys.size) + } + + @Test + fun `window constants match what the contract promises`() { + // 契约 §3.12：单帧上限 trace 100 / log 200，客户端保留窗口与它们对齐。 + assertEquals(100, TRACE_WINDOW) + assertEquals(200, LOG_WINDOW) + assertTrue(TRACE_WINDOW > 0) + } +}