feat(runtime+admin): A2 LLM异步提取接线 + A4 Tracing接线 + A6通知后端聚合
A2 异步记忆提取:
- loop/memory.go: asyncExtractor接口(ExtractAsync)+extractMemoryAsync(60s超时,WithOutCancel,dispatch即返回,后台落库,失败仅Warn)+persistMemories共用
- run.go 三个回合落定点改异步提取; 删除run.go逐步提取(design/03'每轮后', 避免多轮工具N次LLM调用/重复写)
- cmd/cyrene/extractor.go buildMemoryExtractor: llmGroupUsable为真→LLMExtractor, 否则回落ManualExtractor(日志说明); keylessEndpoint放行loopback/私网
A4 可观测性 Tracing:
- cmd/cyrene/observability.go: newAppTraceWriter+traceFanout(多sink)+gatewaySpanSink(走gw.ObserveSpan,喂环形缓冲+LLM/subagent指标)
- wire.go 注入; 未改trace/gateway数据结构
A6 通知后端:
- internal/services/gateway/notifications.go(311行): GET /admin/notifications?type=&before=&limit= (adminAuth)
- 聚合已有源(obs.audit SQLite + s.backthink), createdAt倒序, id=source:id; 类型派生(channel/system/audit/backthink); 无新建表/无新Deps/wire零改动
- 诚实: backthink仅shouldPush计, 未注入源为空, 报错500, 未读游标后置
- 契约与前端Y一致: {id,type,title,body,source,createdAt}, before含游标(<=), type未知400
- Docs/api-client v0.33
全仓 build/vet/test绿(56包)+win/darwin/linux-arm64交叉编译绿; 新增测试(memory_async/extractor/observability/notifications)
This commit is contained in:
AskaEth committed
1 parent
34c8cc4dca
commit
153a9d6211
12 files changed
+1427
-11
No files matched your search
+12
-2
@@ -67,8 +67,9 @@
|
||||
| 13 | `/api/v1/admin/rag/query` | POST | ✅ | 知识库检索（真实检索；配置向量增强走语义路径） |
|
||||
| 14 | `/api/v1/admin/rag/delete` | POST | ✅(写) | 删除知识库文档（**真实删除**：项目索引条目 + 级联章节）；owner-only，`{id}` → `{id,deleted:true}`，不存在 `404` |
|
||||
| 15 | `/api/v1/admin/observability/metrics` | GET | ✅ | 网关请求指标（总数/QPS/延迟/方法·路径·状态计数/LLM·子代理计数） |
|
||||
| 16 | `/api/v1/admin/observability/trace` | GET | ✅ | 最近 trace span（环形缓冲，默认 512 条） |
|
||||
| 16 | `/api/v1/admin/observability/trace` | GET | ✅ | 最近 trace span（环形缓冲，默认 512 条；**loop/工具/编排的真实 span**） |
|
||||
| 17 | `/api/v1/admin/observability/audit` | GET | ✅ | 审计记录（who/what/when/result，SQLite 持久化） |
|
||||
| 17a | `/api/v1/admin/notifications` | GET | ✅ | **统一通知 feed**（审计 + 主动推送聚合为只读 feed，design/10 §3 #16；见下方「通知聚合契约」） |
|
||||
| 18 | `/api/v1/admin/memories` | POST | ✅(写) | 手动添加记忆（owner-only，审计）；body `{personaId?,content,scope?}`，无 store 回 `501` |
|
||||
| 19 | `/api/v1/admin/rag` | POST | ✅(写) | 导入知识文档（owner-only，审计）；body `{name,desc?,content}`，分块后落项目索引可检索；无 store 回 `501` |
|
||||
| 20 | `/api/v1/admin/system/revoke-keys` | POST | ✅(写) | 吊销**全部** API key（owner-only，审计）；返回 `{revoked:n}`——key 表清空，先前签发的所有 key 立即失效 |
|
||||
@@ -76,7 +77,7 @@
|
||||
| 22 | `/api/v1/admin/system/backup` | POST | ✅(写) | 备份数据（owner-only，审计）；用 SQLite `VACUUM INTO` 快照（含已提交 WAL 内容）到 `<dataDir>/backup-<ts>.db` → `{path,size}` |
|
||||
| 23 | `/api/v1/admin/system/restore` | POST | ✅(写) | 恢复数据（owner-only，审计）；body `{path}` → 校验路径在 dataDir 内→**先备份当前**→替换 DB 文件→`{restored:true,backup,note:"需重启后生效"}`（运行进程持有 DB 句柄，热替换需重启可见） |
|
||||
|
||||
> **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` 仍🕐未实现（前端按契约封装、后端 404 时显示"待实现"壳，不报错）。**写操作已补齐**（见下方「管理面板写操作契约」）：人格/API key/设备控制/自动化/LLM 测试/渠道连接与识别测试/记忆检索/会话导出/插件卸载**真实持久化或真实生效**；**系统设置安全与存储类**（吊销 API key / 轮换 JWT 密钥 / 备份 / 恢复）也已**真实实现**（见下方「系统设置契约」）；仍为后置 M10：**新增用户**、**IoT 场景与规则**、**主人档案保存**、**编辑环境变量/重启服务/检查更新**——后端如实返回 `501 not_implemented`，不伪造成功。（**知识库删除已实现**：`POST /admin/rag/delete` 真实删除，见下方写操作契约；**插件安装已实现**：v1 为 MCP server 热安装。）
|
||||
> **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/notifications` 已实现，返回真实数据；`GET /api/v1/admin/logs` 仍🕐未实现（前端按契约封装、后端 404 时显示"待实现"壳，不报错）。**写操作已补齐**（见下方「管理面板写操作契约」）：人格/API key/设备控制/自动化/LLM 测试/渠道连接与识别测试/记忆检索/会话导出/插件卸载**真实持久化或真实生效**；**系统设置安全与存储类**（吊销 API key / 轮换 JWT 密钥 / 备份 / 恢复）也已**真实实现**（见下方「系统设置契约」）；仍为后置 M10：**新增用户**、**IoT 场景与规则**、**主人档案保存**、**编辑环境变量/重启服务/检查更新**——后端如实返回 `501 not_implemented`，不伪造成功。（**知识库删除已实现**：`POST /admin/rag/delete` 真实删除，见下方写操作契约；**插件安装已实现**：v1 为 MCP server 热安装。）
|
||||
|
||||
**管理面板写操作契约**（`POST/PUT/DELETE /api/v1/admin/*`，均 **owner-only**，读为 owner/admin；写操作经 audit 落 `audit_logs`；成功 `{data}` / 失败 `{error:{code,message}}`）：
|
||||
- `POST /admin/personas` body `{id,name,role,systemPrompt?,characterRef?,modes?/mode?,memoryNamespace?}` → `201 {data:{id,name,character,isDefault,mode,memoryNamespace}}`（注册进 persona registry + 持久化到 `configs/personas/<id>.yaml`）；id 已存在回 `409 duplicate`。
|
||||
@@ -126,6 +127,14 @@
|
||||
- `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 时退化为进程内存）。
|
||||
|
||||
**通知聚合契约**（`GET /api/v1/admin/notifications`，**owner/admin 读**；design/10 §3 #16）：
|
||||
|
||||
- `GET /admin/notifications?type=&before=&limit=` → `200 {"data":[{id,type,title,body?,source,createdAt}]}`，按 `createdAt` **倒序**（同毫秒按 `id` 倒序稳定）。
|
||||
- **查询参数**：`type` ∈ `system|channel|backthink|audit`（**空/缺省 = 全部**，未知值回 `400 invalid_type`）；`limit` 缺省 `50`、上限 `200`；`before` 为 `createdAt`（epoch 毫秒）游标，**含游标本身**（`createdAt <= before`——两源时间戳精度不同（审计秒/纳秒、backthink 毫秒），页边界可能落在同一毫秒，**客户端按 `id` 去重**）。缺省/非法 `before` = 从最新开始。
|
||||
- **字段**：`id` = `<source>:<原生 id>`（`audit:<audit_logs.id>` / `backthink:<backthink_records.id>`），**跨页稳定、可去重**；`source` ∈ `audit|backthink`；`createdAt` 为 epoch 毫秒（无法解析的审计时间戳记 `0`，不伪造时间）。`title`/`body` 由各自源派生：审计项 `title = what`（如 `channel.open`）、`body = "<who> · <result>"`；backthink 项 `title = "主动推送（<trigger> · 已送达|未送达）"`、`body =` 结论摘要（≤200 rune，超出加 `…`）。
|
||||
- **类型归类**（审计源按操作前缀派生）：`channel.*` → `channel`；`system.*` / `plugin.*` / `policy.*` → `system`；其余审计操作（persona/user/memory/rag/session/upload/audio/…）→ `audit`；backthink 主动推送 → `backthink`。
|
||||
- **数据源与诚实边界**：**只聚合现有源**（审计 `audit_logs` + backthink 记录），**不新建 notifications 表**。backthink 仅 `shouldPush=true` 的记录算通知（纯思考记录不推给主人）；**未注入的源 → 该类型为空，不伪造**（无 backthink 服务即无 `backthink` 项）；源报错回 `500 notifications_failed`（**不静默少给**）。本版为**只读 feed**：未读/已读游标后置（不建重表）。
|
||||
|
||||
## 3. WebSocket `/ws/chat`（实时）
|
||||
|
||||
- **握手**: `ws://<host>:8080/ws/chat?token=<JWT>` 或 `Authorization: Bearer`；未认证在握手前回 `401` JSON。
|
||||
@@ -231,6 +240,7 @@
|
||||
- 语音：上行 `voice_input`（base64 音频或已转写 text），下行 `voice_output`（合成音频 base64 + text）。
|
||||
|
||||
## 5. 变更记录
|
||||
- `v0.33`（**A2 记忆异步提取 / A4 可观测性 Tracing 接线 / A6 通知页后端**）：① **A6 新增 `GET /api/v1/admin/notifications`**（owner/admin 读，**全新端点**）：把**现有源**（审计 `audit_logs` + backthink 主动推送记录）聚合为**统一只读通知 feed**，`type`/`before`/`limit` 过滤与游标见 §2「通知聚合契约」；**不新建表**，未注入的源该类型为空、不伪造；未读游标后置。② **A4 trace 端点由恒空变真实数据**（**无契约变更**，`GET /admin/observability/trace` 与 `metrics` 的 `llmCalls`/`subagentEvents` 从此反映 loop/工具/编排的真实 span）：此前 `TraceSink` 从未接线、环形缓冲无人喂；现在组合根把 loop 产生的 span 扇出到网关观测缓冲。响应字段/形状不变。③ **A2 LLM 异步记忆提取接线**（**对客户端无变更**）：loop 回合结束后改由 LLM 提取器**后台异步**抽取记忆（不阻塞回复、失败仅记日志），**群聊/陌生人仍只读不写**；无可用模型凭据时回落确定性 `ManualExtractor`（`记住 X` 路径不变）。
|
||||
- `v0.32`（**A1 情绪系统接线：情绪/好感度真跑起来**）：后端把既有的 `persona/emotion` 状态机接进 loop 回合边界——每轮结束按该轮用户消息的情绪信号移动**该角色自己**的情绪状态机（沿配置转移表走一步）并调整好感度（按角色 `affection.delta`）；无信号轮次衰减一步趋向 `calm`。① **消息 `mood` 不再恒空**：assistant 消息落库带该轮情绪，`message_persisted` / `history` 回放 / `GET /sessions/{id}/messages` 的 `mood` 为真实值（user 消息不带）。② **新增下行 `mood` 事件**（`{data:{mood,personaId,affection}}`，**仅情绪变化时**广播，见 §3.7），前端顶栏「心情 · x」与消息气泡接真数据。③ **隔离**：群聊/陌生人（`MemWritable=false`）情绪只读不写——不改情绪/好感度、不发事件、该轮 `mood` 空。④ **持久性**：情绪/好感度为进程内按角色状态，重启回 `initialState`（未落库；design/01 未要求）。**触发词表与「每轮一步」为 design/01 未规定处的最小实现，待主人确认**。**契约新增（向后兼容）：`mood` 事件 + 消息 `mood` 变真实值。**
|
||||
- `v0.31`（**Android 协同批次 ③b：B5 剩余 501 + D1 推送出口钩子**）：① **B5-知识库删除由 501 改为真实实现**：`POST /api/v1/admin/rag/delete` body `{"id"}` → `200 {data:{id,deleted:true}}`——从项目索引删除该文档条目并**级联删除全部章节**（单事务）；不存在回 `404 not_found`，`id` 空回 `400`，未注入 store 回 `501`；路由由 admin 收紧为 **owner-only**（与其它 admin 写一致）+ 审计 `rag.delete`。② **B5-新增用户 / IoT 场景 / IoT 规则仍诚实 501**：无用户表（登录仅 `.env` 单 admin）、IoT 无场景/规则模型与存储（`iot.Provider` 只有设备列表/控制/状态），错误消息明确缺什么，**不伪造 store/假成功**。③ **D1 推送出口钩子**：主动消息在 WS/IM 投递之外异步镜像到外部推送出口，**自建统一推送服务优先 → 个推(Getui)兜底**（个推 SDK 未接入，仅预留 `GETUI_*` 字段，不可达时日志注明待接入）；`.env` 新增 `PUSH_EXPORTER_ENABLED/URL/TOKEN/TIMEOUT_SEC`，token 只走 `Authorization` 头、不落日志；**对客户端无新端点**（详见 §3.6）。
|
||||
- `v0.30`（**Android 协同批次 ③a：D2 消息搜索 / D3 消息编辑删除 / D5 会话置顶**）：① **D2**：`GET /api/v1/sessions/{id}/messages` 新增可选 **`q`**——**会话内**内容搜索（大小写不敏感 ASCII LIKE，`%`/`_`/`\` 转义为字面量），仍倒序 + `before`/`limit` 游标；`q` 与 `after` 互斥（同传 `400`），空 `q` 走普通分页；**严格限定单会话，不新增跨会话搜索**（避免群聊/陌生人隔离会话外泄，红线）。`HistoryStore` 新增 `Search`。② **D3**：新增 `PATCH /api/v1/messages/{id}`（body `{content}`，非空 ≤4000 rune → `{data:{message}}`，只改 content）与 `DELETE /api/v1/messages/{id}`（→ `{data:{deleted:id}}`）；不存在回 `404 not_found`；`HistoryStore` 新增 `UpdateMessage`/`DeleteMessage` + `ErrMessageNotFound`。**不新增重发端点**（客户端用编辑后的 content 再发）。③ **D5**：`chat_sessions` 新增 `pinned INTEGER NOT NULL DEFAULT 0`（迁移 **0009**，down 可逆）；`GET /sessions` 排序改为 **`pinned DESC` → 主会话优先 → `createdAt ASC`**，会话 DTO 新增恒返回的 **`pinned`** bool；`PATCH /sessions/{id}` 扩展为 body `{title?,pinned?}`（至少一个字段，只回显变更字段，title-only 保持 A4 形状）；`HistoryStore` 新增 `SetPinned`。全部为 owner/admin 写、读消息 owner/admin；未注入 HistoryStore 回 `501`。
|
||||
|
||||
@@ -0,0 +1,78 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"net"
|
||||
"net/url"
|
||||
"strings"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||
)
|
||||
|
||||
// memoryOwnerRoleID is the role id the composition root authors extracted
|
||||
// memories with; the owner is the only actor whose private memory is written
|
||||
// (group/stranger conversations never extract at all, design/11 §2).
|
||||
const memoryOwnerRoleID = "owner"
|
||||
|
||||
// buildMemoryExtractor selects the memory extractor the loop runs after each
|
||||
// settled turn (design/03 §1 LLM 异步自动提取):
|
||||
//
|
||||
// - the LLM extractor when the provider stack has a usable active group, so
|
||||
// the model denoises and deduplicates what is worth remembering;
|
||||
// - the deterministic ManualExtractor otherwise, so a keyless deployment still
|
||||
// records the owner's explicit "记住 X" directives instead of failing a model
|
||||
// call on every turn.
|
||||
//
|
||||
// The loop detaches the LLM extractor to a background goroutine (ExtractAsync);
|
||||
// see internal/runtime/loop/memory.go.
|
||||
func (a *app) buildMemoryExtractor(personaID string) memory.Extractor {
|
||||
if llmGroupUsable(a.llm) {
|
||||
a.log.Info("memory extractor: llm", "model", a.llm.Model(), "provider", a.llm.Provider())
|
||||
return memory.NewLLMExtractor(a.llm, personaID, memoryOwnerRoleID)
|
||||
}
|
||||
a.log.Warn("memory extractor: manual (no usable llm model group)")
|
||||
return memory.NewManualExtractor(personaID, memoryOwnerRoleID)
|
||||
}
|
||||
|
||||
// llmGroupUsable reports whether the active model group of the runtime stack can
|
||||
// actually be called: a model and a base URL are mandatory, and either an API key
|
||||
// is configured or the endpoint is a local/private address, where a keyless call
|
||||
// is the normal case (a self-hosted OpenAI-compatible server).
|
||||
//
|
||||
// It is deliberately a cheap structural check, never a probe request: boot must
|
||||
// not spend a model call. A group that looks usable but fails at call time only
|
||||
// logs an extraction error (extraction failures are never fatal).
|
||||
func llmGroupUsable(llm *model.RuntimeLLM) bool {
|
||||
if llm == nil {
|
||||
return false
|
||||
}
|
||||
group := llm.Group()
|
||||
if strings.TrimSpace(group.Model) == "" || strings.TrimSpace(group.BaseURL) == "" {
|
||||
return false
|
||||
}
|
||||
if strings.TrimSpace(group.APIKey) != "" {
|
||||
return true
|
||||
}
|
||||
return keylessEndpoint(group.BaseURL)
|
||||
}
|
||||
|
||||
// keylessEndpoint reports whether baseURL points at a loopback or private-network
|
||||
// host, where a missing API key is expected rather than a misconfiguration.
|
||||
func keylessEndpoint(baseURL string) bool {
|
||||
parsed, err := url.Parse(strings.TrimSpace(baseURL))
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
host := parsed.Hostname()
|
||||
if host == "" {
|
||||
return false
|
||||
}
|
||||
if strings.EqualFold(host, "localhost") {
|
||||
return true
|
||||
}
|
||||
ip := net.ParseIP(host)
|
||||
if ip == nil {
|
||||
return false
|
||||
}
|
||||
return ip.IsLoopback() || ip.IsPrivate()
|
||||
}
|
||||
@@ -0,0 +1,79 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/config"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory"
|
||||
modelpkg "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||
)
|
||||
|
||||
// testRuntimeLLM builds a provider stack from env alone (the pointed-at
|
||||
// models.json does not exist), so a test controls the active group through the
|
||||
// flat LLM config exactly as a keyless .env boot would.
|
||||
func testRuntimeLLM(t *testing.T, env config.LLMConfig) *modelpkg.RuntimeLLM {
|
||||
t.Helper()
|
||||
return modelpkg.NewRuntimeLLM(filepath.Join(t.TempDir(), "models.json"), env, testLog())
|
||||
}
|
||||
|
||||
// TestBuildMemoryExtractor_LLMWhenCredentials proves a configured provider group
|
||||
// selects the LLM extractor (design/03 §1).
|
||||
func TestBuildMemoryExtractor_LLMWhenCredentials(t *testing.T) {
|
||||
t.Parallel()
|
||||
a := &app{
|
||||
log: testLog(),
|
||||
llm: testRuntimeLLM(t, config.LLMConfig{
|
||||
BaseURL: "https://api.example.com/v1",
|
||||
APIKey: "sk-test",
|
||||
Model: "gpt-4o-mini",
|
||||
}),
|
||||
}
|
||||
if _, ok := a.buildMemoryExtractor("persona-a").(*memory.LLMExtractor); !ok {
|
||||
t.Fatalf("extractor = %T, want *memory.LLMExtractor (credentials configured)", a.buildMemoryExtractor("persona-a"))
|
||||
}
|
||||
}
|
||||
|
||||
// TestBuildMemoryExtractor_ManualWhenNoCredentials proves the keyless fallback: a
|
||||
// boot without an API key (and without a local endpoint) keeps the deterministic
|
||||
// ManualExtractor instead of firing a doomed model call on every turn.
|
||||
func TestBuildMemoryExtractor_ManualWhenNoCredentials(t *testing.T) {
|
||||
t.Parallel()
|
||||
a := &app{log: testLog(), llm: testRuntimeLLM(t, config.LLMConfig{})}
|
||||
if _, ok := a.buildMemoryExtractor("persona-a").(*memory.ManualExtractor); !ok {
|
||||
t.Fatalf("extractor = %T, want *memory.ManualExtractor (no usable llm group)", a.buildMemoryExtractor("persona-a"))
|
||||
}
|
||||
|
||||
bare := &app{log: testLog()}
|
||||
if _, ok := bare.buildMemoryExtractor("persona-a").(*memory.ManualExtractor); !ok {
|
||||
t.Fatalf("extractor = %T, want *memory.ManualExtractor (nil provider stack)", bare.buildMemoryExtractor("persona-a"))
|
||||
}
|
||||
}
|
||||
|
||||
// TestLLMGroupUsable_KeylessLocalEndpoint locks the ollama-style case: a local
|
||||
// OpenAI-compatible server needs no API key, so extraction stays LLM-backed.
|
||||
func TestLLMGroupUsable_KeylessLocalEndpoint(t *testing.T) {
|
||||
t.Parallel()
|
||||
cases := []struct {
|
||||
name string
|
||||
env config.LLMConfig
|
||||
ready bool
|
||||
}{
|
||||
{name: "api key", env: config.LLMConfig{BaseURL: "https://api.example.com/v1", APIKey: "sk", Model: "m"}, ready: true},
|
||||
{name: "loopback keyless", env: config.LLMConfig{BaseURL: "http://127.0.0.1:11434/v1", Model: "llama3"}, ready: true},
|
||||
{name: "localhost keyless", env: config.LLMConfig{BaseURL: "http://localhost:1234/v1", Model: "m"}, ready: true},
|
||||
{name: "private keyless", env: config.LLMConfig{BaseURL: "http://192.168.1.20:11434/v1", Model: "m"}, ready: true},
|
||||
{name: "remote keyless", env: config.LLMConfig{BaseURL: "https://api.example.com/v1", Model: "m"}, ready: false},
|
||||
{name: "no model", env: config.LLMConfig{BaseURL: "https://api.example.com/v1", APIKey: "sk"}, ready: false},
|
||||
{name: "no base url", env: config.LLMConfig{APIKey: "sk", Model: "m"}, ready: false},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
tc := tc
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
if got := llmGroupUsable(testRuntimeLLM(t, tc.env)); got != tc.ready {
|
||||
t.Fatalf("llmGroupUsable = %v, want %v", got, tc.ready)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,98 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
"sync"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/services/gateway"
|
||||
)
|
||||
|
||||
// newAppTraceWriter builds the composition root's trace writer: the JSONL stderr
|
||||
// recorder (the local structured log of every span) plus room for the gateway's
|
||||
// observability sink, which can only be registered once the gateway exists (see
|
||||
// wireGatewayTraceSink).
|
||||
func newAppTraceWriter() *traceFanout {
|
||||
return newTraceFanout(trace.NewJSONRecorder(os.Stderr))
|
||||
}
|
||||
|
||||
// wireGatewayTraceSink registers the gateway's observability sink on the app's
|
||||
// trace fanout. It is called from wireFrontend — the first moment the gateway
|
||||
// exists — because the gateway is built after the loop that emits the spans.
|
||||
// Without it the admin trace ring buffer is never fed and
|
||||
// GET /api/v1/admin/observability/trace answers an empty list forever
|
||||
// (design/19 §2). It is a no-op on a boot without a tracer or gateway.
|
||||
func wireGatewayTraceSink(a *app, gw *gateway.Server) {
|
||||
if a.traceWriter == nil || gw == nil {
|
||||
return
|
||||
}
|
||||
a.traceWriter.Add(gatewaySpanSink{srv: gw})
|
||||
a.log.Info("observability trace sink wired")
|
||||
}
|
||||
|
||||
// traceFanout is the composition root's trace.Writer. It fans every completed
|
||||
// span out to the sinks installed at boot (the JSONL stderr recorder) plus any
|
||||
// sink registered later in the boot sequence: the gateway's observability buffer
|
||||
// only exists once wireFrontend has built the gateway, while the loop that emits
|
||||
// the spans is built earlier in wireRuntime.
|
||||
//
|
||||
// It is safe for concurrent use — the loop finishes spans from several
|
||||
// goroutines (per-step, per-tool-call, sub-agents).
|
||||
type traceFanout struct {
|
||||
mu sync.RWMutex
|
||||
writers []trace.Writer
|
||||
}
|
||||
|
||||
// newTraceFanout builds a fanout over the given writers. Nil writers are dropped.
|
||||
func newTraceFanout(writers ...trace.Writer) *traceFanout {
|
||||
f := &traceFanout{}
|
||||
for _, w := range writers {
|
||||
f.Add(w)
|
||||
}
|
||||
return f
|
||||
}
|
||||
|
||||
// Add installs one more sink. A nil writer is ignored.
|
||||
func (f *traceFanout) Add(w trace.Writer) {
|
||||
if w == nil {
|
||||
return
|
||||
}
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.writers = append(f.writers, w)
|
||||
}
|
||||
|
||||
// WriteSpan writes rec to every sink and returns the first error. Every sink is
|
||||
// tried even after a failure, so one broken sink cannot hide a span from the
|
||||
// others; the tracer only logs the returned error (trace.Tracer.emit).
|
||||
func (f *traceFanout) WriteSpan(rec trace.SpanRecord) error {
|
||||
f.mu.RLock()
|
||||
defer f.mu.RUnlock()
|
||||
var firstErr error
|
||||
for _, w := range f.writers {
|
||||
if w == nil {
|
||||
continue
|
||||
}
|
||||
if err := w.WriteSpan(rec); err != nil && firstErr == nil {
|
||||
firstErr = fmt.Errorf("trace fanout: %w", err)
|
||||
}
|
||||
}
|
||||
return firstErr
|
||||
}
|
||||
|
||||
// gatewaySpanSink adapts *gateway.Server to trace.Writer through its ObserveSpan
|
||||
// entry point, so an emitted span lands in the admin panel's trace ring buffer
|
||||
// AND in the LLM/subagent metric counters (ObserveSpan appends to the buffer and
|
||||
// observes the metrics; the buffer is what GET /api/v1/admin/observability/trace
|
||||
// serves).
|
||||
type gatewaySpanSink struct{ srv *gateway.Server }
|
||||
|
||||
// WriteSpan feeds rec to the gateway. A missing server is a no-op.
|
||||
func (g gatewaySpanSink) WriteSpan(rec trace.SpanRecord) error {
|
||||
if g.srv == nil {
|
||||
return nil
|
||||
}
|
||||
g.srv.ObserveSpan(rec)
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,164 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/config"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/services/gateway"
|
||||
)
|
||||
|
||||
const obsTestSecret = "0123456789abcdef0123456789abcdef"
|
||||
|
||||
// recordingWriter captures the spans a fanout hands it.
|
||||
type recordingWriter struct {
|
||||
spans []trace.SpanRecord
|
||||
}
|
||||
|
||||
func (r *recordingWriter) WriteSpan(rec trace.SpanRecord) error {
|
||||
r.spans = append(r.spans, rec)
|
||||
return nil
|
||||
}
|
||||
|
||||
// TestTraceFanout_LateSinkReceivesSpans proves the composition root can register
|
||||
// a sink after the loop's tracer is already running: spans finished before the
|
||||
// sink is added do not reach it, every span after does (the gateway is built
|
||||
// after the loop, so this ordering is what A4 depends on).
|
||||
func TestTraceFanout_LateSinkReceivesSpans(t *testing.T) {
|
||||
t.Parallel()
|
||||
early := &recordingWriter{}
|
||||
fanout := newTraceFanout(early)
|
||||
tracer := trace.New(fanout)
|
||||
|
||||
_, first := tracer.StartSpan(context.Background(), "before", trace.SpanKindStateTransition)
|
||||
first.Finish(trace.SpanStatusOK)
|
||||
|
||||
late := &recordingWriter{}
|
||||
fanout.Add(late)
|
||||
|
||||
_, second := tracer.StartSpan(context.Background(), "after", trace.SpanKindLLM)
|
||||
second.Finish(trace.SpanStatusOK)
|
||||
|
||||
if len(early.spans) != 2 {
|
||||
t.Fatalf("early sink spans = %d, want 2", len(early.spans))
|
||||
}
|
||||
if len(late.spans) != 1 || late.spans[0].ID != second.ID {
|
||||
t.Fatalf("late sink spans = %+v, want exactly the span %q", late.spans, second.ID)
|
||||
}
|
||||
}
|
||||
|
||||
// TestWireGatewayTraceSink_Guards proves the boot helper tolerates a missing
|
||||
// tracer or gateway, so a degraded boot cannot panic on the observability path.
|
||||
func TestWireGatewayTraceSink_Guards(t *testing.T) {
|
||||
t.Parallel()
|
||||
wireGatewayTraceSink(&app{log: testLog()}, nil)
|
||||
a := &app{log: testLog(), traceWriter: newAppTraceWriter()}
|
||||
wireGatewayTraceSink(a, nil)
|
||||
if len(a.traceWriter.writers) != 1 {
|
||||
t.Fatalf("writers = %d, want 1 (the stderr recorder only)", len(a.traceWriter.writers))
|
||||
}
|
||||
}
|
||||
|
||||
// TestGatewaySpanSink_FeedsTraceEndpointAndMetrics is the A4 acceptance: a span
|
||||
// emitted by the tracer is written through the gateway sink (registered by the
|
||||
// same helpers wire.go calls) and surfaces both in
|
||||
// GET /api/v1/admin/observability/trace (a real span, not an empty buffer) and in
|
||||
// the LLM metric counter.
|
||||
func TestGatewaySpanSink_FeedsTraceEndpointAndMetrics(t *testing.T) {
|
||||
t.Parallel()
|
||||
jwt := auth.New(config.AuthConfig{JWTSecret: obsTestSecret})
|
||||
gw, err := gateway.New(&config.Config{}, gateway.Deps{Log: testLog(), Auth: jwt})
|
||||
if err != nil {
|
||||
t.Fatalf("gateway.New: %v", err)
|
||||
}
|
||||
t.Cleanup(gw.Close)
|
||||
ts := httptest.NewServer(gw.Handler())
|
||||
t.Cleanup(ts.Close)
|
||||
|
||||
// Exactly the boot path: the app's fanout is built in wireRuntime and the
|
||||
// gateway sink is registered in wireFrontend.
|
||||
a := &app{log: testLog(), traceWriter: newAppTraceWriter()}
|
||||
wireGatewayTraceSink(a, gw)
|
||||
tracer := trace.New(a.traceWriter)
|
||||
|
||||
ctx := context.Background()
|
||||
_, llmSpan := tracer.StartSpan(ctx, "llm.chat", trace.SpanKindLLM)
|
||||
llmSpan.Finish(trace.SpanStatusOK)
|
||||
_, stepSpan := tracer.StartSpan(ctx, "loop.step", trace.SpanKindStateTransition)
|
||||
_ = stepSpan.SetProp("session", "web")
|
||||
stepSpan.Finish(trace.SpanStatusOK)
|
||||
|
||||
tok, err := jwt.Sign("acct", auth.RoleOwner, time.Hour)
|
||||
if err != nil {
|
||||
t.Fatalf("sign token: %v", err)
|
||||
}
|
||||
|
||||
var traceBody struct {
|
||||
Data []struct {
|
||||
ID string `json:"id"`
|
||||
Name string `json:"name"`
|
||||
Kind string `json:"kind"`
|
||||
} `json:"data"`
|
||||
}
|
||||
status, raw := obsGet(t, ts.URL+"/api/v1/admin/observability/trace", tok)
|
||||
if status != http.StatusOK {
|
||||
t.Fatalf("trace status = %d, want 200 (body=%s)", status, raw)
|
||||
}
|
||||
if err := json.Unmarshal(raw, &traceBody); err != nil {
|
||||
t.Fatalf("decode trace: %v (body=%s)", err, raw)
|
||||
}
|
||||
seen := map[string]string{}
|
||||
for _, sp := range traceBody.Data {
|
||||
seen[sp.ID] = sp.Kind
|
||||
}
|
||||
if seen[stepSpan.ID] != string(trace.SpanKindStateTransition) {
|
||||
t.Fatalf("trace endpoint = %+v, want the loop.step span %q", traceBody.Data, stepSpan.ID)
|
||||
}
|
||||
if seen[llmSpan.ID] != string(trace.SpanKindLLM) {
|
||||
t.Fatalf("trace endpoint = %+v, want the llm span %q", traceBody.Data, llmSpan.ID)
|
||||
}
|
||||
|
||||
var metricsBody struct {
|
||||
Data struct {
|
||||
LLMCalls int64 `json:"llmCalls"`
|
||||
} `json:"data"`
|
||||
}
|
||||
status, raw = obsGet(t, ts.URL+"/api/v1/admin/observability/metrics", tok)
|
||||
if status != http.StatusOK {
|
||||
t.Fatalf("metrics status = %d, want 200 (body=%s)", status, raw)
|
||||
}
|
||||
if err := json.Unmarshal(raw, &metricsBody); err != nil {
|
||||
t.Fatalf("decode metrics: %v (body=%s)", err, raw)
|
||||
}
|
||||
if metricsBody.Data.LLMCalls != 1 {
|
||||
t.Fatalf("llmCalls = %d, want 1 (ObserveSpan counts the emitted llm span)", metricsBody.Data.LLMCalls)
|
||||
}
|
||||
}
|
||||
|
||||
// obsGet performs an authenticated owner GET and returns the status plus the raw
|
||||
// body.
|
||||
func obsGet(t *testing.T, url, token string) (int, []byte) {
|
||||
t.Helper()
|
||||
req, err := http.NewRequest(http.MethodGet, url, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("new request: %v", err)
|
||||
}
|
||||
req.Header.Set("Authorization", "Bearer "+token)
|
||||
resp, err := http.DefaultClient.Do(req)
|
||||
if err != nil {
|
||||
t.Fatalf("GET %s: %v", url, err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
raw, err := io.ReadAll(resp.Body)
|
||||
if err != nil {
|
||||
t.Fatalf("read %s: %v", url, err)
|
||||
}
|
||||
return resp.StatusCode, raw
|
||||
}
|
||||
+14
-3
@@ -6,7 +6,6 @@ import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
@@ -117,6 +116,11 @@ type app struct {
|
||||
// satisfied by *gateway.Server (BroadcastMood); held as an interface so the
|
||||
// composition root is unit-testable with a fake sink.
|
||||
moodSink moodEmitter
|
||||
// traceWriter fans finished trace spans out to the JSONL stderr recorder and,
|
||||
// once the gateway exists, to the gateway's observability buffer behind
|
||||
// GET /api/v1/admin/observability/trace (design/19 §2). It is held so
|
||||
// wireFrontend can register the gateway sink on the already-running tracer.
|
||||
traceWriter *traceFanout
|
||||
}
|
||||
|
||||
// capacityEmitter is the gateway-facing sink for capacity_awaken broadcasts.
|
||||
@@ -360,7 +364,10 @@ func (a *app) wireRuntime() error {
|
||||
personaID := a.defaultPersonaID()
|
||||
store := a.memStore
|
||||
|
||||
tracer := trace.New(trace.NewJSONRecorder(os.Stderr))
|
||||
// Spans fan out to the JSONL stderr recorder and — once wireFrontend has built
|
||||
// the gateway — to its observability buffer (cmd/cyrene/observability.go).
|
||||
a.traceWriter = newAppTraceWriter()
|
||||
tracer := trace.New(a.traceWriter)
|
||||
// Realtime flow visualization (design/19 §1): the publisher accumulates the
|
||||
// loop's trace spans into a per-session flow graph and publishes
|
||||
// ws.graph_update on each span transition. It is gated to the owner's web
|
||||
@@ -413,7 +420,7 @@ func (a *app) wireRuntime() error {
|
||||
OnTurnEnd: a.onTurnEnd,
|
||||
Memory: store,
|
||||
Retriever: a.retriever,
|
||||
Extractor: memory.NewManualExtractor(personaID, "owner"),
|
||||
Extractor: a.buildMemoryExtractor(personaID),
|
||||
Checkpoint: checkpoint,
|
||||
Tracer: tracer,
|
||||
Flow: flowPub,
|
||||
@@ -738,6 +745,10 @@ func (a *app) wireFrontend() error {
|
||||
}
|
||||
gw.SetVoice(a.voice)
|
||||
a.gw = gw // assign BEFORE wiring hooks below, which capture a.gw / gw (nil-receiver panic otherwise)
|
||||
// Observability trace sink (design/19 §2): now that the gateway exists, its
|
||||
// ring buffer must be fed by the loop's spans, or the admin trace/metrics
|
||||
// endpoints stay empty.
|
||||
wireGatewayTraceSink(a, gw)
|
||||
if a.channelBridge != nil {
|
||||
// Route the channel bridge's owner-private persisted messages into the same
|
||||
// `message_persisted` broadcast the orchestrator uses, so the Web client mirrors
|
||||
|
||||
@@ -3,11 +3,26 @@ package loop
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||
)
|
||||
|
||||
// memoryExtractTimeout bounds one background extraction. It only governs the
|
||||
// detached extraction context, so a hung model call cannot leak a goroutine for
|
||||
// the process lifetime.
|
||||
const memoryExtractTimeout = 60 * time.Second
|
||||
|
||||
// asyncExtractor is the extractor capability the loop detaches to a background
|
||||
// goroutine: memory.LLMExtractor implements it through ExtractAsync (design/03 §1
|
||||
// LLM 异步自动提取), returning a channel that delivers exactly one result.
|
||||
// Deterministic extractors (memory.ManualExtractor) do not implement it and stay
|
||||
// on the synchronous path.
|
||||
type asyncExtractor interface {
|
||||
ExtractAsync(ctx context.Context, conversation []model.Message) (<-chan memory.AsyncResult, error)
|
||||
}
|
||||
|
||||
// injectMemory retrieves the persona's relevant long-term memories and prepends
|
||||
// them as a leading system message, so the model can ground its answer in recalled
|
||||
// context (design §6 / §2). It is re-run every round via assembleContext so the
|
||||
@@ -157,6 +172,66 @@ func (l *Loop) extractMemory(ctx context.Context, messages []model.Message, step
|
||||
l.log.Warn("memory extract", "step", step, "error", err)
|
||||
return
|
||||
}
|
||||
l.persistMemories(ctx, items, step)
|
||||
}
|
||||
|
||||
// extractMemoryAsync triggers the extraction hook at the turn boundary without
|
||||
// blocking the reply (design/03 §1: 每轮对话后后台异步抽取). When the configured
|
||||
// extractor can extract asynchronously (the LLM extractor) the conversation is
|
||||
// handed to a background goroutine and this call returns immediately, so a slow
|
||||
// extraction model never delays the response. A deterministic extractor
|
||||
// (ManualExtractor) does not implement asyncExtractor and is run inline, which
|
||||
// keeps its cheap "记住 X" path observable and unchanged.
|
||||
//
|
||||
// The same gates as extractMemory apply: a restricted conversation is never
|
||||
// extracted (read-only isolation, design/02 §3 / design/11 §2), and a missing
|
||||
// extractor or store is a no-op. Extraction failures and store failures are
|
||||
// logged with the loop's source and never propagated.
|
||||
func (l *Loop) extractMemoryAsync(ctx context.Context, messages []model.Message, step int) {
|
||||
env := sessionEnvFrom(ctx)
|
||||
if env.Restricted() {
|
||||
l.log.Debug("memory extract skipped: read-only context", "step", step, "env", env.ActorEnv)
|
||||
return
|
||||
}
|
||||
if l.extractor == nil || l.memory == nil {
|
||||
return
|
||||
}
|
||||
async, ok := l.extractor.(asyncExtractor)
|
||||
if !ok {
|
||||
l.extractMemory(ctx, messages, step)
|
||||
return
|
||||
}
|
||||
// The turn's context dies with the request while the extraction deliberately
|
||||
// outlives it, so the goroutine gets a detached copy of the context (values —
|
||||
// the session env among them — are kept) under its own timeout.
|
||||
dctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), memoryExtractTimeout)
|
||||
// The caller keeps appending to messages as the turn continues, so the
|
||||
// goroutine reads a private snapshot instead of racing on the shared slice.
|
||||
snapshot := append([]model.Message(nil), messages...)
|
||||
results, err := async.ExtractAsync(dctx, snapshot)
|
||||
if err != nil {
|
||||
cancel()
|
||||
l.log.Warn("memory extract dispatch", "step", step, "error", err)
|
||||
return
|
||||
}
|
||||
go func() {
|
||||
defer cancel()
|
||||
res, ok := <-results
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
if res.Err != nil {
|
||||
l.log.Warn("memory extract", "step", step, "error", res.Err)
|
||||
return
|
||||
}
|
||||
l.persistMemories(dctx, res.Items, step)
|
||||
}()
|
||||
}
|
||||
|
||||
// persistMemories writes extracted items to the memory store, scoping any item
|
||||
// the extractor left unscoped to the loop's persona. Store failures are logged
|
||||
// and never fatal, so one rejected item cannot abort a turn.
|
||||
func (l *Loop) persistMemories(ctx context.Context, items []memory.MemoryItem, step int) {
|
||||
for _, it := range items {
|
||||
if it.PersonaID == "" && l.personaID != "" {
|
||||
it.PersonaID = l.personaID
|
||||
|
||||
@@ -0,0 +1,228 @@
|
||||
package loop
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/capability/restrict"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||
)
|
||||
|
||||
// The LLM extractor is the async-capable implementation behind design/03 §1: the
|
||||
// loop type-asserts this capability to detach extraction to a background
|
||||
// goroutine, so the contract is locked here rather than discovered at runtime.
|
||||
var (
|
||||
_ memory.Extractor = (*memory.LLMExtractor)(nil)
|
||||
_ asyncExtractor = (*memory.LLMExtractor)(nil)
|
||||
_ memory.Extractor = (*memory.ManualExtractor)(nil)
|
||||
)
|
||||
|
||||
// gatedExtractor is a memory.Extractor that also implements asyncExtractor. Its
|
||||
// extraction blocks until release is closed (or ctx is canceled), so a test can
|
||||
// prove the loop's turn finishes while extraction is still in flight, and that
|
||||
// the extraction survives the caller's own cancellation.
|
||||
type gatedExtractor struct {
|
||||
release chan struct{}
|
||||
started chan struct{}
|
||||
items []memory.MemoryItem
|
||||
dispatch int
|
||||
}
|
||||
|
||||
func newGatedExtractor() *gatedExtractor {
|
||||
return &gatedExtractor{
|
||||
release: make(chan struct{}),
|
||||
started: make(chan struct{}, 1),
|
||||
items: []memory.MemoryItem{{
|
||||
Scope: memory.ScopeUser,
|
||||
Kind: memory.KindRelationship,
|
||||
Content: "用户习惯：爱喝冰美式",
|
||||
Confidence: 0.9,
|
||||
DecayWeight: 1,
|
||||
}},
|
||||
}
|
||||
}
|
||||
|
||||
func (g *gatedExtractor) Extract(ctx context.Context, _ []model.Message) ([]memory.MemoryItem, error) {
|
||||
select {
|
||||
case g.started <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
select {
|
||||
case <-g.release:
|
||||
return g.items, nil
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
func (g *gatedExtractor) ExtractAsync(ctx context.Context, conv []model.Message) (<-chan memory.AsyncResult, error) {
|
||||
g.dispatch++
|
||||
out := make(chan memory.AsyncResult, 1)
|
||||
go func() {
|
||||
defer close(out)
|
||||
items, err := g.Extract(ctx, conv)
|
||||
out <- memory.AsyncResult{Items: items, Err: err}
|
||||
}()
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// TestExtractMemoryAsync_LLMExtractorThroughLoop wires the real LLM extractor
|
||||
// (not a fake) into the loop: after the turn settles, the extractor's own model
|
||||
// call runs in the background and its parsed items land in the store, so the
|
||||
// production path (design/03 §1 LLM 异步自动提取) is exercised end to end.
|
||||
func TestExtractMemoryAsync_LLMExtractorThroughLoop(t *testing.T) {
|
||||
t.Parallel()
|
||||
store := newTestMemoryStore(t)
|
||||
ad := &mockAdapter{chat: []mgResp{
|
||||
{resp: model.Response{Content: "记住了"}},
|
||||
{resp: model.Response{Content: `[{"content":"用户爱喝冰美式","scope":"user","kind":"relationship","confidence":0.92}]`}},
|
||||
}}
|
||||
a := newTestAgent(t, ad, nil)
|
||||
l := New(Options{
|
||||
Agent: a, Memory: store, PersonaID: "persona-a", MaxRounds: 8,
|
||||
Extractor: memory.NewLLMExtractor(ad, "persona-a", "owner"),
|
||||
})
|
||||
|
||||
if _, err := l.Run(context.Background(), "sess", []model.Message{model.TextMessage(model.RoleUser, "我喜欢冰美式")}); err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
waitForMemoryCount(t, store, "persona-a", 1)
|
||||
|
||||
items, err := store.ListByPersona(context.Background(), "persona-a")
|
||||
if err != nil {
|
||||
t.Fatalf("ListByPersona: %v", err)
|
||||
}
|
||||
if items[0].Content != "用户爱喝冰美式" || items[0].Scope != memory.ScopeUser {
|
||||
t.Fatalf("stored item = %+v, want the extracted user-scope memory", items[0])
|
||||
}
|
||||
if items[0].PersonaID != "persona-a" {
|
||||
t.Fatalf("item persona = %q, want persona-a", items[0].PersonaID)
|
||||
}
|
||||
|
||||
extractCalls := 0
|
||||
for _, req := range ad.reqs {
|
||||
if req.Purpose == "memory_extract" {
|
||||
extractCalls++
|
||||
}
|
||||
}
|
||||
if extractCalls != 1 {
|
||||
t.Fatalf("memory_extract model calls = %d, want 1 (the LLM extractor drove the background call)", extractCalls)
|
||||
}
|
||||
}
|
||||
|
||||
// waitForMemoryCount polls the persona's stored memories until want is reached,
|
||||
// failing the test when the extraction never lands.
|
||||
func waitForMemoryCount(t *testing.T, store *memory.Store, personaID string, want int) {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(2 * time.Second)
|
||||
for {
|
||||
items, err := store.ListByPersona(context.Background(), personaID)
|
||||
if err != nil {
|
||||
t.Fatalf("ListByPersona: %v", err)
|
||||
}
|
||||
if len(items) == want {
|
||||
return
|
||||
}
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatalf("stored memories = %d, want %d (extraction never landed)", len(items), want)
|
||||
}
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
}
|
||||
}
|
||||
|
||||
// TestExtractMemoryAsync_DoesNotBlockTurn proves the turn returns while the LLM
|
||||
// extraction is still running, and that the extraction completes (and persists)
|
||||
// after the caller's context has already been canceled — the detached context
|
||||
// keeps a request-scoped cancellation from aborting the background extraction.
|
||||
func TestExtractMemoryAsync_DoesNotBlockTurn(t *testing.T) {
|
||||
t.Parallel()
|
||||
store := newTestMemoryStore(t)
|
||||
ext := newGatedExtractor()
|
||||
ad := &mockAdapter{chat: []mgResp{{resp: model.Response{Content: "记住了"}}}}
|
||||
a := newTestAgent(t, ad, nil)
|
||||
l := New(Options{Agent: a, Memory: store, Extractor: ext, PersonaID: "persona-a", MaxRounds: 8})
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
done := make(chan error, 1)
|
||||
go func() {
|
||||
_, err := l.Run(ctx, "sess", []model.Message{model.TextMessage(model.RoleUser, "记住 我爱喝冰美式")})
|
||||
done <- err
|
||||
}()
|
||||
|
||||
select {
|
||||
case err := <-done:
|
||||
if err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("Run blocked on memory extraction: the async path must not gate the reply")
|
||||
}
|
||||
|
||||
if ext.dispatch != 1 {
|
||||
t.Fatalf("async dispatch count = %d, want 1 (one extraction per settled turn)", ext.dispatch)
|
||||
}
|
||||
items, err := store.ListByPersona(context.Background(), "persona-a")
|
||||
if err != nil {
|
||||
t.Fatalf("ListByPersona: %v", err)
|
||||
}
|
||||
if len(items) != 0 {
|
||||
t.Fatalf("store already holds %d items after Run, want 0 (extraction still in flight)", len(items))
|
||||
}
|
||||
|
||||
// The turn is over, so its context is canceled; the background extraction
|
||||
// must still finish and persist.
|
||||
cancel()
|
||||
select {
|
||||
case <-ext.started:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("extraction goroutine never started")
|
||||
}
|
||||
close(ext.release)
|
||||
waitForMemoryCount(t, store, "persona-a", 1)
|
||||
}
|
||||
|
||||
// TestExtractMemoryAsync_RestrictedEnvNeverDispatches proves the group-chat /
|
||||
// stranger isolation still gates the async path: nothing is dispatched, so no
|
||||
// owner-private memory can be written from a read-only conversation.
|
||||
func TestExtractMemoryAsync_RestrictedEnvNeverDispatches(t *testing.T) {
|
||||
t.Parallel()
|
||||
store := newTestMemoryStore(t)
|
||||
ext := newGatedExtractor()
|
||||
l := New(Options{Memory: store, Extractor: ext, PersonaID: "persona-a"})
|
||||
|
||||
ctx := WithSessionEnv(context.Background(), SessionEnv{ActorEnv: restrict.ActorGroupMember, MemWritable: false})
|
||||
l.extractMemoryAsync(ctx, []model.Message{model.TextMessage(model.RoleUser, "记住 我爱喝冰美式")}, 1)
|
||||
|
||||
if ext.dispatch != 0 {
|
||||
t.Fatalf("async dispatch count = %d, want 0 (restricted context is read-only)", ext.dispatch)
|
||||
}
|
||||
items, err := store.ListByPersona(context.Background(), "persona-a")
|
||||
if err != nil {
|
||||
t.Fatalf("ListByPersona: %v", err)
|
||||
}
|
||||
if len(items) != 0 {
|
||||
t.Fatalf("restricted async extract wrote %d items, want 0 (read-only isolation)", len(items))
|
||||
}
|
||||
}
|
||||
|
||||
// TestExtractMemoryAsync_ManualExtractorStaysSync proves the deterministic path
|
||||
// is unchanged: an extractor without the async capability is still run inline, so
|
||||
// the explicit "记住 X" memory is stored by the time the call returns.
|
||||
func TestExtractMemoryAsync_ManualExtractorStaysSync(t *testing.T) {
|
||||
t.Parallel()
|
||||
store := newTestMemoryStore(t)
|
||||
l := New(Options{Memory: store, Extractor: memory.NewManualExtractor("persona-a", "owner"), PersonaID: "persona-a"})
|
||||
|
||||
ctx := WithSessionEnv(context.Background(), OwnerEnv())
|
||||
l.extractMemoryAsync(ctx, []model.Message{model.TextMessage(model.RoleUser, "记住 我爱喝冰美式")}, 1)
|
||||
|
||||
items, err := store.ListByPersona(context.Background(), "persona-a")
|
||||
if err != nil {
|
||||
t.Fatalf("ListByPersona: %v", err)
|
||||
}
|
||||
if len(items) != 1 {
|
||||
t.Fatalf("stored memories = %d, want 1 (deterministic extractor runs inline)", len(items))
|
||||
}
|
||||
}
|
||||
@@ -18,8 +18,10 @@ import (
|
||||
// timeout and fed back as RoleTool messages so the model can continue.
|
||||
// A checkpoint is persisted at every step, every step is traced as a
|
||||
// state_transition span (with an llm_call child and tool_call spans for each
|
||||
// invocation), and after each step the memory Extractor runs with errors logged
|
||||
// rather than fatal.
|
||||
// invocation), and when the turn settles the memory Extractor runs over the
|
||||
// turn's conversation (design/03 §1: the LLM extractor on a background
|
||||
// goroutine, so extraction never delays the reply) with errors logged rather
|
||||
// than fatal.
|
||||
//
|
||||
// When a turn settles with a Result, the OnTurnEnd hook is called once for that
|
||||
// turn (never per step) and its mood label is recorded on Result.Mood, so the
|
||||
@@ -84,7 +86,7 @@ func (l *Loop) Run(ctx context.Context, sessionID string, messages []model.Messa
|
||||
res.Text = outcome.text
|
||||
res.StopReason = outcome.stopReason
|
||||
l.saveCheckpoint(ctx, sessionID, step, msgs, outcome.stopReason)
|
||||
l.extractMemory(ctx, msgs, step)
|
||||
l.extractMemoryAsync(ctx, msgs, step)
|
||||
l.settleTurn(ctx, sessionID, msgs, res)
|
||||
return res, nil
|
||||
}
|
||||
@@ -96,7 +98,6 @@ func (l *Loop) Run(ctx context.Context, sessionID string, messages []model.Messa
|
||||
msgs = batch.messages
|
||||
res.ToolCallsExecuted += batch.executed
|
||||
l.saveCheckpoint(ctx, sessionID, step, msgs, agent.StopReasonToolCalls)
|
||||
l.extractMemory(ctx, msgs, step)
|
||||
|
||||
// The critical-tool-loop safety net (or a terminating hook) stops the
|
||||
// round without waiting for a real final answer.
|
||||
@@ -104,7 +105,7 @@ func (l *Loop) Run(ctx context.Context, sessionID string, messages []model.Messa
|
||||
res.Text = criticalToolLoopTerminalReason
|
||||
res.StopReason = StopReasonToolLoopTerminated
|
||||
l.saveCheckpoint(ctx, sessionID, step, msgs, StopReasonToolLoopTerminated)
|
||||
l.extractMemory(ctx, msgs, step)
|
||||
l.extractMemoryAsync(ctx, msgs, step)
|
||||
l.settleTurn(ctx, sessionID, msgs, res)
|
||||
return res, nil
|
||||
}
|
||||
@@ -114,7 +115,7 @@ func (l *Loop) Run(ctx context.Context, sessionID string, messages []model.Messa
|
||||
res.Steps = l.maxRounds
|
||||
res.StopReason = StopReasonMaxRounds
|
||||
l.saveCheckpoint(ctx, sessionID, l.maxRounds, msgs, StopReasonMaxRounds)
|
||||
l.extractMemory(ctx, msgs, l.maxRounds)
|
||||
l.extractMemoryAsync(ctx, msgs, l.maxRounds)
|
||||
l.settleTurn(ctx, sessionID, msgs, res)
|
||||
return res, nil
|
||||
}
|
||||
@@ -0,0 +1,311 @@
|
||||
package gateway
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
"unicode/utf8"
|
||||
|
||||
apperr "git.yeij.top/AskaEth/Cyrene_v2/internal/errors"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/services/backthink"
|
||||
)
|
||||
|
||||
// Notification feed limits (design/10 §3 #16). Compile-time constants so no bare
|
||||
// magic number appears in the run path (project rule §10).
|
||||
const (
|
||||
defaultNotificationLimit = 50
|
||||
maxNotificationLimit = 200
|
||||
// notificationFetchFactor widens the per-source read. The merged page must be
|
||||
// the newest N of the union, which the per-source newest-N always contains,
|
||||
// but with a `before` cursor one source can dominate the newest window and
|
||||
// starve the other — the factor pays for that.
|
||||
notificationFetchFactor = 4
|
||||
// notificationBodyRunes caps a backthink summary in the feed; the full text
|
||||
// stays available on GET /api/v1/admin/backthink.
|
||||
notificationBodyRunes = 200
|
||||
)
|
||||
|
||||
// The unified feed's families, used both as the item `type` and as the `?type=`
|
||||
// filter. They are derived from the sources the composition root already has
|
||||
// (design/10 §3 #16: 先聚合现有源为统一 feed，避免新建重模型).
|
||||
const (
|
||||
// NotificationTypeSystem is a system/security operation: backup & restore,
|
||||
// JWT rotation, API-key revocation, plugin install/uninstall, tool policy.
|
||||
NotificationTypeSystem = "system"
|
||||
// NotificationTypeChannel is a channel lifecycle or connection event
|
||||
// (channel.open / close / save / reject / login).
|
||||
NotificationTypeChannel = "channel"
|
||||
// NotificationTypeBackthink is one proactive push the background thinker
|
||||
// decided to send.
|
||||
NotificationTypeBackthink = "backthink"
|
||||
// NotificationTypeAudit is the remaining admin audit trail: personas, memory,
|
||||
// users, RAG, sessions, uploads, audio.
|
||||
NotificationTypeAudit = "audit"
|
||||
)
|
||||
|
||||
// Provenance labels carried on an item's `source` field.
|
||||
const (
|
||||
notificationSourceAudit = "audit"
|
||||
notificationSourceBackthink = "backthink"
|
||||
// notificationIDPrefixAudit / ...Backthink namespace the source's own record id
|
||||
// so the merged feed has stable, collision-free, client-dedupable ids.
|
||||
notificationIDPrefixAudit = "audit:"
|
||||
notificationIDPrefixBackthink = "backthink:"
|
||||
)
|
||||
|
||||
// AdminNotification is one item of the unified notification feed (design/10 §3
|
||||
// #16). It is a read-only projection of an existing record — audit rows and
|
||||
// backthink records — never a new persisted model: `id` is "<source>:<原生 id>",
|
||||
// `title`/`body` are derived from the source record, and `createdAt` is epoch
|
||||
// milliseconds.
|
||||
type AdminNotification struct {
|
||||
ID string `json:"id"`
|
||||
Type string `json:"type"`
|
||||
Title string `json:"title"`
|
||||
Body string `json:"body,omitempty"`
|
||||
Source string `json:"source"`
|
||||
CreatedAt int64 `json:"createdAt"`
|
||||
}
|
||||
|
||||
// adminListNotifications returns the unified notification feed (owner/admin
|
||||
// read): the admin audit trail and the background thinker's proactive pushes,
|
||||
// merged newest-first and filterable by family and by a `before` cursor.
|
||||
//
|
||||
// A source that is not injected contributes no items — the feed reports what
|
||||
// actually exists instead of fabricating a family — so a deployment without the
|
||||
// backthink service simply has no `backthink` items.
|
||||
func (s *Server) adminListNotifications(w http.ResponseWriter, r *http.Request) {
|
||||
limit := parseNotificationLimit(r)
|
||||
typ := strings.TrimSpace(r.URL.Query().Get("type"))
|
||||
if typ != "" && !validNotificationType(typ) {
|
||||
writeError(w, apperr.Wrap(apperr.ErrBadRequest, "invalid_type",
|
||||
"type must be one of system|channel|backthink|audit"))
|
||||
return
|
||||
}
|
||||
before, hasBefore := parseNotificationBefore(r)
|
||||
|
||||
items, err := s.notificationItems(r.Context(), notificationFetchLimit(limit))
|
||||
if err != nil {
|
||||
s.log.Error("list notifications", "err", err)
|
||||
writeError(w, apperr.Wrap(apperr.ErrInternal, "notifications_failed",
|
||||
"failed to list notifications"))
|
||||
return
|
||||
}
|
||||
|
||||
out := make([]AdminNotification, 0, limit)
|
||||
for _, it := range items {
|
||||
if typ != "" && it.Type != typ {
|
||||
continue
|
||||
}
|
||||
if hasBefore && it.CreatedAt > before {
|
||||
continue
|
||||
}
|
||||
out = append(out, it)
|
||||
if len(out) == limit {
|
||||
break
|
||||
}
|
||||
}
|
||||
writeJSON(w, http.StatusOK, out)
|
||||
}
|
||||
|
||||
// notificationItems reads both sources and returns their merge, newest first. A
|
||||
// source failure is returned (never swallowed into a silently shorter feed), so
|
||||
// the endpoint answers a clear error instead of pretending there was no event.
|
||||
func (s *Server) notificationItems(ctx context.Context, fetch int) ([]AdminNotification, error) {
|
||||
auditItems, err := s.notificationsFromAudit(ctx, fetch)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
btItems, err := s.notificationsFromBackthink(ctx, fetch)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
items := append(auditItems, btItems...)
|
||||
sort.SliceStable(items, func(i, j int) bool {
|
||||
if items[i].CreatedAt != items[j].CreatedAt {
|
||||
return items[i].CreatedAt > items[j].CreatedAt
|
||||
}
|
||||
// Descending id keeps the order deterministic for items sharing a
|
||||
// millisecond (the cursor can split one millisecond across pages).
|
||||
return items[i].ID > items[j].ID
|
||||
})
|
||||
return items, nil
|
||||
}
|
||||
|
||||
// notificationsFromAudit projects the admin audit trail (audit_logs, persisted
|
||||
// by NewSQLiteAuditStore) into feed items. The fallback in-memory store holds
|
||||
// only records really appended in this process, so an unwired store yields an
|
||||
// empty family rather than fabricated entries.
|
||||
func (s *Server) notificationsFromAudit(ctx context.Context, fetch int) ([]AdminNotification, error) {
|
||||
if s.obs == nil || s.obs.audit == nil {
|
||||
return nil, nil
|
||||
}
|
||||
records, err := s.obs.audit.List(ctx, fetch)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("gateway: notification audit source: %w", err)
|
||||
}
|
||||
out := make([]AdminNotification, 0, len(records))
|
||||
for _, rec := range records {
|
||||
out = append(out, AdminNotification{
|
||||
ID: notificationIDPrefixAudit + strconv.FormatInt(rec.ID, 10),
|
||||
Type: notificationTypeOfAudit(rec.What),
|
||||
Title: rec.What,
|
||||
Body: notificationAuditBody(rec),
|
||||
Source: notificationSourceAudit,
|
||||
CreatedAt: notificationAuditAt(rec.When),
|
||||
})
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// notificationsFromBackthink projects the background thinker's records into feed
|
||||
// items. Only records that decided to push are notifications (a plain thinking
|
||||
// record is not an event the owner needs to be told about); each carries the
|
||||
// delivery state so the feed never claims an undelivered push was sent.
|
||||
func (s *Server) notificationsFromBackthink(ctx context.Context, fetch int) ([]AdminNotification, error) {
|
||||
if s.backthink == nil {
|
||||
return nil, nil
|
||||
}
|
||||
records, err := s.backthink.List(ctx, fetch)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("gateway: notification backthink source: %w", err)
|
||||
}
|
||||
out := make([]AdminNotification, 0, len(records))
|
||||
for _, rec := range records {
|
||||
if !rec.ShouldPush {
|
||||
continue
|
||||
}
|
||||
out = append(out, AdminNotification{
|
||||
ID: notificationIDPrefixBackthink + strconv.FormatInt(rec.ID, 10),
|
||||
Type: NotificationTypeBackthink,
|
||||
Title: backthinkNotificationTitle(rec),
|
||||
Body: clipRunes(strings.TrimSpace(rec.Conclusion), notificationBodyRunes),
|
||||
Source: notificationSourceBackthink,
|
||||
CreatedAt: rec.At.UnixMilli(),
|
||||
})
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// notificationTypeOfAudit classifies an audit operation into a feed family: the
|
||||
// channel lifecycle is `channel`, and the system/plugin/policy operations are
|
||||
// `system`; everything else stays `audit`.
|
||||
func notificationTypeOfAudit(what string) string {
|
||||
switch {
|
||||
case strings.HasPrefix(what, "channel."):
|
||||
return NotificationTypeChannel
|
||||
case strings.HasPrefix(what, "system."),
|
||||
strings.HasPrefix(what, "plugin."),
|
||||
strings.HasPrefix(what, "policy."):
|
||||
return NotificationTypeSystem
|
||||
default:
|
||||
return NotificationTypeAudit
|
||||
}
|
||||
}
|
||||
|
||||
// notificationAuditBody derives an audit item's body from its who/what/result
|
||||
// trail, e.g. "owner:owner · ok".
|
||||
func notificationAuditBody(rec AuditRecord) string {
|
||||
parts := make([]string, 0, 2)
|
||||
if who := strings.TrimSpace(rec.Who); who != "" {
|
||||
parts = append(parts, who)
|
||||
}
|
||||
if result := strings.TrimSpace(rec.Result); result != "" {
|
||||
parts = append(parts, result)
|
||||
}
|
||||
return strings.Join(parts, " · ")
|
||||
}
|
||||
|
||||
// notificationAuditAt parses an audit record's RFC3339 timestamp into epoch
|
||||
// milliseconds. An unparseable timestamp yields 0: the item still appears (oldest
|
||||
// first) instead of being dropped or stamped with a made-up time.
|
||||
func notificationAuditAt(when string) int64 {
|
||||
ts, err := time.Parse(time.RFC3339Nano, strings.TrimSpace(when))
|
||||
if err != nil {
|
||||
return 0
|
||||
}
|
||||
return ts.UnixMilli()
|
||||
}
|
||||
|
||||
// backthinkNotificationTitle labels one proactive push with its trigger and
|
||||
// delivery state, so the feed distinguishes "sent" from "decided but not
|
||||
// delivered" (route none / no online client / push failure).
|
||||
func backthinkNotificationTitle(rec backthink.Record) string {
|
||||
state := "未送达"
|
||||
if rec.PushResult.Pushed {
|
||||
state = "已送达"
|
||||
}
|
||||
if trigger := strings.TrimSpace(rec.Trigger); trigger != "" {
|
||||
return "主动推送（" + trigger + " · " + state + "）"
|
||||
}
|
||||
return "主动推送（" + state + "）"
|
||||
}
|
||||
|
||||
// clipRunes truncates s to at most max runes, appending an ellipsis when it was
|
||||
// cut. It never splits a multi-byte rune.
|
||||
func clipRunes(s string, max int) string {
|
||||
if max <= 0 || utf8.RuneCountInString(s) <= max {
|
||||
return s
|
||||
}
|
||||
runes := []rune(s)
|
||||
return string(runes[:max]) + "…"
|
||||
}
|
||||
|
||||
// parseNotificationLimit parses the ?limit= query param, clamping to
|
||||
// [1, maxNotificationLimit] and selecting defaultNotificationLimit when absent or
|
||||
// invalid.
|
||||
func parseNotificationLimit(r *http.Request) int {
|
||||
n, err := strconv.Atoi(r.URL.Query().Get("limit"))
|
||||
if err != nil || n <= 0 {
|
||||
return defaultNotificationLimit
|
||||
}
|
||||
if n > maxNotificationLimit {
|
||||
return maxNotificationLimit
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
// parseNotificationBefore parses the ?before= epoch-millisecond cursor. The
|
||||
// cursor is inclusive (items with createdAt <= before are returned): the sources
|
||||
// resolve to milliseconds, so a page boundary can fall inside one millisecond,
|
||||
// and the client deduplicates by id. An absent or malformed cursor means "from
|
||||
// the newest item".
|
||||
func parseNotificationBefore(r *http.Request) (int64, bool) {
|
||||
raw := strings.TrimSpace(r.URL.Query().Get("before"))
|
||||
if raw == "" {
|
||||
return 0, false
|
||||
}
|
||||
ms, err := strconv.ParseInt(raw, 10, 64)
|
||||
if err != nil || ms <= 0 {
|
||||
return 0, false
|
||||
}
|
||||
return ms, true
|
||||
}
|
||||
|
||||
// notificationFetchLimit widens a page request into the per-source read window,
|
||||
// capped at maxObsLimit (the audit store's own ceiling) and never below the page.
|
||||
func notificationFetchLimit(limit int) int {
|
||||
fetch := limit * notificationFetchFactor
|
||||
if fetch > maxObsLimit {
|
||||
fetch = maxObsLimit
|
||||
}
|
||||
if fetch < limit {
|
||||
fetch = limit
|
||||
}
|
||||
return fetch
|
||||
}
|
||||
|
||||
// validNotificationType reports whether t names a feed family. An unknown family
|
||||
// is rejected with 400 rather than answered with a silently empty feed.
|
||||
func validNotificationType(t string) bool {
|
||||
switch t {
|
||||
case NotificationTypeSystem, NotificationTypeChannel, NotificationTypeBackthink, NotificationTypeAudit:
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,357 @@
|
||||
package gateway
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/services/backthink"
|
||||
)
|
||||
|
||||
// stubAuditFeed is an AuditStore over scripted records (already newest first),
|
||||
// so the feed tests control millisecond timestamps exactly — the in-memory
|
||||
// fallback store only keeps second precision.
|
||||
type stubAuditFeed struct {
|
||||
records []AuditRecord
|
||||
err error
|
||||
}
|
||||
|
||||
func (s *stubAuditFeed) Append(context.Context, string, string, string, time.Time) error { return nil }
|
||||
|
||||
func (s *stubAuditFeed) List(_ context.Context, limit int) ([]AuditRecord, error) {
|
||||
if s.err != nil {
|
||||
return nil, s.err
|
||||
}
|
||||
if limit <= 0 || limit > len(s.records) {
|
||||
limit = len(s.records)
|
||||
}
|
||||
return append([]AuditRecord(nil), s.records[:limit]...), nil
|
||||
}
|
||||
|
||||
// stubBackthinkFeed is a BackthinkStore over scripted records (newest first).
|
||||
type stubBackthinkFeed struct {
|
||||
records []backthink.Record
|
||||
err error
|
||||
}
|
||||
|
||||
func (s *stubBackthinkFeed) List(_ context.Context, limit int) ([]backthink.Record, error) {
|
||||
if s.err != nil {
|
||||
return nil, s.err
|
||||
}
|
||||
if limit <= 0 || limit > len(s.records) {
|
||||
limit = len(s.records)
|
||||
}
|
||||
return append([]backthink.Record(nil), s.records[:limit]...), nil
|
||||
}
|
||||
|
||||
// notificationMS renders epoch milliseconds as the RFC3339 timestamp an audit
|
||||
// record carries.
|
||||
func notificationMS(ms int64) string {
|
||||
return time.UnixMilli(ms).UTC().Format(time.RFC3339Nano)
|
||||
}
|
||||
|
||||
// notificationFeedFixture scripts both sources: channel + system + audit rows in
|
||||
// the audit trail and one delivered / one undelivered / one non-push backthink
|
||||
// record.
|
||||
func notificationFeedFixture() (*stubAuditFeed, *stubBackthinkFeed) {
|
||||
audit := &stubAuditFeed{records: []AuditRecord{
|
||||
{ID: 5, Who: "owner:owner", What: "channel.open", When: notificationMS(3000), Result: "ok"},
|
||||
{ID: 4, Who: "bridge:telegram", What: "channel.reject", When: notificationMS(2900), Result: "群聊白名单外"},
|
||||
{ID: 3, Who: "owner:owner", What: "system.backup", When: notificationMS(2000), Result: "ok"},
|
||||
{ID: 2, Who: "owner:owner", What: "persona.update", When: notificationMS(1000), Result: "ok"},
|
||||
{ID: 1, Who: "owner:owner", What: "plugin.install", When: notificationMS(500), Result: "error"},
|
||||
}}
|
||||
bt := &stubBackthinkFeed{records: []backthink.Record{
|
||||
{ID: 8, Trigger: "idle", At: time.UnixMilli(2600), Conclusion: "该喝水了", ShouldPush: true,
|
||||
PushResult: backthink.PushResult{Pushed: true}},
|
||||
{ID: 7, Trigger: "event", At: time.UnixMilli(2400), Conclusion: "只是想了一下", ShouldPush: false},
|
||||
{ID: 6, Trigger: "heartbeat", At: time.UnixMilli(1500), Conclusion: strings.Repeat("很长的结论", 60), ShouldPush: true},
|
||||
}}
|
||||
return audit, bt
|
||||
}
|
||||
|
||||
// listNotifications calls the endpoint and decodes the feed.
|
||||
func listNotifications(t *testing.T, ts *httptest.Server, tok, query string) (int, []AdminNotification, []byte) {
|
||||
t.Helper()
|
||||
resp, body := doReq(t, http.MethodGet, ts.URL+"/api/v1/admin/notifications"+query, nil, tok)
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return resp.StatusCode, nil, body
|
||||
}
|
||||
var out []AdminNotification
|
||||
if err := json.Unmarshal(decodeBody(t, body)["data"], &out); err != nil {
|
||||
t.Fatalf("decode notifications %s: %v", body, err)
|
||||
}
|
||||
return resp.StatusCode, out, body
|
||||
}
|
||||
|
||||
// TestAdminListNotifications_MergesAuditAndBackthink proves the feed is the two
|
||||
// existing sources merged newest-first with source-namespaced ids, the audit
|
||||
// family classification, and that a backthink record which decided not to push is
|
||||
// not an event.
|
||||
func TestAdminListNotifications_MergesAuditAndBackthink(t *testing.T) {
|
||||
t.Parallel()
|
||||
audit, bt := notificationFeedFixture()
|
||||
_, ts, jwt := buildEnv(t, nil, Deps{AuditStore: audit, Backthink: bt})
|
||||
tok := signAny(t, jwt, "acct", auth.RoleOwner)
|
||||
|
||||
status, items, body := listNotifications(t, ts, tok, "")
|
||||
if status != http.StatusOK {
|
||||
t.Fatalf("status = %d, want 200 (body=%s)", status, body)
|
||||
}
|
||||
wantOrder := []struct {
|
||||
id string
|
||||
typ string
|
||||
src string
|
||||
when int64
|
||||
}{
|
||||
{"audit:5", NotificationTypeChannel, notificationSourceAudit, 3000},
|
||||
{"audit:4", NotificationTypeChannel, notificationSourceAudit, 2900},
|
||||
{"backthink:8", NotificationTypeBackthink, notificationSourceBackthink, 2600},
|
||||
{"audit:3", NotificationTypeSystem, notificationSourceAudit, 2000},
|
||||
{"backthink:6", NotificationTypeBackthink, notificationSourceBackthink, 1500},
|
||||
{"audit:2", NotificationTypeAudit, notificationSourceAudit, 1000},
|
||||
{"audit:1", NotificationTypeSystem, notificationSourceAudit, 500},
|
||||
}
|
||||
if len(items) != len(wantOrder) {
|
||||
t.Fatalf("feed = %d items, want %d (%+v)", len(items), len(wantOrder), items)
|
||||
}
|
||||
for i, want := range wantOrder {
|
||||
got := items[i]
|
||||
if got.ID != want.id || got.Type != want.typ || got.Source != want.src || got.CreatedAt != want.when {
|
||||
t.Fatalf("feed[%d] = %+v, want {id:%s type:%s source:%s createdAt:%d}",
|
||||
i, got, want.id, want.typ, want.src, want.when)
|
||||
}
|
||||
if got.Title == "" {
|
||||
t.Fatalf("feed[%d] has no title: %+v", i, got)
|
||||
}
|
||||
if want.id == "backthink:8" && !strings.Contains(got.Title, "已送达") {
|
||||
t.Fatalf("backthink title = %q, want the delivery state", got.Title)
|
||||
}
|
||||
}
|
||||
// The undelivered push is older and its conclusion is long: both the body cap
|
||||
// and the "未送达" label must show.
|
||||
for _, it := range items {
|
||||
if it.ID == "backthink:6" {
|
||||
if !strings.Contains(it.Title, "未送达") {
|
||||
t.Fatalf("backthink title = %q, want 未送达 for an undelivered push", it.Title)
|
||||
}
|
||||
if len([]rune(it.Body)) > notificationBodyRunes+1 {
|
||||
t.Fatalf("body = %d runes, want <= %d (+ellipsis)", len([]rune(it.Body)), notificationBodyRunes)
|
||||
}
|
||||
}
|
||||
}
|
||||
// A thinking record that decided not to push never becomes a notification.
|
||||
for _, it := range items {
|
||||
if it.ID == "backthink:7" {
|
||||
t.Fatalf("non-push backthink record leaked into the feed: %+v", it)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestAdminListNotifications_TypeFilter proves `?type=` selects one family and
|
||||
// that a family without any source data answers an empty feed (never a fabricated
|
||||
// row).
|
||||
func TestAdminListNotifications_TypeFilter(t *testing.T) {
|
||||
t.Parallel()
|
||||
audit, bt := notificationFeedFixture()
|
||||
_, ts, jwt := buildEnv(t, nil, Deps{AuditStore: audit, Backthink: bt})
|
||||
tok := signAny(t, jwt, "acct", auth.RoleAdmin)
|
||||
|
||||
cases := []struct {
|
||||
query string
|
||||
want []string
|
||||
}{
|
||||
{"?type=channel", []string{"audit:5", "audit:4"}},
|
||||
{"?type=backthink", []string{"backthink:8", "backthink:6"}},
|
||||
{"?type=system", []string{"audit:3", "audit:1"}},
|
||||
{"?type=audit", []string{"audit:2"}},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
status, items, body := listNotifications(t, ts, tok, tc.query)
|
||||
if status != http.StatusOK {
|
||||
t.Fatalf("%s status = %d, want 200 (body=%s)", tc.query, status, body)
|
||||
}
|
||||
if len(items) != len(tc.want) {
|
||||
t.Fatalf("%s = %+v, want %v", tc.query, items, tc.want)
|
||||
}
|
||||
for i, want := range tc.want {
|
||||
if items[i].ID != want {
|
||||
t.Fatalf("%s[%d] = %s, want %s", tc.query, i, items[i].ID, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// An empty source (no backthink service injected) yields an empty family.
|
||||
_, auditOnly, jwtOnly := buildEnv(t, nil, Deps{AuditStore: audit})
|
||||
tokAudit := signAny(t, jwtOnly, "acct", auth.RoleOwner)
|
||||
status, items, body := listNotifications(t, auditOnly, tokAudit, "?type=backthink")
|
||||
if status != http.StatusOK || len(items) != 0 {
|
||||
t.Fatalf("no-backthink feed = %d items / status %d, want 0 / 200 (body=%s)", len(items), status, body)
|
||||
}
|
||||
}
|
||||
|
||||
// TestAdminListNotifications_BeforeCursorPagination proves the `before` cursor:
|
||||
// pages walk older, the cursor item itself is included (the sources resolve to
|
||||
// milliseconds and the client dedupes by id), and `limit` caps a page.
|
||||
func TestAdminListNotifications_BeforeCursorPagination(t *testing.T) {
|
||||
t.Parallel()
|
||||
audit, bt := notificationFeedFixture()
|
||||
_, ts, jwt := buildEnv(t, nil, Deps{AuditStore: audit, Backthink: bt})
|
||||
tok := signAny(t, jwt, "acct", auth.RoleOwner)
|
||||
|
||||
status, page1, body := listNotifications(t, ts, tok, "?limit=2")
|
||||
if status != http.StatusOK {
|
||||
t.Fatalf("page1 status = %d, want 200 (body=%s)", status, body)
|
||||
}
|
||||
if len(page1) != 2 || page1[0].ID != "audit:5" || page1[1].ID != "audit:4" {
|
||||
t.Fatalf("page1 = %+v, want [audit:5 audit:4]", page1)
|
||||
}
|
||||
|
||||
status, page2, body := listNotifications(t, ts, tok, "?limit=2&before=2900")
|
||||
if status != http.StatusOK {
|
||||
t.Fatalf("page2 status = %d, want 200 (body=%s)", status, body)
|
||||
}
|
||||
if len(page2) != 2 || page2[0].ID != "audit:4" || page2[1].ID != "backthink:8" {
|
||||
t.Fatalf("page2 = %+v, want [audit:4 backthink:8] (cursor inclusive)", page2)
|
||||
}
|
||||
|
||||
status, page3, body := listNotifications(t, ts, tok, "?limit=2&before=2600")
|
||||
if status != http.StatusOK {
|
||||
t.Fatalf("page3 status = %d, want 200 (body=%s)", status, body)
|
||||
}
|
||||
if len(page3) != 2 || page3[0].ID != "backthink:8" || page3[1].ID != "audit:3" {
|
||||
t.Fatalf("page3 = %+v, want [backthink:8 audit:3]", page3)
|
||||
}
|
||||
|
||||
// A cursor older than every item ends the walk with an empty page.
|
||||
status, tail, body := listNotifications(t, ts, tok, "?before=1")
|
||||
if status != http.StatusOK || len(tail) != 0 {
|
||||
t.Fatalf("tail = %+v / status %d, want empty / 200 (body=%s)", tail, status, body)
|
||||
}
|
||||
}
|
||||
|
||||
// TestAdminListNotifications_NoSourcesNoFabrication proves the honest empty feed:
|
||||
// with neither source injected the endpoint returns an empty array, not invented
|
||||
// notices.
|
||||
func TestAdminListNotifications_NoSourcesNoFabrication(t *testing.T) {
|
||||
t.Parallel()
|
||||
_, ts, jwt := buildEnv(t, nil, Deps{})
|
||||
tok := signAny(t, jwt, "acct", auth.RoleOwner)
|
||||
|
||||
status, items, body := listNotifications(t, ts, tok, "")
|
||||
if status != http.StatusOK {
|
||||
t.Fatalf("status = %d, want 200 (body=%s)", status, body)
|
||||
}
|
||||
if len(items) != 0 {
|
||||
t.Fatalf("unwired feed = %+v, want empty", items)
|
||||
}
|
||||
if got := strings.TrimSpace(string(decodeBody(t, body)["data"])); got != "[]" {
|
||||
t.Fatalf("data = %s, want []", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestAdminListNotifications_Errors proves the endpoint fails loudly: an unknown
|
||||
// family is a 400, a failing source is a 500 (never a silently shortened feed),
|
||||
// and a viewer is rejected by RBAC.
|
||||
func TestAdminListNotifications_Errors(t *testing.T) {
|
||||
t.Parallel()
|
||||
audit, bt := notificationFeedFixture()
|
||||
_, ts, jwt := buildEnv(t, nil, Deps{AuditStore: audit, Backthink: bt})
|
||||
tok := signAny(t, jwt, "acct", auth.RoleOwner)
|
||||
|
||||
status, _, body := listNotifications(t, ts, tok, "?type=bogus")
|
||||
if status != http.StatusBadRequest {
|
||||
t.Fatalf("unknown type status = %d, want 400 (body=%s)", status, body)
|
||||
}
|
||||
|
||||
broken := &stubAuditFeed{err: errors.New("audit exploded")}
|
||||
_, brokenTS, brokenJWT := buildEnv(t, nil, Deps{AuditStore: broken, Backthink: bt})
|
||||
status, _, body = listNotifications(t, brokenTS, signAny(t, brokenJWT, "acct", auth.RoleOwner), "")
|
||||
if status != http.StatusInternalServerError {
|
||||
t.Fatalf("broken source status = %d, want 500 (body=%s)", status, body)
|
||||
}
|
||||
|
||||
viewer := signAny(t, jwt, "acct", auth.RoleViewer)
|
||||
if status, _, _ := listNotifications(t, ts, viewer, ""); status != http.StatusForbidden {
|
||||
t.Fatalf("viewer status = %d, want 403", status)
|
||||
}
|
||||
}
|
||||
|
||||
// TestNotificationAuditAt_ParsesBothStoreFormats locks the timestamp contract with
|
||||
// the two real audit stores: SQLiteAuditStore writes RFC3339Nano, the in-memory
|
||||
// fallback writes RFC3339, and an unparseable value degrades to 0 (oldest) rather
|
||||
// than being dropped or faked.
|
||||
func TestNotificationAuditAt_ParsesBothStoreFormats(t *testing.T) {
|
||||
t.Parallel()
|
||||
base := time.Date(2026, 9, 30, 10, 0, 0, 123456000, time.UTC)
|
||||
cases := []struct {
|
||||
when string
|
||||
want int64
|
||||
}{
|
||||
{base.Format(time.RFC3339Nano), base.UnixMilli()},
|
||||
{base.Truncate(time.Second).Format(time.RFC3339), base.Truncate(time.Second).UnixMilli()},
|
||||
{"2026-09-30T18:00:00+08:00", time.Date(2026, 9, 30, 10, 0, 0, 0, time.UTC).UnixMilli()},
|
||||
{"", 0},
|
||||
{"not-a-time", 0},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
if got := notificationAuditAt(tc.when); got != tc.want {
|
||||
t.Fatalf("notificationAuditAt(%q) = %d, want %d", tc.when, got, tc.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestNotificationQueryParsing locks the query knobs: the limit defaults to 50 and
|
||||
// clamps at 200, and an absent or malformed `before` means "from newest".
|
||||
func TestNotificationQueryParsing(t *testing.T) {
|
||||
t.Parallel()
|
||||
limitCases := []struct {
|
||||
raw string
|
||||
want int
|
||||
}{
|
||||
{"", defaultNotificationLimit},
|
||||
{"?limit=0", defaultNotificationLimit},
|
||||
{"?limit=-3", defaultNotificationLimit},
|
||||
{"?limit=abc", defaultNotificationLimit},
|
||||
{"?limit=10", 10},
|
||||
{"?limit=1000", maxNotificationLimit},
|
||||
}
|
||||
for _, tc := range limitCases {
|
||||
req := httptest.NewRequest(http.MethodGet, "/api/v1/admin/notifications"+tc.raw, nil)
|
||||
if got := parseNotificationLimit(req); got != tc.want {
|
||||
t.Fatalf("parseNotificationLimit(%q) = %d, want %d", tc.raw, got, tc.want)
|
||||
}
|
||||
}
|
||||
|
||||
beforeCases := []struct {
|
||||
raw string
|
||||
want int64
|
||||
wantOK bool
|
||||
}{
|
||||
{"", 0, false},
|
||||
{"?before=0", 0, false},
|
||||
{"?before=-5", 0, false},
|
||||
{"?before=abc", 0, false},
|
||||
{"?before=1700000000000", 1700000000000, true},
|
||||
}
|
||||
for _, tc := range beforeCases {
|
||||
req := httptest.NewRequest(http.MethodGet, "/api/v1/admin/notifications"+tc.raw, nil)
|
||||
got, ok := parseNotificationBefore(req)
|
||||
if got != tc.want || ok != tc.wantOK {
|
||||
t.Fatalf("parseNotificationBefore(%q) = %d,%v, want %d,%v", tc.raw, got, ok, tc.want, tc.wantOK)
|
||||
}
|
||||
}
|
||||
|
||||
if got := notificationFetchLimit(defaultNotificationLimit); got < defaultNotificationLimit || got > maxObsLimit {
|
||||
t.Fatalf("notificationFetchLimit(50) = %d, want within [50, %d]", got, maxObsLimit)
|
||||
}
|
||||
if got := notificationFetchLimit(maxNotificationLimit); got != maxObsLimit {
|
||||
t.Fatalf("notificationFetchLimit(200) = %d, want the %d cap", got, maxObsLimit)
|
||||
}
|
||||
}
|
||||
@@ -515,6 +515,10 @@ func (s *Server) Handler() http.Handler {
|
||||
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)))
|
||||
// Unified notification feed (design/10 §3 #16): a read-only projection of the
|
||||
// audit trail + backthink pushes, so the panel's notification page needs no
|
||||
// new persisted model.
|
||||
mux.Handle("GET /api/v1/admin/notifications", adminAuth(http.HandlerFunc(s.adminListNotifications)))
|
||||
mux.Handle("POST /api/v1/admin/plugins/policy", s.auth.Authorize(auth.RoleOwner)(http.HandlerFunc(s.adminSetPolicy)))
|
||||
|
||||
// Admin writes: owner-only, audited. Reads above accept owner/admin; writes
|
||||
|
||||
Reference in new issue
Block a user