Files
Cyrene_v2/cmd/cyrene/wire.go
T
Aska 8eaf81bd0d
CI / Backend (build / vet / test / cross-compile) (push) Has been cancelled
CI / Frontend (tsc + vite build) (push) Has been cancelled
feat(engage): 群聊自主参与度引擎核心(design/18 P2a+P2b)
- persona 加 triggerWords(昔涟/Cyrene/有瑕的追忆/追忆);Loader 解析
- light.go: shouldRespond 拆分 legacy 与 engage 路径;LightInput/LightDecision 扩展
- engage.go(新): EngageConfig + IDLE/PROBING/ACTIVE/COOLDOWN 状态机(按 platform:groupID 内存) +
  概率触发(可注入rand) + 角色名提词触发 + 活动期跟随(话题转移/连续无感/窗口) + 负面态度冷避
- engagejudge.go(新): NewLLMLightJudge,一次成本受控轻LLM判 值不值得回(不判身份),
  解析失败/超时 fail-closed 静默
- channelbridge.go: Options(TriggerWords/PersonaPrompt/Engage/EngageFor/EngageRand) + 字段初始化
- wire.go: 接线 Light+TriggerWords+PersonaPrompt+Engage(默认概率0.10)
- 红线: 未触发/未命中概率绝不调LLM; 身份靠actor元数据; 敏感不落日志; 零新依赖
- 测试: engage_test.go(触发/状态机/冷避/成本受控/JUDGE输入) + persona triggerWords 用例
- 验收: go build/vet/test ./... 全绿
2026-09-07 18:16:55 +08:00

774 lines
27 KiB
Go

package main
import (
"context"
"database/sql"
"encoding/json"
"fmt"
"log/slog"
"os"
"path/filepath"
"strings"
"time"
"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
"git.yeij.top/AskaEth/Cyrene_v2/internal/bus"
"git.yeij.top/AskaEth/Cyrene_v2/internal/capability/mcp"
"git.yeij.top/AskaEth/Cyrene_v2/internal/capability/registry"
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session"
"git.yeij.top/AskaEth/Cyrene_v2/internal/config"
"git.yeij.top/AskaEth/Cyrene_v2/internal/config/runtimeconfig"
"git.yeij.top/AskaEth/Cyrene_v2/internal/db"
"git.yeij.top/AskaEth/Cyrene_v2/internal/iot"
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
"git.yeij.top/AskaEth/Cyrene_v2/internal/persona/loader"
personareg "git.yeij.top/AskaEth/Cyrene_v2/internal/persona/registry"
"git.yeij.top/AskaEth/Cyrene_v2/internal/platform"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/capacity"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/embedding"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/exec"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/guardrail"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/knowledge"
"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/vision"
"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/orchestration/audit"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/persist"
"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/runtime/trace"
"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"
"git.yeij.top/AskaEth/Cyrene_v2/plugins/tools/basic"
iotools "git.yeij.top/AskaEth/Cyrene_v2/plugins/tools/io"
)
// defaultOwnerPath and defaultPersonasDir point at the on-disk persona/owner
// definitions shipped with the repo; they are relative to the working directory.
const (
defaultOwnerPath = "configs/users/owner.yaml"
defaultPersonasDir = "configs/personas"
defaultSQLitePath = "./data/cyrene.db"
fallbackInstruction = "You are Cyrene, a warm and attentive companion."
// Bound/knob constants. Compile-time so no bare magic number appears in the
// runtime path (project rule §10).
defaultMaxRounds = 8
defaultMaxParallel = 4
defaultAgentID = "orchestrator"
defaultToolTimeout = 30 * time.Second
)
// app holds every subsystem the single Cyrene binary assembles. It is built by
// newApp and torn down by close on shutdown.
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
dispatcher *tool.Dispatcher
memStore *memory.Store
retriever *memory.Retriever
projStore *projectindex.Store
loop *loop.Loop
scheduler *orchestration.Scheduler
tasks *task.Manager
execProv exec.Provider
iot *iot.Manager
voice *voice.Chain
mcpRegs []*mcp.Registry
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
// capacitySink broadcasts capacity_awaken transitions to the Web UI (design/14
// §5). It is satisfied by *gateway.Server (BroadcastCapacityAwaken); held as an
// interface so the composition root is unit-testable with a fake sink.
capacitySink capacityEmitter
}
// capacityEmitter is the gateway-facing sink for capacity_awaken broadcasts.
// *gateway.Server implements it; tests substitute a fake to capture the args.
type capacityEmitter interface {
BroadcastCapacityAwaken(sessionID, fromState, toState, tool string)
}
// newApp assembles the full Cyrene runtime from cfg and log. It opens the
// database, migrates the schema, wires the persona registry, model adapter, tool
// dispatcher, memory stack, agent loop, orchestration, execution, IoT, channel
// sessions/adapters and the gateway, and returns the ready app.
//
// Fatal wiring failures (database open/migrate, required auth, gateway) are
// returned as errors; optional subsystems degrade gracefully so the binary can
// boot even when a model key or external channel is not configured.
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(),
}
// Release already-opened resources if any later wiring step fails, without
// the full-service shutdown log (a partial boot is not a clean shutdown).
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
}
if err := a.wireChannels(); err != nil {
return a, err
}
if err := a.wireFrontend(); err != nil {
return a, err
}
completed = true
return a, nil
}
// buildPersonas loads the persona definitions and registers them, plus the
// default character and the owner profile used by channel identity resolution.
func (a *app) buildPersonas() error {
a.personas = personareg.New()
dirPersonas, err := loader.LoadDir(defaultPersonasDir)
if err != nil {
a.log.Warn("persona dir load skipped", "error", err)
} else {
for _, p := range dirPersonas {
if rerr := a.personas.RegisterPersona(p); rerr != nil {
a.log.Warn("persona register skipped", "persona", p.ID, "error", rerr)
}
}
}
if owner, oerr := user.LoadOwnerProfile(defaultOwnerPath); oerr == nil {
a.owner = owner
} else {
a.log.Warn("owner profile load skipped", "error", oerr)
}
// Load the default persona explicitly if it was not under the personas dir.
defaultID := "cyrene"
if a.owner != nil && a.owner.DefaultPersona != "" {
defaultID = a.owner.DefaultPersona
}
if _, ok := a.personas.Persona(defaultID); !ok {
if p, lerr := loader.Load(filepath.Join(defaultPersonasDir, defaultID+".yaml")); lerr == nil {
_ = a.personas.RegisterPersona(p)
} else {
a.log.Warn("default persona load skipped", "persona", defaultID, "error", lerr)
}
}
if p, ok := a.personas.Persona(defaultID); ok {
char := &personareg.Character{ID: p.ID, Name: p.Name, Default: true, PersonaID: p.ID}
if cerr := a.personas.RegisterCharacter(char); cerr != nil {
a.log.Warn("default character register skipped", "error", cerr)
}
}
a.log.Info("personas ready", "count", a.personas.CountPersonas(), "default", defaultID)
return nil
}
// wireMemory builds the long-term memory stack: store, embedding manager, vector
// index, project index and the retriever wired into the io knowledge tool.
func (a *app) wireMemory(ctx context.Context) error {
store, err := memory.NewStore(ctx, a.db, a.log)
if err != nil {
return fmt.Errorf("memory store: %w", err)
}
a.memStore = store
embedManager := embedding.New(a.cfg.Embedding)
embedder := embedManager.Embedder()
index, err := vector.New(a.dbCfgForEmbed(), embedder, a.log)
if err != nil {
a.log.Warn("vector index unavailable, falling back to in-memory semantic", "error", err)
} else {
a.vindex = index
}
projStore, err := projectindex.New(ctx, a.db, a.log)
if err != nil {
a.log.Warn("project index unavailable", "error", err)
projStore = nil
}
a.projStore = projStore
opts := []memory.RetrieverOption{memory.WithEmbedder(embedder)}
if a.vindex != nil {
opts = append(opts, memory.WithVectorIndex(a.vindex))
}
if projStore != nil {
opts = append(opts, memory.WithProjectIndex(projStore))
}
a.retriever = memory.NewRetriever(store, opts...)
return nil
}
// wireRuntime builds the model adapter, tool dispatcher, guardrails, the agent
// and its loop, tracing, persistence and the orchestration scheduler.
func (a *app) wireRuntime() error {
// The LLM provider stack is hot-swappable: it loads models.json (or falls
// back to .env) and the panel save endpoint reloads it without a restart
// (design/04 §3.1). a.modelAdapter stays an Adapter (the runtime stack
// implements it) for the channel-bridge review model (design/12 §6 策略四).
runtimeStack := model.NewRuntimeLLM(runtimeconfig.DefaultPath(resolveDataDir(a.cfg)), a.cfg.LLM, a.log)
a.modelAdapter = runtimeStack
a.llm = runtimeStack
dispatcher, derr := tool.New(a.caps, tool.WithDefaultTimeout(defaultToolTimeout))
if derr != nil {
return fmt.Errorf("tool dispatcher: %w", derr)
}
a.dispatcher = dispatcher
if err := basic.Register(a.caps); err != nil {
return fmt.Errorf("register basic tools: %w", err)
}
ioTools, ioErr := iotools.New(a.cfg, iotools.WithKnowledgeRetriever(a.retriever), iotools.WithLogger(a.log))
if ioErr != nil {
return fmt.Errorf("build io tools: %w", ioErr)
}
for _, t := range ioTools {
if rerr := a.dispatcher.Register(t); rerr != nil {
a.log.Warn("io tool register skipped", "tool", t.Name(), "error", rerr)
}
}
a.wireVision()
guards := guardrail.New()
instructions := a.defaultInstructions()
ag, aerr := agent.New(instructions, a.dispatcher, guards,
agent.WithModel(runtimeStack, runtimeStack),
agent.WithOptions(agent.Options{MaxRounds: defaultMaxRounds, Purpose: "chat"}))
if aerr != nil {
return fmt.Errorf("agent: %w", aerr)
}
personaID := a.defaultPersonaID()
store := a.memStore
tracer := trace.New(trace.NewJSONRecorder(os.Stderr))
checkpoint, perr := persist.New(context.Background(), a.db, a.log)
if perr != nil {
return fmt.Errorf("checkpoint store: %w", perr)
}
// Vision routing: the loop rewrites an image-bearing user message per the
// vision Router — a vision-capable main model sees the image directly,
// otherwise the vision/OCR tool's text description replaces it (design/06 §0).
visionRouter := vision.NewRouter(
vision.WithVisionTool(visionToolName),
vision.WithOCRTool(ocrToolName),
)
visionResolver := vision.NewMessageResolver(
visionRouter,
func() vision.ModelInfo {
return vision.ModelInfo{Name: a.cfg.LLM.VisionModel, SupportsVision: a.cfg.LLM.VisionModel != ""}
},
a.dispatcher.Invoke,
)
capDecider := a.buildCapacity()
a.loop = loop.New(loop.Options{
Agent: ag,
Tools: a.dispatcher,
Capacity: capDecider,
OnCapacity: a.onCapacity,
OnAudit: a.onToolAudit,
Memory: store,
Retriever: a.retriever,
Extractor: memory.NewManualExtractor(personaID, "owner"),
Checkpoint: checkpoint,
Tracer: tracer,
Guards: guards,
PersonaID: personaID,
Timezone: a.cfg.Timezone,
MaxRounds: defaultMaxRounds,
WorkingMemory: memory.NewWorkingMemory(0),
OnStream: a.streamChunk,
Vision: visionResolver,
Log: logger.WithSource(a.log, "runtime/loop"),
})
a.scheduler = orchestration.New(orchestration.Options{
MaxParallel: defaultMaxParallel,
Loop: a.loop,
Bus: a.bus,
Memory: store,
Tracer: tracer,
Approver: audit.NewApprover(a.log),
Logger: logger.WithSource(a.log, "orchestration"),
AgentID: defaultAgentID,
})
return nil
}
// buildCapacity constructs the awakened-capacity decider when enabled, or nil
// (which keeps the loop's default all-tools behavior). It is opt-in via
// CAPACITY_ENABLED; the allowed-set mapping comes from CAPACITY_TOOL_SETS and
// falls back to built-in defaults on a parse error.
func (a *app) buildCapacity() *capacity.Decider {
if !a.cfg.Capacity.Enabled {
return nil
}
sets, err := parseCapacityToolSets(a.cfg.Capacity.ToolSetsJSON)
if err != nil {
a.log.Warn("capacity tool sets parse failed, using built-in defaults", "error", err)
sets = nil
}
return capacity.New(capacity.Config{
Enabled: true,
MinConfidence: a.cfg.Capacity.MinConfidence,
ToolSets: sets,
}, capacity.WithLogger(a.log))
}
// onCapacity is the loop's OnCapacity sink: it observes a capacity_awaken
// transition and pushes it to the Web UI (design/14 §5) as a `capacity_awaken`
// frame. The loop only emits on a real state transition (never every round), and
// the event carries the originating session id so it routes to that session, not
// globally. The payload is the frontend AwakenPayload {fromState,toState,tool}.
func (a *app) onCapacity(ev capacity.Event) {
logger.WithSource(a.log, "cmd/cyrene").Debug("capacity event",
"type", ev.Type, "from", ev.FromState, "to", ev.ToState,
"tool", ev.Tool, "at", ev.At, "session", ev.SessionID)
if a.capacitySink == nil {
return
}
a.capacitySink.BroadcastCapacityAwaken(ev.SessionID, string(ev.FromState), string(ev.ToState), ev.Tool)
}
// onToolAudit is the loop's OnAudit sink: it forwards a tool call denied by the
// design/11 §3 security gate in a group/stranger environment into the gateway
// audit trail, so the rejection is "audit 可查" and the frontend can render the
// small "xx 工具在群聊被拒" note (design/11 §8). The gateway is built after the
// loop (wireFrontend), so the sink reads a.gw at call time — by then the app is
// fully assembled and any Run is served through the gateway.
func (a *app) onToolAudit(ev loop.AuditEvent) {
if a.gw == nil {
logger.WithSource(a.log, "cmd/cyrene").Warn("loop audit dropped: gateway not ready",
"who", ev.Who, "what", ev.What, "result", ev.Result)
return
}
a.gw.AuditEvent(ev.Who, ev.What, ev.Result)
}
// parseCapacityToolSets decodes the CAPACITY_TOOL_SETS JSON map of
// state→allowed tool name list. A blank string yields nil (built-in defaults); a
// null value for a state means unrestricted (all tools).
func parseCapacityToolSets(raw string) (map[capacity.State][]string, error) {
if strings.TrimSpace(raw) == "" {
return nil, nil
}
var m map[string][]string
if err := json.Unmarshal([]byte(raw), &m); err != nil {
return nil, fmt.Errorf("capacity tool sets: %w", err)
}
out := make(map[capacity.State][]string, len(m))
for k, v := range m {
out[capacity.State(k)] = v
}
return out, nil
}
// wireAuxiliary builds the execution provider, task manager, IoT manager,
// channel session manager, automations and background thinking services.
func (a *app) wireAuxiliary() error {
provider, err := exec.New(a.cfg.ExecSandbox, a.log)
if err != nil {
return fmt.Errorf("exec provider: %w", err)
}
a.execProv = provider
tmgr, err := task.New(a.cfg, provider, a.log, task.WithBus(a.bus))
if err != nil {
return fmt.Errorf("task manager: %w", err)
}
a.tasks = tmgr
iotMgr, err := iot.New(a.cfg.IoT, a.log, iot.WithBus(a.bus))
if err != nil {
return fmt.Errorf("iot: %w", err)
}
a.iot = iotMgr
sess, err := session.New(context.Background(), a.db, a.log)
if err != nil {
return fmt.Errorf("session manager: %w", err)
}
a.sessions = sess
hist, herr := history.NewStore(context.Background(), a.db, a.log)
if herr != nil {
return fmt.Errorf("history store: %w", herr)
}
a.history = hist
automation, err := automations.NewService(automations.Config{}, automations.Deps{
Bus: a.bus,
Loop: a.loop,
Log: logger.WithSource(a.log, "services/automations"),
})
if err != nil {
return fmt.Errorf("automations: %w", err)
}
a.automation = automation
bt, err := backthink.New(backthink.ConfigFromBackground(a.cfg.Background), backthink.Deps{
Bus: a.bus,
Thinker: backthink.NewLoopThinker(a.loop),
State: backthink.NewIoTStateProvider(iotMgr, a.log),
DB: a.db,
Log: logger.WithSource(a.log, "services/backthink"),
})
if err != nil {
return fmt.Errorf("backthink: %w", err)
}
a.backthink = bt
a.wireVoice()
a.wireMCP()
a.wireSkills()
a.wireTerminalTool()
return nil
}
// wireChannels constructs the four IM channel adapters from environment
// configuration, registers them so the admin API can enumerate them, and wires
// them into the channel bridge so a configured adapter connects and routes
// inbound messages through session→actor→loop→reply. Adapters without
// credentials stay disconnected (logged, never fatal): connecting with an empty
// config would either fail or, for weixin-ilink, trigger a blocking scan-code
// login.
func (a *app) wireChannels() error {
// The channel config is runtime-editable: channels.json (when present and
// valid) is authoritative, otherwise the env channel block is the fallback
// (design/04 §3.1). The factory and the store read the same file so a panel
// save + hot-reload keeps boot and runtime consistent.
channelsPath := runtimeconfig.DefaultChannelsPath(resolveDataDir(a.cfg))
factory := &channelFactory{path: channelsPath, fallback: channelAccountsFromEnv(), log: a.log}
adapters := buildChannelAdapters(factory)
for _, ch := range adapters {
capp := registry.Capability{
Name: ch.Name(),
Kind: registry.KindChannel,
Description: ch.Name(),
Value: ch,
}
if err := a.caps.Register(capp); err != nil {
a.log.Warn("channel register skipped", "channel", ch.Name(), "error", err)
}
}
bridge, berr := channelbridge.New(channelbridge.Options{
Log: logger.WithSource(a.log, "services/channelbridge"),
Loop: a.loop,
Sessions: a.sessions,
Owner: a.owner,
History: a.history,
QuietStartHour: a.cfg.Background.QuietStartHour,
QuietEndHour: a.cfg.Background.QuietEndHour,
// Optional review model prunes low-value queued messages near the per-adapter
// cap (design/12 §6 策略四). Nil adapter → the gate falls back to FIFO evict.
Review: reviewFor(a.modelAdapter, logger.WithSource(a.log, "services/channelbridge/review")),
// Light is the design/18 §3 participation judge: a cost-controlled light LLM
// decision on a triggered group/stranger message (role-name mention / probability
// / active window). Only an actual trigger feeds the group context.
Light: engageLightFor(a.modelAdapter, logger.WithSource(a.log, "services/channelbridge/engage")),
TriggerWords: a.defaultTriggerWords(),
PersonaPrompt: a.defaultInstructions(),
Engage: channelbridge.DefaultEngageConfig(),
// EngageFor is left nil (P2a); per-group overrides arrive with P2c.
// Factory lets the panel save hot-replace a platform's adapters without a
// restart (design/04 §3.1).
Factory: factory,
})
if berr != nil {
return fmt.Errorf("channel bridge: %w", berr)
}
for _, ch := range adapters {
bridge.Add(ch)
}
bridge.Start()
a.channelBridge = bridge
a.channels = &runtimeChannels{path: channelsPath, fallback: factory.fallback, bridge: bridge, log: a.log}
a.log.Info("channel adapters registered", "count", len(adapters))
return nil
}
// runtimeChannels is the gateway-facing view of the runtime-editable channel
// config. Load re-reads channels.json (file wins → env fallback) and Reload
// delegates to the bridge's hot-replace, so the panel's save/reload path drives
// the same source the bridge boots from.
type runtimeChannels struct {
path string
fallback map[string][]config.ChannelAccount
bridge *channelbridge.Bridge
log *slog.Logger
}
func (r *runtimeChannels) Path() string { return r.path }
func (r *runtimeChannels) Load() map[string][]config.ChannelAccount {
accounts, warn := runtimeconfig.LoadChannels(r.path, r.fallback)
if warn != nil {
r.log.Warn("channels.json load fell back to env", "error", warn)
}
return accounts
}
func (r *runtimeChannels) Reload(platform string) error { return r.bridge.Reload(platform) }
// wireFrontend builds the gateway with the embedded/external SPA filesystem and
// the JWT/RBAC authenticator. Session/Persona/Memories are wired to the concrete
// managers so the admin/user list endpoints return real data.
func (a *app) wireFrontend() error {
jwt := auth.New(a.cfg.Auth)
apiKeys := auth.NewAPIKey(a.cfg.Auth)
ragStore := knowledge.New(knowledge.Deps{
Projects: a.projStore,
Retriever: a.retriever,
DefaultPersona: a.defaultPersonaID(),
VectorOn: a.vindex != nil,
Log: a.log,
})
auditStore, aerr := gateway.NewSQLiteAuditStore(context.Background(), a.db, a.log)
if aerr != nil {
a.log.Warn("audit store unavailable, falling back to in-memory", "error", aerr)
auditStore = nil
}
// a.history is built in wireAuxiliary so both the channel bridge (wireChannels)
// and the gateway share the same persistent store. History was a no-op dependency
// before the IM path, so nothing downstream depends on its creation order change.
gw, err := gateway.New(a.cfg, gateway.Deps{
Log: logger.WithSource(a.log, "gateway"),
Auth: jwt,
Bus: a.bus,
Registry: a.caps,
Loop: a.loop,
Session: a.sessions,
Personas: a.personas,
Memories: a.memStore,
History: a.history,
IoT: a.iot,
Scheduler: a.scheduler,
Tasks: a.tasks,
Automations: a.automation,
Users: apiKeys,
Backthink: a.backthink,
RAG: ragStore,
LLM: a.llm,
Channels: a.channels,
AuditStore: auditStore,
Streaming: true,
Frontend: resolveFrontendFS(a.log),
Webhooks: a.channelBridge.WebhookHandlers(),
DatabasePath: a.dbPath,
})
if err != nil {
return fmt.Errorf("gateway: %w", err)
}
gw.SetVoice(a.voice)
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
// IM messages with one, uniform frame (design/16 §4.4).
a.channelBridge.SetOnPersisted(a.gw.BroadcastPersisted)
// design/16 §4.6 §6: a message dropped by the platform's allow/deny
// black-or-white lists is recorded in the gateway audit trail so the
// rejection is "audit 可查" even though it never reaches the loop.
a.channelBridge.SetOnRejected(func(msg adapter.UnifiedMessage, reason string) {
a.gw.AuditEvent("bridge:"+msg.Platform, "channel.reject", reason)
})
}
a.gw = gw
a.capacitySink = gw
return nil
}
// dbCfgForEmbed returns the DB config used by the vector index, which opens the
// same SQLite file as the primary pool.
func (a *app) dbCfgForEmbed() config.DBConfig {
cfg := a.cfg.DB
cfg.SQLitePath = resolveSQLitePath(a.cfg, resolveDataDir(a.cfg))
return cfg
}
// defaultPersonaID returns the id of the default persona, falling back to
// "cyrene".
func (a *app) defaultPersonaID() string {
if c, ok := a.personas.DefaultCharacter(); ok {
return c.PersonaID
}
return "cyrene"
}
// defaultInstructions returns the default persona's system prompt, falling back
// to its role text or a minimal instruction when neither is set.
func (a *app) defaultInstructions() string {
if c, ok := a.personas.DefaultCharacter(); ok {
if p, ok := a.personas.Persona(c.PersonaID); ok {
if p.Identity.SystemPrompt != "" {
return p.Identity.SystemPrompt
}
if p.Identity.Role != "" {
return p.Identity.Role
}
}
}
return fallbackInstruction
}
// close shuts down every subsystem in dependency order and releases resources.
func (a *app) close(ctx context.Context) {
if a.backthink != nil {
_ = a.backthink.Stop(ctx)
}
if a.automation != nil {
_ = a.automation.Stop(ctx)
}
if a.channelBridge != nil {
a.channelBridge.Stop(ctx)
}
if a.scheduler != nil {
_ = a.scheduler.CancelAll(ctx)
}
if a.gw != nil {
a.gw.Close()
}
for _, reg := range a.mcpRegs {
_ = reg.Unregister(ctx)
}
if a.tasks != nil {
_ = a.tasks.Close()
}
if a.vindex != nil {
_ = a.vindex.Close()
}
if a.bus != nil {
_ = a.bus.Close()
}
if a.db != nil {
_ = a.db.Close()
}
a.log.Info("cyrene shutdown complete")
}
// resolveDataDir returns the configured data directory (from .env / env, via
// cfg.DataDir), falling back to the OS-appropriate default only when no
// directory is configured.
func resolveDataDir(cfg *config.Config) string {
if cfg.DataDir != "" {
return cfg.DataDir
}
return platform.DefaultDataDir()
}
// resolveSQLitePath returns the SQLite path to use, deriving it under the data
// directory when the configured default is left untouched.
func resolveSQLitePath(cfg *config.Config, dataDir string) string {
p := cfg.DB.SQLitePath
if strings.TrimSpace(p) == "" || p == defaultSQLitePath {
return filepath.Join(dataDir, "cyrene.db")
}
return p
}
// reviewFor builds the channel-bridge review model (design/12 §6 策略四), or nil
// when no model adapter is available so the gate falls back to FIFO eviction — the
// reviewer is never invoked with a nil adapter (which would panic on .Chat).
func reviewFor(llm model.Adapter, log *slog.Logger) adapter.Reviewer {
if llm == nil {
return nil
}
return channelbridge.NewLLMReviewer(llm, log)
}
// engageLightFor builds the design/18 §3 participation judge, or nil when no model
// adapter is available so the bridge's participation engine fails closed toward
// silence (a triggered message without a judge is never answered).
func engageLightFor(llm model.Adapter, log *slog.Logger) channelbridge.LightJudge {
if llm == nil {
return nil
}
return channelbridge.NewLLMLightJudge(llm, log)
}
// defaultTriggerWords returns the default persona's role-name reminder words
// (design/18 §2) so the channel bridge knows which group messages cause a PROBING.
func (a *app) defaultTriggerWords() []string {
if c, ok := a.personas.DefaultCharacter(); ok {
if p, ok := a.personas.Persona(c.PersonaID); ok {
return append([]string(nil), p.TriggerWords...)
}
}
return nil
}