From e9afca9b29821837e6b38bfbcf70a4fecd253e33 Mon Sep 17 00:00:00 2001 From: AskaEth Date: Sun, 4 Oct 2026 10:50:35 +0800 Subject: [PATCH] =?UTF-8?q?refactor(cyrene):=20=E6=8B=86=20wire.go=20926?= =?UTF-8?q?=20=E8=A1=8C=20=E2=86=92=205=20=E6=96=87=E4=BB=B6=EF=BC=88?= =?UTF-8?q?=E7=BA=AF=E6=90=AC=E7=A7=BB=EF=BC=8C=E9=9B=B6=E8=A1=8C=E4=B8=BA?= =?UTF-8?q?=E5=8F=98=E5=8C=96=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 执行方案阶段 0 · G0-A：组合根 926 行是全仓第一大超限文件，且被 10+ 条裁决争用 （1.4/12.4.1、03.4.3、06.4.2、07.4.4、10.4.5、12.4.2、13.4.3、14.4.1、14.4.3、16.4.2） ⇒ 不先拆，后续全部被迫串行且必然破 500 行红线。 - wire.go(181)：入口与共享 —— const 块、app struct、newApp - wire_runtime.go(362)：loop/agent/model/memory/capacity/trace（buildPersonas、wireMemory、 wireRuntime、buildCapacity、onCapacity、onToolAudit、defaultPersonaID、defaultInstructions…） - wire_services.go(158)：exec/task/iot/session/history/automations/backthink （wireAuxiliary、pushOutletConfig、close、resolveDataDir、resolveSQLitePath） - wire_channels.go(172)：渠道/桥（ownerRegistrar、wireChannels、runtimeChannels、reviewFor…） - wire_frontend.go(117)：gateway/webui/观测扇出（capacityEmitter、wireFrontend） 纯搬移机械证明：去 package/import/空行后两侧 **811=811 行、排序后 diff 为空、md5 相同**； 顶层声明集合 30 条守恒；唯一新增 = 各文件的 package 行与 import 块（原 import 块按声明归属 重分配给新文件）。未重命名/重排/改注释。 验证：go build/vet 绿；go test ./cmd/... 绿；linux/amd64 + darwin/arm64 交叉编译绿； 五文件 181/362/158/172/117 行均 ≤500；gofmt -l cmd/cyrene/ 无输出。 注：全量 go test 当前有 1 处红 —— internal/contract/enums 的 TestModeTableMatchesFrontendModes 读 frontend/web/src/lib/types.ts 找 MODES，而该文件 正被并行的前端拆分任务搬走；与本提交无关，随后修正该测试的读取路径。 --- cmd/cyrene/wire.go | 745 ------------------------------------ cmd/cyrene/wire_channels.go | 172 +++++++++ cmd/cyrene/wire_frontend.go | 117 ++++++ cmd/cyrene/wire_runtime.go | 362 ++++++++++++++++++ cmd/cyrene/wire_services.go | 158 ++++++++ 5 files changed, 809 insertions(+), 745 deletions(-) create mode 100644 cmd/cyrene/wire_channels.go create mode 100644 cmd/cyrene/wire_frontend.go create mode 100644 cmd/cyrene/wire_runtime.go create mode 100644 cmd/cyrene/wire_services.go diff --git a/cmd/cyrene/wire.go b/cmd/cyrene/wire.go index 678af1b..aa5fa39 100644 --- a/cmd/cyrene/wire.go +++ b/cmd/cyrene/wire.go @@ -3,56 +3,36 @@ package main import ( "context" "database/sql" - "encoding/json" "fmt" "log/slog" - "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" 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/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/emotion" - "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 @@ -123,46 +103,6 @@ type app struct { traceWriter *traceFanout } -// 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) -} - -// ownerRegistrar implements gateway.OwnerRegistrar: it persists a channel account -// id into the owner whitelist (configs/users/owner.yaml) AND hot-updates the live -// owner profile the channel bridge holds, so a weixin scan-login confirmation is -// recognized as the owner immediately — no restart, no manual edit. path and the -// live profile are captured at wiring time: the boot-loaded *user.OwnerProfile is -// the same pointer the bridge resolves actors against, so RegisterAccount on it is -// visible to ResolveActor without touching the bridge. -type ownerRegistrar struct { - path string - owner *user.OwnerProfile - log *slog.Logger -} - -func (r *ownerRegistrar) RegisterChannelOwner(_ context.Context, platform, accountID string) error { - if strings.TrimSpace(accountID) == "" { - return nil - } - if err := user.RegisterOwnerAccount(r.path, platform, accountID); err != nil { - return fmt.Errorf("register owner %s:%s: %w", platform, accountID, err) - } - // Hot-effect: extend the live profile the bridge already points at. If the - // profile failed to load at boot (r.owner nil) the persist above still lands - // on disk and is recognized after a restart; log so the operator knows the - // hot path was unavailable. - if r.owner != nil { - r.owner.RegisterAccount(platform, accountID) - } else { - r.log.Warn("owner registrar: no live profile, whitelist persisted to disk only (restart to take effect)", - "platform", platform, "accountId", accountID) - } - r.log.Info("owner whitelist registered", "platform", platform, "accountId", accountID) - return nil -} - // 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 @@ -239,688 +179,3 @@ func newApp(ctx context.Context, cfg *config.Config, log *slog.Logger) (*app, er 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 - // One agent per role, all sharing the dispatcher/guards/adapter above, so a session - // can answer as any role (design/22 §4.2 方案 A); the loop picks per turn. - agents, defaultAgent := a.personaAgentPool(runtimeStack, a.dispatcher, guards, ag) - - // Spans fan out to the JSONL stderr recorder and — once wireFrontend has built - // the gateway — to its observability buffer (cmd/cyrene/observability.go). - a.traceWriter = newAppTraceWriter() - tracer := trace.New(a.traceWriter) - // Realtime flow visualization (design/19 §1): the publisher accumulates the - // loop's trace spans into a per-session flow graph and publishes - // ws.graph_update on each span transition. It is gated to the owner's web - // sessions ("web" and web sub-sessions), so IM / stranger / sub-agent turns - // never leak into the admin panel. A nil flow keeps the loop's prior behavior. - flowPub := loop.NewFlowPublisher( - func(subject string, payload any) { _ = a.bus.Publish(context.Background(), subject, payload) }, - func(sessionID string) bool { - // Publish the web/main flow graph (web chat + IM owner-private mirror) AND the - // autonomous-backthink session, so the backthink page's flow link graph renders - // too (design/20 §2). Other isolated channel sessions (sess-xxx) stay filtered. - return sessionID == history.MainSessionID || - strings.HasPrefix(sessionID, history.MainSessionID+":") || - sessionID == backthink.DefaultSessionID - }, - ) - 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() - // The persona emotion observer (design/01 §2) is built here because it is the - // loop's turn-boundary dependency: after every settled turn the bound persona's - // emotion machine and affection move, and the returned mood label is written - // onto Result.Mood and the assistant message (design/16 §3.3). - a.emotion = a.buildEmotionObserver() - // Reasoning knobs from the default persona (budget / effort / thinking broadcast), - // which are also the per-turn defaults: the loop re-resolves them for each turn's - // role through PersonaKnobs, so a session bound to another role uses that role's - // budget instead. - thinking, reasoningEffort, thinkingShow := a.defaultThinkingOptions() - - a.loop = loop.New(loop.Options{ - Agent: defaultAgent, - Agents: agents, - Tools: a.dispatcher, - Capacity: capDecider, - OnCapacity: a.onCapacity, - OnAudit: a.onToolAudit, - OnTurnEnd: a.onTurnEnd, - Memory: store, - Retriever: a.retriever, - Extractor: a.buildMemoryExtractor(), - Checkpoint: checkpoint, - Tracer: tracer, - Flow: flowPub, - Guards: guards, - PersonaID: personaID, - PersonaName: a.personaName, - PersonaKnobs: a.personaKnobs, - Timezone: a.cfg.Timezone, - MaxRounds: defaultMaxRounds, - WorkingMemory: memory.NewWorkingMemory(0), - OnStream: a.streamChunk, - Vision: visionResolver, - Thinking: thinking, - ReasoningEffort: reasoningEffort, - ThinkingShowToUser: thinkingShow, - Log: logger.WithSource(a.log, "runtime/loop"), - }) - - a.scheduler = orchestration.New(orchestration.Options{ - MaxParallel: defaultMaxParallel, - PerAgentTimeout: defaultSubagentTimeout, - Loop: a.loop, - Bus: a.bus, - Memory: store, - Tracer: tracer, - // A sub-agent dispatched as another role (subagent_spawn `persona`, design/22 - // §8-2) may only name a role this process can actually run with its own - // instructions, i.e. one the pool above has an agent for; anything else is - // rejected instead of silently answering in the default role's voice. - PersonaExists: personaExists(agents), - // The sub-agent's span and its inner llm/tool steps fold into the spawning - // session's live graph, so a branch shows up as a swimlane of that session's - // flow instead of opening a second graph (design/19 §2). - Flow: flowPub, - 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, a.history), - Recent: a.history, - State: backthink.NewIoTStateProvider(iotMgr, a.log), - Exporter: backthink.NewPushOutlet(a.pushOutletConfig()), - 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() - a.wireSubagentTool() - a.wireSessionReadTool() - return nil -} - -// pushOutletConfig maps the environment push-outlet block into the backthink -// exporter config. The Getui credentials are read by config but not consumed: -// the fallback SDK is not integrated yet, so the outlet logs the pending tier -// instead of faking a push. The token is passed through and never logged. -func (a *app) pushOutletConfig() backthink.PushOutletConfig { - po := a.cfg.PushOutlet - return backthink.PushOutletConfig{ - Enabled: po.Enabled, - URL: po.URL, - Token: po.Token, - Timeout: time.Duration(po.TimeoutSec) * time.Second, - Log: a.log, - } -} - -// 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, - // ChannelPersona resolves an account's channel-level role from the same - // config the factory builds from (design/22 §2 方案 b); the bridge caches it - // per connected adapter and refreshes it on every reload. - ChannelPersona: factory.Persona, - }) - 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 - } - // Opaque refresh tokens (B2) share the primary SQLite handle. A failure here - // degrades gracefully: login simply omits refreshToken and /auth/refresh - // reports not-configured, rather than failing the whole boot. The store is - // held as the interface so a constructor failure leaves it nil (a typed nil - // pointer in the interface would be non-nil and panic on first call). - var refreshStore gateway.RefreshGrantStore - if store, rerr := auth.NewRefreshStore(context.Background(), a.db, a.log); rerr != nil { - a.log.Warn("refresh token store unavailable", "error", rerr) - } else { - refreshStore = store - } - // 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, - Refresh: refreshStore, - Backthink: a.backthink, - RAG: ragStore, - LLM: a.llm, - Channels: a.channels, - Plugins: a.plugins, - OwnerRegistrar: &ownerRegistrar{path: defaultOwnerPath, owner: a.owner, log: a.log}, - AuditStore: auditStore, - Streaming: true, - Frontend: resolveFrontendFS(a.log), - Webhooks: a.channelBridge.WebhookHandlers(), - DatabasePath: a.dbPath, - DataDir: resolveDataDir(a.cfg), - }) - if err != nil { - return fmt.Errorf("gateway: %w", err) - } - gw.SetVoice(a.voice) - a.gw = gw // assign BEFORE wiring hooks below, which capture a.gw / gw (nil-receiver panic otherwise) - // Observability trace sink (design/19 §2): now that the gateway exists, its - // ring buffer must be fed by the loop's spans, or the admin trace/metrics - // endpoints stay empty. - wireGatewayTraceSink(a, gw) - 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.capacitySink = gw - a.moodSink = gw - - // Wire the backthink proactive sinks now that the gateway hub (active-client - // route) and the channel bridge (IM fallback) both exist. They were absent when - // backthink.New ran (the hub needs the loop and the bridge needs the loop and the - // owner profile, all assembled later), so without this the pusher would resolve - // RouteNone. Each setter rebuilds the pusher under the service's lifecycle lock. - if a.backthink != nil { - a.backthink.SetActiveSink(gw) - if a.channelBridge != nil { - a.backthink.SetIMSink(a.channelBridge) - } - } - 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 followed by the -// personality/values/speechStyle/catchphrase/background sections (design/01 -// §2.2), falling back to its role text or a minimal instruction when neither is -// set. -func (a *app) defaultInstructions() string { - if p, ok := a.defaultPersona(); ok { - return instructionsFromPersona(p) - } - 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.plugins != nil { - a.plugins.Close(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 -} diff --git a/cmd/cyrene/wire_channels.go b/cmd/cyrene/wire_channels.go new file mode 100644 index 0000000..e56ab6c --- /dev/null +++ b/cmd/cyrene/wire_channels.go @@ -0,0 +1,172 @@ +package main + +import ( + "context" + "fmt" + "log/slog" + "strings" + + "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/config" + "git.yeij.top/AskaEth/Cyrene_v2/internal/config/runtimeconfig" + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" + "git.yeij.top/AskaEth/Cyrene_v2/internal/services/channelbridge" + "git.yeij.top/AskaEth/Cyrene_v2/internal/user" +) + +// ownerRegistrar implements gateway.OwnerRegistrar: it persists a channel account +// id into the owner whitelist (configs/users/owner.yaml) AND hot-updates the live +// owner profile the channel bridge holds, so a weixin scan-login confirmation is +// recognized as the owner immediately — no restart, no manual edit. path and the +// live profile are captured at wiring time: the boot-loaded *user.OwnerProfile is +// the same pointer the bridge resolves actors against, so RegisterAccount on it is +// visible to ResolveActor without touching the bridge. +type ownerRegistrar struct { + path string + owner *user.OwnerProfile + log *slog.Logger +} + +func (r *ownerRegistrar) RegisterChannelOwner(_ context.Context, platform, accountID string) error { + if strings.TrimSpace(accountID) == "" { + return nil + } + if err := user.RegisterOwnerAccount(r.path, platform, accountID); err != nil { + return fmt.Errorf("register owner %s:%s: %w", platform, accountID, err) + } + // Hot-effect: extend the live profile the bridge already points at. If the + // profile failed to load at boot (r.owner nil) the persist above still lands + // on disk and is recognized after a restart; log so the operator knows the + // hot path was unavailable. + if r.owner != nil { + r.owner.RegisterAccount(platform, accountID) + } else { + r.log.Warn("owner registrar: no live profile, whitelist persisted to disk only (restart to take effect)", + "platform", platform, "accountId", accountID) + } + r.log.Info("owner whitelist registered", "platform", platform, "accountId", accountID) + 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, + // ChannelPersona resolves an account's channel-level role from the same + // config the factory builds from (design/22 §2 方案 b); the bridge caches it + // per connected adapter and refreshes it on every reload. + ChannelPersona: factory.Persona, + }) + 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) } + +// 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 +} diff --git a/cmd/cyrene/wire_frontend.go b/cmd/cyrene/wire_frontend.go new file mode 100644 index 0000000..1a92190 --- /dev/null +++ b/cmd/cyrene/wire_frontend.go @@ -0,0 +1,117 @@ +package main + +import ( + "context" + "fmt" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/knowledge" + "git.yeij.top/AskaEth/Cyrene_v2/internal/services/gateway" +) + +// 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) +} + +// 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 + } + // Opaque refresh tokens (B2) share the primary SQLite handle. A failure here + // degrades gracefully: login simply omits refreshToken and /auth/refresh + // reports not-configured, rather than failing the whole boot. The store is + // held as the interface so a constructor failure leaves it nil (a typed nil + // pointer in the interface would be non-nil and panic on first call). + var refreshStore gateway.RefreshGrantStore + if store, rerr := auth.NewRefreshStore(context.Background(), a.db, a.log); rerr != nil { + a.log.Warn("refresh token store unavailable", "error", rerr) + } else { + refreshStore = store + } + // 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, + Refresh: refreshStore, + Backthink: a.backthink, + RAG: ragStore, + LLM: a.llm, + Channels: a.channels, + Plugins: a.plugins, + OwnerRegistrar: &ownerRegistrar{path: defaultOwnerPath, owner: a.owner, log: a.log}, + AuditStore: auditStore, + Streaming: true, + Frontend: resolveFrontendFS(a.log), + Webhooks: a.channelBridge.WebhookHandlers(), + DatabasePath: a.dbPath, + DataDir: resolveDataDir(a.cfg), + }) + if err != nil { + return fmt.Errorf("gateway: %w", err) + } + gw.SetVoice(a.voice) + a.gw = gw // assign BEFORE wiring hooks below, which capture a.gw / gw (nil-receiver panic otherwise) + // Observability trace sink (design/19 §2): now that the gateway exists, its + // ring buffer must be fed by the loop's spans, or the admin trace/metrics + // endpoints stay empty. + wireGatewayTraceSink(a, gw) + 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.capacitySink = gw + a.moodSink = gw + + // Wire the backthink proactive sinks now that the gateway hub (active-client + // route) and the channel bridge (IM fallback) both exist. They were absent when + // backthink.New ran (the hub needs the loop and the bridge needs the loop and the + // owner profile, all assembled later), so without this the pusher would resolve + // RouteNone. Each setter rebuilds the pusher under the service's lifecycle lock. + if a.backthink != nil { + a.backthink.SetActiveSink(gw) + if a.channelBridge != nil { + a.backthink.SetIMSink(a.channelBridge) + } + } + return nil +} diff --git a/cmd/cyrene/wire_runtime.go b/cmd/cyrene/wire_runtime.go new file mode 100644 index 0000000..3d1469a --- /dev/null +++ b/cmd/cyrene/wire_runtime.go @@ -0,0 +1,362 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "path/filepath" + "strings" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/config" + "git.yeij.top/AskaEth/Cyrene_v2/internal/config/runtimeconfig" + "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/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/guardrail" + "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/vision" + "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/tool" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" + "git.yeij.top/AskaEth/Cyrene_v2/internal/services/backthink" + "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" +) + +// 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 + // One agent per role, all sharing the dispatcher/guards/adapter above, so a session + // can answer as any role (design/22 §4.2 方案 A); the loop picks per turn. + agents, defaultAgent := a.personaAgentPool(runtimeStack, a.dispatcher, guards, ag) + + // Spans fan out to the JSONL stderr recorder and — once wireFrontend has built + // the gateway — to its observability buffer (cmd/cyrene/observability.go). + a.traceWriter = newAppTraceWriter() + tracer := trace.New(a.traceWriter) + // Realtime flow visualization (design/19 §1): the publisher accumulates the + // loop's trace spans into a per-session flow graph and publishes + // ws.graph_update on each span transition. It is gated to the owner's web + // sessions ("web" and web sub-sessions), so IM / stranger / sub-agent turns + // never leak into the admin panel. A nil flow keeps the loop's prior behavior. + flowPub := loop.NewFlowPublisher( + func(subject string, payload any) { _ = a.bus.Publish(context.Background(), subject, payload) }, + func(sessionID string) bool { + // Publish the web/main flow graph (web chat + IM owner-private mirror) AND the + // autonomous-backthink session, so the backthink page's flow link graph renders + // too (design/20 §2). Other isolated channel sessions (sess-xxx) stay filtered. + return sessionID == history.MainSessionID || + strings.HasPrefix(sessionID, history.MainSessionID+":") || + sessionID == backthink.DefaultSessionID + }, + ) + 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() + // The persona emotion observer (design/01 §2) is built here because it is the + // loop's turn-boundary dependency: after every settled turn the bound persona's + // emotion machine and affection move, and the returned mood label is written + // onto Result.Mood and the assistant message (design/16 §3.3). + a.emotion = a.buildEmotionObserver() + // Reasoning knobs from the default persona (budget / effort / thinking broadcast), + // which are also the per-turn defaults: the loop re-resolves them for each turn's + // role through PersonaKnobs, so a session bound to another role uses that role's + // budget instead. + thinking, reasoningEffort, thinkingShow := a.defaultThinkingOptions() + + a.loop = loop.New(loop.Options{ + Agent: defaultAgent, + Agents: agents, + Tools: a.dispatcher, + Capacity: capDecider, + OnCapacity: a.onCapacity, + OnAudit: a.onToolAudit, + OnTurnEnd: a.onTurnEnd, + Memory: store, + Retriever: a.retriever, + Extractor: a.buildMemoryExtractor(), + Checkpoint: checkpoint, + Tracer: tracer, + Flow: flowPub, + Guards: guards, + PersonaID: personaID, + PersonaName: a.personaName, + PersonaKnobs: a.personaKnobs, + Timezone: a.cfg.Timezone, + MaxRounds: defaultMaxRounds, + WorkingMemory: memory.NewWorkingMemory(0), + OnStream: a.streamChunk, + Vision: visionResolver, + Thinking: thinking, + ReasoningEffort: reasoningEffort, + ThinkingShowToUser: thinkingShow, + Log: logger.WithSource(a.log, "runtime/loop"), + }) + + a.scheduler = orchestration.New(orchestration.Options{ + MaxParallel: defaultMaxParallel, + PerAgentTimeout: defaultSubagentTimeout, + Loop: a.loop, + Bus: a.bus, + Memory: store, + Tracer: tracer, + // A sub-agent dispatched as another role (subagent_spawn `persona`, design/22 + // §8-2) may only name a role this process can actually run with its own + // instructions, i.e. one the pool above has an agent for; anything else is + // rejected instead of silently answering in the default role's voice. + PersonaExists: personaExists(agents), + // The sub-agent's span and its inner llm/tool steps fold into the spawning + // session's live graph, so a branch shows up as a swimlane of that session's + // flow instead of opening a second graph (design/19 §2). + Flow: flowPub, + 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 +} + +// 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 followed by the +// personality/values/speechStyle/catchphrase/background sections (design/01 +// §2.2), falling back to its role text or a minimal instruction when neither is +// set. +func (a *app) defaultInstructions() string { + if p, ok := a.defaultPersona(); ok { + return instructionsFromPersona(p) + } + return fallbackInstruction +} diff --git a/cmd/cyrene/wire_services.go b/cmd/cyrene/wire_services.go new file mode 100644 index 0000000..cd3b916 --- /dev/null +++ b/cmd/cyrene/wire_services.go @@ -0,0 +1,158 @@ +package main + +import ( + "context" + "fmt" + "path/filepath" + "strings" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session" + "git.yeij.top/AskaEth/Cyrene_v2/internal/config" + "git.yeij.top/AskaEth/Cyrene_v2/internal/iot" + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" + "git.yeij.top/AskaEth/Cyrene_v2/internal/platform" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/exec" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/task" + "git.yeij.top/AskaEth/Cyrene_v2/internal/services/automations" + "git.yeij.top/AskaEth/Cyrene_v2/internal/services/backthink" +) + +// 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, a.history), + Recent: a.history, + State: backthink.NewIoTStateProvider(iotMgr, a.log), + Exporter: backthink.NewPushOutlet(a.pushOutletConfig()), + 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() + a.wireSubagentTool() + a.wireSessionReadTool() + return nil +} + +// pushOutletConfig maps the environment push-outlet block into the backthink +// exporter config. The Getui credentials are read by config but not consumed: +// the fallback SDK is not integrated yet, so the outlet logs the pending tier +// instead of faking a push. The token is passed through and never logged. +func (a *app) pushOutletConfig() backthink.PushOutletConfig { + po := a.cfg.PushOutlet + return backthink.PushOutletConfig{ + Enabled: po.Enabled, + URL: po.URL, + Token: po.Token, + Timeout: time.Duration(po.TimeoutSec) * time.Second, + Log: a.log, + } +} + +// 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.plugins != nil { + a.plugins.Close(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 +}