Cyrene_v2/cmd/cyrene/wire_runtime.go

package main

import (
	"context"
	"fmt"
	"path/filepath"
	"strings"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/capability/registry"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/capability/restrict"
	"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/embedding"
	"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 构建模型适配器、工具 dispatcher、护栏、agent 与其 loop、
// 追踪、持久化与编排 scheduler。
func (a *app) wireRuntime() error {
	a.attachToolPolicyStore()

	// 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 策略四).
	modelsPath := runtimeconfig.DefaultPath(resolveDataDir(a.cfg))
	// 启动即补齐一次（主人裁决）：`.env` 只是默认/播种层，models.json 才是面板的
	// 一等来源——把 `.env` 补齐进面板配置里缺失的键（面板已有的键绝不覆盖，组集合
	// 完全由面板决定），幂等；失败只记日志、不阻断启动。
	materializeModels(modelsPath, a.cfg.LLM, a.log)
	runtimeStack := model.NewRuntimeLLM(modelsPath, 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()

	// 默认护栏策略在 buildGuardrailPipeline 里集中定义（敏感信息阻断 +
	// 长度上限 + 仅当配置提供时的关键词表，见 wire_guardrail.go）。
	guards := a.buildGuardrailPipeline()
	instructions := a.defaultInstructions()
	// 默认角色的 agent 也要绑它自己的 persona 视图：models.json 的 personaModels /
	// personaProviders 可能给默认角色单独指定模型或提供商，用裸栈会漏掉这一层
	// （persona 为空时该视图与裸栈等价，行为不变）。
	personaID := a.defaultPersonaID()
	defaultStack := runtimeStack.ForPersona(personaID)
	ag, aerr := agent.New(instructions, a.dispatcher, guards,
		agent.WithModel(defaultStack, defaultStack),
		agent.WithOptions(agent.Options{MaxRounds: defaultMaxRounds, Purpose: "chat"}))
	if aerr != nil {
		return fmt.Errorf("agent: %w", aerr)
	}

	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.
	//
	// 图源白名单（裁决 16.4.2）：它决定**哪些会话产出实时链路图**，也是后端能提供
	// 完整链路导图的那一侧 —— 主人 web 会话族（"web" 主会话 + "web:<uuid>" 次级会话，
	// IM 主人私聊镜像也落主会话）与自主后台思考会话（design/20 §2）。子代理的 span 由
	// loop.WithFlowBranch 并进**派生它的那个会话**，渲染成该会话图的一条泳道（design/19
	// §2），因此也在这份白名单里。
	//
	// 注意口径（裁决 16.4.2）：design/19 §1 的「仅发布/不发」说的是**用户侧前端**观察到
	// 的现象，不是「后端可以少画图」——管理后台的图源由此保持完整，与用户端发布策略
	// 是两件事。隔离的渠道会话（群聊/陌生人，sess-xxx）不在此白名单内：那是 design/19 §1
	// 发布会话范围（裁决 01.4.5）的既有裁决，本批未改。nil flow 保持 loop 原行为。
	flowPub := loop.NewFlowPublisher(
		func(subject string, payload any) { _ = a.bus.Publish(context.Background(), subject, payload) },
		func(sessionID string) bool {
			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 main model that really accepts image input sees the image
	// directly, otherwise the vision/OCR tool's text description replaces it
	// (design/06 §0)。
	//
	// SupportsVision means "主模型能否收图"（裁决 14.4.1 修语义，C-7）, read from the
	// active models.json group's `multimodal` declaration — NOT from
	// LLM_VISION_MODEL, which configures the vision *tool's* model. It is read per
	// call so a panel save/reload takes effect on the next turn.
	visionRouter := vision.NewRouter(
		vision.WithVisionTool(visionToolName),
		vision.WithOCRTool(ocrToolName),
	)
	visionResolver := vision.NewMessageResolver(
		visionRouter,
		func() vision.ModelInfo {
			group := a.llm.Group()
			return vision.ModelInfo{Name: group.Model, SupportsVision: group.Multimodal}
		},
		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()

	// 敏感操作代审（design/11 §4；裁决 13.4.3）：terminal 这类敏感工具在 loop 的
	// 工具准入门上经主代理代审（复用 orchestration 的 Approver），再由网关把
	// 结果落进 audit_logs 并广播 design/11 §4 的小字。同一个 Approver 也交给
	// 编排层（子代理代审），使两条路径共用一处审计。
	approver := audit.NewApprover(a.log)
	approvalGate := a.buildApprovalGate(approver)

	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,
		PersonaAwakening: a.awakeningFor,
		// 主人身份说明书按**会话**注入（design/11 §0 红线）：只有主人车道的
		// 上下文里有它，群聊 / 陌生人私聊的返回空串 —— 那段知识绝不进它们的
		// 上下文，也绝不随之发给模型供应商。开关与判定见 persona_identity.go。
		PersonaKnowledgeFor: a.personaKnowledgeFor(envPath),
		Timezone:            a.cfg.Timezone,
		MaxRounds:           defaultMaxRounds,
		// 上下文溢出压缩：预算取自 CONTEXT_MAX_TOKENS（非正 = 关闭，
		// 与接线前一致），默认启用（见 wire_context.go）。Compactor 不注入，
		// 由 loop 选用内置的结构化 compactor（离线、不调模型）。
		MaxContextTokens:   a.contextBudget(),
		WorkingMemory:      memory.NewWorkingMemory(0),
		OnStream:           a.streamChunk,
		Vision:             visionResolver,
		Thinking:           thinking,
		ReasoningEffort:    reasoningEffort,
		ThinkingShowToUser: thinkingShow,
		// 按调用（而非整批）准入：敏感与否取决于该次调用的参数
		// （file 的 read 自主、write 代审），见 toolAdmissionHooks。
		Hooks: a.toolAdmissionHooks(approvalGate),
		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: approver,
		Logger:   logger.WithSource(a.log, "orchestration"),
		AgentID:  defaultAgentID,
	})
	return nil
}

// attachToolPolicyStore 接上工具策略的落盘载体（design/11 §3；裁决 1.3/10.4.2）。
//
// 它必须在任何工具注册**之前**调用：registry 接到 store 时会立刻 Load 落盘
// 策略并套到已注册能力上，之后 Register 进来的能力（MCP 热装、插件刷新）也会
// 拿到落盘值——「重启后策略仍在」正是靠这两条路径合起来成立。
//
// 失败只记 WARN 并保留默认值：默认（两个布尔 false = owner-only）就是最严的
// 一侧，读不出一份策略绝不放宽策略，也不因为策略存储不可用而拖垮启动
// （写路径仍会诚实报错，不会假装保存成功）。
func (a *app) attachToolPolicyStore() {
	log := logger.WithSource(a.log, "cmd/cyrene")
	store, err := registry.NewSQLitePolicyStore(a.db, a.log)
	if err != nil {
		log.Warn("tool policy store unavailable, keeping owner-only defaults", "error", err)
		return
	}
	if err := a.caps.AttachPolicyStore(context.Background(), store); err != nil {
		log.Warn("tool policy load failed, keeping owner-only defaults", "error", err)
	}
}

// buildApprovalGate 构建敏感操作代审门（design/11 §4；裁决 13.4.3）。
//
// 归属解析取 Run 上下文里的会话 id 与该 turn 的角色名，使前端小字既投到
// 正确的会话、也写明是谁想动手（design/11 §4 的「xxx(角色名)」）。会话 id 用
// loop.FlowOwnerFromContext：子代理的敏感调用归属到**派生它的那个会话**
// （与 design/19 §2 的 flow 归属同一口径），所以小字出现在主人正在看的对话里，
// 而不是一个没人查看的子代理 id 下。
func (a *app) buildApprovalGate(approver *audit.Approver) *restrict.Gate {
	return restrict.NewGate(
		auditReviewer{approver: approver},
		func(ctx context.Context) (string, string) {
			return loop.FlowOwnerFromContext(ctx), a.approvalActor(ctx)
		},
		restrict.WithGateLogger(a.log),
	)
}

// toolAdmissionHooks 构建 loop 的工具准入钩子（design/11 §4；裁决 13.4.3）：
// 敏感调用先经主代理代审，代审记录交给网关落 audit_logs 并广播前端小字；
// 未获代审的调用以软结果回灌给模型（绝不执行）。非敏感调用直接放行。
//
// 只装 CallAdmit：敏感与否取决于**该次调用的参数**（file 的 read 自主、
// write 代审），批量钩子（BatchAdmit）看不到参数，做不出这个区分。
// 组合根按 loop 的既有契约装配，因此这条链也覆盖子代理的敏感调用
// （编排层派生的 loop 共享同一组钩子）。
func (a *app) toolAdmissionHooks(gate *restrict.Gate) loop.ToolLoopHooks {
	return loop.ToolLoopHooks{
		CallAdmit: func(ctx context.Context, call agent.ToolCall) *loop.ToolCallVeto {
			decision := gate.Admit(ctx, call.Name, call.Args)
			if decision.Approval != nil {
				// a.gw 在 wireFrontend 才建好；RecordApproval 对 nil 接收者安全。
				a.gw.RecordApproval(ctx, *decision.Approval)
			}
			if !decision.Allow {
				return &loop.ToolCallVeto{Block: true, Reason: decision.Reason}
			}
			return nil
		},
	}
}

// auditReviewer 把 orchestration/audit 的 Approver 适配成代审门的 Reviewer
// （design/11 §4；裁决 13.4.3 要求复用该 Approver，而不是另起一套代审）。
// 它只把 Approval 收敛成 (allowed, error)，前端小字文案仍由
// restrict 复用 audit.Approval.Explain 的固定句式产出。
type auditReviewer struct{ approver *audit.Approver }

// ReviewAndApprove 执行一次代审；approver 缺席时复用编排层的 ErrNoApprover
// （门据此 fail-closed 拦截敏感操作，绝不静默放行）。
func (r auditReviewer) ReviewAndApprove(ctx context.Context, actor, action, describe string) (bool, error) {
	if r.approver == nil {
		return false, fmt.Errorf("review-and-approve: %w", orchestration.ErrNoApprover)
	}
	ap, err := r.approver.ReviewAndApprove(ctx, actor, action, describe)
	if err != nil {
		return false, fmt.Errorf("review-and-approve: %w", err)
	}
	return ap.Allowed, nil
}

// approvalActor 解析代审小字里的「谁想动手」：该 turn 绑定的角色名，
// 未绑定时回落到默认角色；角色名解析不出来就用角色 id（小字宁可用 id
// 也要说清是谁，但绝不编造一个名字）。
func (a *app) approvalActor(ctx context.Context) string {
	pid := loop.SessionEnvFrom(ctx).PersonaID
	if pid == "" {
		pid = a.defaultPersonaID()
	}
	if name := a.personaName(pid); name != "" {
		return name
	}
	return pid
}

// buildCapacity / onCapacity / capacity JSON+TTL parsing live in
// wire_capacity.go (AGENTS §10 文件 ≤500 行).

// 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 / parseL3CacheTTL live in wire_capacity.go.

// 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.
//
// 主人身份说明书**不在这里**（P0-10 收窄后按会话注入，见 personaKnowledgeFor）：
// 静态拼进指令会让它进入该角色的每一个会话，含群聊与陌生人私聊。
func (a *app) defaultInstructions() string {
	if p, ok := a.defaultPersona(); ok {
		return a.personaInstructions(p)
	}
	return fallbackInstruction
}