merge: W10 — 接上可观测性下行帧(概览页实时刷新 metrics / trace / 日志)

This commit is contained in:
AskaEth committed 2026-10-04 18:42:58 +08:00
commit 93941cd1ea
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,
@@ -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> = AdminPage.entries.filter { canAccess(role, it) }
@@ -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<AdminTraceSpanDto>,
incoming: List<AdminTraceSpanDto>,
window: Int = TRACE_WINDOW,
): List<AdminTraceSpanDto> = mergeNewestFirst(existing, incoming, window) { it.id }
/**
* 合并日志帧:新条目在前,按业务键(时间 + 等级 + 来源 + 消息)去重,保留最近 [window] 条。
*
* 去重键取整条内容,是因为日志**没有 id**(REST `logs` 的条目形状就是这四个字段)。
* 代价如实说:同一秒内、来源与消息完全相同的两条日志会被收成一条——与 Web 面板
* (`observabilityStream.ts` 按业务键合并)同一口径。
*/
internal fun mergeLogEntries(
existing: List<ObservabilityLogEntryDto>,
incoming: List<ObservabilityLogEntryDto>,
window: Int = LOG_WINDOW,
): List<ObservabilityLogEntryDto> = 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<ObservabilityLogEntryDto>,
minLevel: String?,
): List<ObservabilityLogEntryDto> {
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 <T> mergeNewestFirst(
existing: List<T>,
incoming: List<T>,
window: Int,
key: (T) -> String,
): List<T> {
if (incoming.isEmpty()) return existing
val seen = HashSet<String>(incoming.size + existing.size)
val merged = ArrayList<T>(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))
}
@@ -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<String, Boolean>() }
// 日志等级选择(`null` = 全部)。**在本地过滤**:推送帧是广播给所有 owner/admin 连接的,
// 服务端的 `?level=` 对帧不适用(契约 §3.12)。放这里而不是 LazyColumn 的 item 里,
// 是为了让「读状态 → 重算列表」发生在正常的 compose 作用域里。
var logLevel by remember { mutableStateOf<String?>(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<Pair<String, String?>>(
"全部" 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)}")
@@ -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<T>(
data class AdminOverviewData(
val metrics: AdminMetricsDto? = null,
val trace: List<AdminTraceSpanDto> = emptyList(),
/**
* 日志:本端**没有** logs REST 端点(`data/api` 不在本模块改动面内),
* 所以它**只**来自 `observability_log` 帧累积;一帧都没到过时是空列表。
* UI 据此**整段不渲染**——不写「暂无日志」,那等于把「没有推送」说成「确实没有日志」。
*/
val logs: List<ObservabilityLogEntryDto> = 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<List<AdminBackthinkItemDto>>())
val backthink: StateFlow<AdminUiState<List<AdminBackthinkItemDto>>> = _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 <T> 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() }
}
}
@@ -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"))
}
}
@@ -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)))
}
}
@@ -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)
}
}