agent_deply/pam_deploy_graph/models.py

"""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"]
AgentExecutionMode = Literal["fixed_runtime", "agentic_skill"]
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"]
ModeDecisionRisk = Literal["low", "medium", "high"]
ToolScope = Literal["global", "ip"]


@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 ActionToolSpec:
    """面向 LLM 和 runtime 的统一 action tool 描述。"""

    name: str
    action: str
    scope: ToolScope
    description: str
    risk_level: ModeDecisionRisk = "medium"
    requires_confirmation: bool = False
    required_runtime_fields: tuple[str, ...] = ()
    required_param_fields: tuple[str, ...] = ()
    preferred_backend: str = ""


@dataclass(slots=True)
class SkillPolicy:
    """从 Skill 文档提取出的部署策略约束。"""

    name: str
    source_path: str
    description: str = ""
    allowed_execution_modes: tuple[AgentExecutionMode, ...] = ("fixed_runtime", "agentic_skill")
    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 LlmModeDecision:
    """LLM 给出的执行模式决策。"""

    mode: AgentExecutionMode = "fixed_runtime"
    reason: str = ""
    risk_level: ModeDecisionRisk = "medium"
    requires_confirmation: bool = True


@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
    should_continue: bool = True
    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)
    execution_mode: AgentExecutionMode = "fixed_runtime"
    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 = ""
    planned_actions: list[str] = field(default_factory=list)
    mode_reason: str = ""
    mode_risk_level: ModeDecisionRisk = "medium"
    mode_requires_confirmation: bool = True
    paused: bool = False
    pause_reason: str = ""
    review_context: dict[str, Any] = field(default_factory=dict)
    events: list[dict[str, Any]] = field(default_factory=list)