diff --git a/.gitea/scripts/ci_checkout.sh b/.gitea/scripts/ci_checkout.sh index 37c4b52..aeaa55a 100755 --- a/.gitea/scripts/ci_checkout.sh +++ b/.gitea/scripts/ci_checkout.sh @@ -108,9 +108,23 @@ if [ -n "$BEFORE" ] && [ "$BEFORE" != "$ZERO" ]; then if git cat-file -e "${BEFORE}^{commit}" 2>/dev/null; then echo "✅ 事件 before ${BEFORE:0:12} 已在克隆里（本次推送范围可直接算）" else - gitauth fetch -q --depth 50 origin "$BEFORE" 2>/dev/null \ - && echo "✅ 已补取事件 before ${BEFORE:0:12}" \ - || echo "⚠️ 取不到事件 before ${BEFORE:0:12}：范围校验会明确失败，不会静默收缩" + # ⚠️ 为什么重试一次：浅克隆里用 `fetch --depth` 取一个**未公开**的对象（撤销/改写推送之后 + # 的旧提交就是这种）时，git 会在这次 fetch 中途重写 .git/shallow，随后发现自己读到的 + # 与开始时不同就整体中止：`fatal: shallow file has changed since we read it`。 + # 实测（run 305–317，含原地重跑）这一步必中止，而**同一仓库里紧接着重试同一条命令即成功**。 + # before 取不到 ⇒ commits 检查算不出范围 ⇒ 流水线红，且红的原因与代码质量无关。 + # 重试只负责"把 before 取进克隆"，不收缩范围；两次都失败仍明确报警，由 commits 检查判红。 + FETCH_OK="" + FETCH_ERR="" + for _attempt in 1 2; do + FETCH_ERR="$(gitauth fetch -q --depth 50 origin "$BEFORE" 2>&1 >/dev/null || true)" + if git cat-file -e "${BEFORE}^{commit}" 2>/dev/null; then FETCH_OK=1; break; fi + done + if [ -n "$FETCH_OK" ]; then + echo "✅ 已补取事件 before ${BEFORE:0:12}" + else + echo "⚠️ 取不到事件 before ${BEFORE:0:12}：范围校验会明确失败，不会静默收缩（最后一次错误：${FETCH_ERR:0:160}）" + fi fi else echo "（事件里没有 before / before 为全 0：新分支推送，范围按默认分支为基准判）" diff --git a/CHANGELOG.md b/CHANGELOG.md index 852e20f..586eec4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,27 @@ # Changelog +## 开发快照（发布 HTTP 成功闭环：真实监听 POST /library/syncs → handoff → executor → 幂等重放） + +### Added + +- 新增单一场景用例 `TestFormalPublishHTTPReplaySucceededR39`（`KG_R38_SCENARIO=publish-http`）： + 发布作业只来自真实监听的 `POST /api/v1/library/syncs`（管理员会话 + 真实授权）， + 不调用 `store.CreateLibraryDispatch` 冒充 HTTP，也不再走 httptest。 +- 新增 `r39` 证据链：来源身份先声明后封存（独立推导 `source_id`，来源行真实落进隔离 + PostgreSQL schema 再读回）、目标基线走正式 `POST /api/v1/library/baselines`（真实整库 + 捕获 + 校验 + `target_baseline` 证据，不手写生效版本）、handoff 用真实 `local` transport + 与真实 `libraryctl` 的 `snapshot-pack/unpack/verify`、发布后逐表/引用/索引向量/对象版本 + GET SHA 的独立 oracle，以及对同一 body 原样重放（`reused=true`，治理行/目标内容/索引/ + 对象版本逐字节不变）。 +- 新增外置运行器 `publish_http_runner.py`：沿用已审 `r38_runner.py` 的"创建前登记 + + 创建后补不可变 ID"与四次源码门禁，清理仍交给已审 `r38_cleanup.py`（run 目录名以 + `r38-` 开头），不重跑旧 runner、不改它的证据。 + +### Changed + +- 本批次只新增一个单场景测试与外置 wrapper，旧 r38 故障恢复分支（含 `:775` 的测试直建 + dispatch）一行未改，仍按原语义保留。 + ## 开发快照（DS03 早段 PG 恢复：合法来源 + database 23505 → held → 同窗口恢复） ### Added diff --git a/internal/releaseadapter/v071/formal_publish_http_r39_helpers_test.go b/internal/releaseadapter/v071/formal_publish_http_r39_helpers_test.go new file mode 100644 index 0000000..bf6fac0 --- /dev/null +++ b/internal/releaseadapter/v071/formal_publish_http_r39_helpers_test.go @@ -0,0 +1,649 @@ +package v071 + +// r39 辅助函数：只服务 publish-http 单场景，复用旧分支已审的 fixture/读回/oracle 规则。 + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "net/http" + "os" + "os/exec" + "path/filepath" + "sort" + "strings" + "testing" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/service/s3" + "github.com/jackc/pgx/v5/pgxpool" + + "knowledge-governance/internal/governance" + "knowledge-governance/internal/librarysnapshot" + postgresstore "knowledge-governance/internal/platform/postgres" +) + +// r39BaseEnv 是 API / capture / release 三份进程共用的产品配置。 +// 唯一差异（归档根、角色、worker id）在调用点显式覆盖，避免"两套 env 慢慢分叉"。 +func r39BaseEnv(t *testing.T, fx *writeSetFixture) map[string]string { + t.Helper() + sourceArchive := os.Getenv("KG_R38_SOURCE_ARCHIVE") + return map[string]string{ + "KG_LIBRARY_TEST_ISOLATED": "1", + "DATABASE_URL": os.Getenv("KG_R38_GOVERNANCE_DSN"), + "LIBRARY_WORKER_ROLE": "release-executor", + "WORKER_POLL_INTERVAL": "200ms", "WORKER_LEASE_DURATION": "30s", + "PRODUCTION_RELEASE_REQUESTS_ENABLED": "true", "PRODUCTION_BACKUP_RESTORE_VERIFIED": "true", + "LIBRARY_ARCHIVE_ROOT": sourceArchive, + "LIBRARY_CAPACITY_BUDGET_BYTES": "1073741824", + "LIBRARY_PRODUCTION_DATABASE_URL": os.Getenv("KG_LIBRARY_TEST_DATABASE_URL"), + "LIBRARY_PRODUCTION_ELASTIC_URL": "", + "LIBRARY_PRODUCTION_INDEX": fx.index, + "LIBRARY_PRODUCTION_TENANT_ID": "10001", + "LIBRARY_PRODUCTION_KB_INTERNAL": wsTargetKB, + "LIBRARY_PRODUCTION_S3_PROVIDER": "minio", + "LIBRARY_PRODUCTION_S3_ENDPOINT": os.Getenv("KG_LIBRARY_TEST_S3_ENDPOINT"), + "LIBRARY_PRODUCTION_S3_BUCKET": fx.bucket, + "LIBRARY_PRODUCTION_S3_ACCESS_KEY": os.Getenv("KG_R38_READER_KEY"), + "LIBRARY_PRODUCTION_S3_SECRET_KEY": os.Getenv("KG_R38_READER_SECRET"), + "LIBRARY_TARGET_INSTANCE": os.Getenv("KG_R38_PROJECT"), + "LIBRARY_TARGET_PROJECT": os.Getenv("KG_R38_PROJECT"), + "LIBRARY_TARGET_WRITERS": "writer", + "LIBRARY_TARGET_NON_WRITERS": "ingress=entry gate,redis=queue,minio=object backend", + "LIBRARY_TARGET_INGRESS_URL": os.Getenv("KG_R38_INGRESS"), + "LIBRARY_TARGET_INGRESS_SERVICE": "ingress", + "LIBRARY_TARGET_INGRESS_DRAIN_FILE": filepath.Join(os.Getenv("KG_R38_RUN"), "gate/drain.json"), + "LIBRARY_TARGET_REDIS_ADDR": os.Getenv("KG_R38_REDIS_ADDR"), + "LIBRARY_TARGET_REDIS_PASSWORD": os.Getenv("KG_R38_REDIS_PASSWORD"), + "LIBRARY_TARGET_REDIS_DB": "0", + "LIBRARY_TARGET_JOURNAL_ROOT": filepath.Join(os.Getenv("KG_R38_RUN"), "maintenance"), + "LIBRARY_TARGET_DOCKER_SOCKET": "/var/run/docker.sock", + "LIBRARY_EMBEDDING_DIMENSION": "1024", + "LIBRARY_SOURCE_INSTANCE": os.Getenv("KG_R38_SOURCE_INSTANCE"), + "LIBRARY_SOURCE_PROJECT": os.Getenv("KG_R38_SOURCE_PROJECT"), + "LIBRARY_SOURCE_APP": os.Getenv("KG_R38_SOURCE_APP"), + "LIBRARY_SOURCE_TENANT_ID": "10002", + "LIBRARY_SOURCE_KB_INTERNAL": wsSourceKB, + } +} + +func r39ReadJSON(t *testing.T, path string) map[string]any { + t.Helper() + raw, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read %s: %v", path, err) + } + var out map[string]any + if err := json.Unmarshal(raw, &out); err != nil { + t.Fatalf("decode %s: %v", path, err) + } + return out +} + +func r39DatabaseName(t *testing.T, dsn string) string { + t.Helper() + parsed, err := pgxpool.ParseConfig(dsn) + if err != nil { + t.Fatalf("parse production dsn: %v", err) + } + return parsed.ConnConfig.Database +} + +func r39RegisteredDatabase(t *testing.T, state map[string]any, name string) map[string]any { + t.Helper() + resources, _ := state["resources"].([]any) + for _, item := range resources { + entry, _ := item.(map[string]any) + if entry["kind"] == "database" && entry["name"] == name { + return entry + } + } + t.Fatalf("database %s was never registered before creation", name) + return nil +} + +func r39DatabaseOID(t *testing.T, state map[string]any, name string) string { + t.Helper() + oid, _ := r39RegisteredDatabase(t, state, name)["id"].(string) + return oid +} + +func r39DatabaseOwner(t *testing.T, state map[string]any, name string) string { + t.Helper() + owner, _ := r39RegisteredDatabase(t, state, name)["owner"].(string) + return owner +} + +// r39ArchiveListing 枚举归档本体（清单 + 被引用字节），用于证明 executor 起始为空。 +func r39ArchiveListing(t *testing.T, root string) map[string]any { + t.Helper() + versions := []string{} + blobs := []string{} + if entries, err := os.ReadDir(filepath.Join(root, "versions")); err == nil { + for _, entry := range entries { + versions = append(versions, entry.Name()) + } + } + if entries, err := os.ReadDir(filepath.Join(root, "blobs")); err == nil { + for _, entry := range entries { + blobs = append(blobs, entry.Name()) + } + } + sort.Strings(versions) + sort.Strings(blobs) + return map[string]any{"root": root, "versions": versions, "blobs": blobs, + "versions_present": len(versions) > 0, "blob_count": len(blobs)} +} + +func r39ArchiveHas(t *testing.T, root, id string) bool { + t.Helper() + _, err := os.Stat(filepath.Join(root, "versions", id+".json")) + return err == nil +} + +func r39Manifest(t *testing.T, root, id string) (librarysnapshot.Manifest, []byte) { + t.Helper() + archive, err := librarysnapshot.Open(root, 256<<20) + if err != nil { + t.Fatalf("open archive %s: %v", root, err) + } + defer archive.Close() + manifest, err := archive.Load(id) + if err != nil { + t.Fatalf("load snapshot %s from %s: %v", id, root, err) + } + raw, err := os.ReadFile(filepath.Join(root, "versions", id+".json")) + if err != nil { + t.Fatalf("read manifest %s: %v", id, err) + } + return manifest, raw +} + +// r39WaitLibraryJob 等到业务作业进入期望终态；中途的失败/取消立即报错，不吞。 +func r39WaitLibraryJob(t *testing.T, store *postgresstore.Store, id string, want governance.LibraryJobState, seconds int) governance.LibraryJob { + t.Helper() + deadline := time.Now().Add(time.Duration(seconds) * time.Second) + var last governance.LibraryJob + for time.Now().Before(deadline) { + job, err := store.LibraryJob(context.Background(), id) + if err != nil { + t.Fatalf("read library job %s: %v", id, err) + } + last = job + if job.State == want { + return job + } + if job.State == governance.LibraryJobFailed || job.State == governance.LibraryJobCancelled { + t.Fatalf("library job %s reached %s stage=%s error=%s recovery=%s", id, job.State, job.Stage, job.Error, job.Recovery) + } + time.Sleep(250 * time.Millisecond) + } + t.Fatalf("library job %s timeout in state %s stage=%s error=%s", id, last.State, last.Stage, last.Error) + return last +} + +func r39WaitQueue(t *testing.T, pool *pgxpool.Pool, id, want string, seconds int) string { + t.Helper() + deadline := time.Now().Add(time.Duration(seconds) * time.Second) + var state, message string + for time.Now().Before(deadline) { + if err := pool.QueryRow(context.Background(), + `SELECT state,COALESCE(last_error,'') FROM jobs WHERE id=$1`, id).Scan(&state, &message); err != nil { + t.Fatalf("read queue %s: %v", id, err) + } + if state == want { + return state + } + if state == "failed" || state == "completed" || state == "dead" { + t.Fatalf("queue %s terminal state=%s error=%s (want %s)", id, state, message, want) + } + time.Sleep(200 * time.Millisecond) + } + t.Fatalf("queue %s timeout (state=%s error=%s)", id, state, message) + return state +} + +// r39FirstQueueObservation 记录队列作业第一次被看到的形态：用来证明它入的是 handoff 队列， +// 而不是"直接进 release 也行"。 +func r39FirstQueueObservation(t *testing.T, pool *pgxpool.Pool, id string, seconds int) map[string]any { + t.Helper() + deadline := time.Now().Add(time.Duration(seconds) * time.Second) + for time.Now().Before(deadline) { + var raw string + err := pool.QueryRow(context.Background(), `SELECT to_jsonb(x)::text FROM jobs AS x WHERE id=$1`, id).Scan(&raw) + if err == nil { + var row map[string]any + if err := json.Unmarshal([]byte(raw), &row); err != nil { + t.Fatalf("decode queue row: %v", err) + } + return row + } + time.Sleep(100 * time.Millisecond) + } + t.Fatalf("queue job %s never appeared", id) + return nil +} + +func r39EntryOpen(t *testing.T, url string) map[string]any { + t.Helper() + client := &http.Client{Timeout: 10 * time.Second} + code, detail := 0, "" + for i := 0; i < 30; i++ { + response, err := client.Get(url) + if err == nil { + raw, _ := io.ReadAll(io.LimitReader(response.Body, 4096)) + response.Body.Close() + code = response.StatusCode + detail = strings.TrimSpace(string(raw)) + if code == 200 { + return map[string]any{"url": url, "http": code, "open": true, "body_prefix": truncateR39(detail, 200)} + } + } else { + detail = err.Error() + } + time.Sleep(200 * time.Millisecond) + } + return map[string]any{"url": url, "http": code, "open": false, "detail": truncateR39(detail, 300)} +} + +func truncateR39(value string, limit int) string { + if len(value) <= limit { + return value + } + return value[:limit] +} + +// r39RunCLI 执行真实的 libraryctl 并把原始输出（stdout/stderr/rc/字节数）留在证据里。 +func r39RunCLI(t *testing.T, binary string, args ...string) map[string]any { + t.Helper() + command := exec.Command(binary, args...) + command.Env = append(os.Environ(), "LIBRARY_HANDOFF_FIXTURE=1") + var stdout, stderr strings.Builder + command.Stdout = &stdout + command.Stderr = &stderr + err := command.Run() + code := 0 + if err != nil { + if exit, ok := err.(*exec.ExitError); ok { + code = exit.ExitCode() + } else { + code = -1 + } + } + return map[string]any{"binary": binary, "argv": append([]string{binary}, args...), "exit_code": code, + "stdout": truncateR39(stdout.String(), 4000), "stderr": truncateR39(stderr.String(), 2000)} +} + +func r39BlobDigests(t *testing.T, root string, manifest librarysnapshot.Manifest) map[string]map[string]any { + t.Helper() + out := map[string]map[string]any{} + for _, part := range manifest.Parts { + path := filepath.Join(root, "blobs", part.SHA256) + info, err := os.Stat(path) + if err != nil { + t.Fatalf("blob %s missing in %s: %v", part.SHA256, root, err) + } + raw, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read blob %s: %v", part.SHA256, err) + } + sum := sha256.Sum256(raw) + out[part.SHA256] = map[string]any{"name": part.Name, "kind": part.Kind, "bytes": info.Size(), + "sha256": hex.EncodeToString(sum[:]), "size_matches_manifest": info.Size() == part.Bytes} + } + return out +} + +// r39HandoffEvidence 证明 handoff 真的经由 libraryctl 落到独立 executor archive， +// 并用真实 pack/unpack/verify 复核两端字节（不是"复制目录后声称成功"）。 +func r39HandoffEvidence(t *testing.T, gov *pgxpool.Pool, store *postgresstore.Store, jobID, snapshot, + sourceArchive, executorArchive, libraryctl, repackRoot string) map[string]any { + t.Helper() + ctx := context.Background() + sourceManifest, sourceRaw := r39Manifest(t, sourceArchive, snapshot) + executorManifest, executorRaw := r39Manifest(t, executorArchive, snapshot) + sourceSum, executorSum := sha256.Sum256(sourceRaw), sha256.Sum256(executorRaw) + sourceBlobs := r39BlobDigests(t, sourceArchive, sourceManifest) + executorBlobs := r39BlobDigests(t, executorArchive, executorManifest) + matches := hex.EncodeToString(sourceSum[:]) == snapshot && hex.EncodeToString(executorSum[:]) == snapshot + matches = matches && reflectDeepEqualR39(sourceBlobs, executorBlobs) && len(sourceBlobs) == len(sourceManifest.Parts) + + // 真实 libraryctl：verify(源) → pack(源) → unpack(全新归档) → verify(全新归档)。 + verifySource := r39RunCLI(t, libraryctl, "snapshot-verify", "-archive", sourceArchive, "-snapshot", snapshot) + if err := os.MkdirAll(repackRoot, 0o700); err != nil { + t.Fatalf("prepare repack archive: %v", err) + } + bundleDir := filepath.Join(repackRoot, "bundle") + if err := os.MkdirAll(bundleDir, 0o700); err != nil { + t.Fatalf("prepare bundle dir: %v", err) + } + bundlePath := filepath.Join(bundleDir, snapshot+".tar") + freshArchive := filepath.Join(repackRoot, "archive") + if err := os.MkdirAll(freshArchive, 0o700); err != nil { + t.Fatalf("prepare fresh archive: %v", err) + } + pack := r39RunCLI(t, libraryctl, "snapshot-pack", "-archive", sourceArchive, "-snapshot", snapshot, "-bundle", bundlePath) + var bundleBytes int64 + if info, err := os.Stat(bundlePath); err == nil { + bundleBytes = info.Size() + } + unpack := r39RunCLI(t, libraryctl, "snapshot-unpack", "-archive", freshArchive, "-bundle", bundlePath, "-snapshot", snapshot) + verifyFresh := r39RunCLI(t, libraryctl, "snapshot-verify", "-archive", freshArchive, "-snapshot", snapshot) + verifyExecutor := r39RunCLI(t, libraryctl, "snapshot-verify", "-archive", executorArchive, "-snapshot", snapshot) + freshManifest, freshRaw := r39Manifest(t, freshArchive, snapshot) + freshSum := sha256.Sum256(freshRaw) + freshBlobs := r39BlobDigests(t, freshArchive, freshManifest) + allZero := true + for _, item := range []map[string]any{verifySource, pack, unpack, verifyFresh, verifyExecutor} { + if item["exit_code"] != 0 { + allZero = false + } + } + bundleLeft, err := os.ReadDir(filepath.Join(os.Getenv("KG_R38_EXECUTOR_BUNDLE"))) + bundleAbsent := err == nil && len(bundleLeft) == 0 + + var detail string + stage, err := store.LibraryJob(ctx, jobID) + if err == nil { + detail = stage.Stage + } + stages := ds02Read(t, &writeSetFixture{pool: gov, ctx: ctx}, + `SELECT to_jsonb(x)::text FROM library_job_stages AS x WHERE job_id='`+jobID+`' AND stage='handoff'`) + return map[string]any{ + "snapshot": snapshot, "job": jobID, "job_stage": detail, + "source_archive": sourceArchive, "executor_archive": executorArchive, + "source_manifest_sha256": hex.EncodeToString(sourceSum[:]), "executor_manifest_sha256": hex.EncodeToString(executorSum[:]), + "fresh_manifest_sha256": hex.EncodeToString(freshSum[:]), + "source_parts": sourceManifest.Parts, "executor_parts": executorManifest.Parts, + "source_blobs": sourceBlobs, "executor_blobs": executorBlobs, + "executor_matches_source": matches, + "fresh_matches_source": hex.EncodeToString(freshSum[:]) == snapshot && reflectDeepEqualR39(freshBlobs, sourceBlobs), + "libraryctl_all_zero": allZero, + "libraryctl": map[string]any{"verify_source": verifySource, "pack": pack, "unpack": unpack, + "verify_fresh": verifyFresh, "verify_executor": verifyExecutor}, + "bundle_bytes": bundleBytes, "bundle_absent_after": bundleAbsent, + "handoff_stage_rows": stages, + "libraryctl_identity": func() map[string]any { + raw, err := os.ReadFile(libraryctl) + if err != nil { + t.Fatalf("read libraryctl: %v", err) + } + sum := sha256.Sum256(raw) + return map[string]any{"path": libraryctl, "sha256": hex.EncodeToString(sum[:])} + }(), + } +} + +func reflectDeepEqualR39(a, b any) bool { + left, err := json.Marshal(a) + if err != nil { + return false + } + right, err := json.Marshal(b) + if err != nil { + return false + } + return string(left) == string(right) +} + +// r39CheckPublishedIndex 用声明内容独立推导"发布后索引应有的文档"： +// 目标 tenant/KB、来源 chunk 内容、来源声明的 1024 维向量，逐项核对 _source。 +func r39CheckPublishedIndex(t *testing.T, spec ds02Spec, before, after map[string]map[string]any) map[string]any { + t.Helper() + vectors := r38SourceVectors() + expected := map[string]map[string]any{} + for _, chunk := range spec.rows["chunks"] { + id := fmt.Sprint(chunk["id"]) + vector, ok := vectors[id] + if !ok { + t.Fatalf("source chunk %s has no declared vector", id) + } + expected[id] = map[string]any{"tenant_id": float64(wsTargetTenant), "knowledge_base_id": wsTargetKB, + "knowledge_id": chunk["knowledge_id"], "chunk_id": id, "source_id": id, + "content": chunk["content"], "embedding": vector} + } + problems := []string{} + for id, want := range expected { + doc, ok := after[id] + if !ok { + problems = append(problems, "published document missing: "+id) + continue + } + source, _ := doc["_source"].(map[string]any) + if source == nil { + problems = append(problems, "published document has no _source: "+id) + continue + } + for field, wantValue := range want { + got, exists := source[field] + if !exists { + problems = append(problems, id+"."+field+" missing") + continue + } + if field == "embedding" { + if !r39VectorEqual(wantValue.([]float64), got) { + problems = append(problems, id+".embedding differs from the declared 1024-dim vector") + } + continue + } + if fmt.Sprint(got) != fmt.Sprint(wantValue) { + problems = append(problems, fmt.Sprintf("%s.%s = %v, want %v", id, field, got, wantValue)) + } + } + } + // 目标范围之外的文档必须原样保留；目标范围的旧一代文档必须被替换掉。 + // 判定用独立读到的 _source（tenant/KB），不按 id 名单。 + sentinels, replaced := 0, 0 + for id, doc := range before { + source, _ := doc["_source"].(map[string]any) + if source != nil && fmt.Sprint(source["tenant_id"]) == fmt.Sprint(wsTargetTenant) && + fmt.Sprint(source["knowledge_base_id"]) == wsTargetKB { + replaced++ + if _, stillThere := after[id]; stillThere { + problems = append(problems, "replaced target-generation document survived: "+id) + } + continue + } + sentinels++ + got, ok := after[id] + if !ok { + problems = append(problems, "sentinel document disappeared: "+id) + continue + } + if !reflectDeepEqualR39(doc["_source"], got["_source"]) { + problems = append(problems, "sentinel document changed: "+id) + } + } + if len(after) != len(expected)+sentinels { + problems = append(problems, fmt.Sprintf("index document count = %d, want %d published + %d sentinel", + len(after), len(expected), sentinels)) + } + return map[string]any{"match": len(problems) == 0, "problems": problems, "published_documents": len(expected), + "sentinel_documents": sentinels, "replaced_documents": replaced, "after_documents": len(after)} +} + +func r39VectorEqual(want []float64, got any) bool { + values, ok := got.([]any) + if !ok || len(values) != len(want) { + return false + } + for i, value := range values { + number, ok := value.(float64) + if !ok { + return false + } + delta := number - want[i] + if delta < 0 { + delta = -delta + } + if delta > 1e-9 { + return false + } + } + return true +} + +// r39CheckPublishedObjects 逐资源按真实版本 GET 校验字节：发布后的对象必须就是来源字节。 +func r39CheckPublishedObjects(t *testing.T, ctx context.Context, fx *writeSetFixture, spec ds02Spec, + expectation ds02Expectation) map[string]any { + t.Helper() + problems := []string{} + observed := map[string]any{} + ids := make([]string, 0, len(spec.objects)) + for id := range spec.objects { + ids = append(ids, id) + } + sort.Strings(ids) + for _, id := range ids { + path, ok := expectation.Paths[id] + if !ok { + problems = append(problems, "no expected locator for resource "+id) + continue + } + reader, err := fx.objects.s3.Open(ctx, "minio", path) + if err != nil { + problems = append(problems, fmt.Sprintf("open published object %s: %v", id, err)) + continue + } + body, readErr := io.ReadAll(reader) + reader.Close() + if readErr != nil { + problems = append(problems, fmt.Sprintf("read published object %s: %v", id, readErr)) + continue + } + want := sha256.Sum256([]byte(spec.objects[id])) + got := sha256.Sum256(body) + if hex.EncodeToString(want[:]) != hex.EncodeToString(got[:]) { + problems = append(problems, fmt.Sprintf("object %s bytes differ from the source", id)) + } + observed[id] = map[string]any{"locator": path, "bytes": len(body), "sha256": hex.EncodeToString(got[:]), + "matches_source": hex.EncodeToString(want[:]) == hex.EncodeToString(got[:])} + } + inventory := objectInventoryR38(t, ctx, fx) + digests := r38ObjectVersionDigests(t, ctx, fx, inventory) + return map[string]any{"match": len(problems) == 0, "problems": problems, "objects": observed, + "versions_after": len(digests), "version_digests": digests} +} + +// r39ObjectVersionReadback 逐 key+versionId 实取 GET 并本地算 SHA256， +// 返回完整版本清单（含 key/versionId/ETag/size/SHA256）与同一键的摘要映射。 +// GET 失败、字节数与清单不符、清单里有条目没取到摘要都直接失败（非零）： +// ETag/size 只是旁证，不能代替字节。 +func r39ObjectVersionReadback(t *testing.T, ctx context.Context, fx *writeSetFixture, + inventory *s3.ListObjectVersionsOutput) ([]map[string]any, map[string]string) { + t.Helper() + digests := r38ObjectVersionDigests(t, ctx, fx, inventory) + manifest := make([]map[string]any, 0, len(inventory.Versions)) + for _, version := range inventory.Versions { + key, id := aws.ToString(version.Key), aws.ToString(version.VersionId) + digest, ok := digests[key+"@"+id] + if !ok || digest == "" { + t.Fatalf("version %s@%s has no measured SHA256", key, id) + } + manifest = append(manifest, map[string]any{"key": key, "version_id": id, + "etag": aws.ToString(version.ETag), "size": aws.ToInt64(version.Size), "sha256": digest}) + } + sort.Slice(manifest, func(i, j int) bool { + return fmt.Sprint(manifest[i]["key"], "@", manifest[i]["version_id"]) < + fmt.Sprint(manifest[j]["key"], "@", manifest[j]["version_id"]) + }) + return manifest, digests +} + +// r39CompareObjectVersionBytes 逐 key+versionId 比较重放前后的完整清单和实取 SHA256。 +// 版本增减/消失、摘要缺失、字节不同都返回问题；空清单也算问题—— +// 否则"一个对象都没读到"会被当成"完全相同"。 +func r39CompareObjectVersionBytes(label string, before, after []map[string]any, + beforeDigests, afterDigests map[string]string) []string { + problems := []string{} + if len(before) == 0 || len(after) == 0 { + return []string{fmt.Sprintf("%s object version manifest is empty: %d -> %d", label, len(before), len(after))} + } + if len(before) != len(after) { + problems = append(problems, fmt.Sprintf("%s object versions %d -> %d", label, len(before), len(after))) + } + ids := func(manifest []map[string]any) []string { + out := make([]string, 0, len(manifest)) + for _, row := range manifest { + out = append(out, fmt.Sprint(row["key"], "@", row["version_id"])) + } + sort.Strings(out) + return out + } + beforeIDs, afterIDs := ids(before), ids(after) + beforeSet := make(map[string]bool, len(beforeIDs)) + afterSet := make(map[string]bool, len(afterIDs)) + for _, id := range beforeIDs { + beforeSet[id] = true + } + for _, id := range afterIDs { + afterSet[id] = true + } + for _, id := range beforeIDs { + if !afterSet[id] { + problems = append(problems, fmt.Sprintf("%s object version disappeared: %s", label, id)) + continue + } + if beforeDigests[id] == "" || afterDigests[id] == "" { + problems = append(problems, fmt.Sprintf("%s object version %s has no measured SHA256", label, id)) + continue + } + if beforeDigests[id] != afterDigests[id] { + problems = append(problems, fmt.Sprintf("%s object version %s bytes changed: %s -> %s", + label, id, beforeDigests[id], afterDigests[id])) + } + } + for _, id := range afterIDs { + if !beforeSet[id] { + problems = append(problems, fmt.Sprintf("%s object version appeared: %s", label, id)) + } + } + return problems +} + +func r39Flatten(values map[string][]string) []string { + tables := make([]string, 0, len(values)) + for table := range values { + tables = append(tables, table) + } + sort.Strings(tables) + out := make([]string, 0, 64) + for _, table := range tables { + for _, row := range values[table] { + out = append(out, table+"\x00"+row) + } + } + return out +} + +func r39DiffStrings(label string, before, after []string) []string { + if len(before) != len(after) { + return []string{fmt.Sprintf("%s rows %d -> %d", label, len(before), len(after))} + } + for i := range before { + if before[i] != after[i] { + return []string{fmt.Sprintf("%s row %d changed: %s", label, i, after[i])} + } + } + return nil +} + +var _ = aws.ToString + +// r39PublishedScope 返回与期望同口径的实际行集合：resources/resource_bindings 按 +// 目标租户取（产品按内容寻址复用同租户行），knowledge_tag_relations 没有租户列、 +// 期望里已含目标范围外的关系，所以取整表；其余表按目标知识库范围。 +func r39PublishedScope(t *testing.T, fx *writeSetFixture, table string, targetRows []map[string]any) []map[string]any { + t.Helper() + switch table { + case "resources", "resource_bindings": + return ds02Read(t, fx, `SELECT to_jsonb(x)::text FROM `+table+` AS x WHERE tenant_id=$1 ORDER BY 1`, wsTargetTenant) + case "knowledge_tag_relations": + return ds02Read(t, fx, `SELECT to_jsonb(x)::text FROM knowledge_tag_relations AS x ORDER BY 1`) + } + return targetRows +} diff --git a/internal/releaseadapter/v071/formal_publish_http_r39_replay_bytes_control_test.go b/internal/releaseadapter/v071/formal_publish_http_r39_replay_bytes_control_test.go new file mode 100644 index 0000000..7fe2049 --- /dev/null +++ b/internal/releaseadapter/v071/formal_publish_http_r39_replay_bytes_control_test.go @@ -0,0 +1,75 @@ +package v071 + +// r39 重放字节判据的离线负控：原判据只比 Listing 里的 key@versionId#etag#size， +// 同一个 versionId 的字节被改写它完全看不见。这里先把那个盲区固定下来， +// 再要求新判据逐 key+versionId 比较实取 SHA256，缺失/空清单也必须报问题。 + +import ( + "crypto/sha256" + "encoding/hex" + "reflect" + "strings" + "testing" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/service/s3" + "github.com/aws/aws-sdk-go-v2/service/s3/types" +) + +func r39ControlVersion() types.ObjectVersion { + return types.ObjectVersion{ + Key: aws.String("kb-source/docs/1"), VersionId: aws.String("v-before"), + ETag: aws.String("\"etag-same\""), Size: aws.Int64(4)} +} + +func r39ControlDigest(body string) string { + sum := sha256.Sum256([]byte(body)) + return hex.EncodeToString(sum[:]) +} + +func r39ControlManifest(digest string) ([]map[string]any, map[string]string) { + manifest := []map[string]any{{"key": "kb-source/docs/1", "version_id": "v-before", + "etag": "\"etag-same\"", "size": int64(4), "sha256": digest}} + return manifest, map[string]string{"kb-source/docs/1@v-before": digest} +} + +func TestR39ReplayObjectByteControl(t *testing.T) { + // 1) 旧判据的盲区：Listing 只有 key/versionId/etag/size，没有字节。 + // 同一 versionId 的字节被改写，旧判据仍然报"完全相同"。 + beforeListing := &s3.ListObjectVersionsOutput{Versions: []types.ObjectVersion{r39ControlVersion()}} + afterListing := &s3.ListObjectVersionsOutput{Versions: []types.ObjectVersion{r39ControlVersion()}} + if !reflect.DeepEqual(objectVersionIDs(beforeListing), objectVersionIDs(afterListing)) { + t.Fatalf("control setup is wrong: the legacy version-ID predicate is expected to see identical metadata") + } + beforeManifest, beforeDigests := r39ControlManifest(r39ControlDigest("abcd")) + afterManifest, afterDigests := r39ControlManifest(r39ControlDigest("abce")) + if beforeDigests["kb-source/docs/1@v-before"] == afterDigests["kb-source/docs/1@v-before"] { + t.Fatalf("control setup is wrong: the two bodies must hash differently") + } + + // 2) 新判据必须在"版本元数据完全相同 + 字节不同"的输入上报出差异。 + problems := r39CompareObjectVersionBytes("replay", beforeManifest, afterManifest, beforeDigests, afterDigests) + if len(problems) == 0 { + t.Fatalf("replay byte comparison missed a body change that kept key/versionId/etag/size identical") + } + if !strings.Contains(strings.Join(problems, "\n"), "bytes changed") { + t.Fatalf("byte change must be reported as bytes changed, got: %v", problems) + } + + // 3) 正例对照：字节相同不得报问题。 + if same := r39CompareObjectVersionBytes("replay", beforeManifest, beforeManifest, beforeDigests, beforeDigests); len(same) != 0 { + t.Fatalf("identical manifests must not report problems: %v", same) + } + + // 4) 版本消失、空清单、摘要缺失都必须非零。 + if got := r39CompareObjectVersionBytes("replay", beforeManifest, nil, beforeDigests, map[string]string{}); len(got) == 0 { + t.Fatalf("a disappeared/empty version manifest must be reported") + } + if got := r39CompareObjectVersionBytes("replay", nil, nil, nil, nil); len(got) == 0 { + t.Fatalf("an empty version manifest must be reported") + } + noDigest := map[string]string{} + if got := r39CompareObjectVersionBytes("replay", beforeManifest, beforeManifest, beforeDigests, noDigest); len(got) == 0 { + t.Fatalf("a version without a measured SHA256 must be reported") + } +} diff --git a/internal/releaseadapter/v071/formal_publish_http_r39_test.go b/internal/releaseadapter/v071/formal_publish_http_r39_test.go new file mode 100644 index 0000000..ce8ac92 --- /dev/null +++ b/internal/releaseadapter/v071/formal_publish_http_r39_test.go @@ -0,0 +1,1030 @@ +package v071 + +// r39：单一本地专用场景 —— 真实 HTTP 发布成功闭环 + 同一请求原样重放。 +// +// 链路（全部由真实进程/真实存储产生，测试只声明来源、播种 before-image、观察读回）： +// +// POST /api/v1/library/syncs（真实监听、管理员会话） +// → library.snapshot.handoff 队列（capture 角色 worker 消费） +// → 真实 libraryctl pack/unpack/verify → 独立 executor archive +// → library.sync 队列（release-executor 角色 worker 消费） +// → 业务 succeeded / 两队列 completed / production_effective / 维护释放 / 入口恢复 +// → 同一 body 原样重放：复用同 operation+job，目标内容与执行次数不增加 +// +// 与旧 r38 分支的边界： +// - 旧分支用 store.CreateLibraryDispatch 直建作业证明"故障恢复"；本场景的发布必须来自 +// 真实 HTTP handler，测试不得直接建 publish 作业； +// - 旧分支的隔离装配/鉴权/读回/provenance 规则沿用，但目标基线走正式 HTTP +// POST /api/v1/library/baselines（真实整库捕获 + 校验 + 登记），不手写生效版本； +// - 本地 handoff 使用已实现的 local transport + 真实 libraryctl，明确不证明 SSH/云传输。 + +import ( + "bytes" + "context" + "crypto/ed25519" + "crypto/rand" + "crypto/sha256" + "encoding/base64" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/cookiejar" + "os" + "os/exec" + "path/filepath" + "reflect" + "sort" + "strings" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + "golang.org/x/crypto/bcrypt" + + "knowledge-governance/internal/appconfig" + "knowledge-governance/internal/governance" + "knowledge-governance/internal/librarysnapshot" + "knowledge-governance/internal/objectgateway" + postgresstore "knowledge-governance/internal/platform/postgres" +) + +// r39SourceSchema 是本轮"真实本地来源实例"的落地位置：业务库里一张独立 schema， +// 装的就是本轮声明的合法来源行。产品不读它；它只负责证明"manifest 的内容来自真实 +// PostgreSQL 行"，而不是凭空构造。 +const r39SourceSchema = "r39_source" + +// r39Result 是本场景的逐项断言结果；失败项必须留在文件里，不能只体现在日志。 +type r39Result struct { + Run string `json:"run"` + Scenario string `json:"scenario"` + Checks []map[string]any `json:"checks"` + Failed []string `json:"failed"` + Status string `json:"status"` + Finished string `json:"finished"` + Assertion int `json:"assertion_count"` +} + +func r39Write(t *testing.T, path string, value any) { + t.Helper() + if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil { + t.Fatalf("prepare evidence dir: %v", err) + } + raw, err := json.MarshalIndent(value, "", " ") + if err != nil { + t.Fatalf("encode %s: %v", path, err) + } + if err := os.WriteFile(path, raw, 0o600); err != nil { + t.Fatalf("write %s: %v", path, err) + } +} + +// r39SeedSourceStore 把本轮声明的来源行真的写进独立 schema，再独立读回： +// 封存清单的内容必须来自这次读回，不能来自"声明对象自己"。 +func r39SeedSourceStore(t *testing.T, fx *writeSetFixture, spec ds02Spec) (map[string][]map[string]any, map[string]any) { + t.Helper() + ctx := fx.ctx + if _, err := fx.pool.Exec(ctx, `DROP SCHEMA IF EXISTS `+r39SourceSchema+` CASCADE; CREATE SCHEMA `+r39SourceSchema); err != nil { + t.Fatalf("create source schema: %v", err) + } + for _, table := range ds02TargetTables { + statement := `CREATE TABLE ` + r39SourceSchema + `.` + table + ` (LIKE public.` + table + ` INCLUDING ALL)` + if _, err := fx.pool.Exec(ctx, statement); err != nil { + t.Fatalf("create source table %s: %v", table, err) + } + } + insert := `INSERT INTO ` + r39SourceSchema + `.%s SELECT * FROM jsonb_populate_record(NULL::` + r39SourceSchema + `.%s, $1::jsonb)` + canonical := `SELECT to_jsonb(jsonb_populate_record(NULL::` + r39SourceSchema + `.%s, $1::jsonb))::text` + declared := map[string][]string{} + for _, table := range ds02TargetTables { + rows := spec.rows[table] + if len(rows) == 0 { + t.Fatalf("source table %s is empty: a source capture claim would be vacuous", table) + } + for _, row := range rows { + raw, err := json.Marshal(row) + if err != nil { + t.Fatal(err) + } + if _, err := fx.pool.Exec(ctx, fmt.Sprintf(insert, table, table), raw); err != nil { + t.Fatalf("seed source %s row %v: %v", table, row["id"], err) + } + var text string + if err := fx.pool.QueryRow(ctx, fmt.Sprintf(canonical, table), raw).Scan(&text); err != nil { + t.Fatalf("canonical source %s: %v", table, err) + } + declared[table] = append(declared[table], text) + } + } + readback := map[string][]map[string]any{} + counts := map[string]int64{} + for _, table := range ds02TargetTables { + var rows int64 + if err := fx.pool.QueryRow(ctx, `SELECT count(*) FROM `+r39SourceSchema+`.`+table).Scan(&rows); err != nil { + t.Fatalf("count source %s: %v", table, err) + } + counts[table] = rows + // 两侧都用 PostgreSQL 自己的 jsonb 文本：键序与数字格式由服务器决定， + // 不拿 Go 的 map 排序冒充"逐字节一致"。 + readbackRows, err := fx.pool.Query(ctx, `SELECT to_jsonb(x)::text FROM `+r39SourceSchema+`.`+table+` AS x ORDER BY 1`) + if err != nil { + t.Fatalf("read back source %s: %v", table, err) + } + gotText := []string{} + for readbackRows.Next() { + var text string + if err := readbackRows.Scan(&text); err != nil { + readbackRows.Close() + t.Fatalf("scan source %s: %v", table, err) + } + gotText = append(gotText, text) + } + readbackRows.Close() + if err := readbackRows.Err(); err != nil { + t.Fatalf("read source %s: %v", table, err) + } + want := append([]string(nil), declared[table]...) + sort.Strings(want) + sort.Strings(gotText) + if !reflect.DeepEqual(want, gotText) { + t.Fatalf("source store readback for %s differs from the declaration: want %v / got %v", table, want, gotText) + } + if int64(len(want)) != rows { + t.Fatalf("source store %s counted %d rows but read %d", table, rows, len(want)) + } + readback[table] = ds02Read(t, fx, `SELECT to_jsonb(x)::text FROM `+r39SourceSchema+`.`+table+` AS x ORDER BY 1`) + } + return readback, map[string]any{ + "schema": r39SourceSchema, + "tables": ds02TargetTables, + "row_counts": counts, + "declared_matches": true, + "independent_readback": "SELECT to_jsonb(x)::text ... ORDER BY 1, compared row by row with the declared canonical rows", + "note": "本轮来源行的落地位置；产品不读它，只用于证明封存内容来自真实 PostgreSQL 行", + } +} + +// r39JobRow 读一条队列作业的完整行（payload 用原文，便于逐字节留证）。 +func r39JobRow(t *testing.T, pool *pgxpool.Pool, id string) map[string]any { + t.Helper() + rows := ds02Read(t, &writeSetFixture{pool: pool, ctx: context.Background()}, + `SELECT to_jsonb(x)::text FROM jobs AS x WHERE id='`+id+`'`) + if len(rows) == 0 { + t.Fatalf("queue job %s not found", id) + } + return rows[0] +} + +func r39QueueState(t *testing.T, pool *pgxpool.Pool, id string) string { + t.Helper() + var state string + if err := pool.QueryRow(context.Background(), `SELECT state FROM jobs WHERE id=$1`, id).Scan(&state); err != nil { + t.Fatalf("read queue state %s: %v", id, err) + } + return state +} + +// r39GovernanceTables 是需要逐行比对的治理/网关表：重放前后任何新增执行都会出现在这里。 +var r39GovernanceTables = []string{ + "jobs", "library_jobs", "library_job_stages", "library_operations", "library_attempts", + "library_versions", "library_current_versions", "library_maintenance", "library_instance_maintenance", + "library_release_authorizations", "library_dead_letter_releases", "library_executor_status", + "kg_object_gateway_intents", "kg_object_gateway_receipts", "kg_object_gateway_deliveries", +} + +func r39DumpTables(t *testing.T, pool *pgxpool.Pool, tables []string) map[string][]string { + t.Helper() + out := map[string][]string{} + for _, table := range tables { + rows, err := pool.Query(context.Background(), `SELECT to_jsonb(x)::text FROM "`+table+`" AS x ORDER BY 1`) + if err != nil { + t.Fatalf("dump governance table %s: %v", table, err) + } + values := make([]string, 0, 16) + for rows.Next() { + var raw string + if err := rows.Scan(&raw); err != nil { + rows.Close() + t.Fatalf("scan %s: %v", table, err) + } + values = append(values, raw) + } + rows.Close() + if err := rows.Err(); err != nil { + t.Fatalf("dump %s: %v", table, err) + } + out[table] = values + } + return out +} + +func r39DiffDump(before, after map[string][]string) []string { + problems := []string{} + tables := make([]string, 0, len(before)) + for table := range before { + tables = append(tables, table) + } + sort.Strings(tables) + for _, table := range tables { + if len(before[table]) != len(after[table]) { + problems = append(problems, fmt.Sprintf("%s rows %d -> %d", table, len(before[table]), len(after[table]))) + continue + } + for i := range before[table] { + if before[table][i] != after[table][i] { + problems = append(problems, fmt.Sprintf("%s row %d content changed: %s", table, i, after[table][i])) + } + } + } + return problems +} + +func TestFormalPublishHTTPReplaySucceededR39(t *testing.T) { + root := os.Getenv("KG_R38_RUN") + if root == "" { + t.Skip("requires the dedicated publish-http isolated deployment (KG_R38_RUN)") + } + if scenario := os.Getenv("KG_R38_SCENARIO"); scenario != "publish-http" { + t.Fatalf("publish-http branch requires KG_R38_SCENARIO=publish-http, got %q", scenario) + } + ctx := context.Background() + result := &r39Result{Run: filepath.Base(root), Scenario: "publish-http", Checks: []map[string]any{}} + defer func() { + result.Finished = time.Now().UTC().Format(time.RFC3339Nano) + result.Assertion = len(result.Checks) + if len(result.Failed) == 0 { + result.Status = "passed" + } else { + result.Status = "failed" + } + raw, err := json.MarshalIndent(result, "", " ") + if err != nil { + t.Logf("encode assertions: %v", err) + return + } + if err := os.WriteFile(filepath.Join(root, "result-assertions.json"), raw, 0600); err != nil { + t.Logf("write assertions: %v", err) + } + }() + evidence := func(name string) string { return filepath.Join(root, name) } + agree := func(id, detail, path string, ok bool) { + t.Helper() + result.Checks = append(result.Checks, map[string]any{"id": id, "detail": detail, "passed": ok, "evidence": path}) + if !ok { + result.Failed = append(result.Failed, id) + } + } + require := func(id, detail, path string, ok bool) { + t.Helper() + agree(id, detail, path, ok) + if !ok { + t.Fatalf("%s: %s (evidence %s)", id, detail, path) + } + } + must := func(e error) { + t.Helper() + if e != nil { + t.Fatal(e) + } + } + save := func(name string, value any) { r39Write(t, evidence(name), value) } + saveIn := func(dir, name string, value any) { r39Write(t, filepath.Join(root, dir, name), value) } + + // ---------- 1) 来源/目标身份：先生成、先声明，再封存与登记 ---------- + sourceInstance := strings.TrimSpace(os.Getenv("KG_R38_SOURCE_INSTANCE")) + sourceProject := strings.TrimSpace(os.Getenv("KG_R38_SOURCE_PROJECT")) + sourceApp := strings.TrimSpace(os.Getenv("KG_R38_SOURCE_APP")) + sourceIdentity := governance.LibrarySourceIdentity{Project: sourceProject, App: sourceApp, + Scope: "internal", TenantID: wsSourceTenant, KnowledgeBaseID: wsSourceKB} + sourceID, err := sourceIdentity.ID() + must(err) + project := os.Getenv("KG_R38_PROJECT") + sourceArchive := os.Getenv("KG_R38_SOURCE_ARCHIVE") + executorArchive := os.Getenv("KG_R38_EXECUTOR_ARCHIVE") + require("source-identity-declared", "来源实例/项目/app 在生成前由运行器独立注入", evidence("01-source-target-contract.json"), + sourceInstance != "" && sourceProject != "" && sourceApp != "" && sourceID != "") + require("archives-separated", "源 archive 与 executor archive 必须是两个绝对路径目录", evidence("01-source-target-contract.json"), + filepath.IsAbs(sourceArchive) && filepath.IsAbs(executorArchive) && filepath.Clean(sourceArchive) != filepath.Clean(executorArchive)) + must(os.MkdirAll(executorArchive, 0o700)) + executorBefore := r39ArchiveListing(t, executorArchive) + require("executor-starts-without-snapshot", "executor archive 起始不得已有本次快照，否则首发只证明复用", + evidence("03-before/executor-archive-before.json"), executorBefore["versions_present"] == false) + + fx := openWriteSetFixtureWith(t, writeSetFixtureOptions{reset: true, archiveRoot: sourceArchive, embeddingDimension: 1024}) + declaredBefore := r38SeedFullShape(t, fx) + // 适配层解析 embedding 模型/向量库需要这两列与两张表（与 r38 fullShape 同一装配）。 + _, err = fx.pool.Exec(ctx, `ALTER TABLE knowledge_bases ADD COLUMN embedding_model_id text, ADD COLUMN vector_store_id text; +CREATE TABLE models(id text PRIMARY KEY,name text,parameters jsonb,deleted_at timestamptz); +CREATE TABLE vector_stores(id text PRIMARY KEY,engine_type text,index_config jsonb,deleted_at timestamptz); +INSERT INTO models VALUES('r38-model','text-embedding-v4','{"embedding_parameters":{"dimension":1024}}',NULL); +INSERT INTO vector_stores VALUES('r38-vector','elasticsearch_v8','{}',NULL); +UPDATE knowledge_bases SET embedding_model_id='r38-model',vector_store_id='r38-vector';`) + must(err) + + // ---------- 2) 目标契约（冻结）：真实隔离端点推导，不硬编码历史指纹 ---------- + env := r39BaseEnv(t, fx) + // 发布闸口开启时 appconfig 要求显式授权方与令牌文件：先落私有文件再解析配置， + // 后面的进程与本测试用同一份 env（凭据只在本轮私有文件里）。 + env["PRODUCTION_RELEASE_AUTHORIZER_REF"] = project + env["PRODUCTION_RELEASE_AUTH_TOKEN_FILE"] = filepath.Join(root, "authorization-token") + must(os.WriteFile(env["PRODUCTION_RELEASE_AUTH_TOKEN_FILE"], + []byte(objectgateway.Digest([]byte(root+"approval"))), 0600)) + cfg, err := appconfig.Load(func(key string) string { return env[key] }) + must(err) + targetFingerprint := cfg.LibraryTargetIdentity() + stateOfRun := r39ReadJSON(t, filepath.Join(root, "state.json")) + contract := map[string]any{ + "declared_at": time.Now().UTC().Format(time.RFC3339Nano), + "declared_before": "封存、版本登记与任何 HTTP 请求之前；以下值来自运行器注入的环境与本轮已建端点", + "source": map[string]any{ + "instance": sourceInstance, "project": sourceProject, "app": sourceApp, + "scope": "internal", "tenant_id": wsSourceTenant, "knowledge_base_id": wsSourceKB, + "source_id": sourceID, + "source_id_origin": "governance.LibrarySourceIdentity{project,app,scope,tenant,kb}.ID()，由上面的独立配置推导，不从 manifest 反填", + "source_store": r39SourceSchema + " schema in " + r39DatabaseName(t, env["LIBRARY_PRODUCTION_DATABASE_URL"]), + }, + "target": map[string]any{ + "instance": cfg.LibraryTargetInstance, "project": cfg.LibraryTargetProject, + "tenant_id": cfg.LibraryProductionTenantID, "knowledge_base_id": cfg.LibraryProductionInternalKB, + "index": cfg.LibraryProductionIndex, "bucket": cfg.LibraryProductionS3Bucket, + "database": map[string]any{"name": r39DatabaseName(t, env["LIBRARY_PRODUCTION_DATABASE_URL"]), + "oid": r39DatabaseOID(t, stateOfRun, r39DatabaseName(t, env["LIBRARY_PRODUCTION_DATABASE_URL"])), + "owner": r39DatabaseOwner(t, stateOfRun, r39DatabaseName(t, env["LIBRARY_PRODUCTION_DATABASE_URL"]))}, + "writers": cfg.LibraryTargetWriters, "ingress_service": cfg.LibraryTargetIngressService, + "ingress_url": cfg.LibraryTargetIngressURL, "journal_root": cfg.LibraryTargetJournalRoot, + "gateway": env["LIBRARY_OBJECT_GATEWAY_URL"], "fingerprint": targetFingerprint, + }, + "fingerprint_rule": "appconfig.Config.LibraryTargetIdentity() over instance/project/database/tenant/KB/index/provider/endpoint/bucket；API、release worker 与本测试用同一份 env 计算", + "boundary_note": "本轮只证明本地专用链（local transport + 真实 libraryctl + 真实库/索引/对象）；不声称 SSH/云传输、不声称 Capture/Export HTTP 已验收", + } + save("01-source-target-contract.json", contract) + + // before-image 与外围哨兵：真实行，不是捏造 baseline。 + beforePG := universeDump(t, fx.pool) + saveIn("03-before", "pg-before.json", beforePG) + beforeTarget := r38TargetRows(t, fx) + beforeSentinels := r38SentinelRows(t, fx) + // 声明 → 读回：种子行（目标范围 + 邻域哨兵）必须原样落库，否则 before-image 只是文案。 + for _, table := range ds02TargetTables { + if len(declaredBefore[table]) == 0 { + t.Fatalf("declared before-image for %s is empty", table) + } + union := append(append([]map[string]any{}, beforeTarget[table]...), beforeSentinels[table]...) + if diff := r38CompareRows("seeded vs declared "+table, declaredBefore[table], union); diff != "" { + t.Fatalf("seeded before-image %s does not match its own declaration: %s", table, diff) + } + } + for _, table := range ds02TargetTables { + require("before-image-non-empty-"+table, "before-image 十表必须非空，否则全形状断言是空的", + evidence("03-before/pg-before.json"), len(beforePG[table]) > 0) + require("sentinel-non-empty-"+table, "外围哨兵必须真实存在", evidence("03-before/sentinels-before.json"), len(beforeSentinels[table]) > 0) + } + saveIn("03-before", "sentinels-before.json", beforeSentinels) + beforeIndex := recoveryIndexReadback(t, fx) + saveIn("03-before", "index-before.json", beforeIndex) + beforeObjects := objectInventoryR38(t, ctx, fx) + saveIn("03-before", "objects-before.json", beforeObjects) + beforeDigests := r38ObjectVersionDigests(t, ctx, fx, beforeObjects) + require("before-objects-have-versions", "before-image 必须有真实对象版本可供逐版本 GET", evidence("03-before/object-digests-before.json"), len(beforeDigests) > 0) + saveIn("03-before", "object-digests-before.json", beforeDigests) + saveIn("03-before", "executor-archive-before.json", executorBefore) + + // ---------- 3) 合法 fullshape 来源封存（来源行先落真实 PG，再封存） ---------- + spec := r38LegalSourceSpec(t, fx) + saveIn("03-before", "legal-source-proof.json", r38ProveLegalSourceContract(t, fx, spec)) + readback, sourceStoreEvidence := r39SeedSourceStore(t, fx, spec) + saveIn("03-before", "source-store.json", sourceStoreEvidence) + sealSpec := ds02Spec{rows: readback, vectors: ds02Vectors, objects: ds02Objects} + mapping, err := fx.target(t).Search().MappingSHA256(ctx) + must(err) + source := r38SealFullShape(t, fx, sealSpec, mapping) + saveIn("03-before", "sealed-source.json", map[string]any{"snapshot": source, "index_mapping_sha256": mapping, + "full_shape": true, "sealed_from": "r39_source independent readback"}) + sourceParts := r38SourcePartCounts(t, fx, source) + manifest, manifestRaw := r39Manifest(t, sourceArchive, source) + require("source-parts-non-empty", "十表部件必须全部非空", evidence("03-before/source-parts.json"), len(sourceParts) == len(ds02TargetTables)) + require("source-manifest-identity", "manifest 的 scope/tenant/KB 必须等于独立声明的来源身份", + evidence("03-before/source-manifest.json"), + manifest.Scope == "internal" && manifest.TenantID == wsSourceTenant && manifest.KnowledgeBaseID == wsSourceKB) + manifestSum := sha256.Sum256(manifestRaw) + require("source-manifest-address", "清单字节摘要必须等于快照 ID", evidence("03-before/source-manifest.json"), + hex.EncodeToString(manifestSum[:]) == source) + saveIn("03-before", "source-parts.json", sourceParts) + saveIn("03-before", "source-manifest.json", map[string]any{"snapshot": source, + "manifest_sha256": hex.EncodeToString(manifestSum[:]), "manifest": json.RawMessage(manifestRaw)}) + expectation := ds02Expected(t, fx, spec) + // 产品契约：knowledges.embedding_model_id 换成目标知识库绑定的模型 + // （library_scope_remap.retargetEmbeddingModelField）。期望侧独立读取目标 KB 的 + // 真实配置再施加同一规则，而不是从发布结果反填。 + targetModel := "" + must(fx.pool.QueryRow(ctx, `SELECT coalesce(embedding_model_id,'') FROM knowledge_bases WHERE id=$1 AND tenant_id=$2`, + wsTargetKB, wsTargetTenant).Scan(&targetModel)) + if targetModel != "" { + for _, row := range expectation.Rows["knowledges"] { + if _, ok := row["embedding_model_id"]; ok { + row["embedding_model_id"] = targetModel + } + } + } + // 复用 DS02 已验的期望组合：resources/resource_bindings 没有知识库列，发布按 + // 内容寻址复用同租户行；knowledge_tag_relations 按父文档归属删除。目标范围之外 + // 的这些行必须在期望里原样保留，否则"保住了"会被误判成"多出来"。 + expectation.Rows["resources"] = append(expectation.Rows["resources"], ds02SameTenantRows(t, fx, "resources")...) + // 绑定行按"所有者是否在目标范围"消失：被替换文档/分块的绑定必须删掉， + // 其它知识库的绑定原样保留（独立 SQL 先声明，再看发布结果）。 + expectation.Rows["resource_bindings"] = append(expectation.Rows["resource_bindings"], + ds02Read(t, fx, `SELECT to_jsonb(x)::text FROM resource_bindings AS x + WHERE x.tenant_id=$1 + AND NOT ((x.owner_type='knowledge' AND x.owner_id IN (SELECT id FROM knowledges WHERE tenant_id=$1 AND knowledge_base_id=$2)) + OR (x.owner_type='chunk' AND x.owner_id IN (SELECT id FROM chunks WHERE tenant_id=$1 AND knowledge_base_id=$2)))`, + wsTargetTenant, wsTargetKB)...) + expectation.Rows["knowledge_tag_relations"] = append(expectation.Rows["knowledge_tag_relations"], + ds02Read(t, fx, `SELECT to_jsonb(x)::text FROM knowledge_tag_relations AS x + WHERE knowledge_id NOT IN (SELECT id FROM knowledges WHERE tenant_id=$1 AND knowledge_base_id=$2)`, + wsTargetTenant, wsTargetKB)...) + saveIn("03-before", "expected-target-rows.json", map[string]any{"remap": map[string]any{ + "embedding_model_id": targetModel, "rule": "knowledges.embedding_model_id -> 目标知识库绑定的模型", + "source": "knowledge_bases 真实 before-image 配置"}, + "preserved_rule": "resources 追加同租户现存行（内容寻址复用）；resource_bindings 只追加所有者不在目标范围的行；knowledge_tag_relations 追加目标范围外关系", + "rows": expectation.Rows}) + require("executor-lacks-source-snapshot", "executor archive 在发布前不得已有本次快照", + evidence("03-before/executor-archive-before.json"), !r39ArchiveHas(t, executorArchive, source)) + + // ---------- 4) 目标当前版本：正式基线（HTTP，不动内容、不手写版本） ---------- + govDSN := os.Getenv("KG_R38_GOVERNANCE_DSN") + gov, err := pgxpool.New(ctx, govDSN) + must(err) + t.Cleanup(gov.Close) + must(postgresstore.ApplyMigrations(ctx, gov)) + _, err = gov.Exec(ctx, objectgateway.Schema) + must(err) + reader := gatewayReadOnlyGovernance(t, ctx, gov, govDSN) + proxy := newR38RefusalProxy(t, root, fx.searchBase) + proxy.setRefuseClean(false) + objectProxy := newR38ObjectRefusalProxy(t, root, "http://"+os.Getenv("KG_R38_GATEWAY_ADDR")) + store := postgresstore.NewStore(gov).WithLibraryTarget(targetFingerprint) + defer func() { + if !t.Failed() { + return + } + dump := map[string]any{} + for _, table := range r39GovernanceTables { + rows, err := gov.Query(context.Background(), `SELECT to_jsonb(x)::text FROM "`+table+`" AS x ORDER BY 1`) + if err != nil { + continue + } + values := []string{} + for rows.Next() { + var raw string + if rows.Scan(&raw) == nil { + values = append(values, raw) + } + } + rows.Close() + dump[table] = values + } + if raw, err := json.MarshalIndent(dump, "", " "); err == nil { + _ = os.WriteFile(filepath.Join(root, "99-failure-governance.json"), raw, 0600) + } + }() + + // 测试封存版本：登记的是本轮真实封存的快照与独立推导的 source_id。 + // 这不是 Capture/Export HTTP 的产物，证据里必须写明装配方式，不能冒充 agent 捕获。 + versionID := governance.LibraryVersionID(governance.LibraryVersionTestSealed, "internal", source) + must(store.RecordLibraryVersion(ctx, governance.LibraryVersion{ID: versionID, Scope: "internal", + Kind: governance.LibraryVersionTestSealed, SnapshotID: source, + Evidence: map[string]any{ + "source_id": sourceID, "source_identity": sourceIdentity, + "capture_job": "r39-fixture-assembled-fullshape", + "assembly": map[string]any{"kind": "fixture-assembled-legal-fullshape", + "source_store": r39SourceSchema, + "settlement": "r38ProveLegalSourceContract（产品同一份结算 SQL）", + "note": "由本轮隔离 fixture 按 DS02 十表契约装配的合法 fullshape；不是 WeKnora agent 捕获，也不声称 Capture/Export HTTP 已验收", + "manifest_sha256": source, + "source_parts_row_counts": sourceParts}, + }, + CreatedBy: "admin", CreatedAt: time.Now().UTC()})) + candidate, found, err := store.LibrarySourceCandidate(ctx, "internal", sourceID) + must(err) + require("candidate-registered", "独立 source_id 必须命中唯一 test_sealed 候选", evidence("03-before/candidate.json"), + found && candidate.SnapshotID == source) + saveIn("03-before", "candidate.json", map[string]any{"version": candidate, "source_id": sourceID, + "source_identity": sourceIdentity, "requested_snapshot": source, "found": found}) + save("01a-source-target-bound.json", map[string]any{ + "frozen_contract": evidence("01-source-target-contract.json"), + "frozen_source_id": sourceID, + "frozen_source_identity": sourceIdentity, + "source_snapshot": source, + "manifest_sha256": hex.EncodeToString(manifestSum[:]), + "manifest_path": evidence("03-before/source-manifest.json"), + "parts_path": evidence("03-before/source-parts.json"), + "source_store_path": evidence("03-before/source-store.json"), + "candidate_registration": evidence("03-before/candidate.json"), + "candidate_http_readback": evidence("03-before/overview-before-baseline.json"), + "note": "声明先冻结；这里只记录封存后的实际绑定，身份字段必须与 01 逐项相同", + "source_identity_unchanged": candidate.SnapshotID == source && candidate.Evidence["source_id"] == sourceID, + }) + + // ---------- 5) 网关 / API / 两个角色 worker（真实进程，身份逐字节核对） ---------- + save("00-scenario.json", map[string]any{"scenario": "publish-http", "label": os.Getenv("KG_R38_LABEL"), + "commit": os.Getenv("KG_R38_COMMIT"), "version": os.Getenv("KG_R38_VERSION"), + "release_version": os.Getenv("KG_R38_RELEASE_VERSION"), + "publish_entry": "POST /api/v1/library/syncs (real listener, admin session)", + "handoff": "LIBRARY_HANDOFF_TRANSPORT=local + real libraryctl; SSH/cloud transport not proven"}) + testExecutable, err := os.Executable() + must(err) + testRaw, err := os.ReadFile(testExecutable) + must(err) + testSum := sha256.Sum256(testRaw) + testIdentity := map[string]any{"label": "test", "pid": os.Getpid(), "executable": testExecutable, + "executable_sha256": hex.EncodeToString(testSum[:]), "expected_sha256": os.Getenv("KG_R38_TEST_BINARY_SHA"), + "commit": os.Getenv("KG_R38_COMMIT"), "source_tree": os.Getenv("KG_R38_SOURCE_TREE")} + require("test-process-identity", "测试进程字节必须等于本轮构建产物", evidence("00-test-process-identity.json"), + testIdentity["expected_sha256"] != "" && testIdentity["executable_sha256"] == testIdentity["expected_sha256"]) + save("00-test-process-identity.json", testIdentity) + + // 授权令牌/审计文件：真实网关要读，凭据只在私有文件里。 + dataToken := objectgateway.Digest([]byte(root + "data")) + controlToken := objectgateway.Digest([]byte(root + "control")) + env["LIBRARY_PRODUCTION_ELASTIC_URL"] = proxy.url + env["LIBRARY_OBJECT_GATEWAY_URL"] = objectProxy.url + env["LIBRARY_OBJECT_GATEWAY_TOKEN"] = dataToken + env["LIBRARY_HANDOFF_LIBRARYCTL"] = filepath.Join(os.Getenv("KG_R38_BIN"), "libraryctl") + env["LIBRARY_HANDOFF_TRANSPORT"] = "local" + env["LIBRARY_HANDOFF_LOCAL_ROOT"] = executorArchive + env["LIBRARY_HANDOFF_REMOTE_ARCHIVE"] = executorArchive + env["LIBRARY_HANDOFF_BUNDLE_DIR"] = os.Getenv("KG_R38_EXECUTOR_BUNDLE") + env["LIBRARY_HANDOFF_STAGING_DIR"] = os.Getenv("KG_R38_SOURCE_STAGING") + for _, dir := range []string{env["LIBRARY_HANDOFF_BUNDLE_DIR"], env["LIBRARY_HANDOFF_STAGING_DIR"]} { + must(os.MkdirAll(dir, 0o700)) + } + cfg, err = appconfig.Load(func(key string) string { return env[key] }) + must(err) + require("config-fingerprint-stable", "加入网关/搬运配置后目标指纹必须不变（指纹只绑数据身份）", + evidence("01-source-target-contract.json"), cfg.LibraryTargetIdentity() == targetFingerprint) + env["LIBRARY_HANDOFF_LIBRARYCTL"] = filepath.Join(os.Getenv("KG_R38_BIN"), "libraryctl") + + // worker 的对象身份只读：对象写入只能经网关，receipt 因此必然来自真实生产写。 + readStore, err := NewS3ObjectStore(S3ObjectStoreConfig{Provider: "minio", Endpoint: env["LIBRARY_PRODUCTION_S3_ENDPOINT"], + Bucket: fx.bucket, AccessKeyID: env["LIBRARY_PRODUCTION_S3_ACCESS_KEY"], SecretAccessKey: env["LIBRARY_PRODUCTION_S3_SECRET_KEY"], + Region: "us-east-1", ForcePathStyle: true}) + must(err) + allowedRead, err := readStore.Open(ctx, "minio", wsLocator("backend-target", "minio", fx.bucket, 10001, "res-target")) + must(err) + allowedBody, err := io.ReadAll(allowedRead) + allowedRead.Close() + must(err) + if denied := readStore.Put(ctx, "minio", wsLocator("backend-target", "minio", fx.bucket, 10001, "r39-denied"), + bytes.NewReader([]byte("denied")), 6); denied == nil { + t.Fatal("worker readonly identity unexpectedly wrote an object") + } + save("02-target-identity.json", map[string]any{"worker_object_identity": "readonly", "gateway_only_writer": true, + "bucket": fx.bucket, "index": fx.index, "target": targetFingerprint, "project": project, + "readonly_read_bytes": string(allowedBody), "readonly_put_denied": true}) + + scope := objectgateway.Digest([]byte("minio\x00" + env["LIBRARY_PRODUCTION_S3_ENDPOINT"] + "\x00" + fx.bucket)) + pub, priv, err := ed25519.GenerateKey(rand.Reader) + must(err) + digestFile := func(name string) string { + raw, err := os.ReadFile(filepath.Join(root, name)) + must(err) + return objectgateway.Digest(raw) + } + initialDrain := filepath.Join(root, "initial-drain.json") + if _, err := os.Stat(initialDrain); err != nil { + t.Fatalf("runner must provide real initial drain evidence: %v", err) + } + payload, err := json.Marshal(objectgateway.Attestation{Protocol: objectgateway.Protocol, Scope: scope, GatewayID: project, + IssuedAt: time.Now().Add(-time.Second), ExpiresAt: time.Now().Add(50 * time.Minute), + WriterInventory: digestFile("02-target-identity.json"), PolicyEvidence: digestFile("02-target-identity.json"), + DenialEvidence: digestFile("02-target-identity.json"), DrainEvidence: digestFile("initial-drain.json")}) + must(err) + signedRaw, err := json.Marshal(objectgateway.SignedAttestation{Payload: payload, Signature: ed25519.Sign(priv, payload)}) + must(err) + must(os.WriteFile(filepath.Join(root, "audit.json"), signedRaw, 0600)) + + processes := []map[string]any{} + expectedBinaries := map[string]string{} + must(json.Unmarshal([]byte(os.Getenv("KG_R38_BINARY_SHA")), &expectedBinaries)) + start := func(binary, label string, values map[string]string) *exec.Cmd { + t.Helper() + cmd := exec.Command(filepath.Join(os.Getenv("KG_R38_BIN"), binary)) + cmd.Env = []string{"PATH=" + os.Getenv("PATH"), "HOME=" + os.Getenv("HOME")} + for k, v := range values { + cmd.Env = append(cmd.Env, k+"="+v) + } + f, err := os.OpenFile(filepath.Join(root, label+".log"), os.O_CREATE|os.O_WRONLY|os.O_EXCL, 0600) + must(err) + cmd.Stdout = f + cmd.Stderr = f + must(cmd.Start()) + executable, err := os.Readlink(fmt.Sprintf("/proc/%d/exe", cmd.Process.Pid)) + must(err) + raw, err := os.ReadFile(executable) + must(err) + sum := sha256.Sum256(raw) + digest := hex.EncodeToString(sum[:]) + record := map[string]any{"label": label, "role": label, "pid": cmd.Process.Pid, "binary": binary, + "executable": executable, "executable_sha256": digest, "expected_sha256": expectedBinaries[binary], + "match": expectedBinaries[binary] != "" && digest == expectedBinaries[binary], + "commit": os.Getenv("KG_R38_COMMIT"), "source_tree": os.Getenv("KG_R38_SOURCE_TREE"), + "started_at": time.Now().UTC()} + if !record["match"].(bool) { + t.Fatalf("%s process identity mismatch: %v", label, record) + } + processes = append(processes, record) + save("02-process-provenance.json", processes) + t.Cleanup(func() { _ = cmd.Process.Kill(); _ = cmd.Wait(); f.Close() }) + return cmd + } + genv := map[string]string{"OBJECT_GATEWAY_LISTEN": os.Getenv("KG_R38_GATEWAY_ADDR"), "OBJECT_GATEWAY_DATABASE_URL": govDSN, + "OBJECT_GATEWAY_GOVERNANCE_DATABASE_URL": reader.Config().ConnString(), "OBJECT_GATEWAY_ARCHIVE_ROOT": executorArchive, + "OBJECT_GATEWAY_TARGET_ID": targetFingerprint, "OBJECT_GATEWAY_PROVIDER": "minio", + "OBJECT_GATEWAY_ENDPOINT": env["LIBRARY_PRODUCTION_S3_ENDPOINT"], "OBJECT_GATEWAY_BUCKET": fx.bucket, + "OBJECT_GATEWAY_REGION": "us-east-1", "OBJECT_GATEWAY_ACCESS_KEY": os.Getenv("KG_LIBRARY_TEST_S3_ACCESS_KEY"), + "OBJECT_GATEWAY_SECRET_KEY": os.Getenv("KG_LIBRARY_TEST_S3_SECRET_KEY"), "OBJECT_GATEWAY_DATA_TOKEN": dataToken, + "OBJECT_GATEWAY_CONTROL_TOKEN": controlToken, "OBJECT_GATEWAY_ID": project, + "OBJECT_GATEWAY_AUDIT_FILE": filepath.Join(root, "audit.json"), + "OBJECT_GATEWAY_AUDIT_PUBLIC_KEY": base64.StdEncoding.EncodeToString(pub)} + start("object-gateway", "gateway", genv) + client, err := objectgateway.NewClient(env["LIBRARY_OBJECT_GATEWAY_URL"], dataToken, scope) + must(err) + ready := false + for i := 0; i < 100; i++ { + if client.ReadyForAcquisition(ctx) == nil { + ready = true + break + } + time.Sleep(100 * time.Millisecond) + } + if !ready { + t.Fatal("object gateway not ready; inspect gateway.log") + } + + apiAddr := os.Getenv("KG_R38_API_ADDR") + apiEnv := map[string]string{} + for k, v := range env { + apiEnv[k] = v + } + apiEnv["APP_ADDR"] = apiAddr + apiEnv["GOVERNANCE_SPOOL_ROOT"] = filepath.Join(root, "spool") + apiEnv["RELEASE_WORKER_ENABLED"] = "false" + apiEnv["LIBRARY_ARCHIVE_ROOT"] = sourceArchive + must(os.MkdirAll(apiEnv["GOVERNANCE_SPOOL_ROOT"], 0700)) + start("api", "api", apiEnv) + apiURL := "http://" + apiAddr + for i := 0; i < 120; i++ { + if response, err := http.Get(apiURL + "/health"); err == nil { + response.Body.Close() + if response.StatusCode == 200 { + break + } + } + if i == 119 { + t.Fatal("api /health never became 200") + } + time.Sleep(200 * time.Millisecond) + } + captureEnv := map[string]string{} + for k, v := range env { + captureEnv[k] = v + } + captureEnv["WORKER_ID"] = project + "-capture" + captureEnv["LIBRARY_WORKER_ROLE"] = "capture" + captureEnv["LIBRARY_ARCHIVE_ROOT"] = sourceArchive + start("worker", "capture", captureEnv) + releaseEnv := map[string]string{} + for k, v := range env { + releaseEnv[k] = v + } + releaseEnv["WORKER_ID"] = project + "-release" + releaseEnv["LIBRARY_WORKER_ROLE"] = "release-executor" + releaseEnv["LIBRARY_ARCHIVE_ROOT"] = executorArchive + start("worker", "release", releaseEnv) + libraryctlRaw, err := os.ReadFile(env["LIBRARY_HANDOFF_LIBRARYCTL"]) + must(err) + libraryctlSum := sha256.Sum256(libraryctlRaw) + require("libraryctl-identity", "capture 使用的 libraryctl 必须是本轮真实构建产物", + evidence("02-process-provenance.json"), hex.EncodeToString(libraryctlSum[:]) == expectedBinaries["libraryctl"]) + processes = append(processes, map[string]any{"label": "libraryctl", "role": "real-cli", + "binary": "libraryctl", "executable": env["LIBRARY_HANDOFF_LIBRARYCTL"], + "executable_sha256": hex.EncodeToString(libraryctlSum[:]), "expected_sha256": expectedBinaries["libraryctl"], + "match": true, "note": "由 capture 角色在每次 push 时跨进程调用（snapshot-unpack/snapshot-verify）"}) + save("02-process-provenance.json", processes) + + // 管理员会话（只改这一次性治理库里的 admin 口令；口令只存在内存里）。 + password := hex.EncodeToString(func() []byte { b := make([]byte, 18); _, _ = rand.Read(b); return b }()) + hash, err := bcrypt.GenerateFromPassword([]byte(password), 10) + must(err) + _, err = gov.Exec(ctx, `UPDATE users SET password_hash=$1, must_change_password=false, failed_attempts=0, + locked_until=NULL, state='active' WHERE username='admin'`, string(hash)) + must(err) + jar, err := cookiejar.New(nil) + must(err) + api := r38API{base: apiURL, client: &http.Client{Jar: jar, Timeout: 60 * time.Second}} + code, loginBody := api.do(t, http.MethodPost, "/api/v1/auth/login", map[string]string{"username": "admin", "password": password}) + loginUser, _ := loginBody["data"].(map[string]any) + require("admin-login", "管理员会话必须来自真实登录接口", evidence("07-session.json"), + code == http.StatusOK && loginUser != nil && loginUser["role"] == "admin") + save("07-session.json", map[string]any{"login_http": code, "role": "admin", "auth": "session cookie (password not recorded)"}) + + // HTTP 读回源候选：必须命中同一声明身份，而不是另一个更方便的候选。 + overviewCode, overview := api.do(t, http.MethodGet, "/api/v1/library/overview?scope=internal", nil) + saveIn("03-before", "overview-before-baseline.json", map[string]any{"http": overviewCode, "body": overview}) + candidateView, _ := overview["test_sealed"].(map[string]any) + require("candidate-http-readback", "overview.test_sealed 必须是本轮快照且带同一声明 source_id", + evidence("03-before/overview-before-baseline.json"), + overviewCode == http.StatusOK && candidateView != nil && candidateView["snapshot_id"] == source && candidateView["source_id"] == sourceID) + + // 发布授权：管理员通过真实接口签发，绑定目标快照与参数摘要。 + code, authBody := api.do(t, http.MethodPost, "/api/v1/library/authorizations", + map[string]any{"scope": "internal", "snapshot_id": source, "action": "sync", "expires_in_minutes": 120}) + authCode := code + require("authorization-issued", "sync 授权必须由真实接口签发", evidence("04-baseline-and-auth.json"), authCode == http.StatusCreated) + save("04-baseline-and-auth.json", map[string]any{"authorization_http": authCode, "authorization": authBody}) + + // 目标基线：目标非空且没有生效版本 → 只能走正式基线捕获。 + code, baseBody := api.do(t, http.MethodPost, "/api/v1/library/baselines", map[string]any{"scope": "internal"}) + baselineCode := code + require("baseline-accepted", "目标基线的唯一正式入口必须接受本次请求", evidence("04-baseline-and-auth.json"), baselineCode == http.StatusAccepted) + baseJob, _ := baseBody["job"].(map[string]any) + baselineJobID, _ := baseJob["id"].(string) + if baselineJobID == "" { + t.Fatalf("baseline response carries no job id: %v", baseBody) + } + baseline := r39WaitLibraryJob(t, store, baselineJobID, governance.LibraryJobSucceeded, 420) + save("04-baseline-and-auth.json", map[string]any{"authorization_http": authCode, "authorization": authBody, + "baseline": map[string]any{"http": baselineCode, "job": baselineJobID, "terminal": baseline}}) + require("baseline-succeeded", "基线必须由 release worker 真实捕获+校验+登记后成功", + evidence("04-baseline-and-auth.json"), baseline.State == governance.LibraryJobSucceeded) + baselineVersion, hasBaseline, err := store.LibraryCurrentVersion(ctx, "internal") + must(err) + _, err = librarysnapshot.Open(executorArchive, 256<<20) + must(err) + require("baseline-current-version", "基线必须是 production_effective 的目标整库真实捕获（带 target_baseline 证据）", + evidence("04-baseline-and-auth.json"), + hasBaseline && baselineVersion.Kind == governance.LibraryVersionProductionEffective && + baselineVersion.Evidence["action"] == "target_baseline_init" && baselineVersion.Evidence["origin"] == "target_baseline") + maintenanceAfterBaseline, err := store.LibraryMaintenanceOf(ctx, "internal") + must(err) + instanceAfterBaseline, err := store.InstanceMaintenanceOf(ctx, cfg.LibraryTargetInstance) + must(err) + entryAfterBaseline := r39EntryOpen(t, cfg.LibraryTargetIngressURL) + saveIn("06-terminal", "baseline-release.json", map[string]any{"job": baselineJobID, + "current_version": baselineVersion, "scope_maintenance": maintenanceAfterBaseline, + "instance_maintenance": instanceAfterBaseline, "entry": entryAfterBaseline}) + require("baseline-window-released", "基线收尾必须释放两把维护占用并把入口恢复", + evidence("06-terminal/baseline-release.json"), + maintenanceAfterBaseline != nil && maintenanceAfterBaseline.ReleasedAt != nil && + instanceAfterBaseline != nil && instanceAfterBaseline.ReleasedAt != nil && entryAfterBaseline["open"] == true) + + // ---------- 6) 唯一发布请求：真实监听 + 原 body ---------- + overviewCode, overview = api.do(t, http.MethodGet, "/api/v1/library/overview?scope=internal", nil) + gate, _ := overview["gate"].(map[string]any) + save("04-gate-before-publish.json", map[string]any{"http": overviewCode, "gate": gate, "production": overview["production"]}) + require("gate-open", "闸口必须放行（当前生效版本+有效授权+目标可读）", evidence("04-gate-before-publish.json"), gate["sync_allowed"] == true) + + operationKey := project + "-publish-op" + requestBody := map[string]any{"scope": "internal", "target_snapshot_id": source, + "operation_key": operationKey, "source_id": sourceID} + rawRequest, err := json.Marshal(requestBody) + must(err) + publishCode, publishBody := api.do(t, http.MethodPost, "/api/v1/library/syncs", json.RawMessage(rawRequest)) + published, _ := publishBody["job"].(map[string]any) + jobID, _ := published["id"].(string) + operationID, _ := published["operation_id"].(string) + require("publish-http-accepted", "POST /api/v1/library/syncs 必须返回 202 且新建作业（不是复用）", + evidence("04-publish-http.json"), + publishCode == http.StatusAccepted && jobID != "" && operationID != "" && publishBody["reused"] == false) + processes = append(processes, map[string]any{"label": "publish-request", "role": "http-request", + "at": time.Now().UTC(), "job": jobID, "operation": operationID, + "processes_in_this_request": []string{"api", "capture", "release", "gateway"}, + "note": "本条把四个真实进程与这一次 POST 关联起来；时间取自同一台机器"}) + save("02-process-provenance.json", processes) + save("04-publish-http.json", map[string]any{ + "method": "POST", "url": apiURL + "/api/v1/library/syncs", + "auth": "admin session cookie (credentials not recorded)", "status": publishCode, + "request_body": json.RawMessage(rawRequest), "response": publishBody, + "operation_id": operationID, "job_id": jobID, "reused": publishBody["reused"], + "handoff_enabled": true, "note": "真实监听 HTTP，不是 httptest，也不是直接调用 store.CreateLibraryDispatch"}) + + // ---------- 7) 自然消费：handoff → release → 业务成功 ---------- + handoffQueue := governance.LibraryHandoffQueueJobID(jobID) + releaseQueue := governance.QueueJobIDFor(jobID) + firstHandoff := r39FirstQueueObservation(t, gov, handoffQueue, 60) + require("handoff-enqueued", "发布作业必须先进入 library.snapshot.handoff 队列，而不是直接进 release", + evidence("05-handoff/queue.json"), firstHandoff["type"] == "library.snapshot.handoff") + r39WaitQueue(t, gov, handoffQueue, "completed", 300) + handoffQueueRow := r39JobRow(t, gov, handoffQueue) + require("handoff-completed", "capture 角色的 handoff 队列作业必须 completed", evidence("05-handoff/queue.json"), + handoffQueueRow["state"] == "completed") + r39WaitQueue(t, gov, releaseQueue, "completed", 420) + releaseQueueRow := r39JobRow(t, gov, releaseQueue) + require("release-completed", "release-executor 的发布队列作业必须 completed", evidence("06-terminal/queues.json"), + releaseQueueRow["state"] == "completed") + business := r39WaitLibraryJob(t, store, jobID, governance.LibraryJobSucceeded, 120) + save("06-terminal/job.json", map[string]any{"business": business, "handoff_queue": handoffQueueRow, + "release_queue": releaseQueueRow, "handoff_queue_id": handoffQueue, "release_queue_id": releaseQueue}) + require("business-succeeded", "业务作业必须是 succeeded（不是 pending/202/日志）", evidence("06-terminal/job.json"), + business.State == governance.LibraryJobSucceeded) + + // 持久行：阶段 / 代次 / 版本 / 当前指针 / 维护。 + stages := ds02Read(t, &writeSetFixture{pool: gov, ctx: ctx}, + `SELECT to_jsonb(x)::text FROM library_job_stages AS x WHERE job_id='`+jobID+`' ORDER BY stage`) + attempts := ds02Read(t, &writeSetFixture{pool: gov, ctx: ctx}, + `SELECT to_jsonb(x)::text FROM library_attempts AS x WHERE job_id='`+jobID+`' ORDER BY epoch`) + operations := ds02Read(t, &writeSetFixture{pool: gov, ctx: ctx}, + `SELECT to_jsonb(x)::text FROM library_operations AS x WHERE id='`+operationID+`'`) + versionsOfSnapshot := ds02Read(t, &writeSetFixture{pool: gov, ctx: ctx}, + `SELECT to_jsonb(x)::text FROM library_versions AS x WHERE scope='internal' AND snapshot_id='`+source+`' ORDER BY kind`) + currentRows := ds02Read(t, &writeSetFixture{pool: gov, ctx: ctx}, + `SELECT to_jsonb(x)::text FROM library_current_versions AS x WHERE scope='internal'`) + maintenanceRows := ds02Read(t, &writeSetFixture{pool: gov, ctx: ctx}, + `SELECT to_jsonb(x)::text FROM library_maintenance AS x WHERE scope='internal' ORDER BY acquired_at`) + instanceRows := ds02Read(t, &writeSetFixture{pool: gov, ctx: ctx}, + `SELECT to_jsonb(x)::text FROM library_instance_maintenance AS x ORDER BY acquired_at`) + journal := r39ReadJSON(t, filepath.Join(root, "maintenance", jobID+".json")) + save("06-terminal/durable.json", map[string]any{"stages": stages, "attempts": attempts, "operations": operations, + "versions_of_snapshot": versionsOfSnapshot, "current": currentRows, "maintenance": maintenanceRows, + "instance_maintenance": instanceRows, "journal": journal}) + effectiveCount := 0 + for _, row := range versionsOfSnapshot { + if row["kind"] == "production_effective" && row["superseded_at"] == nil { + effectiveCount++ + } + } + require("production-effective", "发布必须登记一条未 superseded 的 production_effective（绑定本轮 job/operation）", + evidence("06-terminal/durable.json"), effectiveCount == 1) + currentSnapshot := "" + for _, row := range currentRows { + currentSnapshot, _ = row["snapshot_id"].(string) + } + require("current-pointer", "当前生效版本指针必须指向本轮快照", evidence("06-terminal/durable.json"), currentSnapshot == source) + scopeRow, err := store.LibraryMaintenanceOf(ctx, "internal") + must(err) + instanceRow, err := store.InstanceMaintenanceOf(ctx, cfg.LibraryTargetInstance) + must(err) + entryAfter := r39EntryOpen(t, cfg.LibraryTargetIngressURL) + require("maintenance-released", "scope/instance 两把维护占用都必须释放", + evidence("06-terminal/maintenance.json"), + scopeRow != nil && scopeRow.ReleasedAt != nil && instanceRow != nil && instanceRow.ReleasedAt != nil) + require("entry-restored", "真实入口必须读回 200（不是只看进程健康）", evidence("06-terminal/entry.json"), entryAfter["open"] == true) + save("06-terminal/maintenance.json", map[string]any{"scope": scopeRow, "instance": instanceRow}) + save("06-terminal/entry.json", entryAfter) + save("06-terminal/journal.json", journal) + for _, pair := range [][2]string{{"api", "api.log"}, {"capture", "capture.log"}, {"release", "release.log"}, {"gateway", "gateway.log"}} { + raw, err := os.ReadFile(filepath.Join(root, pair[1])) + if err != nil { + t.Fatalf("read %s log: %v", pair[0], err) + } + must(os.WriteFile(filepath.Join(root, "06-terminal", pair[0]+".log"), raw, 0600)) + } + + // ---------- 8) 05-handoff：两端字节 + 真实 pack/unpack/verify ---------- + handoffEvidence := r39HandoffEvidence(t, gov, store, jobID, source, sourceArchive, executorArchive, + env["LIBRARY_HANDOFF_LIBRARYCTL"], filepath.Join(root, "05-handoff", "repack")) + saveIn("05-handoff", "queue.json", map[string]any{"handoff_queue": handoffQueue, "first_observation": firstHandoff, + "terminal": handoffQueueRow, "payload": handoffQueueRow["payload"], + "release_handover": map[string]any{"queue": releaseQueue, "idempotency_key": releaseQueueRow["idempotency_key"], + "type": releaseQueueRow["type"], "payload": releaseQueueRow["payload"], "state": releaseQueueRow["state"], + "handoff_stage": stages}}) + saveIn("05-handoff", "evidence.json", handoffEvidence) + require("handoff-executor-verified", "executor 端清单与每个 blob 必须逐字节等于源端", + evidence("05-handoff/evidence.json"), handoffEvidence["executor_matches_source"] == true) + require("handoff-libraryctl-pack-unpack-verify", "真实 libraryctl 的 pack/unpack/verify 必须全部 rc=0 且摘要一致", + evidence("05-handoff/libraryctl.json"), handoffEvidence["libraryctl_all_zero"] == true) + require("handoff-bundle-cleaned", "传输暂存 bundle 必须被清理（只留归档本体）", + evidence("05-handoff/evidence.json"), handoffEvidence["bundle_absent_after"] == true) + + // ---------- 9) 07-after：独立内容/映射/索引/对象读回 ---------- + afterPG := universeDump(t, fx.pool) + saveIn("07-after", "pg-after.json", afterPG) + afterTarget := r38TargetRows(t, fx) + afterSentinels := r38SentinelRows(t, fx) + pgDiffs, sentinelDiffs := []string{}, []string{} + for _, table := range ds02TargetTables { + actual := r39PublishedScope(t, fx, table, afterTarget[table]) + if diff := r38CompareRows("published "+table, expectation.Rows[table], actual); diff != "" { + pgDiffs = append(pgDiffs, diff) + } + if diff := r38CompareRows("sentinel "+table, beforeSentinels[table], afterSentinels[table]); diff != "" { + sentinelDiffs = append(sentinelDiffs, diff) + } + } + saveIn("07-after", "pg-verdict.json", map[string]any{"match": len(pgDiffs) == 0, "diffs": pgDiffs, + "tables": ds02TargetTables}) + require("published-pg-tables", "PG 十表的行身份与逐列内容必须与独立期望一致", + evidence("07-after/pg-verdict.json"), len(pgDiffs) == 0) + require("sentinels-intact", "目标范围之外的行/索引文档/对象必须逐列不变", + evidence("07-after/pg-verdict.json"), len(sentinelDiffs) == 0) + if len(pgDiffs) > 0 || len(sentinelDiffs) > 0 { + t.Fatalf("published content differs from the independent expectation:\n%s\n%s", + strings.Join(pgDiffs, "\n"), strings.Join(sentinelDiffs, "\n")) + } + saveIn("07-after", "sentinels-after.json", afterSentinels) + changedByPublish := 0 + for _, table := range ds02TargetTables { + if r38CompareRows("before-vs-after "+table, beforeTarget[table], afterTarget[table]) != "" { + changedByPublish++ + } + } + require("publish-changed-target", "发布必须真的改写目标范围内容（不是 before=after 的空操作）", + evidence("07-after/pg-after.json"), changedByPublish > 0) + references := r38RequireReferences(t, "after publish", afterTarget) + saveIn("07-after", "reference-promises.json", references) + afterIndex := recoveryIndexReadback(t, fx) + saveIn("07-after", "index-after.json", afterIndex) + indexDiff := r39CheckPublishedIndex(t, sealSpec, beforeIndex, afterIndex) + saveIn("07-after", "index-verdict.json", indexDiff) + require("index-published", "索引必须与独立期望逐文档一致（含向量），外围文档不变", + evidence("07-after/index-verdict.json"), indexDiff["match"] == true) + objectVerdict := r39CheckPublishedObjects(t, ctx, fx, sealSpec, expectation) + saveIn("07-after", "object-verdict.json", objectVerdict) + require("objects-published", "发布后的资源对象必须按版本 GET 到与来源逐字节相同的字节", + evidence("07-after/object-verdict.json"), objectVerdict["match"] == true) + ledger := r38FindWriteSet(t, executorArchive, jobID) + mappingDiffs := ds02CheckLocalIDMapping(r38LocalColumns(t, fx, spec, expectation), ledger.LocalIDs) + saveIn("06-terminal", "ledger.json", map[string]any{"writeset_digest": ledger.Digest, "job": ledger.JobID, + "target_snapshot": ledger.TargetSnapshotID, "backup_snapshot": ledger.BackupSnapshotID, + "local_ids": ledger.LocalIDs}) + saveIn("07-after", "publish-ledger.json", map[string]any{"writeset_digest": ledger.Digest, "job": ledger.JobID, + "target_snapshot": ledger.TargetSnapshotID, "backup_snapshot": ledger.BackupSnapshotID, + "local_ids": ledger.LocalIDs, "mapping_diffs": mappingDiffs}) + require("ledger-mapping", "写集合必须绑定本轮 job/operation/target 且通过独立映射 oracle", + evidence("07-after/publish-ledger.json"), len(ledger.LocalIDs) > 0 && len(mappingDiffs) == 0 && ledger.JobID == jobID) + receipts := ds02Read(t, &writeSetFixture{pool: gov, ctx: ctx}, + `SELECT to_jsonb(x)::text FROM kg_object_gateway_receipts AS x ORDER BY operation`) + saveIn("07-after", "receipts.json", receipts) + saveIn("06-terminal", "receipts.json", receipts) + require("gateway-receipts", "对象写入必须留下真实网关 receipt", evidence("07-after/receipts.json"), len(receipts) > 0) + + // ---------- 10) 08-replay：同一 body 原样重放 ---------- + beforeReplayGovernance := r39DumpTables(t, gov, r39GovernanceTables) + beforeReplayPG := universeDump(t, fx.pool) + beforeReplayIndex := recoveryIndexReadback(t, fx) + beforeReplayInventory := objectInventoryR38(t, ctx, fx) + beforeReplayObjects, beforeReplayDigests := r39ObjectVersionReadback(t, ctx, fx, beforeReplayInventory) + saveIn("08-replay", "before.json", map[string]any{"governance": beforeReplayGovernance, "pg": beforeReplayPG, + "index": beforeReplayIndex, "object_versions": beforeReplayObjects}) + replayCode, replayBody := api.do(t, http.MethodPost, "/api/v1/library/syncs", json.RawMessage(rawRequest)) + replayed, _ := replayBody["job"].(map[string]any) + require("replay-reused", "同一 body 重放必须 reused=true 并复用同一 operation/job", evidence("08-replay-http.json"), + replayCode == http.StatusAccepted && replayBody["reused"] == true && + replayed["id"] == jobID && replayed["operation_id"] == operationID) + afterReplayGovernance := r39DumpTables(t, gov, r39GovernanceTables) + afterReplayPG := universeDump(t, fx.pool) + afterReplayIndex := recoveryIndexReadback(t, fx) + afterReplayInventory := objectInventoryR38(t, ctx, fx) + afterReplayObjects, afterReplayDigests := r39ObjectVersionReadback(t, ctx, fx, afterReplayInventory) + objectByteProblems := r39CompareObjectVersionBytes("replay", beforeReplayObjects, afterReplayObjects, + beforeReplayDigests, afterReplayDigests) + saveIn("08-replay", "object-bytes.json", map[string]any{ + "before_count": len(beforeReplayObjects), "after_count": len(afterReplayObjects), + "before_versions": beforeReplayObjects, "after_versions": afterReplayObjects, + "match": len(objectByteProblems) == 0, "problems": objectByteProblems, + "method": "每个 key+versionId 单独 GET，本地 SHA256；ETag/size 只作旁证"}) + problems := r39DiffDump(beforeReplayGovernance, afterReplayGovernance) + problems = append(problems, r39DiffStrings("pg", r39Flatten(beforeReplayPG), r39Flatten(afterReplayPG))...) + if !reflect.DeepEqual(beforeReplayIndex, afterReplayIndex) { + problems = append(problems, "index readback changed on replay") + } + if !reflect.DeepEqual(beforeReplayObjects, afterReplayObjects) { + problems = append(problems, fmt.Sprintf("object versions changed on replay: %d -> %d", len(beforeReplayObjects), len(afterReplayObjects))) + } + problems = append(problems, objectByteProblems...) + replayRecord := map[string]any{"request_body": json.RawMessage(rawRequest), "status": replayCode, + "response": replayBody, "reused": replayBody["reused"], "job_id": replayed["id"], "operation_id": replayed["operation_id"], + "original_job_id": jobID, "original_operation_id": operationID, + "governance_diff": problems, "object_versions_before": len(beforeReplayObjects), + "object_versions_after": len(afterReplayObjects), "object_byte_problems": objectByteProblems, + "new_execution_evidence": map[string]any{ + "jobs": len(afterReplayGovernance["jobs"]) - len(beforeReplayGovernance["jobs"]), + "library_jobs": len(afterReplayGovernance["library_jobs"]) - len(beforeReplayGovernance["library_jobs"]), + "library_attempts": len(afterReplayGovernance["library_attempts"]) - len(beforeReplayGovernance["library_attempts"]), + "versions": len(afterReplayGovernance["library_versions"]) - len(beforeReplayGovernance["library_versions"]), + "receipts": len(afterReplayGovernance["kg_object_gateway_receipts"]) - len(beforeReplayGovernance["kg_object_gateway_receipts"]), + }} + save("08-replay-http.json", replayRecord) + saveIn("08-replay", "after.json", map[string]any{"governance": afterReplayGovernance, "pg": afterReplayPG, + "index": afterReplayIndex, "object_versions": afterReplayObjects}) + require("replay-object-version-bytes", "重放前后每个 key+versionId 实取 GET 的完整版本清单与 SHA256 必须完全相同", + evidence("08-replay/object-bytes.json"), len(objectByteProblems) == 0) + require("replay-zero-side-effects", "重放不得新增执行/写入：治理行、目标内容、索引、对象版本必须逐字节不变", + evidence("08-replay-http.json"), len(problems) == 0) + afterOrderPG := universeDump(t, fx.pool) + if diff := diffUniverse(afterPG, afterOrderPG); diff != "" { + t.Fatalf("target content changed during replay: %s", diff) + } + + // ---------- 11) 边界：不声称的能力 ---------- + save("10-boundaries.json", map[string]any{ + "candidate_ui": "not_tested", + "formal_8088": "not_deployed", + "ssh_cloud_transport": "not_proven (local transport only)", + "capture_export_http": "not_tested", + "ds01_kill_put_matrix": "open, not re-run in this scenario", + "deployment": "not_deployed; handed to GPT-6 Astra for review", + "production_or_cloud_write": "none", + }) +}