diff --git a/CHANGELOG.md b/CHANGELOG.md
index e79a661..8d557d4 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,28 @@
# Changelog
+## 开发快照 0.1.366（启动收敛误判修复与误判核对审计）
+
+### Fixed
+
+- `ReconcileLibraryJobs` 启动收敛不再把仍为 pending/running、但 capture queue 已
+ `completed` 的封存业务作业误判为“队列不存在”；completed 队列保留待核对状态，
+ failed/cancelled/missing 仍按真实队列状态处置。
+- 管理员核对入口增加唯一允许的“系统误判”失败形态：原 state=failed、stage 为空、
+ snapshot 为空且 error/recovery 精确匹配历史启动误判；仍须 completed queue、
+ export succeeded、唯一 test_sealed 版本、来源身份和归档字节复核。真实失败/取消/
+ 其他 error 一律拒绝。
+- 新增 `library_job_repairs` 不可变审计表（migration 80）：在同一事务内 CAS 比对原
+ state/stage/error/recovery/finished_at，保存原值、actor、原因和精确证据后再修复原行；
+ 审计写入失败则回滚，不静默覆盖。
+- UI 仅在该精确误判形态显示“核对封存结果”，真实失败没有入口；成功后刷新原作业
+ 和快照状态。
+
+### Tests
+
+- 真实隔离 PostgreSQL：启动收敛 pending+completed 保留 RED/GREEN；真实 handler
+ 纵向启动→核对 pending 与系统误判 failed；审计原字段与重复幂等；错误 failed/取消/
+ 缺失/错证据拒绝。
+
## 开发快照 0.1.365（封存终态回写与精确核对）
### Fixed
diff --git a/internal/httpapi/library_capture_reconcile.go b/internal/httpapi/library_capture_reconcile.go
index b387033..7686527 100644
--- a/internal/httpapi/library_capture_reconcile.go
+++ b/internal/httpapi/library_capture_reconcile.go
@@ -24,8 +24,14 @@ type libraryCaptureReconciler interface {
LibraryTestSealedVersionForJob(ctx context.Context, scope, jobID string) (governance.LibraryVersion, bool, error)
EnsureLibraryOperation(ctx context.Context, op governance.LibraryOperation) (governance.LibraryOperation, bool, error)
ReconcileLibraryCaptureTerminal(ctx context.Context, jobID, scope, queueJobID, snapshotID string, now time.Time) (bool, error)
+ ReconcileMisclassifiedLibraryCapture(ctx context.Context, jobID, scope, queueJobID, snapshotID, expectedStage, expectedError, expectedRecovery string, expectedFinished *time.Time, auditID, actor, reason string, evidence map[string]any, now time.Time) (bool, error)
}
+const (
+ legacyOrphanError = "队列中已无对应作业，业务作业未收到结果"
+ legacyOrphanRecovery = "封存作业由 agent 恢复来源；确认来源 healthy 后可重发"
+)
+
type libraryCaptureQueueReader interface {
QueueJob(ctx context.Context, id string) (jobs.Job, bool, error)
}
@@ -61,8 +67,11 @@ func libraryCaptureReconcile(w http.ResponseWriter, r *http.Request, config Conf
conflict(w, "只有封存作业可以核对封存结果")
return
}
- if job.State != governance.LibraryJobPending && job.State != governance.LibraryJobRunning && job.State != governance.LibraryJobSucceeded {
- conflict(w, "作业已是失败/取消终态，不能改写成成功")
+ systemMisclassified := job.State == governance.LibraryJobFailed && job.Stage == "" &&
+ job.Error == legacyOrphanError && job.Recovery == legacyOrphanRecovery && job.SnapshotID == ""
+ if job.State != governance.LibraryJobPending && job.State != governance.LibraryJobRunning &&
+ job.State != governance.LibraryJobSucceeded && !systemMisclassified {
+ conflict(w, "作业是真实失败/取消终态，不能改写成成功")
return
}
if job.QueueJobID == "" {
@@ -176,6 +185,30 @@ func libraryCaptureReconcile(w http.ResponseWriter, r *http.Request, config Conf
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot build reconcile audit digest"})
return
}
+ if systemMisclassified {
+ auditID := "librepair-" + digest[:16]
+ updated, repairErr := store.ReconcileMisclassifiedLibraryCapture(r.Context(), job.ID, job.Scope,
+ job.QueueJobID, version.SnapshotID, job.Stage, legacyOrphanError, legacyOrphanRecovery,
+ job.FinishedAt, auditID, actor, "completed capture queue + export succeeded + exact version + archive verified; legacy orphan misclassification", map[string]any{
+ "original_state": job.State, "original_error": job.Error, "original_recovery": job.Recovery,
+ "original_finished_at": job.FinishedAt, "queue_job_id": queue.ID,
+ "snapshot_id": version.SnapshotID, "version_id": version.ID, "archive_verified": true,
+ }, now)
+ if repairErr != nil {
+ writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "cannot persist misclassification repair"})
+ return
+ }
+ if !updated {
+ conflict(w, "原误判行状态/错误/完成时间已变化，拒绝覆盖")
+ return
+ }
+ writeJSON(w, http.StatusOK, map[string]any{
+ "repaired": true, "already_succeeded": false, "misclassification_repaired": true,
+ "job_id": job.ID, "snapshot_id": version.SnapshotID, "queue_job_id": queue.ID,
+ "stage": "export", "version_id": version.ID, "audit_repair_id": auditID,
+ })
+ return
+ }
opID := "libop-capture-reconcile-" + digest[:16]
op, _, err := store.EnsureLibraryOperation(r.Context(), governance.LibraryOperation{
ID: opID, Scope: job.Scope, Kind: governance.LibraryJobRecover,
diff --git a/internal/httpapi/library_capture_reconcile_test.go b/internal/httpapi/library_capture_reconcile_test.go
index a5dfe6a..7ccd1be 100644
--- a/internal/httpapi/library_capture_reconcile_test.go
+++ b/internal/httpapi/library_capture_reconcile_test.go
@@ -99,6 +99,30 @@ func (f *fakeLibraryStore) ReconcileLibraryCaptureTerminal(_ context.Context, jo
}
return false, nil
}
+
+func (f *fakeLibraryStore) ReconcileMisclassifiedLibraryCapture(_ context.Context, jobID, scope, queueJobID, snapshotID, expectedStage, expectedError, expectedRecovery string, expectedFinished *time.Time, _ string, _ string, _ string, _ map[string]any, _ time.Time) (bool, error) {
+ if f.reconcileDenied {
+ return false, nil
+ }
+ for index := range f.jobs {
+ j := &f.jobs[index]
+ if j.ID != jobID || j.Scope != scope || j.Kind != governance.LibraryJobSeal || j.QueueJobID != queueJobID ||
+ j.State != governance.LibraryJobFailed || j.Stage != expectedStage || j.Error != expectedError ||
+ j.Recovery != expectedRecovery || j.SnapshotID != "" || (j.FinishedAt == nil) != (expectedFinished == nil) ||
+ (j.FinishedAt != nil && expectedFinished != nil && !j.FinishedAt.Equal(*expectedFinished)) {
+ return false, nil
+ }
+ j.State = governance.LibraryJobSucceeded
+ j.SnapshotID = snapshotID
+ j.Stage = "export"
+ f.lastState = governance.LibraryJobSucceeded
+ return true, nil
+ }
+ return false, nil
+}
+
+var _ libraryCaptureReconciler = (*fakeLibraryStore)(nil)
+
func runReconcile(t *testing.T, config Config, jobID string) *httptest.ResponseRecorder {
t.Helper()
req := httptest.NewRequest(http.MethodPost, "/api/v1/library/jobs/"+jobID+"/reconcile-capture", nil)
@@ -271,3 +295,67 @@ func TestLibraryCaptureReconcileConditionalWriteRejectsConcurrentState(t *testin
t.Fatalf("code=%d advanced=%q body=%s", rec.Code, store.lastState, rec.Body.String())
}
}
+
+func TestLibraryCaptureReconcileRepairsExactLegacyMisclassification(t *testing.T) {
+ store, config, job, _ := reconcileFixture(t)
+ finished := time.Now().UTC()
+ job.State = governance.LibraryJobFailed
+ job.Stage = ""
+ job.Error = legacyOrphanError
+ job.Recovery = legacyOrphanRecovery
+ job.FinishedAt = &finished
+ store.jobs[0] = job
+ rec := runReconcile(t, config, job.ID)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("code=%d body=%s", rec.Code, rec.Body.String())
+ }
+ var body map[string]any
+ _ = json.Unmarshal(rec.Body.Bytes(), &body)
+ if body["misclassification_repaired"] != true || store.lastState != governance.LibraryJobSucceeded {
+ t.Fatalf("body=%v advanced=%q", body, store.lastState)
+ }
+}
+
+func TestLibraryCaptureReconcileRejectsOtherFailedError(t *testing.T) {
+ store, config, job, _ := reconcileFixture(t)
+ finished := time.Now().UTC()
+ job.State = governance.LibraryJobFailed
+ job.Stage = ""
+ job.Error = "capture export failed"
+ job.Recovery = legacyOrphanRecovery
+ job.FinishedAt = &finished
+ store.jobs[0] = job
+ rec := runReconcile(t, config, job.ID)
+ if rec.Code != http.StatusConflict || store.lastState != "" {
+ t.Fatalf("code=%d advanced=%q body=%s", rec.Code, store.lastState, rec.Body.String())
+ }
+}
+
+func TestLibraryCaptureReconcileRejectsRealFailureAndCancel(t *testing.T) {
+ for _, state := range []governance.LibraryJobState{governance.LibraryJobFailed, governance.LibraryJobCancelled} {
+ store, config, job, _ := reconcileFixture(t)
+ job.State = state
+ job.Error = "real failure"
+ store.jobs[0] = job
+ rec := runReconcile(t, config, job.ID)
+ if rec.Code != http.StatusConflict || store.lastState != "" {
+ t.Fatalf("state=%s code=%d advanced=%q", state, rec.Code, store.lastState)
+ }
+ }
+}
+
+func TestLibraryCaptureReconcileMisclassificationCASRejectsChange(t *testing.T) {
+ store, config, job, _ := reconcileFixture(t)
+ finished := time.Now().UTC()
+ job.State = governance.LibraryJobFailed
+ job.Stage = ""
+ job.Error = legacyOrphanError
+ job.Recovery = legacyOrphanRecovery
+ job.FinishedAt = &finished
+ store.jobs[0] = job
+ store.reconcileDenied = true
+ rec := runReconcile(t, config, job.ID)
+ if rec.Code != http.StatusConflict || store.lastState != "" {
+ t.Fatalf("code=%d advanced=%q", rec.Code, store.lastState)
+ }
+}
diff --git a/internal/httpapi/library_capture_reconcile_vertical_pg_test.go b/internal/httpapi/library_capture_reconcile_vertical_pg_test.go
new file mode 100644
index 0000000..848f4d8
--- /dev/null
+++ b/internal/httpapi/library_capture_reconcile_vertical_pg_test.go
@@ -0,0 +1,204 @@
+package httpapi
+
+import (
+ "context"
+ "encoding/json"
+ "net/http"
+ "net/http/httptest"
+ "os"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/jackc/pgx/v5/pgxpool"
+
+ "knowledge-governance/internal/governance"
+ "knowledge-governance/internal/librarysnapshot"
+ postgresstore "knowledge-governance/internal/platform/postgres"
+)
+
+// Vertical PostgreSQL path: the real worker startup reconciliation must leave
+// a pending seal with a completed capture queue intact, then the actual HTTP
+// handler must verify archive/version/stage and repair it. This is not a fake
+// store or a direct helper test.
+func verticalReconcileFixture(t *testing.T) (*postgresstore.Store, Config, string, *pgxpool.Pool) {
+ t.Helper()
+ dsn := os.Getenv("KG_LIBRARY_TEST_DATABASE_URL")
+ if dsn == "" {
+ t.Skip("requires explicitly isolated kg_library_test_ PostgreSQL database")
+ }
+ cfg, err := pgxpool.ParseConfig(dsn)
+ if err != nil || !strings.HasPrefix(cfg.ConnConfig.Database, "kg_library_test_") || os.Getenv("KG_LIBRARY_TEST_ISOLATED") != "1" {
+ t.Fatal("isolated database required")
+ }
+ cfg.MaxConns = 1
+ cfg.ConnConfig.RuntimeParams["search_path"] = "pg_temp"
+ ctx := context.Background()
+ pool, err := pgxpool.NewWithConfig(ctx, cfg)
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(pool.Close)
+ if _, err := pool.Exec(ctx, `
+CREATE TEMP TABLE library_jobs(
+ id varchar(64) PRIMARY KEY, scope varchar(16) NOT NULL, kind varchar(16) NOT NULL,
+ state varchar(16) NOT NULL, stage varchar(32) NOT NULL DEFAULT '',
+ idempotency_key varchar(255) NOT NULL DEFAULT '', request_digest varchar(64) NOT NULL DEFAULT '',
+ queue_job_id varchar(64) NOT NULL DEFAULT '', snapshot_id varchar(64) NOT NULL DEFAULT '',
+ target_snapshot_id varchar(64) NOT NULL DEFAULT '', actor varchar(128) NOT NULL DEFAULT '',
+ error text NOT NULL DEFAULT '', recovery varchar(255) NOT NULL DEFAULT '',
+ operation_id varchar(64) NOT NULL DEFAULT '', attempt_epoch integer NOT NULL DEFAULT 0,
+ created_at timestamptz NOT NULL, updated_at timestamptz NOT NULL, finished_at timestamptz,
+ target_id text NOT NULL DEFAULT '');
+CREATE TEMP TABLE library_job_stages(
+ job_id varchar(64) NOT NULL, stage varchar(32) NOT NULL, state varchar(16) NOT NULL,
+ total bigint, completed bigint, bytes_total bigint, bytes_done bigint,
+ detail text NOT NULL DEFAULT '', updated_at timestamptz NOT NULL, PRIMARY KEY(job_id,stage));
+CREATE TEMP TABLE library_versions(
+ id varchar(64) PRIMARY KEY, scope varchar(16) NOT NULL, kind varchar(24) NOT NULL,
+ snapshot_id varchar(64) NOT NULL DEFAULT '', version_label varchar(64) NOT NULL DEFAULT '',
+ source_job_id varchar(64) NOT NULL DEFAULT '', operation_id varchar(64) NOT NULL DEFAULT '',
+ attempt_epoch integer NOT NULL DEFAULT 0, evidence jsonb NOT NULL DEFAULT '{}'::jsonb,
+ created_by varchar(128) NOT NULL DEFAULT '', created_at timestamptz NOT NULL, superseded_at timestamptz);
+CREATE TEMP TABLE library_job_repairs(
+ id varchar(64) PRIMARY KEY, job_id varchar(64) NOT NULL, scope varchar(16) NOT NULL,
+ kind varchar(16) NOT NULL, queue_job_id varchar(64) NOT NULL, snapshot_id varchar(64) NOT NULL,
+ original_state varchar(16) NOT NULL, original_stage varchar(32) NOT NULL DEFAULT '',
+ original_error text NOT NULL DEFAULT '', original_recovery varchar(255) NOT NULL DEFAULT '',
+ original_finished_at timestamptz, reason varchar(255) NOT NULL, actor varchar(128) NOT NULL DEFAULT '',
+ evidence jsonb NOT NULL DEFAULT '{}'::jsonb, created_at timestamptz NOT NULL);
+CREATE TEMP TABLE library_operations(
+ id varchar(64) PRIMARY KEY, scope varchar(16) NOT NULL, kind varchar(16) NOT NULL,
+ target_snapshot_id varchar(64) NOT NULL DEFAULT '', request_digest varchar(64) NOT NULL,
+ operation_key varchar(255) NOT NULL, actor varchar(128) NOT NULL DEFAULT '',
+ created_at timestamptz NOT NULL, target_id text NOT NULL DEFAULT '',
+ first_backup_snapshot_id varchar(64), first_backup_bytes bigint, first_backup_at timestamptz,
+ writeset_digest varchar(64), writeset_rows bigint, target_fingerprint varchar(128), anchors_at timestamptz);
+CREATE UNIQUE INDEX library_operations_key_test ON library_operations(scope,kind,operation_key);
+CREATE TEMP TABLE jobs(
+ id varchar(64) PRIMARY KEY, type varchar(64) NOT NULL, payload jsonb NOT NULL,
+ state varchar(16) NOT NULL, priority integer NOT NULL DEFAULT 0, available_at timestamptz NOT NULL,
+ lease_owner varchar(128), lease_until timestamptz, attempt integer NOT NULL DEFAULT 0,
+ max_attempts integer NOT NULL DEFAULT 1, last_error text, idempotency_key varchar(255) NOT NULL,
+ created_at timestamptz NOT NULL, updated_at timestamptz NOT NULL);`); err != nil {
+ t.Fatal(err)
+ }
+ root := libraryFixtureRoot(t)
+ snapshotID := sealSharedVersions(t, root, 1, 4)[0]
+ archive, err := librarysnapshot.Open(root, 256*1024*1024)
+ if err != nil {
+ t.Fatal(err)
+ }
+ manifest, err := archive.Load(snapshotID)
+ archive.Close()
+ if err != nil {
+ t.Fatal(err)
+ }
+ config := Config{LibraryArchiveRoot: root,
+ LibrarySourceProject: "fixture-project", LibrarySourceApp: "fixture-app",
+ LibrarySourceTenantID: manifest.TenantID, LibrarySourceInternalKB: manifest.KnowledgeBaseID}
+ source := configuredLibrarySource(config, "internal")
+ sourceID, err := source.ID()
+ if err != nil {
+ t.Fatal(err)
+ }
+ now := time.Now().UTC()
+ payload, _ := json.Marshal(map[string]string{"scope": "internal", "capture_job": "vertical"})
+ if _, err := pool.Exec(ctx, `INSERT INTO jobs(id,type,payload,state,available_at,idempotency_key,created_at,updated_at)
+ VALUES('q-vertical','library.capture',$1,'completed',$2,'library.capture:vertical',$2,$2)`, payload, now); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := pool.Exec(ctx, `INSERT INTO library_versions(id,scope,kind,snapshot_id,source_job_id,evidence,created_at)
+ VALUES('ver-vertical','internal','test_sealed',$1,'libseal-vertical',$2,$3)`, snapshotID,
+ map[string]any{"source_id": sourceID, "capture_job": "vertical"}, now); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := pool.Exec(ctx, `INSERT INTO library_job_stages(job_id,stage,state,detail,updated_at)
+ VALUES('libseal-vertical','export','succeeded','已封存快照', $1)`, now); err != nil {
+ t.Fatal(err)
+ }
+ store := postgresstore.NewStore(pool)
+ config.Store = store
+ if _, err := pool.Exec(ctx, `INSERT INTO library_jobs(id,scope,kind,state,idempotency_key,request_digest,queue_job_id,created_at,updated_at)
+ VALUES('libseal-vertical','internal','seal','pending','k','d','q-vertical',$1,$1)`, now); err != nil {
+ t.Fatal(err)
+ }
+ return store, config, snapshotID, pool
+}
+
+func TestStartupReconcileThenHandlerRepairsPendingCapture(t *testing.T) {
+ store, config, _, _ := verticalReconcileFixture(t)
+ ctx := context.Background()
+ now := time.Now().UTC()
+ if n, err := store.ReconcileLibraryJobs(ctx, now, []string{"seal"}); err != nil || n != 0 {
+ t.Fatalf("startup reconcile n=%d err=%v", n, err)
+ }
+ job, err := store.LibraryJob(ctx, "libseal-vertical")
+ if err != nil || job.State != governance.LibraryJobPending {
+ t.Fatalf("startup changed job: %+v err=%v", job, err)
+ }
+ req := httptest.NewRequest(http.MethodPost, "/api/v1/library/jobs/libseal-vertical/reconcile-capture", nil)
+ req.SetPathValue("id", "libseal-vertical")
+ rec := httptest.NewRecorder()
+ libraryCaptureReconcile(rec, req, config)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("handler code=%d body=%s", rec.Code, rec.Body.String())
+ }
+ job, err = store.LibraryJob(ctx, "libseal-vertical")
+ if err != nil || job.State != governance.LibraryJobSucceeded || job.SnapshotID == "" {
+ t.Fatalf("handler did not repair: %+v err=%v", job, err)
+ }
+}
+
+func TestStartupReconcilePreservesFailedCapturesForExactReconcile(t *testing.T) {
+ store, config, snapshotID, pool := verticalReconcileFixture(t)
+ ctx := context.Background()
+ now := time.Now().UTC()
+ source := configuredLibrarySource(config, "internal")
+ sourceID, err := source.ID()
+ if err != nil {
+ t.Fatal(err)
+ }
+ payload, _ := json.Marshal(map[string]string{"scope": "internal", "capture_job": "vertical-failed"})
+ if _, err := pool.Exec(ctx, `INSERT INTO jobs(id,type,payload,state,available_at,idempotency_key,created_at,updated_at)
+ VALUES('q-failed','library.capture',$1,'completed',$2,'library.capture:vertical-failed',$2,$2)`, payload, now); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := pool.Exec(ctx, `INSERT INTO library_versions(id,scope,kind,snapshot_id,source_job_id,evidence,created_at)
+ VALUES('ver-failed','internal','test_sealed',$1,'libseal-vertical-failed',$2,$3)`,
+ snapshotID, map[string]any{"source_id": sourceID, "capture_job": "vertical-failed"}, now); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := pool.Exec(ctx, `INSERT INTO library_job_stages(job_id,stage,state,detail,updated_at)
+ VALUES('libseal-vertical-failed','export','succeeded','已封存快照', $1)`, now); err != nil {
+ t.Fatal(err)
+ }
+ finished := now.Truncate(time.Microsecond)
+ if _, err := pool.Exec(ctx, `INSERT INTO library_jobs(id,scope,kind,state,idempotency_key,request_digest,queue_job_id,error,recovery,created_at,updated_at,finished_at)
+ VALUES('libseal-vertical-failed','internal','seal','failed','kf','df','q-failed',$1,$2,$3,$3,$3)`,
+ "队列中已无对应作业，业务作业未收到结果", "封存作业由 agent 恢复来源；确认来源 healthy 后可重发", finished); err != nil {
+ t.Fatal(err)
+ }
+ if n, err := store.ReconcileLibraryJobs(ctx, now, []string{"seal"}); err != nil || n != 0 {
+ t.Fatalf("startup reconcile n=%d err=%v", n, err)
+ }
+ req := httptest.NewRequest(http.MethodPost, "/api/v1/library/jobs/libseal-vertical-failed/reconcile-capture", nil)
+ req.SetPathValue("id", "libseal-vertical-failed")
+ rec := httptest.NewRecorder()
+ libraryCaptureReconcile(rec, req, config)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("handler code=%d body=%s", rec.Code, rec.Body.String())
+ }
+ job, err := store.LibraryJob(ctx, "libseal-vertical-failed")
+ if err != nil || job.State != governance.LibraryJobSucceeded || job.SnapshotID != snapshotID {
+ t.Fatalf("failed misclassification not repaired: %+v err=%v", job, err)
+ }
+ var originalState, originalError string
+ if err := pool.QueryRow(ctx, `SELECT original_state, original_error FROM library_job_repairs WHERE job_id='libseal-vertical-failed'`).
+ Scan(&originalState, &originalError); err != nil {
+ t.Fatal(err)
+ }
+ if originalState != "failed" || originalError == "" {
+ t.Fatalf("audit missing original: state=%q error=%q", originalState, originalError)
+ }
+}
diff --git a/internal/platform/postgres/library_capture_reconcile_pg_test.go b/internal/platform/postgres/library_capture_reconcile_pg_test.go
index fcbbefd..6c87ec5 100644
--- a/internal/platform/postgres/library_capture_reconcile_pg_test.go
+++ b/internal/platform/postgres/library_capture_reconcile_pg_test.go
@@ -35,10 +35,17 @@ CREATE TEMP TABLE library_jobs(
id varchar(64) PRIMARY KEY, scope varchar(16) NOT NULL, kind varchar(16) NOT NULL,
state varchar(16) NOT NULL, queue_job_id varchar(64) NOT NULL DEFAULT '',
stage varchar(32) NOT NULL DEFAULT '', snapshot_id varchar(64) NOT NULL DEFAULT '',
- error text NOT NULL DEFAULT '', updated_at timestamptz NOT NULL, finished_at timestamptz);
+ error text NOT NULL DEFAULT '', recovery varchar(255) NOT NULL DEFAULT '', updated_at timestamptz NOT NULL, finished_at timestamptz);
CREATE TEMP TABLE library_job_stages(
job_id varchar(64) NOT NULL, stage varchar(32) NOT NULL, state varchar(16) NOT NULL,
detail text NOT NULL DEFAULT '', updated_at timestamptz NOT NULL, PRIMARY KEY(job_id,stage));
+CREATE TEMP TABLE library_job_repairs(
+ id varchar(64) PRIMARY KEY, job_id varchar(64) NOT NULL, scope varchar(16) NOT NULL,
+ kind varchar(16) NOT NULL, queue_job_id varchar(64) NOT NULL, snapshot_id varchar(64) NOT NULL,
+ original_state varchar(16) NOT NULL, original_stage varchar(32) NOT NULL DEFAULT '',
+ original_error text NOT NULL DEFAULT '', original_recovery varchar(255) NOT NULL DEFAULT '',
+ original_finished_at timestamptz, reason varchar(255) NOT NULL, actor varchar(128) NOT NULL DEFAULT '',
+ evidence jsonb NOT NULL DEFAULT '{}'::jsonb, created_at timestamptz NOT NULL);
CREATE TEMP TABLE library_versions(
id varchar(64) PRIMARY KEY, scope varchar(16) NOT NULL, kind varchar(24) NOT NULL,
snapshot_id varchar(64) NOT NULL DEFAULT '', version_label varchar(64) NOT NULL DEFAULT '',
@@ -94,4 +101,31 @@ INSERT INTO library_versions(id,scope,kind,snapshot_id,source_job_id,created_at)
if updated, err := store.ReconcileLibraryCaptureTerminal(ctx, "libseal-other", "internal", "libcap-3", "snap-3", now); err != nil || updated {
t.Fatalf("different existing snapshot must not be overwritten: updated=%v err=%v", updated, err)
}
+
+ legacyFinished := now.Add(-time.Minute).Truncate(time.Microsecond)
+ if _, err := pool.Exec(ctx, `INSERT INTO library_jobs(id,scope,kind,state,queue_job_id,error,recovery,updated_at,finished_at)
+ VALUES('libseal-misclassified','internal','seal','failed','libcap-4',$1,$2,$3,$3)`,
+ "队列中已无对应作业，业务作业未收到结果", "封存作业由 agent 恢复来源；确认来源 healthy 后可重发", legacyFinished); err != nil {
+ t.Fatal(err)
+ }
+ repaired, err := store.ReconcileMisclassifiedLibraryCapture(ctx, "libseal-misclassified", "internal", "libcap-4", "snap-4",
+ "", "队列中已无对应作业，业务作业未收到结果", "封存作业由 agent 恢复来源；确认来源 healthy 后可重发", &legacyFinished,
+ "librepair-test", "tester", "test repair", map[string]any{"archive_verified": true}, now)
+ if err != nil || !repaired {
+ t.Fatalf("misclassification repair=%v err=%v", repaired, err)
+ }
+ var originalState, originalError string
+ var originalFinished *time.Time
+ if err := pool.QueryRow(ctx, `SELECT original_state, original_error, original_finished_at FROM library_job_repairs WHERE id='librepair-test'`).
+ Scan(&originalState, &originalError, &originalFinished); err != nil {
+ t.Fatal(err)
+ }
+ if originalState != "failed" || originalError == "" || originalFinished == nil {
+ t.Fatalf("audit lost original fields: state=%q error=%q finished=%v", originalState, originalError, originalFinished)
+ }
+ if repaired, err := store.ReconcileMisclassifiedLibraryCapture(ctx, "libseal-misclassified", "internal", "libcap-4", "snap-4",
+ "", "队列中已无对应作业，业务作业未收到结果", "封存作业由 agent 恢复来源；确认来源 healthy 后可重发", &legacyFinished,
+ "librepair-test-2", "tester", "test repair", map[string]any{}, now); err != nil || repaired {
+ t.Fatalf("second repair must be false: repaired=%v err=%v", repaired, err)
+ }
}
diff --git a/internal/platform/postgres/library_jobs.go b/internal/platform/postgres/library_jobs.go
index 320c56d..fc61cb2 100644
--- a/internal/platform/postgres/library_jobs.go
+++ b/internal/platform/postgres/library_jobs.go
@@ -346,6 +346,73 @@ WHERE id=$1 AND scope=$2 AND kind='seal' AND queue_job_id=$3
return tag.RowsAffected() == 1, nil
}
+// ReconcileMisclassifiedLibraryCapture repairs ONLY the known startup
+// misclassification: a failed seal whose queue actually completed, with no
+// stage/snapshot and the exact legacy error/recovery. The original fields are
+// copied to the immutable audit row and the job update is CAS-bound to them;
+// any other failed/cancelled row is rejected.
+func (s *Store) ReconcileMisclassifiedLibraryCapture(ctx context.Context, jobID, scope, queueJobID, snapshotID,
+ expectedStage, expectedError, expectedRecovery string, expectedFinished *time.Time,
+ auditID, actor, reason string, evidence map[string]any, now time.Time) (bool, error) {
+ if jobID == "" || scope == "" || queueJobID == "" || snapshotID == "" || auditID == "" {
+ return false, errors.New("misclassification repair requires job/scope/queue/snapshot/audit")
+ }
+ rawEvidence, err := json.Marshal(evidence)
+ if err != nil {
+ return false, fmt.Errorf("encode repair evidence: %w", err)
+ }
+ tx, err := s.db.Begin(ctx)
+ if err != nil {
+ return false, fmt.Errorf("begin repair: %w", err)
+ }
+ defer func() { _ = tx.Rollback(ctx) }()
+ var state, stage, errText, recovery, currentSnapshot string
+ var finished *time.Time
+ if err := tx.QueryRow(ctx, `
+SELECT state, stage, error, recovery, snapshot_id, finished_at
+FROM library_jobs WHERE id=$1 AND scope=$2 AND kind='seal' AND queue_job_id=$3
+FOR UPDATE`, jobID, scope, queueJobID).Scan(&state, &stage, &errText, &recovery, ¤tSnapshot, &finished); err != nil {
+ return false, fmt.Errorf("read repair target: %w", err)
+ }
+ if state != "failed" || stage != expectedStage || errText != expectedError ||
+ recovery != expectedRecovery || currentSnapshot != "" ||
+ (finished == nil) != (expectedFinished == nil) ||
+ (finished != nil && expectedFinished != nil && !finished.Equal(*expectedFinished)) {
+ return false, nil
+ }
+ tag, err := tx.Exec(ctx, `
+UPDATE library_jobs
+SET state='succeeded', stage='export', error='', recovery='', snapshot_id=$3,
+ updated_at=$4, finished_at=$4
+WHERE id=$1 AND scope=$2 AND kind='seal' AND queue_job_id=$5
+ AND state='failed' AND stage=$6 AND error=$7 AND recovery=$8 AND snapshot_id=''
+ AND finished_at IS NOT DISTINCT FROM $9`,
+ jobID, scope, snapshotID, now.UTC(), queueJobID, expectedStage, expectedError, expectedRecovery, expectedFinished)
+ if err != nil {
+ return false, fmt.Errorf("repair misclassified seal: %w", err)
+ }
+ if tag.RowsAffected() != 1 {
+ return false, nil
+ }
+ tag, err = tx.Exec(ctx, `
+INSERT INTO library_job_repairs
+ (id, job_id, scope, kind, queue_job_id, snapshot_id, original_state, original_stage,
+ original_error, original_recovery, original_finished_at, reason, actor, evidence, created_at)
+VALUES ($1,$2,$3,'seal',$4,$5,'failed',$6,$7,$8,$9,$10,$11,$12,$13)
+ON CONFLICT (id) DO NOTHING`, auditID, jobID, scope, queueJobID, snapshotID,
+ expectedStage, expectedError, expectedRecovery, expectedFinished, reason, actor, rawEvidence, now.UTC())
+ if err != nil {
+ return false, fmt.Errorf("write repair audit: %w", err)
+ }
+ if tag.RowsAffected() != 1 {
+ return false, errors.New("repair audit already exists")
+ }
+ if err := tx.Commit(ctx); err != nil {
+ return false, fmt.Errorf("commit repair: %w", err)
+ }
+ return true, nil
+}
+
// RecordLibraryVersion 写入一条**不可变版本事件**，并在需要时推进 current 指针。
//
// 与旧实现的区别（F02/F09 的根因）：
@@ -724,7 +791,7 @@ func (s *Store) ReconcileLibraryJobs(ctx context.Context, now time.Time, kinds [
if len(kinds) == 0 {
return 0, errors.New("reconcile library jobs requires the kinds this executor owns")
}
- if _, err := s.db.Exec(ctx, `
+ tag, err := s.db.Exec(ctx, `
UPDATE library_jobs AS lj
SET state='failed', error='队列中已无对应作业，业务作业未收到结果',
recovery=CASE WHEN lj.kind IN ('sync','restore','recover','baseline')
@@ -737,12 +804,15 @@ WHERE lj.state IN ('pending','running')
AND lj.queue_job_id <> ''
AND NOT EXISTS (
SELECT 1 FROM jobs AS j
- WHERE j.id = lj.queue_job_id AND j.state IN ('pending','running','retry')
- )`, now.UTC(), kinds, s.libraryTargetID); err != nil {
+ WHERE j.id = lj.queue_job_id
+ AND j.state IN ('pending','running','retry','completed')
+ )`, now.UTC(), kinds, s.libraryTargetID)
+ if err != nil {
return 0, fmt.Errorf("reconcile library jobs: %w", err)
}
+ first := int(tag.RowsAffected())
// 建了业务作业但队列投递没完成（queue_job_id 为空）超过 5 分钟，也是孤儿。
- tag, err := s.db.Exec(ctx, `
+ tag, err = s.db.Exec(ctx, `
UPDATE library_jobs
SET state='failed', error='队列投递未完成，作业从未开始', recovery='无需恢复来源；可直接重发', updated_at=$1, finished_at=$1
WHERE state IN ('pending','running') AND queue_job_id='' AND kind = ANY($3) AND (kind='seal' OR target_id=$4) AND created_at < $2`,
@@ -750,7 +820,7 @@ WHERE state IN ('pending','running') AND queue_job_id='' AND kind = ANY($3) AND
if err != nil {
return 0, fmt.Errorf("reconcile unattached library jobs: %w", err)
}
- return int(tag.RowsAffected()), nil
+ return first + int(tag.RowsAffected()), nil
}
// LibraryVersionByRef 按 (scope, kind, 作业, 快照) 精确定位一条版本记录。
diff --git a/internal/platform/postgres/library_jobs_startup_reconcile_pg_test.go b/internal/platform/postgres/library_jobs_startup_reconcile_pg_test.go
new file mode 100644
index 0000000..3574176
--- /dev/null
+++ b/internal/platform/postgres/library_jobs_startup_reconcile_pg_test.go
@@ -0,0 +1,79 @@
+package postgres
+
+import (
+ "context"
+ "os"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/jackc/pgx/v5/pgxpool"
+)
+
+// The worker startup reconciliation must not turn a completed capture queue
+// into the legacy "queue does not exist" failure. This is a real PostgreSQL
+// fixture with the actual varchar shapes; the RED case is the old SQL which
+// only accepted pending/running/retry.
+func TestStartupReconcileKeepsCompletedCaptureJob(t *testing.T) {
+ dsn := os.Getenv("KG_LIBRARY_TEST_DATABASE_URL")
+ if dsn == "" {
+ t.Skip("requires explicitly isolated kg_library_test_ PostgreSQL database")
+ }
+ cfg, err := pgxpool.ParseConfig(dsn)
+ if err != nil || !strings.HasPrefix(cfg.ConnConfig.Database, "kg_library_test_") || os.Getenv("KG_LIBRARY_TEST_ISOLATED") != "1" {
+ t.Fatal("isolated database required")
+ }
+ cfg.MaxConns = 1
+ cfg.ConnConfig.RuntimeParams["search_path"] = "pg_temp"
+ ctx := context.Background()
+ pool, err := pgxpool.NewWithConfig(ctx, cfg)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer pool.Close()
+ now := time.Now().UTC()
+ if _, err := pool.Exec(ctx, `
+CREATE TEMP TABLE library_jobs(
+ id varchar(64) PRIMARY KEY, scope varchar(16) NOT NULL, kind varchar(16) NOT NULL,
+ state varchar(16) NOT NULL, stage varchar(32) NOT NULL DEFAULT '',
+ queue_job_id varchar(64) NOT NULL DEFAULT '', target_id text NOT NULL DEFAULT '',
+ error text NOT NULL DEFAULT '', recovery varchar(255) NOT NULL DEFAULT '',
+ snapshot_id varchar(64) NOT NULL DEFAULT '', updated_at timestamptz NOT NULL, finished_at timestamptz, created_at timestamptz NOT NULL DEFAULT now());
+CREATE TEMP TABLE jobs(id varchar(64) PRIMARY KEY, type varchar(64) NOT NULL, state varchar(16) NOT NULL);`); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := pool.Exec(ctx, `INSERT INTO library_jobs(id,scope,kind,state,queue_job_id,updated_at)
+ VALUES('libseal-completed','internal','seal','pending','q-completed',$1),
+ ('libseal-failed','internal','seal','pending','q-failed',$1),
+ ('libseal-missing','internal','seal','pending','q-missing',$1)`, now); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := pool.Exec(ctx, `INSERT INTO jobs(id,type,state) VALUES('q-completed','library.capture','completed'),
+ ('q-failed','library.capture','failed')`); err != nil {
+ t.Fatal(err)
+ }
+ store := NewStore(pool)
+ n, err := store.ReconcileLibraryJobs(ctx, now, []string{"seal"})
+ if err != nil {
+ t.Fatal(err)
+ }
+ var state, errText string
+ if err := pool.QueryRow(ctx, `SELECT state, error FROM library_jobs WHERE id='libseal-completed'`).Scan(&state, &errText); err != nil {
+ t.Fatal(err)
+ }
+ if state != "pending" || errText != "" {
+ t.Fatalf("completed queue was misclassified: state=%q error=%q", state, errText)
+ }
+ if n != 2 {
+ t.Fatalf("reconciled=%d, want only failed/missing queue jobs (2)", n)
+ }
+ for _, id := range []string{"libseal-failed", "libseal-missing"} {
+ var got string
+ if err := pool.QueryRow(ctx, `SELECT state FROM library_jobs WHERE id=$1`, id).Scan(&got); err != nil {
+ t.Fatal(err)
+ }
+ if got != "failed" {
+ t.Fatalf("%s state=%q, want failed", id, got)
+ }
+ }
+}
diff --git a/internal/platform/postgres/migrations/000080_library_job_repair_audit.down.sql b/internal/platform/postgres/migrations/000080_library_job_repair_audit.down.sql
new file mode 100644
index 0000000..1ff3720
--- /dev/null
+++ b/internal/platform/postgres/migrations/000080_library_job_repair_audit.down.sql
@@ -0,0 +1 @@
+DROP TABLE IF EXISTS library_job_repairs;
diff --git a/internal/platform/postgres/migrations/000080_library_job_repair_audit.up.sql b/internal/platform/postgres/migrations/000080_library_job_repair_audit.up.sql
new file mode 100644
index 0000000..6362738
--- /dev/null
+++ b/internal/platform/postgres/migrations/000080_library_job_repair_audit.up.sql
@@ -0,0 +1,19 @@
+-- Immutable audit for the only permitted automatic/system-misclassification repair.
+CREATE TABLE library_job_repairs (
+ id VARCHAR(64) PRIMARY KEY,
+ job_id VARCHAR(64) NOT NULL,
+ scope VARCHAR(16) NOT NULL,
+ kind VARCHAR(16) NOT NULL,
+ queue_job_id VARCHAR(64) NOT NULL,
+ snapshot_id VARCHAR(64) NOT NULL,
+ original_state VARCHAR(16) NOT NULL,
+ original_stage VARCHAR(32) NOT NULL DEFAULT '',
+ original_error TEXT NOT NULL DEFAULT '',
+ original_recovery VARCHAR(255) NOT NULL DEFAULT '',
+ original_finished_at TIMESTAMPTZ,
+ reason VARCHAR(255) NOT NULL,
+ actor VARCHAR(128) NOT NULL DEFAULT '',
+ evidence JSONB NOT NULL DEFAULT '{}'::jsonb,
+ created_at TIMESTAMPTZ NOT NULL
+);
+CREATE UNIQUE INDEX library_job_repairs_identity ON library_job_repairs(job_id, snapshot_id);
diff --git a/internal/platform/postgres/migrations_test.go b/internal/platform/postgres/migrations_test.go
index 402bde7..e3e11ec 100644
--- a/internal/platform/postgres/migrations_test.go
+++ b/internal/platform/postgres/migrations_test.go
@@ -11,11 +11,14 @@ func TestEmbeddedMigrationsAreVersionSorted(t *testing.T) {
if err != nil {
t.Fatalf("loadMigrations() error = %v", err)
}
- if len(migrations) != 79 {
- t.Fatalf("migration count = %d, want 79", len(migrations))
+ if len(migrations) != 80 {
+ t.Fatalf("migration count = %d, want 80", len(migrations))
}
// 79 是恢复验证：verify 成为独立业务作业并与写作业共用互斥，
// 证据按 target_id 落库（换目标旧证据自动失效）。
+ if repair := migrations[79]; repair.Version != 80 {
+ t.Fatalf("migration 80 = %#v", repair)
+ }
if verification := migrations[78]; verification.Version != 79 ||
!strings.Contains(verification.UpSQL, "library_restore_verifications") ||
!strings.Contains(verification.UpSQL, "'verify'") {
diff --git a/web/src/ReleaseTargets.test.tsx b/web/src/ReleaseTargets.test.tsx
index ee11eb8..f3d72e0 100644
--- a/web/src/ReleaseTargets.test.tsx
+++ b/web/src/ReleaseTargets.test.tsx
@@ -744,6 +744,32 @@ describe('ReleaseTargetsView（确认稿布局）', () => {
expect(engineer).not.toContain('核对封存结果')
})
+ it('系统误判的封存作业（已完成队列+成功导出）仍可由管理员核对；真实失败不显示', () => {
+ const misclassified: LibraryJobRecord = {
+ ...persistedSeal,
+ state: 'failed',
+ stage: '',
+ snapshot_id: '',
+ error: '队列中已无对应作业，业务作业未收到结果',
+ recovery: '封存作业由 agent 恢复来源；确认来源 healthy 后可重发',
+ stages: [{ ...persistedSeal.stages[0], state: 'succeeded' }],
+ }
+ const admin = render({
+ canReconcileCapture: true,
+ overview: { ...overview, job: misclassified, jobs: [misclassified] },
+ })
+ expect(admin).toContain('核对封存结果')
+ const realFailure = render({
+ canReconcileCapture: true,
+ overview: {
+ ...overview,
+ job: { ...misclassified, error: '真实导出失败' },
+ jobs: [{ ...misclassified, error: '真实导出失败' }],
+ },
+ })
+ expect(realFailure).not.toContain('核对封存结果')
+ })
+
it('封存作业只显示只读导出步骤，不冒充生产备份/同步/生效或目标维护', () => {
const html = render({
jobs: [runningJob],
diff --git a/web/src/ReleaseTargets.tsx b/web/src/ReleaseTargets.tsx
index 14d338f..beed84a 100644
--- a/web/src/ReleaseTargets.tsx
+++ b/web/src/ReleaseTargets.tsx
@@ -1107,6 +1107,14 @@ function ProgressBody({
// 不是"只有 sync 才有"。写死 sync 会让恢复作业的分项全显示"未接入"。
const dataJob = job && job.kind !== 'seal' ? job : undefined
const running = job?.state === 'running' || job?.state === 'pending'
+ // Startup reconciliation used to rewrite a completed capture as "队列不存在".
+ // Only that exact legacy shape gets an admin reconcile action; real failures do not.
+ const legacyOrphan =
+ sealJob?.state === 'failed' &&
+ sealJob.stage === '' &&
+ sealJob.snapshot_id === '' &&
+ sealJob.error === '队列中已无对应作业，业务作业未收到结果' &&
+ sealJob.recovery === '封存作业由 agent 恢复来源；确认来源 healthy 后可重发'
const stage = (name: string) => (dataJob ?? sealJob)?.stages.find((item) => item.stage === name)
// F11: counts must belong to the snapshot this job actually used. A newly
@@ -1197,7 +1205,7 @@ function ProgressBody({
{sealJob &&
canReconcileCapture &&
- (sealJob.state === 'pending' || sealJob.state === 'running') &&
+ (sealJob.state === 'pending' || sealJob.state === 'running' || legacyOrphan) &&
stage('export')?.state === 'succeeded' ? (
队列已完成、导出阶段已成功，但作业终态尚未回写。
@@ -1869,7 +1877,17 @@ export function ReleaseTargetsView(props: ViewProps) {
const internal = scopeView(page, jobs, 'internal', scopedOverview('internal'))
const external = scopeView(page, jobs, 'external', scopedOverview('external'))
const view = scope === 'internal' ? internal : external
- const active = Boolean(view.activeJob) || view.job?.state === 'running' || view.job?.state === 'pending'
+ const legacyOrphanJob = Boolean(
+ view.job &&
+ view.job.kind === 'seal' &&
+ view.job.state === 'failed' &&
+ view.job.stage === '' &&
+ view.job.snapshot_id === '' &&
+ view.job.error === '队列中已无对应作业，业务作业未收到结果' &&
+ view.job.recovery === '封存作业由 agent 恢复来源；确认来源 healthy 后可重发'
+ )
+ const active =
+ Boolean(view.activeJob) || view.job?.state === 'running' || view.job?.state === 'pending' || legacyOrphanJob
// F06: 运行中禁止再次启动，也禁止把主按钮继续显示成可启动的绿色动作。
const runningJob = active && view.job ? view.job : undefined
// 三类版本严格分开：target 是测试封存，production 是生产权威版本，prebackup 是本次预备份。