test(r39): 重放前后逐对象实取字节摘要比对

原重放段只比 Listing 的 key@versionId#etag#size，同一版本号下的字节
被改写它看不见，因此"重放后字节不变"缺实测证据。

改为重放前后各调一次 r38ObjectVersionDigests：逐 key+versionId 真实
GET 并本地算 SHA256，落完整版本清单（含版本、ETag、size、摘要），
再逐项比较；GET 失败、清单缺条目、版本增减或摘要不同都判失败。
治理 PG/ES/ledger/receipt 零新增断言保持原样。

离线负控 TestR39ReplayObjectByteControl：固定"旧判据对字节改写报相同"
的盲区，并要求新判据在同一元数据上报 bytes changed、正例不误报、
空清单/缺摘要必须红；用变异（去掉摘要比较）确认该负控会红后恢复。

台账：04-baseline-and-auth.json 分开记录授权 201 与基线 202，
不再让后一次状态码覆盖前一次。

.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：新分支推送，范围按默认分支为基准判）"

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

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
+}

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")
+	}
+}