Feat/3.0.0 beta1

features/v3.0.0-beta1/048-rebac-permission-model-grants/design.md

@@ -1057,8 +1057,15 @@ higher consistency。Grant/mode/lifecycle command 仍只允许从 `CURRENT` clai
 `scripts/reconcile_f048_projection_operations.py --tenant-id <tenant> <operation...>`；默认
 dry-run 必须先核对 ledger、CURRENT Catalog 的 Store/model pin 与 scope fence，只有显式
 `--apply` 才逐 operation 调用领域 `reconcile_operation()`。脚本不得直接 UPDATE operation
-状态、资源镜像或增删 OpenFGA tuple；`FAILED_CLOSED`、pin/scope 不匹配或 ledger 不完整时
-保持阻断并转人工分析。重复传入 `FINALIZED` operation 只验证并跳过。
+状态、资源镜像或增删 OpenFGA tuple；普通 reconcile 对 `FAILED_CLOSED`、pin/scope 不匹配或
+ledger 不完整继续保持阻断。`FAILED_CLOSED` resource operation 经人工确认其冻结请求仍是目标后，
+改用 `recover_f048_failed_closed_projection.py`：先 dry-run 用 higher consistency 读取 exact tuple，
+操作者只输入 tenant/resource type/resource ID，脚本从资源 mode row 的 operation fence 解析 ledger；
+按 ledger AFTER 集计算单次原子前向修正并输出绑定 live correction 的确认 checksum。apply 必须同时
+确认 Store、model 和该 checksum，任一资源绑定或 tuple 事实漂移即拒绝。领域恢复入口只补缺失 WRITE/多余
+DELETE，不重放已满足的 visible/action tuple；完整 AFTER verify 后才执行原 SQL finalizer，并以
+`FINALIZED + CURRENT(target_version)` 收口。超过 90 个 terminal correction、外部业务 scope、
+scope/version/operation 不匹配继续转人工分析。重复传入 `FINALIZED` operation 只验证并跳过。
 
 #### Mode switch
 

features/v3.0.0-beta1/048-rebac-permission-model-grants/tasks.md

@@ -1657,6 +1657,32 @@
     arch-guard 与 diff-check 通过。
   - **依赖**：T195
 
+---
+
+## Wave 16 — FAILED_CLOSED 资源受控前向恢复
+
+- [x] **T197：FAILED_CLOSED 前向恢复合同测试**
+  - **文件**：`src/backend/test/permission/test_f048_projection_service.py`,
+    `src/backend/test/permission/test_f048_projection_sql_runtime.py`,
+    `src/backend/test/permission/test_f048_failed_closed_recovery_cli.py`
+  - **测试**：覆盖 mixed live tuple 仅补 terminal difference、已存在 visible 不重复写、dry-run
+    checksum 漂移拒绝、FAILED_CLOSED SQL mirror 原子 finalize、CLI 按资源解析 operation、
+    Store/model/checksum 二次确认和默认不写。
+  - **覆盖 AC**：AC-166, AC-167, AC-170, AC-179
+  - **依赖**：T159, T196
+
+- [x] **T198：实现通用 resource FAILED_CLOSED 恢复脚本**
+  - **文件**：`src/backend/bisheng/permission/domain/services/projection_service.py`,
+    `src/backend/bisheng/permission/application/sql_runtime.py`,
+    `src/backend/scripts/recover_f048_failed_closed_projection.py`,
+    `src/backend/scripts/README.md`
+  - **逻辑**：只接受仍持有 expected version/operation fence 的 resource operation；以 durable
+    ledger AFTER 集和 higher-consistency exact reads 生成至多 90 条单次原子 correction，dry-run
+    checksum 绑定 live proposal，apply 后完整 verify 再复用 SQL finalizer 收口。普通 reconcile
+    继续拒绝 FAILED_CLOSED，department 等外部业务 scope 不自动恢复。
+  - **验收**：T197、F048 permission focused suite、Ruff、arch-guard 与 diff-check 通过。
+  - **依赖**：T197
+
 ## 实际偏差记录
 
 > 只记录一句话指针；设计原因和反直觉事实回写 [design.md](./design.md)。

src/backend/bisheng/permission/application/sql_runtime.py

@@ -9,7 +9,7 @@
 from uuid import uuid4
 
 from loguru import logger
-from sqlalchemy import func, update
+from sqlalchemy import and_, func, or_, update
 from sqlmodel import select
 
 from bisheng.common.errcode.permission import (
@@ -445,6 +445,36 @@ async def is_expected_version(
             row_id = (await session.execute(statement)).scalar_one_or_none()
         return row_id is not None
 
+    async def is_failed_closed_recovery_scope(
+        self,
+        plan: ProjectionPlan,
+        operation_id: int,
+    ) -> bool:
+        if plan.scope_type != "resource":
+            return False
+        resource_type, separator, resource_id = plan.scope_key.partition(":")
+        if not separator:
+            return False
+        async with get_async_db_session() as session:
+            statement = select(ResourcePermissionMode.id).where(
+                ResourcePermissionMode.tenant_id == plan.tenant_id,
+                ResourcePermissionMode.resource_type == resource_type,
+                ResourcePermissionMode.resource_id == resource_id,
+                ResourcePermissionMode.operation_id == operation_id,
+                or_(
+                    and_(
+                        ResourcePermissionMode.version == plan.expected_version,
+                        ResourcePermissionMode.projection_state == "FAILED_CLOSED",
+                    ),
+                    and_(
+                        ResourcePermissionMode.version == plan.target_version,
+                        ResourcePermissionMode.projection_state == "CURRENT",
+                    ),
+                ),
+            )
+            row_id = (await session.execute(statement)).scalar_one_or_none()
+        return row_id is not None
+
     async def fail_closed(
         self,
         plan: ProjectionPlan,
@@ -588,9 +618,14 @@ async def finalize(
                     )
                 if mode_row.version == plan.target_version and mode_row.projection_state == "CURRENT":
                     return
-                if mode_row.version != plan.expected_version or mode_row.projection_state != "PROJECTING":
+                if mode_row.version != plan.expected_version or mode_row.projection_state not in {
+                    "PROJECTING",
+                    "FAILED_CLOSED",
+                }:
                     raise PermissionPublishNotReadyError(msg=("Permission scope changed before projection finalize"))
 
+                source_projection_state = mode_row.projection_state
+
                 grant_ids = tuple(
                     (
                         await session.execute(
@@ -733,7 +768,7 @@ async def finalize(
                         ResourcePermissionMode.id == mode_row.id,
                         ResourcePermissionMode.operation_id == operation_id,
                         ResourcePermissionMode.version == plan.expected_version,
-                        ResourcePermissionMode.projection_state == "PROJECTING",
+                        ResourcePermissionMode.projection_state == source_projection_state,
                     )
                     .values(**mode_values)
                 )

src/backend/bisheng/permission/domain/services/projection_service.py

@@ -2,7 +2,9 @@
 
 from __future__ import annotations
 
-from dataclasses import replace
+import json
+from dataclasses import dataclass, replace
+from hashlib import sha256
 from typing import Protocol
 
 from loguru import logger
@@ -54,6 +56,12 @@ async def is_expected_version(
         operation_id: int,
     ) -> bool: ...
 
+    async def is_failed_closed_recovery_scope(
+        self,
+        plan: ProjectionPlan,
+        operation_id: int,
+    ) -> bool: ...
+
     async def fail_closed(self, plan: ProjectionPlan, reason: str) -> None: ...
 
 
@@ -99,6 +107,28 @@ async def emit(self, name: str, fields: dict) -> None:
         return None
 
 
+@dataclass(frozen=True, slots=True)
+class FailedClosedRecoveryPreview:
+    """Exact terminal-state correction proposed for one fenced resource."""
+
+    operation_id: int
+    tenant_id: int
+    operation_type: str
+    scope_type: str
+    scope_key: str
+    expected_version: int
+    target_version: int
+    store_id: str
+    model_id: str
+    operation_status: str
+    request_checksum: str
+    after_checksum: str
+    observed_state: str
+    target_tuple_count: int
+    correction_deltas: tuple[ProjectionTupleDelta, ...]
+    confirmation_checksum: str
+
+
 def restore_projection_plan(
     operation: PermissionProjectionOperation,
     tuple_rows,
@@ -269,6 +299,153 @@ async def reconcile_operation(
         plan = restore_projection_plan(operation, tuple_rows)
         return await self.reconcile(plan)
 
+    async def inspect_failed_closed_recovery(
+        self,
+        operation_id: int,
+    ) -> FailedClosedRecoveryPreview:
+        """Build a read-only, checksum-bound forward recovery proposal."""
+
+        operation, plan = await self._load_operation_plan(operation_id)
+        status = str(operation.status)
+        if status not in {
+            ProjectionOperationStatus.FAILED_CLOSED.value,
+            ProjectionOperationStatus.COMMITTED.value,
+        }:
+            raise PermissionPublishNotReadyError(
+                msg=f"Projection operation is not recoverable from status {status}",
+            )
+        if plan.scope_type != "resource":
+            raise PermissionPublishNotReadyError(
+                msg="FAILED_CLOSED forward recovery only supports resource scopes",
+            )
+
+        correction = await self._terminal_correction(plan.deltas)
+        observed_state = "AFTER" if not correction else await self._classify(plan.deltas)
+        confirmation_checksum = self._recovery_confirmation_checksum(
+            operation=operation,
+            correction=correction,
+        )
+        return FailedClosedRecoveryPreview(
+            operation_id=int(operation.id),
+            tenant_id=plan.tenant_id,
+            operation_type=plan.operation_type,
+            scope_type=plan.scope_type,
+            scope_key=plan.scope_key,
+            expected_version=plan.expected_version,
+            target_version=plan.target_version,
+            store_id=plan.store_id,
+            model_id=plan.model_id,
+            operation_status=status,
+            request_checksum=operation.request_checksum,
+            after_checksum=operation.after_checksum,
+            observed_state=observed_state,
+            target_tuple_count=len(
+                projection_state_expectations(plan.deltas, after=True),
+            ),
+            correction_deltas=correction,
+            confirmation_checksum=confirmation_checksum,
+        )
+
+    async def recover_failed_closed_operation(
+        self,
+        operation_id: int,
+        *,
+        confirmation_checksum: str,
+    ) -> ProjectionOutcome:
+        """Forward-complete one fenced resource to its frozen AFTER state."""
+
+        operation, plan = await self._load_operation_plan(operation_id)
+        request_checksum = projection_request_checksum(plan)
+        status = str(operation.status)
+        if status == ProjectionOperationStatus.FINALIZED.value:
+            return self._outcome(
+                plan,
+                operation,
+                request_checksum=request_checksum,
+                idempotent=True,
+                reconciled=True,
+            )
+
+        preview = await self.inspect_failed_closed_recovery(operation_id)
+        if preview.confirmation_checksum != confirmation_checksum:
+            raise PermissionVersionConflictError(
+                msg="FAILED_CLOSED recovery confirmation checksum changed",
+            )
+        if len(preview.correction_deltas) > MAX_ATOMIC_TUPLES:
+            raise PermissionPublishNotReadyError(
+                msg=(
+                    "FAILED_CLOSED recovery requires more than "
+                    f"{MAX_ATOMIC_TUPLES} atomic tuple corrections"
+                ),
+            )
+        if not await self._scope_guard.is_failed_closed_recovery_scope(
+            plan,
+            int(operation.id),
+        ):
+            raise PermissionPublishNotReadyError(
+                msg="FAILED_CLOSED resource scope no longer owns the operation fence",
+            )
+
+        if status == ProjectionOperationStatus.COMMITTED.value:
+            if preview.correction_deltas:
+                raise PermissionProjectionFailedError(
+                    msg="Committed recovery operation no longer has its full AFTER state",
+                )
+            return await self._finalize(
+                plan,
+                operation,
+                request_checksum=request_checksum,
+                reconciled=True,
+            )
+
+        if not await self._marker.is_ready():
+            raise PermissionPublishNotReadyError(
+                msg="Permission recent-change marker sentinel is not ready",
+            )
+        try:
+            await self._marker.arm(plan)
+            commit_checksum = (
+                await self._write(preview.correction_deltas)
+                if preview.correction_deltas
+                else operation.after_checksum
+            )
+        except Exception as exc:
+            if not await self._is_after(plan.deltas):
+                await self._emit(plan, operation, "FAILED_CLOSED_RECOVERY_FAILED", exc)
+                raise PermissionProjectionFailedError(exception=exc) from exc
+            commit_checksum = operation.after_checksum
+
+        if not await self._is_after(plan.deltas):
+            error = PermissionProjectionFailedError(
+                msg="FAILED_CLOSED recovery did not reach the frozen AFTER state",
+            )
+            await self._emit(plan, operation, "FAILED_CLOSED_RECOVERY_FAILED", error)
+            raise error
+        await self._transition(
+            operation,
+            expected=ProjectionOperationStatus.FAILED_CLOSED.value,
+            target=ProjectionOperationStatus.COMMITTED.value,
+            commit_checksum=commit_checksum,
+        )
+        return await self._finalize(
+            plan,
+            operation,
+            request_checksum=request_checksum,
+            reconciled=True,
+        )
+
+    async def _load_operation_plan(
+        self,
+        operation_id: int,
+    ) -> tuple[PermissionProjectionOperation, ProjectionPlan]:
+        operation = await self._repository.aget_operation(operation_id)
+        if operation is None:
+            raise PermissionPublishNotReadyError(
+                msg="Projection operation does not exist",
+            )
+        tuple_rows = await self._repository.aget_operation_tuples(operation_id)
+        return operation, restore_projection_plan(operation, tuple_rows)
+
     async def _run_prepared(
         self,
         plan: ProjectionPlan,
@@ -495,6 +672,63 @@ async def _pending_stage_deltas(
         )
         return tuple(delta for delta in deltas if (delta.key in present) != (delta.action == "WRITE"))
 
+    async def _terminal_correction(
+        self,
+        deltas: tuple[ProjectionTupleDelta, ...],
+    ) -> tuple[ProjectionTupleDelta, ...]:
+        """Return one exact mutation for every tuple not at terminal AFTER."""
+
+        expected = projection_state_expectations(deltas, after=True)
+        representatives: dict[tuple[str, str, str], ProjectionTupleDelta] = {}
+        for delta in deltas:
+            representatives[delta.key] = delta
+        present = await self._fga.read_present(
+            tuple(representatives[key] for key in sorted(representatives)),
+            consistency=HIGHER_CONSISTENCY,
+        )
+        correction: list[ProjectionTupleDelta] = []
+        for sequence, (key, should_exist) in enumerate(sorted(expected.items())):
+            if (key in present) == should_exist:
+                continue
+            correction.append(
+                replace(
+                    representatives[key],
+                    phase="COMMIT",
+                    sequence=sequence,
+                    action="WRITE" if should_exist else "DELETE",
+                )
+            )
+        return tuple(correction)
+
+    @staticmethod
+    def _recovery_confirmation_checksum(
+        *,
+        operation: PermissionProjectionOperation,
+        correction: tuple[ProjectionTupleDelta, ...],
+    ) -> str:
+        payload = {
+            "after_checksum": operation.after_checksum,
+            "correction": [
+                {
+                    "action": delta.action,
+                    "object": delta.object,
+                    "relation": delta.relation,
+                    "user": delta.user,
+                }
+                for delta in correction
+            ],
+            "operation_id": int(operation.id),
+            "operation_status": str(operation.status),
+            "request_checksum": operation.request_checksum,
+        }
+        canonical = json.dumps(
+            payload,
+            ensure_ascii=True,
+            separators=(",", ":"),
+            sort_keys=True,
+        )
+        return sha256(canonical.encode("utf-8")).hexdigest()
+
     async def _classify(
         self,
         deltas: tuple[ProjectionTupleDelta, ...],