aiagent/backend/app/services/company_orchestrator.py

"""
虚拟公司编排引擎 — CEO 战略规划 → 部门并行执行 → CEO 汇总审查

用法:
    orchestrator = CompanyOrchestrator(db, company_id, user_id)
    result = await orchestrator.execute("开发新产品X")
    async for event in orchestrator.execute_stream("开发新产品X"):
        ...
"""
from __future__ import annotations

import asyncio
import json
import logging
import re
import difflib
from datetime import datetime
from pathlib import Path
from typing import Any, AsyncGenerator, Dict, List, Optional

from sqlalchemy.orm import Session

from app.core.database import SessionLocal
from app.models.agent import Agent
from app.models.company import Company, CompanyProject
from app.models.company_knowledge import CompanyKnowledge
from app.models.team import Team, TeamMember
from app.agent_runtime.core import AgentRuntime
from app.agent_runtime.schemas import (
    AgentConfig,
    AgentLLMConfig,
    AgentToolConfig,
    AgentMemoryConfig,
)

logger = logging.getLogger(__name__)

# ─── CEO 系统提示词 ───

CEO_SYSTEM_PROMPT = """You are the Chief Executive Officer (CEO) of a virtual company. Your role is to receive high-level business objectives and decompose them into executable departmental plans.

Your responsibilities:
1. **Strategic Analysis**: Understand the company's goal, market context, and constraints
2. **Task Decomposition**: Break the objective into department-level tasks with clear deliverables
3. **Dependency Mapping**: Identify which departments depend on others' outputs
4. **Resource Allocation**: Assign priorities and suggest which department tackles what

When you receive a company objective:
1. First, analyze the objective — what is the desired outcome? What are the constraints? What industry are we in?
2. Then, identify which departments need to be involved
3. For each department, define: the specific goal, expected deliverables, priority, and any dependencies
4. Output your plan in strict JSON format

Your company has the following departments (with their available agent roles):
{departments_desc}

IMPORTANT: Each department can ONLY execute tasks that match its available_roles. Tasks assigned to roles that don't exist in that department will be SKIPPED entirely. For example, if a department only has ["产品经理", "UI/UX设计师"], do NOT assign ceo/cto/backend tasks to it. Plan work that the actual agents can do.

Output format — respond with ONLY the JSON object below, no markdown, no explanation, no code fences:
{{
  "analysis": "Brief strategic analysis of the objective (2-3 sentences)",
  "overall_timeline": "Estimated timeline for the entire project",
  "key_success_metrics": ["metric 1", "metric 2"],
  "risks": ["risk 1", "risk 2"],
  "departments": [
    {{
      "department_name": "Department name (exact match from the list above)",
      "department_type": "executive/product/engineering/marketing/operations/sales/finance/hr",
      "goal": "Specific goal for this department — MUST be achievable by its available_roles",
      "deliverables": ["deliverable 1", "deliverable 2"],
      "priority": 1,
      "dependencies": ["other_department_name or empty"],
      "key_questions": ["question this department should answer"]
    }}
  ]
}}

CRITICAL RULES:
- Output ONLY the JSON object. No prefix text, no suffix text, no markdown code fences.
- Only include departments that are actually needed for this objective
- department_name MUST exactly match the department name from the list above
- Assign priority 1 to the most critical/blocking departments
- Be specific about deliverables — vague goals lead to poor execution
- Dependencies should be department names (not agent names)
- DO NOT plan tasks that require roles not in the department's available_roles list"""


def _parse_ceo_json(raw_output: str) -> Dict[str, Any]:
    """Extract JSON from CEO output (with markdown and fallback)."""
    text = raw_output.strip()

    # Try direct parse
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        pass

    # Try all ```json ... ``` code blocks
    for m in re.finditer(r'```(?:json)?\s*\n?(.*?)```', text, re.DOTALL):
        try:
            return json.loads(m.group(1).strip())
        except json.JSONDecodeError:
            continue

    # Try bracket-counting: find balanced { } containing "departments"
    dep_idx = text.find('"departments"')
    if dep_idx >= 0:
        brace_start = text.rfind('{', 0, dep_idx)
        if brace_start >= 0:
            depth = 0
            for i, c in enumerate(text[brace_start:], brace_start):
                if c == '{':
                    depth += 1
                elif c == '}':
                    depth -= 1
                    if depth == 0:
                        try:
                            return json.loads(text[brace_start:i + 1])
                        except json.JSONDecodeError:
                            # Try fixing common LLM JSON issues
                            candidate = text[brace_start:i + 1]
                            fixed = re.sub(r',\s*}', '}', candidate)
                            fixed = re.sub(r',\s*]', ']', fixed)
                            try:
                                return json.loads(fixed)
                            except json.JSONDecodeError:
                                pass
                        break

    logger.warning("Failed to parse CEO JSON plan from output: %.500s", text)
    return {}


# ─── 四档-②：历史知识相关性筛选（学习闭环）───

# 领域信号词——用于粗粒度中文主题匹配（与 knowledge_extractor._extract_tags 同源）
_MEMORY_DOMAIN_KEYWORDS = [
    "市场", "产品", "技术", "营销", "销售", "运营", "风险", "增长", "用户", "创新",
    "架构", "前端", "后端", "数据", "设计", "测试", "安全", "成本", "内容", "策略",
]

# category → 中文标签 + 权重（review/insight 是"教训"，规划最该参考，权重更高）
_MEMORY_CATEGORY_META = {
    "review": ("审查教训", 2.0),
    "insight": ("风险与指标", 2.0),
    "ceo_plan": ("战略分析", 1.0),
    "dept_output": ("部门产出", 0.5),
    "general": ("经验", 0.5),
}


def _select_relevant_knowledge(
    entries: List[Dict[str, Any]],
    project_desc: str,
    max_entries: int,
    max_chars: int,
) -> str:
    """从本公司历史知识里挑与新项目相关的，格式化成注入 CEO prompt 的「历史项目经验」块。

    纯函数（不碰 DB），便于隔离测试。相关性 = 领域词在知识正文命中 + tag 命中新项目描述；
    再叠加 category 权重（review/insight 优先）。**topical<=0 的条目不注入**（只喂真相关的，避免噪声）。
    无相关条目 → 返回 ""（回归旧行为，不改 CEO prompt）。
    """
    if not entries or max_entries <= 0 or max_chars <= 0:
        return ""
    desc = project_desc or ""
    proj_terms = [kw for kw in _MEMORY_DOMAIN_KEYWORDS if kw in desc]

    scored = []
    for e in entries:
        title = (e.get("title") or "").strip()
        content = (e.get("content") or "").strip()
        category = e.get("category") or "general"
        tags = e.get("tags") or []
        hay = f"{title}\n{content}"
        # 主题命中：领域词出现在知识正文 + 该知识 tag 出现在新项目描述
        keyword_hits = sum(1 for t in proj_terms if t in hay)
        tag_hits = sum(1 for tg in tags if tg and str(tg) in desc)
        topical = keyword_hits * 1.0 + tag_hits * 2.0
        if topical <= 0:
            continue
        _, cat_w = _MEMORY_CATEGORY_META.get(category, ("经验", 0.5))
        final = topical + cat_w
        scored.append((final, e.get("_recency", 0), e))

    if not scored:
        return ""
    scored.sort(key=lambda x: (x[0], x[1]), reverse=True)

    sections: List[str] = []
    used = 0
    for _, _rec, e in scored[:max_entries]:
        cat_cn, _ = _MEMORY_CATEGORY_META.get(e.get("category") or "general", ("经验", 0.5))
        title = (e.get("title") or "").strip()
        content = (e.get("content") or "").strip()
        block = f"【{cat_cn}】{title}\n{content}"
        remaining = max_chars - used
        if remaining <= 0:
            break
        if len(block) > remaining:
            block = block[:remaining]
        sections.append(block)
        used += len(block)

    if not sections:
        return ""
    return (
        "## 历史项目经验（本公司过往项目沉淀，供规划参考：复用被验证有效的做法、规避已知风险与低分项）\n"
        + "\n\n---\n\n".join(sections)
    )


class CompanyOrchestrator:
    """虚拟公司编排器 — 多部门并行协作执行。"""

    def __init__(
        self,
        db: Session,
        company_id: str,
        user_id: str,
        auto_approve_files: bool = True,
    ):
        self.db = db
        self.company_id = company_id
        self.user_id = user_id
        self.auto_approve_files = auto_approve_files
        self._dept_sem: Optional[asyncio.Semaphore] = None  # 三档-b：每波部门并发信号量（lazy）

    def _get_dept_sem(self) -> asyncio.Semaphore:
        """取（缺则建）本 run 的部门并发信号量：限制每一波并行运行的部门数。

        每个编排实例独立，避免不同项目相互耦合；也覆盖 rework_single_department 复用
        _run_department_stream 的路径。lazy 构造绑定到运行时事件循环。
        """
        if self._dept_sem is None:
            from app.core.config import settings
            self._dept_sem = asyncio.Semaphore(max(1, int(settings.MAX_CONCURRENT_DEPARTMENTS)))
        return self._dept_sem

    def _load_company(self) -> Company:
        company = self.db.query(Company).filter(Company.id == self.company_id).first()
        if not company:
            raise ValueError(f"公司不存在: {self.company_id}")
        return company

    def _load_departments(self) -> List[Team]:
        return (
            self.db.query(Team)
            .filter(Team.parent_company_id == self.company_id)
            .order_by(Team.position if hasattr(Team, 'position') else Team.created_at)
            .all()
        )

    def _auto_extract_knowledge(self, project) -> None:
        """四档-②：项目终态时自动把本次项目沉淀为知识（闭合学习闭环）。

        受 settings 门控；按 project_id 去重（避免与手动 generate 或返工重复落库）；
        任何异常一律吞掉（学习沉淀失败绝不影响项目收尾）。
        """
        from app.core.config import settings
        if not settings.COMPANY_AUTO_EXTRACT_KNOWLEDGE:
            return
        try:
            exists = (
                self.db.query(CompanyKnowledge)
                .filter(CompanyKnowledge.project_id == project.id)
                .first()
            )
            if exists:
                return  # 已沉淀过，跳过
            from app.services.company_knowledge_extractor import extract_from_project
            entries = extract_from_project(project, self.db)
            self.db.commit()
            logger.info("项目终态自动沉淀知识: project=%s entries=%d", project.id, len(entries))
        except Exception as e:
            try:
                self.db.rollback()
            except Exception:
                pass
            logger.warning("自动沉淀知识失败(忽略): project=%s err=%s", getattr(project, "id", "?"), e)

    def _load_ceo_memory(self, project_desc: str) -> str:
        """四档-②：检索本公司历史知识，返回注入 CEO prompt 的「历史项目经验」块（或 ""）。

        受 settings 门控与上限约束；DB/检索异常一律降级为空串（学习闭环失败不阻断规划）。
        """
        from app.core.config import settings
        if not settings.COMPANY_CEO_MEMORY_ENABLED:
            return ""
        try:
            rows = (
                self.db.query(CompanyKnowledge)
                .filter(CompanyKnowledge.company_id == self.company_id)
                .order_by(CompanyKnowledge.created_at.desc())
                .limit(200)
                .all()
            )
        except Exception as e:
            logger.warning("加载 CEO 历史知识失败(降级为空): company=%s err=%s", self.company_id, e)
            return ""
        entries = [
            {
                "title": r.title,
                "content": r.content,
                "category": r.category,
                "tags": r.tags or [],
                "_recency": r.created_at.timestamp() if r.created_at else 0,
            }
            for r in rows
        ]
        return _select_relevant_knowledge(
            entries,
            project_desc,
            int(settings.COMPANY_CEO_MEMORY_MAX_ENTRIES),
            int(settings.COMPANY_CEO_MEMORY_MAX_CHARS),
        )

    async def _run_ceo_planning(self, company: Company, departments: List[Team], project_desc: str) -> Dict[str, Any]:
        """Phase 1: CEO 战略规划"""
        # Build CEO agent config
        ceo_agent = None
        if company.ceo_agent_id:
            ceo_agent = self.db.query(Agent).filter(Agent.id == company.ceo_agent_id).first()

        if not ceo_agent:
            # Find any agent with CEO-like role
            ceo_agent = self.db.query(Agent).filter(
                Agent.user_id == self.user_id,
                Agent.name.like("%CEO%")
            ).first()

        if not ceo_agent:
            raise ValueError("公司没有 CEO Agent，请先指定 CEO Agent")

        # Build department description for CEO prompt — include available agent roles
        dept_descs = []
        for d in departments:
            members = d.members if hasattr(d, 'members') else []
            member_roles = []
            for m in members:
                agent = m.agent if hasattr(m, 'agent') else None
                if agent:
                    role = m.role or agent.name or "unknown"
                    member_roles.append(role)
            dept_descs.append(
                f"- **{d.name}** (type: {d.department_type or 'general'}, "
                f"available_roles: {member_roles if member_roles else ['none']}, "
                f"members: {', '.join(member_roles) if member_roles else 'none'})"
            )
        dept_desc_str = "\n".join(dept_descs) if dept_descs else "No departments configured"

        # Build CEO prompt
        ceo_prompt = CEO_SYSTEM_PROMPT.replace("{departments_desc}", dept_desc_str)

        # 四档-②：学习闭环——把本公司历史项目知识按相关性注入 CEO prompt
        memory_block = self._load_ceo_memory(project_desc)
        if memory_block:
            ceo_prompt = f"{ceo_prompt}\n\n{memory_block}"
            logger.info("CEO 规划注入历史知识: company=%s chars=%d", self.company_id, len(memory_block))

        # Create runtime config
        wf = ceo_agent.workflow_config or {}
        nodes = wf.get("nodes", [])
        llm_node = None
        for n in nodes:
            if n.get("type") == "llm":
                llm_node = n
                break
        llm_data = llm_node.get("data", {}) if llm_node else {}

        config = AgentConfig(
            name=ceo_agent.name,
            system_prompt=ceo_prompt,
            user_id=self.user_id,
            memory_scope_id=f"company_ceo_{self.company_id}",
            llm=AgentLLMConfig(
                provider=llm_data.get("provider", "deepseek"),
                model=llm_data.get("model", "deepseek-v4-pro"),
                temperature=float(llm_data.get("temperature", 0.4)),
                max_tokens=4096,  # Ensure enough output for full department plan
                max_iterations=3,  # CEO planning should be concise
            ),
            tools=AgentToolConfig(include_tools=["web_search", "text_analyze", "task_plan"]),
            memory=AgentMemoryConfig(enabled=True, max_history_messages=10),
        )

        runtime = AgentRuntime(config)
        result = await runtime.run(project_desc)
        plan = _parse_ceo_json(result.content)

        logger.info("CEO 规划完成: company=%s departments=%d", self.company_id, len(plan.get("departments", [])))
        return plan

    # 四档-①：部门间交接产物——把已完成上游依赖部门的交付物注入下游部门的 context
    _HANDOFF_PER_DEPT_CHARS = 2000   # 每个上游交付物注入上限
    _HANDOFF_TOTAL_CHARS = 6000      # 全部上游交付物合计上限（防 prompt 膨胀）

    def _build_dept_context(
        self,
        dept_plan: Dict[str, Any],
        ceo_analysis: str,
        completed_deliverables: Dict[str, str],
    ) -> str:
        """构造某部门的执行 context = CEO 总体分析 + 其依赖的上游部门交付物摘要。

        只按该部门声明的 `dependencies` 精准注入（无依赖 → 仅 CEO 分析，回归旧行为）。
        逐项与合计各有字数上限，避免把下游 prompt 撑爆。
        """
        base = ceo_analysis or ""
        deps = dept_plan.get("dependencies", [])
        if isinstance(deps, str):
            deps = [deps] if deps else []
        if not deps:
            return base

        sections: List[str] = []
        used = 0
        for dep_name in deps:
            deliverable = (completed_deliverables.get(dep_name) or "").strip()
            if not deliverable:
                continue
            snippet = deliverable[: self._HANDOFF_PER_DEPT_CHARS]
            if used + len(snippet) > self._HANDOFF_TOTAL_CHARS:
                snippet = snippet[: max(0, self._HANDOFF_TOTAL_CHARS - used)]
            if not snippet:
                break
            sections.append(f"— 上游部门「{dep_name}」的交付物：\n{snippet}")
            used += len(snippet)
            if used >= self._HANDOFF_TOTAL_CHARS:
                break

        if not sections:
            return base
        handoff = (
            "【上游部门已完成的交付物——请在此基础上衔接，勿重复其工作】\n"
            + "\n\n".join(sections)
        )
        return f"{base}\n\n{handoff}" if base else handoff

    async def _run_department(
        self,
        department: Team,
        dept_plan: Dict[str, Any],
        context: str,
        project_path: Path,
    ) -> Dict[str, Any]:
        """Phase 2: 运行单个部门（复用 TeamOrchestrator，非流式）。"""
        from app.services.team_orchestrator import TeamOrchestrator

        goal = dept_plan.get("goal", context)
        deliverables = dept_plan.get("deliverables", [])
        dept_task = f"{goal}\n\nExpected deliverables:\n" + "\n".join(f"- {d}" for d in deliverables)
        if context:
            dept_task = f"Company context: {context}\n\nDepartment task: {dept_task}"

        # per-dept 独立 session：execute() 用 asyncio.gather 并行多部门，
        # 共享 self.db 会产生 SQLAlchemy 事务态竞态；各部门自建、用完即关。
        dept_db = SessionLocal()
        try:
            orchestrator = TeamOrchestrator(
                db=dept_db,
                team_id=department.id,
                user_id=self.user_id,
                auto_approve_files=self.auto_approve_files,
            )
            result = await orchestrator.execute(dept_task, auto_approve_files=self.auto_approve_files)
            logger.info("Department %s completed: success=%s", department.name, result.get("success"))
            return {
                "department_id": department.id,
                "department_name": department.name,
                "department_type": department.department_type,
                "success": result.get("success", False),
                "plan": dept_plan,
                "result": result,
            }
        except Exception as e:
            logger.error("Department %s failed: %s", department.name, e)
            return {
                "department_id": department.id,
                "department_name": department.name,
                "department_type": department.department_type,
                "success": False,
                "error": str(e),
                "plan": dept_plan,
                "result": None,
            }
        finally:
            try:
                dept_db.close()
            except Exception:
                pass

    async def _run_department_stream(
        self,
        department: Team,
        dept_plan: Dict[str, Any],
        context: str,
        project_path: Path,
        event_queue: asyncio.Queue,
    ):
        """Phase 2 (stream): 运行单个部门并实时推送 Agent 事件到队列。"""
        from app.services.team_orchestrator import TeamOrchestrator
        from app.services.cost_estimator import estimate_cost_yuan

        goal = dept_plan.get("goal", context)
        deliverables = dept_plan.get("deliverables", [])
        dept_task = f"{goal}\n\nExpected deliverables:\n" + "\n".join(f"- {d}" for d in deliverables)
        if context:
            dept_task = f"Company context: {context}\n\nDepartment task: {dept_task}"
        # M2-b 返工注入：本轮为返工时，把 CEO 评审反馈 + 人工意见拼进任务，驱动针对性改进
        rework_fb = dept_plan.get("rework_feedback", "")
        if rework_fb:
            dept_task += ("\n\n【本轮为返工——请针对以下反馈做针对性改进，不要照抄上一版】\n" + rework_fb)

        dept_result = {
            "department_id": department.id,
            "department_name": department.name,
            "department_type": department.department_type,
            "success": False,
            "plan": dept_plan,
            "result": None,
        }

        # Track live state for multi-panel monitoring
        last_thought = ""
        current_phase = 0
        total_phases = 0
        current_phase_name = ""
        dept_real_tokens = 0  # 真实 token 累加（来自 Agent final.token_usage）
        dept_cost_yuan = 0.0  # 真实成本累加(¥)，按每个 agent 各自 model 分别计价

        sem = self._get_dept_sem()  # 三档-b：每波部门并发上限
        acquired = False
        # per-dept 独立 session：同波并行部门若共享 db_orch，SQLAlchemy Session
        # 在交错 await 下会交叉污染事务态（竞态）。各部门自建、用完即关。
        dept_db = SessionLocal()
        try:
            await sem.acquire()
            acquired = True
            orchestrator = TeamOrchestrator(
                db=dept_db,
                team_id=department.id,
                user_id=self.user_id,
                auto_approve_files=self.auto_approve_files,
            )
            async for team_event in orchestrator.execute_stream(dept_task, auto_approve_files=self.auto_approve_files):
                t_type = team_event.get("type", "unknown")

                # ─── Extract live state for monitoring panels ───
                if t_type == "agent_event":
                    inner = team_event.get("data", team_event)
                    inner_type = inner.get("type", "")
                    if inner_type == "think":
                        last_thought = inner.get("content", "")[:2000]
                    elif inner_type == "final":
                        tu = inner.get("token_usage") or {}
                        try:
                            dept_real_tokens += int(tu.get("total_tokens") or 0)
                        except (TypeError, ValueError):
                            pass
                        try:
                            dept_cost_yuan += estimate_cost_yuan(
                                tu.get("model") or "",
                                int(tu.get("prompt_tokens") or 0),
                                int(tu.get("completion_tokens") or 0),
                            )
                        except (TypeError, ValueError):
                            pass
                elif t_type == "phase_start":
                    current_phase = team_event.get("phase", 0)
                    total_phases = team_event.get("total_phases", total_phases or 1)
                    current_phase_name = team_event.get("name", "")
                elif t_type == "phase_done":
                    current_phase_name = ""

                # Forward as dept-level event
                company_event = {
                    "type": "dept_agent_event",
                    "department_name": department.name,
                    "department_type": department.department_type,
                    "sub_type": t_type,
                    "data": team_event,
                    "last_thought": last_thought,
                    "phase_progress": f"{current_phase}/{total_phases}" if total_phases > 0 else "",
                    "phase_name": current_phase_name,
                    "_ts": __import__("time").time(),
                }
                await event_queue.put(company_event)

                # Capture final result from team events
                if team_event.get("type") == "final":
                    dept_result["success"] = True
                    dept_result["result"] = {
                        "deliverable": team_event.get("deliverable", ""),
                        "files": team_event.get("files", []),
                        "phase_count": team_event.get("phase_count", 0),
                    }
                elif team_event.get("type") == "error":
                    dept_result["success"] = False
                    dept_result["error"] = team_event.get("content", str(team_event))

            # If no final event, mark as success anyway
            if dept_result["result"] is None:
                dept_result["success"] = True

            logger.info("Department (stream) %s completed: success=%s", department.name, dept_result["success"])
        except asyncio.CancelledError:
            # M2-c 单部门取消：吞掉 CancelledError（协作式），仍发 dept_done 让波次推进、
            # 下游依赖照常解锁；标 cancelled 供 UI 显示「已取消」而非 error。
            logger.info("Department (stream) %s cancelled", department.name)
            dept_result["success"] = False
            dept_result["error"] = "cancelled"
            dept_result["cancelled"] = True
            event_queue.put_nowait({
                "type": "dept_agent_event",
                "department_name": department.name,
                "sub_type": "error",
                "data": {"content": "该部门已被取消"},
                "last_thought": last_thought,
                "phase_progress": f"{current_phase}/{total_phases}" if total_phases > 0 else "",
                "phase_name": current_phase_name,
                "_ts": __import__("time").time(),
            })
        except Exception as e:
            logger.error("Department (stream) %s failed: %s", department.name, e)
            dept_result["success"] = False
            dept_result["error"] = str(e)
            await event_queue.put({
                "type": "dept_agent_event",
                "department_name": department.name,
                "sub_type": "error",
                "data": {"content": str(e)},
                "last_thought": last_thought,
                "phase_progress": f"{current_phase}/{total_phases}" if total_phases > 0 else "",
                "phase_name": current_phase_name,
                "_ts": __import__("time").time(),
            })
        finally:
            try:
                dept_db.close()
            except Exception:
                pass
            if acquired:
                sem.release()  # 释放部门名额（在 dept_done 之前），供本波下一个部门启动

        # Signal completion
        await event_queue.put({
            "type": "dept_done",
            "department_name": dept_result["department_name"],
            "success": dept_result["success"],
            "result": dept_result.get("result"),
            "error": dept_result.get("error"),
            "cancelled": dept_result.get("cancelled", False),
            "real_tokens": dept_real_tokens,
            "cost_yuan": round(dept_cost_yuan, 4),
            "_ts": __import__("time").time(),
        })

    async def rework_single_department(self, project_id: str, dept_name: str):
        """M2-c：跑完后重跑单个部门，事件续写进同一流，替换其交付物与打分。

        不走 execute_stream 全流程；只驱动单个 _run_department_stream + 单部门重打分，
        再把该部门的 review_scores 条目替换、状态翻回 completed/failed。
        """
        import time as _time
        project = self.db.query(CompanyProject).filter(CompanyProject.id == project_id).first()
        if project is None:
            yield {"type": "error", "message": "项目不存在", "_ts": _time.time()}
            return
        ceo_plan = project.ceo_plan or {}
        dept_plans = ceo_plan.get("departments", [])
        if not dept_plans:
            yield {"type": "error", "message": "项目无 CEO 计划，无法返工", "_ts": _time.time()}
            return

        company = self._load_company()
        departments = self._load_departments()

        # 匹配 (Team, plan)：精确名/型
        target_plan = next(
            (dp for dp in dept_plans
             if dp.get("department_name") == dept_name or dp.get("department_type") == dept_name),
            None,
        )
        target_dept = next(
            (d for d in departments if d.name == dept_name or d.department_type == dept_name),
            None,
        )
        if target_plan is None or target_dept is None:
            yield {"type": "error", "message": f"部门 {dept_name} 不在计划或已删除", "_ts": _time.time()}
            return

        from app.services import company_runner  # 惰性导入，复用取消能力
        control = company_runner.get_control(project_id)

        context = ceo_plan.get("analysis", "")
        project_path = Path(f"./company_projects/{self.company_id}/{_safe_name(project.description or '')}")
        project_path.mkdir(parents=True, exist_ok=True)

        # 运行单个部门（复用 _run_department_stream）
        q: asyncio.Queue = asyncio.Queue()
        t_start = _time.time()
        task = asyncio.create_task(
            self._run_department_stream(target_dept, target_plan, context, project_path, q)
        )
        control.dept_tasks[target_dept.name] = task
        dept_result: Optional[Dict[str, Any]] = None
        while True:
            evt = await q.get()
            yield evt
            if evt.get("type") == "dept_done":
                dept_result = evt
                break
        control.dept_tasks.pop(target_dept.name, None)
        dur = int((_time.time() - t_start) * 1000)

        # 组装该部门结果供重打分
        one_result = {
            "department_name": (dept_result or {}).get("department_name", dept_name),
            "department_type": target_dept.department_type or "",
            "success": bool((dept_result or {}).get("success")),
            "plan": target_plan,
            "result": (dept_result or {}).get("result"),
            "error": (dept_result or {}).get("error"),
            "duration_ms": dur,
            "tokens_used": int((dept_result or {}).get("real_tokens") or 0),
            "cost_yuan": float((dept_result or {}).get("cost_yuan") or 0.0),
        }

        # 单部门重打分，替换 review_scores 中该条目（知识/洞察读它）
        try:
            review = await self._ceo_review_with_scoring(company, ceo_plan, [one_result], project.description or "")
            new_scores = review.get("departments", [])
            existing = list(project.review_scores or [])
            for ns in new_scores:
                nm = ns.get("name")
                replaced = False
                for i, es in enumerate(existing):
                    if es.get("name") == nm:
                        existing[i] = ns
                        replaced = True
                        break
                if not replaced:
                    existing.append(ns)
            project.review_scores = existing
            self.db.commit()
            yield {
                "type": "ceo_score_done",
                "scores": new_scores,
                "overall_score": review.get("overall_score", 0),
                "summary": review.get("summary", ""),
                "next_steps": review.get("next_steps", ""),
                "round_index": 0,
                "_ts": _time.time(),
            }
        except Exception as e:
            yield {"type": "error", "message": f"返工打分失败: {e}", "phase": "rework_score", "_ts": _time.time()}

        # 收尾：状态翻回 completed/failed
        success = bool(dept_result.get("success")) if dept_result else False
        project.status = "completed" if success else "failed"
        project.completed_at = datetime.now()
        self.db.commit()
        yield {
            "type": "rework_done",
            "department_name": dept_name,
            "success": success,
            "duration_ms": dur,
            "cost_yuan": float(dept_result.get("cost_yuan") or 0.0) if dept_result else 0.0,
            "_ts": _time.time(),
        }

    async def preview_plan(self, project_description: str) -> Dict[str, Any]:
        """仅运行 CEO 规划并返回计划，不派发部门、不落库（无状态预览）。

        供「执行前预览+确认」使用：前端拿到计划后可微调，再携带该计划正式执行。
        """
        company = self._load_company()
        departments = self._load_departments()
        if not departments:
            raise ValueError("公司没有部门，请先添加部门")
        ceo_plan = await self._run_ceo_planning(company, departments, project_description)
        return {
            "company_name": company.name,
            "department_names": [d.name for d in departments],
            "ceo_plan": ceo_plan,
        }

    async def execute(
        self,
        project_description: str,
        auto_approve_files: bool = True,
        ceo_plan: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, Any]:
        """执行公司级项目（非流式）。

        Returns:
            {
                "company_id": str,
                "company_name": str,
                "project_description": str,
                "ceo_plan": {...},
                "departments": [{department_name, success, result}],
                "final_summary": str,
                "success": bool,
            }
        """
        self.auto_approve_files = auto_approve_files
        company = self._load_company()
        departments = self._load_departments()

        if not departments:
            raise ValueError("公司没有部门，请先添加部门")

        # Create project record
        project = CompanyProject(
            company_id=self.company_id,
            name=project_description[:200],
            description=project_description,
            status="planning",
        )
        self.db.add(project)
        self.db.commit()

        # Phase 1: CEO Planning（若已传入已确认的计划则跳过，直接采用）
        if ceo_plan is None:
            logger.info("公司编排 [%s]: Phase 1 — CEO 规划开始", self.company_id)
            ceo_plan = await self._run_ceo_planning(company, departments, project_description)
        else:
            logger.info("公司编排 [%s]: Phase 1 — 采用已确认的 CEO 计划（跳过规划）", self.company_id)
        project.ceo_plan = ceo_plan
        project.status = "in_progress"
        self.db.commit()

        # Phase 2: Department Execution — handle dependencies
        logger.info("公司编排 [%s]: Phase 2 — 部门并行执行 (%d departments)", self.company_id, len(departments))
        dept_plans = ceo_plan.get("departments", [])

        # Build department map
        dept_map: Dict[str, Team] = {}
        for d in departments:
            dept_map[d.name] = d
            if d.department_type:
                dept_map[d.department_type] = d
        norm_dept_map: Dict[str, Team] = {}
        for k, v in dept_map.items():
            nk = (k or "").strip().lower()
            if nk:
                norm_dept_map.setdefault(nk, v)

        # Match plans to departments (精确 → 归一化 → 模糊)
        matched_tasks = []
        for dp in dept_plans:
            dept_name = dp.get("department_name", "")
            dept_type = dp.get("department_type", "")
            target_dept = dept_map.get(dept_name) or dept_map.get(dept_type)
            if not target_dept:
                for key in (dept_name, dept_type):
                    nk = (key or "").strip().lower()
                    if nk and nk in norm_dept_map:
                        target_dept = norm_dept_map[nk]
                        break
            if not target_dept:
                cand = difflib.get_close_matches(
                    (dept_name or dept_type or "").strip().lower(),
                    list(norm_dept_map.keys()), n=1, cutoff=0.6,
                )
                if cand:
                    target_dept = norm_dept_map[cand[0]]
                    logger.warning("Department fuzzy-matched: plan=%s → %s", dept_name or dept_type, target_dept.name)
            if target_dept:
                matched_tasks.append((target_dept, dp))
            else:
                logger.warning("Department not found: name=%s type=%s", dept_name, dept_type)

        # Sort by dependency: no-deps first, then dependent
        dept_results: List[Dict] = []
        completed_names: set = set()
        completed_deliverables: Dict[str, str] = {}  # 四档-①：name→交付物，供下游交接

        # Execute in waves based on dependencies
        while matched_tasks:
            ready = []
            remaining = []
            for dept, plan in matched_tasks:
                deps = plan.get("dependencies", [])
                if isinstance(deps, str):
                    deps = [deps] if deps else []
                if all(d in completed_names for d in deps):
                    ready.append((dept, plan))
                else:
                    remaining.append((dept, plan))

            if not ready:
                # Circular dependency or all remaining have unmet deps — execute remaining
                logger.warning("Possible circular dependency, executing remaining departments")
                ready = remaining
                remaining = []

            # Execute ready departments in parallel
            project_path = Path(f"./company_projects/{self.company_id}/{_safe_name(project_description)}")
            project_path.mkdir(parents=True, exist_ok=True)

            ceo_analysis = ceo_plan.get("analysis", "")
            tasks = [
                self._run_department(
                    dept, plan,
                    self._build_dept_context(plan, ceo_analysis, completed_deliverables),
                    project_path,
                )
                for dept, plan in ready
            ]
            batch_results = await asyncio.gather(*tasks)
            dept_results.extend(batch_results)
            for r in batch_results:
                completed_names.add(r["department_name"])
                # 四档-①：记录交付物供下一波下游交接（非流式 result 形状为 TeamOrchestrator.execute 返回）
                _res = r.get("result")
                if isinstance(_res, dict):
                    _dlv = _res.get("deliverable") or _res.get("final_output") or ""
                    if _dlv:
                        completed_deliverables[r["department_name"]] = _dlv

            matched_tasks = remaining

        # Phase 3: CEO Review
        logger.info("公司编排 [%s]: Phase 3 — CEO 审查汇总", self.company_id)
        review = await self._ceo_review_with_scoring(company, ceo_plan, dept_results, project_description)
        summary = review.get("summary", "")
        project.review_scores = review.get("departments", [])
        project.round_count = "1"

        # Update project
        all_success = all(d["success"] for d in dept_results)
        project.status = "completed" if all_success else "failed"
        project.completed_at = datetime.now()
        self.db.commit()

        return {
            "company_id": self.company_id,
            "company_name": company.name,
            "project_id": project.id,
            "project_description": project_description,
            "ceo_plan": ceo_plan,
            "departments": dept_results,
            "final_summary": summary,
            "success": all_success,
        }

    async def _halt(self, project, event_type: str, extra: Dict[str, Any]):
        """M2-b 优雅自我终止：置项目 stopped + commit，再 yield 终止事件。

        必须 yield 事件后由调用方 `return`（勿 raise——raise 会进 runner 的
        except 分支把状态覆盖成 failed）。runner 视编排器正常结束为成功、不改状态，
        故这里先把 status 落成 stopped。
        """
        import time as _time
        try:
            project.status = "stopped"
            project.completed_at = datetime.now()
            self.db.commit()
        except Exception:
            self.db.rollback()
        yield {"type": event_type, **extra, "_ts": _time.time()}

    async def execute_stream(
        self,
        project_description: str,
        auto_approve_files: bool = True,
        max_rounds: int = 3,
        ceo_plan: Optional[Dict[str, Any]] = None,
        project_id: Optional[str] = None,
        limits: Optional[Dict[str, Any]] = None,
    ) -> AsyncGenerator[Dict[str, Any], None]:
        """流式执行公司级项目，SSE 事件实时推送。支持多轮迭代+打分+预算。

        Events:
            {type: "start", message}
            {type: "company_start", company_name, department_count, departments, _ts}
            {type: "ceo_plan_start", message, _ts}
            {type: "ceo_plan_done", plan, _ts}
            {type: "round_start", round_index, rework_depts, _ts}
            {type: "dept_start", department_name, goal, _ts}
            {type: "dept_waiting", department_name, waiting_for, _ts}
            {type: "dept_agent_event", department_name, sub_type, data, tokens_used, budget_remaining, _ts}
            {type: "dept_budget", department_name, used, remaining, percentage, _ts}
            {type: "dept_budget_exceeded", department_name, max_tokens, tokens_used, _ts}
            {type: "dept_done", department_name, success, result, duration_ms, tokens_used, _ts}
            {type: "ceo_score_start", message, _ts}
            {type: "ceo_score_done", scores, overall_score, dept_pass_map, _ts}
            {type: "round_done", round_index, scores, rework_needed, _ts}
            {type: "company_done", success, summary, departments, project_id, total_duration_ms, round_count, _ts}
            {type: "error", message}
        """
        import time as _time
        t0 = _time.time()
        self.auto_approve_files = auto_approve_files
        # ─── M2-b 运行限额（人在环审批 + 预算/超时熔断）；默认全关，回归安全 ───
        limits = limits or {}
        max_cost = limits.get("max_cost_yuan")          # ¥ 上限，None=不限
        max_dur = limits.get("max_duration_min")        # 分钟上限，None=不限
        approve_plan = bool(limits.get("approve_plan"))
        approve_rework = bool(limits.get("approve_rework"))
        yield {"type": "start", "message": "Loading company...", "_ts": _time.time()}

        try:
            company = self._load_company()
        except Exception as e:
            yield {"type": "error", "message": str(e), "_ts": _time.time()}
            return

        departments = self._load_departments()
        yield {
            "type": "company_start",
            "company_name": company.name,
            "department_count": len(departments),
            "departments": [d.name for d in departments],
            "_ts": _time.time(),
        }

        # Phase 1: CEO Planning（若已传入已确认的计划则跳过，直接采用）
        t_ceo_start = _time.time()
        if ceo_plan is None:
            yield {"type": "ceo_plan_start", "message": "CEO is analyzing the objective...", "_ts": _time.time()}
            try:
                ceo_plan = await self._run_ceo_planning(company, departments, project_description)
            except Exception as e:
                yield {"type": "error", "message": f"CEO planning failed: {e}", "phase": "ceo_planning", "_ts": _time.time()}
                return
        else:
            yield {"type": "ceo_plan_start", "message": "Using approved plan...", "approved": True, "_ts": _time.time()}

        yield {"type": "ceo_plan_done", "plan": ceo_plan, "_ts": _time.time(), "duration_ms": int((_time.time() - t_ceo_start) * 1000)}

        # Create project（若调用方已预建项目行则复用它，避免重复创建）
        project = None
        if project_id:
            project = self.db.query(CompanyProject).filter(CompanyProject.id == project_id).first()
        if project is None:
            project = CompanyProject(
                company_id=self.company_id,
                name=project_description[:200],
                description=project_description,
                status="in_progress",
                ceo_plan=ceo_plan,
            )
            self.db.add(project)
        else:
            project.ceo_plan = ceo_plan
            project.status = "in_progress"
        self.db.add(project)
        self.db.commit()

        # ─── M2-a 运行时控制（进程内注册表：暂停/继续/跳过）───
        from app.services import company_runner  # 惰性导入，避免与 runner 循环依赖
        control = company_runner.get_control(project_id) if project_id else company_runner.ProjectControl()

        # ─── M2-b CEO 计划审批门：开门等人裁决，批准才派部门；拒绝→优雅终止 ───
        if approve_plan:
            control.open_gate("plan")
            yield {"type": "awaiting_approval", "gate": "plan", "plan": ceo_plan, "_ts": _time.time()}
            await control.approval_event.wait()
            dec = control.approval_decision or "rejected"
            control.pending_gate = ""
            yield {"type": "approval_resolved", "gate": "plan", "decision": dec, "_ts": _time.time()}
            if dec != "approved":
                async for ev in self._halt(project, "company_terminated", {"reason": "plan_rejected"}):
                    yield ev
                return

        # ─── Budget setup ───
        company_config = company.config or {}
        budget_config = company_config.get("budget", {})
        budget_enabled = budget_config.get("enabled", False)
        dept_budgets: Dict[str, int] = budget_config.get("departments", {})
        dept_token_usage: Dict[str, int] = {}
        dept_token_total: Dict[str, int] = {}

        # ─── Build department map ───
        dept_plans = ceo_plan.get("departments", [])
        dept_map: Dict[str, Team] = {}
        for d in departments:
            dept_map[d.name] = d
            if d.department_type:
                dept_map[d.department_type] = d
        # 归一化键（小写去空白）+ 候选列表，供模糊匹配
        norm_dept_map: Dict[str, Team] = {}
        for k, v in dept_map.items():
            nk = (k or "").strip().lower()
            if nk:
                norm_dept_map.setdefault(nk, v)

        def match_plans(plans):
            """返回 (matched, unmatched)。匹配阶梯：精确 → 归一化 → 模糊。"""
            matched = []
            unmatched = []
            for dp in plans:
                name = dp.get("department_name", "")
                dtype = dp.get("department_type", "")
                target = dept_map.get(name) or dept_map.get(dtype)
                if not target:
                    for key in (name, dtype):
                        nk = (key or "").strip().lower()
                        if nk and nk in norm_dept_map:
                            target = norm_dept_map[nk]
                            break
                if not target:
                    cand = difflib.get_close_matches(
                        (name or dtype or "").strip().lower(),
                        list(norm_dept_map.keys()), n=1, cutoff=0.6,
                    )
                    if cand:
                        target = norm_dept_map[cand[0]]
                        logger.warning("Department fuzzy-matched: plan=%s → %s", name or dtype, target.name)
                if target:
                    matched.append((target, dp))
                else:
                    unmatched.append(dp)
            return matched, unmatched

        project_path = Path(f"./company_projects/{self.company_id}/{_safe_name(project_description)}")
        project_path.mkdir(parents=True, exist_ok=True)

        # ─── Multi-round iteration loop ───
        all_dept_results: List[Dict] = []
        previous_scores: List[Dict] = []
        current_plans = dept_plans
        round_index = 0
        all_pass = False
        company_cost_yuan = 0.0  # 公司总成本累加(¥)，跨轮次
        human_rework_feedback = ""  # M2-b 返工门注入的人工反馈，供下一轮返工部门 prompt

        while round_index < max_rounds and not all_pass:
            round_index += 1

            # M2-a 暂停闸（轮次顶）：暂停时 hold 住不开新轮次
            if not control.pause_event.is_set():
                yield {"type": "paused", "round_index": round_index, "_ts": _time.time()}
                await control.pause_event.wait()
                yield {"type": "resumed", "round_index": round_index, "_ts": _time.time()}

            # M2-b 熔断守卫（轮次顶）：超时/预算任一超阈 → 优雅终止
            if max_dur and (_time.time() - t0) >= max_dur * 60:
                async for ev in self._halt(project, "timeout_tripped",
                                           {"elapsed_min": round((_time.time() - t0) / 60, 2), "limit_min": max_dur}):
                    yield ev
                return
            if max_cost and company_cost_yuan >= max_cost:
                async for ev in self._halt(project, "budget_tripped",
                                           {"cost_yuan": round(company_cost_yuan, 4), "limit_yuan": max_cost}):
                    yield ev
                return

            rework_depts = []
            if round_index > 1:
                # Identify departments that need rework
                failed_names = {d["name"] for d in previous_scores if not d.get("pass", True)}
                rework_depts = list(failed_names)
                yield {
                    "type": "round_start",
                    "round_index": round_index,
                    "max_rounds": max_rounds,
                    "rework_depts": rework_depts,
                    "message": f"Round {round_index}: {len(rework_depts)} departments need rework",
                    "_ts": _time.time(),
                }
                # Filter plans to only rework departments（复制 plan 字典，避免污染已存的 ceo_plan）
                filtered_plans = [dict(p) for p in dept_plans if p.get("department_name", "") in failed_names or p.get("department_type", "") in failed_names]
                if filtered_plans:
                    # M2-b 返工注入：把 CEO 评审反馈 + improvement_areas + 人工意见拼进各失败部门的 plan
                    failed_map = {d.get("name"): d for d in previous_scores if not d.get("pass", True)}
                    for p in filtered_plans:
                        entry = failed_map.get(p.get("department_name")) or failed_map.get(p.get("department_type"))
                        parts = []
                        if entry:
                            if entry.get("feedback"):
                                parts.append("CEO 评审反馈：" + str(entry["feedback"]))
                            areas = entry.get("improvement_areas") or []
                            if areas:
                                parts.append("改进方向：" + "；".join(str(a) for a in areas))
                        if human_rework_feedback:
                            parts.append("人工意见：" + human_rework_feedback)
                        if parts:
                            p["rework_feedback"] = "\n".join(parts)
                    current_plans = filtered_plans
            else:
                yield {
                    "type": "round_start",
                    "round_index": 1,
                    "max_rounds": max_rounds,
                    "rework_depts": [],
                    "message": "Round 1: Initial execution",
                    "_ts": _time.time(),
                }

            # Phase 2: Department Execution (wave-based + budget tracking)
            matched, unmatched = match_plans(current_plans)
            for dp in unmatched:
                logger.warning("Department not found for plan: name=%s type=%s",
                               dp.get("department_name"), dp.get("department_type"))
                yield {
                    "type": "dept_unmatched",
                    "department_name": dp.get("department_name", ""),
                    "department_type": dp.get("department_type", ""),
                    "goal": dp.get("goal", ""),
                    "_ts": _time.time(),
                }
            dept_results: List[Dict] = []
            completed_names: set = set()
            completed_deliverables: Dict[str, str] = {}  # 四档-①：name→交付物，供下游交接
            dept_timings: Dict[str, float] = {}

            while matched:
                ready = []
                remaining = []
                for dept, plan in matched:
                    # M2-a 跳过闸：跳过尚未开始的部门，标记为已完成让下游依赖可推进
                    if dept.name in control.skip_depts:
                        control.skip_depts.discard(dept.name)
                        completed_names.add(dept.name)
                        yield {
                            "type": "dept_skipped",
                            "department_name": dept.name,
                            "round_index": round_index,
                            "_ts": _time.time(),
                        }
                        continue
                    deps = plan.get("dependencies", [])
                    if isinstance(deps, str):
                        deps = [deps] if deps else []
                    if all(d in completed_names for d in deps):
                        ready.append((dept, plan))
                    else:
                        unmet = [d for d in deps if d not in completed_names]
                        yield {
                            "type": "dept_waiting",
                            "department_name": dept.name,
                            "waiting_for": unmet,
                            "_ts": _time.time(),
                        }
                        remaining.append((dept, plan))

                if not ready and remaining:
                    logger.warning("Possible circular dependency, executing remaining departments")
                    yield {
                        "type": "dept_waiting",
                        "department_name": "all_remaining",
                        "waiting_for": ["circular_dependency_resolved"],
                        "message": "Circular dependency detected, executing all remaining departments",
                        "_ts": _time.time(),
                    }
                    ready = remaining
                    remaining = []

                # Budget checks before execution
                for dept, plan in ready:
                    if budget_enabled:
                        dept_budget_limit = dept_budgets.get(dept.name, dept_budgets.get("default", 0))
                        dept_token_total[dept.name] = dept_budget_limit
                        used = dept_token_usage.get(dept.name, 0)
                        if dept_budget_limit > 0 and used >= dept_budget_limit:
                            yield {
                                "type": "dept_budget_exceeded",
                                "department_name": dept.name,
                                "max_tokens": dept_budget_limit,
                                "tokens_used": used,
                                "_ts": _time.time(),
                            }

                # Execute ready departments in parallel with stream forwarding
                ceo_analysis = ceo_plan.get("analysis", "")
                dept_event_queue: asyncio.Queue = asyncio.Queue()
                # 计时起点：部门被派发时（而非首个事件到达时），避免 duration≈0
                for dept, plan in ready:
                    dept_timings.setdefault(dept.name, __import__("time").time())
                tasks = {
                    asyncio.create_task(
                        self._run_department_stream(
                            dept, plan,
                            self._build_dept_context(plan, ceo_analysis, completed_deliverables),
                            project_path, dept_event_queue,
                        )
                    ): (dept, plan)
                    for dept, plan in ready
                }
                # M2-c 登记部门 task 句柄，供单部门硬取消（cancel_department）
                for task, (dept, _plan) in tasks.items():
                    control.dept_tasks[dept.name] = task

                done_count = 0
                while done_count < len(tasks):
                    evt = await dept_event_queue.get()
                    # Inject budget info into dept_agent_event
                    if evt.get("type") == "dept_agent_event" and budget_enabled:
                        dept_name = evt.get("department_name", "")
                        current_used = dept_token_usage.get(dept_name, 0) + 1
                        dept_token_usage[dept_name] = current_used
                        limit = dept_token_total.get(dept_name, 0)
                        evt["tokens_used"] = current_used
                        evt["budget_remaining"] = max(0, limit - current_used) if limit > 0 else -1
                        if limit > 0 and current_used % 5 == 0:
                            evt["budget_percentage"] = round(current_used / limit * 100, 1)

                    yield evt

                    if evt.get("type") == "dept_done":
                        done_count += 1
                        r = evt
                        completed_names.add(r["department_name"])
                        # 四档-①：记录该部门交付物，供同轮下游部门交接
                        _dlv = (r.get("result") or {}).get("deliverable", "") if isinstance(r.get("result"), dict) else ""
                        if _dlv:
                            completed_deliverables[r["department_name"]] = _dlv
                        control.dept_tasks.pop(r["department_name"], None)  # M2-c 注销句柄
                        dur = int((__import__("time").time() - dept_timings.get(r["department_name"], __import__("time").time())) * 1000)
                        approx_used = dept_token_usage.get(r["department_name"], 0)
                        real_used = int(r.get("real_tokens") or 0)
                        used_display = real_used if real_used > 0 else approx_used
                        dept_cost = float(r.get("cost_yuan") or 0.0)
                        company_cost_yuan += dept_cost
                        dept_results.append({
                            "department_name": r["department_name"],
                            "department_type": "",
                            "success": r["success"],
                            "plan": {},
                            "result": r.get("result"),
                            "error": r.get("error"),
                            "duration_ms": dur,
                            "tokens_used": used_display,
                            "tokens_used_approx": approx_used,
                            "cost_yuan": round(dept_cost, 4),
                        })
                        # Emit budget summary for this department
                        dept_name = r["department_name"]
                        used = approx_used
                        limit = dept_token_total.get(dept_name, 0)
                        if limit > 0:
                            yield {
                                "type": "dept_budget",
                                "department_name": dept_name,
                                "used": used,
                                "remaining": max(0, limit - used),
                                "percentage": round(used / limit * 100, 1) if limit > 0 else 0,
                                "_ts": __import__("time").time(),
                            }
                        yield {
                            "type": "dept_done",
                            "department_name": r["department_name"],
                            "success": r["success"],
                            "result": r.get("result"),
                            "duration_ms": dur,
                            "tokens_used": used_display,
                            "tokens_used_approx": approx_used,
                            "cost_yuan": round(dept_cost, 4),
                            "_ts": __import__("time").time(),
                        }
                        # M2-b 预算熔断（部门结束粒度，最灵敏）：累计成本超阈 → 优雅终止
                        if max_cost and company_cost_yuan >= max_cost:
                            async for ev in self._halt(project, "budget_tripped",
                                                       {"cost_yuan": round(company_cost_yuan, 4), "limit_yuan": max_cost}):
                                yield ev
                            return
                    elif evt.get("type") == "dept_agent_event":
                        dept_name = evt.get("department_name", "")
                        if dept_name and dept_name not in dept_timings:
                            dept_timings[dept_name] = __import__("time").time()

                matched = remaining

                # M2-a 暂停闸（波次边界）：在飞部门已跑完，暂停时 hold 住不开下一波
                if matched and not control.pause_event.is_set():
                    yield {"type": "paused", "round_index": round_index, "_ts": _time.time()}
                    await control.pause_event.wait()
                    yield {"type": "resumed", "round_index": round_index, "_ts": _time.time()}

                # M2-b 超时熔断（波次边界，粒度更细）：总耗时超阈 → 优雅终止
                if max_dur and (_time.time() - t0) >= max_dur * 60:
                    async for ev in self._halt(project, "timeout_tripped",
                                               {"elapsed_min": round((_time.time() - t0) / 60, 2), "limit_min": max_dur}):
                        yield ev
                    return

            # Phase 3: CEO Review with Scoring
            yield {
                "type": "ceo_score_start",
                "message": f"CEO is scoring round {round_index} outputs...",
                "round_index": round_index,
                "_ts": _time.time(),
            }
            t_review_start = _time.time()
            try:
                review = await self._ceo_review_with_scoring(company, ceo_plan, dept_results, project_description)
            except Exception as e:
                review = {"summary": f"Review failed: {e}", "overall_score": 0, "departments": [], "next_steps": ""}

            scores = review.get("departments", [])
            review_dur = int((_time.time() - t_review_start) * 1000)
            dept_pass_map = {d["name"]: d.get("pass", True) for d in scores}

            yield {
                "type": "ceo_score_done",
                "scores": scores,
                "overall_score": review.get("overall_score", 0),
                "summary": review.get("summary", ""),
                "next_steps": review.get("next_steps", ""),
                "dept_pass_map": dept_pass_map,
                "duration_ms": review_dur,
                "round_index": round_index,
                "_ts": _time.time(),
            }

            previous_scores = scores
            all_dept_results = dept_results
            all_pass = all(d.get("pass", True) for d in scores)

            yield {
                "type": "round_done",
                "round_index": round_index,
                "scores": scores,
                "overall_score": review.get("overall_score", 0),
                "rework_needed": not all_pass and round_index < max_rounds,
                "passed_depts": [d["name"] for d in scores if d.get("pass", True)],
                "failed_depts": [d["name"] for d in scores if not d.get("pass", True)],
                "_ts": _time.time(),
            }

            # 评审解析失败：不基于不可信打分盲目返工（会空转+重复失败），直接保守收尾
            if review.get("parse_failed"):
                yield {"type": "review_parse_failed", "round_index": round_index, "_ts": _time.time()}
                break

            # ─── M2-b 返工审批门：有失败部门且还能再迭代时，问是否返工 ───
            if approve_rework and not all_pass and round_index < max_rounds:
                control.open_gate("rework")
                yield {
                    "type": "awaiting_approval", "gate": "rework",
                    "failed_depts": [d["name"] for d in scores if not d.get("pass", True)],
                    "round_index": round_index, "_ts": _time.time(),
                }
                await control.approval_event.wait()
                dec = control.approval_decision or "rejected"
                human_rework_feedback = getattr(control, "approval_feedback", "") or ""
                control.pending_gate = ""
                yield {"type": "approval_resolved", "gate": "rework", "decision": dec,
                       "feedback": human_rework_feedback, "_ts": _time.time()}
                if dec != "approved":
                    # 拒绝返工 = 就地收尾（按当前产出定 completed/failed），非失败
                    yield {"type": "rework_skipped", "round_index": round_index, "_ts": _time.time()}
                    break

        # ─── Finalize ───
        all_success = all(d["success"] for d in all_dept_results)
        project.review_scores = previous_scores
        project.round_count = str(round_index)
        project.status = "completed" if all_success else "failed"
        project.completed_at = datetime.now()
        self.db.commit()

        # 四档-②：终态自动沉淀知识，供后续项目的 CEO 规划回喂（闭合学习闭环）
        self._auto_extract_knowledge(project)

        total_duration = int((_time.time() - t0) * 1000)
        yield {
            "type": "company_done",
            "success": all_success,
            "summary": previous_scores[0].get("summary", "") if previous_scores else "",
            "departments": all_dept_results,
            "project_id": project.id,
            "total_duration_ms": total_duration,
            "round_count": round_index,
            "overall_score": previous_scores[0].get("overall_score", 0) if previous_scores else 0,
            "review_scores": previous_scores,
            "total_cost_yuan": round(company_cost_yuan, 4),
            "_ts": _time.time(),
        }

    async def _ceo_review_with_scoring(
        self,
        company: Company,
        ceo_plan: Dict,
        dept_results: List[Dict],
        project_desc: str,
    ) -> Dict[str, Any]:
        """Phase 3: CEO 审查所有部门产出 + 打分反馈。

        Returns:
            {summary, overall_score, departments: [{name, score, feedback, pass}], next_steps}
        """
        ceo_agent = None
        if company.ceo_agent_id:
            ceo_agent = self.db.query(Agent).filter(Agent.id == company.ceo_agent_id).first()
        if not ceo_agent:
            ceo_agent = self.db.query(Agent).filter(
                Agent.user_id == self.user_id,
                Agent.name.like("%CEO%")
            ).first()
        if not ceo_agent:
            return {"summary": "CEO Agent not available for review", "overall_score": 0, "departments": [], "next_steps": ""}

        # Build review prompt with scoring requirement
        dept_summaries = []
        for dr in dept_results:
            result = dr.get("result", {})
            deliverable = result.get("deliverable", "")[:800] if result else ""
            error = dr.get("error", "")
            # 把真实产出文件清单喂给评审：仅凭 deliverable 文本会误判「没做出东西」
            files = (result.get("files") or []) if result else []
            file_line = ""
            if files:
                names = [Path(fp).name for fp in files[:25]]
                more = f" 等{len(files)}个" if len(files) > 25 else ""
                file_line = f"Produced files ({len(files)}): {', '.join(names)}{more}\n"
            dept_summaries.append(
                f"### {dr['department_name']}\n"
                f"Success: {dr['success']}\n"
                f"{file_line}"
                f"Output: {deliverable or error or 'No output'}"
            )

        review_prompt = f"""You are the CEO reviewing your company's execution of the project:
"{project_desc}"

Your original plan:
{json.dumps(ceo_plan, ensure_ascii=False, indent=2)[:1500]}

Department Results:
{chr(10).join(dept_summaries)[:3000]}

You MUST output a strict JSON object (no markdown, no extra text):
{{
  "overall_score": <1-10>,
  "summary": "<2-3 sentence overall assessment>",
  "next_steps": "<recommended next actions>",
  "departments": [
    {{
      "name": "<department_name>",
      "score": <1-10>,
      "feedback": "<specific feedback, what was good and what needs improvement>",
      "pass": <true if score >= 6, false otherwise>,
      "improvement_areas": ["<area1>", "<area2>"]
    }}
  ]
}}

Scoring criteria:
- 9-10: Exceptional output, exceeds expectations
- 7-8: Good quality, meets all requirements
- 6: Acceptable, minor issues
- 4-5: Below standard, needs significant revision
- 1-3: Poor quality, fundamental issues

A department passes if score >= 6. Be honest but fair."""

        wf = ceo_agent.workflow_config or {}
        nodes = wf.get("nodes", [])
        llm_node = None
        for n in nodes:
            if n.get("type") == "llm":
                llm_node = n
                break
        llm_data = llm_node.get("data", {}) if llm_node else {}

        # 直接单次 LLM 调用（非 ReAct）：评审是「产出严格 JSON」的一次性任务。
        # 走 AgentRuntime 的 ReAct 循环会把 max_iterations 耗在 Thought/Action 上、
        # 从不产出最终 JSON（返回「已达最大迭代次数(3)，任务被截断」或中途「Let me check...」），
        # 解析失败 → 旧逻辑静默 auto-pass(score=5) 污染打分/返工判断。
        from app.agent_runtime.core import _LLMClient

        review_llm = AgentLLMConfig(
            provider=llm_data.get("provider", "deepseek"),
            model=llm_data.get("model", "deepseek-v4-pro"),
            temperature=0.2,
            max_tokens=1500,
            request_timeout=90.0,
        )
        client = _LLMClient(review_llm)
        messages = [
            {"role": "system", "content": "You are the CEO. Review and score your company's work. Output ONLY a single valid JSON object — no markdown fences, no explanation, no tool use."},
            {"role": "user", "content": review_prompt},
        ]

        raw = ""
        review: Dict[str, Any] = {}
        for attempt in range(2):
            try:
                resp = await client.chat(messages=messages, tools=None, iteration=0)
            except Exception as e:
                logger.warning("CEO review LLM call failed (attempt %d): %s", attempt + 1, e)
                break
            raw = getattr(resp, "content", "") or (resp.get("content", "") if isinstance(resp, dict) else "")
            review = _parse_ceo_json(raw)
            if review:
                break
            # 一次纠正性重试：明确告知上一轮不是合法 JSON
            messages.append({"role": "assistant", "content": (raw or "")[:2000]})
            messages.append({"role": "user", "content": "Your previous response was not a valid JSON object. Output ONLY the JSON object described above, nothing else."})

        if not review:
            # 兜底（正常几乎不触发）：解析仍失败 → 保守判「未通过」并显式标注，绝不静默 auto-pass；
            # 上层 round 循环遇 parse_failed 会直接收尾，避免基于不可信打分盲目返工空转。
            logger.warning("CEO review JSON parsing failed after retries, raw (first 500 chars): %.500s", raw)
            score_match = re.search(r'"overall_score"\s*:\s*(\d+)', raw)
            extracted_score = int(score_match.group(1)) if score_match else 0
            fallback_depts = []
            for dr_item in dept_results:
                fallback_depts.append({
                    "name": dr_item.get("department_name", ""),
                    "score": extracted_score,
                    "feedback": "CEO 评审输出无法解析为 JSON，已保守判为未通过，请人工复核。",
                    "pass": False,
                    "improvement_areas": [],
                })
            review = {
                "overall_score": extracted_score,
                "summary": (raw or "Review failed")[:500],
                "departments": fallback_depts,
                "next_steps": "",
                "parse_failed": True,
            }
        return review


def _safe_name(text: str) -> str:
    """Generate a safe directory name from project description."""
    import re
    safe = re.sub(r'[^\w\s-]', '', text[:40])
    safe = re.sub(r'\s+', '_', safe)
    return safe or "untitled"