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

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

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

	agenttools "agent-native/runtime/tools"
	"agent-system-backend/internal/bus"
	"agent-system-backend/internal/cardsvc"
	"agent-system-backend/internal/cmdbus"
	"agent-system-backend/internal/intentgate"
	"agent-system-backend/internal/pgbus"
	"agent-system-backend/internal/plugins"
	"agent-system-backend/internal/profiles"
	"agent-system-backend/internal/prompts"
	"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/squadadapter"
	"agent-system-backend/internal/store"
	"agent-system-backend/internal/tools"
	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
}

// Run 装配并阻塞服务直至 SIGINT/SIGTERM（优雅停机：中断在途 run、冲净事件缓冲）。
func Run(opt Options) error {
	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)
	}
	setSvc := settings.NewService(st)
	setSvc.SetSystemOverride(map[string]any{"storage": map[string]any{"dataDir": opt.DataDir}})
	// 网络搜索（Tavily）：配置密钥池才注册（对齐 agent-native 语义：未配置不静默降级）。
	eff := setSvc.Effective(store.OwnerDefault, "")
	if keys := splitKeyPool(eff.GetString("tools.tavily.apiKeys", "")); 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
	}
	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}
	subRunner := &runtime.SubLLMRunner{Prompts: promptSvc}
	mockRunner := &runtime.MockRunner{}
	mgr := runmgr.New(st, b, reg, profSvc, setSvc, convRunner, subRunner, mockRunner, &runtime.MockSubagent{})

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

	pluginHost := plugins.New(st, reg, setSvc, opt.PluginsDir)
	pluginHost.OnIntents = func(overlay []json.RawMessage) error { return gate.Rebuild(context.Background(), overlay) }
	for id, err := range pluginHost.LoadDir() {
		if err != nil {
			log.Printf("[plugins] %s: %v", id, err) // fail-closed：单个失败不阻断启动
		}
	}

	// 检索管线（4.12）：分块+全文+（配置时）向量混合与 rerank；未配置自动退化为全文。
	pipe := rag.NewPipeline(st, setSvc, nil, nil)
	pipe.SetEmbedder(rag.NewHTTPEmbedder(pipe.EmbeddingRoute))
	pipe.SetReranker(rag.NewHTTPReranker(pipe.RerankRoute))
	mgr.SetRetriever(pipe)
	mgr.SetIndexer(pipe)
	// fs 预留能力沙箱根：<DataDir>/workspaces/<owner>/<conversationID>/。
	mgr.SetWorkspaceRoot(filepath.Join(opt.DataDir, "workspaces"))

	// 收件箱任务卡：事件走会话事件管道（持久化/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() })
	pluginHost.OnCommands = bus.RegisterPluginCommands

	srv := server.New(st, mgr, reg, profSvc, setSvc, pluginHost)
	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 := opt.AuthMode == "on" || (opt.AuthMode != "off" && opt.StorageDSN != "")
	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, Handler: srv.Handler(), ReadHeaderTimeout: 10 * time.Second}

	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 空段）。
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
}