AskaEth
6e28b38b08
【MT-A：护栏 + 诚实截断 + 试算异步】
- maxTidyCompareItems 400(写死)→2000 并开放为 MEMORY_TIDY_COMPARE_LIMIT；
MEMORY_TIDY_CANDIDATE_LIMIT 默认 20→200（语义不变，仍每分区）
- 依据是首跑实测：459 条的分区被 400 截断 ⇒ compared(705) < scanned(784)，
候选数(77)由上限而不是规则决定；三处上限的 WARN 都补齐了总数/上限/分区名
- 新增「上限 + 时间预算」双闸（实测 2000 条/分区 ≈234s > 3 分钟墙钟）：
整分区跳过 / 两两比对中途收手，一律如实记 (budget) + WARN
- ?async=1 ⇒ 202 {runId,status}；GET …/{runId} 轮询（writeAuth 同档）；
done 的 result 与同步 data 逐字节同形；并发=复用既有 runId、TTL 30min 自完成起算、
重启即如实 404 tidy_run_not_found；不带 async 时同步路径逐字不变
【MT-B：D3 入库前审查（design/23 §15.1）】
- 新记忆经 Store.Add（仓库里唯一 INSERT INTO memory_items）落库后，由独立记忆管理
会话（sess-memorytidy）在有界后台队列异步判定 skip/merge/modify/create
- 软合并回写（status + merged_into/merged_from/merge_count，可 revert、绝不改 content）
- 隔离来源（群聊/陌生人）**概不审查**、召回/超时失败**零改写**（fail-safe 结构性成立）
- 写入侧仅 +62.7µs/条（非阻塞入队）；默认 ENABLED=true（主人裁定：入库前先审查）
【耦合说明（重要）】两片同包且互相依赖：MT-B 的 preinsert_recall.go 用 MT-A 的
collectTidyPairs 新签名 + tidyBudget ⇒ **各自单独都 build 不过**，故一并落地。
MT-A 为让整包编译，最小幅度适配了 MT-B 的 preinsert_recall.go / preinsert_test.go
（零值预算=行为不变；符号重名改名）。
【我的验收】隔离 worktree（37 文件）build/vet/linux/darwin 全过，切片包全绿；
我亲手打红两个变异：① 去掉 (compare) 截断标记 ⇒ TestTidyCompareLimit_HitIsLoud 红；
② 中和隔离来源守卫 ⇒ TestReviewRestrictedSourceIsNeverTouched 红。
唯一红是 internal/runtime/tool 的两条超时用例（该包无人改）⇒ 主仓库 -count=8 全绿，
判定为负载敏感 flake（当时两个整树验收 + 子代理并发），已记入开发习惯与约定 §7。
.env.example 补 MEMORY_REVIEW_* 四键（MT-B 无该文件权限，由我按其代码语义补写）。
225 lines
9.6 KiB
Go
225 lines
9.6 KiB
Go
package main
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"fmt"
|
||
"log/slog"
|
||
"time"
|
||
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/bus"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/capability/mcp"
|
||
pluginmgr "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/plugin"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/capability/registry"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/capability/transcode"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/config"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/db"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/iot"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/persona/emotion"
|
||
personareg "git.yeij.top/AskaEth/Cyrene_v2/internal/persona/registry"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/embedding"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/events"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/exec"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/flowgraph"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory/projectindex"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory/vector"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/multimodal/voice"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/orchestration"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/task"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/tool"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/services/automations"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/services/backthink"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/services/channelbridge"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/services/gateway"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/user"
|
||
)
|
||
|
||
// defaultOwnerPath 与 defaultPersonasDir 指向随仓库分发的磁盘上人格/主人定义，
|
||
// 路径相对工作目录。
|
||
const (
|
||
defaultOwnerPath = "configs/users/owner.yaml"
|
||
defaultPersonasDir = "configs/personas"
|
||
defaultSQLitePath = "./data/cyrene.db"
|
||
fallbackInstruction = "You are Cyrene, a warm and attentive companion."
|
||
|
||
// 上界与旋钮常量。编译期定值，运行时路径上不出现裸魔法数字（项目规范 §10）。
|
||
defaultMaxRounds = 8
|
||
defaultMaxParallel = 4
|
||
defaultAgentID = "orchestrator"
|
||
defaultToolTimeout = 30 * time.Second
|
||
)
|
||
|
||
// app 持有单二进制 Cyrene 组装的全部子系统：由 newApp 构建，关闭时由 close 拆除。
|
||
type app struct {
|
||
cfg *config.Config
|
||
log *slog.Logger
|
||
db *sql.DB
|
||
dbPath string
|
||
bus bus.Bus
|
||
modelAdapter model.Adapter
|
||
llm *model.RuntimeLLM
|
||
personas *personareg.Registry
|
||
caps *registry.Registry
|
||
// transcoder 是本地转码引擎（design/25）：进程内、同步、纯本地，供渠道与其它
|
||
// 模块经依赖注入调用；同一个引擎也以 `transcode` 工具注册进 caps。装配只告警、
|
||
// 绝不阻塞启动，见 wire_transcode.go。
|
||
transcoder transcode.Transcoder
|
||
dispatcher *tool.Dispatcher
|
||
memStore *memory.Store
|
||
retriever *memory.Retriever
|
||
// embedder 与 memoryTidy 属于记忆整理（design/23）：整理只在 embedder
|
||
// 上算召回相似度，绝不接触会话/历史读数。
|
||
embedder embedding.Embedder
|
||
memoryTidy *memory.TidyService
|
||
// memoryReview 是 D3「入库前审查」（design/23 §15.1 D3）：新记忆落库后由
|
||
// 独立的记忆管理会话判定 merge/modify/create/skip。它同样只依赖记忆存储
|
||
// 与 embedder，拿不到任何会话/历史读数。未启用时为 nil（写入路径不变）。
|
||
memoryReview *memory.PreInsertReviewer
|
||
projStore *projectindex.Store
|
||
loop *loop.Loop
|
||
scheduler *orchestration.Scheduler
|
||
tasks *task.Manager
|
||
execProv exec.Provider
|
||
iot *iot.Manager
|
||
voice *voice.Chain
|
||
mcpRegs []*mcp.Registry
|
||
plugins *pluginmgr.Manager
|
||
sessions *session.SessionManager
|
||
automation *automations.Service
|
||
backthink *backthink.Backthink
|
||
vindex vector.Index
|
||
owner *user.OwnerProfile
|
||
history *history.Store
|
||
gw *gateway.Server
|
||
channelBridge *channelbridge.Bridge
|
||
channels gateway.ChannelConfigStore
|
||
// docGC 是文档回收的常驻任务（切片 S7）：未被引用且超过保留期的文档连同
|
||
// sidecar 与元数据行一起回收。GC 被配置关闭时为 nil（见 wire_document_gc.go）。
|
||
docGC *documentGC
|
||
// channelEvents 是渠道事件留存面（design/26 §4；切片 E3+E4）：非消息事件逐条
|
||
// 落行 / T1 计数都经它，桥接层由 wire_channels.go 注入。库未接时为 nil
|
||
// （桥接层随即走内存计数降级，见 wire_channel_events.go）。
|
||
channelEvents *events.SQLStore
|
||
// channelEventKeeper 是事件留存的常驻维护任务（保留期清扫 + 计数落盘），
|
||
// 由 main 的 startChannelEvents 拉起（见 wire_channel_events.go）。
|
||
channelEventKeeper *channelEventKeeper
|
||
// usage 是 LLM 用量的异步落库出口（切片 U1）：注入模型栈后，每次成功模型
|
||
// 调用都会把非敏感计数样本塞进它的有界队列。库未接时为 nil（= 只记日志）。
|
||
usage *model.UsageWriter
|
||
// usageGC 是用量样本的保留期清理常驻任务。`LLM_USAGE_RETENTION ≤ 0` 时为
|
||
// nil（清理关闭，见 wire_usage.go）。
|
||
usageGC *usageRetentionGC
|
||
// capacitySink 把 capacity_awaken 状态迁移广播给 Web UI（design/14 §5）。
|
||
// 由 *gateway.Server 满足（BroadcastCapacityAwaken）；以接口形式持有，
|
||
// 组合根才能用假出口做单元测试。
|
||
capacitySink capacityEmitter
|
||
// emotion 把每一轮落定的对话施加到所绑人格的情绪机与好感度上（design/01 §2），
|
||
// 并把变化交给 moodSink。没有注册人格时为 nil，此时不追踪、不持久化任何心情。
|
||
emotion *emotion.Observer
|
||
// moodSink 把心情变化广播给 Web UI（design/07 §3）。由 *gateway.Server 满足
|
||
// （BroadcastMood）；以接口形式持有，组合根才能用假出口做单元测试。
|
||
moodSink moodEmitter
|
||
// traceWriter 把完成的 trace span 扇出到 JSONL stderr 记录器，网关出现后也送往
|
||
// 网关的可观测性缓冲（挂在 GET /api/v1/admin/observability/trace 后面，
|
||
// design/19 §2）。持有它是为了让 wireFrontend 能往已在运行的 tracer 上注册
|
||
// 网关出口。
|
||
traceWriter *traceFanout
|
||
// flow 是 loop 的实时链路图发布器（design/19 §1）。持有它是因为网关还要一个
|
||
// **只读**出口：`GET /api/v1/admin/flow/graph` 回「最近一轮」的快照，补上
|
||
// 「轮次结束后才打开的页面」看不到图的那个洞（契约 v0.71）。组合根把**同一个**
|
||
// 发布器同时交给 loop/orchestration（写入面）与网关（查询面），因此查询到的
|
||
// 就是实时发布的那张图，不会出现两份状态。
|
||
flow *loop.FlowPublisher
|
||
// flowStore 是链路图的**本地落盘**（design/19 §4 P3；切片 F2；迁移 0022）：发布器
|
||
// 每轮收尾经它落一条，网关按 turnId / turns 读回。库未接时为 nil（发布器随即走
|
||
// 纯内存路径，见 wire_flow_graphs.go）。它与 flow 是两件事：flow 是实时内存图，
|
||
// flowStore 是收尾存档。
|
||
flowStore *flowgraph.SQLStore
|
||
// flowGraphKeeper 是链路图留存的常驻维护任务（保留期清扫），由 main 的
|
||
// startFlowGraphs 拉起（见 wire_flow_graphs.go）。
|
||
flowGraphKeeper *flowGraphKeeper
|
||
}
|
||
|
||
// newApp 依据 cfg 与 log 组装完整的 Cyrene 运行时：打开数据库、迁移表结构，接线
|
||
// 人格注册表、模型适配器、工具分发器、记忆栈、agent 循环、编排、执行、IoT、
|
||
// 渠道会话/适配器与网关，最后返回就绪的 app。
|
||
//
|
||
// 致命的接线失败（数据库打开/迁移、必需的认证、网关）以错误返回；可选子系统则
|
||
// 优雅降级，使得未配置模型密钥或外部渠道时二进制仍能启动。
|
||
func newApp(ctx context.Context, cfg *config.Config, log *slog.Logger) (*app, error) {
|
||
dataDir := resolveDataDir(cfg)
|
||
databasePath := resolveSQLitePath(cfg, dataDir)
|
||
dbCfg := cfg.DB
|
||
dbCfg.SQLitePath = databasePath
|
||
|
||
sqlDB, err := db.Open(ctx, databasePath, log)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("open database: %w", err)
|
||
}
|
||
if err := db.MigrateUp(ctx, sqlDB, log); err != nil {
|
||
_ = sqlDB.Close()
|
||
return nil, fmt.Errorf("migrate database: %w", err)
|
||
}
|
||
|
||
eventBus := bus.New()
|
||
a := &app{
|
||
cfg: cfg,
|
||
log: log,
|
||
db: sqlDB,
|
||
dbPath: databasePath,
|
||
bus: eventBus,
|
||
caps: registry.New(),
|
||
}
|
||
|
||
// 后续任一接线步骤失败时，释放已经打开的资源，但不打整套服务关闭日志
|
||
// （半途启动不等同于一次干净退出）。
|
||
completed := false
|
||
defer func() {
|
||
if completed {
|
||
return
|
||
}
|
||
if a.vindex != nil {
|
||
_ = a.vindex.Close()
|
||
}
|
||
if a.tasks != nil {
|
||
_ = a.tasks.Close()
|
||
}
|
||
if a.bus != nil {
|
||
_ = a.bus.Close()
|
||
}
|
||
if a.db != nil {
|
||
_ = a.db.Close()
|
||
}
|
||
}()
|
||
|
||
if err := a.buildPersonas(); err != nil {
|
||
return a, err
|
||
}
|
||
if err := a.wireMemory(ctx); err != nil {
|
||
return a, err
|
||
}
|
||
if err := a.wireRuntime(); err != nil {
|
||
return a, err
|
||
}
|
||
if err := a.wireAuxiliary(); err != nil {
|
||
return a, err
|
||
}
|
||
a.wirePlugins(ctx)
|
||
// 渠道事件留存面（design/26 §4；切片 E3+E4）：必须在 wireChannels 之前构造 ——
|
||
// 桥接层在构造时就把它注入（Options.EventStore），运行时不再重取。
|
||
a.wireChannelEvents()
|
||
if err := a.wireChannels(); err != nil {
|
||
return a, err
|
||
}
|
||
if err := a.wireFrontend(); err != nil {
|
||
return a, err
|
||
}
|
||
completed = true
|
||
return a, nil
|
||
}
|