FR-038 Wave 1 — jedi-oscal-v2 套件實作 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: 從零建立 jedi-oscal-v2 套件，把 FR-038 契約層定稿的 OSCAL v1.2.2 關聯式 schema 以 DDD 四層（entity / model / mapper / repository / service）落地 —— 本期涵蓋稽核生命週期主流程的 6 個 root（catalog / profile / ssp / ap / ar / poam）+ 共用層 + 全部 delta 表，並提供 snapshot / clone / profile-resolution / AR 全量判定矩陣 / 匯入匯出等 primitive 供主專案使用。

本期不做（follow-up）：component-definition（component_definitions + 5 個 cd_* 表）。它是「可重用元件描述」的獨立 OSCAL model，不在 Catalog→Profile→SSP→AP→AR→POA&M 稽核主流程上，CMMC 內部稽核流程用不到（design §1 四邊界與 PM 教材皆未涉及）。等有「元件庫」需求再開。本期落地表數 = base 36（42 − 6 個 cd_*）+ delta 6 = 42 表 + compliance.project_audit_rounds（後者僅建 DB，Wave 2 才建 model）。

Architecture: DDD + Hexagonal，沿用舊 jedi-oscal 的分層慣例（domain/ entity+repo interface+service、infra/ model+mapper+repo impl+adapter、app/ app service+dto、ports/ parser、common/ enum+code）。Repo 繼承 jedi_common.session.database.repository.base_repository_impl.BaseRepositoryImpl；session 由主專案 @transaction 開啟，套件 method 為被呼叫端。import 名 jedi_oscal_v2，dist jedi-oscal-v2。dev 走 poetry path dependency，feature 完成才推 Nexus。

Tech Stack: Python 3.11+ / SQLAlchemy 2.0 (mapped_column) / PostgreSQL (schema oscal + compliance) / Poetry / pytest / jedi-common (BaseRepositoryImpl, BaseModel, exception, transaction)。

Scope: 本計畫只涵蓋 Wave 1（jedi-oscal-v2 套件）。主專案 BE（Wave 2）、FE（Wave 3）另立計畫。契約基準：design.md + oscal-v2-deltas.sql + requirement-analysis.md。

計畫撰寫到開工可能有時間差；method 改名 / 欄位改 shape / base 套件 signature 變動都可能發生。每個 module agent 開工第一步：

• 讀 jedi_common.session.database.repository.base_repository_impl.BaseRepositoryImpl 的實際 signature（__init__(mapper, model)、add/update/get_by_uid/get_all/delete 簽章、update(locale=...) 是否存在）
• 讀舊 jedi-oscal（~/Projects/Jedicogy/module/jedi-python-package/jedi-oscal/）對應 module 的 entity/mapper/repo 寫法當「同款 pattern」參考（只參考寫法，不抄欄位 — schema 已改）
• 確認 delta SQL 已套到 dev DB（見 Phase 0 Task 0.2），用 \d oscal.<table> 對照欄位名再寫 model

Phase 0  套件骨架 + DB 套用 + 共用層          【序列瓶頸，先做完】
   │
   ├─ A1 catalog 樹 + framework + profile + resolution   ┐
   ├─ A2 SSP + 12 子表 + UI→OSCAL 落點                    │ 契約定稿後
   ├─ A3 AP + reviewed-controls + subjects + tasks       ├ 5 模組可平行
   ├─ A4 AR 全量矩陣 + risk + remediation + milestone + POA&M │ （各自獨立 model）
   └─ A5 parser adapter（CMMC/ISO/NIST）+ 匯入匯出        ┘
   │
A6  OscalSnapshotService（clone/snapshot primitives）   【依賴 A1+A2，故排在其後】
   │
Phase Z  套件層整合 smoke（全 model import + DI wire 驗證）  【序列收尾】

A6 為什麼不在 A1-A5 平行群：clone_resource_library 要同時複製 catalog(A1)+profile(A1)+ssp(A2) 三件組、snapshot_ssp 要深複製 SSP 12 子表(A2)，橫跨 A1+A2 的 entity，不屬單一模組。故排在 A1+A2 的 model+mapper 落地後做（可由 A2 的 agent 接手，因它已有 SSP deep-copy 程式）。A5 的 export round-trip 測試 fixture 依賴 A6 產出的 cloned/frozen 物件，A6 完成後再補 A5 export 測試。

A1-A5 為什麼能平行：各模組對應不同 OSCAL model 的表，entity/model/mapper/repo/service 檔案不重疊；跨模組只靠「token / uuid 軟參照」（不建 FK 到別模組的 Python 物件），無共享可變狀態。唯一共用 = Phase 0 的 base class / common enum / DI container 骨架，故 Phase 0 必須先完成。 A4 內部順序：AR finding → risk → remediation → milestone 有 FK 依賴，同一 agent 內依序做；POA&M item 可獨立。 A1 的 catalog 是 A2/A3/A4 的「token 來源」但只是 token 字串參照，不阻塞平行（A2-A5 用測試假資料的 control_id token 即可）。

~/Projects/Jedicogy/module/jedi-python-package/jedi-oscal-v2/
├── pyproject.toml                         # dist=jedi-oscal-v2, packages=jedi_oscal_v2
├── jedi_oscal_v2/
│   ├── common/
│   │   ├── enum/oscal_enums.py            # StrEnum: RoundStatus, RoundType, FindingState, RiskStatus, ...
│   │   └── code/oscal_v2_error_code.py    # 套件層 ErrorCode（OSCAL_V2_*）
│   ├── domain/
│   │   ├── entity/<model>/...             # @dataclass entity（純資料，無 ORM）
│   │   ├── repository/<model>/..._repo.py # abstract repo interface
│   │   └── service/<model>/..._service.py # domain service（業務邏輯）
│   ├── infra/
│   │   ├── model/<model>/...              # SQLAlchemy model（mapped_column）
│   │   ├── mapper/<model>/..._mapper.py   # entity ↔ model 雙向
│   │   ├── repository/<model>/..._impl.py # BaseRepositoryImpl 衍生
│   │   └── adapter/{cmmc,iso,nist}/       # parser adapter（A5）
│   ├── app/
│   │   ├── service/<model>/...            # app service（對外，主專案 import 點）
│   │   └── dto/<model>/...                # app DTO
│   └── ports/oscal_parser_factory.py      # parser factory（A5）
└── tests/
    └── <model>/test_*.py

<model> ∈ framework / catalog / profile / ssp / ap / ar / poam / common（共用 metadata/roles/parties/resources 歸 common）。

3.1 Entity（domain，純 dataclass）

from dataclasses import dataclass, field
@dataclass
class CatalogControlEntity:
    id: int | None = None
    uid: str | None = None            # 若該表有對外 uid（root 有 uuid；子物件用自身 uuid）
    control_id: str | None = None     # OSCAL token
    title: str | None = None
    # ... 對齊 delta/base 欄位；JSONB 欄用 dict | list | None

3.2 Model（infra，SQLAlchemy 2.0）

from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy import BigInteger, Text, ForeignKey
from jedi_common.session.database.base_model import BaseModel  # pre-flight 確認實際 base
class CatalogControlModel(BaseModel):
    __tablename__ = "catalog_controls"
    __table_args__ = {"schema": "oscal"}
    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    catalog_id: Mapped[int] = mapped_column(BigInteger, ForeignKey("oscal.catalogs.id", ondelete="CASCADE"))
    control_id: Mapped[str] = mapped_column(Text, nullable=False)
    # Python 保留字：class → mapped_column("class", ...) attr class_；metadata → metadata_id + attr metadata_

• NOT NULL ⟺ delta/base 標 [1]；JSONB 用 mapped_column(JSONB)。
• 跨 schema FK（如 compliance.project_audit_rounds → oscal.*）字串必帶 schema 前綴（記憶 feedback_cross_schema_fk_must_qualify）。

3.3 Repo（infra）

class CatalogControlRepoImpl(ICatalogControlRepo,
        BaseRepositoryImpl[CatalogControlEntity, CatalogControlQuery, CatalogControlModel, CatalogControlMapper]):
    def __init__(self):
        super().__init__(mapper=CatalogControlMapper, model=CatalogControlModel)
    # self.session 自動有（lazy）；禁止 __init__ 寫 self.session = get_session()

3.4 App Service（對外）

class CatalogService:
    def __init__(self, catalog_domain_service, ...):
        self._svc = catalog_domain_service
    # 注意：套件 app service 不自己加 @transaction —— 由主專案 caller 開 scope。
    # 但套件可提供「被 @transaction 包住才安全」的 method，docstring 標明「caller 必須在 @transaction scope 內」。

3.5 測試

• 每個 model：entity↔︎model mapper round-trip test + repo CRUD test（用 dev DB 或 in-memory 視 base 慣例）。
• app service test：加 logger patch autouse fixture（記憶 feedback_test_logger_patch_db_handler），patch jedi_oscal_v2.app.service.<m>.<svc>.logger。
• commit 顆粒：一個表的 entity+model+mapper+repo+test 綠 → 一個 commit。

Task 0.1: 建立套件骨架 + pyproject

Files:

• Create: ~/Projects/Jedicogy/module/jedi-python-package/jedi-oscal-v2/pyproject.toml
• Create: jedi-oscal-v2/jedi_oscal_v2/__init__.py（+ 各層 __init__.py）

• Step 1: 複製舊 jedi-oscal 的 pyproject.toml 當底，改 name="jedi-oscal-v2"、packages=[{include="jedi_oscal_v2"}]、version 2.0.0、保留 jedi-common 依賴與 Nexus source 設定。
• Step 2: 建立 §2 的目錄樹（空 __init__.py）。
• Step 3: 在主專案 pyproject.toml 加 path dependency（dev-only，不 commit，比照現有 jedi-oscal 第 95 行）： jedi-oscal-v2 = { path = "/Users/chouraymond/Projects/Jedicogy/module/jedi-python-package/jedi-oscal-v2", develop = true } 並在 [project].dependencies 暫加 "jedi-oscal-v2"（pin 版本待發版才填）。
• Step 4: poetry update jedi-oscal-v2（不用 poetry lock，記憶 feedback_poetry_update_only）。
• Step 5: smoke：python -c "import jedi_oscal_v2" 無錯 → commit（套件 repo）。

Task 0.2: DB schema 套用（dev，drop & rebuild）

• Step 1: 備份 dev oscal schema（pg_dump -n oscal；D1 已說 dev 可任意，仍備份）。
• Step 2: psql --single-transaction -v ON_ERROR_STOP=1 用 cmmgr 跑 DROP SCHEMA oscal CASCADE; → base oscal-catalog-and-roots-schema-v2.sql → oscal-v2-deltas.sql（記憶 feedback_sql_migration_use_cmmgr）。
• Step 3: 驗證 DB：\dt oscal.* 應有 48 表（base 42 + delta 6：frameworks/framework_versions/ap_assessment_subjects/assessment_finding_risks/assessment_remediations/poam_milestones）+ \dt compliance.project_audit_rounds。注意：DB 有 48 表，但本期套件只建 42 表的 Python model（component_definitions + 5 個 cd_* 共 6 表存在於 DB、本期不建 model，見 Goal 的「本期不做」）。
• Step 4: 驗 CHECK 約束存在（\d compliance.project_audit_rounds 看 ck_audit_rounds_*）。
• Step 5: INSERT public.schema_migrations（記憶 SQL 鐵則）。dev 套用不 commit code，schema 已在 DB。

Task 0.3: common 層（enum + error code）

Files: Create jedi_oscal_v2/common/enum/oscal_enums.py、common/code/oscal_v2_error_code.py

• Step 1: 寫 StrEnum：RoundType(initial/close_out/surveillance)、RoundStatus(7 態)、FindingState(met/not_met/pending)（對 OSCAL satisfied/not-satisfied + 產品 pending）、RiskStatus(open/investigating/remediating/closed)、MilestoneStatus(open/in_progress/done)、PublishStatus(draft/published/deprecated)。
• Step 2: error code OSCAL_V2_*（套件層；HTTP 對映用 jedi-common exception）。
• Step 3: test：import 全 enum 值正確 → commit。

Task 0.4: BaseEntity / BaseQuery / mapper 基底（若舊套件有共用基底就 port）

• Step 1: pre-flight 讀舊套件是否有 BaseEntity/BaseMapper/audit mixin 慣例，port 一份到 v2（DRY）。
• Step 2: 建 DI container 骨架 jedi_oscal_v2（若套件自帶 container；否則主專案 Wave 2 wire）。
• Step 3: commit。

這是「一張表」的標準 5 步 TDD micro-cycle。A1-A5 的每個表 task 都展開成這 5 步（agent 執行時照此 pattern，不需在計畫重抄）。以 framework_versions 為例：

• Step 1（test first）: tests/framework/test_framework_version_mapper.py：建 entity → mapper.from_entity → model → mapper.to_entity round-trip，assert 欄位相等（含 JSONB、含 parent_id/catalog_id 軟參照）。
• Step 2: 跑測試 → FAIL（mapper/model 未建）。
• Step 3: 建 entity → model（對齊 \d oscal.framework_versions）→ mapper → repository interface + impl。
• Step 4: 跑 mapper test + repo CRUD test（add/get_by_uid/list/update/delete）→ PASS。
• Step 5: commit feat(jedi-oscal-v2): framework_version model+mapper+repo。

JSONB 欄：entity 用 dict|list|None，mapper 直通，不展開。 自巢狀表（catalog_groups/controls/parts、ap_tasks）：parent_id self-FK，repo 提供 get_children(parent_id) + get_tree(root_id)。

負責 agent：A1。Files: */common/*、*/framework/*、*/catalog/*、*/profile/*、tests/{common,framework,catalog,profile}/*

⚠️ A1 是所有 root 的共同前置：OSCAL 共用層 oscal.metadata / roles / parties / resources 被每個 root（catalog/profile/ssp/ap/ar/poam）的 model FK 參照（metadata_id 等）。A1 必須先建這 4 張 common 表的 model，A2-A5 才能建各自 root（依賴 A1 的 metadata model）。這使「A1 先行、A2-A5 再平行」成為硬順序。

A1-0 表清單：OSCAL 共用層（§4 pattern，A1 第一批做）

• metadata（含 revisions/locations/responsible_parties/document_ids/actions/props/links JSONB）
• roles（metadata_id FK）
• parties（metadata_id FK、email_addresses text[]、location_uuids/member_of_organizations uuid[]）
• resources（back-matter，metadata_id FK）

A1 表清單（各表照 §4 canonical pattern 各一輪 TDD + commit）

• frameworks、framework_versions（含 parent_id 血緣、catalog_id 回填、main_version_id）
• catalogs（含 framework_version_id、status）
• catalog_groups（self-FK 樹）
• catalog_controls（self-FK enhancement、source_identifier、catalog_group_id）
• catalog_control_parts（AO 在此；self-FK；idx_catalog_parts_ao 查詢）
• catalog_control_params
• profiles、profile_imports

A1 特殊邏輯（非 boilerplate，full TDD）

Task A1-R: Profile Resolution（邊界①）

Files: Create domain/service/profile/profile_resolution_service.py、tests/profile/test_profile_resolution.py

• Step 1（test）: 建 catalog（2 group / 4 control / 含 1 enhancement）+ profile（profile_imports include-controls with-ids=[ac-1,ac-2]）→ resolve(profile_id) → assert 回傳 resolved control set 只含 ac-1/ac-2（+ enhancement 視 with-child-controls）。
• Step 2: FAIL。
• Step 3: 實作 resolution：讀 profile_imports（include-all / include-controls.with-ids / matching / exclude），對 source catalog 的 controls 過濾，回傳 resolved set（回傳內容，非只 id，為 clone 鋪路）。
• Step 4: PASS（含 edge：include-all、exclude、with-child-controls=true 帶 enhancement）。
• Step 5: commit。

Task A1-S: AO 查詢

• list_aos(catalog_control_id) → 用 idx_catalog_parts_ao（name ∈ assessment-objective/method/objects）回該 control 全 AO；test 驗證 method prop（EXAMINE/INTERVIEW/TEST）解析。

Task A1-App: FrameworkService / CatalogService / ProfileService（app 層）

• add/get/list/update_framework、add_framework_version、publish_version、get_control_tree、list_aos、add_profile、resolve_profile — 各一 app service test（含 logger patch fixture）。

負責 agent：A2。Files: */ssp/*、tests/ssp/*

A2 表清單（§4 pattern）

• system_security_plans（root、import_profile_id）
• D1: ssp_system_characteristics（1:1，多欄）、ssp_information_types、ssp_diagrams
• D2: ssp_system_implementations(1:1)、ssp_leveraged_authorizations、ssp_system_users、ssp_components、ssp_inventory_items
• D3: ssp_control_implementations(1:1)、ssp_implemented_requirements、ssp_statements、ssp_by_components（雙父擇一 FK）

A2 特殊邏輯

Task A2-SoA: SoA / 實作狀態落點（requirement-analysis §5）

• test：給定一條控制的 SoA（applicability + inclusion-justification）+ implementation-status + 實作描述 + 證據連結 → 寫入 implemented_requirements(props) + statements.by_components(description+links) → 讀回正確。對齊 requirement-analysis §5 表。

Task A2-App: SspService + 子物件 CRUD app service（含 deep_clone_ssp 的 entity 級複製能力，供 A4/Wave2 用）

負責 agent：A3。Files: */ap/*、tests/ap/*

A3 表清單（§4 pattern）

• assessment_plans（root、import_ssp_id）
• ap_reviewed_controls(1:1)、ap_local_objectives、ap_assessment_activities、ap_tasks(self-FK)
• ap_assessment_subjects（delta D-3，抽查名單）

A3 特殊邏輯

Task A3-Draft: AP 草稿自動生成（requirement-analysis §4.5）

• test：給定 SSP 快照（含適用性表）→ generate_draft_from_ssp(ssp_snapshot_id) → 產出 AP + ap_reviewed_controls（= 適用控制集）+ 空 subjects/tasks 待稽核員補。assert reviewed-controls = 該 SSP 的 applicable 控制。

Task A3-App: AssessmentPlanService（create / generate_draft / set_reviewed_controls / set_assessment_subjects / set_tasks）

負責 agent：A4（內部依序：finding→risk→remediation→milestone）。Files: */ar/*、*/poam/*、common/(observation/risk/finding 共用)、tests/{ar,poam}/*

A4 表清單（§4 pattern）

• assessment_results(root、import_ap_id)、ar_results
• 共用：assessment_observations、assessment_risks、assessment_findings（雙 owner ar_result/poam 擇一）
• delta：assessment_finding_risks(D-4)、assessment_remediations(D-4b.1)、poam_milestones(D-4b.2)
• poams(root)、poam_items

A4 特殊邏輯（核心，full TDD）

Task A4-Matrix: AO 全量判定矩陣（Q3）

• test：給 ar_result + 一組 in-scope AO（從 reviewed-controls 展開）→ init_finding_matrix(ar_result_id, ao_list) 為每個 AO 建一筆 finding（state=pending、target_type=objective-id、target_id=AO）→ assert 筆數 = AO 數、皆 pending。
• upsert_finding(ar_result_id, ao, state) 改 met/not_met/pending；list_findings 回全量含統計（met/not_met/pending 計數，報表分母用）。

Task A4-Risk: 風險手動組（Q2）

• add_risk(ar_result_id, …) → link_findings(risk_id, finding_ids[]) 寫 assessment_finding_risks（多對多）；test：一 risk 關聯 3 finding、一 finding 屬 2 risk 都成立；風險等級存 risk.characterizations/props。

Task A4-Remed: 整改三層（PM2 / B 方案）

• upsert_remediation(risk_id, …)(response) → add_milestone(remediation_id, assignee_user_id, …)(task)；test：risk→remediation→milestone 三層 round-trip；per-milestone assignee 不同；匯出結構 = risk.remediations[].tasks[]。

Task A4-Poam: POA&M 從 findings 生成（不複製）

• generate_from_findings(ar_result_id)：對 not_met findings 建 poam_items（related_findings 用 FK/uuid 指回，不複製內容）；test 驗證 poam_item 數 = not_met finding 數、指回正確。

Task A4-App: AssessmentResultService / AssessmentRiskService / PoamService（含 list_remediations / list_milestones）

負責 agent：A5。Files: infra/adapter/{cmmc,iso,nist}/*、ports/oscal_parser_factory.py、app/service/oscal_io_service.py、tests/adapter/*

• port factory：get_oscal_parser_adapter(code) → CMMC/ISO/NIST adapter（pre-flight 讀舊套件 adapter 當 pattern）。
• CMMC adapter（本期重點，D5）：pdf_parser / excel_parser / convert_to_oscal_catalog_entity → 產出新 catalog/group/control/part(AO) 結構（AO 落 catalog_control_parts、method 存 props）。test：用一份 CMMC 樣本 PDF/Excel parse → assert control 數 + AO 數 + method 正確。
• ISO / NIST adapter：搬移舊邏輯對齊新結構（CMMC 綠後再做；ISO 深度 follow-up）。
• OscalIoService.export_oscal：entity tree → OSCAL JSON（先 JSON，XML/YAML 後續）；test：catalog/ssp/ar export round-trip 過 OSCAL schema 驗證（用 skill references/json-schemas/）。
• import_ssp_docx/excel：落地新 schema（搬舊邏輯，對齊 v2 表）。

負責 agent：A6（建議由 A2 agent 接手，重用 SSP deep-copy）。依賴：A1（catalog/profile entity+mapper）+ A2（SSP 12 子表 entity+mapper + deep_clone_ssp）已落地。Files: Create app/service/snapshot/oscal_snapshot_service.py、tests/snapshot/*

這是 design §1 四個「複製並固定」邊界中的②③，是架構核心。clone=可編副本、snapshot=凍結副本，但兩者都實際複製內容（獨立 rows、不存 id join 回母表）。

Task A6-Clone: clone_resource_library(framework_version_uid)（邊界②，clone）

• Step 1（test）: 建一份資源庫三件組（catalog 副本 + profile + SSP 範本）→ clone_resource_library(fw_version_uid) → assert 回傳新的 (catalog_id, profile_id, ssp_id) 三個新 row，內容與來源相等但 id 不同；改來源 catalog 一個 control title → 副本不變（驗證脫鉤）。
• Step 2: FAIL。
• Step 3: 實作：深複製 catalog 樹（groups/controls/parts/params，保留 self-FK 結構、重映射 parent_id）+ profile(+imports) + SSP 範本（呼叫 A2 的 deep_clone_ssp）；framework_version_id 沿用作 provenance（design §2.2 已定）。
• Step 4: PASS（含 edge：空 SSP 範本、catalog 多層巢狀）。
• Step 5: commit。

Task A6-Snapshot: snapshot_ssp(ssp_id) → frozen_ssp_id（邊界③，snapshot）

• Step 1（test）: 建 living SSP（含 12 子表資料）→ snapshot_ssp(ssp_id) → assert 回傳新 frozen ssp_id（全子樹複製、新 uuid）；frozen 的 status 標 frozen/locked（凍結語意）；改 living SSP → frozen 不變。
• Step 2: FAIL。
• Step 3: 實作：呼叫 deep_clone_ssp 複製全 12 子表 + 標記凍結（status）。回傳 frozen ssp_id（供主專案 Wave 2 寫入 project_audit_rounds.ssp_id）。
• Step 4: PASS。
• Step 5: commit。

Task A6-App: 對外 OscalSnapshotService（clone_resource_library / snapshot_ssp / deep_clone_ssp 統一出口）+ app service test（logger patch fixture）。

完成後回頭補 A5 的 export round-trip 測試 fixture（用 A6 產出的 cloned/frozen 物件當輸入）。

• Step 1: 全 model import smoke：python -c "import jedi_oscal_v2.app.service ..."（依依賴順序，避免誤判循環 import，記憶 feedback_audit_test_import_order_circular）。
• Step 2: DI wire 驗證（若套件自帶 container）。
• Step 3: 跑全 pytest tests/ 綠。
• Step 4: baseline 零回歸確認（本套件全新，無 baseline；記錄測試數）。
• Step 5: 收尾 commit；不推 Nexus（feature 完成 + user 明示才發版，記憶 feedback_jedi_package_publish_flow）。

• 顯式 git add 檔名、禁用 -am（記憶 feedback_subagent_explicit_git_add）；套件 repo 與主專案分開 commit；push 等 user。
• 不切 branch；pyproject path-dep 改動 dev-only 不 commit。
• 改 service 後若主專案要驗，提醒 user 重啟 BE（套件源碼改 BE 重啟即生效）。
• 跨 schema FK 字串必帶 schema 前綴。