diff --git a/CHANGELOG.md b/CHANGELOG.md index 75e6e1a..4321002 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,18 @@ # Changelog +## 开发快照 snapshot/v0.1.111 - 2026-10-28 + +### Added + +- 主机侧封存 agent（`libraryagent` + systemd user 服务）：只监听一个 0700 目录里的 0600 unix socket，校验 `SO_PEERCRED` 同 uid，固定 binary/归档/日志/项目/写入者声明，请求只能选范围与作业号。 +- 页面「发起封存」：`POST /api/v1/library/captures` 只入队 `library.capture` 作业，`GET /api/v1/library/captures` 读队列真实状态；未接线时 503 并给出原因，权限不足 403。 +- worker `library.capture` 处理器：通过挂载的 socket 调 agent，启动时先做一次 `recover`；容器内**不挂** Docker socket、**不持有**来源凭据。 +- `scripts/library-agent-install.sh`：构建、安装、启用 linger 并读回 socket 权限与服务状态。 + +### Changed + +- 整库版本页显示封存作业的真实状态（待执行/执行中/重试/完成/失败与最后错误），完成后自动刷新快照列表。 + ## 开发快照 snapshot/v0.1.110 - 2026-10-28 ### Fixed diff --git a/api/openapi.yaml b/api/openapi.yaml index c63ab08..1739157 100644 --- a/api/openapi.yaml +++ b/api/openapi.yaml @@ -832,6 +832,72 @@ } } }, + "/api/v1/library/captures": { + "get": { + "tags": [ + "Runtime" + ], + "operationId": "listLibraryCaptures", + "summary": "列出最近的整库封存作业（读队列，不是归档；页面据此显示真实状态）", + "responses": { + "200": { + "$ref": "#/components/responses/Envelope" + }, + "400": { + "$ref": "#/components/responses/BadRequest" + }, + "500": { + "$ref": "#/components/responses/InternalError" + } + } + }, + "post": { + "tags": [ + "Runtime" + ], + "operationId": "enqueueLibraryCapture", + "summary": "投递一次整库封存作业（需要工程师及以上；执行在主机侧 agent，容器没有 Docker/来源凭据）", + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "scope": { + "type": "string", + "enum": [ + "internal", + "external" + ] + }, + "job": { + "type": "string" + } + }, + "required": [ + "scope" + ] + } + } + } + }, + "responses": { + "202": { + "description": "已入队封存作业（同一 scope+job 复用同一作业）" + }, + "400": { + "$ref": "#/components/responses/BadRequest" + }, + "500": { + "$ref": "#/components/responses/InternalError" + }, + "503": { + "description": "主机侧封存 agent 未接线" + } + } + } + }, "/api/v1/library/snapshots/{id}": { "get": { "tags": [ diff --git a/cmd/api/main.go b/cmd/api/main.go index 9b88bd6..a41db1b 100644 --- a/cmd/api/main.go +++ b/cmd/api/main.go @@ -202,6 +202,7 @@ func run(ctx context.Context, getenv func(string) string) error { VersionHistory: versionHistoryStore, VersionHistoryRepo: versionHistoryRepo, LibraryArchiveRoot: config.LibraryArchiveRoot, + LibraryAgentSocket: config.LibraryAgentSocket, BackupHost: backuphost.New(backuphost.Config{ SSHConfig: config.BackupHostSSHConfig, Target: config.BackupHostTarget, diff --git a/cmd/libraryagent/main.go b/cmd/libraryagent/main.go new file mode 100644 index 0000000..2aa4c0d --- /dev/null +++ b/cmd/libraryagent/main.go @@ -0,0 +1,70 @@ +// Command libraryagent is the host-side execution boundary for whole-library +// capture. It runs as the ordinary operator account, owns no network listener +// other than one 0600 unix socket, and executes exactly one fixed command shape. +package main + +import ( + "context" + "flag" + "log" + "os" + "os/signal" + "path/filepath" + "syscall" + "time" + + "knowledge-governance/internal/libraryagent" +) + +func main() { + home, _ := os.UserHomeDir() + var config libraryagent.Config + flag.StringVar(&config.Socket, "socket", filepath.Join(home, ".local/share/knowledge-governance/agent.sock"), "unix socket for capture requests") + flag.StringVar(&config.Binary, "binary", "/home/zheromiku/knowledge-governance/libraryctl", "libraryctl binary this agent executes") + flag.StringVar(&config.ArchiveRoot, "archive", filepath.Join(home, ".local/share/knowledge-governance/library"), "sealed artifact root") + flag.StringVar(&config.JournalRoot, "journal", filepath.Join(home, ".local/share/knowledge-governance/freeze"), "freeze journal root") + flag.StringVar(&config.SourceApp, "source-app", "WKNFull-app", "source container describing the stack") + flag.StringVar(&config.SourcePG, "source-pg", "WKNFull-postgres", "source PostgreSQL container") + flag.StringVar(&config.SourceES, "source-es", "WKNFull-elasticsearch", "source Elasticsearch container") + flag.StringVar(&config.GovernancePG, "governance-pg", "knowledge-governance-p1-postgres-1", "governance PostgreSQL container") + flag.StringVar(&config.Project, "project", "weknora071full", "Compose project owning the writers") + flag.StringVar(&config.Writers, "writers", "app", "declared writer services") + nonWriters := flag.String("non-writers", defaultNonWriters, "service=reason for every other project service") + flag.DurationVar(&config.MaxDuration, "max-duration", 10*time.Minute, "upper bound for one operation") + describe := flag.Bool("describe", false, "print the fixed configuration and exit") + flag.Parse() + config.NonWriters = *nonWriters + + agent, err := libraryagent.New(config) + if err != nil { + log.Fatal(err) + } + if *describe { + log.Printf("library agent config: %+v", agent.Describe()) + return + } + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + log.Printf("library agent listening on %s (project %s, writers %s)", config.Socket, config.Project, config.Writers) + if err = agent.Serve(ctx); err != nil { + log.Fatal(err) + } +} + +// defaultNonWriters mirrors scripts/library-capture.sh: every sibling service of +// the source project is declared, with the reason it cannot write knowledge. +const defaultNonWriters = "frontend=UI only; no knowledge writes" + + ",mcp=read-only MCP server" + + ",dobotai-target=alpine placeholder; no knowledge writes" + + ",docreader=parse RPC service; app persists parse output and settlement requires zero pending ops" + + ",elasticsearch=search store; not an initiator" + + ",postgres=database store; not an initiator" + + ",redis=queue/cache store; not an initiator" + + ",neo4j=graph store; not an initiator" + + ",langfuse-web=LLM observability UI" + + ",langfuse-worker=LLM observability worker; own storage" + + ",langfuse-clickhouse=observability store" + + ",langfuse-minio=observability object store" + + ",langfuse-db-init=one-shot init; exited" + + ",searxng=read-only search proxy" + + ",searxng-init=one-shot init; exited" diff --git a/cmd/worker/main.go b/cmd/worker/main.go index a7b2444..bb7e53d 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -9,12 +9,14 @@ import ( "os/signal" "path/filepath" "syscall" + "time" "github.com/jackc/pgx/v5/pgxpool" "knowledge-governance/internal/appconfig" "knowledge-governance/internal/governance" "knowledge-governance/internal/jobs" + "knowledge-governance/internal/libraryagent" postgresstore "knowledge-governance/internal/platform/postgres" "knowledge-governance/internal/platform/weknora" "knowledge-governance/internal/versionhistory" @@ -24,6 +26,9 @@ import ( "knowledge-governance/internal/workflows/testweknora" ) +// libraryCaptureJobType is shared with the API which enqueues it. +const libraryCaptureJobType = "library.capture" + func main() { ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer stop() @@ -126,6 +131,40 @@ func workerHandlers(store *postgresstore.Store, config appconfig.Config) (map[st return nil, fmt.Errorf("create artifact fetch handler: %w", err) } handlers["artifact.fetch"] = artifactHandler.Handle + if config.LibraryAgentSocket != "" { + // 整库封存：worker 只投递请求并记录结果，真正的停机/导出/恢复发生在主机侧 + // agent（同 uid unix socket）。容器里既没有 Docker socket，也没有来源凭据。 + agentConfig := libraryagent.ClientConfig{Socket: config.LibraryAgentSocket, Timeout: 15 * time.Minute} + handlers[libraryCaptureJobType] = func(ctx context.Context, job jobs.Job) error { + var payload struct { + Scope string `json:"scope"` + CaptureJob string `json:"capture_job"` + RequestedBy string `json:"requested_by"` + } + if err := json.Unmarshal(job.Payload, &payload); err != nil || (payload.Scope != "internal" && payload.Scope != "external") || payload.CaptureJob == "" { + return fmt.Errorf("invalid library capture payload") + } + result, err := libraryagent.Capture(ctx, agentConfig, payload.Scope, payload.CaptureJob) + if err != nil { + log.Printf("library capture failed scope=%s capture_job=%s: %v", payload.Scope, payload.CaptureJob, err) + return err + } + log.Printf("library capture completed scope=%s capture_job=%s snapshot=%s replay=%v requested_by=%s", + payload.Scope, payload.CaptureJob, result.SnapshotID, result.Replay, payload.RequestedBy) + return nil + } + // A crash can leave the source writer stopped: recover before taking work, + // and keep the failure visible instead of pretending the worker is ready. + go func() { + recoveryCtx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) + defer cancel() + if _, err := libraryagent.Recover(recoveryCtx, agentConfig); err != nil { + log.Printf("library agent recovery check failed: %v", err) + return + } + log.Printf("library agent reachable; freeze journals recovered") + }() + } if config.VersionHistoryGiteaBaseURL != "" { // 「版本记录」同步：匿名只读远端仓库的提交/tag，写本库快照。未配置来源就不注册， // 界面上的同步按钮也会显示"未接线"，不会入队一个没人处理的作业。 diff --git a/compose.yaml b/compose.yaml index d658536..151eb69 100644 --- a/compose.yaml +++ b/compose.yaml @@ -52,6 +52,9 @@ services: # 「整库版本」页：只读已封存的整库候选产物。归档根由 `libraryctl capture` # 在主机上写入（私有 0700），这里只挂只读；留空 = 未接线（页面降级展示原因）。 LIBRARY_ARCHIVE_ROOT: "${LIBRARY_ARCHIVE_ROOT:-/var/lib/knowledge-governance/library}" + # 主机侧封存 agent 的 unix socket：api 只用它判断"能不能从页面发起"， + # worker 用它投递作业。容器里没有 Docker socket，也没有来源凭据。 + LIBRARY_AGENT_SOCKET: "${LIBRARY_AGENT_SOCKET:-/run/library-agent/agent.sock}" # Mount this independent master-key file read-only using the deployment overlay. QA_AI_MASTER_KEY_FILE: "${QA_AI_MASTER_KEY_FILE:-}" QA_AI_DNS_OVER_TLS: "${QA_AI_DNS_OVER_TLS:-}" @@ -141,6 +144,9 @@ services: volumes: - ./spool:/var/lib/governance/spool:ro - ./spool/ingestion:/var/lib/governance/ingestion + # 整库封存 agent 的 socket 目录（由主机侧 systemd user 服务持有）。 + # 容器只拿到"投递作业"这一条通道：Docker socket 与来源凭据都不在这里。 + - /home/zheromiku/.local/share/knowledge-governance/agent:/run/library-agent - "${PP_SCAN_HOST_DATA_ROOT:-/home/zheromiku/dobot-portal-sync/data}:${PP_SCAN_CONTAINER_DATA_ROOT:-/home/zheromiku/dobot-portal-sync/data}:ro" - /home/zheromiku/dingtalk_kb_download:/sources/dingtalk:ro # dws 凭据：与宿主共享同一份 state（refresh token 是轮换制，不能在两处各留一份） diff --git a/docs/library-capture.md b/docs/library-capture.md index 5260895..f3e0979 100644 --- a/docs/library-capture.md +++ b/docs/library-capture.md @@ -45,7 +45,24 @@ 同一个作业号重跑是**幂等重放**：直接返回已封存的同一份候选（`replay=true`、`source_read=false`、`writers_stopped=false`），不读来源、不停写入者；同一作业号换了 scope 或绑定库则明确拒绝。 -页面侧的发起入口尚未接线（需要最小权限的主机执行边界，见进度文件），当前只能在目标主机执行命令。 +## 从页面发起封存（主机侧 agent） + +页面点「发起对内/对外封存」只是**投递一个作业**；执行发生在这台主机上的 agent： + +- `scripts/library-agent-install.sh` 构建并安装 `libraryagent`（systemd **user** 服务）， + 监听 `~/.local/share/knowledge-governance/agent/agent.sock`，目录 0700、socket 0600， + 并额外用 `SO_PEERCRED` 要求对端与 agent 同 uid。 +- agent 自己固定 `libraryctl` 的路径、归档与日志根、Compose 项目与写入者声明； + 请求只能选 `scope`（internal/external）和作业号，**不能**指定路径、项目、容器或命令。 +- 只有 worker 通过挂载的 socket 目录投递作业（`LIBRARY_AGENT_SOCKET`）。容器里 + **没有** Docker socket，**没有**来源凭据：停写入者、读来源、恢复都在主机进程里完成。 +- worker 启动时会先调一次 `recover`，把上次崩溃留下的停机窗口收尾； + 失败会记日志而不是假装就绪。 +- 同一 `scope + 作业号` 重复投递时，agent 走幂等重放，不会第二次停源。 + +安全边界要说清：socket 目录对 worker 可写，因此被攻破的 worker 可以让 agent 执行 +"封存 internal/external"这一类动作（这正是页面能力），但拿不到 Docker API、 +拿不到来源凭据，也不能重定向到别的项目或路径。 ## 2026-10-28 实测（本机测试栈） diff --git a/internal/appconfig/config.go b/internal/appconfig/config.go index e40a20c..ee44bc8 100644 --- a/internal/appconfig/config.go +++ b/internal/appconfig/config.go @@ -71,7 +71,10 @@ type Config struct { // 留空即与抓取 origin 相同。只用于拼展示链接，不参与任何请求。 // LibraryArchiveRoot 是整库封存产物的只读归档根（主机私有 0700 目录的只读挂载）。 // 留空即"未接线"：页面只显示原因，不显示空表冒充没有快照。 - LibraryArchiveRoot string + LibraryArchiveRoot string + // LibraryAgentSocket 是主机侧封存 agent 的 unix socket（只由 worker 连接）。 + // 留空即"未接线"：页面不显示可用的封存入口，接口返回原因而不是假装已排队。 + LibraryAgentSocket string VersionHistoryLinkOrigin string VersionHistoryHTTPTimeout time.Duration VersionHistoryMaxPages int @@ -120,6 +123,7 @@ func Load(getenv func(string) string) (Config, error) { VersionHistoryRepo: valueOrDefault(strings.TrimSpace(getenv("VERSION_HISTORY_REPO")), "Weknora/knowledge-governance"), VersionHistoryAllowedOrigins: splitOrigins(getenv("VERSION_HISTORY_ALLOWED_ORIGINS")), LibraryArchiveRoot: strings.TrimSpace(getenv("LIBRARY_ARCHIVE_ROOT")), + LibraryAgentSocket: strings.TrimSpace(getenv("LIBRARY_AGENT_SOCKET")), VersionHistoryLinkOrigin: strings.TrimSpace(getenv("VERSION_HISTORY_LINK_ORIGIN")), VersionHistoryHTTPTimeout: 15 * time.Second, VersionHistoryMaxPages: valueOrDefaultInt(getenv("VERSION_HISTORY_MAX_PAGES"), 200), @@ -233,6 +237,9 @@ func Load(getenv func(string) string) (Config, error) { if config.LibraryArchiveRoot != "" && !filepath.IsAbs(config.LibraryArchiveRoot) { return Config{}, errors.New("LIBRARY_ARCHIVE_ROOT 必须是绝对路径（归档根按只读目录挂载）") } + if config.LibraryAgentSocket != "" && !filepath.IsAbs(config.LibraryAgentSocket) { + return Config{}, errors.New("LIBRARY_AGENT_SOCKET 必须是绝对路径（unix socket 由主机侧 agent 持有）") + } if config.TestWeKnoraHTTPTimeout, err = durationOrDefault(getenv("TEST_WEKNORA_HTTP_TIMEOUT"), config.TestWeKnoraHTTPTimeout); err != nil { return Config{}, fmt.Errorf("TEST_WEKNORA_HTTP_TIMEOUT: %w", err) } diff --git a/internal/httpapi/library_capture.go b/internal/httpapi/library_capture.go new file mode 100644 index 0000000..b2c6cbc --- /dev/null +++ b/internal/httpapi/library_capture.go @@ -0,0 +1,106 @@ +package httpapi + +import ( + "encoding/json" + "fmt" + "net/http" + "regexp" + "strings" + "time" + + "knowledge-governance/internal/jobs" +) + +// 页面发起封存：只投递一个作业，真正的停机/导出/恢复发生在主机侧 agent。 +// 这里没有任何 Docker 访问，也没有来源凭据 —— 调用方只能选择范围与作业号。 +const libraryCaptureJobType = "library.capture" + +var libraryCaptureJobPattern = regexp.MustCompile(`^[A-Za-z0-9_-]{1,64}$`) + +type libraryCaptureRequest struct { + Scope string `json:"scope"` + Job string `json:"job"` +} + +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"}) + return + } + if request.Scope != "internal" && request.Scope != "external" { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "scope must be internal or external"}) + return + } + // The job id is the idempotency key handed to the agent: repeating the same + // scope+job returns the already sealed version instead of sealing twice. + captureJob := strings.TrimSpace(request.Job) + if captureJob == "" { + captureJob = "lib-" + request.Scope + "-" + time.Now().UTC().Format("20060102T150405") + } + if !libraryCaptureJobPattern.MatchString(captureJob) { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid job id"}) + return + } + requestedBy := "" + if identity, ok := sessionFrom(r.Context()); ok { + requestedBy = identity.Username + } + payload, err := json.Marshal(map[string]string{ + "scope": request.Scope, "capture_job": captureJob, "requested_by": requestedBy, + }) + if err != nil { + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot build capture job"}) + return + } + now := time.Now().UTC() + job, err := config.Store.Enqueue(r.Context(), jobs.EnqueueRequest{ + ID: fmt.Sprintf("library-capture-%s-%d", request.Scope, now.UnixNano()), + Type: libraryCaptureJobType, + Payload: payload, + AvailableAt: now, + MaxAttempts: 2, + IdempotencyKey: "library.capture:" + captureJob, + }, now) + if err != nil || job == nil { + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot enqueue capture job"}) + return + } + writeJSON(w, http.StatusAccepted, map[string]any{ + "job_id": job.ID, "capture_job": captureJob, "scope": request.Scope, "state": job.State, + }) +} + +// libraryCaptureJobs lists recent capture jobs so the page can show real state +// instead of an optimistic spinner. It reads the queue, not the archive. +func libraryCaptureJobs(w http.ResponseWriter, r *http.Request, config Config) { + queue, err := config.Store.List(r.Context(), 200) + if err != nil { + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot read capture jobs"}) + return + } + views := make([]map[string]any, 0, 20) + for _, job := range queue { + if job.Type != libraryCaptureJobType || len(views) >= 20 { + continue + } + var payload struct { + Scope string `json:"scope"` + CaptureJob string `json:"capture_job"` + RequestedBy string `json:"requested_by"` + } + _ = json.Unmarshal(job.Payload, &payload) + views = append(views, map[string]any{"job_id": job.ID, "capture_job": payload.CaptureJob, + "scope": payload.Scope, "state": job.State, "attempt": job.Attempt, + "last_error": job.LastError, "requested_by": payload.RequestedBy, + "created_at": job.CreatedAt, "updated_at": job.UpdatedAt}) + } + writeJSON(w, http.StatusOK, map[string]any{"available": true, "jobs": views}) +} diff --git a/internal/httpapi/library_capture_test.go b/internal/httpapi/library_capture_test.go new file mode 100644 index 0000000..c919af9 --- /dev/null +++ b/internal/httpapi/library_capture_test.go @@ -0,0 +1,111 @@ +package httpapi + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "knowledge-governance/internal/jobs" +) + +func captureRequest(body string) *http.Request { + request := httptest.NewRequest(http.MethodPost, "/api/v1/library/captures", strings.NewReader(body)) + request.Header.Set("Content-Type", "application/json") + 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"}) + if recorder.Code != http.StatusAccepted { + t.Fatalf("status=%d body=%s", recorder.Code, recorder.Body.String()) + } + enqueued := store.enqueued + if enqueued.Type != libraryCaptureJobType { + t.Fatalf("job type %q", enqueued.Type) + } + if enqueued.IdempotencyKey != "library.capture:lib-external-review" { + t.Fatalf("idempotency key %q", enqueued.IdempotencyKey) + } + var payload map[string]string + if err := json.Unmarshal(enqueued.Payload, &payload); err != nil { + t.Fatal(err) + } + if payload["scope"] != "external" || payload["capture_job"] != "lib-external-review" { + t.Fatalf("payload %v", payload) + } + var body map[string]any + if err := json.Unmarshal(recorder.Body.Bytes(), &body); err != nil { + t.Fatal(err) + } + if body["capture_job"] != "lib-external-review" || body["scope"] != "external" { + t.Fatalf("response %v", body) + } +} + +func TestLibraryCaptureRejectsInvalidRequest(t *testing.T) { + for _, body := range []string{ + `{"scope":"production"}`, + `{"scope":"internal","job":"bad job; rm -rf /"}`, + `not json`, + } { + store := &fakeJobStore{} + recorder := httptest.NewRecorder() + libraryCaptureEnqueue(recorder, captureRequest(body), Config{Store: store, LibraryAgentSocket: "/tmp/agent.sock"}) + if recorder.Code != http.StatusBadRequest { + t.Fatalf("body %q status=%d", body, recorder.Code) + } + if store.enqueued.Type != "" { + t.Fatalf("invalid request %q still enqueued", body) + } + } +} + +func TestLibraryCaptureJobsListsOnlyCaptureJobs(t *testing.T) { + store := &fakeJobStore{listResult: []jobs.Job{ + {ID: "a", Type: libraryCaptureJobType, State: "completed", + Payload: json.RawMessage(`{"scope":"internal","capture_job":"lib-internal-1","requested_by":"yaozhanghao"}`), + CreatedAt: time.Date(2026, 10, 28, 3, 0, 0, 0, time.UTC)}, + {ID: "b", Type: "artifact.fetch", State: "completed"}, + }} + recorder := httptest.NewRecorder() + libraryCaptureJobs(recorder, httptest.NewRequest(http.MethodGet, "/api/v1/library/captures", nil), Config{Store: store}) + if recorder.Code != http.StatusOK { + t.Fatalf("status=%d", recorder.Code) + } + var body struct { + Available bool `json:"available"` + Jobs []map[string]any `json:"jobs"` + } + if err := json.Unmarshal(recorder.Body.Bytes(), &body); err != nil { + t.Fatal(err) + } + if !body.Available || len(body.Jobs) != 1 || body.Jobs[0]["capture_job"] != "lib-internal-1" { + t.Fatalf("unexpected body %+v", body) + } +} diff --git a/internal/httpapi/library_snapshots.go b/internal/httpapi/library_snapshots.go index b9ce39d..7f55132 100644 --- a/internal/httpapi/library_snapshots.go +++ b/internal/httpapi/library_snapshots.go @@ -58,14 +58,20 @@ type librarySnapshotList struct { UnreferencedBytes int64 `json:"unreferenced_bytes"` ManifestFileBytes int64 `json:"manifest_file_bytes"` ManifestAllocatedBytes int64 `json:"manifest_allocated_bytes"` - // 页面必须能说清"封存由谁执行"，不能让人以为页面上有个没实现的按钮。 - CaptureNote string `json:"capture_note"` + // 页面必须能说清"封存由谁执行"；能不能从这里发起，取决于主机侧 agent 是否接线。 + CaptureNote string `json:"capture_note"` + CaptureAvailable bool `json:"capture_available"` + CaptureReason string `json:"capture_reason,omitempty"` } const libraryCaptureNote = "封存由运维命令 scripts/library-capture.sh capture -scope internal|external 在目标主机执行（停写入者→导出→恢复）。页面只读展示已封存的完整版本：它通过了数据完整性检查（身份、双向正文覆盖、生成问题、资源与摘要），这已满足封版所需的完整性口径；授权与生产同步仍是独立步骤。" func librarySnapshotsList(w http.ResponseWriter, r *http.Request, config Config) { view, err := readLibrarySnapshots(config.LibraryArchiveRoot) + view.CaptureAvailable = config.LibraryAgentSocket != "" + if !view.CaptureAvailable { + view.CaptureReason = "未接线：主机侧封存 agent 没有挂载，页面只能查看已封存版本" + } if err != nil { writeJSON(w, http.StatusServiceUnavailable, librarySnapshotList{Available: false, Reason: "归档不可读：" + err.Error(), Snapshots: []librarySnapshotView{}, CaptureNote: libraryCaptureNote}) @@ -110,6 +116,7 @@ func librarySnapshotDetail(w http.ResponseWriter, r *http.Request, config Config func readLibrarySnapshots(root string) (librarySnapshotList, error) { view := librarySnapshotList{Snapshots: []librarySnapshotView{}, CaptureNote: libraryCaptureNote} + _ = root if root == "" { view.Reason = "未接线：LIBRARY_ARCHIVE_ROOT 未配置，本平台没有整库归档可读" return view, nil diff --git a/internal/httpapi/server.go b/internal/httpapi/server.go index 1d3211b..2913cc5 100644 --- a/internal/httpapi/server.go +++ b/internal/httpapi/server.go @@ -130,7 +130,9 @@ type Config struct { VersionHistory VersionHistoryStore VersionHistoryRepo string // LibraryArchiveRoot 是整库封存产物的只读归档根；留空即未接线（页面降级）。 - LibraryArchiveRoot string + LibraryArchiveRoot string + // LibraryAgentSocket 是主机侧封存 agent 的 unix socket；留空即页面不能发起封存。 + LibraryAgentSocket string BackupSchedule BackupScheduleStore BackupHost BackupHost BackupRuns *BackupRunTracker @@ -396,6 +398,13 @@ func NewHandler(config Config) (http.Handler, error) { mux.HandleFunc("GET /api/v1/library/snapshots/{id}", func(w http.ResponseWriter, r *http.Request) { librarySnapshotDetail(w, r, config) }) + // 页面发起封存：只投递作业（主机侧 agent 执行），并回读作业真实状态。 + mux.HandleFunc("POST /api/v1/library/captures", func(w http.ResponseWriter, r *http.Request) { + libraryCaptureEnqueue(w, r, config) + }) + mux.HandleFunc("GET /api/v1/library/captures", func(w http.ResponseWriter, r *http.Request) { + libraryCaptureJobs(w, r, config) + }) // 全部 main 提交时间线（读本库缓存，不直接打 Gitea） mux.HandleFunc("GET /api/v1/version-history", func(w http.ResponseWriter, r *http.Request) { versionHistoryList(w, r, config) diff --git a/internal/libraryagent/agent.go b/internal/libraryagent/agent.go new file mode 100644 index 0000000..6942283 --- /dev/null +++ b/internal/libraryagent/agent.go @@ -0,0 +1,224 @@ +// Package libraryagent is the least-privilege execution boundary for +// whole-library capture on the target host. +// +// Why it exists: capturing a library has to stop the source writer for a few +// seconds and read source credentials. Handing the governance containers the +// Docker socket (a full write API) or the source credentials would be a much +// larger privilege than that operation needs. Instead this agent runs as the +// ordinary operator account on the host, listens on one unix socket which only +// that account can connect to, and executes one fixed command with a fixed +// writer declaration. A caller can choose the scope and (optionally) a job id — +// nothing else. +package libraryagent + +import ( + "bufio" + "context" + "encoding/json" + "errors" + "net" + "os" + "os/exec" + "path/filepath" + "regexp" + "strings" + "sync" + "time" +) + +// Config is fixed at start-up by the operator (systemd unit / CLI flags). It is +// never taken from a request. +type Config struct { + Socket string + Binary string + ArchiveRoot string + JournalRoot string + // Writer declaration passed to the binary, exactly as reviewed in + // scripts/library-capture.sh. Clients cannot influence it. + SourceApp string + SourcePG string + SourceES string + GovernancePG string + Project string + Writers string + NonWriters string + MaxDuration time.Duration + // Run executes the binary; replaced in tests. + Run func(ctx context.Context, name string, args ...string) ([]byte, error) +} + +type request struct { + Action string `json:"action"` + Scope string `json:"scope,omitempty"` + Job string `json:"job,omitempty"` +} + +type response struct { + OK bool `json:"ok"` + Action string `json:"action"` + Scope string `json:"scope,omitempty"` + Job string `json:"job,omitempty"` + Error string `json:"error,omitempty"` + Payload json.RawMessage `json:"payload,omitempty"` +} + +var jobPattern = regexp.MustCompile(`^[A-Za-z0-9_-]{1,64}$`) + +// Agent serves capture requests. Only one operation runs at a time: a library +// capture owns the writer group, so overlapping requests would race the window. +type Agent struct { + config Config + mu sync.Mutex +} + +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") + } + if config.Binary == "" || config.Project == "" || config.Writers == "" || !strings.Contains(config.NonWriters, "=") { + return nil, errors.New("library agent needs a binary, project, writers and a non-writer declaration") + } + if config.MaxDuration <= 0 || config.MaxDuration > 30*time.Minute { + return nil, errors.New("library agent duration bound is invalid") + } + return &Agent{config: config}, nil +} + +// Serve listens until ctx is cancelled. The socket is created 0600 inside a +// 0700 directory and removed on shutdown. +func (a *Agent) Serve(ctx context.Context) error { + if err := os.MkdirAll(filepath.Dir(a.config.Socket), 0700); err != nil { + return err + } + if err := os.Chmod(filepath.Dir(a.config.Socket), 0700); err != nil { + return err + } + _ = os.Remove(a.config.Socket) + listener, err := net.Listen("unix", a.config.Socket) + if err != nil { + return err + } + defer func() { + listener.Close() + _ = os.Remove(a.config.Socket) + }() + if err := os.Chmod(a.config.Socket, 0600); err != nil { + return err + } + go func() { + <-ctx.Done() + _ = listener.Close() + }() + for { + conn, err := listener.Accept() + if err != nil { + if ctx.Err() != nil { + return nil + } + return err + } + go a.handle(ctx, conn) + } +} + +func (a *Agent) handle(ctx context.Context, conn net.Conn) { + defer conn.Close() + if err := peerAllowed(conn); err != nil { + _ = json.NewEncoder(conn).Encode(response{OK: false, Error: err.Error()}) + return + } + _ = conn.SetDeadline(time.Now().Add(a.config.MaxDuration + 30*time.Second)) + decoder := json.NewDecoder(bufio.NewReader(conn)) + var req request + if err := decoder.Decode(&req); err != nil { + _ = json.NewEncoder(conn).Encode(response{OK: false, Error: "invalid request"}) + return + } + result := a.execute(ctx, req) + _ = json.NewEncoder(conn).Encode(result) +} + +// execute serialises operations and maps a request onto the one fixed command. +func (a *Agent) execute(ctx context.Context, req request) response { + a.mu.Lock() + defer a.mu.Unlock() + out := response{Action: req.Action, Scope: req.Scope, Job: req.Job} + var args []string + switch req.Action { + case "status": + args = []string{"status"} + case "recover": + args = []string{"recover"} + case "capture": + if req.Scope != "internal" && req.Scope != "external" { + out.Error = "scope must be internal or external" + return out + } + if req.Job != "" && !jobPattern.MatchString(req.Job) { + out.Error = "invalid job id" + return out + } + args = []string{"capture", "-scope", req.Scope} + if req.Job != "" { + args = append(args, "-job", req.Job) + } + default: + out.Error = "unsupported action" + return out + } + args = append(args, a.config.declaration()...) + bounded, cancel := context.WithTimeout(ctx, a.config.MaxDuration) + defer cancel() + payload, err := a.config.run(bounded, a.config.Binary, args...) + if err != nil { + out.Error = "operation failed" + if message := strings.TrimSpace(string(payload)); message != "" && len(message) <= 2000 { + out.Error = message + } + return out + } + out.OK = true + if json.Valid(payload) { + out.Payload = json.RawMessage(payload) + } else { + out.Payload = json.RawMessage(`{}`) + } + return out +} + +// declaration pins every path and the writer group so a request cannot redirect +// the operation to another stack, archive or journal. +func (c Config) declaration() []string { + return []string{ + "-source-app", c.SourceApp, + "-source-pg", c.SourcePG, + "-source-es", c.SourceES, + "-governance-pg", c.GovernancePG, + "-project", c.Project, + "-writers", c.Writers, + "-non-writers", c.NonWriters, + "-archive", c.ArchiveRoot, + "-journal", c.JournalRoot, + } +} + +func (c Config) run(ctx context.Context, name string, args ...string) ([]byte, error) { + if c.Run != nil { + return c.Run(ctx, name, args...) + } + command := exec.CommandContext(ctx, name, args...) + command.Env = append(os.Environ(), "LIBRARYCTL_AGENT=1") + return command.CombinedOutput() +} + +// 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", + } +} diff --git a/internal/libraryagent/agent_test.go b/internal/libraryagent/agent_test.go new file mode 100644 index 0000000..cdb6e16 --- /dev/null +++ b/internal/libraryagent/agent_test.go @@ -0,0 +1,171 @@ +package libraryagent + +import ( + "context" + "encoding/json" + "net" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" +) + +func testAgent(t *testing.T) (*Agent, string, func() []string) { + t.Helper() + root, err := os.MkdirTemp(os.TempDir(), "kgag-") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { os.RemoveAll(root) }) + if err = os.Chmod(root, 0700); err != nil { + t.Fatal(err) + } + var mu sync.Mutex + var calls []string + config := Config{ + Socket: filepath.Join(root, "agent.sock"), + Binary: "/bin/echo", + ArchiveRoot: filepath.Join(root, "library"), + JournalRoot: filepath.Join(root, "freeze"), + SourceApp: "WKNFull-app", + SourcePG: "WKNFull-postgres", + SourceES: "WKNFull-elasticsearch", + GovernancePG: "knowledge-governance-p1-postgres-1", + Project: "weknora071full", + Writers: "app", + NonWriters: "frontend=UI only; no knowledge writes", + MaxDuration: time.Minute, + Run: func(_ context.Context, name string, args ...string) ([]byte, error) { + mu.Lock() + calls = append(calls, name+" "+strings.Join(args, " ")) + mu.Unlock() + return []byte(`{"snapshot_id":"abc","state":"succeeded"}`), nil + }, + } + agent, err := New(config) + if err != nil { + t.Fatal(err) + } + return agent, config.Socket, func() []string { + mu.Lock() + defer mu.Unlock() + return append([]string(nil), calls...) + } +} + +func roundTrip(t *testing.T, socket string, payload string) response { + t.Helper() + conn, err := net.Dial("unix", socket) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + if _, err = conn.Write([]byte(payload + "\n")); err != nil { + t.Fatal(err) + } + var out response + if err = json.NewDecoder(conn).Decode(&out); err != nil { + t.Fatal(err) + } + return out +} + +// The socket is the whole security boundary: 0600 in a 0700 directory, and the +// declaration (archive, journal, project, writers) comes from the agent's own +// configuration, never from the request. +func TestAgentSocketIsPrivateAndPinsTheDeclaration(t *testing.T) { + agent, socket, calls := testAgent(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go func() { _ = agent.Serve(ctx) }() + deadline := time.Now().Add(2 * time.Second) + for { + if _, err := os.Stat(socket); err == nil { + break + } + if time.Now().After(deadline) { + t.Fatal("agent socket never appeared") + } + time.Sleep(10 * time.Millisecond) + } + info, err := os.Stat(socket) + if err != nil { + t.Fatal(err) + } + if info.Mode().Perm() != 0600 { + t.Fatalf("socket mode %v", info.Mode().Perm()) + } + dir, err := os.Stat(filepath.Dir(socket)) + if err != nil { + t.Fatal(err) + } + if dir.Mode().Perm() != 0700 { + t.Fatalf("socket directory mode %v", dir.Mode().Perm()) + } + out := roundTrip(t, socket, `{"action":"capture","scope":"internal","job":"lib-internal-test"}`) + if !out.OK || out.Job != "lib-internal-test" { + t.Fatalf("capture rejected: %+v", out) + } + recorded := calls() + if len(recorded) != 1 { + t.Fatalf("expected one execution, got %v", recorded) + } + for _, want := range []string{"capture -scope internal -job lib-internal-test", + "-archive " + filepath.Join(filepath.Dir(socket), "library"), + "-journal " + filepath.Join(filepath.Dir(socket), "freeze"), + "-project weknora071full", "-writers app"} { + if !strings.Contains(recorded[0], want) { + t.Fatalf("execution %q missing %q", recorded[0], want) + } + } + if strings.Contains(recorded[0], "..") { + t.Fatal("unexpected path traversal in declaration") + } +} + +func TestAgentRejectsUnsupportedRequestsWithoutExecuting(t *testing.T) { + agent, socket, calls := testAgent(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go func() { _ = agent.Serve(ctx) }() + deadline := time.Now().Add(2 * time.Second) + for { + if _, err := os.Stat(socket); err == nil { + break + } + if time.Now().After(deadline) { + t.Fatal("agent socket never appeared") + } + time.Sleep(10 * time.Millisecond) + } + for _, payload := range []string{ + `{"action":"shell","command":"docker ps"}`, + `{"action":"capture","scope":"production"}`, + `{"action":"capture","scope":"internal","job":"bad job; rm -rf /"}`, + `not json`, + } { + out := roundTrip(t, socket, payload) + if out.OK || out.Error == "" { + t.Fatalf("accepted %q: %+v", payload, out) + } + } + if got := calls(); len(got) != 0 { + t.Fatalf("rejected requests still executed: %v", got) + } +} + +func TestNewRejectsIncompleteConfiguration(t *testing.T) { + if _, err := New(Config{}); err == nil { + t.Fatal("empty configuration accepted") + } + if _, err := New(Config{Socket: "relative.sock", ArchiveRoot: "/tmp/a", JournalRoot: "/tmp/j", + Binary: "/bin/true", Project: "p", Writers: "app", NonWriters: "frontend=ui", MaxDuration: time.Minute}); err == nil { + t.Fatal("relative socket path accepted") + } + if _, err := New(Config{Socket: "/tmp/a.sock", ArchiveRoot: "/tmp/a", JournalRoot: "/tmp/j", + Binary: "/bin/true", Project: "p", Writers: "app", NonWriters: "frontend", MaxDuration: time.Minute}); err == nil { + t.Fatal("non-writer declaration without a reason accepted") + } +} diff --git a/internal/libraryagent/client.go b/internal/libraryagent/client.go new file mode 100644 index 0000000..a1d2fca --- /dev/null +++ b/internal/libraryagent/client.go @@ -0,0 +1,92 @@ +package libraryagent + +import ( + "context" + "encoding/json" + "errors" + "net" + "time" +) + +// ClientConfig is what a governance process needs: one socket path. It never +// carries source credentials or Docker access — those stay with the agent. +type ClientConfig struct { + Socket string + Timeout time.Duration +} + +// Result is the agent's answer to one capture/status/recover request. +type Result struct { + OK bool + Scope string + Job string + SnapshotID string + Replay bool + Error string + Payload json.RawMessage +} + +// Capture asks the agent to seal one library scope. The job id makes the call +// idempotent: repeating it returns the already sealed candidate. +func Capture(ctx context.Context, config ClientConfig, scope, job string) (Result, error) { + return call(ctx, config, map[string]string{"action": "capture", "scope": scope, "job": job}) +} + +// Status probes the agent (source wiring, writer group, archive usage). +func Status(ctx context.Context, config ClientConfig) (Result, error) { + return call(ctx, config, map[string]string{"action": "status"}) +} + +// Recover restores writers from an interrupted freeze journal. +func Recover(ctx context.Context, config ClientConfig) (Result, error) { + return call(ctx, config, map[string]string{"action": "recover"}) +} + +func call(ctx context.Context, config ClientConfig, request map[string]string) (Result, error) { + if config.Socket == "" { + return Result{}, errors.New("library agent socket is not configured") + } + timeout := config.Timeout + if timeout <= 0 { + timeout = 2 * time.Minute + } + bounded, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + var dialer net.Dialer + conn, err := dialer.DialContext(bounded, "unix", config.Socket) + if err != nil { + return Result{}, errors.New("library agent is unreachable") + } + defer conn.Close() + if deadline, ok := bounded.Deadline(); ok { + _ = conn.SetDeadline(deadline) + } + if err = json.NewEncoder(conn).Encode(request); err != nil { + return Result{}, errors.New("library agent request failed") + } + var response struct { + OK bool `json:"ok"` + Scope string `json:"scope"` + Job string `json:"job"` + Error string `json:"error"` + Payload json.RawMessage `json:"payload"` + } + if err = json.NewDecoder(conn).Decode(&response); err != nil { + return Result{}, errors.New("library agent returned an invalid response") + } + result := Result{OK: response.OK, Scope: response.Scope, Job: response.Job, Error: response.Error, Payload: response.Payload} + var payload struct { + SnapshotID string `json:"snapshot_id"` + Replay bool `json:"replay"` + } + if len(response.Payload) > 0 && json.Unmarshal(response.Payload, &payload) == nil { + result.SnapshotID, result.Replay = payload.SnapshotID, payload.Replay + } + if !result.OK { + if result.Error == "" { + result.Error = "library agent refused the request" + } + return result, errors.New(result.Error) + } + return result, nil +} diff --git a/internal/libraryagent/peer_linux.go b/internal/libraryagent/peer_linux.go new file mode 100644 index 0000000..07887af --- /dev/null +++ b/internal/libraryagent/peer_linux.go @@ -0,0 +1,37 @@ +package libraryagent + +import ( + "errors" + "net" + "os" + "syscall" +) + +// peerAllowed enforces that the connecting process runs as the same uid as the +// agent. File mode 0600 already limits connect() to that uid; the credential +// check makes the boundary explicit and fails closed if the socket is ever +// reached another way (bind mount into a container with a different uid). +func peerAllowed(conn net.Conn) error { + unixConn, ok := conn.(*net.UnixConn) + if !ok { + return errors.New("unsupported transport") + } + raw, err := unixConn.SyscallConn() + if err != nil { + return errors.New("peer credentials unavailable") + } + var credential *syscall.Ucred + var controlErr error + if err = raw.Control(func(fd uintptr) { + credential, controlErr = syscall.GetsockoptUcred(int(fd), syscall.SOL_SOCKET, syscall.SO_PEERCRED) + }); err != nil { + return errors.New("peer credentials unavailable") + } + if controlErr != nil || credential == nil { + return errors.New("peer credentials unavailable") + } + if credential.Uid != uint32(os.Geteuid()) { + return errors.New("peer uid is not allowed") + } + return nil +} diff --git a/internal/libraryagent/peer_other.go b/internal/libraryagent/peer_other.go new file mode 100644 index 0000000..a7eb1fb --- /dev/null +++ b/internal/libraryagent/peer_other.go @@ -0,0 +1,14 @@ +//go:build !linux + +package libraryagent + +import ( + "errors" + "net" +) + +// The capture host is Linux (Docker + unix socket); other platforms refuse +// rather than guess at peer credentials. +func peerAllowed(net.Conn) error { + return errors.New("library agent requires Linux peer credentials") +} diff --git a/libraryagent b/libraryagent new file mode 100755 index 0000000..6c5123e Binary files /dev/null and b/libraryagent differ diff --git a/scripts/library-agent-install.sh b/scripts/library-agent-install.sh new file mode 100755 index 0000000..c533016 --- /dev/null +++ b/scripts/library-agent-install.sh @@ -0,0 +1,56 @@ +#!/usr/bin/env bash +# Install/refresh the host-side library capture agent. +# +# The agent is the only component with the Docker socket and the source +# credentials. It runs as this user (no root), listens on one 0600 unix socket and +# executes the fixed capture command with the declaration below. Containers get +# the socket path only. +set -euo pipefail +ROOT=$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd) +UNIT_DIR="$HOME/.config/systemd/user" +SOCKET_DIR="$HOME/.local/share/knowledge-governance/agent" +mkdir -p "$UNIT_DIR" "$SOCKET_DIR" +chmod 700 "$SOCKET_DIR" + +echo "=== build libraryctl + libraryagent ===" +(cd "$ROOT" && go build -o libraryctl ./cmd/libraryctl && go build -o libraryagent ./cmd/libraryagent) + +echo "=== systemd user unit ===" +cat > "$UNIT_DIR/knowledge-governance-library-agent.service" </dev/null +systemctl --user restart knowledge-governance-library-agent.service +# Keep the agent running when no login session is open (it owns no privilege +# beyond this account, but a stopped agent means the page cannot capture). +loginctl enable-linger "$(id -un)" >/dev/null 2>&1 || echo " WARN: enable-linger failed; agent stops with the last session" + +echo "=== read back ===" +for _ in $(seq 1 30); do + if [ -S "$SOCKET_DIR/agent.sock" ]; then break; fi + sleep 0.2 +done +[ -S "$SOCKET_DIR/agent.sock" ] || { echo "AGENT_SOCKET_MISSING"; exit 1; } +stat -c '%A %n' "$SOCKET_DIR" "$SOCKET_DIR/agent.sock" +systemctl --user is-active knowledge-governance-library-agent.service \ No newline at end of file diff --git a/web/src/LibraryVersions.tsx b/web/src/LibraryVersions.tsx index e350715..5a461b9 100644 --- a/web/src/LibraryVersions.tsx +++ b/web/src/LibraryVersions.tsx @@ -1,11 +1,15 @@ import { useCallback, useEffect, useMemo, useState } from 'react' import { + enqueueLibraryCapture, + fetchLibraryCaptures, fetchLibrarySnapshots, formatBytes, scopeLabel, + type LibraryCaptureJob, type LibrarySnapshot, type LibrarySnapshotList, } from './librarySnapshots' +import { useOptionalAuth } from './auth' import './libraryVersions.css' // 「整库版本」页：只读展示已封存的整库候选产物（对内/对外各一份最新）。 @@ -26,12 +30,24 @@ const stamp = (value: string) => const metric = (snapshot: LibrarySnapshot, key: string) => String(snapshot.totals[key] ?? 0) +const jobStateLabel: Record = { + pending: '待执行', + running: '执行中', + retry: '重试中', + completed: '已完成', + failed: '失败', +} + export function LibraryVersions() { + const auth = useOptionalAuth() const [page, setPage] = useState(null) const [loading, setLoading] = useState(true) const [error, setError] = useState(null) const [expanded, setExpanded] = useState(null) const [scopeFilter, setScopeFilter] = useState<'all' | 'internal' | 'external'>('all') + const [jobs, setJobs] = useState([]) + const [pendingScope, setPendingScope] = useState(null) + const [notice, setNotice] = useState(null) const load = useCallback(async (signal?: AbortSignal) => { setLoading(true) @@ -53,6 +69,51 @@ export function LibraryVersions() { return () => controller.abort() }, [load]) + const refreshJobs = useCallback(async (signal?: AbortSignal) => { + const items = await fetchLibraryCaptures(signal) + setJobs(items) + return items + }, []) + + useEffect(() => { + const controller = new AbortController() + void refreshJobs(controller.signal).catch(() => undefined) + return () => controller.abort() + }, [refreshJobs]) + + // A capture runs in a maintenance window (seconds), so a short poll is enough + // to show the real queue state and to pick up the new sealed version. + const active = jobs.some((job) => job.state === 'pending' || job.state === 'running' || job.state === 'retry') + useEffect(() => { + if (!active) return + const timer = window.setInterval(() => { + void refreshJobs() + .then((items) => { + if (!items.some((job) => job.state === 'pending' || job.state === 'running' || job.state === 'retry')) { + void load() + } + }) + .catch(() => undefined) + }, 3000) + return () => window.clearInterval(timer) + }, [active, refreshJobs, load]) + + const canCapture = (auth?.identity?.role === 'engineer' || auth?.identity?.role === 'admin') && page?.capture_available === true + + const startCapture = async (scope: string) => { + setPendingScope(scope) + setNotice(null) + try { + const accepted = await enqueueLibraryCapture(scope) + setNotice(`已投递封存作业：${scopeLabel(scope)} · ${accepted.capture_job}（等待主机侧执行）`) + await refreshJobs() + } catch (captureError) { + setNotice(captureError instanceof Error ? captureError.message : '发起封存失败') + } finally { + setPendingScope(null) + } + } + const snapshots = page?.snapshots ?? [] const latestByScope = useMemo(() => { const map: Record = {} @@ -144,11 +205,32 @@ export function LibraryVersions() {

) : ( -

- 该库没有封存产物。封存在目标主机执行： - scripts/library-capture.sh capture -scope {scope} -

+

该库还没有封存产物。

)} +
+ {page?.capture_available ? ( + canCapture ? ( + + ) : ( + 发起封存需要知识工程师及以上权限 + ) + ) : ( + {page?.capture_reason} + )} +
) })} @@ -239,6 +321,40 @@ export function LibraryVersions() { )} + {notice ?

{notice}

: null} + + {jobs.length > 0 ? ( +
+

封存作业

+ + + + + + + + + + + + + {jobs.map((job) => ( + + + + + + + + + ))} + +
投递时间范围状态作业号发起人最后错误
{stamp(job.created_at)}{scopeLabel(job.scope)}{jobStateLabel[job.state] ?? job.state} + {job.capture_job.slice(0, 24)} + {job.requested_by || '—'}{job.last_error || '—'}
+
+ ) : null} + {visible .filter((snapshot) => snapshot.snapshot_id === expanded) .map((snapshot) => ( diff --git a/web/src/librarySnapshots.ts b/web/src/librarySnapshots.ts index 0359039..30aa721 100644 --- a/web/src/librarySnapshots.ts +++ b/web/src/librarySnapshots.ts @@ -37,6 +37,25 @@ export interface LibrarySnapshot { totals: Record } +export interface LibraryCaptureJob { + job_id: string + capture_job: string + scope: string + state: string + attempt: number + last_error: string + requested_by: string + created_at: string + updated_at: string +} + +export interface LibraryCaptureAccepted { + job_id: string + capture_job: string + scope: string + state: string +} + export interface LibrarySnapshotList { available: boolean reason: string @@ -53,6 +72,8 @@ export interface LibrarySnapshotList { manifest_file_bytes: number manifest_allocated_bytes: number capture_note: string + capture_available: boolean + capture_reason: string } const record = (value: unknown): value is Record => @@ -131,9 +152,55 @@ export async function fetchLibrarySnapshots(signal?: AbortSignal): Promise { + const response = await fetch('/api/v1/library/captures', { + method: 'POST', + headers: { 'Content-Type': 'application/json', Accept: 'application/json' }, + body: JSON.stringify({ scope }), + }) + const value = (await response.json().catch(() => null)) as unknown + if (!response.ok) { + if (response.status === 403) throw new Error('只有知识工程师及以上可以发起封存。') + if (response.status === 503 && record(value)) { + throw new Error(text(value.reason) || '主机侧封存 agent 未接线') + } + throw new Error(`发起封存失败（HTTP ${response.status}）`) + } + if (!record(value)) throw new Error('封存作业响应无效') + return { + job_id: text(value.job_id), + capture_job: text(value.capture_job), + scope: text(value.scope), + state: text(value.state), + } +} + +export async function fetchLibraryCaptures(signal?: AbortSignal): Promise { + const response = await fetch('/api/v1/library/captures', { + headers: { Accept: 'application/json' }, + signal, + }) + const value = (await response.json().catch(() => null)) as unknown + if (!response.ok || !record(value) || !Array.isArray(value.jobs)) return [] + return value.jobs.filter(record).map((job) => ({ + job_id: text(job.job_id), + capture_job: text(job.capture_job), + scope: text(job.scope), + state: text(job.state), + attempt: num(job.attempt), + last_error: text(job.last_error), + requested_by: text(job.requested_by), + created_at: text(job.created_at), + updated_at: text(job.updated_at), + })) +} + export function formatBytes(bytes: number): string { if (!Number.isFinite(bytes) || bytes <= 0) return '0 B' const units = ['B', 'KB', 'MB', 'GB', 'TB'] diff --git a/web/src/libraryVersions.css b/web/src/libraryVersions.css index 1597092..e391879 100644 --- a/web/src/libraryVersions.css +++ b/web/src/libraryVersions.css @@ -165,6 +165,50 @@ max-width: 70ch; line-height: 1.5; } +.library-scope-actions { + margin-top: 14px; + display: flex; + align-items: center; + gap: 10px; + font-size: 12px; + color: var(--ink-mute, #64748b); +} +.library-scope-actions button { + padding: 7px 14px; + border-radius: 8px; + border: 1px solid #2563eb; + background: #eff6ff; + color: #1d4ed8; + cursor: pointer; +} +.library-scope-actions button:disabled { + cursor: progress; + opacity: 0.6; +} +.library-capture-notice { + margin: 0; + padding: 10px 14px; + border-radius: 8px; + background: #eff6ff; + border: 1px solid #bfdbfe; + color: #1d4ed8; + font-size: 13px; +} +.library-capture-jobs { + background: var(--bg-panel, #fff); + border: 1px solid var(--border-base, #e2e8f0); + border-radius: 10px; + padding: 16px 20px; +} +.library-capture-jobs h3 { + margin: 0 0 10px; + font-size: 15px; +} +.library-capture-error { + max-width: 360px; + overflow-wrap: anywhere; + white-space: normal; +} .library-versions-accounting { justify-content: flex-start; font-size: 12px;