AskaEth
c1fa1e9b9a
- 语义：只把 .env **确实声明过**的键覆盖回 env 值；env 未声明的键（fallback/multimodal/maxTokens/reasoning/personaModels）原样保留；**组集合不增减**（不复活被面板删的组、不删面板新增的组）；无键可改时不写盘。成功即热重载 LLM 栈。返回 {reset:[{group,keys}], skipped:[{group,reason}], path}，**只回键名、不回显凭据值**。
- 护栏：必须 confirm:true（缺失/false/非法 body ⇒ 400，且**先于任何读写**）；**owner-only**（与同页 POST /admin/llm、/admin/llm/test 同门 —— 这是破坏性写，门不比同族更宽）；审计动作 llm.resetFromEnv（camelCase，全仓多词动作的既有约定）。失败路径明确报错（env_defaults_unreadable / models_unreadable / llm_unavailable），不假成功。
- 组合根注入 Deps.EnvPath（网关不 fopen .env）：请求时按既有解析入口重读，语义 =「退回 .env 现在声明的默认值」；.env 层里的 provider 是同义反复，不计入「env 为该组声明了什么」的判定（有专门用例）。
- 前端：LLM 页「重置为 env 默认」卡片（中文二次确认 → 提交 → 按 reset/skipped 渲染 → 成功后重拉配置），标签 owner 专属（与 System.tsx 同款）。
- 契约 v0.62：api-client（权限矩阵改回 12+10+38=60、端点表 9b、重置契约小节、写操作契约、changelog、banner）与 api-reference（端点表 #70、§5.13a 详情含 curl、§8.2/§8.3/§8.4）同步；字段名三方逐字对齐。
- 主会话独立验收：隔离 worktree（R1 的 19 文件）build/vet 绿、gateway+config 三包绿、R1 用例 10 PASS、reset_from_env 残留 0；**变异亲自复现**（去掉 confirm 校验 ⇒ RequiresConfirm 红：status = 200, want 400，且响应体显示面板配置真的被改）。
- 子代理自报并已处理：测试文件触 500 行上限 ⇒ 拆成语义侧 + 护栏侧两份；门改回 owner-only 后前端标签会误导 admin ⇒ 改「owner 专属」并重建 bundle。任务书本身仍是旧口径（owner|admin / snake_case）—— 已由主会话登记修正说明，避免压缩后有人按任务书回退。
- 未做（诚实边界）：不支持按组重置；不纳入 channels.json/plugins.json；不自动重启（contextWindow 的预算推导仍只在启动时算一次）。新端点需下一次 build+restart 才在 :10942 生效。
221 lines
8.2 KiB
Go
221 lines
8.2 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
|
||
// db 是组合根注入的活 SQLite 连接池（见 Deps.Database）。管理侧
|
||
// 「存储状态」只读端点用它读页级统计与 documents 表；为 nil 时该
|
||
// 端点回 500 storage_unavailable（不另开池、不假装有数据）。
|
||
db *sql.DB
|
||
// vector 是组合根报告的 pgvector 增强运行时状态（见
|
||
// Deps.Vector / VectorState）。零值 = 未配置。
|
||
vector VectorState
|
||
// 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,
|
||
dbPath: deps.DatabasePath,
|
||
dataDir: deps.DataDir,
|
||
db: deps.Database,
|
||
vector: deps.Vector,
|
||
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)
|
||
}
|