Agent-System-Backend/internal/app/app.go

// Package app 组合根：装配存储（嵌入式/PG 档）、总线、注册表、服务、runmgr、
// 认证、插件宿主与 HTTP 服务。
package app

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"log"
	"net"
	"net/http"
	"os"
	"os/signal"
	"path/filepath"
	"strings"
	"sync"
	"syscall"
	"time"

	"error-code-system/errcode"
	"logging-system/logsys"
	"logging-system/sink"

	"agent-native/runtime/checkpoint"
	agentllm "agent-native/runtime/llm"
	agentmcp "agent-native/runtime/mcp"
	"agent-native/runtime/policy"
	"agent-native/runtime/telemetry"
	agenttools "agent-native/runtime/tools"
	nativeadp "agent-squad/adapters/native"
	"agent-system-backend/internal/bus"
	"agent-system-backend/internal/cardsvc"
	"agent-system-backend/internal/cmdbus"
	"agent-system-backend/internal/config"
	"agent-system-backend/internal/hostapi"
	"agent-system-backend/internal/intentgate"
	"agent-system-backend/internal/intentsem"
	"agent-system-backend/internal/pgbus"
	"agent-system-backend/internal/plugins"
	"agent-system-backend/internal/profiles"
	"agent-system-backend/internal/prompts"
	"agent-system-backend/internal/quality"
	"agent-system-backend/internal/rag"
	"agent-system-backend/internal/ratelimit"
	"agent-system-backend/internal/runmgr"
	"agent-system-backend/internal/runtime"
	"agent-system-backend/internal/server"
	"agent-system-backend/internal/settings"
	"agent-system-backend/internal/skills"
	"agent-system-backend/internal/squadadapter"
	"agent-system-backend/internal/store"
	"agent-system-backend/internal/tools"
	"agent-system-backend/internal/workflow"
	core "agent-system-core"
)

type Options struct {
	Addr           string
	DataDir        string
	PluginsDir     string
	ForceMock      bool
	StorageDSN     string // 空 = 嵌入式 SQLite 档；非空 = PostgreSQL 档（多用户/多实例）
	ArtifactsDir   string // PG 档工件对象目录（空 = <DataDir>/artifacts）
	AuthMode       string // auto（默认：PG 档开、嵌入式关）| on | off
	InstanceID     string // 空 = 主机名:监听地址；写入 runs.instance_id，启动对账只收口本实例
	InsecureListen bool
	AllowedHosts   []string
	// DisableUploadIndexing EXT-9：true 时上传端点不自动写检索索引（发布决定
	// 移交下游命令层）；零值=false=现状。由 link 层自 extension.Options 透传。
	DisableUploadIndexing bool
	// DisableConvCardMirror EXT-12：true 时会话卡不镜像进项目公共流（不发布
	// 即不持久，重放不含会话卡）；零值=false=现状全量镜像。
	DisableConvCardMirror bool
	// DisableExecTools 启动级硬关闭执行原语（run_shell / run_python /
	// register_artifact 与小队内核执行配置）：用户设置 runtime.exec 无法覆盖；
	// 零值=false=现状。
	DisableExecTools bool
}

// Run 装配并阻塞服务直至 SIGINT/SIGTERM（优雅停机：中断在途 run、冲净事件缓冲）。
func Run(opt Options) error {
	logger := logsys.New(logsys.Options{
		Service: "agent-system", Module: "app", Level: logsys.LevelInfo,
		Sinks: []logsys.Sink{&sink.Console{Out: os.Stderr, Format: "json"}},
	})
	logger.Info("agent-system starting", "addr", opt.Addr, "data", opt.DataDir)
	resolved := config.Resolved{Config: config.Config{
		Addr: opt.Addr, AuthMode: opt.AuthMode, StorageDSN: opt.StorageDSN, InsecureListen: opt.InsecureListen,
	}}
	if msg := resolved.ListenRefusal(); msg != "" {
		return fmt.Errorf("listen guard: %s", msg)
	}
	if opt.InsecureListen && !config.IsLoopbackListen(opt.Addr) && !resolved.AuthEnabled() {
		log.Printf("warning: listening on %s without authentication because insecureListen is set", opt.Addr)
	}
	var st *store.Store
	var err error
	if opt.StorageDSN != "" {
		artifacts := opt.ArtifactsDir
		if artifacts == "" {
			artifacts = filepath.Join(opt.DataDir, "artifacts")
		}
		st, err = store.OpenPostgres(opt.StorageDSN, artifacts)
	} else {
		st, err = store.OpenSQLite(filepath.Join(opt.DataDir, "agent-system.db"))
	}
	if err != nil {
		return fmt.Errorf("open store: %w", err)
	}
	defer st.Close()
	log.Printf("storage tier: %s", st.Dialect())

	b := bus.New()
	reg := tools.NewRegistry()
	if err := reg.RegisterBuiltin(); err != nil {
		return fmt.Errorf("builtin tools: %w", err)
	}
	// fs 预留能力：工具族实现但 Hidden（runtime.fsTools 未开时模型与 /api/tools 均不可见）。
	if err := tools.RegisterFSTools(reg); err != nil {
		return fmt.Errorf("fs tools: %w", err)
	}
	// 执行原语（agent-native 交付）：shell/python 在会话工作区执行 + 产物登记桥，
	// runtime.exec=false 可整体禁用（fail-closed）。生成文档/表格/演示稿的方法
	// 在提示词库 base.deliverables，不在系统工具里。启动参数 DisableExecTools
	// 为不受用户设置影响的硬关闭（多用户部署的默认安全档）。
	tools.SetExecDisabled(opt.DisableExecTools)
	if opt.DisableExecTools {
		log.Printf("exec tools: disabled by startup option (runtime.exec settings are ignored)")
	}
	if err := tools.RegisterExecTools(reg); err != nil {
		return fmt.Errorf("exec tools: %w", err)
	}
	if err := tools.RegisterInterpreterTools(reg); err != nil {
		return fmt.Errorf("interpreter tools: %w", err)
	}
	if err := tools.RegisterCodeTool(reg); err != nil {
		return fmt.Errorf("code orchestration tool: %w", err)
	}
	setSvc := settings.NewService(st)
	effBoot := setSvc.Effective(store.OwnerDefault, "")
	traceFn := telemetry.StartFunc(logger,
		telemetry.WithServiceName(effBoot.GetString("telemetry.serviceName", "agent-system-backend")),
		telemetry.WithEndpoint(effBoot.GetString("telemetry.otlpEndpoint", "")))
	setSvc.SetSystemOverride(map[string]any{"storage": map[string]any{"dataDir": opt.DataDir}})
	// 点路径键展开必须在读 tools.tavily.apiKeys 之前，否则启动仍注册旧池。
	norm, err := setSvc.NormalizeStored()
	if err != nil {
		return fmt.Errorf("normalize settings: %w", err)
	}
	for _, p := range norm.MaskPaths {
		log.Printf("settings: warning: stored value is the mask placeholder at %s", p)
	}
	if norm.DottedKeys > 0 {
		log.Printf("settings: normalized %d dotted key(s) in %d scope(s)", norm.DottedKeys, norm.Scopes)
	}
	if err := setSvc.SnapshotAndMigrateSecrets(); err != nil {
		return fmt.Errorf("migrate secrets: %w", err)
	}
	runtime.SecretLookup = func(owner, service string) (string, bool) {
		if owner == "" {
			owner = store.OwnerDefault
		}
		return st.ActiveSecret(owner, service)
	}
	rag.SecretLookup = func(service, fallback string) string {
		return runtime.LookupSecret(store.OwnerDefault, service, fallback)
	}
	// 网络搜索（Tavily）：配置密钥池才注册（对齐 agent-native 语义：未配置不静默降级）。
	eff := setSvc.Effective(store.OwnerDefault, "")
	tavilyRaw := eff.GetString("tools.tavily.apiKeys", "")
	if active, ok := st.ActiveSecret(store.OwnerDefault, "tools.tavily"); ok {
		tavilyRaw = active
	}
	if keys := splitKeyPool(tavilyRaw); len(keys) > 0 {
		provider := agenttools.NewTavilyProvider(eff.GetString("tools.tavily.endpoint", ""), keys)
		if err := tools.RegisterSearchTools(reg, provider); err != nil {
			return fmt.Errorf("search tools: %w", err)
		}
		log.Printf("tools: tavily_search/web_search registered (%d key(s))", len(keys))
	}
	promptLib, err := prompts.DefaultLibrary()
	if err != nil {
		return fmt.Errorf("prompt library: %w", err)
	}
	promptSvc, err := prompts.NewService(st, promptLib)
	if err != nil {
		return fmt.Errorf("prompt service: %w", err)
	}
	// 小队角色指令接提示词库（agent.squad.<kind>；未命中回落内联文案）。
	squadadapter.RoleInstructions = func(kind string) (string, bool) {
		body, err := promptSvc.AgentPrompt(store.OwnerDefault, "agent.squad."+kind)
		return body, err == nil
	}
	// 技能种子：交付配方与 house-style 落 <data>/skills/。
	// 未改过的内置文件随版本升级；用户改过的保留并记 drift。
	skillsRoot := filepath.Join(opt.DataDir, "skills")
	if res, serr := skills.Seed(skillsRoot); serr != nil {
		log.Printf("skills: seed builtin recipes failed: %v", serr)
	} else if res.Written > 0 || res.Upgraded > 0 || len(res.Drifted) > 0 {
		log.Printf("skills: wrote %d, upgraded %d into %s", res.Written, res.Upgraded, skillsRoot)
		for _, path := range res.Drifted {
			log.Printf("skills: kept user revision (drift) %s", path)
		}
	}
	tools.SetSkillsRoot(skillsRoot)
	// 终轮评分的 lint 子进程与 run_python 同一技能根：不注入的话 housestyle
	// 在后端进程里 import 失败，lint 静默变空结果（评分门禁半边失效）。
	quality.SetLintEnvRoot(skillsRoot)
	// 小队的 agent-native 内核工厂（组合根装配，逐次运行时从 settings 解析）：
	// 模型走 settings.models.chat 同一路由；搜索/解释器/MCP/技能根复用产品配置；
	// 委派桥接产品 SpawnSubagent。主模型未配置时返回未启用（不阻塞产品）。
	var pluginHost *plugins.Host
	squadadapter.AgentNativeOptions = func(rc squadadapter.Ctx) (nativeadp.Config, bool) {
		effNow := setSvc.Effective(store.OwnerDefault, "")
		route := runtime.ResolveChatModel(core.AgentProfile{}, effNow)
		if route.Model == "" || route.Base == "" {
			return nativeadp.Config{}, false
		}
		mode := "default"
		if src, ok := rc.(interface{ PermissionMode() string }); ok {
			mode = src.PermissionMode()
		}
		cfg := nativeadp.Config{
			Client:       &agentllm.Client{BaseURL: route.Base, APIKey: route.Key, HTTPClient: &http.Client{}},
			DefaultModel: route.Model,
			MaxTurns:     0,
			Timeout:      0,
			Subagent:     squadadapter.SubagentFactory(rc),
			RepeatGuard:  true,
			BeforeTool: func(call agentllm.ToolCall) error {
				effect := policy.Decide(policy.Request{
					Mode: policy.Normalize(mode), Name: call.Name, Args: call.Arguments,
					Traits: policy.TraitsFor(call.Name),
				}, nil)
				if effect == policy.EffectDeny {
					return fmt.Errorf("计划模式拒绝了工具 %s", call.Name)
				}
				return nil
			},
		}
		tavilyNow := effNow.GetString("tools.tavily.apiKeys", "")
		if active, ok := st.ActiveSecret(store.OwnerDefault, "tools.tavily"); ok {
			tavilyNow = active
		}
		if keys := splitKeyPool(tavilyNow); len(keys) > 0 {
			cfg.Search = agenttools.NewTavilyProvider(effNow.GetString("tools.tavily.endpoint", ""), keys)
		}
		if effNow.GetBool("runtime.exec", true) {
			cfg.Exec = tools.InterpreterExecConfig()
		}
		if st2, err := os.Stat(skillsRoot); err == nil && st2.IsDir() {
			cfg.SkillsRoots = []string{skillsRoot}
		}
		rawMCP := strings.TrimSpace(effNow.GetString("tools.mcp.servers", ""))
		if pluginHost != nil {
			rawMCP = mergeMCPServers(rawMCP, pluginHost.MCPServers())
		}
		if mgr := mcpManagerFor(rawMCP, strings.TrimSpace(effNow.GetString("tools.mcp.overrides", ""))); mgr != nil {
			cfg.MCP = mgr
		}
		cfg.CodeOrchestration = effNow.GetBool("runtime.codeOrchestration", false)
		if ask, ok := rc.(interface {
			AskUser(card core.HITLCard) (core.HITLResponse, error)
		}); ok {
			cfg.Ask = func(_ context.Context, question string) (string, error) {
				resp, err := ask.AskUser(core.HITLCard{
					Question: question, Kind: core.HITLKindInput, AllowFreeInput: true,
				})
				if err != nil {
					return "", err
				}
				if strings.TrimSpace(resp.FreeText) != "" {
					return strings.TrimSpace(resp.FreeText), nil
				}
				return resp.ChoiceID, nil
			}
		}
		cfg.Trace = traceFn
		// N1 WP4：小队成员的内核 ask 决策经 GateOptions.PermissionAsk 到达
		// runCtx 的 HITL 确认卡（宿主无实现则 nil，内核保持 ask=拒绝旧行为）。
		cfg.PermissionAsk = runtime.PermissionAskOf(rc)
		if host, ok := rc.(interface {
			OpenCheckpoints() (*checkpoint.Log, error)
			CheckpointAnchor() string
		}); ok {
			if lg, err := host.OpenCheckpoints(); err == nil {
				cfg.Checkpoints = lg
				anchor := host.CheckpointAnchor()
				cfg.CheckpointAnchor = func() string { return anchor }
			}
		}
		return cfg, true
	}
	profSvc := profiles.New(st)
	if err := profSvc.EnsureDefault(); err != nil {
		return fmt.Errorf("default profile: %w", err)
	}
	if opt.ForceMock {
		if err := setSvc.MergeUser(store.OwnerDefault, map[string]any{"runtime": map[string]any{"mock": true}}); err != nil {
			return err
		}
	}

	convRunner := &runtime.LLMRunner{Prompts: promptSvc, SkillsDir: skillsRoot, Trace: traceFn}
	subRunner := &runtime.SubLLMRunner{Prompts: promptSvc, Trace: traceFn}
	mockRunner := &runtime.MockRunner{}
	mgr := runmgr.New(st, b, reg, profSvc, setSvc, convRunner, subRunner, mockRunner, &runtime.MockSubagent{})
	instanceID := opt.InstanceID
	if instanceID == "" {
		host, err := os.Hostname()
		if err != nil || host == "" {
			host = "local"
		}
		instanceID = host + ":" + opt.Addr
	}
	st.SetInstanceID(instanceID)
	mgr.SetInstanceID(instanceID)
	if err := mgr.ReconcileInterrupted(); err != nil {
		return fmt.Errorf("reconcile interrupted runs: %w", err)
	}
	// 后台子代理通知重启重投（N2 WP5）：未投递结果装回队列并按空闲语义投递。
	mgr.LoadPendingNotices()
	mgr.SetSquadRunner(&runtime.SquadRunner{})

	// 意图闸门：插件意图 overlay → agent-intent 注册表（未装载意图 = 行为回退）。
	gate := intentgate.New()
	mgr.SetGate(gate)

	pluginHost = plugins.New(st, reg, setSvc, opt.PluginsDir)
	pluginHost.SetSkillsRoot(skillsRoot)
	hostTok := make([]byte, 16)
	_, _ = rand.Read(hostTok)
	hostSrv := &hostapi.Server{Token: hex.EncodeToString(hostTok), Store: st}
	hostLn, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		return fmt.Errorf("host api: %w", err)
	}
	defer hostLn.Close()
	go func() { _ = http.Serve(hostLn, hostSrv.Handler()) }()
	pluginHost.SetHostAPI("http://"+hostLn.Addr().String(), hostSrv.Token)
	pluginHost.OnIntents = func(overlay []json.RawMessage) error { return gate.Rebuild(context.Background(), overlay) }
	wfReg := workflow.NewRegistry()
	pluginHost.OnWorkflows = func(defs []core.WorkflowDef, scripts map[string]string) error {
		wfReg.Set(defs, scripts)
		return nil
	}
	for id, perr := range pluginHost.LoadDir() {
		if perr != nil {
			log.Printf("[plugins] %s: %s: %v", id, errcode.SysUnavailable, perr) // fail-closed：单个失败不阻断启动
			logger.Error("plugin load failed", perr, "plugin", id, "code", string(errcode.SysUnavailable))
		}
	}

	// 检索管线（4.12）：分块+全文+（配置时）向量混合与 rerank；未配置自动退化为全文。
	pipe := rag.NewPipeline(st, setSvc, nil, nil)
	embedder := rag.SwitchingEmbedder{HTTP: rag.NewHTTPEmbedder(pipe.EmbeddingRoute), Provider: func() string {
		return setSvc.Effective(store.OwnerDefault, "").GetString("models.embedding.provider", "")
	}}
	pipe.SetEmbedder(embedder)
	pipe.SetReranker(rag.NewHTTPReranker(pipe.RerankRoute))
	mgr.SetRetriever(pipe)
	mgr.SetIndexer(pipe)

	// 意图慢通道（I2）：语义向量裁决 + 轻量模型命令生成。
	// 两个开关都默认关——此时识别行为与 I1 完全一致（fail-closed 分层）；
	// 开启语义需已配置 embedding 端点，开启生成需可用的轻量/主模型档。
	if setSvc.Effective(store.OwnerDefault, "").GetBool("runtime.intentSemantic", false) {
		gate.SetSemantic(intentsem.EmbeddingScorer{Provider: embedder}, 0)
	}
	if setSvc.Effective(store.OwnerDefault, "").GetBool("runtime.intentGenerator", false) {
		effNow := setSvc.Effective(store.OwnerDefault, "")
		light := runtime.ResolveLightweightModel(core.AgentProfile{}, effNow)
		client := &agentllm.Client{BaseURL: light.Base, APIKey: light.Key, HTTPClient: &http.Client{}}
		gate.SetGenerator(intentsem.LightGenerator{Stream: client.Stream, Model: light.Model})
	}
	// fs 预留能力沙箱根：<DataDir>/workspaces/<owner>/<conversationID>/。
	mgr.SetWorkspaceRoot(filepath.Join(opt.DataDir, "workspaces"))
	// 中断子代理续跑（N2 WP5）：依赖转录目录（workspaceRoot）与派生请求。
	mgr.ResumeInterruptedChildren()

	// 收件箱任务卡：事件走会话事件管道（持久化/seq/重放）；关闭即 T0 基线。
	cardSvc := cardsvc.New(st, mgr.EmitCardEvent)
	mgr.SetCards(cardSvc)

	// 命令总线（提案 T3/D27/D30）：四入口同源分发 + 审计 + 危险内联确认卡。
	bus := cmdbus.New(st, cardSvc, func() time.Time { return time.Now().UTC() })
	bus.SetPendingStore(st)
	pluginHost.OnCommands = bus.RegisterPluginCommands
	// hostcaps：hostapi 闭包与进程内插件句柄共用的 owner 作用域能力（EXT-1 批注 1）。
	authzSlot := &projectAuthz{}
	caps := hostcaps{st: st, pipe: pipe, set: setSvc, authz: authzSlot}
	// EXT-11：公开受限 Host 工厂（下游恢复路径经 extension.NewScopedHost 重建句柄）。
	wireScopedHostFactory(caps)
	applyCompiled(skillsRoot, bus, caps)
	mgr.OnUsage = func(owner string, tokens int) {
		budget := int(setSvc.Effective(owner, "").GetNumber("llm.dailyTokenBudget", 0))
		status, aerr := st.AccountTokens(owner, tokens, budget)
		if aerr != nil {
			logger.Error("llm budget", aerr, "code", string(errcode.SysInternal))
			return
		}
		if status.Over {
			_ = st.AppendAuditEvent(store.AuditEvent{Owner: owner, Action: "llm.budget_exceeded", Evidence: fmt.Sprintf(`{"total":%d,"budget":%d}`, status.Total, budget)})
			logger.Warn("llm daily budget exceeded", "owner", owner, "total", status.Total, "budget", budget)
		}
	}
	hostSrv.Dispatch = func(ctx context.Context, pluginID, owner, commandID string, params map[string]any) error {
		resp := bus.Dispatch(ctx, cmdbus.Request{ID: commandID, Params: params, Entrance: "api", Actor: pluginID, Owner: owner})
		if resp.Receipt.State == "rejected" || resp.Receipt.State == "unknown" {
			return fmt.Errorf("%s", resp.Receipt.Reason)
		}
		return nil
	}
	hostSrv.Search = func(ctx context.Context, owner, text string) ([]core.RetrievalHit, error) {
		return caps.searchDefault(ctx, owner, text)
	}
	// EXT-8 余项：hostapi 检索面同样吃适配器共享（本人 + 放行的他人项目知识）。
	hostSrv.SharedSearch = func(ctx context.Context, owner, text string) ([]core.RetrievalHit, error) {
		own, err := caps.searchDefault(ctx, owner, text)
		if err != nil {
			return nil, err
		}
		shared, err := st.SearchKnowledgeShared(owner, core.RetrievalQuery{Text: text, TopK: 5}, func(docOwner, projectID core.ID) bool {
			ok, aerr := authzSlot.Authorize(owner, projectID)
			return aerr == nil && ok
		})
		if err != nil {
			return nil, err
		}
		return append(own, shared...), nil
	}
	hostSrv.Complete = func(ctx context.Context, owner, prompt string) (string, error) {
		return caps.complete(ctx, owner, prompt)
	}

	srv := server.New(st, mgr, reg, profSvc, setSvc, pluginHost)
	// EXT-8：装载期注册的资源授权适配器注入读取入口（未注册=默认拒绝，行为不变）。
	srv.SetProjectAuthorizer(authzSlot.Authorize)
	// EXT-9：宿主可关闭上传自动索引（发布决定移交下游命令层）。
	srv.DisableUploadIndexing = opt.DisableUploadIndexing
	// EXT-12 最小开关：会话卡不镜像进项目公共流（cardSvc 在上方已建）。
	cardSvc.MirrorConvCards = !opt.DisableConvCardMirror

	srv.Workflows = wfReg
	mgr.SetWorkflows(wfReg)
	srv.Cards = cardSvc
	srv.Bus = bus
	srv.Gate = gate
	srv.Prompts = promptSvc
	srv.WireBuiltinCommands()
	srv.Retriever = pipe
	srv.Indexer = pipe

	// 限流（阶段 3）：默认进程内档；设置 runtime.redisAddr 切原生 Redis 共享档。
	var limiter ratelimit.Limiter = ratelimit.NewMemory()
	if addr := setSvc.Effective(store.OwnerDefault, "").GetString("runtime.redisAddr", ""); addr != "" {
		rl := ratelimit.NewRedis(addr, "", 0)
		if err := rl.Ping(context.Background()); err != nil {
			return fmt.Errorf("redis %s: %w", addr, err)
		}
		defer rl.Close()
		limiter = rl
		log.Printf("rate limit: shared via redis %s", addr)
	}
	srv.Limiter = limiter
	srv.LimitLookup = cachedInt(5*time.Second, func() int {
		return int(setSvc.Effective(store.OwnerDefault, "").GetNumber("runtime.rateLimitPerMin", 0))
	})
	// 认证策略（auto）：PG 档默认开启（多用户面不能裸奔）；嵌入式档默认关闭（本地单用户）。
	authOn := resolved.AuthEnabled()
	srv.Auth = server.NewAuthenticator(st, authOn)
	if authOn {
		log.Printf("auth: API key required (create with: agent-system apikey create --owner <name>)")
	}
	srv.Subscribe = func(conversationID string) (<-chan core.StreamEvent, func()) {
		return b.Subscribe(conversationID)
	}

	httpSrv := &http.Server{Addr: opt.Addr, ReadHeaderTimeout: 10 * time.Second, Handler: server.RequestGuard{
		AuthOn: authOn, AllowedHosts: opt.AllowedHosts, Next: srv.Handler(),
	}}

	go func() {
		log.Printf("agent-system serving on http://%s (data: %s, plugins: %s)", opt.Addr, opt.DataDir, opt.PluginsDir)
		if err := httpSrv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
			log.Fatalf("http: %v", err)
		}
	}()

	// PG 档：事件跨实例扇出（NOTIFY/LISTEN；SQLite 档不需要）。
	if st.Dialect() == "postgres" {
		mgr.SetEventNotifier(pgbus.NewNotifier(st).Notify)
		listener, err := pgbus.Start(context.Background(), opt.StorageDSN, st, b)
		if err != nil {
			return fmt.Errorf("pgbus listen: %w", err)
		}
		defer listener.Close(context.Background())
		log.Printf("pgbus: LISTEN agent_events（跨实例事件扇出）")
	}

	stop := make(chan os.Signal, 1)
	signal.Notify(stop, os.Interrupt, syscall.SIGTERM)
	<-stop
	log.Printf("shutting down…")
	mgr.InterruptRunning()
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	if err := httpSrv.Shutdown(ctx); err != nil {
		return err
	}
	mgr.Shutdown() // 冲净事件缓冲后再关存储
	return nil
}

// CreateUserAccount CLI 建号（管理员发放制）：未给密码则生成随机 12 位并返回明文（仅此一次输出）。
func CreateUserAccount(storageDSN, dataDir, owner, name, password string, isAdmin bool) (string, error) {
	if password == "" {
		password = server.RandomPassword()
	}
	hash, err := store.HashPassword(password)
	if err != nil {
		return "", err
	}
	if err := withStore(storageDSN, dataDir, func(st *store.Store) error {
		return st.CreateAccount(owner, name, hash, isAdmin)
	}); err != nil {
		return "", err
	}
	return password, nil
}

// ResetUserPassword CLI 重置密码：must_change=1（首登强制改密），吊销全部会话。
func ResetUserPassword(storageDSN, dataDir, owner, password string) (string, error) {
	if password == "" {
		password = server.RandomPassword()
	}
	hash, err := store.HashPassword(password)
	if err != nil {
		return "", err
	}
	if err := withStore(storageDSN, dataDir, func(st *store.Store) error {
		if err := st.SetUserPassword(owner, hash); err != nil {
			return err
		}
		st.DeleteUserSessions(owner, "")
		return nil
	}); err != nil {
		return "", err
	}
	return password, nil
}

// ListUserAccounts CLI 用户列表。
func ListUserAccounts(storageDSN, dataDir string) ([]store.AdminUser, error) {
	var out []store.AdminUser
	err := withStore(storageDSN, dataDir, func(st *store.Store) error {
		var e error
		out, e = st.ListUsersAdmin()
		return e
	})
	return out, err
}

func withStore(storageDSN, dataDir string, fn func(*store.Store) error) error {
	var st *store.Store
	var err error
	if storageDSN != "" {
		st, err = store.OpenPostgres(storageDSN, filepath.Join(dataDir, "artifacts"))
	} else {
		st, err = store.OpenSQLite(filepath.Join(dataDir, "agent-system.db"))
	}
	if err != nil {
		return err
	}
	defer st.Close()
	return fn(st)
}

// CreateAPIKey 是 apikey 子命令的实现：在指定存储档上为 owner 创建密钥（明文仅打印一次）。
func CreateAPIKey(storageDSN, dataDir, owner, name string) (string, error) {
	var st *store.Store
	var err error
	if storageDSN != "" {
		st, err = store.OpenPostgres(storageDSN, filepath.Join(dataDir, "artifacts"))
	} else {
		st, err = store.OpenSQLite(filepath.Join(dataDir, "agent-system.db"))
	}
	if err != nil {
		return "", err
	}
	defer st.Close()
	return st.CreateAPIKey(owner, name)
}

// ListAPIKeys 列出租户密钥（摘要视图）。
func ListAPIKeys(storageDSN, dataDir, owner string) ([]store.APIKeyInfo, error) {
	var st *store.Store
	var err error
	if storageDSN != "" {
		st, err = store.OpenPostgres(storageDSN, filepath.Join(dataDir, "artifacts"))
	} else {
		st, err = store.OpenSQLite(filepath.Join(dataDir, "agent-system.db"))
	}
	if err != nil {
		return nil, err
	}
	defer st.Close()
	if _, err := st.GetUser(owner); err != nil {
		return nil, fmt.Errorf("owner %q 不存在（先 apikey create 创建）", owner)
	}
	return st.ListAPIKeys(owner)
}

// RevokeAPIKey 撤销指定名称密钥。
func RevokeAPIKey(storageDSN, dataDir, owner, name string) (bool, error) {
	var st *store.Store
	var err error
	if storageDSN != "" {
		st, err = store.OpenPostgres(storageDSN, filepath.Join(dataDir, "artifacts"))
	} else {
		st, err = store.OpenSQLite(filepath.Join(dataDir, "agent-system.db"))
	}
	if err != nil {
		return false, err
	}
	defer st.Close()
	return st.RevokeAPIKeyByName(owner, name)
}

// cachedInt 带短 TTL 的整型缓存读（限流值热读取免打库）。
func cachedInt(ttl time.Duration, fn func() int) func() int {
	var mu sync.Mutex
	var val int
	var at time.Time
	return func() int {
		mu.Lock()
		defer mu.Unlock()
		if time.Since(at) > ttl {
			val = fn()
			at = time.Now()
		}
		return val
	}
}

// splitKeyPool 拆逗号分隔密钥池（trim 空段）。
// mcpCache 按配置串缓存 MCP Manager：配置不变复用同一实例（后端解析器
// per-role 缓存会长期持有它，进程型 server 不重复拉起）；配置变更时关闭旧
// Manager（杀掉其进程型 server）再重建。配置错误返回 nil 并留日志——
// MCP 是可选能力，坏配置禁用该能力而不是阻塞产品启动。
var (
	mcpMu      sync.Mutex
	mcpKey     string
	mcpManager *agentmcp.Manager
)

func mcpManagerFor(raw, overrides string) *agentmcp.Manager {
	mcpMu.Lock()
	defer mcpMu.Unlock()
	key := raw + "\n" + overrides
	if raw == "" {
		if mcpManager != nil {
			mcpManager.Close()
			mcpManager = nil
		}
		mcpKey = ""
		return nil
	}
	if key == mcpKey && mcpManager != nil {
		return mcpManager
	}
	var servers []agentmcp.ServerConfig
	if err := json.Unmarshal([]byte(raw), &servers); err != nil {
		log.Printf("squad agent-native: tools.mcp.servers is not valid JSON, MCP disabled: %v", err)
		return nil
	}
	mgr, err := agentmcp.NewManager(servers)
	if err != nil {
		log.Printf("squad agent-native: MCP config invalid, MCP disabled: %v", err)
		return nil
	}
	if err := mgr.ApplyOverrides(overrides); err != nil {
		log.Printf("squad agent-native: MCP overrides invalid, MCP disabled: %v", err)
		return nil
	}
	if mcpManager != nil {
		mcpManager.Close()
	}
	mcpKey, mcpManager = key, mgr
	return mgr
}

func splitKeyPool(s string) []string {
	var out []string
	for _, k := range strings.Split(s, ",") {
		if k = strings.TrimSpace(k); k != "" {
			out = append(out, k)
		}
	}
	return out
}