feat(library): 封存执行器可用性与还原条件来自真实观测

页面上的"可发起封存"原来只判 LIBRARY_AGENT_SOCKET 非空：正式栈的 API 根本没挂
那个 socket，而 agent 曾顶着正式单元名冻结候选实例——配置非空、页面可点、真发起
却封错库。

现在：agent 新增 health 动作（不占封存操作锁、不跑子进程，只回答身份/降级/忙），
worker 每 30 秒探测并写入迁移 074 的 library_executor_status；overview、快照列表与
POST /library/captures 共用同一份判定（过期、不可达、实例不符、降级、正忙都拒绝并
给原因）。封存结果由 agent 附上自报身份，worker 在停写入者之前再探测一次、执行后
用结果复核，不一致就不登记测试封存——心跳只能证明"刚才"。

期望身份用 LIBRARY_SOURCE_*（封存源），不是 LIBRARY_TARGET_*（同步目标），否则
正确的执行器会被判成不匹配。另加 GET /library/restore-availability：恢复能否执行由
双闸、该快照的恢复授权与实例维护状态决定，页面不再写死 disabled。

真跑（候选隔离栈，两处 socket 隔离不变）：overview 返回 executor
{kgmaint-weknora/kgmaint-app}；停 agent 后转不可达并 503 带原因；重启后恢复。

api/openapi.yaml

@@ -1312,6 +1312,51 @@
     }
    }
   },
+  "/api/v1/library/restore-availability": {
+   "get": {
+    "tags": [
+     "Runtime"
+    ],
+    "operationId": "getLibraryRestoreAvailability",
+    "summary": "这份快照现在能不能恢复到生产：双闸、恢复授权与目标实例维护状态逐条说明",
+    "parameters": [
+     {
+      "name": "scope",
+      "in": "query",
+      "required": false,
+      "schema": {
+       "type": "string",
+       "enum": [
+        "internal",
+        "external"
+       ]
+      }
+     },
+     {
+      "name": "snapshot_id",
+      "in": "query",
+      "required": true,
+      "schema": {
+       "type": "string"
+      }
+     }
+    ],
+    "responses": {
+     "200": {
+      "$ref": "#/components/responses/Envelope"
+     },
+     "400": {
+      "$ref": "#/components/responses/BadRequest"
+     },
+     "500": {
+      "$ref": "#/components/responses/InternalError"
+     },
+     "503": {
+      "description": "读取恢复授权失败（不是\"可恢复\"，是\"无法判定\"）"
+     }
+    }
+   }
+  },
   "/api/v1/library/authorizations": {
    "get": {
     "tags": [

cmd/api/main.go

@@ -204,7 +204,9 @@ func run(ctx context.Context, getenv func(string) string) error {
 		LibraryArchiveRoot:           config.LibraryArchiveRoot,
 		LibraryTargetInstance:        config.LibraryTargetInstance,
 		LibrarySourceInstance:        config.SourceInstance(),
+		LibrarySourceProject:         config.SourceProject(),
 		LibraryAgentSocket:           config.LibraryAgentSocket,
+		LibraryExecutorStatusMaxAge:  config.LibraryExecutorStatusMaxAge,
 		LibraryCapacityBudgetBytes:   config.LibraryCapacityBudgetBytes,
 		LibraryProductionDatabaseURL: config.LibraryProductionDatabaseURL,
 		LibraryProductionInternalKB:  config.LibraryProductionInternalKB,

cmd/worker/library_capture_identity.go

@@ -0,0 +1,89 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"knowledge-governance/internal/libraryagent"
+)
+
+// 封存执行的身份核对分两次：执行前一次、执行后一次。
+//
+// 为什么心跳不够：心跳说明"上一次探测时"那个 socket 后面是哪个实例，而从心跳到
+// 真正停写入者之间，socket 可能被换成另一个 agent（换挂载、换单元、误启正式单元）。
+// 执行前探测把窗口缩到最小；执行后复核 agent 自报的身份，则把"换过 socket"这件事
+// 变成一次显式的作业失败，而不是悄悄封错了库还登记成测试封存。
+
+// captureExecutorPrecheck 在动源写入者之前确认执行器就是期望的那个实例。
+func captureExecutorPrecheck(ctx context.Context, client libraryagent.ClientConfig, expectedProject string, probeTimeout time.Duration) error {
+	if strings.TrimSpace(expectedProject) == "" {
+		return errors.New("未配置封存来源实例身份（LIBRARY_SOURCE_PROJECT），拒绝在无法核对执行器的情况下停写入者")
+	}
+	if probeTimeout <= 0 {
+		probeTimeout = 3 * time.Second
+	}
+	bounded, cancel := context.WithTimeout(ctx, probeTimeout)
+	defer cancel()
+	result, err := libraryagent.Health(bounded, client)
+	if err != nil {
+		return fmt.Errorf("封存执行器不可达，拒绝停写入者：%w", err)
+	}
+	identity, err := agentIdentityOf(result.Payload)
+	if err != nil {
+		return err
+	}
+	if identity.Project != expectedProject {
+		return fmt.Errorf("封存执行器指向的实例不匹配（agent=%q，期望=%q），拒绝停写入者",
+			identity.Project, expectedProject)
+	}
+	if identity.Degraded != "" {
+		return fmt.Errorf("封存执行器处于降级状态（%s），拒绝停写入者", identity.Degraded)
+	}
+	return nil
+}
+
+// captureExecutorPostcheck 用封存结果里 agent 自报的身份复核：不一致就当作
+// "封存可能来自别的实例"，调用方不得把它登记成测试封存。
+func captureExecutorPostcheck(payload json.RawMessage, expectedProject string) error {
+	identity, err := agentIdentityOf(payload)
+	if err != nil {
+		return err
+	}
+	if identity.Project == "" {
+		return errors.New("封存结果里没有执行器身份，无法证明这次封存来自期望实例")
+	}
+	if identity.Project != expectedProject {
+		return fmt.Errorf("封存结果来自另一个实例（agent=%q，期望=%q），该快照不能当作本实例的测试封存",
+			identity.Project, expectedProject)
+	}
+	return nil
+}
+
+// agentIdentity 是 agent 在 health/封存结果里自报的身份（无凭据）。
+type agentIdentity struct {
+	Project   string `json:"project"`
+	SourceApp string `json:"source_app"`
+	Degraded  string `json:"degraded"`
+}
+
+func agentIdentityOf(payload json.RawMessage) (agentIdentity, error) {
+	if len(payload) == 0 {
+		return agentIdentity{}, errors.New("执行器应答没有身份信息")
+	}
+	var document struct {
+		Identity agentIdentity `json:"agent_identity"`
+		agentIdentity
+	}
+	if err := json.Unmarshal(payload, &document); err != nil {
+		return agentIdentity{}, errors.New("执行器身份无法解析")
+	}
+	// health 把身份摊平在顶层；封存结果把它放在 agent_identity 下。
+	if document.Identity.Project != "" {
+		return document.Identity, nil
+	}
+	return document.agentIdentity, nil
+}

cmd/worker/library_capture_identity_test.go

@@ -0,0 +1,98 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"net"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"knowledge-governance/internal/libraryagent"
+)
+
+func shortDir(t *testing.T) string {
+	t.Helper()
+	// unix socket 路径有长度上限，测试目录要短。
+	dir, err := os.MkdirTemp("/tmp", "kgcap-")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { os.RemoveAll(dir) })
+	return dir
+}
+
+func fakeAgent(t *testing.T, payload map[string]any) string {
+	t.Helper()
+	socket := filepath.Join(shortDir(t), "agent.sock")
+	listener, err := net.Listen("unix", socket)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { listener.Close() })
+	go func() {
+		for {
+			conn, acceptErr := listener.Accept()
+			if acceptErr != nil {
+				return
+			}
+			var request map[string]any
+			_ = json.NewDecoder(conn).Decode(&request)
+			body, _ := json.Marshal(payload)
+			_ = json.NewEncoder(conn).Encode(map[string]any{"ok": true, "action": "health", "payload": json.RawMessage(body)})
+			conn.Close()
+		}
+	}()
+	return socket
+}
+
+// 心跳说"可用"不等于执行时还是那个实例：执行前的探测必须能挡住换过的 socket。
+func TestCaptureExecutorPrecheckRejectsAWrongInstance(t *testing.T) {
+	socket := fakeAgent(t, map[string]any{"project": "weknora071full", "source_app": "WKNFull-app"})
+	client := libraryagent.ClientConfig{Socket: socket, Timeout: 2 * time.Second}
+	err := captureExecutorPrecheck(context.Background(), client, "kgmaint-weknora", 2*time.Second)
+	if err == nil || !strings.Contains(err.Error(), "指向的实例不匹配") {
+		t.Fatalf("err = %v", err)
+	}
+	if !strings.Contains(err.Error(), "weknora071full") {
+		t.Fatalf("err should name the actual agent project: %v", err)
+	}
+}
+
+func TestCaptureExecutorPrecheckAcceptsTheExpectedInstance(t *testing.T) {
+	socket := fakeAgent(t, map[string]any{"project": "kgmaint-weknora", "source_app": "kgmaint-app"})
+	client := libraryagent.ClientConfig{Socket: socket, Timeout: 2 * time.Second}
+	if err := captureExecutorPrecheck(context.Background(), client, "kgmaint-weknora", 2*time.Second); err != nil {
+		t.Fatal(err)
+	}
+	// 没有期望身份就不许停写入者：那不是"跳过检查"，是不能动生产。
+	if err := captureExecutorPrecheck(context.Background(), client, "", 2*time.Second); err == nil ||
+		!strings.Contains(err.Error(), "LIBRARY_SOURCE_PROJECT") {
+		t.Fatalf("empty expectation accepted: %v", err)
+	}
+	// 降级同样挡住。
+	degraded := fakeAgent(t, map[string]any{"project": "kgmaint-weknora", "degraded": "写入者恢复失败"})
+	if err := captureExecutorPrecheck(context.Background(), libraryagent.ClientConfig{Socket: degraded, Timeout: 2 * time.Second},
+		"kgmaint-weknora", 2*time.Second); err == nil || !strings.Contains(err.Error(), "降级") {
+		t.Fatalf("degraded executor accepted: %v", err)
+	}
+}
+
+// 执行后复核：结果里自报的身份不符，这次封存就不能被登记成测试封存。
+func TestCaptureExecutorPostcheckUsesTheIdentityInTheResult(t *testing.T) {
+	good := json.RawMessage(`{"snapshot_id":"abc","agent_identity":{"project":"kgmaint-weknora"}}`)
+	if err := captureExecutorPostcheck(good, "kgmaint-weknora"); err != nil {
+		t.Fatalf("matching identity rejected: %v", err)
+	}
+	swapped := json.RawMessage(`{"snapshot_id":"abc","agent_identity":{"project":"weknora071full"}}`)
+	err := captureExecutorPostcheck(swapped, "kgmaint-weknora")
+	if err == nil || !strings.Contains(err.Error(), "另一个实例") {
+		t.Fatalf("err = %v", err)
+	}
+	if err := captureExecutorPostcheck(json.RawMessage(`{"snapshot_id":"abc"}`), "kgmaint-weknora"); err == nil ||
+		!strings.Contains(err.Error(), "没有执行器身份") {
+		t.Fatalf("missing identity accepted: %v", err)
+	}
+}

cmd/worker/library_executor_status.go

@@ -0,0 +1,52 @@
+package main
+
+import (
+	"context"
+	"log"
+	"time"
+
+	"knowledge-governance/internal/appconfig"
+	"knowledge-governance/internal/libraryexecutor"
+	postgresstore "knowledge-governance/internal/platform/postgres"
+)
+
+// 为什么由 worker 观测、页面读库：agent socket 只挂载在 worker 容器里，API 那个
+// LIBRARY_AGENT_SOCKET 即使配了也 dial 不到。谁真能执行就由谁写下观测，
+// 否则页面上的"可发起"只是配置字符串的回声。
+const libraryExecutorProbeInterval = 30 * time.Second
+
+// libraryExecutorProbeTimeout 是一次 health 往返的上限：它只是同 uid unix socket
+// 上的一次 JSON 往返，慢到秒级就说明执行器有问题，不该拖住心跳。
+const libraryExecutorProbeTimeout = 3 * time.Second
+
+func recordLibraryExecutorStatus(ctx context.Context, store *postgresstore.Store, config appconfig.Config) {
+	// 归属用**封存源**实例：agent 冻结的是源库写入者，和同步目标（kgseal）不是一套。
+	instance := config.SourceInstance()
+	if instance == "" {
+		// 没有实例身份，观测就无处归属：写进去只会让页面拿别的实例的状态。
+		log.Print("library executor probe skipped: LIBRARY_SOURCE_INSTANCE is not configured")
+		return
+	}
+	observation := libraryexecutor.Probe(ctx, instance, config.LibraryAgentSocket, libraryExecutorProbeTimeout, nil)
+	if err := store.UpsertLibraryExecutorStatus(ctx, observation, config.WorkerID, time.Now().UTC()); err != nil {
+		log.Printf("library executor status not recorded: %v", err)
+	}
+}
+
+// runLibraryExecutorProbe 先观测一次再进循环：页面在 worker 刚起来时就该有真实状态。
+func runLibraryExecutorProbe(ctx context.Context, store *postgresstore.Store, config appconfig.Config, interval time.Duration) {
+	if interval <= 0 {
+		interval = libraryExecutorProbeInterval
+	}
+	recordLibraryExecutorStatus(ctx, store, config)
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			recordLibraryExecutorStatus(ctx, store, config)
+		}
+	}
+}

cmd/worker/main.go

@@ -8,7 +8,6 @@ import (
 	"os"
 	"os/signal"
 	"path/filepath"
-	"strings"
 	"syscall"
 	"time"
 
@@ -58,21 +57,6 @@ func (r libraryStageRecorder) Stage(ctx context.Context, stage, state, detail st
 
 func int64Ptr(value int64) *int64 { return &value }
 
-// redactDSN 只保留库名，避免把连接串写进维护记录。
-func redactDSN(dsn string) string {
-	if dsn == "" {
-		return ""
-	}
-	if index := strings.LastIndex(dsn, "/"); index >= 0 && index+1 < len(dsn) {
-		name := dsn[index+1:]
-		if cut := strings.Index(name, "?"); cut >= 0 {
-			name = name[:cut]
-		}
-		return name
-	}
-	return "configured"
-}
-
 // stageSize 是某一类数据在快照里的真实总量（行数/字节）。
 type stageSize struct {
 	count int64
@@ -205,6 +189,8 @@ func run(ctx context.Context, getenv func(string) string) error {
 		go runRecordsSyncScheduler(ctx, mirrorStore, func(error) { log.Print("QA local mirror scheduler unavailable; last committed data retained") })
 	}
 	go weeklyMailLoop(ctx, pool, getenv)
+	// 封存执行器状态：worker 是唯一能连 agent socket 的一方，它写观测、页面读观测。
+	go runLibraryExecutorProbe(ctx, store, config, libraryExecutorProbeInterval)
 	go runSourceRefreshScheduler(ctx, store, config.SourceRefreshSchedulerInterval, func(err error) {
 		log.Printf("worker %s source refresh scheduler error: %v", config.WorkerID, err)
 	})
@@ -284,6 +270,13 @@ func workerHandlers(store *postgresstore.Store, config appconfig.Config) (map[st
 				_ = store.AdvanceLibraryJob(ctx, libraryJobID, governance.LibraryJobFailed, "export", "dead letter release storage unavailable", "", "", "", time.Now().UTC())
 				return fmt.Errorf("load dead letter releases: %w", releaseErr)
 			}
+			// 心跳只能证明"刚才"是哪个实例：动写入者之前再核一次身份。
+			if identityErr := captureExecutorPrecheck(ctx, agentConfig, config.SourceProject(), 5*time.Second); identityErr != nil {
+				_ = store.UpdateLibraryStage(ctx, libraryJobID, "export", "failed", nil, nil, nil, nil, identityErr.Error(), time.Now().UTC())
+				_ = store.AdvanceLibraryJob(ctx, libraryJobID, governance.LibraryJobFailed, "export", identityErr.Error(),
+					"未停任何写入者；修正执行器身份后重试", "", "", time.Now().UTC())
+				return identityErr
+			}
 			result, err := libraryagent.CaptureReleased(ctx, agentConfig, payload.Scope, payload.CaptureJob, released)
 			finishedAt := time.Now().UTC()
 			if err != nil {
@@ -294,6 +287,14 @@ func workerHandlers(store *postgresstore.Store, config appconfig.Config) (map[st
 				log.Printf("library capture failed scope=%s capture_job=%s: %v", payload.Scope, payload.CaptureJob, err)
 				return err
 			}
+			// 执行后复核：结果里 agent 自报的身份必须还是期望实例，否则这次封存
+			// 可能来自别的栈，绝不能登记成"本实例的测试封存"。
+			if identityErr := captureExecutorPostcheck(result.Payload, config.SourceProject()); identityErr != nil {
+				_ = store.UpdateLibraryStage(ctx, libraryJobID, "export", "failed", nil, nil, nil, nil, identityErr.Error(), finishedAt)
+				_ = store.AdvanceLibraryJob(ctx, libraryJobID, governance.LibraryJobFailed, "export", identityErr.Error(),
+					"快照已产生但来源身份不符，未登记为测试封存；核对执行器后重新封存", "", "", finishedAt)
+				return identityErr
+			}
 			// 封存产物成版本记录：只登记"测试封存"，绝不写成生产生效版本。
 			versionErr := store.RecordLibraryVersion(ctx, governance.LibraryVersion{
 				ID:    governance.LibraryVersionID(governance.LibraryVersionTestSealed, payload.Scope, result.SnapshotID),

internal/appconfig/config.go

@@ -99,7 +99,10 @@ type Config struct {
 	LibraryTargetInstance string
 	// LibrarySourceInstance 是死信解除记录绑定的"来源实例"名字（封存源，不是同步目标）。
 	// 留空时回落到 LibraryTargetInstance，保持单实例部署的旧行为。
-	LibrarySourceInstance     string
+	LibrarySourceInstance string
+	// LibrarySourceProject 是封存执行器应该指向的实例名（agent -project）。
+	// 留空时等于来源实例名：环境里两者本来就同名（kgmaint-weknora）。
+	LibrarySourceProject      string
 	LibraryTargetProject      string
 	LibraryTargetWriters      []string
 	LibraryTargetNonWriters   map[string]string
@@ -115,10 +118,13 @@ type Config struct {
 	LibraryTargetJournalRoot   string
 	// LibraryAgentSocket 是主机侧封存 agent 的 unix socket（只由 worker 连接）。
 	// 留空即"未接线"：页面不显示可用的封存入口，接口返回原因而不是假装已排队。
-	LibraryAgentSocket        string
-	VersionHistoryLinkOrigin  string
-	VersionHistoryHTTPTimeout time.Duration
-	VersionHistoryMaxPages    int
+	LibraryAgentSocket string
+	// LibraryExecutorStatusMaxAge 是执行器状态的有效期：超过它就是"无法确认"，
+	// 不是"仍然可用"。worker 每 30 秒观测一次，默认 120 秒足够容错。
+	LibraryExecutorStatusMaxAge time.Duration
+	VersionHistoryLinkOrigin    string
+	VersionHistoryHTTPTimeout   time.Duration
+	VersionHistoryMaxPages      int
 }
 
 func Load(getenv func(string) string) (Config, error) {
@@ -203,6 +209,7 @@ func Load(getenv func(string) string) (Config, error) {
 		LibraryProductionS3SecretKey:  strings.TrimSpace(getenv("LIBRARY_PRODUCTION_S3_SECRET_KEY")),
 		LibraryTargetInstance:         strings.TrimSpace(getenv("LIBRARY_TARGET_INSTANCE")),
 		LibrarySourceInstance:         strings.TrimSpace(getenv("LIBRARY_SOURCE_INSTANCE")),
+		LibrarySourceProject:          strings.TrimSpace(getenv("LIBRARY_SOURCE_PROJECT")),
 		LibraryTargetProject:          strings.TrimSpace(getenv("LIBRARY_TARGET_PROJECT")),
 		LibraryTargetWriters:          splitList(getenv("LIBRARY_TARGET_WRITERS")),
 		LibraryTargetNonWriters:       splitDeclarations(getenv("LIBRARY_TARGET_NON_WRITERS")),
@@ -219,6 +226,7 @@ func Load(getenv func(string) string) (Config, error) {
 		LibraryCapacityBudgetBytes:    capacityBudget,
 		LibraryRetentionVersions:      retentionVersions,
 		LibraryRetentionDays:          retentionDays,
+		LibraryExecutorStatusMaxAge:   2 * time.Minute,
 		VersionHistoryLinkOrigin:      strings.TrimSpace(getenv("VERSION_HISTORY_LINK_ORIGIN")),
 		VersionHistoryHTTPTimeout:     15 * time.Second,
 		VersionHistoryMaxPages:        valueOrDefaultInt(getenv("VERSION_HISTORY_MAX_PAGES"), 200),
@@ -241,6 +249,9 @@ func Load(getenv func(string) string) (Config, error) {
 	if config.SourceRefreshSchedulerInterval < 5*time.Second {
 		return Config{}, errors.New("SOURCE_REFRESH_SCHEDULER_INTERVAL must be at least 5s")
 	}
+	if config.LibraryExecutorStatusMaxAge, err = durationOrDefault(getenv("LIBRARY_EXECUTOR_STATUS_MAX_AGE"), config.LibraryExecutorStatusMaxAge); err != nil {
+		return Config{}, fmt.Errorf("LIBRARY_EXECUTOR_STATUS_MAX_AGE: %w", err)
+	}
 	if raw := getenv("RELEASE_WORKER_ENABLED"); raw != "" {
 		config.ReleaseWorkerEnabled, err = strconv.ParseBool(raw)
 		if err != nil {
@@ -575,3 +586,13 @@ func (config Config) SourceInstance() string {
 	}
 	return config.LibraryTargetInstance
 }
+
+// SourceProject 是封存执行器自报 project 的期望值。它描述的是**封存源**
+// （agent 冻结哪一套写入者），不是同步目标：拿目标实例去比会把正确的执行器判成
+// 不匹配（本机候选栈就是这种形状：源 kgmaint、目标 kgseal）。
+func (config Config) SourceProject() string {
+	if config.LibrarySourceProject != "" {
+		return config.LibrarySourceProject
+	}
+	return config.SourceInstance()
+}

internal/httpapi/library_capture.go

@@ -26,13 +26,6 @@ type libraryCaptureRequest struct {
 }
 
 func libraryCaptureEnqueue(w http.ResponseWriter, r *http.Request, config Config) {
-	if config.LibraryAgentSocket == "" {
-		writeJSON(w, http.StatusServiceUnavailable, map[string]any{
-			"code":   "library_capture_unwired",
-			"reason": "未接线：主机侧封存 agent 没有挂载，页面无法发起封存",
-		})
-		return
-	}
 	var request libraryCaptureRequest
 	if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 4096)).Decode(&request); err != nil {
 		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid request body"})
@@ -52,6 +45,17 @@ func libraryCaptureEnqueue(w http.ResponseWriter, r *http.Request, config Config
 		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid job id"})
 		return
 	}
+	// 请求形状先校验（400），再看执行器（503）：不只看"配了 socket"——执行器可能
+	// 不可达、可能指向别的实例、可能降级。页面与接口必须用同一份判定，否则又会出现
+	// "点了才失败"或"封错库"。
+	if available, reason, executor := libraryExecutorAvailability(r.Context(), config); !available {
+		writeJSON(w, http.StatusServiceUnavailable, map[string]any{
+			"code":     "library_capture_unavailable",
+			"reason":   reason,
+			"executor": executor,
+		})
+		return
+	}
 	requestedBy := ""
 	if identity, ok := sessionFrom(r.Context()); ok {
 		requestedBy = identity.Username

internal/httpapi/library_capture_test.go

@@ -17,31 +17,11 @@ func captureRequest(body string) *http.Request {
 	return request
 }
 
-// An unwired agent must refuse clearly instead of queueing a job nobody can run.
-func TestLibraryCaptureRefusedWhenAgentUnwired(t *testing.T) {
-	store := &fakeJobStore{}
-	recorder := httptest.NewRecorder()
-	libraryCaptureEnqueue(recorder, captureRequest(`{"scope":"internal"}`), Config{Store: store})
-	if recorder.Code != http.StatusServiceUnavailable {
-		t.Fatalf("status=%d body=%s", recorder.Code, recorder.Body.String())
-	}
-	var body map[string]any
-	if err := json.Unmarshal(recorder.Body.Bytes(), &body); err != nil {
-		t.Fatal(err)
-	}
-	if body["code"] != "library_capture_unwired" || body["reason"] == "" {
-		t.Fatalf("unexpected body %v", body)
-	}
-	if store.enqueued.Type != "" {
-		t.Fatal("unwired agent still enqueued a job")
-	}
-}
-
 func TestLibraryCaptureEnqueuesFixedJobShape(t *testing.T) {
 	store := &fakeJobStore{enqueueEchoRequestedID: true}
 	recorder := httptest.NewRecorder()
 	libraryCaptureEnqueue(recorder, captureRequest(`{"scope":"external","job":"lib-external-review"}`),
-		Config{Store: store, LibraryAgentSocket: "/run/library-agent/agent.sock"})
+		captureReadyConfig(store))
 	if recorder.Code != http.StatusAccepted {
 		t.Fatalf("status=%d body=%s", recorder.Code, recorder.Body.String())
 	}
@@ -76,7 +56,7 @@ func TestLibraryCaptureRejectsInvalidRequest(t *testing.T) {
 	} {
 		store := &fakeJobStore{}
 		recorder := httptest.NewRecorder()
-		libraryCaptureEnqueue(recorder, captureRequest(body), Config{Store: store, LibraryAgentSocket: "/tmp/agent.sock"})
+		libraryCaptureEnqueue(recorder, captureRequest(body), captureReadyConfig(store))
 		if recorder.Code != http.StatusBadRequest {
 			t.Fatalf("body %q status=%d", body, recorder.Code)
 		}
@@ -92,8 +72,7 @@ func TestLibraryCaptureQueueIDFitsTheColumn(t *testing.T) {
 	for _, scope := range []string{"internal", "external"} {
 		store := &fakeJobStore{enqueueEchoRequestedID: true}
 		recorder := httptest.NewRecorder()
-		libraryCaptureEnqueue(recorder, captureRequest(`{"scope":"`+scope+`"}`),
-			Config{Store: store, LibraryAgentSocket: "/tmp/agent.sock"})
+		libraryCaptureEnqueue(recorder, captureRequest(`{"scope":"`+scope+`"}`), captureReadyConfig(store))
 		if recorder.Code != http.StatusAccepted {
 			t.Fatalf("%s status=%d body=%s", scope, recorder.Code, recorder.Body.String())
 		}
@@ -146,7 +125,7 @@ func TestLibraryCaptureRejectsScopeConflictForReusedJob(t *testing.T) {
 		Payload: json.RawMessage(`{"scope":"internal","capture_job":"lib-shared","requested_by":"yaozhanghao"}`)}}
 	recorder := httptest.NewRecorder()
 	libraryCaptureEnqueue(recorder, captureRequest(`{"scope":"external","job":"lib-shared"}`),
-		Config{Store: store, LibraryAgentSocket: "/run/library-agent/agent.sock"})
+		captureReadyConfig(store))
 	if recorder.Code != http.StatusConflict {
 		t.Fatalf("status=%d body=%s", recorder.Code, recorder.Body.String())
 	}
@@ -166,7 +145,7 @@ func TestLibraryCaptureReplayReportsStoredJob(t *testing.T) {
 		Payload: json.RawMessage(`{"scope":"internal","capture_job":"lib-shared","requested_by":"yaozhanghao"}`)}}
 	recorder := httptest.NewRecorder()
 	libraryCaptureEnqueue(recorder, captureRequest(`{"scope":"internal","job":"lib-shared"}`),
-		Config{Store: store, LibraryAgentSocket: "/run/library-agent/agent.sock"})
+		captureReadyConfig(store))
 	if recorder.Code != http.StatusAccepted {
 		t.Fatalf("status=%d body=%s", recorder.Code, recorder.Body.String())
 	}

internal/httpapi/library_executor.go

@@ -0,0 +1,77 @@
+package httpapi
+
+import (
+	"context"
+	"time"
+
+	"knowledge-governance/internal/libraryexecutor"
+)
+
+// libraryExecutorView 是"封存执行器到底能不能用"的页面口径。
+//
+// 为什么不是 `LIBRARY_AGENT_SOCKET != ""`：正式栈里 API 根本没有挂载那个 socket，
+// 配置非空并不代表链路通；更危险的是 agent 可能顶着正式单元名却在冻结另一个实例。
+// 这里只认 worker 写入的真实观测，并把观测时间与执行器自报的身份一起给出来。
+type libraryExecutorView struct {
+	Available   bool       `json:"available"`
+	Reason      string     `json:"reason,omitempty"`
+	Instance    string     `json:"instance,omitempty"`
+	Project     string     `json:"project,omitempty"`
+	SourceApp   string     `json:"source_app,omitempty"`
+	ArchiveRoot string     `json:"archive_root,omitempty"`
+	Busy        bool       `json:"busy"`
+	Degraded    string     `json:"degraded,omitempty"`
+	ObservedAt  *time.Time `json:"observed_at,omitempty"`
+}
+
+type libraryExecutorStore interface {
+	LibraryExecutorStatus(ctx context.Context, instance string) (*libraryexecutor.Observation, error)
+}
+
+// libraryExecutorAvailability 把"配置里的期望"与"worker 观测到的真实身份"合起来判定：
+// 观测不存在、过期、不可达、实例不符、降级、正忙，任何一条都不可用，并给出原因。
+func libraryExecutorAvailability(ctx context.Context, config Config) (bool, string, libraryExecutorView) {
+	// 期望身份是**封存源**：agent 冻结的是源库写入者，与同步目标（LIBRARY_TARGET_*）
+	// 本来就是两套名字（本机候选栈：源 kgmaint、目标 kgseal）。
+	instance := config.LibrarySourceInstance
+	view := libraryExecutorView{Instance: instance}
+	if instance == "" {
+		return false, "未配置封存来源实例身份（LIBRARY_SOURCE_INSTANCE），无法确认执行器归属", view
+	}
+	store, ok := config.Store.(libraryExecutorStore)
+	if !ok {
+		return false, "执行器状态存储不可用", view
+	}
+	observation, err := store.LibraryExecutorStatus(ctx, instance)
+	if err != nil {
+		return false, "读取执行器状态失败：" + err.Error(), view
+	}
+	if observation == nil {
+		return false, "封存执行器还没有上报状态（worker 未运行，或实例身份与 worker 不一致）", view
+	}
+	available, reason := libraryexecutor.Evaluate(*observation,
+		libraryexecutor.Expectation{Instance: instance, Project: config.LibrarySourceProject,
+			ArchiveRoot: config.LibraryArchiveRoot},
+		executorStatusMaxAge(config), time.Now().UTC())
+	observedAt := observation.ObservedAt
+	view.Available = available
+	view.Reason = reason
+	view.Project = observation.Identity.Project
+	view.SourceApp = observation.Identity.SourceApp
+	view.ArchiveRoot = observation.Identity.ArchiveRoot
+	view.Busy = observation.Busy
+	view.Degraded = observation.Degraded
+	view.ObservedAt = &observedAt
+	return available, reason, view
+}
+
+// defaultLibraryExecutorStatusMaxAge 与 worker 的探测间隔（30s）配套：
+// 没配上限绝不等于"永不过期"——那会把一次陈旧的"可用"永久留在页面上。
+const defaultLibraryExecutorStatusMaxAge = 2 * time.Minute
+
+func executorStatusMaxAge(config Config) time.Duration {
+	if config.LibraryExecutorStatusMaxAge > 0 {
+		return config.LibraryExecutorStatusMaxAge
+	}
+	return defaultLibraryExecutorStatusMaxAge
+}

internal/httpapi/library_executor_test.go

@@ -0,0 +1,184 @@
+package httpapi
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"knowledge-governance/internal/libraryexecutor"
+)
+
+// fakeJobStore 只是"存储"替身；执行器状态同样是它的一条记录。
+func (s *fakeJobStore) LibraryExecutorStatus(_ context.Context, instance string) (*libraryexecutor.Observation, error) {
+	if s.executorStatus == nil || s.executorStatus.Instance != instance {
+		return nil, nil
+	}
+	return s.executorStatus, nil
+}
+
+// healthyExecutor 返回一份"worker 刚观测到、指向本实例"的记录。
+func healthyExecutor(now time.Time) *libraryexecutor.Observation {
+	return &libraryexecutor.Observation{
+		Instance: "kgmaint-weknora", Reachable: true, ObservedAt: now,
+		Identity: libraryexecutor.Identity{Project: "kgmaint-weknora", SourceApp: "kgmaint-app",
+			ArchiveRoot: "/tmp/kg-e2e/archive", Writers: "app"},
+	}
+}
+
+// captureReadyConfig 是"执行器真的可用"时的配置：实例、期望项目与观测三者一致。
+func captureReadyConfig(store *fakeJobStore) Config {
+	store.executorStatus = healthyExecutor(time.Now().UTC())
+	return Config{
+		Store:                 store,
+		LibraryAgentSocket:    "/run/library-agent/agent.sock",
+		LibrarySourceInstance: "kgmaint-weknora",
+		LibrarySourceProject:  "kgmaint-weknora",
+		LibraryArchiveRoot:    "/var/lib/knowledge-governance/library",
+	}
+}
+
+func decodeCaptureRefusal(t *testing.T, recorder *httptest.ResponseRecorder) (string, string) {
+	t.Helper()
+	var body struct {
+		Code   string `json:"code"`
+		Reason string `json:"reason"`
+	}
+	if err := json.Unmarshal(recorder.Body.Bytes(), &body); err != nil {
+		t.Fatalf("body %s: %v", recorder.Body.String(), err)
+	}
+	return body.Code, body.Reason
+}
+
+// 页面上的"可发起"必须等于接口真的会接受：配置了 socket 但执行器不可达、
+// 指向别的实例、降级或过期时，都必须带着原因拒绝，而不是先排队再失败。
+func TestLibraryCaptureRefusesWhenExecutorIsNotVerified(t *testing.T) {
+	now := time.Now().UTC()
+	tests := []struct {
+		name   string
+		config func(*Config)
+		text   string
+	}{
+		{name: "no instance identity", config: func(c *Config) {
+			c.LibrarySourceInstance = ""
+		}, text: "LIBRARY_SOURCE_INSTANCE"},
+		{name: "nothing reported yet", config: func(c *Config) {
+			c.Store.(*fakeJobStore).executorStatus = nil
+		}, text: "还没有上报状态"},
+		{name: "unreachable", config: func(c *Config) {
+			c.Store.(*fakeJobStore).executorStatus.Reachable = false
+			c.Store.(*fakeJobStore).executorStatus.Error = "library agent is unreachable"
+		}, text: "不可达"},
+		{name: "stale observation", config: func(c *Config) {
+			c.Store.(*fakeJobStore).executorStatus.ObservedAt = now.Add(-10 * time.Minute)
+		}, text: "已过期"},
+		{name: "points at another instance", config: func(c *Config) {
+			c.Store.(*fakeJobStore).executorStatus.Identity.Project = "weknora071full"
+		}, text: "指向的实例不匹配"},
+		{name: "degraded", config: func(c *Config) {
+			c.Store.(*fakeJobStore).executorStatus.Degraded = "写入者恢复失败"
+		}, text: "降级"},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			store := &fakeJobStore{}
+			config := captureReadyConfig(store)
+			test.config(&config)
+			recorder := httptest.NewRecorder()
+			libraryCaptureEnqueue(recorder, captureRequest(`{"scope":"internal"}`), config)
+			if recorder.Code != http.StatusServiceUnavailable {
+				t.Fatalf("status=%d body=%s", recorder.Code, recorder.Body.String())
+			}
+			code, reason := decodeCaptureRefusal(t, recorder)
+			if code != "library_capture_unavailable" {
+				t.Fatalf("code=%q body=%s", code, recorder.Body.String())
+			}
+			if reason == "" || !strings.Contains(reason, test.text) {
+				t.Fatalf("reason=%q want it to mention %q", reason, test.text)
+			}
+			if store.enqueued.Type != "" {
+				t.Fatal("refused capture still enqueued a job")
+			}
+		})
+	}
+}
+
+// The overview must report the same reason the capture endpoint would use.
+func TestLibraryOverviewReportsExecutorIdentityAndReason(t *testing.T) {
+	store := &fakeLibraryStore{}
+	config := captureReadyConfig(&store.fakeJobStore)
+	// libraryJobStore 的方法在 fakeLibraryStore 上，执行器状态在嵌入的 fakeJobStore 上。
+	config.Store = store
+	recorder := httptest.NewRecorder()
+	libraryOverviewHandler(recorder, httptest.NewRequest(http.MethodGet, "/api/v1/library/overview?scope=internal", nil), config)
+	if recorder.Code != http.StatusOK {
+		t.Fatalf("status=%d body=%s", recorder.Code, recorder.Body.String())
+	}
+	var body struct {
+		CaptureAvailable bool `json:"capture_available"`
+		Executor         struct {
+			Available   bool   `json:"available"`
+			Instance    string `json:"instance"`
+			Project     string `json:"project"`
+			SourceApp   string `json:"source_app"`
+			ArchiveRoot string `json:"archive_root"`
+		} `json:"executor"`
+	}
+	if err := json.Unmarshal(recorder.Body.Bytes(), &body); err != nil {
+		t.Fatal(err)
+	}
+	if !body.CaptureAvailable || !body.Executor.Available {
+		t.Fatalf("overview did not report an available executor: %s", recorder.Body.String())
+	}
+	if body.Executor.Project != "kgmaint-weknora" || body.Executor.SourceApp != "kgmaint-app" ||
+		body.Executor.ArchiveRoot != "/tmp/kg-e2e/archive" || body.Executor.Instance != "kgmaint-weknora" {
+		t.Fatalf("executor view = %+v", body.Executor)
+	}
+
+	store.fakeJobStore.executorStatus.Identity.Project = "weknora071full"
+	recorder = httptest.NewRecorder()
+	libraryOverviewHandler(recorder, httptest.NewRequest(http.MethodGet, "/api/v1/library/overview?scope=internal", nil), config)
+	if err := json.Unmarshal(recorder.Body.Bytes(), &body); err != nil {
+		t.Fatal(err)
+	}
+	if body.CaptureAvailable {
+		t.Fatalf("overview still says available after the executor moved to another instance: %s", recorder.Body.String())
+	}
+}
+
+// "恢复至该版本"必须由真实条件决定：没有 restore 授权就明确说出来。
+func TestLibraryRestoreAvailabilityExplainsMissingAuthorization(t *testing.T) {
+	store := &fakeJobStore{}
+	config := Config{Store: store, ProductionReleaseRequestsEnabled: true, ProductionBackupRestoreVerified: true}
+	recorder := httptest.NewRecorder()
+	libraryRestoreAvailability(recorder, httptest.NewRequest(http.MethodGet,
+		"/api/v1/library/restore-availability?scope=internal&snapshot_id=snap-1", nil), config)
+	if recorder.Code != http.StatusOK {
+		t.Fatalf("status=%d body=%s", recorder.Code, recorder.Body.String())
+	}
+	var body struct {
+		Allowed       bool     `json:"allowed"`
+		Reasons       []string `json:"reasons"`
+		Authorization struct {
+			State string `json:"state"`
+		} `json:"authorization"`
+	}
+	if err := json.Unmarshal(recorder.Body.Bytes(), &body); err != nil {
+		t.Fatal(err)
+	}
+	if body.Allowed || body.Authorization.State != "none" || len(body.Reasons) == 0 {
+		t.Fatalf("body=%s", recorder.Body.String())
+	}
+	found := false
+	for _, reason := range body.Reasons {
+		if strings.Contains(reason, "恢复授权") {
+			found = true
+		}
+	}
+	if !found {
+		t.Fatalf("reasons = %v", body.Reasons)
+	}
+}

internal/httpapi/library_flow.go

@@ -307,6 +307,7 @@ type libraryOverview struct {
 	Authorization    libraryAuthorizationView `json:"authorization"`
 	CaptureAvailable bool                     `json:"capture_available"`
 	CaptureReason    string                   `json:"capture_reason,omitempty"`
+	Executor         libraryExecutorView      `json:"executor"`
 	Now              time.Time                `json:"now"`
 }
 
@@ -563,12 +564,10 @@ func libraryOverviewHandler(w http.ResponseWriter, r *http.Request, config Confi
 	}
 	now := time.Now().UTC()
 	overview := libraryOverview{Scope: scope, Jobs: []governance.LibraryJob{}, Now: now,
-		CaptureAvailable: strings.TrimSpace(config.LibraryAgentSocket) != "",
-		Gate:             libraryGateView{Reasons: []string{}},
-		Diff:             libraryDiffView{State: "unknown", Reason: "生产与测试的文档级差异读取未接线", Added: []any{}, Updated: []any{}, Removed: []any{}}}
-	if !overview.CaptureAvailable {
-		overview.CaptureReason = "主机侧封存 agent 未接线"
-	}
+		Gate: libraryGateView{Reasons: []string{}},
+		Diff: libraryDiffView{State: "unknown", Reason: "生产与测试的文档级差异读取未接线", Added: []any{}, Updated: []any{}, Removed: []any{}}}
+	// 可发起封存 = worker 真实探测到执行器可用且指向本实例；原因逐条给全。
+	overview.CaptureAvailable, overview.CaptureReason, overview.Executor = libraryExecutorAvailability(r.Context(), config)
 
 	// 测试封存来自归档：它只是测试侧事实。
 	snapshots, snapErr := readLibrarySnapshots(r.Context(), config.LibraryArchiveRoot)

internal/httpapi/library_restore.go

@@ -146,6 +146,64 @@ func libraryRestoreEnqueue(w http.ResponseWriter, r *http.Request, config Config
 	writeJSON(w, http.StatusAccepted, map[string]any{"job": job, "reused": false})
 }
 
+// libraryRestoreAvailability 回答"现在能不能把生产恢复到这份快照"。
+//
+// 为什么需要它：页面原来把"恢复至该版本"按钮写死为 disabled，理由是"未接线"——
+// 但接口、授权、恢复作业其实都已经在跑。按钮该由真实条件决定：双闸打开、
+// 有这份快照的恢复授权、目标实例没有在维护。
+type libraryRestoreAvailabilityView struct {
+	Allowed       bool                     `json:"allowed"`
+	Reasons       []string                 `json:"reasons"`
+	Scope         string                   `json:"scope"`
+	SnapshotID    string                   `json:"snapshot_id"`
+	Authorization libraryAuthorizationView `json:"authorization"`
+}
+
+func libraryRestoreAvailability(w http.ResponseWriter, r *http.Request, config Config) {
+	scope, err := scopeFromQuery(r)
+	if err != nil {
+		writeJSON(w, http.StatusBadRequest, map[string]string{"error": err.Error()})
+		return
+	}
+	snapshotID := strings.TrimSpace(r.URL.Query().Get("snapshot_id"))
+	if snapshotID == "" {
+		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "snapshot_id required"})
+		return
+	}
+	view := libraryRestoreAvailabilityView{Scope: scope, SnapshotID: snapshotID, Reasons: []string{}}
+	if !config.ProductionReleaseRequestsEnabled {
+		view.Reasons = append(view.Reasons, "生产写入闸口关闭（PRODUCTION_RELEASE_REQUESTS_ENABLED=false）")
+	}
+	if !config.ProductionBackupRestoreVerified {
+		view.Reasons = append(view.Reasons, "没有已核验的生产恢复演练证据（PRODUCTION_BACKUP_RESTORE_VERIFIED=false）")
+	}
+	digest, err := libraryRestoreParamsDigest(scope, snapshotID)
+	if err != nil {
+		writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot digest params"})
+		return
+	}
+	grant, found := governance.LibraryReleaseAuthorization{}, false
+	if storeWithAuth, ok := config.Store.(interface {
+		ActiveLibraryAuthorization(ctx context.Context, scope, snapshotID, digest string, now time.Time) (governance.LibraryReleaseAuthorization, bool, error)
+	}); ok {
+		grant, found, err = storeWithAuth.ActiveLibraryAuthorization(r.Context(), scope, snapshotID, digest, time.Now().UTC())
+		if err != nil {
+			writeJSON(w, http.StatusServiceUnavailable, map[string]any{"allowed": false,
+				"reasons": []string{"读取恢复授权失败：" + err.Error()}})
+			return
+		}
+	}
+	view.Authorization = authorizationViewOf(grant, found)
+	if !found {
+		view.Reasons = append(view.Reasons, "没有针对该快照的恢复授权：先签发 restore 授权（绑定同一快照）")
+	}
+	if _, held := instanceMaintenanceGuard(r.Context(), config, scope); held {
+		view.Reasons = append(view.Reasons, "目标实例正在维护中，等当前作业释放后再恢复")
+	}
+	view.Allowed = len(view.Reasons) == 0
+	writeJSON(w, http.StatusOK, view)
+}
+
 // 崩溃续作：失败作业可以按原参数重新排队，生成新作业（旧作业保留为审计记录）。
 func libraryJobResume(w http.ResponseWriter, r *http.Request, config Config) {
 	identity, ok := sessionFrom(r.Context())

internal/httpapi/library_snapshots.go

@@ -85,15 +85,14 @@ const libraryCaptureNote = "封存由运维命令 scripts/library-capture.sh cap
 
 func librarySnapshotsList(w http.ResponseWriter, r *http.Request, config Config) {
 	view, err := readLibrarySnapshots(r.Context(), config.LibraryArchiveRoot)
-	view.CaptureAvailable = config.LibraryAgentSocket != ""
-	if !view.CaptureAvailable {
-		view.CaptureReason = "未接线：主机侧封存 agent 没有挂载，页面只能查看已封存版本"
-	}
+	// 能不能发起封存只认 worker 的真实观测，不认"配置里写了 socket 路径"。
+	view.CaptureAvailable, view.CaptureReason, _ = libraryExecutorAvailability(r.Context(), config)
 	if err != nil {
 		status, reason := libraryArchiveErrorStatus(err, "归档不可读")
 		writeJSON(w, status, librarySnapshotList{Available: false, Reason: reason,
 			Snapshots: []librarySnapshotView{}, Verification: libraryVerificationManifest,
-			VerificationNote: libraryVerificationNote, CaptureNote: libraryCaptureNote})
+			VerificationNote: libraryVerificationNote, CaptureNote: libraryCaptureNote,
+			CaptureAvailable: view.CaptureAvailable, CaptureReason: view.CaptureReason})
 		return
 	}
 	writeJSON(w, http.StatusOK, view)

internal/httpapi/server.go

@@ -152,6 +152,11 @@ type Config struct {
 	LibrarySourceInstance string
 	// LibraryAgentSocket 是主机侧封存 agent 的 unix socket；留空即页面不能发起封存。
 	LibraryAgentSocket string
+	// LibrarySourceProject 是执行器应指向的实例名（agent -project，封存源）。
+	// 与它不一致时按"封错库"处理：不可发起，并说明差在哪。
+	LibrarySourceProject string
+	// LibraryExecutorStatusMaxAge 是执行器观测的有效期；超过即"无法确认"。
+	LibraryExecutorStatusMaxAge time.Duration
 	// LibraryCapacityBudgetBytes 是整库归档容量预算；0 = 未设置（容量门禁不放行）。
 	LibraryCapacityBudgetBytes int64
 	// 整库生产目标的只读探测配置：首次发布要能证明"生产可读且为空"。
@@ -510,6 +515,10 @@ func NewHandler(config Config) (http.Handler, error) {
 	mux.HandleFunc("POST /api/v1/library/restores", func(w http.ResponseWriter, r *http.Request) {
 		libraryRestoreEnqueue(w, r, config)
 	})
+	// 页面问"这份快照现在能不能恢复"：条件来自双闸、授权与实例维护，而不是写死的按钮。
+	mux.HandleFunc("GET /api/v1/library/restore-availability", func(w http.ResponseWriter, r *http.Request) {
+		libraryRestoreAvailability(w, r, config)
+	})
 	// 失败/取消作业续作：按原参数重新排队，旧作业保留审计。
 	mux.HandleFunc("POST /api/v1/library/jobs/{id}/resume", func(w http.ResponseWriter, r *http.Request) {
 		libraryJobResume(w, r, config)

internal/httpapi/server_test.go

@@ -15,6 +15,7 @@ import (
 
 	"knowledge-governance/internal/governance"
 	"knowledge-governance/internal/jobs"
+	"knowledge-governance/internal/libraryexecutor"
 	postgresstore "knowledge-governance/internal/platform/postgres"
 )
 
@@ -1501,30 +1502,32 @@ type fakeJobStore struct {
 	activateSettingsErr          error
 	// enqueueEchoRequestedID：幂等"首次入队"语义 —— 返回调用方请求的那个 id
 	// （默认 false 时返回 enqueueResult，即"复用已有任务"的路径）。
-	enqueueEchoRequestedID        bool
-	listSettingsConnectionID      string
-	listSettingsErr               error
-	settingsRevisions             []governance.SourceConnectionRevision
-	jobFailureFilter              governance.JobFailureFilter
-	jobFailurePage                governance.JobFailurePage
-	jobFailureErr                 error
-	retryBatchRequest             *governance.RetryPublicationsBatchRequest
-	retryBatchOperation           string
-	retryBatchResult              governance.RetryPublicationsBatchResult
-	cancelBatchRequest            *governance.CancelPublicationsBatchRequest
-	cancelBatchOperation          string
-	cancelBatchResult             governance.RetryPublicationsBatchResult
-	cancelBatchReplayed           bool
-	cancelBatchErr                error
-	retryBatchReplayed            bool
-	retryBatchErr                 error
-	requeueFailureRequest         *governance.RequeueArtifactFetchRequest
-	requeueFailureOperation       string
-	requeueFailureResult          governance.RequeueArtifactFetchResult
-	requeueFailureReplayed        bool
-	requeueFailureErr             error
-	enqueued                      jobs.EnqueueRequest
-	enqueueResult                 *jobs.Job
+	enqueueEchoRequestedID   bool
+	listSettingsConnectionID string
+	listSettingsErr          error
+	settingsRevisions        []governance.SourceConnectionRevision
+	jobFailureFilter         governance.JobFailureFilter
+	jobFailurePage           governance.JobFailurePage
+	jobFailureErr            error
+	retryBatchRequest        *governance.RetryPublicationsBatchRequest
+	retryBatchOperation      string
+	retryBatchResult         governance.RetryPublicationsBatchResult
+	cancelBatchRequest       *governance.CancelPublicationsBatchRequest
+	cancelBatchOperation     string
+	cancelBatchResult        governance.RetryPublicationsBatchResult
+	cancelBatchReplayed      bool
+	cancelBatchErr           error
+	retryBatchReplayed       bool
+	retryBatchErr            error
+	requeueFailureRequest    *governance.RequeueArtifactFetchRequest
+	requeueFailureOperation  string
+	requeueFailureResult     governance.RequeueArtifactFetchResult
+	requeueFailureReplayed   bool
+	requeueFailureErr        error
+	enqueued                 jobs.EnqueueRequest
+	enqueueResult            *jobs.Job
+	// executorStatus 是封存执行器的"最近一次观测"；nil 表示还没有任何上报。
+	executorStatus                *libraryexecutor.Observation
 	listResult                    []jobs.Job
 	listType                      string
 	listByTypeResult              []jobs.Job

internal/libraryagent/agent.go

@@ -89,13 +89,45 @@ const maxReleasedDeadLetters = 64
 type Agent struct {
 	config Config
 	mu     sync.Mutex
+	// stateMu 只保护下面两个"给人看"的字段：健康应答不能被一次 15 分钟的封存
+	// 卡住（否则页面会把"正在封存"读成"执行器不可达"）。
+	stateMu sync.RWMutex
 	// degraded is non-empty when a failed operation could not be followed by a
 	// verified recovery. While set, capture requests are refused: leaving the
 	// source writer stopped is worse than refusing work.
 	degraded string
+	busy     bool
 	wg       sync.WaitGroup
 }
 
+// captureState 是健康应答要暴露的执行器状态：是否在跑、是否降级。
+type captureState struct {
+	Busy     bool   `json:"busy"`
+	Degraded string `json:"degraded"`
+}
+
+func (a *Agent) snapshotState() captureState {
+	a.stateMu.RLock()
+	defer a.stateMu.RUnlock()
+	return captureState{Busy: a.busy, Degraded: a.degraded}
+}
+
+func (a *Agent) setBusy(busy bool) {
+	a.stateMu.Lock()
+	defer a.stateMu.Unlock()
+	a.busy = busy
+}
+
+func (a *Agent) setDegraded(reason string) {
+	a.stateMu.Lock()
+	defer a.stateMu.Unlock()
+	a.degraded = reason
+}
+
+func (a *Agent) clearDegraded() {
+	a.setDegraded("")
+}
+
 func New(config Config) (*Agent, error) {
 	if !filepath.IsAbs(config.Socket) || !filepath.IsAbs(config.ArchiveRoot) || !filepath.IsAbs(config.JournalRoot) {
 		return nil, errors.New("library agent paths must be absolute")
@@ -143,7 +175,7 @@ func (a *Agent) Serve(ctx context.Context) error {
 	// writer from a previous killed run. Recover before accepting work; if that
 	// fails, stay degraded and refuse captures instead of pretending to be ready.
 	if err := a.recoverWriters(ctx); err != nil {
-		a.degraded = err.Error()
+		a.setDegraded(err.Error())
 		log.Printf("library agent: pending freeze recovery failed: %v", err)
 	}
 	go func() {
@@ -194,13 +226,18 @@ func (a *Agent) handle(ctx context.Context, conn net.Conn) {
 
 // execute serialises operations and maps a request onto the one fixed command.
 func (a *Agent) execute(ctx context.Context, req request) response {
+	// health 不进操作锁、不跑子进程：调用方要的是"这个执行器指向谁、能不能干活"，
+	// 而一次正在进行的封存不该让这个回答变成"不可达"。
+	if req.Action == "health" {
+		return a.health()
+	}
 	a.mu.Lock()
 	defer a.mu.Unlock()
 	out := response{Action: req.Action, Scope: req.Scope, Job: req.Job}
 	// Degraded blocks new captures, but recover and status must stay reachable:
 	// otherwise there is no way back to a verified state.
-	if a.degraded != "" && req.Action == "capture" {
-		out.Error = "封存执行边界处于降级状态（上次恢复失败：" + a.degraded + "），先执行 recover"
+	if state := a.snapshotState(); state.Degraded != "" && req.Action == "capture" {
+		out.Error = "封存执行边界处于降级状态（上次恢复失败：" + state.Degraded + "），先执行 recover"
 		return out
 	}
 	var args []string
@@ -242,6 +279,10 @@ func (a *Agent) execute(ctx context.Context, req request) response {
 	args = append(args, a.config.declaration()...)
 	bounded, cancel := context.WithTimeout(ctx, a.config.MaxDuration)
 	defer cancel()
+	if req.Action == "capture" || req.Action == "recover" {
+		a.setBusy(true)
+		defer a.setBusy(false)
+	}
 	payload, err := a.config.run(bounded, a.config.Binary, args...)
 	if err != nil {
 		detail := "operation failed"
@@ -254,7 +295,7 @@ func (a *Agent) execute(ctx context.Context, req request) response {
 		// verify.
 		if req.Action == "capture" || req.Action == "recover" {
 			if recoveryErr := a.recoverWriters(ctx); recoveryErr != nil {
-				a.degraded = recoveryErr.Error()
+				a.setDegraded(recoveryErr.Error())
 				out.Error = detail + "；写入者恢复失败：" + recoveryErr.Error() + "（已阻断后续封存，需人工处理）"
 				return out
 			}
@@ -266,7 +307,7 @@ func (a *Agent) execute(ctx context.Context, req request) response {
 		return out
 	}
 	if req.Action == "recover" {
-		a.degraded = ""
+		a.clearDegraded()
 	}
 	out.OK = true
 	if json.Valid(payload) {
@@ -274,9 +315,28 @@ func (a *Agent) execute(ctx context.Context, req request) response {
 	} else {
 		out.Payload = json.RawMessage(`{}`)
 	}
+	// 封存结果必须带上"这次是谁执行的"：调用方据此复核，防止心跳与执行之间
+	// socket 被换成另一个 agent（真正的封错库事故就是这样发生的）。
+	if req.Action == "capture" {
+		out.Payload = a.attachIdentity(out.Payload)
+	}
 	return out
 }
 
+// attachIdentity 把执行器身份并进封存结果（保留 CLI 原有的字段，如 snapshot_id）。
+func (a *Agent) attachIdentity(payload json.RawMessage) json.RawMessage {
+	var document map[string]any
+	if err := json.Unmarshal(payload, &document); err != nil {
+		return payload
+	}
+	document["agent_identity"] = a.config.identity()
+	encoded, err := json.Marshal(document)
+	if err != nil {
+		return payload
+	}
+	return json.RawMessage(encoded)
+}
+
 // recoverWriters runs the idempotent recovery command with its own budget, even
 // when the triggering context is already cancelled (timeout / shutdown).
 func (a *Agent) recoverWriters(ctx context.Context) error {
@@ -285,7 +345,7 @@ func (a *Agent) recoverWriters(ctx context.Context) error {
 	if _, err := a.config.run(bounded, a.config.Binary, append([]string{"recover"}, a.config.declaration()...)...); err != nil {
 		return err
 	}
-	a.degraded = ""
+	a.clearDegraded()
 	return nil
 }
 
@@ -332,12 +392,38 @@ func (c Config) run(ctx context.Context, name string, args ...string) ([]byte, e
 
 // Describe returns a redacted view for logs and health checks.
 func (a *Agent) Describe() map[string]any {
-	return map[string]any{
-		"socket":       a.config.Socket,
-		"archive_root": a.config.ArchiveRoot,
-		"journal_root": a.config.JournalRoot,
-		"project":      a.config.Project,
-		"writers":      a.config.Writers,
-		"scope_choice": "request may select internal or external only",
-	}
+	view := a.config.identity()
+	view["scope_choice"] = "request may select internal or external only"
+	return view
+}
+
+// health 是"这个执行器指向谁、现在能不能干活"的唯一回答。它只暴露容器名、
+// 路径与声明，不含任何凭据；调用方据此核对源身份，而不是靠配置猜。
+func (a *Agent) health() response {
+	state := a.snapshotState()
+	payload := a.config.identity()
+	payload["busy"] = state.Busy
+	payload["degraded"] = state.Degraded
+	encoded, err := json.Marshal(payload)
+	if err != nil {
+		return response{Action: "health", Error: "cannot encode health"}
+	}
+	return response{OK: true, Action: "health", Payload: json.RawMessage(encoded)}
+}
+
+// identity 被日志与健康应答共用一份，避免两处描述漂移。
+func (c Config) identity() map[string]any {
+	identity := map[string]any{
+		"project":      c.Project,
+		"archive_root": c.ArchiveRoot,
+		"journal_root": c.JournalRoot,
+		"source_app":   c.SourceApp,
+		"source_pg":    c.SourcePG,
+		"source_es":    c.SourceES,
+		"writers":      c.Writers,
+	}
+	if c.SourceS3Endpoint != "" {
+		identity["source_s3_endpoint"] = c.SourceS3Endpoint
+	}
+	return identity
 }

internal/libraryagent/client.go

@@ -46,6 +46,12 @@ func Status(ctx context.Context, config ClientConfig) (Result, error) {
 	return call(ctx, config, map[string]any{"action": "status"})
 }
 
+// Health 只问三件事：这个执行器指向谁（project/容器/归档）、是否在跑、是否降级。
+// 它不跑子进程、不占操作锁，所以可以当心跳用；Status 会去查源库，代价高得多。
+func Health(ctx context.Context, config ClientConfig) (Result, error) {
+	return call(ctx, config, map[string]any{"action": "health"})
+}
+
 // Recover restores writers from an interrupted freeze journal.
 func Recover(ctx context.Context, config ClientConfig) (Result, error) {
 	return call(ctx, config, map[string]any{"action": "recover"})

internal/libraryagent/health_test.go

@@ -0,0 +1,108 @@
+package libraryagent
+
+import (
+	"context"
+	"encoding/json"
+	"os"
+	"testing"
+	"time"
+)
+
+func serveAgent(t *testing.T, agent *Agent, socket string) {
+	t.Helper()
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+	go func() { _ = agent.Serve(ctx) }()
+	deadline := time.Now().Add(2 * time.Second)
+	for {
+		if _, err := os.Stat(socket); err == nil {
+			return
+		}
+		if time.Now().After(deadline) {
+			t.Fatal("agent socket never appeared")
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+}
+
+// 健康应答必须不排队等正在跑的封存：一次封存最长 15 分钟，如果 health 也去抢
+// 操作锁，页面就会把"正在封存"看成"执行器不可达"，进而误导运维。
+func TestHealthAnswersWhileACaptureHoldsTheOperationLock(t *testing.T) {
+	agent, socket, calls := testAgent(t)
+	serveAgent(t, agent, socket)
+	// 占住操作锁并置 busy，模拟一次正在进行的封存。
+	agent.mu.Lock()
+	defer agent.mu.Unlock()
+	agent.setBusy(true)
+	defer agent.setBusy(false)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+	result, err := Health(ctx, ClientConfig{Socket: socket, Timeout: 2 * time.Second})
+	if err != nil {
+		t.Fatalf("health while busy: %v", err)
+	}
+	var payload struct {
+		Project string `json:"project"`
+		Busy    bool   `json:"busy"`
+	}
+	if err = json.Unmarshal(result.Payload, &payload); err != nil {
+		t.Fatalf("health payload: %v", err)
+	}
+	if payload.Project != "weknora071full" || !payload.Busy {
+		t.Fatalf("health payload = %+v", payload)
+	}
+	if len(captureCalls(calls())) != 0 {
+		t.Fatalf("health must not run the capture binary, calls = %v", calls())
+	}
+}
+
+// 降级状态必须能被调用方看见：否则页面只能看到"可发起"，而真发起会被拒。
+func TestHealthReportsDegradedState(t *testing.T) {
+	agent, socket, _ := testAgent(t)
+	serveAgent(t, agent, socket)
+	// 启动恢复成功会清空降级；这里在它之后模拟"恢复失败"留下的降级状态。
+	agent.setDegraded("写入者恢复失败")
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+	result, err := Health(ctx, ClientConfig{Socket: socket, Timeout: 2 * time.Second})
+	if err != nil {
+		t.Fatal(err)
+	}
+	var payload struct {
+		Degraded  string `json:"degraded"`
+		SourceApp string `json:"source_app"`
+		Writers   string `json:"writers"`
+	}
+	if err = json.Unmarshal(result.Payload, &payload); err != nil {
+		t.Fatal(err)
+	}
+	if payload.Degraded != "写入者恢复失败" || payload.SourceApp != "WKNFull-app" || payload.Writers != "app" {
+		t.Fatalf("health payload = %+v", payload)
+	}
+}
+
+// 封存结果必须自带"这次是谁执行的"：调用方据此在事后复核，防止心跳与执行之间
+// socket 被换成另一个实例。
+func TestCaptureResultCarriesTheAgentIdentity(t *testing.T) {
+	agent, socket, _ := testAgent(t)
+	serveAgent(t, agent, socket)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	result, err := Capture(ctx, ClientConfig{Socket: socket, Timeout: 5 * time.Second}, "internal", "lib-internal-identity")
+	if err != nil {
+		t.Fatal(err)
+	}
+	var payload struct {
+		SnapshotID string `json:"snapshot_id"`
+		Identity   struct {
+			Project string `json:"project"`
+		} `json:"agent_identity"`
+	}
+	if err = json.Unmarshal(result.Payload, &payload); err != nil {
+		t.Fatal(err)
+	}
+	if payload.SnapshotID != "abc" || payload.Identity.Project != "weknora071full" {
+		t.Fatalf("payload = %s", result.Payload)
+	}
+}

internal/libraryexecutor/executor.go

@@ -0,0 +1,122 @@
+// Package libraryexecutor 回答一个问题：主机侧封存执行器现在是不是真的可用，
+// 而且是不是指向我们这个实例。
+//
+// 为什么不能在页面里读一个配置字符串了事：正式栈曾经出现过"agent 顶着正式单元名
+// 却在冻结候选实例"的事故——配置非空、页面显示可发起、真发起却封错了库。可用的
+// 定义必须是**运行时观测**：socket 通、执行器自报的实例与期望一致、没有降级。
+package libraryexecutor
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"knowledge-governance/internal/libraryagent"
+)
+
+// Identity 是执行器自报的身份：它要冻结哪个容器的写入者、归档写到哪。
+// 只有容器名与路径，没有任何凭据。
+type Identity struct {
+	Project          string `json:"project"`
+	ArchiveRoot      string `json:"archive_root"`
+	JournalRoot      string `json:"journal_root"`
+	SourceApp        string `json:"source_app"`
+	SourcePG         string `json:"source_pg"`
+	SourceES         string `json:"source_es"`
+	Writers          string `json:"writers"`
+	SourceS3Endpoint string `json:"source_s3_endpoint,omitempty"`
+}
+
+// Observation 是一次真实探测的结果，会被落库供页面读取。
+type Observation struct {
+	Instance   string    `json:"instance"`
+	Reachable  bool      `json:"reachable"`
+	Busy       bool      `json:"busy"`
+	Degraded   string    `json:"degraded,omitempty"`
+	Identity   Identity  `json:"identity"`
+	Error      string    `json:"error,omitempty"`
+	ObservedAt time.Time `json:"observed_at"`
+}
+
+// Expectation 是治理侧对执行器的要求。ArchiveRoot 不参与判定：容器里的路径是
+// 挂载后的路径，和宿主机上的真实路径本来就不是一个字符串；用它当门禁只会造成
+// 永久误报，所以只作为观测值展示给运维比对。
+type Expectation struct {
+	Instance    string
+	Project     string
+	ArchiveRoot string
+}
+
+// Probe 用一次短超时的 health 往返取回执行器身份。它不跑子进程、不碰源库，
+// 所以可以当心跳反复调用。instance 是"这次观测属于哪个实例"，由调用方的
+// LIBRARY_TARGET_INSTANCE 决定。
+func Probe(ctx context.Context, instance, socket string, timeout time.Duration, now func() time.Time) Observation {
+	if now == nil {
+		now = time.Now
+	}
+	observation := Observation{Instance: instance, ObservedAt: now().UTC()}
+	if socket == "" {
+		observation.Error = "未配置主机侧封存 agent socket（LIBRARY_AGENT_SOCKET）"
+		return observation
+	}
+	if timeout <= 0 {
+		timeout = 3 * time.Second
+	}
+	result, err := libraryagent.Health(ctx, libraryagent.ClientConfig{Socket: socket, Timeout: timeout})
+	if err != nil {
+		observation.Error = err.Error()
+		return observation
+	}
+	var payload struct {
+		Identity
+		Busy     bool   `json:"busy"`
+		Degraded string `json:"degraded"`
+	}
+	if err = json.Unmarshal(result.Payload, &payload); err != nil {
+		observation.Error = "执行器健康应答无法解析"
+		return observation
+	}
+	observation.Reachable = true
+	observation.Busy = payload.Busy
+	observation.Degraded = payload.Degraded
+	observation.Identity = payload.Identity
+	return observation
+}
+
+// Evaluate 把"观测 + 期望 + 新鲜度"换算成页面上的可用性与原因。
+// 返回的原因必须能直接告诉运维该去改什么，而不是"不可用"三个字。
+func Evaluate(observation Observation, want Expectation, maxAge time.Duration, now time.Time) (bool, string) {
+	if want.Instance == "" {
+		return false, "未配置封存来源实例身份（LIBRARY_SOURCE_INSTANCE），无法确认执行器归属"
+	}
+	if observation.Instance != want.Instance {
+		return false, fmt.Sprintf("执行器状态来自另一个实例（%q），不是 %q", observation.Instance, want.Instance)
+	}
+	age := now.Sub(observation.ObservedAt)
+	if maxAge > 0 && age > maxAge {
+		return false, fmt.Sprintf("执行器状态已过期：%d 秒前的观测，上限 %d 秒（worker 未上报？）",
+			int(age.Seconds()), int(maxAge.Seconds()))
+	}
+	if !observation.Reachable {
+		reason := "主机侧封存执行器不可达"
+		if observation.Error != "" {
+			reason += "：" + observation.Error
+		}
+		return false, reason
+	}
+	if want.Project == "" {
+		return false, "未配置执行器源身份（LIBRARY_SOURCE_PROJECT），无法核对 agent 指向哪个实例"
+	}
+	if observation.Identity.Project != want.Project {
+		return false, fmt.Sprintf("执行器指向的实例不匹配：agent=%q，期望=%q",
+			observation.Identity.Project, want.Project)
+	}
+	if observation.Degraded != "" {
+		return false, "执行器处于降级状态（上次恢复失败）：" + observation.Degraded
+	}
+	if observation.Busy {
+		return false, "执行器正在执行另一个封存/恢复，等它结束再发起"
+	}
+	return true, ""
+}

internal/libraryexecutor/executor_test.go

@@ -0,0 +1,133 @@
+package libraryexecutor
+
+import (
+	"context"
+	"encoding/json"
+	"net"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+)
+
+func shortSocketDir(t *testing.T) string {
+	t.Helper()
+	// unix socket 路径上限 ~104 字节，TMPDIR（本机是 .hermes/cache/scratch）太长，
+	// 测试里用一个短目录并立刻清理。
+	dir, err := os.MkdirTemp("/tmp", "kgex-")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { os.RemoveAll(dir) })
+	return dir
+}
+
+// 可用性必须来自运行时观测，而不是"配置里写了 socket 路径"。
+func TestProbeReadsIdentityFromTheAgentOverTheSocket(t *testing.T) {
+	root := shortSocketDir(t)
+	socket := filepath.Join(root, "agent.sock")
+	listener, err := net.Listen("unix", socket)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer listener.Close()
+	go func() {
+		for {
+			conn, acceptErr := listener.Accept()
+			if acceptErr != nil {
+				return
+			}
+			var request map[string]any
+			_ = json.NewDecoder(conn).Decode(&request)
+			payload, _ := json.Marshal(map[string]any{
+				"project": "kgmaint-weknora", "archive_root": "/tmp/kg-e2e/archive",
+				"journal_root": "/tmp/kg-e2e/journal", "source_app": "kgmaint-app",
+				"source_pg": "kgmaint-postgres", "source_es": "WKNFull-elasticsearch",
+				"writers": "app", "busy": true, "degraded": "",
+			})
+			_ = json.NewEncoder(conn).Encode(map[string]any{"ok": true, "action": "health", "payload": json.RawMessage(payload)})
+			conn.Close()
+		}
+	}()
+
+	observation := Probe(context.Background(), "kgmaint-weknora", socket, 2*time.Second, nil)
+	if !observation.Reachable {
+		t.Fatalf("observation = %+v", observation)
+	}
+	if observation.Identity.Project != "kgmaint-weknora" || observation.Identity.SourceApp != "kgmaint-app" || !observation.Busy {
+		t.Fatalf("identity = %+v", observation.Identity)
+	}
+	if observation.Instance != "kgmaint-weknora" || observation.ObservedAt.IsZero() {
+		t.Fatalf("observation = %+v", observation)
+	}
+}
+
+func TestProbeReportsUnreachableAgentWithoutFakingSuccess(t *testing.T) {
+	observation := Probe(context.Background(), "kgmaint-weknora", filepath.Join(t.TempDir(), "absent.sock"), time.Second, nil)
+	if observation.Reachable || observation.Error == "" {
+		t.Fatalf("observation = %+v", observation)
+	}
+	available, reason := Evaluate(observation, Expectation{Instance: "kgmaint-weknora", Project: "kgmaint-weknora"},
+		time.Minute, time.Now().UTC())
+	if available || !strings.Contains(reason, "不可达") {
+		t.Fatalf("available=%v reason=%q", available, reason)
+	}
+}
+
+func TestEvaluateRejectsEveryIdentityAndFreshnessProblem(t *testing.T) {
+	now := time.Date(2026, 10, 29, 8, 0, 0, 0, time.UTC)
+	healthy := Observation{
+		Instance: "kgmaint-weknora", Reachable: true, ObservedAt: now,
+		Identity: Identity{Project: "kgmaint-weknora", SourceApp: "kgmaint-app"},
+	}
+	want := Expectation{Instance: "kgmaint-weknora", Project: "kgmaint-weknora"}
+	if available, reason := Evaluate(healthy, want, 2*time.Minute, now); !available || reason != "" {
+		t.Fatalf("healthy executor rejected: %v %q", available, reason)
+	}
+
+	tests := []struct {
+		name string
+		mut  func(*Observation)
+		want func(Expectation) Expectation
+		text string
+	}{
+		{name: "missing instance config", want: func(Expectation) Expectation { return expectEmptyInstance },
+			text: "LIBRARY_SOURCE_INSTANCE"},
+		{name: "row from another instance", mut: func(o *Observation) { o.Instance = "someone-else" },
+			text: "另一个实例"},
+		{name: "stale", mut: func(o *Observation) { o.ObservedAt = now.Add(-10 * time.Minute) },
+			text: "已过期"},
+		{name: "unreachable", mut: func(o *Observation) { o.Reachable = false; o.Error = "socket missing" },
+			text: "不可达：socket missing"},
+		{name: "project not configured", want: func(Expectation) Expectation { return Expectation{Instance: "kgmaint-weknora"} },
+			text: "LIBRARY_SOURCE_PROJECT"},
+		{name: "wrong source project", mut: func(o *Observation) { o.Identity.Project = "weknora071full" },
+			text: "指向的实例不匹配"},
+		{name: "degraded", mut: func(o *Observation) { o.Degraded = "写入者恢复失败" },
+			text: "降级状态"},
+		{name: "busy", mut: func(o *Observation) { o.Busy = true },
+			text: "正在执行另一个封存"},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			observation := healthy
+			if test.mut != nil {
+				test.mut(&observation)
+			}
+			expectation := want
+			if test.want != nil {
+				expectation = test.want(want)
+			}
+			available, reason := Evaluate(observation, expectation, 2*time.Minute, now)
+			if available {
+				t.Fatalf("accepted: %+v", observation)
+			}
+			if !strings.Contains(reason, test.text) {
+				t.Fatalf("reason = %q, want it to mention %q", reason, test.text)
+			}
+		})
+	}
+}
+
+var expectEmptyInstance = Expectation{Project: "kgmaint-weknora"}

internal/platform/postgres/library_executor_status.go

@@ -0,0 +1,71 @@
+package postgres
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+
+	"knowledge-governance/internal/libraryexecutor"
+)
+
+// 为什么落库而不是 API 直接探测：agent socket 只挂在 worker 容器里，API 那个
+// LIBRARY_AGENT_SOCKET 即使配了也 dial 不到。谁真能执行就由谁观测，页面读结果。
+const executorStatusColumns = `instance, reachable, busy, degraded, identity, status_error, observed_at`
+
+// UpsertLibraryExecutorStatus 记录一次真实探测。降级/不可达同样要写：
+// "最新一次观测是什么"比"最后一条好消息"更有用。
+func (s *Store) UpsertLibraryExecutorStatus(ctx context.Context, observation libraryexecutor.Observation, observer string, now time.Time) error {
+	identity, err := json.Marshal(observation.Identity)
+	if err != nil {
+		return fmt.Errorf("encode library executor identity: %w", err)
+	}
+	if observation.Instance == "" {
+		return errors.New("library executor observation needs an instance")
+	}
+	observedAt := observation.ObservedAt
+	if observedAt.IsZero() {
+		observedAt = now
+	}
+	_, err = s.db.Exec(ctx, `
+INSERT INTO library_executor_status (instance, reachable, busy, degraded, identity, status_error, observer, observed_at)
+VALUES ($1,$2,$3,$4,$5::jsonb,$6,$7,$8)
+ON CONFLICT (instance) DO UPDATE SET
+    reachable = EXCLUDED.reachable,
+    busy = EXCLUDED.busy,
+    degraded = EXCLUDED.degraded,
+    identity = EXCLUDED.identity,
+    status_error = EXCLUDED.status_error,
+    observer = EXCLUDED.observer,
+    observed_at = EXCLUDED.observed_at`,
+		observation.Instance, observation.Reachable, observation.Busy, observation.Degraded,
+		string(identity), observation.Error, observer, observedAt.UTC())
+	if err != nil {
+		return fmt.Errorf("record library executor status: %w", err)
+	}
+	return nil
+}
+
+// LibraryExecutorStatus 读某个实例最近一次观测；没有记录返回 nil（不是"可用"）。
+func (s *Store) LibraryExecutorStatus(ctx context.Context, instance string) (*libraryexecutor.Observation, error) {
+	row := s.db.QueryRow(ctx, `SELECT `+executorStatusColumns+` FROM library_executor_status WHERE instance=$1`, instance)
+	var observation libraryexecutor.Observation
+	var raw []byte
+	err := row.Scan(&observation.Instance, &observation.Reachable, &observation.Busy, &observation.Degraded,
+		&raw, &observation.Error, &observation.ObservedAt)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, fmt.Errorf("read library executor status: %w", err)
+	}
+	if len(raw) > 0 {
+		if err := json.Unmarshal(raw, &observation.Identity); err != nil {
+			return nil, fmt.Errorf("library executor identity unreadable: %w", err)
+		}
+	}
+	return &observation, nil
+}

internal/platform/postgres/library_executor_status_integration_test.go

@@ -0,0 +1,70 @@
+package postgres
+
+import (
+	"context"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"knowledge-governance/internal/libraryexecutor"
+)
+
+// 执行器状态是"最新一次观测"，不是"最后一条好消息"：
+// 降级与不可达必须覆盖上一次的健康记录，否则页面会一直显示可发起。
+func TestLibraryExecutorStatusKeepsTheLatestObservation(t *testing.T) {
+	databaseURL := isolatedIntegrationDatabaseURL(t, os.Getenv("TEST_DATABASE_URL"))
+	if databaseURL == "" {
+		t.Skip("TEST_DATABASE_URL is not set")
+	}
+	ctx := context.Background()
+	pool, err := pgxpool.New(ctx, databaseURL)
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer pool.Close()
+	if err := ApplyMigrations(ctx, pool); err != nil {
+		t.Fatalf("migrate: %v", err)
+	}
+	store := NewStore(pool)
+	now := time.Date(2026, 10, 29, 8, 0, 0, 0, time.UTC)
+
+	if missing, err := store.LibraryExecutorStatus(ctx, "kgmaint-weknora"); err != nil || missing != nil {
+		t.Fatalf("missing row = %+v, %v", missing, err)
+	}
+
+	healthy := libraryexecutor.Observation{
+		Instance: "kgmaint-weknora", Reachable: true, ObservedAt: now,
+		Identity: libraryexecutor.Identity{Project: "kgmaint-weknora", SourceApp: "kgmaint-app", ArchiveRoot: "/tmp/kg-e2e/archive"},
+	}
+	if err := store.UpsertLibraryExecutorStatus(ctx, healthy, "worker-1", now); err != nil {
+		t.Fatalf("upsert healthy: %v", err)
+	}
+	read, err := store.LibraryExecutorStatus(ctx, "kgmaint-weknora")
+	if err != nil {
+		t.Fatalf("read healthy: %v", err)
+	}
+	if read == nil || !read.Reachable || read.Identity.Project != "kgmaint-weknora" || read.Identity.SourceApp != "kgmaint-app" {
+		t.Fatalf("healthy row = %+v", read)
+	}
+	if !read.ObservedAt.Equal(now) {
+		t.Fatalf("observed_at = %v, want %v", read.ObservedAt, now)
+	}
+
+	degraded := healthy
+	degraded.Reachable = false
+	degraded.ObservedAt = now.Add(30 * time.Second)
+	degraded.Degraded = ""
+	degraded.Error = "library agent is unreachable"
+	if err := store.UpsertLibraryExecutorStatus(ctx, degraded, "worker-1", degraded.ObservedAt); err != nil {
+		t.Fatalf("upsert unreachable: %v", err)
+	}
+	read, err = store.LibraryExecutorStatus(ctx, "kgmaint-weknora")
+	if err != nil || read == nil {
+		t.Fatalf("read after unreachable: %+v %v", read, err)
+	}
+	if read.Reachable || read.Error != "library agent is unreachable" || !read.ObservedAt.Equal(degraded.ObservedAt) {
+		t.Fatalf("row kept a stale good state: %+v", read)
+	}
+}

internal/platform/postgres/migrations/000074_library_executor_status.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS library_executor_status;

internal/platform/postgres/migrations/000074_library_executor_status.up.sql

@@ -0,0 +1,16 @@
+-- 封存执行器的运行时状态：页面上的"可以发起封存"必须来自真实观测，而不是
+-- "配置里写了 socket 路径"。正式栈曾出现过 agent 顶着正式单元名却在冻结候选实例：
+-- 配置非空、页面显示可发起、真发起却封错了库。
+--
+-- worker 探测 agent 的 health（不跑子进程、不碰源库），把"它指向哪个实例、
+-- 是否降级/正在跑、什么时候观测的"落库；API 只读这张表并核对身份，不再猜。
+CREATE TABLE IF NOT EXISTS library_executor_status (
+    instance        VARCHAR(128) PRIMARY KEY,
+    reachable       BOOLEAN      NOT NULL DEFAULT FALSE,
+    busy            BOOLEAN      NOT NULL DEFAULT FALSE,
+    degraded        TEXT         NOT NULL DEFAULT '',
+    identity        JSONB        NOT NULL DEFAULT '{}'::jsonb,
+    status_error    TEXT         NOT NULL DEFAULT '',
+    observer        VARCHAR(64)  NOT NULL DEFAULT '',
+    observed_at     TIMESTAMPTZ  NOT NULL
+);

internal/platform/postgres/migrations_test.go

@@ -11,40 +11,45 @@ func TestEmbeddedMigrationsAreVersionSorted(t *testing.T) {
 	if err != nil {
 		t.Fatalf("loadMigrations() error = %v", err)
 	}
-	if len(migrations) != 73 {
-		t.Fatalf("migration count = %d, want 73", len(migrations))
+	if len(migrations) != 74 {
+		t.Fatalf("migration count = %d, want 74", len(migrations))
+	}
+	// 74 记录封存执行器的运行时状态：页面可用性来自真实观测，不来自配置字符串。
+	if executor := migrations[len(migrations)-1]; executor.Version != 74 ||
+		!strings.Contains(executor.UpSQL, "library_executor_status") {
+		t.Fatalf("migration 74 = %#v", executor)
 	}
 	// 73 记录死信精确解除：保留上游行，只绑定指纹、重放引用与观测摘要。
-	if release := migrations[len(migrations)-1]; release.Version != 73 ||
+	if release := migrations[len(migrations)-2]; release.Version != 73 ||
 		!strings.Contains(release.UpSQL, "library_dead_letter_releases") {
 		t.Fatalf("migration 73 = %#v", release)
 	}
 	// 72 让维护占用按实例互斥（internal/external 不再各有各的锁）。
-	if instance := migrations[len(migrations)-2]; instance.Version != 72 ||
+	if instance := migrations[len(migrations)-3]; instance.Version != 72 ||
 		!strings.Contains(instance.UpSQL, "library_instance_maintenance") {
 		t.Fatalf("migration 72 = %#v", instance)
 	}
 	// 71 让维护记录能区分实例级维护与仅治理库互斥。
-	if level := migrations[len(migrations)-3]; level.Version != 71 ||
+	if level := migrations[len(migrations)-4]; level.Version != 71 ||
 		!strings.Contains(level.UpSQL, "level") || !strings.Contains(level.UpSQL, "ingress_state") {
 		t.Fatalf("migration 71 = %#v", level)
 	}
-	if maint := migrations[len(migrations)-4]; maint.Version != 70 || !strings.Contains(maint.UpSQL, "library_maintenance") {
+	if maint := migrations[len(migrations)-5]; maint.Version != 70 || !strings.Contains(maint.UpSQL, "library_maintenance") {
 		t.Fatalf("migration 70 = %#v", maint)
 	}
-	if tail := migrations[len(migrations)-5]; tail.Version != 69 ||
+	if tail := migrations[len(migrations)-6]; tail.Version != 69 ||
 		!strings.Contains(tail.UpSQL, "library_jobs_write_target") {
 		t.Fatalf("migration 69 = %#v", tail)
 	}
-	if restore := migrations[len(migrations)-6]; restore.Version != 68 || !strings.Contains(restore.UpSQL, "restore") {
+	if restore := migrations[len(migrations)-7]; restore.Version != 68 || !strings.Contains(restore.UpSQL, "restore") {
 		t.Fatalf("migration 68 = %#v", restore)
 	}
-	if last := migrations[len(migrations)-7]; last.Version != 67 ||
+	if last := migrations[len(migrations)-8]; last.Version != 67 ||
 		!strings.Contains(last.UpSQL, "library_release_authorizations") ||
 		!strings.Contains(last.UpSQL, "params_digest") {
 		t.Fatalf("migration 67 = %#v", last)
 	}
-	if flow := migrations[len(migrations)-8]; flow.Version != 66 ||
+	if flow := migrations[len(migrations)-9]; flow.Version != 66 ||
 		!strings.Contains(flow.UpSQL, "CREATE TABLE library_versions") ||
 		!strings.Contains(flow.UpSQL, "library_jobs_active") {
 		t.Fatalf("migration 66 = %#v", flow)