diff --git a/backend/access_control.py b/backend/access_control.py index 5f99378..2de9c0e 100644 --- a/backend/access_control.py +++ b/backend/access_control.py @@ -27,6 +27,8 @@ GLOBAL_APPROVAL_DECIDE = "approval.global.decide" ERP_EXCHANGE_EXPORT = "erp.exchange.export" ERP_EXCHANGE_RECONCILE = "erp.exchange.reconcile" +DECISION_EVIDENCE_READ = "decision.evidence.read" +DECISION_RECORD_WRITE = "decision.record.write" L1_MONITOR = "l1_monitor" L2_DECISION = "l2_decision" @@ -46,13 +48,15 @@ GLOBAL_APPROVAL_DECIDE: L3_GOVERNED_ACTION, ERP_EXCHANGE_EXPORT: L3_GOVERNED_ACTION, ERP_EXCHANGE_RECONCILE: L3_GOVERNED_ACTION, + DECISION_EVIDENCE_READ: L1_MONITOR, + DECISION_RECORD_WRITE: L2_DECISION, } _ALL_CAPABILITIES = frozenset(_CAPABILITY_ENTITLEMENT) _ROLE_CAPABILITIES = { - "risk_viewer": frozenset({RISK_OVERVIEW_READ, RISK_ALERT_ACK}), + "risk_viewer": frozenset({RISK_OVERVIEW_READ, RISK_ALERT_ACK, DECISION_EVIDENCE_READ}), "supply_planner": frozenset( { RISK_OVERVIEW_READ, @@ -61,6 +65,8 @@ RISK_WHAT_IF_RUN, RISK_WORKSPACE_WRITE, ERP_EXCHANGE_PROPOSE, + DECISION_EVIDENCE_READ, + DECISION_RECORD_WRITE, } ), "procurement_approver": frozenset( @@ -72,6 +78,7 @@ APPROVAL_DECIDE, ERP_EXCHANGE_EXPORT, ERP_EXCHANGE_RECONCILE, + DECISION_EVIDENCE_READ, } ), # Preserve the existing demo accounts while routing the new accounts @@ -88,6 +95,8 @@ APPROVAL_QUEUE_READ, ERP_EXCHANGE_EXPORT, ERP_EXCHANGE_RECONCILE, + DECISION_EVIDENCE_READ, + DECISION_RECORD_WRITE, } ), "admin": _ALL_CAPABILITIES, diff --git a/backend/database.py b/backend/database.py index 8738a5d..80ae5d4 100644 --- a/backend/database.py +++ b/backend/database.py @@ -280,6 +280,52 @@ def init_db(): version INTEGER NOT NULL DEFAULT 0 )''') + # AI 決策可驗證性：建議本身、當下證據快照與人員回饋分開保存。 + # snapshot_json / snapshot_digest 一經建立不再更新，避免事後資料變動 + # 讓人誤以為 AI 當時是依據新資料做出判斷。 + c.execute('''CREATE TABLE IF NOT EXISTS decision_records ( + decision_id TEXT PRIMARY KEY, + organization_id TEXT NOT NULL, + created_by TEXT NOT NULL, + decision_type TEXT NOT NULL, + model_name TEXT NOT NULL, + model_output_json TEXT NOT NULL, + output_schema_version INTEGER NOT NULL, + status TEXT NOT NULL DEFAULT 'proposed', + decision_reason TEXT NOT NULL DEFAULT '', + decided_by TEXT, + decided_at TEXT, + created_at TEXT NOT NULL + )''') + c.execute('''CREATE TABLE IF NOT EXISTS decision_evidence_snapshots ( + decision_id TEXT PRIMARY KEY, + snapshot_json TEXT NOT NULL, + snapshot_digest TEXT NOT NULL, + data_as_of TEXT NOT NULL, + created_at TEXT NOT NULL, + FOREIGN KEY(decision_id) REFERENCES decision_records(decision_id) + )''') + c.execute('''CREATE TABLE IF NOT EXISTS decision_feedback ( + feedback_id INTEGER PRIMARY KEY AUTOINCREMENT, + decision_id TEXT NOT NULL, + outcome TEXT NOT NULL, + note TEXT NOT NULL DEFAULT '', + action_taken TEXT NOT NULL DEFAULT '', + outcome_evidence TEXT NOT NULL DEFAULT '', + recorded_by TEXT NOT NULL, + recorded_at TEXT NOT NULL, + FOREIGN KEY(decision_id) REFERENCES decision_records(decision_id) + )''') + c.execute('''CREATE INDEX IF NOT EXISTS ix_decision_records_created + ON decision_records(organization_id, created_at DESC)''') + + # 讓已建立的舊資料庫也可逐步加入較完整的回饋欄位。 + for column_name in ("action_taken", "outcome_evidence"): + try: + c.execute(f"ALTER TABLE decision_feedback ADD COLUMN {column_name} TEXT NOT NULL DEFAULT ''") + except sqlite3.OperationalError: + pass + c.execute('''CREATE TABLE IF NOT EXISTS effect_receipts ( receipt_id INTEGER PRIMARY KEY AUTOINCREMENT, operation_id TEXT NOT NULL UNIQUE, diff --git a/backend/decision_evidence.py b/backend/decision_evidence.py new file mode 100644 index 0000000..22b2854 --- /dev/null +++ b/backend/decision_evidence.py @@ -0,0 +1,334 @@ +"""可驗證 AI 決策紀錄。 + +AI 回覆是「建議」而非執行指令。這個模組保存不可變的資料證據快照， +驗證模型輸出結構，並記錄人員最後採納、拒絕與觀察到的結果。 +""" + +from __future__ import annotations + +from datetime import datetime, timezone +import hashlib +import json +import sqlite3 +import uuid +from typing import Any, Mapping + +from backend import database +from backend.access_control import ( + DECISION_EVIDENCE_READ, + DECISION_RECORD_WRITE, + require_capability, +) + + +OUTPUT_SCHEMA_VERSION = 1 +_ALLOWED_RECOMMENDATIONS = { + "monitor", + "request_review", + "propose_alternative_purchase", +} +_ALLOWED_DECISIONS = {"adopted", "rejected", "needs_more_evidence"} +_ALLOWED_FEEDBACK = {"effective", "ineffective", "inconclusive"} + + +def _now() -> str: + return datetime.now(timezone.utc).replace(microsecond=0).isoformat() + + +def _canonical_json(value: Mapping[str, Any]) -> str: + return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + + +def _digest(value: Mapping[str, Any]) -> str: + return hashlib.sha256(_canonical_json(value).encode("utf-8")).hexdigest() + + +def build_what_if_decision_draft( + *, + question: str, + answer: str, + model_name: str, +) -> dict[str, Any]: + """Turn a completed What-if response into a reviewable, *unpersisted* draft. + + The What-if prompt currently returns prose, not a reliable numeric risk score or + a specific affected SKU. This adapter deliberately uses a neutral provisional + score and a ``request_review`` recommendation. A human must verify those + fields and explicitly submit the draft before it becomes a Decision Record. + """ + question = str(question or "").strip() + answer = str(answer or "").strip() + model_name = str(model_name or "").strip() + if not question or not answer or not model_name: + raise ValueError("What-if 草稿需要情境問題、AI 回覆與模型名稱。") + if answer.startswith("模擬分析暫時無法產生："): + raise ValueError("What-if 分析失敗，不能建立決策草稿。") + + captured_at = _now() + evidence_id = f"what-if:{hashlib.sha256((question + answer).encode('utf-8')).hexdigest()[:16]}" + return { + "decision_type": "supply_chain_what_if_response", + "model_name": model_name, + "ai_output": { + "recommendation": "request_review", + "reasoning": answer, + "risk_level": "medium", + "evidence_ids": [evidence_id], + "limitations": ( + "What-if 回覆為情境推估；風險分數、受影響項目與最終處置" + "必須由人員依當下 ERP 資料覆核。" + ), + }, + "evidence_snapshot": { + "risk_score": 50, + "data_as_of": captured_at, + "sources": [ + { + "name": "What-if 情境分析（ERP 供應商、採購單與庫存快照）", + "as_of": captured_at, + } + ], + "affected_entity": question, + }, + } + + +def build_heatmap_alert_draft( + *, + region_name: str, + risk_score: float, + ai_summary: str | None = None, + data_as_of: str | None = None, +) -> dict[str, Any]: + """Create an unpersisted alert draft from a high-risk supply-map node.""" + region_name = str(region_name or "").strip() + if not region_name: + raise ValueError("高風險預警需要供應商據點名稱。") + if isinstance(risk_score, bool) or not isinstance(risk_score, (int, float)) or not 0 <= risk_score <= 100: + raise ValueError("高風險預警的分數必須是 0 到 100。") + score = float(risk_score) + if score < 70: + raise ValueError("只有風險分數達 70 的供應據點可建立高風險預警。") + captured_at = str(data_as_of or _now()).strip() + recommendation = "propose_alternative_purchase" if score >= 85 else "request_review" + reasoning = str(ai_summary or "").strip() or ( + f"供應鏈風險地圖顯示「{region_name}」的影響程度為 {score:.0f}%。" + ) + return { + "decision_type": "supply_chain_heatmap_alert", + "model_name": "supply-chain-risk-map", + "ai_output": { + "recommendation": recommendation, + "reasoning": reasoning, + "risk_level": "high", + "evidence_ids": [f"risk-map:{hashlib.sha256(region_name.encode('utf-8')).hexdigest()[:16]}"], + "limitations": "此預警依地區／供應據點風險彙整，仍須人工確認特定供應商、採購單與庫存影響。", + }, + "evidence_snapshot": { + "risk_score": score, + "data_as_of": captured_at, + "sources": [{"name": "供應鏈風險地圖", "as_of": captured_at}], + "affected_entity": region_name, + }, + } + + +def validate_ai_output(output: Mapping[str, Any]) -> dict[str, Any]: + """Accept only the small, reviewable AI recommendation contract.""" + if not isinstance(output, Mapping): + raise ValueError("AI 輸出必須是 JSON 物件。") + recommendation = str(output.get("recommendation", "")).strip() + reasoning = str(output.get("reasoning", "")).strip() + limitations = str(output.get("limitations", "")).strip() + evidence_ids = output.get("evidence_ids", []) + risk_level = str(output.get("risk_level", "")).strip().lower() + if recommendation not in _ALLOWED_RECOMMENDATIONS: + raise ValueError("AI recommendation 不在允許的決策類型內。") + if not reasoning or not limitations: + raise ValueError("AI 輸出必須包含 reasoning 與 limitations。") + if risk_level not in {"low", "medium", "high"}: + raise ValueError("AI 輸出 risk_level 必須是 low、medium 或 high。") + if not isinstance(evidence_ids, list) or not all( + isinstance(item, str) and item.strip() for item in evidence_ids + ): + raise ValueError("AI 輸出 evidence_ids 必須是非空字串陣列。") + return { + "recommendation": recommendation, + "reasoning": reasoning, + "risk_level": risk_level, + "evidence_ids": list(evidence_ids), + "limitations": limitations, + } + + +def _validate_snapshot(snapshot: Mapping[str, Any]) -> dict[str, Any]: + if not isinstance(snapshot, Mapping): + raise ValueError("evidence snapshot 必須是 JSON 物件。") + score = snapshot.get("risk_score") + if isinstance(score, bool) or not isinstance(score, (int, float)) or not 0 <= score <= 100: + raise ValueError("risk_score 必須是 0 到 100 的數字。") + data_as_of = str(snapshot.get("data_as_of", "")).strip() + sources = snapshot.get("sources") + if not data_as_of: + raise ValueError("evidence snapshot 必須包含 data_as_of。") + if not isinstance(sources, list) or not sources: + raise ValueError("evidence snapshot 必須至少包含一個資料來源。") + normalized_sources = [] + for source in sources: + if not isinstance(source, Mapping): + raise ValueError("每個資料來源必須是 JSON 物件。") + name = str(source.get("name", "")).strip() + as_of = str(source.get("as_of", "")).strip() + if not name or not as_of: + raise ValueError("資料來源必須包含 name 與 as_of。") + normalized_sources.append({"name": name, "as_of": as_of}) + return { + "risk_score": float(score), + "data_as_of": data_as_of, + "sources": normalized_sources, + "affected_entity": str(snapshot.get("affected_entity", "")).strip(), + } + + +def create_decision_record( + *, + actor: str, + decision_type: str, + model_name: str, + ai_output: Mapping[str, Any], + evidence_snapshot: Mapping[str, Any], + decision_id: str | None = None, +) -> dict[str, Any]: + """Store validated AI output and immutable evidence from the same moment.""" + principal = require_capability(actor, DECISION_RECORD_WRITE) + structured_output = validate_ai_output(ai_output) + snapshot = _validate_snapshot(evidence_snapshot) + decision_type = str(decision_type or "").strip() + model_name = str(model_name or "").strip() + if not decision_type or not model_name: + raise ValueError("decision_type 與 model_name 不可空白。") + decision_id = str(decision_id or f"DEC-{uuid.uuid4().hex[:12]}").strip() + now = _now() + with database.transaction(immediate=True) as conn: + conn.execute( + """INSERT INTO decision_records ( + decision_id, organization_id, created_by, decision_type, model_name, + model_output_json, output_schema_version, status, created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, 'proposed', ?)""", + ( + decision_id, + principal.organization_id, + principal.username, + decision_type, + model_name, + _canonical_json(structured_output), + OUTPUT_SCHEMA_VERSION, + now, + ), + ) + conn.execute( + """INSERT INTO decision_evidence_snapshots ( + decision_id, snapshot_json, snapshot_digest, data_as_of, created_at + ) VALUES (?, ?, ?, ?, ?)""", + (decision_id, _canonical_json(snapshot), _digest(snapshot), snapshot["data_as_of"], now), + ) + return get_decision_record(actor=actor, decision_id=decision_id) + + +def get_decision_record(*, actor: str, decision_id: str) -> dict[str, Any]: + principal = require_capability(actor, DECISION_EVIDENCE_READ) + with sqlite3.connect(database.DB_FILE) as conn: + row = conn.execute( + """SELECT r.decision_id, r.organization_id, r.created_by, r.decision_type, + r.model_name, r.model_output_json, r.output_schema_version, r.status, + r.decision_reason, r.decided_by, r.decided_at, r.created_at, + s.snapshot_json, s.snapshot_digest, s.data_as_of + FROM decision_records r JOIN decision_evidence_snapshots s + ON s.decision_id = r.decision_id + WHERE r.decision_id = ?""", + (decision_id,), + ).fetchone() + if row is None or row[1] != principal.organization_id: + raise ValueError("找不到決策紀錄。") + feedback = conn.execute( + """SELECT outcome, note, action_taken, outcome_evidence, recorded_by, recorded_at + FROM decision_feedback WHERE decision_id = ? ORDER BY feedback_id""", + (decision_id,), + ).fetchall() + return { + "decision_id": row[0], "organization_id": row[1], "created_by": row[2], + "decision_type": row[3], "model_name": row[4], + "ai_output": json.loads(row[5]), "output_schema_version": row[6], + "status": row[7], "decision_reason": row[8], "decided_by": row[9], + "decided_at": row[10], "created_at": row[11], + "evidence_snapshot": json.loads(row[12]), "snapshot_digest": row[13], + "data_as_of": row[14], + "feedback": [ + { + "outcome": item[0], "note": item[1], "action_taken": item[2], + "outcome_evidence": item[3], "recorded_by": item[4], "recorded_at": item[5], + } + for item in feedback + ], + } + + +def list_decision_records(*, actor: str, limit: int = 100) -> list[dict[str, Any]]: + principal = require_capability(actor, DECISION_EVIDENCE_READ) + with sqlite3.connect(database.DB_FILE) as conn: + rows = conn.execute( + """SELECT r.decision_id FROM decision_records r + WHERE r.organization_id = ? ORDER BY r.created_at DESC LIMIT ?""", + (principal.organization_id, max(1, min(int(limit), 200))), + ).fetchall() + return [get_decision_record(actor=actor, decision_id=row[0]) for row in rows] + + +def decide_decision_record(*, actor: str, decision_id: str, outcome: str, reason: str = "") -> dict[str, Any]: + principal = require_capability(actor, DECISION_RECORD_WRITE) + outcome = str(outcome or "").strip() + if outcome not in _ALLOWED_DECISIONS: + raise ValueError("決定結果不合法。") + if outcome in {"rejected", "needs_more_evidence"} and not str(reason).strip(): + raise ValueError("拒絕或要求更多證據時必須填寫原因。") + with database.transaction(immediate=True) as conn: + record = conn.execute("SELECT status, organization_id FROM decision_records WHERE decision_id = ?", (decision_id,)).fetchone() + if record is None or record[1] != principal.organization_id: + raise ValueError("找不到可決定的決策紀錄。") + if record[0] != "proposed": + raise ValueError("此決策已完成處理，不能重複決定。") + conn.execute( + """UPDATE decision_records SET status=?, decision_reason=?, decided_by=?, decided_at=? + WHERE decision_id=?""", + (outcome, str(reason).strip(), principal.username, _now(), decision_id), + ) + return get_decision_record(actor=actor, decision_id=decision_id) + + +def add_outcome_feedback( + *, actor: str, decision_id: str, outcome: str, action_taken: str, outcome_evidence: str, + note: str = "", +) -> dict[str, Any]: + principal = require_capability(actor, DECISION_RECORD_WRITE) + outcome = str(outcome or "").strip() + if outcome not in _ALLOWED_FEEDBACK: + raise ValueError("回饋結果不合法。") + action_taken = str(action_taken or "").strip() + outcome_evidence = str(outcome_evidence or "").strip() + if not action_taken: + raise ValueError("請填寫實際採取的動作。") + if not outcome_evidence: + raise ValueError("請填寫結果依據，例如交期、庫存或採購紀錄。") + with database.transaction(immediate=True) as conn: + record = conn.execute("SELECT status, organization_id FROM decision_records WHERE decision_id = ?", (decision_id,)).fetchone() + if record is None or record[1] != principal.organization_id: + raise ValueError("找不到決策紀錄。") + if record[0] != "adopted": + raise ValueError("只有已採納的建議可新增結果回饋。") + conn.execute( + """INSERT INTO decision_feedback + (decision_id, outcome, note, action_taken, outcome_evidence, recorded_by, recorded_at) + VALUES (?, ?, ?, ?, ?, ?, ?)""", + (decision_id, outcome, str(note).strip(), action_taken, outcome_evidence, principal.username, _now()), + ) + return get_decision_record(actor=actor, decision_id=decision_id) diff --git a/backend/prompts.py b/backend/prompts.py index d503904..49bbb72 100644 --- a/backend/prompts.py +++ b/backend/prompts.py @@ -73,6 +73,19 @@ WHAT_IF_SYSTEM_PROMPT = """你是一個供應鏈風險分析師。請根據以下 ERP 資料（供應商名單與地區、未結案採購單、庫存與安全庫存設定），回答使用者的「如果…會怎樣」情境問題。 重點：指出哪些訂單/物料會斷貨、影響比例，並給出具體建議（例如：將 X 物料的安全庫存從 30 天提高到 60 天）。回覆用繁體中文、條列清晰。""" +WHAT_IF_DECISION_JSON_INSTRUCTIONS = """ +你正在提出「供人員審核」的 AI 建議，不可聲稱已執行採購、庫存或其他 ERP 異動。 +只輸出 JSON 物件，不要 Markdown 或其他文字，格式必須是： +{ + "recommendation": "monitor" | "request_review" | "propose_alternative_purchase", + "risk_level": "low" | "medium" | "high", + "risk_score": 0 到 100 的整數, + "affected_entity": "受影響的供應商、採購單或物料；不確定時清楚說明範圍", + "reasoning": "以繁體中文說明判斷依據與可能影響", + "limitations": "以繁體中文說明資料不足、不確定性或需要人工確認的事項" +} +""" + WHAT_IF_USER_PROMPT = """【供應商名單與地區】 {supplier_text} diff --git a/backend/supply_chain_risk.py b/backend/supply_chain_risk.py index 26ffd91..90fdf96 100644 --- a/backend/supply_chain_risk.py +++ b/backend/supply_chain_risk.py @@ -4,10 +4,11 @@ 職責：供應鏈地圖資料、風險事件與交期、風險係數管理、風險報告產出 """ +import json import sqlite3 import re import pandas as pd -from datetime import datetime, timedelta +from datetime import datetime, timedelta, timezone from typing import List, Optional, Any from backend.database import DB_FILE, run_query from backend.access_control import ( @@ -21,6 +22,7 @@ BATCH_INFER_WITH_PRECEDENTS_PROMPT, PO_ALTERNATIVE_SUGGESTION_PROMPT, WHAT_IF_SYSTEM_PROMPT, + WHAT_IF_DECISION_JSON_INSTRUCTIONS, WHAT_IF_USER_PROMPT ) @@ -838,6 +840,67 @@ def get_impacted_pos(region_key=None, country=None, supplier_id=None): return out +def get_what_if_erp_evidence(user_question: str, *, actor=None) -> dict: + """Return the ERP rows shown beside a What-if answer. + + This intentionally does not ask an LLM to identify a purchase order. A + location named in the question narrows the rows deterministically; when + there is no ERP location match, the UI explicitly labels the open orders as + items that still need human confirmation. + """ + require_capability(actor, RISK_WHAT_IF_RUN) + question = str(user_question or "").casefold() + with sqlite3.connect(DB_FILE) as conn: + rows = conn.execute( + """SELECT p.po_id, s.name, s.country, s.region, p.estimated_delay_days, + p.alternative_suggestion + FROM purchase_orders p JOIN suppliers s ON s.supplier_id = p.supplier_id + WHERE p.status IS NULL OR p.status NOT IN ('已完成', '已取消') + ORDER BY p.po_id""" + ).fetchall() + item_rows = conn.execute( + """SELECT poi.po_id, i.product_id, i.name, i.stock, i.daily_sales + FROM purchase_order_items poi JOIN inventory i ON i.product_id = poi.product_id""" + ).fetchall() + + materials_by_po: dict[str, list[dict]] = {} + for po_id, product_id, name, stock, daily_sales in item_rows: + stock = float(stock or 0) + daily_sales = float(daily_sales or 0) + days_left = int(stock / daily_sales) if daily_sales > 0 else None + materials_by_po.setdefault(po_id, []).append({ + "product_id": product_id, "product_name": name, + "stock": stock, "daily_sales": daily_sales, "days_left": days_left, + }) + + matched_locations: list[str] = [] + for _, _, country, region, _, _ in rows: + for value in (country, region): + normalized = str(value or "").strip().casefold() + if len(normalized) >= 2 and normalized in question and normalized not in matched_locations: + matched_locations.append(normalized) + + evidence = [] + for po_id, supplier_name, country, region, delay, alternative in rows: + location_values = {str(country or "").strip().casefold(), str(region or "").strip().casefold()} + if matched_locations and not location_values.intersection(matched_locations): + continue + material_rows = materials_by_po.get(po_id, []) + evidence.append({ + "po_id": po_id, + "supplier_name": supplier_name, + "supplier_location": " ".join(part for part in (str(country or "").strip(), str(region or "").strip()) if part), + "materials": material_rows, + "estimated_delay_days": delay if delay is not None else None, + "alternative_suggestion": str(alternative or "").strip() or None, + }) + return { + "matched_locations": matched_locations, + "match_status": "location_matched" if matched_locations else "needs_human_confirmation", + "purchase_orders": evidence, + } + + def update_po_impact(po_id, estimated_delay_days=None, alternative_suggestion=None, *, actor=None): """Planner may change assessment notes, never the underlying transaction.""" require_capability(actor, RISK_WORKSPACE_WRITE) @@ -937,6 +1000,18 @@ def what_if_simulation( ): """依使用者情境問題，結合 ERP 供應商、未結案採購單、庫存安全天數，由 AI 回覆影響與建議。model 為 Gemini 模型 ID。""" require_capability(actor, RISK_WHAT_IF_RUN) + prompt = _what_if_prompt(user_question) + try: + # issue #27：統一 LLM 入口（api_key 參數棄用，.env 驅動） + from backend.llm_client import complete_text + return (complete_text(prompt, system=WHAT_IF_SYSTEM_PROMPT, temperature=0.2, + tag="analysis:whatif") or "").strip() + except Exception as e: + return f"模擬分析暫時無法產生：{e}" + + +def _what_if_prompt(user_question: str) -> str: + """Build one ERP evidence snapshot prompt for either prose or JSON What-if analysis.""" conn = connect_db(DB_FILE) suppliers = __pd_read("SELECT supplier_id, name, country, region FROM suppliers", conn) pos = __pd_read( @@ -953,20 +1028,60 @@ def what_if_simulation( supplier_text = suppliers.to_string(index=False) if suppliers is not None and not suppliers.empty else "無" po_text = pos.to_string(index=False) if pos is not None and not pos.empty else "無進行中採購單" inv_text = inv.to_string(index=False) if inv is not None and not inv.empty else "無庫存資料" - system = WHAT_IF_SYSTEM_PROMPT - prompt = WHAT_IF_USER_PROMPT.format( + return WHAT_IF_USER_PROMPT.format( supplier_text=supplier_text, po_text=po_text, inv_text=inv_text, user_question=user_question ) + + +def what_if_decision_analysis(api_key, user_question, model: str | None = "gemini-2.5-flash", *, actor=None): + """Ask the model for a validated, reviewable What-if decision draft. + + This function creates no database record and never performs an ERP write. + The caller must show the result to a human for confirmation first. + """ + require_capability(actor, RISK_WHAT_IF_RUN) + prompt = _what_if_prompt(user_question) try: - # issue #27：統一 LLM 入口（api_key 參數棄用，.env 驅動） from backend.llm_client import complete_text - return (complete_text(prompt, system=system, temperature=0.2, - tag="analysis:whatif") or "").strip() - except Exception as e: - return f"模擬分析暫時無法產生：{e}" + raw = complete_text( + prompt, + system=WHAT_IF_SYSTEM_PROMPT + "\n\n" + WHAT_IF_DECISION_JSON_INSTRUCTIONS, + temperature=0.2, + json_mode=True, + tag="analysis:whatif-decision", + ) + payload = json.loads(raw or "{}") + from backend.decision_evidence import validate_ai_output + risk_score = payload.get("risk_score") + if isinstance(risk_score, bool) or not isinstance(risk_score, (int, float)) or not 0 <= risk_score <= 100: + raise ValueError("AI 回覆的 risk_score 必須是 0 到 100 的數字。") + affected_entity = str(payload.get("affected_entity", "")).strip() + if not affected_entity: + raise ValueError("AI 回覆缺少 affected_entity。") + ai_output = validate_ai_output({ + "recommendation": payload.get("recommendation"), + "risk_level": payload.get("risk_level"), + "reasoning": payload.get("reasoning"), + "limitations": payload.get("limitations"), + "evidence_ids": [f"what-if:{hash(str(user_question) + str(raw)) & 0xffffffff:08x}"], + }) + captured_at = datetime.now(timezone.utc).replace(microsecond=0).isoformat() + return { + "decision_type": "supply_chain_what_if_response", + "model_name": f"gemini/{model or 'default'}", + "ai_output": ai_output, + "evidence_snapshot": { + "risk_score": float(risk_score), + "data_as_of": captured_at, + "sources": [{"name": "What-if 情境分析（ERP 供應商、採購單與庫存快照）", "as_of": captured_at}], + "affected_entity": affected_entity, + }, + } + except Exception as exc: + raise ValueError(f"模擬分析暫時無法產生結構化 AI 判斷：{exc}") from exc # ── 風險事件與交期 ──────────────────────────────────────────────────── diff --git a/frontend/components/supply_map.py b/frontend/components/supply_map.py index c66a45d..9a67504 100644 --- a/frontend/components/supply_map.py +++ b/frontend/components/supply_map.py @@ -3,6 +3,7 @@ import streamlit as st import pandas as pd import plotly.express as px +from datetime import datetime, timezone from backend.supply_chain_news import get_news_from_db from backend.supply_chain_risk import ( get_risk_heatmap_data, @@ -14,7 +15,8 @@ get_impacted_pos, update_po_impact, get_ai_alternative_suggestions, - what_if_simulation, + get_what_if_erp_evidence, + what_if_decision_analysis, ) @@ -94,6 +96,28 @@ def render_risk_heatmap(key: str = "risk_heatmap", heatmap_rows=None): ) st.plotly_chart(fig, use_container_width=True, key=key) + +def _render_heatmap_data_freshness(heatmap_rows) -> None: + """在既有地圖下說明風險數值的來源與最近更新時間。""" + st.caption("資料來源：供應商據點、已登錄風險事件、地區風險係數與區域採購集中度。") + timestamps = [] + for row in heatmap_rows: + value = row.get("updated_at") + if not value: + continue + try: + timestamps.append(datetime.fromisoformat(str(value).replace("Z", "+00:00")).astimezone(timezone.utc)) + except ValueError: + continue + if not timestamps: + st.caption("資料時間：依目前 ERP 資料即時計算；尚無 AI 摘要更新時間。") + return + newest = max(timestamps) + age = datetime.now(timezone.utc) - newest + st.caption(f"最近 AI 摘要更新：{newest.isoformat(timespec='seconds')}（UTC）") + if age.total_seconds() > 24 * 60 * 60: + st.warning("這份 AI 風險摘要已超過 24 小時未更新；請先更新情報或重新產生摘要，再據此做決策。") + _BADGE_STYLE = "font-size: 0.7rem; padding: 2px 6px; border-radius: 4px; display: inline-block; margin-bottom: 5px;" _CARD_BADGES = { "view": ("#D1FAE5", "#065F46", "✅ 應變執行中"), @@ -304,6 +328,31 @@ def render_supply_chain_map( # ── 即時風險熱圖 (Risk Heatmap) ───────────────────────────────── render_risk_heatmap(key="detail_heatmap", heatmap_rows=heatmap_rows) + _render_heatmap_data_freshness(heatmap_rows) + + # 高風險供應據點可直接送入既有的「決策證據與回饋」流程，但不直接執行採購。 + high_risk_nodes = [row for row in heatmap_rows if float(row.get("risk_pct") or 0) >= 70] + if high_risk_nodes: + with st.expander("🚨 建立高風險供應據點預警", expanded=False): + st.caption("選取風險達 70% 以上的據點，建立待人員覆核的決策草稿。") + node_by_label = { + f"{row['display_name']}（風險 {float(row.get('risk_pct') or 0):.0f}%）": row + for row in high_risk_nodes + } + selected_label = st.selectbox("高風險供應據點", list(node_by_label), key="heatmap_alert_node") + if st.button("🧾 建立高風險預警草稿", key="heatmap_to_decision"): + from backend.decision_evidence import build_heatmap_alert_draft + row = node_by_label[selected_label] + try: + st.session_state["decision_prefill_pending"] = build_heatmap_alert_draft( + region_name=row["display_name"], + risk_score=float(row.get("risk_pct") or 0), + ai_summary=row.get("ai_summary"), + data_as_of=row.get("updated_at"), + ) + st.success("已建立預警草稿；請到「決策證據與回饋」確認後建立正式紀錄。") + except ValueError as exc: + st.error(str(exc)) # AI 摘要（使用最近最新新聞） st.markdown("**AI 摘要**") @@ -371,12 +420,24 @@ def render_supply_chain_map( if heatmap_rows_for_update: st.markdown("##### 🎯 審核並套用 AI 風險建議") - st.caption("套用會保存各據點的風險與延遲天數；空白代表未知，0 代表確認為零。正式事件需另行登錄。") + st.caption("下表依照您的供應商據點清單產生，AI 的建議風險值已對應至每個確切節點。套用會保存各據點的風險與延遲天數；空白代表未知，0 代表確認為零。正式事件需另行登錄。") table_rows = build_heatmap_review_rows( h_updates_raw, st.session_state.get("suggested_events", []), heatmap_rows_for_update ) + # #23：在審核表加上目前風險、變動幅度與 AI 依據，方便覆核（不改 build_heatmap_review_rows 的輸出契約） + _current_by_name = {r.get("display_name", ""): float(r.get("risk_pct") or 0) for r in heatmap_rows_for_update} + _events = st.session_state.get("suggested_events", []) or [] + for _row in table_rows: + _c, _r = split_location(_row["地區"]) + _cur = _current_by_name.get(_row["地區"], 0.0) + _ai = _row.get("預估風險 (%)") + _ev = next((e for e in _events if matches_location(_c, _r, e.get("region"), e.get("country"))), None) + _row["目前風險 (%)"] = _cur + _row["風險變動 (%)"] = (float(_ai) - _cur) if _ai is not None else None + _row["變動依據"] = (_ev or {}).get("description") or "AI 摘要建議變更此供應據點的風險。" + if table_rows: df_upd = pd.DataFrame(table_rows) edited_risk_df = st.data_editor( @@ -384,8 +445,11 @@ def render_supply_chain_map( column_config={ "套用": st.column_config.CheckboxColumn("是否套用", default=True), "地區": st.column_config.TextColumn("熱點名稱", disabled=True), - "預估風險 (%)": st.column_config.NumberColumn("影響 %", min_value=0, max_value=100, step=1, required=True), - "預估延遲 (天)": st.column_config.NumberColumn("延遲天數", min_value=0, max_value=365, step=1) + "目前風險 (%)": st.column_config.NumberColumn("目前風險", disabled=True), + "預估風險 (%)": st.column_config.NumberColumn("AI 建議風險", min_value=0, max_value=100, step=1, required=True), + "風險變動 (%)": st.column_config.NumberColumn("變動", disabled=True), + "預估延遲 (天)": st.column_config.NumberColumn("延遲天數", min_value=0, max_value=365, step=1), + "變動依據": st.column_config.TextColumn("AI 依據", disabled=True), }, hide_index=True, use_container_width=True, @@ -493,13 +557,56 @@ def render_what_if_analysis( key="whatif_question" ) if st.button("執行 What-If 模擬分析", key="whatif_btn"): - with st.spinner("AI 正在依供應商、採購單與庫存資料分析情境…"): - answer = what_if_simulation( - api_key, user_question, model=gemini_model, actor=actor + try: + with st.spinner("AI 正在依供應商、採購單與庫存資料分析情境…"): + draft = what_if_decision_analysis( + api_key, user_question, model=gemini_model, actor=actor + ) + erp_evidence = get_what_if_erp_evidence(user_question, actor=actor) + draft["evidence_snapshot"]["erp_impact_evidence"] = erp_evidence + st.markdown("**AI 判斷與依據**") + clean_answer = draft["ai_output"]["reasoning"] + st.info(clean_answer) + st.caption( + f"AI 建議：{draft['ai_output']['recommendation']}｜" + f"風險等級：{draft['ai_output']['risk_level']}｜" + f"風險分數：{draft['evidence_snapshot']['risk_score']:.0f}/100" ) - st.markdown("**AI 回覆**") - # 隱藏技術後綴 - clean_answer = answer.split("【自動化指令】")[0].strip() - st.info(clean_answer) - st.caption("範例回覆：「這將影響您 40% 的原材料供應。建議現在就將 X 物料的安全庫存從 30 天提高到 60 天。」") + st.caption("範例回覆：「這將影響您 40% 的原材料供應。建議現在就將 X 物料的安全庫存從 30 天提高到 60 天。」") + st.session_state["whatif_decision_draft"] = draft + except (ValueError, PermissionError) as exc: + st.error(str(exc)) + + draft = st.session_state.get("whatif_decision_draft") + if draft: + erp_evidence = draft["evidence_snapshot"].get("erp_impact_evidence", {}) + orders = erp_evidence.get("purchase_orders", []) + st.markdown("#### 可查核的 ERP 影響資料") + if erp_evidence.get("match_status") == "needs_human_confirmation": + st.warning("此情境沒有直接對應到 ERP 的供應商地區。下方列出所有未結採購單供人工覆核，系統不會宣稱它們已受影響。") + else: + locations = "、".join(erp_evidence.get("matched_locations", [])) + st.caption(f"依問題中的地區「{locations}」篩選未結採購單。") + if orders: + evidence_rows = [] + for order in orders: + materials = order.get("materials", []) + material_text = "、".join( + f"{item['product_name']}（庫存約剩 {item['days_left']} 天）" + if item.get("days_left") is not None else item["product_name"] + for item in materials + ) or "尚無品項資料" + evidence_rows.append({ + "採購單": order["po_id"], "供應商": order["supplier_name"], + "據點": order["supplier_location"] or "未填寫", "相關品項／庫存": material_text, + "既有延遲": f"{order['estimated_delay_days']} 天" if order.get("estimated_delay_days") is not None else "未設定", + "既有替代建議": order.get("alternative_suggestion") or "未設定", + }) + st.dataframe(pd.DataFrame(evidence_rows), hide_index=True, use_container_width=True) + else: + st.info("目前沒有符合條件的未結採購單。") + st.success("已準備好可驗證的決策草稿；請先覆核風險分數與受影響項目，再送出供人員決定。") + if st.button("🧾 將此分析帶入『決策證據與回饋』", key="whatif_to_decision"): + st.session_state["decision_prefill_pending"] = draft + st.rerun() diff --git a/frontend/page_decision_evidence.py b/frontend/page_decision_evidence.py new file mode 100644 index 0000000..54fe851 --- /dev/null +++ b/frontend/page_decision_evidence.py @@ -0,0 +1,180 @@ +"""Streamlit surface for reviewable AI decision records.""" + +from __future__ import annotations + +from datetime import datetime, timezone + +import streamlit as st + +from backend.access_control import ( + DECISION_EVIDENCE_READ, + DECISION_RECORD_WRITE, + load_principal, +) +from backend.decision_evidence import ( + add_outcome_feedback, + create_decision_record, + decide_decision_record, + list_decision_records, +) + + +_RECOMMENDATION_LABELS = { + "monitor": "持續監控", + "request_review": "請求人工覆核", + "propose_alternative_purchase": "提出替代採購建議", +} +_STATUS_LABELS = { + "proposed": "待人員決定", + "adopted": "已採納", + "rejected": "已拒絕", + "needs_more_evidence": "需要更多證據", +} + + +def _render_record(record: dict, actor: str, can_write: bool) -> None: + output = record["ai_output"] + snapshot = record["evidence_snapshot"] + with st.container(border=True): + st.markdown(f"#### 🧾 決策 `{record['decision_id']}`｜{_STATUS_LABELS.get(record['status'], record['status'])}") + left, right = st.columns(2) + left.metric("風險分數", f"{snapshot['risk_score']:.0f} / 100") + right.metric("資料截至", snapshot["data_as_of"]) + st.markdown(f"**AI 建議**：{_RECOMMENDATION_LABELS.get(output['recommendation'], output['recommendation'])}") + st.markdown(f"**依據說明**：{output['reasoning']}") + st.caption(f"模型：{record['model_name']}｜產生時間：{record['created_at']}") + st.caption("資料來源：" + "、".join(f"{item['name']}（{item['as_of']}）" for item in snapshot["sources"])) + st.caption(f"證據快照 digest：`{record['snapshot_digest'][:16]}…`｜限制：{output['limitations']}") + + if record["status"] == "proposed" and can_write: + reason = st.text_input("決定原因", key=f"decision_reason_{record['decision_id']}") + adopt, reject, more = st.columns(3) + if adopt.button("採納", key=f"adopt_{record['decision_id']}", use_container_width=True): + decide_decision_record(actor=actor, decision_id=record["decision_id"], outcome="adopted") + st.rerun() + if reject.button("拒絕", key=f"reject_{record['decision_id']}", use_container_width=True): + try: + decide_decision_record(actor=actor, decision_id=record["decision_id"], outcome="rejected", reason=reason) + st.rerun() + except ValueError as exc: + st.error(str(exc)) + if more.button("要求補證", key=f"more_{record['decision_id']}", use_container_width=True): + try: + decide_decision_record(actor=actor, decision_id=record["decision_id"], outcome="needs_more_evidence", reason=reason) + st.rerun() + except ValueError as exc: + st.error(str(exc)) + + if record["status"] == "adopted" and can_write: + with st.expander("新增實際結果回饋"): + result = st.selectbox("觀察結果", ["effective", "ineffective", "inconclusive"], format_func=lambda value: {"effective": "有效", "ineffective": "無效", "inconclusive": "尚無結論"}[value], key=f"feedback_{record['decision_id']}") + action_taken = st.text_area("實際採取的動作（必填）", placeholder="例如：將 PO-102 改由備援供應商出貨。", key=f"feedback_action_{record['decision_id']}") + outcome_evidence = st.text_area("結果依據（必填）", placeholder="例如：新供應商確認交期縮短為 7 天，採購單紀錄已更新。", key=f"feedback_evidence_{record['decision_id']}") + note = st.text_area("補充說明（選填）", key=f"feedback_note_{record['decision_id']}") + if st.button("儲存回饋", key=f"save_feedback_{record['decision_id']}"): + try: + add_outcome_feedback( + actor=actor, decision_id=record["decision_id"], outcome=result, + action_taken=action_taken, outcome_evidence=outcome_evidence, note=note, + ) + st.rerun() + except ValueError as exc: + st.error(str(exc)) + + if record["decided_by"]: + st.caption(f"人員決定：{record['decided_by']}｜{record['decided_at']}｜{record['decision_reason'] or '採納未填原因'}") + for feedback in record["feedback"]: + st.info( + f"結果回饋：{feedback['outcome']}｜{feedback['recorded_by']}｜{feedback['recorded_at']}\n\n" + f"採取動作：{feedback.get('action_taken') or '舊紀錄未填寫'}\n\n" + f"結果依據：{feedback.get('outcome_evidence') or '舊紀錄未填寫'}" + + (f"\n\n補充：{feedback['note']}" if feedback['note'] else "") + ) + + +def render(*, username: str) -> None: + principal = load_principal(username) + if principal is None or not principal.can(DECISION_EVIDENCE_READ): + st.error("你沒有查看 AI 決策證據的權限。") + return + can_write = principal.can(DECISION_RECORD_WRITE) + st.markdown("
🧠 AI 決策證據與回饋
", unsafe_allow_html=True) + st.caption("AI 只提出結構化建議；資料快照、人員決定與結果回饋均可追溯。") + + records_tab, create_tab = st.tabs(["決策紀錄", "建立／匯入建議"]) + with records_tab: + records = list_decision_records(actor=principal.username) + if not records: + st.info("尚無決策紀錄。請由具有 L2 決策權限的人建立第一筆示範建議。") + else: + adopted = sum(record["status"] == "adopted" for record in records) + decided = sum(record["status"] in {"adopted", "rejected", "needs_more_evidence"} for record in records) + feedback = [item for record in records for item in record["feedback"]] + effective = sum(item["outcome"] == "effective" for item in feedback) + metrics = st.columns(4) + metrics[0].metric("決策紀錄", len(records)) + metrics[1].metric("已完成決定", f"{decided} / {len(records)}") + metrics[2].metric("已採納", adopted) + metrics[3].metric("已有有效回饋", effective) + st.caption("摘要只統計目前帳號可查看的紀錄；回饋需同時附上實際動作與結果依據。") + for record in records: + _render_record(record, principal.username, can_write) + + with create_tab: + if not can_write: + st.info("你可查看證據與決策，但建立、採納與回饋需要 L2 決策權限。") + return + draft = st.session_state.pop("decision_prefill_pending", None) + if draft: + snapshot = draft["evidence_snapshot"] + output = draft["ai_output"] + st.session_state.update({ + "decision_entity": snapshot["affected_entity"], + "decision_score": int(snapshot["risk_score"]), + "decision_source": snapshot["sources"][0]["name"], + "decision_as_of": snapshot["data_as_of"], + "decision_recommendation": output["recommendation"], + "decision_risk_level": output["risk_level"], + "decision_reasoning": output["reasoning"], + "decision_limitations": output["limitations"], + "decision_model_name": draft["model_name"], + "decision_type": draft["decision_type"], + }) + st.success("已從 What-if 分析帶入草稿。請確認或調整欄位後，再建立正式決策紀錄。") + else: + st.caption("可手動建立建議，或從 L2 What-if 模擬分析帶入草稿。正式紀錄建立前仍須由人員確認。") + with st.form("create_decision_record"): + entity = st.text_input("受影響項目／供應商", placeholder="例如：SUP-021 / 零件 P-100", key="decision_entity") + score = st.slider("風險分數", 0, 100, 70, key="decision_score") + source_name = st.text_input("資料來源", value="供應鏈風險地圖", key="decision_source") + as_of = st.text_input("資料截至時間（UTC）", value=datetime.now(timezone.utc).replace(microsecond=0).isoformat(), key="decision_as_of") + recommendation = st.selectbox("AI 建議", list(_RECOMMENDATION_LABELS), format_func=_RECOMMENDATION_LABELS.get, key="decision_recommendation") + risk_level = st.selectbox("AI 判定風險等級", ["low", "medium", "high"], index=2, key="decision_risk_level") + reasoning = st.text_area("AI 依據說明", placeholder="請說明為何提出此建議。", key="decision_reasoning") + limitations = st.text_area("資料限制", value="此建議僅供人工覆核，未直接執行任何 ERP 異動。", key="decision_limitations") + model_name = st.text_input("模型名稱", value="manual-structured-demo", key="decision_model_name") + decision_type = st.text_input("決策類型", value="supply_chain_risk_response", key="decision_type") + submit = st.form_submit_button("建立可驗證建議") + if submit: + try: + create_decision_record( + actor=principal.username, + decision_type=decision_type, + model_name=model_name, + ai_output={ + "recommendation": recommendation, + "reasoning": reasoning, + "risk_level": risk_level, + "evidence_ids": [f"source:{source_name}"], + "limitations": limitations, + }, + evidence_snapshot={ + "risk_score": score, + "data_as_of": as_of, + "sources": [{"name": source_name, "as_of": as_of}], + "affected_entity": entity, + }, + ) + st.success("已建立決策紀錄；可回到「決策紀錄」採納、拒絕或補上結果回饋。") + except (ValueError, PermissionError) as exc: + st.error(str(exc)) diff --git a/frontend/page_supply_chain_risk.py b/frontend/page_supply_chain_risk.py index 7526093..0478d3b 100644 --- a/frontend/page_supply_chain_risk.py +++ b/frontend/page_supply_chain_risk.py @@ -6,12 +6,13 @@ """ import streamlit as st -from backend.access_control import load_principal +from backend.access_control import DECISION_EVIDENCE_READ, load_principal from frontend.access_navigation import risk_sections from frontend.components.supply_map import render_supply_chain_map, render_what_if_analysis from frontend.components.risk_dashboard import render_intelligence_gathering, render_response_execution from frontend.components.risk_overview import render_risk_overview from frontend.components.purchase_proposal_workbench import render_purchase_proposal_workbench +from frontend.page_decision_evidence import render as render_decision_evidence def render( sub_menu: str, @@ -33,16 +34,25 @@ def render( from frontend.components.news_acceptance import render_news_acceptance render_news_acceptance() - if "analysis" not in sections and "what_if" not in sections: - render_risk_overview(actor=principal.username) - return - - overview_tab, analysis_tab = st.tabs(["📊 L1 風險總覽", "🧭 L2 情報與決策"]) + tabs = ["📊 L1 風險總覽"] + if principal.can(DECISION_EVIDENCE_READ): + tabs.append("🧠 決策證據與回饋") + if "analysis" in sections or "what_if" in sections: + tabs.append("🧭 L2 情報與決策") + rendered_tabs = st.tabs(tabs) - with overview_tab: + with rendered_tabs[0]: render_risk_overview(actor=principal.username) - with analysis_tab: + evidence_index = 1 if principal.can(DECISION_EVIDENCE_READ) else None + if evidence_index is not None: + with rendered_tabs[evidence_index]: + render_decision_evidence(username=principal.username) + + analysis_index = len(rendered_tabs) - 1 + if "analysis" not in sections and "what_if" not in sections: + return + with rendered_tabs[analysis_index]: # Step 1: Intelligence Hub st.markdown("### 📡 步驟 1: 即時情報獲取與 AI 摘要") render_intelligence_gathering( diff --git a/tests/test_decision_evidence.py b/tests/test_decision_evidence.py new file mode 100644 index 0000000..e68c676 --- /dev/null +++ b/tests/test_decision_evidence.py @@ -0,0 +1,165 @@ +import pytest + +from backend import database +from backend.decision_evidence import ( + add_outcome_feedback, + build_heatmap_alert_draft, + build_what_if_decision_draft, + create_decision_record, + decide_decision_record, + get_decision_record, + validate_ai_output, +) + + +@pytest.fixture +def decision_db(tmp_path, monkeypatch): + db_path = tmp_path / "decision-evidence.db" + monkeypatch.setattr(database, "DB_FILE", str(db_path)) + database.init_db() + return db_path + + +def _output(**overrides): + value = { + "recommendation": "propose_alternative_purchase", + "reasoning": "供應商延遲事件與風險分數皆上升。", + "risk_level": "high", + "evidence_ids": ["risk-map:tw-central", "news:123"], + "limitations": "新聞資料可能延遲，需人工確認。", + } + value.update(overrides) + return value + + +def _snapshot(): + return { + "risk_score": 82, + "data_as_of": "2026-09-14T12:00:00+00:00", + "sources": [ + {"name": "供應鏈風險地圖", "as_of": "2026-09-14T12:00:00+00:00"}, + {"name": "新聞事件", "as_of": "2026-09-14T11:30:00+00:00"}, + ], + "affected_entity": "SUP-021 / P-100", + } + + +def test_decision_keeps_verified_snapshot_through_human_decision(decision_db): + created = create_decision_record( + actor="planner", decision_type="supply_chain_risk_response", + model_name="gemini-test", ai_output=_output(), evidence_snapshot=_snapshot(), + decision_id="DEC-TEST-1", + ) + assert created["status"] == "proposed" + assert created["snapshot_digest"] + assert created["evidence_snapshot"]["risk_score"] == 82 + + decided = decide_decision_record( + actor="planner", decision_id="DEC-TEST-1", outcome="adopted" + ) + assert decided["status"] == "adopted" + assert decided["evidence_snapshot"] == created["evidence_snapshot"] + assert decided["snapshot_digest"] == created["snapshot_digest"] + + +def test_invalid_structured_output_and_invalid_snapshot_are_rejected(decision_db): + with pytest.raises(ValueError, match="recommendation"): + validate_ai_output(_output(recommendation="free-form-action")) + with pytest.raises(ValueError, match="risk_score"): + create_decision_record( + actor="planner", decision_type="supply_chain_risk_response", + model_name="gemini-test", ai_output=_output(), + evidence_snapshot={**_snapshot(), "risk_score": 101}, + ) + + +def test_what_if_response_becomes_reviewable_but_unpersisted_draft(): + draft = build_what_if_decision_draft( + question="紅海航線中斷兩週，哪些採購單會受影響？", + answer="採購單 PO-101 與 PO-102 可能延遲，建議人工覆核。", + model_name="gemini/gemini-2.5-flash", + ) + + assert draft["decision_type"] == "supply_chain_what_if_response" + assert draft["ai_output"]["recommendation"] == "request_review" + assert draft["ai_output"]["reasoning"].startswith("採購單 PO-101") + assert draft["evidence_snapshot"]["risk_score"] == 50 + assert "紅海航線" in draft["evidence_snapshot"]["affected_entity"] + assert draft["ai_output"] == validate_ai_output(draft["ai_output"]) + + with pytest.raises(ValueError, match="分析失敗"): + build_what_if_decision_draft( + question="測試", answer="模擬分析暫時無法產生：缺少金鑰", model_name="gemini/test" + ) + + +def test_high_risk_heatmap_node_becomes_reviewable_alert_draft(): + draft = build_heatmap_alert_draft( + region_name="越南 東南亞", + risk_score=88, + ai_summary="港口罷工可能延誤零件交期。", + data_as_of="2026-09-14T12:00:00+00:00", + ) + assert draft["decision_type"] == "supply_chain_heatmap_alert" + assert draft["ai_output"]["recommendation"] == "propose_alternative_purchase" + assert draft["evidence_snapshot"]["risk_score"] == 88 + assert draft["ai_output"] == validate_ai_output(draft["ai_output"]) + + with pytest.raises(ValueError, match="達 70"): + build_heatmap_alert_draft(region_name="台灣 東亞", risk_score=69) + + +def test_viewer_cannot_create_or_decide_and_adopted_record_accepts_feedback(decision_db): + with pytest.raises(PermissionError, match="decision.record.write"): + create_decision_record( + actor="viewer", decision_type="supply_chain_risk_response", + model_name="gemini-test", ai_output=_output(), evidence_snapshot=_snapshot(), + ) + create_decision_record( + actor="planner", decision_type="supply_chain_risk_response", + model_name="gemini-test", ai_output=_output(), evidence_snapshot=_snapshot(), + decision_id="DEC-TEST-2", + ) + with pytest.raises(PermissionError, match="decision.record.write"): + decide_decision_record(actor="viewer", decision_id="DEC-TEST-2", outcome="adopted") + decide_decision_record(actor="planner", decision_id="DEC-TEST-2", outcome="adopted") + record = add_outcome_feedback( + actor="planner", decision_id="DEC-TEST-2", outcome="effective", + action_taken="改由備援供應商出貨。", outcome_evidence="採購單顯示延遲已降低。", note="實際延遲已降低。" + ) + assert record["feedback"][0]["outcome"] == "effective" + assert record["feedback"][0]["action_taken"] == "改由備援供應商出貨。" + assert get_decision_record(actor="viewer", decision_id="DEC-TEST-2")["status"] == "adopted" + + +def test_feedback_requires_action_and_outcome_evidence(decision_db): + create_decision_record( + actor="planner", decision_type="supply_chain_risk_response", + model_name="gemini-test", ai_output=_output(), evidence_snapshot=_snapshot(), decision_id="DEC-TEST-4", + ) + decide_decision_record(actor="planner", decision_id="DEC-TEST-4", outcome="adopted") + with pytest.raises(ValueError, match="實際採取"): + add_outcome_feedback( + actor="planner", decision_id="DEC-TEST-4", outcome="effective", + action_taken="", outcome_evidence="交期已確認。", + ) + with pytest.raises(ValueError, match="結果依據"): + add_outcome_feedback( + actor="planner", decision_id="DEC-TEST-4", outcome="effective", + action_taken="改由備援供應商出貨。", outcome_evidence="", + ) + + +def test_rejection_requires_reason_and_record_cannot_be_decided_twice(decision_db): + create_decision_record( + actor="planner", decision_type="supply_chain_risk_response", + model_name="gemini-test", ai_output=_output(), evidence_snapshot=_snapshot(), + decision_id="DEC-TEST-3", + ) + with pytest.raises(ValueError, match="必須填寫原因"): + decide_decision_record(actor="planner", decision_id="DEC-TEST-3", outcome="rejected") + decide_decision_record( + actor="planner", decision_id="DEC-TEST-3", outcome="rejected", reason="成本資料不足。" + ) + with pytest.raises(ValueError, match="不能重複"): + decide_decision_record(actor="planner", decision_id="DEC-TEST-3", outcome="adopted") diff --git a/tests/test_supply_chain_authorization.py b/tests/test_supply_chain_authorization.py index 51a5927..9c5e0e2 100644 --- a/tests/test_supply_chain_authorization.py +++ b/tests/test_supply_chain_authorization.py @@ -249,6 +249,37 @@ def fake_llm(*args, **kwargs): assert len(llm_calls) == 1 +def test_what_if_can_return_a_validated_ai_decision_draft(supply_db, monkeypatch): + monkeypatch.setattr( + "backend.llm_client.complete_text", + lambda *args, **kwargs: '''{"recommendation":"request_review","risk_level":"high","risk_score":82,"affected_entity":"PO-101 / SUP-021","reasoning":"交期與庫存風險升高。","limitations":"需確認供應商最新復工日期。"}''', + ) + + draft = risk.what_if_decision_analysis("", "台灣地震造成停工", actor="planner") + + assert draft["ai_output"]["recommendation"] == "request_review" + assert draft["ai_output"]["risk_level"] == "high" + assert draft["evidence_snapshot"]["risk_score"] == 82 + assert draft["evidence_snapshot"]["affected_entity"] == "PO-101 / SUP-021" + + +def test_what_if_erp_evidence_matches_location_without_llm(supply_db): + evidence = risk.get_what_if_erp_evidence("Taiwan 地震造成停工", actor="planner") + + assert evidence["match_status"] == "location_matched" + assert evidence["purchase_orders"][0]["po_id"] == "PO-AUTH" + material = evidence["purchase_orders"][0]["materials"][0] + assert material["product_id"] == "P-AUTH" + assert material["days_left"] == 25 + + +def test_what_if_erp_evidence_marks_unmatched_scenario_for_review(supply_db): + evidence = risk.get_what_if_erp_evidence("紅海航線中斷兩週", actor="planner") + + assert evidence["match_status"] == "needs_human_confirmation" + assert evidence["purchase_orders"][0]["po_id"] == "PO-AUTH" + + def test_entitlement_revocation_is_immediate(supply_db, monkeypatch): llm_calls = [] diff --git a/tests/test_tier_authorization.py b/tests/test_tier_authorization.py index c878c0d..cbb2fd9 100644 --- a/tests/test_tier_authorization.py +++ b/tests/test_tier_authorization.py @@ -11,6 +11,8 @@ from backend.access_control import ( APPROVAL_DECIDE, APPROVAL_QUEUE_READ, + DECISION_EVIDENCE_READ, + DECISION_RECORD_WRITE, ERP_EXCHANGE_EXPORT, ERP_EXCHANGE_PROPOSE, ERP_EXCHANGE_RECONCILE, @@ -117,11 +119,13 @@ def test_demo_roles_have_context_visibility_without_inheriting_actions(): planner = capabilities_for_role("supply_planner") approver = capabilities_for_role("procurement_approver") - # L1 告警確認是監控狀態、不是 ERP 寫入，viewer 仍無任何 L2/L3 能力 - assert viewer == {RISK_OVERVIEW_READ, RISK_ALERT_ACK} + # L1 告警確認是監控狀態、不是 ERP 寫入；決策證據為唯讀。viewer 仍無任何 L2/L3 能力 + assert viewer == {RISK_OVERVIEW_READ, RISK_ALERT_ACK, DECISION_EVIDENCE_READ} assert {RISK_OVERVIEW_READ, RISK_ANALYSIS_READ, RISK_WHAT_IF_RUN} <= planner assert ERP_EXCHANGE_PROPOSE in planner + assert DECISION_EVIDENCE_READ in planner + assert DECISION_RECORD_WRITE in planner assert APPROVAL_DECIDE not in planner assert ERP_EXCHANGE_EXPORT not in planner @@ -131,6 +135,8 @@ def test_demo_roles_have_context_visibility_without_inheriting_actions(): assert APPROVAL_DECIDE in approver assert ERP_EXCHANGE_EXPORT in approver assert ERP_EXCHANGE_RECONCILE in approver + assert DECISION_EVIDENCE_READ in approver + assert DECISION_RECORD_WRITE not in approver assert RISK_ANALYSIS_READ not in approver assert RISK_WHAT_IF_RUN not in approver assert ERP_EXCHANGE_PROPOSE not in approver diff --git a/tests/test_tier_navigation.py b/tests/test_tier_navigation.py index c045b80..c9390f0 100644 --- a/tests/test_tier_navigation.py +++ b/tests/test_tier_navigation.py @@ -178,9 +178,10 @@ def test_tier_pages_derive_sections_from_live_principal(): assert "principal = load_principal(username)" in risk_source assert "sections = risk_sections(principal)" in risk_source - # L2 五個渲染器 + L1 總覽的兩個呼叫點（單頁／分頁）都必須轉發 live actor - assert risk_source.count("actor=principal.username") == 7 - assert risk_source.count("render_risk_overview(actor=principal.username)") == 2 + # L2 五個渲染器 + L1 總覽（#20 起單一呼叫點、所有角色共用）都必須轉發 live actor + assert risk_source.count("actor=principal.username") == 6 + assert risk_source.count("render_risk_overview(") == 1 + assert risk_source.count("render_risk_overview(actor=principal.username)") == 1 assert "principal = load_principal(username)" in exchange_source assert "sections = exchange_sections(principal)" in exchange_source assert "actor=current_actor" in exchange_source