From 21d76dc8efc6c52a033f56acd88699d103c06e3a Mon Sep 17 00:00:00 2001 From: falltwo Date: Tue, 21 Jul 2026 06:03:18 +0800 Subject: [PATCH] feat: complete governed L1-L3 supply chain workflow --- .env.example | 5 + .gitignore | 4 + README.en.md | 203 +++++ README.md | 286 +++--- backend/access_control.py | 6 + backend/database.py | 88 ++ backend/l1_monitoring.py | 145 +++ backend/procurement.py | 53 +- backend/purchase_proposals.py | 854 ++++++++++++++++++ backend/tool_classification.py | 2 +- backend/tool_gateway.py | 81 +- docs/releases/v1.0.md | 82 ++ frontend/access_navigation.py | 1 + .../components/purchase_proposal_workbench.py | 216 +++++ frontend/components/risk_overview.py | 158 +++- frontend/page_agent_dashboard.py | 171 +++- frontend/page_supply_chain_risk.py | 6 + scripts/seed_e_day1_demo_data.py | 66 +- tests/test_demo_seed_schema.py | 76 ++ tests/test_l1_monitoring.py | 142 +++ tests/test_purchase_order_approval.py | 2 +- tests/test_purchase_proposal_ui.py | 176 ++++ tests/test_purchase_proposals.py | 803 ++++++++++++++++ tests/test_tier_authorization.py | 22 + tests/test_tier_navigation.py | 2 +- 25 files changed, 3472 insertions(+), 178 deletions(-) create mode 100644 README.en.md create mode 100644 backend/l1_monitoring.py create mode 100644 backend/purchase_proposals.py create mode 100644 docs/releases/v1.0.md create mode 100644 frontend/components/purchase_proposal_workbench.py create mode 100644 tests/test_l1_monitoring.py create mode 100644 tests/test_purchase_proposal_ui.py create mode 100644 tests/test_purchase_proposals.py diff --git a/.env.example b/.env.example index e13d4f5..de37a5d 100644 --- a/.env.example +++ b/.env.example @@ -32,6 +32,11 @@ LINE_BRIEFING_ENABLED=false # Local competition/demo only. Safe default is false; never enable on a public service. ERP_DEMO_MODE=false +# PoC tenant boundary: one SQLite database belongs to one organization. +# Demo Mode automatically uses demo-org. Existing non-demo databases must set this, +# then provision user_organizations and organization_entitlements before startup. +# ERP_ORGANIZATION_ID=your-organization-id + # Optional service identity for the 24-hour supply-chain news refresh. # The account must have risk.workspace.write; for the local demo use planner. ERP_SCHEDULER_ACTOR= diff --git a/.gitignore b/.gitignore index cbe61e1..643aace 100644 --- a/.gitignore +++ b/.gitignore @@ -228,3 +228,7 @@ data/erp.db-wal data/*preview*.db* .preview/ line bot/ngrok.exe + +# Local AI-assistant and planning state must never enter the public repository. +.claude/ +.planning/ diff --git a/README.en.md b/README.en.md new file mode 100644 index 0000000..a03a9fc --- /dev/null +++ b/README.en.md @@ -0,0 +1,203 @@ +# AI-Risk-Based-Inventory-ERP + +[繁體中文](README.md) | [English](README.en.md) + +[![Tests](https://github.com/falltwo/AI-Risk-Based-Inventory-ERP/actions/workflows/tests.yml/badge.svg)](https://github.com/falltwo/AI-Risk-Based-Inventory-ERP/actions/workflows/tests.yml) +[![Release](https://img.shields.io/github/v/release/falltwo/AI-Risk-Based-Inventory-ERP?display_name=tag)](https://github.com/falltwo/AI-Risk-Based-Inventory-ERP/releases) +![Python](https://img.shields.io/badge/Python-3.11%2B-3776AB?logo=python&logoColor=white) +![Streamlit](https://img.shields.io/badge/UI-Streamlit-FF4B4B?logo=streamlit&logoColor=white) + +> **v1.0 — A governed supply-chain AI decision loop** +> AI makes judgments and proposals; humans retain execution authority. Protected AI/Gateway procurement writes can be approved, replayed, and traced. + +This project is a governance-first AI Agent ERP. It connects external supply-chain risk, internal procurement data, AI recommendations, human approval, and ERP execution into one verifiable workflow—rather than stopping at a chatbot or a risk dashboard. + +> [!IMPORTANT] +> v1.0 is a competition and research proof of concept. Its deployment boundary is one SQLite database per organization. It does not provide shared-database row-level multi-tenancy, external IAM/SSO, or distributed transactions, and must not be treated as a production identity or authorization service on the public internet. + +## v1.0 at a glance + +| Tier | Demo account | What it can do | What it cannot do | +|---|---|---|---| +| **L1 Risk Observer** | `viewer / viewer` | Risk KPIs, heatmap, alerts, read-only CSV mapping, notification preview | Cannot create proposals or modify ERP data | +| **L2 Intelligence & Decision** | `planner / planner` | Impact analysis, What-if, alternative-supplier comparison, durable Proposal submission | Cannot approve or directly execute ERP writes | +| **L3 Approval & Execution** | `approver / approver` | Review evidence, approve/reject, Gateway execution, audit timeline | Cannot approve its own proposal | + +### End-to-end decision flow + +```mermaid +flowchart LR + NEWS["External risk intelligence"] --> L1["L1 Observe
alerts, map, read-only mapping"] + L1 --> L2["L2 Recommend
What-if, supplier alternatives"] + L2 --> PROP["Durable Proposal
source line, price, digest"] + PROP --> L3["L3 Approve / Reject
human decision"] + L3 --> GATE["Tool Gateway
authorization, CAS, transaction, idempotency"] + GATE --> ERP["ERP Effect
purchase order + receipt + audit"] + L3 -. "not approved" .-> STOP["zero ERP writes"] +``` + +## v0.1 → v1.0 + +v1.0 preserves the governance harness completed in v0.1 and turns it into an operable supply-chain decision product. + +| Area | v0.1 — Governance Harness Complete | v1.0 — Governed Decision Loop | +|---|---|---| +| Primary outcome | Closed governance bypasses across Web, LINE, and rollback paths | Connected the governance foundation into a complete L1→L2→L3 product flow | +| AI honesty | Code-enforced pending/denied disclosure | Separate Proposal, Approval, and Execution objects keep UI and database state aligned | +| Supply-chain UX | Intelligence, heatmap, affected records, and recommendations existed as separate capabilities | An affected procurement line can become a governed alternative-purchase Proposal | +| Human approval | Generic write approval with auditable state | L3 reviews source PO, supplier change, quantity, unit price, reason, and digest | +| Execution safety | Gateway, hash-chain logs, and transaction baseline | Exact line/price identity, live revocation checks, one effect per source line, idempotent receipts | +| Product tiers | Governance roles and capabilities | Three accounts with distinct views and least-privilege behavior | +| Automated tests | 55 tests in the v0.1 release | **327 passing tests** in v1.0 release verification | +| Documentation | Chinese README and architecture diagrams | Bilingual README, version comparison, honest limits, and English release notes | + +The v0.1 column is based on the archived release record retained by the maintainer. The private archive is intentionally not linked from public documentation. + +## Governance and security design + +- **Server-side capability checks:** role, organization membership, and entitlements are reloaded from the database; missing or revoked access fails closed. +- **Separation of duties:** L2 proposes and L3 decides. The original proposer cannot self-approve, even after a role change. +- **Immutable approval evidence:** a canonical payload digest covers effectful fields and binds the source PO line, supplier price row, and operation ID. +- **Atomic execution:** protected purchase approval performs CAS state transition, ERP write, business-effect claim, execution receipt, and terminal status in one SQLite transaction. +- **Idempotent replay:** the same operation returns its existing receipt instead of creating a second purchase order. +- **End-to-end audit:** Proposal, approval, and execution share one operation ID; public UI surfaces expose only redacted summaries. +- **34 governed tools:** 27 `read_only`, 1 `suggestion`, 6 `write`, and 0 `dangerous`; eight specialist Agents receive task-specific allowlists. + +## Architecture + +```mermaid +flowchart TB + subgraph ENTRY["Entry and identity"] + WEB["Streamlit Web"] + LINE["LINE Bot"] + WEB_ACCESS["Web: Role + Membership + Entitlement"] + LINE_ACCESS["LINE: source allowlist + role policy"] + end + + subgraph AI["AI orchestration"] + ORCH["Orchestrator Agent"] + AGENTS["8 specialist Agents"] + end + + subgraph GOV["Governance"] + REG["Tool Registry / Allowlist"] + GATE["Tool Gateway"] + APPROVAL["Proposal / Approval / Execution"] + end + + subgraph DATA["Data and evidence"] + ERP["ERP Modules"] + DB["SQLite"] + AUDIT["Audit Logs + Receipts"] + end + + WEB --> WEB_ACCESS + LINE --> LINE_ACCESS + WEB_ACCESS --> ORCH + LINE_ACCESS --> ORCH + ORCH --> AGENTS --> REG --> GATE --> APPROVAL --> ERP --> DB + GATE --> AUDIT + APPROVAL --> AUDIT +``` + +The governance claims above are scoped to the protected AI/Gateway procurement workflow. Existing manual Web ERP forms have role-based access controls, but not every manual write produces a Proposal, Approval, and execution receipt. + +Original diagrams: [system architecture PNG](docs/images/erp_current_clean_architecture.png) · [governance flow PNG](docs/images/erp_current_standard_flowchart.png) + +## Quick start + +### 1. Install + +```bash +git clone https://github.com/falltwo/AI-Risk-Based-Inventory-ERP.git +cd AI-Risk-Based-Inventory-ERP + +python -m venv .venv +# Windows +.venv\Scripts\activate +# macOS / Linux +# source .venv/bin/activate + +pip install -r requirements.txt +``` + +### 2. Configure a local demo + +```bash +cp .env.example .env +``` + +Set at least: + +```dotenv +ERP_DEMO_MODE=true +LLM_MODEL=gemini/gemini-2.5-flash +GEMINI_API_KEY=replace_with_your_key +``` + +### 3. Run + +```bash +streamlit run app.py +``` + +Known credentials such as `viewer`, `planner`, and `approver` are created and displayed only in Demo Mode. **Use this mode only on localhost; never expose it to the public internet.** + +## Key configuration + +| Environment variable | Purpose | Default / requirement | +|---|---|---| +| `ERP_DEMO_MODE` | Seeds synthetic data and demo users | `false`; localhost only | +| `ERP_ORGANIZATION_ID` | Binds a SQLite database to one organization | Demo uses `demo-org`; existing non-demo databases must set it and then provision memberships and entitlements | +| `ERP_DB_PATH` | Custom SQLite path | `data/erp.db` | +| `LLM_MODEL` | Primary LiteLLM model | `gemini/gemini-2.5-flash` | +| `LLM_FALLBACK_MODELS` | Comma-separated fallback models | See `.env.example` | +| `LLM_ANALYSIS_MODEL` | Optional model for classification/translation | Primary chain when unset | +| `GEMINI_API_KEY` / `OPENAI_API_KEY` | Provider credentials | Depends on the selected model | +| `GNEWS_API_KEY` | Supply-chain news source | Optional | +| `ERP_SCHEDULER_ACTOR` | Service identity for scheduled risk refresh | Disabled when unset | +| `LINE_CHANNEL_ACCESS_TOKEN` / `LINE_CHANNEL_SECRET` | LINE Bot | Optional | + +## Tests and verification + +```bash +pip install -r requirements-dev.txt +python -m pytest -q +``` + +v1.0 local release verification: **327 passed**. CI runs on every pull request. + +Coverage includes: + +- L1/L2/L3 navigation and negative server-side authorization tests +- Self-approval, revoked access, and cross-organization denial +- Payload, resource-version, source-line, and price tamper rejection +- Concurrent approval, CAS, rollback, and receipt replay +- A single full replacement effect per source procurement line +- Demo seed integrity with no orphan items and stable approved source-line identity across replays + +## Repository layout + +```text +backend/ access control, Agents, Gateway, Proposal, ERP, database +frontend/ Streamlit pages and L1/L2/L3 interfaces +line bot/ FastAPI + LINE Messaging API +scripts/ demo seed and operations utilities +tests/ governance, authorization, transaction, and UI-contract tests +docs/ architecture diagrams, runbooks, and release notes +``` + +## Known limitations + +- One SQLite database represents one organization; this is not shared-database row-level multi-tenancy. +- Application audit data is tamper-evident, but a host or database administrator can still alter files directly. +- SQLite atomicity does not automatically extend to an external ERP API; cross-system execution still needs outbox, worker, and reconciliation patterns. +- Demo users and synthetic data must not exist in production. Production identity, membership, entitlement, and secret provisioning are deployment responsibilities. +- Upgrading an older non-demo database without an organization boundary fails fast. Set `ERP_ORGANIZATION_ID`, then provision `user_organizations` and `organization_entitlements` before startup. + +## Versions + +- [v1.0 Releases](https://github.com/falltwo/AI-Risk-Based-Inventory-ERP/releases) +- [v1.0 English release notes](docs/releases/v1.0.md) + +Stack: Python 3.11 · Streamlit · SQLite · LiteLLM · FastAPI · LINE Messaging API · Plotly diff --git a/README.md b/README.md index c884737..50cd0ac 100644 --- a/README.md +++ b/README.md @@ -1,199 +1,203 @@ # AI-Risk-Based-Inventory-ERP -![tests](https://github.com/falltwo/AI-Risk-Based-Inventory-ERP/actions/workflows/tests.yml/badge.svg) +[繁體中文](README.md) | [English](README.en.md) -**治理優先的 AI Agent 進銷存系統** —— 讓 AI 助理能實際操作 ERP（查庫存、開訂單、評估供應鏈風險），同時把每一個 AI 自主行動納入「可控、可審批、可稽核」的治理鏈。 +[![Tests](https://github.com/falltwo/AI-Risk-Based-Inventory-ERP/actions/workflows/tests.yml/badge.svg)](https://github.com/falltwo/AI-Risk-Based-Inventory-ERP/actions/workflows/tests.yml) +[![Release](https://img.shields.io/github/v/release/falltwo/AI-Risk-Based-Inventory-ERP?display_name=tag)](https://github.com/falltwo/AI-Risk-Based-Inventory-ERP/releases) +![Python](https://img.shields.io/badge/Python-3.11%2B-3776AB?logo=python&logoColor=white) +![Streamlit](https://img.shields.io/badge/UI-Streamlit-FF4B4B?logo=streamlit&logoColor=white) -## 這套系統解決什麼問題 +> **v1.0 — 可治理的供應鏈 AI 決策閉環** +> AI 負責判斷與提案，人類保留執行權；受保護的 AI／Gateway 採購寫入可被審批、重放與追溯。 -企業導入 AI Agent 最大的顧慮不是能力，是控制：AI 建議可以參考，但讓 AI 直接改庫存、開訂單，誰來把關？出了問題怎麼追？ +本專案是一套治理優先的 AI Agent 進銷存系統。它把外部供應鏈風險、企業採購資料、AI 決策提案與 ERP 執行接成一條可驗證的流程，而不是只做聊天機器人或風險儀表板。 -本系統目前實作一條治理流程： +> [!IMPORTANT] +> 本版本是競賽與研究型 PoC。它採「一個 SQLite 資料庫對應一個組織」的部署邊界，尚未提供共享資料庫的多租戶隔離、外部 IAM/SSO 或跨系統分散式交易，因此不得直接當成公開網路服務的正式身分與授權系統。 -- Web／LINE Agent 路徑的工具呼叫都會經過 **Tool Gateway**；Web 另檢查 Agent 白名單，LINE 在模型前先縮減工具範圍 -- 會改動資料的操作一律**先攔截、送人工審批**，核准後才執行 -- 全程寫入**稽核紀錄**（派工決策、工具呼叫、審批流程三層） -- 審批狀態由**程式層保留並揭露** —— 降低模型把 pending 說成 completed 的風險 +## v1.0 一眼看懂 -## 核心特性 +| 層級 | Demo 帳號 | 能做什麼 | 明確不能做什麼 | +|---|---|---|---| +| **L1 風險觀測** | `viewer / viewer` | 風險 KPI、熱圖、最新告警、唯讀 CSV 對映與通知預覽 | 不建立提案、不修改 ERP | +| **L2 情報與決策** | `planner / planner` | 影響分析、What-if、替代供應商比較、建立不可變 Proposal 並送審 | 不核准、不直接執行 ERP 寫入 | +| **L3 核准與執行** | `approver / approver` | 檢視核准證據、核准／拒絕、Gateway 執行、稽核時間線 | 不能核准自己的提案 | -| 特性 | 說明 | -|------|------| -| 總管 Agent 派工 | 自然語言任務由 LLM 語意路由到專責 Agent，跨領域任務自動串接多個 Agent 並彙整 | -| 8 個專責 Agent | 庫存、採購、銷售、財務、人資、ESG、供應鏈風險、客服 —— 各自僅持有職責內的工具白名單 | -| Tool Gateway | 32 個登記工具的集中政策入口：檢查工具存在 → Agent 白名單 → 角色權限 → 工具政策分類；尚未以系統級方法證明不存在所有旁路 | -| 四類工具 taxonomy | `read_only` / `suggestion` 直接執行；`write` / `dangerous` 攔截送審批。目前四類只形成兩種執行結果，`dangerous` 工具數為 0 | -| 人工審批流程 | 待審批清單、核准 / 拒絕（附原因）、沖銷與重試，皆於 Dashboard 操作 | -| 三層稽核紀錄 | 派工決策、工具呼叫與審批歷程；目前 requester 主要記錄角色，尚未完成個人層級職責分離 | -| 審批狀態揭露 | 操作被攔下時，回覆由系統層強制附上審批單號與「尚未執行」告示，不依賴 LLM 自律 | -| 多供應商容錯 | LLM 供應商掉線自動依序切換備援模型（LiteLLM，一行設定換模型） | -| 供應鏈風險分析 | 外部新聞情資 → 風險熱圖 → 受影響採購單 → 替代建議 | -| 雙入口 | Web（Streamlit）與 LINE Bot 最終共用 Gateway；LINE 另有來源專用 allowlist，兩個入口的前置控制並不完全相同 | +### 完整決策鏈 -## 系統架構 +```mermaid +flowchart LR + NEWS["外部風險情資"] --> L1["L1 Observe
告警、熱圖、唯讀對映"] + L1 --> L2["L2 Recommend
What-if、替代供應商"] + L2 --> PROP["Durable Proposal
來源明細、價格、digest"] + PROP --> L3["L3 Approve / Reject
人工決策"] + L3 --> GATE["Tool Gateway
權限、CAS、交易、冪等"] + GATE --> ERP["ERP Effect
採購單 + receipt + audit"] + L3 -. "未核准" .-> STOP["零 ERP 寫入"] +``` + +## v0.1 → v1.0 + +v1.0 延續 v0.1 已完成的治理 harness，將它推進成可展示、可操作的供應鏈決策產品。 -### GitHub 可渲染版 +| 面向 | v0.1 — Governance Harness Complete | v1.0 — Governed Decision Loop | +|---|---|---| +| 核心成果 | 關閉 Web、LINE、rollback 等治理旁路 | 將治理底座接成 L1→L2→L3 完整產品流程 | +| AI 誠實性 | 由程式強制揭露 pending／denied，不依賴 prompt | Proposal、Approval、Execution 分離，畫面與資料庫狀態一致 | +| 供應鏈體驗 | 情資、熱圖、受影響單據與建議各自存在 | 受影響採購明細可直接形成替代採購 Proposal | +| 人工核准 | 通用寫入審批與可稽核狀態 | L3 顯示來源單據、供應商變更、數量、單價、理由與 digest | +| 執行安全 | Gateway、hash-chain log、transaction baseline | exact line/price identity、即時撤權檢查、同來源明細唯一 effect、冪等 receipt | +| 產品分層 | 角色與治理能力為主要重點 | 三個獨立帳號、三種可見功能與最小權限 | +| 自動化測試 | 55 tests（v0.1 release） | **327 passing tests**（v1.0 release verification） | +| 文件 | 中文 README 與架構圖 | 雙語 README、版本比較、誠實邊界與 v1.0 Release notes | -**目前系統架構圖** +v0.1 欄位根據維護者保留的封存 Release 紀錄整理；私有封存庫不列入公開文件連結。 + +## 治理與安全設計 + +- **伺服器端能力檢查**：角色、組織 membership 與 entitlement 每次從資料庫重新載入；缺值或撤權後一律 fail closed。 +- **職責分離**：L2 只能提案，L3 才能決策；同一帳號即使換角色也不能核准自己的提案。 +- **不可變核准證據**：canonical payload digest 覆蓋真正決定效果的欄位，並綁定來源採購明細、替代供應商價格與 operation ID。 +- **原子執行**：受保護採購單在同一 SQLite transaction 內完成 CAS 狀態轉移、ERP 寫入、business-effect claim、execution receipt 與終態。 +- **冪等重放**：相同 operation 重送時回傳既有 receipt，不會建立第二張採購單。 +- **端到端稽核**：L2 Proposal、L3 決策與 Gateway 執行以同一 operation ID 串接；公開畫面只顯示脫敏摘要。 +- **34 個受治理工具**：27 `read_only`、1 `suggestion`、6 `write`、0 `dangerous`；8 個專責 Agent 僅持有職責內白名單。 + +## 系統架構 ```mermaid flowchart TB - subgraph ENTRY["入口層"] - WEB["Streamlit Web
ERP 功能頁、AI 助理、語音草稿"] - DASH["Agent Dashboard
Agent 狀態、審批、派工與工具 log"] - LINE["LINE Bot
FastAPI Webhook、低權限遠端查詢"] - RBAC["RBAC
admin / warehouse / sales / hr"] + subgraph ENTRY["入口與身分"] + WEB["Streamlit Web"] + LINE["LINE Bot"] + WEB_ACCESS["Web: Role + Membership + Entitlement"] + LINE_ACCESS["LINE: source allowlist + role policy"] end - subgraph AI["AI 編排層"] - ORCH["總管 Agent
LiteLLM 語意路由、關鍵字 fallback"] - AGENTS["8 個專責 Agent
庫存、採購、銷售、財務、人資、ESG、供應鏈風險、客服"] + subgraph AI["AI 編排"] + ORCH["總管 Agent"] + AGENTS["8 個專責 Agent"] end subgraph GOV["治理層"] - GATE["Tool Gateway
工具登記、Agent 白名單、角色權限、風險等級檢核"] - APPROVAL["審批控制
write / dangerous 建立 pending approvals"] - SAFE["安全補強
prompt 防注入、LINE 工具收斂、錯誤脫敏、密碼雜湊"] + REG["Tool Registry / Allowlist"] + GATE["Tool Gateway"] + APPROVAL["Proposal / Approval / Execution"] end - subgraph DATA["資料與稽核層"] - TOOLS["ERP 工具模組
庫存、訂單、採購、財務、人資、製造、ESG、供應鏈風險"] - DB["SQLite 資料庫
ERP_DB_PATH、init_db、使用者與營運資料表"] - LOGS["稽核紀錄
agent_dispatch_logs、agent_action_logs、pending_approvals、llm_usage_logs"] + subgraph DATA["資料與證據"] + ERP["ERP Modules"] + DB["SQLite"] + AUDIT["Audit Logs + Receipts"] end - ENTRY --> AI --> GOV --> DATA + WEB --> WEB_ACCESS + LINE --> LINE_ACCESS + WEB_ACCESS --> ORCH + LINE_ACCESS --> ORCH + ORCH --> AGENTS --> REG --> GATE --> APPROVAL --> ERP --> DB + GATE --> AUDIT + APPROVAL --> AUDIT ``` -**AI 任務治理標準流程圖** +治理宣稱的邊界是上圖中的受保護 AI／Gateway 採購流程。現有手動 Web ERP 表單另有角色權限控制，但並非每個手動寫入都會產生 Proposal、Approval 與 execution receipt。 -```mermaid -flowchart TD - START["開始
使用者提出 ERP 任務"] --> ROLE["任務入口與角色確認
Web / Dashboard / LINE，取得 role"] - ROLE --> LINEQ{"來源是否為 LINE？"} - LINEQ -->|否| WEBROLE["使用登入角色權限
admin / warehouse / sales / hr"] - LINEQ -->|是| LINELIMIT["套用 LINE 任務白名單
僅開放低敏感、非寫入工具"] - WEBROLE --> ORCH["總管 Agent 判斷需求
single / multi / smalltalk"] - LINELIMIT --> ORCH - ORCH --> DISPATCH["派工到專責 Agent
寫入 agent_dispatch_logs"] - DISPATCH --> AGENT["專責 Agent 產生工具呼叫"] - AGENT --> GATE["Tool Gateway 檢核
工具存在、Agent 白名單、角色權限、風險等級"] - GATE --> RISK{"風險等級"} - RISK -->|read_only / suggestion| EXEC["直接執行"] - RISK -->|write / dangerous| PENDING["建立 pending_approvals
等待人工核准"] - PENDING --> APPROVE{"人工審批"} - APPROVE -->|核准| EXEC - APPROVE -->|拒絕| REJECT["拒絕並記錄原因"] - EXEC --> LOG["寫入 agent_action_logs
必要時寫入業務資料"] - REJECT --> LOG - LOG --> RESP["回覆使用者
揭露已執行或尚未執行狀態"] -``` - -### PNG 原圖 - -**目前系統架構圖** - -
-展開架構圖原始 PNG +原始架構圖：[系統架構 PNG](docs/images/erp_current_clean_architecture.png) · [治理流程 PNG](docs/images/erp_current_standard_flowchart.png) -

- AI 風險控管進銷存 ERP 系統架構圖 -

- -[開啟架構圖原始大圖](docs/images/erp_current_clean_architecture.png) +## 快速開始 -
+### 1. 安裝 -**AI 任務治理標準流程圖** +```bash +git clone https://github.com/falltwo/AI-Risk-Based-Inventory-ERP.git +cd AI-Risk-Based-Inventory-ERP -
-展開流程圖原始 PNG +python -m venv .venv +# Windows +.venv\Scripts\activate +# macOS / Linux +# source .venv/bin/activate -

- 目前 ERP 的 AI 任務治理流程圖 -

+pip install -r requirements.txt +``` -[開啟流程圖原始大圖](docs/images/erp_current_standard_flowchart.png) +### 2. 建立本機 Demo 設定 -
+```bash +cp .env.example .env +``` -### 文字版 +在 `.env` 至少設定： +```dotenv +ERP_DEMO_MODE=true +LLM_MODEL=gemini/gemini-2.5-flash +GEMINI_API_KEY=replace_with_your_key ``` -使用者（Web / LINE） - │ 自然語言任務 - ▼ -總管 Agent ─── 語意路由：判斷派給哪個專責 Agent（派工紀錄落庫） - ▼ -8 個專責 Agent ─── 各自僅能使用白名單內的工具 - │ tool call - ▼ -Tool Gateway ─── 白名單 → 角色權限 → 風險分級（呼叫紀錄落庫） - ├── read_only / suggestion ──→ 直接執行 - └── write / dangerous ──→ 待審批 ──→ 人工核准 ──→ 執行（審批歷程落庫） - ▼ -SQLite（業務資料 + 三層稽核紀錄） -``` - -**治理邊界**：治理鏈的對象是「AI Agent 的自主行動」。傳統的人工操作表單（手動開單、記帳、維護主檔）由登入者的角色權限（RBAC）管理，操作者本人即決策者，不重複進審批。 -## 快速開始 +### 3. 啟動 ```bash -# 1. 環境（Python 3.11+） -python -m venv .venv -.venv/Scripts/activate # Windows；macOS/Linux 用 source .venv/bin/activate -pip install -r requirements.txt - -# 2. 設定模型（.env） -cp .env.example .env -# LLM_MODEL=gemini/gemini-2.5-flash ← 填你的供應商/模型與對應金鑰 -# 本機比賽 Demo 才設定 ERP_DEMO_MODE=true（會建立並顯示已知測試帳密） - -# 3. 啟動 streamlit run app.py ``` -登入後左側選單進入「AI 智能助理」即可用自然語言操作；「Agent Dashboard」檢視派工、稽核與待審批。 +Demo 模式才會建立並顯示 `viewer`、`planner`、`approver` 等已知測試帳密。**只能在本機展示使用，不得開放至公網。** -當且僅當 `.env` 明確設定 `ERP_DEMO_MODE=true` 時，系統才會建立並顯示測試帳號。此模式只供本機比賽展示，不得用於公開部署。 +## 重要設定 -若某個既有資料庫曾以 Demo 模式初始化，之後把旗標改回 `false` 不會自動刪除帳號；公開或正式部署前必須改用乾淨資料庫，或由管理者移除／輪替所有測試帳密。現階段的 L1/L2/L3 權限模型是單一組織、本機展示邊界，尚未提供多租戶資料列隔離或外部 IAM/SSO，不能直接當成網路服務的正式身分系統。 +| 環境變數 | 用途 | 預設／要求 | +|---|---|---| +| `ERP_DEMO_MODE` | 建立合成資料與 Demo 帳號 | `false`；僅限本機 | +| `ERP_ORGANIZATION_ID` | 綁定此 SQLite DB 所屬組織 | Demo 自動使用 `demo-org`；既有非 Demo DB 必須設定後再配置 membership 與 entitlement | +| `ERP_DB_PATH` | 自訂 SQLite 路徑 | `data/erp.db` | +| `LLM_MODEL` | LiteLLM 主模型 | `gemini/gemini-2.5-flash` | +| `LLM_FALLBACK_MODELS` | 逗號分隔的備援模型 | 見 `.env.example` | +| `LLM_ANALYSIS_MODEL` | 新聞歸類／翻譯等副任務模型 | 未設時沿用主模型鏈 | +| `GEMINI_API_KEY` / `OPENAI_API_KEY` | 對應模型供應商金鑰 | 依模型選擇 | +| `GNEWS_API_KEY` | 供應鏈新聞來源 | 選用 | +| `ERP_SCHEDULER_ACTOR` | 24 小時新聞刷新服務身分 | 未設定時停用 | +| `LINE_CHANNEL_ACCESS_TOKEN` / `LINE_CHANNEL_SECRET` | LINE Bot | 選用 | -### LINE Bot（選用） +## 測試與驗證 ```bash -# .env 需另設 LINE_CHANNEL_ACCESS_TOKEN / LINE_CHANNEL_SECRET / GEMINI_API_KEY -python "line bot/bot_server.py" # FastAPI 於 :8000，webhook 需公開網址（如 ngrok） +pip install -r requirements-dev.txt +python -m pytest -q ``` -## 設定一覽 +v1.0 本機 release verification：**327 passed**。CI 會在每個 PR 自動執行。 -| 環境變數 | 用途 | 預設 | -|---------|------|------| -| `ERP_DEMO_MODE` | 建立合成資料與已知 Demo 帳密；僅限本機展示 | `false` | -| `ERP_SCHEDULER_ACTOR` | 背景新聞刷新使用的 ERP 服務身分；未設定時排程停用 | 未設定 | -| `LINE_RICH_MENU_IMAGE_PATH` | 執行 LINE Rich Menu 設定腳本時使用的本機 PNG 路徑 | 未設定 | -| `LLM_MODEL` | 主模型（LiteLLM 格式 `provider/model`），AI 助理與分析頁共用 | `gemini/gemini-2.5-flash` | -| `LLM_FALLBACK_MODELS` | 備援模型（逗號分隔，主模型失敗時依序切換） | `openai/kimi-k2.6,gemini/gemini-2.5-flash` | -| `LLM_ANALYSIS_MODEL` | 分析副任務別名（選填；新聞歸類/翻譯可指到較便宜模型） | 未設＝用主模型鏈 | -| `OPENAI_API_KEY` / `OPENAI_API_BASE` | OpenAI 相容供應商的金鑰與端點 | — | -| `GEMINI_API_KEY` | Gemini 金鑰（選 gemini 系模型時） | — | -| `GNEWS_API_KEY` | 供應鏈新聞來源（選用） | — | -| `ERP_DB_PATH` | 資料庫路徑（企業可指定既有 .db） | `data/erp.db` | -| `LINE_CHANNEL_ACCESS_TOKEN` / `LINE_CHANNEL_SECRET` | LINE Bot 憑證（選用） | — | +測試包含： -若一般（非採購／ERP 交換）寫入在效果完成後、執行收據落庫前中斷，審批會安全停在 `executing`，系統不會自動重試。處理方式見 [Generic approval reconciliation runbook](docs/generic_approval_reconciliation.md)。 +- L1/L2/L3 導覽與伺服器端授權負向測試 +- 提案人自審、撤權後執行與跨組織拒絕 +- payload／resource version／來源明細／價格竄改拒絕 +- 併發核准、CAS、rollback 與 receipt 冪等重放 +- 同一來源採購明細只能產生一個完整替代 effect +- Demo 種子資料不得產生孤兒採購品項，重播也不得改變已核准來源明細的識別碼 -## 測試 +## 專案結構 -```bash -pip install -r requirements-dev.txt -python -m pytest tests/ -v +```text +backend/ 權限、Agent、Gateway、Proposal、ERP 與資料庫 +frontend/ Streamlit 頁面與 L1/L2/L3 操作介面 +line bot/ FastAPI + LINE Messaging API +scripts/ Demo 種子與維運工具 +tests/ 治理、授權、交易、UI contract 測試 +docs/ 架構圖、runbook 與 Release notes ``` -測試涵蓋治理關鍵路徑：審批狀態揭露、彙整層治理訊號保留、供應商容錯切換。CI 於每個 PR 自動執行。 +## 已知限制 + +- 一個 SQLite DB 只代表一個 organization；不是共享 DB 的 row-level multi-tenancy。 +- 應用層 audit 是 tamper-evident，但不能抵擋擁有主機／資料庫管理權限的人直接改檔。 +- SQLite 原子交易證據不能直接外推到外部 ERP API；跨系統執行仍需要 outbox／worker／對帳策略。 +- Demo 帳號與合成資料不應存在於正式部署；正式環境需另行配置身分、membership、entitlement 與秘密管理。 +- 從早期非 Demo 資料庫升級時，若尚未建立組織邊界，啟動會 fail fast；必須先設定 `ERP_ORGANIZATION_ID`，再配置 `user_organizations` 與 `organization_entitlements`。 + +## 版本 -## 技術組成 +- [v1.0 Releases](https://github.com/falltwo/AI-Risk-Based-Inventory-ERP/releases) +- [v1.0 English release notes](docs/releases/v1.0.md) -Python 3.11 · Streamlit · SQLite · LiteLLM（多供應商模型層）· FastAPI + LINE Messaging API · Plotly +技術組成：Python 3.11 · Streamlit · SQLite · LiteLLM · FastAPI · LINE Messaging API · Plotly diff --git a/backend/access_control.py b/backend/access_control.py index 87b5b1d..e76536d 100644 --- a/backend/access_control.py +++ b/backend/access_control.py @@ -128,6 +128,12 @@ def _load(active_conn: sqlite3.Connection) -> AccessContext | None: if row is None: return None organization_id = row[3] + deployment = active_conn.execute( + "SELECT value FROM app_metadata " + "WHERE key = 'deployment_organization_id'" + ).fetchone() + if deployment is None or deployment[0] != organization_id: + return None entitlement_rows = active_conn.execute( """ SELECT entitlement_key diff --git a/backend/database.py b/backend/database.py index 6632e81..4af17ec 100644 --- a/backend/database.py +++ b/backend/database.py @@ -73,6 +73,39 @@ def init_db(): key TEXT PRIMARY KEY, value TEXT NOT NULL )''') + # PoC deployment boundary: one SQLite database belongs to one organization. + # Production deployments set ERP_ORGANIZATION_ID; the demo is demo-org. + deployment_org = os.environ.get("ERP_ORGANIZATION_ID", "").strip() + demo_mode = is_demo_mode_enabled() + if demo_mode and not deployment_org: + deployment_org = "demo-org" + existing_deployment = c.execute( + "SELECT value FROM app_metadata " + "WHERE key = 'deployment_organization_id'" + ).fetchone() + if ( + not demo_mode + and not deployment_org + and existing_deployment is None + and c.execute("SELECT COUNT(*) FROM users").fetchone()[0] > 0 + ): + conn.close() + raise RuntimeError( + "Existing non-demo database has no organization boundary. " + "Set ERP_ORGANIZATION_ID, then provision user_organizations and " + "organization_entitlements before starting the application." + ) + if deployment_org: + if existing_deployment is not None and existing_deployment[0] != deployment_org: + conn.close() + raise RuntimeError( + "ERP_ORGANIZATION_ID does not match this database deployment" + ) + c.execute( + "INSERT OR IGNORE INTO app_metadata (key, value) " + "VALUES ('deployment_organization_id', ?)", + (deployment_org,), + ) # 進銷存：商品、倉庫 c.execute('''CREATE TABLE IF NOT EXISTS warehouses (warehouse_id TEXT PRIMARY KEY, name TEXT, address TEXT)''') c.execute('''CREATE TABLE IF NOT EXISTS inventory (product_id TEXT PRIMARY KEY, name TEXT, stock INTEGER, price INTEGER, cost REAL, reorder_point INTEGER, baseline_reorder_point INTEGER, daily_sales INTEGER, barcode TEXT, warehouse_id TEXT)''') @@ -190,6 +223,61 @@ def init_db(): created_at TEXT NOT NULL )''') + # L2 purchase proposals are independent business objects. They correlate + # with approval/execution through an adapter-derived operation ID, but do + # not own Gateway metadata, approval state, or an approval identifier. + c.execute('''CREATE TABLE IF NOT EXISTS purchase_proposals ( + proposal_id TEXT PRIMARY KEY, + proposal_type TEXT NOT NULL, + schema_version INTEGER NOT NULL, + organization_id TEXT NOT NULL, + proposer_username TEXT NOT NULL, + proposer_role TEXT NOT NULL, + affected_po_id TEXT NOT NULL, + source_po_item_id INTEGER NOT NULL, + proposed_po_id TEXT NOT NULL UNIQUE, + original_supplier_id TEXT NOT NULL, + alternative_supplier_id TEXT NOT NULL, + alternative_supplier_product_id INTEGER NOT NULL, + product_id TEXT NOT NULL, + qty INTEGER NOT NULL, + unit_price REAL NOT NULL, + currency TEXT NOT NULL, + order_date TEXT NOT NULL, + proposed_status TEXT NOT NULL, + reason TEXT NOT NULL, + estimated_delay_days INTEGER, + source_event_id INTEGER, + source_po_version TEXT NOT NULL, + proposal_digest TEXT NOT NULL, + created_at TEXT NOT NULL + )''') + c.execute('''CREATE INDEX IF NOT EXISTS ix_purchase_proposals_affected_po + ON purchase_proposals(affected_po_id, created_at)''') + + # Unreleased Proposal v1 development databases may predate exact line/price + # identity. New rows always populate these columns; old incomplete rows fail + # closed during integrity validation rather than being guessed. + for column_name, column_type in ( + ("source_po_item_id", "INTEGER"), + ("alternative_supplier_product_id", "INTEGER"), + ): + try: + c.execute( + f"ALTER TABLE purchase_proposals ADD COLUMN {column_name} {column_type}" + ) + except sqlite3.OperationalError: + pass + + # Business-effect claim: several proposals may be compared, but only one + # full replacement PO may commit for a given original PO line. + c.execute('''CREATE TABLE IF NOT EXISTS purchase_proposal_effects ( + source_po_item_id INTEGER PRIMARY KEY, + proposal_id TEXT NOT NULL UNIQUE, + operation_id TEXT NOT NULL UNIQUE, + created_at TEXT NOT NULL + )''') + # L3 ERP CSV exchange: validated staging is kept separate from live PO data. # A stable (source_system, external_id) identifies one external record while # version/content_digest identify the exact revision submitted for approval. diff --git a/backend/l1_monitoring.py b/backend/l1_monitoring.py new file mode 100644 index 0000000..98af2a1 --- /dev/null +++ b/backend/l1_monitoring.py @@ -0,0 +1,145 @@ +"""Read-only helpers for the L1 supply-chain monitoring surface.""" + +from __future__ import annotations + + +def _text(value) -> str: + if value is None: + return "" + normalized = str(value).strip() + if normalized.casefold() in {"nan", "none", ""}: + return "" + return normalized + + +def _location_matches(left, right) -> bool: + left_text = "".join(_text(left).casefold().split()) + right_text = "".join(_text(right).casefold().split()) + if not left_text or not right_text: + return False + if left_text == right_text: + return True + return min(len(left_text), len(right_text)) >= 2 and ( + left_text in right_text or right_text in left_text + ) + + +def _event_matches_supplier(event: dict, supplier: dict) -> bool: + event_country = _text(event.get("country")) + event_region = _text(event.get("region")) + supplier_country = _text(supplier.get("country")) + supplier_region = _text(supplier.get("region")) + + country_matches = _location_matches(event_country, supplier_country) + region_matches = _location_matches(event_region, supplier_region) + + if event_region and supplier_region: + if event_country and supplier_country: + return country_matches and region_matches + return region_matches + if event_country and supplier_country: + return country_matches + return region_matches + + +def _impact_days(event: dict) -> int: + try: + return max(0, int(event.get("impact_days") or 0)) + except (TypeError, ValueError): + return 0 + + +def _event_id(event: dict) -> int: + try: + return int(event.get("id") or 0) + except (TypeError, ValueError): + return 0 + + +def map_purchase_rows_to_events( + purchase_rows: list[dict], + *, + supplier_context: dict[str, dict], + events: list[dict], +) -> list[dict]: + """Enrich imported PO rows with deterministic, non-persistent alert matches.""" + mapped_rows: list[dict] = [] + event_records = [dict(event) for event in events] + + for purchase_row in purchase_rows: + row = dict(purchase_row) + supplier_id = _text(row.get("supplier_id")) + supplier = dict(supplier_context.get(supplier_id) or {}) + country = _text(supplier.get("country")) + region = _text(supplier.get("region")) + risk_level = _text(supplier.get("risk_level")) or "未設定" + row.update( + { + "supplier_country": country or "未設定", + "supplier_region": region or "未設定", + "supplier_risk_level": risk_level, + } + ) + + if not country and not region: + row.update( + { + "match_status": "資料待補", + "matched_event_id": None, + "event_type": "未命中", + "impact_days": 0, + "notification_status": "無法判定", + "notification": ( + f"採購單 {_text(row.get('po_id')) or _text(row.get('external_id'))}：" + f"供應商 {supplier_id or '未設定'} 缺少供應商地區資料，" + "目前無法完成事件對映。" + ), + } + ) + mapped_rows.append(row) + continue + + matches = [ + event + for event in event_records + if _event_matches_supplier(event, supplier) + ] + if not matches: + location = "／".join(part for part in (country, region) if part) + row.update( + { + "match_status": "正常", + "matched_event_id": None, + "event_type": "未命中", + "impact_days": 0, + "notification_status": "無需通知", + "notification": ( + f"採購單 {_text(row.get('po_id')) or _text(row.get('external_id'))}：" + f"供應商 {supplier_id} 位於{location}，未命中目前風險事件。" + ), + } + ) + mapped_rows.append(row) + continue + + matched_event = max(matches, key=lambda event: (_impact_days(event), _event_id(event))) + event_type = _text(matched_event.get("event_type")) or "未分類事件" + impact_days = _impact_days(matched_event) + location = "／".join(part for part in (country, region) if part) + po_reference = _text(row.get("po_id")) or _text(row.get("external_id")) + row.update( + { + "match_status": "需關注", + "matched_event_id": matched_event.get("id"), + "event_type": event_type, + "impact_days": impact_days, + "notification_status": "待人工確認", + "notification": ( + f"採購單 {po_reference}：供應商 {supplier_id} 位於{location}，" + f"命中{event_type}風險，預估延遲 {impact_days} 天。" + ), + } + ) + mapped_rows.append(row) + + return mapped_rows diff --git a/backend/procurement.py b/backend/procurement.py index ed8518a..54104e3 100644 --- a/backend/procurement.py +++ b/backend/procurement.py @@ -61,7 +61,7 @@ def create_purchase_order( executing_approval = conn.execute( """ - SELECT 1 FROM pending_approvals + SELECT requester_username FROM pending_approvals WHERE operation_id = ? AND tool_name = 'create_purchase_order' AND status = 'executing' @@ -73,6 +73,12 @@ def create_purchase_order( raise PermissionError( "create_purchase_order requires a matching executing approval" ) + requester_username = str(executing_approval[0] or "").strip() + from backend.access_control import ERP_EXCHANGE_PROPOSE, load_principal + + requester = load_principal(requester_username, conn=conn) + if requester is None or not requester.can(ERP_EXCHANGE_PROPOSE): + raise PermissionError("purchase order requester no longer has proposal access") if isinstance(note, str): note = note.strip() @@ -89,6 +95,38 @@ def create_purchase_order( if not math.isfinite(unit_price) or unit_price < 0: raise ValueError("unit_price must be a non-negative finite number") + proposal_id = _internal.get("proposal_id") + if requester.role == "supply_planner" and proposal_id is None: + raise PermissionError("L2 purchase order execution requires a bound Proposal") + validated_proposal = None + if proposal_id is not None: + from backend.purchase_proposals import validate_purchase_proposal_execution + + execution_args = { + "po_id": po_id, + "supplier_id": supplier_id, + "product_id": product_id, + "qty": qty, + "unit_price": unit_price, + "order_date": order_date, + "status": status, + "note": note, + } + execution_args.update( + { + key: value + for key, value in _internal.items() + if not str(key).startswith("_") + } + ) + validated_proposal = validate_purchase_proposal_execution( + conn, + operation_id=operation_id, + proposal_id=str(proposal_id), + execution_args=execution_args, + requester_username=requester.username, + ) + supplier = conn.execute( "SELECT is_official FROM suppliers WHERE supplier_id = ?", (supplier_id,), @@ -106,6 +144,19 @@ def create_purchase_order( raise ValueError(f"unknown product_id: {product_id}") total_amount = qty * unit_price + if validated_proposal is not None: + conn.execute( + """ + INSERT INTO purchase_proposal_effects ( + source_po_item_id, proposal_id, operation_id, created_at + ) VALUES (?, ?, ?, datetime('now', 'localtime')) + """, + ( + validated_proposal.source_po_item_id, + validated_proposal.proposal_id, + operation_id, + ), + ) conn.execute( """ INSERT INTO purchase_orders ( diff --git a/backend/purchase_proposals.py b/backend/purchase_proposals.py new file mode 100644 index 0000000..e040a71 --- /dev/null +++ b/backend/purchase_proposals.py @@ -0,0 +1,854 @@ +"""Independent L2 proposal domain plus adapters to the existing Gateway. + +The proposal, approval decision, and execution request are different immutable +objects. The proposal never stores Gateway operation IDs or approval state; +the adapter derives a stable operation ID from the proposal ID when submitting. +""" + +from __future__ import annotations + +from dataclasses import asdict, dataclass +from datetime import date, datetime +import hashlib +import hmac +import json +import math +import re +import sqlite3 +from types import MappingProxyType +from typing import Mapping + +from backend import database +from backend.access_control import ( + APPROVAL_DECIDE, + ERP_EXCHANGE_PROPOSE, + PROPOSAL_EVIDENCE_READ, + load_principal, + require_capability, +) + + +PROPOSAL_TYPE = "alternative_purchase_order" +PROPOSAL_SCHEMA_VERSION = 1 +EXECUTION_CONTRACT_VERSION = "v1" +_OPERATION_PREFIX = "proposal:create-po:" +_PROPOSAL_ID_RE = re.compile(r"^[A-Za-z0-9._-]{1,80}$") + + +@dataclass(frozen=True) +class PurchaseProposal: + proposal_id: str + proposal_type: str + schema_version: int + organization_id: str + proposer_username: str + proposer_role: str + affected_po_id: str + source_po_item_id: int + proposed_po_id: str + original_supplier_id: str + alternative_supplier_id: str + alternative_supplier_product_id: int + product_id: str + qty: int + unit_price: float + currency: str + order_date: str + proposed_status: str + reason: str + estimated_delay_days: int | None + source_event_id: int | None + source_po_version: str + proposal_digest: str + created_at: str + + +@dataclass(frozen=True) +class ApprovalDecision: + proposal_id: str + outcome: str + reason: str = "" + + +@dataclass(frozen=True) +class PurchaseOrderExecutionRequest: + tool_name: str + operation_id: str + contract_version: str + args: Mapping[str, object] + + +_PROPOSAL_COLUMNS = ( + "proposal_id", + "proposal_type", + "schema_version", + "organization_id", + "proposer_username", + "proposer_role", + "affected_po_id", + "source_po_item_id", + "proposed_po_id", + "original_supplier_id", + "alternative_supplier_id", + "alternative_supplier_product_id", + "product_id", + "qty", + "unit_price", + "currency", + "order_date", + "proposed_status", + "reason", + "estimated_delay_days", + "source_event_id", + "source_po_version", + "proposal_digest", + "created_at", +) + + +def _required_text(value, field_name: str) -> str: + text = str(value or "").strip() + if not text: + raise ValueError(f"{field_name} 不可為空白。") + return text + + +def _validate_proposal_id(proposal_id: str) -> str: + proposal_id = _required_text(proposal_id, "proposal_id") + if not _PROPOSAL_ID_RE.fullmatch(proposal_id): + raise ValueError("proposal_id 格式不合法。") + return proposal_id + + +def _canonical_digest(payload: dict) -> str: + canonical = json.dumps( + payload, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ) + return hashlib.sha256(canonical.encode("utf-8")).hexdigest() + + +def _proposal_payload(proposal: PurchaseProposal | dict) -> dict: + payload = asdict(proposal) if isinstance(proposal, PurchaseProposal) else dict(proposal) + payload.pop("proposal_digest", None) + return payload + + +def _proposal_digest(proposal: PurchaseProposal | dict) -> str: + return _canonical_digest(_proposal_payload(proposal)) + + +def proposal_operation_id(proposal_id: str) -> str: + proposal_id = _validate_proposal_id(proposal_id) + return f"{_OPERATION_PREFIX}{proposal_id}:{EXECUTION_CONTRACT_VERSION}" + + +def _proposal_id_from_operation_id(operation_id: str) -> str: + operation_id = _required_text(operation_id, "operation_id") + suffix = f":{EXECUTION_CONTRACT_VERSION}" + if not operation_id.startswith(_OPERATION_PREFIX) or not operation_id.endswith(suffix): + raise ValueError("operation_id 不是受支援的採購提案操作。") + proposal_id = operation_id[len(_OPERATION_PREFIX) : -len(suffix)] + return _validate_proposal_id(proposal_id) + + +def _source_po_snapshot( + conn: sqlite3.Connection, + affected_po_id: str, + product_id: str, + source_po_item_id: int | None = None, +) -> dict: + conn.row_factory = sqlite3.Row + where = "p.po_id = ? AND i.product_id = ?" + params: list[object] = [affected_po_id, product_id] + if source_po_item_id is not None: + if isinstance(source_po_item_id, bool) or int(source_po_item_id) <= 0: + raise ValueError("source_po_item_id 格式錯誤。") + where += " AND i.id = ?" + params.append(int(source_po_item_id)) + rows = conn.execute( + """ + SELECT i.id AS source_po_item_id, + p.po_id, p.supplier_id, p.order_date, p.status, + p.estimated_delay_days, i.product_id, i.qty, i.unit_price + FROM purchase_orders p + JOIN purchase_order_items i ON i.po_id = p.po_id + WHERE """ + + where + + """ + ORDER BY i.id + """, + tuple(params), + ).fetchall() + if not rows: + raise ValueError("找不到受影響採購單或指定品項。") + if source_po_item_id is None and len(rows) != 1: + raise ValueError("採購單含多筆同品項明細，必須指定 source_po_item_id。") + row = rows[0] + if str(row["status"] or "").strip() in {"已完成", "已取消"}: + raise ValueError("受影響採購單已結案，不能建立替代提案。") + return dict(row) + + +def source_po_version( + conn: sqlite3.Connection, + affected_po_id: str, + product_id: str, + source_po_item_id: int | None = None, +) -> str: + return "sha256:" + _canonical_digest( + _source_po_snapshot( + conn, affected_po_id, product_id, source_po_item_id + ) + ) + + +def _row_to_proposal(row: sqlite3.Row | tuple) -> PurchaseProposal: + values = dict(row) if isinstance(row, sqlite3.Row) else dict(zip(_PROPOSAL_COLUMNS, row)) + values["schema_version"] = int(values["schema_version"]) + if values["source_po_item_id"] is not None: + values["source_po_item_id"] = int(values["source_po_item_id"]) + if values["alternative_supplier_product_id"] is not None: + values["alternative_supplier_product_id"] = int( + values["alternative_supplier_product_id"] + ) + values["qty"] = int(values["qty"]) + values["unit_price"] = float(values["unit_price"]) + if values["estimated_delay_days"] is not None: + values["estimated_delay_days"] = int(values["estimated_delay_days"]) + if values["source_event_id"] is not None: + values["source_event_id"] = int(values["source_event_id"]) + return PurchaseProposal(**values) + + +def _load_proposal_with_conn( + conn: sqlite3.Connection, proposal_id: str +) -> PurchaseProposal | None: + conn.row_factory = sqlite3.Row + row = conn.execute( + f"SELECT {', '.join(_PROPOSAL_COLUMNS)} FROM purchase_proposals WHERE proposal_id = ?", + (_validate_proposal_id(proposal_id),), + ).fetchone() + return _row_to_proposal(row) if row is not None else None + + +def _assert_proposal_integrity(proposal: PurchaseProposal) -> None: + if proposal.proposal_type != PROPOSAL_TYPE: + raise ValueError("不支援的提案類型。") + if proposal.schema_version != PROPOSAL_SCHEMA_VERSION: + raise ValueError("不支援的提案 schema 版本。") + if not hmac.compare_digest(proposal.proposal_digest, _proposal_digest(proposal)): + raise PermissionError("採購提案內容完整性驗證失敗。") + + +def prepare_alternative_purchase_proposal( + *, + affected_po_id: str, + product_id: str, + source_po_item_id: int | None = None, + alternative_supplier_id: str, + alternative_supplier_product_id: int | None = None, + reason: str, + actor: str, + proposal_id: str, + estimated_delay_days: int | None = None, + source_event_id: int | None = None, +) -> PurchaseProposal: + """Build one immutable proposal from server-side ERP master data.""" + require_capability(actor, ERP_EXCHANGE_PROPOSE) + principal = load_principal(actor) + if principal is None: + raise PermissionError("提案人身分無效。") + + proposal_id = _validate_proposal_id(proposal_id) + affected_po_id = _required_text(affected_po_id, "affected_po_id") + product_id = _required_text(product_id, "product_id") + alternative_supplier_id = _required_text( + alternative_supplier_id, "alternative_supplier_id" + ) + if source_po_item_id is not None: + if isinstance(source_po_item_id, bool) or int(source_po_item_id) <= 0: + raise ValueError("source_po_item_id 格式錯誤。") + source_po_item_id = int(source_po_item_id) + if alternative_supplier_product_id is not None: + if ( + isinstance(alternative_supplier_product_id, bool) + or int(alternative_supplier_product_id) <= 0 + ): + raise ValueError("alternative_supplier_product_id 格式錯誤。") + alternative_supplier_product_id = int(alternative_supplier_product_id) + reason = _required_text(reason, "reason") + if len(reason) > 1000: + raise ValueError("reason 不可超過 1000 字。") + if estimated_delay_days is not None: + if isinstance(estimated_delay_days, bool): + raise ValueError("estimated_delay_days 格式錯誤。") + estimated_delay_days = int(estimated_delay_days) + if not 0 <= estimated_delay_days <= 3650: + raise ValueError("estimated_delay_days 超出允許範圍。") + if source_event_id is not None: + source_event_id = int(source_event_id) + + with database.transaction() as conn: + existing = _load_proposal_with_conn(conn, proposal_id) + if existing is not None: + _assert_proposal_integrity(existing) + same_request = ( + hmac.compare_digest(existing.proposer_username, principal.username) + and hmac.compare_digest( + existing.organization_id, principal.organization_id + ) + and existing.affected_po_id == affected_po_id + and existing.product_id == product_id + and existing.alternative_supplier_id == alternative_supplier_id + and existing.reason == reason + and existing.estimated_delay_days == estimated_delay_days + and existing.source_event_id == source_event_id + ) + if source_po_item_id is not None: + same_request = same_request and ( + existing.source_po_item_id == source_po_item_id + ) + if alternative_supplier_product_id is not None: + same_request = same_request and ( + existing.alternative_supplier_product_id + == alternative_supplier_product_id + ) + if not same_request: + raise ValueError("proposal_id 已綁定另一份採購提案。") + return existing + + source = _source_po_snapshot( + conn, affected_po_id, product_id, source_po_item_id + ) + source_po_item_id = int(source["source_po_item_id"]) + original_supplier_id = str(source["supplier_id"]) + if hmac.compare_digest(original_supplier_id, alternative_supplier_id): + raise ValueError("替代供應商必須與原供應商不同。") + alternative_where = "s.supplier_id = ? AND sp.product_id = ?" + alternative_params: list[object] = [alternative_supplier_id, product_id] + if alternative_supplier_product_id is not None: + alternative_where += " AND sp.id = ?" + alternative_params.append(alternative_supplier_product_id) + alternative_rows = conn.execute( + """ + SELECT sp.id AS supplier_product_id, s.is_official, sp.price + FROM suppliers s + JOIN supplier_products sp ON sp.supplier_id = s.supplier_id + WHERE """ + + alternative_where + + " ORDER BY sp.id", + tuple(alternative_params), + ).fetchall() + if not alternative_rows: + raise ValueError("替代供應商未提供指定品項。") + if alternative_supplier_product_id is None and len(alternative_rows) != 1: + raise ValueError( + "替代供應商有多筆同品項報價，必須指定 supplier_product_id。" + ) + alternative = alternative_rows[0] + alternative_supplier_product_id = int(alternative["supplier_product_id"]) + if int(alternative["is_official"] or 0) != 1: + raise PermissionError("替代供應商不是有效的正式供應商。") + unit_price = float(alternative["price"]) + if not math.isfinite(unit_price) or unit_price < 0: + raise ValueError("供應商主檔單價格式錯誤。") + if source_event_id is not None: + if conn.execute( + "SELECT 1 FROM supply_chain_events WHERE id = ?", (source_event_id,) + ).fetchone() is None: + raise ValueError("找不到來源風險事件。") + source_version = f"sha256:{_canonical_digest(source)}" + + stable_suffix = hashlib.sha256(proposal_id.encode("utf-8")).hexdigest()[:16] + values = { + "proposal_id": proposal_id, + "proposal_type": PROPOSAL_TYPE, + "schema_version": PROPOSAL_SCHEMA_VERSION, + "organization_id": principal.organization_id, + "proposer_username": principal.username, + "proposer_role": principal.role, + "affected_po_id": affected_po_id, + "source_po_item_id": source_po_item_id, + "proposed_po_id": f"ALT-{stable_suffix}", + "original_supplier_id": original_supplier_id, + "alternative_supplier_id": alternative_supplier_id, + "alternative_supplier_product_id": alternative_supplier_product_id, + "product_id": product_id, + "qty": int(source["qty"]), + "unit_price": unit_price, + "currency": "TWD", + "order_date": date.today().isoformat(), + "proposed_status": "待入庫", + "reason": reason, + "estimated_delay_days": estimated_delay_days, + "source_event_id": source_event_id, + "source_po_version": source_version, + "created_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), + } + values["proposal_digest"] = _canonical_digest(values) + return PurchaseProposal(**values) + + +def proposal_to_execution_request( + proposal: PurchaseProposal, +) -> PurchaseOrderExecutionRequest: + """Map proposal payload to the unchanged create-PO Gateway contract.""" + _assert_proposal_integrity(proposal) + args = { + "po_id": proposal.proposed_po_id, + "supplier_id": proposal.alternative_supplier_id, + "product_id": proposal.product_id, + "qty": proposal.qty, + "unit_price": proposal.unit_price, + "order_date": proposal.order_date, + "status": proposal.proposed_status, + "note": ( + f"替代採購提案 {proposal.proposal_id}；" + f"受影響採購單 {proposal.affected_po_id}" + ), + "proposal_id": proposal.proposal_id, + "proposal_digest": proposal.proposal_digest, + "affected_po_id": proposal.affected_po_id, + "source_po_item_id": proposal.source_po_item_id, + "alternative_supplier_product_id": ( + proposal.alternative_supplier_product_id + ), + "source_po_version": proposal.source_po_version, + } + return PurchaseOrderExecutionRequest( + tool_name="create_purchase_order", + operation_id=proposal_operation_id(proposal.proposal_id), + contract_version=EXECUTION_CONTRACT_VERSION, + args=MappingProxyType(args), + ) + + +def _validate_current_proposal( + conn: sqlite3.Connection, proposal: PurchaseProposal +) -> None: + _assert_proposal_integrity(proposal) + source = _source_po_snapshot( + conn, + proposal.affected_po_id, + proposal.product_id, + proposal.source_po_item_id, + ) + current_source = "sha256:" + _canonical_digest(source) + if not hmac.compare_digest(proposal.source_po_version, current_source): + raise PermissionError("受影響採購單已變更，請重新建立提案。") + if not hmac.compare_digest( + proposal.original_supplier_id, str(source["supplier_id"]) + ): + raise PermissionError("原供應商與受影響採購單不一致。") + if isinstance(proposal.qty, bool) or proposal.qty != int(source["qty"]): + raise PermissionError("提案數量與受影響採購明細不一致。") + expected_po_id = ( + "ALT-" + hashlib.sha256(proposal.proposal_id.encode("utf-8")).hexdigest()[:16] + ) + if not hmac.compare_digest(proposal.proposed_po_id, expected_po_id): + raise PermissionError("替代採購單編號不符合伺服器命名規則。") + if hmac.compare_digest( + proposal.original_supplier_id, proposal.alternative_supplier_id + ): + raise PermissionError("替代供應商必須與原供應商不同。") + if proposal.currency != "TWD": + raise PermissionError("替代採購提案幣別必須為 TWD。") + if proposal.proposed_status != "待入庫": + raise PermissionError("替代採購提案狀態必須為待入庫。") + if not str(proposal.reason or "").strip() or len(proposal.reason) > 1000: + raise PermissionError("提案理由格式不合法。") + if proposal.estimated_delay_days is not None and ( + isinstance(proposal.estimated_delay_days, bool) + or not 0 <= proposal.estimated_delay_days <= 3650 + ): + raise PermissionError("預估延誤天數超出允許範圍。") + if proposal.source_event_id is not None and conn.execute( + "SELECT 1 FROM supply_chain_events WHERE id = ?", + (proposal.source_event_id,), + ).fetchone() is None: + raise PermissionError("找不到來源風險事件。") + supplier = conn.execute( + """ + SELECT s.is_official, sp.price + FROM suppliers s + JOIN supplier_products sp ON sp.supplier_id = s.supplier_id + WHERE s.supplier_id = ? AND sp.product_id = ? AND sp.id = ? + """, + ( + proposal.alternative_supplier_id, + proposal.product_id, + proposal.alternative_supplier_product_id, + ), + ).fetchone() + if supplier is None or int(supplier[0] or 0) != 1: + raise PermissionError("替代供應商不是有效的正式供應商。") + if float(supplier[1]) != proposal.unit_price: + raise PermissionError("替代供應商報價已變更，請重新建立提案。") + claimed = conn.execute( + "SELECT proposal_id FROM purchase_proposal_effects " + "WHERE source_po_item_id = ?", + (proposal.source_po_item_id,), + ).fetchone() + if claimed is not None and not hmac.compare_digest( + str(claimed[0]), proposal.proposal_id + ): + raise PermissionError("此來源採購明細已有另一份替代提案完成執行。") + + +def _persist_proposal(proposal: PurchaseProposal) -> PurchaseProposal: + with database.transaction(immediate=True) as conn: + _validate_current_proposal(conn, proposal) + existing = _load_proposal_with_conn(conn, proposal.proposal_id) + if existing is not None: + _assert_proposal_integrity(existing) + if not hmac.compare_digest( + existing.proposal_digest, proposal.proposal_digest + ): + raise ValueError("proposal_id 已綁定另一份採購提案。") + return existing + try: + proposed_order_date = date.fromisoformat(proposal.order_date) + datetime.strptime(proposal.created_at, "%Y-%m-%d %H:%M:%S") + except (TypeError, ValueError) as exc: + raise PermissionError("提案日期或建立時間格式不合法。") from exc + if proposed_order_date != date.today(): + raise PermissionError("新提案的採購日期必須為今日。") + placeholders = ", ".join("?" for _ in _PROPOSAL_COLUMNS) + conn.execute( + f"INSERT INTO purchase_proposals ({', '.join(_PROPOSAL_COLUMNS)}) VALUES ({placeholders})", + tuple(getattr(proposal, column) for column in _PROPOSAL_COLUMNS), + ) + return proposal + + +def submit_purchase_proposal(proposal: PurchaseProposal, *, actor: str): + """Persist a proposal, then submit its execution request for approval.""" + require_capability(actor, ERP_EXCHANGE_PROPOSE) + principal = load_principal(actor) + if principal is None or not hmac.compare_digest( + principal.username, proposal.proposer_username + ): + raise PermissionError("只有原提案人可以送出此提案。") + if principal.role != proposal.proposer_role: + raise PermissionError("提案人的角色已變更，請重新建立提案。") + if not hmac.compare_digest( + principal.organization_id, proposal.organization_id + ): + raise PermissionError("提案人的組織已變更，請重新建立提案。") + durable = _persist_proposal(proposal) + request = proposal_to_execution_request(durable) + + from backend.tool_gateway import gateway + + return gateway.call( + request.tool_name, + dict(request.args), + role=principal.role, + actor=principal.username, + agent_name="procurement_agent", + operation_id=request.operation_id, + ) + + +def validate_purchase_proposal_execution( + conn: sqlite3.Connection, + *, + operation_id: str, + proposal_id: str, + execution_args: Mapping[str, object], + requester_username: str | None = None, +) -> PurchaseProposal: + """Verify immutable proposal evidence and live source preconditions.""" + proposal = _load_proposal_with_conn(conn, proposal_id) + if proposal is None: + raise PermissionError("找不到採購提案。") + _assert_proposal_integrity(proposal) + if requester_username is not None: + requester = load_principal(requester_username, conn=conn) + if requester is None or not requester.can(ERP_EXCHANGE_PROPOSE): + raise PermissionError("提案人的即時權限已失效。") + if not hmac.compare_digest( + requester.username, proposal.proposer_username + ) or not hmac.compare_digest( + requester.organization_id, proposal.organization_id + ): + raise PermissionError("目前提案人不是此 Proposal 的擁有者。") + if not hmac.compare_digest( + proposal_operation_id(proposal.proposal_id), str(operation_id or "") + ): + raise PermissionError("執行操作與採購提案不一致。") + expected = proposal_to_execution_request(proposal).args + if set(execution_args) != set(expected): + raise PermissionError("執行請求欄位集合與提案不一致。") + for field in expected: + if execution_args.get(field) != expected.get(field): + raise PermissionError(f"執行請求欄位 {field} 與提案不一致。") + _validate_current_proposal(conn, proposal) + return proposal + + +def validate_purchase_proposal_gateway_request( + *, + operation_id: str, + proposal_id: str, + execution_args: Mapping[str, object], + actor: str, +) -> None: + """Reload and verify a planner's canonical Proposal at the Gateway boundary.""" + with database.transaction() as conn: + validate_purchase_proposal_execution( + conn, + operation_id=operation_id, + proposal_id=proposal_id, + execution_args=execution_args, + requester_username=actor, + ) + + +def validate_purchase_proposal_decision_scope( + conn: sqlite3.Connection, *, proposal_id: str, actor: str +) -> PurchaseProposal: + """Enforce Proposal organization scope inside a Gateway decision transaction.""" + principal = load_principal(actor, conn=conn) + if principal is None or not principal.can(APPROVAL_DECIDE): + raise PermissionError("決策者目前不具採購核准權限。") + proposal = _load_proposal_with_conn(conn, proposal_id) + if proposal is None: + raise PermissionError("找不到採購提案。") + _assert_proposal_integrity(proposal) + if not hmac.compare_digest( + principal.organization_id, proposal.organization_id + ): + raise PermissionError("不可處理其他組織的採購提案。") + return proposal + + +def get_purchase_proposal_evidence( + proposal_id: str, *, actor: str +) -> PurchaseProposal | None: + principal = load_principal(actor) + if principal is None: + raise PermissionError("讀取提案證據需要有效身分。") + with database.transaction() as conn: + proposal = _load_proposal_with_conn(conn, proposal_id) + if proposal is None: + return None + if not hmac.compare_digest( + principal.organization_id, proposal.organization_id + ): + raise PermissionError("不可讀取其他組織的採購提案。") + can_review = principal.can(PROPOSAL_EVIDENCE_READ) + owns_proposal = principal.can(ERP_EXCHANGE_PROPOSE) and hmac.compare_digest( + principal.username, proposal.proposer_username + ) + if not (can_review or owns_proposal): + raise PermissionError("沒有讀取此提案證據的權限。") + _assert_proposal_integrity(proposal) + return proposal + + +def get_purchase_proposal_for_operation( + operation_id: str | None, *, actor: str +) -> PurchaseProposal | None: + """Resolve proposal evidence for a Gateway operation; legacy rows return None.""" + if not operation_id: + return None + try: + proposal_id = _proposal_id_from_operation_id(operation_id) + except ValueError: + return None + return get_purchase_proposal_evidence(proposal_id, actor=actor) + + +def _load_bound_approval(proposal: PurchaseProposal) -> dict | None: + rows = database.run_query( + """ + SELECT approval_id, tool_name, parameters, requester_username, status, + approver, created_at, updated_at, reason + FROM pending_approvals WHERE operation_id = ? + """, + (proposal_operation_id(proposal.proposal_id),), + ) + if not rows: + return None + row = rows[0] + return { + "approval_id": row[0], + "tool_name": row[1], + "parameters": row[2], + "requester_username": row[3], + "status": row[4], + "approver": row[5], + "created_at": row[6], + "updated_at": row[7], + "reason": row[8], + } + + +def decide_purchase_proposal(decision: ApprovalDecision, *, actor: str): + """Apply an L3 decision without placing PO args inside the decision DTO.""" + require_capability(actor, APPROVAL_DECIDE) + proposal = get_purchase_proposal_evidence(decision.proposal_id, actor=actor) + if proposal is None: + raise ValueError("找不到採購提案。") + approval = _load_bound_approval(proposal) + if approval is None: + raise ValueError("採購提案尚未送交 Gateway 審批。") + execution = proposal_to_execution_request(proposal) + try: + approval_args = json.loads(approval["parameters"]) + except (TypeError, json.JSONDecodeError) as exc: + raise PermissionError("審批執行內容已損壞。") from exc + if ( + approval["tool_name"] != execution.tool_name + or approval["requester_username"] != proposal.proposer_username + or approval_args != dict(execution.args) + ): + raise PermissionError("審批內容與採購提案不一致。") + + outcome = str(decision.outcome or "").strip().lower() + from backend.tool_gateway import gateway + + if outcome == "approve": + return gateway.approve_action(approval["approval_id"], approver=actor) + if outcome == "reject": + reason = _required_text(decision.reason, "拒絕原因") + return gateway.reject_action( + approval["approval_id"], reason, approver=actor + ) + raise ValueError("outcome 必須是 approve 或 reject。") + + +def get_purchase_operation_timeline(operation_id: str, *, actor: str) -> list[dict]: + """Return a redacted, correlation-based L3 audit timeline.""" + require_capability(actor, PROPOSAL_EVIDENCE_READ) + proposal_id = _proposal_id_from_operation_id(operation_id) + proposal = get_purchase_proposal_evidence(proposal_id, actor=actor) + if proposal is None: + return [] + approval = _load_bound_approval(proposal) + receipts = database.run_query( + "SELECT receipt_id, approval_id, created_at FROM effect_receipts WHERE operation_id = ?", + (operation_id,), + ) + + events = [ + { + "kind": "proposal_created", + "operation_id": operation_id, + "time": proposal.created_at, + "actor": proposal.proposer_username, + "proposal_id": proposal.proposal_id, + "summary": ( + f"{proposal.affected_po_id} -> {proposal.proposed_po_id}; " + f"{proposal.original_supplier_id} -> {proposal.alternative_supplier_id}" + ), + } + ] + if approval is not None: + events.append( + { + "kind": "approval_submitted", + "operation_id": operation_id, + "time": approval["created_at"], + "actor": approval["requester_username"], + "approval_id": approval["approval_id"], + "summary": f"approval status: {approval['status']}", + } + ) + if approval["status"] == "rejected": + events.append( + { + "kind": "approval_rejected", + "operation_id": operation_id, + "time": approval["updated_at"], + "actor": approval["approver"], + "approval_id": approval["approval_id"], + "summary": "Proposal rejected by an authorized reviewer.", + } + ) + if receipts: + receipt = receipts[0] + events.append( + { + "kind": "execution_completed", + "operation_id": operation_id, + "time": receipt[2], + "actor": approval["approver"] if approval else None, + "approval_id": receipt[1], + "receipt_id": receipt[0], + "summary": "Gateway committed one ERP effect and receipt.", + } + ) + return events + + +def list_impacted_purchase_options(*, actor: str) -> list[dict]: + """Return structured open PO lines for the L2 proposal workbench.""" + require_capability(actor, ERP_EXCHANGE_PROPOSE) + with database.transaction() as conn: + conn.row_factory = sqlite3.Row + rows = conn.execute( + """ + SELECT i.id AS source_po_item_id, + p.po_id, p.supplier_id AS original_supplier_id, + s.name AS supplier_name, s.country, s.region, + p.estimated_delay_days, p.alternative_suggestion, + i.product_id, inv.name AS product_name, i.qty, i.unit_price + FROM purchase_orders p + JOIN suppliers s ON s.supplier_id = p.supplier_id + JOIN purchase_order_items i ON i.po_id = p.po_id + LEFT JOIN inventory inv ON inv.product_id = i.product_id + WHERE (p.status IS NULL OR p.status NOT IN ('已完成', '已取消')) + AND ( + p.estimated_delay_days IS NOT NULL + OR TRIM(COALESCE(p.alternative_suggestion, '')) <> '' + ) + ORDER BY COALESCE(p.estimated_delay_days, 0) DESC, p.po_id, i.id + """ + ).fetchall() + result = [] + for row in rows: + item = dict(row) + item["source_po_version"] = source_po_version( + conn, + item["po_id"], + item["product_id"], + item["source_po_item_id"], + ) + result.append(item) + return result + + +def list_alternative_suppliers( + *, + affected_po_id: str, + product_id: str, + actor: str, + source_po_item_id: int | None = None, +) -> list[dict]: + """Return active supplier-master candidates for one affected PO line.""" + require_capability(actor, ERP_EXCHANGE_PROPOSE) + with database.transaction() as conn: + conn.row_factory = sqlite3.Row + source = _source_po_snapshot( + conn, affected_po_id, product_id, source_po_item_id + ) + rows = conn.execute( + """ + SELECT sp.id AS supplier_product_id, + s.supplier_id, s.name, s.country, s.region, s.risk_level, + sp.price, sp.carbon_factor + FROM suppliers s + JOIN supplier_products sp ON sp.supplier_id = s.supplier_id + WHERE sp.product_id = ? AND s.is_official = 1 + AND s.supplier_id <> ? + ORDER BY COALESCE(s.risk_level, ''), sp.price, s.supplier_id + """, + (product_id, source["supplier_id"]), + ).fetchall() + return [dict(row) for row in rows] diff --git a/backend/tool_classification.py b/backend/tool_classification.py index 518d217..8741535 100644 --- a/backend/tool_classification.py +++ b/backend/tool_classification.py @@ -125,7 +125,7 @@ "create_purchase_order": { "module": "procurement", "risk_level": "write", - "allowed_roles": ["admin", "warehouse"], + "allowed_roles": ["admin", "warehouse", "supply_planner"], "description": "建立採購單與明細，核准後才會寫入資料庫", }, "sync_external_purchase_order": { diff --git a/backend/tool_gateway.py b/backend/tool_gateway.py index b5b4500..312ea0f 100644 --- a/backend/tool_gateway.py +++ b/backend/tool_gateway.py @@ -437,6 +437,32 @@ def call( return GatewayResult(status="denied", message=msg) actor = principal.username + # The L2 planner role may create a PO only through a durable, + # canonical PurchaseProposal. Admin/warehouse keep the existing + # manual protected-PO path for backward compatibility. + if tool_name == "create_purchase_order" and principal.role == "supply_planner": + proposal_id = str(args.get("proposal_id") or "").strip() + operation_id = str(operation_id or "").strip() + if not proposal_id: + msg = "L2 採購建立必須綁定已保存的 Proposal。" + _write_log(tool_name, args, role, msg, success=False) + return GatewayResult(status="denied", message=msg) + try: + from backend.purchase_proposals import ( + validate_purchase_proposal_gateway_request, + ) + + validate_purchase_proposal_gateway_request( + operation_id=operation_id, + proposal_id=proposal_id, + execution_args=args, + actor=actor, + ) + except (PermissionError, ValueError) as exc: + msg = f"L2 採購提案驗證失敗：{exc}" + _write_log(tool_name, args, role, msg, success=False) + return GatewayResult(status="denied", message=msg) + # Step 5：依風險等級決定行為 risk_level = registry.get_risk_level(tool_name) @@ -892,6 +918,27 @@ def _approve_purchase_order( status="error", message="審批內容完整性驗證失敗，操作未執行。" ) + if expected_tool == "create_purchase_order": + proposal_id = str(args.get("proposal_id") or "").strip() + if str(operation_id).startswith("proposal:create-po:"): + if not proposal_id: + return GatewayResult( + status="denied", + message="採購提案審批缺少 Proposal 綁定。", + ) + try: + from backend.purchase_proposals import ( + validate_purchase_proposal_decision_scope, + ) + + validate_purchase_proposal_decision_scope( + conn, + proposal_id=proposal_id, + actor=approver_username, + ) + except (PermissionError, ValueError) as exc: + return GatewayResult(status="denied", message=str(exc)) + receipt = conn.execute( """ SELECT result, approval_id, payload_digest @@ -1120,7 +1167,7 @@ def _reject_purchase_order( row = conn.execute( """ SELECT tool_name, parameters, requester, status, version, - requester_username + requester_username, operation_id FROM pending_approvals WHERE approval_id = ? """, (approval_id,), @@ -1164,6 +1211,37 @@ def _reject_purchase_order( status="error", message=f"該審批項目的狀態為 {row[3]}，無法拒絕。", ) + try: + args = json.loads(row[1]) + except (TypeError, json.JSONDecodeError): + return GatewayResult( + status="error", message="審批參數格式已損壞，無法拒絕。" + ) + if not isinstance(args, dict): + return GatewayResult( + status="error", message="審批參數格式不合法，無法拒絕。" + ) + proposal_id = str(args.get("proposal_id") or "").strip() + if row[0] == "create_purchase_order" and str(row[6] or "").startswith( + "proposal:create-po:" + ): + if not proposal_id: + return GatewayResult( + status="denied", + message="採購提案審批缺少 Proposal 綁定。", + ) + try: + from backend.purchase_proposals import ( + validate_purchase_proposal_decision_scope, + ) + + validate_purchase_proposal_decision_scope( + conn, + proposal_id=proposal_id, + actor=approver_username, + ) + except (PermissionError, ValueError) as exc: + return GatewayResult(status="denied", message=str(exc)) recorded_reason = ( f"[legacy originator unavailable] {reason}" if legacy_originator_missing @@ -1180,7 +1258,6 @@ def _reject_purchase_order( approval_context=_PROTECTED_APPROVAL_CONTEXT, ): raise RuntimeError("審批狀態競態，拒絕未生效。") - args = json.loads(row[1]) msg = ( f"操作遭管理者「{approver_username}」拒絕，原因：{recorded_reason}。" "該工具執行已作廢。" diff --git a/docs/releases/v1.0.md b/docs/releases/v1.0.md new file mode 100644 index 0000000..44fd6eb --- /dev/null +++ b/docs/releases/v1.0.md @@ -0,0 +1,82 @@ +# v1.0 — Governed Supply-Chain Decision Loop + +v1.0 turns the governance harness established in v0.1 into an end-to-end supply-chain decision product. + +The system can now move from external risk intelligence to an affected procurement line, create an AI-assisted alternative-sourcing Proposal, request a distinct human decision, and commit exactly one traceable ERP effect after approval. + +## Highlights + +### Three product tiers with real separation of duties + +- **L1 Risk Observer** — risk KPIs, heatmap, alerts, read-only CSV mapping, and notification preview. L1 cannot create proposals or modify ERP data. +- **L2 Intelligence & Decision** — impact analysis, What-if simulation, alternative-supplier comparison, and durable Proposal submission. L2 cannot approve or execute ERP writes. +- **L3 Approval & Execution** — immutable proposal evidence, approve/reject controls, governed Gateway execution, and an end-to-end audit timeline. The proposer cannot self-approve. + +### Durable Proposal-to-ERP workflow + +- Separate `PurchaseProposal`, `ApprovalDecision`, and `PurchaseOrderExecutionRequest` domain objects. +- Exact binding to the affected PO line and selected supplier-price row. +- Canonical payload digest and stable operation identity. +- Live requester-capability and organization-scope revalidation at execution time. +- One full replacement effect per source procurement line. + +### Atomic and idempotent execution + +- Protected purchase approval uses one SQLite transaction for CAS state transition, ERP write, effect claim, execution receipt, audit record, and terminal approval state. +- Replaying the same approved operation returns the existing receipt without creating a second purchase order. +- Cross-organization access, self-approval, stale permissions, and tampered evidence fail closed. + +### L1 workflow completion + +- Added recent-event alerts and read-only procurement CSV mapping. +- Mapping and notification preview remain in memory and never create ERP or proposal records. +- Demo seed data now guarantees that referenced purchase items exist in the product master. + +### Documentation and developer experience + +- New Traditional Chinese and English README files. +- Clear v0.1-to-v1.0 comparison and honest PoC boundaries. +- Improved local Demo Mode instructions and tier-account walkthrough. +- Failed approval decisions remain visible instead of being erased by an immediate UI rerun. + +## From v0.1 to v1.0 + +| Area | v0.1 | v1.0 | +|---|---|---| +| Product focus | Governance harness and bypass closure | Governed supply-chain decision loop | +| Workflow | Governed tools and generic approval | Observe → Recommend → Propose → Approve → Execute | +| Supply-chain handoff | Analysis and recommendations were separate from execution | Affected PO lines become durable alternative-purchase Proposals | +| Approval evidence | Generic action payload and status | Source PO, supplier change, quantity, price, reason, digest, and timeline | +| Execution integrity | Gateway, hash-chain audit, transaction baseline | Exact source identity, live revalidation, effect uniqueness, idempotent receipt | +| Test suite | 55 passing tests reported by the v0.1 release | **327 passing tests** in v1.0 release verification | +| Documentation | Chinese README and diagrams | Bilingual README, release comparison, and explicit limitations | + +The comparison is based on the v0.1 release record retained by the maintainer. That milestone remains the historical governance baseline for this release. + +## Verification + +- **327 automated tests passed** locally before release. +- Python compile verification completed successfully. +- End-to-end browser walkthrough completed with the L1, L2, and L3 demo accounts. +- L2 submission produced one Proposal and one pending approval with zero ERP effects. +- L3 approval produced one purchase order, one proposal-effect claim, and one execution receipt. +- Replaying the same approval kept all three effect counts at exactly one. + +## Upgrade notes + +- Proposal and effect-claim tables are added without deleting existing business data. However, an older non-demo database with users but no organization boundary now fails fast instead of silently locking out every principal. +- Demo deployments automatically use organization `demo-org`. +- For an existing non-demo database, set `ERP_ORGANIZATION_ID`, then provision `user_organizations` and `organization_entitlements` before startup. +- Use a clean database when moving from local Demo Mode to a non-demo deployment; known demo credentials must never be exposed publicly. + +## Known limitations + +- v1.0 is a competition and research PoC, not a production authorization service. +- One SQLite database maps to one organization; shared-database multi-tenancy is not implemented. +- Audit records are tamper-evident at the application level, not tamper-proof against a host/database administrator. +- SQLite transaction guarantees do not automatically extend to external ERP APIs; future integrations require an outbox, worker, and reconciliation design. + +## Documentation + +- [Traditional Chinese README](https://github.com/falltwo/AI-Risk-Based-Inventory-ERP/blob/main/README.md) +- [English README](https://github.com/falltwo/AI-Risk-Based-Inventory-ERP/blob/main/README.en.md) diff --git a/frontend/access_navigation.py b/frontend/access_navigation.py index 95cfcf6..755a1bb 100644 --- a/frontend/access_navigation.py +++ b/frontend/access_navigation.py @@ -143,6 +143,7 @@ def clear_identity_session_state(state: MutableMapping[str, object]) -> None: if ( key.startswith("erp_csv_") or key.startswith("po_") + or key.startswith("purchase_proposal_") or key.startswith("radio_") ): state.pop(key, None) diff --git a/frontend/components/purchase_proposal_workbench.py b/frontend/components/purchase_proposal_workbench.py new file mode 100644 index 0000000..b9b9e00 --- /dev/null +++ b/frontend/components/purchase_proposal_workbench.py @@ -0,0 +1,216 @@ +"""L2 workbench for turning affected PO lines into governed proposals.""" + +from __future__ import annotations + +import uuid + +import pandas as pd +import streamlit as st + +from backend.agent_logger import get_pending_approval_by_id +from backend.purchase_proposals import ( + list_alternative_suppliers, + list_impacted_purchase_options, + prepare_alternative_purchase_proposal, + submit_purchase_proposal, +) + + +def _new_proposal_id() -> str: + return f"PROP-{uuid.uuid4().hex[:20].upper()}" + + +def ensure_purchase_proposal_id(state) -> str: + """Keep one proposal ID stable across Streamlit reruns.""" + proposal_id = state.get("purchase_proposal_id") + if not proposal_id: + proposal_id = _new_proposal_id() + state["purchase_proposal_id"] = proposal_id + return proposal_id + + +def start_new_purchase_proposal(state) -> str: + """Rotate identity only after an explicit new-proposal action.""" + proposal_id = _new_proposal_id() + state["purchase_proposal_id"] = proposal_id + state.pop("purchase_proposal_last_id", None) + state.pop("purchase_proposal_last_approval_id", None) + return proposal_id + + +def remember_purchase_proposal_submission(state, proposal_id: str, result) -> bool: + """Persist durable Gateway identity after submit or an idempotent replay.""" + approval_id = str(getattr(result, "approval_id", "") or "").strip() + if getattr(result, "status", None) not in {"pending", "ok", "denied"}: + return False + if not approval_id: + return False + state["purchase_proposal_last_id"] = proposal_id + state["purchase_proposal_last_approval_id"] = approval_id + return True + + +def _render_submission_state() -> bool: + """Render the durable approval state; return whether a prior submission exists.""" + proposal_id = st.session_state.get("purchase_proposal_last_id") + approval_id = st.session_state.get("purchase_proposal_last_approval_id") + if not proposal_id or not approval_id: + return False + + approval = get_pending_approval_by_id(approval_id) + if approval is None: + st.error(f"找不到審批單 `{approval_id}`，請交由管理者確認。") + elif approval["status"] == "pending": + st.info( + f"提案 `{proposal_id}` 已送審（`{approval_id}`）；尚未寫入 ERP。" + ) + elif approval["status"] == "approved": + st.success( + f"提案 `{proposal_id}` 已由 L3 核准，Gateway 已完成受控執行。" + ) + elif approval["status"] == "rejected": + st.warning( + f"提案 `{proposal_id}` 已被拒絕，ERP 未因本提案產生寫入。" + ) + else: + st.error( + f"提案 `{proposal_id}` 的審批狀態為 `{approval['status']}`，請確認稽核紀錄。" + ) + + if st.button("建立下一份替代採購提案", key="purchase_proposal_next"): + start_new_purchase_proposal(st.session_state) + st.rerun() + return True + + +def render_purchase_proposal_workbench(*, actor: str) -> None: + """Render affected PO evidence, candidate suppliers, and proposal submit.""" + st.subheader("🧾 替代採購決策提案") + st.caption( + "L2 只建立可稽核提案，沒有核准或執行權；L3 核准前不會寫入 ERP。" + ) + if _render_submission_state(): + return + + try: + impacted = list_impacted_purchase_options(actor=actor) + except (PermissionError, ValueError) as exc: + st.error(str(exc)) + return + if not impacted: + st.info("目前沒有含延遲或替代建議的受影響採購單。") + return + + option_keys = list(range(len(impacted))) + selected_index = st.selectbox( + "選擇受影響採購品項", + option_keys, + format_func=lambda index: ( + f"{impacted[index]['po_id']}｜" + f"{impacted[index]['product_id']} {impacted[index].get('product_name') or ''}｜" + f"明細 #{impacted[index]['source_po_item_id']} × {impacted[index]['qty']}｜" + f"原供應商 {impacted[index]['original_supplier_id']}｜" + f"延遲 {impacted[index].get('estimated_delay_days') or 0} 天" + ), + key="purchase_proposal_affected_line", + ) + selected = impacted[selected_index] + st.dataframe( + pd.DataFrame( + [ + { + "受影響採購單": selected["po_id"], + "原供應商": selected["supplier_name"], + "品項": selected.get("product_name") or selected["product_id"], + "數量": selected["qty"], + "預估延遲天數": selected.get("estimated_delay_days") or 0, + "既有應變建議": selected.get("alternative_suggestion") or "—", + } + ] + ), + use_container_width=True, + hide_index=True, + ) + + try: + alternatives = list_alternative_suppliers( + affected_po_id=selected["po_id"], + product_id=selected["product_id"], + source_po_item_id=selected["source_po_item_id"], + actor=actor, + ) + except (PermissionError, ValueError) as exc: + st.error(str(exc)) + return + if not alternatives: + st.warning("正式供應商主檔中沒有可供應此品項的替代來源。") + return + + proposal_id = ensure_purchase_proposal_id(st.session_state) + with st.form("purchase_proposal_form"): + alternative_index = st.selectbox( + "替代供應商", + list(range(len(alternatives))), + format_func=lambda index: ( + f"{alternatives[index]['supplier_id']} - {alternatives[index]['name']}｜" + f"{alternatives[index].get('country') or '地區未填'}｜" + f"風險 {alternatives[index].get('risk_level') or '未評'}｜" + f"單價 NT${float(alternatives[index]['price']):,.2f}" + ), + key="purchase_proposal_alternative_supplier", + ) + default_reason = str(selected.get("alternative_suggestion") or "").strip() + reason = st.text_area( + "提案理由", + value=default_reason + or "因供應鏈事件造成延遲，建議改由正式備援供應商供貨。", + max_chars=1000, + ) + delay_days = st.number_input( + "預估延遲天數", + min_value=0, + max_value=3650, + value=int(selected.get("estimated_delay_days") or 0), + step=1, + ) + st.caption(f"提案識別碼：`{proposal_id}`") + if st.form_submit_button( + "送交 L3 人工核准", type="primary", use_container_width=True + ): + alternative = alternatives[alternative_index] + try: + proposal = prepare_alternative_purchase_proposal( + proposal_id=proposal_id, + affected_po_id=selected["po_id"], + product_id=selected["product_id"], + source_po_item_id=selected["source_po_item_id"], + alternative_supplier_id=alternative["supplier_id"], + alternative_supplier_product_id=alternative[ + "supplier_product_id" + ], + reason=reason, + estimated_delay_days=int(delay_days), + actor=actor, + ) + result = submit_purchase_proposal(proposal, actor=actor) + except (PermissionError, ValueError) as exc: + st.error(str(exc)) + else: + if remember_purchase_proposal_submission( + st.session_state, proposal.proposal_id, result + ): + if result.status == "pending": + st.success( + f"提案已送審（`{result.approval_id}`）；尚未寫入 ERP。" + ) + elif result.status == "ok": + st.success( + f"已恢復提案（`{result.approval_id}`）的完成狀態。" + ) + else: + st.warning( + f"已恢復提案（`{result.approval_id}`）的拒絕狀態。" + ) + st.rerun() + else: + st.error(result.message or "提案送審失敗。") diff --git a/frontend/components/risk_overview.py b/frontend/components/risk_overview.py index c709486..148e732 100644 --- a/frontend/components/risk_overview.py +++ b/frontend/components/risk_overview.py @@ -1,10 +1,150 @@ +import sqlite3 + +import pandas as pd import streamlit as st -from frontend.ui_utils import show_error -from backend.supply_chain_risk import get_supply_chain_summary_kpis + +from backend.database import DB_FILE +from backend.erp_exchange import ( + build_purchase_order_template_csv, + parse_purchase_order_csv, +) +from backend.l1_monitoring import map_purchase_rows_to_events +from backend.supply_chain_risk import ( + get_risk_events_list, + get_supply_chain_summary_kpis, +) from frontend.components.supply_map import render_risk_heatmap +from frontend.ui_utils import show_error + + +_L1_DISPLAY_COLUMNS = { + "po_id": "採購單", + "supplier_id": "供應商", + "product_id": "物料", + "supplier_country": "國家", + "supplier_region": "地區", + "event_type": "命中事件", + "impact_days": "預估延遲天數", + "match_status": "對映結果", + "notification_status": "通知狀態", +} + + +def _load_supplier_context(supplier_ids: set[str]) -> dict[str, dict]: + if not supplier_ids: + return {} + placeholders = ",".join("?" for _ in supplier_ids) + with sqlite3.connect(DB_FILE) as conn: + conn.row_factory = sqlite3.Row + rows = conn.execute( + "SELECT supplier_id, country, region, risk_level " + f"FROM suppliers WHERE supplier_id IN ({placeholders})", + tuple(sorted(supplier_ids)), + ).fetchall() + return {row["supplier_id"]: dict(row) for row in rows} + + +def _render_latest_event_alerts(events: list[dict]) -> None: + st.markdown("#### 🚨 最新事件告警") + if not events: + st.info("目前尚無已登錄的供應鏈風險事件。") + return + + event_rows = [] + for event in events[:5]: + event_rows.append( + { + "事件": event.get("event_type") or "未分類", + "地區": event.get("region") or event.get("country") or "未設定", + "預估延遲": f"{int(event.get('impact_days') or 0)} 天", + "事件說明": event.get("description") or "未提供", + } + ) + st.dataframe(pd.DataFrame(event_rows), width="stretch", hide_index=True) + + +def _render_read_only_mapping(events: list[dict]) -> None: + st.markdown("#### 🔔 L1 告警與通知中心") + st.caption( + "上傳資料只會在記憶體中進行格式驗證、事件對映與通知預覽，" + "不會寫入 ERP 或提案暫存區。Excel 資料請先另存為 UTF-8 CSV。" + ) + st.download_button( + "下載唯讀對映 CSV 範本", + data=build_purchase_order_template_csv(), + file_name="l1_purchase_order_monitoring_template.csv", + mime="text/csv", + key="l1_monitor_download_template", + ) + uploaded = st.file_uploader( + "上傳採購資料 CSV", + type=["csv"], + key="l1_monitor_csv_upload", + help="檔案必須為 UTF-8；上傳與對映均不會修改 ERP。", + ) + if uploaded is None: + st.info("可下載範本後匯入採購資料，以預覽事件對映與通知結果。") + return + + try: + purchase_rows = parse_purchase_order_csv(uploaded.getvalue()) + supplier_context = _load_supplier_context( + {row["supplier_id"] for row in purchase_rows} + ) + mapped_rows = map_purchase_rows_to_events( + purchase_rows, + supplier_context=supplier_context, + events=events, + ) + except ValueError as exc: + st.error(f"CSV 驗證失敗：{exc}") + return + except sqlite3.Error: + st.error("目前無法讀取供應商地區資料，請稍後再試。") + return + + alert_rows = [row for row in mapped_rows if row["match_status"] == "需關注"] + incomplete_rows = [ + row for row in mapped_rows if row["match_status"] == "資料待補" + ] + metric_a, metric_b, metric_c = st.columns(3) + metric_a.metric("完成對映", f"{len(mapped_rows)} 筆") + metric_b.metric("需通知", f"{len(alert_rows)} 筆") + metric_c.metric("資料待補", f"{len(incomplete_rows)} 筆") + + display = pd.DataFrame(mapped_rows).rename(columns=_L1_DISPLAY_COLUMNS) + st.dataframe( + display[list(_L1_DISPLAY_COLUMNS.values())], + width="stretch", + hide_index=True, + ) + + st.markdown("##### 通知預覽（尚未發送）") + if alert_rows: + for row in alert_rows: + st.warning(row["notification"]) + else: + st.success("本次匯入資料未命中已登錄事件，無需發送風險通知。") + + export_rows = pd.DataFrame( + { + "採購單": [row.get("po_id") for row in mapped_rows], + "供應商": [row.get("supplier_id") for row in mapped_rows], + "對映結果": [row["match_status"] for row in mapped_rows], + "通知狀態": [row["notification_status"] for row in mapped_rows], + "通知內容": [row["notification"] for row in mapped_rows], + } + ) + st.download_button( + "下載告警與通知清單", + data=export_rows.to_csv(index=False).encode("utf-8-sig"), + file_name="l1_alert_notifications.csv", + mime="text/csv", + key="l1_monitor_download_alerts", + ) def render_risk_overview(): - """渲染供應鏈風險總覽：KPI 卡片 + 即時風險熱圖。""" + """渲染 L1 唯讀閉環：事件告警、熱圖、資料對映與通知預覽。""" st.markdown("#### 📊 供應鏈風險總覽 (Risk Overview)") # 取得 KPI 數據 @@ -35,3 +175,15 @@ def render_risk_overview(): with st.container(border=True): st.markdown("**🌍 全球即時風險熱圖**") render_risk_heatmap(key="overview_heatmap") + + st.markdown("
", unsafe_allow_html=True) + try: + event_frame = get_risk_events_list(limit=30) + events = [] if event_frame is None or event_frame.empty else event_frame.to_dict("records") + except Exception as exc: + show_error("風險事件讀取失敗", exc) + events = [] + + _render_latest_event_alerts(events) + st.markdown("
", unsafe_allow_html=True) + _render_read_only_mapping(events) diff --git a/frontend/page_agent_dashboard.py b/frontend/page_agent_dashboard.py index ae59bd0..b9d1e21 100644 --- a/frontend/page_agent_dashboard.py +++ b/frontend/page_agent_dashboard.py @@ -20,6 +20,12 @@ write_action_log, ) from backend.database import run_query +from backend.purchase_proposals import ( + ApprovalDecision, + decide_purchase_proposal, + get_purchase_operation_timeline, + get_purchase_proposal_for_operation, +) from frontend.access_navigation import dashboard_mode @@ -37,6 +43,44 @@ def _filter_purchase_proposals(records): ] +def _scope_purchase_records(records, principal): + """Remove cross-organization Proposal records before any detail is rendered.""" + visible = [] + for item in records: + operation_id = str(item.get("operation_id") or "") + if operation_id.startswith("proposal:create-po:"): + try: + proposal = get_purchase_proposal_for_operation( + operation_id, actor=principal.username + ) + except (PermissionError, ValueError): + continue + if proposal is None: + continue + visible.append(item) + return visible + + +def _safe_purchase_proposal_decision(decision, *, actor: str) -> tuple[str, str]: + """Convert expected domain/authorization failures into renderable UI state.""" + try: + result = decide_purchase_proposal(decision, actor=actor) + except (PermissionError, ValueError) as exc: + return "error", str(exc) + return result.status, result.message + + +def _should_refresh_after_decision(outcome: str, status: str) -> bool: + """Keep failed decisions visible; refresh only after a terminal success.""" + normalized_outcome = str(outcome or "").strip().lower() + normalized_status = str(status or "").strip().lower() + if normalized_outcome == "approve": + return normalized_status in {"ok", "pending"} + if normalized_outcome == "reject": + return normalized_status == "denied" + return False + + def _history_action_kind(status: str, tool_name: str, role: str) -> str: """判斷審批歷程卡片可提供的後續動作。""" if status == "rejected": @@ -138,6 +182,52 @@ def format_parameters_to_chinese(tool_name: str, args) -> str: return ", ".join(parts) +def _render_domain_proposal_evidence(proposal) -> None: + """Show the immutable business evidence separately from approval state.""" + st.markdown(f"**受影響採購單**：`{proposal.affected_po_id}`") + st.markdown( + f"**供應來源變更**：`{proposal.original_supplier_id}` → " + f"**替代供應商** `{proposal.alternative_supplier_id}`" + ) + st.markdown( + f"**品項與條件**：`{proposal.product_id}` × {proposal.qty}｜" + f"單價 {proposal.currency} {proposal.unit_price:,.2f}｜" + f"預估延遲 {proposal.estimated_delay_days or 0} 天" + ) + st.markdown(f"**L2 提案理由**：{proposal.reason}") + st.caption( + f"Proposal `{proposal.proposal_id}`｜schema v{proposal.schema_version}｜" + f"digest `{proposal.proposal_digest[:16]}…`" + ) + + +def _render_operation_timeline(operation_id: str, principal) -> None: + """Show a redacted operation timeline without raw payloads or secrets.""" + try: + timeline = get_purchase_operation_timeline( + operation_id, actor=principal.username + ) + except (PermissionError, ValueError) as exc: + st.error(f"無法驗證稽核時間線：{exc}") + return + if not timeline: + return + labels = { + "proposal_created": "L2 建立 Proposal", + "approval_submitted": "送交 L3 審批", + "approval_rejected": "L3 拒絕", + "execution_completed": "Gateway 執行完成", + } + with st.expander("🔎 端到端稽核時間線", expanded=False): + for event in timeline: + st.markdown( + f"- **{labels.get(event['kind'], event['kind'])}**｜" + f"{event.get('time') or '時間未記錄'}｜" + f"actor `{event.get('actor') or 'system'}`｜" + f"{event.get('summary') or ''}" + ) + + def _render_purchase_approval_dashboard(principal, pending_list, approval_history): """Focused L3 surface: proposal evidence and decisions, without global logs.""" st.markdown( @@ -163,8 +253,22 @@ def _render_purchase_approval_dashboard(principal, pending_list, approval_histor st.markdown( f"**核准證據**：`{format_parameters_to_chinese(item['tool'], item['args'])}`" ) + domain_proposal = None + evidence_error = None if item.get("operation_id"): st.caption(f"🔗 操作識別碼：`{item['operation_id']}`") + try: + domain_proposal = get_purchase_proposal_for_operation( + item["operation_id"], actor=principal.username + ) + except (PermissionError, ValueError) as exc: + evidence_error = str(exc) + if evidence_error: + st.error(f"提案證據驗證失敗，已停止決策：{evidence_error}") + continue + if domain_proposal is not None: + _render_domain_proposal_evidence(domain_proposal) + _render_operation_timeline(item["operation_id"], principal) if item.get("requester_username") == principal.username: st.warning("提案人不得核准自己的提案，請由另一位核准者處理。") @@ -181,12 +285,25 @@ def _render_purchase_approval_dashboard(principal, pending_list, approval_histor key=f"tier_approve_{item['id']}", use_container_width=True, ): - result = approve_action(item["id"], approver=principal.username) - if result.get("status") in {"ok", "pending"}: + if domain_proposal is not None: + result_status, result_message = _safe_purchase_proposal_decision( + ApprovalDecision( + proposal_id=domain_proposal.proposal_id, + outcome="approve", + ), + actor=principal.username, + ) + else: + result = approve_action( + item["id"], approver=principal.username + ) + result_status = result.get("status") + result_message = result.get("message") + if _should_refresh_after_decision("approve", result_status): st.toast(f"提案 {item['id']} 已核准。") + st.rerun() else: - st.error(result.get("message") or "核准失敗。") - st.rerun() + st.error(result_message or "核准失敗。") with reject_col: if st.button( "❌ 拒絕", @@ -196,14 +313,26 @@ def _render_purchase_approval_dashboard(principal, pending_list, approval_histor if not reason.strip(): st.warning("請先填寫拒絕原因。") else: - result = reject_action( - item["id"], reason, approver=principal.username - ) - if result.get("status") == "denied": + if domain_proposal is not None: + result_status, result_message = _safe_purchase_proposal_decision( + ApprovalDecision( + proposal_id=domain_proposal.proposal_id, + outcome="reject", + reason=reason, + ), + actor=principal.username, + ) + else: + result = reject_action( + item["id"], reason, approver=principal.username + ) + result_status = result.get("status") + result_message = result.get("message") + if _should_refresh_after_decision("reject", result_status): st.toast(f"提案 {item['id']} 已拒絕。") + st.rerun() else: - st.error(result.get("message") or "拒絕失敗。") - st.rerun() + st.error(result_message or "拒絕失敗。") st.markdown("---") st.markdown("### 🕒 採購提案審批歷史") @@ -223,6 +352,20 @@ def _render_purchase_approval_dashboard(principal, pending_list, approval_histor st.markdown( f"**提案內容**：`{format_parameters_to_chinese(item['tool'], item['raw_args'])}`" ) + domain_proposal = None + evidence_error = None + if item.get("operation_id"): + try: + domain_proposal = get_purchase_proposal_for_operation( + item["operation_id"], actor=principal.username + ) + except (PermissionError, ValueError) as exc: + evidence_error = str(exc) + if evidence_error: + st.error(f"提案證據驗證失敗：{evidence_error}") + elif domain_proposal is not None: + _render_domain_proposal_evidence(domain_proposal) + _render_operation_timeline(item["operation_id"], principal) if item["reason"]: st.markdown(f"**拒絕原因**：{item['reason']}") @@ -248,7 +391,9 @@ def render(username: str = ""): # ── 從資料庫取得最新審批資料 ────────────────────────────── pending_list = get_pending_list() if mode == "approvals": - pending_list = _filter_purchase_proposals(pending_list) + pending_list = _scope_purchase_records( + _filter_purchase_proposals(pending_list), principal + ) pending_count = len(pending_list) # 讀取歷史審批紀錄 @@ -276,7 +421,9 @@ def render(username: str = ""): _render_purchase_approval_dashboard( principal, pending_list, - _filter_purchase_proposals(approval_history), + _scope_purchase_records( + _filter_purchase_proposals(approval_history), principal + ), ) return diff --git a/frontend/page_supply_chain_risk.py b/frontend/page_supply_chain_risk.py index 88365c0..e506373 100644 --- a/frontend/page_supply_chain_risk.py +++ b/frontend/page_supply_chain_risk.py @@ -11,6 +11,7 @@ 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 def render( sub_menu: str, @@ -85,3 +86,8 @@ def render( actor=principal.username, ) + st.markdown("
", unsafe_allow_html=True) + st.markdown("---") + st.markdown("### 🧾 步驟 5: 建立受治理的替代採購提案") + render_purchase_proposal_workbench(actor=principal.username) + diff --git a/scripts/seed_e_day1_demo_data.py b/scripts/seed_e_day1_demo_data.py index abbef48..a92556d 100644 --- a/scripts/seed_e_day1_demo_data.py +++ b/scripts/seed_e_day1_demo_data.py @@ -13,6 +13,9 @@ LOW_STOCK_PRODUCTS = [ { "product_id": "P001", + "name": "筆記型電腦", + "price": 45000, + "cost": 35000, "stock": 12, "reorder_point": 60, "daily_sales": 8, @@ -20,6 +23,9 @@ }, { "product_id": "P004", + "name": "電腦螢幕", + "price": 6000, + "cost": 4500, "stock": 8, "reorder_point": 60, "daily_sales": 10, @@ -27,6 +33,9 @@ }, { "product_id": "P019", + "name": "USB-C 控制模組", + "price": 1200, + "cost": 690, "stock": 15, "reorder_point": 96, "daily_sales": 14, @@ -173,19 +182,29 @@ def seed_low_stock(conn: sqlite3.Connection) -> None: for item in LOW_STOCK_PRODUCTS: conn.execute( """ - UPDATE inventory - SET stock = ?, - reorder_point = ?, - daily_sales = ?, - baseline_reorder_point = COALESCE(baseline_reorder_point, ?) - WHERE product_id = ? + INSERT INTO inventory ( + product_id, name, stock, price, cost, reorder_point, + baseline_reorder_point, daily_sales, barcode + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(product_id) DO UPDATE SET + stock = excluded.stock, + reorder_point = excluded.reorder_point, + daily_sales = excluded.daily_sales, + baseline_reorder_point = COALESCE( + inventory.baseline_reorder_point, + excluded.baseline_reorder_point + ) """, ( + item["product_id"], + item["name"], item["stock"], + item["price"], + item["cost"], item["reorder_point"], - item["daily_sales"], item["baseline_reorder_point"], - item["product_id"], + item["daily_sales"], + f"E-DAY1-{item['product_id']}", ), ) @@ -309,15 +328,30 @@ def seed_purchase_order(conn: sqlite3.Connection, today: datetime) -> None: ), ) - conn.execute("DELETE FROM purchase_order_items WHERE po_id = ?", (po["po_id"],)) for item in po["items"]: - conn.execute( - """ - INSERT INTO purchase_order_items (po_id, product_id, qty, unit_price) - VALUES (?, ?, ?, ?) - """, - (po["po_id"], item["product_id"], item["qty"], item["unit_price"]), - ) + existing_items = conn.execute( + "SELECT id FROM purchase_order_items " + "WHERE po_id = ? AND product_id = ? ORDER BY id", + (po["po_id"], item["product_id"]), + ).fetchall() + if len(existing_items) > 1: + raise RuntimeError( + "Demo purchase order has duplicate product lines; " + "refusing to replace durable source evidence." + ) + if existing_items: + conn.execute( + "UPDATE purchase_order_items SET qty = ?, unit_price = ? WHERE id = ?", + (item["qty"], item["unit_price"], existing_items[0][0]), + ) + else: + conn.execute( + """ + INSERT INTO purchase_order_items (po_id, product_id, qty, unit_price) + VALUES (?, ?, ?, ?) + """, + (po["po_id"], item["product_id"], item["qty"], item["unit_price"]), + ) def seed_orders(conn: sqlite3.Connection, today: datetime) -> None: diff --git a/tests/test_demo_seed_schema.py b/tests/test_demo_seed_schema.py index a9f17dc..7f01921 100644 --- a/tests/test_demo_seed_schema.py +++ b/tests/test_demo_seed_schema.py @@ -3,6 +3,8 @@ import sqlite3 from datetime import datetime +import pytest + from backend import database from scripts import seed_e_day1_demo_data @@ -28,9 +30,83 @@ def test_fresh_database_has_every_column_required_by_demo_seeder( assert conn.execute( "SELECT COUNT(*) FROM supply_chain_events" ).fetchone()[0] >= 1 + orphan_item_count = conn.execute( + """ + SELECT COUNT(*) + FROM purchase_order_items item + LEFT JOIN inventory product + ON product.product_id = item.product_id + WHERE item.po_id = ? AND product.product_id IS NULL + """, + (seed_e_day1_demo_data.DEMO_PURCHASE_ORDER["po_id"],), + ).fetchone()[0] + assert orphan_item_count == 0 conn.rollback() +def test_demo_seed_preserves_source_line_identity_after_approval( + tmp_path, monkeypatch +): + from backend.purchase_proposals import ( + ApprovalDecision, + decide_purchase_proposal, + prepare_alternative_purchase_proposal, + submit_purchase_proposal, + ) + + db_path = tmp_path / "stable-demo-source.db" + monkeypatch.setattr(database, "DB_FILE", str(db_path)) + database.init_db() + now = datetime.now() + + with sqlite3.connect(db_path) as conn: + seed_e_day1_demo_data.seed_low_stock(conn) + seed_e_day1_demo_data.seed_suppliers(conn) + seed_e_day1_demo_data.seed_purchase_order(conn, now) + source_id = conn.execute( + "SELECT id FROM purchase_order_items " + "WHERE po_id = ? AND product_id = 'P019'", + (seed_e_day1_demo_data.DEMO_PURCHASE_ORDER["po_id"],), + ).fetchone()[0] + + proposal = prepare_alternative_purchase_proposal( + proposal_id="PROP-SEED-STABLE-A", + affected_po_id=seed_e_day1_demo_data.DEMO_PURCHASE_ORDER["po_id"], + product_id="P019", + source_po_item_id=source_id, + alternative_supplier_id="SUP-E-DEMO-TW", + reason="Verify stable demo source identity", + actor="planner", + ) + submit_purchase_proposal(proposal, actor="planner") + approved = decide_purchase_proposal( + ApprovalDecision(proposal_id=proposal.proposal_id, outcome="approve"), + actor="approver", + ) + assert approved.status == "ok" + + with sqlite3.connect(db_path) as conn: + seed_e_day1_demo_data.seed_purchase_order(conn, now) + replayed_source_id = conn.execute( + "SELECT id FROM purchase_order_items " + "WHERE po_id = ? AND product_id = 'P019'", + (seed_e_day1_demo_data.DEMO_PURCHASE_ORDER["po_id"],), + ).fetchone()[0] + assert replayed_source_id == source_id + + second = prepare_alternative_purchase_proposal( + proposal_id="PROP-SEED-STABLE-B", + affected_po_id=seed_e_day1_demo_data.DEMO_PURCHASE_ORDER["po_id"], + product_id="P019", + source_po_item_id=source_id, + alternative_supplier_id="SUP-E-DEMO-TW", + reason="A second full replacement must remain blocked", + actor="planner", + ) + with pytest.raises(PermissionError, match="已有另一份替代提案"): + submit_purchase_proposal(second, actor="planner") + + def test_init_db_additively_upgrades_old_demo_schema_without_data_loss( tmp_path, monkeypatch ): diff --git a/tests/test_l1_monitoring.py b/tests/test_l1_monitoring.py new file mode 100644 index 0000000..7c878bd --- /dev/null +++ b/tests/test_l1_monitoring.py @@ -0,0 +1,142 @@ +from copy import deepcopy +from pathlib import Path + +from backend.l1_monitoring import map_purchase_rows_to_events + + +ROOT = Path(__file__).resolve().parents[1] + + +def _purchase_row(**overrides): + row = { + "external_id": "EXT-001", + "po_id": "PO-001", + "supplier_id": "SUP01", + "product_id": "P001", + "qty": 100, + "unit_price": 25.0, + "order_date": "2026-07-21", + "status": "待交貨", + "note": "", + } + row.update(overrides) + return row + + +def test_l1_csv_rows_map_to_highest_impact_location_event_without_mutating_inputs(): + purchase_rows = [_purchase_row()] + supplier_context = { + "SUP01": { + "country": "日本", + "region": "關東", + "risk_level": "中", + } + } + events = [ + { + "id": 10, + "event_type": "交通", + "country": "日本", + "region": "關西", + "impact_days": 3, + "description": "港口壅塞", + }, + { + "id": 11, + "event_type": "地震", + "country": "日本", + "region": "關東", + "impact_days": 14, + "description": "區域物流中斷", + }, + ] + original_rows = deepcopy(purchase_rows) + original_context = deepcopy(supplier_context) + original_events = deepcopy(events) + + result = map_purchase_rows_to_events( + purchase_rows, supplier_context=supplier_context, events=events + ) + + assert result == [ + { + **purchase_rows[0], + "supplier_country": "日本", + "supplier_region": "關東", + "supplier_risk_level": "中", + "match_status": "需關注", + "matched_event_id": 11, + "event_type": "地震", + "impact_days": 14, + "notification_status": "待人工確認", + "notification": ( + "採購單 PO-001：供應商 SUP01 位於日本／關東，" + "命中地震風險，預估延遲 14 天。" + ), + } + ] + assert purchase_rows == original_rows + assert supplier_context == original_context + assert events == original_events + + +def test_l1_unknown_supplier_remains_unmatched_and_does_not_invent_an_alert(): + result = map_purchase_rows_to_events( + [_purchase_row(supplier_id="UNKNOWN")], + supplier_context={}, + events=[ + { + "id": 1, + "event_type": "戰爭", + "country": "日本", + "region": "關東", + "impact_days": 30, + } + ], + ) + + assert result[0]["match_status"] == "資料待補" + assert result[0]["matched_event_id"] is None + assert result[0]["event_type"] == "未命中" + assert result[0]["impact_days"] == 0 + assert result[0]["notification_status"] == "無法判定" + assert "缺少供應商地區資料" in result[0]["notification"] + + +def test_l1_known_supplier_without_matching_event_is_reported_as_normal(): + result = map_purchase_rows_to_events( + [_purchase_row()], + supplier_context={ + "SUP01": {"country": "台灣", "region": "中部", "risk_level": "低"} + }, + events=[ + { + "id": 1, + "event_type": "罷工", + "country": "德國", + "region": "漢堡", + "impact_days": 7, + } + ], + ) + + assert result[0]["match_status"] == "正常" + assert result[0]["matched_event_id"] is None + assert result[0]["event_type"] == "未命中" + assert result[0]["notification_status"] == "無需通知" + assert "未命中目前風險事件" in result[0]["notification"] + + +def test_l1_overview_source_contains_read_only_csv_mapping_and_notification_flow(): + source = (ROOT / "frontend/components/risk_overview.py").read_text( + encoding="utf-8" + ) + + assert "build_purchase_order_template_csv" in source + assert "parse_purchase_order_csv" in source + assert "map_purchase_rows_to_events" in source + assert 'type=["csv"]' in source + assert "不會寫入 ERP" in source + assert "L1 告警與通知中心" in source + assert "stage_purchase_order_rows" not in source + assert "submit_exchange_record" not in source diff --git a/tests/test_purchase_order_approval.py b/tests/test_purchase_order_approval.py index a538393..0a60758 100644 --- a/tests/test_purchase_order_approval.py +++ b/tests/test_purchase_order_approval.py @@ -435,7 +435,7 @@ def test_po_tool_is_registered_as_governed_write_and_hidden_from_line(): assert info == { "module": "procurement", "risk_level": "write", - "allowed_roles": ["admin", "warehouse"], + "allowed_roles": ["admin", "warehouse", "supply_planner"], "description": info["description"], } assert callable(tools_mapping["create_purchase_order"]) diff --git a/tests/test_purchase_proposal_ui.py b/tests/test_purchase_proposal_ui.py new file mode 100644 index 0000000..d8fd445 --- /dev/null +++ b/tests/test_purchase_proposal_ui.py @@ -0,0 +1,176 @@ +"""UI contracts for the L2 proposal workbench and L3 evidence surface.""" + +import ast +from pathlib import Path +from types import SimpleNamespace + + +ROOT = Path(__file__).resolve().parents[1] + + +def _source(relative_path: str) -> str: + return (ROOT / relative_path).read_text(encoding="utf-8") + + +def _call_is_guarded_by_submit(tree: ast.AST, function_name: str) -> bool: + for node in ast.walk(tree): + if not isinstance(node, ast.If): + continue + if "form_submit_button" not in ast.dump(node.test): + continue + if any( + isinstance(child, ast.Call) + and isinstance(child.func, ast.Name) + and child.func.id == function_name + for statement in node.body + for child in ast.walk(statement) + ): + return True + return False + + +def test_proposal_id_is_stable_until_user_starts_another_proposal(): + from frontend.components.purchase_proposal_workbench import ( + ensure_purchase_proposal_id, + start_new_purchase_proposal, + ) + + state = {} + first = ensure_purchase_proposal_id(state) + assert ensure_purchase_proposal_id(state) == first + + state["purchase_proposal_last_id"] = "PROP-OLD" + second = start_new_purchase_proposal(state) + assert second != first + assert "purchase_proposal_last_id" not in state + + +def test_l2_page_renders_proposal_workbench_after_what_if(): + source = _source("frontend/page_supply_chain_risk.py") + + assert "render_purchase_proposal_workbench" in source + assert source.rindex("render_what_if_analysis(") < source.rindex( + "render_purchase_proposal_workbench(" + ) + assert "actor=principal.username" in source + + +def test_workbench_submits_domain_proposal_only_after_explicit_form_submit(): + source = _source("frontend/components/purchase_proposal_workbench.py") + tree = ast.parse(source) + + assert "list_impacted_purchase_options" in source + assert "list_alternative_suppliers" in source + assert "prepare_alternative_purchase_proposal" in source + assert _call_is_guarded_by_submit(tree, "submit_purchase_proposal") + assert "gateway.call" not in source + assert "create_purchase_order(" not in source + assert "尚未寫入 ERP" in source + + +def test_l3_dashboard_uses_domain_decision_and_redacted_timeline_for_new_proposals(): + source = _source("frontend/page_agent_dashboard.py") + + assert "ApprovalDecision" in source + assert "decide_purchase_proposal" in source + assert "get_purchase_proposal_for_operation" in source + assert "get_purchase_operation_timeline" in source + assert "受影響採購單" in source + assert "替代供應商" in source + + +def test_logout_clears_purchase_proposal_session_state(): + source = _source("frontend/access_navigation.py") + + assert 'key.startswith("purchase_proposal_")' in source + + +def test_successful_replay_restores_durable_submission_tracking(): + from frontend.components.purchase_proposal_workbench import ( + remember_purchase_proposal_submission, + ) + + state = {} + replay = SimpleNamespace(status="ok", approval_id="APPROVAL-REPLAY") + + assert remember_purchase_proposal_submission(state, "PROP-REPLAY", replay) + assert state["purchase_proposal_last_id"] == "PROP-REPLAY" + assert state["purchase_proposal_last_approval_id"] == "APPROVAL-REPLAY" + + +def test_failed_submission_does_not_create_false_session_tracking(): + from frontend.components.purchase_proposal_workbench import ( + remember_purchase_proposal_submission, + ) + + state = {} + failed = SimpleNamespace(status="error", approval_id=None) + + assert not remember_purchase_proposal_submission(state, "PROP-FAILED", failed) + assert state == {} + + +def test_rejected_replay_also_restores_durable_submission_tracking(): + from frontend.components.purchase_proposal_workbench import ( + remember_purchase_proposal_submission, + ) + + state = {} + rejected = SimpleNamespace(status="denied", approval_id="APPROVAL-REJECTED") + + assert remember_purchase_proposal_submission(state, "PROP-REJECTED", rejected) + assert state["purchase_proposal_last_id"] == "PROP-REJECTED" + assert state["purchase_proposal_last_approval_id"] == "APPROVAL-REJECTED" + + +def test_l3_records_are_scoped_before_rendering_proposal_details(monkeypatch): + import frontend.page_agent_dashboard as dashboard + + records = [ + {"id": "same-org", "operation_id": "proposal:create-po:SAME:v1"}, + {"id": "other-org", "operation_id": "proposal:create-po:OTHER:v1"}, + {"id": "legacy", "operation_id": "legacy-operation"}, + ] + + def fake_lookup(operation_id, *, actor): + if "OTHER" in operation_id: + raise PermissionError("cross organization") + if "SAME" in operation_id: + return SimpleNamespace(proposal_id="SAME") + return None + + monkeypatch.setattr( + dashboard, "get_purchase_proposal_for_operation", fake_lookup + ) + principal = SimpleNamespace(username="approver") + + visible = dashboard._scope_purchase_records(records, principal) + + assert [item["id"] for item in visible] == ["same-org", "legacy"] + + +def test_l3_expected_domain_error_is_renderable_instead_of_crashing(monkeypatch): + import frontend.page_agent_dashboard as dashboard + + def revoked(*args, **kwargs): + raise PermissionError("approval role revoked") + + monkeypatch.setattr(dashboard, "decide_purchase_proposal", revoked) + + status, message = dashboard._safe_purchase_proposal_decision( + SimpleNamespace(), actor="approver" + ) + + assert status == "error" + assert message == "approval role revoked" + + +def test_l3_refreshes_only_after_a_successful_decision(): + from frontend.page_agent_dashboard import _should_refresh_after_decision + + assert _should_refresh_after_decision("approve", "ok") + assert _should_refresh_after_decision("approve", "pending") + assert _should_refresh_after_decision("reject", "denied") + assert not _should_refresh_after_decision("approve", "error") + assert not _should_refresh_after_decision("approve", "denied") + assert not _should_refresh_after_decision("reject", "error") diff --git a/tests/test_purchase_proposals.py b/tests/test_purchase_proposals.py new file mode 100644 index 0000000..3344313 --- /dev/null +++ b/tests/test_purchase_proposals.py @@ -0,0 +1,803 @@ +"""Contracts for the L2 proposal -> L3 approval -> Gateway execution flow.""" + +from dataclasses import FrozenInstanceError, replace +from datetime import datetime as real_datetime +import json +import sqlite3 + +import pytest + +from backend import database + + +@pytest.fixture +def proposal_db(tmp_path, monkeypatch): + db_path = tmp_path / "purchase-proposals.db" + monkeypatch.setattr(database, "DB_FILE", str(db_path)) + database.init_db() + + with database.transaction() as conn: + conn.execute( + "UPDATE suppliers SET is_official = 1 WHERE supplier_id IN ('SUP01', 'SUP02')" + ) + conn.execute( + "DELETE FROM supplier_products WHERE supplier_id IN ('SUP01', 'SUP02') " + "AND product_id = 'P001'" + ) + conn.execute( + "INSERT INTO supplier_products (supplier_id, product_id, price, carbon_factor) " + "VALUES ('SUP01', 'P001', 500, 3.5)" + ) + conn.execute( + "INSERT INTO supplier_products (supplier_id, product_id, price, carbon_factor) " + "VALUES ('SUP02', 'P001', 475, 2.1)" + ) + conn.execute( + """ + INSERT INTO purchase_orders ( + po_id, supplier_id, order_date, status, total_amount, note, + estimated_delay_days, alternative_suggestion + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + "PO-RISK-001", + "SUP01", + "2026-07-20", + "待入庫", + 1000.0, + "affected by port disruption", + 12, + "switch to SUP02", + ), + ) + conn.execute( + "INSERT INTO purchase_order_items (po_id, product_id, qty, unit_price) " + "VALUES ('PO-RISK-001', 'P001', 2, 500)" + ) + return db_path + + +def _prepare(proposal_id="PROP-TEST-001", *, actor="planner"): + from backend.purchase_proposals import prepare_alternative_purchase_proposal + + return prepare_alternative_purchase_proposal( + affected_po_id="PO-RISK-001", + product_id="P001", + alternative_supplier_id="SUP02", + reason="港口中斷，改由低風險正式供應商供貨", + estimated_delay_days=12, + actor=actor, + proposal_id=proposal_id, + ) + + +def test_fresh_schema_keeps_proposal_separate_from_approval(proposal_db): + with sqlite3.connect(proposal_db) as conn: + columns = { + row[1] for row in conn.execute("PRAGMA table_info(purchase_proposals)") + } + + assert { + "proposal_id", + "proposal_type", + "schema_version", + "organization_id", + "proposer_username", + "proposer_role", + "affected_po_id", + "proposed_po_id", + "original_supplier_id", + "alternative_supplier_id", + "product_id", + "qty", + "unit_price", + "currency", + "reason", + "estimated_delay_days", + "source_po_version", + "proposal_digest", + "created_at", + } <= columns + assert "operation_id" not in columns + assert "approval_id" not in columns + assert "approval_status" not in columns + assert "tool_name" not in columns + + +def test_prepare_returns_immutable_domain_object_and_separate_execution_request( + proposal_db, +): + from backend.purchase_proposals import ( + ApprovalDecision, + proposal_to_execution_request, + ) + + proposal = _prepare() + assert proposal.affected_po_id == "PO-RISK-001" + assert proposal.original_supplier_id == "SUP01" + assert proposal.alternative_supplier_id == "SUP02" + assert proposal.product_id == "P001" + assert proposal.qty == 2 + assert proposal.unit_price == 475.0 + assert proposal.currency == "TWD" + assert proposal.source_po_version.startswith("sha256:") + assert not hasattr(proposal, "operation_id") + assert not hasattr(proposal, "tool_name") + + with pytest.raises(FrozenInstanceError): + proposal.qty = 99 + + execution = proposal_to_execution_request(proposal) + assert execution.tool_name == "create_purchase_order" + assert execution.operation_id == "proposal:create-po:PROP-TEST-001:v1" + assert execution.args["proposal_id"] == proposal.proposal_id + assert execution.args["affected_po_id"] == proposal.affected_po_id + assert execution.args["source_po_version"] == proposal.source_po_version + assert execution.args["supplier_id"] == "SUP02" + assert "approval_id" not in execution.args + assert "approver" not in execution.args + assert "approval_status" not in execution.args + assert "reason" not in execution.args + assert "estimated_delay_days" not in execution.args + assert "source_event_id" not in execution.args + assert set(ApprovalDecision.__dataclass_fields__) == { + "proposal_id", + "outcome", + "reason", + } + + +@pytest.mark.parametrize("actor", ["viewer", "approver", None]) +def test_only_l2_proposer_can_prepare_alternative_proposal(proposal_db, actor): + with pytest.raises(PermissionError): + _prepare(actor=actor) + + +def test_alternative_supplier_must_be_official_and_different(proposal_db): + from backend.purchase_proposals import prepare_alternative_purchase_proposal + + with pytest.raises(ValueError, match="不同"): + prepare_alternative_purchase_proposal( + affected_po_id="PO-RISK-001", + product_id="P001", + alternative_supplier_id="SUP01", + reason="same supplier", + actor="planner", + proposal_id="PROP-SAME-SUPPLIER", + ) + + database.run_query( + "UPDATE suppliers SET is_official = 0 WHERE supplier_id = 'SUP02'", + fetch=False, + ) + with pytest.raises(PermissionError, match="正式供應商"): + _prepare(proposal_id="PROP-INACTIVE-SUPPLIER") + + +def test_l2_submission_persists_proposal_but_not_live_erp_effect(proposal_db): + from backend.purchase_proposals import submit_purchase_proposal + + proposal = _prepare() + result = submit_purchase_proposal(proposal, actor="planner") + + assert result.status == "pending" + assert result.approval_id + assert database.run_query( + "SELECT COUNT(*) FROM purchase_proposals WHERE proposal_id = ?", + (proposal.proposal_id,), + )[0][0] == 1 + approval = database.run_query( + "SELECT tool_name, requester_username, parameters, status " + "FROM pending_approvals WHERE approval_id = ?", + (result.approval_id,), + )[0] + args = json.loads(approval[2]) + assert approval[:2] == ("create_purchase_order", "planner") + assert approval[3] == "pending" + assert args["proposal_id"] == proposal.proposal_id + assert database.run_query( + "SELECT COUNT(*) FROM purchase_orders WHERE po_id = ?", + (proposal.proposed_po_id,), + )[0][0] == 0 + assert database.run_query( + "SELECT COUNT(*) FROM effect_receipts WHERE operation_id = ?", + ("proposal:create-po:PROP-TEST-001:v1",), + )[0][0] == 0 + + +def test_submit_rejects_forged_effect_fields_even_with_a_valid_digest(proposal_db): + from backend.purchase_proposals import _proposal_digest, submit_purchase_proposal + + proposal = _prepare(proposal_id="PROP-FORGED-QTY") + forged = replace(proposal, qty=proposal.qty + 999, proposal_digest="") + forged = replace(forged, proposal_digest=_proposal_digest(forged)) + + with pytest.raises(PermissionError, match="數量"): + submit_purchase_proposal(forged, actor="planner") + + assert database.run_query( + "SELECT COUNT(*) FROM purchase_proposals WHERE proposal_id = ?", + (proposal.proposal_id,), + )[0][0] == 0 + assert database.run_query( + "SELECT COUNT(*) FROM pending_approvals WHERE operation_id = ?", + ("proposal:create-po:PROP-FORGED-QTY:v1",), + )[0][0] == 0 + + +def test_same_operation_replays_one_proposal_and_one_approval(proposal_db): + from backend.purchase_proposals import submit_purchase_proposal + + proposal = _prepare() + first = submit_purchase_proposal(proposal, actor="planner") + second = submit_purchase_proposal(proposal, actor="planner") + + assert second.status == "pending" + assert second.approval_id == first.approval_id + assert database.run_query( + "SELECT COUNT(*) FROM purchase_proposals WHERE proposal_id = ?", + (proposal.proposal_id,), + )[0][0] == 1 + assert database.run_query( + "SELECT COUNT(*) FROM pending_approvals WHERE operation_id = ?", + ("proposal:create-po:PROP-TEST-001:v1",), + )[0][0] == 1 + + +def test_l3_approval_executes_once_and_builds_correlated_timeline(proposal_db): + from backend.purchase_proposals import ( + ApprovalDecision, + decide_purchase_proposal, + get_purchase_operation_timeline, + proposal_to_execution_request, + submit_purchase_proposal, + ) + + proposal = _prepare() + execution = proposal_to_execution_request(proposal) + pending = submit_purchase_proposal(proposal, actor="planner") + decision = ApprovalDecision( + proposal_id=proposal.proposal_id, outcome="approve", reason="" + ) + approved = decide_purchase_proposal(decision, actor="approver") + replay = decide_purchase_proposal(decision, actor="approver") + + assert approved.status == "ok" + assert replay.status == "ok" + assert database.run_query( + "SELECT supplier_id, operation_id FROM purchase_orders WHERE po_id = ?", + (proposal.proposed_po_id,), + ) == [("SUP02", execution.operation_id)] + assert database.run_query( + "SELECT COUNT(*) FROM effect_receipts WHERE operation_id = ?", + (execution.operation_id,), + )[0][0] == 1 + + timeline = get_purchase_operation_timeline( + execution.operation_id, actor="approver" + ) + assert [event["kind"] for event in timeline] == [ + "proposal_created", + "approval_submitted", + "execution_completed", + ] + assert all(event["operation_id"] == execution.operation_id for event in timeline) + assert "parameters" not in json.dumps(timeline, ensure_ascii=False) + + +@pytest.mark.parametrize("tamper_target", ["source_po", "proposal"]) +def test_approval_fails_closed_when_source_or_proposal_changes( + proposal_db, tamper_target +): + from backend.purchase_proposals import submit_purchase_proposal + from backend.tool_gateway import gateway + + proposal = _prepare(proposal_id=f"PROP-TAMPER-{tamper_target.upper()}") + pending = submit_purchase_proposal(proposal, actor="planner") + from backend.purchase_proposals import proposal_to_execution_request + + execution = proposal_to_execution_request(proposal) + + if tamper_target == "source_po": + database.run_query( + "UPDATE purchase_order_items SET qty = 99 " + "WHERE po_id = 'PO-RISK-001' AND product_id = 'P001'", + fetch=False, + ) + else: + database.run_query( + "UPDATE purchase_proposals SET alternative_supplier_id = 'SUP01' " + "WHERE proposal_id = ?", + (proposal.proposal_id,), + fetch=False, + ) + + result = gateway.approve_action(pending.approval_id, approver="approver") + assert result.status in {"denied", "error"} + assert database.run_query( + "SELECT COUNT(*) FROM purchase_orders WHERE po_id = ?", + (proposal.proposed_po_id,), + )[0][0] == 0 + assert database.run_query( + "SELECT COUNT(*) FROM effect_receipts WHERE operation_id = ?", + (execution.operation_id,), + )[0][0] == 0 + + +@pytest.mark.parametrize( + ("proposal_id", "operation_id"), + [ + (None, "proposal:create-po:RAW-MISSING:v1"), + ("PROP-NOT-FOUND", "proposal:create-po:PROP-NOT-FOUND:v1"), + ], +) +def test_l2_raw_gateway_requires_a_durable_bound_proposal( + proposal_db, proposal_id, operation_id +): + """A planner cannot bypass the Proposal adapter with a raw Gateway call.""" + from backend.tool_gateway import gateway + + args = { + "po_id": "ALT-RAW-BYPASS", + "supplier_id": "SUP02", + "product_id": "P001", + "qty": 2, + "unit_price": 475.0, + "order_date": "2026-07-21", + "status": "待入庫", + "note": "raw L2 bypass attempt", + } + if proposal_id is not None: + args.update( + { + "proposal_id": proposal_id, + "proposal_digest": "0" * 64, + "affected_po_id": "PO-RISK-001", + "source_po_version": "sha256:" + "0" * 64, + } + ) + + result = gateway.call( + "create_purchase_order", + args, + role="supply_planner", + actor="planner", + agent_name="procurement_agent", + operation_id=operation_id, + ) + + assert result.status in {"denied", "error"} + assert database.run_query( + "SELECT COUNT(*) FROM pending_approvals WHERE operation_id = ?", + (operation_id,), + )[0][0] == 0 + assert database.run_query( + "SELECT COUNT(*) FROM purchase_orders WHERE po_id = 'ALT-RAW-BYPASS'" + )[0][0] == 0 + + +def test_l2_raw_gateway_rejects_payload_mismatched_to_durable_proposal(proposal_db): + from backend.purchase_proposals import ( + _persist_proposal, + proposal_to_execution_request, + ) + from backend.tool_gateway import gateway + + proposal = _persist_proposal(_prepare(proposal_id="PROP-RAW-MISMATCH")) + execution = proposal_to_execution_request(proposal) + tampered_args = dict(execution.args) + tampered_args["qty"] = proposal.qty + 99 + + result = gateway.call( + execution.tool_name, + tampered_args, + role="supply_planner", + actor="planner", + agent_name="procurement_agent", + operation_id=execution.operation_id, + ) + + assert result.status in {"denied", "error"} + assert database.run_query( + "SELECT COUNT(*) FROM pending_approvals WHERE operation_id = ?", + (execution.operation_id,), + )[0][0] == 0 + assert database.run_query( + "SELECT COUNT(*) FROM purchase_orders WHERE po_id = ?", + (proposal.proposed_po_id,), + )[0][0] == 0 + + +def test_other_l2_actor_cannot_submit_someone_elses_durable_proposal(proposal_db): + from backend.purchase_proposals import ( + _persist_proposal, + proposal_to_execution_request, + ) + from backend.tool_gateway import gateway + + database.run_query( + "INSERT INTO users (username, password, role, name) " + "VALUES ('planner2', 'unused', 'supply_planner', '第二位規劃員')", + fetch=False, + ) + database.run_query( + "INSERT INTO user_organizations (username, organization_id) " + "VALUES ('planner2', 'demo-org')", + fetch=False, + ) + proposal = _persist_proposal(_prepare(proposal_id="PROP-OWNER-BOUND")) + execution = proposal_to_execution_request(proposal) + + result = gateway.call( + execution.tool_name, + dict(execution.args), + role="supply_planner", + actor="planner2", + agent_name="procurement_agent", + operation_id=execution.operation_id, + ) + + assert result.status == "denied" + assert database.run_query( + "SELECT COUNT(*) FROM pending_approvals WHERE operation_id = ?", + (execution.operation_id,), + )[0][0] == 0 + + +def test_approval_rechecks_proposal_owner_for_preexisting_pending_request(proposal_db): + from backend.purchase_proposals import ( + _persist_proposal, + proposal_to_execution_request, + ) + from backend.tool_gateway import ( + PO_APPROVAL_POLICY_VERSION, + _create_pending_approval, + gateway, + ) + + database.run_query( + "INSERT INTO users (username, password, role, name) " + "VALUES ('planner2', 'unused', 'supply_planner', '第二位規劃員')", + fetch=False, + ) + database.run_query( + "INSERT INTO user_organizations (username, organization_id) " + "VALUES ('planner2', 'demo-org')", + fetch=False, + ) + proposal = _persist_proposal(_prepare(proposal_id="PROP-OWNER-EXECUTE")) + execution = proposal_to_execution_request(proposal) + approval_id = _create_pending_approval( + execution.tool_name, + dict(execution.args), + "supply_planner", + requester_username="planner2", + operation_id=execution.operation_id, + resource_version="absent", + policy_version=PO_APPROVAL_POLICY_VERSION, + ) + + result = gateway.approve_action(approval_id, approver="approver") + + assert result.status in {"denied", "error"} + assert database.run_query( + "SELECT status FROM pending_approvals WHERE approval_id = ?", + (approval_id,), + )[0][0] == "pending" + assert database.run_query( + "SELECT COUNT(*) FROM purchase_orders WHERE po_id = ?", + (proposal.proposed_po_id,), + )[0][0] == 0 + assert database.run_query( + "SELECT COUNT(*) FROM effect_receipts WHERE operation_id = ?", + (execution.operation_id,), + )[0][0] == 0 + + +def test_same_actor_cannot_self_approve_after_switching_to_l3_role(proposal_db): + from backend.purchase_proposals import ( + ApprovalDecision, + decide_purchase_proposal, + submit_purchase_proposal, + ) + + proposal = _prepare(proposal_id="PROP-SELF-SWITCH") + pending = submit_purchase_proposal(proposal, actor="planner") + database.run_query( + "UPDATE users SET role = 'procurement_approver' WHERE username = 'planner'", + fetch=False, + ) + + result = decide_purchase_proposal( + ApprovalDecision(proposal_id=proposal.proposal_id, outcome="approve"), + actor="planner", + ) + + assert result.status == "denied" + assert database.run_query( + "SELECT status FROM pending_approvals WHERE approval_id = ?", + (pending.approval_id,), + )[0][0] == "pending" + assert database.run_query( + "SELECT COUNT(*) FROM purchase_orders WHERE po_id = ?", + (proposal.proposed_po_id,), + )[0][0] == 0 + + +def test_l3_decision_uses_live_role_not_stale_session_claim(proposal_db): + from backend.purchase_proposals import ( + ApprovalDecision, + decide_purchase_proposal, + submit_purchase_proposal, + ) + + proposal = _prepare(proposal_id="PROP-LIVE-ROLE") + submit_purchase_proposal(proposal, actor="planner") + database.run_query( + "UPDATE users SET role = 'read_only_viewer' WHERE username = 'approver'", + fetch=False, + ) + + with pytest.raises(PermissionError): + decide_purchase_proposal( + ApprovalDecision(proposal_id=proposal.proposal_id, outcome="approve"), + actor="approver", + ) + + assert database.run_query( + "SELECT COUNT(*) FROM purchase_orders WHERE po_id = ?", + (proposal.proposed_po_id,), + )[0][0] == 0 + + +def test_timeline_is_an_allowlist_and_does_not_echo_rejection_reason(proposal_db): + from backend.purchase_proposals import ( + ApprovalDecision, + decide_purchase_proposal, + get_purchase_operation_timeline, + proposal_to_execution_request, + submit_purchase_proposal, + ) + + proposal = _prepare(proposal_id="PROP-TIMELINE-REDACT") + execution = proposal_to_execution_request(proposal) + submit_purchase_proposal(proposal, actor="planner") + sensitive_reason = ( + "API_TOKEN=timeline-secret; C:\\private\\erp.env; " + "SELECT * FROM credentials; " + ) + decide_purchase_proposal( + ApprovalDecision( + proposal_id=proposal.proposal_id, + outcome="reject", + reason=sensitive_reason, + ), + actor="approver", + ) + + timeline = get_purchase_operation_timeline( + execution.operation_id, actor="approver" + ) + allowed_fields = { + "kind", + "operation_id", + "time", + "actor", + "proposal_id", + "approval_id", + "receipt_id", + "summary", + } + assert all(set(event) <= allowed_fields for event in timeline) + serialized = json.dumps(timeline, ensure_ascii=False) + assert "timeline-secret" not in serialized + assert "private\\erp.env" not in serialized + assert "credentials" not in serialized + assert "