AskaEth
fe1ff33115
主人 2026-10-03 拍板「注释一律中文」。本次先把本会话动过的文件就地翻； 不做全仓一次性翻译，存量按包分批排期（见 工作小结.md 待办）。 范围：ws_frame_scope.go / ws_frame_scope_test.go / broadcast.go / ws_hub_turn.go / ws_hub_message_event.go / session_read.go / server.go / ws_hub_stream_scope_test.go / history_write_broadcast_test.go / mood_test.go / extra_test.go 含中文的整行注释 18 → 382。 校验（判据 = "只改注释、零代码字节改动"）： - `git diff -U0 | 排除 +++/--- | 排除注释行/空行/行尾注释` → 为空 - 独立实现：两侧剥掉注释后逐行比对代码文本 → 11/11 文件全等，不一致数 0 - 行尾注释 7 行逐条列出，代码列与原文逐字节相同（gofmt 未要求重排） - gofmt -l 干净；全树 go build/vet 绿；go test ./... = 58 包 ok / 0 FAIL； linux/amd64 + darwin/arm64 交叉编译绿 语义未删减：部分折行英文长句合并为更少行，故整行注释数 438 → 399。
207 lines
7.5 KiB
Go
207 lines
7.5 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"
|
||
"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
|
||
channels ChannelConfigStore
|
||
// plugins 是可在运行时安装的外部工具提供方管理器。见
|
||
// Deps.Plugins。
|
||
plugins *pluginmgr.Manager
|
||
// ownerRegistrar 把扫码登录绑定的 owner id 写入 owner 白名单
|
||
// （尽力而为）。见 Deps.OwnerRegistrar。
|
||
ownerRegistrar OwnerRegistrar
|
||
webhooks map[string]http.Handler
|
||
// 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,
|
||
channels: deps.Channels,
|
||
plugins: deps.Plugins,
|
||
ownerRegistrar: deps.OwnerRegistrar,
|
||
webhooks: deps.Webhooks,
|
||
dbPath: deps.DatabasePath,
|
||
dataDir: deps.DataDir,
|
||
log: sourcedLog(deps.Log),
|
||
}
|
||
// 一个角色解析器由 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})
|
||
},
|
||
OnPersisted: s.BroadcastPersisted,
|
||
Log: s.log,
|
||
History: s.history,
|
||
})
|
||
}
|
||
|
||
// 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)
|
||
}
|