"""Run-scoped execution context shared by the engine and node executors."""

from __future__ import annotations

import asyncio
from collections.abc import Awaitable, Callable
from dataclasses import dataclass, field
from typing import Any

from deerflow.workflows.errors import WorkflowError
from deerflow.workflows.events import WorkflowEventType
from deerflow.workflows.schemas import ArtifactRef, WorkflowGraph, WorkflowNode

EmitFn = Callable[..., Awaitable[Any]]


def _public_artifact_payload(ref: ArtifactRef) -> dict[str, Any]:
    """Wire payload for ``artifact.created``.

    The sandbox/host ``path`` stays internal: clients address artifacts by id
    through the run-scoped content endpoints, never by filesystem path."""
    from deerflow.workflows.events import artifact_kind

    return {
        "artifactId": ref.artifact_id,
        "name": ref.name,
        "kind": artifact_kind(ref.mime_type),
        "mimeType": ref.mime_type,
        "sizeBytes": ref.size_bytes,
        "preview": ref.preview,
    }


class WorkflowPaused(Exception):  # noqa: N818 - control-flow signal, not an error
    """Raised by ``human_input`` to suspend the run until a resume arrives."""

    def __init__(self, node_id: str, pending_input: dict[str, Any]) -> None:
        super().__init__(f"workflow paused at {node_id}")
        self.node_id = node_id
        self.pending_input = pending_input


class CancelToken:
    """Cooperative cancellation shared between the executor and node code."""

    def __init__(self) -> None:
        self._event = asyncio.Event()
        self._reason = "cancelled"

    def cancel(self, reason: str = "cancelled") -> None:
        self._reason = reason
        self._event.set()

    @property
    def cancelled(self) -> bool:
        return self._event.is_set()

    async def wait(self) -> None:
        await self._event.wait()

    def raise_if_cancelled(self, node_id: str | None = None) -> None:
        if self._event.is_set():
            raise WorkflowError("WORKFLOW_CANCELLED", "运行已被取消", node_id=node_id)


@dataclass
class WorkflowRuntimeDeps:
    """Injected capabilities. The harness never imports the app layer, so every
    integration point (agents, HTTP egress, SQL, sandbox, subworkflows) arrives
    as an async callable supplied by the caller. ``None`` means "not available",
    which node executors turn into a typed ``WORKFLOW_RESOURCE_MISSING``."""

    run_agent: Callable[..., Awaitable[dict[str, Any]]] | None = None
    run_http: Callable[..., Awaitable[dict[str, Any]]] | None = None
    run_sql: Callable[..., Awaitable[dict[str, Any]]] | None = None
    run_code: Callable[..., Awaitable[dict[str, Any]]] | None = None
    run_subworkflow: Callable[..., Awaitable[dict[str, Any]]] | None = None
    # Starts (or reattaches to) a durable Deep Research report-writing job.
    # The app adapter owns its session/job/event stores; workflow nodes only
    # receive the final report plus safe progress callbacks.
    run_deep_research: Callable[..., Awaitable[dict[str, Any]]] | None = None
    save_artifact: Callable[..., Awaitable[dict[str, Any]]] | None = None
    resolve_credential: Callable[[str], Awaitable[dict[str, Any] | None]] | None = None
    resolve_data_source: Callable[[str], Awaitable[dict[str, Any] | None]] | None = None


@dataclass
class RunContext:
    """Everything a node executor is allowed to see."""

    run_id: str
    workflow_id: str
    version_id: str
    owner_id: str
    graph: WorkflowGraph
    inputs: dict[str, Any]
    deps: WorkflowRuntimeDeps
    cancel: CancelToken
    emit: EmitFn
    node_results: dict[str, dict[str, Any]] = field(default_factory=dict)
    loop_state: dict[str, int] = field(default_factory=dict)
    resume_payload: dict[str, Any] | None = None
    node_run_ids: dict[str, str] = field(default_factory=dict)
    current_node_id: str | None = None
    env: dict[str, Any] = field(default_factory=dict)

    def state(self) -> dict[str, Any]:
        """Whitelisted tree visible to expressions. Credentials never appear."""
        return {
            "inputs": self.inputs,
            "nodes": self.node_results,
            "run": {"id": self.run_id, "workflowId": self.workflow_id, "versionId": self.version_id},
            "loop": dict(self.loop_state),
            "env": self.env,
        }

    async def emit_event(
        self,
        event_type: WorkflowEventType,
        *,
        data: dict[str, Any] | None = None,
        node_id: str | None = None,
    ) -> None:
        node = node_id or self.current_node_id
        await self.emit(
            event_type,
            data=data or {},
            node_id=node,
            node_run_id=self.node_run_ids.get(node) if node else None,
        )

    async def register_artifact(
        self,
        node: WorkflowNode,
        *,
        name: str,
        mime_type: str = "application/octet-stream",
        path: str = "",
        size_bytes: int | None = None,
        preview: str | None = None,
    ) -> ArtifactRef | None:
        """Persist an artifact reference and emit ``artifact.created``."""
        if self.deps.save_artifact is None:
            return None
        row = await self.deps.save_artifact(
            run_id=self.run_id,
            node_id=node.id,
            name=name,
            mime_type=mime_type,
            path=path,
            size_bytes=size_bytes,
            preview=preview,
        )
        ref = ArtifactRef(
            artifactId=str(row.get("id") or ""),
            name=str(row.get("name") or name),
            mimeType=str(row.get("mime_type") or mime_type),
            path=str(row.get("path") or path),
            sizeBytes=row.get("size_bytes"),
            preview=row.get("preview"),
        )
        await self.emit_event("artifact.created", data=_public_artifact_payload(ref), node_id=node.id)
        return ref


__all__ = ["CancelToken", "RunContext", "WorkflowPaused", "WorkflowRuntimeDeps"]