diff --git a/.gitignore b/.gitignore index c7f1e6c..5f2e9f9 100644 --- a/.gitignore +++ b/.gitignore @@ -29,6 +29,7 @@ /release-diag /worker /libraryctl +/libraryagent /apply_s5_patch.py /tmp_* .style-backup-*/ diff --git a/CHANGELOG.md b/CHANGELOG.md index 4321002..917cde0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,23 @@ # Changelog +## 开发快照 snapshot/v0.1.112 - 2026-10-28 + +### Changed + +- 发布管理改为一版可操作的整库视图：只给「对内 / 对外」两个目标，把整库版本页的封存信息、完整性结果、分项数量/字节/摘要、封存作业状态与归档存储口径并入同一页；链路按真实能力标注「封存已接通、生产预备份未接线、同步未授权、校验/恢复未接线」，禁用项写明原因，不用假进度。 +- 「整库版本」独立导航并入发布管理（不再保留第三个相近入口）；文档级旧流程折叠保留在页底，并明确单篇文档发布成功不等于整库发布成功。 + +### Fixed + +- R4：封存子进程不再被默认 SIGKILL。超时/取消先发 SIGTERM 并给宽限期（`StopGrace`），`libraryctl` 把 SIGTERM 转成上下文取消，让停机窗口走恢复路径；任何失败的封存/恢复操作之后 agent 都会再执行一次有界的 `recover`，恢复失败则进入降级状态、阻断后续封存并给出原因（`recover`/`status` 仍可用）；agent 启动时先恢复上次遗留的停机窗口，关闭时等待在飞操作完成其恢复责任。 +- R5：同一作业号换范围不再被回显成新范围。接口以已存 payload 为准，范围不一致返回 409（`library_capture_scope_conflict`）且不入新队；相同请求重放返回真实 `reused`/存储范围。 +- R6：封存作业列表改为按类型查询（`ListByType`），不再先取全局 200 条再过滤；存储不支持按类型查询时明确返回不可用，页面不据此判断空闲。 + +### Added + +- `ListByType` 作用域查询与测试；封存闸口单测（未接线 503、非法 400、范围冲突 409、重放回显）。 +- 发布管理新增 6 条渲染测试（两个目标、真实数量、链路禁用原因、权限/未接线分支、作业态、存储口径）。 + ## 开发快照 snapshot/v0.1.111 - 2026-10-28 ### Added diff --git a/cmd/libraryctl/main.go b/cmd/libraryctl/main.go index 7a57212..b8a82f2 100644 --- a/cmd/libraryctl/main.go +++ b/cmd/libraryctl/main.go @@ -23,9 +23,11 @@ import ( "fmt" "os" "os/exec" + "os/signal" "path/filepath" "sort" "strings" + "syscall" "time" "github.com/jackc/pgx/v5/pgxpool" @@ -67,7 +69,12 @@ func main() { os.Exit(2) } _ = rest - ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute) + // SIGTERM must not kill this process mid-window: the capture path restores the + // source writers through a cancelled-context recovery, so the signal is turned + // into a context cancellation and the process exits only after that cleanup. + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + ctx, cancel := context.WithTimeout(ctx, 30*time.Minute) defer cancel() var runErr error switch command { diff --git a/internal/httpapi/library_capture.go b/internal/httpapi/library_capture.go index b2c6cbc..2cac99c 100644 --- a/internal/httpapi/library_capture.go +++ b/internal/httpapi/library_capture.go @@ -1,6 +1,7 @@ package httpapi import ( + "context" "encoding/json" "fmt" "net/http" @@ -61,8 +62,13 @@ func libraryCaptureEnqueue(w http.ResponseWriter, r *http.Request, config Config return } now := time.Now().UTC() + requestedID := fmt.Sprintf("library-capture-%s-%d", request.Scope, now.UnixNano()) + // The idempotency key is the capture job id: the same logical capture must map + // to one queue row. Binding the scope into the key would silently create a + // second job instead of reporting a conflicting request, so the conflict is + // detected by comparing the stored payload below. job, err := config.Store.Enqueue(r.Context(), jobs.EnqueueRequest{ - ID: fmt.Sprintf("library-capture-%s-%d", request.Scope, now.UnixNano()), + ID: requestedID, Type: libraryCaptureJobType, Payload: payload, AvailableAt: now, @@ -73,24 +79,55 @@ func libraryCaptureEnqueue(w http.ResponseWriter, r *http.Request, config Config writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot enqueue capture job"}) return } + // Enqueue returns the existing row for a reused key. Answer from that row, not + // from the current request: a reused job id with another scope must be refused + // instead of being reported as a job for the new scope. + var stored struct { + Scope string `json:"scope"` + CaptureJob string `json:"capture_job"` + RequestedBy string `json:"requested_by"` + } + if err = json.Unmarshal(job.Payload, &stored); err != nil || stored.CaptureJob != captureJob { + writeJSON(w, http.StatusConflict, map[string]any{ + "code": "library_capture_conflict", "error": "该作业号已被另一个封存请求占用", + "existing_scope": stored.Scope, "requested_scope": request.Scope, + }) + return + } + if stored.Scope != request.Scope { + writeJSON(w, http.StatusConflict, map[string]any{ + "code": "library_capture_scope_conflict", "error": "同一作业号已按另一个范围投递", + "existing_scope": stored.Scope, "requested_scope": request.Scope, + }) + return + } writeJSON(w, http.StatusAccepted, map[string]any{ - "job_id": job.ID, "capture_job": captureJob, "scope": request.Scope, "state": job.State, + "job_id": job.ID, "capture_job": stored.CaptureJob, "scope": stored.Scope, + "state": job.State, "reused": job.ID != requestedID, "requested_by": stored.RequestedBy, }) } // 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) + // Ask the store for this type only. Reading the global queue and filtering in + // the handler would hide capture jobs behind unrelated traffic and make the + // page claim "idle" while a capture is still queued. + byType, ok := config.Store.(interface { + ListByType(ctx context.Context, jobType string, limit int) ([]jobs.Job, error) + }) + if !ok { + writeJSON(w, http.StatusOK, map[string]any{"available": false, + "reason": "任务存储不支持按类型查询，页面不会据此判断空闲", "jobs": []any{}}) + return + } + queue, err := byType.ListByType(r.Context(), libraryCaptureJobType, 20) if err != nil { writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot read capture jobs"}) return } - views := make([]map[string]any, 0, 20) + views := make([]map[string]any, 0, len(queue)) for _, job := range queue { - if job.Type != libraryCaptureJobType || len(views) >= 20 { - continue - } var payload struct { Scope string `json:"scope"` CaptureJob string `json:"capture_job"` diff --git a/internal/httpapi/library_capture_test.go b/internal/httpapi/library_capture_test.go index c919af9..060d56f 100644 --- a/internal/httpapi/library_capture_test.go +++ b/internal/httpapi/library_capture_test.go @@ -86,18 +86,25 @@ func TestLibraryCaptureRejectsInvalidRequest(t *testing.T) { } } -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"}, - }} +func TestLibraryCaptureJobsAsksStoreForItsOwnType(t *testing.T) { + // The global queue is busy with unrelated traffic; the capture job must still + // be visible because the query is type-scoped, not filtered after the fact. + store := &fakeJobStore{ + listResult: []jobs.Job{{ID: "busy", Type: "artifact.fetch", State: "running"}}, + listByTypeResult: []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)}, + }, + } 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) } + if store.listType != libraryCaptureJobType { + t.Fatalf("store queried with type %q", store.listType) + } var body struct { Available bool `json:"available"` Jobs []map[string]any `json:"jobs"` @@ -109,3 +116,44 @@ func TestLibraryCaptureJobsListsOnlyCaptureJobs(t *testing.T) { t.Fatalf("unexpected body %+v", body) } } + +// Reusing a capture job id with another scope must be refused, and the response +// must report the scope that was actually stored. +func TestLibraryCaptureRejectsScopeConflictForReusedJob(t *testing.T) { + store := &fakeJobStore{enqueueResult: &jobs.Job{ID: "existing-job", Type: libraryCaptureJobType, + State: jobs.StateCompleted, + 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"}) + if recorder.Code != http.StatusConflict { + 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_scope_conflict" || body["existing_scope"] != "internal" || body["requested_scope"] != "external" { + t.Fatalf("unexpected conflict body %v", body) + } +} + +// A reused job id with an identical request is answered from the stored row. +func TestLibraryCaptureReplayReportsStoredJob(t *testing.T) { + store := &fakeJobStore{enqueueResult: &jobs.Job{ID: "existing-job", Type: libraryCaptureJobType, + State: jobs.StateCompleted, + 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"}) + if recorder.Code != http.StatusAccepted { + 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["scope"] != "internal" || body["capture_job"] != "lib-shared" || body["reused"] != true || body["job_id"] != "existing-job" { + t.Fatalf("unexpected replay body %v", body) + } +} diff --git a/internal/httpapi/server_test.go b/internal/httpapi/server_test.go index 3122788..b4bfef6 100644 --- a/internal/httpapi/server_test.go +++ b/internal/httpapi/server_test.go @@ -1526,6 +1526,10 @@ type fakeJobStore struct { enqueued jobs.EnqueueRequest enqueueResult *jobs.Job listResult []jobs.Job + listType string + listByTypeResult []jobs.Job + listByTypeLimit int + listByTypeErr error listLimit int assetResult governance.SourceAssetPage assetFilter governance.SourceAssetFilter @@ -1871,7 +1875,9 @@ func (s *fakeJobStore) ActivateUISettingsWithOperation(_ context.Context, revisi func (s *fakeJobStore) Enqueue(_ context.Context, request jobs.EnqueueRequest, _ time.Time) (*jobs.Job, error) { s.enqueued = request if s.enqueueEchoRequestedID { - return &jobs.Job{ID: request.ID, Type: request.Type, State: jobs.StatePending}, nil + // The real store returns the stored row, including its payload: a reused + // idempotency key must be judged by that payload, not by the request. + return &jobs.Job{ID: request.ID, Type: request.Type, State: jobs.StatePending, Payload: request.Payload}, nil } return s.enqueueResult, nil } @@ -1881,6 +1887,12 @@ func (s *fakeJobStore) List(_ context.Context, limit int) ([]jobs.Job, error) { return s.listResult, nil } +func (s *fakeJobStore) ListByType(_ context.Context, jobType string, limit int) ([]jobs.Job, error) { + s.listType = jobType + s.listByTypeLimit = limit + return s.listByTypeResult, s.listByTypeErr +} + func (s *fakeJobStore) ListTestPublications(_ context.Context, filter governance.TestPublicationFilter) (governance.TestPublicationPage, error) { s.publicationFilter = filter return s.publicationResult, nil diff --git a/internal/libraryagent/agent.go b/internal/libraryagent/agent.go index 6942283..edc279a 100644 --- a/internal/libraryagent/agent.go +++ b/internal/libraryagent/agent.go @@ -16,6 +16,7 @@ import ( "context" "encoding/json" "errors" + "log" "net" "os" "os/exec" @@ -23,6 +24,7 @@ import ( "regexp" "strings" "sync" + "syscall" "time" ) @@ -43,6 +45,13 @@ type Config struct { Writers string NonWriters string MaxDuration time.Duration + // StopGrace bounds the graceful stop of one operation: SIGTERM first (the + // child aborts the maintenance window and restores writers), then SIGKILL + // only after this grace period. It must exceed the child's own restore budget. + StopGrace time.Duration + // RecoverTimeout bounds the recovery run performed after a failed or killed + // operation, and at agent start-up. + RecoverTimeout time.Duration // Run executes the binary; replaced in tests. Run func(ctx context.Context, name string, args ...string) ([]byte, error) } @@ -69,6 +78,11 @@ var jobPattern = regexp.MustCompile(`^[A-Za-z0-9_-]{1,64}$`) type Agent struct { config Config mu sync.Mutex + // 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 + wg sync.WaitGroup } func New(config Config) (*Agent, error) { @@ -81,6 +95,15 @@ func New(config Config) (*Agent, error) { if config.MaxDuration <= 0 || config.MaxDuration > 30*time.Minute { return nil, errors.New("library agent duration bound is invalid") } + if config.StopGrace == 0 { + config.StopGrace = 3 * time.Minute + } + if config.RecoverTimeout == 0 { + config.RecoverTimeout = 3 * time.Minute + } + if config.StopGrace <= 0 || config.StopGrace > 10*time.Minute || config.RecoverTimeout <= 0 || config.RecoverTimeout > 10*time.Minute { + return nil, errors.New("library agent recovery/stop budget is invalid") + } return &Agent{config: config}, nil } @@ -105,6 +128,13 @@ func (a *Agent) Serve(ctx context.Context) error { if err := os.Chmod(a.config.Socket, 0600); err != nil { return err } + // Startup responsibility: an agent restart may have inherited a stopped + // 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() + log.Printf("library agent: pending freeze recovery failed: %v", err) + } go func() { <-ctx.Done() _ = listener.Close() @@ -113,11 +143,24 @@ func (a *Agent) Serve(ctx context.Context) error { conn, err := listener.Accept() if err != nil { if ctx.Err() != nil { + // Drain in-flight operations: each of them still owes a bounded + // recovery before the agent may exit. + done := make(chan struct{}) + go func() { a.wg.Wait(); close(done) }() + select { + case <-done: + case <-time.After(a.config.StopGrace + a.config.RecoverTimeout): + log.Printf("library agent: shutdown deadline exceeded with operations in flight") + } return nil } return err } - go a.handle(ctx, conn) + a.wg.Add(1) + go func(conn net.Conn) { + defer a.wg.Done() + a.handle(ctx, conn) + }(conn) } } @@ -143,6 +186,12 @@ 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} + // 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" + return out + } var args []string switch req.Action { case "status": @@ -171,12 +220,30 @@ func (a *Agent) execute(ctx context.Context, req request) response { defer cancel() payload, err := a.config.run(bounded, a.config.Binary, args...) if err != nil { - out.Error = "operation failed" + detail := "operation failed" if message := strings.TrimSpace(string(payload)); message != "" && len(message) <= 2000 { - out.Error = message + detail = message } + // A killed or aborted child may have left the source writer stopped. The + // recovery responsibility does not end with the failed request: run the + // idempotent recovery now, and refuse further captures if it does not + // verify. + if req.Action == "capture" || req.Action == "recover" { + if recoveryErr := a.recoverWriters(ctx); recoveryErr != nil { + a.degraded = recoveryErr.Error() + out.Error = detail + "；写入者恢复失败：" + recoveryErr.Error() + "（已阻断后续封存，需人工处理）" + return out + } + detail += "；已执行 recover 并确认写入者恢复" + } else { + // status 失败不需要恢复；它不改变停机状态。 + } + out.Error = detail return out } + if req.Action == "recover" { + a.degraded = "" + } out.OK = true if json.Valid(payload) { out.Payload = json.RawMessage(payload) @@ -186,6 +253,18 @@ func (a *Agent) execute(ctx context.Context, req request) response { return out } +// 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 { + bounded, cancel := context.WithTimeout(context.WithoutCancel(ctx), a.config.RecoverTimeout) + defer cancel() + if _, err := a.config.run(bounded, a.config.Binary, append([]string{"recover"}, a.config.declaration()...)...); err != nil { + return err + } + a.degraded = "" + return nil +} + // 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 { @@ -208,6 +287,17 @@ func (c Config) run(ctx context.Context, name string, args ...string) ([]byte, e } command := exec.CommandContext(ctx, name, args...) command.Env = append(os.Environ(), "LIBRARYCTL_AGENT=1") + // Default CommandContext behaviour is SIGKILL, which would skip the child's + // window restore and leave the source writer stopped. Ask for a graceful stop + // first and only escalate after the grace period. + command.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} + command.Cancel = func() error { + if command.Process == nil { + return nil + } + return command.Process.Signal(syscall.SIGTERM) + } + command.WaitDelay = c.StopGrace return command.CombinedOutput() } diff --git a/internal/libraryagent/agent_test.go b/internal/libraryagent/agent_test.go index cdb6e16..8af94d7 100644 --- a/internal/libraryagent/agent_test.go +++ b/internal/libraryagent/agent_test.go @@ -3,6 +3,7 @@ package libraryagent import ( "context" "encoding/json" + "errors" "net" "os" "path/filepath" @@ -108,9 +109,9 @@ func TestAgentSocketIsPrivateAndPinsTheDeclaration(t *testing.T) { if !out.OK || out.Job != "lib-internal-test" { t.Fatalf("capture rejected: %+v", out) } - recorded := calls() + recorded := captureCalls(calls()) if len(recorded) != 1 { - t.Fatalf("expected one execution, got %v", recorded) + t.Fatalf("expected one capture execution, got %v", recorded) } for _, want := range []string{"capture -scope internal -job lib-internal-test", "-archive " + filepath.Join(filepath.Dir(socket), "library"), @@ -151,11 +152,177 @@ func TestAgentRejectsUnsupportedRequestsWithoutExecuting(t *testing.T) { t.Fatalf("accepted %q: %+v", payload, out) } } - if got := calls(); len(got) != 0 { + if got := captureCalls(calls()); len(got) != 0 { t.Fatalf("rejected requests still executed: %v", got) } } +func captureCalls(calls []string) []string { + kept := make([]string, 0, len(calls)) + for _, call := range calls { + if strings.Contains(call, " capture ") { + kept = append(kept, call) + } + } + return kept +} + +func recoverCalls(calls []string) []string { + kept := make([]string, 0, len(calls)) + for _, call := range calls { + if strings.Contains(call, " recover ") { + kept = append(kept, call) + } + } + return kept +} + +// failingAgent builds an agent whose capture always fails and whose recovery +// outcome can be switched, to prove the recovery responsibility after a +// killed/aborted child. +func failingAgent(t *testing.T) (*Agent, string, func() []string, func(error)) { + t.Helper() + root, err := os.MkdirTemp(os.TempDir(), "kgrf-") + 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 + var recoverErr error + agent, err := New(Config{ + Socket: filepath.Join(root, "agent.sock"), Binary: "/bin/echo", + ArchiveRoot: filepath.Join(root, "library"), JournalRoot: filepath.Join(root, "freeze"), + SourceApp: "app-container", SourcePG: "pg", SourceES: "es", GovernancePG: "gov", + Project: "weknora071full", Writers: "app", NonWriters: "frontend=UI only", + MaxDuration: time.Minute, StopGrace: time.Minute, RecoverTimeout: time.Minute, + Run: func(_ context.Context, name string, args ...string) ([]byte, error) { + mu.Lock() + calls = append(calls, name+" "+strings.Join(args, " ")) + current := recoverErr + mu.Unlock() + if len(args) > 0 && args[0] == "recover" { + if current != nil { + return []byte("writer restart failed"), current + } + return []byte(`{"recovered":true}`), nil + } + return []byte("LIBRARYCTL_FAILED: stop outcome uncertain"), errors.New("exit status 1") + }, + }) + if err != nil { + t.Fatal(err) + } + return agent, filepath.Join(root, "agent.sock"), func() []string { + mu.Lock() + defer mu.Unlock() + return append([]string(nil), calls...) + }, func(err error) { + mu.Lock() + defer mu.Unlock() + recoverErr = err + } +} + +// A failed capture must be followed by a recovery attempt, not just an error. +func TestAgentRecoversWritersAfterFailedCapture(t *testing.T) { + agent, socket, calls, _ := failingAgent(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go func() { _ = agent.Serve(ctx) }() + waitSocket(t, socket) + out := roundTrip(t, socket, `{"action":"capture","scope":"internal","job":"lib-fail-1"}`) + if out.OK { + t.Fatalf("failed capture reported success: %+v", out) + } + if !strings.Contains(out.Error, "已执行 recover 并确认写入者恢复") { + t.Fatalf("error does not state the recovery outcome: %q", out.Error) + } + if len(recoverCalls(calls())) == 0 { + t.Fatal("no recovery ran after the failed capture") + } +} + +// If recovery itself fails, the boundary must refuse further captures instead of +// leaving the source stopped and merely marking a job failed. +func TestAgentBlocksCapturesWhenRecoveryFails(t *testing.T) { + agent, socket, calls, setRecoverErr := failingAgent(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go func() { _ = agent.Serve(ctx) }() + waitSocket(t, socket) + setRecoverErr(errors.New("docker restart failed")) + out := roundTrip(t, socket, `{"action":"capture","scope":"internal","job":"lib-fail-2"}`) + if out.OK || !strings.Contains(out.Error, "写入者恢复失败") || !strings.Contains(out.Error, "已阻断后续封存") { + t.Fatalf("recovery failure not surfaced: %+v", out) + } + before := len(captureCalls(calls())) + blocked := roundTrip(t, socket, `{"action":"capture","scope":"external","job":"lib-fail-3"}`) + if blocked.OK || !strings.Contains(blocked.Error, "降级") { + t.Fatalf("degraded agent accepted a capture: %+v", blocked) + } + if got := len(captureCalls(calls())); got != before { + t.Fatalf("blocked capture still executed (%d -> %d)", before, got) + } + // A verified recovery clears the degraded state and lets work resume. + setRecoverErr(nil) + if recovered := roundTrip(t, socket, `{"action":"recover"}`); !recovered.OK { + t.Fatalf("explicit recovery failed: %+v", recovered) + } + if resumed := roundTrip(t, socket, `{"action":"capture","scope":"internal","job":"lib-fail-4"}`); strings.Contains(resumed.Error, "降级") { + t.Fatalf("recovery did not clear the degraded state: %+v", resumed) + } +} + +// The child is stopped with SIGTERM and the grace period, not an immediate kill. +func TestRunStopsChildWithSigterm(t *testing.T) { + dir := t.TempDir() + script := filepath.Join(dir, "child.sh") + // A real child that only exits 0 when it receives SIGTERM. + child := "#!/bin/sh\ntrap 'exit 0' TERM\nwhile :; do sleep 0.05; done\n" + if err := os.WriteFile(script, []byte(child), 0o700); err != nil { + t.Fatal(err) + } + agent, err := New(Config{Socket: "/tmp/irrelevant.sock", Binary: script, + ArchiveRoot: filepath.Join(dir, "library"), JournalRoot: filepath.Join(dir, "freeze"), + SourceApp: "app-container", SourcePG: "pg", SourceES: "es", GovernancePG: "gov", + Project: "weknora071full", Writers: "app", NonWriters: "frontend=UI only", + MaxDuration: 2 * time.Second, StopGrace: 5 * time.Second, RecoverTimeout: 5 * time.Second}) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond) + defer cancel() + started := time.Now() + _, err = agent.config.run(ctx, script) + if err == nil { + t.Fatal("cancelled run reported success") + } + if elapsed := time.Since(started); elapsed > 4*time.Second { + t.Fatalf("child was not stopped within the graceful window: %s", elapsed) + } + if ctx.Err() == nil { + t.Fatal("test did not exercise cancellation") + } +} + +func waitSocket(t *testing.T, socket string) { + t.Helper() + 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) + } +} + func TestNewRejectsIncompleteConfiguration(t *testing.T) { if _, err := New(Config{}); err == nil { t.Fatal("empty configuration accepted") diff --git a/internal/platform/postgres/queue_sql.go b/internal/platform/postgres/queue_sql.go index 38a0037..e2c0c80 100644 --- a/internal/platform/postgres/queue_sql.go +++ b/internal/platform/postgres/queue_sql.go @@ -132,6 +132,18 @@ RETURNING job_id SELECT id FROM updated ` +// One job type, newest first: a caller which needs "the recent jobs of MY type" +// must not page the global queue (other traffic would hide its own rows). +const listJobsByTypeSQL = ` +SELECT id, type, payload, state, priority, available_at, + lease_owner, lease_until, attempt, max_attempts, + last_error, idempotency_key, created_at, updated_at +FROM jobs +WHERE type = $1 +ORDER BY created_at DESC, id DESC +LIMIT $2 +` + const listJobsSQL = ` SELECT id, type, payload, state, priority, available_at, lease_owner, lease_until, attempt, max_attempts, diff --git a/internal/platform/postgres/store.go b/internal/platform/postgres/store.go index b619f13..a045abe 100644 --- a/internal/platform/postgres/store.go +++ b/internal/platform/postgres/store.go @@ -65,6 +65,31 @@ func (s *Store) Claim(ctx context.Context, workerID string, allowedTypes []strin return job, nil } +// ListByType lists only the requested job type, so a busy queue cannot hide the +// caller's own recent jobs behind unrelated traffic. +func (s *Store) ListByType(ctx context.Context, jobType string, limit int) ([]jobs.Job, error) { + if strings.TrimSpace(jobType) == "" || limit < 1 || limit > 200 { + return nil, errors.New("list jobs by type requires a type and a bounded limit") + } + rows, err := s.db.Query(ctx, listJobsByTypeSQL, jobType, limit) + if err != nil { + return nil, fmt.Errorf("list jobs by type: %w", err) + } + defer rows.Close() + result := make([]jobs.Job, 0, limit) + for rows.Next() { + job, err := scanJob(rows) + if err != nil { + return nil, fmt.Errorf("scan listed job: %w", err) + } + result = append(result, *job) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterate jobs: %w", err) + } + return result, nil +} + func (s *Store) List(ctx context.Context, limit int) ([]jobs.Job, error) { rows, err := s.db.Query(ctx, listJobsSQL, limit) if err != nil { diff --git a/libraryagent b/libraryagent deleted file mode 100755 index 6c5123e..0000000 Binary files a/libraryagent and /dev/null differ diff --git a/web/src/App.tsx b/web/src/App.tsx index 582db36..bfaeeaf 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -4,7 +4,7 @@ import { DocumentWorkbench } from './DocumentWorkbench' import { SourceCatalogBrowser } from './SourceCatalogBrowser' import { WeKnoraMonitor } from './WeKnoraMonitor' import { VersionHistory } from './VersionHistory' -import { LibraryVersions } from './LibraryVersions' +import { ReleaseTargets } from './ReleaseTargets' import { QAOperations } from './QAOperations' import { SourceLinkManager } from './SourceLinkManager' import { SystemSettings } from './SystemSettings' @@ -63,7 +63,6 @@ const modules: ReadonlyArray<{ id: ModuleId; label: string; icon: string }> = [ { id: 'tasks', label: '解析监控', icon: '↻' }, { id: 'qaops', label: '问答运营', icon: '▤' }, { id: 'releases', label: '发布管理', icon: '⇧' }, - { id: 'library', label: '整库版本', icon: '▦' }, { id: 'settings', label: '系统设置', icon: '⚙' }, { id: 'versions', label: '版本记录', icon: '⌥' }, ] @@ -409,7 +408,11 @@ export function ModulePage({ } if (moduleId === 'releases') { return ( - + +
+ 文档级发布（旧流程，折叠保留）：单篇文档发布成功不等于整库发布成功 + + onCancelReleaseAction={onCancelReleaseAction} + /> +
+ ) } if (moduleId === 'settings') { @@ -2972,8 +2977,6 @@ export default function App() { ) : workspace.activeTab === 'knowledge' ? ( - ) : workspace.activeTab === 'library' ? ( - ) : workspace.activeTab === 'versions' ? ( ) : workspace.activeTab === 'qaops' ? ( diff --git a/web/src/LibraryVersions.tsx b/web/src/LibraryVersions.tsx deleted file mode 100644 index 5a461b9..0000000 --- a/web/src/LibraryVersions.tsx +++ /dev/null @@ -1,387 +0,0 @@ -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' - -// 「整库版本」页：只读展示已封存的整库候选产物（对内/对外各一份最新）。 -// 页面不提供封存按钮：封存要停真实写入者并读来源凭据，只能在目标主机用 -// scripts/library-capture.sh 执行。通过完整性检查 != 已发布，授权仍走发布管理。 -const stamp = (value: string) => - value && Number.isFinite(new Date(value).getTime()) - ? new Date(value).toLocaleString('zh-CN', { - timeZone: 'Asia/Shanghai', - year: 'numeric', - month: '2-digit', - day: '2-digit', - hour: '2-digit', - minute: '2-digit', - hour12: false, - }) - : '—' - -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) - try { - const result = await fetchLibrarySnapshots(signal) - setPage(result) - setError(null) - } catch (loadError) { - if ((loadError as Error).name === 'AbortError') return - setError(loadError instanceof Error ? loadError.message : '加载整库版本失败') - } finally { - setLoading(false) - } - }, []) - - useEffect(() => { - const controller = new AbortController() - void load(controller.signal) - 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 = {} - for (const snapshot of snapshots) { - if (!map[snapshot.scope]) map[snapshot.scope] = snapshot - } - return map - }, [snapshots]) - const visible = snapshots.filter((snapshot) => scopeFilter === 'all' || snapshot.scope === scopeFilter) - - return ( -
-
-
-

整库版本

-

- 每个条目是一份完整封存的整库候选：数据库十表、原始索引正文与向量、对象资源与 Wiki 快照， - 带逐项摘要与兼容契约。封存期间真实写入者被停住，完成后已恢复。 -

-
-
- - 共 {snapshots.length} 份 · 被引用的去重内容 {formatBytes(page?.content_bytes ?? 0)} - - -
-
- - {page && !page.available ? ( -
- 未接线 -

{page.reason || '归档根未配置'}

-
- ) : null} - {error ? ( -
- 读取失败 -

{error}

-
- ) : null} - - {page?.available ? ( - <> -
- {(['internal', 'external'] as const).map((scope) => { - const latest = latestByScope[scope] - return ( -
-
-

{scopeLabel(scope)}

- {latest ? '已封存' : '尚无封存'} -
- {latest ? ( - <> -

封存于 {stamp(latest.created_at)}

-
-
-
文档
-
{metric(latest, 'documents')}
-
-
-
分块
-
{metric(latest, 'chunks')}
-
-
-
索引
-
{metric(latest, 'index_documents')}
-
-
-
资源
-
{metric(latest, 'resources')}
-
-
-
Wiki 页
-
{metric(latest, 'wiki_pages')}
-
-
-
体积
-
{formatBytes(latest.logical_bytes)}
-
-
-

- 快照 {latest.snapshot_id.slice(0, 16)} -

- - ) : ( -

该库还没有封存产物。

- )} -
- {page?.capture_available ? ( - canCapture ? ( - - ) : ( - 发起封存需要知识工程师及以上权限 - ) - ) : ( - {page?.capture_reason} - )} -
-
- ) - })} -
- -
-
- {( - [ - { id: 'all', label: '全部' }, - { id: 'internal', label: '对内' }, - { id: 'external', label: '对外' }, - ] as const - ).map((item) => ( - - ))} -
- {page.capture_note} -
- -
- - 磁盘事实（不是可回收空间）：清单 {formatBytes(page.manifest_file_bytes)} · blob 文件{' '} - {formatBytes(page.blob_file_bytes)}（实际分配 {formatBytes(page.blob_allocated_bytes)}）· 未被任何版本引用{' '} - {page.unreferenced_blobs} 个 / {formatBytes(page.unreferenced_bytes)} - -
- - {visible.length === 0 ? ( -

还没有封存产物。

- ) : ( - - - - - - - - - - - - - - - - - {visible.map((snapshot) => ( - - - - - - - - - - - - - ))} - -
封存时间范围文档分块索引资源Wiki体积模型 / 维度快照
{stamp(snapshot.created_at)}{scopeLabel(snapshot.scope)}{metric(snapshot, 'documents')}{metric(snapshot, 'chunks')}{metric(snapshot, 'index_documents')}{metric(snapshot, 'resources')}{metric(snapshot, 'wiki_pages')} - {formatBytes(snapshot.logical_bytes)} - - {snapshot.compatibility.embedding_model || '—'} / {snapshot.compatibility.embedding_dimension} - - -
- )} - - {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) => ( -
-
- {scopeLabel(snapshot.scope)} · {snapshot.snapshot_id} - - 策略 {snapshot.validation_policy || '—'} · WeKnora {snapshot.compatibility.weknora_version} · - schema {snapshot.compatibility.schema_version} · 映射{' '} - {snapshot.compatibility.index_mapping_sha256.slice(0, 12)} - -
-
    - {snapshot.parts.map((part) => ( -
  • - {part.kind} - {part.name} - {part.count} 项 - {formatBytes(part.bytes)} - {part.sha256.slice(0, 12)} -
    • - ))} -
-
- ))} - - ) : null} -
- ) -} \ No newline at end of file diff --git a/web/src/ReleaseTargets.test.tsx b/web/src/ReleaseTargets.test.tsx new file mode 100644 index 0000000..1aba14e --- /dev/null +++ b/web/src/ReleaseTargets.test.tsx @@ -0,0 +1,133 @@ +import { renderToStaticMarkup } from 'react-dom/server' +import { describe, expect, it, vi } from 'vitest' +import { ReleaseTargetsView } from './ReleaseTargets' +import type { LibraryCaptureJob, LibrarySnapshot, LibrarySnapshotList } from './librarySnapshots' + +const snapshot: LibrarySnapshot = { + snapshot_id: 'f992d64ea45bf737a0721a465b5aef2b150ef7757157eb84f3aaf8b6972d98f9', + scope: 'internal', + knowledge_base_id: '0c84a098-d3ed-4b82-a165-e6de9691d6a8', + tenant_id: 10000, + created_at: '2026-10-28T03:34:57.314178223Z', + validation_policy: 'artifact-acceptance-v1', + format: 'weknora-library-snapshot/v1', + compatibility: { + weknora_version: 'v0.7.1', + schema_version: 74, + retriever_engine: 'elasticsearch_v8', + embedding_model: 'text-embedding-v4', + embedding_dimension: 1024, + index_mapping_sha256: '137cde2965f3b56521f5fb722b4e0617f46d0e509e571b217fe01e090d5e4d01', + }, + parts: [ + { kind: 'database', name: 'knowledges', sha256: 'a'.repeat(64), count: 4, bytes: 10654 }, + { kind: 'index', name: 'chunks', sha256: 'b'.repeat(64), count: 122, bytes: 1624880 }, + ], + logical_bytes: 55309044, + unique_bytes: 55309040, + shared_bytes: 2, + totals: { documents: 4, chunks: 38, index_documents: 122, resources: 21, wiki_pages: 31 }, +} + +const job: LibraryCaptureJob = { + job_id: 'library-capture-internal-1', + capture_job: 'lib-internal-20261028T033457', + scope: 'internal', + state: 'running', + attempt: 1, + last_error: '', + requested_by: 'yaozhanghao', + created_at: '2026-10-28T03:34:50Z', + updated_at: '2026-10-28T03:34:51Z', +} + +const page: LibrarySnapshotList = { + available: true, + reason: '', + archive_root: '/var/lib/knowledge-governance/library', + snapshots: [snapshot], + versions: 2, + content_bytes: 55309693, + blob_file_bytes: 55309693, + blob_allocated_bytes: 55373824, + unreferenced_blobs: 0, + unreferenced_bytes: 0, + manifest_file_bytes: 7544, + manifest_allocated_bytes: 12288, + capture_note: '', + capture_available: true, + capture_reason: '', +} + +const baseProps = { + page, + jobs: [] as LibraryCaptureJob[], + loading: false, + error: null, + notice: null, + expanded: null, + pendingScope: null, + canCapture: true, + onRefresh: () => undefined, + onCapture: vi.fn(), + onToggleExpanded: () => undefined, +} + +const render = (overrides: Partial = {}) => + renderToStaticMarkup() + +describe('ReleaseTargetsView', () => { + it('shows exactly the two library targets with real counts and integrity wording', () => { + const html = render() + expect(html).toContain('对内') + expect(html).toContain('对外') + expect(html).not.toContain('第三目标') + expect(html).toContain('122') + expect(html).toContain('31') + expect(html).toMatch(/完整性：读取时已重算 2 项摘要/) + expect(html).toContain('text-embedding-v4') + }) + + it('keeps the unbuilt chain stages disabled instead of faking progress', () => { + const html = render() + expect(html).toMatch(/生产预备份[\s\S]*未接线/) + expect(html).toMatch(/同步到生产[\s\S]*未授权/) + expect(html).toMatch(/校验 \/ 恢复[\s\S]*未接线/) + expect(html).not.toMatch(/\d+%/) + // Disabled actions carry a reason in the title attribute. + expect(html).toContain('title="生产独立预备份与恢复演练未接线"') + expect(html).toContain('title="生产写入需要独立授权闸口，当前未开放"') + }) + + it('offers the capture action only when the agent is wired and the role allows it', () => { + expect(render()).toContain('发起对内封存') + const readonly = render({ canCapture: false }) + expect(readonly).toContain('发起封存需要知识工程师及以上权限') + const unwired = render({ + page: { ...page, capture_available: false, capture_reason: '未接线：主机侧封存 agent 没有挂载' }, + }) + expect(unwired).toContain('未接线：主机侧封存 agent 没有挂载') + expect(unwired).not.toContain('发起对内封存') + }) + + it('reports the live job instead of a guessed stage', () => { + const html = render({ jobs: [job] }) + expect(html).toContain('执行中') + expect(html).toContain('lib-internal-20261028T033457') + expect(html).toContain('yaozhanghao') + // A running job disables the capture button for that scope. + expect(html).toMatch(/disabled[^>]*>发起对内封存/) + }) + + it('states that storage numbers are content accounting, not reclaimable space', () => { + const html = render() + expect(html).toContain('被版本引用的去重内容') + expect(html).toContain('不是可回收空间') + expect(html).toMatch(/保留策略、依赖保护、清理预览都没有接线/) + }) + + it('shows an empty target as an invitation, with the blocking reason', () => { + const html = render({ page: { ...page, snapshots: [], available: false, reason: '未接线：归档根未配置' } }) + expect(html).toContain('未接线：归档根未配置') + }) +}) \ No newline at end of file diff --git a/web/src/ReleaseTargets.tsx b/web/src/ReleaseTargets.tsx new file mode 100644 index 0000000..90d981d --- /dev/null +++ b/web/src/ReleaseTargets.tsx @@ -0,0 +1,446 @@ +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 './releaseTargets.css' + +// 发布管理的主视图：只服务『对内 / 对外』两个整库目标。 +// +// 这条链有四段：封存 → 生产预备份 → 同步 → 校验/恢复。目前只有第一段接通， +// 其余三段一律禁用并写明"未接线/未授权"——不用假进度、不推算阶段、不显示假成功。 +// 文档级旧流程折叠保留在页面底部，它成功也不代表整库发布成功。 + +const stamp = (value: string) => + value && Number.isFinite(new Date(value).getTime()) + ? new Date(value).toLocaleString('zh-CN', { + timeZone: 'Asia/Shanghai', + month: '2-digit', + day: '2-digit', + hour: '2-digit', + minute: '2-digit', + hour12: false, + }) + : '—' + +const jobStateLabel: Record = { + pending: '排队中', + running: '执行中', + retry: '重试中', + completed: '已完成', + failed: '失败', +} + +const ACTIVE_STATES = ['pending', 'running', 'retry'] +const metric = (snapshot: LibrarySnapshot, key: string) => snapshot.totals[key] ?? 0 + +interface StageProps { + id: string + label: string + state: 'ready' | 'not-wired' | 'not-authorized' + detail: string +} + +function Stage({ label, state, detail }: StageProps) { + const chip = + state === 'ready' ? '已接通' : state === 'not-authorized' ? '未授权' : '未接线' + return ( +
  • + {label} + {chip} + {detail} +
    • + ) +} + +interface ReleaseTargetsViewProps { + page: LibrarySnapshotList | null + jobs: LibraryCaptureJob[] + loading: boolean + error: string | null + notice: string | null + expanded: string | null + pendingScope: string | null + canCapture: boolean + onRefresh: () => void + onCapture: (scope: string) => void + onToggleExpanded: (snapshotID: string) => void +} + +export function ReleaseTargetsView(props: ReleaseTargetsViewProps) { + const { page, jobs, loading, error, notice, expanded, pendingScope, canCapture, onRefresh, onCapture, onToggleExpanded } = props + const latestByScope = useMemo(() => { + const map: Record = {} + for (const snapshot of page?.snapshots ?? []) { + if (!map[snapshot.scope]) map[snapshot.scope] = snapshot + } + return map + }, [page]) + + return ( +
    +
    +
    +

    发布管理

    +

    + 两个目标：对内与对外。每个目标是一份完整封存的整库版本，不是若干单篇文档。 + 这条链只有第一段接通，其余动作按真实状态禁用。 +

    +
    + +
    + +
      + + + + +
    + + {error ? ( +
    + 读取失败 +

    {error}

    +
    + ) : null} + {page && !page.available ? ( +
    + 归档未接线 +

    {page.reason || '整库归档根未配置'}

    +
    + ) : null} + {notice ?

    {notice}

    : null} + +
    + {(['internal', 'external'] as const).map((scope) => { + const snapshot = latestByScope[scope] + const activeJob = jobs.find((job) => job.scope === scope && ACTIVE_STATES.includes(job.state)) + const lastJob = jobs.find((job) => job.scope === scope) + const expandedHere = snapshot && expanded === snapshot.snapshot_id + const blockedReason = !page + ? '正在读取归档' + : !page.available + ? page.reason || '归档根未配置' + : !page.capture_available + ? page.capture_reason || '主机侧封存 agent 未接线' + : null + return ( +
    +
    +

    {scopeLabel(scope)}

    + + {snapshot ? '已封存' : '尚无封存'} + +
    + + {snapshot ? ( + <> +

    + 封存于 {stamp(snapshot.created_at)} · 快照 {snapshot.snapshot_id.slice(0, 16)} +

    +
    +
    +
    文档
    +
    {metric(snapshot, 'documents')}
    +
    +
    +
    分块
    +
    {metric(snapshot, 'chunks')}
    +
    +
    +
    索引
    +
    {metric(snapshot, 'index_documents')}
    +
    +
    +
    资源
    +
    {metric(snapshot, 'resources')}
    +
    +
    +
    Wiki 页
    +
    {metric(snapshot, 'wiki_pages')}
    +
    +
    +
    内容字节
    +
    {formatBytes(snapshot.logical_bytes)}
    +
    +
    +

    + 完整性：读取时已重算 {snapshot.parts.length} 项摘要并逐项匹配（A04 口径）。 + 模型 {snapshot.compatibility.embedding_model || '未知'} /{' '} + {snapshot.compatibility.embedding_dimension || '?'} 维 · schema{' '} + {snapshot.compatibility.schema_version || '?'} · 策略 {snapshot.validation_policy || '—'} +

    + + {expandedHere ? ( +
    + + + + + + + + + + + + {snapshot.parts.map((part) => ( + + + + + + + + ))} + +
    类别分项数量字节摘要（SHA256）
    {part.kind}{part.name}{part.count}{part.bytes.toLocaleString('zh-CN')} + {part.sha256.slice(0, 12)} +
    +

    + 校验结果：{snapshot.parts.length} 项全部可读且摘要一致；未发现异常。 + 这证明的是数据完整性，不是效果测试（A06 已取消）。 +

    +
    + ) : null} + + ) : ( +

    + {blockedReason || '还没有封存产物。先发起一次封存，产出一份完整版本。'} +

    + )} + +
    + {activeJob ? ( + <> + + {jobStateLabel[activeJob.state] ?? activeJob.state} + + {activeJob.capture_job} + {activeJob.requested_by || '—'} + + ) : lastJob ? ( + <> + + {jobStateLabel[lastJob.state] ?? lastJob.state} + + 上次作业 {lastJob.capture_job} + {lastJob.last_error ? ( + + {lastJob.last_error.slice(0, 80)} + + ) : null} + + ) : ( + 当前没有封存作业 + )} +
    + +
    + {page?.capture_available ? ( + canCapture ? ( + + ) : ( + 发起封存需要知识工程师及以上权限 + ) + ) : ( + {page?.capture_reason || '封存未接线'} + )} + + + +
    +

    + 禁用项不是流程建议，而是能力状态：预备份、同步、校验恢复都没有接线，点击不会发生任何事。 +

    +
    + ) + })} +
    + + {jobs.length > 0 ? ( +
    +

    封存作业

    + + + + + + + + + + + + + {jobs.map((job) => ( + + + + + + + + + ))} + +
    投递时间目标状态作业号发起人最后错误
    {stamp(job.created_at)}{scopeLabel(job.scope)}{jobStateLabel[job.state] ?? job.state} + {job.capture_job} + {job.requested_by || '—'}{job.last_error || '—'}
    +
    + ) : null} + +
    +

    归档存储

    + {page?.available ? ( + <> +
    +
    +
    被版本引用的去重内容
    +
    {formatBytes(page.content_bytes)}
    +
    +
    +
    blob 表观字节
    +
    {formatBytes(page.blob_file_bytes)}
    +
    +
    +
    实际分配字节
    +
    {formatBytes(page.blob_allocated_bytes)}
    +
    +
    +
    未被任何版本引用
    +
    + {page.unreferenced_blobs} 个 / {formatBytes(page.unreferenced_bytes)} +
    +
    +
    +
    清单文件
    +
    {formatBytes(page.manifest_file_bytes)}
    +
    +
    +
    版本数
    +
    {page.versions}
    +
    +
    +

    + 去重内容按跨版本重新计算；这些都不是可回收空间。 + 保留策略、依赖保护、清理预览都没有接线，自动清理保持关闭，页面不会删除任何东西。 +

    + + ) : ( +

    {page?.reason || '归档未接线，存储口径不可读。'}

    + )} +
    +
    + ) +} + +export function ReleaseTargets() { + const auth = useOptionalAuth() + const [page, setPage] = useState(null) + const [jobs, setJobs] = useState([]) + const [loading, setLoading] = useState(true) + const [error, setError] = useState(null) + const [expanded, setExpanded] = useState(null) + const [pendingScope, setPendingScope] = useState(null) + const [notice, setNotice] = useState(null) + + const load = useCallback(async (signal?: AbortSignal) => { + setLoading(true) + try { + const [snapshots, captures] = await Promise.all([ + fetchLibrarySnapshots(signal), + fetchLibraryCaptures(signal), + ]) + setPage(snapshots) + setJobs(captures) + setError(null) + } catch (loadError) { + if ((loadError as Error).name === 'AbortError') return + setError(loadError instanceof Error ? loadError.message : '加载整库发布状态失败') + } finally { + setLoading(false) + } + }, []) + + useEffect(() => { + const controller = new AbortController() + void load(controller.signal) + return () => controller.abort() + }, [load]) + + const busy = jobs.some((job) => ACTIVE_STATES.includes(job.state)) + useEffect(() => { + if (!busy) return + const timer = window.setInterval(() => { + void load().catch(() => undefined) + }, 3000) + return () => window.clearInterval(timer) + }, [busy, load]) + + const role = auth?.identity?.role + const canCapture = (role === 'engineer' || role === 'admin') && page?.capture_available === true + + const startCapture = async (scope: string) => { + setPendingScope(scope) + setNotice(null) + try { + const accepted = await enqueueLibraryCapture(scope) + setNotice( + accepted.reused + ? `该作业已存在：${scopeLabel(scope)} · ${accepted.capture_job}（状态 ${jobStateLabel[accepted.state] ?? accepted.state}）` + : `已投递封存作业：${scopeLabel(scope)} · ${accepted.capture_job}` + ) + await load() + } catch (captureError) { + setNotice(captureError instanceof Error ? captureError.message : '发起封存失败') + } finally { + setPendingScope(null) + } + } + + return ( + void load()} + onCapture={(scope) => void startCapture(scope)} + onToggleExpanded={(snapshotID) => setExpanded(expanded === snapshotID ? null : snapshotID)} + /> + ) +} diff --git a/web/src/librarySnapshots.ts b/web/src/librarySnapshots.ts index 30aa721..aa04522 100644 --- a/web/src/librarySnapshots.ts +++ b/web/src/librarySnapshots.ts @@ -54,6 +54,9 @@ export interface LibraryCaptureAccepted { capture_job: string scope: string state: string + // true when the idempotency key matched an existing job; the response then + // reports the stored job's scope, not the requested one. + reused: boolean } export interface LibrarySnapshotList { @@ -178,6 +181,7 @@ export async function enqueueLibraryCapture(scope: string): Promise summary { + padding: 14px 18px; + cursor: pointer; + font-size: 13px; + color: var(--ink-mute, #64748b); +} +.release-legacy[open] > summary { + border-bottom: 1px solid var(--border-base, #e2e8f0); +} + +@media (max-width: 860px) { + .release-targets { + padding: 12px; + } + .release-chain { + grid-template-columns: repeat(2, minmax(0, 1fr)); + } + .release-legacy { + margin: 0 12px 12px; + } +} \ No newline at end of file diff --git a/web/src/workspace.ts b/web/src/workspace.ts index 086d982..ff6f0bd 100644 --- a/web/src/workspace.ts +++ b/web/src/workspace.ts @@ -4,7 +4,6 @@ export type ModuleId = | 'knowledge' | 'tasks' | 'releases' - | 'library' | 'settings' | 'versions' | 'qaops' @@ -14,7 +13,7 @@ export interface WorkspaceState { activeTab: ModuleId } -const moduleIds = new Set(['dashboard', 'tasks', 'releases', 'library', 'settings', 'qaops']) +const moduleIds = new Set(['dashboard', 'tasks', 'releases', 'settings', 'qaops']) export const initialWorkspace: WorkspaceState = { openTabs: ['dashboard'],