fix(release): 完成分发身份与 CAS legacy ownership 收尾

2026-09-18 07:24:55 +08:00 · 2026-09-13 10:44:24 +08:00
parent 786b739f99
commit 5bae90cb14
14 changed files with 867 additions and 109 deletions
@@ -52,13 +52,18 @@ TextUnit scope、source history 和 approved review。worker、TM 复用、人
 `release.status/list/distribution/cleanup` 已由 Rust 从既有状态、manifest、文件系统和
 CAS/reference 元数据统一生成，Go 仅 typed 转发。CAS repository 的对象文件、引用计数
 和 GC 通过跨进程操作锁协调，release-local CAS 引用以 `(ownership_id, ordinal)` ownership
-记录幂等释放；新清单持久化 `ownership_id`，旧清单走不生成新 identity 的 legacy
-cleanup path；localized publish/rollback 通过 output-root 单写者锁和事务日志恢复
+记录幂等释放；新清单持久化 `ownership_id`，旧清单按 output-root scope、稳定 source
+identity 和 generation-aware legacy cleanup path 迁移；localized publish/rollback 通过
+output-root 单写者锁和事务日志恢复
 current、version-state、version 目录，publish 只有最终 `verified` phase 才能恢复为已提交。
-localized release 还写入实际 bytes/BLAKE3 的 `localized-distribution-manifest.json`，
-`release.distribution(destination=...)` 使用单条轻量 metadata lookup，返回 exactly one
-entry 且不在分发热路径执行完整 release audit；`release.cleanup execute` 与 official
-sync 共用同一个 `.official-sync.lock`，localized cleanup 继续使用 `.localized-release.lock`。
+localized release 还写入实际 bytes/BLAKE3、完整 source mapping identity 和 destination
+index 的 `localized-distribution-manifest.json`；official download manifest 同样持久化
+canonical mapping identity 和 destination index。`release.distribution(destination=...)`
+只比较 persisted identity、查目标索引并校验单文件，返回 exactly one entry 且不在分发
+热路径执行完整 release audit；`release.cleanup execute` 与 official sync 共用同一个
+`.official-sync.lock`，localized cleanup 继续使用 `.localized-release.lock`。legacy CAS
+manifest 首次 cleanup 时按 output-root scope、release generation 和稳定 source identity
+迁移；已有 basename ledger 的部分 cleanup 保持兼容 key，完成后同名新 generation 不再复用。

 ---

@@ -389,6 +389,17 @@ impl SqliteRefCounter {
        transaction.commit().await?;
        Ok(true)
    }
+
+    /// Returns whether the durable ledger contains any row for an ownership.
+    pub async fn has_release_ownership(&self, ownership_id: &str) -> Result<bool> {
+        let exists: i64 = sqlx::query_scalar(
+            "SELECT EXISTS(SELECT 1 FROM cas_release_references WHERE release_id = ?1)",
+        )
+        .bind(ownership_id)
+        .fetch_one(&self.pool)
+        .await?;
+        Ok(exists != 0)
+    }
 }

 #[cfg(test)]
@@ -176,6 +176,12 @@ impl FileSystemCasRepository {
            .await
    }

+    /// Returns whether the durable release ownership ledger has any row.
+    pub async fn has_release_ownership(&self, ownership_id: &str) -> Result<bool> {
+        let _lock = self.acquire_operation_lock().await?;
+        self.ref_counter.has_release_ownership(ownership_id).await
+    }
+
    /// 获取存储统计信息。
    pub async fn stats(&self) -> Result<StorageStats> {
        let _lock = self.acquire_operation_lock().await?;
@@ -243,7 +243,7 @@ trusted 和 release/TextUnit/provider/run provenance；`translation.tasks` 优
 12. 将 staging rename 为 `<output>/versions/<id>`，再原子替换 `<output>/current` symlink 指向该 versioned 目录。
 13. 发布完成后先对比上一完整 release 和当前 release 的 `official-download-manifest.json`，写出 `official-resource-changes.json` 和 `crowdin-translation-handoff.json`。同一 destination 只有 size 或 BLAKE3 变化才算 modified；新增+变更资源进入解析/翻译 handoff，删除资源只进入差异记录。当前只预留 Crowdin 本地 handoff，不发外部 API 请求。
 14. 随后刷新 active release 下的 `official-parse-cache.json` 和 `official-textunit-index.json`，并从 Added/Modified 资源、parse cache 与 TextUnit 明细索引派生 `official-textunit-tasks.json`、`crowdin-textunit-queue.json` 和版本化的 `translation-tasks.sqlite`；up-to-date 轮询在已有有效解析缓存、TextUnit 明细索引和 TextUnit 队列时只读取摘要，不重复解析，重新同步队列时保留已有 worker 状态。历史 release 复用只允许不可变资源 payload/sidecar 硬链接；download manifest、snapshot、parse/textunit cache、queue、handoff、bootstrap、CAS reuse references 以及 `translation-tasks.sqlite`、WAL/SHM 都必须独立复制，不能共享可变 inode。
-15. 若启用 `--import-repository`，已校验 release 会被导入 CAS + `ResourceRepository`，并可经 `resource.index` 查询。历史 release 候选失效时，已有 CAS 对象会先经过完整性和元数据校验，再增加 release 引用并原子物化；当前 release 在 `official-cas-reuse-references.json` 中记录引用和首次生成后持久化的 `ownership_id`，staging/release 清理时按 `ownership_id + ordinal` 递减，旧无 identity 清单走明确 legacy key 路径，失败则回退网络并保留诊断。
+15. 若启用 `--import-repository`，已校验 release 会被导入 CAS + `ResourceRepository`，并可经 `resource.index` 查询。历史 release 候选失效时，已有 CAS 对象会先经过完整性和元数据校验，再增加 release 引用并原子物化；当前 release 在 `official-cas-reuse-references.json` 中记录引用和首次生成后持久化的 `ownership_id`，staging/release 清理时按 `ownership_id + ordinal` 递减。旧无 identity 清单按 output-root scope、稳定 source mapping identity 和 generation counter 迁移；已有 basename ledger 的部分 cleanup 继续使用兼容 key，直到该 generation 完成，后续同名 generation 不复用该 key，失败则回退网络并保留诊断。
 16. 官方同步报告默认给出 `localized_release_status=not_localized`，表示原版资源已发布、汉化资源未发布；generic manifest 驱动的 Binary/JSON/Text 以及当前支持的 UnityFS TextAsset、TypeTree string field 和 managed-reference string field patch 发布成功并通过 `localized-patch-manifest.json`、current symlink、release ID 及 ZIP 内层最终重解析校验后，`localized.status` 才返回 `localized`，表示原版和汉化两套资源都已发布。`localized.status` 分开返回 `patch_manifest_contract_status` 与 `artifact_integrity_status`；state/current/identity 存在但文件被截断或手工修改时返回 `localized.degraded`，只读检查不回滚、不删除、不修复。`translation.proofread` 只会把 workflow 标记成 `manual_proofreading` / `translation.manual_proofreading`，不会回退已发布汉化 release 的发布状态。

 维护期特殊分支：如果官方 launcher/server-info 已经指向新资源根，但 client-patch seed marker 或必需 seed catalog 仍返回 403/404 等未开放状态，`bat` 返回 `waiting_for_official_resources`，保留现有 `current`，不创建失败 staging；若本轮启用 `--auto-discover`，会在 `<output>/official-launcher-bootstrap.pending.json` 写入待处理 launcher bootstrap 证据，供后续排障和自研客户端开发使用。
@@ -104,9 +104,11 @@ source relation 正确且 manifest/artifact integrity 通过的 release 才能
 损坏、缺失、symlink/path escape 或未验证历史项不会回退到另一 channel。Rust 使用已发布
 manifest 做轻量选择，HTTP 热路径不重新执行完整 release audit；localized 必须额外满足
 `localized-distribution-manifest.json` 与 source official manifest 的 destination/URL
-集合一致，并返回实际 localized bytes/hash。返回的 `resource_root` 和 manifest entry
+集合及 deterministic source mapping identity 一致，并返回实际 localized bytes/hash。
+manifest 同时保存 localized mapping identity 和 destination index；返回的 `resource_root` 和 manifest entry
 由 Rust 决定，Go 只做 typed forwarding；传入 `destination` 时 Rust 会重新校验该文件的
-实际 bytes/BLAKE3。
+实际 bytes/BLAKE3。单条请求只比较发布时持久化的 identity、通过 destination index
+定位 entry，不重新遍历全量映射或资源文件。

 localized publish/rollback 先取得 `.localized-release.lock`，并在 output root 下记录
 `.localized-transaction.json`。current、version-state 和 version 目录的切换按日志阶段
@@ -137,7 +139,11 @@ reference、无法确认 ownership 的对象均保留；只删除重新验证后

 从 CAS 物化资源时，`official-cas-reuse-references.json` 首次创建时生成并持久化
 `ownership_id`，记录每个获取的对象引用；文件带版本字段且允许重复 object ID。
-没有 `ownership_id` 的旧清单保留 legacy ownership key，不在读取时随机迁移。
+没有 `ownership_id` 的旧清单首次 cleanup 按 output-root scope、release ID、稳定 source
+mapping identity 和 generation counter 建立 persistent legacy generation identity；
+若已存在 basename ledger，则在该 generation 完成前继续使用 basename compatibility key，
+完成后同名新 generation 使用新的 ownership。`.cas-owner-scope` 是 output-root 私有状态，
+不会复制到另一个 release 或 staging。
 孤儿 staging 或显式 release 清理必须先按
 清单减少 CAS 引用，再删除目录；cleanup execute 与官方同步共用
 `.official-sync.lock`，localized cleanup 使用 `.localized-release.lock`。CAS 对象损坏、缺失或元数据不一致时只产生诊断，
@@ -228,8 +228,8 @@ cargo run -p bat-infrastructure --example official_pull_plan -- \
 - `<output>/official-launcher-bootstrap.pending.json`：官方 launcher/server-info 已前进但 client-patch seed marker 或必需 seed catalog 尚未开放时写入的待处理 bootstrap 证据；它不代表资源已发布，也不会改变 `current`。
 - `<output>/official-bootstrap-cache.json`：`--auto-discover` 的 `GameMainConfig` 解析缓存。launcher metadata 与 remote manifest 文件列表 digest 都未变时复用缓存；任一变化时才通过官方 HTTP 按 manifest 下载必要 `resources.assets` 或旧版 game zip 到临时目录解析。
 - `<output>/official-version-state.json`：资源发布根目录的持久版本状态，包含当前已完成版本、正在拉取版本、上一个可用版本和失败版本。
- `<output>/current/official-download-manifest.json`：本地下载强校验清单，记录 URL、相对路径、size 和 BLAKE3。
- `<output>/current/official-cas-reuse-references.json`：当前 release 获取的 CAS 引用清单，首次创建时包含持久化 `ownership_id`；每个复用项占一条记录，release 清理或孤儿 staging GC 时据此按 ownership/ordinal 递减引用。
+- `<output>/current/official-download-manifest.json`：本地下载强校验清单，记录 URL、相对路径、size、BLAKE3、deterministic distribution mapping identity 和 destination index。
+- `<output>/current/official-cas-reuse-references.json`：当前 release 获取的 CAS 引用清单，首次创建时包含持久化 `ownership_id`；每个复用项占一条记录，release 清理或孤儿 staging GC 时据此按 ownership/ordinal 递减引用。旧无 identity 清单使用 output-root 的 `.cas-owner-scope` 完成 generation-aware legacy cleanup；该文件不会被复制到新 release。
 - `<output>/current/official-resource-changes.json`：当前 release 相对上一完整 release 的资源差异，记录新增、变更、删除以及解析/翻译候选计数。
 - `<output>/current/crowdin-translation-handoff.json`：为后续 Crowdin worker 预留的本地队列，只包含新增+变更资源；它不是 Crowdin API 调用结果。
 - `<output>/current/official-parse-cache.json`：官方资源发布后的派生解析缓存，记录 bundle/zip 条目解析摘要和缓存复用情况；它不是汉化产物。
@@ -178,7 +178,7 @@ SQLite `ResourceRepository`，索引不存在时返回 `ok=true` 且
 |---|---|---|---|
 | `release.status` | 已实现 | `null` | official/localized current、source relation、match、历史 release 和 manifest/artifact/distribution integrity 统一视图。 |
 | `release.list` | 已实现 | `{ "channel": "official" }` 或 `{ "channel": "localized" }`，可省略 | 对应 namespace 的历史 release 摘要，包含 stable ID、created/published、current pointer、`rollback_available`、lifecycle、`stale`/`damaged`/`referenced`/`unknown`、legacy 和诊断。 |
-| `release.distribution` | 已实现 | `{ "channel": "official", "release_id": "...", "destination": "...", "offset": 0, "limit": 1000 }`，均可省略 | Rust 选择的 verified `resource_root` 和 download manifest entries；有 `destination` 时是 single-entry lookup，响应固定 `total=1, offset=0, limit=1, entries.length=1`，只校验该实际文件的 bytes/BLAKE3；无 `destination` 时保留管理查询分页语义。localized 使用发布时生成的实际字节 metadata，不复用 official size/hash；默认 channel 为 official，选择失败返回 `available=false`，不跨 channel fallback。 |
+| `release.distribution` | 已实现 | `{ "channel": "official", "release_id": "...", "destination": "...", "offset": 0, "limit": 1000 }`，均可省略 | Rust 选择的 verified `resource_root` 和 download manifest entries；有 `destination` 时是 single-entry lookup，响应固定 `total=1, offset=0, limit=1, entries.length=1`，使用 persisted source mapping identity/destination index，只校验该实际文件的 bytes/BLAKE3，不重新执行全量映射或资源 audit；无 `destination` 时保留管理查询分页语义。localized 使用发布时生成的实际字节 metadata，不复用 official size/hash；默认 channel 为 official，选择失败返回 `available=false`，不跨 channel fallback。 |
 | `release.cleanup` | 已实现 | dry-run `{ "execute": false }`；执行 `{ "execute": true, "plan_id": "..." }` | cleanup plan、candidate/retain reasons、blocking references 和 removed paths；执行前会重新生成并比对 `plan_id`。 |

 `release.status`、`release.list` 和 `release.distribution` 只读现有 official/localized
@@ -127,14 +127,17 @@ rollback 与 cleanup 保持独立；缺少 generic manifest 的旧 localized rel

 本轮 P1 一致性修复已完成：CAS repository 的 store/get/reference/GC 使用跨进程操作锁，
 release-local CAS 引用通过持久化 `ownership_id + ordinal` ledger 幂等释放；没有
-`ownership_id` 的旧清单走明确 legacy cleanup path，不随机迁移 ownership。官方历史复用
-只对不可变文件使用 hard link，`translation-tasks.sqlite` 及 WAL/SHM 始终独立复制；
+`ownership_id` 的旧清单按持久化 output-root scope、source mapping 和 generation identity
+迁移，已有 basename ledger 的部分 cleanup 保持 legacy compatibility key，完成后同名新
+generation 使用新的 ownership，不随机迁移已开始的 cleanup。官方历史复用只对不可变文件使用 hard link，`translation-tasks.sqlite` 及 WAL/SHM 始终独立复制；
 localized output 使用单写者锁和事务日志恢复 publish/rollback，publish 只有最终
 `verified` phase 才能 roll-forward，并在发布时写入实际 localized bytes/BLAKE3 的
-distribution manifest。`release.distribution(destination=...)` 是单条 lookup，返回
-exactly one entry；分发读取只使用轻量发布 metadata，保留 path ownership、symlink 和
-文件完整性检查；`release.cleanup execute` 与 official sync 共用 `.official-sync.lock`。
-P2 尚未由本轮处理：ResourceRepository
+distribution manifest。official/localized distribution manifest 还持久化 deterministic
+source/localized mapping identity 和 destination index；`release.distribution(destination=...)`
+是单条 lookup，返回 exactly one entry，source identity mismatch 会阻断 localized
+distribution，完整 identity 校验只在 publish/status/audit 路径执行；分发读取保留 path
+ownership、symlink 和文件完整性检查；`release.cleanup execute` 与 official sync 共用
+`.official-sync.lock`。本轮已关闭两个 Release/CAS P2。其他 P2 尚未由本轮处理：ResourceRepository
 更完整的查询/权限/损坏恢复、模糊 TM、bat.sock peer credential/perms、FFI 生命周期、
 资源大小/限额与更强的持久化 fsync 语义仍按后续专项推进。

@@ -45,6 +45,15 @@ impl FileSystemCasRepository {
            .map_err(Self::map_error)
    }

+    /// Returns whether the durable release ownership ledger has any row.
+    pub async fn has_release_ownership(&self, ownership_id: &str) -> bat_core::Result<bool> {
+        self.engine()
+            .await?
+            .has_release_ownership(ownership_id)
+            .await
+            .map_err(Self::map_error)
+    }
+
    async fn engine(&self) -> bat_core::Result<&engine_repository::FileSystemCasRepository> {
        self.inner
            .get_or_try_init(|| async {
@@ -80,15 +80,15 @@ pub use official_changes::{
    OFFICIAL_RESOURCE_CHANGES_VERSION,
 };
 pub use official_download::{
-    read_cas_reuse_reference_manifest_at, read_download_manifest_at, release_cas_reuse_references,
-    DownloadError, OfficialCasReuseReferenceManifest, OfficialDownloadManifest,
-    OfficialDownloadManifestEntry, OfficialLocalManifestAuditItem,
-    OfficialLocalManifestAuditReport, OfficialLocalManifestAuditStatus,
-    OfficialLocalVerificationReport, OfficialResourceHashAlgorithm,
-    OfficialResourceHashVerification, OfficialResourcePullItem, OfficialResourcePullProgress,
-    OfficialResourcePullProgressKind, OfficialResourcePullReport, OfficialResourcePullService,
-    OfficialResourcePullStatus, OfficialResourceReuseWarning, OfficialResourceVerification,
-    OFFICIAL_CAS_REUSE_REFERENCES_FILE,
+    official_distribution_mapping_identity, read_cas_reuse_reference_manifest_at,
+    read_download_manifest_at, release_cas_reuse_references, DownloadError,
+    OfficialCasReuseReferenceManifest, OfficialDownloadManifest, OfficialDownloadManifestEntry,
+    OfficialLocalManifestAuditItem, OfficialLocalManifestAuditReport,
+    OfficialLocalManifestAuditStatus, OfficialLocalVerificationReport,
+    OfficialResourceHashAlgorithm, OfficialResourceHashVerification, OfficialResourcePullItem,
+    OfficialResourcePullProgress, OfficialResourcePullProgressKind, OfficialResourcePullReport,
+    OfficialResourcePullService, OfficialResourcePullStatus, OfficialResourceReuseWarning,
+    OfficialResourceVerification, OFFICIAL_CAS_REUSE_REFERENCES_FILE,
 };
 pub use official_game_main_config::OfficialGameMainConfigBootstrapService;
 pub use official_launcher::{
@@ -400,6 +400,15 @@ pub struct LocalizedDistributionManifest {
    pub official_release_id: String,
    /// Localized release identity.
    pub localized_release_id: String,
+    /// Deterministic identity of the complete source official mapping.
+    #[serde(default)]
+    pub source_mapping_identity: String,
+    /// Deterministic identity of this localized mapping and its source.
+    #[serde(default)]
+    pub localized_mapping_identity: String,
+    /// Persisted destination-to-entry index for single-entry lookup.
+    #[serde(default)]
+    pub destination_index: BTreeMap<String, usize>,
    /// Actual metadata for every official manifest entry.
    pub entries: Vec<LocalizedDistributionEntry>,
 }
@@ -841,7 +850,7 @@ impl LocalizedPatchService {
            )
        })?;
        verify_localized_release_files(&version_path, &manifest)?;
-        verify_localized_distribution_manifest_at(&version_path, &current_release_id)?;
+        verify_localized_distribution_manifest_at(None, &version_path, &current_release_id)?;
        if manifest.localized_release_id != current_release_id {
            return Err(anyhow::anyhow!(
                "manifest release={} 与当前状态 release={} 不一致",
@@ -900,6 +909,7 @@ impl LocalizedPatchService {
            );
            verify_localized_release_files(&previous_path, restored_manifest.as_ref().unwrap())?;
            verify_localized_distribution_manifest_at(
+                None,
                &previous_path,
                restored_release_id.as_deref().unwrap_or_default(),
            )?;
@@ -2178,7 +2188,11 @@ fn verify_published_localized_release(
        &manifest,
        unzip_command,
    )?;
-    verify_localized_distribution_manifest_at(version_path, &manifest.localized_release_id)?;
+    verify_localized_distribution_manifest_at(
+        Some(official_release_root),
+        version_path,
+        &manifest.localized_release_id,
+    )?;
    integrity.current_points_to_release = current_points_to_version(current_path, version_path)?;
    if !integrity.current_points_to_release {
        return Err(anyhow::anyhow!(
@@ -2218,6 +2232,8 @@ fn build_localized_distribution_manifest(
    else {
        return Ok(None);
    };
+    let source_mapping_identity =
+        crate::official_download::official_distribution_mapping_identity(&official_manifest);
    let mut entries = Vec::with_capacity(official_manifest.entries.len());
    for entry in official_manifest.entries.values() {
        let path = staging_root.join(&entry.destination);
@@ -2232,10 +2248,25 @@ fn build_localized_distribution_manifest(
            blake3: blake3::hash(&bytes).to_hex().to_string(),
        });
    }
+    let destination_index = entries
+        .iter()
+        .enumerate()
+        .map(|(index, entry)| (entry.destination.clone(), index))
+        .collect::<BTreeMap<_, _>>();
+    if destination_index.len() != entries.len() {
+        return Err(anyhow::anyhow!(
+            "official distribution manifest 存在重复 destination"
+        ));
+    }
+    let localized_mapping_identity =
+        localized_distribution_mapping_identity(&source_mapping_identity, &entries);
    Ok(Some(LocalizedDistributionManifest {
        version: 1,
        official_release_id: config.release_id.clone(),
        localized_release_id: config.published_release_id().to_string(),
+        source_mapping_identity,
+        localized_mapping_identity,
+        destination_index,
        entries,
    }))
 }
@@ -2267,6 +2298,7 @@ fn verify_localized_release_files(
 }

 fn verify_localized_distribution_manifest_at(
+    official_release_root: Option<&Path>,
    version_path: &Path,
    localized_release_id: &str,
 ) -> anyhow::Result<()> {
@@ -2286,6 +2318,65 @@ fn verify_localized_distribution_manifest_at(
            manifest.localized_release_id
        ));
    }
+    let source_mapping_identity = if let Some(official_release_root) = official_release_root {
+        let official_manifest =
+            crate::official_download::read_download_manifest_at(official_release_root)
+                .map_err(anyhow::Error::msg)?
+                .ok_or_else(|| {
+                    anyhow::anyhow!("localized distribution 缺少 source official manifest")
+                })?;
+        let source_mapping_identity =
+            crate::official_download::official_distribution_mapping_identity(&official_manifest);
+        let expected_destination_index =
+            crate::official_download::official_distribution_destination_index(&official_manifest)
+                .map_err(anyhow::Error::msg)?;
+        if manifest.source_mapping_identity != source_mapping_identity
+            || official_manifest.distribution_mapping_identity.as_deref()
+                != Some(source_mapping_identity.as_str())
+            || official_manifest.destination_index != expected_destination_index
+        {
+            return Err(anyhow::anyhow!(
+                "localized distribution source mapping/index 不一致：expected={} actual={} official={:?}",
+                source_mapping_identity,
+                manifest.source_mapping_identity,
+                official_manifest.distribution_mapping_identity
+            ));
+        }
+        if !localized_distribution_entries_match_official(&manifest, &official_manifest) {
+            return Err(anyhow::anyhow!(
+                "localized distribution manifest 与 source official mapping 不一致"
+            ));
+        }
+        source_mapping_identity
+    } else {
+        manifest.source_mapping_identity.clone()
+    };
+    if !manifest.source_mapping_identity.is_empty() {
+        let localized_mapping_identity =
+            localized_distribution_mapping_identity(&source_mapping_identity, &manifest.entries);
+        if manifest.localized_mapping_identity != localized_mapping_identity {
+            return Err(anyhow::anyhow!(
+                "localized distribution mapping identity 不一致：expected={} actual={}",
+                localized_mapping_identity,
+                manifest.localized_mapping_identity
+            ));
+        }
+        if manifest.destination_index.len() != manifest.entries.len()
+            || manifest
+                .destination_index
+                .iter()
+                .any(|(destination, index)| {
+                    manifest
+                        .entries
+                        .get(*index)
+                        .is_none_or(|entry| entry.destination != *destination)
+                })
+        {
+            return Err(anyhow::anyhow!(
+                "localized distribution destination index 不一致"
+            ));
+        }
+    }
    let mut destinations = BTreeSet::new();
    for entry in &manifest.entries {
        if !destinations.insert(entry.destination.as_str()) {
@@ -2314,6 +2405,69 @@ fn verify_localized_distribution_manifest_at(
    Ok(())
 }

+pub(crate) fn verify_localized_distribution_manifest_for_status(
+    official_release_root: &Path,
+    version_path: &Path,
+    localized_release_id: &str,
+) -> anyhow::Result<()> {
+    verify_localized_distribution_manifest_at(
+        Some(official_release_root),
+        version_path,
+        localized_release_id,
+    )
+}
+
+fn localized_distribution_entries_match_official(
+    localized: &LocalizedDistributionManifest,
+    official: &crate::official_download::OfficialDownloadManifest,
+) -> bool {
+    if localized.entries.len() != official.entries.len() {
+        return false;
+    }
+    let mut localized_by_destination = BTreeMap::new();
+    for entry in &localized.entries {
+        if localized_by_destination
+            .insert(entry.destination.as_str(), entry.url.as_str())
+            .is_some()
+        {
+            return false;
+        }
+    }
+    official.entries.values().all(|entry| {
+        localized_by_destination.get(entry.destination.as_str()) == Some(&entry.url.as_str())
+    })
+}
+
+fn localized_distribution_mapping_identity(
+    source_mapping_identity: &str,
+    entries: &[LocalizedDistributionEntry],
+) -> String {
+    let mut ordered = entries.iter().collect::<Vec<_>>();
+    ordered.sort_by(|left, right| {
+        left.destination
+            .cmp(&right.destination)
+            .then_with(|| left.url.cmp(&right.url))
+            .then_with(|| left.bytes.cmp(&right.bytes))
+            .then_with(|| left.blake3.cmp(&right.blake3))
+    });
+    let mut hasher = blake3::Hasher::new();
+    hasher.update(b"localized-distribution-mapping-v1");
+    update_distribution_identity_string(&mut hasher, source_mapping_identity);
+    hasher.update(&(ordered.len() as u64).to_be_bytes());
+    for entry in ordered {
+        update_distribution_identity_string(&mut hasher, &entry.destination);
+        update_distribution_identity_string(&mut hasher, &entry.url);
+        hasher.update(&entry.bytes.to_be_bytes());
+        update_distribution_identity_string(&mut hasher, &entry.blake3);
+    }
+    format!("ldm-v1-{}", hasher.finalize().to_hex())
+}
+
+fn update_distribution_identity_string(hasher: &mut blake3::Hasher, value: &str) {
+    hasher.update(&(value.len() as u64).to_be_bytes());
+    hasher.update(value.as_bytes());
+}
+
 fn write_localized_transaction(
    localized_output_root: &Path,
    transaction: &LocalizedReleaseTransaction,
@@ -3996,42 +4150,50 @@ mod tests {
        fs::write(target.join("data.json"), json_target).unwrap();
        fs::write(official.join("text.txt"), text_source).unwrap();
        fs::write(target.join("text.txt"), text_target).unwrap();
+        let mut official_manifest = crate::OfficialDownloadManifest {
+            version: 1,
+            entries: [
+                (
+                    "https://example.invalid/data.bin".to_string(),
+                    "data.bin",
+                    binary_source.as_slice(),
+                ),
+                (
+                    "https://example.invalid/data.json".to_string(),
+                    "data.json",
+                    json_source.as_slice(),
+                ),
+                (
+                    "https://example.invalid/text.txt".to_string(),
+                    "text.txt",
+                    text_source.as_bytes(),
+                ),
+            ]
+            .into_iter()
+            .map(|(url, destination, bytes)| {
+                (
+                    url.clone(),
+                    crate::OfficialDownloadManifestEntry {
+                        url,
+                        destination: destination.to_string(),
+                        bytes: bytes.len() as u64,
+                        blake3: blake3::hash(bytes).to_hex().to_string(),
+                    },
+                )
+            })
+            .collect(),
+            destination_index: BTreeMap::new(),
+            distribution_mapping_identity: None,
+        };
+        official_manifest.distribution_mapping_identity = Some(
+            crate::official_distribution_mapping_identity(&official_manifest),
+        );
+        official_manifest.destination_index =
+            crate::official_download::official_distribution_destination_index(&official_manifest)
+                .unwrap();
        fs::write(
            official.join("official-download-manifest.json"),
-            serde_json::to_vec(&crate::OfficialDownloadManifest {
-                version: 1,
-                entries: [
-                    (
-                        "https://example.invalid/data.bin".to_string(),
-                        "data.bin",
-                        binary_source.as_slice(),
-                    ),
-                    (
-                        "https://example.invalid/data.json".to_string(),
-                        "data.json",
-                        json_source.as_slice(),
-                    ),
-                    (
-                        "https://example.invalid/text.txt".to_string(),
-                        "text.txt",
-                        text_source.as_bytes(),
-                    ),
-                ]
-                .into_iter()
-                .map(|(url, destination, bytes)| {
-                    (
-                        url.clone(),
-                        crate::OfficialDownloadManifestEntry {
-                            url,
-                            destination: destination.to_string(),
-                            bytes: bytes.len() as u64,
-                            blake3: blake3::hash(bytes).to_hex().to_string(),
-                        },
-                    )
-                })
-                .collect(),
-            })
-            .unwrap(),
+            serde_json::to_vec(&official_manifest).unwrap(),
        )
        .unwrap();

@@ -4142,6 +4304,12 @@ mod tests {
        )
        .unwrap();
        assert_eq!(distribution.entries.len(), 3);
+        assert_eq!(
+            distribution.source_mapping_identity,
+            crate::official_distribution_mapping_identity(&official_manifest)
+        );
+        assert_eq!(distribution.destination_index.len(), 3);
+        assert!(!distribution.localized_mapping_identity.is_empty());
        for (path, expected) in [
            ("data.bin", binary_target.as_slice()),
            ("data.json", json_target.as_slice()),
@@ -4155,6 +4323,24 @@ mod tests {
            assert_eq!(entry.bytes, expected.len() as u64);
            assert_eq!(entry.blake3, blake3::hash(expected).to_hex().to_string());
        }
+
+        let mut tampered_official = official_manifest;
+        tampered_official
+            .entries
+            .get_mut("https://example.invalid/data.json")
+            .unwrap()
+            .bytes += 1;
+        fs::write(
+            official.join("official-download-manifest.json"),
+            serde_json::to_vec(&tampered_official).unwrap(),
+        )
+        .unwrap();
+        assert!(verify_localized_distribution_manifest_for_status(
+            &official,
+            &report.version_path,
+            "localized-v1"
+        )
+        .is_err());
    }

    #[cfg(unix)]
@@ -20,7 +20,7 @@ use bat_adapters::official::{
 };
 use bat_core::repositories::CasRepository;
 use serde::{Deserialize, Serialize};
-use std::collections::{BTreeMap, HashMap, HashSet};
+use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
 use std::fs::{self, File};
 use std::io::Read;
 use std::path::{Path, PathBuf};
@@ -75,9 +75,78 @@ const OFFICIAL_CAS_REUSE_REFERENCES_VERSION: u32 = 1;
 const DOWNLOAD_MANIFEST_VERSION: u32 = 1;
 const DOWNLOAD_QUARANTINE_VERSION: u32 = 1;
 const DEFAULT_RETRY_ATTEMPTS: usize = 3;
+const CAS_OWNER_SCOPE_FILE: &str = ".cas-owner-scope";
+const CAS_OWNER_SCOPE_VERSION: u32 = 1;
 static CAS_OWNERSHIP_SEQUENCE: AtomicU64 = AtomicU64::new(1);

-fn new_cas_ownership_id(output_root: &Path) -> String {
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct CasOwnerScopeState {
+    version: u32,
+    scope_id: String,
+    #[serde(default)]
+    next_generation: BTreeMap<String, u64>,
+    #[serde(default)]
+    completed_legacy_release_ids: BTreeSet<String>,
+    #[serde(default)]
+    legacy_basename_compatibility: BTreeSet<String>,
+}
+
+fn ownership_scope_root(release_root: &Path) -> PathBuf {
+    let Some(parent) = release_root.parent() else {
+        return release_root.to_path_buf();
+    };
+    match parent.file_name().and_then(|name| name.to_str()) {
+        Some("versions" | ".staging") => parent.parent().unwrap_or(parent).to_path_buf(),
+        _ => release_root.to_path_buf(),
+    }
+}
+
+fn load_owner_scope_state(release_root: &Path) -> Result<CasOwnerScopeState, String> {
+    let scope_root = ownership_scope_root(release_root);
+    ensure_safe_directory_path(&scope_root, "CAS ownership scope 根目录")?;
+    fs::create_dir_all(&scope_root)
+        .map_err(|error| format!("创建 CAS ownership scope 根目录失败：{error}"))?;
+    let path = scope_root.join(CAS_OWNER_SCOPE_FILE);
+    let Some(bytes) = read_file_no_symlink(&path, "CAS ownership scope")? else {
+        let sequence = CAS_OWNERSHIP_SEQUENCE.fetch_add(1, Ordering::Relaxed);
+        let now = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .map(|duration| duration.as_nanos())
+            .unwrap_or_default();
+        let material = format!("{}:{}:{}", scope_root.display(), std::process::id(), now);
+        let state = CasOwnerScopeState {
+            version: CAS_OWNER_SCOPE_VERSION,
+            scope_id: format!(
+                "cas-scope-{}",
+                blake3::hash(format!("{material}:{sequence}").as_bytes()).to_hex()
+            ),
+            next_generation: BTreeMap::new(),
+            completed_legacy_release_ids: BTreeSet::new(),
+            legacy_basename_compatibility: BTreeSet::new(),
+        };
+        write_owner_scope_state(&scope_root, &state)?;
+        return Ok(state);
+    };
+    let state: CasOwnerScopeState = serde_json::from_slice(&bytes)
+        .map_err(|error| format!("解析 CAS ownership scope 失败 {}：{error}", path.display()))?;
+    if state.version != CAS_OWNER_SCOPE_VERSION || state.scope_id.is_empty() {
+        return Err(format!("不支持的 CAS ownership scope：{}", path.display()));
+    }
+    Ok(state)
+}
+
+fn write_owner_scope_state(release_root: &Path, state: &CasOwnerScopeState) -> Result<(), String> {
+    let scope_root = ownership_scope_root(release_root);
+    let path = scope_root.join(CAS_OWNER_SCOPE_FILE);
+    ensure_path_within_root(&scope_root, &path)?;
+    ensure_safe_file_target(&scope_root, &path, "CAS ownership scope")?;
+    let bytes = serde_json::to_vec_pretty(state)
+        .map_err(|error| format!("序列化 CAS ownership scope 失败：{error}"))?;
+    write_file_atomic(&path, &bytes, STATE_FILE_MODE, "CAS ownership scope")
+}
+
+fn new_cas_ownership_id(output_root: &Path) -> Result<String, String> {
+    let scope = load_owner_scope_state(output_root)?;
    let sequence = CAS_OWNERSHIP_SEQUENCE.fetch_add(1, Ordering::Relaxed);
    let now = SystemTime::now()
        .duration_since(UNIX_EPOCH)
@@ -85,12 +154,65 @@ fn new_cas_ownership_id(output_root: &Path) -> String {
        .unwrap_or_default();
    let material = format!(
        "{}:{}:{}:{}",
-        output_root.display(),
+        scope.scope_id,
        std::process::id(),
        now,
-        sequence
+        sequence,
    );
-    format!("cas-owner-{}", blake3::hash(material.as_bytes()).to_hex())
+    Ok(format!(
+        "cas-owner-{}",
+        blake3::hash(material.as_bytes()).to_hex()
+    ))
+}
+
+/// Returns the deterministic identity of an official distribution mapping.
+///
+/// The framing is deliberately independent from JSON serialization and map
+/// iteration details. Destination is the primary sort key; the remaining
+/// fields make duplicate destinations deterministic as well.
+pub fn official_distribution_mapping_identity(manifest: &OfficialDownloadManifest) -> String {
+    let mut entries = manifest.entries.values().collect::<Vec<_>>();
+    entries.sort_by(|left, right| {
+        left.destination
+            .cmp(&right.destination)
+            .then_with(|| left.url.cmp(&right.url))
+            .then_with(|| left.bytes.cmp(&right.bytes))
+            .then_with(|| left.blake3.cmp(&right.blake3))
+    });
+
+    let mut hasher = blake3::Hasher::new();
+    hasher.update(b"official-distribution-mapping-v1");
+    hasher.update(&(entries.len() as u64).to_be_bytes());
+    for entry in entries {
+        update_identity_string(&mut hasher, &entry.destination);
+        update_identity_string(&mut hasher, &entry.url);
+        hasher.update(&entry.bytes.to_be_bytes());
+        update_identity_string(&mut hasher, &entry.blake3);
+    }
+    format!("odm-v1-{}", hasher.finalize().to_hex())
+}
+
+pub(crate) fn official_distribution_destination_index(
+    manifest: &OfficialDownloadManifest,
+) -> Result<BTreeMap<String, String>, String> {
+    let mut index = BTreeMap::new();
+    for entry in manifest.entries.values() {
+        if index
+            .insert(entry.destination.clone(), entry.url.clone())
+            .is_some()
+        {
+            return Err(format!(
+                "official distribution manifest 存在重复 destination：{}",
+                entry.destination
+            ));
+        }
+    }
+    Ok(index)
+}
+
+fn update_identity_string(hasher: &mut blake3::Hasher, value: &str) {
+    hasher.update(&(value.len() as u64).to_be_bytes());
+    hasher.update(value.as_bytes());
 }

 /// Outcome for one official resource pull item.
@@ -416,19 +538,36 @@ pub fn read_cas_reuse_reference_manifest_at(
    Ok(Some(manifest))
 }

-/// Decrements and removes CAS references recorded for a release.
-///
-/// Each decrement is committed together with a durable `(ownership, ordinal)`
-/// record in CAS metadata. The release-local manifest remains a resumable
-/// progress cursor, so a crash before its rewrite cannot decrement the same
-/// ownership twice. Legacy manifests without `ownership_id` intentionally use
-/// the historical release-basename key and are never assigned a new identity
-/// during cleanup.
-pub fn release_cas_reuse_references(release_root: &Path, cas_root: &Path) -> Result<usize, String> {
-    let Some(mut manifest) = read_cas_reuse_reference_manifest_at(release_root)? else {
-        return Ok(0);
+fn cas_has_ownership(cas_root: &Path, ownership_id: &str) -> Result<bool, String> {
+    let cas_root = cas_root.to_path_buf();
+    let ownership_id = ownership_id.to_string();
+    let runtime = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .map_err(|error| format!("创建 CAS ownership 查询 runtime 失败：{error}"))?;
+    runtime.block_on(async move {
+        let cas = crate::FileSystemCasRepository::new(cas_root);
+        cas.has_release_ownership(&ownership_id)
+            .await
+            .map_err(|error| format!("查询 CAS ownership ledger 失败：{error}"))
+    })
+}
+
+fn legacy_source_mapping_identity(release_root: &Path) -> Result<String, String> {
+    let Some(manifest) = read_download_manifest_at(release_root)? else {
+        return Ok("missing-official-manifest".to_string());
    };
-    let legacy_release_id = release_root
+    Ok(manifest
+        .distribution_mapping_identity
+        .clone()
+        .unwrap_or_else(|| official_distribution_mapping_identity(&manifest)))
+}
+
+fn resolve_legacy_ownership(
+    release_root: &Path,
+    cas_root: &Path,
+) -> Result<(String, bool), String> {
+    let release_id = release_root
        .file_name()
        .and_then(|name| name.to_str())
        .filter(|name| !name.is_empty() && *name != "." && *name != "..")
@@ -439,12 +578,94 @@ pub fn release_cas_reuse_references(release_root: &Path, cas_root: &Path) -> Res
            )
        })?
        .to_string();
-    let ownership_id = manifest.ownership_id.clone().unwrap_or(legacy_release_id);
+    let mut scope = load_owner_scope_state(release_root)?;
+    if scope.completed_legacy_release_ids.contains(&release_id) {
+        let source_identity = legacy_source_mapping_identity(release_root)?;
+        let generation_key = format!("{release_id}\0{source_identity}");
+        let generation = scope.next_generation.entry(generation_key).or_insert(0);
+        let current_generation = *generation;
+        *generation = generation.saturating_add(1);
+        write_owner_scope_state(release_root, &scope)?;
+        let owner_material = format!(
+            "cas-legacy-owner-v1:{}:{}:{}:{}",
+            scope.scope_id, release_id, source_identity, current_generation
+        );
+        return Ok((
+            format!(
+                "cas-legacy-owner-v1-{}",
+                blake3::hash(owner_material.as_bytes()).to_hex()
+            ),
+            false,
+        ));
+    }
+    if scope.legacy_basename_compatibility.contains(&release_id) {
+        return Ok((release_id, true));
+    }
+    if cas_has_ownership(cas_root, &release_id)? {
+        scope
+            .legacy_basename_compatibility
+            .insert(release_id.clone());
+        write_owner_scope_state(release_root, &scope)?;
+        return Ok((release_id, true));
+    }
+
+    let source_identity = legacy_source_mapping_identity(release_root)?;
+    let generation_key = format!("{release_id}\0{source_identity}");
+    let generation = scope.next_generation.entry(generation_key).or_insert(0);
+    let current_generation = *generation;
+    *generation = generation.saturating_add(1);
+    write_owner_scope_state(release_root, &scope)?;
+    let owner_material = format!(
+        "cas-legacy-owner-v1:{}:{}:{}:{}",
+        scope.scope_id, release_id, source_identity, current_generation
+    );
+    Ok((
+        format!(
+            "cas-legacy-owner-v1-{}",
+            blake3::hash(owner_material.as_bytes()).to_hex()
+        ),
+        false,
+    ))
+}
+
+fn mark_legacy_ownership_completed(release_root: &Path, release_id: &str) -> Result<(), String> {
+    let mut scope = load_owner_scope_state(release_root)?;
+    scope
+        .completed_legacy_release_ids
+        .insert(release_id.to_string());
+    write_owner_scope_state(release_root, &scope)
+}
+
+/// Decrements and removes CAS references recorded for a release.
+///
+/// Each decrement is committed together with a durable `(ownership, ordinal)`
+/// record in CAS metadata. The release-local manifest remains a resumable
+/// progress cursor, so a crash before its rewrite cannot decrement the same
+/// ownership twice. Legacy manifests without `ownership_id` use the historical
+/// release-basename key only when an existing ledger requires compatibility;
+/// otherwise cleanup persists a generation-aware identity before decrementing.
+pub fn release_cas_reuse_references(release_root: &Path, cas_root: &Path) -> Result<usize, String> {
+    let Some(mut manifest) = read_cas_reuse_reference_manifest_at(release_root)? else {
+        return Ok(0);
+    };
    let objects_root = cas_root.join("objects");
    let metadata_path = cas_root.join("metadata.sqlite");
    require_existing_directory(cas_root, "CAS 根目录")?;
    require_existing_directory(&objects_root, "CAS 对象目录")?;
    require_existing_file(cas_root, &metadata_path, "CAS 元数据库")?;
+    let (ownership_id, legacy_basename_compatibility) = match manifest.ownership_id.clone() {
+        Some(ownership_id) => (ownership_id, false),
+        None => resolve_legacy_ownership(release_root, cas_root)?,
+    };
+    if manifest.ownership_id.is_none() && !legacy_basename_compatibility {
+        manifest.ownership_id = Some(ownership_id.clone());
+        write_cas_reuse_reference_manifest(release_root, &manifest)?;
+    }
+    let legacy_release_id = release_root
+        .file_name()
+        .and_then(|name| name.to_str())
+        .unwrap_or_default()
+        .to_string();
    let mut released = 0usize;
    while let Some(object_id) = manifest.object_ids.pop() {
        let ordinal = manifest.object_ids.len() as u64;
@@ -467,6 +688,9 @@ pub fn release_cas_reuse_references(release_root: &Path, cas_root: &Path) -> Res
        }
    }
    let path = release_root.join(OFFICIAL_CAS_REUSE_REFERENCES_FILE);
+    if legacy_basename_compatibility {
+        mark_legacy_ownership_completed(release_root, &legacy_release_id)?;
+    }
    match fs::symlink_metadata(&path) {
        Ok(metadata) if metadata.file_type().is_symlink() => {
            return Err(format!(
@@ -889,7 +1113,7 @@ impl OfficialResourcePullService {
            Some(manifest) => manifest,
            None => OfficialCasReuseReferenceManifest {
                version: OFFICIAL_CAS_REUSE_REFERENCES_VERSION,
-                ownership_id: Some(new_cas_ownership_id(&self.output_root)),
+                ownership_id: Some(new_cas_ownership_id(&self.output_root)?),
                object_ids: Vec::new(),
            },
        };
@@ -2139,7 +2363,12 @@ impl OfficialResourcePullService {
        }
        ensure_safe_file_target(&self.output_root, &path, "下载 manifest")?;

-        let bytes = serde_json::to_vec_pretty(manifest)
+        let mut persisted_manifest = manifest.clone();
+        persisted_manifest.distribution_mapping_identity =
+            Some(official_distribution_mapping_identity(&persisted_manifest));
+        persisted_manifest.destination_index =
+            official_distribution_destination_index(&persisted_manifest)?;
+        let bytes = serde_json::to_vec_pretty(&persisted_manifest)
            .map_err(|error| format!("序列化下载 manifest 失败 {}：{error}", path.display()))?;
        write_file_atomic(&path, &bytes, STATE_FILE_MODE, "下载 manifest")?;

@@ -2625,6 +2854,12 @@ pub struct OfficialDownloadManifest {
    /// 按 URL 为键的资源条目。
    #[serde(default)]
    pub entries: BTreeMap<String, OfficialDownloadManifestEntry>,
+    /// Persisted destination-to-URL index for single-entry distribution lookup.
+    #[serde(default)]
+    pub destination_index: BTreeMap<String, String>,
+    /// Deterministic identity of the complete distribution mapping.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub distribution_mapping_identity: Option<String>,
 }

 impl Default for OfficialDownloadManifest {
@@ -2632,6 +2867,8 @@ impl Default for OfficialDownloadManifest {
        Self {
            version: DOWNLOAD_MANIFEST_VERSION,
            entries: BTreeMap::new(),
+            destination_index: BTreeMap::new(),
+            distribution_mapping_identity: None,
        }
    }
 }
@@ -3491,6 +3728,75 @@ exit 22
        }
    }

+    #[test]
+    fn official_distribution_mapping_identity_is_deterministic_and_sensitive() {
+        let entries = [
+            ("https://example.invalid/z", "z", b"z-bytes".as_slice()),
+            ("https://example.invalid/a", "a", b"a-bytes".as_slice()),
+        ];
+        let mut first = OfficialDownloadManifest::default();
+        for (url, destination, bytes) in entries {
+            first.entries.insert(
+                url.to_string(),
+                OfficialDownloadManifestEntry {
+                    url: url.to_string(),
+                    destination: destination.to_string(),
+                    bytes: bytes.len() as u64,
+                    blake3: blake3::hash(bytes).to_hex().to_string(),
+                },
+            );
+        }
+        let mut second = OfficialDownloadManifest::default();
+        for (url, destination, bytes) in entries.into_iter().rev() {
+            second.entries.insert(
+                url.to_string(),
+                OfficialDownloadManifestEntry {
+                    url: url.to_string(),
+                    destination: destination.to_string(),
+                    bytes: bytes.len() as u64,
+                    blake3: blake3::hash(bytes).to_hex().to_string(),
+                },
+            );
+        }
+        let identity = official_distribution_mapping_identity(&first);
+        assert_eq!(identity, official_distribution_mapping_identity(&second));
+        let mutators: &[fn(&mut OfficialDownloadManifest)] = &[
+            |manifest: &mut OfficialDownloadManifest| {
+                manifest
+                    .entries
+                    .get_mut("https://example.invalid/a")
+                    .unwrap()
+                    .destination = "changed".to_string()
+            },
+            |manifest: &mut OfficialDownloadManifest| {
+                manifest
+                    .entries
+                    .get_mut("https://example.invalid/a")
+                    .unwrap()
+                    .url = "https://example.invalid/changed".to_string()
+            },
+            |manifest: &mut OfficialDownloadManifest| {
+                manifest
+                    .entries
+                    .get_mut("https://example.invalid/a")
+                    .unwrap()
+                    .bytes += 1
+            },
+            |manifest: &mut OfficialDownloadManifest| {
+                manifest
+                    .entries
+                    .get_mut("https://example.invalid/a")
+                    .unwrap()
+                    .blake3 = "0".repeat(64)
+            },
+        ];
+        for mutate in mutators {
+            let mut changed = first.clone();
+            mutate(&mut changed);
+            assert_ne!(identity, official_distribution_mapping_identity(&changed));
+        }
+    }
+
    fn one_file_zip(name: &[u8], data: &[u8]) -> Vec<u8> {
        let mut bytes = Vec::new();
        bytes.extend_from_slice(&0x0403_4b50u32.to_le_bytes());
@@ -3573,6 +3879,11 @@ exit 22
        let manifest = service.read_download_manifest().unwrap();
        assert_eq!(manifest.version, DOWNLOAD_MANIFEST_VERSION);
        assert_eq!(manifest.entries.len(), report.items.len());
+        assert_eq!(
+            manifest.distribution_mapping_identity.as_deref(),
+            Some(official_distribution_mapping_identity(&manifest).as_str())
+        );
+        assert_eq!(manifest.destination_index.len(), manifest.entries.len());
        for item in &report.items {
            let entry = manifest.entries.get(&item.url).unwrap();
            assert_eq!(entry.url, item.url);
@@ -4801,6 +5112,132 @@ exit 22
        assert_eq!(cas_reference_count(&cas_root, &object_id), 1);
    }

+    #[test]
+    fn legacy_manifest_migration_persists_identity_before_retry() {
+        let temp = TempDir::new().unwrap();
+        let cas_root = temp.path().join("cas");
+        let release_root = temp.path().join("versions/release-migrate");
+        fs::create_dir_all(&release_root).unwrap();
+        let object_id = store_cas_object(&cas_root, b"migrated legacy object");
+        let missing_object = "f".repeat(64);
+        fs::write(
+            release_root.join(OFFICIAL_CAS_REUSE_REFERENCES_FILE),
+            serde_json::to_vec(&OfficialCasReuseReferenceManifest {
+                version: OFFICIAL_CAS_REUSE_REFERENCES_VERSION,
+                ownership_id: None,
+                object_ids: vec![missing_object],
+            })
+            .unwrap(),
+        )
+        .unwrap();
+
+        assert!(release_cas_reuse_references(&release_root, &cas_root).is_err());
+        let migrated = read_cas_reuse_reference_manifest_at(&release_root)
+            .unwrap()
+            .unwrap();
+        let ownership_id = migrated.ownership_id.clone().unwrap();
+        assert!(ownership_id.starts_with("cas-legacy-owner-v1-"));
+
+        fs::write(
+            release_root.join(OFFICIAL_CAS_REUSE_REFERENCES_FILE),
+            serde_json::to_vec(&OfficialCasReuseReferenceManifest {
+                version: OFFICIAL_CAS_REUSE_REFERENCES_VERSION,
+                ownership_id: Some(ownership_id),
+                object_ids: vec![object_id.clone()],
+            })
+            .unwrap(),
+        )
+        .unwrap();
+        assert_eq!(
+            release_cas_reuse_references(&release_root, &cas_root).unwrap(),
+            1
+        );
+        assert_eq!(cas_reference_count(&cas_root, &object_id), 0);
+    }
+
+    #[test]
+    fn legacy_partial_cleanup_keeps_basename_ledger_and_new_generation_isolated() {
+        let temp = TempDir::new().unwrap();
+        let cas_root = temp.path().join("cas");
+        let first = temp.path().join("versions").join("release-x");
+        fs::create_dir_all(&first).unwrap();
+        let first_object = store_cas_object(&cas_root, b"legacy-first");
+        let second_object = store_cas_object(&cas_root, b"legacy-second");
+        let cas = crate::FileSystemCasRepository::new(&cas_root);
+        tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .unwrap()
+            .block_on(async {
+                cas.add_reference(&first_object).await.unwrap();
+                cas.add_reference(&second_object).await.unwrap();
+                assert!(cas
+                    .release_reference_once("release-x", 0, &first_object)
+                    .await
+                    .unwrap());
+            });
+        fs::write(
+            first.join(OFFICIAL_CAS_REUSE_REFERENCES_FILE),
+            serde_json::to_vec(&OfficialCasReuseReferenceManifest {
+                version: OFFICIAL_CAS_REUSE_REFERENCES_VERSION,
+                ownership_id: None,
+                object_ids: vec![first_object.clone()],
+            })
+            .unwrap(),
+        )
+        .unwrap();
+
+        assert_eq!(release_cas_reuse_references(&first, &cas_root).unwrap(), 0);
+        assert_eq!(cas_reference_count(&cas_root, &first_object), 1);
+
+        fs::create_dir_all(&first).unwrap();
+        fs::write(
+            first.join(OFFICIAL_CAS_REUSE_REFERENCES_FILE),
+            serde_json::to_vec(&OfficialCasReuseReferenceManifest {
+                version: OFFICIAL_CAS_REUSE_REFERENCES_VERSION,
+                ownership_id: None,
+                object_ids: vec![second_object.clone()],
+            })
+            .unwrap(),
+        )
+        .unwrap();
+        assert_eq!(release_cas_reuse_references(&first, &cas_root).unwrap(), 1);
+        assert_eq!(cas_reference_count(&cas_root, &second_object), 1);
+        assert_eq!(cas_reference_count(&cas_root, &first_object), 1);
+    }
+
+    #[test]
+    fn legacy_generations_in_different_output_roots_do_not_share_scope() {
+        let temp = TempDir::new().unwrap();
+        let cas_root = temp.path().join("cas");
+        let first = temp.path().join("output-a/versions/release-x");
+        let second = temp.path().join("output-b/versions/release-x");
+        let first_object = store_cas_object(&cas_root, b"root-a");
+        let second_object = store_cas_object(&cas_root, b"root-b");
+        for (root, object_id) in [(&first, &first_object), (&second, &second_object)] {
+            fs::create_dir_all(root).unwrap();
+            fs::write(
+                root.join(OFFICIAL_CAS_REUSE_REFERENCES_FILE),
+                serde_json::to_vec(&OfficialCasReuseReferenceManifest {
+                    version: OFFICIAL_CAS_REUSE_REFERENCES_VERSION,
+                    ownership_id: None,
+                    object_ids: vec![object_id.clone()],
+                })
+                .unwrap(),
+            )
+            .unwrap();
+        }
+
+        assert_eq!(release_cas_reuse_references(&first, &cas_root).unwrap(), 1);
+        assert_eq!(release_cas_reuse_references(&second, &cas_root).unwrap(), 1);
+        assert_eq!(cas_reference_count(&cas_root, &first_object), 0);
+        assert_eq!(cas_reference_count(&cas_root, &second_object), 0);
+        assert_ne!(
+            load_owner_scope_state(&first).unwrap().scope_id,
+            load_owner_scope_state(&second).unwrap().scope_id
+        );
+    }
+
    #[test]
    fn corrupted_cas_falls_back_to_network_with_diagnostic() {
        let temp = TempDir::new().unwrap();
@@ -3209,6 +3209,7 @@ fn copy_tree_no_symlink(
                        | OFFICIAL_VERSIONS_DIR
                        | OFFICIAL_CURRENT_LINK
                        | ".official-sync.lock"
+                        | ".cas-owner-scope"
                ) {
                    continue;
                }
@@ -466,15 +466,19 @@ fn select_release_distribution_metadata(
            let manifest = read_download_manifest_at(&path)
                .map_err(anyhow::Error::msg)?
                .ok_or_else(|| anyhow::anyhow!("official release 缺少官方下载 manifest"))?;
+            if manifest.distribution_mapping_identity.is_none()
+                || manifest.destination_index.len() != manifest.entries.len()
+            {
+                return Ok(None);
+            }
            let entries = if let Some(destination) = destination {
-                let mut matching_entries = manifest
-                    .entries
-                    .values()
-                    .filter(|entry| entry.destination == destination);
-                let Some(entry) = matching_entries.next() else {
+                let Some(url) = manifest.destination_index.get(destination) else {
                    return Ok(None);
                };
-                if matching_entries.next().is_some() {
+                let Some(entry) = manifest.entries.get(url) else {
+                    return Ok(None);
+                };
+                if entry.destination != destination || entry.url != *url {
                    return Ok(None);
                }
                vec![ReleaseDistributionEntry {
@@ -532,25 +536,37 @@ fn select_release_distribution_metadata(
            }) else {
                return Ok(None);
            };
+            let Some(source_mapping_identity) =
+                official_manifest.distribution_mapping_identity.as_deref()
+            else {
+                return Ok(None);
+            };
+            if manifest.source_mapping_identity != source_mapping_identity
+                || manifest.localized_mapping_identity.is_empty()
+                || manifest.destination_index.len() != manifest.entries.len()
+                || official_manifest.destination_index.len() != official_manifest.entries.len()
+            {
+                return Ok(None);
+            }
            let entries = if let Some(destination) = destination {
-                let mut matching_entries = manifest
-                    .entries
-                    .iter()
-                    .filter(|entry| entry.destination == destination);
-                let Some(entry) = matching_entries.next() else {
+                let Some(entry_index) = manifest.destination_index.get(destination) else {
                    return Ok(None);
                };
-                if matching_entries.next().is_some() {
+                let Some(entry) = manifest.entries.get(*entry_index) else {
+                    return Ok(None);
+                };
+                if entry.destination != destination {
                    return Ok(None);
                }
-                let mut official_entries = official_manifest
-                    .entries
-                    .values()
-                    .filter(|official_entry| official_entry.destination == entry.destination);
-                let Some(official_entry) = official_entries.next() else {
+                let Some(official_url) = official_manifest.destination_index.get(destination)
+                else {
                    return Ok(None);
                };
-                if official_entries.next().is_some() {
+                let Some(official_entry) = official_manifest.entries.get(official_url) else {
+                    return Ok(None);
+                };
+                if official_entry.destination != destination || official_entry.url != *official_url
+                {
                    return Ok(None);
                }
                if !localized_distribution_entry_matches_official(entry, official_entry) {
@@ -784,7 +800,7 @@ pub fn cleanup_releases(
                .map_err(anyhow::Error::msg)?
                .is_some()
        {
-            let _ = release_cas_reuse_references(&path, cas_root).map_err(anyhow::Error::msg)?;
+            release_cas_reuse_references(&path, cas_root).map_err(anyhow::Error::msg)?;
        }
        fs::remove_dir_all(&path)?;
        removed.push(path);
@@ -1385,6 +1401,21 @@ fn verify_download_manifest(
    root: &Path,
    manifest: &OfficialDownloadManifest,
 ) -> anyhow::Result<()> {
+    let expected_identity =
+        crate::official_download::official_distribution_mapping_identity(manifest);
+    if manifest.distribution_mapping_identity.as_deref() != Some(expected_identity.as_str()) {
+        return Err(anyhow::anyhow!(
+            "official distribution mapping identity mismatch"
+        ));
+    }
+    let expected_index =
+        crate::official_download::official_distribution_destination_index(manifest)
+            .map_err(anyhow::Error::msg)?;
+    if manifest.destination_index != expected_index {
+        return Err(anyhow::anyhow!(
+            "official distribution destination index mismatch"
+        ));
+    }
    for entry in manifest.entries.values() {
        let path = safe_manifest_file_path(root, &entry.destination)?;
        let bytes = fs::read(&path)?;
@@ -1403,6 +1434,21 @@ fn verify_full_distribution(official_root: &Path, localized_root: &Path) -> Stri
    let Ok(Some(manifest)) = read_download_manifest_at(official_root) else {
        return "unknown".to_string();
    };
+    if let Some(localized_release_id) = read_localized_patch_manifest_at(localized_root)
+        .ok()
+        .flatten()
+        .map(|manifest| manifest.localized_release_id)
+    {
+        if crate::localized_patch::verify_localized_distribution_manifest_for_status(
+            official_root,
+            localized_root,
+            &localized_release_id,
+        )
+        .is_err()
+        {
+            return "invalid".to_string();
+        }
+    }
    let changed_paths = read_localized_patch_manifest_at(localized_root)
        .ok()
        .flatten()
@@ -1533,7 +1579,7 @@ mod tests {
        fs::create_dir_all(&version).unwrap();
        let data = b"official";
        fs::write(version.join("data.bin"), data).unwrap();
-        let manifest = OfficialDownloadManifest {
+        let mut manifest = OfficialDownloadManifest {
            version: 1,
            entries: [(
                "https://example.invalid/data.bin".to_string(),
@@ -1546,7 +1592,13 @@ mod tests {
            )]
            .into_iter()
            .collect(),
+            destination_index: BTreeMap::new(),
+            distribution_mapping_identity: None,
        };
+        manifest.distribution_mapping_identity =
+            Some(crate::official_distribution_mapping_identity(&manifest));
+        manifest.destination_index =
+            crate::official_download::official_distribution_destination_index(&manifest).unwrap();
        fs::write(
            version.join("official-download-manifest.json"),
            serde_json::to_vec(&manifest).unwrap(),
@@ -1608,6 +1660,42 @@ mod tests {
        assert!(!localized.rollback_available);
        assert!(!localized.unknown);
        assert_eq!(report.releases.len(), 2);
+
+        let mut tampered = crate::read_download_manifest_at(&official_version)
+            .unwrap()
+            .unwrap();
+        tampered
+            .entries
+            .get_mut("https://example.invalid/data.bin")
+            .unwrap()
+            .url = "https://example.invalid/tampered.bin".to_string();
+        fs::write(
+            official_version.join("official-download-manifest.json"),
+            serde_json::to_vec(&tampered).unwrap(),
+        )
+        .unwrap();
+        let tampered_report =
+            build_release_status(&official_root, &localized_root, Path::new("unzip")).unwrap();
+        assert!(!tampered_report.localized_distribution_ready);
+        let tampered_localized = tampered_report
+            .releases
+            .iter()
+            .find(|release| release.channel == "localized")
+            .unwrap();
+        assert_eq!(tampered_localized.distribution_integrity_status, "invalid");
+        assert!(tampered_localized.damaged);
+        let blocked = select_release_distribution(
+            &official_root,
+            &localized_root,
+            &ReleaseDistributionParams {
+                channel: Some("localized".to_string()),
+                destination: Some("data.bin".to_string()),
+                ..ReleaseDistributionParams::default()
+            },
+            Path::new("unzip"),
+        )
+        .unwrap();
+        assert!(!blocked.available);
    }

    #[test]
@@ -1682,13 +1770,19 @@ mod tests {
                },
            );
        }
+        let mut manifest = OfficialDownloadManifest {
+            version: 1,
+            entries,
+            destination_index: BTreeMap::new(),
+            distribution_mapping_identity: None,
+        };
+        manifest.distribution_mapping_identity =
+            Some(crate::official_distribution_mapping_identity(&manifest));
+        manifest.destination_index =
+            crate::official_download::official_distribution_destination_index(&manifest).unwrap();
        fs::write(
            version.join("official-download-manifest.json"),
-            serde_json::to_vec(&OfficialDownloadManifest {
-                version: 1,
-                entries,
-            })
-            .unwrap(),
+            serde_json::to_vec(&manifest).unwrap(),
        )
        .unwrap();
        fs::create_dir_all(&official_root).unwrap();