# FR-050 稽核輪次階段回退（Stage Rollback）— Implementation Plan

> 依 [design.md](./design.md) 五決策（§2）與回退邊界表（§3）展開。BE 為主，FE 只列整合點。
> 開工前提查證：技術未知數已解——`jedi-flow-engine WorkflowExecutionService.revert_job`
> （`~/Projects/Jedicogy/module/jedi-python-package/jedi-flow-engine/jedi_flow_engine/app/service/workflow_execution_service.py:329`）
> 已支援「目標 job 設回 PROCESSING、清 end_time，其後 job 依 BPMN 順序設回 TODO」，design §4.1
> 的 fallback 備案不需要。BE 既有呼叫（`app/flow_engine/service/workflow_execution_service.py:816`
> `revert_job`）是單一 job 層級 revert，與本案 stage 邊界層級的 rollback 概念不同層次，
> 只參考其防禦模式（`assert_project_participant` + 輪次凍結 gate），不可混淆呼叫對象。

## 0. 現況盤點（開工當下已核實，程式碼若已變動以當下為準）

| 項目 | 現況 |
|---|---|
| BPMN 回退能力 | `jedi-flow-engine` app service 層 `WorkflowExecutionService.revert_job(workflow_execution_uid, job_id, revert_to_job_id, comment, user, user_nickname)` — 目標 job → PROCESSING + `end_time=None`；BPMN `job_execution_order` 中目標之後的 job → TODO |
| stage_objects 現況（DEV DB 核對） | 6 筆：`planning(launch_audit) → ap_authoring(submit_ap) → audit(confirm_audit) → poam(close_round)`，另有平行的 `task_execution` / `review`（builtin-full-audit 用，非本案 round 主流程）。round 主流程走 `planning → ap_authoring → audit → poam` |
| round.status 狀態機 | `planning → audit_planning → auditing → (remediation｜closed) → pending_reverify → closed`（`app/grc/service/audit_round_app_service.py:36-41`），DB CHECK 強制值域，無 rollback 方向的 CHECK（新增值不需要，仍在既有 7 態內移動） |
| StageAdvanceService | `app/flow_engine/service/stage_advance_service.py`（682 行），advance 流程：①角色驗證 →②force 檢查→③precondition→④handler dispatch→⑤`complete_main_workflow_job`→⑥`[AUDIT:STAGE_ADVANCE]` log→⑦回傳新 stage info |
| StageRegistry | `domain/flow_engine/service/stage_completion_registry.py`，`IStageCompletionHandler` / `IStagePreconditionCheck` + `StageRegistry` singleton，key-based lookup，log warning 不 raise on duplicate |
| DI 註冊點 | `di_containers/grc/grc_containers.py` `stage_advance_service`（Factory，約 L554-566）+ `register_stage_hooks_to_registry()`（L570-607，`core/app_factory.py` container.wire 後呼叫一次） |
| route 現況 | `api/flow_engine/routes/stage_advance_route.py` + `api/flow_engine/__init__.py:80-87` 掛 `/project/<uid>/audit-round/<uid>/stage/{info,advance}` |
| `_sync_project_status` | `app/grc/service/audit_round_app_service.py:112-144`，依全部輪次是否 closed 決定 project.status；rollback 呼叫即可，**不寫新邏輯**（user 決策 B） |
| error code 序號 | `common/code/grc_error_code.py` 目前最大：400→`GRC_400076`、403→`GRC_403053`、404→`GRC_404036`、409→`GRC_409033`、412→`GRC_412028`。本案新碼往後接（見 §5） |
| event_code | `common/enum/event_code.py` `STAGE_ADVANCE_*` = 6050-6054；本案 `STAGE_ROLLBACK_*` 接續用 6090 起（避開 6060-6082 已用區段） |
| `project_audit_rounds` 表現況（DEV DB 核對） | 無 `tenant_id` / `org_unit_id` / RLS（`relrowsecurity=f`），跨 schema soft-ref 慣例（`project_id` 不建 FK constraint）。**新表 `round_stage_transitions` 沿用同一「軟隔離」慣例**，不需 tenant_id/org_unit_id/RLS——與母表一致（見 §2 決策 D1） |

---

## 1. 前置技術決策（plan 時裁定，design §3/§5 交辦事項）

### D1：`round_stage_transitions` 是否加 tenant_id / org_unit_id / RLS？

**裁定：不加。** `compliance.project_audit_rounds` 本身無 tenant_id/org_unit_id/RLS（純 soft-ref
`project_id`，租戶隔離由上層 `compliance.projects` 負責）。`round_stage_transitions.round_id`
FK 指向 `project_audit_rounds.id`，屬同一「軟隔離、由 project 上層把關」家族，不應該单獨加碼出
不對稱的隔離層級（CLAUDE.md「tenant-scoped 表必含 org_unit_id」鐵則的前提是
`TenantScopedMixinModel` 表；本表 ORM 不繼承該 mixin，故不適用）。查詢一律經
`round_id → project_id` 的既有 authz 鏈（`assert_project_role`）把關，不依賴 RLS。

### D2：rollback handler 的 registry 機制——複用 `StageRegistry` 或另開？

**裁定：複用同一個 `StageRegistry` singleton，加兩個新方法 `register_rollback_handler` /
`get_rollback_handler`（+ 對應 `_rollback_handlers: dict[str, IStageRollbackHandler]`），
不另開新 registry class。**

理由：
- `StageRegistry` 已是「DI 啟動時掛 handler，flow_engine 執行期查表」的既定模式，rollback
  handler 生命週期與 advance handler 完全一致（皆由 grc 模組在啟動時註冊），沒有理由拆兩個
  registry 增加 DI wiring 複雜度。
- key 命名沿用 `stage_object.code`（如 `"ap_authoring"` 代表「退到 planning」的 handler），
  **不新增 stage_objects 表欄位**（design 提到「stage_registry 加 rollback_handler_key」，
  plan 裁定改用「key = 當前 stage code」直接查，因為回退邊界只有 3 條且都是「退到相鄰上一
  stage」，不需要 stage_objects 表額外存一個 rollback_handler_key 欄位——**只在
  `StageRollbackService` 內部維護一個 Python dict 常數 `_ROLLBACK_ALLOWED = {"ap_authoring", "auditing", "remediation"}`**
  對照決策表判斷是否允許，rollback handler 走 registry 查，key 用 stage code）。
- `IStageRollbackHandler` 新 interface（見 Task 2），與 `IStageCompletionHandler` 平行、不繼承
  （方法簽名不同：rollback 不需要 `execute(ap_uid, ...)` 語意的「完成副作用」，而是「復原副
  作用」，回傳值也不同）。

### D3：三個回退邊界，v1 開放幾條？

**裁定：三條全開**（audit_planning→planning / auditing→audit_planning /
remediation→auditing），對應 user 已拍板決策 A。design §5「remediation → auditing 邊界是否
v1 開放，plan 時裁決」——**開放**，因為 POA&M 作廢標記機制與前兩條邊界的「作廢標記」模式一致，
不額外增加技術風險，且此邊界對應「稽核已定案但發現誤判」的真實情境，v1 排除反而讓最常見的
「recall」情境無法使用。

### D4：`round_stage_transitions` 是否含 advance 方向？

**裁定：含。** design §4.2 已定案「advance 也同步落」。本 plan 在 Task 1 於既有 advance 路徑
（`StageAdvanceService.advance_stage`）補一行落表呼叫，**非破壞性 delta**（advance 現有邏輯
不變，只加一次 domain service 呼叫）。

---

## 2. Migration：`compliance.round_stage_transitions`

**檔案**：`scripts/sql/2026-07-19-fr050-round-stage-transitions.sql`

```sql
-- Date: 2026-07-19
-- FR-050 稽核輪次階段回退（Stage Rollback）— 階段歷程時間軸表
-- 新建 compliance.round_stage_transitions：advance / rollback 皆落表，UI 時間軸資料源。
-- 不含 tenant_id / org_unit_id / RLS — 比照 compliance.project_audit_rounds 軟隔離慣例
-- （round_id soft-ref project_audit_rounds.id，租戶隔離由上層 project 把關；見 plan §1 D1）。

CREATE TABLE compliance.round_stage_transitions (
    id             BIGSERIAL PRIMARY KEY,
    uid            VARCHAR(36)  NOT NULL UNIQUE,                       -- 2026-07-19
    round_id       BIGINT       NOT NULL,                              -- 2026-07-19 → compliance.project_audit_rounds.id（soft ref，比照母表慣例不建 FK constraint，避免 CASCADE）
    from_stage     VARCHAR(50)  NOT NULL,                              -- 2026-07-19 stage_objects.code
    to_stage       VARCHAR(50)  NOT NULL,                              -- 2026-07-19 stage_objects.code
    direction      VARCHAR(10)  NOT NULL,                              -- 2026-07-19 advance | rollback
    operator       VARCHAR(255) NOT NULL,                              -- 2026-07-19 login_name（curr_user）
    operator_nickname VARCHAR(255),                                    -- 2026-07-19 顯示用快照（審計欄位規範：response 需 *_name）
    reason         TEXT,                                               -- 2026-07-19 rollback 必填；advance 恆 NULL
    created_at     TIMESTAMPTZ  NOT NULL DEFAULT NOW(),                -- 2026-07-19
    CONSTRAINT ck_round_stage_transitions_direction
        CHECK (direction IN ('advance', 'rollback')),
    CONSTRAINT ck_round_stage_transitions_reason_required_on_rollback
        CHECK (direction <> 'rollback' OR (reason IS NOT NULL AND btrim(reason) <> ''))
);
COMMENT ON TABLE compliance.round_stage_transitions IS
  'FR-050 稽核輪次階段歷程時間軸（advance + rollback 皆落表）：UI「階段歷程」資料源，rollback 必含 reason';
COMMENT ON COLUMN compliance.round_stage_transitions.round_id IS
  '→ compliance.project_audit_rounds.id（soft ref，比照母表慣例）';
COMMENT ON COLUMN compliance.round_stage_transitions.direction IS
  'advance（推進）｜rollback（退回）— decision==''reject''的 review 退回不算本表 rollback（那是 BPMN reject 分支，非 stage 邊界回退），本表 direction=rollback 專指 FR-050 的相鄰上一步回退';
COMMENT ON COLUMN compliance.round_stage_transitions.reason IS
  'rollback 必填（DB CHECK 強制）；advance 恆 NULL';

CREATE INDEX idx_round_stage_transitions_round_id
  ON compliance.round_stage_transitions (round_id, created_at);

GRANT SELECT, INSERT, UPDATE, DELETE ON compliance.round_stage_transitions TO cm_app;
GRANT USAGE, SELECT ON SEQUENCE compliance.round_stage_transitions_id_seq TO cm_app;

INSERT INTO public.schema_migrations(filename, note) VALUES
  ('2026-07-19-fr050-round-stage-transitions.sql',
   'FR-050 新建 compliance.round_stage_transitions（階段歷程時間軸，advance+rollback 皆落表）')
ON CONFLICT (filename) DO NOTHING;
```

**套用**（依 CLAUDE.md SQL migration 鐵則）：
```bash
PGPASSWORD='<查 .env DB_SECRET>' psql -h 192.168.50.188 -p 25432 -U cmmgr -d guidant_ai_dev \
  --single-transaction -v ON_ERROR_STOP=1 -f scripts/sql/2026-07-19-fr050-round-stage-transitions.sql
```
STG（同 188、db=`guidant_ai_stg`）/ POC（189、`guidant_ai_poc`）待 DEV 驗證通過後比照套用
（三環境對齊走 `sql-migration` skill 或手動比對 `schema_migrations`）。

---

## 3. Domain 層

### 3.1 `domain/grc/entities/round_stage_transition_entity.py`（新檔）

```python
from dataclasses import dataclass
from datetime import datetime
from typing import Optional


@dataclass
class RoundStageTransitionEntity:
    id: Optional[int] = None
    uid: Optional[str] = None
    round_id: Optional[int] = None
    from_stage: Optional[str] = None
    to_stage: Optional[str] = None
    direction: Optional[str] = None  # "advance" | "rollback"
    operator: Optional[str] = None
    operator_nickname: Optional[str] = None
    reason: Optional[str] = None
    created_at: Optional[datetime] = None
```

### 3.2 `domain/grc/entities/round_stage_transition_query_entity.py`（新檔）

鏡像既有 query entity 慣例（如 `ProjectAuditRoundEntity` 的查詢面，或 `JobExecutionQueryEntity`
的欄位式 query object）：

```python
from dataclasses import dataclass
from typing import Optional


@dataclass
class RoundStageTransitionQueryEntity:
    round_id: Optional[int] = None
    direction: Optional[str] = None
```

### 3.3 `domain/grc/repository/i_round_stage_transition_repo.py`（新檔）

```python
from abc import ABC, abstractmethod
from typing import List, Optional

from domain.grc.entities.round_stage_transition_entity import RoundStageTransitionEntity


class IRoundStageTransitionRepo(ABC):
    @abstractmethod
    def add(self, entity: RoundStageTransitionEntity) -> RoundStageTransitionEntity: ...

    @abstractmethod
    def list_by_round(self, round_id: int) -> List[RoundStageTransitionEntity]: ...
```

### 3.4 `domain/grc/service/round_stage_transition_domain_service.py`（新檔）

薄 wrapper，鏡像專案內既有 domain service pattern（如
`domain/participant/service/task_assignee_domain_service.py` 的形狀——建構子吃 repo、方法直接
委派）：

```python
class RoundStageTransitionDomainService:
    def __init__(self, repo):
        self._repo = repo

    def add(self, entity):
        return self._repo.add(entity)

    def list_by_round(self, round_id: int):
        return self._repo.list_by_round(round_id)
```

### 3.5 `domain/flow_engine/service/stage_completion_registry.py`（既有檔，加方法）

在 `StageRegistry` class 內加：

```python
    def __init__(self):
        self._handlers: dict[str, IStageCompletionHandler] = {}
        self._preconditions: dict[str, IStagePreconditionCheck] = {}
        self._rollback_handlers: dict[str, "IStageRollbackHandler"] = {}  # 新增

    def register_rollback_handler(self, handler: "IStageRollbackHandler") -> None:
        if handler.key in self._rollback_handlers:
            logger.warning(
                f"StageRegistry: rollback handler key {handler.key!r} 已存在，將被覆蓋 "
                f"(old={type(self._rollback_handlers[handler.key]).__name__}, "
                f"new={type(handler).__name__})"
            )
        self._rollback_handlers[handler.key] = handler

    def get_rollback_handler(self, key: str) -> Optional["IStageRollbackHandler"]:
        return self._rollback_handlers.get(key)

    def has_rollback_handler(self, key: str) -> bool:
        return key in self._rollback_handlers
```

新增 interface（同檔案，`IStageCompletionHandler` 定義之後）：

```python
class IStageRollbackHandler(ABC):
    """副作用復原 handler — stage 回退時觸發（FR-050）。

    key 慣例＝「當前 stage code」（要退出的那個 stage），例如 key="ap_authoring" 代表
    「從 ap_authoring 退回 planning」的復原邏輯。與 IStageCompletionHandler 的 key（handler_key，
    綁 stage_objects.complete_handler_key）不是同一命名空間，各自查各自的 dict。
    """

    @property
    @abstractmethod
    def key(self) -> str:
        """對應要退出的 stage_objects.code"""

    @abstractmethod
    def execute(
        self,
        round_uid: str,
        project_uid: str,
        user_id: int,
        curr_user: str,
        reason: str,
    ) -> dict:
        """執行副作用復原（作廢標記，不物理刪）。回傳 dict 拼進 rollback response。"""
```

---

## 4. App 層

### 4.1 `app/grc/service/oscal_stage_rollback_handlers.py`（新檔，鏡像 `oscal_stage_handlers.py`）

三個 handler class，一對一鏡像 §3 回退邊界表：

**Key 命名裁定**：handler registry key 用 **round.status 值本身**（`"audit_planning"` /
`"auditing"` / `"remediation"`），**不**用 `stage_object.code`——因為 round.status 命名與
BPMN stage_object.code 命名不是 1:1 字面對應（DEV DB 核對：round.status=`audit_planning`
時活躍 BPMN stage 是 `ap_authoring`；status=`auditing` 對應 BPMN stage `audit`；
status=`remediation` 對應 BPMN stage `poam`——三組名字都不同）。用 round.status 當 key 對
`StageRollbackService`（讀 `rnd.status` 決定走哪個 handler）最直觀，BPMN stage code 的映射
單獨在 `StageRollbackService` 內部一個 dict 管理（見 Task 6），不混進 handler key。

```python
"""Round stage rollback handlers（FR-050 稽核輪次階段回退）。

每個 handler 的 key＝要退出的 round.status 值（非 stage_object.code——兩者命名不 1:1，
見 plan §4.1 說明）。由 grc DI container 啟動時註冊到 flow_engine StageRegistry 的
rollback 分支。回退鐵則：絕不物理刪資料，一律「作廢/superseded 標記 + 留痕」（design §3）。
"""
from domain.flow_engine.service.stage_completion_registry import IStageRollbackHandler


class RollbackApAuthoringHandler(IStageRollbackHandler):
    """audit_planning → planning：frozen SSP + AP 草稿標 superseded；round.status 回 planning。

    對應推進時的 launch_audit（snapshot living SSP + clone 程序書池 + 建 AP 草稿）。
    living SSP 本未動（snapshot 是另建一份），不需處理。
    """

    def __init__(self, audit_round_app_service):
        self._rounds = audit_round_app_service

    @property
    def key(self) -> str:
        return "audit_planning"  # round.status（此 status 下要回退到 planning）

    def execute(self, round_uid, project_uid, user_id, curr_user, reason) -> dict:
        return self._rounds.rollback_to_planning(
            round_uid=round_uid, curr_user=curr_user, curr_user_id=user_id, reason=reason,
        )


class RollbackAuditingHandler(IStageRollbackHandler):
    """auditing → audit_planning：AR 矩陣保留 + 標記（決策 3，不刪不重建）；round.status 回 audit_planning。

    對應推進時的 start_auditing（建 AR 全量矩陣 + ar_result_id 回填）。
    """

    def __init__(self, audit_round_app_service):
        self._rounds = audit_round_app_service

    @property
    def key(self) -> str:
        return "auditing"  # round.status

    def execute(self, round_uid, project_uid, user_id, curr_user, reason) -> dict:
        return self._rounds.rollback_to_audit_planning(
            round_uid=round_uid, curr_user=curr_user, curr_user_id=user_id, reason=reason,
        )


class RollbackRemediationHandler(IStageRollbackHandler):
    """remediation → auditing：POA&M 標作廢 + 已有整改紀錄保留；round.status 回 auditing。

    對應推進時的 finalize_audit（AR 定版 + not_met 生 POA&M）。
    """

    def __init__(self, audit_round_app_service):
        self._rounds = audit_round_app_service

    @property
    def key(self) -> str:
        return "remediation"  # round.status

    def execute(self, round_uid, project_uid, user_id, curr_user, reason) -> dict:
        return self._rounds.rollback_to_auditing(
            round_uid=round_uid, curr_user=curr_user, curr_user_id=user_id, reason=reason,
        )
```

### 4.2 `AuditRoundAppService` 加三個 rollback 方法（既有檔 `app/grc/service/audit_round_app_service.py`）

鏡像既有 `launch_audit` / `start_auditing` / `finalize_audit` 的 `@transaction` + 前置條件檢查
+ manager 守門 + `_sync_project_status` 收尾模式。**這三個方法是 rollback 業務邏輯的實作點**
（handler 只是 registry 掛勾，實際邏輯在這裡）：

```python
    @transaction
    def rollback_to_planning(self, round_uid: str, curr_user: str, curr_user_id: int,
                             reason: str, locale=None) -> dict:
        """audit_planning → planning：frozen SSP / AP 草稿標 superseded，回填清空。

        前置：status == audit_planning（否則 GRC_ROUND_ROLLBACK_INVALID_STATE）。
        manager only（決策 4）。reason 必填（route schema 層已擋，這裡防呆再檢一次）。
        """
        e = self._rounds.get_by_uid(round_uid)
        if e is None:
            raise NotFound(GrcErrorCode.GRC_ROUND_NOT_FOUND)
        self._check_role(e.project_id, curr_user_id, "manager")
        if not (reason or "").strip():
            raise BadRequestError(GrcErrorCode.GRC_ROUND_ROLLBACK_REASON_REQUIRED)
        if e.status != STATUS_AUDIT_PLANNING:
            raise PreconditionFailedError(GrcErrorCode.GRC_ROUND_ROLLBACK_INVALID_STATE)

        # frozen SSP 標 superseded（不物理刪；deep_clone 出來的獨立副本，標記後 UI 濾掉）
        if e.ssp_id and self._oscal_snapshot_service is not None:
            self._oscal_snapshot_service.mark_ssp_superseded(e.ssp_id, curr_user)
        # AP 草稿標 superseded（PublishStatus 無此值域 — 見 Task 4.4 套件異動）
        if e.assessment_plan_id and self._ap_app_service is not None:
            self._ap_app_service.mark_ap_superseded(e.assessment_plan_id, curr_user)

        e.ssp_id = None
        e.assessment_plan_id = None
        e.status = STATUS_PLANNING
        e.updated_user = curr_user
        updated = self._rounds.update(e)
        self._record_transition(e.id, from_stage="ap_authoring", to_stage="planning",
                                curr_user=curr_user, reason=reason)
        logger.info("rollback_to_planning round=%s -> planning (superseded ssp/ap)", round_uid)
        self._sync_project_status(e.project_id, curr_user)
        return self._to_dto(updated)

    @transaction
    def rollback_to_audit_planning(self, round_uid: str, curr_user: str, curr_user_id: int,
                                   reason: str, locale=None) -> dict:
        """auditing → audit_planning：AR 矩陣保留 + 標記（決策 3），round.status 回 audit_planning。"""
        e = self._rounds.get_by_uid(round_uid)
        if e is None:
            raise NotFound(GrcErrorCode.GRC_ROUND_NOT_FOUND)
        self._check_role(e.project_id, curr_user_id, "manager")
        if not (reason or "").strip():
            raise BadRequestError(GrcErrorCode.GRC_ROUND_ROLLBACK_REASON_REQUIRED)
        if e.status != STATUS_AUDITING:
            raise PreconditionFailedError(GrcErrorCode.GRC_ROUND_ROLLBACK_INVALID_STATE)

        if e.ar_result_id and self._ar_app_service is not None:
            self._ar_app_service.mark_ar_superseded(e.ar_result_id, curr_user)

        e.status = STATUS_AUDIT_PLANNING
        e.updated_user = curr_user
        updated = self._rounds.update(e)
        self._record_transition(e.id, from_stage="auditing", to_stage="ap_authoring",
                                curr_user=curr_user, reason=reason)
        logger.info("rollback_to_audit_planning round=%s -> audit_planning (ar retained+marked)",
                    round_uid)
        self._sync_project_status(e.project_id, curr_user)
        return self._to_dto(updated)

    @transaction
    def rollback_to_auditing(self, round_uid: str, curr_user: str, curr_user_id: int,
                             reason: str, locale=None) -> dict:
        """remediation → auditing：POA&M 標作廢 + 已有整改紀錄保留，round.status 回 auditing。"""
        e = self._rounds.get_by_uid(round_uid)
        if e is None:
            raise NotFound(GrcErrorCode.GRC_ROUND_NOT_FOUND)
        self._check_role(e.project_id, curr_user_id, "manager")
        if not (reason or "").strip():
            raise BadRequestError(GrcErrorCode.GRC_ROUND_ROLLBACK_REASON_REQUIRED)
        if e.status != STATUS_REMEDIATION:
            raise PreconditionFailedError(GrcErrorCode.GRC_ROUND_ROLLBACK_INVALID_STATE)

        if e.poam_id and self._poam_app_service is not None:
            self._poam_app_service.mark_poam_superseded(e.poam_id, curr_user)

        e.status = STATUS_AUDITING
        e.poam_id = None
        e.updated_user = curr_user
        updated = self._rounds.update(e)
        self._record_transition(e.id, from_stage="poam", to_stage="audit",
                                curr_user=curr_user, reason=reason)
        logger.info("rollback_to_auditing round=%s -> auditing (poam superseded)", round_uid)
        self._sync_project_status(e.project_id, curr_user)
        return self._to_dto(updated)

    def _record_transition(self, round_id: int, from_stage: str, to_stage: str,
                           curr_user: str, reason: Optional[str]) -> None:
        """落 round_stage_transitions 一行（advance/rollback 共用）。caller 須在 @transaction scope。"""
        from uuid import uuid4
        from domain.grc.entities.round_stage_transition_entity import RoundStageTransitionEntity
        nickname = None
        if self._user_domain_service is not None:
            users = self._user_domain_service.get_users(
                UserQueryEntity(_in_login_name=[curr_user])
            )
            nickname = users[0].nickname if users else None
        self._transitions.add(RoundStageTransitionEntity(
            uid=str(uuid4()), round_id=round_id,
            from_stage=from_stage, to_stage=to_stage,
            direction="rollback" if reason else "advance",
            operator=curr_user, operator_nickname=nickname, reason=reason,
        ))
```

**建構子加一個依賴** `round_stage_transition_domain_service`（`self._transitions`），DI 於 §6
wiring。

### 4.3 既有 advance 路徑補落表（Task D4 的非破壞性 delta）

`StageAdvanceService.advance_stage`（`app/flow_engine/service/stage_advance_service.py:184`）
第 6 步「回新 stage info」之前，加一次 domain service 呼叫：

```python
        # 6.5 落階段歷程（FR-050 D4：advance 也記，UI 時間軸資料源）
        if self._round_stage_transition_domain_service is not None:
            self._round_stage_transition_domain_service.add(RoundStageTransitionEntity(
                uid=str(uuid4()),
                round_id=wf_ctx["round"].id,
                from_stage=stage_object.code,
                to_stage=next_stage_code or stage_object.code,
                direction="advance",
                operator=curr_user,
                operator_nickname=ctx.get("user_nickname"),
                reason=None,
            ))
```

**注意**：`StageAdvanceService` 建構子需新增依賴 `round_stage_transition_domain_service`（可選，
`Optional[...] = None`，DI 未注入時 no-op skip，避免測試/舊 DI 設定炸裂）。`wf_ctx["round"]`
已由 `_resolve_workflow_context` 回傳（見既有 §509-529），有 `.id` 可用，不需要額外查詢。

### 4.4 三個 `mark_*_superseded` 方法（跨套件異動，需 user 點頭）

**⚠️ 這是套件層異動，依 CLAUDE.md「外部套件異動規範」必須提醒 user 做決策，不可直接動手。**

| 方法 | 落點 | 異動內容 |
|---|---|---|
| `OscalSnapshotService.mark_ssp_superseded(ssp_id, curr_user)` | `jedi-oscal-v2` `oscal_snapshot_service.py` | frozen SSP 的 `status` 欄（現有 `FROZEN_STATUS="frozen"` 同一個自由 String(20) 欄，無 CHECK）多寫一個值 `"superseded"`，同一 pattern，风险低 |
| `AssessmentPlanAppService.mark_ap_superseded(ap_id, curr_user)` | 若 AP status 欄位空間足夠：`ApEntity.status` 現用 `PublishStatus`（draft/published/deprecated），**superseded 需要新值**——需在 `jedi-oscal-v2` `oscal_enums.py` `PublishStatus` 加 `SUPERSEDED = "superseded"`，或本專案 app 層另立業務層 status 判斷（不動套件列舉，改用本專案 AP entity 外的 marker，例如另建極簡 `oscal.ap_supersessions` 標記表——**待 user 裁決走哪條**） |
| `PoamAppService.mark_poam_superseded(poam_id, curr_user)` | 同上，`oscal.poams.status` 現用 `PublishStatus`，同樣需要決定「動套件列舉」vs「本專案側標記表」 |

**Plan 交付前必須讓 user 決策的兩個選項**（本 plan 先列選項，不擅自選）：

- **選項 A（動套件列舉）**：`jedi-oscal-v2` `PublishStatus` 加 `SUPERSEDED`。影響範圍：所有讀
  `ApEntity.status` / `PoamEntity.status` / `SspEntity.status` 做 `== PublishStatus.PUBLISHED`
  等窮舉比對的地方需要 grep 確認不會誤判 superseded 為其他態（如 list 頁篩選 draft 是否會意外
  撈到 superseded）。跨套件異動，走 poetry path dependency dev-loop，需 user 點頭 + 事後決定是否
  發版。
- **選項 B（本專案側標記表）**：新建極簡 `compliance.round_rollback_supersessions`（entity_type
  + entity_id + round_id + superseded_at + reason），本專案業務層判斷「這筆 AP/POA&M 是否被
  superseded」查這張表，**不動套件**、風險更低，但查詢時多一次 join（AP/POA&M list 頁若要濾掉
  superseded 項目需要額外查詢）。

本 plan **建議選項 B**（風險更低、不跨套件），但按 CLAUDE.md 規範，動手前需 user 明確選定，
不可由執行 session 自行拍板。

> **✅ 裁決定案（user 2026-07-19 拍板）：選項 B。** 理由：superseded 是 FR-050 回退流程的
> 業務概念（天然帶 round_id + reason 脈絡），非 OSCAL 文件生命週期通用狀態——塞進套件
> `PublishStatus` 屬 caller 業務知識下沉，違反套件異動分界原則；且選項 A 的 grep 面
> （3 entity × 全部窮舉比對讀取點）成本不小。
>
> **執行落點（實作時依此，不再停下確認）**：
> 1. 新表 `compliance.round_rollback_supersessions`（uid, entity_type CHECK IN
>    ('ssp','ap','poam'), entity_id BIGINT, round_id BIGINT, reason TEXT,
>    superseded_at TIMESTAMPTZ DEFAULT NOW(), operator VARCHAR(255)）——**併入 §2 同一支
>    migration 檔**（GRANT cm_app + sequence + COMMENT 比照 §2 慣例；idx on
>    (entity_type, entity_id) + (round_id)）。
> 2. 三個 `mark_*_superseded` 全部落在**本專案** `AuditRoundAppService` private helper
>    （或獨立 domain service），一律 INSERT 本表，**不動 jedi-oscal-v2 任何檔**。
>    §4.4 表格中 SSP 那列原「寫 status 欄 superseded」也改走本表（三 entity 統一機制，
>    不做兩套；frozen SSP 的 status 欄維持 "frozen" 不動）。
> 3. 「是否被 supersede」的讀取封裝在 `RoundRollbackSupersessionDomainService.is_superseded(
>    entity_type, entity_id)` / `list_by_round(round_id)`，AP/POA&M list 頁若需濾除，
>    由 caller 帶集合查詢（batch in），不逐筆 join。

其餘皆可依 plan 直接執行，**本 plan 已無未決分岔點**。

---

## 5. Error Code 新增（`common/code/grc_error_code.py`）

接續現有最大序號（§0 已核對）：

```python
GRC_ROUND_ROLLBACK_INVALID_STATE   = ("目前輪次狀態不允許回退到此階段", "GRC_412029")
GRC_ROUND_ROLLBACK_REASON_REQUIRED = ("回退原因為必填",                 "GRC_400077")
GRC_ROUND_ROLLBACK_NOT_ALLOWED     = ("此邊界不允許回退（closed 輪次或無對應 handler）", "GRC_412030")
GRC_ROUND_ROLLBACK_HANDLER_NOT_REGISTERED = ("rollback handler 未註冊", "GRC_400078")
```

（`GRC_NOT_MANAGER`、`GRC_ROUND_NOT_FOUND` 沿用既有碼，不新增。）

`common/enum/event_code.py` 新增：

```python
    # Flow Engine — Stage Rollback（FR-050，log handler 寫進 public.system_logs.event_code）
    STAGE_ROLLBACK_REQUESTED = 6090  # 回退 API 收到請求
    STAGE_ROLLBACK_SUCCESS   = 6091  # 回退成功
    STAGE_ROLLBACK_DENIED    = 6092  # 各種被擋（reason 由 message 內 reason= 區分）
```

---

## 6. `StageRollbackService`（新檔 `app/flow_engine/service/stage_rollback_service.py`）

鏡像 `StageAdvanceService` 結構（不繼承，因流程形狀不同：advance 靠 stage_object 精確定位下一
節點，rollback 靠 §1 D2 的 3 條白名單邊界）。

```python
"""Stage rollback service（FR-050 稽核輪次階段回退）。

「返回上一階段」按鈕的 BE 入口。設計：
- 只允許退回相鄰上一步（決策 1）；跨階段＝連按多次，每次各留一筆 round_stage_transitions
- closed 輪次不可回退（決策 2，DB CHECK 之外此處也業務層擋，防呆雙保險）
- manager only（決策 4）
- BPMN 回退委派 jedi-flow-engine WorkflowExecutionService.revert_job
- 副作用復原委派 stage_registry 查到的 IStageRollbackHandler（AuditRoundAppService 對應方法）
"""
from __future__ import annotations

import logging
from typing import Optional

from jedi_common.handler.exception import (
    BadRequestError,
    ForbiddenError,
    NotFound,
    PreconditionFailedError,
)
from jedi_common.session.database.db import transaction
from jedi_flow_engine.common.utils.bpmn_uilts import BpmnUtils
from jedi_flow_engine.domain.entity.workflow_execution_query_entity import (
    WorkflowExecutionQueryEntity,
)

from common.code.grc_error_code import GrcErrorCode
from common.enum.event_code import EventCode
from common.enum.participant_enum import ParticipantRole
from domain.participant.entity.project_participant_query_entity import (
    ProjectParticipantQueryEntity,
)

logger = logging.getLogger(__name__)

# 決策 1/2/3（design §3）：允許回退的 round.status 白名單，一律退回相鄰上一步
# （跨階段靠連按達成，不做多級跳轉）。三欄：
#   status_key      = rnd.status 現值（決定走哪個 rollback handler，見 handler.key）
#   from_stage_code = 目前活躍的 BPMN stage_object.code（revert_job 的 job_id 來源）
#   to_stage_code   = 回退目標的 BPMN stage_object.code（revert_job 的 revert_to_job_id 來源）
# status code 與 stage_object.code 命名不是 1:1（§0 已用 DEV DB 核對 stage_objects seed），
# 兩者關係在此表一次性對照清楚，避免散落各處各自猜測導致錯位。
_ROLLBACK_BOUNDARIES = {
    "audit_planning": {"from_stage_code": "ap_authoring", "to_stage_code": "planning"},
    "auditing":        {"from_stage_code": "audit",        "to_stage_code": "ap_authoring"},
    "remediation":     {"from_stage_code": "poam",         "to_stage_code": "audit"},
}


class StageRollbackService:
    def __init__(
        self,
        stage_registry,
        audit_round_domain_service,
        workflow_execution_service,
        workflow_execution_domain_service,
        project_domain_service,
        project_participant_domain_service,
    ):
        self._stage_registry = stage_registry
        self._audit_round_domain_service = audit_round_domain_service
        self._workflow_execution_service = workflow_execution_service
        self._workflow_execution_domain_service = workflow_execution_domain_service
        self._project_domain_service = project_domain_service
        self._participant_domain_service = project_participant_domain_service

    @transaction
    def rollback_stage(
        self,
        project_uid: str,
        round_uid: str,
        user_id: int,
        curr_user: str,
        reason: str,
        ctx: Optional[dict] = None,
    ) -> dict:
        ctx = dict(ctx or {})
        logger.info(
            "[AUDIT:STAGE_ROLLBACK] requested user_id=%s curr_user=%s project_uid=%s "
            "round_uid=%s reason=%r",
            user_id, curr_user, project_uid, round_uid, (reason or "")[:200],
            extra={'event_code': EventCode.STAGE_ROLLBACK_REQUESTED},
        )

        rnd = self._audit_round_domain_service.get_by_uid(round_uid)
        if rnd is None:
            raise NotFound(GrcErrorCode.GRC_ROUND_NOT_FOUND)

        # 1. manager only（決策 4；不做 role fallback，精確查該專案 participant）
        user_role = self._get_user_role(project_uid, user_id)
        if user_role != ParticipantRole.MANAGER:
            logger.warning(
                "[AUDIT:STAGE_ROLLBACK] denied reason=not_manager user_id=%s round_uid=%s",
                user_id, round_uid, extra={'event_code': EventCode.STAGE_ROLLBACK_DENIED},
            )
            raise ForbiddenError(GrcErrorCode.GRC_NOT_MANAGER)

        # 2. reason 必填（決策 5 前提）
        if not (reason or "").strip():
            raise BadRequestError(GrcErrorCode.GRC_ROUND_ROLLBACK_REASON_REQUIRED)

        # 3. closed 不可退（決策 2，業務層雙保險；round.status CHECK 本身不擋「從哪個態轉」）
        if rnd.status == "closed":
            raise PreconditionFailedError(GrcErrorCode.GRC_ROUND_ROLLBACK_NOT_ALLOWED)

        # 4. 依 round.status 判斷這是哪個回退邊界（round.status 是 canonical 真相；
        #    handler registry key＝round.status 本身，見 §4.1 handler 定義 + _ROLLBACK_BOUNDARIES）
        boundary = _ROLLBACK_BOUNDARIES.get(rnd.status)
        if boundary is None:
            raise PreconditionFailedError(GrcErrorCode.GRC_ROUND_ROLLBACK_NOT_ALLOWED)

        handler = self._stage_registry.get_rollback_handler(rnd.status)
        if handler is None:
            logger.error(
                "[AUDIT:STAGE_ROLLBACK] denied reason=handler_not_registered "
                "round_uid=%s status=%s",
                round_uid, rnd.status,
                extra={'event_code': EventCode.STAGE_ROLLBACK_DENIED},
            )
            raise PreconditionFailedError(GrcErrorCode.GRC_ROUND_ROLLBACK_HANDLER_NOT_REGISTERED)

        # 5. 副作用復原（AuditRoundAppService.rollback_to_* — 標記 superseded + round.status 回寫
        #    + 落 round_stage_transitions；見 app/grc/service/audit_round_app_service.py）
        handler_result = handler.execute(
            round_uid=round_uid, project_uid=project_uid,
            user_id=user_id, curr_user=curr_user, reason=reason,
        )

        # 6. BPMN 回退（jedi-flow-engine revert_job；目標 job 設回 PROCESSING、清 end_time，
        #    其後 job 依 job_execution_order 設回 TODO）
        if rnd.workflow_execution_uid:
            wf = self._workflow_execution_domain_service.get_workflow_execution(
                WorkflowExecutionQueryEntity(uid=str(rnd.workflow_execution_uid))
            )
            if wf is not None:
                bpmn_util = BpmnUtils(wf.workflow_template_xml)
                curr_job = self._find_job_by_stage_code(
                    bpmn_util, boundary["from_stage_code"]
                )
                target_job = self._find_job_by_stage_code(
                    bpmn_util, boundary["to_stage_code"]
                )
                if target_job is not None and curr_job is not None:
                    self._workflow_execution_service.revert_job(
                        workflow_execution_uid=str(rnd.workflow_execution_uid),
                        job_id=curr_job.id,
                        revert_to_job_id=target_job.id,
                        comment=reason,
                        user=curr_user,
                        user_nickname=ctx.get("user_nickname") or curr_user,
                    )

        logger.info(
            "[AUDIT:STAGE_ROLLBACK] success user_id=%s curr_user=%s round_uid=%s "
            "from_status=%s",
            user_id, curr_user, round_uid, rnd.status,
            extra={'event_code': EventCode.STAGE_ROLLBACK_SUCCESS},
        )
        return {"rolled_back": True, "handler_result": handler_result}

    def _get_user_role(self, project_uid: str, user_id: int) -> Optional[str]:
        from jedi_project.domain.entity.project_query_entity import ProjectQueryEntity
        project = self._project_domain_service.get_one(ProjectQueryEntity(uid=project_uid))
        if project is None:
            raise NotFound(GrcErrorCode.GRC_PROJECT_NOT_FOUND)
        participant = self._participant_domain_service.get_one(
            ProjectParticipantQueryEntity(project_id=project.id, user_id=user_id)
        )
        return participant.role if participant is not None else None

    def _find_job_by_stage_code(self, bpmn_util: BpmnUtils, stage_code: Optional[str]):
        if not stage_code:
            return None
        for job in bpmn_util.jobs:
            if (job.properties or {}).get("stage_object_code") == stage_code:
                return job
        return None
```

**待開工時再核實一項**（plan 交付後、實作前）：`_ROLLBACK_BOUNDARIES` dict 裡的
`from_stage_code` / `to_stage_code` 值是依 §0 DEV DB 核對的 `compliance.stage_objects`
seed 資料（`planning/ap_authoring/audit/poam`）逐一對照 `round.status` 值
（`audit_planning/auditing/remediation`）推導，STG/POC 若 seed 資料有差異（理論上不該有，
stage_objects 是 builtin seed，但仍應開工時 SELECT 一次 `compliance.stage_objects` 現況
確認 code 值與此表一致），再動手接線。

---

## 7. Serializer（`api/flow_engine/serializers/stage_rollback.py`，新檔）

鏡像 `api/flow_engine/serializers/stage_advance.py`：

```python
from marshmallow import Schema, fields, validate


class StageRollbackRequestSchema(Schema):
    """POST /project/<uid>/audit-round/<uid>/stage/rollback body."""
    reason = fields.Str(required=True, validate=validate.Length(min=1))
    ctx = fields.Dict(load_default=dict)


class StageRollbackResponseSchema(Schema):
    rolled_back = fields.Bool(required=True)
    handler_result = fields.Dict(allow_none=True)


class RoundStageTransitionEntrySchema(Schema):
    uid = fields.Str(required=True)
    from_stage = fields.Str(required=True)
    to_stage = fields.Str(required=True)
    direction = fields.Str(required=True)  # advance | rollback
    operator = fields.Str(required=True)
    operator_nickname = fields.Str(allow_none=True)
    reason = fields.Str(allow_none=True)
    created_at = fields.DateTime(required=True)


class RoundStageTransitionsResponseSchema(Schema):
    """GET /project/<uid>/audit-round/<uid>/stage/transitions response body."""
    data = fields.List(fields.Nested(RoundStageTransitionEntrySchema), load_default=list)
```

`reason` 必填由 marshmallow `required=True` + `Length(min=1)` 在 schema 層先擋一次 400
（`unknown` 情境交給 jedi-common ValidationError handler），service 層 §6 Step 2 仍保留
strip 檢查作雙保險（鏡像既有 `StageAdvanceRequestSchema._validate_reject_requires_comment`
的雙層防禦模式）。

---

## 8. Route + DI wiring

### 8.1 `api/flow_engine/routes/stage_rollback_route.py`（新檔）

```python
from dependency_injector.wiring import Provide, inject
from flask import request
from flask_jwt_extended import jwt_required
from flask_restful import Resource
from jedi_common.session.auth.auth_context import get_user_context

from api.flow_engine.serializers.stage_rollback import (
    StageRollbackRequestSchema,
    StageRollbackResponseSchema,
    RoundStageTransitionsResponseSchema,
)
from app.flow_engine.service.stage_rollback_service import StageRollbackService
from app.grc.service.audit_round_app_service import AuditRoundAppService
from common.util.response_util import return_response
from di_containers.containers import Containers


class StageRollbackResource(Resource):
    method_decorators = [jwt_required()]

    @inject
    def post(
        self,
        project_uid: str,
        round_uid: str,
        service: StageRollbackService = Provide[
            Containers.grc_container.stage_rollback_service
        ],
    ):
        body = StageRollbackRequestSchema().load(request.get_json(silent=True) or {})
        user_ctx = get_user_context()
        ctx = dict(body.get("ctx") or {})
        ctx.setdefault("user_nickname", user_ctx.nickname)

        result = service.rollback_stage(
            project_uid=project_uid,
            round_uid=round_uid,
            user_id=user_ctx.id,
            curr_user=user_ctx.login_name,
            reason=body["reason"],
            ctx=ctx,
        )
        return return_response(True, StageRollbackResponseSchema().dump(result))


class RoundStageTransitionsResource(Resource):
    method_decorators = [jwt_required()]

    @inject
    def get(
        self,
        project_uid: str,
        round_uid: str,
        audit_round_app_service: AuditRoundAppService = Provide[
            Containers.grc_container.audit_round_app_service
        ],
    ):
        transitions = audit_round_app_service.list_stage_transitions(round_uid)
        return return_response(
            True, RoundStageTransitionsResponseSchema().dump({"data": transitions})
        )
```

`list_stage_transitions(round_uid)` 為 `AuditRoundAppService` 再加一個薄讀方法（`@transaction`，
`self._rounds.get_by_uid` 解 round_id 後呼叫 `self._transitions.list_by_round`，回傳 dto list，
含 §1 D 的 enrich nickname，鏡像 `_enrich_users` pattern）。

### 8.2 `api/flow_engine/__init__.py` 加 route 掛載

`create_module()` import 區塊加：
```python
    from api.flow_engine.routes.stage_rollback_route import (
        StageRollbackResource, RoundStageTransitionsResource,
    )
```
掛載區塊（緊接既有 `StageAdvanceResource` 之後）：
```python
    api.add_resource(
        StageRollbackResource,
        '/project/<string:project_uid>/audit-round/<string:round_uid>/stage/rollback',
    )
    api.add_resource(
        RoundStageTransitionsResource,
        '/project/<string:project_uid>/audit-round/<string:round_uid>/stage/transitions',
    )
```

### 8.3 DI Container（`di_containers/grc/grc_containers.py`）

在既有 `stage_advance_service` provider 之後加：

```python
    # ── Stage Rollback Service（FR-050，round-scoped）──────────────────────
    round_stage_transition_repo = providers.Factory(RoundStageTransitionRepoImpl)
    round_stage_transition_domain_service = providers.Factory(
        RoundStageTransitionDomainService,
        repo=round_stage_transition_repo,
    )
    stage_rollback_service = providers.Factory(
        StageRollbackService,
        stage_registry=flow_template_container.stage_registry,
        audit_round_domain_service=project_audit_round_domain_service,
        workflow_execution_service=workflow_execution_container.workflow_execution_service,
        workflow_execution_domain_service=workflow_execution_container.workflow_execution_domain_service,
        project_domain_service=project_container.project_domain_service,
        project_participant_domain_service=project_participant_container.project_participant_domain_service,
    )
```

`audit_round_app_service` provider（既有）加一個 kwarg
`round_stage_transition_domain_service=round_stage_transition_domain_service`。

`stage_advance_service` provider（既有）加一個 kwarg
`round_stage_transition_domain_service=round_stage_transition_domain_service`（Optional，§4.3）。

### 8.4 `register_stage_hooks_to_registry()` 加三行

既有函式（`grc_containers.py:568-607`）尾端加：
```python
    _safe_register("rollback_handler", grc_container.rollback_ap_authoring_handler)
    _safe_register("rollback_handler", grc_container.rollback_auditing_handler)
    _safe_register("rollback_handler", grc_container.rollback_remediation_handler)
```
`_safe_register` 內部的 `if kind == "handler": ... else: ...` 二分需擴成三分
（`"handler"` / `"precondition"` / `"rollback_handler"`），對應呼叫
`registry.register_rollback_handler(obj)`。

三個 handler provider（`rollback_ap_authoring_handler` 等）比照既有
`launch_audit_on_complete_handler` 等 provider 形狀，`providers.Singleton(RollbackApAuthoringHandler, audit_round_app_service=audit_round_app_service)`。

---

## 9. Infra 層：`RoundStageTransitionRepoImpl`

### 9.1 `infra/grc/model/round_stage_transition.py`（新檔）

```python
from datetime import datetime
from typing import Optional

from sqlalchemy import BigInteger, DateTime, String, Text
from sqlalchemy.orm import Mapped, mapped_column

from jedi_common.session.database.declarative_base import Base


class RoundStageTransition(Base):
    """compliance.round_stage_transitions — 階段歷程時間軸（FR-050）。"""

    __tablename__ = "round_stage_transitions"
    __table_args__ = {"schema": "compliance", "comment": "FR-050 階段歷程時間軸"}

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    uid: Mapped[str] = mapped_column(String(36), nullable=False, unique=True)
    round_id: Mapped[int] = mapped_column(BigInteger, nullable=False, comment="→ compliance.project_audit_rounds.id（soft ref）")
    from_stage: Mapped[str] = mapped_column(String(50), nullable=False)
    to_stage: Mapped[str] = mapped_column(String(50), nullable=False)
    direction: Mapped[str] = mapped_column(String(10), nullable=False)
    operator: Mapped[str] = mapped_column(String(255), nullable=False)
    operator_nickname: Mapped[Optional[str]] = mapped_column(String(255), nullable=True)
    reason: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
```

（不用 `BaseModel` — 本表無 `updated_at`/`updated_user`，是純 append-only log，鏡像
`JobExecutionComment` 的「不繼承 BaseModel、自己列欄位」做法。）

### 9.2 `infra/grc/mapper/round_stage_transition_mapper.py`（新檔）

鏡像 `GrcJobCommentMapper` 的 entity↔model 轉換 pattern（`to_entity` / `to_model`）。

### 9.3 `infra/grc/repository/round_stage_transition_repo_impl.py`（新檔）

```python
from typing import List

from domain.grc.entities.round_stage_transition_entity import RoundStageTransitionEntity
from domain.grc.repository.i_round_stage_transition_repo import IRoundStageTransitionRepo
from infra.grc.mapper.round_stage_transition_mapper import RoundStageTransitionMapper
from infra.grc.model.round_stage_transition import RoundStageTransition


class RoundStageTransitionRepoImpl(IRoundStageTransitionRepo):
    """CLAUDE.md session 規範：session 必須 lazy — @property，不在 __init__ 存 self.session。"""

    @property
    def session(self):
        from jedi_common.session.database.db import get_session
        return get_session()

    def add(self, entity: RoundStageTransitionEntity) -> RoundStageTransitionEntity:
        from datetime import datetime
        model = RoundStageTransitionMapper.to_model(entity)
        model.created_at = model.created_at or datetime.utcnow()
        self.session.add(model)
        self.session.flush()
        self.session.refresh(model)
        return RoundStageTransitionMapper.to_entity(model)

    def list_by_round(self, round_id: int) -> List[RoundStageTransitionEntity]:
        rows = (
            self.session.query(RoundStageTransition)
            .filter(RoundStageTransition.round_id == round_id)
            .order_by(RoundStageTransition.created_at.asc())
            .all()
        )
        return [RoundStageTransitionMapper.to_entity(r) for r in rows]
```

（DDD checklist：不繼承 `BaseRepositoryImpl` 是因為本表沒有標準 update/get_by_uid 需求
——純 append + list，比照 `GrcJobCommentRepoImpl` 直接寫 raw session query 也可，但用
`@property session` 滿足「repo 層 session 必須 lazy」鐵則即可。）

---

## 10. FE 整合點（只列位置與契約，不展開實作）

| 項目 | 檔案 | 內容 |
|---|---|---|
| Service 方法 | `~/Projects/Billows/Audit-Manager/compliance-manager-fe/src/service/StageService.js` | 加 `rollbackStage(projectUid, roundUid, { reason, ctx })` → `POST .../stage/rollback`；`getStageTransitions(projectUid, roundUid)` → `GET .../stage/transitions` |
| Banner 回退入口 | `src/components/grc/FlowPhaseBanner.vue` | manager only 可見的「返回上一階段」次按鈕（比照既有 review reject 按鈕的 `v-if="isManager"` 條件）；點擊開 Dialog 收 `reason`（必填，鏡像既有 `reviewDialogVisible` + reject comment 必填的 pattern，§297-306 已有前例） |
| 階段時間軸元件 | 新檔 `src/components/grc/StageTransitionTimeline.vue`（或 banner 展開區塊） | 垂直 timeline：`direction=advance` 藍色、`direction=rollback` 橘/紅色；rollback 項目顯示 `reason` + `operator_nickname` + `created_at` |
| API 契約 | — | `POST /api/1.0/project/<uid>/audit-round/<uid>/stage/rollback` body `{reason, ctx}` → `{rolled_back, handler_result}`；`GET /api/1.0/project/<uid>/audit-round/<uid>/stage/transitions` → `{data: [...]}` |
| 錯誤碼 i18n | `config/translations/` | 新增 `GRC_412029`（狀態不允許回退）、`GRC_400077`（reason 必填）、`GRC_412030`（不允許回退）、`GRC_400078`（handler 未註冊）四碼中英文案 |

---

## 11. 驗收情境對應（design §6 六條逐一對應到 task）

| # | 驗收情境 | 對應 Task |
|---|---|---|
| 1 | audit_planning 回退 planning → frozen SSP/AP 標 superseded、輪次回 planning、重推產生新 snapshot、時間軸兩筆 | Task 4.2 `rollback_to_planning` + Task 4.4（superseded 標記機制，需 user 選 A/B）+ Task 4.3（advance 落表）+ Task 9（時間軸讀取） |
| 2 | auditing 回退 → AR 保留可見（帶標記）、回 audit_planning 可改 AP、重推後稽核員接續 | Task 4.2 `rollback_to_audit_planning` + Task 4.4（AR marker）；「重推後接續」驗證點＝BPMN revert_job 是否正確把 job 設回 PROCESSING（Task 6 §「待開工核實」項） |
| 3 | 非 manager 按回退 → 403；reason 空 → 400 | Task 6 `StageRollbackService.rollback_stage` Step 1/2 + Task 7 schema 層雙保險 |
| 4 | closed 輪次無回退入口（API 打也擋） | Task 6 Step 3（`rnd.status == "closed"` 擋）；FE 入口見 Task 10（Banner 條件式渲染，closed 輪次 banner 本就不顯示，見 `isFlowClosed` 既有邏輯） |
| 5 | 時間軸完整呈現多次往返（誰/何時/方向/原因），順序正確 | Task 2/9（`list_by_round` order by `created_at asc`）+ Task 7（`RoundStageTransitionsResponseSchema`） |
| 6 | 回退不物理刪任何資料（DB 對帳：僅新增標記欄/記錄，零 DELETE） | 全 plan 貫穿鐵則——Task 4.2 三個 `rollback_to_*` 方法皆用 `mark_*_superseded`（UPDATE status 欄，非 DELETE）+ Task 2 `round_stage_transitions` 只 INSERT；驗收時對帳腳本：`SELECT count(*) FROM pg_stat_user_tables WHERE relname IN (...) AND n_tup_del > 0`（migration 前後對比不應變動） |

---

## 12. 測試（依 `feature-test-planner` agent 或手動列，plan 階段先列覆蓋點）

**BE pytest**（`test/` 目錄，鏡像 `test_stage_advance_service.py` 的 MagicMock DI pattern）：
- `test_stage_rollback_service.py`：manager-only 403、reason 空 400、closed 輪次 412、
  handler 未註冊 412、正常路徑呼叫鏈（mock `stage_registry.get_rollback_handler` + mock
  `workflow_execution_service.revert_job`，斷言呼叫參數）
- `test_audit_round_app_service.py`（既有檔擴充，若無則新建）：三個 `rollback_to_*` 方法的
  前置狀態檢查（非對應 status 時 412）、`_sync_project_status` 有被呼叫、superseded 標記方法
  有被呼叫（mock 驗證，不測套件內部邏輯）

**FE E2E**（`compliance-manager-test` repo，本 plan 不展開，僅標記需求）：
- manager 對 auditing 輪次按「返回上一階段」→ 確認 UI 顯示 reason 必填 dialog → 送出後輪次
  狀態列顯示回到 audit_planning、時間軜出現新增一筆 rollback 記錄
- 非 manager 登入 → 「返回上一階段」按鈕不可見或點擊回 403 toast
