"""Frozen workflow graph schema v1.0 (internal DTO).

Coze canvas DTO is converted into this structure at the Gateway boundary.
The runtime must never depend on Coze-private fields.

Wire JSON uses camelCase aliases; Python code uses snake_case.
"""

from __future__ import annotations

from hashlib import sha256
from typing import Any, Literal

from pydantic import BaseModel, Field, field_validator, model_validator

SCHEMA_VERSION = "1.0"

NodeType = Literal[
    "start",
    "output",
    "agent",
    "skill",
    "http",
    "sql_read",
    "code",
    "transform",
    "evidence_normalizer",
    "deep_research_write",
    "condition",
    "merge",
    "human_input",
    "subworkflow",
    "loop",
]

ALL_NODE_TYPES: tuple[NodeType, ...] = (
    "start",
    "output",
    "agent",
    "skill",
    "http",
    "sql_read",
    "code",
    "transform",
    "evidence_normalizer",
    "deep_research_write",
    "condition",
    "merge",
    "human_input",
    "subworkflow",
    "loop",
)

RunStatus = Literal[
    "queued",
    "running",
    "awaiting_input",
    "completed",
    "failed",
    "cancel_requested",
    "cancelled",
]

ALL_RUN_STATUSES: tuple[RunStatus, ...] = (
    "queued",
    "running",
    "awaiting_input",
    "completed",
    "failed",
    "cancel_requested",
    "cancelled",
)

TERMINAL_RUN_STATUSES: frozenset[RunStatus] = frozenset({"completed", "failed", "cancelled"})

NodeRunStatus = Literal[
    "queued",
    "running",
    "completed",
    "failed",
    "skipped",
    "cancelled",
]

SkillMode = Literal["agent_skill", "callable_skill"]
MergeStrategy = Literal["all", "first_success", "any_success"]
HttpMethod = Literal["GET", "POST", "PUT", "PATCH", "DELETE"]
AgentResponseMode = Literal["text", "json"]
AgentThreadMode = Literal["isolated_per_node", "shared_run_thread"]
ExecutionMode = Literal["normal"]


class WorkflowGraphSettings(BaseModel):
    """Per-graph resource limits (still capped by system limits at publish)."""

    run_timeout_seconds: int = Field(default=1800, ge=1, alias="runTimeoutSeconds")
    node_timeout_seconds: int = Field(default=300, ge=1, alias="nodeTimeoutSeconds")
    max_steps: int = Field(default=200, ge=1, alias="maxSteps")
    max_loop_iterations: int = Field(default=5, ge=1, alias="maxLoopIterations")
    max_parallelism: int = Field(default=4, ge=1, alias="maxParallelism")

    model_config = {"extra": "forbid", "populate_by_name": True}


class WorkflowNode(BaseModel):
    """A single node on the internal standard graph."""

    id: str = Field(min_length=1)
    type: NodeType
    name: str = ""
    config: dict[str, Any] = Field(default_factory=dict)

    model_config = {"extra": "forbid", "populate_by_name": True}

    @field_validator("id")
    @classmethod
    def _id_not_blank(cls, value: str) -> str:
        stripped = value.strip()
        if not stripped:
            raise ValueError("node id must be non-empty")
        return stripped


class WorkflowEdge(BaseModel):
    """Directed edge. ``source_port`` names a condition / loop outlet when set."""

    id: str = Field(min_length=1)
    source: str
    target: str
    source_port: str | None = Field(default=None, alias="sourcePort")

    model_config = {"extra": "forbid", "populate_by_name": True}


class WorkflowGraph(BaseModel):
    """Internal standard workflow graph (publish / run snapshot)."""

    schema_version: Literal["1.0"] = Field(default=SCHEMA_VERSION, alias="schemaVersion")
    id: str = Field(min_length=1)
    name: str = ""
    description: str = ""
    input_schema: dict[str, Any] = Field(default_factory=dict, alias="inputSchema")
    output_schema: dict[str, Any] = Field(default_factory=dict, alias="outputSchema")
    nodes: list[WorkflowNode] = Field(default_factory=list)
    edges: list[WorkflowEdge] = Field(default_factory=list)
    settings: WorkflowGraphSettings = Field(default_factory=WorkflowGraphSettings)

    model_config = {"extra": "forbid", "populate_by_name": True}

    @model_validator(mode="after")
    def _unique_ids(self) -> WorkflowGraph:
        node_ids = [n.id for n in self.nodes]
        if len(node_ids) != len(set(node_ids)):
            raise ValueError("duplicate node id")
        edge_ids = [e.id for e in self.edges]
        if len(edge_ids) != len(set(edge_ids)):
            raise ValueError("duplicate edge id")
        known = set(node_ids)
        for edge in self.edges:
            if edge.source not in known:
                raise ValueError(f"edge {edge.id} source not found: {edge.source}")
            if edge.target not in known:
                raise ValueError(f"edge {edge.id} target not found: {edge.target}")
        return self

    def node_map(self) -> dict[str, WorkflowNode]:
        return {node.id: node for node in self.nodes}

    def canonical_json(self) -> str:
        """Deterministic JSON used for ``graph_hash`` (sorted keys, no whitespace)."""
        return self.model_dump_json(by_alias=True, exclude_none=True)

    def graph_hash(self) -> str:
        return sha256(self.canonical_json().encode("utf-8")).hexdigest()


class ArtifactRef(BaseModel):
    """Reference to a large/binary result; full bytes live outside events."""

    artifact_id: str = Field(alias="artifactId")
    name: str
    mime_type: str = Field(default="application/octet-stream", alias="mimeType")
    path: str = ""
    size_bytes: int | None = Field(default=None, alias="sizeBytes")
    preview: str | None = None

    model_config = {"extra": "forbid", "populate_by_name": True}


class NodeResult(BaseModel):
    """Uniform result returned by every node executor."""

    data: dict[str, Any] = Field(default_factory=dict)
    messages: list[dict[str, Any]] = Field(default_factory=list)
    artifacts: list[ArtifactRef] = Field(default_factory=list)
    metadata: dict[str, Any] = Field(default_factory=dict)
    warnings: list[str] = Field(default_factory=list)

    model_config = {"extra": "forbid", "populate_by_name": True}


class StartRunRequest(BaseModel):
    """Body for ``POST /api/workflows/{workflow_id}/runs``."""

    version_id: str = Field(alias="versionId")
    inputs: dict[str, Any] = Field(default_factory=dict)
    idempotency_key: str | None = Field(default=None, alias="idempotencyKey")
    execution_mode: ExecutionMode = Field(default="normal", alias="executionMode")

    model_config = {"extra": "forbid", "populate_by_name": True}


class ResumeRunRequest(BaseModel):
    """Body for ``POST /api/workflows/runs/{run_id}/resume``."""

    resume_token: str = Field(alias="resumeToken")
    action: str
    payload: dict[str, Any] = Field(default_factory=dict)

    model_config = {"extra": "forbid", "populate_by_name": True}


# ── Typed node config helpers (optional; validators may use later) ───────────


class AgentNodeConfig(BaseModel):
    agent_id: str = Field(alias="agentId")
    prompt_template: str = Field(default="", alias="promptTemplate")
    input_bindings: dict[str, Any] = Field(default_factory=dict, alias="inputBindings")
    response_mode: AgentResponseMode = Field(default="text", alias="responseMode")
    response_schema: dict[str, Any] = Field(default_factory=dict, alias="responseSchema")
    thread_mode: AgentThreadMode = Field(default="isolated_per_node", alias="threadMode")
    timeout_seconds: int = Field(default=600, ge=1, alias="timeoutSeconds")
    skill_names: list[str] = Field(default_factory=list, alias="skillNames")

    model_config = {"extra": "forbid", "populate_by_name": True}


class SkillNodeConfig(BaseModel):
    mode: SkillMode = "agent_skill"
    agent_id: str | None = Field(default=None, alias="agentId")
    skill_names: list[str] = Field(default_factory=list, alias="skillNames")
    prompt_template: str = Field(default="", alias="promptTemplate")
    input_bindings: dict[str, Any] = Field(default_factory=dict, alias="inputBindings")
    # callable_skill fields (phase 4+; accepted in schema, not executed in phase 0)
    entry: str | None = None
    input_schema: dict[str, Any] = Field(default_factory=dict, alias="inputSchema")
    output_schema: dict[str, Any] = Field(default_factory=dict, alias="outputSchema")
    timeout_seconds: int = Field(default=300, ge=1, alias="timeoutSeconds")

    model_config = {"extra": "forbid", "populate_by_name": True}


class HttpNodeConfig(BaseModel):
    method: HttpMethod = "GET"
    url: str
    headers: dict[str, str] = Field(default_factory=dict)
    query: dict[str, Any] = Field(default_factory=dict)
    body: Any = None
    credential_ref: str | None = Field(default=None, alias="credentialRef")
    timeout_seconds: int = Field(default=60, ge=1, alias="timeoutSeconds")
    max_response_bytes: int = Field(default=1_048_576, ge=1, alias="maxResponseBytes")
    allow_redirects: bool = Field(default=False, alias="allowRedirects")

    model_config = {"extra": "forbid", "populate_by_name": True}


class SqlReadNodeConfig(BaseModel):
    data_source_id: str = Field(alias="dataSourceId")
    statement: str
    parameters: dict[str, Any] = Field(default_factory=dict)
    max_rows: int = Field(default=1000, ge=1, alias="maxRows")

    model_config = {"extra": "forbid", "populate_by_name": True}


class CodeNodeConfig(BaseModel):
    language: Literal["python"] = "python"
    source: str
    timeout_seconds: int = Field(default=60, ge=1, alias="timeoutSeconds")
    allow_network: bool = Field(default=False, alias="allowNetwork")

    model_config = {"extra": "forbid", "populate_by_name": True}


class TransformNodeConfig(BaseModel):
    operations: list[dict[str, Any]] = Field(default_factory=list)
    template: str | None = None

    model_config = {"extra": "forbid", "populate_by_name": True}


class ConditionBranch(BaseModel):
    name: str
    expression: str

    model_config = {"extra": "forbid", "populate_by_name": True}


class ConditionNodeConfig(BaseModel):
    branches: list[ConditionBranch] = Field(default_factory=list)
    default_branch: str | None = Field(default=None, alias="defaultBranch")

    model_config = {"extra": "forbid", "populate_by_name": True}


class MergeNodeConfig(BaseModel):
    strategy: MergeStrategy = "all"
    sources: list[str] = Field(default_factory=list)

    model_config = {"extra": "forbid", "populate_by_name": True}


class HumanInputNodeConfig(BaseModel):
    form_schema: dict[str, Any] = Field(default_factory=dict, alias="formSchema")
    prompt: str = ""
    actions: list[str] = Field(default_factory=lambda: ["submit"])

    model_config = {"extra": "forbid", "populate_by_name": True}


class SubworkflowNodeConfig(BaseModel):
    workflow_id: str = Field(alias="workflowId")
    version_id: str = Field(alias="versionId")
    input_mapping: dict[str, str] = Field(default_factory=dict, alias="inputMapping")

    model_config = {"extra": "forbid", "populate_by_name": True}


class LoopNodeConfig(BaseModel):
    max_iterations: int = Field(default=5, ge=1, alias="maxIterations")
    continue_when: str = Field(default="", alias="continueWhen")
    body_entry: str = Field(alias="bodyEntry")

    model_config = {"extra": "forbid", "populate_by_name": True}


class OutputNodeConfig(BaseModel):
    mapping: dict[str, str] = Field(default_factory=dict)
    llm_repair: bool = Field(default=False, alias="llmRepair")

    model_config = {"extra": "forbid", "populate_by_name": True}


__all__ = [
    "SCHEMA_VERSION",
    "ALL_NODE_TYPES",
    "ALL_RUN_STATUSES",
    "TERMINAL_RUN_STATUSES",
    "AgentNodeConfig",
    "AgentResponseMode",
    "AgentThreadMode",
    "ArtifactRef",
    "CodeNodeConfig",
    "ConditionBranch",
    "ConditionNodeConfig",
    "ExecutionMode",
    "HttpMethod",
    "HttpNodeConfig",
    "HumanInputNodeConfig",
    "LoopNodeConfig",
    "MergeNodeConfig",
    "MergeStrategy",
    "NodeResult",
    "NodeRunStatus",
    "NodeType",
    "OutputNodeConfig",
    "ResumeRunRequest",
    "RunStatus",
    "SkillMode",
    "SkillNodeConfig",
    "SqlReadNodeConfig",
    "StartRunRequest",
    "SubworkflowNodeConfig",
    "TransformNodeConfig",
    "WorkflowEdge",
    "WorkflowGraph",
    "WorkflowGraphSettings",
    "WorkflowNode",
]