diff --git a/backend/access_control.py b/backend/access_control.py
index e76536d..53a8f92 100644
--- a/backend/access_control.py
+++ b/backend/access_control.py
@@ -25,6 +25,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"
@@ -43,13 +45,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_viewer": frozenset({RISK_OVERVIEW_READ, DECISION_EVIDENCE_READ}),
"supply_planner": frozenset(
{
RISK_OVERVIEW_READ,
@@ -57,6 +61,8 @@
RISK_WHAT_IF_RUN,
RISK_WORKSPACE_WRITE,
ERP_EXCHANGE_PROPOSE,
+ DECISION_EVIDENCE_READ,
+ DECISION_RECORD_WRITE,
}
),
"procurement_approver": frozenset(
@@ -67,6 +73,7 @@
APPROVAL_DECIDE,
ERP_EXCHANGE_EXPORT,
ERP_EXCHANGE_RECONCILE,
+ DECISION_EVIDENCE_READ,
}
),
# Preserve the existing demo accounts while routing the new accounts
@@ -82,6 +89,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 4af17ec..4008add 100644
--- a/backend/database.py
+++ b/backend/database.py
@@ -214,6 +214,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 0afa71a..2dac3e6 100644
--- a/backend/prompts.py
+++ b/backend/prompts.py
@@ -72,6 +72,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 bcbf9a3..d4f4b2b 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,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 = sqlite3.connect(DB_FILE)
suppliers = __pd_read("SELECT supplier_id, name, country, region FROM suppliers", conn)
pos = __pd_read(
@@ -854,20 +868,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 0511191..c3600ba 100644
--- a/frontend/components/supply_map.py
+++ b/frontend/components/supply_map.py
@@ -1,6 +1,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,
@@ -11,7 +12,7 @@
get_impacted_pos,
update_po_impact,
get_ai_alternative_suggestions,
- what_if_simulation,
+ what_if_decision_analysis,
)
@@ -70,6 +71,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 小時未更新；請先更新情報或重新產生摘要，再據此做決策。")
+
def render_risk_shortcuts(key: str, heatmap_rows=None, *, actor: str):
"""區域風險快速分析小卡。"""
if heatmap_rows is None:
@@ -276,6 +299,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 摘要**")
@@ -490,13 +538,28 @@ 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
+ )
+ 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:
+ 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 e506373..bb8f11d 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,
@@ -31,16 +32,25 @@ def render(
st.markdown("🌱 供應鏈與風險監控
", unsafe_allow_html=True)
- if "analysis" not in sections and "what_if" not in sections:
- render_risk_overview()
- 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()
- 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 9cda089..72ef5cd 100644
--- a/tests/test_supply_chain_authorization.py
+++ b/tests/test_supply_chain_authorization.py
@@ -248,6 +248,20 @@ 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_entitlement_revocation_is_immediate(supply_db, monkeypatch):
llm_calls = []
diff --git a/tests/test_tier_authorization.py b/tests/test_tier_authorization.py
index 97bd378..8967944 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,
@@ -116,10 +118,12 @@ def test_demo_roles_have_context_visibility_without_inheriting_actions():
planner = capabilities_for_role("supply_planner")
approver = capabilities_for_role("procurement_approver")
- assert viewer == {RISK_OVERVIEW_READ}
+ assert viewer == {RISK_OVERVIEW_READ, 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
@@ -129,6 +133,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