diff --git a/features/v3.0.0-beta1/048-rebac-permission-model-grants/design.md b/features/v3.0.0-beta1/048-rebac-permission-model-grants/design.md
index b6ddd836c5..907eed45b2 100644
--- a/features/v3.0.0-beta1/048-rebac-permission-model-grants/design.md
+++ b/features/v3.0.0-beta1/048-rebac-permission-model-grants/design.md
@@ -1041,7 +1041,17 @@ HTTP 每次最多 50 个 change item；单槽 visible tuple 也计入编译后
 FGA Write 超时属于 `COMMIT_UNKNOWN`：reconciler 用 higher consistency 对比 operation
 记录的 before/after tuple checksum。全为 after 则转 COMMITTED，全为 before 才可按原
 operation 重试；出现混合集或 scope version 已被外部改变时标 FAILED_CLOSED 并 fence
-该资源，不能盲目补写覆盖后来状态。
+该资源的后续权限配置写入，不能盲目补写覆盖后来状态。资源的具体 action/visible 决策不复用
+该写栅栏：`PROJECTING`、`COMMIT_UNKNOWN`、`COMMITTED` 和 `FAILED_CLOSED` 期间仍由唯一
+OpenFGA 执行面返回 ALLOW/DENY，并强制 higher consistency。OpenFGA 或 CURRENT Catalog/model
+不可用、verified identity/parent 不匹配时仍 fail closed；SQL Grant 与待处理来源不参与决策兜底。
+
+`ResourcePermissionMode.projection_state` 同时承担 operation 归属和版本提交记录，但不再作为
+普通业务鉴权的全局停服开关。decision fence 接受 `CURRENT/PROJECTING/FAILED_CLOSED` 三种资源
+镜像状态；后两者以及同一请求内观察到的 permission version 变化返回 degraded 标记并强制
+higher consistency。Grant/mode/lifecycle command 仍只允许从 `CURRENT` claim 新 operation，
+因此同一资源的权限写保持串行。`my-permissions` 在 degraded 状态只返回 OpenFGA 逐 action
+结果和投影状态，不读取 staged SQL source explanation。
 
 人工恢复统一从 backend 容器运行
 `scripts/reconcile_f048_projection_operations.py --tenant-id <tenant> <operation...>`；默认
@@ -1872,7 +1882,7 @@ D5 必须同时满足：
 | D2 数据脚本 | 保持运维停流；修复源数据/映射或脚本后，以同一 run/checkpoint 续跑 `migrate --apply` |
 | D3～D4 | 保持运维停流；按 migration item 对 SQL/同一 Store 的新 tuple 与 legacy delete 做幂等前向修正，重跑完整 D4 |
 | D5 重启或 smoke | 立即保持/恢复运维停流并停止全部权限写；新 model 仍是唯一目标，修复配置/代码/投影后重跑 D4+D5 |
-| D6 运行中 | 由新业务 Service + `permission_projection_operation/tuple` 执行 forward-fix；`FAILED_CLOSED` 资源保持 fenced，修复并 higher-consistency 验证后再开放 |
+| D6 运行中 | 由新业务 Service + `permission_projection_operation/tuple` 执行 forward-fix；`FAILED_CLOSED` 资源冻结新的权限配置写入，但具体 action/visible 继续以 higher-consistency OpenFGA 决策；修复验证后恢复正常一致性与权限写入 |
 
 禁止重新 pin 旧 model、恢复 Config 第二 PDP、把新授权 down-convert 成四档 tuple、
 逐请求询问旧 model，或只回退应用代码。旧 model 无法从 OpenFGA 删除，保留它只是产品
@@ -1934,6 +1944,7 @@ D3 已完成全部旧运行数据退役，D6 没有延后的 cleanup 窗口。
 |---|---|---|
 | 2026-08-13 | 对单槽浅层 visible、inactive 既有授权保持、删除零引用门禁、旧系统单次迁移、列表路径、契约/依赖/测试/可观测执行 24 项 Design 接手测试与 Constitution Check；复审 LGTM，停在 Design ★ | `/sdd-review ... design` |
 | 2026-08-13 | 将模型 `active` 收窄为“是否可用于新增/变更授权”：停用不影响已有 Grant；删除必须先撤销或替换全部绑定，并在引用/source projection/live tuple 清零后完成。可见执行关系改为单槽浅层 `visible`，移除 A/B 槽、switch、双写和 Catalog 4-tuple 切换；保留来源引用计数、ledger、reconcile 与旧系统单次迁移 | 用户确认界面语义“关闭后不能再用它授权，已有授权不受影响；删除必须先清理绑定关系” |
+| 2026-08-20 | 将资源权限投影状态从普通鉴权停服条件中解耦：非 CURRENT 期间继续执行 higher-consistency OpenFGA action/visible 决策，只冻结同资源新的权限配置写；`my-permissions` 降级为 OpenFGA 逐动作结果且不展示 staged SQL 来源 | 用户确认增加查看者等权限修改不得影响已有授权的正常鉴权 |
 | 2026-08-13 | 按 F048 未上线事实重新完成 24 项 Design 接手测试：旧系统单次迁移、A/B 可见投影、订阅/历史 membership 证据边界、D4 完整性门禁和 Release Contract 均一致；复审 LGTM，停在 Design ★ | `/sdd-review ... design` |
 | 2026-08-13 | 用户澄清 F048 尚未上线：删除“旧 F048 → successor model”二次迁移和独立 migration run/checkpoint，把 A/B 可见 source projection、switch 与完整性校验合并到原 `migrate_f048_permission_data.py` 从旧系统执行的唯一正式 run | 用户迁移拓扑纠正 |
 | 2026-08-13 | 首轮 24 项 Design 接手测试修正现状快照、跨模型同来源 fingerprint 冲突与完整枚举错误码缺口；迁移拓扑后续按上一行重新修订并复审 | `/sdd-review ... design` |
diff --git a/features/v3.0.0-beta1/048-rebac-permission-model-grants/spec.md b/features/v3.0.0-beta1/048-rebac-permission-model-grants/spec.md
index fccb8f1591..8757698c18 100644
--- a/features/v3.0.0-beta1/048-rebac-permission-model-grants/spec.md
+++ b/features/v3.0.0-beta1/048-rebac-permission-model-grants/spec.md
@@ -322,6 +322,18 @@
 - **AC-68** — WHEN 两个管理员基于同一旧版本并发修改模型、Grant 或权限模式, THE SYSTEM SHALL 只接受符合当前版本的更新，并明确拒绝过期覆盖。
 - **AC-69** — WHEN 授权或撤销成功返回, THE SYSTEM SHALL 让随后用于安全决策的读取观察到该新状态，不得因旧缓存继续产生与已确认变更相反的 ALLOW。
 - **AC-70** — IF 业务数据变更已发生但权限状态未能完成一致更新, THEN THE SYSTEM SHALL 不把该权限变化报告为已生效，并产生可恢复、可审计的异常；业务资源结果继续遵守其 Owner Feature 与 Constitution C4 的失败补偿契约。
+- **AC-178** — WHILE 一个已存在资源的权限投影处于 `PROJECTING`、`COMMIT_UNKNOWN`、
+  `COMMITTED` 或 `FAILED_CLOSED`, THE SYSTEM SHALL 继续通过唯一 OpenFGA 执行面对该资源执行
+  具体 action、`visible` 与批量鉴权；投影控制面状态本身不得暂停未参与本次变更的既有授权。
+- **AC-179** — WHEN 非 `CURRENT` 投影状态或权限版本在同一请求内发生切换, THE SYSTEM SHALL
+  对具体资源鉴权强制使用 higher consistency；OpenFGA、CURRENT Catalog/model 或已验证资源
+  identity/parent 不可用时仍须 fail closed，不得使用 SQL Grant 或待处理记录补充 ALLOW。
+- **AC-180** — WHILE 资源权限投影不是 `CURRENT`, THE SYSTEM SHALL 拒绝新的 Grant mutation、
+  mode switch 及其他权限配置写入，只允许原 operation 的幂等继续或受控恢复；普通资源业务
+  读写不得复用该权限写锁。
+- **AC-181** — WHEN `my-permissions` 在非 `CURRENT` 投影状态下读取当前用户有效权限,
+  THE SYSTEM SHALL 从 OpenFGA 逐动作计算结果并返回投影降级状态，不得把 SQL 中的
+  `PENDING`、`PENDING_DELETE` 或 `PROJECTING` 来源明细包装成最终授权来源。
 
 ### 3.9 PRD 显式交互与新建规则
 
diff --git a/features/v3.0.0-beta1/048-rebac-permission-model-grants/tasks.md b/features/v3.0.0-beta1/048-rebac-permission-model-grants/tasks.md
index 23d8fe05c2..7d3678925d 100644
--- a/features/v3.0.0-beta1/048-rebac-permission-model-grants/tasks.md
+++ b/features/v3.0.0-beta1/048-rebac-permission-model-grants/tasks.md
@@ -1631,6 +1631,32 @@
 
 ---
 
+---
+
+## Wave 15 — 投影写锁与普通鉴权解耦
+
+- [x] **T195：非 CURRENT 投影鉴权合同测试**
+  - **文件**：`src/backend/test/permission/test_stale_projection_fail_soft.py`,
+    `src/backend/test/permission/test_f048_resource_api.py`
+  - **测试**：覆盖 `CURRENT/PROJECTING/FAILED_CLOSED` decision fence、请求内 version 切换、
+    higher-consistency 强制、PENDING 拒绝、FAILED_CLOSED 新权限写拒绝，以及 degraded
+    `my-permissions` 不读取 staged source explanation。
+  - **覆盖 AC**：AC-178, AC-179, AC-180, AC-181
+
+- [x] **T196：解耦资源投影写锁与 OpenFGA 决策路径**
+  - **文件**：`src/backend/bisheng/permission/domain/models/grant.py`,
+    `src/backend/bisheng/permission/application/control_state.py`,
+    `src/backend/bisheng/permission/application/sql_runtime.py`,
+    `src/backend/bisheng/permission/domain/services/permission_action_service.py`,
+    `src/backend/bisheng/permission/application/resource_api.py`
+  - **逻辑**：允许现有资源在 `PROJECTING/FAILED_CLOSED` 镜像状态下生成 verified target 并执行
+    OpenFGA action/visible；非 CURRENT 或版本切换强制 higher consistency；Grant/mode/lifecycle
+    新 operation 仍只从 CURRENT claim；degraded `my-permissions` 返回实际 action 与投影状态，
+    不把 PENDING/PENDING_DELETE 来源作为最终事实。
+  - **验收**：T195、F048 permission service、projection SQL runtime 定向回归、Ruff、
+    arch-guard 与 diff-check 通过。
+  - **依赖**：T195
+
 ## 实际偏差记录
 
 > 只记录一句话指针；设计原因和反直觉事实回写 [design.md](./design.md)。
diff --git a/features/v3.0.0-beta1/release-contract.md b/features/v3.0.0-beta1/release-contract.md
index 47a08cf07c..d66edd9f56 100644
--- a/features/v3.0.0-beta1/release-contract.md
+++ b/features/v3.0.0-beta1/release-contract.md
@@ -63,7 +63,7 @@
 | INV-16 | 权限继承复用资源既有的直接 `parent` 语义，不建立第二套 `permission_parent` 层级；`CUSTOM` 只切断权限继承，不能改变业务结构父子关系 | ResourcePermissionMode | F048 |
 | INV-17 | 任一有效 Grant 可以产生资源列表/基础元数据可见性，但可见性不能替代下载、搜索、RAG 或业务变更动作的具体鉴权。文件预览不设置 PermissionAction；只有原件/打包下载必须检查 `download`，不得由“可预览”推导下载能力 | PermissionAction, PermissionGrant | F048 |
 | INV-18 | 权限升级采用应用自动阻断业务访问后的单向正式数据迁移：更新镜像并启动进程后，旧 model 只能进入 `MIGRATION_REQUIRED/NOT_READY` 运维态，不初始化 F048 权限运行时、不发布 ready heartbeat，HTTP/WS 迁移门禁除 `/health` 外统一拒绝访问，Celery/Linsight 暂停消费任务；schema upgrade 成功后，由 `src/backend/scripts/` 专用脚本沿用现有 Store 发布一个新 model ID，原地转换 tuple 并退役旧运行数据，校验通过后重启全部进程并自动恢复访问/任务消费。F048 不提供独立迁移预演、旧/新 model 影子运行、应用级回滚、新→旧转换、dual/legacy model client、长期双写、旧动作别名、Config 第二 PDP 或逐请求旧系统 ALLOW fallback；失败保持维护并前向修复 | PermissionMigrationRun | F048 |
-| INV-19 | 对需要进入资源 ReBAC 的请求，权限服务不可用、模型未生效、动作未分级、迁移记录不明确或授权状态不可判定时必须 fail closed | PermissionAction, PermissionModel, PermissionGrant | F048 |
+| INV-19 | 对需要进入资源 ReBAC 的请求，权限服务不可用、模型未生效、动作未分级、迁移记录不明确或 OpenFGA 具体决策不可获得时必须 fail closed；资源投影处于 PROJECTING/COMMIT_UNKNOWN/COMMITTED/FAILED_CLOSED 本身不等于具体决策不可判定，普通 action/visible 继续通过唯一 OpenFGA 执行面以 higher consistency 决策，但新的权限配置写持续冻结且 SQL Grant/待处理来源不得补充 ALLOW | PermissionAction, PermissionModel, PermissionGrant | F048 |
 | INV-20 | 动作、模型、模型动作、资源 Grant、Grant 主体和权限模式的运行时事实必须存于规范化关系表；`permission_relation_models_v1`、`permission_relation_model_bindings_v1` 及任何新的大 JSON 不得继续作为运行时真相 | PermissionAction, PermissionModel, PermissionGrant, ResourcePermissionMode | F048 |
 | INV-21 | 所有生产 OpenFGA Check、List 和 Write 必须显式指定经发布门禁确认的 Authorization Model ID；发布新模型不得依赖“自动使用最新模型”完成切换 | AuthorizationModelRelease | F048 |
 | INV-22 | 旧 `owner/manager/editor/viewer` 只迁移直接关系事实；由旧模型计算出的层级、父级或角色蕴含结果不得展开为新的 Grant assignee | PermissionGrant, PermissionGrantAssignee | F048 |
@@ -144,3 +144,4 @@
 | 2026-08-13 | 明确模型停用/删除语义：停用只禁止新增或变更授权，已有 Grant 保持有效；删除必须先清零或替换全部绑定并完成残留投影对账。因停用不再触发批量撤权，F048 可见执行投影采用单槽浅层 `visible`，不引入 A/B 槽与运行时 switch | F048 |
 | 2026-08-14 | 登记 F049 知识空间目录与搜索读取优化：指定资源的超级管理员按 C4 系统身份策略放行、去重空间鉴权、页大小驱动的有界候选扫描、移除未展示的文件夹数量统计并保留失败存在性、增加分段性能观测；普通用户候选最终可见性继续统一使用 OpenFGA BatchCheck，不新增继承捷径 | F049、F027、F040、F048 |
 | 2026-08-14 | 登记 F050 统一权限设置入口：保留 v2.6.0 F044 页面目标，创建/编辑权限完全改接 F048，并增加创建后初始 Grant 部分失败与持久幂等约束 INV-28 | F050、F048 |
+| 2026-08-20 | 澄清 INV-19 的决策与写入边界：资源权限投影非 CURRENT 时冻结新的权限配置写，但不暂停普通资源鉴权；具体 action/visible 继续使用 higher-consistency OpenFGA，OpenFGA/Catalog/model/verified identity 不可用时仍 fail closed，SQL 不兜底 ALLOW | F048 |
diff --git a/src/backend/bisheng/permission/api/endpoints/decision.py b/src/backend/bisheng/permission/api/endpoints/decision.py
index 414d021593..b588eab152 100644
--- a/src/backend/bisheng/permission/api/endpoints/decision.py
+++ b/src/backend/bisheng/permission/api/endpoints/decision.py
@@ -35,11 +35,16 @@ async def check_permission_action(
     except ValidationError:
         return InvalidCatalogActionError.return_resp()
     try:
+        actor = await permission_actor(login_user)
+        if request.action == "visible" and actor.super_admin:
+            # This HTTP contract answers privileged operational access. Keep the
+            # domain visible Check/BatchCheck/ListObjects relation unchanged.
+            return resp_200({"allowed": True})
         allowed = await api.check(
             resource_type=request.resource_type,
             resource_id=request.resource_id,
             action=request.action,
-            actor=await permission_actor(login_user),
+            actor=actor,
         )
     except BaseErrorCode as error:
         return permission_error_response(error)
diff --git a/src/backend/bisheng/permission/application/control_state.py b/src/backend/bisheng/permission/application/control_state.py
index cae0a9048a..30ef81ab32 100644
--- a/src/backend/bisheng/permission/application/control_state.py
+++ b/src/backend/bisheng/permission/application/control_state.py
@@ -19,6 +19,7 @@
     stable_grant_key,
 )
 from bisheng.permission.domain.models import (
+    DECIDABLE_PROJECTION_STATES,
     AuthorizationModelRelease,
     PermissionAction,
     PermissionCatalogRelease,
@@ -200,7 +201,7 @@ async def permission_version(
                 ResourcePermissionMode.resource_id == resource_id,
             )
             row = (await session.execute(statement)).scalars().first()
-        if row is None or row.projection_state != "CURRENT":
+        if row is None or row.projection_state not in DECIDABLE_PROJECTION_STATES:
             raise PermissionPublishNotReadyError(msg="Resource permission projection is not current")
         context = "|".join(
             (
@@ -1283,6 +1284,47 @@ async def _upsert_assignee(
             .scalars()
             .first()
         )
+        if row is not None and row.id != source.source_id:
+            signature = (
+                row.subject_type,
+                row.subject_id,
+                row.userset_relation,
+                row.source_locator,
+                bool(row.protected),
+            )
+            expected = (
+                source.subject_type,
+                source.subject_id,
+                source.userset_relation,
+                source.source_locator,
+                source.protected,
+            )
+            if signature != expected:
+                raise PermissionVersionConflictError(msg="Permission source fingerprint collision")
+            if row.state != "INACTIVE":
+                raise PermissionVersionConflictError(
+                    msg="Target Grant already contains another current assignee identity",
+                )
+            visible_owner_key = f"grant_assignee:{row.id}"
+            current_visible_source = (
+                await session.execute(
+                    select(PermissionVisibleSourceProjection.id)
+                    .where(
+                        PermissionVisibleSourceProjection.tenant_id == grant_row.tenant_id,
+                        PermissionVisibleSourceProjection.source_kind == "GRANT_ASSIGNEE",
+                        PermissionVisibleSourceProjection.source_owner_key == visible_owner_key,
+                        PermissionVisibleSourceProjection.state != "RETIRED",
+                    )
+                    .limit(1)
+                )
+            ).scalar_one_or_none()
+            if current_visible_source is not None:
+                raise PermissionVersionConflictError(
+                    msg="Historical target assignee still owns a current visible projection",
+                )
+            await session.delete(row)
+            await session.flush()
+            row = None
         if row is None:
             id_collision = await session.get(
                 PermissionGrantAssignee,
diff --git a/src/backend/bisheng/permission/application/resource_api.py b/src/backend/bisheng/permission/application/resource_api.py
index b277c6d3e6..6d50095e38 100644
--- a/src/backend/bisheng/permission/application/resource_api.py
+++ b/src/backend/bisheng/permission/application/resource_api.py
@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import base64
 import json
 from datetime import UTC, datetime, timedelta
@@ -64,13 +65,13 @@ async def resource_display_names(
     ) -> dict[tuple[str, str], str]: ...
 
 
-
 def _split_resource_key(resource_key: str) -> tuple[str, str]:
     """Split "knowledge_space:3377" into its type and id."""
 
     resource_type, _, resource_id = resource_key.partition(":")
     return resource_type, resource_id
 
+
 def _encode_cursor(payload: dict[str, object]) -> str:
     raw = json.dumps(
         payload,
@@ -196,9 +197,7 @@ async def list_grants(
         # roster used to render "knowledge_space:3377" at users. Resolved through
         # the business side, the same way subject names already are.
         parents = tuple(
-            dict.fromkeys(
-                _split_resource_key(row.inherited_from) for row in selected if row.inherited_from
-            )
+            dict.fromkeys(_split_resource_key(row.inherited_from) for row in selected if row.inherited_from)
         )
         parent_names = await self._subjects.resource_display_names(parents) if parents else {}
         model_names = {item.snapshot.model_key: item.name for item in catalog.models}
@@ -267,19 +266,37 @@ async def get_my_permissions(
             "visible",
         )
         await self._require_visible(actor, target)
+        mode = await self._runtime.mode_for_target(target)
+        projection_degraded = mode.projection_state != "CURRENT" or mode.version != target.resource_version
         if self._privileged(actor, target):
             # A super admin / tenant admin is authorized on identity and holds
             # no grant rows, so the grant-derived explanation would report an
             # empty action set — "visible but powerless", which is exactly what
             # made the client show them as having no permissions. Report the
             # full effective action set for the resource type instead.
-            mode = await self._runtime.current_mode(target)
             actions = await self._runtime.effective_actions(resource_type)
             return {
                 "mode": mode.mode,
                 "actions": list(actions),
                 "sources": [],
                 "roster_complete": False,
+                "projection_state": mode.projection_state,
+                "projection_degraded": projection_degraded,
+            }
+        if projection_degraded:
+            effective_actions = await self._runtime.effective_actions(resource_type)
+            allowed = await asyncio.gather(
+                *(self._runtime.check_action(actor, target, action) for action in effective_actions)
+            )
+            return {
+                "mode": mode.mode,
+                "actions": [
+                    action for action, is_allowed in zip(effective_actions, allowed, strict=True) if is_allowed
+                ],
+                "sources": [],
+                "roster_complete": False,
+                "projection_state": mode.projection_state,
+                "projection_degraded": True,
             }
         explanation = await self._explanation(
             actor,
@@ -297,6 +314,8 @@ async def get_my_permissions(
                 for row in explanation.sources
             ],
             "roster_complete": False,
+            "projection_state": mode.projection_state,
+            "projection_degraded": False,
         }
 
     async def mutate_grants(
@@ -484,10 +503,7 @@ def _privileged(actor: PermissionActor, target) -> bool:
         """
         if actor.super_admin:
             return True
-        return (
-            target.tenant_id == actor.current_tenant_id
-            and target.tenant_id in actor.tenant_admin_tenant_ids
-        )
+        return target.tenant_id == actor.current_tenant_id and target.tenant_id in actor.tenant_admin_tenant_ids
 
     async def _require_visible(self, actor, target) -> None:
         if self._privileged(actor, target):
diff --git a/src/backend/bisheng/permission/application/sql_runtime.py b/src/backend/bisheng/permission/application/sql_runtime.py
index 25761d5c9c..d9c53d695a 100644
--- a/src/backend/bisheng/permission/application/sql_runtime.py
+++ b/src/backend/bisheng/permission/application/sql_runtime.py
@@ -20,6 +20,7 @@
 from bisheng.core.database import get_async_db_session
 from bisheng.core.openfga.client import FGAClient
 from bisheng.permission.domain.models import (
+    DECIDABLE_PROJECTION_STATES,
     AuthorizationModelRelease,
     AuthorizationModelReleaseStatus,
     PermissionAction,
@@ -174,12 +175,12 @@ async def effective_actions(self, resource_type: str) -> tuple[str, ...]:
 
 
 class SqlPermissionScopeFence:
-    """Trust only a CURRENT permission-owned mirror of a verified target."""
+    """Validate decision identity while permission writes remain serialized."""
 
     async def ensure_readable(
         self,
         target: VerifiedPermissionTarget,
-    ) -> None:
+    ) -> bool:
         async with get_async_db_session() as session:
             statement = select(ResourcePermissionMode).where(
                 ResourcePermissionMode.tenant_id == target.tenant_id,
@@ -189,8 +190,7 @@ async def ensure_readable(
             row = (await session.execute(statement)).scalars().first()
         if (
             row is None
-            or row.version != target.resource_version
-            or row.projection_state != "CURRENT"
+            or row.projection_state not in DECIDABLE_PROJECTION_STATES
             or row.parent_type != target.parent_type
             or row.parent_id != target.parent_id
         ):
@@ -203,8 +203,9 @@ async def ensure_readable(
                 expected_parent_type=target.parent_type,
                 expected_parent_id=target.parent_id,
                 expected_version=target.resource_version,
-                expected_projection_state="CURRENT",
+                expected_projection_state="CURRENT|PROJECTING|FAILED_CLOSED",
             )
+        return row.projection_state != "CURRENT" or row.version != target.resource_version
 
 
 class RedisConsistencyMarker:
diff --git a/src/backend/bisheng/permission/domain/models/__init__.py b/src/backend/bisheng/permission/domain/models/__init__.py
index b9d18706c3..b8edab903f 100644
--- a/src/backend/bisheng/permission/domain/models/__init__.py
+++ b/src/backend/bisheng/permission/domain/models/__init__.py
@@ -10,6 +10,7 @@
     PermissionModelAction,
 )
 from .grant import (
+    DECIDABLE_PROJECTION_STATES,
     GrantState,
     PermissionGrant,
     PermissionGrantAssignee,
@@ -34,6 +35,7 @@
 )
 
 __all__ = [
+    "DECIDABLE_PROJECTION_STATES",
     "AuthorizationModelRelease",
     "AuthorizationModelReleaseStatus",
     "CatalogReleaseStatus",
diff --git a/src/backend/bisheng/permission/domain/models/grant.py b/src/backend/bisheng/permission/domain/models/grant.py
index eac1fb04d0..ade8172b57 100644
--- a/src/backend/bisheng/permission/domain/models/grant.py
+++ b/src/backend/bisheng/permission/domain/models/grant.py
@@ -37,6 +37,15 @@ class ProjectionState(StrEnum):
     FAILED_CLOSED = "FAILED_CLOSED"
 
 
+DECIDABLE_PROJECTION_STATES = frozenset(
+    {
+        ProjectionState.CURRENT.value,
+        ProjectionState.PROJECTING.value,
+        ProjectionState.FAILED_CLOSED.value,
+    }
+)
+
+
 class PermissionGrant(SQLModelSerializable, table=True):
     __tablename__ = "permission_grant"
     __table_args__ = (
diff --git a/src/backend/bisheng/permission/domain/services/permission_action_service.py b/src/backend/bisheng/permission/domain/services/permission_action_service.py
index dee1353d54..af07498696 100644
--- a/src/backend/bisheng/permission/domain/services/permission_action_service.py
+++ b/src/backend/bisheng/permission/domain/services/permission_action_service.py
@@ -57,7 +57,7 @@ class PermissionScopeFencePort(Protocol):
     async def ensure_readable(
         self,
         target: VerifiedPermissionTarget,
-    ) -> None: ...
+    ) -> bool: ...
 
 
 class PermissionConsistencyMarkerPort(Protocol):
@@ -165,8 +165,11 @@ async def check_action(
             )
             return allowed
 
-        await self._prepare_action_target(target, action)
-        consistency = await self._consistency(target)
+        force_higher_consistency = await self._prepare_action_target(target, action)
+        consistency = await self._consistency(
+            target,
+            force_higher_consistency=force_higher_consistency,
+        )
         try:
             allowed = await self._fga.check(
                 user=f"user:{actor.user_id}",
@@ -206,8 +209,11 @@ async def check_visible(
             )
             return False
         await self._catalog.ensure_runtime_ready()
-        await self._scope_fence.ensure_readable(target)
-        consistency = await self._consistency(target)
+        force_higher_consistency = bool(await self._scope_fence.ensure_readable(target))
+        consistency = await self._consistency(
+            target,
+            force_higher_consistency=force_higher_consistency,
+        )
         try:
             allowed = await self._fga.check(
                 user=f"user:{actor.user_id}",
@@ -257,12 +263,15 @@ async def batch_check_actions(
                 results[index] = shortcut[0]
                 continue
             try:
-                await self._prepare_action_target(target, action)
+                force_higher_consistency = await self._prepare_action_target(target, action)
             except PermissionPublishNotReadyError as exc:
                 results[index] = False
                 self._handle_stale_projection(target, exc)
                 continue
-            target_consistency = await self._consistency(target)
+            target_consistency = await self._consistency(
+                target,
+                force_higher_consistency=force_higher_consistency,
+            )
             if target_consistency == HIGHER_CONSISTENCY:
                 consistency = HIGHER_CONSISTENCY
             unresolved.append((index, target))
@@ -305,12 +314,15 @@ async def batch_check_visible(
                 continue
             await self._catalog.ensure_runtime_ready()
             try:
-                await self._scope_fence.ensure_readable(target)
+                force_higher_consistency = bool(await self._scope_fence.ensure_readable(target))
             except PermissionPublishNotReadyError as exc:
                 results[index] = False
                 self._handle_stale_projection(target, exc)
                 continue
-            target_consistency = await self._consistency(target)
+            target_consistency = await self._consistency(
+                target,
+                force_higher_consistency=force_higher_consistency,
+            )
             if target_consistency == HIGHER_CONSISTENCY:
                 consistency = HIGHER_CONSISTENCY
             unresolved.append((index, target))
@@ -551,19 +563,31 @@ async def _prepare_action_target(
         self,
         target: VerifiedPermissionTarget,
         action: str,
-    ) -> None:
+    ) -> bool:
         await self._catalog.ensure_runtime_ready()
-        await self._scope_fence.ensure_readable(target)
+        force_higher_consistency = bool(await self._scope_fence.ensure_readable(target))
         if not await self._catalog.is_action_effective(
             target.resource_type,
             action,
         ):
             raise InvalidCatalogActionError(msg=f"Action {action} is unavailable for {target.resource_type}")
+        return force_higher_consistency
 
     async def _consistency(
         self,
         target: VerifiedPermissionTarget,
+        *,
+        force_higher_consistency: bool = False,
     ) -> str | None:
+        if force_higher_consistency:
+            emit_metric(
+                "permission",
+                event="degraded_projection_decision",
+                resource_type=target.resource_type,
+                resource_id=target.resource_id,
+                tenant_id=str(target.tenant_id),
+            )
+            return HIGHER_CONSISTENCY
         return await self._scope_consistency(
             target.tenant_id,
             target.resource_type,
diff --git a/src/backend/bisheng/workflow/nodes/report/docx_replace.py b/src/backend/bisheng/workflow/nodes/report/docx_replace.py
index 74b6ab6305..958abff12b 100644
--- a/src/backend/bisheng/workflow/nodes/report/docx_replace.py
+++ b/src/backend/bisheng/workflow/nodes/report/docx_replace.py
@@ -6,6 +6,7 @@
 from docx import Document
 from docx.enum.style import WD_STYLE_TYPE
 from docx.oxml import OxmlElement
+from docx.oxml.ns import qn
 from docx.shared import Inches, Pt, RGBColor
 from docx.table import _Cell
 from docx.text.paragraph import Paragraph
@@ -227,6 +228,28 @@ def _get_paragraph_index(self, paragraph: Paragraph) -> int:
         parent = paragraph._element.getparent()
         return parent.index(paragraph._element)
 
+    def _run_spans(self, paragraph: Paragraph) -> list[tuple[int, int, Any]]:
+        """Character range each run covers, over the paragraph's run text."""
+        spans = []
+        offset = 0
+        for run in paragraph.runs:
+            spans.append((offset, offset + len(run.text), run._element))
+            offset += len(run.text)
+        return spans
+
+    def _append_run(self, paragraph: Paragraph, template_element, text: str, format_data: dict[str, Any]):
+        """Append a run, cloning `template_element` so its rPr carries over."""
+        if template_element is None:
+            run = paragraph.add_run(text)
+        else:
+            element = copy.deepcopy(template_element)
+            paragraph._element.append(element)
+            run = Run(element, paragraph)
+            run.text = text
+        if format_data:
+            self._apply_run_format(run, format_data)
+        return run
+
     def _replace_paragraph_placeholders(
         self,
         paragraph: Paragraph,
@@ -234,59 +257,91 @@ def _replace_paragraph_placeholders(
         variables: dict[str, list[dict[str, Any]]],
         insert_index: int,
     ):
+        """Split a paragraph around block-level values (table / image / heading).
+
+        The paragraph has to be taken apart here, so every piece is rebuilt from
+        the original XML rather than from plain text: each new paragraph clones
+        the source paragraph's pPr (style, numbering, borders) and each run is
+        cloned from the run the text actually came from, which is what keeps a
+        mid-paragraph underline or font change alive.
+        """
         parent = paragraph._element.getparent()
-        text = paragraph.text
+        run_spans = self._run_spans(paragraph)
+        runs_text = "".join(paragraph.runs[i].text for i in range(len(paragraph.runs)))
+        run_matches = list(self.placeholder_pattern.finditer(runs_text))
+
+        if len(run_matches) == len(matches):
+            # Offsets line up with the runs, so text can keep its own formatting.
+            text, matches = runs_text, run_matches
+        else:
+            # Placeholder lives somewhere runs don't reach (a hyperlink, a field).
+            # Fall back to the flat text; formatting fidelity is lost but the
+            # value still lands.
+            text, run_spans = paragraph.text, []
 
         segments = []
         last_end = 0
-
         for match in matches:
             var_name = match.group(1)
             start, end = match.span()
-
             if start > last_end:
-                segments.append({"type": "text_segment", "content": text[last_end:start], "paragraph": paragraph})
-
+                segments.append({"type": "text_range", "start": last_end, "end": start})
             if var_name in variables:
-                segments.append({"type": "variable", "content": variables[var_name], "paragraph": paragraph})
+                segments.append(
+                    {
+                        "type": "variable",
+                        "items": variables[var_name],
+                        "template": self._run_element_at(run_spans, start),
+                    }
+                )
             else:
-                segments.append({"type": "text_segment", "content": match.group(0), "paragraph": paragraph})
-
+                segments.append({"type": "text_range", "start": start, "end": end})
             last_end = end
-
         if last_end < len(text):
-            segments.append({"type": "text_segment", "content": text[last_end:], "paragraph": paragraph})
+            segments.append({"type": "text_range", "start": last_end, "end": len(text)})
 
+        pPr = paragraph._element.find(qn("w:pPr"))
+        pPr_template = copy.deepcopy(pPr) if pPr is not None else None
         original_format = self._extract_paragraph_format(paragraph)
-        original_run_format = self._extract_run_format(paragraph.runs[0] if paragraph.runs else None)
+        fallback_run_format = self._extract_run_format(paragraph.runs[0] if paragraph.runs else None)
 
         parent.remove(paragraph._element)
 
         current_insert_index = insert_index
         current_paragraph = None
 
-        for segment in segments:
-            if segment["type"] == "text_segment":
-                if current_paragraph is None:
-                    current_paragraph = self._insert_paragraph_at_index(parent, current_insert_index, original_format)
-                    current_insert_index += 1
+        def ensure_paragraph():
+            nonlocal current_paragraph, current_insert_index
+            if current_paragraph is None:
+                current_paragraph = self._insert_paragraph_at_index(
+                    parent, current_insert_index, original_format, pPr_template
+                )
+                current_insert_index += 1
+            return current_paragraph
 
-                run = current_paragraph.add_run(segment["content"])
-                self._apply_run_format(run, original_run_format)
+        for segment in segments:
+            if segment["type"] == "text_range":
+                start, end = segment["start"], segment["end"]
+                if start >= end:
+                    continue
+                target = ensure_paragraph()
+                if run_spans:
+                    for run_start, run_end, element in run_spans:
+                        if run_end <= start or run_start >= end:
+                            continue
+                        piece = self._element_text(element)[max(start, run_start) - run_start : min(end, run_end) - run_start]
+                        if piece:
+                            self._append_run(target, element, piece, {})
+                else:
+                    self._append_run(target, None, text[start:end], fallback_run_format)
 
             elif segment["type"] == "variable":
-                for item in segment["content"]:
+                for item in segment["items"]:
                     item_type = item.get("type")
 
                     if item_type == "text":
-                        if current_paragraph is None:
-                            current_paragraph = self._insert_paragraph_at_index(
-                                parent, current_insert_index, original_format
-                            )
-                            current_insert_index += 1
-
-                        run = current_paragraph.add_run(item["content"])
-                        self._apply_run_format(run, item)
+                        target = ensure_paragraph()
+                        self._append_run(target, segment["template"], item["content"], item)
 
                     elif item_type in ["table", "image", "heading"]:
                         if current_paragraph is not None and current_paragraph.text.strip():
@@ -302,6 +357,18 @@ def _replace_paragraph_placeholders(
                         current_insert_index += 1
                         current_paragraph = None
 
+    @staticmethod
+    def _run_element_at(run_spans, position: int):
+        """The run element covering `position`, or None when spans are unusable."""
+        for run_start, run_end, element in run_spans:
+            if run_start <= position < run_end:
+                return element
+        return None
+
+    @staticmethod
+    def _element_text(element) -> str:
+        return Run(element, None).text
+
     def _extract_paragraph_format(self, paragraph: Paragraph) -> dict[str, Any]:
         return {
             "alignment": paragraph.alignment,
@@ -326,8 +393,14 @@ def _extract_run_format(self, run) -> dict[str, Any]:
             "font_color": run.font.color.rgb if run.font.color.rgb else None,
         }
 
-    def _insert_paragraph_at_index(self, parent, index: int, format_dict: dict[str, Any]) -> Paragraph:
+    def _insert_paragraph_at_index(
+        self, parent, index: int, format_dict: dict[str, Any], pPr_template=None
+    ) -> Paragraph:
         p_element = OxmlElement("w:p")
+        if pPr_template is not None:
+            # Carrying the whole pPr keeps the style, numbering and borders that
+            # the seven copied properties below cannot express.
+            p_element.append(copy.deepcopy(pPr_template))
         parent.insert(index, p_element)
         paragraph = Paragraph(p_element, self.doc)
 
diff --git a/src/backend/scripts/README.md b/src/backend/scripts/README.md
index 5d7d29eeaf..b6ce98123e 100644
--- a/src/backend/scripts/README.md
+++ b/src/backend/scripts/README.md
@@ -51,12 +51,16 @@ explicit Store scan to report missing and orphan direct `visible` tuple keys:
 
 ```bash
 PYTHONPATH=./ .venv/bin/python scripts/reconcile_f048_visible_projection.py \
-  --audit-orphan-tuples
+  --audit-orphan-tuples \
+  --orphan-object folder:97394
 ```
 
 The audit treats the union of canonical Grant sources and every ACTIVE SQL
 visible-source contribution as supported, so non-Grant system/resource sources
-are not classified as orphans. It prints exact tuple keys and stable checksums.
+are not classified as orphans. With `--orphan-object`, it uses an exact OpenFGA
+Read for those resources and prints the scoped anomaly set plus a stable cleanup
+checksum. Repeat the flag to select more than one reviewed resource. Omit it to
+scan the whole Store and select every audited orphan.
 
 For apply, stop ingress traffic and all API/Worker/Linsight processes, wait for
 their F048 heartbeat TTL to expire, and copy the dry-run `store_id` into the
@@ -91,18 +95,21 @@ dry-run:
 PYTHONPATH=./ .venv/bin/python scripts/reconcile_f048_visible_projection.py \
   --apply \
   --audit-orphan-tuples \
+  --orphan-object folder:97394 \
   --cleanup-orphan-tuples \
   --confirm-orphan-checksum <orphan-tuple-checksum> \
   --confirm-store-id <store-id> \
   --operator-id <operator-user-id>
 ```
 
-Cleanup is refused if the reviewed orphan set changes. Each resource is fenced
-by its current permission version, each exact delete is recorded as a
+Cleanup is an independent operation: it does not backfill or retire SQL source
+rows and does not publish an Authorization Model/Catalog release. It is refused
+if the selected orphan set changes. Each resource is fenced by its current
+permission version, each exact delete is recorded as a
 `VISIBLE_ORPHAN_CLEANUP` projection operation, and higher-consistency reads must
-confirm that no orphan direct tuple remains. Effective `visible` checks can
-still be true through an inherited parent; the audit verifies exact direct
-tuple presence instead.
+confirm that no selected orphan direct tuple remains. Other unselected anomalies
+are untouched. Effective `visible` checks can still be true through an inherited
+parent; the audit verifies exact direct tuple presence instead.
 Restart all permission-using processes after success; they discover the latest
 model through the stable Store name and validate the new SQL CURRENT Catalog
 pin.
diff --git a/src/backend/scripts/reconcile_f048_visible_projection.py b/src/backend/scripts/reconcile_f048_visible_projection.py
index e8a83e9f93..09603d8c6f 100644
--- a/src/backend/scripts/reconcile_f048_visible_projection.py
+++ b/src/backend/scripts/reconcile_f048_visible_projection.py
@@ -15,7 +15,8 @@
     PYTHONPATH=./ .venv/bin/python scripts/reconcile_f048_visible_projection.py
     PYTHONPATH=./ .venv/bin/python scripts/reconcile_f048_visible_projection.py --audit-orphan-tuples
     PYTHONPATH=./ .venv/bin/python scripts/reconcile_f048_visible_projection.py --apply --confirm-store-id <store-id> --operator-id <user-id> --allow-model-upgrade
-    PYTHONPATH=./ .venv/bin/python scripts/reconcile_f048_visible_projection.py --apply --audit-orphan-tuples --cleanup-orphan-tuples --confirm-orphan-checksum <checksum> --confirm-store-id <store-id> --operator-id <user-id>
+    PYTHONPATH=./ .venv/bin/python scripts/reconcile_f048_visible_projection.py --audit-orphan-tuples --orphan-object folder:97394
+    PYTHONPATH=./ .venv/bin/python scripts/reconcile_f048_visible_projection.py --apply --audit-orphan-tuples --orphan-object folder:97394 --cleanup-orphan-tuples --confirm-orphan-checksum <checksum> --confirm-store-id <store-id> --operator-id <user-id>
 
 Dry-run is the default.  Apply refuses active runtime heartbeats or in-flight
 permission projection operations. Orphan cleanup deletes only exact direct
@@ -158,6 +159,7 @@ class ReconcileReport:
 
 @dataclass(frozen=True, slots=True)
 class OrphanTupleAudit:
+    object_filters: tuple[str, ...]
     live_direct_visible_count: int
     supported_tuple_count: int
     missing_tuple_count: int
@@ -168,6 +170,14 @@ class OrphanTupleAudit:
     orphan_tuples: tuple[tuple[str, str, str], ...]
 
 
+@dataclass(frozen=True, slots=True)
+class OrphanCleanupSelection:
+    object_filters: tuple[str, ...]
+    tuple_count: int
+    tuple_checksum: str
+    tuples: tuple[tuple[str, str, str], ...]
+
+
 def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
     parser = argparse.ArgumentParser(
         description=__doc__,
@@ -210,6 +220,12 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
         action="store_true",
         help="With --apply, delete audited orphan tuple keys through resource-scoped projection operations",
     )
+    parser.add_argument(
+        "--orphan-object",
+        action="append",
+        default=[],
+        help="Limit the reported cleanup selection to an exact resource key; repeatable",
+    )
     parser.add_argument(
         "--confirm-orphan-checksum",
         default=None,
@@ -232,6 +248,8 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
         parser.error("--cleanup-orphan-tuples requires --confirm-orphan-checksum")
     if args.confirm_orphan_checksum and not args.cleanup_orphan_tuples:
         parser.error("--confirm-orphan-checksum requires --cleanup-orphan-tuples")
+    if args.orphan_object and not args.audit_orphan_tuples:
+        parser.error("--orphan-object requires --audit-orphan-tuples")
     return args
 
 
@@ -490,8 +508,23 @@ async def _audit_orphan_tuples(
     *,
     canonical_sources: tuple[Any, ...],
     persisted: tuple[PermissionVisibleSourceProjection, ...],
+    object_filters: tuple[str, ...] = (),
 ) -> OrphanTupleAudit:
-    rows = await client.read_tuples(consistency=HIGHER_CONSISTENCY)
+    normalized_filters = tuple(sorted(dict.fromkeys(object_filters)))
+    invalid = [value for value in normalized_filters if not all(value.partition(":"))]
+    _require(not invalid, f"invalid --orphan-object resource keys: {invalid}")
+    if normalized_filters:
+        rows = [
+            row
+            for object_key in normalized_filters
+            for row in await client.read_tuples(
+                relation="visible",
+                object=object_key,
+                consistency=HIGHER_CONSISTENCY,
+            )
+        ]
+    else:
+        rows = await client.read_tuples(consistency=HIGHER_CONSISTENCY)
     live = frozenset(
         (str(row["user"]), "visible", str(row["object"]))
         for row in rows
@@ -500,9 +533,12 @@ async def _audit_orphan_tuples(
     canonical = frozenset(_tuple_key(row) for row in canonical_sources)
     persisted_active = frozenset(_tuple_key(row) for row in persisted if row.state == "ACTIVE")
     supported = canonical | persisted_active
+    if normalized_filters:
+        supported = frozenset(row for row in supported if row[2] in normalized_filters)
     missing = tuple(sorted(supported - live))
     orphans = tuple(sorted(live - supported))
     return OrphanTupleAudit(
+        object_filters=normalized_filters,
         live_direct_visible_count=len(live),
         supported_tuple_count=len(supported),
         missing_tuple_count=len(missing),
@@ -514,6 +550,21 @@ async def _audit_orphan_tuples(
     )
 
 
+def _select_orphan_tuples(
+    audit: OrphanTupleAudit,
+    *,
+    object_filters: tuple[str, ...],
+) -> OrphanCleanupSelection:
+    normalized_filters = tuple(sorted(dict.fromkeys(object_filters)))
+    selected = tuple(row for row in audit.orphan_tuples if not normalized_filters or row[2] in normalized_filters)
+    return OrphanCleanupSelection(
+        object_filters=normalized_filters,
+        tuple_count=len(selected),
+        tuple_checksum=_checksum(selected),
+        tuples=selected,
+    )
+
+
 async def _load_cleanup_scope(object_key: str) -> ResourcePermissionMode:
     resource_type, separator, resource_id = object_key.partition(":")
     _require(
@@ -598,13 +649,13 @@ async def _cleanup_orphan_tuples(
     client: FGAClient,
     *,
     current: CurrentRelease,
-    audit: OrphanTupleAudit,
+    selection: OrphanCleanupSelection,
     operator_id: int,
 ) -> tuple[int, ...]:
-    if not audit.orphan_tuples:
+    if not selection.tuples:
         return ()
     grouped: dict[str, list[tuple[str, str, str]]] = defaultdict(list)
-    for tuple_key in audit.orphan_tuples:
+    for tuple_key in selection.tuples:
         grouped[tuple_key[2]].append(tuple_key)
 
     sql_projection = await build_sql_projection_runtime(client)
@@ -633,7 +684,7 @@ async def _cleanup_orphan_tuples(
             resource_version=plan.expected_version,
             parent_type=scope.parent_type,
             parent_id=scope.parent_id,
-            context_version=f"visible-orphan-{audit.orphan_tuple_checksum[:40]}",
+            context_version=f"visible-orphan-{selection.tuple_checksum[:40]}",
         )
         with bypass_tenant_filter():
             operation = await projection.prepare(plan)
@@ -921,11 +972,13 @@ async def execute(args: argparse.Namespace, *, live_settings: Any = settings) ->
         )
         print(json.dumps(asdict(report), ensure_ascii=False, sort_keys=True))
         orphan_audit: OrphanTupleAudit | None = None
+        cleanup_selection: OrphanCleanupSelection | None = None
         if args.audit_orphan_tuples:
             orphan_audit = await _audit_orphan_tuples(
                 source_client,
                 canonical_sources=canonical_sources,
                 persisted=persisted,
+                object_filters=tuple(args.orphan_object),
             )
             print(
                 json.dumps(
@@ -934,16 +987,62 @@ async def execute(args: argparse.Namespace, *, live_settings: Any = settings) ->
                     sort_keys=True,
                 )
             )
+            cleanup_selection = _select_orphan_tuples(
+                orphan_audit,
+                object_filters=tuple(args.orphan_object),
+            )
+            print(
+                json.dumps(
+                    {"event": "orphan_cleanup_selection", **asdict(cleanup_selection)},
+                    ensure_ascii=False,
+                    sort_keys=True,
+                )
+            )
         if not args.apply:
             print("[dry-run] no SQL, OpenFGA, Authorization Model, or Catalog mutations were requested")
             return EXIT_OK
 
         if args.cleanup_orphan_tuples:
-            _require(orphan_audit is not None, "orphan cleanup requires a completed orphan audit")
+            _require(cleanup_selection is not None, "orphan cleanup requires a completed orphan selection")
+            _require(not current.write_fenced, "CURRENT Catalog is write fenced")
+            _require(
+                args.confirm_orphan_checksum == cleanup_selection.tuple_checksum,
+                "--confirm-orphan-checksum does not match the selected orphan tuple set",
+            )
+            cleanup_operation_ids = await _cleanup_orphan_tuples(
+                source_client,
+                current=current,
+                selection=cleanup_selection,
+                operator_id=args.operator_id,
+            )
+            persisted_after = await _load_persisted_sources()
+            audit_after = await _audit_orphan_tuples(
+                source_client,
+                canonical_sources=canonical_sources,
+                persisted=persisted_after,
+                object_filters=tuple(args.orphan_object),
+            )
+            remaining_selected = tuple(sorted(set(cleanup_selection.tuples) & set(audit_after.orphan_tuples)))
             _require(
-                args.confirm_orphan_checksum == orphan_audit.orphan_tuple_checksum,
-                "--confirm-orphan-checksum does not match the audited orphan tuple set",
+                not remaining_selected,
+                f"{len(remaining_selected)} selected orphan visible tuples remain after cleanup",
             )
+            print(
+                json.dumps(
+                    {
+                        "event": "orphan_tuple_cleanup",
+                        "deleted_tuple_count": cleanup_selection.tuple_count,
+                        "operation_ids": cleanup_operation_ids,
+                        "orphan_tuple_checksum": cleanup_selection.tuple_checksum,
+                        "remaining_audited_orphan_tuple_count": audit_after.orphan_tuple_count,
+                        "remaining_selected_orphan_tuple_count": len(remaining_selected),
+                    },
+                    ensure_ascii=False,
+                    sort_keys=True,
+                )
+            )
+            return EXIT_OK
+
         _require(
             not retires,
             f"{len(retires)} stale Grant source projections require classified removal; no writes applied",
@@ -1001,38 +1100,6 @@ async def execute(args: argparse.Namespace, *, live_settings: Any = settings) ->
             "final CURRENT Catalog/model pin verification failed",
         )
         await _retire_other_active_models(final)
-        cleanup_operation_ids: tuple[int, ...] = ()
-        if args.cleanup_orphan_tuples:
-            _require(orphan_audit is not None, "orphan cleanup audit disappeared")
-            cleanup_operation_ids = await _cleanup_orphan_tuples(
-                target_client,
-                current=final,
-                audit=orphan_audit,
-                operator_id=args.operator_id,
-            )
-            persisted_after = await _load_persisted_sources()
-            audit_after = await _audit_orphan_tuples(
-                target_client,
-                canonical_sources=canonical_sources,
-                persisted=persisted_after,
-            )
-            _require(
-                audit_after.orphan_tuple_count == 0,
-                f"{audit_after.orphan_tuple_count} orphan visible tuples remain after cleanup",
-            )
-            print(
-                json.dumps(
-                    {
-                        "event": "orphan_tuple_cleanup",
-                        "deleted_tuple_count": orphan_audit.orphan_tuple_count,
-                        "operation_ids": cleanup_operation_ids,
-                        "orphan_tuple_checksum": orphan_audit.orphan_tuple_checksum,
-                        "remaining_orphan_tuple_count": audit_after.orphan_tuple_count,
-                    },
-                    ensure_ascii=False,
-                    sort_keys=True,
-                )
-            )
         print(
             json.dumps(
                 {
@@ -1045,11 +1112,6 @@ async def execute(args: argparse.Namespace, *, live_settings: Any = settings) ->
                     "source_retires": len(retires),
                     "visible_tuples_ensured": len(expected),
                     "visible_tuples_verified": len(expected),
-                    "orphan_tuples_deleted": (
-                        orphan_audit.orphan_tuple_count
-                        if args.cleanup_orphan_tuples and orphan_audit is not None
-                        else 0
-                    ),
                 },
                 ensure_ascii=False,
                 sort_keys=True,
diff --git a/src/backend/test/permission/test_f048_decision_api.py b/src/backend/test/permission/test_f048_decision_api.py
index e73a397679..8aedbee0da 100644
--- a/src/backend/test/permission/test_f048_decision_api.py
+++ b/src/backend/test/permission/test_f048_decision_api.py
@@ -53,7 +53,7 @@ async def check_action(self, actor, target, action):
         return self.allowed
 
 
-def _app(decision_api) -> FastAPI:
+def _app(decision_api, *, super_admin: bool = False) -> FastAPI:
     app = FastAPI()
     app.include_router(router, prefix="/api/v1/permissions")
     app.dependency_overrides[get_permission_decision_api] = lambda: decision_api
@@ -62,6 +62,7 @@ def _app(decision_api) -> FastAPI:
         user_name="member",
         user_role=[],
         tenant_id=5,
+        is_global_super=super_admin,
     )
     return app
 
@@ -106,6 +107,34 @@ def test_concrete_action_check_returns_true_and_normal_false_as_200() -> None:
         assert coordinator.targets[0].context_version == "business-v7"
 
 
+def test_super_admin_visible_check_short_circuits_only_the_http_decision() -> None:
+    decision, business, coordinator = _decision(allowed=False)
+    with TestClient(_app(decision, super_admin=True)) as client:
+        visible = client.post(
+            "/api/v1/permissions/check",
+            json={
+                "resource_type": "knowledge_library",
+                "resource_id": "4192",
+                "action": "visible",
+            },
+        )
+        edit = client.post(
+            "/api/v1/permissions/check",
+            json={
+                "resource_type": "knowledge_file",
+                "resource_id": "file-1",
+                "action": "edit",
+            },
+        )
+
+    assert visible.status_code == 200
+    assert visible.json()["data"] == {"allowed": True}
+    assert edit.status_code == 200
+    assert edit.json()["data"] == {"allowed": False}
+    assert [call["action"] for call in business.calls] == ["edit"]
+    assert len(coordinator.targets) == 1
+
+
 def test_client_cannot_forge_verified_target_fields_or_legacy_aliases() -> None:
     decision, business, _ = _decision(allowed=True)
     forged_payloads = (
diff --git a/src/backend/test/permission/test_f048_projection_sql_runtime.py b/src/backend/test/permission/test_f048_projection_sql_runtime.py
index 7b4bee10bf..ddb765bda8 100644
--- a/src/backend/test/permission/test_f048_projection_sql_runtime.py
+++ b/src/backend/test/permission/test_f048_projection_sql_runtime.py
@@ -510,6 +510,115 @@ async def test_assignee_move_preserves_identity_and_advances_version(
                 assert moved.version == 2
 
 
+@pytest.mark.asyncio
+async def test_assignee_move_replaces_inactive_target_identity(
+    session_factory,
+) -> None:
+    source = GrantSourceService().canonicalize_source(
+        source_id=101,
+        subject_type="user",
+        subject_id="11",
+        source_type="DIRECT",
+    )
+    with bypass_tenant_filter():
+        async with session_factory() as session:
+            async with session.begin():
+                old_grant = PermissionGrant(
+                    tenant_id=7,
+                    resource_type="folder",
+                    resource_id="42",
+                    model_key="viewer",
+                    state="ACTIVE",
+                    projection_state="CURRENT",
+                )
+                target_grant = PermissionGrant(
+                    tenant_id=7,
+                    resource_type="folder",
+                    resource_id="42",
+                    model_key="editor",
+                    state="ACTIVE",
+                    projection_state="CURRENT",
+                )
+                session.add_all((old_grant, target_grant))
+                await session.flush()
+                session.add_all(
+                    (
+                        PermissionGrantAssignee(
+                            id=source.source_id,
+                            tenant_id=7,
+                            grant_id=int(old_grant.id),
+                            subject_type=source.subject_type,
+                            subject_id=source.subject_id,
+                            userset_relation=source.userset_relation,
+                            include_children=source.include_children,
+                            source_type=source.source_type,
+                            source_ref=source.source_ref,
+                            source_locator=source.source_locator,
+                            source_fingerprint=source.source_fingerprint,
+                            projected_subject=source.projected_subject,
+                            protected=source.protected,
+                            state="ACTIVE",
+                            version=1,
+                        ),
+                        PermissionGrantAssignee(
+                            id=202,
+                            tenant_id=7,
+                            grant_id=int(target_grant.id),
+                            subject_type=source.subject_type,
+                            subject_id=source.subject_id,
+                            userset_relation=source.userset_relation,
+                            include_children=source.include_children,
+                            source_type=source.source_type,
+                            source_ref=source.source_ref,
+                            source_locator=source.source_locator,
+                            source_fingerprint=source.source_fingerprint,
+                            projected_subject=source.projected_subject,
+                            protected=source.protected,
+                            state="INACTIVE",
+                            version=4,
+                        ),
+                        PermissionVisibleSourceProjection(
+                            tenant_id=7,
+                            resource_type="folder",
+                            resource_id="42",
+                            visibility_class="ordinary",
+                            projected_subject=source.projected_subject,
+                            source_kind="GRANT_ASSIGNEE",
+                            source_owner_key="grant_assignee:202",
+                            source_locator=source.source_locator,
+                            source_fingerprint=source.source_fingerprint,
+                            contribution_fingerprint="8" * 64,
+                            model_key="editor",
+                            source_version=4,
+                            tuple_fingerprint="9" * 64,
+                            state="RETIRED",
+                        ),
+                    )
+                )
+                await session.flush()
+
+                moved = await SqlPermissionControlState._upsert_assignee(
+                    session,
+                    grant_row=target_grant,
+                    source=replace(source, version=2),
+                    state="PENDING",
+                )
+
+                assert moved.id == 101
+                assert moved.grant_id == target_grant.id
+                assert moved.version == 2
+
+            rows = list(
+                (await session.execute(select(PermissionGrantAssignee).order_by(PermissionGrantAssignee.id)))
+                .scalars()
+                .all()
+            )
+
+    assert [(row.id, row.grant_id, row.state, row.version) for row in rows] == [
+        (101, target_grant.id, "PENDING", 2),
+    ]
+
+
 @pytest.mark.asyncio
 async def test_visible_source_after_state_is_frozen_then_finalized(
     session_factory,
diff --git a/src/backend/test/permission/test_f048_resource_api.py b/src/backend/test/permission/test_f048_resource_api.py
index 8e5b922807..4dcd79f9e1 100644
--- a/src/backend/test/permission/test_f048_resource_api.py
+++ b/src/backend/test/permission/test_f048_resource_api.py
@@ -198,6 +198,14 @@ async def current_mode(self, target):
         del target
         return SimpleNamespace(mode="CUSTOM", projection_state="READY")
 
+    async def mode_for_target(self, target):
+        del target
+        return SimpleNamespace(
+            mode="CUSTOM",
+            projection_state="CURRENT",
+            version=3,
+        )
+
     async def effective_actions(self, resource_type):
         del resource_type
         # 'visible' is a base relation, not a registered action, so it never
@@ -311,6 +319,7 @@ async def test_super_admin_my_permissions_returns_full_effective_actions() -> No
     # would be empty; the full effective action set is reported instead.
     assert result["actions"] == ["use", "edit", "delete", "manage_permission"]
     assert result["sources"] == []
+    assert result["projection_degraded"] is False
     assert runtime.visible_checks == 0
 
 
@@ -337,6 +346,60 @@ async def test_ordinary_user_my_permissions_stays_grant_derived() -> None:
     assert result["actions"] == []
 
 
+class _DegradedPermissionRuntime(_ExplainRuntime):
+    def __init__(self) -> None:
+        super().__init__()
+        self.explain_calls = 0
+
+    async def mode_for_target(self, target):
+        del target
+        return SimpleNamespace(
+            mode="CUSTOM",
+            projection_state="FAILED_CLOSED",
+            version=3,
+        )
+
+    async def check_action(self, actor, target, action):
+        del actor, target
+        if action == "visible":
+            self.visible_checks += 1
+            return True
+        return action in {"use", "edit"}
+
+    async def explain_permissions(self, **kwargs):
+        del kwargs
+        self.explain_calls += 1
+        raise AssertionError("degraded projection must not read staged grant explanations")
+
+
+@pytest.mark.asyncio
+async def test_degraded_my_permissions_uses_openfga_actions_without_staged_sources() -> None:
+    runtime = _DegradedPermissionRuntime()
+    api = F048ResourcePermissionApi(
+        resources=_Resources(),
+        runtime=runtime,
+        subjects=_Subjects(),
+    )
+    actor = PermissionActor(user_id=7, current_tenant_id=9)
+
+    result = await api.get_my_permissions(
+        resource_type="workflow",
+        resource_id="wf-1",
+        actor=actor,
+    )
+
+    assert result == {
+        "mode": "CUSTOM",
+        "actions": ["use", "edit"],
+        "sources": [],
+        "roster_complete": False,
+        "projection_state": "FAILED_CLOSED",
+        "projection_degraded": True,
+    }
+    assert runtime.visible_checks == 1
+    assert runtime.explain_calls == 0
+
+
 @pytest.mark.asyncio
 async def test_roster_uses_bounded_sql_page_instead_of_full_explanation() -> None:
     runtime = _Runtime()
diff --git a/src/backend/test/permission/test_f048_visible_reconcile_cli.py b/src/backend/test/permission/test_f048_visible_reconcile_cli.py
index de69868ff2..8a926d117f 100644
--- a/src/backend/test/permission/test_f048_visible_reconcile_cli.py
+++ b/src/backend/test/permission/test_f048_visible_reconcile_cli.py
@@ -74,6 +74,8 @@ def test_parse_defaults_to_dry_run_and_apply_requires_store_confirmation() -> No
             "--operator-id",
             "7",
             "--audit-orphan-tuples",
+            "--orphan-object",
+            "folder:97394",
             "--cleanup-orphan-tuples",
             "--confirm-orphan-checksum",
             "a" * 64,
@@ -81,6 +83,7 @@ def test_parse_defaults_to_dry_run_and_apply_requires_store_confirmation() -> No
     )
     assert args.cleanup_orphan_tuples is True
     assert args.confirm_orphan_checksum == "a" * 64
+    assert args.orphan_object == ["folder:97394"]
 
 
 def test_report_deduplicates_only_the_same_projected_subject_tuple() -> None:
@@ -143,9 +146,14 @@ async def batch_check(self, checks, consistency=None):
         self.checks.append((checks, consistency))
         return [True] * len(checks)
 
-    async def read_tuples(self, consistency=None):
+    async def read_tuples(self, user=None, relation=None, object=None, consistency=None):
+        del user
         assert consistency == cli.HIGHER_CONSISTENCY
-        return self.tuples
+        return [
+            row
+            for row in self.tuples
+            if (relation is None or row["relation"] == relation) and (object is None or row["object"] == object)
+        ]
 
 
 @pytest.mark.asyncio
@@ -201,11 +209,76 @@ async def test_orphan_audit_uses_canonical_and_all_active_persisted_sources() ->
     )
 
     assert audit.live_direct_visible_count == 3
+    assert audit.object_filters == ()
     assert audit.supported_tuple_count == 2
     assert audit.missing_tuple_count == 0
     assert audit.orphan_tuples == (("user:9", "visible", "knowledge_space:42"),)
 
 
+def test_orphan_cleanup_selection_can_limit_one_reviewed_resource() -> None:
+    tuples = (
+        ("user:7", "visible", "folder:97394"),
+        ("user:8", "visible", "knowledge_space:4166"),
+    )
+    audit = cli.OrphanTupleAudit(
+        object_filters=(),
+        live_direct_visible_count=2,
+        supported_tuple_count=0,
+        missing_tuple_count=0,
+        orphan_tuple_count=2,
+        missing_tuple_checksum=cli._checksum(()),
+        orphan_tuple_checksum=cli._checksum(tuples),
+        missing_tuples=(),
+        orphan_tuples=tuples,
+    )
+
+    selection = cli._select_orphan_tuples(
+        audit,
+        object_filters=("folder:97394",),
+    )
+
+    assert selection.object_filters == ("folder:97394",)
+    assert selection.tuple_count == 1
+    assert selection.tuples == (("user:7", "visible", "folder:97394"),)
+    assert selection.tuple_checksum == cli._checksum(selection.tuples)
+
+    clean_selection = cli._select_orphan_tuples(
+        cli.OrphanTupleAudit(
+            object_filters=("folder:97394",),
+            live_direct_visible_count=1,
+            supported_tuple_count=1,
+            missing_tuple_count=0,
+            orphan_tuple_count=0,
+            missing_tuple_checksum=cli._checksum(()),
+            orphan_tuple_checksum=cli._checksum(()),
+            missing_tuples=(),
+            orphan_tuples=(),
+        ),
+        object_filters=("folder:97394",),
+    )
+    assert clean_selection.tuple_count == 0
+    assert clean_selection.tuples == ()
+
+
+@pytest.mark.asyncio
+async def test_orphan_audit_reads_only_selected_resource() -> None:
+    client = _FGAClient()
+    client.tuples = [
+        {"user": "user:7", "relation": "visible", "object": "folder:97394"},
+        {"user": "user:8", "relation": "visible", "object": "knowledge_space:4166"},
+    ]
+
+    audit = await cli._audit_orphan_tuples(
+        client,
+        canonical_sources=(),
+        persisted=(),
+        object_filters=("folder:97394",),
+    )
+
+    assert audit.object_filters == ("folder:97394",)
+    assert audit.orphan_tuples == (("user:7", "visible", "folder:97394"),)
+
+
 def test_build_orphan_cleanup_plan_is_exact_and_resource_fenced() -> None:
     current = cli.CurrentRelease(
         catalog_id=1,
diff --git a/src/backend/test/permission/test_stale_projection_fail_soft.py b/src/backend/test/permission/test_stale_projection_fail_soft.py
index ec30d64f3e..eeca68c265 100644
--- a/src/backend/test/permission/test_stale_projection_fail_soft.py
+++ b/src/backend/test/permission/test_stale_projection_fail_soft.py
@@ -2,11 +2,17 @@
 
 from __future__ import annotations
 
+from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 
-from bisheng.common.errcode.permission import PermissionPublishNotReadyError
+from bisheng.common.errcode.permission import (
+    PermissionPublishNotReadyError,
+    PermissionVersionConflictError,
+)
+from bisheng.permission.application import control_state as control_state_module
+from bisheng.permission.application.control_state import SqlPermissionControlState
 from bisheng.permission.application.sql_runtime import SqlPermissionScopeFence
 from bisheng.permission.domain.schemas.f048 import VerifiedPermissionTarget
 from bisheng.permission.domain.services.permission_action_service import (
@@ -188,7 +194,152 @@ async def test_ensure_readable_error_carries_diagnostic_fields():
     assert exc.kwargs.get("expected_parent_type") == "knowledge_space"
     assert exc.kwargs.get("expected_parent_id") == "3377"
     assert exc.kwargs.get("expected_version") == 1
-    assert exc.kwargs.get("expected_projection_state") == "CURRENT"
+    assert exc.kwargs.get("expected_projection_state") == "CURRENT|PROJECTING|FAILED_CLOSED"
+
+
+@pytest.mark.parametrize(
+    ("projection_state", "stored_version", "requires_higher_consistency"),
+    (
+        ("CURRENT", 1, False),
+        ("CURRENT", 2, True),
+        ("PROJECTING", 1, True),
+        ("FAILED_CLOSED", 1, True),
+    ),
+)
+async def test_decision_fence_keeps_non_current_projection_readable(
+    projection_state: str,
+    stored_version: int,
+    requires_higher_consistency: bool,
+):
+    from unittest.mock import patch
+
+    target = _make_target(resource_version=1)
+    row = SimpleNamespace(
+        version=stored_version,
+        projection_state=projection_state,
+        parent_type=target.parent_type,
+        parent_id=target.parent_id,
+    )
+    mock_session = AsyncMock()
+    mock_result = MagicMock()
+    mock_result.scalars.return_value.first.return_value = row
+    mock_session.execute = AsyncMock(return_value=mock_result)
+    mock_ctx = AsyncMock()
+    mock_ctx.__aenter__ = AsyncMock(return_value=mock_session)
+    mock_ctx.__aexit__ = AsyncMock(return_value=None)
+
+    with patch(
+        "bisheng.permission.application.sql_runtime.get_async_db_session",
+        return_value=mock_ctx,
+    ):
+        assert await SqlPermissionScopeFence().ensure_readable(target) is requires_higher_consistency
+
+
+async def test_decision_fence_still_rejects_pending_projection():
+    from unittest.mock import patch
+
+    target = _make_target(resource_version=1)
+    row = SimpleNamespace(
+        version=1,
+        projection_state="PENDING",
+        parent_type=target.parent_type,
+        parent_id=target.parent_id,
+    )
+    mock_session = AsyncMock()
+    mock_result = MagicMock()
+    mock_result.scalars.return_value.first.return_value = row
+    mock_session.execute = AsyncMock(return_value=mock_result)
+    mock_ctx = AsyncMock()
+    mock_ctx.__aenter__ = AsyncMock(return_value=mock_session)
+    mock_ctx.__aexit__ = AsyncMock(return_value=None)
+
+    with (
+        patch(
+            "bisheng.permission.application.sql_runtime.get_async_db_session",
+            return_value=mock_ctx,
+        ),
+        pytest.raises(PermissionPublishNotReadyError),
+    ):
+        await SqlPermissionScopeFence().ensure_readable(target)
+
+
+@pytest.mark.parametrize("projection_state", ("CURRENT", "PROJECTING", "FAILED_CLOSED"))
+async def test_permission_version_exposes_decidable_projection_state(projection_state: str):
+    from unittest.mock import patch
+
+    row = SimpleNamespace(
+        version=7,
+        mode="CUSTOM",
+        parent_type=None,
+        parent_id=None,
+        projection_state=projection_state,
+    )
+    mock_session = AsyncMock()
+    mock_result = MagicMock()
+    mock_result.scalars.return_value.first.return_value = row
+    mock_session.execute = AsyncMock(return_value=mock_result)
+    mock_ctx = AsyncMock()
+    mock_ctx.__aenter__ = AsyncMock(return_value=mock_session)
+    mock_ctx.__aexit__ = AsyncMock(return_value=None)
+
+    with patch.object(
+        control_state_module,
+        "get_async_db_session",
+        return_value=mock_ctx,
+    ):
+        version, context = await SqlPermissionControlState().permission_version(
+            tenant_id=1,
+            resource_type="knowledge_space",
+            resource_id="4166",
+        )
+
+    assert version == 7
+    assert context.endswith(projection_state)
+
+
+async def test_non_current_decision_forces_higher_consistency():
+    scope_fence = AsyncMock()
+    scope_fence.ensure_readable = AsyncMock(return_value=True)
+    fga = AsyncMock()
+    fga.check = AsyncMock(return_value=True)
+    service = _make_service(scope_fence=scope_fence, fga=fga)
+
+    assert await service.check_action(_make_actor(), _make_target(), "download")
+    assert fga.check.await_args.kwargs["consistency"] == "HIGHER_CONSISTENCY"
+
+
+async def test_non_current_visible_and_batch_decisions_force_higher_consistency():
+    scope_fence = AsyncMock()
+    scope_fence.ensure_readable = AsyncMock(return_value=True)
+    fga = AsyncMock()
+    fga.check = AsyncMock(return_value=True)
+    fga.batch_check = AsyncMock(return_value=[True, True])
+    service = _make_service(scope_fence=scope_fence, fga=fga)
+    actor = _make_actor()
+    targets = (_make_target(resource_id="1"), _make_target(resource_id="2"))
+
+    assert await service.check_visible(actor, targets[0])
+    assert fga.check.await_args.kwargs["consistency"] == "HIGHER_CONSISTENCY"
+    assert await service.batch_check_actions(actor, targets, "download") == (True, True)
+    assert fga.batch_check.await_args.kwargs["consistency"] == "HIGHER_CONSISTENCY"
+    assert await service.batch_check_visible(actor, targets) == (True, True)
+    assert fga.batch_check.await_args.kwargs["consistency"] == "HIGHER_CONSISTENCY"
+
+
+async def test_failed_closed_resource_still_rejects_new_permission_operation():
+    row = SimpleNamespace(
+        version=3,
+        projection_state="FAILED_CLOSED",
+        operation_id=405,
+    )
+
+    with pytest.raises(PermissionVersionConflictError):
+        SqlPermissionControlState._claim_projection_operation(
+            row,
+            expected_version=3,
+            operation_id=406,
+            allowed_initial_states=("CURRENT",),
+        )
 
 
 # ── P1: reconciler repairs root-parent mismatch ────────────────────────────
diff --git a/src/backend/test/workflow/nodes/test_docx_replace_formatting.py b/src/backend/test/workflow/nodes/test_docx_replace_formatting.py
index 542259d361..c21e884bd7 100644
--- a/src/backend/test/workflow/nodes/test_docx_replace_formatting.py
+++ b/src/backend/test/workflow/nodes/test_docx_replace_formatting.py
@@ -143,3 +143,56 @@ def build(doc):
     assert len(rendered.tables) == 1
     assert rendered.tables[0].rows[0].cells[0].text == "列1"
     assert any("见下表：" in paragraph.text for paragraph in rendered.paragraphs)
+
+
+def test_block_split_keeps_paragraph_style_and_run_formatting():
+    """Splitting around a heading must not flatten the surviving text."""
+
+    def build(doc):
+        paragraph = doc.add_paragraph(style="Quote")
+        paragraph.add_run("前缀 ").underline = True
+        paragraph.add_run(PLACEHOLDER).underline = True
+        paragraph.add_run(" 后缀____").underline = True
+
+    rendered = _render(
+        build,
+        {
+            KEY: [
+                {"type": "text", "content": "正文"},
+                {"type": "heading", "content": "小标题", "level": 3},
+                {"type": "text", "content": "结尾"},
+            ]
+        },
+    )
+    before, heading, after = rendered.paragraphs[:3]
+
+    assert before.style.name == "Quote"
+    assert after.style.name == "Quote"
+    assert heading.style.name == "Heading 3"
+    # Text on both sides of the heading keeps the underline it had, and so does
+    # the substituted value — it used to come out plain.
+    assert all(run.underline is True for run in before.runs if run.text)
+    assert all(run.underline is True for run in after.runs if run.text)
+    assert before.text == "前缀 正文"
+    assert after.text == "结尾 后缀____"
+
+
+def test_block_split_keeps_each_run_its_own_format():
+    """Mid-paragraph format changes used to be re-stamped with run[0]'s format."""
+
+    def build(doc):
+        paragraph = doc.add_paragraph()
+        paragraph.add_run("普通 ")
+        paragraph.add_run("加粗 ").bold = True
+        paragraph.add_run(PLACEHOLDER).underline = True
+
+    rendered = _render(
+        build,
+        {KEY: [{"type": "text", "content": "值"}, {"type": "heading", "content": "标题", "level": 2}]},
+    )
+    runs = [run for run in rendered.paragraphs[0].runs if run.text]
+
+    assert [run.text for run in runs] == ["普通 ", "加粗 ", "值"]
+    assert runs[0].bold is None
+    assert runs[1].bold is True
+    assert runs[2].underline is True
diff --git a/src/frontend/client/src/components/Chat/AiChatInput.tsx b/src/frontend/client/src/components/Chat/AiChatInput.tsx
index 8fad3a8fa8..f792cfc792 100644
--- a/src/frontend/client/src/components/Chat/AiChatInput.tsx
+++ b/src/frontend/client/src/components/Chat/AiChatInput.tsx
@@ -349,8 +349,15 @@ const AiChatInput = memo(
             const trimmed = text.trim();
             // Workbench: uploaded files require accompanying text before send.
             if (!trimmed || disabled || sendDisabled || isStreaming || isParsingMedia || fileUploading || filesParsing) return;
-            // Pass files through to parent
-            onSend(trimmed, chatFiles);
+            // Pass files through to parent. The local first-frame poster is a blob
+            // that this component revokes on the very next line, and it outranks
+            // the server cover in the message bubble — so it stops here, and the
+            // bubble goes back to `cover_filepath` once parsing produces it.
+            onSend(trimmed, chatFiles?.map((file) => (
+                file?.mediaCoverUrl?.startsWith('blob:')
+                    ? { ...file, mediaCoverUrl: undefined }
+                    : file
+            )) ?? chatFiles);
             setText("");
             setChatFiles(null);
             setUploadingFiles([]);
diff --git a/src/frontend/client/src/components/Chat/attachments/MediaAttachmentChip.tsx b/src/frontend/client/src/components/Chat/attachments/MediaAttachmentChip.tsx
index eebf703451..ff60360a50 100644
--- a/src/frontend/client/src/components/Chat/attachments/MediaAttachmentChip.tsx
+++ b/src/frontend/client/src/components/Chat/attachments/MediaAttachmentChip.tsx
@@ -1,5 +1,5 @@
 import { Loader2, Play, Video } from 'lucide-react';
-import { useState } from 'react';
+import { useEffect, useState } from 'react';
 import { useNavigate } from 'react-router-dom';
 import useLocalize from '~/hooks/useLocalize';
 import usePrefersMobileLayout from '~/hooks/usePrefersMobileLayout';
@@ -63,7 +63,15 @@ export function MediaAttachmentChip({
     const isUploading = !!file.isUploading;
     const isParsing = file.parsingState === 'parsing';
     const playbackUrl = resolveMediaPlaybackUrl(file);
-    const coverUrl = kind === 'video' ? resolveMediaCoverUrl(file) : undefined;
+    const resolvedCoverUrl = kind === 'video' ? resolveMediaCoverUrl(file) : undefined;
+    // A composer poster is a local blob that is revoked once the message is sent;
+    // the sent bubble then holds a dead URL until the server cover arrives. Fall
+    // back to the icon instead of rendering a broken image.
+    const [coverFailed, setCoverFailed] = useState(false);
+    useEffect(() => {
+        setCoverFailed(false);
+    }, [resolvedCoverUrl]);
+    const coverUrl = coverFailed ? undefined : resolvedCoverUrl;
     const mediaFilepath = extractMediaFilepath(file);
     const canPlay = !!playbackUrl && !isUploading;
     const parsingLabel = localize('com_chat.media_parsing');
@@ -118,7 +126,12 @@ export function MediaAttachmentChip({
                     }
                 >
                     {coverUrl ? (
-                        <img src={coverUrl} alt="" className="size-full object-cover" />
+                        <img
+                            src={coverUrl}
+                            alt=""
+                            className="size-full object-cover"
+                            onError={() => setCoverFailed(true)}
+                        />
                     ) : (
                         <>
                             <div className="flex size-full items-center justify-center text-[#999]">
diff --git a/src/frontend/client/src/pages/appChat/components/InputFiles.tsx b/src/frontend/client/src/pages/appChat/components/InputFiles.tsx
index ab45878261..6728a65e2f 100644
--- a/src/frontend/client/src/pages/appChat/components/InputFiles.tsx
+++ b/src/frontend/client/src/pages/appChat/components/InputFiles.tsx
@@ -26,6 +26,7 @@ import {
     TASK_MODE_MAX_FOLDER_FILES,
 } from "~/utils/folderUpload";
 import {
+    captureVideoPosterFromFile,
     getMediaKind,
     readMediaDurationFromFile,
     isMediaAttachmentFile,
@@ -160,7 +161,11 @@ const InputFiles = forwardRef(({ v, showVoice, accepts, disabled = false, size,
                 : undefined,
         previewUrl: f.previewUrl,
         mediaPreviewUrl: f.mediaPreviewUrl,
-        mediaCoverUrl: f.mediaCoverUrl,
+        // The composer poster is a local blob that is revoked the moment the
+        // message is sent. Handing it to the message would pin a dead URL that
+        // also outranks the server cover once that arrives, leaving the bubble
+        // with no thumbnail at all.
+        mediaCoverUrl: f.mediaCoverUrl?.startsWith('blob:') ? undefined : f.mediaCoverUrl,
         cover_filepath: f.cover_filepath,
         mediaDurationSec: f.mediaDurationSec,
     }));
@@ -289,6 +294,34 @@ const InputFiles = forwardRef(({ v, showVoice, accepts, disabled = false, size,
             });
         });
 
+        // Local first-frame poster for video chips. The server cover only exists
+        // after the turn is submitted and parsed, so without this the chip sits
+        // as a bare icon for the whole time the user is composing. Best effort:
+        // a codec the browser cannot decode just leaves the icon in place, and
+        // the server poster replaces this blob as soon as it lands.
+        filesWithProgress.forEach(({ file, id }: { file: File; id: string }) => {
+            if (!file || getMediaKind(file.name) !== 'video') {
+                return;
+            }
+            captureVideoPosterFromFile(file).then((mediaCoverUrl) => {
+                if (!mediaCoverUrl) {
+                    return;
+                }
+                const target = filesRef.current.find((f) => f.id === id);
+                // Removed while decoding, or the server poster won the race.
+                if (!target || target.cover_filepath) {
+                    URL.revokeObjectURL(mediaCoverUrl);
+                    return;
+                }
+                const updated = filesRef.current.map((f) =>
+                    f.id === id ? { ...f, mediaCoverUrl } : f,
+                );
+                filesRef.current = updated;
+                setFiles(updated);
+                onFilesStateChange?.(updated);
+            });
+        });
+
         // Keep track of the number of remaining uploads across concurrent batches.
         remainingUploadsRef.current += validFiles.length;
 
diff --git a/src/frontend/client/src/utils/mediaAttachmentUtils.ts b/src/frontend/client/src/utils/mediaAttachmentUtils.ts
index 8af68da0b8..bb3360d0af 100644
--- a/src/frontend/client/src/utils/mediaAttachmentUtils.ts
+++ b/src/frontend/client/src/utils/mediaAttachmentUtils.ts
@@ -72,7 +72,13 @@ export function readMediaDurationFromFile(file: File): Promise<number | undefine
     });
 }
 
-/** Capture a JPEG poster from the first decoded video frame for local chip preview. */
+/** How long to wait for the browser to decode a first frame before giving up.
+ *  A codec it cannot handle usually errors out at once, but some containers just
+ *  never fire an event — without this the promise would hang and leak the URL. */
+const VIDEO_POSTER_TIMEOUT_MS = 5000;
+
+/** Capture a JPEG poster from the first decoded video frame for local chip preview.
+ *  Best effort: resolves undefined when the browser cannot decode the file. */
 export function captureVideoPosterFromFile(file: File): Promise<string | undefined> {
     return new Promise((resolve) => {
         const url = URL.createObjectURL(file);
@@ -81,12 +87,25 @@ export function captureVideoPosterFromFile(file: File): Promise<string | undefin
         video.playsInline = true;
         video.preload = 'auto';
 
+        let settled = false;
+        const settle = (poster?: string) => {
+            if (settled) return;
+            settled = true;
+            window.clearTimeout(timeoutId);
+            resolve(poster);
+        };
+
         const cleanup = () => {
             URL.revokeObjectURL(url);
             video.removeAttribute('src');
             video.load();
         };
 
+        const timeoutId = window.setTimeout(() => {
+            cleanup();
+            settle(undefined);
+        }, VIDEO_POSTER_TIMEOUT_MS);
+
         video.onloadeddata = () => {
             video.currentTime = 0.001;
         };
@@ -96,7 +115,7 @@ export function captureVideoPosterFromFile(file: File): Promise<string | undefin
                 const height = video.videoHeight;
                 if (!width || !height) {
                     cleanup();
-                    resolve(undefined);
+                    settle(undefined);
                     return;
                 }
                 const canvas = document.createElement('canvas');
@@ -105,26 +124,26 @@ export function captureVideoPosterFromFile(file: File): Promise<string | undefin
                 const ctx = canvas.getContext('2d');
                 if (!ctx) {
                     cleanup();
-                    resolve(undefined);
+                    settle(undefined);
                     return;
                 }
                 ctx.drawImage(video, 0, 0, width, height);
                 canvas.toBlob(
                     (blob) => {
                         cleanup();
-                        resolve(blob ? URL.createObjectURL(blob) : undefined);
+                        settle(blob ? URL.createObjectURL(blob) : undefined);
                     },
                     'image/jpeg',
                     0.85,
                 );
             } catch {
                 cleanup();
-                resolve(undefined);
+                settle(undefined);
             }
         };
         video.onerror = () => {
             cleanup();
-            resolve(undefined);
+            settle(undefined);
         };
         video.src = url;
     });
diff --git a/src/frontend/packages/file-viewers/src/ExcelPreview.tsx b/src/frontend/packages/file-viewers/src/ExcelPreview.tsx
index fd605f122d..e6321c7eef 100644
--- a/src/frontend/packages/file-viewers/src/ExcelPreview.tsx
+++ b/src/frontend/packages/file-viewers/src/ExcelPreview.tsx
@@ -27,6 +27,31 @@ interface SheetCoordinateMaps {
   colMap: number[];
 }
 
+/**
+ * One download per URL while it is in flight. A preview that gets remounted (a
+ * re-rendering parent, a suspended boundary) would otherwise fire a fresh request
+ * for the same signed URL on every mount, which reads as a request storm against
+ * object storage.
+ */
+const inFlightDownloads = new Map<string, Promise<ArrayBuffer>>();
+
+function downloadOnce(url: string, onNotOk: (status: number) => Error): Promise<ArrayBuffer> {
+  const pending = inFlightDownloads.get(url);
+  if (pending) return pending;
+
+  const request = fetch(url)
+    .then((response) => {
+      if (!response.ok) throw onNotOk(response.status);
+      return response.arrayBuffer();
+    })
+    .finally(() => {
+      inFlightDownloads.delete(url);
+    });
+
+  inFlightDownloads.set(url, request);
+  return request;
+}
+
 function DefaultSpinner() {
   return (
     <svg className="size-20 animate-spin text-blue-500" viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
@@ -37,7 +62,15 @@ function DefaultSpinner() {
 }
 
 export function ExcelPreview({ filePath, fileExt: fileExtProp, loadingIcon }: ExcelPreviewProps) {
-  const { t } = useTranslation('shared', { keyPrefix: 'knowledge.excelPreview' });
+  // useSuspense: false — a suspending viewer gets torn down and remounted by the
+  // nearest boundary while the namespace settles, and every remount refetches the
+  // file. These labels are error text only, so resolving them a frame late is fine.
+  const { t } = useTranslation('shared', { keyPrefix: 'knowledge.excelPreview', useSuspense: false });
+  // The parse effect must not depend on `t`: react-i18next returns a new `t`
+  // whenever the namespace or language settles, and that identity change alone
+  // re-downloaded and re-parsed the whole workbook.
+  const tRef = useRef(t);
+  tRef.current = t;
 
   const [loading, setLoading] = useState(true);
   const [error, setError] = useState<string | null>(null);
@@ -117,15 +150,15 @@ export function ExcelPreview({ filePath, fileExt: fileExtProp, loadingIcon }: Ex
         setSheets([]);
         setActiveSheet('');
 
-        if (!filePath) throw new Error(t('filePathEmpty'));
+        if (!filePath) throw new Error(tRef.current('filePathEmpty'));
 
-        const response = await fetch(filePath);
-        if (!response.ok) throw new Error(`${t('fileLoadFailed')}: ${response.status}`);
-
-        const arrayBuffer = await response.arrayBuffer();
+        const arrayBuffer = await downloadOnce(
+          filePath,
+          (status) => new Error(`${tRef.current('fileLoadFailed')}: ${status}`),
+        );
 
         if (isCSV) {
-          if (arrayBuffer.byteLength === 0) throw new Error(t('fileContentEmpty'));
+          if (arrayBuffer.byteLength === 0) throw new Error(tRef.current('fileContentEmpty'));
 
           const uint8Array = new Uint8Array(arrayBuffer);
           let decodedStr = '';
@@ -155,7 +188,7 @@ export function ExcelPreview({ filePath, fileExt: fileExtProp, loadingIcon }: Ex
             wb = XLSX.read(arrayBuffer, { type: 'array' });
           } catch (e) {
             console.error('SheetJS parsing failed:', e);
-            throw new Error(t('excelParseFailed'));
+            throw new Error(tRef.current('excelParseFailed'));
           }
 
           const sheetNames = wb.SheetNames;
@@ -188,13 +221,13 @@ export function ExcelPreview({ filePath, fileExt: fileExtProp, loadingIcon }: Ex
             }
           }
         } else {
-          throw new Error(t('unsupportedType', { type: fileExt }));
+          throw new Error(tRef.current('unsupportedType', { type: fileExt }));
         }
 
         setError(null);
       } catch (err) {
         console.error('File parsing failed:', err);
-        setError(err instanceof Error ? err.message : t('unknownError'));
+        setError(err instanceof Error ? err.message : tRef.current('unknownError'));
       } finally {
         setLoading(false);
       }
@@ -204,9 +237,9 @@ export function ExcelPreview({ filePath, fileExt: fileExtProp, loadingIcon }: Ex
       fetchAndParseFile();
     } else {
       setLoading(false);
-      setError(t('filePathEmpty'));
+      setError(tRef.current('filePathEmpty'));
     }
-  }, [filePath, fileExt, isCSV, isXLSX, t]);
+  }, [filePath, fileExt, isCSV, isXLSX]);
 
   const renderContent = () => {
     const sheetData = excelData[activeSheet];