證據分類分析結果報表(FR-031)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: 把既有 Python 報表演算法(validate.py / gen_ao_adjudication.py)移植進 BE,每次分類 run 落 DB,並提供 2 個系統內獨立唯讀報表頁(驗證報表 + 各 AO 誰對誰錯)。

Architecture: 新增 compliance.evidence_classification_runs 主表(JSONB 鏡像三個 Drive 產物)+ 完整 DDD 持久層;既有 _finalize_drive_output/put_state/archive_run 旁路補寫 DB(不改既有 Drive 行為與畫面);報表為純函式 builder 吃 DB row(report_original + state)× catalog × CANON → 結構化 JSON → FE PrimeVue 渲染。

Tech Stack: Python 3.11 / Flask-RESTful (flask_apispec MethodResource) / SQLAlchemy + jedi_common BaseModel/BaseRepositoryImpl / dependency-injector / PostgreSQL JSONB / pytest|FE: Vue 3 + PrimeVue + Pinia。

⚠️ 實作後追加(見 design.md §13 Reconciliation):本計畫 Phase 1-7 完成後,依實機對照外部 script + user 回饋追加了:正解匯入(Option B)(新表 evidence_classification_ground_truth + import API + 報告①②都吃匯入正解)、報告② 同源正解修正報告① 結構強化(ao_by_control / 未分類彙整 / 移除 attention)、FE 改成 1:1 移植 script HTML 版面(非原 D7 的 generic PrimeVue 重畫)。細節以 design §13 為準。

規範錨點(每個 commit 都遵守):

  • DDD 嚴格分層(route 不碰 DB、app service @transaction、repo 繼承 BaseRepositoryImpl),見 CLAUDE.md
  • error code 走 EvidenceClassificationErrorCode(前綴 EC_)。
  • commit 顯式 git add <檔名>,禁用 -am(memory feedback_subagent_explicit_git_add)。
  • app service test 必加 logger patch autouse fixture(memory feedback_test_logger_patch_db_handler)。
  • 改 BE service 後提醒 user 重啟 BE(無 hot reload);服務一律 user 自己起,計畫不附啟動命令。
  • 不切 branch;push 等 user 明示。

§1

參考來源(implementer 必讀)

用途 路徑
SD 設計書(本計畫母文件) docs/features/FR-031-2605-evidence-classify-reports/design.md
白話需求 docs/features/FR-031-2605-evidence-classify-reports/raw-requirement.md
報告① 演算法原始碼 ~/Desktop/AirAsia 真實證據/CMMC佐證/_驗證/validate.py
報告② 演算法 + CANON 原始碼 ~/Desktop/AirAsia 真實證據/CMMC佐證/_驗證/gen_ao_adjudication.py
JSON 欄位權威 schema docs/api/evidence-classification/state-json-schema.md
既有 service app/evidence_classification/service/evidence_classification_service.py
DDD 範本 infra/subtask_status_history/infra/project_summary_report/(compliance schema + JSONB 案例)
catalog docs/reference/CMMC-Level 1-Evidences/cmmc_l1_aos.jsondomains[]→controls[]→assessment_objectives

§2

Phase 0 — Pre-flight 驗證(不寫 code,先確認假設)

plan 寫好到實作常隔多日,先驗證 method/欄位假設(memory feedback_plan_vs_reality_verify_first)。

    • _finalize_drive_output(self, job_uid, tenant_id, evidence_folder_id, state, catalog_data, copy_files=True)
    • put_state(self, run_folder_id, new_state, current_user_id) -> dict
    • archive_run(self, run_folder_id, current_user_id) -> dict Run: grep -n "def _finalize_drive_output\|def put_state\|def archive_run" app/evidence_classification/service/evidence_classification_service.py

§3

Phase 1 — DB 持久層(DDD,模組目前零 infra)

Task 1.1:SQL migration

Files: Create scripts/sql/2026-05-31_fr031_evidence_classification_runs.sql

-- Date: 2026-05-31
-- FR-031 證據分類分析結果報表 — run 結果落 DB
CREATE TABLE compliance.evidence_classification_runs (
    id                      BIGSERIAL    PRIMARY KEY,                       -- (2026-05-31)
    run_folder_id           VARCHAR(128) NOT NULL,                          -- Drive run folder id(自然鍵)(2026-05-31)
    run_folder_name         VARCHAR(255),
    tenant_id               INTEGER      NOT NULL,
    project_id              INTEGER      NOT NULL,
    project_uid             VARCHAR(36),                                   -- FE 列表/URL (2026-05-31)
    ap_uid                  VARCHAR(36),                                   -- AP uid soft ref (2026-05-31)
    org_unit_id             INTEGER,
    framework_id            VARCHAR(64)  NOT NULL DEFAULT 'cmmc-l1',
    model                   VARCHAR(64),
    confidence_threshold    NUMERIC(4,2),
    status                  VARCHAR(32)  NOT NULL DEFAULT 'completed',
    archive_files           BOOLEAN      NOT NULL DEFAULT TRUE,
    input_file_count        INTEGER,
    classified_count        INTEGER,
    estimated_cost_usd      NUMERIC(10,4),
    triggered_at            TIMESTAMPTZ,
    completed_at            TIMESTAMPTZ,
    last_edited_at          TIMESTAMPTZ,                                    -- 報告① 降級判定 (2026-05-31)
    archived_at             TIMESTAMPTZ,
    triggered_by_user_id    INTEGER,
    last_edited_by_user_id  INTEGER,
    archived_by_user_id     INTEGER,
    report_original         JSONB,                                         -- AI 原始, immutable (2026-05-31)
    state                   JSONB,                                         -- 當前狀態, 同步 Drive (2026-05-31)
    container_log           TEXT,
    created_at              TIMESTAMPTZ  NOT NULL DEFAULT now(),
    updated_at              TIMESTAMPTZ  NOT NULL DEFAULT now(),
    created_user            VARCHAR(50),
    updated_user            VARCHAR(50)
);
CREATE UNIQUE INDEX uq_ecr_run_folder_id ON compliance.evidence_classification_runs (run_folder_id);  -- (2026-05-31)
CREATE INDEX ix_ecr_project_ap ON compliance.evidence_classification_runs (project_uid, ap_uid);      -- (2026-05-31)
CREATE INDEX ix_ecr_tenant ON compliance.evidence_classification_runs (tenant_id);                    -- (2026-05-31)
GRANT SELECT, INSERT, UPDATE, DELETE ON compliance.evidence_classification_runs TO cm_app;
GRANT USAGE, SELECT ON SEQUENCE compliance.evidence_classification_runs_id_seq TO cm_app;
git add scripts/sql/2026-05-31_fr031_evidence_classification_runs.sql
git commit -m "feat(evidence-classify): FR-031 run 落 DB migration(evidence_classification_runs)"

Task 1.2:ORM model

Files: Create infra/evidence_classification/model/classification_run_model.py, infra/evidence_classification/model/__init__.py

from datetime import datetime
from typing import Optional
from sqlalchemy import String, Integer, Numeric, Boolean, Text, DateTime
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column
from jedi_common.session.database.model.base_model import BaseModel


class ClassificationRunModel(BaseModel):
    """證據分類 run 結果(鏡像 Drive run folder)"""
    __tablename__ = "evidence_classification_runs"
    __table_args__ = {"schema": "compliance", "comment": "證據分類 run 結果"}

    run_folder_id: Mapped[str] = mapped_column(String(128), nullable=False, comment="Drive run folder id(自然鍵)")
    run_folder_name: Mapped[Optional[str]] = mapped_column(String(255))
    tenant_id: Mapped[int] = mapped_column(Integer, nullable=False)
    project_id: Mapped[int] = mapped_column(Integer, nullable=False)
    project_uid: Mapped[Optional[str]] = mapped_column(String(36))
    ap_uid: Mapped[Optional[str]] = mapped_column(String(36))
    org_unit_id: Mapped[Optional[int]] = mapped_column(Integer)
    framework_id: Mapped[str] = mapped_column(String(64), nullable=False, server_default="cmmc-l1")
    model: Mapped[Optional[str]] = mapped_column(String(64))
    confidence_threshold: Mapped[Optional[float]] = mapped_column(Numeric(4, 2))
    status: Mapped[str] = mapped_column(String(32), nullable=False, server_default="completed")
    archive_files: Mapped[bool] = mapped_column(Boolean, nullable=False, server_default="true")
    input_file_count: Mapped[Optional[int]] = mapped_column(Integer)
    classified_count: Mapped[Optional[int]] = mapped_column(Integer)
    estimated_cost_usd: Mapped[Optional[float]] = mapped_column(Numeric(10, 4))
    triggered_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True))
    completed_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True))
    last_edited_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True))
    archived_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True))
    triggered_by_user_id: Mapped[Optional[int]] = mapped_column(Integer)
    last_edited_by_user_id: Mapped[Optional[int]] = mapped_column(Integer)
    archived_by_user_id: Mapped[Optional[int]] = mapped_column(Integer)
    report_original: Mapped[Optional[dict]] = mapped_column(JSONB)
    state: Mapped[Optional[dict]] = mapped_column(JSONB)
    container_log: Mapped[Optional[str]] = mapped_column(Text)

    def __repr__(self) -> str:
        return f"<ClassificationRunModel {self.run_folder_id}>"
git add infra/evidence_classification/model/classification_run_model.py infra/evidence_classification/model/__init__.py
git commit -m "feat(evidence-classify): FR-031 ClassificationRunModel ORM"

Task 1.3:Entity + QueryEntity

Files: Create domain/evidence_classification/entity/classification_run_entity.py, classification_run_query_entity.py, domain/evidence_classification/__init__.py + entity/__init__.py

Task 1.4:Mapper

Files: Create infra/evidence_classification/mapper/classification_run_mapper.py + mapper/__init__.py

Task 1.5:Repo interface + impl

Files: Create domain/evidence_classification/repository/classification_run_repository.py, infra/evidence_classification/repository/classification_run_repository_impl.py (+ __init__)

from jedi_common.session.database.repository.base_repository import IBaseRepo
from typing import TypeVar
T = TypeVar("T"); Q = TypeVar("Q")
class IClassificationRunRepo(IBaseRepo[T, Q]):
    pass

Task 1.6:Domain service

Files: Create domain/evidence_classification/service/classification_run_domain_service.py + service/__init__.py

    • get_by_run_folder_id(run_folder_id) -> Entity | Noneget_one_by_fields
    • upsert_run(entity) -> Entity(先查 run_folder_id,有則帶 id 走 update、無則 add
    • update_state(run_folder_id, state: dict, last_edited_at, last_edited_by_user_id, archive_files=None, archived_at=None, archived_by_user_id=None)(查到 entity → 覆寫對應欄位 → update
    • list_by_project_ap(project_id, ap_id) -> list[Entity](報表/列表用,可選)

Task 1.7:DI wiring

Files: Modify di_containers/evidence_classification/evidence_classification_containers.py


§4

Phase 2 — DB 寫入整合(既有 service 旁路;決策 D2)

Task 2.0:trigger 把 ap_uid 存進 JobRegistry(Phase 0 驗證補)

Files: Modify app/evidence_classification/service/evidence_classification_service.pyapp/evidence_classification/service/job_registry.py

  • 取值對照(_finalize upsert 用):project_id/project_uid/model/confidence_threshold/framework_id/archive_files/triggered_by(started_by_user_id) ← JobRegistry.get(job_uid)org_unit_idget_user_context().org_unit_idap_uid ← registry(本 task 補存後);report_originalself._runner.read_report(job_uid)container_logself._runner.read_container_log(job_uid)state ← 參數。

Task 2.1:service 注入 domain service + upsert at finalize

Files: Modify app/evidence_classification/service/evidence_classification_service.py

    • report_original/container_log 依 Phase 0.2 發現的來源。
    • DB 失敗不可回滾 Drive:包 try/except Exception as e: logger.error(...)(不 re-raise)。

Task 2.2:put_state / archive_run 同步 DB

Files: Modify same service

Task 2.3:歷史 run lazy 回填(開放問題1,預設做)

Files: Modify same service(get_state 或新 helper _ensure_run_persisted


§5

Phase 3 — 報表共用工具(純函式,TDD 主場)

Task 3.1:report_common.py

Files: Create app/evidence_classification/service/report/__init__.py, report_common.py; Test test/test_report_common.py

from app.evidence_classification.service.report.report_common import doc_key, normalize_aoid, ai_placements, user_placements

def test_doc_key_same_doc_diff_ext():
    assert doc_key("X_20260518.pdf") == doc_key("X.docx")        # 去日期/副檔名後同鍵

def test_normalize_aoid_single_ao_autofill():
    assert normalize_aoid("PE.L1-3.10.4", {"PE.L1-3.10.4[a]"}) == "PE.L1-3.10.4[a]"

def test_ai_placements_threshold_filter():
    ro = {"metadata": {"confidence_threshold": 0.8},
          "files": [{"file_name": "a.png", "matches": [{"ao_id": "AC.L1-3.1.1[a]", "confidence": 0.9}, {"ao_id": "IA.L1-3.5.2[a]", "confidence": 0.4}]}]}
    ph, unplaced, diag = ai_placements(ro, 0.8, {"AC.L1-3.1.1[a]", "IA.L1-3.5.2[a]"})
    assert ph == {"AC.L1-3.1.1[a]": ["a.png"]}
    assert unplaced == []

def test_user_placements_from_state():
    st = {"files": [{"file_name": "a.png", "is_deleted": False,
                     "placements": [{"ao_id": "AC.L1-3.1.1[a]"}]}]}
    assert user_placements(st) == {"AC.L1-3.1.1[a]": ["a.png"]}

§6

Phase 4 — 報告①:驗證報表 builder

Task 4.1:validation_report_builder.py

Files: Create app/evidence_classification/service/report/validation_report_builder.py; Test test/test_validation_report_builder.py

def test_full_mode_recall_precision():
    # state.metadata.last_edited_at 有值 → full;AI 對 1/2 → recall/precision 可算
    report = build_validation_report(report_original, state_reviewed, catalog)
    assert report["mode"] == "full"
    assert report["summary"]["control_level"]["hit"] == 1

def test_degraded_when_unreviewed():
    # last_edited_at/archived_at 皆 None → degraded,無 summary
    report = build_validation_report(report_original, state_raw, catalog)
    assert report["mode"] == "degraded"
    assert "summary" not in report or report["summary"] is None
    assert report["unclassified"] is not None         # 降級仍給未分類診斷 + 放置清單

def test_missing_four_levels():
    # 構造涵蓋 版本格式/AO選錯/分到其他控制項/AI完全未分類 四級
    ...
def test_unclassified_three_categories_and_recover():
    # catA/catB/catC + 降門檻可否救回
    ...
    • GT = user_placements(state);AI = ai_placements(report_original);catalog 給 valid_aoids + control_name + ao_text。
    • 降級判定:state["metadata"].get("last_edited_at")archived_at 皆 falsy → mode="degraded",跳過 recall/precision 與四級彙整,仍輸出 unclassified + AI 放置 + cost
    • 演算法逐段對應 validate.py main()+write_reports()(控制項/AO 層級、四級 label、未分類三類、docker log 解析)。輸出 §5.2 JSON shape。
    • 不產 HTML/MD/CSV(只回 dict)。

§7

Phase 5 — 報告②:各 AO 誰對誰錯 builder + CANON 資源

Task 5.1:CANON 資源檔

Files: Create app/evidence_classification/resources/cmmc_l1_canon.json + resources/__init__.py

{ "framework_id": "cmmc-l1",
  "rules": { "Login_Fido_Setting.png": {"controls": ["IA.L1-3.5.2"], "rationale": "FIDO 登入=身分鑑別"},
             "NTP_Setting.png": {"controls": [], "rationale": "L2 範圍外"} } }

(逐筆照搬,set(...) → array、空 set → []、理由原文)。

Task 5.2:adjudication_report_builder.py

Files: Create app/evidence_classification/service/report/adjudication_report_builder.py; Test test/test_adjudication_report_builder.py

def test_user_wrong_detected():
    # 權限申請表 User 放 IA.3.5.2(鑑別),CANON 說應在 AC.3.1.1 → user_wrong 抓到
    r = build_adjudication_report(report_original, state, catalog, canon)
    assert any(w["file"] == "perm_form.pdf" for w in r["user_wrong"])

def test_ai_over_placement():
    # AI 放 CANON 外 → ai_wrong
def test_out_of_scope_empty_canon():
    # NTP(CANON 空集合)兩邊放 → out_scope
def test_undecided_when_not_in_canon():
    # 檔不在 CANON → undecided,不判對錯
def test_available_false_when_no_canon():
    assert build_adjudication_report(ro, st, catalog, canon=None)["available"] is False
def test_note_user_equals_ai_when_unreviewed():
    # last_edited_at None → note_user_equals_ai True

Task 5.3:CANON loader

Files: Modify report_common.py(或新 canon_loader.py


§8

Phase 6 — 報表 API + app service method

Task 6.1:app service 兩個 report method

Files: Modify evidence_classification_service.py; Test test/test_evidence_classification_report_endpoints.py(logger fixture)

@transaction
def get_validation_report(self, run_folder_id, current_user_id) -> dict:
    run = self._ensure_run_persisted_for_read(run_folder_id, current_user_id)
    catalog = load_catalog(run.framework_id)
    return build_validation_report(run.report_original, run.state, catalog)

@transaction
def get_adjudication_report(self, run_folder_id, current_user_id) -> dict:
    run = self._ensure_run_persisted_for_read(run_folder_id, current_user_id)
    catalog = load_catalog(run.framework_id)
    canon = load_canon(run.framework_id)
    return build_adjudication_report(run.report_original, run.state, catalog, canon)

Task 6.2:error code + route

Files: Modify common/code/evidence_classification_error_code.py, api/evidence_classification/routes/evidence_classification_route.py

    • GET /api/1.0/classification-run/<run_folder_id>/report/validationservice.get_validation_report(run_folder_id, user.id)
    • GET /api/1.0/classification-run/<run_folder_id>/report/adjudicationservice.get_adjudication_report(...)create_module() 註冊兩 route(照既有 add_url_rule 樣式)。

§9

Phase 7 — FE:2 報表頁 + 2 入口(compliance-manager-fe)

動工前先讀 FE CLAUDE.md(memory feedback_cross_repo_read_claude_md_first)。FE 同 branch feature/ai-analysis-result-landing只新增頁面與導覽連結,不動既有審閱/歸檔/比較邏輯。

Task 7.1:Service method

Files: Modify src/service/EvidenceClassificationService.js

Task 7.2:i18n

Files: Modify src/config/locales/i18n/{zh-tw,en}/evidence-classification.json

Task 7.3:驗證報表頁

Files: Create src/views/evidence-classification/EvidenceClassificationValidationReport.vue + route

Task 7.4:各 AO 誰對誰錯頁

Files: Create src/views/evidence-classification/EvidenceClassificationAdjudicationReport.vue + route

Task 7.5:入口連結(僅加導覽)

Files: Modify src/views/evidence-classification/EvidenceClassificationReview.vue(header 加兩報表 Button)、src/components/grc/project/AIEvidenceClassificationDialog.vue(job history row 既有「審閱」Button 旁加兩報表連結,status==='completed' && run_folder_id

Task 7.6:FE 手測


§10

收尾(等 user 明確下令才做)


§11

風險 / 注意

風險 緩解
_finalize_drive_outputreport_original/project_id 取得點與假設不符 Phase 0.2/0.3 先驗證,不符先回報
DB 寫入失敗連帶弄壞既有 Drive 流程 try/except 不回滾、只 log(Task 2.1 Step 3)
報告② User=AI(未審閱)誤讀 note_user_equals_ai + FE 提示(Task 5.2 / 7.4)
catalog 中文控制項名缺 移植 CTRL_NAME 16 條(Phase 3)
改 BE service 後 user 報「沒生效」 每次改 service 提醒重啟 BE