knowledge-governance/cmd/libraryctl/main.go

// Command libraryctl is the operator entry point for the whole-library capture
// chain: it reads one test knowledge base (internal or external) from the real
// WeKnora test stack, freezes its writers for the duration of the export and
// seals a content-addressed candidate artifact.
//
// Boundaries this command keeps:
//   - the source is read through a read-only PostgreSQL session; no source
//     schema, data or program is modified;
//   - only SIGTERM is sent to the declared writer group, and only when every
//     service of the Compose project is declared (a new sibling container is
//     refused, not ignored);
//   - credentials are read from the running containers at the moment of the
//     call and are never printed, logged or written to disk;
//   - the sealed artifact is a *candidate*: it is not a release, and it does
//     not replace snapshot-bound content, Q&A and performance acceptance.
package main

import (
	"context"
	"encoding/json"
	"errors"
	"flag"
	"fmt"
	"os"
	"os/exec"
	"os/signal"
	"path/filepath"
	"sort"
	"strings"
	"syscall"
	"time"

	"github.com/jackc/pgx/v5/pgxpool"

	"knowledge-governance/internal/libraryfreeze"
	"knowledge-governance/internal/librarymaintenance"
	"knowledge-governance/internal/librarysnapshot"
	v071 "knowledge-governance/internal/releaseadapter/v071"
	releaseworkflow "knowledge-governance/internal/workflows/release"
)

type options struct {
	sourceApp      string
	sourcePG       string
	sourceES       string
	governancePG   string
	project        string
	writers        string
	nonWriters     string
	archiveRoot    string
	journalRoot    string
	dockerSocket   string
	maxDocuments   int
	maxObjectBytes int64
	scope          string
	job            string
	dryRun         bool
	snapshot       string
}

func main() {
	if len(os.Args) < 2 {
		usage()
		os.Exit(2)
	}
	command := os.Args[1]
	opts, rest, err := parseFlags(command, os.Args[2:])
	if err != nil {
		fmt.Fprintln(os.Stderr, "LIBRARYCTL_FLAGS:", err)
		os.Exit(2)
	}
	_ = rest
	// SIGTERM must not kill this process mid-window: the capture path restores the
	// source writers through a cancelled-context recovery, so the signal is turned
	// into a context cancellation and the process exits only after that cleanup.
	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
	defer stop()
	ctx, cancel := context.WithTimeout(ctx, 30*time.Minute)
	defer cancel()
	var runErr error
	switch command {
	case "status":
		runErr = runStatus(ctx, opts)
	case "preflight":
		runErr = runPreflight(ctx, opts)
	case "capture":
		runErr = runCapture(ctx, opts)
	case "list":
		runErr = runList(opts)
	case "verify":
		runErr = runVerify(opts)
	case "recover":
		runErr = runRecover(ctx, opts)
	default:
		usage()
		os.Exit(2)
	}
	if runErr != nil {
		fmt.Fprintln(os.Stderr, "LIBRARYCTL_FAILED:", runErr)
		os.Exit(1)
	}
}

func usage() {
	fmt.Fprint(os.Stderr, `libraryctl <status|preflight|capture|list|verify|recover> [flags]

  status      probe the real source and print what is configured / blocked
  preflight   read the source, run settlement + candidate validation (no writes, no stop)
  capture     freeze the declared writer group, archive a candidate, restore writers
  list        list sealed candidates and stored bytes
  verify      re-verify one sealed candidate (digests, coverage, identity)
  recover     restore writers from an interrupted freeze journal

Flags (defaults target the local WKNFull test stack):
  -source-app WKNFull-app         container whose env describes the source
  -source-pg WKNFull-postgres     source PostgreSQL container (address resolution)
  -source-es WKNFull-elasticsearch
  -governance-pg knowledge-governance-p1-postgres-1   routing bindings read here
  -project weknora071full         Compose project owning the writer group
  -writers WKNFull-app            declared writer services (comma separated)
  -non-writers WKNFull-frontend=reason,...            every other project service
  -archive ~/.local/share/knowledge-governance/library
  -journal ~/.local/share/knowledge-governance/freeze
  -docker-socket /var/run/docker.sock
  -scope internal|external        (preflight/capture)
  -job <id>                       (capture; default: lib-<scope>-<UTC timestamp>)
  -dry-run                        (capture: export + validate only, no archive, no stop)
  -snapshot <id>                  (verify)
`)
}

func parseFlags(command string, args []string) (options, []string, error) {
	home, _ := os.UserHomeDir()
	opts := options{
		sourceApp:      "WKNFull-app",
		sourcePG:       "WKNFull-postgres",
		sourceES:       "WKNFull-elasticsearch",
		governancePG:   "knowledge-governance-p1-postgres-1",
		project:        "weknora071full",
		writers:        "WKNFull-app",
		archiveRoot:    filepath.Join(home, ".local/share/knowledge-governance/library"),
		journalRoot:    filepath.Join(home, ".local/share/knowledge-governance/freeze"),
		dockerSocket:   "/var/run/docker.sock",
		maxDocuments:   100000,
		maxObjectBytes: 134217728,
	}
	set := flag.NewFlagSet(command, flag.ContinueOnError)
	set.SetOutput(os.Stderr)
	set.StringVar(&opts.sourceApp, "source-app", opts.sourceApp, "source container describing the stack")
	set.StringVar(&opts.sourcePG, "source-pg", opts.sourcePG, "source PostgreSQL container")
	set.StringVar(&opts.sourceES, "source-es", opts.sourceES, "source Elasticsearch container")
	set.StringVar(&opts.governancePG, "governance-pg", opts.governancePG, "governance PostgreSQL container")
	set.StringVar(&opts.project, "project", opts.project, "Compose project owning the writers")
	set.StringVar(&opts.writers, "writers", opts.writers, "declared writer services")
	set.StringVar(&opts.nonWriters, "non-writers", "", "service=reason for every other project service")
	set.StringVar(&opts.archiveRoot, "archive", opts.archiveRoot, "sealed artifact root")
	set.StringVar(&opts.journalRoot, "journal", opts.journalRoot, "freeze journal root")
	set.StringVar(&opts.dockerSocket, "docker-socket", opts.dockerSocket, "Docker engine socket")
	set.IntVar(&opts.maxDocuments, "max-documents", opts.maxDocuments, "maximum indexed documents per scope")
	set.Int64Var(&opts.maxObjectBytes, "max-object-bytes", opts.maxObjectBytes, "maximum bytes per stored object")
	set.StringVar(&opts.scope, "scope", "", "internal|external")
	set.StringVar(&opts.job, "job", "", "capture job id")
	set.BoolVar(&opts.dryRun, "dry-run", false, "export and validate without archiving or stopping")
	set.StringVar(&opts.snapshot, "snapshot", "", "snapshot id for verify")
	if err := set.Parse(args); err != nil {
		return opts, nil, err
	}
	return opts, set.Args(), nil
}

// ---------------------------------------------------------------- discovery

func dockerOutput(ctx context.Context, args ...string) (string, error) {
	bounded, cancel := context.WithTimeout(ctx, 60*time.Second)
	defer cancel()
	out, err := exec.CommandContext(bounded, "docker", args...).Output()
	if err != nil {
		var exit *exec.ExitError
		if errors.As(err, &exit) && len(exit.Stderr) > 0 {
			return "", fmt.Errorf("docker %s failed", args[0])
		}
		return "", fmt.Errorf("docker %s failed", args[0])
	}
	return strings.TrimSpace(string(out)), nil
}

func containerEnv(ctx context.Context, name string) (map[string]string, error) {
	raw, err := dockerOutput(ctx, "inspect", "--format", "{{json .Config.Env}}", name)
	if err != nil {
		return nil, err
	}
	var entries []string
	if err := json.Unmarshal([]byte(raw), &entries); err != nil {
		return nil, errors.New("invalid container environment")
	}
	env := map[string]string{}
	for _, entry := range entries {
		if key, value, ok := strings.Cut(entry, "="); ok {
			env[key] = value
		}
	}
	return env, nil
}

func containerAddress(ctx context.Context, name string) (string, error) {
	raw, err := dockerOutput(ctx, "inspect", "--format",
		"{{range .NetworkSettings.Networks}}{{.IPAddress}} {{end}}", name)
	if err != nil {
		return "", err
	}
	for _, field := range strings.Fields(raw) {
		if field != "" && field != "0.0.0.0" {
			return field, nil
		}
	}
	return "", fmt.Errorf("container %s has no reachable address", name)
}

func containerImageID(ctx context.Context, name string) (string, error) {
	return dockerOutput(ctx, "inspect", "--format", "{{.Image}}", name)
}

// sourceConfig is everything needed to read the test library. Credential fields
// are never printed: String() masks them and no log line formats the struct.
type sourceConfig struct {
	Scopes       map[string]string
	DatabaseURL  string
	Search       v071.SearchExporterConfig
	Object       v071.S3ObjectStoreConfig
	AppImageID   string
	SearchIndex  string
	ObjectDetail string
}

func discoverSource(ctx context.Context, opts options, binding map[string]string) (sourceConfig, error) {
	var config sourceConfig
	if binding == nil {
		return config, errors.New("routing bindings are required")
	}
	config.Scopes = binding
	app, err := containerEnv(ctx, opts.sourceApp)
	if err != nil {
		return config, err
	}
	if app["DB_DRIVER"] != "postgres" {
		return config, errors.New("source database driver is not postgres")
	}
	address, err := containerAddress(ctx, opts.sourcePG)
	if err != nil {
		return config, err
	}
	dsn := strings.Join([]string{
		"postgres://" + urlEscape(app["DB_USER"]) + ":" + urlEscape(app["DB_PASSWORD"]) +
			"@" + address + ":" + app["DB_PORT"] + "/" + urlEscape(app["DB_NAME"]) + "?sslmode=disable",
	}, "")
	config.DatabaseURL = dsn
	config.AppImageID, err = containerImageID(ctx, opts.sourceApp)
	if err != nil {
		return config, err
	}
	esAddress, err := containerAddress(ctx, opts.sourceES)
	if err != nil {
		return config, err
	}
	config.SearchIndex = app["ELASTICSEARCH_INDEX"]
	config.Search = v071.SearchExporterConfig{BaseURL: "http://" + esAddress + ":9200", Index: app["ELASTICSEARCH_INDEX"],
		AuthorizationToken: app["ELASTICSEARCH_PASSWORD"], PageSize: 500}
	config.Object, config.ObjectDetail, err = objectStoreFromSourceEnv(app, opts.maxObjectBytes)
	if err != nil {
		return config, err
	}
	return config, nil
}

// objectStoreFromSourceEnv 把源应用的存储环境映射成对象存储配置。
//
// 只认 OSS 时，S3 兼容的 MinIO 目标（本机专用验收实例）会在发现阶段就被拒，
// 验收链只能跑在生产存储上——这不是安全性，而是能力缺口。两种都要能读，
// 读不出端点/桶就明确报错，不退回任何默认值。
func objectStoreFromSourceEnv(app map[string]string, maxObjectBytes int64) (v071.S3ObjectStoreConfig, string, error) {
	storage := strings.ToLower(strings.TrimSpace(app["STORAGE_TYPE"]))
	var config v071.S3ObjectStoreConfig
	var endpoint, bucket string
	switch storage {
	case "oss":
		config = v071.S3ObjectStoreConfig{Provider: "oss", Region: app["OSS_REGION"],
			AccessKeyID: app["OSS_ACCESS_KEY"], SecretAccessKey: app["OSS_SECRET_KEY"],
			Bucket: app["OSS_BUCKET_NAME"], MaxObjectBytes: maxObjectBytes}
		endpoint, bucket = app["OSS_ENDPOINT"], app["OSS_BUCKET_NAME"]
		if !strings.Contains(endpoint, "://") {
			endpoint = "https://" + endpoint
		}
	case "minio":
		accessKey := firstNonEmpty(app["MINIO_ACCESS_KEY_ID"], app["MINIO_ACCESS_KEY"])
		secretKey := firstNonEmpty(app["MINIO_SECRET_ACCESS_KEY"], app["MINIO_SECRET_KEY"])
		config = v071.S3ObjectStoreConfig{Provider: "minio", Region: app["MINIO_REGION"],
			AccessKeyID: accessKey, SecretAccessKey: secretKey, Bucket: app["MINIO_BUCKET_NAME"],
			MaxObjectBytes: maxObjectBytes}
		endpoint, bucket = app["MINIO_ENDPOINT"], app["MINIO_BUCKET_NAME"]
		if strings.TrimSpace(endpoint) == "" {
			return v071.S3ObjectStoreConfig{}, "", errors.New("source minio endpoint is not configured")
		}
		scheme := "http"
		if parseBool(app["MINIO_USE_SSL"]) {
			scheme = "https"
		}
		if !strings.Contains(endpoint, "://") {
			endpoint = scheme + "://" + endpoint
		}
	default:
		return v071.S3ObjectStoreConfig{}, "", fmt.Errorf("unsupported source storage type %q", app["STORAGE_TYPE"])
	}
	if strings.TrimSpace(endpoint) == "" {
		return v071.S3ObjectStoreConfig{}, "", errors.New("source object endpoint is not configured")
	}
	config.Endpoint = endpoint
	config.Secure = strings.HasPrefix(endpoint, "https://")
	config.ForcePathStyle = storage == "minio"
	if config.Bucket == "" {
		return v071.S3ObjectStoreConfig{}, "", errors.New("source object bucket is not configured")
	}
	return config, "provider=" + config.Provider + " bucket=" + bucket, nil
}

func firstNonEmpty(values ...string) string {
	for _, value := range values {
		if strings.TrimSpace(value) != "" {
			return value
		}
	}
	return ""
}

func parseBool(value string) bool {
	switch strings.ToLower(strings.TrimSpace(value)) {
	case "1", "true", "yes", "on":
		return true
	}
	return false
}

func urlEscape(value string) string {
	replacer := strings.NewReplacer("%", "%25", ":", "%3A", "@", "%40", "/", "%2F", "?", "%3F", "#", "%23", "&", "%26", " ", "%20")
	return replacer.Replace(value)
}

// scopeBindings 返回 scope→知识库 映射。治理 API 用的是 LIBRARY_PRODUCTION_KB_*，
// CLI 读的是治理库里的 routing bindings；同一台机器上两份不一致时，
// CLI 抓到的会是"配置里那个库"而不是 API 认为的目标——所以环境变量优先，
// 读不到才退回 bindings。两边讲同一份身份，验收才可比。
func scopeBindings(ctx context.Context, container string) (map[string]string, error) {
	binding, err := readRoutingBindings(ctx, container)
	if err != nil {
		return nil, err
	}
	for scope, key := range map[string]string{"internal": "LIBRARY_PRODUCTION_KB_INTERNAL", "external": "LIBRARY_PRODUCTION_KB_EXTERNAL"} {
		if value := strings.TrimSpace(os.Getenv(key)); value != "" {
			binding[scope] = value
		}
	}
	return binding, nil
}

func readRoutingBindings(ctx context.Context, container string) (map[string]string, error) {
	raw, err := dockerOutput(ctx, "exec", container, "psql", "-U", "governance", "-d", "governance", "-tAc",
		"SELECT internal_kb_id||'|'||external_kb_id FROM routing_kb_bindings WHERE id='active'")
	if err != nil {
		return nil, errors.New("read routing bindings failed")
	}
	internal, external, ok := strings.Cut(strings.TrimSpace(raw), "|")
	if !ok || internal == "" || external == "" {
		return nil, errors.New("routing bindings are incomplete")
	}
	return map[string]string{"internal": internal, "external": external}, nil
}

func openSourceDatabase(ctx context.Context, config sourceConfig) (*pgxpool.Pool, error) {
	parsed, err := pgxpool.ParseConfig(config.DatabaseURL)
	if err != nil {
		return nil, errors.New("invalid source database url")
	}
	// The capture chain never writes to the source. The read-only session makes
	// that a database-enforced property rather than a convention.
	parsed.ConnConfig.RuntimeParams["default_transaction_read_only"] = "on"
	parsed.ConnConfig.RuntimeParams["statement_timeout"] = "120000"
	pool, err := pgxpool.NewWithConfig(ctx, parsed)
	if err != nil {
		return nil, errors.New("open source database failed")
	}
	var readOnly string
	if err = pool.QueryRow(ctx, "SHOW default_transaction_read_only").Scan(&readOnly); err != nil || readOnly != "on" {
		pool.Close()
		return nil, errors.New("source session is not read-only")
	}
	return pool, nil
}

func tenantAndKnowledgeBase(ctx context.Context, pool *pgxpool.Pool, knowledgeBaseID string) (uint64, error) {
	var tenant uint64
	err := pool.QueryRow(ctx, "SELECT tenant_id FROM knowledge_bases WHERE id=$1 AND deleted_at IS NULL", knowledgeBaseID).Scan(&tenant)
	if err != nil || tenant == 0 {
		return 0, errors.New("knowledge base not found in source")
	}
	return tenant, nil
}

func writerDeclaration(opts options) ([]string, map[string]string, error) {
	writers := []string{}
	for _, name := range strings.Split(opts.writers, ",") {
		if trimmed := strings.TrimSpace(name); trimmed != "" {
			writers = append(writers, trimmed)
		}
	}
	if len(writers) == 0 {
		return nil, nil, errors.New("no writer service declared")
	}
	nonWriters := map[string]string{}
	for _, entry := range strings.Split(opts.nonWriters, ",") {
		entry = strings.TrimSpace(entry)
		if entry == "" {
			continue
		}
		name, reason, ok := strings.Cut(entry, "=")
		if !ok || strings.TrimSpace(reason) == "" {
			return nil, nil, fmt.Errorf("non-writer %q needs service=reason", entry)
		}
		nonWriters[strings.TrimSpace(name)] = strings.TrimSpace(reason)
	}
	return writers, nonWriters, nil
}

func openFreeze(ctx context.Context, opts options) (*libraryfreeze.Coordinator, *libraryfreeze.ProjectController, error) {
	writers, nonWriters, err := writerDeclaration(opts)
	if err != nil {
		return nil, nil, err
	}
	if err = os.MkdirAll(opts.journalRoot, 0700); err != nil {
		return nil, nil, err
	}
	if err = os.Chmod(opts.journalRoot, 0700); err != nil {
		return nil, nil, err
	}
	control, err := libraryfreeze.NewProjectController(opts.dockerSocket, opts.project, writers, nonWriters, 2*time.Minute)
	if err != nil {
		return nil, nil, err
	}
	coordinator, err := libraryfreeze.Open(opts.journalRoot, opts.project, control)
	if err != nil {
		control.Close()
		return nil, nil, err
	}
	return coordinator, control, nil
}

// captureMaintenanceConfig 把"这次封存动的是哪个实例"落成实例级维护目标。
//
// 为什么必须实例级：被代理的 WeKnora 收到 SIGTERM 会直接 exit(1)，写者的退出码
// 因此永远不能当作"在途请求已排空"。排空证据只能在真正的入口上取——关掉闸门、
// 等它把在途连接放完。没有闸门就没有证据，那种封存不允许以"库级窗口"的名义跑
// （真跑过：被 exit=1 拒掉，这是校验在起作用，不是要放松的地方）。
//
// 入口/队列的声明与容器内 worker 共用同一组 LIBRARY_TARGET_* 名称，
// 数据库/索引/桶则取自本次真实发现的源，避免"关的是 A 的门、读的是 B 的队列"。
func captureMaintenanceConfig(opts options, source sourceConfig, get func(string) string) (librarymaintenance.Config, error) {
	target, err := librarymaintenance.TargetFromEnv(get)
	if err != nil {
		return librarymaintenance.Config{}, err
	}
	writers, nonWriters, err := writerDeclaration(opts)
	if err != nil {
		return librarymaintenance.Config{}, err
	}
	identity, err := librarymaintenance.DatabaseIdentity(source.DatabaseURL)
	if err != nil {
		return librarymaintenance.Config{}, err
	}
	return librarymaintenance.Config{
		Instance:         target.Instance,
		Project:          opts.project,
		Writers:          writers,
		NonWriters:       nonWriters,
		DockerSocket:     opts.dockerSocket,
		IngressURL:       target.IngressURL,
		IngressService:   target.IngressService,
		IngressDrainFile: target.IngressDrainFile,
		Database:         identity,
		Index:            source.SearchIndex,
		Bucket:           source.Object.Bucket,
		// 后台静默检查连的是"正在被封的这个源"：未结清的后台账本与队列都在它身上。
		DatabaseURL:   source.DatabaseURL,
		RedisAddr:     target.RedisAddr,
		RedisPassword: target.RedisPassword,
		RedisDB:       target.RedisDB,
	}, nil
}

// openMaintenance 用实例级目标装配协调器：入口闸门、后台静默检查、目标指纹、
// 入口探针全部来自上面那份声明，与发布链路用的是同一套实现（不是第二份协议）。
func openMaintenance(ctx context.Context, opts options, source sourceConfig) (*libraryfreeze.Coordinator, *libraryfreeze.ProjectController, *librarymaintenance.Manager, error) {
	config, err := captureMaintenanceConfig(opts, source, os.Getenv)
	if err != nil {
		return nil, nil, nil, err
	}
	manager, err := librarymaintenance.New(config)
	if err != nil {
		return nil, nil, nil, err
	}
	// 拓扑先过关：写入者发布了端口 = 存在绕闸直连，这样的"维护"只关了半扇门。
	if err = manager.VerifyTopology(ctx); err != nil {
		manager.Close()
		return nil, nil, nil, fmt.Errorf("目标拓扑不满足唯一入口约束，拒绝开维护窗口：%w", err)
	}
	coordinator, control, err := manager.Open(opts.journalRoot)
	if err != nil {
		manager.Close()
		return nil, nil, nil, err
	}
	return coordinator, control, manager, nil
}

// discoverMaintenanceSource 只读"这个源是什么"，不碰治理库的 scope 绑定：
// 恢复/收尾不该因为治理库读不到就被卡住。
func discoverMaintenanceSource(ctx context.Context, opts options) (sourceConfig, error) {
	return discoverSource(ctx, opts, map[string]string{})
}

func openArchive(opts options) (*librarysnapshot.Archive, error) {
	if err := os.MkdirAll(opts.archiveRoot, 0700); err != nil {
		return nil, err
	}
	if err := os.Chmod(opts.archiveRoot, 0700); err != nil {
		return nil, err
	}
	return librarysnapshot.Open(opts.archiveRoot, 256*1024*1024)
}

// ---------------------------------------------------------------- commands

func runStatus(ctx context.Context, opts options) error {
	report := map[string]any{}
	binding, bindingErr := scopeBindings(ctx, opts.governancePG)
	var config sourceConfig
	var err error
	if bindingErr != nil {
		err = bindingErr
	} else {
		config, err = discoverSource(ctx, opts, binding)
	}
	if err != nil {
		report["source_discovery"] = err.Error()
	} else {
		report["source"] = map[string]any{
			"app_image":          config.AppImageID,
			"scopes":             config.Scopes,
			"search_index":       config.SearchIndex,
			"elasticsearch":      config.Search.BaseURL,
			"object_store":       config.ObjectDetail,
			"database":           "postgres (read-only session, credentials withheld)",
			"credentials_logged": false,
		}
		pool, poolErr := openSourceDatabase(ctx, config)
		if poolErr != nil {
			report["database"] = poolErr.Error()
		} else {
			defer pool.Close()
			settled := "settled"
			if settleErr := v071.LibrarySourceSettled(ctx, pool); settleErr != nil {
				settled = settleErr.Error()
			}
			report["settlement"] = settled
			counts := map[string]int{}
			for _, scope := range []string{"internal", "external"} {
				kb := config.Scopes[scope]
				tenant, tenantErr := tenantAndKnowledgeBase(ctx, pool, kb)
				if tenantErr != nil {
					counts[scope+"_error"] = -1
					continue
				}
				pg, exportErr := v071.ExportLibrary(ctx, pool, v071.PostgresIdentity{TenantID: tenant, KnowledgeBaseID: kb})
				if exportErr != nil {
					counts[scope+"_error"] = -1
					continue
				}
				for _, table := range pg.Tables {
					if table.Name == "chunks" || table.Name == "knowledges" {
						counts[scope+"_"+table.Name] = len(table.Rows)
					}
				}
				_ = kb
			}
			report["postgres_rows"] = counts
		}
	}
	archive, err := openArchive(opts)
	if err != nil {
		report["archive"] = err.Error()
	} else {
		defer archive.Close()
		usage, usageErr := archive.Usage()
		if usageErr != nil {
			report["archive"] = usageErr.Error()
		} else {
			total := int64(0)
			for _, item := range usage {
				total += item.UniqueBytes
			}
			report["archive"] = map[string]any{"root": opts.archiveRoot, "snapshots": len(usage), "stored_bytes": total}
		}
	}
	if coordinator, control, freezeErr := openFreeze(ctx, opts); freezeErr != nil {
		report["freeze"] = freezeErr.Error()
	} else {
		defer coordinator.Close()
		defer control.Close()
		items, inventoryErr := control.Inventory(ctx)
		if inventoryErr != nil {
			report["freeze"] = inventoryErr.Error()
		} else {
			group := []map[string]any{}
			for _, item := range items {
				group = append(group, map[string]any{"id": item.ID, "running": item.Running, "healthy": item.Healthy})
			}
			report["freeze"] = map[string]any{"project": opts.project, "writers": group, "journal": opts.journalRoot}
		}
	}
	printJSON(report)
	return nil
}

func runPreflight(ctx context.Context, opts options) error {
	if opts.scope != "internal" && opts.scope != "external" {
		return errors.New("-scope must be internal or external")
	}
	binding, err := scopeBindings(ctx, opts.governancePG)
	if err != nil {
		return err
	}
	config, err := discoverSource(ctx, opts, binding)
	if err != nil {
		return err
	}
	pool, err := openSourceDatabase(ctx, config)
	if err != nil {
		return err
	}
	defer pool.Close()
	kb := config.Scopes[opts.scope]
	tenant, err := tenantAndKnowledgeBase(ctx, pool, kb)
	if err != nil {
		return err
	}
	if err = v071.LibrarySourceSettled(ctx, pool); err != nil {
		return err
	}
	search, err := v071.NewSearchExporter(config.Search)
	if err != nil {
		return err
	}
	contract, err := compatibilityContract(ctx, pool, search, config.SearchIndex, tenant, kb)
	if err != nil {
		return err
	}
	pg, err := v071.ExportLibrary(ctx, pool, v071.PostgresIdentity{TenantID: tenant, KnowledgeBaseID: kb})
	if err != nil {
		return err
	}
	es, err := search.ExportLibrary(ctx, v071.SearchIdentity{TenantID: tenant, KnowledgeBaseID: kb}, opts.maxDocuments)
	if err != nil {
		return err
	}
	validation := "candidate-valid"
	if err = v071.ValidateLibraryCandidateForPreflight(v071.PostgresIdentity{TenantID: tenant, KnowledgeBaseID: kb}, pg, es); err != nil {
		validation = err.Error()
	}
	tables := map[string]int{}
	for _, table := range pg.Tables {
		tables[table.Name] = len(table.Rows)
	}
	printJSON(map[string]any{"scope": opts.scope, "knowledge_base_id": kb, "tenant_id": tenant,
		"compatibility": contract, "postgres_rows": tables, "index_documents": len(es.Documents),
		"mapping_sha256": es.MappingSHA256, "validation": validation, "writes": 0, "writers_stopped": 0})
	if validation != "candidate-valid" {
		return errors.New("candidate validation failed: " + validation)
	}
	return nil
}

func compatibilityContract(ctx context.Context, pool *pgxpool.Pool, search *v071.SearchExporter, index string, tenant uint64, kb string) (v071.LibrarySnapshotCompatibility, error) {
	inspector, err := v071.NewTargetContractInspector(v071.TargetContractInspectorConfig{DB: pool, Mapping: search,
		KnowledgeBaseID: kb, TenantID: tenant, IndexName: index, ReadTestedBase: true})
	if err != nil {
		return v071.LibrarySnapshotCompatibility{}, err
	}
	contract, err := inspector.Contract(ctx, releaseworkflow.PlanItem{TestKnowledgeBaseID: kb})
	if err != nil {
		return v071.LibrarySnapshotCompatibility{}, err
	}
	return v071.LibrarySnapshotCompatibility{WeKnoraVersion: contract.WeKnoraVersion, SchemaVersion: contract.MigrationVersion,
		RetrieverEngine: contract.RetrieverEngine, EmbeddingModel: contract.EmbeddingModel,
		EmbeddingDimension: contract.EmbeddingDimension, IndexMappingSHA256: contract.ElasticsearchMappingSHA256}, nil
}

func runCapture(ctx context.Context, opts options) error {
	if opts.scope != "internal" && opts.scope != "external" {
		return errors.New("-scope must be internal or external")
	}
	if opts.dryRun {
		return runPreflight(ctx, opts)
	}
	job := opts.job
	if job == "" {
		job = "lib-" + opts.scope + "-" + time.Now().UTC().Format("20060102T150405")
	}
	archive, err := openArchive(opts)
	if err != nil {
		return err
	}
	defer archive.Close()
	binding, err := scopeBindings(ctx, opts.governancePG)
	if err != nil {
		return err
	}
	config, err := discoverSource(ctx, opts, binding)
	if err != nil {
		return err
	}
	// 维护窗口必须先按"实例级"接好：入口闸门给出在途排空证据，后台静默检查
	// 给出停写者那一刻队列已空。缺任何一项都拒绝封存——没有证据的封存不是候选。
	coordinator, control, manager, err := openMaintenance(ctx, opts, config)
	if err != nil {
		return err
	}
	defer manager.Close()
	defer coordinator.Close()
	defer control.Close()
	if err = coordinator.Recover(ctx); err != nil {
		return fmt.Errorf("pending freeze requires recovery: %w", err)
	}
	// Replay is answered from the persisted journal + sealed archive: the same job
	// re-used with the same scope/binding returns the original candidate without
	// reading the source or stopping a writer, and a job re-used with different
	// parameters is refused instead of silently sealing under a stale identity.
	existing, found, err := coordinator.Inspect(job)
	if err != nil {
		return err
	}
	scopedKB := binding[opts.scope]
	if found {
		manifest, err := replayCapture(job, opts.scope, scopedKB, existing, archive)
		if err != nil {
			return err
		}
		return printCapture(job, existing.SnapshotID, manifest, true)
	}
	pool, err := openSourceDatabase(ctx, config)
	if err != nil {
		return err
	}
	defer pool.Close()
	kb := config.Scopes[opts.scope]
	tenant, err := tenantAndKnowledgeBase(ctx, pool, kb)
	if err != nil {
		return err
	}
	search, err := v071.NewSearchExporter(config.Search)
	if err != nil {
		return err
	}
	objects, err := v071.NewS3ObjectStore(config.Object)
	if err != nil {
		return err
	}
	contract, err := compatibilityContract(ctx, pool, search, config.SearchIndex, tenant, kb)
	if err != nil {
		return err
	}
	// The contract above is read before the maintenance window. Re-read it inside
	// the stopped window and refuse if a model/index change happened in between:
	// sealing a candidate whose contract no longer matches the source would hide
	// exactly the change a later production sync cannot reproduce.
	verifyContract := func(ctx context.Context) error {
		live, err := compatibilityContract(ctx, pool, search, config.SearchIndex, tenant, kb)
		if err != nil {
			return err
		}
		if live != contract {
			return errors.New("source contract changed between preflight and the frozen window; capture aborted")
		}
		return nil
	}
	// sourceReady proves the declared writer group is complete and no other
	// running container uses a declared writer image. Recorded work settlement is
	// enforced separately by CaptureLibrary; neither check can see unrecorded or
	// remote work, so this is not a cross-system consistency proof.
	sourceReady := func(ctx context.Context) error {
		items, err := control.Inventory(ctx)
		if err != nil {
			return err
		}
		if len(items) == 0 {
			return errors.New("writer group is empty")
		}
		declared := make(map[string]bool, len(items))
		for _, item := range items {
			declared[item.ID] = true
		}
		images, err := writerImageIDs(ctx, items)
		if err != nil {
			return err
		}
		app, err := containerEnv(ctx, opts.sourceApp)
		if err != nil {
			return err
		}
		return checkNoUndeclaredWriter(ctx, images, declared, opts.project, sourceTargetOf(app))
	}
	result, err := v071.CaptureLibrary(ctx, coordinator, archive, pool, search, objects, sourceReady,
		v071.LibraryCaptureRequest{JobID: job, Scope: opts.scope,
			Identity:      v071.PostgresIdentity{TenantID: tenant, KnowledgeBaseID: kb},
			Compatibility: contract, CreatedAt: time.Now().UTC(), MaxDocuments: opts.maxDocuments,
			VerifyContract: verifyContract})
	if err != nil {
		printJSON(map[string]any{"job": job, "state": result.State, "snapshot_id": result.SnapshotID, "error": err.Error()})
		return err
	}
	manifest, err := archive.Load(result.SnapshotID)
	if err != nil {
		return err
	}
	return printCapture(job, result.SnapshotID, manifest, false)
}

// replayCapture answers a repeated job from persisted state. It never reads the
// source and never stops a writer; conflicting parameters are refused.
func replayCapture(job, scope, expectedKB string, result libraryfreeze.Result, archive *librarysnapshot.Archive) (librarysnapshot.Manifest, error) {
	if result.State != "succeeded" || result.SnapshotID == "" {
		return librarysnapshot.Manifest{}, fmt.Errorf("job %q already exists with state %q: recover it or use a new job id", job, result.State)
	}
	manifest, err := archive.Load(result.SnapshotID)
	if err != nil {
		return librarysnapshot.Manifest{}, errors.New("captured library archive no longer verifies")
	}
	if manifest.ValidationPolicy != librarysnapshot.ArtifactAcceptancePolicy {
		return librarysnapshot.Manifest{}, errors.New("existing job was sealed under a different validation policy")
	}
	if manifest.Scope != scope {
		return librarysnapshot.Manifest{}, fmt.Errorf("job %q was captured for scope %q, not %q", job, manifest.Scope, scope)
	}
	if manifest.KnowledgeBaseID != expectedKB {
		return librarysnapshot.Manifest{}, fmt.Errorf("job %q was captured for knowledge base %q, which is no longer the %q binding", job, manifest.KnowledgeBaseID, scope)
	}
	return manifest, nil
}

func printCapture(job, snapshotID string, manifest librarysnapshot.Manifest, replay bool) error {
	parts := make([]map[string]any, 0, len(manifest.Parts))
	var total int64
	for _, part := range manifest.Parts {
		parts = append(parts, map[string]any{"kind": part.Kind, "name": part.Name, "count": part.Count, "bytes": part.Bytes, "sha256": part.SHA256})
		total += part.Bytes
	}
	sort.Slice(parts, func(i, j int) bool {
		return fmt.Sprint(parts[i]["kind"])+fmt.Sprint(parts[i]["name"]) < fmt.Sprint(parts[j]["kind"])+fmt.Sprint(parts[j]["name"])
	})
	printJSON(map[string]any{"job": job, "state": "succeeded", "snapshot_id": snapshotID,
		"scope": manifest.Scope, "knowledge_base_id": manifest.KnowledgeBaseID, "tenant_id": manifest.TenantID,
		"compatibility": manifest.Compatibility, "parts": parts, "logical_bytes": total,
		"validation_policy": manifest.ValidationPolicy, "writers_restored": true,
		"replay": replay, "source_read": !replay, "writers_stopped": !replay})
	return nil
}

func writerImageIDs(ctx context.Context, items []libraryfreeze.Service) ([]string, error) {
	images := make([]string, 0, len(items))
	seen := map[string]bool{}
	for _, item := range items {
		image, err := containerImageOf(ctx, item.ID)
		if err != nil {
			return nil, err
		}
		if !seen[image] {
			seen[image] = true
			images = append(images, image)
		}
	}
	return images, nil
}

// checkNoUndeclaredWriter refuses to run when a running container uses a declared
// writer image but is not part of the declared group: such a container could keep
// writing during the maintenance window.
func checkNoUndeclaredWriter(ctx context.Context, images []string, declared map[string]bool,
	sourceProject string, source map[string]string) error {
	for _, image := range images {
		// 必须 --no-trunc：`docker ps` 默认只给 12 位短 ID，和 declared 里的
		// 64 位容器 ID 永远不相等——那样连"已声明的写入者"都会被判成未声明。
		out, err := dockerOutput(ctx, "ps", "-aq", "--no-trunc", "--filter", "ancestor="+image)
		if err != nil {
			return err
		}
		for _, id := range strings.Fields(out) {
			if declaredWriter(declared, id) {
				continue
			}
			project, err := dockerOutput(ctx, "inspect", "--format", `{{index .Config.Labels "com.docker.compose.project"}}`, id)
			if err != nil {
				return err
			}
			env, err := containerEnv(ctx, id)
			if err != nil {
				return err
			}
			if conflict, reason := undeclaredWriterConflict(false, project, sourceProject, env, source); conflict {
				return fmt.Errorf("container %s uses a declared writer image but is not in the declared writer group: %s", id[:12], reason)
			}
		}
	}
	return nil
}

// declaredWriter 同时接受完整 ID 与短 ID：证据里两种写法都出现过，
// 只比一种会把同一容器判成两个身份。
func declaredWriter(declared map[string]bool, id string) bool {
	if declared[id] {
		return true
	}
	for key := range declared {
		if strings.HasPrefix(key, id) || strings.HasPrefix(id, key) {
			return true
		}
	}
	return false
}

// sourceTargetOf 只取"写到哪儿"的身份：同一个数据库或同一个队列才是真冲突。
func sourceTargetOf(env map[string]string) map[string]string {
	return map[string]string{
		"DB_HOST": env["DB_HOST"], "DB_NAME": env["DB_NAME"], "DB_PORT": env["DB_PORT"],
		"REDIS_ADDR": env["REDIS_ADDR"], "REDIS_NAMESPACE": env["REDIS_NAMESPACE"],
	}
}

// undeclaredWriterConflict 判断一个同镜像容器是否构成"未声明的并发写入者"。
//
// 同一 compose 项目或没有项目标签 ⇒ 无法证明隔离，直接冲突。
// 另一个项目 ⇒ 只有数据目标（数据库/队列）与源相同才算冲突：
// 镜像相同不等于写同一个库，本机共享业务栈与专用验收实例就是这种关系。
func undeclaredWriterConflict(declared bool, project, sourceProject string, env, source map[string]string) (bool, string) {
	if declared {
		return false, ""
	}
	if project == "" {
		return true, "没有 compose 项目标签，无法证明它不写同一个目标"
	}
	if project == sourceProject {
		return true, "属于同一 compose 项目，却不是声明的写入服务"
	}
	if source["DB_HOST"] != "" && env["DB_HOST"] == source["DB_HOST"] && env["DB_NAME"] == source["DB_NAME"] {
		return true, "另一个项目的同镜像容器指向同一个源数据库"
	}
	if source["REDIS_ADDR"] != "" && env["REDIS_ADDR"] == source["REDIS_ADDR"] &&
		env["REDIS_NAMESPACE"] == source["REDIS_NAMESPACE"] {
		return true, "另一个项目的同镜像容器共用同一个源队列"
	}
	return false, "另一套隔离栈（数据目标不同）"
}

func containerImageOf(ctx context.Context, id string) (string, error) {
	image, err := dockerOutput(ctx, "inspect", "--format", "{{.Image}}", id)
	if err != nil {
		return "", err
	}
	if image == "" {
		return "", errors.New("declared writer has no image identity")
	}
	return image, nil
}
func runList(opts options) error {
	archive, err := openArchive(opts)
	if err != nil {
		return err
	}
	defer archive.Close()
	usage, err := archive.Usage()
	if err != nil {
		return err
	}
	total, err := archive.ContentUsage()
	if err != nil {
		return err
	}
	items := make([]map[string]any, 0, len(usage))
	for _, item := range usage {
		manifest, err := archive.Load(item.VersionID)
		if err != nil {
			return err
		}
		items = append(items, map[string]any{"snapshot_id": item.VersionID, "scope": manifest.Scope,
			"knowledge_base_id": manifest.KnowledgeBaseID, "created_at": manifest.CreatedAt,
			"logical_bytes": item.LogicalBytes, "unique_bytes": item.UniqueBytes, "shared_bytes": item.SharedBytes,
			"parts": len(manifest.Parts), "policy": manifest.ValidationPolicy})
	}
	printJSON(map[string]any{"archive": opts.archiveRoot, "snapshots": items,
		// 跨版本去重后的内容字节；文件系统事实单列，任何一项都不是可回收空间。
		"versions": total.Versions, "content_bytes": total.ContentBytes,
		"blob_file_bytes": total.BlobFileBytes, "blob_allocated_bytes": total.BlobAllocatedBytes,
		"unreferenced_blobs": total.UnreferencedBlobs, "unreferenced_bytes": total.UnreferencedBytes,
		"manifest_file_bytes": total.ManifestFileBytes, "manifest_allocated_bytes": total.ManifestAllocatedBytes})
	return nil
}

func runVerify(opts options) error {
	if opts.snapshot == "" {
		return errors.New("-snapshot is required")
	}
	archive, err := openArchive(opts)
	if err != nil {
		return err
	}
	defer archive.Close()
	manifest, err := archive.Load(opts.snapshot)
	if err != nil {
		return err
	}
	parts := make([]map[string]any, 0, len(manifest.Parts))
	for _, part := range manifest.Parts {
		parts = append(parts, map[string]any{"kind": part.Kind, "name": part.Name, "count": part.Count, "bytes": part.Bytes, "sha256": part.SHA256})
	}
	printJSON(map[string]any{"snapshot_id": opts.snapshot, "verified": true, "format": manifest.Format,
		"policy": manifest.ValidationPolicy, "scope": manifest.Scope, "knowledge_base_id": manifest.KnowledgeBaseID,
		"tenant_id": manifest.TenantID, "created_at": manifest.CreatedAt, "compatibility": manifest.Compatibility, "parts": parts})
	return nil
}

func runRecover(ctx context.Context, opts options) error {
	// 收尾也必须走同一份实例级目标：崩溃留下的窗口是"入口已关、写入者已停"，
	// 恢复要用闸门把入口真的重新打开，没有闸门就没有"目标已恢复"的证据。
	source, err := discoverMaintenanceSource(ctx, opts)
	if err != nil {
		return err
	}
	coordinator, control, manager, err := openMaintenance(ctx, opts, source)
	if err != nil {
		return err
	}
	defer manager.Close()
	defer coordinator.Close()
	defer control.Close()
	if err = coordinator.Recover(ctx); err != nil {
		return err
	}
	items, err := control.Inventory(ctx)
	if err != nil {
		return err
	}
	group := []map[string]any{}
	for _, item := range items {
		group = append(group, map[string]any{"id": item.ID, "running": item.Running, "healthy": item.Healthy})
	}
	printJSON(map[string]any{"recovered": true, "writers": group})
	return nil
}

func printJSON(value any) {
	encoder := json.NewEncoder(os.Stdout)
	encoder.SetIndent("", "  ")
	_ = encoder.Encode(value)
}