AskaEth
a6cb44b8a8
主人要求「链路流程应该有本地数据，不然每次重载之后页面就没数据了」： - 落盘：迁移 0022_flow_graphs（session_id/turn_id/graph_json/时间戳 + 唯一键 upsert）+ internal/runtime/flowgraph（RecordRound/Graph/Latest/Turns/Turn + 保留期 GC） - 写入：loop.EndRound **每轮一条**（锁外深拷贝后写；失败 WARN 且不打印图内容）；只落发布会话白名单（复用 flowSessionAllowed） - 读回：端点加 turnId + source(live/persisted/none) + turns[]（最新在前，空为 []）；**内存优先、miss 才落盘**（最新一轮永远最新） - 前端：flowGraphs 纳入 persistOptions.partialize（**刷新即刻有图**，version 3→4 + migrate）+ 历史轮选择器 + 诚实文案；顺带收 L1 转交的 flow/NodeDetail.tsx 节点日志倒序 - 新 env 键：FLOW_GRAPH_RETENTION=168h / FLOW_GRAPH_GC_INTERVAL=1h - 变异实测：每 span 一行 / 保留期守卫失效 / partialize 去掉 flowGraphs / 去掉内存优先 / 白名单失效 ⇒ 全部打红 - 3 个禁动面文件的纯追加改动（Flow.tsx 挂选择器、FlowSnapshotHint 纠已成假话的文案、其渲染断言随之更新）由主会话裁定认可 - 登记存疑：图桶持久化无体积上限；useFlowTurns 的 effect 无自动化证据（项目无 jsdom）
249 lines
9.9 KiB
Go
249 lines
9.9 KiB
Go
// Package gateway 实现 Cyrene v2 的 HTTP 网关：单二进制唯一打开的套接字。
|
||
// 它在同一个端口（:8080）上提供版本化 REST API（/api/v1）、实时
|
||
// WebSocket hub（/ws/chat）以及受 RBAC 保护的管理 API（/api/v1/admin/*）。
|
||
//
|
||
// 它是刻意精简的。所有后端能力都经由注入的依赖（构造期依赖注入）和
|
||
// 进程内 bus 流动，绝不跨进程 IPC，因此本包除网关端口外永不自行
|
||
// 打开监听器。
|
||
//
|
||
// 客户端契约（DevDocs/02-development-constraints.md §九、§十一）：
|
||
// - REST 成功响应为 {data}，错误响应为 {error:{code,message}}。
|
||
// - 认证是 JWT bearer（internal/auth）；授权是 owner>admin>viewer 的
|
||
// RBAC 层级（internal/auth）。
|
||
// - 管理写操作有审计（谁/做了什么/何时/结果）。
|
||
// - 实时事件遵循 DevDocs/design/07 §3（WS 协议）。
|
||
//
|
||
// 文件布局：contracts.go 声明注入的依赖接口，deps.go 声明 Deps 组合，
|
||
// server.go 是 Server 生命周期（构造 / 服务 / 关闭），routes.go 是路由表
|
||
// 加外层中间件，broadcast.go 是下行实时推送。
|
||
package gateway
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"fmt"
|
||
"io"
|
||
"io/fs"
|
||
"log/slog"
|
||
"net"
|
||
"net/http"
|
||
"strconv"
|
||
"time"
|
||
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/bus"
|
||
pluginmgr "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/plugin"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/config"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||
)
|
||
|
||
// source 是本包每条日志记录的来源标签。
|
||
const source = "gateway"
|
||
|
||
// Server 是网关本体。它持有聚合 REST + WS + admin 的 ServeMux 与一个
|
||
// WebSocket hub。构造完成后可安全并发使用。
|
||
type Server struct {
|
||
cfg *config.Config
|
||
auth Authenticator
|
||
bus bus.Bus
|
||
caps CapabilityStore
|
||
runner Chatter
|
||
sessions SessionStore
|
||
personas PersonaStore
|
||
memories MemoryStore
|
||
history HistoryStore
|
||
iot IoTStore
|
||
auditor SubagentAuditor
|
||
taskOutputs TaskOutputStore
|
||
automations AutomationStore
|
||
users UserStore
|
||
refresh RefreshGrantStore
|
||
backthink BackthinkStore
|
||
rag RagStore
|
||
obs *observability
|
||
voice VoiceService
|
||
streaming bool
|
||
frontend fs.FS
|
||
dbPath string
|
||
// dataDir 是支撑上传存储的已解析数据目录（<dataDir>/uploads）。
|
||
// 由组合根注入；cfg.DataDir 是回退值。见 Deps.DataDir。
|
||
dataDir string
|
||
// llm 是可热切换的提供方栈（models.json）。为 nil 时 admin 的 /llm
|
||
// 端点回退到环境变量配置。它是可选的，以便未接它的测试仍能工作。
|
||
llm *model.RuntimeLLM
|
||
// envPath 是启动时解析的 `.env` 路径（见 Deps.EnvPath）：只供
|
||
// 「重置为 env 默认」动作按既有解析入口重读 env 默认层，网关不 fopen `.env`。
|
||
envPath string
|
||
channels ChannelConfigStore
|
||
// plugins 是可在运行时安装的外部工具提供方管理器。见
|
||
// Deps.Plugins。
|
||
plugins *pluginmgr.Manager
|
||
// ownerRegistrar 把扫码登录绑定的 owner id 写入 owner 白名单
|
||
// （尽力而为）。见 Deps.OwnerRegistrar。
|
||
ownerRegistrar OwnerRegistrar
|
||
webhooks map[string]http.Handler
|
||
// usage 是 LLM 用量汇总的只读出口（切片 U1，见 Deps.Usage）。为 nil 时
|
||
// GET /api/v1/admin/llm/usage 回 500 usage_unavailable。
|
||
usage LlmUsageStore
|
||
// channelEvents 是渠道事件留存的只读查询出口（切片 E6，见 Deps.EventStore）。
|
||
// 为 nil 时 GET /api/v1/admin/channels/events（及其 /counters）回
|
||
// 500 channel_events_unavailable —— 未接线就说未接线，不回全 0 假汇总。
|
||
channelEvents ChannelEventStore
|
||
// db 是组合根注入的活 SQLite 连接池（见 Deps.Database）。管理侧
|
||
// 「存储状态」只读端点用它读页级统计与 documents 表；为 nil 时该
|
||
// 端点回 500 storage_unavailable（不另开池、不假装有数据）。
|
||
db *sql.DB
|
||
// vector 是组合根报告的 pgvector 增强运行时状态（见
|
||
// Deps.Vector / VectorState）。零值 = 未配置。
|
||
vector VectorState
|
||
// maxParallelTurns 是单会话在途并行路数的上限（见 Deps.MaxParallelTurns）。
|
||
// 零值 = 未接线 ⇒ 编排器用默认 2。
|
||
maxParallelTurns int
|
||
// flow 是「最近一轮链路图」的只读出口（见 Deps.Flow / FlowGraphSource）。
|
||
// 为 nil 时 GET /api/v1/admin/flow/graph 回 500 flow_unavailable。
|
||
flow FlowGraphSource
|
||
// flowGraphs 是**已落盘轮次图**的只读出口（契约 v0.73；见 Deps.FlowGraphs）。
|
||
// 为 nil 时该端点回退到只有内存的行为（turns 为空、turnId 只能命中内存那一轮）。
|
||
flowGraphs FlowGraphStore
|
||
// envResolver 解析会话每轮的 loop 环境（其角色）。见
|
||
// persona_env.go / design/22 §4.5。
|
||
envResolver turnEnvResolver
|
||
log *slog.Logger
|
||
hub *hub
|
||
}
|
||
|
||
// New 校验 cfg 与 deps 并返回就绪的 Server。Auth 必填；WebSocket hub
|
||
// 会立即订阅 deps.Bus，因此 ListenAndServe 之前发布的事件仍会转发给
|
||
// 客户端。
|
||
func New(cfg *config.Config, deps Deps) (*Server, error) {
|
||
if cfg == nil {
|
||
return nil, fmt.Errorf("gateway: nil config")
|
||
}
|
||
if deps.Auth == nil {
|
||
return nil, fmt.Errorf("gateway: auth dependency required")
|
||
}
|
||
s := &Server{
|
||
cfg: cfg,
|
||
auth: deps.Auth,
|
||
bus: deps.Bus,
|
||
caps: deps.Registry,
|
||
runner: deps.Loop,
|
||
sessions: deps.Session,
|
||
personas: deps.Personas,
|
||
memories: deps.Memories,
|
||
history: deps.History,
|
||
iot: deps.IoT,
|
||
auditor: deps.Scheduler,
|
||
taskOutputs: deps.Tasks,
|
||
automations: deps.Automations,
|
||
users: deps.Users,
|
||
refresh: deps.Refresh,
|
||
backthink: deps.Backthink,
|
||
rag: deps.RAG,
|
||
obs: newObservability(deps),
|
||
streaming: deps.Streaming,
|
||
frontend: deps.Frontend,
|
||
llm: deps.LLM,
|
||
envPath: deps.EnvPath,
|
||
channels: deps.Channels,
|
||
plugins: deps.Plugins,
|
||
ownerRegistrar: deps.OwnerRegistrar,
|
||
webhooks: deps.Webhooks,
|
||
usage: deps.Usage,
|
||
channelEvents: deps.EventStore,
|
||
dbPath: deps.DatabasePath,
|
||
dataDir: deps.DataDir,
|
||
db: deps.Database,
|
||
vector: deps.Vector,
|
||
log: sourcedLog(deps.Log),
|
||
maxParallelTurns: deps.MaxParallelTurns,
|
||
flow: deps.Flow,
|
||
flowGraphs: deps.FlowGraphs,
|
||
}
|
||
// 一个角色解析器由 Server 各处理器与 hub 共用，使每个面向 owner
|
||
// 的入口（REST、WS、语音）以同样方式解析会话角色（design/22 §4.5），
|
||
// 而不是各自硬编码 loop.OwnerEnv()。
|
||
s.envResolver = turnEnvResolver{history: deps.History, personas: deps.Personas, log: s.log}
|
||
s.hub = newHub(s)
|
||
s.hub.orchestrator = newOrchestrator(s)
|
||
s.hub.start()
|
||
return s, nil
|
||
}
|
||
|
||
// Start 在 goroutine 中启动网关，返回携带 ListenAndServe 结果的
|
||
// channel。服务器运行至 ctx 被取消。
|
||
func (s *Server) Start(ctx context.Context) (<-chan error, error) {
|
||
errCh := make(chan error, 1)
|
||
go func() { errCh <- s.ListenAndServe(ctx) }()
|
||
return errCh, nil
|
||
}
|
||
|
||
// ListenAndServe 绑定配置的 host:port（默认 :8080）并阻塞至 ctx 被
|
||
// 取消，随后优雅地排空在途请求。
|
||
func (s *Server) ListenAndServe(ctx context.Context) error {
|
||
if ctx == nil {
|
||
ctx = context.Background()
|
||
}
|
||
addr := net.JoinHostPort(s.cfg.Server.Host, strconv.Itoa(s.cfg.Server.Port))
|
||
httpSrv := &http.Server{Addr: addr, Handler: s.Handler()}
|
||
|
||
done := make(chan struct{})
|
||
go func() {
|
||
<-ctx.Done()
|
||
s.log.Info("gateway shutting down", "addr", addr)
|
||
s.hub.close()
|
||
shCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
|
||
defer cancel()
|
||
_ = httpSrv.Shutdown(shCtx)
|
||
close(done)
|
||
}()
|
||
|
||
s.log.Info("gateway listening", "addr", addr)
|
||
err := httpSrv.ListenAndServe()
|
||
if err != nil && err != http.ErrServerClosed {
|
||
return fmt.Errorf("gateway: listen: %w", err)
|
||
}
|
||
<-done
|
||
return nil
|
||
}
|
||
|
||
// Close 释放 hub 的 bus 订阅并关闭存活连接。
|
||
func (s *Server) Close() { s.hub.close() }
|
||
|
||
// newOrchestrator 构建绑定到本 server 的 hub 的会话消息 orchestrator
|
||
// （design/15）。TurnRunner 复用 hub 的 runTurn（因此一轮总是发出相同的
|
||
// 流契约），StateSink 则把每会话状态迁移经由 hub 广播扇出给前端 ——
|
||
// 这些迁移属于 design/11 §7.1 的「状态机信号」，因此每个已认证连接都
|
||
// 可以跟随它们。它被包在函数里，以便组合根之后能覆盖注入（一个显式
|
||
// 的接线点，而不是硬编码的构造）。
|
||
func newOrchestrator(s *Server) *Orchestrator {
|
||
return NewOrchestrator(OrchestratorOptions{
|
||
Run: func(ctx context.Context, sessionID string, messages []model.Message, env loop.SessionEnv) (*loop.Result, error) {
|
||
return s.hub.runTurn(ctx, sessionID, messages, env)
|
||
},
|
||
Emit: func(sessionID, typ string, data map[string]any) {
|
||
s.hub.broadcast(wsEvent{Type: typ, SessionID: sessionID, Data: data})
|
||
},
|
||
// EmitTurn 是带 turnId 的变体（design/15 §2.5.9）：帧壳上的 turnId 让客户端
|
||
// 把「这条状态帧属于哪一路」与流帧对齐 —— 真并行下同一会话有两条交错的路，
|
||
// 只按 sessionId 分桶会把它们糊在一起。
|
||
EmitTurn: func(sessionID, turnID, typ string, data map[string]any) {
|
||
s.hub.broadcast(wsEvent{Type: typ, SessionID: sessionID, TurnID: turnID, Data: data})
|
||
},
|
||
OnPersisted: s.BroadcastPersisted,
|
||
Log: s.log,
|
||
History: s.history,
|
||
MaxParallelTurns: s.maxParallelTurns,
|
||
})
|
||
}
|
||
|
||
// sourcedLog 返回带 source "gateway" 标签的 logger，nil 时替换为丢弃型
|
||
// logger，使 Server 总能安全地打日志。
|
||
func sourcedLog(l *slog.Logger) *slog.Logger {
|
||
if l == nil {
|
||
l = slog.New(slog.NewTextHandler(io.Discard, nil))
|
||
}
|
||
return logger.WithSource(l, source)
|
||
}
|