""" FreeExile Design Pipeline Manager & Cross-Department Autonomous Dispatcher. Enforces the 2026 Autonomous Game Studio Directives for the Art & UI/UX Department. Manages the end-to-end lifecycle of Design Requests (DRQ): Inbox -> In-Progress -> Pending Review -> 2-Stage Approval (QA + CEO) -> Approved/Rejected. """ from __future__ import annotations import json import os from pathlib import Path import re import shutil import sys import time from typing import Any if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8") # Ensure repository root is on sys.path for direct script invocation _REPO_ROOT = Path(__file__).resolve().parent.parent.parent if str(_REPO_ROOT) not in sys.path: sys.path.insert(0, str(_REPO_ROOT)) from tools.asset_pipeline.design_pipeline_types import ( ApprovalVerdict, AssetCategory, AssetDeliverable, AuditReport, DesignRequest, Priority, RequestStatus, is_power_of_two, ) class DesignPipelineManager: """Core autonomous workflow engine for FreeExile Design Department.""" def __init__(self, workspace_root: Path | str | None = None) -> None: if workspace_root is None: # Default to repo root (2 levels up from tools/asset_pipeline) self.workspace_root = Path(__file__).resolve().parent.parent.parent else: self.workspace_root = Path(workspace_root).resolve() self.docs_dir = self.workspace_root / "docs" self.design_requests_dir = self.docs_dir / "design_requests" self.templates_dir = self.design_requests_dir / "templates" self.inbox_dir = self.design_requests_dir / "inbox" self.in_progress_dir = self.design_requests_dir / "in_progress" self.pending_review_dir = self.design_requests_dir / "pending_review" self.approved_dir = self.design_requests_dir / "approved" self.rejected_dir = self.design_requests_dir / "rejected" self.assets_root = self.workspace_root / "assets" self.assets_pending = self.assets_root / "pending_review" self.assets_approved = self.assets_root / "approved" self.client_webapp_assets = self.workspace_root / "client" / "webapp" / "assets" self._ensure_directories() def _ensure_directories(self) -> None: """Creates all necessary directory pipelines if missing.""" dirs = ( self.templates_dir, self.inbox_dir, self.in_progress_dir, self.pending_review_dir, self.approved_dir, self.rejected_dir, self.assets_pending, self.assets_approved, ) for d in dirs: d.mkdir(parents=True, exist_ok=True) def parse_request_file(self, file_path: Path) -> DesignRequest: """Parses Markdown Design Request file into structured DesignRequest object.""" content = file_path.read_text(encoding="utf-8") # Determine status from parent folder name parent_name = file_path.parent.name try: status = RequestStatus(parent_name) except ValueError: status = RequestStatus.INBOX # Extract fields using regex req_id_match = re.search(r"\*\*Mã Yêu Cầu \(Request ID\)\*\*:\s*`?([A-Za-z0-9_\-]+)`?", content) request_id = req_id_match.group(1) if req_id_match else file_path.stem title_match = re.search(r"\*\*Tên Yêu Cầu \(Title\)\*\*:\s*(.+)", content) title = title_match.group(1).strip() if title_match else file_path.stem dept_match = re.search(r"\*\*Phòng Ban Khởi Tạo \(Requesting Dept\)\*\*:\s*(.+)", content) requesting_dept = dept_match.group(1).strip() if dept_match else "Documentation" source_match = re.search(r"\*\*Tài Liệu Nguồn Căn Cứ \(Source Document\)\*\*:\s*(.+)", content) source_doc = source_match.group(1).strip() if source_match else "" cat_match = re.search(r"\*\*Phân Loại Tài Nguyên \(Asset Category\)\*\*:\s*`?([A-Za-z0-9_]+)`?", content) cat_str = cat_match.group(1).lower() if cat_match else "concept" try: category = AssetCategory(cat_str) except ValueError: category = AssetCategory.CONCEPT prio_match = re.search(r"\*\*Mức Độ Ưu Tiên \(Priority\)\*\*:\s*`?([A-Za-z0-9_]+)`?", content) prio_str = prio_match.group(1).lower() if prio_match else "medium" try: priority = Priority(prio_str) except ValueError: priority = Priority.MEDIUM agent_match = re.search(r"\*\*Sub-Agent Tiếp Nhận Chỉ Định \(Assigned Subagent\)\*\*:\s*`?([A-Za-z0-9_]+)`?", content) assigned_agent = agent_match.group(1).strip() if agent_match else "art_ui_director" date_match = re.search(r"\*\*Ngày Khởi Tạo \(Created Date\)\*\*:\s*`?([0-9\-]+)`?", content) created_at = date_match.group(1).strip() if date_match else "2026-09-29" sla_match = re.search(r"\*\*Hạn Định Mục Tiêu \(Target SLA\)\*\*:\s*`?([^`\n]+)`?", content) target_sla = sla_match.group(1).strip() if sla_match else "48h" return DesignRequest( request_id=request_id, title=title, source_doc=source_doc, requesting_dept=requesting_dept, category=category, priority=priority, assigned_agent=assigned_agent, status=status, created_at=created_at, target_sla=target_sla, file_path=file_path, ) def create_request( self, request_id: str, title: str, source_doc: str, requesting_dept: str, category: AssetCategory, priority: Priority, assigned_agent: str, lore_context: str, specs_dict: dict[str, str], deliverables: list[str], target_sla: str = "48h", ) -> DesignRequest: """Creates a standardized design request markdown in inbox/.""" target_file = self.inbox_dir / f"{request_id}.md" specs_rows = "\n".join(f"| **{k}** | {v} |" for k, v in specs_dict.items()) deliverables_checklist = "\n".join(f"- [ ] `{d}`" for d in deliverables) markdown_body = f"""# FREEEXILE DESIGN REQUEST: {title} ## 1. THÔNG TIN YÊU CẦU (METADATA) - **Mã Yêu Cầu (Request ID)**: `{request_id}` - **Tên Yêu Cầu (Title)**: {title} - **Phòng Ban Khởi Tạo (Requesting Dept)**: {requesting_dept} - **Tài Liệu Nguồn Căn Cứ (Source Document)**: {source_doc} - **Phân Loại Tài Nguyên (Asset Category)**: `{category.value.upper()}` - **Mức Độ Ưu Tiên (Priority)**: `{priority.value.upper()}` - **Sub-Agent Tiếp Nhận Chỉ Định (Assigned Subagent)**: `{assigned_agent}` - **Ngày Khởi Tạo (Created Date)**: 2026-09-29 - **Hạn Định Mục Tiêu (Target SLA)**: `{target_sla}` --- ## 2. BỐI CẢNH NGHIỆP VỤ & CỐT TRUYỆN (CONTEXT & LORE) ### 2.1. Bối Cảnh Võ Học & Tính Năng {lore_context} --- ## 3. THÔNG SỐ KỸ THUẬT CHI TIẾT (TECHNICAL SPECIFICATIONS) | Tiêu Chí Kỹ Thuật | Yêu Cầu Chi Tiết | | :--- | :--- | {specs_rows} --- ## 4. DANH MỤC SẢN PHẨM ĐẦU RA MONG ĐỢI (DELIVERABLES) {deliverables_checklist} --- ## 5. BIÊN BẢN NGHIỆM THU 2 CẤP (TWO-STAGE APPROVAL GATE) ### Cấp 1: Thẩm Định Kỹ Thuật & Hiệu Năng (QA Technical Audit) *Người thực hiện: `qa_stress_tester`* - [ ] Kích thước tệp tối ưu. - [ ] Kích thước ảnh chuẩn lũy thừa 2 ($2^n$). - [ ] Không gây drop frame (< 8.33ms/frame 120Hz). - [ ] Normal map chuẩn Tangent-Space. - **Kết Quả Cấp 1**: `PENDING` - **Ghi Chú Kỹ Thuật**: Chưa nghiệm thu ### Cấp 2: Nghiệm Thu Thẩm Mỹ & Cốt Truyện (CEO & Art Director Audit) *Người thực hiện: `studio_producer` & `art_ui_director`* - [ ] Đúng phong cách Cổ Võ Hắc Ám 2.5D. - [ ] Đúng tinh thần VLTK1 hiện đại hóa. - [ ] 100% Nguyên bản (Original IP). - **Kết Quả Cấp 2**: `PENDING` - **Ý Kiến Phê Duyệt**: Chưa phê duyệt """ target_file.write_text(markdown_body, encoding="utf-8") return self.parse_request_file(target_file) def scan_inbox(self) -> list[DesignRequest]: """Scans the inbox folder and returns requests sorted by priority (CRITICAL -> HIGH -> MEDIUM -> LOW).""" requests: list[DesignRequest] = [] for file in self.inbox_dir.glob("*.md"): try: req = self.parse_request_file(file) requests.append(req) except Exception as e: print(f"[WARN] Failed to parse request file {file.name}: {e}", file=sys.stderr) # Sort by priority rank ascending (0: critical, 1: high...) requests.sort(key=lambda r: r.priority.rank) return requests def dispatch_request(self, request_id: str, assigned_agent: str | None = None) -> DesignRequest: """Moves a request from inbox/ to in_progress/ and updates assigned subagent if provided.""" src_file = self.inbox_dir / f"{request_id}.md" if not src_file.exists(): # Check if already in progress prog_file = self.in_progress_dir / f"{request_id}.md" if prog_file.exists(): return self.parse_request_file(prog_file) raise FileNotFoundError(f"Request {request_id} not found in inbox ({src_file})") dst_file = self.in_progress_dir / f"{request_id}.md" content = src_file.read_text(encoding="utf-8") if assigned_agent: content = re.sub( r"(\*\*Sub-Agent Tiếp Nhận Chỉ Định \(Assigned Subagent\)\*\*:\s*)`?[A-Za-z0-9_]+`?", f"\\g<1>`{assigned_agent}`", content, ) dst_file.write_text(content, encoding="utf-8") src_file.unlink() # Create assets working directory req_asset_dir = self.assets_pending / request_id req_asset_dir.mkdir(parents=True, exist_ok=True) return self.parse_request_file(dst_file) def submit_for_review(self, request_id: str, deliverables: list[str | Path]) -> DesignRequest: """Moves request from in_progress/ to pending_review/ and records submitted assets.""" src_file = self.in_progress_dir / f"{request_id}.md" if not src_file.exists(): raise FileNotFoundError(f"Request {request_id} not found in in_progress ({src_file})") dst_file = self.pending_review_dir / f"{request_id}.md" content = src_file.read_text(encoding="utf-8") # Mark deliverables as checked for item in deliverables: item_name = Path(item).name content = content.replace(f"- [ ] `{item_name}`", f"- [x] `{item_name}`") content = content.replace(f"- [ ] {item_name}", f"- [x] {item_name}") dst_file.write_text(content, encoding="utf-8") src_file.unlink() return self.parse_request_file(dst_file) def qa_technical_audit( self, request_id: str, auditor: str = "qa_stress_tester", max_file_size_kb: int = 2048, require_power_of_two: bool = True, override_verdict: ApprovalVerdict | None = None, notes: str = "", ) -> tuple[bool, str, list[AssetDeliverable]]: """ Stage 1 Gate: QA performs technical verification on submitted assets in assets/pending_review/{request_id}/. Checks: file sizes, power-of-two image resolutions, normal map presence. """ req_file = self.pending_review_dir / f"{request_id}.md" if not req_file.exists(): raise FileNotFoundError(f"Request {request_id} not found in pending_review ({req_file})") req_assets_dir = self.assets_pending / request_id deliverables: list[AssetDeliverable] = [] errors: list[str] = [] if not req_assets_dir.exists() or not any(req_assets_dir.iterdir()): errors.append(f"No asset files found in {req_assets_dir}") has_image = False has_normal = False if req_assets_dir.exists(): for f in req_assets_dir.glob("*"): if f.is_file(): size_kb = f.stat().st_size / 1024.0 width, height = 0, 0 is_pot = True is_img = f.suffix.lower() in [".png", ".webp", ".jpg", ".jpeg"] if is_img: has_image = True try: from PIL import Image with Image.open(f) as img: width, height = img.size if require_power_of_two: if not (is_power_of_two(width) and is_power_of_two(height)): is_pot = False errors.append(f"File {f.name} dimensions ({width}x{height}) are not power of 2.") except Exception as e: errors.append(f"Could not read image {f.name}: {e}") if "normal" in f.name.lower(): has_normal = True if size_kb > max_file_size_kb: errors.append(f"File {f.name} size ({size_kb:.1f}KB) exceeds limit of {max_file_size_kb}KB.") deliverables.append( AssetDeliverable( relative_path=f.name, file_size_bytes=f.stat().st_size, width=width, height=height, format=f.suffix.lower(), is_power_of_two=is_pot, has_normal_map=has_normal, ) ) # Decide verdict if override_verdict is not None: passed = override_verdict == ApprovalVerdict.PASS else: passed = len(errors) == 0 verdict_str = "PASS" if passed else "FAIL" audit_note = notes or ("; ".join(errors) if not passed else f"Checked {len(deliverables)} files, all passed specs.") # Update markdown content = req_file.read_text(encoding="utf-8") if "- **Kết Quả Cấp 1**:" in content: content = re.sub(r"- \*\*Kết Quả Cấp 1\*\*:\s*`?[^`\n]+`?", f"- **Kết Quả Cấp 1**: `{verdict_str}`", content) content = re.sub(r"- \*\*Ghi Chú Kỹ Thuật\*\*:\s*.+", f"- **Ghi Chú Kỹ Thuật**: {audit_note}", content) else: content += f"\n- **Kết Quả Cấp 1**: `{verdict_str}`\n- **Ghi Chú Kỹ Thuật**: {audit_note}\n" if not passed: # Move to rejected/ rej_file = self.rejected_dir / f"{request_id}.md" rej_file.write_text(content, encoding="utf-8") req_file.unlink() else: req_file.write_text(content, encoding="utf-8") return passed, audit_note, deliverables def ceo_aesthetic_audit( self, request_id: str, approver: str = "studio_producer", verdict: ApprovalVerdict = ApprovalVerdict.APPROVE, notes: str = "Approved - Thẩm mỹ Cổ Võ Hắc Ám xuất sắc, chuẩn phong cách VLTK1 hiện đại.", sync_to_client_subfolder: str | None = None, ) -> DesignRequest: """ Stage 2 Gate: CEO & Art Director perform final aesthetic, lore and IP audit. If APPROVED: Moves to approved/ and synchronizes assets to assets/approved/ and client/webapp/assets. If REJECTED: Moves to rejected/ with reviewer feedback. """ req_file = self.pending_review_dir / f"{request_id}.md" if not req_file.exists(): raise FileNotFoundError(f"Request {request_id} not found in pending_review ({req_file})") content = req_file.read_text(encoding="utf-8") # Verify Stage 1 passed first if "- **Kết Quả Cấp 1**: `PASS`" not in content: raise ValueError(f"Request {request_id} cannot be reviewed by CEO because Stage 1 QA did not PASS.") verdict_str = "APPROVED" if verdict == ApprovalVerdict.APPROVE else "REJECTED" if "- **Kết Quả Cấp 2**:" in content: content = re.sub(r"- \*\*Kết Quả Cấp 2\*\*:\s*`?[^`\n]+`?", f"- **Kết Quả Cấp 2**: `{verdict_str}`", content) content = re.sub(r"- \*\*Ý Kiến Phê Duyệt\*\*:\s*.+", f"- **Ý Kiến Phê Duyệt**: ({approver}) {notes}", content) else: content += f"\n- **Kết Quả Cấp 2**: `{verdict_str}`\n- **Ý Kiến Phê Duyệt**: ({approver}) {notes}\n" if verdict == ApprovalVerdict.APPROVE: dst_file = self.approved_dir / f"{request_id}.md" dst_file.write_text(content, encoding="utf-8") req_file.unlink() # Synchronize assets src_assets = self.assets_pending / request_id dst_assets = self.assets_approved / request_id dst_assets.mkdir(parents=True, exist_ok=True) if src_assets.exists(): for f in src_assets.glob("*"): if f.is_file(): shutil.copy2(f, dst_assets / f.name) # Sync to client webapp if subfolder specified if sync_to_client_subfolder: client_dir = self.client_webapp_assets / sync_to_client_subfolder client_dir.mkdir(parents=True, exist_ok=True) shutil.copy2(f, client_dir / f.name) return self.parse_request_file(dst_file) else: dst_file = self.rejected_dir / f"{request_id}.md" dst_file.write_text(content, encoding="utf-8") req_file.unlink() return self.parse_request_file(dst_file) def get_pipeline_summary(self) -> dict[str, int]: """Returns the count of requests across each stage.""" return { "inbox": len(list(self.inbox_dir.glob("*.md"))), "in_progress": len(list(self.in_progress_dir.glob("*.md"))), "pending_review": len(list(self.pending_review_dir.glob("*.md"))), "approved": len(list(self.approved_dir.glob("*.md"))), "rejected": len(list(self.rejected_dir.glob("*.md"))), } def generate_status_report(self) -> str: """Generates a comprehensive markdown report of the entire design pipeline.""" summary = self.get_pipeline_summary() lines = [ "# FREEEXILE DESIGN PIPELINE STATUS REPORT", f"*Generated At: 2026-09-29* | *Total Requests: {sum(summary.values())}*", "", "## 1. THỐNG KÊ TỔNG QUAN THEO GIAI ĐOẠN", f"- **Inbox (Yêu cầu mới)**: `{summary['inbox']}`", f"- **In Progress (Đang thực hiện)**: `{summary['in_progress']}`", f"- **Pending Review (Chờ nghiệm thu 2 cấp)**: `{summary['pending_review']}`", f"- **Approved (Đã nghiệm thu chính thức)**: `{summary['approved']}`", f"- **Rejected (Từ chối / Cần chỉnh sửa)**: `{summary['rejected']}`", "", "## 2. DANH SÁCH CHI TIẾT THEO TỪNG GIAI ĐOẠN", ] stages = [ ("Inbox", self.inbox_dir), ("In Progress", self.in_progress_dir), ("Pending Review", self.pending_review_dir), ("Approved", self.approved_dir), ("Rejected", self.rejected_dir), ] for stage_name, folder in stages: lines.append(f"### {stage_name} ({len(list(folder.glob('*.md')))})") files = list(folder.glob("*.md")) if not files: lines.append("_Không có yêu cầu nào trong giai đoạn này._\n") continue for f in files: try: req = self.parse_request_file(f) lines.append( f"- **`{req.request_id}`**: {req.title} | Priority: `{req.priority.value.upper()}` | " f"Assigned: `{req.assigned_agent}` | Dept: `{req.requesting_dept}`" ) except Exception: lines.append(f"- **{f.stem}**: (Parse error)") lines.append("") return "\n".join(lines) # ============================================================================= # CLI ENTRY POINT # ============================================================================= def main() -> None: from tools.asset_pipeline.design_pipeline_cli import main as cli_main cli_main() if __name__ == "__main__": main()