# Stage 4：多輪稽核 Implementation Plan > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** 實作多輪稽核歷史保留 — 每次 close-round 建立新一輪 ar_data，保留前一輪完整快照，支援 round 切換瀏覽與歷史 findings 參考。 **Architecture:** DDD 四層（Route → Serializer → App Service → Domain Service → Repo Impl）。close_round 在 OscalAuditService（app/project 層），AR 控制項查閱在 AuditService + GrcAuditRepoImpl（app/grc + infra/grc 層）。新增 ArRoundListResource route + _check_latest_round 寫入保護。 **Tech Stack:** Python 3.11, Flask-RESTful, SQLAlchemy, Marshmallow, dependency-injector **Spec:** `docs/design_spec/stage4-multi-round-audit-design.md` --- ## File Structure ### 修改檔案 | 檔案 | 變更摘要 | |------|---------| | `common/code/grc_error_code.py` | +1 error code (GRC_412008) | | `domain/grc/entities/grc_audit_entity.py` | GrcArControlEntity +is_inherited; GrcArControlDetailEntity +previous_findings; 新增 GrcRoundEntity | | `app/grc/dto/audit_dto.py` | ArControlDto +is_inherited; ArControlDetailDto +previous_findings; 新增 RoundDto, PreviousFindingDto | | `infra/grc/mapper/grc_audit_mapper.py` | to_ar_control_entity +is_inherited 參數; 新增 to_round_entity, to_previous_finding_entity | | `domain/grc/repository/i_grc_audit_repo.py` | list_ar_controls +run_no; 新增 list_rounds, get_previous_findings, check_latest_round, check_latest_round_by_finding | | `domain/grc/service/grc_audit_domain_service.py` | list_ar_controls +run_no; 新增 list_rounds, get_previous_findings, check_latest_round, check_latest_round_by_finding | | `infra/grc/repository/grc_audit_repo_impl.py` | _resolve_ar_data_id 支援 run_no; list_ar_controls +run_no+is_inherited; 新增 list_rounds, get_previous_findings, check_latest_round, check_latest_round_by_finding | | `app/grc/service/audit_service.py` | list_ar_controls +run_no; 新增 list_rounds, check_latest_round, check_latest_round_by_finding; get_ar_control_detail +previous_findings | | `api/grc/serializers/audit.py` | ArControlListRequestSchema +run_no; ArControlResponseSchema +is_inherited; ArControlDetailInnerSchema +previous_findings; 新增 RoundResponseSchema 等 | | `api/grc/serializers/poam.py` | CloseRoundResponseSchema +run_no, +inherited_count | | `api/grc/routes/audit_route.py` | 新增 ArRoundListResource; ArVerdictResource/ArFindingCreateResource/ArFindingDetailResource 加 _check_latest_round | | `api/grc/__init__.py` | 註冊 ArRoundListResource URL | | `app/project/service/oscal_audit_service.py` | close_round 完全重寫（建新 ar_data + 複製 ar_controls）; confirm_audit 改查最新 run_no | --- ## Task 1: Error Code + Entity 層擴充 新增 error code，擴充 entity 支援 is_inherited、previous_findings、round 資料。 **Files:** - Modify: `common/code/grc_error_code.py:74-77` - Modify: `domain/grc/entities/grc_audit_entity.py` - [ ] **Step 1: 新增 error code** 在 `common/code/grc_error_code.py` Close Round 區塊末尾新增： ```python # ── Multi-Round ───────────────────────────────────────────────────── GRC_AR_NOT_LATEST_ROUND = ("只能操作最新一輪的稽核結果", "GRC_412008") ``` - [ ] **Step 2: 擴充 GrcArControlEntity 加 is_inherited** 在 `domain/grc/entities/grc_audit_entity.py` 的 `GrcArControlEntity.__init__` 新增參數 `is_inherited=False`，並設為 `self.is_inherited = is_inherited`。 ```python class GrcArControlEntity: def __init__( self, ar_control_uid=None, control_id=None, control_title=None, verdict=None, confidence=None, rationale=None, remarks=None, findings_count=0, group_uid=None, group_name=None, is_inherited=False, ): self.ar_control_uid = ar_control_uid self.control_id = control_id self.control_title = control_title self.verdict = verdict self.confidence = confidence self.rationale = rationale self.remarks = remarks self.findings_count = findings_count self.group_uid = group_uid self.group_name = group_name self.is_inherited = is_inherited ``` - [ ] **Step 3: 擴充 GrcArControlDetailEntity 加 previous_findings** 在 `GrcArControlDetailEntity.__init__` 加 `previous_findings=None`，設為 `self.previous_findings = previous_findings or []`。 ```python class GrcArControlDetailEntity(GrcArControlEntity): def __init__( self, description=None, guidance=None, ssp_implementation_status=None, ssp_implementation_description=None, findings=None, assessment_objects=None, previous_findings=None, **kwargs, ): super().__init__(**kwargs) self.description = description self.guidance = guidance self.ssp_implementation_status = ssp_implementation_status self.ssp_implementation_description = ssp_implementation_description self.findings = findings or [] self.assessment_objects = assessment_objects or [] self.previous_findings = previous_findings or [] ``` - [ ] **Step 4: 新增 GrcRoundEntity 和 GrcPreviousFindingEntity** 在同一檔案末尾新增： ```python class GrcPreviousFindingEntity: """上一輪的 finding（唯讀）""" def __init__(self, uid=None, title=None, category=None, severity=None, description=None, recommendation=None, run_no=None): self.uid = uid self.title = title self.category = category self.severity = severity self.description = description self.recommendation = recommendation self.run_no = run_no class GrcRoundEntity: """單一輪次摘要""" def __init__(self, run_no=None, title=None, started_at=None, completed_at=None, total_controls=0, pass_count=0, fail_count=0, partial_count=0, na_count=0, pending_count=0, is_current=False): self.run_no = run_no self.title = title self.started_at = started_at self.completed_at = completed_at self.total_controls = total_controls self.pass_count = pass_count self.fail_count = fail_count self.partial_count = partial_count self.na_count = na_count self.pending_count = pending_count self.is_current = is_current ``` - [ ] **Step 5: Commit** ```bash git add common/code/grc_error_code.py domain/grc/entities/grc_audit_entity.py git commit -m "feat(stage4): add GRC_412008 error code, extend entities with is_inherited/previous_findings/round" ``` --- ## Task 2: DTO 層擴充 擴充 DTO 支援 is_inherited、previous_findings、round。 **Files:** - Modify: `app/grc/dto/audit_dto.py` - [ ] **Step 1: ArControlDto 加 is_inherited** 在 `ArControlDto` dataclass 加欄位 `is_inherited: bool = False`，並更新 `from_entity`： ```python @dataclass class ArControlDto: ar_control_uid: str control_id: str control_title: str verdict: Optional[str] = None confidence: Optional[int] = None findings_count: int = 0 group_uid: Optional[str] = None group_name: Optional[str] = None is_inherited: bool = False @staticmethod def from_entity(entity) -> "ArControlDto": return ArControlDto( ar_control_uid=entity.ar_control_uid, control_id=entity.control_id, control_title=entity.control_title, verdict=entity.verdict, confidence=entity.confidence, findings_count=entity.findings_count, group_uid=entity.group_uid, group_name=entity.group_name, is_inherited=entity.is_inherited, ) @staticmethod def from_entity_list(entities) -> list: return [ArControlDto.from_entity(e) for e in entities] ``` - [ ] **Step 2: 新增 PreviousFindingDto** ```python @dataclass class PreviousFindingDto: uid: str title: Optional[str] = None category: Optional[str] = None severity: Optional[str] = None description: Optional[str] = None recommendation: Optional[str] = None run_no: Optional[int] = None @staticmethod def from_entity(entity) -> "PreviousFindingDto": return PreviousFindingDto( uid=entity.uid, title=entity.title, category=entity.category, severity=entity.severity, description=entity.description, recommendation=entity.recommendation, run_no=entity.run_no, ) @staticmethod def from_entity_list(entities) -> list: return [PreviousFindingDto.from_entity(e) for e in entities] ``` - [ ] **Step 3: ArControlDetailDto 加 previous_findings** 在 `ArControlDetailDto` 加欄位 `previous_findings: List[PreviousFindingDto] = field(default_factory=list)`，更新 `from_entity`： ```python @dataclass class ArControlDetailDto(ArControlDto): description: Optional[str] = None guidance: Optional[str] = None rationale: Optional[str] = None remarks: Optional[str] = None ssp_implementation: Optional[SspImplementationDto] = None findings: List[ArFindingDto] = field(default_factory=list) assessment_objects: List[ArAoDto] = field(default_factory=list) previous_findings: List[PreviousFindingDto] = field(default_factory=list) @staticmethod def from_entity(entity) -> "ArControlDetailDto": return ArControlDetailDto( ar_control_uid=entity.ar_control_uid, control_id=entity.control_id, control_title=entity.control_title, verdict=entity.verdict, confidence=entity.confidence, findings_count=entity.findings_count, group_uid=entity.group_uid, group_name=entity.group_name, is_inherited=entity.is_inherited, description=entity.description, guidance=entity.guidance, rationale=entity.rationale, remarks=entity.remarks, ssp_implementation=SspImplementationDto( implementation_status=entity.ssp_implementation_status, implementation_description=entity.ssp_implementation_description, ) if entity.ssp_implementation_status else None, findings=[ArFindingDto.from_entity(f) for f in entity.findings], assessment_objects=[ArAoDto.from_entity(ao) for ao in entity.assessment_objects], previous_findings=[PreviousFindingDto.from_entity(pf) for pf in entity.previous_findings], ) ``` - [ ] **Step 4: 新增 RoundDto** ```python @dataclass class RoundDto: run_no: int title: Optional[str] = None started_at: Optional[datetime] = None completed_at: Optional[datetime] = None total_controls: int = 0 pass_count: int = 0 fail_count: int = 0 partial_count: int = 0 na_count: int = 0 pending_count: int = 0 is_current: bool = False @staticmethod def from_entity(entity) -> "RoundDto": return RoundDto( run_no=entity.run_no, title=entity.title, started_at=entity.started_at, completed_at=entity.completed_at, total_controls=entity.total_controls, pass_count=entity.pass_count, fail_count=entity.fail_count, partial_count=entity.partial_count, na_count=entity.na_count, pending_count=entity.pending_count, is_current=entity.is_current, ) @staticmethod def from_entity_list(entities) -> list: return [RoundDto.from_entity(e) for e in entities] ``` - [ ] **Step 5: Commit** ```bash git add app/grc/dto/audit_dto.py git commit -m "feat(stage4): extend DTOs with is_inherited, previous_findings, round" ``` --- ## Task 3: Mapper 層擴充 更新 GrcAuditMapper 支援新增的 entity 欄位。 **Files:** - Modify: `infra/grc/mapper/grc_audit_mapper.py` - [ ] **Step 1: to_ar_control_entity 加 is_inherited** ```python @staticmethod def to_ar_control_entity(ar_control, group_uid, group_name, findings_count=0, is_inherited=False) -> GrcArControlEntity: return GrcArControlEntity( ar_control_uid=str(ar_control.uid), control_id=ar_control.control_id, control_title=ar_control.control_title, verdict=ar_control.verdict.value if ar_control.verdict else None, confidence=ar_control.confidence, rationale=ar_control.rationale, remarks=ar_control.remarks, findings_count=findings_count, group_uid=group_uid, group_name=group_name, is_inherited=is_inherited, ) ``` - [ ] **Step 2: 新增 to_previous_finding_entity 和 to_round_entity** 在 mapper 新增 import 和兩個方法： ```python from domain.grc.entities.grc_audit_entity import ( GrcArControlEntity, GrcArFindingEntity, GrcArAoEvidenceEntity, GrcArAoEntity, GrcPreviousFindingEntity, GrcRoundEntity, ) # ... 在 class GrcAuditMapper 內新增： @staticmethod def to_previous_finding_entity(finding, run_no) -> GrcPreviousFindingEntity: return GrcPreviousFindingEntity( uid=str(finding.uid), title=finding.title, category=finding.category if isinstance(finding.category, str) else (finding.category.value if finding.category else None), severity=finding.severity if isinstance(finding.severity, str) else (finding.severity.value if finding.severity else None), description=finding.description, recommendation=finding.recommendation, run_no=run_no, ) @staticmethod def to_round_entity(ar_data, verdict_counts, max_run_no) -> GrcRoundEntity: return GrcRoundEntity( run_no=ar_data.run_no, title=ar_data.title, started_at=ar_data.started_at, completed_at=ar_data.completed_at, total_controls=verdict_counts.get("total", 0), pass_count=verdict_counts.get("pass", 0), fail_count=verdict_counts.get("fail", 0), partial_count=verdict_counts.get("partial", 0), na_count=verdict_counts.get("na", 0), pending_count=verdict_counts.get("pending", 0), is_current=(ar_data.run_no == max_run_no), ) ``` - [ ] **Step 3: Commit** ```bash git add infra/grc/mapper/grc_audit_mapper.py git commit -m "feat(stage4): extend mapper with is_inherited, previous_finding, round" ``` --- ## Task 4: Repository Interface 擴充 擴充 IGrcAuditRepo 介面，新增 run_no 參數和新方法。 **Files:** - Modify: `domain/grc/repository/i_grc_audit_repo.py` - [ ] **Step 1: 修改 list_ar_controls 加 run_no，新增 list_rounds / get_previous_findings / check_latest_round** ```python from abc import abstractmethod from typing import Optional, List from jedi_common.interfaces.entities import PageDataEntity from jedi_common.interfaces.spec.base_query_spec import PageSpec, SortSpec from domain.grc.entities.grc_audit_entity import ( GrcArControlEntity, GrcArControlDetailEntity, GrcArFindingEntity, GrcRoundEntity, GrcPreviousFindingEntity, ) class IGrcAuditRepo: @abstractmethod def list_ar_controls( self, ap_uid: str, pager: PageSpec, sorts: List[SortSpec], search: str = None, verdict: str = None, run_no: int = None, ) -> PageDataEntity: pass @abstractmethod def get_ar_control_detail(self, ap_uid: str, ar_control_uid: str) -> Optional[GrcArControlDetailEntity]: pass @abstractmethod def update_verdict( self, ar_control_uid: str, verdict: str, confidence: int, rationale: str, remarks: str, curr_user: str, ) -> Optional[GrcArControlEntity]: pass @abstractmethod def create_finding( self, ar_control_uid: str, category: str, severity: str, title: str, description: str, recommendation: str, ao_uid: str, curr_user: str, ) -> Optional[GrcArFindingEntity]: pass @abstractmethod def get_finding(self, finding_uid: str) -> Optional[GrcArFindingEntity]: pass @abstractmethod def update_finding( self, finding_uid: str, category: str, severity: str, title: str, description: str, recommendation: str, ao_uid: str, curr_user: str, ) -> Optional[GrcArFindingEntity]: pass @abstractmethod def delete_finding(self, finding_uid: str) -> bool: pass @abstractmethod def list_rounds(self, ap_uid: str) -> List[GrcRoundEntity]: pass @abstractmethod def get_previous_findings(self, ar_control_uid: str) -> List[GrcPreviousFindingEntity]: pass @abstractmethod def check_latest_round(self, ar_control_uid: str) -> bool: """回傳 True 表示 ar_control 屬於最新一輪；False 表示非最新""" pass @abstractmethod def check_latest_round_by_finding(self, finding_uid: str) -> bool: """透過 finding_uid 反查 ar_control，檢查是否屬於最新一輪""" pass ``` - [ ] **Step 2: Commit** ```bash git add domain/grc/repository/i_grc_audit_repo.py git commit -m "feat(stage4): extend IGrcAuditRepo with run_no, rounds, previous_findings, check_latest_round" ``` --- ## Task 5: Repository Impl — _resolve_ar_data_id 支援 run_no + list_rounds 修改核心查詢方法，支援指定 run_no 或預設查最新。新增 list_rounds。 **Files:** - Modify: `infra/grc/repository/grc_audit_repo_impl.py` - [ ] **Step 1: 修改 _resolve_ar_data_id 支援 run_no 參數** 將 `_resolve_ar_data_id(self, ap_uid)` 改為 `_resolve_ar_data_id(self, ap_uid, run_no=None)`。 ```python def _resolve_ar_data_id(self, ap_uid: str, run_no: int = None) -> Optional[int]: """ap_uid → AR → ARData.id run_no=None 時取最新一輪；指定 run_no 取該輪。 """ query = ( self.session.query(OscalAssessmentResultData.id) .join(OscalAssessmentResult, OscalAssessmentResult.id == OscalAssessmentResultData.assessment_result_id) .join(OscalAssessmentPlan, OscalAssessmentPlan.id == OscalAssessmentResult.assessment_plan_id) .filter(OscalAssessmentPlan.uid == ap_uid) ) if run_no is not None: query = query.filter(OscalAssessmentResultData.run_no == run_no) else: query = query.order_by(OscalAssessmentResultData.run_no.desc()) row = query.first() return row.id if row else None ``` - [ ] **Step 2: 新增 _get_max_run_no helper** ```python def _get_max_run_no(self, ap_uid: str) -> int: """查詢 AP 下最大 run_no""" row = ( self.session.query(sa_func.max(OscalAssessmentResultData.run_no)) .join(OscalAssessmentResult, OscalAssessmentResult.id == OscalAssessmentResultData.assessment_result_id) .join(OscalAssessmentPlan, OscalAssessmentPlan.id == OscalAssessmentResult.assessment_plan_id) .filter(OscalAssessmentPlan.uid == ap_uid) .first() ) return row[0] if row and row[0] else 0 ``` - [ ] **Step 3: 新增 _get_poam_control_ids_for_round helper** 用於判斷 is_inherited：查詢觸發當前輪次的 POA&M control_identifier 集合。 邏輯：找上一輪（run_no - 1）ar_data 下 fail/partial 控制項的 findings 對應的 POA&M。 這確保多輪累積的 POA&M 不會影響後續輪次的 is_inherited 判定。 ```python def _get_poam_control_ids_for_round(self, ap_id: int, ar_id: int, current_run_no: int) -> set: """取得觸發當前輪次覆核的 POA&M control_identifier 集合。 查詢上一輪 ar_data 下的 findings 所對應的 POA&M。 run_no=1 時回傳空集合。 """ if current_run_no <= 1: return set() from infra.grc.model.poam_model import PoamModel prev_run_no = current_run_no - 1 # 找上一輪 ar_data prev_ar_data = ( self.session.query(OscalAssessmentResultData) .filter( OscalAssessmentResultData.assessment_result_id == ar_id, OscalAssessmentResultData.run_no == prev_run_no, ) .first() ) if prev_ar_data is None: return set() # 找上一輪 ar_controls 的 findings prev_finding_ids = ( self.session.query(OscalAssessmentResultFinding.id) .join(OscalAssessmentResultControl, OscalAssessmentResultControl.id == OscalAssessmentResultFinding.assessment_result_control_id) .filter(OscalAssessmentResultControl.assessment_result_data_id == prev_ar_data.id) .subquery() ) # 查這些 findings 對應的 POA&M 的 control_identifier rows = ( self.session.query(PoamModel.control_identifier) .filter( PoamModel.assessment_plan_id == ap_id, PoamModel.ar_finding_id.in_(prev_finding_ids), ) .distinct() .all() ) return {r[0] for r in rows} ``` - [ ] **Step 4: 修改 list_ar_controls 支援 run_no + is_inherited** ```python def list_ar_controls(self, ap_uid, pager, sorts, search=None, verdict=None, run_no=None): ar_data_id = self._resolve_ar_data_id(ap_uid, run_no) page = pager.page if pager else 1 page_size = pager.page_size if pager else 25 if ar_data_id is None: meta = PageMetaEntity(paging=True, page=page, page_size=page_size, total=0, total_pages=0, has_next=False, has_prev=False) return PageDataEntity(meta=meta, data=[]) # 判斷 is_inherited：需要知道 run_no 和 POA&M 集合 actual_run_no = run_no if actual_run_no is None: # 查詢此 ar_data 的 run_no ar_data_row = self.session.query(OscalAssessmentResultData.run_no).filter( OscalAssessmentResultData.id == ar_data_id ).first() actual_run_no = ar_data_row[0] if ar_data_row else 1 poam_control_ids = set() if actual_run_no > 1: # 查 AP id 和 AR id ap_ar_row = ( self.session.query(OscalAssessmentPlan.id, OscalAssessmentResult.id) .join(OscalAssessmentResult, OscalAssessmentResult.assessment_plan_id == OscalAssessmentPlan.id) .filter(OscalAssessmentPlan.uid == ap_uid) .first() ) if ap_ar_row: poam_control_ids = self._get_poam_control_ids_for_round(ap_ar_row[0], ap_ar_row[1], actual_run_no) # findings count subquery findings_count_sq = ( self.session.query( OscalAssessmentResultFinding.assessment_result_control_id, sa_func.count(OscalAssessmentResultFinding.id).label("cnt"), ) .group_by(OscalAssessmentResultFinding.assessment_result_control_id) .subquery() ) query = ( self.session.query( OscalAssessmentResultControl, OscalAssessmentPlanGroup.uid.label("group_uid"), OscalAssessmentPlanGroup.name.label("group_name"), sa_func.coalesce(findings_count_sq.c.cnt, 0).label("findings_count"), ) .filter(OscalAssessmentResultControl.assessment_result_data_id == ar_data_id) .outerjoin( OscalAssessmentPlanControl, OscalAssessmentPlanControl.control_id == OscalAssessmentResultControl.control_id, ) .outerjoin(OscalAssessmentPlanGroup, OscalAssessmentPlanGroup.id == OscalAssessmentPlanControl.group_id) .outerjoin(findings_count_sq, findings_count_sq.c.assessment_result_control_id == OscalAssessmentResultControl.id) ) if search: like = f"%{search}%" query = query.filter( (OscalAssessmentResultControl.control_id.ilike(like)) | (OscalAssessmentResultControl.control_title.ilike(like)) ) if verdict: query = query.filter(OscalAssessmentResultControl.verdict == verdict) total = query.count() rows = query.order_by(OscalAssessmentResultControl.control_id).offset((page - 1) * page_size).limit(page_size).all() entities = [] for row in rows: arc = row[0] # is_inherited: run_no>1 且 verdict 非 null 且 control_id 不在 POA&M 集合 is_inherited = ( actual_run_no > 1 and arc.verdict is not None and arc.control_id not in poam_control_ids ) entities.append( GrcAuditMapper.to_ar_control_entity( arc, row.group_uid, row.group_name, row.findings_count, is_inherited ) ) meta = PageMetaEntity( paging=True, page=page, page_size=page_size, total=total, total_pages=math.ceil(total / page_size) if page_size else 0, has_next=page * page_size < total, has_prev=page > 1, ) return PageDataEntity(meta=meta, data=entities) ``` - [ ] **Step 5: 新增 list_rounds 方法** ```python def list_rounds(self, ap_uid): """查詢 AP 下所有 round 的摘要（含 verdict 統計）""" # 找 AR ar_row = ( self.session.query(OscalAssessmentResult.id) .join(OscalAssessmentPlan, OscalAssessmentPlan.id == OscalAssessmentResult.assessment_plan_id) .filter(OscalAssessmentPlan.uid == ap_uid) .first() ) if ar_row is None: return [] ar_datas = ( self.session.query(OscalAssessmentResultData) .filter(OscalAssessmentResultData.assessment_result_id == ar_row[0]) .order_by(OscalAssessmentResultData.run_no.asc()) .all() ) if not ar_datas: return [] max_run_no = max(d.run_no for d in ar_datas) entities = [] for ar_data in ar_datas: # 聚合 verdict 統計 controls = ( self.session.query(OscalAssessmentResultControl.verdict) .filter(OscalAssessmentResultControl.assessment_result_data_id == ar_data.id) .all() ) counts = {"total": len(controls), "pass": 0, "fail": 0, "partial": 0, "na": 0, "pending": 0} for (v,) in controls: if v is None: counts["pending"] += 1 else: key = v.value if hasattr(v, "value") else str(v) if key in counts: counts[key] += 1 entities.append(GrcAuditMapper.to_round_entity(ar_data, counts, max_run_no)) return entities ``` - [ ] **Step 6: Commit** ```bash git add infra/grc/repository/grc_audit_repo_impl.py git commit -m "feat(stage4): parameterize _resolve_ar_data_id, add list_rounds, is_inherited logic" ``` --- ## Task 6: Repository Impl — get_previous_findings + check_latest_round **Files:** - Modify: `infra/grc/repository/grc_audit_repo_impl.py` - [ ] **Step 1: 新增 get_previous_findings** 查詢邏輯：ar_control_uid → ar_data → run_no → run_no-1 的 ar_data → 同 control_id 的 ar_control → findings。 ```python def get_previous_findings(self, ar_control_uid): """查詢上一輪同 control_id 的 findings""" # 1. 找當前 ar_control ar_control = ( self.session.query(OscalAssessmentResultControl) .filter(OscalAssessmentResultControl.uid == ar_control_uid) .first() ) if ar_control is None: return [] # 2. 找所屬 ar_data 的 run_no ar_data = ( self.session.query(OscalAssessmentResultData) .filter(OscalAssessmentResultData.id == ar_control.assessment_result_data_id) .first() ) if ar_data is None or ar_data.run_no <= 1: return [] prev_run_no = ar_data.run_no - 1 # 3. 找上一輪的 ar_data prev_ar_data = ( self.session.query(OscalAssessmentResultData) .filter( OscalAssessmentResultData.assessment_result_id == ar_data.assessment_result_id, OscalAssessmentResultData.run_no == prev_run_no, ) .first() ) if prev_ar_data is None: return [] # 4. 找上一輪同 control_id 的 ar_control prev_ar_control = ( self.session.query(OscalAssessmentResultControl) .filter( OscalAssessmentResultControl.assessment_result_data_id == prev_ar_data.id, OscalAssessmentResultControl.control_id == ar_control.control_id, ) .first() ) if prev_ar_control is None: return [] # 5. 撈 findings findings = ( self.session.query(OscalAssessmentResultFinding) .filter(OscalAssessmentResultFinding.assessment_result_control_id == prev_ar_control.id) .all() ) return [GrcAuditMapper.to_previous_finding_entity(f, prev_run_no) for f in findings] ``` - [ ] **Step 2: 新增 check_latest_round** ```python def check_latest_round(self, ar_control_uid): """檢查 ar_control 是否屬於最新一輪的 ar_data""" ar_control = ( self.session.query(OscalAssessmentResultControl) .filter(OscalAssessmentResultControl.uid == ar_control_uid) .first() ) if ar_control is None: return False ar_data = ( self.session.query(OscalAssessmentResultData) .filter(OscalAssessmentResultData.id == ar_control.assessment_result_data_id) .first() ) if ar_data is None: return False max_run = ( self.session.query(sa_func.max(OscalAssessmentResultData.run_no)) .filter(OscalAssessmentResultData.assessment_result_id == ar_data.assessment_result_id) .scalar() ) return ar_data.run_no == max_run ``` - [ ] **Step 3: 新增 check_latest_round_by_finding** ```python def check_latest_round_by_finding(self, finding_uid): """透過 finding_uid 反查 ar_control，檢查是否屬於最新一輪""" finding = ( self.session.query(OscalAssessmentResultFinding) .filter(OscalAssessmentResultFinding.uid == finding_uid) .first() ) if finding is None: return False ar_control = ( self.session.query(OscalAssessmentResultControl) .filter(OscalAssessmentResultControl.id == finding.assessment_result_control_id) .first() ) if ar_control is None: return False return self.check_latest_round(str(ar_control.uid)) ``` - [ ] **Step 4: Commit** ```bash git add infra/grc/repository/grc_audit_repo_impl.py git commit -m "feat(stage4): add get_previous_findings, check_latest_round, check_latest_round_by_finding" ``` --- ## Task 7: Domain Service 擴充 透傳新方法到 domain service。 **Files:** - Modify: `domain/grc/service/grc_audit_domain_service.py` - [ ] **Step 1: 更新 domain service** ```python from typing import Optional, List from jedi_common.interfaces.entities import PageDataEntity from jedi_common.interfaces.spec.base_query_spec import PageSpec, SortSpec from domain.grc.entities.grc_audit_entity import ( GrcArControlEntity, GrcArControlDetailEntity, GrcArFindingEntity, GrcRoundEntity, GrcPreviousFindingEntity, ) from domain.grc.repository.i_grc_audit_repo import IGrcAuditRepo class GrcAuditDomainService: def __init__(self, grc_audit_repo: IGrcAuditRepo): self._repo = grc_audit_repo def list_ar_controls(self, ap_uid, pager, sorts, run_no=None, **filters) -> PageDataEntity: return self._repo.list_ar_controls( ap_uid, pager, sorts, search=filters.get("search"), verdict=filters.get("verdict"), run_no=run_no, ) def get_ar_control_detail(self, ap_uid, ar_control_uid) -> Optional[GrcArControlDetailEntity]: return self._repo.get_ar_control_detail(ap_uid, ar_control_uid) def update_verdict(self, ar_control_uid, verdict, confidence, rationale, remarks, curr_user): return self._repo.update_verdict(ar_control_uid, verdict, confidence, rationale, remarks, curr_user) def create_finding(self, ar_control_uid, category, severity, title, description, recommendation, ao_uid, curr_user): return self._repo.create_finding(ar_control_uid, category, severity, title, description, recommendation, ao_uid, curr_user) def get_finding(self, finding_uid): return self._repo.get_finding(finding_uid) def update_finding(self, finding_uid, category, severity, title, description, recommendation, ao_uid, curr_user): return self._repo.update_finding(finding_uid, category, severity, title, description, recommendation, ao_uid, curr_user) def delete_finding(self, finding_uid): return self._repo.delete_finding(finding_uid) def list_rounds(self, ap_uid) -> List[GrcRoundEntity]: return self._repo.list_rounds(ap_uid) def get_previous_findings(self, ar_control_uid) -> List[GrcPreviousFindingEntity]: return self._repo.get_previous_findings(ar_control_uid) def check_latest_round(self, ar_control_uid) -> bool: return self._repo.check_latest_round(ar_control_uid) def check_latest_round_by_finding(self, finding_uid) -> bool: return self._repo.check_latest_round_by_finding(finding_uid) ``` - [ ] **Step 2: Commit** ```bash git add domain/grc/service/grc_audit_domain_service.py git commit -m "feat(stage4): extend domain service with run_no, rounds, previous_findings, check_latest_round" ``` --- ## Task 8: App Service 擴充 在 AuditService 增加 run_no 參數、list_rounds、previous_findings 整合到 detail、check_latest_round。 **Files:** - Modify: `app/grc/service/audit_service.py` - [ ] **Step 1: 完整更新 AuditService** ```python from jedi_common.interfaces.dto import PageDataDto, PageDto from jedi_common.interfaces.spec.base_query_spec import PageSpec, SortSpec from jedi_common.session.database.db import transaction from app.grc.dto.audit_dto import ( ArControlDto, ArControlDetailDto, ArFindingDto, RoundDto, PreviousFindingDto, ) from domain.grc.service.grc_audit_domain_service import GrcAuditDomainService class AuditService: def __init__(self, grc_audit_domain_service: GrcAuditDomainService): self._domain_service = grc_audit_domain_service @transaction def list_ar_controls(self, ap_uid: str, pager: dict, sorts: list, run_no: int = None, **filters) -> PageDataDto: page_spec = PageSpec(**pager) if pager else None sorts_spec = [SortSpec(**s) for s in sorts] if sorts else None result = self._domain_service.list_ar_controls(ap_uid, page_spec, sorts_spec, run_no=run_no, **filters) return PageDataDto( meta=PageDto.from_entity(result.meta), data=ArControlDto.from_entity_list(result.data), ) @transaction def get_ar_control_detail(self, ap_uid: str, ar_control_uid: str): entity = self._domain_service.get_ar_control_detail(ap_uid, ar_control_uid) if entity is None: return None # 注入 previous_findings prev_findings = self._domain_service.get_previous_findings(ar_control_uid) entity.previous_findings = prev_findings return ArControlDetailDto.from_entity(entity) @transaction def list_rounds(self, ap_uid: str): entities = self._domain_service.list_rounds(ap_uid) return RoundDto.from_entity_list(entities) @transaction def check_latest_round(self, ar_control_uid: str) -> bool: return self._domain_service.check_latest_round(ar_control_uid) @transaction def check_latest_round_by_finding(self, finding_uid: str) -> bool: return self._domain_service.check_latest_round_by_finding(finding_uid) @transaction def update_verdict(self, ar_control_uid: str, verdict: str, confidence: int, rationale: str, remarks: str, curr_user: str): entity = self._domain_service.update_verdict(ar_control_uid, verdict, confidence, rationale, remarks, curr_user) if entity is None: return None return ArControlDto.from_entity(entity) @transaction def create_finding(self, ar_control_uid: str, curr_user: str, **data): entity = self._domain_service.create_finding( ar_control_uid, data["category"], data["severity"], data["title"], data["description"], data.get("recommendation"), data["ao_uid"], curr_user, ) if entity is None: return None return ArFindingDto.from_entity(entity) @transaction def get_finding(self, finding_uid: str): entity = self._domain_service.get_finding(finding_uid) if entity is None: return None return ArFindingDto.from_entity(entity) @transaction def update_finding(self, finding_uid: str, curr_user: str, **data): entity = self._domain_service.update_finding( finding_uid, data["category"], data["severity"], data["title"], data["description"], data.get("recommendation"), data["ao_uid"], curr_user, ) if entity is None: return None return ArFindingDto.from_entity(entity) @transaction def delete_finding(self, finding_uid: str): return self._domain_service.delete_finding(finding_uid) ``` - [ ] **Step 2: Commit** ```bash git add app/grc/service/audit_service.py git commit -m "feat(stage4): extend AuditService with run_no, list_rounds, previous_findings, check_latest_round" ``` --- ## Task 9: Serializer 擴充 更新 marshmallow schemas 支援新欄位。 **Files:** - Modify: `api/grc/serializers/audit.py` - Modify: `api/grc/serializers/poam.py` - [ ] **Step 1: ArControlListRequestSchema 加 run_no** ```python class ArControlListRequestSchema(RequestMetaSchema): filters = fields.Nested(ArControlFiltersSchema, missing={}, allow_none=True) run_no = fields.Integer(load_default=None, allow_none=True) ``` - [ ] **Step 2: ArControlResponseSchema 加 is_inherited** ```python class ArControlResponseSchema(Schema): ar_control_uid = fields.String() control_id = fields.String() control_title = fields.String() verdict = fields.String(allow_none=True) confidence = fields.Integer(allow_none=True) findings_count = fields.Integer() group_uid = fields.String(allow_none=True) group_name = fields.String(allow_none=True) is_inherited = fields.Boolean(dump_default=False) ``` - [ ] **Step 3: 新增 PreviousFindingResponseSchema** ```python class PreviousFindingResponseSchema(Schema): uid = fields.String() title = fields.String(allow_none=True) category = fields.String(allow_none=True) severity = fields.String(allow_none=True) description = fields.String(allow_none=True) recommendation = fields.String(allow_none=True) run_no = fields.Integer(allow_none=True) ``` - [ ] **Step 4: ArControlDetailInnerSchema 加 previous_findings** ```python class ArControlDetailInnerSchema(ArControlResponseSchema): description = fields.String(allow_none=True) guidance = fields.String(allow_none=True) rationale = fields.String(allow_none=True) remarks = fields.String(allow_none=True) ssp_implementation = fields.Nested(SspImplementationResponseSchema, allow_none=True) findings = fields.Nested(ArFindingResponseSchema, many=True, dump_default=[]) assessment_objects = fields.Nested(ArAoResponseSchema, many=True, dump_default=[]) previous_findings = fields.Nested(PreviousFindingResponseSchema, many=True, dump_default=[]) ``` - [ ] **Step 5: 新增 Round schemas** ```python class RoundResponseSchema(Schema): run_no = fields.Integer() title = fields.String(allow_none=True) started_at = fields.DateTime(allow_none=True) completed_at = fields.DateTime(allow_none=True) total_controls = fields.Integer(dump_default=0) pass_count = fields.Integer(dump_default=0) fail_count = fields.Integer(dump_default=0) partial_count = fields.Integer(dump_default=0) na_count = fields.Integer(dump_default=0) pending_count = fields.Integer(dump_default=0) is_current = fields.Boolean(dump_default=False) ``` 注意：此 API 不分頁，route 層直接使用 `RoundResponseSchema(many=True).dump(result)` 回傳陣列，不需額外的 envelope schema。 - [ ] **Step 6: 更新 CloseRoundResponseSchema 加 run_no + inherited_count** 在 `api/grc/serializers/poam.py`： ```python class CloseRoundResponseSchema(Schema): ap_uid = fields.String() status = fields.String() run_no = fields.Integer() reset_count = fields.Integer() inherited_count = fields.Integer() ``` - [ ] **Step 7: Commit** ```bash git add api/grc/serializers/audit.py api/grc/serializers/poam.py git commit -m "feat(stage4): extend serializers with run_no, is_inherited, previous_findings, round schemas" ``` --- ## Task 10: Route 層 — ArRoundListResource + _check_latest_round 新增 round 列表 route，在寫入 routes 加歷史輪次保護。 **Files:** - Modify: `api/grc/routes/audit_route.py` - Modify: `api/grc/__init__.py` - [ ] **Step 1: 在 audit_route.py 新增 _check_latest_round helper** 在檔案頂部 import 區域新增： ```python from api.grc.serializers.audit import ( # ... 既有 imports ... RoundResponseSchema, ) ``` 在 `_check_ap_status_auditing` 之後新增： ```python def _check_latest_round(ar_control_uid: str, audit_service: AuditService): """驗證 ar_control 屬於最新一輪，否則 412""" is_latest = audit_service.check_latest_round(ar_control_uid) if not is_latest: raise PreconditionFailedError(GrcErrorCode.GRC_AR_NOT_LATEST_ROUND) ``` - [ ] **Step 2: ArControlListResource.post 傳入 run_no** 修改 `ArControlListResource.post` 方法，取出 `run_no` 傳給 service： ```python class ArControlListResource(MethodResource): @doc(description="取得 AR 控制項列表（含 verdict 狀態）", tags=["GRC Audit"], params=AUTH_PARAMS) @use_kwargs(ArControlListRequestSchema, location="json", apply=False) @marshal_with(ArControlListResponseSchema, apply=False) @jwt_required() @inject def post( self, project_uid: str, ap_uid: str, audit_service: AuditService = Provide[Containers.grc_container.audit_service], ): payload = request.get_json(silent=True) or {} data = ArControlListRequestSchema().load(payload) pager = data.get("pager") sorts = data.get("sort") filters = data.get("filters") or {} run_no = data.get("run_no") result = audit_service.list_ar_controls(ap_uid, pager, sorts, run_no=run_no, **filters) return return_response(True, ArControlListResponseSchema().dump(result)) ``` - [ ] **Step 3: ArVerdictResource.put 加 _check_latest_round** 在 `_check_ap_status_auditing(ap_uid)` 之後加一行： ```python _check_latest_round(ar_control_uid, audit_service) ``` - [ ] **Step 4: ArFindingCreateResource.post 加 _check_latest_round** 在 `_check_ap_status_auditing(ap_uid)` 之後加： ```python _check_latest_round(ar_control_uid, audit_service) ``` - [ ] **Step 5: ArFindingDetailResource.put 和 .delete 加 _check_latest_round** finding 的 put/delete 需要透過 finding_uid 反查 ar_control。使用 Task 6 中已新增的 `check_latest_round_by_finding` 方法（已在 repo interface、repo impl、domain service、app service 四層完整定義）。 在 `ArFindingDetailResource` 的 `put` 和 `delete` 方法中，`_check_ap_status_auditing` 之後加： ```python # 透過 finding_uid 反查，驗證屬於最新一輪 is_latest = audit_service.check_latest_round_by_finding(finding_uid) if not is_latest: raise PreconditionFailedError(GrcErrorCode.GRC_AR_NOT_LATEST_ROUND) ``` - [ ] **Step 6: 新增 ArRoundListResource** ```python class ArRoundListResource(MethodResource): """GET /grc/project//ap//ar/rounds""" @doc(description="取得稽核輪次列表", tags=["GRC Audit"], params=AUTH_PARAMS) @marshal_with(RoundResponseSchema(many=True), apply=False) @jwt_required() @inject def get( self, project_uid: str, ap_uid: str, audit_service: AuditService = Provide[Containers.grc_container.audit_service], ): result = audit_service.list_rounds(ap_uid) return return_response(True, RoundResponseSchema(many=True).dump(result)) ``` - [ ] **Step 7: 在 api/grc/__init__.py 註冊 ArRoundListResource** import 加上 `ArRoundListResource`： ```python from api.grc.routes.audit_route import ( LaunchAuditResource, ConfirmAuditResource, ArControlListResource, ArControlDetailResource, ArVerdictResource, ArFindingCreateResource, ArFindingDetailResource, ArRoundListResource, ) ``` 在 Audit (AP-scoped) 區塊內加： ```python api.add_resource( ArRoundListResource, "/project//ap//ar/rounds", ) ``` - [ ] **Step 8: Commit** ```bash git add api/grc/routes/audit_route.py api/grc/__init__.py git commit -m "feat(stage4): add ArRoundListResource, _check_latest_round on write routes" ``` --- ## Task 11: OscalAuditService — close_round 重寫 將 Stage 3 的 close_round（重置 verdict）改為 Stage 4（建新 ar_data + 複製 ar_controls）。 **Files:** - Modify: `app/project/service/oscal_audit_service.py` - [ ] **Step 1: 重寫 close_round 方法** 完全替換 `close_round` 方法（從 `@transaction` 到 return）： ```python @transaction def close_round(self, ap_uid: str, curr_user) -> dict: session = get_session() # 1. 查 AP ap = self._ap_service.get_assessment_plan(uid=ap_uid) if ap is None: raise NotFound(GrcErrorCode.GRC_AP_NOT_FOUND) if ap.status != "remediation": raise PreconditionFailedError(GrcErrorCode.GRC_AP_NOT_REMEDIATION) # 2. 查該 AP 所有 POA&M from infra.grc.model.poam_model import PoamModel poams = ( session.query(PoamModel) .filter(PoamModel.assessment_plan_id == ap.id) .all() ) # 3. 驗證全部 closed not_closed = [p for p in poams if p.status != "closed"] if not_closed: raise BadRequestError(GrcErrorCode.GRC_POAM_NOT_ALL_CLOSED) # 4. 收集有 POA&M 的 control_identifiers poam_control_ids = {p.control_identifier for p in poams} # 5. 查當前 ar_data（最大 run_no = N） ar = self._ar_service.get_assessment_result(assessment_plan_id=ap.id) from jedi_oscal.infra.model.ar.assessment_result_data import OscalAssessmentResultData from jedi_oscal.infra.model.ar.assessment_result_control import OscalAssessmentResultControl curr_ar_data = ( session.query(OscalAssessmentResultData) .filter(OscalAssessmentResultData.assessment_result_id == ar.id) .order_by(OscalAssessmentResultData.run_no.desc()) .first() ) if curr_ar_data is None: raise PreconditionFailedError(GrcErrorCode.GRC_AR_NOT_FOUND) curr_run_no = curr_ar_data.run_no new_run_no = curr_run_no + 1 # 6. 建立新 ar_data (run_no = N+1) from jedi_oscal.domain.entity.ar.assessment_result_data_entity import AssessmentResultDataEntity new_ar_data = AssessmentResultDataEntity( uid=str(uuid.uuid4()), assessment_result_id=ar.id, run_no=new_run_no, title="覆核", started_at=datetime.now(), ) new_ar_data = self._ar_data_service.add_assessment_result_data(new_ar_data, curr_user) # 7. 複製 ar_controls 到新一輪 curr_controls = ( session.query(OscalAssessmentResultControl) .filter(OscalAssessmentResultControl.assessment_result_data_id == curr_ar_data.id) .all() ) from jedi_oscal.domain.entity.ar.assessment_result_control_entity import AssessmentResultControlEntity reset_count = 0 inherited_count = 0 for arc in curr_controls: if arc.control_id in poam_control_ids: # 有 POA&M → verdict=null，清除 confidence/rationale/remarks new_arc = AssessmentResultControlEntity( uid=str(uuid.uuid4()), assessment_result_data_id=new_ar_data.id, control_id=arc.control_id, control_title=arc.control_title, verdict=None, ) reset_count += 1 else: # 無 POA&M → 完整繼承 new_arc = AssessmentResultControlEntity( uid=str(uuid.uuid4()), assessment_result_data_id=new_ar_data.id, control_id=arc.control_id, control_title=arc.control_title, verdict=arc.verdict.value if arc.verdict else None, confidence=arc.confidence, rationale=arc.rationale, remarks=arc.remarks, ) inherited_count += 1 self._ar_control_service.add_assessment_result_control(new_arc, curr_user) # 8. AP status → auditing self._ap_service.update_assessment_plan(ap.id, {"status": "auditing"}, curr_user) return { "ap_uid": ap_uid, "status": "auditing", "run_no": new_run_no, "reset_count": reset_count, "inherited_count": inherited_count, } ``` - [ ] **Step 2: Commit** ```bash git add app/project/service/oscal_audit_service.py git commit -m "feat(stage4): rewrite close_round to create new ar_data with control inheritance" ``` --- ## Task 12: OscalAuditService — confirm_audit 改查最新 run_no **Files:** - Modify: `app/project/service/oscal_audit_service.py` - [ ] **Step 1: 修改 confirm_audit 中的 ar_data 查詢** 將 confirm_audit 方法中的 hardcoded `run_no == 1` 改為查最新 run_no。 找到這段（約 line 116-120）： ```python ar_data = ( session.query(OscalAssessmentResultData) .filter(OscalAssessmentResultData.assessment_result_id == ar.id, OscalAssessmentResultData.run_no == 1) .first() ) ``` 替換為： ```python ar_data = ( session.query(OscalAssessmentResultData) .filter(OscalAssessmentResultData.assessment_result_id == ar.id) .order_by(OscalAssessmentResultData.run_no.desc()) .first() ) ``` - [ ] **Step 2: Commit** ```bash git add app/project/service/oscal_audit_service.py git commit -m "feat(stage4): confirm_audit queries latest run_no instead of hardcoded 1" ``` --- ## Task 13: Repo Impl — get_ar_control_detail 整合 previous_findings + is_inherited 更新 detail 查詢，回傳 previous_findings 和 is_inherited。 **Files:** - Modify: `infra/grc/repository/grc_audit_repo_impl.py` - [ ] **Step 1: 修改 get_ar_control_detail** 更新 `get_ar_control_detail` 方法，使其查詢 ar_control 時不再依賴 `_resolve_ar_data_id`（因為 detail 是用 uid 查的），但需要判斷 is_inherited。 現有邏輯先用 `_resolve_ar_data_id` 取 ar_data_id 再過濾 ar_control，這保證 ar_control 屬於該 AP。改為：從 ar_control_uid 直接查 ar_control，然後驗證它屬於正確的 AP。同時計算 is_inherited。 ```python def get_ar_control_detail(self, ap_uid, ar_control_uid): # 查 ar_control ar_control = ( self.session.query(OscalAssessmentResultControl) .filter(OscalAssessmentResultControl.uid == ar_control_uid) .first() ) if ar_control is None: return None # 驗證屬於正確的 AP（透過 ar_data → ar → ap） ar_data = ( self.session.query(OscalAssessmentResultData) .filter(OscalAssessmentResultData.id == ar_control.assessment_result_data_id) .first() ) if ar_data is None: return None ar = ( self.session.query(OscalAssessmentResult) .filter(OscalAssessmentResult.id == ar_data.assessment_result_id) .first() ) if ar is None: return None ap = ( self.session.query(OscalAssessmentPlan) .filter(OscalAssessmentPlan.id == ar.assessment_plan_id, OscalAssessmentPlan.uid == ap_uid) .first() ) if ap is None: return None # is_inherited 判斷 is_inherited = False if ar_data.run_no > 1 and ar_control.verdict is not None: poam_control_ids = self._get_poam_control_ids_for_round(ap.id, ar.id, ar_data.run_no) is_inherited = ar_control.control_id not in poam_control_ids # group info ap_ctrl = ( self.session.query(OscalAssessmentPlanControl, OscalAssessmentPlanGroup) .join(OscalAssessmentPlanGroup, OscalAssessmentPlanGroup.id == OscalAssessmentPlanControl.group_id) .filter(OscalAssessmentPlanControl.control_id == ar_control.control_id) .first() ) group_uid = str(ap_ctrl[1].uid) if ap_ctrl else None group_name = ap_ctrl[1].name if ap_ctrl else None # findings findings_rows = ( self.session.query(OscalAssessmentResultFinding) .filter(OscalAssessmentResultFinding.assessment_result_control_id == ar_control.id) .all() ) findings = [GrcAuditMapper.to_finding_entity(f) for f in findings_rows] return GrcArControlDetailEntity( ar_control_uid=str(ar_control.uid), control_id=ar_control.control_id, control_title=ar_control.control_title, verdict=ar_control.verdict.value if ar_control.verdict else None, confidence=ar_control.confidence, rationale=ar_control.rationale, remarks=ar_control.remarks, findings_count=len(findings), group_uid=group_uid, group_name=group_name, is_inherited=is_inherited, description=ap_ctrl[0].description if ap_ctrl else None, guidance=ap_ctrl[0].guidance if ap_ctrl else None, ssp_implementation_status=None, ssp_implementation_description=None, findings=findings, assessment_objects=[], ) ``` - [ ] **Step 2: Commit** ```bash git add infra/grc/repository/grc_audit_repo_impl.py git commit -m "feat(stage4): update get_ar_control_detail with is_inherited calculation" ``` --- ## Task 14: 整合驗證 驗證語法正確，import 無誤。 **Files:** - All modified files - [ ] **Step 1: Python 語法檢查** ```bash python -c " import py_compile, sys files = [ 'common/code/grc_error_code.py', 'domain/grc/entities/grc_audit_entity.py', 'app/grc/dto/audit_dto.py', 'infra/grc/mapper/grc_audit_mapper.py', 'domain/grc/repository/i_grc_audit_repo.py', 'domain/grc/service/grc_audit_domain_service.py', 'infra/grc/repository/grc_audit_repo_impl.py', 'app/grc/service/audit_service.py', 'api/grc/serializers/audit.py', 'api/grc/serializers/poam.py', 'api/grc/routes/audit_route.py', 'api/grc/__init__.py', 'app/project/service/oscal_audit_service.py', ] ok = True for f in files: try: py_compile.compile(f, doraise=True) except py_compile.PyCompileError as e: print(f'FAIL {f}: {e}') ok = False if ok: print('All files compile OK') else: sys.exit(1) " ``` Expected: `All files compile OK` - [ ] **Step 2: 確認 check_latest_round_by_finding DDD 鏈完整** 搜尋確認以下方法都已實作且簽名一致： - `IGrcAuditRepo.check_latest_round_by_finding(finding_uid: str) -> bool` (Task 4) - `GrcAuditRepoImpl.check_latest_round_by_finding(finding_uid)` (Task 6) - `GrcAuditDomainService.check_latest_round_by_finding(finding_uid)` (Task 7) - `AuditService.check_latest_round_by_finding(finding_uid)` (Task 8) ```bash grep -n "check_latest_round_by_finding" domain/grc/repository/i_grc_audit_repo.py infra/grc/repository/grc_audit_repo_impl.py domain/grc/service/grc_audit_domain_service.py app/grc/service/audit_service.py api/grc/routes/audit_route.py ``` Expected: 至少 5 處匹配（4 層定義 + route 呼叫）。 - [ ] **Step 4: 確認 DI container 不需修改** `AuditService.__init__` 和 `GrcAuditDomainService.__init__` 簽名不變（只加方法，不加建構子參數），DI container 不需更新。 - [ ] **Step 3: 確認所有新增 import 無遺漏** 重點確認： - `grc_audit_entity.py` 匯出 `GrcPreviousFindingEntity`, `GrcRoundEntity` - `grc_audit_mapper.py` import 這兩個 entity - `i_grc_audit_repo.py` import 這兩個 entity - `audit_route.py` import `RoundResponseSchema` - `api/grc/__init__.py` import `ArRoundListResource` --- ## 變更摘要 ### 新增 API | Method | URL | 說明 | |--------|-----|------| | GET | `/api/1.0/grc/project//ap//ar/rounds` | 稽核輪次列表 | ### 修改 API | Method | URL | 變更 | |--------|-----|------| | POST | `.../ar/controls/list` | Request +run_no; Response +is_inherited | | GET | `.../ar/control/` | Response +previous_findings | | PUT | `.../ar/control//verdict` | 加 _check_latest_round 保護 | | POST | `.../ar/control//findings` | 加 _check_latest_round 保護 | | PUT/DELETE | `.../ar/finding/` | 加 _check_latest_round 保護 | | POST | `.../close-round` | Response +run_no, +inherited_count；邏輯重寫 | | POST | `.../confirm-audit` | 內部改查最新 run_no |