"""Unified tool for agent operations with automatic state detection."""

import logging
from typing import Any

from pydantic import BaseModel, Field, field_validator

from backend.api.features.library.model import LibraryAgentPresetCreatable
from backend.copilot.config import ChatConfig
from backend.copilot.constants import MAX_TOOL_WAIT_SECONDS
from backend.copilot.model import ChatSession
from backend.copilot.tracking import track_agent_run_success, track_agent_scheduled
from backend.data.db_accessors import execution_db, graph_db, library_db, user_db
from backend.data.execution import ExecutionStatus, GraphExecutionWithNodes
from backend.data.graph import GraphModel
from backend.data.model import CredentialsMetaInput
from backend.executor import utils as execution_utils
from backend.executor.utils import is_credential_validation_error_message
from backend.util.clients import get_database_manager_async_client, get_scheduler_client
from backend.util.exceptions import DatabaseError, GraphValidationError, NotFoundError
from backend.util.timezone_utils import (
    convert_utc_time_to_user_timezone,
    get_user_timezone_or_utc,
)

from .base import BaseTool
from .execution_utils import get_execution_outputs, wait_for_execution
from .helpers import get_inputs_from_schema
from .models import (
    AgentDetails,
    AgentDetailsResponse,
    AgentOutputResponse,
    ErrorResponse,
    ExecutionOptions,
    ExecutionOutputInfo,
    ExecutionStartedResponse,
    InputValidationErrorResponse,
    SetupInfo,
    SetupRequirementsResponse,
    ToolResponseBase,
    UserReadiness,
)
from .utils import (
    build_missing_credentials_from_graph,
    extract_credentials_from_schema,
    fetch_graph_from_store_slug,
    get_or_create_library_agent,
    match_user_credentials_to_graph,
)

logger = logging.getLogger(__name__)
config = ChatConfig()


async def _safe_link_to_chat_share(session_id: str, execution_id: str) -> None:
    """Best-effort client-side wrapper around the chat-share auto-link hook.

    The server-side function ``link_new_execution_to_chat_share`` has its
    own try/except (db.py) so server-raised exceptions never bubble.  But
    the call goes over the DatabaseManager RPC boundary, and transport-
    layer failures (``ClientNotConnectedError``, RabbitMQ flap, retry
    exhaustion) raise CLIENT-side — before the wrapped function runs.

    Six callsites in this file dispatch the hook from inside ``run_agent``,
    which must never crash because the wrapper SDK turns an orphan
    ``tool_use`` into "The model returned an empty response."  So catch
    everything here and log; the owner can recover by re-toggling share
    (``_collect_execution_ids_from_messages`` backfills at re-enable time).
    """
    try:
        await get_database_manager_async_client().link_new_execution_to_chat_share(
            session_id=session_id, execution_id=execution_id
        )
    except Exception:
        logger.warning(
            "link_new_execution_to_chat_share RPC failed for session=%s "
            "execution=%s; owner can re-share to recover",
            session_id,
            execution_id,
            exc_info=True,
        )


# Constants for response messages
MSG_DO_NOT_RUN_AGAIN = "Do not run again unless explicitly requested."
MSG_DO_NOT_SCHEDULE_AGAIN = "Do not schedule again unless explicitly requested."
SCHEDULED_STATUS = "SCHEDULED"
MSG_ASK_USER_FOR_VALUES = (
    "Ask the user what values to use, or call again with use_defaults=true "
    "to run with default values."
)
MSG_WHAT_VALUES_TO_USE = (
    "What values would you like to use, or would you like to run with defaults?"
)


class RunAgentInput(BaseModel):
    """Input parameters for the run_agent tool."""

    username_agent_slug: str = ""
    library_agent_id: str = ""
    preset_id: str = ""
    inputs: dict[str, Any] = Field(default_factory=dict)
    use_defaults: bool = False
    schedule_name: str = ""
    cron: str = ""
    timezone: str = "UTC"
    wait_for_result: int = Field(default=0, ge=0, le=MAX_TOOL_WAIT_SECONDS)
    dry_run: bool = Field(default=False)
    save_as_preset: bool = False
    preset_name: str = ""
    preset_description: str = ""

    @field_validator(
        "username_agent_slug",
        "library_agent_id",
        "preset_id",
        "schedule_name",
        "cron",
        "timezone",
        "preset_name",
        mode="before",
    )
    @classmethod
    def strip_strings(cls, v: Any) -> Any:
        """Strip whitespace from string fields."""
        return v.strip() if isinstance(v, str) else v


class RunAgentTool(BaseTool):
    """Unified tool for agent operations with automatic state detection.

    The tool automatically determines what to do based on provided parameters:
    1. Fetches agent details (always, silently)
    2. Checks if required inputs are provided
    3. Checks if user has required credentials
    4. Runs immediately OR schedules (if cron is provided)

    The response tells the caller what's missing or confirms execution.
    """

    @property
    def name(self) -> str:
        return "run_agent"

    @property
    def description(self) -> str:
        return (
            "Run or schedule an agent. Automatically checks inputs and credentials "
            "and surfaces the inline credentials-setup card if anything is missing — "
            "do NOT redirect to the Builder for credential setup. "
            "Identify by username_agent_slug ('user/agent') or library_agent_id. "
            "For scheduling, provide schedule_name + cron. To run a saved preset, "
            "pass preset_id (alone). Pass save_as_preset=true (+ preset_name) to save "
            "these inputs as a reusable preset while running."
        )

    @property
    def parameters(self) -> dict[str, Any]:
        return {
            "type": "object",
            "properties": {
                "username_agent_slug": {
                    "type": "string",
                    "description": "Marketplace format 'username/agent-name'.",
                },
                "library_agent_id": {
                    "type": "string",
                    "description": "Library agent ID.",
                },
                "preset_id": {
                    "type": "string",
                    "description": (
                        "Run a saved preset by ID (uses its stored inputs + "
                        "credentials; 'inputs' override individual fields). Use "
                        "alone — not with an agent identifier or save_as_preset."
                    ),
                },
                "inputs": {
                    "type": "object",
                    "description": "Input values for the agent.",
                    "additionalProperties": True,
                },
                "use_defaults": {
                    "type": "boolean",
                    "description": "Run with default values (confirm with user first).",
                },
                "schedule_name": {
                    "type": "string",
                    "description": "Name for scheduled execution. Providing this triggers scheduling mode (also requires cron).",
                },
                "cron": {
                    "type": "string",
                    "description": "Cron expression (min hour day month weekday).",
                },
                "timezone": {
                    "type": "string",
                    "description": "IANA timezone (default: UTC).",
                },
                "wait_for_result": {
                    "type": "integer",
                    "description": (
                        f"Seconds to wait (0-{MAX_TOOL_WAIT_SECONDS}). "
                        "0 = fire-and-forget (returns execution_id). "
                        ">0 blocks for final status/outputs, plus "
                        "node_executions when dry_run. "
                        "Prefer 120 for dry-run, 0 for real runs."
                    ),
                    "minimum": 0,
                    "maximum": MAX_TOOL_WAIT_SECONDS,
                },
                "dry_run": {
                    "type": "boolean",
                    "description": "Simulate the agent run without executing real actions (default: false). Use when testing agent behaviour or when the user explicitly asks for a dry run.",
                },
            },
            "required": [],
        }

    @property
    def requires_auth(self) -> bool:
        """All operations require authentication."""
        return True

    async def _execute(
        self,
        user_id: str | None,
        session: ChatSession,
        **kwargs,
    ) -> ToolResponseBase:
        """Execute the tool with automatic state detection.

        Note: This tool accepts **kwargs and delegates to RunAgentInput for
        validation because the parameter set is complex with cross-field
        validators defined in the Pydantic model.
        """
        params = RunAgentInput(**kwargs)
        # Session-level dry_run forces all runs to be dry. In normal sessions
        # the LLM may still request dry_run=True on individual calls.
        if session.dry_run:
            params.dry_run = True
        session_id = session.session_id

        # Running a saved preset is a distinct path (uses the preset's stored
        # graph + inputs + credentials). Handle it before agent-identifier
        # resolution below.
        if params.preset_id:
            return await self._handle_preset_run(user_id, session, params)

        # Validate at least one identifier is provided
        has_slug = params.username_agent_slug and "/" in params.username_agent_slug
        has_library_id = bool(params.library_agent_id)

        # Builder-bound sessions can omit the identifier — default to the
        # bound graph so the LLM doesn't have to pass IDs the user never sees.
        builder_graph_id = session.metadata.builder_graph_id
        if builder_graph_id and user_id and not has_slug and not has_library_id:
            library_agent = await library_db().get_library_agent_by_graph_id(
                user_id, builder_graph_id
            )
            if library_agent:
                params.library_agent_id = library_agent.id
                has_library_id = True

        if not has_slug and not has_library_id:
            return ErrorResponse(
                message=(
                    "Please provide either a username_agent_slug "
                    "(format 'username/agent-name') or a library_agent_id"
                ),
                session_id=session_id,
            )

        # Auth is required
        if not user_id:
            return ErrorResponse(
                message="Authentication required. Please sign in to use this tool.",
                session_id=session_id,
            )

        # Determine if this is a schedule request
        is_schedule = bool(params.schedule_name or params.cron)

        # Session-level dry-run blocks scheduling — schedules create real
        # side effects that cannot be simulated.
        if params.dry_run and is_schedule:
            return ErrorResponse(
                message=(
                    "Scheduling is disabled in dry-run mode because it creates "
                    "real side effects. Remove cron/schedule_name to simulate "
                    "a run, or disable dry-run to create a real schedule."
                ),
                session_id=session_id,
            )

        try:
            # Step 1: Fetch agent details
            graph: GraphModel | None = None
            library_agent = None

            # Priority: library_agent_id if provided
            if has_library_id:
                library_agent = await library_db().get_library_agent(
                    params.library_agent_id, user_id
                )
                if not library_agent:
                    return ErrorResponse(
                        message=f"Library agent '{params.library_agent_id}' not found",
                        session_id=session_id,
                    )
                # Get the graph from the library agent
                graph = await graph_db().get_graph(
                    library_agent.graph_id,
                    library_agent.graph_version,
                    user_id=user_id,
                )
            else:
                # Fetch from marketplace slug
                username, agent_name = params.username_agent_slug.split("/", 1)
                graph, _ = await fetch_graph_from_store_slug(username, agent_name)

            if not graph:
                identifier = (
                    params.library_agent_id
                    if has_library_id
                    else params.username_agent_slug
                )
                return ErrorResponse(
                    message=f"Agent '{identifier}' not found",
                    session_id=session_id,
                )

            # Builder-bound sessions can only run their bound agent.  We
            # resolve the graph first so the user sees a precise error that
            # references the agent they actually asked to run, rather than
            # pre-emptively rejecting every run request.
            if builder_graph_id and graph.id != builder_graph_id:
                return ErrorResponse(
                    message=(
                        "This chat is bound to the builder's current agent. "
                        "Running a different agent is not allowed here."
                    ),
                    error="builder_session_graph_mismatch",
                    session_id=session_id,
                )

            # Webhook-trigger agents can't be run or scheduled directly — they
            # fire on incoming HTTP events. Hand off to the trigger-setup tool,
            # surfacing the same AgentDetails (with trigger_info) that run_agent
            # uses elsewhere so AutoPilot has the provider + config schema ready.
            if graph.has_external_trigger:
                credentials = extract_credentials_from_schema(
                    graph.credentials_input_schema
                )
                return AgentDetailsResponse(
                    message=(
                        f"Agent '{graph.name}' runs on a webhook trigger, so it "
                        "can't be run or scheduled directly. Set it up with "
                        "setup_agent_webhook_trigger using the trigger block's "
                        "config (see trigger_info.config_schema). For provider "
                        "webhooks (e.g. GitHub), ask the user which connected "
                        "account to register the webhook under — never auto-pick."
                    ),
                    session_id=session_id,
                    agent=self._build_agent_details(graph, credentials),
                    user_authenticated=True,
                    graph_id=graph.id,
                    graph_version=graph.version,
                )

            # Step 2: Check credentials and inputs
            graph_credentials, prereq_error = await self._check_prerequisites(
                graph=graph,
                user_id=user_id,
                params=params,
                session_id=session_id,
            )
            if prereq_error:
                return prereq_error

            # Step 3: Execute or Schedule
            if is_schedule:
                result = await self._schedule_agent(
                    user_id=user_id,
                    session=session,
                    graph=graph,
                    graph_credentials=graph_credentials,
                    inputs=params.inputs,
                    schedule_name=params.schedule_name,
                    cron=params.cron,
                    timezone=params.timezone,
                )
            else:
                result = await self._run_agent(
                    user_id=user_id,
                    session=session,
                    graph=graph,
                    graph_credentials=graph_credentials,
                    inputs=params.inputs,
                    wait_for_result=params.wait_for_result,
                    dry_run=params.dry_run,
                )

            # Step 4: persist the validated config as a reusable preset — only
            # after the run/schedule actually started, so a failed operation
            # (e.g. schedule validation, credential race) doesn't orphan a preset.
            if isinstance(result, ExecutionStartedResponse):
                saved_preset_id = await self._maybe_save_preset(
                    user_id=user_id,
                    graph=graph,
                    graph_credentials=graph_credentials,
                    params=params,
                )
                if saved_preset_id:
                    result.saved_preset_id = saved_preset_id
            return result

        except NotFoundError as e:
            return ErrorResponse(
                message=f"Agent '{params.username_agent_slug}' not found",
                error=str(e) if str(e) else "not_found",
                session_id=session_id,
            )
        except DatabaseError as e:
            logger.error("Database error: %s", e, exc_info=True)
            return ErrorResponse(
                message=f"Failed to process request: {e!s}",
                error=str(e),
                session_id=session_id,
            )
        except Exception as e:
            logger.error("Error processing agent request: %s", e, exc_info=True)
            return ErrorResponse(
                message=f"Failed to process request: {e!s}",
                error=str(e),
                session_id=session_id,
            )

    def _get_execution_modes(self, graph: GraphModel) -> list[str]:
        """Get available execution modes for the graph."""
        trigger_info = graph.trigger_setup_info
        if trigger_info is None:
            return ["manual", "scheduled"]
        return ["webhook"]

    def _build_inputs_message(
        self,
        graph: GraphModel,
        suffix: str,
    ) -> str:
        """Build a message describing available inputs for an agent."""
        inputs_list = get_inputs_from_schema(graph.input_schema)
        required_names = [i["name"] for i in inputs_list if i["required"]]
        optional_names = [i["name"] for i in inputs_list if not i["required"]]

        message_parts = [f"Agent '{graph.name}' accepts the following inputs:"]
        if required_names:
            message_parts.append(f"Required: {', '.join(required_names)}.")
        if optional_names:
            message_parts.append(
                f"Optional (have defaults): {', '.join(optional_names)}."
            )
        if not inputs_list:
            message_parts = [f"Agent '{graph.name}' has no required inputs."]
        message_parts.append(suffix)

        return " ".join(message_parts)

    def _build_agent_details(
        self,
        graph: GraphModel,
        credentials: list[CredentialsMetaInput],
    ) -> AgentDetails:
        """Build AgentDetails from a graph."""
        trigger_info = graph.trigger_setup_info
        return AgentDetails(
            id=graph.id,
            name=graph.name,
            description=graph.description,
            inputs=graph.input_schema,
            credentials=credentials,
            execution_options=ExecutionOptions(
                manual=trigger_info is None,
                scheduled=trigger_info is None,
                webhook=trigger_info is not None,
            ),
            trigger_info=trigger_info,
        )

    def _build_setup_requirements_from_validation_error(
        self,
        graph: GraphModel,
        error: GraphValidationError,
        session_id: str,
    ) -> SetupRequirementsResponse | None:
        """Turn a credential-only ``GraphValidationError`` into the inline
        setup-requirements card; return ``None`` if *any* non-credential
        error is present so the caller falls back to the plain text path
        (otherwise structural errors would be hidden)."""
        messages = [
            msg
            for node_errors in error.node_errors.values()
            for msg in node_errors.values()
        ]
        if not messages or not all(
            is_credential_validation_error_message(msg) for msg in messages
        ):
            return None

        # Show ALL credential fields as missing — the previously-matched
        # creds are now invalid, so narrowing to `error.node_errors` would
        # leak the stale mapping. Passing ``None`` means no field is
        # treated as "already connected".
        credentials_dict = build_missing_credentials_from_graph(graph, None)
        return SetupRequirementsResponse(
            message=(
                f"Agent '{graph.name}' has credentials that are missing or "
                "no longer valid. Please connect the required account(s) "
                "and try again."
            ),
            session_id=session_id,
            setup_info=SetupInfo(
                agent_id=graph.id,
                agent_name=graph.name,
                user_readiness=UserReadiness(
                    has_all_credentials=False,
                    missing_credentials=credentials_dict,
                    ready_to_run=False,
                ),
                requirements={
                    "credentials": list(credentials_dict.values()),
                    "inputs": get_inputs_from_schema(graph.input_schema),
                    "execution_modes": self._get_execution_modes(graph),
                },
            ),
            graph_id=graph.id,
            graph_version=graph.version,
        )

    def _handle_graph_validation_race(
        self,
        error: GraphValidationError,
        graph: GraphModel,
        user_id: str,
        session_id: str,
        action_verb: str,
    ) -> ToolResponseBase:
        """Handle a ``GraphValidationError`` that slipped past the prereq check.

        Shared by both the run and schedule paths — logs the race, attempts to
        rebuild the credential setup card, and falls back to a user-friendly
        ``ErrorResponse`` when the error is structural (not credential-related).
        """
        logger.warning(
            "Race: GraphValidationError after prereq check passed "
            "(user_id=%s graph_id=%s failing_fields=%s)",
            user_id,
            graph.id,
            {node_id: list(fields) for node_id, fields in error.node_errors.items()},
        )
        creds_setup = self._build_setup_requirements_from_validation_error(
            graph=graph,
            error=error,
            session_id=session_id,
        )
        if creds_setup is not None:
            return creds_setup
        return ErrorResponse(
            message=(
                f"Agent has configuration issues that need to be resolved "
                f"before {action_verb}: {error}"
            ),
            error="graph_validation_failed",
            session_id=session_id,
        )

    async def _check_prerequisites(
        self,
        graph: GraphModel,
        user_id: str,
        params: "RunAgentInput",
        session_id: str,
    ) -> tuple[dict[str, CredentialsMetaInput], ToolResponseBase | None]:
        """Validate credentials and inputs before execution.

        Dry runs skip all prerequisite gates (credentials, input prompts).
        The dry_run flag is read from params.dry_run (which may be set by the
        LLM per-call, or forced to True when session.dry_run is True).

        Returns:
            (graph_credentials, error_response) — error_response is None when ready.
        """
        graph_credentials, missing_creds = await match_user_credentials_to_graph(
            user_id, graph
        )

        # --- Reject unknown input fields (always, even for dry runs) ---
        input_properties = graph.input_schema.get("properties", {})
        provided_inputs = set(params.inputs.keys())
        valid_fields = set(input_properties.keys())
        unrecognized_fields = provided_inputs - valid_fields
        if unrecognized_fields:
            return graph_credentials, InputValidationErrorResponse(
                message=(
                    f"Unknown input field(s) provided: {', '.join(sorted(unrecognized_fields))}. "
                    f"Agent was not executed. Please use the correct field names from the schema."
                ),
                session_id=session_id,
                unrecognized_fields=sorted(unrecognized_fields),
                inputs=graph.input_schema,
                graph_id=graph.id,
                graph_version=graph.version,
            )

        # Dry runs bypass remaining prerequisite gates (credentials, missing inputs)
        if params.dry_run:
            return graph_credentials, None

        # --- Credential gate ---
        if missing_creds:
            requirements_creds_dict = build_missing_credentials_from_graph(graph, None)
            missing_credentials_dict = build_missing_credentials_from_graph(
                graph, graph_credentials
            )
            return graph_credentials, SetupRequirementsResponse(
                message=self._build_inputs_message(graph, MSG_WHAT_VALUES_TO_USE),
                session_id=session_id,
                setup_info=SetupInfo(
                    agent_id=graph.id,
                    agent_name=graph.name,
                    user_readiness=UserReadiness(
                        has_all_credentials=False,
                        missing_credentials=missing_credentials_dict,
                        ready_to_run=False,
                    ),
                    requirements={
                        "credentials": list(requirements_creds_dict.values()),
                        "inputs": get_inputs_from_schema(graph.input_schema),
                        "execution_modes": self._get_execution_modes(graph),
                    },
                ),
                graph_id=graph.id,
                graph_version=graph.version,
            )

        # --- Input gates ---
        required_fields = set(graph.input_schema.get("required", []))

        # Prompt user when inputs exist but none were provided
        if input_properties and not provided_inputs and not params.use_defaults:
            credentials = extract_credentials_from_schema(
                graph.credentials_input_schema
            )
            return graph_credentials, AgentDetailsResponse(
                message=self._build_inputs_message(graph, MSG_ASK_USER_FOR_VALUES),
                session_id=session_id,
                agent=self._build_agent_details(graph, credentials),
                user_authenticated=True,
                graph_id=graph.id,
                graph_version=graph.version,
            )

        # Required inputs missing
        missing_inputs = required_fields - provided_inputs
        if missing_inputs and not params.use_defaults:
            credentials = extract_credentials_from_schema(
                graph.credentials_input_schema
            )
            return graph_credentials, AgentDetailsResponse(
                message=(
                    f"Agent '{graph.name}' is missing required inputs: "
                    f"{', '.join(missing_inputs)}. "
                    "Please provide these values to run the agent."
                ),
                session_id=session_id,
                agent=self._build_agent_details(graph, credentials),
                user_authenticated=True,
                graph_id=graph.id,
                graph_version=graph.version,
            )

        return graph_credentials, None

    async def _handle_preset_run(
        self,
        user_id: str | None,
        session: ChatSession,
        params: RunAgentInput,
    ) -> ToolResponseBase:
        """Run a saved preset by id (mirrors POST /presets/{id}/execute)."""
        session_id = session.session_id
        if not user_id:
            return ErrorResponse(
                message="Authentication required. Please sign in to use this tool.",
                session_id=session_id,
            )
        if params.username_agent_slug or params.library_agent_id:
            return ErrorResponse(
                message=(
                    "Use either preset_id or an agent identifier "
                    "(username_agent_slug / library_agent_id), not both."
                ),
                session_id=session_id,
            )
        if params.save_as_preset:
            return ErrorResponse(
                message=(
                    "save_as_preset can't be combined with preset_id — "
                    "the preset already exists."
                ),
                session_id=session_id,
            )
        if params.schedule_name or params.cron:
            return ErrorResponse(
                message=(
                    "preset_id runs the preset now; schedule it separately instead."
                ),
                session_id=session_id,
            )

        preset = await library_db().get_preset(
            user_id=user_id, preset_id=params.preset_id
        )
        if not preset:
            return ErrorResponse(
                message=f"Preset '{params.preset_id}' not found.",
                error="preset_not_found",
                session_id=session_id,
            )
        graph = await graph_db().get_graph(
            preset.graph_id, preset.graph_version, user_id=user_id
        )
        if not graph:
            return ErrorResponse(
                message=(
                    f"The agent for preset '{params.preset_id}' is not "
                    "accessible (anymore)."
                ),
                session_id=session_id,
            )

        # Builder-bound sessions can only run their bound agent — enforce the
        # same guard as the regular run path so a preset for a different graph
        # can't execute in a builder-bound chat.
        builder_graph_id = session.metadata.builder_graph_id
        if builder_graph_id and graph.id != builder_graph_id:
            return ErrorResponse(
                message=(
                    "This chat is bound to the builder's current agent. "
                    "Running a preset for a different agent is not allowed here."
                ),
                error="builder_session_graph_mismatch",
                session_id=session_id,
            )

        # A webhook-triggered preset fires on its external event; it has no
        # runnable payload here, so executing it directly would fail downstream.
        # Reject cleanly (matching run_agent's has_external_trigger guard).
        if graph.has_external_trigger:
            return ErrorResponse(
                message=(
                    f"Preset '{params.preset_id}' is a webhook trigger — it runs "
                    "automatically when its event fires, so it can't be run on "
                    "demand. Use update_preset to reconfigure or pause it "
                    "(is_active=false), or delete_preset to remove it."
                ),
                error="preset_is_webhook_trigger",
                session_id=session_id,
            )

        merged_inputs = {**preset.inputs, **params.inputs}
        return await self._run_agent(
            user_id=user_id,
            session=session,
            graph=graph,
            graph_credentials=preset.credentials,
            inputs=merged_inputs,
            wait_for_result=params.wait_for_result,
            dry_run=params.dry_run,
            preset_id=preset.id,
        )

    async def _maybe_save_preset(
        self,
        *,
        user_id: str,
        graph: GraphModel,
        graph_credentials: dict[str, CredentialsMetaInput],
        params: RunAgentInput,
    ) -> str | None:
        """Persist the validated run config as a reusable preset when requested.

        Returns the new preset id, or None when save_as_preset wasn't set.
        """
        if not params.save_as_preset:
            return None
        created = await library_db().create_preset(
            user_id=user_id,
            preset=LibraryAgentPresetCreatable(
                graph_id=graph.id,
                graph_version=graph.version,
                name=params.preset_name or graph.name,
                description=params.preset_description,
                inputs=params.inputs,
                credentials=graph_credentials,
                is_active=True,
            ),
        )
        return created.id

    async def _run_agent(
        self,
        user_id: str,
        session: ChatSession,
        graph: GraphModel,
        graph_credentials: dict[str, CredentialsMetaInput],
        inputs: dict[str, Any],
        dry_run: bool,
        wait_for_result: int = 0,
        preset_id: str | None = None,
    ) -> ToolResponseBase:
        """Execute an agent immediately, optionally waiting for completion."""
        session_id = session.session_id

        # Check rate limits (dry runs don't count against the session limit)
        if (
            not dry_run
            and session.successful_agent_runs.get(graph.id, 0) >= config.max_agent_runs
        ):
            return ErrorResponse(
                message="Maximum agent runs reached for this session. Please try again later.",
                session_id=session_id,
            )

        # Get or create library agent
        library_agent = await get_or_create_library_agent(graph, user_id)

        # Execute — ``add_graph_execution`` ultimately calls
        # ``validate_and_construct_node_execution_input`` which raises
        # ``GraphValidationError`` on missing/invalid credentials.  The
        # common case is caught by ``_check_prerequisites`` above, but
        # defend against a race (creds deleted between prereq and
        # execute) by turning credential errors back into the inline
        # setup card.
        try:
            execution = await execution_utils.add_graph_execution(
                graph_id=library_agent.graph_id,
                user_id=user_id,
                inputs=inputs,
                graph_credentials_inputs=graph_credentials,
                dry_run=dry_run,
                preset_id=preset_id,
            )
        except GraphValidationError as e:
            return self._handle_graph_validation_race(
                error=e,
                graph=graph,
                user_id=user_id,
                session_id=session_id,
                action_verb="running",
            )

        # Track successful run (dry runs don't count against the session limit)
        if not dry_run:
            session.successful_agent_runs[library_agent.graph_id] = (
                session.successful_agent_runs.get(library_agent.graph_id, 0) + 1
            )

        # Track in PostHog
        track_agent_run_success(
            user_id=user_id,
            session_id=session_id,
            graph_id=library_agent.graph_id,
            graph_name=library_agent.name,
            execution_id=execution.id,
            library_agent_id=library_agent.id,
        )

        library_agent_link = f"/library/agents/{library_agent.id}"

        # If wait_for_result is requested, wait for execution to complete
        if wait_for_result > 0:
            logger.info(
                f"Waiting up to {wait_for_result}s for execution {execution.id}"
            )
            completed = await wait_for_execution(
                user_id=user_id,
                graph_id=library_agent.graph_id,
                execution_id=execution.id,
                timeout_seconds=wait_for_result,
            )

            if completed and completed.status == ExecutionStatus.COMPLETED:
                outputs = get_execution_outputs(completed)
                # Inline the per-node execution trace on dry-runs so the
                # LLM can inspect "did every block run, what did each
                # produce?" without a follow-up view_agent_output call.
                # Empty final outputs on a COMPLETED dry-run almost always
                # mean a node silently produced nothing / a link was wired
                # wrong — the trace is what lets the model debug that.
                node_executions_data = None
                if dry_run:
                    try:
                        detailed = await execution_db().get_graph_execution(
                            user_id=user_id,
                            execution_id=execution.id,
                            include_node_executions=True,
                        )
                        if isinstance(detailed, GraphExecutionWithNodes):
                            node_executions_data = [
                                {
                                    "node_id": ne.node_id,
                                    "block_id": ne.block_id,
                                    "status": ne.status.value,
                                    "input_data": ne.input_data,
                                    "output_data": dict(ne.output_data),
                                    "start_time": (
                                        ne.start_time.isoformat()
                                        if ne.start_time
                                        else None
                                    ),
                                    "end_time": (
                                        ne.end_time.isoformat() if ne.end_time else None
                                    ),
                                }
                                for ne in detailed.node_executions
                            ]
                    except Exception:
                        logger.warning(
                            "run_agent: failed to load node executions for "
                            "dry-run %s; returning summary only",
                            execution.id,
                            exc_info=True,
                        )
                await _safe_link_to_chat_share(
                    session_id=session_id, execution_id=execution.id
                )
                return AgentOutputResponse(
                    message=(
                        f"Agent '{library_agent.name}' completed successfully. "
                        f"View at {library_agent_link}."
                    ),
                    session_id=session_id,
                    agent_name=library_agent.name,
                    agent_id=library_agent.graph_id,
                    library_agent_id=library_agent.id,
                    library_agent_link=library_agent_link,
                    execution=ExecutionOutputInfo(
                        execution_id=execution.id,
                        status=completed.status.value,
                        started_at=completed.started_at,
                        ended_at=completed.ended_at,
                        outputs=outputs or {},
                        node_executions=node_executions_data,
                    ),
                )
            elif completed and completed.status == ExecutionStatus.FAILED:
                error_detail = completed.stats.error if completed.stats else None
                # Auto-share the failed run too — share-modal users may
                # want public viewers to drill into the failure.  Without
                # this hook, ``_collect_execution_ids_from_messages`` can't
                # backfill failed runs either (ErrorResponse payload type
                # isn't in the scanned set).
                await _safe_link_to_chat_share(
                    session_id=session_id, execution_id=execution.id
                )
                return ErrorResponse(
                    message=(
                        f"Agent '{library_agent.name}' execution failed. "
                        f"View details at {library_agent_link}."
                    ),
                    session_id=session_id,
                    execution_id=execution.id,
                    error=error_detail,
                )
            elif completed and completed.status == ExecutionStatus.TERMINATED:
                error_detail = completed.stats.error if completed.stats else None
                # Auto-share terminated runs (cancelled / killed) for the
                # same reason as the FAILED branch — backfill at re-share
                # time won't pick them up.
                await _safe_link_to_chat_share(
                    session_id=session_id, execution_id=execution.id
                )
                return ErrorResponse(
                    message=(
                        f"Agent '{library_agent.name}' execution was terminated. "
                        f"View details at {library_agent_link}."
                    ),
                    session_id=session_id,
                    execution_id=execution.id,
                    error=error_detail,
                )
            elif completed and completed.status == ExecutionStatus.REVIEW:
                await _safe_link_to_chat_share(
                    session_id=session_id, execution_id=execution.id
                )
                return ExecutionStartedResponse(
                    message=(
                        f"Agent '{library_agent.name}' is awaiting human review. "
                        f"The user can approve or reject inline. After approval, "
                        f"the execution resumes automatically. Use view_agent_output "
                        f"with execution_id='{execution.id}' to check the result."
                    ),
                    session_id=session_id,
                    execution_id=execution.id,
                    graph_id=library_agent.graph_id,
                    graph_name=library_agent.name,
                    library_agent_id=library_agent.id,
                    library_agent_link=library_agent_link,
                    status=ExecutionStatus.REVIEW.value,
                )
            else:
                status = completed.status.value if completed else "unknown"
                await _safe_link_to_chat_share(
                    session_id=session_id, execution_id=execution.id
                )
                return ExecutionStartedResponse(
                    message=(
                        f"Agent '{library_agent.name}' is still {status} after "
                        f"{wait_for_result}s. Check results later at "
                        f"{library_agent_link}. "
                        f"Use view_agent_output with wait_if_running to check again."
                    ),
                    session_id=session_id,
                    execution_id=execution.id,
                    graph_id=library_agent.graph_id,
                    graph_name=library_agent.name,
                    library_agent_id=library_agent.id,
                    library_agent_link=library_agent_link,
                    status=status,
                )

        await _safe_link_to_chat_share(session_id=session_id, execution_id=execution.id)
        return ExecutionStartedResponse(
            message=(
                f"Agent '{library_agent.name}' execution started successfully. "
                f"View at {library_agent_link}. "
                f"{MSG_DO_NOT_RUN_AGAIN}"
            ),
            session_id=session_id,
            execution_id=execution.id,
            graph_id=library_agent.graph_id,
            graph_name=library_agent.name,
            library_agent_id=library_agent.id,
            library_agent_link=library_agent_link,
        )

    async def _schedule_agent(
        self,
        user_id: str,
        session: ChatSession,
        graph: GraphModel,
        graph_credentials: dict[str, CredentialsMetaInput],
        inputs: dict[str, Any],
        schedule_name: str,
        cron: str,
        timezone: str,
    ) -> ToolResponseBase:
        """Set up scheduled execution for an agent."""
        session_id = session.session_id

        # Validate schedule params
        if not schedule_name:
            return ErrorResponse(
                message="schedule_name is required for scheduled execution",
                session_id=session_id,
            )
        if not cron:
            return ErrorResponse(
                message="cron expression is required for scheduled execution",
                session_id=session_id,
            )

        # Check rate limits
        if (
            session.successful_agent_schedules.get(graph.id, 0)
            >= config.max_agent_schedules
        ):
            return ErrorResponse(
                message="Maximum agent schedules reached for this session.",
                session_id=session_id,
            )

        # Get or create library agent
        library_agent = await get_or_create_library_agent(graph, user_id)

        # Get user timezone
        user = await user_db().get_user_by_id(user_id)
        user_timezone = get_user_timezone_or_utc(user.timezone if user else timezone)

        # Create schedule — the scheduler re-validates credentials via
        # ``validate_and_construct_node_execution_input`` and will raise
        # ``GraphValidationError`` if any required credential is missing
        # or invalid.  ``_check_prerequisites`` already catches the
        # common case at the top of ``_execute``, but a race (creds
        # deleted between prereq check and scheduler call) or any other
        # validation drift could hit here — turn credential errors back
        # into the inline ``SetupRequirementsResponse`` so the user
        # sees the credential setup card instead of a generic error.
        try:
            result = await get_scheduler_client().add_execution_schedule(
                user_id=user_id,
                graph_id=library_agent.graph_id,
                graph_version=library_agent.graph_version,
                name=schedule_name,
                cron=cron,
                input_data=inputs,
                input_credentials=graph_credentials,
                user_timezone=user_timezone,
            )
        except GraphValidationError as e:
            return self._handle_graph_validation_race(
                error=e,
                graph=graph,
                user_id=user_id,
                session_id=session_id,
                action_verb="scheduling",
            )

        # Convert next_run_time to user timezone for display
        if result.next_run_time:
            result.next_run_time = convert_utc_time_to_user_timezone(
                result.next_run_time, user_timezone
            )

        # Track successful schedule
        session.successful_agent_schedules[library_agent.graph_id] = (
            session.successful_agent_schedules.get(library_agent.graph_id, 0) + 1
        )

        # Track in PostHog
        track_agent_scheduled(
            user_id=user_id,
            session_id=session_id,
            graph_id=library_agent.graph_id,
            graph_name=library_agent.name,
            schedule_id=result.id,
            schedule_name=schedule_name,
            cron=cron,
            library_agent_id=library_agent.id,
        )

        library_agent_link = f"/library/agents/{library_agent.id}"
        return ExecutionStartedResponse(
            message=(
                f"Agent '{library_agent.name}' scheduled successfully as '{schedule_name}'. "
                f"View at {library_agent_link}. "
                f"{MSG_DO_NOT_SCHEDULE_AGAIN}"
            ),
            session_id=session_id,
            execution_id=result.id,
            graph_id=library_agent.graph_id,
            graph_name=library_agent.name,
            library_agent_id=library_agent.id,
            library_agent_link=library_agent_link,
            status=SCHEDULED_STATUS,
        )
