"""Strategic brain for AI PMO decision runtime.""" from __future__ import annotations from .shared import ( CanonicalIntent, INTENT_PROJECT_CREATE, INTENT_SIGNAL_SOURCE_ONBOARD, Any, normalize_workload_ratio, safe_int, ) class StrategicBrain: """组合级判断。""" @staticmethod def evaluate(*, intent: CanonicalIntent, world_model: dict[str, Any], message_text: str) -> dict[str, Any]: project = world_model.get("project") if isinstance(world_model.get("project"), dict) else None budget = world_model.get("budget") if isinstance(world_model.get("budget"), dict) else {} capacity = world_model.get("capacity") if isinstance(world_model.get("capacity"), list) else [] risk_count = int((world_model.get("risk_summary") or {}).get("open_risk_count") or 0) active_projects = int((world_model.get("portfolio") or {}).get("active_project_count") or 0) budget_ratio = float(budget.get("usage_ratio") or 0.0) overloaded_people = [row for row in capacity if normalize_workload_ratio(row.get("workload_ratio")) >= 0.85] risk_score = min(1.0, 0.18 + risk_count * 0.08 + budget_ratio * 0.25 + len(overloaded_people) * 0.1) if intent.name == INTENT_PROJECT_CREATE: risk_score = max(risk_score, 0.62) if intent.name == INTENT_SIGNAL_SOURCE_ONBOARD: risk_score = max(risk_score, 0.58) return { "decision_level": "project" if project else "portfolio", "decision_object_type": "project" if project else "portfolio", "decision_object_id": safe_int((project or {}).get("id")), "risk_score": round(risk_score, 3), "budget_impact": { "budget_id": budget.get("budget_id"), "usage_ratio": budget_ratio, "hard_limit_enabled": bool(budget.get("hard_limit_enabled")), }, "resource_impact": { "overloaded_user_ids": [row.get("user_id") for row in overloaded_people], "active_project_count": active_projects, }, "rationale": ( f"intent={intent.name}; " f"project_scope={'project' if project else 'portfolio'}; " f"budget_ratio={budget_ratio:.2f}; " f"open_risks={risk_count}; " f"overloaded={len(overloaded_people)}" ), "message_excerpt": message_text[:240], }