from __future__ import annotations from .shared import * from .signal_core import _as_dict async def _project_task_metrics( db: AsyncSession, *, tenant_id: int, project_id: int, ) -> dict[str, int]: result = await db.execute( select(Task.status).where( and_( Task.project_id == project_id, Task.tenant_id == tenant_id, ) ) ) statuses = [str(row[0] or "").lower() for row in result.all()] total = len(statuses) done = len([s for s in statuses if s in DONE_STATUSES]) return {"total": total, "done": done} async def _build_governance_snapshot( db: AsyncSession, *, tenant_id: int, project_id: int, charter_id: int | None = None, ) -> dict[str, Any]: charter: TruthNode | None = None if charter_id: charter = ( await db.execute( select(TruthNode).where( and_( TruthNode.id == charter_id, TruthNode.project_id == project_id, TruthNode.tenant_id == tenant_id, TruthNode.node_type == "charter", ) ) ) ).scalar_one_or_none() if charter is None: charter = ( await db.execute( select(TruthNode) .where( and_( TruthNode.project_id == project_id, TruthNode.tenant_id == tenant_id, TruthNode.node_type == "charter", ) ) .order_by(TruthNode.created_at.desc()) .limit(1) ) ).scalar_one_or_none() confidence_score = None confidence_risk = False charter_id_value = int(charter.id) if charter else None if charter: charter_content = _as_dict(charter.content) metadata = charter_content.get("metadata") if isinstance(metadata, dict): raw_score = metadata.get("confidence_score") try: confidence_score = float(raw_score) except (TypeError, ValueError): confidence_score = None confidence_risk = confidence_score is not None and confidence_score < 0.8 binding = None if charter_id_value: binding = ( await db.execute( select(ModelRequirementBinding, ModelVersion) .join( ModelVersion, ModelVersion.id == ModelRequirementBinding.model_version_id, isouter=True, ) .where( and_( ModelRequirementBinding.tenant_id == tenant_id, ModelRequirementBinding.truth_node_id == charter_id_value, ) ) .order_by(ModelRequirementBinding.created_at.desc()) .limit(1) ) ).first() model_block = False model_info: dict[str, Any] = { "bound": False, "compatibility_score": None, "model_version_id": None, "model_name": None, "provider": None, } est_cost = 0.01 if binding: binding_row, model_row = binding score = int(binding_row.compatibility_score or 0) model_block = score < 70 model_info = { "bound": True, "compatibility_score": score, "model_version_id": int(binding_row.model_version_id), "model_name": getattr(model_row, "name", None), "provider": getattr(model_row, "provider", None), } try: est_cost = max(float(binding_row.estimated_cost or 0.01), 0.01) except (TypeError, ValueError): est_cost = 0.01 budget_service = BudgetEnforcementService(db) budget_check = await budget_service.check_budget_before_ai_call( tenant_id=tenant_id, estimated_cost_usd=est_cost, project_id=project_id, ) budget_block = not bool(budget_check.get("allowed", True)) intervention_required = confidence_risk or model_block or budget_block blockers: list[str] = [] warnings: list[str] = [] if confidence_risk: warnings.append("宪章 AI 置信度低于 0.80，建议补充关键事实后再推进。") if model_block: blockers.append("模型匹配分低于 70，禁止自动推进发布验证。") elif not model_info["bound"]: warnings.append("当前宪章未绑定模型版本，建议先完成模型能力矩阵校验。") if budget_block: blockers.append(f"预算管控阻断：{budget_check.get('reason') or 'budget_blocked'}") elif str(budget_check.get("reason") or "") == "within_budget": usage = budget_check.get("usage_percent") if isinstance(usage, (int, float)) and float(usage) >= 80: warnings.append("预算已接近阈值，建议执行成本优化方案。") return { "project_id": project_id, "charter_id": charter_id_value, "intervention_required": intervention_required, "blockers": blockers, "warnings": warnings, "confidence": { "score": confidence_score, "risk": confidence_risk, }, "model_binding": model_info, "budget": budget_check, } async def _assert_release_transition_allowed( db: AsyncSession, *, tenant_id: int, project_id: int, release_node: TruthNode, target_status: str, ) -> None: current = str(release_node.status or "pending").lower() target = str(target_status or "").lower() if target == "verified": raise HTTPException(status_code=409, detail="请通过验证完成接口标记为已验证") if target not in RELEASE_STATUS_TRANSITIONS.get(current, set()): raise HTTPException( status_code=409, detail=f"非法状态流转：{current} -> {target}", ) content = _as_dict(release_node.content) if target in {"releasing", "online"}: task_ids = [int(tid) for tid in (content.get("task_ids") or []) if str(tid).isdigit()] if not task_ids: raise HTTPException(status_code=409, detail="发布计划未绑定任务") task_result = await db.execute( select(Task.id, Task.status).where( and_( Task.tenant_id == tenant_id, Task.project_id == project_id, Task.id.in_(task_ids), ) ) ) task_rows = list(task_result.all()) if len(task_rows) != len(task_ids): raise HTTPException(status_code=409, detail="发布计划存在无效任务绑定") not_done = [row for row in task_rows if str(row[1] or "").lower() not in DONE_STATUSES] if not_done: raise HTTPException(status_code=409, detail="存在未完成任务，禁止进入发布流程") governance = await _build_governance_snapshot( db, tenant_id=tenant_id, project_id=project_id, charter_id=int(content.get("charter_id") or 0) or None, ) if governance.get("blockers"): raise HTTPException( status_code=409, detail="；".join([str(item) for item in governance.get("blockers", [])]), ) verification_items = content.get("verification_items") if not isinstance(verification_items, list) or len(verification_items) == 0: raise HTTPException(status_code=409, detail="请先生成发布验证清单") def _assert_charter_aligned(charter: TruthNode) -> None: if (charter.status or "").lower() not in ALIGNED_STATUSES: raise HTTPException( status_code=409, detail="宪章未对齐，任务环节未解锁", ) def _is_manager(user: User) -> bool: return has_any_role(str(user.role or ""), ["OWNER", "REVIEWER"]) def _require_manager(user: User) -> None: if not _is_manager(user): raise HTTPException(status_code=403, detail="仅项目负责人或管理员可执行该操作")