Agent-System/cmd/control-center/main.go

// Control Center: personal project control plane, independent of the product
// system. Subcommands: serve, migrate, register-agent, import-docs, plus the
// agent CLI (next / instruction / claim / heartbeat / release / event /
// artifact / advance-request / decision / report).
package main

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"flag"
	"fmt"
	"log/slog"
	"net"
	"net/http"
	"os"
	"os/signal"
	"syscall"
	"time"

	"chat-agent-system/internal/controlcenter/audit"
	"chat-agent-system/internal/controlcenter/config"
	"chat-agent-system/internal/controlcenter/docs"
	"chat-agent-system/internal/controlcenter/embed"
	"chat-agent-system/internal/controlcenter/importer"
	"chat-agent-system/internal/controlcenter/llm"
	"chat-agent-system/internal/controlcenter/server"
	"chat-agent-system/internal/controlcenter/store"
)

func main() {
	if len(os.Args) < 2 {
		usage()
		os.Exit(2)
	}
	var err error
	switch os.Args[1] {
	case "serve":
		err = cmdServe(os.Args[2:])
	case "migrate":
		err = cmdMigrate(os.Args[2:])
	case "register-agent":
		err = cmdRegisterAgent(os.Args[2:])
	case "import-docs":
		err = cmdImportDocs(os.Args[2:])
	case "next":
		err = cmdAgentNext(os.Args[2:])
	case "instruction":
		err = cmdAgentInstruction(os.Args[2:])
	case "claim":
		err = simpleTaskAction("claim")(os.Args[2:])
	case "heartbeat":
		err = simpleTaskAction("heartbeat")(os.Args[2:])
	case "release":
		err = simpleTaskAction("release")(os.Args[2:])
	case "event":
		err = cmdAgentEvent(os.Args[2:])
	case "artifact":
		err = cmdAgentArtifact(os.Args[2:])
	case "advance-request":
		err = cmdAgentAdvanceRequest(os.Args[2:])
	case "decision":
		err = cmdAgentDecision(os.Args[2:])
	case "report":
		err = cmdAgentReport(os.Args[2:])
	default:
		usage()
		os.Exit(2)
	}
	if err != nil {
		slog.Error("control-center", "err", err)
		os.Exit(1)
	}
}

func usage() {
	fmt.Fprintln(os.Stderr, `control-center — personal project control plane

usage:
  control-center serve            run HTTP server (CC_HTTP_ADDR, default 127.0.0.1:8790)
  control-center migrate          apply goose migrations only
  control-center register-agent -name X [-kind K] [-token T]   create/rotate an agent token
  control-center import-docs      import .doc task docs into cc_tasks (also runs periodically in serve)

agent CLI (common flags: -server, default $CC_SERVER or http://127.0.0.1:8790;
                          -token, default $CC_AGENT_TOKEN; -task takes a task uuid — use 'next' to discover ids):
  next                                     print next claimable task (slug/title/id/stage)
  instruction -task ID [-md]               print the instruction package (JSON, or readable with -md)
  claim|heartbeat|release -task ID         lease lifecycle
  event -task ID -phase P -message M [-payload '<json>']
  artifact -task ID -kind K -title T [-file F | -content MD]
  advance-request -task ID -summary S [-artifacts id1,id2]
  decision -task ID -question Q -context MD -suggestion "选项||理由||影响" [-suggestion ...]
  report -task ID -kind result|test-report|handoff [-file F | -content MD]`)
}

func cmdServe(args []string) error {
	fs := flag.NewFlagSet("serve", flag.ContinueOnError)
	if err := fs.Parse(args); err != nil {
		return err
	}
	cfg, err := config.Load()
	if err != nil {
		return err
	}
	ctx := context.Background()
	if err := store.Migrate(ctx, cfg.DBDSN, cfg.MigrationsDir); err != nil {
		return err
	}
	st, err := store.Connect(ctx, cfg.DBDSN)
	if err != nil {
		return err
	}
	defer st.Close()

	loopback := isLoopbackAddr(cfg.HTTPAddr)
	if cfg.ConsoleToken == "" {
		if !loopback {
			return fmt.Errorf("refusing to bind %s without a fixed console token: set CC_CONSOLE_TOKEN", cfg.HTTPAddr)
		}
		cfg.ConsoleToken = randomToken()
		slog.Warn("CC_CONSOLE_TOKEN not set; generated ephemeral console token")
	}
	if !loopback {
		slog.Warn("control-center is reachable from the network; all /api routes require tokens")
	}
	srv := server.New(cfg, st)

	var emb *embed.HTTPClient
	if cfg.EmbeddingEndpoint != "" {
		emb = embed.NewHTTP(cfg.EmbeddingEndpoint, cfg.EmbeddingAPIKey, cfg.EmbeddingModel, embed.Dim)
	} else {
		slog.Warn("embedding endpoint not configured; search runs keyword-only")
	}
	syncer := docs.New(st, cfg.RepoRoot, emb)
	var embIface embed.Embedder
	if emb != nil {
		embIface = emb
	}
	srv.SetDocs(embIface, syncer)

	chat := llm.NewHTTP(cfg.LLMEndpoint, cfg.LLMAPIKey, cfg.LLMModel)
	if !chat.Configured() {
		slog.Warn("LLM endpoint/model not configured; audits run deterministic checks only")
	}
	auditRunner := audit.New(st, cfg.RepoRoot, chat)
	srv.SetAudit(auditRunner)

	syncCtx, stopSync := context.WithCancel(ctx)
	defer stopSync()
	syncer.StartPeriodic(syncCtx, time.Duration(cfg.DocSyncSecond)*time.Second)
	importer.New(st, cfg.RepoRoot).StartPeriodic(syncCtx, time.Duration(cfg.DocSyncSecond)*time.Second)
	if cfg.AuditEnabled {
		auditRunner.StartScheduler(syncCtx, cfg.AuditDailyAt)
	}

	reapCtx, stopReaper := context.WithCancel(ctx)
	defer stopReaper()
	srv.StartLeaseReaper(reapCtx)

	httpSrv := &http.Server{Addr: cfg.HTTPAddr, Handler: srv.Handler(), ReadHeaderTimeout: 10 * time.Second}
	go func() {
		sig := make(chan os.Signal, 1)
		signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)
		<-sig
		stopSync()
		stopReaper()
		shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		_ = httpSrv.Shutdown(shutdownCtx)
	}()
	slog.Info("control-center listening", "addr", cfg.HTTPAddr)
	fmt.Printf("console token: %s\n", cfg.ConsoleToken)
	if err := httpSrv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
		return err
	}
	return nil
}

func cmdMigrate(args []string) error {
	fs := flag.NewFlagSet("migrate", flag.ContinueOnError)
	if err := fs.Parse(args); err != nil {
		return err
	}
	cfg, err := config.Load()
	if err != nil {
		return err
	}
	return store.Migrate(context.Background(), cfg.DBDSN, cfg.MigrationsDir)
}

func cmdImportDocs(args []string) error {
	fs := flag.NewFlagSet("import-docs", flag.ContinueOnError)
	if err := fs.Parse(args); err != nil {
		return err
	}
	cfg, err := config.Load()
	if err != nil {
		return err
	}
	ctx := context.Background()
	st, err := store.Connect(ctx, cfg.DBDSN)
	if err != nil {
		return err
	}
	defer st.Close()
	stats, err := importer.New(st, cfg.RepoRoot).Import(ctx)
	if err != nil {
		return err
	}
	fmt.Printf("import-docs: added=%d updated=%d skipped=%d\n", stats.Added, stats.Updated, stats.Skipped)
	return nil
}

func cmdRegisterAgent(args []string) error {
	fs := flag.NewFlagSet("register-agent", flag.ContinueOnError)
	name := fs.String("name", "", "agent name (required)")
	kind := fs.String("kind", "generic", "agent kind, e.g. cursor / kimi / pi")
	token := fs.String("token", "", "plaintext token; generated when empty")
	if err := fs.Parse(args); err != nil {
		return err
	}
	if *name == "" {
		return fmt.Errorf("-name is required")
	}
	cfg, err := config.Load()
	if err != nil {
		return err
	}
	ctx := context.Background()
	if err := store.Migrate(ctx, cfg.DBDSN, cfg.MigrationsDir); err != nil {
		return err
	}
	st, err := store.Connect(ctx, cfg.DBDSN)
	if err != nil {
		return err
	}
	defer st.Close()
	if *token == "" {
		*token = randomToken()
	}
	if err := server.RegisterAgent(ctx, st, *name, *kind, *token); err != nil {
		return err
	}
	fmt.Printf("agent %q registered. token (shown once): %s\n", *name, *token)
	return nil
}

func randomToken() string {
	b := make([]byte, 24)
	if _, err := rand.Read(b); err != nil {
		panic(err)
	}
	return hex.EncodeToString(b)
}

// isLoopbackAddr reports whether addr ("host:port") binds only to loopback.
func isLoopbackAddr(addr string) bool {
	host, _, err := net.SplitHostPort(addr)
	if err != nil {
		host = addr
	}
	if host == "" {
		return false // ":8790" listens on all interfaces
	}
	if ip := net.ParseIP(host); ip != nil {
		return ip.IsLoopback()
	}
	return host == "localhost"
}