# Spec 1 流程管理 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. > > **Spec 文件**：`docs/features/FR-026-2605-project-flow-engine-integrate/01-flow-template-management/design.md` > **整體脈絡**：`docs/features/FR-026-2605-project-flow-engine-integrate/README.md` > **建立日期**：2026-05-12 > **Branch**：`feature/project-flow-engine-integrate`（已在此 branch，無需切） **Goal:** 讓使用者管理「專案流程範本」 — DB schema、CRUD API、RBAC、BPMN 編輯器 UI（FE 另 session），交付一個 admin 可建範本但尚無地方套用的可獨立驗證版本。 **Architecture:** 沿用既有 `flow_engine` 模組目錄，新增兩張表 `compliance.stage_objects`（系統層級 seed）+ `compliance.flow_templates`（tenant 隔離 + builtin tenant_id IS NULL）。後端 DDD 4 層（api/app/domain/infra/di_containers）+ 既有 RBAC 三表（capabilities / ui_routes / route_capabilities）。BPMN XML 純文字儲存，解析借用 `jedi_flow_engine.common.utils.bpmn_uilts.BpmnUtils`。 **Tech Stack:** Flask-RESTful / SQLAlchemy / dependency-injector / marshmallow / Poetry / pytest；DB PostgreSQL（`192.168.50.188:25432` dev，schema `compliance`）；RBAC PostgreSQL RLS；jedi-flow-engine（BPMN parser 重用）。 --- ## 鎖定的設計決策（§八 開放問題收斂） | 主題 | v1 決定 | 影響範圍 | |---|---|---| | Gateway 編輯 | **完整支援**（FE 可拖 ExclusiveGateway / 改條件） | FE 工作 +1-2 週；BE schema 不變，BPMN XML 純文字儲存通吃 | | 範本分類 / tag | **不做** | schema 不留 `category` 欄位；FE 列表頁只有「內建 / 自訂」filter + name 搜尋 | | Cross-tenant 共用 | **不支援，強制 tenant 隔離** | `tenant_id` NULLABLE：NULL=builtin，NOT NULL=tenant 自訂；RLS policy 允許 NULL 公開讀 | | 草稿 vs 發佈 | **不分** | 只有 `is_active`；Spec 3 wizard 列所有 `is_active=true` 範本 | --- ## 前置 verify（執行前必跑） > ⚠️ Plan 寫好到開工經常 days/weeks，期間 jedi-flow-engine method 改名 / error code 序號被佔 / DB schema 漂移皆可能。每個 phase 開始前先驗證以下假設。 - [ ] **error code 序號未被佔用**：grep `common/code/grc_error_code.py` 確認 `GRC_404027` / `GRC_404028` / `GRC_409030` / `GRC_403040` / `GRC_403041` / `GRC_400040` 都還沒人用。被佔走改取下個可用序號。 - [ ] **`compliance` schema 存在**：`psql -U cmmgr -d guidant_ai_stg -c "\dn compliance"`，列得到代表 OK。 - [ ] **`public.capabilities` / `public.ui_routes` / `public.route_capabilities` / `public.roles` 都存在**：`\dt public.capabilities` 等。 - [ ] **`jedi_flow_engine.common.utils.bpmn_uilts.BpmnUtils` 仍存在**：`grep -r "class BpmnUtils" ~/Projects/Jedicogy/module/jedi-python-package/jedi-flow-engine/`（CLAUDE.md memory 指向 `~/Projects/Jedicogy/module/jedi-python-package/`）。被改名 → 改用既有 method 名。 - [ ] **既有 `flow_engine` Blueprint url_prefix 是 `/api/1.0`**：`Read api/flow_engine/__init__.py:25`。Spec 2 / 3 沿用同一個 prefix。 - [ ] **`get_user_context()` 仍回 UserContextDTO（含 `id` / `login_name` / `tenant_id`）**：grep `jedi_common.session.auth.auth_context.get_user_context`。 --- ## File Structure（最終樣貌） ### BE 新檔（共 23 個檔） ``` api/flow_engine/ ├── routes/ │ ├── flow_template_route.py # 新：6 個 Resource │ └── stage_object_route.py # 新：1 個 Resource └── serializers/ ├── flow_template.py # 新：Create/Update/Duplicate/Response/List schemas └── stage_object.py # 新：Response/List schemas app/flow_engine/ ├── service/ │ ├── flow_template_app_service.py # 新：list/get/create/update/soft_delete/duplicate │ └── stage_object_app_service.py # 新：list/get └── dto/ ├── flow_template_dto.py # 新 └── stage_object_dto.py # 新 domain/flow_engine/ ├── entity/ │ ├── flow_template_entity.py # 新 │ ├── flow_template_query_entity.py # 新 │ ├── stage_object_entity.py # 新 │ └── stage_object_query_entity.py # 新 ├── repository/ │ ├── flow_template_repo.py # 新：IFlowTemplateRepo │ └── stage_object_repo.py # 新：IStageObjectRepo └── service/ ├── flow_template_domain_service.py # 新 └── stage_object_domain_service.py # 新 infra/flow_engine/ ├── models/ │ ├── flow_template.py # 新：SQLAlchemy ORM │ └── stage_object.py # 新 ├── mapper/ │ ├── flow_template_mapper.py # 新 │ └── stage_object_mapper.py # 新 └── repository/ ├── flow_template_repo_impl.py # 新 └── stage_object_repo_impl.py # 新 di_containers/flow_engine/ └── flow_template_containers.py # 新：FlowTemplateContainer common/code/ └── grc_error_code.py # 修改：加 6 個 error code scripts/sql/ ├── 2026-MM-DD-flow-template-management-schema.sql # 新：DDL + seeds（單檔） └── 2026-MM-DD-flow-template-permissions.sql # 新：RBAC migration di_containers/containers.py # 修改：wire flow_template_container api/flow_engine/__init__.py # 修改：register 7 個新 Resource docs/changelog/ └── 2026-MM-DD-feat-flow-template-management.md # 新 docs/api/flow-engine/ ├── api-spec.md # 修改：加範本 / 階段物件 SA └── design.md # 修改：加範本 / 階段物件 SD docs/features/FR-026-2605-project-flow-engine-integrate/01-flow-template-management/ └── frontend-spec.md # 新：FE handoff（給 FE session 用） ``` ### BE 測試檔（共 5 個檔） ``` test/ ├── test_flow_template_serializer.py # 新：marshmallow schema 單元測試 ├── test_flow_template_app_service.py # 新：mock domain service，覆蓋 CRUD + builtin 守門 ├── test_flow_template_domain_service.py # 新：mock repo，覆蓋 query + duplicate 邏輯 ├── test_api_flow_template_route.py # 新：直接 invoke Resource，mock service └── test_api_stage_object_route.py # 新：list / detail ``` > **E2E / FE 測試不在此 plan**：依 CLAUDE.md「測試專案路徑」段，E2E feature / step 全部在 `~/Projects/Billows/Audit-Manager/compliance-manager-test/` 進行，由另一 session 處理。 --- ## DB Schema 設計 ### `compliance.stage_objects` | 欄位 | 型別 | 規則 | |---|---|---| | `id` | SERIAL PK | — | | `uid` | UUID NOT NULL UNIQUE | `gen_random_uuid()` default | | `code` | VARCHAR(50) NOT NULL UNIQUE | `planning` / `task_execution` / ... | | `name_i18n` | JSONB NOT NULL | `{"zh_Hant_TW": "規劃", "en": "Planning"}` | | `kind` | VARCHAR(20) NOT NULL | CHECK IN (`routed`, `stateful`) | | `route_pattern` | VARCHAR(255) NULL | `kind='routed'` 才有值；含 `:projectUID` / `:apUID` placeholder | | `complete_button_label_i18n` | JSONB NOT NULL | `{"zh_Hant_TW": "啟動專案", ...}` | | `complete_handler_key` | VARCHAR(100) NOT NULL | 對應 §3.2 既有 API 識別字串（Spec 2 實作 handler） | | `precondition_key` | VARCHAR(100) NULL | Spec 2 實作前置條件 function | | `default_main_roles` | JSONB NOT NULL DEFAULT `'[]'` | 角色 code 陣列，如 `["manager"]` | | `is_builtin` | BOOLEAN NOT NULL DEFAULT TRUE | v1 全部 true | | `is_active` | BOOLEAN NOT NULL DEFAULT TRUE | 軟刪除 | | `sort` | INT NOT NULL DEFAULT 0 | 顯示排序 | | `created_user` / `updated_user` | VARCHAR(100) | 審計 | | `created_at` / `updated_at` | TIMESTAMPTZ DEFAULT NOW() | 審計 | **Indexes**：`code`（已 UNIQUE）、`is_active`。 **RLS**：**不啟用**（系統層級 seed，所有 tenant 共讀；v1 不開 CRUD）。 **GRANT**：`GRANT SELECT, INSERT, UPDATE, DELETE ON compliance.stage_objects TO cm_app;` + sequence GRANT。 ### `compliance.flow_templates` | 欄位 | 型別 | 規則 | |---|---|---| | `id` | SERIAL PK | — | | `uid` | UUID NOT NULL UNIQUE | `gen_random_uuid()` default | | `tenant_id` | INT NULL | NULL=builtin，NOT NULL=tenant 自訂 | | `name` | VARCHAR(200) NOT NULL | 顯示名稱 | | `description` | TEXT NULL | — | | `is_builtin` | BOOLEAN NOT NULL DEFAULT FALSE | — | | `is_active` | BOOLEAN NOT NULL DEFAULT TRUE | 軟刪除 | | `bpmn_xml` | TEXT NOT NULL | BPMN 2.0 XML | | `created_user` / `updated_user` | VARCHAR(100) | 審計 | | `created_at` / `updated_at` | TIMESTAMPTZ DEFAULT NOW() | 審計 | **Constraints**： - `chk_builtin_tenant_null` CHECK (`(is_builtin = TRUE AND tenant_id IS NULL) OR (is_builtin = FALSE AND tenant_id IS NOT NULL)`) - `uq_flow_template_name_per_tenant` UNIQUE (`tenant_id`, `name`) — partial index 排除 `is_active=FALSE`：`CREATE UNIQUE INDEX uq_flow_template_name_active ON compliance.flow_templates (tenant_id, name) WHERE is_active = TRUE;` **Indexes**：`tenant_id, is_active`、`is_builtin, is_active`。 **RLS**： ```sql -- SELECT：builtin (tenant_id IS NULL) 公開讀；tenant 自訂走 helper CREATE POLICY flow_templates_select ON compliance.flow_templates FOR SELECT USING ( COALESCE(current_setting('app.is_super_admin', true), 'f') = 't' OR tenant_id IS NULL OR app_tenant_allowed_for_session(tenant_id) ); -- INSERT / UPDATE / DELETE：必須是 super_admin 或 tenant_id 屬於 session CREATE POLICY flow_templates_modify ON compliance.flow_templates FOR ALL USING ( COALESCE(current_setting('app.is_super_admin', true), 'f') = 't' OR (tenant_id IS NOT NULL AND app_tenant_allowed_for_session(tenant_id)) ); ``` **GRANT**：`GRANT SELECT, INSERT, UPDATE, DELETE ON compliance.flow_templates TO cm_app;` + sequence GRANT。 --- ## Error Code 規劃 加到 `common/code/grc_error_code.py` 的 `GrcErrorCode` class（依命名規則 `序號`，序號參考既有最大值 +1）： ```python # ── Flow Template ────────────────────────────────────────────────── GRC_FLOW_TEMPLATE_NOT_FOUND = ("流程範本不存在", "GRC_404027") GRC_STAGE_OBJECT_NOT_FOUND = ("階段物件不存在", "GRC_404028") GRC_FLOW_TEMPLATE_BPMN_INVALID = ("BPMN XML 格式無效", "GRC_400040") GRC_FLOW_TEMPLATE_BUILTIN_READONLY = ("內建範本不可編輯或刪除", "GRC_403040") GRC_FLOW_TEMPLATE_NO_PERMISSION = ("無流程範本管理權限", "GRC_403041") GRC_FLOW_TEMPLATE_NAME_DUPLICATED = ("流程範本名稱已存在", "GRC_409030") ``` --- ## API 規格（§七 對外介面實作） | Method | Path | Capability | 說明 | |---|---|---|---| | GET | `/api/1.0/flow-engine/stage-objects` | `flow_template.read` | 列出全部 `is_active=true` 階段物件 | | GET | `/api/1.0/flow-engine/flow-templates` | `flow_template.read` | List：query params `name`、`is_builtin`、`page`、`size`；回傳分頁 envelope | | GET | `/api/1.0/flow-engine/flow-templates/` | `flow_template.read` | 詳情含 BPMN XML | | POST | `/api/1.0/flow-engine/flow-templates` | `flow_template.create` | Body：`{name, description, bpmn_xml}` → 自動帶 `tenant_id=current_user.tenant_id`、`is_builtin=false` | | PUT | `/api/1.0/flow-engine/flow-templates/` | `flow_template.update` | 不可改 builtin（→ 403）、不可改 `tenant_id` / `is_builtin` | | DELETE | `/api/1.0/flow-engine/flow-templates/` | `flow_template.delete` | 軟刪除（`is_active=false`）；不可刪 builtin | | POST | `/api/1.0/flow-engine/flow-templates//duplicate` | `flow_template.create` | Body：`{name}`（必填，新名稱）→ 新建 `is_builtin=false`、`tenant_id=current_user.tenant_id`、copy `bpmn_xml` | **權限檢查位置**：App service 層用 `get_user_context()` 取 user，比對 `user.capabilities`（已由 jedi-auth middleware 載入）— 若 user 無對應 capability，raise `ForbiddenError(GrcErrorCode.GRC_FLOW_TEMPLATE_NO_PERMISSION)`。 **Response envelope**（依 CLAUDE.md API Patterns）： - 成功：`return_response(True, {...})` → `{"code": 1, "data": {...}}` - 失敗：jedi exception → `{"code": 0, "msg": "...", "data": {...}}` - 分頁：用既有 `EnvelopeSchema` + `RequestMetaSchema` pattern（見 `docs/claude/api-patterns.md`） **BPMN 驗證**：Create / Update / Duplicate 時呼叫 `BpmnUtils` 解析 XML，至少要含 1 個 StartEvent / 1 個 EndEvent / `>=1` 個 UserTask；解析失敗 → `BadRequestError(GRC_FLOW_TEMPLATE_BPMN_INVALID)`。 --- ## 高階任務拆解（Phases） > 每個 phase 完成後執行 `pytest` 確認 GREEN，並產 1 個 commit。整體目標 8-10 個 commit。 - **Phase A**：DB migration + seed SQL（不動程式碼） - **Phase B**：Error code + 域層（entity + repo interface + domain service） - **Phase C**：基礎層（ORM model + mapper + repo impl） - **Phase D**：應用層（app service + DTO + 權限檢查） - **Phase E**：API 層（route + serializer + BPMN 驗證） - **Phase F**：DI wiring + 整合測試 + smoke run - **Phase G**：RBAC migration + 驗證 - **Phase H**：文件交付（changelog / API docs / FE handoff） --- ## Phase A：DB Migration + Seed > 目標：跑完 `psql -U cmmgr -d guidant_ai_stg -f scripts/sql/.sql` → 兩張表 + 4 stage_object seeds + 3 flow_template seeds + 全部 GRANT + RLS 正常運作。 ### Task A1：建立 stage_objects + flow_templates schema SQL **Files:** - Create: `scripts/sql/2026-05-12-flow-template-management-schema.sql` - [ ] **Step 1：寫 schema SQL** ```sql -- Date: 2026-05-12 -- Spec 1：流程管理 — DDL + seeds -- 執行身份：cmmgr（系統管理員） -- PGPASSWORD='jedi@123!' psql -h 192.168.50.188 -p 25432 -U cmmgr -d guidant_ai_stg \ -- -f scripts/sql/2026-05-12-flow-template-management-schema.sql BEGIN; SET LOCAL app.is_super_admin = 't'; -- 1. compliance.stage_objects (2026-05-12) CREATE TABLE IF NOT EXISTS compliance.stage_objects ( id SERIAL PRIMARY KEY, uid UUID NOT NULL UNIQUE DEFAULT gen_random_uuid(), code VARCHAR(50) NOT NULL UNIQUE, name_i18n JSONB NOT NULL, kind VARCHAR(20) NOT NULL, route_pattern VARCHAR(255), complete_button_label_i18n JSONB NOT NULL, complete_handler_key VARCHAR(100) NOT NULL, precondition_key VARCHAR(100), default_main_roles JSONB NOT NULL DEFAULT '[]', is_builtin BOOLEAN NOT NULL DEFAULT TRUE, is_active BOOLEAN NOT NULL DEFAULT TRUE, sort INT NOT NULL DEFAULT 0, created_user VARCHAR(100), updated_user VARCHAR(100), created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() ); ALTER TABLE compliance.stage_objects DROP CONSTRAINT IF EXISTS chk_stage_object_kind; ALTER TABLE compliance.stage_objects ADD CONSTRAINT chk_stage_object_kind CHECK (kind IN ('routed', 'stateful')); CREATE INDEX IF NOT EXISTS idx_stage_objects_active ON compliance.stage_objects(is_active); GRANT SELECT, INSERT, UPDATE, DELETE ON compliance.stage_objects TO cm_app; GRANT USAGE, SELECT ON SEQUENCE compliance.stage_objects_id_seq TO cm_app; -- 2. compliance.flow_templates (2026-05-12) CREATE TABLE IF NOT EXISTS compliance.flow_templates ( id SERIAL PRIMARY KEY, uid UUID NOT NULL UNIQUE DEFAULT gen_random_uuid(), tenant_id INT, name VARCHAR(200) NOT NULL, description TEXT, is_builtin BOOLEAN NOT NULL DEFAULT FALSE, is_active BOOLEAN NOT NULL DEFAULT TRUE, bpmn_xml TEXT NOT NULL, created_user VARCHAR(100), updated_user VARCHAR(100), created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() ); ALTER TABLE compliance.flow_templates DROP CONSTRAINT IF EXISTS chk_flow_template_builtin; ALTER TABLE compliance.flow_templates ADD CONSTRAINT chk_flow_template_builtin CHECK ( (is_builtin = TRUE AND tenant_id IS NULL) OR (is_builtin = FALSE AND tenant_id IS NOT NULL) ); CREATE UNIQUE INDEX IF NOT EXISTS uq_flow_template_name_active ON compliance.flow_templates (tenant_id, name) WHERE is_active = TRUE; CREATE INDEX IF NOT EXISTS idx_flow_templates_tenant_active ON compliance.flow_templates (tenant_id, is_active); CREATE INDEX IF NOT EXISTS idx_flow_templates_builtin_active ON compliance.flow_templates (is_builtin, is_active); GRANT SELECT, INSERT, UPDATE, DELETE ON compliance.flow_templates TO cm_app; GRANT USAGE, SELECT ON SEQUENCE compliance.flow_templates_id_seq TO cm_app; -- 3. RLS policies for flow_templates (2026-05-12) ALTER TABLE compliance.flow_templates ENABLE ROW LEVEL SECURITY; DROP POLICY IF EXISTS flow_templates_select ON compliance.flow_templates; CREATE POLICY flow_templates_select ON compliance.flow_templates FOR SELECT USING ( COALESCE(current_setting('app.is_super_admin', true), 'f') = 't' OR tenant_id IS NULL OR app_tenant_allowed_for_session(tenant_id) ); DROP POLICY IF EXISTS flow_templates_insert ON compliance.flow_templates; CREATE POLICY flow_templates_insert ON compliance.flow_templates FOR INSERT WITH CHECK ( COALESCE(current_setting('app.is_super_admin', true), 'f') = 't' OR (tenant_id IS NOT NULL AND app_tenant_allowed_for_session(tenant_id)) ); DROP POLICY IF EXISTS flow_templates_update ON compliance.flow_templates; CREATE POLICY flow_templates_update ON compliance.flow_templates FOR UPDATE USING ( COALESCE(current_setting('app.is_super_admin', true), 'f') = 't' OR (tenant_id IS NOT NULL AND app_tenant_allowed_for_session(tenant_id)) ); DROP POLICY IF EXISTS flow_templates_delete ON compliance.flow_templates; CREATE POLICY flow_templates_delete ON compliance.flow_templates FOR DELETE USING ( COALESCE(current_setting('app.is_super_admin', true), 'f') = 't' OR (tenant_id IS NOT NULL AND app_tenant_allowed_for_session(tenant_id)) ); -- 4. Seed 4 個 stage_objects (2026-05-12) INSERT INTO compliance.stage_objects ( code, name_i18n, kind, route_pattern, complete_button_label_i18n, complete_handler_key, precondition_key, default_main_roles, is_builtin, sort ) VALUES ('planning', '{"zh_Hant_TW": "規劃", "en": "Planning"}'::jsonb, 'routed', '/project/projects/:projectUID/settings', '{"zh_Hant_TW": "啟動專案", "en": "Activate Project"}'::jsonb, 'activate_project', NULL, '["manager"]'::jsonb, TRUE, 10), ('task_execution', '{"zh_Hant_TW": "執行任務", "en": "Task Execution"}'::jsonb, 'stateful', NULL, '{"zh_Hant_TW": "啟動稽核", "en": "Launch Audit"}'::jsonb, 'launch_audit', 'all_tasks_completed', '["manager"]'::jsonb, TRUE, 20), ('audit', '{"zh_Hant_TW": "稽核", "en": "Audit"}'::jsonb, 'routed', '/project/projects/:projectUID/ap/:apUID/audit', '{"zh_Hant_TW": "提交稽核", "en": "Confirm Audit"}'::jsonb, 'confirm_audit', 'all_controls_verdict_filled', '["auditor"]'::jsonb, TRUE, 30), ('poam', '{"zh_Hant_TW": "缺失改善", "en": "POA&M"}'::jsonb, 'routed', '/project/projects/:projectUID/ap/:apUID/poam', '{"zh_Hant_TW": "完成改善", "en": "Close Round"}'::jsonb, 'close_round', 'all_poam_closed', '["manager"]'::jsonb, TRUE, 40) ON CONFLICT (code) DO NOTHING; -- 5. Seed 3 個 builtin flow_templates (2026-05-12) -- -- ⚠️ 不用 ON CONFLICT：tenant_id IS NULL 不參與 unique 比對（NULL ≠ NULL）， -- partial unique index 的 inference 條件也不易對齊。 -- 改用 idempotent guard：每個 builtin 一條 INSERT ... SELECT ... WHERE NOT EXISTS -- -- BPMN XML 來源（已用 jedi-flow-engine `BpmnUtils` 驗證 parse 通過，含 stage_object_code + -- main_role camunda:property + full-audit 的 ExclusiveGateway loop back）： -- scripts/sql/seeds/bpmn/builtin-full-audit.bpmn -- scripts/sql/seeds/bpmn/builtin-internal-check.bpmn -- scripts/sql/seeds/bpmn/builtin-self-assessment.bpmn -- -- SQL 用 PostgreSQL dollar-quoting（$xml$...$xml$）內嵌 XML，避免 single-quote 跳脫。 -- 每段 XML 用 `cat` 貼進 SQL 檔，或寫 SQL 時直接從 .bpmn 檔複製貼上。 INSERT INTO compliance.flow_templates (tenant_id, name, description, is_builtin, bpmn_xml, created_user, updated_user) SELECT NULL, 'builtin-full-audit', '完整稽核流程：規劃 → 執行 → 稽核 → 缺失改善 → 結案', TRUE, $xml$<從 scripts/sql/seeds/bpmn/builtin-full-audit.bpmn 完整貼上>$xml$, 'system', 'system' WHERE NOT EXISTS ( SELECT 1 FROM compliance.flow_templates WHERE is_builtin = TRUE AND name = 'builtin-full-audit' AND is_active = TRUE ); INSERT INTO compliance.flow_templates (tenant_id, name, description, is_builtin, bpmn_xml, created_user, updated_user) SELECT NULL, 'builtin-internal-check', '內部自查流程：規劃 → 執行 → 結案', TRUE, $xml$<從 scripts/sql/seeds/bpmn/builtin-internal-check.bpmn 完整貼上>$xml$, 'system', 'system' WHERE NOT EXISTS ( SELECT 1 FROM compliance.flow_templates WHERE is_builtin = TRUE AND name = 'builtin-internal-check' AND is_active = TRUE ); INSERT INTO compliance.flow_templates (tenant_id, name, description, is_builtin, bpmn_xml, created_user, updated_user) SELECT NULL, 'builtin-self-assessment', '自我評估稽核：規劃 → 執行 → 稽核 → 結案（無缺失改善）', TRUE, $xml$<從 scripts/sql/seeds/bpmn/builtin-self-assessment.bpmn 完整貼上>$xml$, 'system', 'system' WHERE NOT EXISTS ( SELECT 1 FROM compliance.flow_templates WHERE is_builtin = TRUE AND name = 'builtin-self-assessment' AND is_active = TRUE ); COMMIT; -- 驗證： -- SELECT code, kind, jsonb_path_query_first(name_i18n, '$.zh_Hant_TW') AS name FROM compliance.stage_objects; -- SELECT name, is_builtin, is_active FROM compliance.flow_templates; ``` > ⚠️ **BPMN XML placeholder**：上面 3 個 builtin 範本的 `bpmn_xml` 是 placeholder；正式 seed 前由 FE 用 BPMN editor 匯出 3 份 XML 字串（或 BE engineer 手寫對應 jedi-flow-engine 範例格式），再以 UPDATE statement 補進 SQL 檔。**這是 Phase A 完成前必須補齊的 blocking item**。 - [ ] **Step 2：本機 dev DB 試跑** ```bash set -a; source .env; set +a PGPASSWORD="$(echo $DB_SECRET | python -c "import sys,json; print(json.loads(sys.stdin.read())['rds_master_password'])")" \ psql -h 192.168.50.188 -p 25432 -U cmmgr -d guidant_ai_stg \ -f scripts/sql/2026-05-12-flow-template-management-schema.sql ``` Expected：BEGIN / 多個 CREATE / GRANT / INSERT 訊息 / COMMIT，沒有 ERROR。 - [ ] **Step 3：驗證 seed 結果** ```bash psql ... -c "SELECT code, kind FROM compliance.stage_objects ORDER BY sort;" psql ... -c "SELECT name, is_builtin FROM compliance.flow_templates WHERE is_active = true ORDER BY id;" ``` Expected：4 個 stage_objects（planning / task_execution / audit / poam），3 個 builtin flow_templates。 - [ ] **Step 4：驗證 RLS — cm_app 用戶看得到 builtin 但看不到別 tenant** ```bash PGPASSWORD='jedi@123!' psql -h 192.168.50.188 -p 25432 -U cm_app -d guidant_ai_stg \ -c "SET app.allowed_tenant_paths = '101'; SELECT name, tenant_id FROM compliance.flow_templates;" ``` Expected：3 個 builtin（tenant_id IS NULL）；無 tenant 101 自訂（尚未建立）。 - [ ] **Step 5：commit** ```bash git add scripts/sql/2026-05-12-flow-template-management-schema.sql git commit -m "feat(flow-template): add DB schema + 4 stage_object + 3 builtin flow_template seeds" ``` --- ## Phase B：Error Code + 域層 ### Task B1：擴充 GrcErrorCode **Files:** - Modify: `common/code/grc_error_code.py` - [ ] **Step 1：grep 確認序號未被佔** ```bash ``` - [ ] **Step 2：在 file 尾部加入新 section** ```python # ── Flow Template (Spec 1 流程管理) ─────────────────────────────────── GRC_FLOW_TEMPLATE_NOT_FOUND = ("流程範本不存在", "GRC_404027") GRC_STAGE_OBJECT_NOT_FOUND = ("階段物件不存在", "GRC_404028") GRC_FLOW_TEMPLATE_BPMN_INVALID = ("BPMN XML 格式無效", "GRC_400040") GRC_FLOW_TEMPLATE_BUILTIN_READONLY = ("內建範本不可編輯或刪除", "GRC_403040") GRC_FLOW_TEMPLATE_NO_PERMISSION = ("無流程範本管理權限", "GRC_403041") GRC_FLOW_TEMPLATE_NAME_DUPLICATED = ("流程範本名稱已存在", "GRC_409030") ``` - [ ] **Step 3：跑 error code 唯一性測試** Run: `pytest test/test_error_code_uniqueness.py -v` Expected: PASS（每個 code 字串唯一）。 ### Task B2：新增 domain entity & query entity **Files:** - Create: `domain/flow_engine/entity/stage_object_entity.py` - Create: `domain/flow_engine/entity/stage_object_query_entity.py` - Create: `domain/flow_engine/entity/flow_template_entity.py` - Create: `domain/flow_engine/entity/flow_template_query_entity.py` - [ ] **Step 1：寫 StageObjectEntity（pure dataclass，鏡射 schema）** ```python # domain/flow_engine/entity/stage_object_entity.py from dataclasses import dataclass, field from datetime import datetime from typing import Optional from uuid import UUID @dataclass class StageObjectEntity: id: Optional[int] = None uid: Optional[str] = None code: str = "" name_i18n: dict = field(default_factory=dict) kind: str = "routed" # routed | stateful route_pattern: Optional[str] = None complete_button_label_i18n: dict = field(default_factory=dict) complete_handler_key: str = "" precondition_key: Optional[str] = None default_main_roles: list = field(default_factory=list) is_builtin: bool = True is_active: bool = True sort: int = 0 created_user: Optional[str] = None updated_user: Optional[str] = None created_at: Optional[datetime] = None updated_at: Optional[datetime] = None ``` - [ ] **Step 2：寫 StageObjectQueryEntity（查詢條件物件）** ```python # domain/flow_engine/entity/stage_object_query_entity.py from dataclasses import dataclass from typing import Optional @dataclass class StageObjectQueryEntity: uid: Optional[str] = None code: Optional[str] = None is_active: Optional[bool] = True ``` - [ ] **Step 3：寫 FlowTemplateEntity + Query** ```python # domain/flow_engine/entity/flow_template_entity.py from dataclasses import dataclass from datetime import datetime from typing import Optional @dataclass class FlowTemplateEntity: id: Optional[int] = None uid: Optional[str] = None tenant_id: Optional[int] = None # NULL = builtin name: str = "" description: Optional[str] = None is_builtin: bool = False is_active: bool = True bpmn_xml: str = "" created_user: Optional[str] = None updated_user: Optional[str] = None created_at: Optional[datetime] = None updated_at: Optional[datetime] = None ``` ```python # domain/flow_engine/entity/flow_template_query_entity.py from dataclasses import dataclass from typing import Optional @dataclass class FlowTemplateQueryEntity: uid: Optional[str] = None tenant_id: Optional[int] = None is_builtin: Optional[bool] = None is_active: Optional[bool] = True name_like: Optional[str] = None # for ILIKE search include_builtin: bool = True # union builtin + tenant ``` - [ ] **Step 4：commit** ```bash git add domain/flow_engine/entity/ git commit -m "feat(flow-template): add stage_object & flow_template domain entities" ``` ### Task B3：Repo interface **Files:** - Create: `domain/flow_engine/repository/stage_object_repo.py` - Create: `domain/flow_engine/repository/flow_template_repo.py` - [ ] **Step 1：寫 IStageObjectRepo** ```python # domain/flow_engine/repository/stage_object_repo.py from abc import ABC, abstractmethod from typing import List, Optional from domain.flow_engine.entity.stage_object_entity import StageObjectEntity from domain.flow_engine.entity.stage_object_query_entity import StageObjectQueryEntity class IStageObjectRepo(ABC): @abstractmethod def find_all(self, query: StageObjectQueryEntity) -> List[StageObjectEntity]: ... @abstractmethod def get_by_uid(self, uid: str) -> Optional[StageObjectEntity]: ... @abstractmethod def get_by_code(self, code: str) -> Optional[StageObjectEntity]: ... ``` - [ ] **Step 2：寫 IFlowTemplateRepo** ```python # domain/flow_engine/repository/flow_template_repo.py from abc import ABC, abstractmethod from typing import List, Optional, Tuple from domain.flow_engine.entity.flow_template_entity import FlowTemplateEntity from domain.flow_engine.entity.flow_template_query_entity import FlowTemplateQueryEntity class IFlowTemplateRepo(ABC): @abstractmethod def find_page(self, query: FlowTemplateQueryEntity, page: int, size: int) -> Tuple[List[FlowTemplateEntity], int]: ... @abstractmethod def get_by_uid(self, uid: str) -> Optional[FlowTemplateEntity]: ... @abstractmethod def exists_active_name(self, tenant_id: int, name: str, exclude_uid: Optional[str] = None) -> bool: ... @abstractmethod def create(self, entity: FlowTemplateEntity) -> FlowTemplateEntity: ... @abstractmethod def update(self, entity: FlowTemplateEntity) -> FlowTemplateEntity: ... @abstractmethod def soft_delete(self, uid: str, curr_user: str) -> bool: ... ``` - [ ] **Step 3：commit** ```bash git add domain/flow_engine/repository/stage_object_repo.py \ domain/flow_engine/repository/flow_template_repo.py git commit -m "feat(flow-template): add stage_object & flow_template repo interfaces" ``` ### Task B4：Domain service **Files:** - Create: `domain/flow_engine/service/stage_object_domain_service.py` - Create: `domain/flow_engine/service/flow_template_domain_service.py` - Test: `test/test_flow_template_domain_service.py` - [ ] **Step 1：寫 StageObjectDomainService** ```python # domain/flow_engine/service/stage_object_domain_service.py from typing import List, Optional from domain.flow_engine.entity.stage_object_entity import StageObjectEntity from domain.flow_engine.entity.stage_object_query_entity import StageObjectQueryEntity from domain.flow_engine.repository.stage_object_repo import IStageObjectRepo class StageObjectDomainService: def __init__(self, stage_object_repo: IStageObjectRepo): self._repo = stage_object_repo def list_active(self) -> List[StageObjectEntity]: return self._repo.find_all(StageObjectQueryEntity(is_active=True)) def get_one(self, query: StageObjectQueryEntity) -> Optional[StageObjectEntity]: if query.uid: return self._repo.get_by_uid(query.uid) if query.code: return self._repo.get_by_code(query.code) return None ``` - [ ] **Step 2：寫 FlowTemplateDomainService**（純委派，不含商業邏輯；商業邏輯放 app service） ```python # domain/flow_engine/service/flow_template_domain_service.py from typing import List, Optional, Tuple from domain.flow_engine.entity.flow_template_entity import FlowTemplateEntity from domain.flow_engine.entity.flow_template_query_entity import FlowTemplateQueryEntity from domain.flow_engine.repository.flow_template_repo import IFlowTemplateRepo class FlowTemplateDomainService: def __init__(self, flow_template_repo: IFlowTemplateRepo): self._repo = flow_template_repo def find_page(self, query: FlowTemplateQueryEntity, page: int, size: int) -> Tuple[List[FlowTemplateEntity], int]: return self._repo.find_page(query, page, size) def get_by_uid(self, uid: str) -> Optional[FlowTemplateEntity]: return self._repo.get_by_uid(uid) def exists_active_name(self, tenant_id: int, name: str, exclude_uid: Optional[str] = None) -> bool: return self._repo.exists_active_name(tenant_id, name, exclude_uid) def create(self, entity: FlowTemplateEntity) -> FlowTemplateEntity: return self._repo.create(entity) def update(self, entity: FlowTemplateEntity) -> FlowTemplateEntity: return self._repo.update(entity) def soft_delete(self, uid: str, curr_user: str) -> bool: return self._repo.soft_delete(uid, curr_user) ``` - [ ] **Step 3：寫 domain service 單元測試（mock repo）** ```python # test/test_flow_template_domain_service.py from unittest.mock import MagicMock import pytest from domain.flow_engine.entity.flow_template_entity import FlowTemplateEntity from domain.flow_engine.entity.flow_template_query_entity import FlowTemplateQueryEntity from domain.flow_engine.service.flow_template_domain_service import FlowTemplateDomainService @pytest.fixture def mock_repo(): return MagicMock() def test_find_page_delegates_to_repo(mock_repo): mock_repo.find_page.return_value = ([], 0) svc = FlowTemplateDomainService(flow_template_repo=mock_repo) q = FlowTemplateQueryEntity(tenant_id=101) result = svc.find_page(q, page=1, size=10) assert result == ([], 0) mock_repo.find_page.assert_called_once_with(q, 1, 10) def test_exists_active_name_excludes_uid(mock_repo): mock_repo.exists_active_name.return_value = False svc = FlowTemplateDomainService(flow_template_repo=mock_repo) result = svc.exists_active_name(tenant_id=101, name="abc", exclude_uid="some-uid") assert result is False mock_repo.exists_active_name.assert_called_once_with(101, "abc", "some-uid") ``` - [ ] **Step 4：跑測試 GREEN** Run: `pytest test/test_flow_template_domain_service.py -v` Expected: PASS（2 tests） - [ ] **Step 5：commit** ```bash git add domain/flow_engine/service/stage_object_domain_service.py \ domain/flow_engine/service/flow_template_domain_service.py \ test/test_flow_template_domain_service.py git commit -m "feat(flow-template): add domain services + unit tests" ``` --- ## Phase C：基礎層（ORM + Mapper + Repo Impl） ### Task C1：SQLAlchemy ORM models **Files:** - Create: `infra/flow_engine/models/stage_object.py` - Create: `infra/flow_engine/models/flow_template.py` - [ ] **Step 1：寫 StageObject ORM model** ```python # infra/flow_engine/models/stage_object.py from sqlalchemy import Column, Integer, String, Boolean, DateTime, Text from sqlalchemy.dialects.postgresql import UUID, JSONB from sqlalchemy.sql import func from jedi_common.session.database.base_model import BaseModel class StageObject(BaseModel): __tablename__ = "stage_objects" __table_args__ = {"schema": "compliance"} id = Column(Integer, primary_key=True, autoincrement=True) uid = Column(UUID(as_uuid=False), nullable=False, unique=True, server_default=func.gen_random_uuid()) code = Column(String(50), nullable=False, unique=True) name_i18n = Column(JSONB, nullable=False) kind = Column(String(20), nullable=False) route_pattern = Column(String(255)) complete_button_label_i18n = Column(JSONB, nullable=False) complete_handler_key = Column(String(100), nullable=False) precondition_key = Column(String(100)) default_main_roles = Column(JSONB, nullable=False, default=list) is_builtin = Column(Boolean, nullable=False, default=True) is_active = Column(Boolean, nullable=False, default=True) sort = Column(Integer, nullable=False, default=0) created_user = Column(String(100)) updated_user = Column(String(100)) created_at = Column(DateTime(timezone=True), server_default=func.now()) updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now()) ``` - [ ] **Step 2：寫 FlowTemplate ORM model** ```python # infra/flow_engine/models/flow_template.py from sqlalchemy import Column, Integer, String, Boolean, DateTime, Text from sqlalchemy.dialects.postgresql import UUID from sqlalchemy.sql import func from jedi_common.session.database.base_model import BaseModel class FlowTemplate(BaseModel): __tablename__ = "flow_templates" __table_args__ = {"schema": "compliance"} id = Column(Integer, primary_key=True, autoincrement=True) uid = Column(UUID(as_uuid=False), nullable=False, unique=True, server_default=func.gen_random_uuid()) tenant_id = Column(Integer, nullable=True) # NULL = builtin name = Column(String(200), nullable=False) description = Column(Text) is_builtin = Column(Boolean, nullable=False, default=False) is_active = Column(Boolean, nullable=False, default=True) bpmn_xml = Column(Text, nullable=False) created_user = Column(String(100)) updated_user = Column(String(100)) created_at = Column(DateTime(timezone=True), server_default=func.now()) updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now()) ``` > **驗證假設**：先確認 `jedi_common.session.database.base_model.BaseModel` 名字仍是這個；如不在，grep 找實際 BaseModel 名稱（可能是 `Base` 或 `DeclarativeBase`）。參考既有 `infra/flow_engine/models/job_evidence.py:1` 看 import。 - [ ] **Step 3：commit** ```bash git add infra/flow_engine/models/stage_object.py infra/flow_engine/models/flow_template.py git commit -m "feat(flow-template): add SQLAlchemy ORM models" ``` ### Task C2：Mappers **Files:** - Create: `infra/flow_engine/mapper/stage_object_mapper.py` - Create: `infra/flow_engine/mapper/flow_template_mapper.py` - [ ] **Step 1：寫 StageObjectMapper（model ↔ entity 雙向轉換）** ```python # infra/flow_engine/mapper/stage_object_mapper.py from domain.flow_engine.entity.stage_object_entity import StageObjectEntity from infra.flow_engine.models.stage_object import StageObject class StageObjectMapper: @staticmethod def to_entity(m: StageObject) -> StageObjectEntity: if m is None: return None return StageObjectEntity( id=m.id, uid=str(m.uid), code=m.code, name_i18n=m.name_i18n or {}, kind=m.kind, route_pattern=m.route_pattern, complete_button_label_i18n=m.complete_button_label_i18n or {}, complete_handler_key=m.complete_handler_key, precondition_key=m.precondition_key, default_main_roles=m.default_main_roles or [], is_builtin=m.is_builtin, is_active=m.is_active, sort=m.sort, created_user=m.created_user, updated_user=m.updated_user, created_at=m.created_at, updated_at=m.updated_at, ) @staticmethod def to_model(e: StageObjectEntity) -> StageObject: m = StageObject() if e.id is not None: m.id = e.id m.code = e.code m.name_i18n = e.name_i18n m.kind = e.kind m.route_pattern = e.route_pattern m.complete_button_label_i18n = e.complete_button_label_i18n m.complete_handler_key = e.complete_handler_key m.precondition_key = e.precondition_key m.default_main_roles = e.default_main_roles m.is_builtin = e.is_builtin m.is_active = e.is_active m.sort = e.sort m.created_user = e.created_user m.updated_user = e.updated_user return m ``` - [ ] **Step 2：寫 FlowTemplateMapper** ```python # infra/flow_engine/mapper/flow_template_mapper.py from domain.flow_engine.entity.flow_template_entity import FlowTemplateEntity from infra.flow_engine.models.flow_template import FlowTemplate class FlowTemplateMapper: @staticmethod def to_entity(m: FlowTemplate) -> FlowTemplateEntity: if m is None: return None return FlowTemplateEntity( id=m.id, uid=str(m.uid), tenant_id=m.tenant_id, name=m.name, description=m.description, is_builtin=m.is_builtin, is_active=m.is_active, bpmn_xml=m.bpmn_xml, created_user=m.created_user, updated_user=m.updated_user, created_at=m.created_at, updated_at=m.updated_at, ) @staticmethod def to_model(e: FlowTemplateEntity) -> FlowTemplate: m = FlowTemplate() if e.id is not None: m.id = e.id m.tenant_id = e.tenant_id m.name = e.name m.description = e.description m.is_builtin = e.is_builtin m.is_active = e.is_active m.bpmn_xml = e.bpmn_xml m.created_user = e.created_user m.updated_user = e.updated_user return m ``` - [ ] **Step 3：commit** ```bash git add infra/flow_engine/mapper/ git commit -m "feat(flow-template): add entity ↔ model mappers" ``` ### Task C3：Repo implementations **Files:** - Create: `infra/flow_engine/repository/stage_object_repo_impl.py` - Create: `infra/flow_engine/repository/flow_template_repo_impl.py` - [ ] **Step 1：寫 StageObjectRepoImpl** ```python # infra/flow_engine/repository/stage_object_repo_impl.py from typing import List, Optional from jedi_common.session.database.db import get_session from domain.flow_engine.entity.stage_object_entity import StageObjectEntity from domain.flow_engine.entity.stage_object_query_entity import StageObjectQueryEntity from domain.flow_engine.repository.stage_object_repo import IStageObjectRepo from infra.flow_engine.mapper.stage_object_mapper import StageObjectMapper from infra.flow_engine.models.stage_object import StageObject class StageObjectRepoImpl(IStageObjectRepo): """Caller must be inside a @transaction scope (uses get_session()).""" @property def session(self): return get_session() def find_all(self, query: StageObjectQueryEntity) -> List[StageObjectEntity]: q = self.session.query(StageObject) if query.is_active is not None: q = q.filter(StageObject.is_active == query.is_active) q = q.order_by(StageObject.sort.asc(), StageObject.id.asc()) return [StageObjectMapper.to_entity(m) for m in q.all()] def get_by_uid(self, uid: str) -> Optional[StageObjectEntity]: m = self.session.query(StageObject).filter_by(uid=uid).first() return StageObjectMapper.to_entity(m) def get_by_code(self, code: str) -> Optional[StageObjectEntity]: m = self.session.query(StageObject).filter_by(code=code).first() return StageObjectMapper.to_entity(m) ``` - [ ] **Step 2：寫 FlowTemplateRepoImpl** ```python # infra/flow_engine/repository/flow_template_repo_impl.py from typing import List, Optional, Tuple from sqlalchemy import or_ from jedi_common.session.database.db import get_session from domain.flow_engine.entity.flow_template_entity import FlowTemplateEntity from domain.flow_engine.entity.flow_template_query_entity import FlowTemplateQueryEntity from domain.flow_engine.repository.flow_template_repo import IFlowTemplateRepo from infra.flow_engine.mapper.flow_template_mapper import FlowTemplateMapper from infra.flow_engine.models.flow_template import FlowTemplate class FlowTemplateRepoImpl(IFlowTemplateRepo): """Caller must be inside a @transaction scope.""" @property def session(self): return get_session() def find_page(self, query: FlowTemplateQueryEntity, page: int, size: int) -> Tuple[List[FlowTemplateEntity], int]: q = self.session.query(FlowTemplate) # tenant + builtin union（RLS 已濾跨 tenant，這裡只需 union builtin） if query.include_builtin and query.tenant_id is not None: q = q.filter(or_( FlowTemplate.tenant_id == query.tenant_id, FlowTemplate.tenant_id.is_(None), )) elif query.tenant_id is not None: q = q.filter(FlowTemplate.tenant_id == query.tenant_id) if query.is_active is not None: q = q.filter(FlowTemplate.is_active == query.is_active) if query.is_builtin is not None: q = q.filter(FlowTemplate.is_builtin == query.is_builtin) if query.name_like: q = q.filter(FlowTemplate.name.ilike(f"%{query.name_like}%")) total = q.count() rows = (q.order_by(FlowTemplate.is_builtin.desc(), FlowTemplate.name.asc()) .offset((page - 1) * size).limit(size).all()) return [FlowTemplateMapper.to_entity(m) for m in rows], total def get_by_uid(self, uid: str) -> Optional[FlowTemplateEntity]: m = self.session.query(FlowTemplate).filter_by(uid=uid, is_active=True).first() return FlowTemplateMapper.to_entity(m) def exists_active_name(self, tenant_id: int, name: str, exclude_uid: Optional[str] = None) -> bool: q = self.session.query(FlowTemplate).filter( FlowTemplate.tenant_id == tenant_id, FlowTemplate.name == name, FlowTemplate.is_active.is_(True), ) if exclude_uid: q = q.filter(FlowTemplate.uid != exclude_uid) return self.session.query(q.exists()).scalar() def create(self, entity: FlowTemplateEntity) -> FlowTemplateEntity: m = FlowTemplateMapper.to_model(entity) self.session.add(m) self.session.flush() return FlowTemplateMapper.to_entity(m) def update(self, entity: FlowTemplateEntity) -> FlowTemplateEntity: m = self.session.query(FlowTemplate).filter_by(uid=entity.uid).first() if not m: return None m.name = entity.name m.description = entity.description m.bpmn_xml = entity.bpmn_xml m.updated_user = entity.updated_user self.session.flush() return FlowTemplateMapper.to_entity(m) def soft_delete(self, uid: str, curr_user: str) -> bool: m = self.session.query(FlowTemplate).filter_by(uid=uid, is_active=True).first() if not m: return False m.is_active = False m.updated_user = curr_user self.session.flush() return True ``` - [ ] **Step 3：commit** ```bash git add infra/flow_engine/repository/stage_object_repo_impl.py \ infra/flow_engine/repository/flow_template_repo_impl.py git commit -m "feat(flow-template): add repo implementations" ``` --- ## Phase D：應用層 ### Task D1：DTOs **Files:** - Create: `app/flow_engine/dto/stage_object_dto.py` - Create: `app/flow_engine/dto/flow_template_dto.py` - [ ] **Step 1：寫 StageObjectDto**（envelope-ready，含 i18n resolved label） ```python # app/flow_engine/dto/stage_object_dto.py from dataclasses import dataclass, asdict from typing import Optional, List from domain.flow_engine.entity.stage_object_entity import StageObjectEntity @dataclass class StageObjectDto: uid: str code: str name: str # resolved per locale kind: str route_pattern: Optional[str] complete_button_label: str # resolved per locale complete_handler_key: str precondition_key: Optional[str] default_main_roles: List[str] is_builtin: bool sort: int @staticmethod def from_entity(e: StageObjectEntity, locale: str = "zh_Hant_TW") -> "StageObjectDto": return StageObjectDto( uid=e.uid, code=e.code, name=(e.name_i18n or {}).get(locale) or (e.name_i18n or {}).get("zh_Hant_TW") or e.code, kind=e.kind, route_pattern=e.route_pattern, complete_button_label=(e.complete_button_label_i18n or {}).get(locale) or (e.complete_button_label_i18n or {}).get("zh_Hant_TW") or "", complete_handler_key=e.complete_handler_key, precondition_key=e.precondition_key, default_main_roles=e.default_main_roles or [], is_builtin=e.is_builtin, sort=e.sort, ) def to_dict(self): return asdict(self) ``` - [ ] **Step 2：寫 FlowTemplateDto** ```python # app/flow_engine/dto/flow_template_dto.py from dataclasses import dataclass, asdict from datetime import datetime from typing import Optional from domain.flow_engine.entity.flow_template_entity import FlowTemplateEntity @dataclass class FlowTemplateDto: uid: str name: str description: Optional[str] is_builtin: bool bpmn_xml: Optional[str] # 列表頁不帶；詳情才帶 created_user: Optional[str] created_user_name: Optional[str] # nickname（app service enrich） updated_user: Optional[str] updated_user_name: Optional[str] created_at: Optional[datetime] updated_at: Optional[datetime] @staticmethod def from_entity(e: FlowTemplateEntity, include_xml: bool = False, user_nickname_map: dict = None) -> "FlowTemplateDto": m = user_nickname_map or {} return FlowTemplateDto( uid=e.uid, name=e.name, description=e.description, is_builtin=e.is_builtin, bpmn_xml=e.bpmn_xml if include_xml else None, created_user=e.created_user, created_user_name=m.get(e.created_user), updated_user=e.updated_user, updated_user_name=m.get(e.updated_user), created_at=e.created_at, updated_at=e.updated_at, ) def to_dict(self): return asdict(self) ``` - [ ] **Step 3：commit** ```bash git add app/flow_engine/dto/ git commit -m "feat(flow-template): add DTOs with i18n + audit nickname enrichment" ``` ### Task D2：StageObjectAppService **Files:** - Create: `app/flow_engine/service/stage_object_app_service.py` - [ ] **Step 1：實作 service** ```python # app/flow_engine/service/stage_object_app_service.py from typing import List from jedi_common.session.database.db import transaction from app.flow_engine.dto.stage_object_dto import StageObjectDto from domain.flow_engine.service.stage_object_domain_service import StageObjectDomainService class StageObjectAppService: def __init__(self, stage_object_domain_service: StageObjectDomainService): self._domain_service = stage_object_domain_service @transaction def list_active(self, locale: str = "zh_Hant_TW") -> List[StageObjectDto]: entities = self._domain_service.list_active() return [StageObjectDto.from_entity(e, locale=locale) for e in entities] ``` - [ ] **Step 2：commit** ```bash git add app/flow_engine/service/stage_object_app_service.py git commit -m "feat(flow-template): add StageObjectAppService (list only)" ``` ### Task D3：FlowTemplateAppService（核心商業邏輯） **Files:** - Create: `app/flow_engine/service/flow_template_app_service.py` - Test: `test/test_flow_template_app_service.py` - [ ] **Step 1：寫先失敗的測試 — duplicate 必須複製成 tenant 自訂（非 builtin）** ```python # test/test_flow_template_app_service.py from unittest.mock import MagicMock, patch import pytest from app.flow_engine.service.flow_template_app_service import FlowTemplateAppService from domain.flow_engine.entity.flow_template_entity import FlowTemplateEntity from jedi_common.handler.exception import ForbiddenError, NotFound, ConflictError, BadRequestError @pytest.fixture def mock_domain(): return MagicMock() @pytest.fixture def mock_user_service(): return MagicMock() @pytest.fixture def service(mock_domain, mock_user_service): return FlowTemplateAppService( flow_template_domain_service=mock_domain, user_service=mock_user_service, ) @pytest.fixture def builtin_template(): return FlowTemplateEntity( id=1, uid="builtin-uid", tenant_id=None, name="builtin-full-audit", description="full", is_builtin=True, is_active=True, bpmn_xml="", ) @pytest.fixture def mock_user_context(): ctx = MagicMock() ctx.id = 10 ctx.login_name = "raymond" ctx.tenant_id = 101 return ctx def test_duplicate_builtin_creates_tenant_custom(service, mock_domain, builtin_template, mock_user_context): mock_domain.get_by_uid.return_value = builtin_template mock_domain.exists_active_name.return_value = False mock_domain.create.side_effect = lambda e: FlowTemplateEntity( **{**e.__dict__, "id": 99, "uid": "new-uid"}) with patch("app.flow_engine.service.flow_template_app_service.get_user_context", return_value=mock_user_context): result = service.duplicate("builtin-uid", new_name="my copy") created_arg = mock_domain.create.call_args[0][0] assert created_arg.is_builtin is False assert created_arg.tenant_id == 101 assert created_arg.name == "my copy" assert created_arg.bpmn_xml == builtin_template.bpmn_xml assert created_arg.created_user == "raymond" def test_update_builtin_raises_forbidden(service, mock_domain, builtin_template, mock_user_context): mock_domain.get_by_uid.return_value = builtin_template with patch("app.flow_engine.service.flow_template_app_service.get_user_context", return_value=mock_user_context): with pytest.raises(ForbiddenError): service.update("builtin-uid", name="x", description="", bpmn_xml="") def test_delete_builtin_raises_forbidden(service, mock_domain, builtin_template, mock_user_context): mock_domain.get_by_uid.return_value = builtin_template with patch("app.flow_engine.service.flow_template_app_service.get_user_context", return_value=mock_user_context): with pytest.raises(ForbiddenError): service.soft_delete("builtin-uid") def test_create_name_duplicate_raises_conflict(service, mock_domain, mock_user_context): mock_domain.exists_active_name.return_value = True with patch("app.flow_engine.service.flow_template_app_service.get_user_context", return_value=mock_user_context): with pytest.raises(ConflictError): service.create(name="dup", description="", bpmn_xml="") def test_create_invalid_bpmn_raises_bad_request(service, mock_domain, mock_user_context): mock_domain.exists_active_name.return_value = False with patch("app.flow_engine.service.flow_template_app_service.get_user_context", return_value=mock_user_context): with pytest.raises(BadRequestError): service.create(name="x", description="", bpmn_xml=" Tuple[List[FlowTemplateDto], int]: ctx = get_user_context() q = FlowTemplateQueryEntity( tenant_id=ctx.tenant_id, is_builtin=is_builtin, name_like=name, include_builtin=True, is_active=True, ) entities, total = self._domain.find_page(q, page=page, size=size) nickname_map = self._build_nickname_map(entities) dtos = [FlowTemplateDto.from_entity(e, include_xml=False, user_nickname_map=nickname_map) for e in entities] return dtos, total @transaction def get(self, uid: str) -> FlowTemplateDto: entity = self._require_active(uid) nickname_map = self._build_nickname_map([entity]) return FlowTemplateDto.from_entity(entity, include_xml=True, user_nickname_map=nickname_map) # ── Write ─────────────────────────────────────────────────────────── @transaction def create(self, name: str, description: Optional[str], bpmn_xml: str) -> FlowTemplateDto: self._validate_bpmn(bpmn_xml) ctx = get_user_context() if self._domain.exists_active_name(ctx.tenant_id, name): raise ConflictError(GrcErrorCode.GRC_FLOW_TEMPLATE_NAME_DUPLICATED) entity = FlowTemplateEntity( tenant_id=ctx.tenant_id, name=name, description=description, is_builtin=False, is_active=True, bpmn_xml=bpmn_xml, created_user=ctx.login_name, updated_user=ctx.login_name, ) created = self._domain.create(entity) nickname_map = self._build_nickname_map([created]) return FlowTemplateDto.from_entity(created, include_xml=True, user_nickname_map=nickname_map) @transaction def update(self, uid: str, name: str, description: Optional[str], bpmn_xml: str) -> FlowTemplateDto: existing = self._require_active(uid) self._guard_not_builtin(existing) self._validate_bpmn(bpmn_xml) ctx = get_user_context() if self._domain.exists_active_name(ctx.tenant_id, name, exclude_uid=uid): raise ConflictError(GrcErrorCode.GRC_FLOW_TEMPLATE_NAME_DUPLICATED) existing.name = name existing.description = description existing.bpmn_xml = bpmn_xml existing.updated_user = ctx.login_name updated = self._domain.update(existing) nickname_map = self._build_nickname_map([updated]) return FlowTemplateDto.from_entity(updated, include_xml=True, user_nickname_map=nickname_map) @transaction def soft_delete(self, uid: str) -> bool: existing = self._require_active(uid) self._guard_not_builtin(existing) ctx = get_user_context() return self._domain.soft_delete(uid, ctx.login_name) @transaction def duplicate(self, source_uid: str, new_name: str) -> FlowTemplateDto: source = self._require_active(source_uid) ctx = get_user_context() if self._domain.exists_active_name(ctx.tenant_id, new_name): raise ConflictError(GrcErrorCode.GRC_FLOW_TEMPLATE_NAME_DUPLICATED) new_entity = FlowTemplateEntity( tenant_id=ctx.tenant_id, name=new_name, description=source.description, is_builtin=False, is_active=True, bpmn_xml=source.bpmn_xml, created_user=ctx.login_name, updated_user=ctx.login_name, ) created = self._domain.create(new_entity) nickname_map = self._build_nickname_map([created]) return FlowTemplateDto.from_entity(created, include_xml=True, user_nickname_map=nickname_map) # ── Helpers ───────────────────────────────────────────────────────── def _require_active(self, uid: str) -> FlowTemplateEntity: e = self._domain.get_by_uid(uid) if not e or not e.is_active: raise NotFound(GrcErrorCode.GRC_FLOW_TEMPLATE_NOT_FOUND) return e def _guard_not_builtin(self, e: FlowTemplateEntity): if e.is_builtin: raise ForbiddenError(GrcErrorCode.GRC_FLOW_TEMPLATE_BUILTIN_READONLY) def _validate_bpmn(self, xml: str): """簡單驗證：解析得起來、含 StartEvent / EndEvent / >=1 UserTask。 ⚠️ Verified (2026-05-12)：`jedi_flow_engine.common.utils.bpmn_uilts.BpmnUtils` 是 **instance class**（不是 static），method 為 `parser_bpmn_xml(self)`。 用法：`util = BpmnUtils(xml); parsed = util.parser_bpmn_xml()` 若開工時 method 名/簽名又改，退化為 xmltodict.parse + 字串檢查。 """ from jedi_flow_engine.common.utils.bpmn_uilts import BpmnUtils try: util = BpmnUtils(xml) parsed = util.parser_bpmn_xml() except Exception: raise BadRequestError(GrcErrorCode.GRC_FLOW_TEMPLATE_BPMN_INVALID) # parser_bpmn_xml() 回 dict — 對齊真實 key 名（執行時用 print(parsed.keys()) 確認） # 常見 key：start_event / end_event / jobs（含 UserTask + ServiceTask） if not parsed.get("start_event") or not parsed.get("end_event"): raise BadRequestError(GrcErrorCode.GRC_FLOW_TEMPLATE_BPMN_INVALID) jobs = parsed.get("jobs") or parsed.get("user_tasks") or [] if not jobs: raise BadRequestError(GrcErrorCode.GRC_FLOW_TEMPLATE_BPMN_INVALID) def _build_nickname_map(self, entities: List[FlowTemplateEntity]) -> dict: """根據 CLAUDE.md 審計欄位規範：login_name → nickname ⚠️ Verified (2026-05-12)：jedi-auth `UserAppService` 沒有 `get_users_by_login_names`。實際 API 是 `get_users(query_entity)` + `UserQueryEntity(login_names=[...])`。本 method 假設 `self._user_service` 提供 `get_users_by_login_names` 為 wrapper；wrapper 需於 user_service 加 thin method（或改在這裡 import UserQueryEntity 直接呼叫 get_users）。 """ login_names = {e.created_user for e in entities if e.created_user} \ | {e.updated_user for e in entities if e.updated_user} login_names.discard(None) login_names.discard("system") if not login_names: return {"system": "系統"} # 開工時兩個選一個（不要兩個都寫）： # 選項 1：在 jedi-auth user_service 加 thin wrapper `get_users_by_login_names(list)` # 選項 2：在這裡直接組 UserQueryEntity，呼叫 get_users users = self._user_service.get_users_by_login_names(list(login_names)) return {"system": "系統", **{u.login_name: u.nickname for u in users}} ``` > ⚠️ **Pre-flight 已 verify (2026-05-12)**： > - `BpmnUtils` 是 instance class、method `parser_bpmn_xml(self)`（不是 static `parse()`）；plan code 已對齊。 > - `user_service.get_users_by_login_names` 在 jedi-auth **不存在**，實際是 `get_users(query_entity=UserQueryEntity(login_names=[...]))`。開工二選一： > - 在 jedi-auth `UserAppService` 加 thin wrapper（要 jedi-auth 進版，user 須核准）；或 > - `_build_nickname_map` 直接組 `UserQueryEntity` 呼叫 `get_users()`，省進版。 > - **推薦選後者**（不動 jedi-auth）— plan 開工時把上面 method call 改成 `self._user_service.get_users(UserQueryEntity(login_names=list(login_names)))`。 > - **連動修改**：`test_flow_template_app_service.py` 的 `mock_user_service` mock target 也要對齊（mock `get_users` 而非 `get_users_by_login_names`）。 - [ ] **Step 4：跑測試 GREEN** Run: `pytest test/test_flow_template_app_service.py -v` Expected: 5 tests PASS - [ ] **Step 5：commit** ```bash git add app/flow_engine/service/flow_template_app_service.py \ test/test_flow_template_app_service.py git commit -m "feat(flow-template): add FlowTemplateAppService + CRUD/duplicate + tests" ``` --- ## Phase E：API 層 ### Task E1：Serializers **Files:** - Create: `api/flow_engine/serializers/stage_object.py` - Create: `api/flow_engine/serializers/flow_template.py` - Test: `test/test_flow_template_serializer.py` - [ ] **Step 1：寫 StageObject serializer（read-only response only）** ```python # api/flow_engine/serializers/stage_object.py from marshmallow import Schema, fields class StageObjectResponseSchema(Schema): uid = fields.Str(required=True) code = fields.Str(required=True) name = fields.Str(required=True) kind = fields.Str(required=True) route_pattern = fields.Str(allow_none=True) complete_button_label = fields.Str(required=True) complete_handler_key = fields.Str(required=True) precondition_key = fields.Str(allow_none=True) default_main_roles = fields.List(fields.Str(), load_default=list) is_builtin = fields.Bool(required=True) sort = fields.Int(required=True) class StageObjectListResponseSchema(Schema): items = fields.List(fields.Nested(StageObjectResponseSchema), required=True) ``` - [ ] **Step 2：寫 FlowTemplate serializer** ```python # api/flow_engine/serializers/flow_template.py from marshmallow import Schema, fields, validate # ── Request ───────────────────────────────────────────────────────── class FlowTemplateCreateRequestSchema(Schema): name = fields.Str(required=True, validate=validate.Length(min=1, max=200)) description = fields.Str(load_default=None, allow_none=True) bpmn_xml = fields.Str(required=True, validate=validate.Length(min=1)) class FlowTemplateUpdateRequestSchema(Schema): name = fields.Str(required=True, validate=validate.Length(min=1, max=200)) description = fields.Str(load_default=None, allow_none=True) bpmn_xml = fields.Str(required=True, validate=validate.Length(min=1)) class FlowTemplateDuplicateRequestSchema(Schema): name = fields.Str(required=True, validate=validate.Length(min=1, max=200)) class FlowTemplateListRequestSchema(Schema): name = fields.Str(load_default=None) is_builtin = fields.Bool(load_default=None) page = fields.Int(load_default=1, validate=validate.Range(min=1)) size = fields.Int(load_default=20, validate=validate.Range(min=1, max=100)) # ── Response ──────────────────────────────────────────────────────── class FlowTemplateResponseSchema(Schema): uid = fields.Str(required=True) name = fields.Str(required=True) description = fields.Str(allow_none=True) is_builtin = fields.Bool(required=True) bpmn_xml = fields.Str(allow_none=True) # only in detail created_user = fields.Str(allow_none=True) created_user_name = fields.Str(allow_none=True) updated_user = fields.Str(allow_none=True) updated_user_name = fields.Str(allow_none=True) created_at = fields.DateTime(allow_none=True) updated_at = fields.DateTime(allow_none=True) class FlowTemplateListResponseSchema(Schema): items = fields.List(fields.Nested(FlowTemplateResponseSchema), required=True) total = fields.Int(required=True) page = fields.Int(required=True) size = fields.Int(required=True) ``` - [ ] **Step 3：寫 serializer 測試（必填欄位、長度上下界、bool 轉型）** ```python # test/test_flow_template_serializer.py import pytest from marshmallow import ValidationError from api.flow_engine.serializers.flow_template import ( FlowTemplateCreateRequestSchema, FlowTemplateListRequestSchema, FlowTemplateDuplicateRequestSchema, ) def test_create_requires_name_and_bpmn(): with pytest.raises(ValidationError) as exc: FlowTemplateCreateRequestSchema().load({}) err = exc.value.messages assert "name" in err and "bpmn_xml" in err def test_create_name_max_length(): with pytest.raises(ValidationError): FlowTemplateCreateRequestSchema().load({"name": "x" * 201, "bpmn_xml": ""}) def test_list_defaults_page_size(): data = FlowTemplateListRequestSchema().load({}) assert data["page"] == 1 and data["size"] == 20 def test_duplicate_requires_name(): with pytest.raises(ValidationError): FlowTemplateDuplicateRequestSchema().load({}) ``` - [ ] **Step 4：跑測試 GREEN** Run: `pytest test/test_flow_template_serializer.py -v` Expected: 4 tests PASS - [ ] **Step 5：commit** ```bash git add api/flow_engine/serializers/stage_object.py \ api/flow_engine/serializers/flow_template.py \ test/test_flow_template_serializer.py git commit -m "feat(flow-template): add marshmallow serializers + tests" ``` ### Task E2：Routes（7 個 Resource） **Files:** - Create: `api/flow_engine/routes/stage_object_route.py` - Create: `api/flow_engine/routes/flow_template_route.py` - Modify: `api/flow_engine/__init__.py`（加 register） - Test: `test/test_api_flow_template_route.py`、`test/test_api_stage_object_route.py` - [ ] **Step 1：StageObjectsRoute** ```python # api/flow_engine/routes/stage_object_route.py from flask import request from flask_jwt_extended import jwt_required from flask_restful import Resource from dependency_injector.wiring import Provide, inject from flask_babel import get_locale from api.flow_engine.serializers.stage_object import StageObjectListResponseSchema from app.flow_engine.service.stage_object_app_service import StageObjectAppService from common.response.return_response import return_response from di_containers.containers import Containers class StageObjectsRoute(Resource): method_decorators = [jwt_required()] @inject def get(self, service: StageObjectAppService = Provide[ Containers.flow_template_container.stage_object_app_service]): locale = str(get_locale()) items = service.list_active(locale=locale) body = StageObjectListResponseSchema().dump({"items": [i.to_dict() for i in items]}) return return_response(True, body) ``` - [ ] **Step 2：FlowTemplate routes（6 個 Resource）** ```python # api/flow_engine/routes/flow_template_route.py from flask import request from flask_jwt_extended import jwt_required from flask_restful import Resource from dependency_injector.wiring import Provide, inject from api.flow_engine.serializers.flow_template import ( FlowTemplateCreateRequestSchema, FlowTemplateUpdateRequestSchema, FlowTemplateDuplicateRequestSchema, FlowTemplateListRequestSchema, FlowTemplateResponseSchema, FlowTemplateListResponseSchema, ) from app.flow_engine.service.flow_template_app_service import FlowTemplateAppService from common.response.return_response import return_response from di_containers.containers import Containers class FlowTemplatesRoute(Resource): method_decorators = [jwt_required()] @inject def get(self, service: FlowTemplateAppService = Provide[ Containers.flow_template_container.flow_template_app_service]): params = FlowTemplateListRequestSchema().load(request.args.to_dict(flat=True)) items, total = service.list( name=params.get("name"), is_builtin=params.get("is_builtin"), page=params["page"], size=params["size"], ) body = FlowTemplateListResponseSchema().dump({ "items": [i.to_dict() for i in items], "total": total, "page": params["page"], "size": params["size"], }) return return_response(True, body) @inject def post(self, service: FlowTemplateAppService = Provide[ Containers.flow_template_container.flow_template_app_service]): body = FlowTemplateCreateRequestSchema().load(request.get_json() or {}) dto = service.create(name=body["name"], description=body.get("description"), bpmn_xml=body["bpmn_xml"]) return return_response(True, FlowTemplateResponseSchema().dump(dto.to_dict())) class FlowTemplateRoute(Resource): method_decorators = [jwt_required()] @inject def get(self, uid, service: FlowTemplateAppService = Provide[ Containers.flow_template_container.flow_template_app_service]): dto = service.get(uid) return return_response(True, FlowTemplateResponseSchema().dump(dto.to_dict())) @inject def put(self, uid, service: FlowTemplateAppService = Provide[ Containers.flow_template_container.flow_template_app_service]): body = FlowTemplateUpdateRequestSchema().load(request.get_json() or {}) dto = service.update(uid, name=body["name"], description=body.get("description"), bpmn_xml=body["bpmn_xml"]) return return_response(True, FlowTemplateResponseSchema().dump(dto.to_dict())) @inject def delete(self, uid, service: FlowTemplateAppService = Provide[ Containers.flow_template_container.flow_template_app_service]): ok = service.soft_delete(uid) return return_response(True, {"deleted": ok}) class FlowTemplateDuplicateRoute(Resource): method_decorators = [jwt_required()] @inject def post(self, uid, service: FlowTemplateAppService = Provide[ Containers.flow_template_container.flow_template_app_service]): body = FlowTemplateDuplicateRequestSchema().load(request.get_json() or {}) dto = service.duplicate(uid, new_name=body["name"]) return return_response(True, FlowTemplateResponseSchema().dump(dto.to_dict())) ``` - [ ] **Step 3：在 `api/flow_engine/__init__.py:create_module()` 加 register** 在 file:25 的 `bp = Blueprint(...)` 後、return 前加： ```python from api.flow_engine.routes.stage_object_route import StageObjectsRoute from api.flow_engine.routes.flow_template_route import ( FlowTemplatesRoute, FlowTemplateRoute, FlowTemplateDuplicateRoute, ) # Stage Objects（Spec 1） api.add_resource(StageObjectsRoute, '/flow-engine/stage-objects') # Flow Templates（Spec 1） api.add_resource(FlowTemplatesRoute, '/flow-engine/flow-templates') api.add_resource(FlowTemplateRoute, '/flow-engine/flow-templates/') api.add_resource(FlowTemplateDuplicateRoute, '/flow-engine/flow-templates//duplicate') ``` - [ ] **Step 4：寫 route 單元測試（用 Flask test client + DI override，避開 `__wrapped__` fragility）** ```python # test/test_api_flow_template_route.py """ Route layer tests — 用 Flask test_client + DI override，覆蓋 serializer 驗證 + service 呼叫參數正確性。conftest.py 的 client / headers fixtures 已提供 jwt token。 mock pattern 參考既有 test/test_api_module_frame_reference_document_route.py。 """ from unittest.mock import MagicMock import pytest from app.flow_engine.dto.flow_template_dto import FlowTemplateDto from di_containers.containers import Containers @pytest.fixture def sample_dto(): return FlowTemplateDto( uid="u-1", name="t1", description="d", is_builtin=False, bpmn_xml="", created_user="raymond", created_user_name="Raymond", updated_user="raymond", updated_user_name="Raymond", created_at=None, updated_at=None, ) @pytest.fixture def mock_service(sample_dto): """override DI provider — 整個 test function 期間 inject 此 mock""" mock = MagicMock() mock.list.return_value = ([sample_dto], 1) mock.get.return_value = sample_dto mock.create.return_value = sample_dto mock.update.return_value = sample_dto mock.duplicate.return_value = sample_dto mock.soft_delete.return_value = True Containers.flow_template_container.flow_template_app_service.override(mock) yield mock Containers.flow_template_container.flow_template_app_service.reset_override() def test_get_list(client, headers, mock_service): r = client.get("/api/1.0/flow-engine/flow-templates", headers=headers) assert r.status_code == 200 assert r.json["code"] == 1 assert r.json["data"]["total"] == 1 mock_service.list.assert_called_once() def test_post_create_validates_body(client, headers, mock_service): # 缺 bpmn_xml → serializer 應抓住 r = client.post("/api/1.0/flow-engine/flow-templates", json={"name": "t1"}, headers=headers) assert r.json["code"] == 0 # ValidationError → envelope 失敗 mock_service.create.assert_not_called() def test_post_create_happy_path(client, headers, mock_service): r = client.post("/api/1.0/flow-engine/flow-templates", json={"name": "t1", "description": "d", "bpmn_xml": ""}, headers=headers) assert r.status_code == 200 mock_service.create.assert_called_once_with( name="t1", description="d", bpmn_xml="") def test_delete_returns_envelope(client, headers, mock_service): r = client.delete("/api/1.0/flow-engine/flow-templates/u-1", headers=headers) assert r.status_code == 200 assert r.json["data"]["deleted"] is True mock_service.soft_delete.assert_called_once_with("u-1") def test_duplicate_requires_name(client, headers, mock_service): r = client.post("/api/1.0/flow-engine/flow-templates/u-1/duplicate", json={}, headers=headers) assert r.json["code"] == 0 mock_service.duplicate.assert_not_called() def test_duplicate_happy_path(client, headers, mock_service): r = client.post("/api/1.0/flow-engine/flow-templates/u-1/duplicate", json={"name": "copy"}, headers=headers) assert r.status_code == 200 mock_service.duplicate.assert_called_once_with("u-1", new_name="copy") ``` > **DI override pattern**：用 `Containers...override(mock)` + `reset_override()`（dependency-injector 內建）— 比 patching `request` 或穿透 `__wrapped__` 都穩。conftest.py 既有 `client` / `headers` fixtures（依 memory `reference_dev_login.md` 使用 blsit 帳號取 token）已支援此測試風格。 ```python # test/test_api_stage_object_route.py from unittest.mock import MagicMock import pytest from app.flow_engine.dto.stage_object_dto import StageObjectDto from di_containers.containers import Containers @pytest.fixture def mock_service(): mock = MagicMock() mock.list_active.return_value = [ StageObjectDto(uid="u", code="planning", name="規劃", kind="routed", route_pattern="/x", complete_button_label="啟動", complete_handler_key="activate_project", precondition_key=None, default_main_roles=["manager"], is_builtin=True, sort=10), ] Containers.flow_template_container.stage_object_app_service.override(mock) yield mock Containers.flow_template_container.stage_object_app_service.reset_override() def test_get_stage_objects_envelope(client, headers, mock_service): r = client.get("/api/1.0/flow-engine/stage-objects", headers=headers) assert r.status_code == 200 assert r.json["code"] == 1 items = r.json["data"]["items"] assert len(items) == 1 and items[0]["code"] == "planning" mock_service.list_active.assert_called_once() ``` - [ ] **Step 5：跑測試 GREEN** Run: `pytest test/test_api_flow_template_route.py test/test_api_stage_object_route.py -v` Expected: 5 tests PASS - [ ] **Step 6：commit** ```bash git add api/flow_engine/routes/stage_object_route.py \ api/flow_engine/routes/flow_template_route.py \ api/flow_engine/__init__.py \ test/test_api_flow_template_route.py test/test_api_stage_object_route.py git commit -m "feat(flow-template): add 7 API routes + unit tests" ``` --- ## Phase F：DI Wiring + Smoke ### Task F1：FlowTemplateContainer **Files:** - Create: `di_containers/flow_engine/flow_template_containers.py` - Modify: `di_containers/containers.py` - [ ] **Step 1：寫 FlowTemplateContainer** ```python # di_containers/flow_engine/flow_template_containers.py from dependency_injector import containers, providers from app.flow_engine.service.flow_template_app_service import FlowTemplateAppService from app.flow_engine.service.stage_object_app_service import StageObjectAppService from domain.flow_engine.service.flow_template_domain_service import FlowTemplateDomainService from domain.flow_engine.service.stage_object_domain_service import StageObjectDomainService from infra.flow_engine.repository.flow_template_repo_impl import FlowTemplateRepoImpl from infra.flow_engine.repository.stage_object_repo_impl import StageObjectRepoImpl class FlowTemplateContainer(containers.DeclarativeContainer): config = providers.Configuration() auth_container = providers.DependenciesContainer() stage_object_repo = providers.Singleton(StageObjectRepoImpl) flow_template_repo = providers.Singleton(FlowTemplateRepoImpl) stage_object_domain_service = providers.Factory( StageObjectDomainService, stage_object_repo=stage_object_repo, ) flow_template_domain_service = providers.Factory( FlowTemplateDomainService, flow_template_repo=flow_template_repo, ) stage_object_app_service = providers.Factory( StageObjectAppService, stage_object_domain_service=stage_object_domain_service, ) flow_template_app_service = providers.Factory( FlowTemplateAppService, flow_template_domain_service=flow_template_domain_service, user_service=auth_container.user_service, ) ``` - [ ] **Step 2：在 `di_containers/containers.py` 註冊** 加 import： ```python from di_containers.flow_engine.flow_template_containers import FlowTemplateContainer ``` 在 `Containers` class 內適當位置（auth_container 後）加： ```python flow_template_container: FlowTemplateContainer = providers.Container( FlowTemplateContainer, config=config, auth_container=auth_container, ) ``` - [ ] **Step 3：smoke run — 啟動 BE 看有沒有 wiring 錯誤** ```bash # 先停掉舊 process（CLAUDE.md memory 提到 nohup 留 orphan PID 的雷） lsof -i :8000 -t | xargs -r kill -9 set -a; source .env; set +a python main_app.py > /tmp/be-smoke.log 2>&1 & sleep 5 tail -50 /tmp/be-smoke.log ``` Expected：log 顯示 server 啟動於 port 8000、沒有 ImportError / ContainerError / unresolved provider 訊息。若有 wiring 錯誤訊息（如 "Provider not found"），grep 對照修正。 - [ ] **Step 4：smoke API hit — 列 stage_objects（未登入應 401，斷言 status code）** ```bash HTTP_CODE=$(curl -s -o /dev/null -w "%{http_code}" http://localhost:8000/api/1.0/flow-engine/stage-objects) test "$HTTP_CODE" = "401" && echo "PASS: route registered, jwt guarded" || \ { echo "FAIL: expected 401, got $HTTP_CODE"; exit 1; } ``` Expected：印 `PASS`。代表 routing 成功 register、jwt_required 守門正常。 - [ ] **Step 5：commit** ```bash git add di_containers/flow_engine/flow_template_containers.py \ di_containers/containers.py git commit -m "feat(flow-template): wire FlowTemplateContainer into root DI" ``` ### Task F2：整合測試（用真實 DB，跑完整 round trip） **Files:** - Create: `test/test_flow_template_integration.py`（標記 `@pytest.mark.integration` — 需要 dev DB） - [ ] **Step 1：以 blsit 帳號取 token，端到端跑 list / create / get / update / duplicate / delete** ```python # test/test_flow_template_integration.py """Integration tests — 需要 dev DB 開著、seed 已執行。 跑法：pytest test/test_flow_template_integration.py -v -m integration """ import pytest pytestmark = pytest.mark.integration def test_full_round_trip(client, headers): # 1. List builtin r = client.get("/api/1.0/flow-engine/flow-templates", headers=headers) assert r.status_code == 200 initial_total = r.json["data"]["total"] assert initial_total >= 3 # 至少 3 個 builtin # 2. Duplicate builtin-full-audit 為 tenant 自訂 builtin = next(x for x in r.json["data"]["items"] if x["name"] == "builtin-full-audit") r = client.post(f"/api/1.0/flow-engine/flow-templates/{builtin['uid']}/duplicate", json={"name": "my-copy"}, headers=headers) assert r.status_code == 200 new_uid = r.json["data"]["uid"] # 3. Get detail 含 bpmn_xml r = client.get(f"/api/1.0/flow-engine/flow-templates/{new_uid}", headers=headers) assert r.status_code == 200 assert r.json["data"]["bpmn_xml"] is not None assert r.json["data"]["is_builtin"] is False # 4. Update name r = client.put(f"/api/1.0/flow-engine/flow-templates/{new_uid}", json={"name": "my-copy-v2", "description": "updated", "bpmn_xml": r.json["data"]["bpmn_xml"]}, headers=headers) assert r.status_code == 200 # 5. 改 builtin → 403 r = client.put(f"/api/1.0/flow-engine/flow-templates/{builtin['uid']}", json={"name": "x", "description": "", "bpmn_xml": ""}, headers=headers) assert r.status_code in (200, 403) # status_code policy 依 jedi handler if r.status_code == 200: assert r.json["code"] == 0 # 失敗 envelope # 6. 刪自訂 r = client.delete(f"/api/1.0/flow-engine/flow-templates/{new_uid}", headers=headers) assert r.status_code == 200 # 7. 刪 builtin → 403 / code=0 r = client.delete(f"/api/1.0/flow-engine/flow-templates/{builtin['uid']}", headers=headers) assert r.json["code"] == 0 def test_stage_objects_list(client, headers): r = client.get("/api/1.0/flow-engine/stage-objects", headers=headers) assert r.status_code == 200 items = r.json["data"]["items"] codes = {i["code"] for i in items} assert {"planning", "task_execution", "audit", "poam"}.issubset(codes) ``` > **conftest 假設**：BE 既有 `test/conftest.py` 有 `client` / `headers` fixtures，用 `blsit / Billows@123!` 取 token（memory 提及）。先 `Read test/conftest.py:1-50` 對齊真實 fixture 名稱。 - [ ] **Step 2：跑整合測試** Run: `pytest test/test_flow_template_integration.py -v -m integration` Expected：2 tests PASS（前提 dev DB 跑過 Phase A migration） - [ ] **Step 3：commit** ```bash git add test/test_flow_template_integration.py git commit -m "test(flow-template): integration round-trip + builtin guards" ``` --- ## Phase G：RBAC Migration ### Task G1：權限 SQL + 驗證 **Files:** - Create: `scripts/sql/2026-05-12-flow-template-permissions.sql` - [ ] **Step 1：寫 RBAC migration（仿照 2026-04-27 audit-and-cloud-integration 的格式）** ```sql -- Date: 2026-05-12 -- Spec 1：流程管理 — RBAC（capabilities + ui_routes + role_capabilities） -- 執行身份：cmmgr BEGIN; SET LOCAL app.is_super_admin = 't'; -- 1. flow_template 模組能力點 (2026-05-12) INSERT INTO public.capabilities (name, resource_type, action, description) VALUES ('flow_template.read', 'flow_template', 'read', '檢視流程範本'), ('flow_template.create', 'flow_template', 'create', '建立流程範本'), ('flow_template.update', 'flow_template', 'update', '編輯流程範本'), ('flow_template.delete', 'flow_template', 'delete', '刪除流程範本') ON CONFLICT (name) DO NOTHING; -- 2. flow-template-manage 路由 (2026-05-12) INSERT INTO public.ui_routes (uid, pid, name, url, icon, enable, description, sort) SELECT gen_random_uuid(), 0, 'flow-template-manage', '/flow/template-manage', 'pi-sitemap', 1, '流程範本管理', 190 WHERE NOT EXISTS ( SELECT 1 FROM public.ui_routes WHERE name = 'flow-template-manage' ); -- 3. route_capabilities (read=ALL, 寫=ANY) (2026-05-12) WITH r AS (SELECT id FROM public.ui_routes WHERE name = 'flow-template-manage'), caps AS ( SELECT id, name FROM public.capabilities WHERE resource_type = 'flow_template' ) INSERT INTO public.route_capabilities (route_id, capability_id, requirement) SELECT r.id, caps.id, CASE WHEN caps.name LIKE '%.read' THEN 'ALL' ELSE 'ANY' END FROM r, caps ON CONFLICT (route_id, capability_id) DO NOTHING; -- 4. Administrator 預設擁有全部 flow_template 能力 (2026-05-12) INSERT INTO public.role_capabilities (role_id, capability_id) SELECT r.id, c.id FROM public.roles r JOIN public.capabilities c ON c.resource_type = 'flow_template' WHERE r.name = 'Administrator' AND r.tenant_id IS NULL ON CONFLICT (role_id, capability_id) DO NOTHING; COMMIT; -- 驗證： -- SELECT * FROM public.capabilities WHERE resource_type = 'flow_template'; -- SELECT * FROM public.ui_routes WHERE name = 'flow-template-manage'; -- SELECT rc.* FROM public.route_capabilities rc -- JOIN public.capabilities c ON c.id = rc.capability_id -- WHERE c.resource_type = 'flow_template'; ``` - [ ] **Step 2：跑 migration** ```bash PGPASSWORD='jedi@123!' psql -h 192.168.50.188 -p 25432 -U cmmgr -d guidant_ai_stg \ -f scripts/sql/2026-05-12-flow-template-permissions.sql ``` Expected: BEGIN / INSERT / COMMIT，4 capabilities + 1 ui_route + 4 route_capabilities + 4 role_capabilities。 - [ ] **Step 3：驗證 web-menu API 看得到新項** ```bash # 用 blsadmin 取 token（Administrator 角色） TOKEN=$(curl -s -X POST http://localhost:8000/api/1.0/auth/login \ -H "Content-Type: application/json" \ -d '{"login_name": "blsadmin", "password": "Billows@123!"}' | jq -r .data.token) curl -s http://localhost:8000/api/1.0/user/web-menu \ -H "Authorization: Bearer $TOKEN" | jq '.data[] | select(.name == "flow-template-manage")' ``` Expected：印出該 node。 - [ ] **Step 4：commit** ```bash git add scripts/sql/2026-05-12-flow-template-permissions.sql git commit -m "feat(flow-template): add RBAC migration (capabilities + ui_route + bindings)" ``` --- ## Phase H：文件交付 ### Task H1：Changelog **Files:** - Create: `docs/changelog/2026-05-12-feat-flow-template-management.md` - [ ] **Step 1：寫 changelog（依 CLAUDE.md 格式）** ```markdown --- type: feat modules: [flow_engine, rbac] commit: TBD --- # feat(flow-template): Spec 1 流程管理 — 範本 CRUD + BPMN 編輯器後端 ## 需求說明 實作 `docs/features/FR-026-2605-project-flow-engine-integrate/01-flow-template-management/design.md` Spec 1： - 4 個 seed 階段物件（planning / task_execution / audit / poam） - 3 套 builtin 流程範本（full-audit / internal-check / self-assessment） - 流程範本 CRUD + duplicate API（tenant 隔離，builtin 唯讀） - RBAC 整合（左側選單「流程範本管理」+ 4 個 capability） ## 變更範圍 ### Schema - 新表：`compliance.stage_objects`、`compliance.flow_templates`（含 RLS policy） - 新 migration：`scripts/sql/2026-05-12-flow-template-management-schema.sql`、`scripts/sql/2026-05-12-flow-template-permissions.sql` ### 後端 - DDD 4 層完整檔案清單見 `docs/features/FR-026-2605-project-flow-engine-integrate/01-flow-template-management/implementation-plan.md` - 新 DI：`FlowTemplateContainer` - 6 個新 error code（GRC_404027 / 404028 / 400040 / 403040 / 403041 / 409030） ### API - 7 個新 endpoint（GET stage-objects / 6 個 flow-template CRUD + duplicate） ## API 變更 新增 endpoint，無破壞性改動。Endpoint 規格見 `docs/api/flow-engine/api-spec.md`。 ## 測試結果 - `pytest test/test_flow_template_*.py test/test_api_flow_template_route.py test/test_api_stage_object_route.py` → PASS - 整合測試（需 dev DB）：`pytest -m integration test/test_flow_template_integration.py` → PASS ## 後續工作 - 3 個 builtin 範本的 BPMN XML 內容由 FE BPMN editor 產出後 UPDATE 進 DB（Phase A 註記） - Spec 2：階段抽象整合 — Banner + 推進按鈕搬家 - FE 實作：左側選單 / 範本列表 / 編輯器（含 Gateway 編輯）/ 預覽 Dialog ``` - [ ] **Step 2：commit** ```bash git add docs/changelog/2026-05-12-feat-flow-template-management.md git commit -m "docs(flow-template): add changelog" ``` ### Task H2：API spec / design 更新 **Files:** - Modify: `docs/api/flow-engine/api-spec.md`（補 7 個 endpoint SA） - Modify: `docs/api/flow-engine/design.md`（補 stage_object / flow_template 模組 SD） - [ ] **Step 1：在 `docs/api/flow-engine/api-spec.md` 加新 endpoint 段**（依該檔現有格式對齊：每個 endpoint 含 Path / Method / Request Schema / Response Schema / Error Code） - [ ] **Step 2：在 `docs/api/flow-engine/design.md` 加新模組設計段**（schema / DI / 商業邏輯：builtin readonly guard / duplicate semantic / BPMN 驗證） - [ ] **Step 3：commit** ```bash git add docs/api/flow-engine/api-spec.md docs/api/flow-engine/design.md git commit -m "docs(flow-template): update api-spec.md + design.md" ``` > **i18n 註記**：6 個新 `GrcErrorCode` 的中文訊息直接寫死在 `grc_error_code.py` 的 tuple 內（既有 GrcErrorCode 全部如此處理），**不走 `_()` / babel extract**。FE 用 error code 字串對應自家 `errorCode.json`，**不需要跑 `pybabel extract/update/compile`**。 ### Task H3：FE handoff spec **Files:** - Create: `docs/features/FR-026-2605-project-flow-engine-integrate/01-flow-template-management/frontend-spec.md` - [ ] **Step 1：寫 FE handoff 文件**，含： - 7 個 API endpoint 規格 + sample request/response - **完整 Gateway 編輯需求**（user 鎖定的 decision C）： - BPMN editor palette 含 ExclusiveGateway - UserTask 屬性面板：階段物件下拉（call `GET /flow-engine/stage-objects`）+ 主要角色單選 - Gateway 條件編輯 UI（如 `${context.has_findings == true}`） - 左側選單 i18n key：`flow-template-manage` → 「流程範本管理」/ 「Flow Template Management」（FE `menu.json` 補） - 預覽 Dialog：唯讀 BPMN renderer + UserTask 列表（顯示綁的階段名稱） - 「複製為自訂」按鈕：呼叫 `POST /flow-engine/flow-templates//duplicate`，body `{name}` - error code i18n key 對照表（依 GrcErrorCode 新增的 6 個） - [ ] **Step 2：commit** ```bash git add docs/features/FR-026-2605-project-flow-engine-integrate/01-flow-template-management/frontend-spec.md git commit -m "docs(flow-template): add FE handoff spec (full Gateway editing scope)" ``` --- ## 完成條件（Definition of Done） - [ ] 所有 BE pytest unit + integration 全 GREEN - [ ] Smoke：BE 啟動無 wiring 錯誤；`curl /flow-engine/stage-objects` 401（jwt 守門正常）；`/user/web-menu`（blsadmin）回傳含 `flow-template-manage` - [ ] DB：dev DB 含 4 stage_objects + 3 builtin flow_templates（含實際 BPMN XML，非 placeholder） - [ ] Migration SQL 可 idempotent 重跑（系統層 seed 用 `ON CONFLICT DO NOTHING`；builtin flow_templates 用 `WHERE NOT EXISTS` 防 NULL key 不等價問題） - [ ] 文件：changelog ✓ / api-spec ✓ / design ✓ / frontend-spec ✓ 全在 - [ ] git working tree 乾淨；branch `feature/project-flow-engine-integrate` 領先 main N 個 commit（user 自行決定 push 時機） --- ## 不在此 plan 範圍（後續 spec 處理） - 處置未追蹤檔 `domain/flow_engine/repository/i_ext_assessment_plan_repo.py` / `infra/flow_engine/repository/ext_assessment_plan_repo_impl.py` → **Spec 2** 動工時決定刪除或重寫 - `docs/fail_feature_backup/` → 純歷史，不引用 - AP 套用範本流程 → **Spec 3** - Stage 推進 runtime / 既有 5 個 view 改造 → **Spec 2** - 階段物件 CRUD UI → 後續 spec（v1 只 seed，不開介面） - FE 實作（BPMN 編輯器 / 列表 / 預覽 Dialog） → 另一 session 在 `compliance-manager-fe` - E2E feature / step → 另一 session 在 `compliance-manager-test` --- ## 風險與緩解 | 風險 | 緩解 | |---|---| | 3 個 builtin BPMN XML 內容 | **已 resolved (2026-05-12)**：3 份 XML 已 handcraft 並用 `jedi_flow_engine.common.utils.bpmn_uilts.BpmnUtils` 驗證 parse 通過（含 full-audit 的 ExclusiveGateway loop back：audit 有 2 incoming）。檔案位於 `scripts/sql/seeds/bpmn/builtin-{full-audit,internal-check,self-assessment}.bpmn`。Phase A1 SQL 用 PG `$xml$...$xml$` dollar-quoting 內嵌。Spec 2 跑 stage 推進時若 runtime 需求不合，再修正 seed XML | | `BpmnUtils` API 與 plan 假設不一致 | Phase D `_validate_bpmn` 內標 `⚠️ Pre-flight verify`；開工 grep 對齊真實簽名後再寫死。若沒提供，退化為 `xmltodict.parse(xml)` + 字串檢查必要 BPMN 元素 | | RLS policy 第一次部署測不出 | Phase A Step 4 強制用 cm_app + 假 tenant 跑一遍 SELECT；尚未上 prod 前驗證 builtin 公開讀、tenant 不可見其他 tenant | | FE Gateway 編輯 +1-2 週成本（user decision C）| 不影響 BE plan；FE handoff spec 內標明 scope，FE session 評估 | | `user_service.get_users_by_login_names` method 名不一致 | Phase D `_build_nickname_map` 內標 `⚠️ Pre-flight verify`；開工對齊 | --- ## Execution Handoff Plan 完成。兩個執行選項： **1. Subagent-Driven（推薦）** — 我每個 task 派一個 fresh subagent，task 之間 review checkpoint，迭代快 **2. Inline Execution** — 此 session 直接跑，phase 之間 checkpoint 請選一個方式繼續，或先讓 raymond review 整份 plan 再決定。