"""PAM 部署 Agent 共享数据模型。""" from __future__ import annotations from dataclasses import dataclass, field from typing import Any, Literal BackendName = Literal["mcp", "script", "fake"] ExecutionStrategy = Literal["hybrid_node_mcp", "script_only", "fake"] IntentName = Literal["deploy", "show_usage", "preview", "query_node_ips", "rollback"] ModePreference = Literal["MCP", "API脚本", "未指定"] StrategyPreference = Literal["hybrid_node_mcp", "script_only", "fake", "未指定"] ActionAnalysisSeverity = Literal["info", "low", "medium", "high"] @dataclass(slots=True) class ActionResult: """单个 action 的统一执行结果。""" action: str backend: BackendName ok: bool values: dict[str, Any] = field(default_factory=dict) exit_code: int = 0 tool_name: str = "" stdout: str = "" stderr: str = "" raw_output: str = "" error_summary: str = "" @dataclass(slots=True) class SkillPolicy: """从 Skill 文档提取出的部署策略约束。""" name: str source_path: str description: str = "" allowed_modes: tuple[str, ...] = ("MCP", "API脚本") allowed_actions: tuple[str, ...] = () required_confirmations: tuple[str, ...] = ( "params", "target_scope", "rollback", ) required_params: tuple[str, ...] = () optional_params: dict[str, Any] = field(default_factory=dict) action_sequence: tuple[str, ...] = () ip_action_sequence: tuple[str, ...] = () forbidden_actions: tuple[str, ...] = ( "script-main-flow", "auto-rollback", "modify-deploy-scripts", ) @dataclass(slots=True) class LlmIntentResult: """LLM 意图识别结果。""" intent: IntentName mode_preference: ModePreference = "未指定" strategy_preference: StrategyPreference = "未指定" confidence: float = 0.0 reasons: list[str] = field(default_factory=list) needs_clarification: bool = False clarification_questions: list[str] = field(default_factory=list) @dataclass(slots=True) class LlmParamResult: """LLM 参数抽取结果。""" extracted_params: dict[str, Any] = field(default_factory=dict) extracted_control: dict[str, Any] = field(default_factory=dict) missing_required_params: list[str] = field(default_factory=list) ambiguous_fields: list[str] = field(default_factory=list) sensitive_fields_present: list[str] = field(default_factory=list) @dataclass(slots=True) class LlmDeployPlan: """LLM 生成的部署计划。""" summary: str risk_notes: list[str] = field(default_factory=list) planned_actions: list[str] = field(default_factory=list) requires_confirmation: bool = True execution_strategy: StrategyPreference = "未指定" @dataclass(slots=True) class LlmActionAnalysis: """LLM 或规则对单次 action 结果的诊断建议。""" action: str has_anomaly: bool = False severity: ActionAnalysisSeverity = "info" possible_reason: str = "" suggested_action: str = "" requires_confirmation: bool = False notes: list[str] = field(default_factory=list) @dataclass(slots=True) class AgentState: """一次部署运行的完整状态,可序列化到 checkpoint。""" run_id: str params: dict[str, Any] execution_strategy: ExecutionStrategy action_backends: dict[str, BackendName] script_entry: str = "" script_base_dir: str = "." config_path: str = "" trace_file_path: str = "" node_mcp_server_name: str = "" node_mcp_tool_names: dict[str, str] = field(default_factory=dict) completed_global_steps: list[str] = field(default_factory=list) hash_code: str = "" node_url: str = "" online_ips: list[str] = field(default_factory=list) target_ips: list[str] = field(default_factory=list) ip_states: dict[str, dict[str, Any]] = field(default_factory=dict) pending_confirmation: str = "" last_success_step: str = "" last_failed_step: str = "" checkpoint_path: str = "" events: list[dict[str, Any]] = field(default_factory=list)