import uuid
from datetime import datetime
from typing import Any, Literal

from pydantic import BaseModel, ConfigDict, Field, model_validator

PublishStatus = Literal["draft", "published"]

# ---- Tasks (Task Designer Studio) -------------------------------------------------------------
#
# A task is a node graph — the same canvas/transition/condition model as
# Workflow's step_definitions (app/schemas/workflow.py::WorkflowStepDef),
# but executed autonomously in one synchronous pass (no human-in-the-loop
# approval steps): POST /tasks/{id}/run starts at the "start" node and
# follows next_nodes — the first transition whose condition evaluates true
# against the shared context (or an unconditional one) — until a node has no
# matching transition, a node fails, or a step-count guard trips (a cycle
# guard, not a supported looping construct).
#
# Every node declares its data contract generically instead of one-off
# "context key" fields: input_params (each with a name, a type, which
# context key to read it from, and whether it's required) and output_params
# (name + type). app/crud/params.py resolves+type-checks input_params into a
# local `inputs` dict before a node runs, and extracts+type-checks
# output_params from its result into the shared context afterward — a
# mismatch or a missing required input fails the node with a clear detail
# rather than running with bad data. Each node type still keeps the fields
# that are structural configuration, not runtime data (which Connector,
# which Master, which SOAP operation, literal template text) — see each
# field's comment below for which named inputs a node type looks for.

TaskNodeType = Literal[
    "start", "condition", "loop",
    "email_read", "email_send", "sms_send",
    "data_get", "data_list", "data_update", "data_delete",
    "excel_read",
    "pdf_read", "pdf_generate",
    "api_rest", "api_soap", "api_mqtt",
    "create_notification", "generate_sequence", "payment_initiate",
    "rpa_run", "rpa_job_status",
    "sql_query", "sql_execute", "ftp_transfer",
    "google_sheets", "google_docs", "google_slides", "google_drive", "google_calendar", "google_gmail", "google_forms",
    "firebase_push", "firebase_topic",
    "custom_code",
    "navigate",
    # Sets task variables (or any context value) mid-run: each input param's
    # name is the variable, its fixed value / source field the new value.
    "set_variable",
]
# navigate — where the caller's browser goes once the run finishes.
TaskNavigateTarget = Literal["page", "form", "dashboard", "master", "url"]
TaskConditionOperator = Literal[
    "equals", "not_equals", "gt", "gte", "lt", "lte", "in", "not_in",
    "contains", "not_contains", "starts_with", "ends_with",
    "is_empty", "is_not_empty", "is_true", "is_false",
]
# Which exit of an If / Else ("true"/"false") or Loop ("body" = run for each
# item, "done" = after the loop) a transition is; None on every other node.
TransitionBranch = Literal["true", "false", "body", "done"]
LoopMode = Literal["for_each", "repeat", "while"]
RulesMatch = Literal["all", "any"]
TaskCodeLanguage = Literal["python", "node"]
# number = int or decimal; object = a whole record (dict); any = no check.
ParamType = Literal["int", "number", "string", "boolean", "list", "object", "any"]
SequenceResetPeriod = Literal["never", "yearly", "monthly"]
TaskDatasourceType = Literal["master", "form"]
EmailBodyFormat = Literal["text", "html"]
RpaAction = Literal["start_job", "add_queue_item"]
FtpOperation = Literal["list", "download", "upload", "move", "delete"]
SoapMode = Literal["wsdl", "raw"]
SoapAuth = Literal["none", "basic", "digest", "wsse_text", "wsse_digest", "bearer"]
SoapVersion = Literal["1.1", "1.2"]


class TaskNodeCondition(BaseModel):
    driver_field: str  # a variable name or dotted path, e.g. "item.amount"
    operator: TaskConditionOperator
    value: object = None  # compared loosely; "{{name}}" = another variable


class TaskNodeTransition(BaseModel):
    to: str
    condition: TaskNodeCondition | None = None  # None on an unconditional single transition
    branch: TransitionBranch | None = None


class NodeInputParam(BaseModel):
    name: str  # local name the node's config/template refers to this value by
    type: ParamType
    source_field: str = ""  # context key to read the value from
    required: bool = True
    # A fixed value instead of a context key — plain text, where {{variable}}
    # placeholders are filled from the running context (app/crud/params.py).
    # Set = used in place of source_field.
    value: str | None = None


class NodeOutputParam(BaseModel):
    name: str  # also the context key this output is written to
    type: ParamType
    # Where in the node's result the value comes from: blank = the result's
    # key of the same name (or the whole result when it isn't an object),
    # "*" = the whole result, "a.b" = a path inside it.
    source: str = ""


# set_variable — where its values go: "task" — task variables, read by every
# later node; "session" — also the signed-in user's session data, readable
# afterwards as {{session.<name>}} on every page, form and task (the browser
# keeps it, as with sign-in values); "context" — also handed back to the page
# or form that ran the task, as its own values ({{name}} there).
SetVariableScope = Literal["task", "session", "context"]


class TaskNodeDef(BaseModel):
    key: str  # client-generated, stable — canvas position + transition target, not a DB row id
    type: TaskNodeType
    label: str
    variable_scope: SetVariableScope = "task"  # set_variable only
    next_nodes: list[TaskNodeTransition] = []
    position_x: float = 0
    position_y: float = 0
    input_params: list[NodeInputParam] = []
    output_params: list[NodeOutputParam] = []

    # condition (If / Else) — evaluates `rules` (all or any, per
    # rules_match) against the variables and follows its "true" or "false"
    # transition.
    # loop — runs the transitions marked "body" once per iteration, then
    # continues along "done":
    #   for_each — over a required list input named "items" (e.g. a
    #              Data — List node's rows); each iteration sets
    #              loop_item_var (default "item") and loop_index_var
    #              (default "index", 0-based)
    #   repeat   — a required int input named "count" times
    #   while    — while `rules` hold (checked before each iteration)
    # An iteration ends where its path ends, or when it links back to the
    # loop node. loop_collect names variables to gather after each
    # iteration into the output "results" (a list of values for one name, a
    # list of objects for several); "count" and "failed" are outputs too.
    # A failing iteration stops the task unless loop_continue_on_error.
    rules: list[TaskNodeCondition] = []
    rules_match: RulesMatch = "all"
    loop_mode: LoopMode = "for_each"
    loop_item_var: str = "item"
    loop_index_var: str = "index"
    loop_max_iterations: int = 1000
    loop_collect: list[str] = []
    loop_continue_on_error: bool = False

    # email_read (IMAP) / email_send (SMTP) — both via a Connector
    connector_id: uuid.UUID | None = None
    # email_read — no inputs required; declare one output (any type — the
    # whole fetched message list is written under its name) to capture it
    folder: str = "INBOX"
    search_criteria: str = "UNSEEN"
    limit: int = 10
    # email_send — looks for a required string input named "to"; subject/body
    # are literal templates rendered against every declared input — the
    # template IS the input_params-driven placeholder mechanism every other
    # node's template field already uses, so an HTML body's "{{name}}" is a
    # placeholder exactly the same way a plain-text one is. body_format picks
    # which MIME subtype the body is sent as; "text" (the default) keeps
    # every already-persisted node's behavior unchanged.
    # sms_send — via a Connector (type "sms"); looks for a required string
    # input named "to"; reuses `body` (rendered the same way) as the message
    # text — no separate subject, and always sent as plain text regardless
    # of body_format (SMS has no HTML concept).
    subject: str = ""
    body: str = ""
    body_format: EmailBodyFormat = "text"
    # email_send, body_format "html" only — wraps the rendered body in one
    # styled <div>, since email clients routinely strip <style> blocks/CSS
    # classes but reliably keep inline styles. None means "don't set that
    # style at all" (inherits the recipient's mail client default), same
    # nullable convention as PageBlockDef's own font_family/font_color/
    # background_color (Page Designer's block Style tab).
    email_font_family: str | None = None
    email_font_size: int | None = None
    email_font_color: str | None = None
    email_background_color: str | None = None

    # create_notification — no Connector; writes directly to the in-app
    # notification list (same table Workflow step notifications use).
    # Reuses `subject` as the notification title and `body` as its message,
    # both rendered as templates against every declared input. Recipients:
    # every active member of notify_role_ids (internal or Public Portal
    # roles), plus notify_user_ids, plus an optional input named
    # "recipient_user_id" (a user id, or a list of them — e.g. source_field
    # "currentUserId", which POST /tasks/{id}/run always sets to the caller,
    # so a portal Form's on_submit task can notify the vendor who submitted
    # it). No recipients configured at all = the original tenant-wide
    # broadcast to internal users.
    notify_role_ids: list[uuid.UUID] = []
    notify_user_ids: list[uuid.UUID] = []

    # generate_sequence — atomically increments a durable per-tenant counter
    # named `sequence_key` and declares ONE output (any type — the formatted
    # string is written under its name) to capture the result, e.g.
    # "RTI-2026-0007". `reset_period` starts the counter over at 1 each new
    # calendar year/month (the counter key itself is period-suffixed, so a
    # past period's count is never lost, just no longer incremented).
    sequence_key: str = ""
    sequence_prefix: str = ""
    sequence_padding: int = 4
    sequence_reset_period: SequenceResetPeriod = "never"

    # payment_initiate — via a Connector (type "payment_gateway"); looks for
    # a required int input named "amount" (minor currency units, e.g. paise/
    # cents), an optional string "currency" (default "INR") and "reference".
    # POSTs to configuration.api_base_url + checkout_path; declare one
    # output to capture the gateway's {checkout_url, payment_reference}
    # response — send the user to checkout_url with an "Open external URL"
    # Page/Form action (see PageAction/FormEventAction's open_external_url).
    checkout_path: str = "/checkout"

    # rpa_run — via a Connector (type "rpa"; app/crud/rpa_providers.py).
    # "start_job" runs rpa_process (a UiPath process name, Automation
    # Anywhere bot id, Power Automate flow name or Blue Prism web-service
    # name) with every declared input passed as an input argument by name;
    # "add_queue_item" (UiPath) adds an item to rpa_queue instead, with the
    # inputs as its specific content and an optional input named
    # "reference" as its reference. With rpa_wait the node polls until the
    # job finishes or rpa_timeout_seconds pass. Its result is {job_id,
    # status, provider_status, info} plus every output argument by name —
    # declare outputs with those names. status is queued | running |
    # submitted | succeeded | failed | stopped; a failed/stopped job fails
    # the node unless rpa_fail_on_error is off (then branch on "status").
    # rpa_job_status — same connector/wait/result, for a required string
    # input named "job_id" (e.g. an earlier rpa_run output).
    rpa_action: RpaAction = "start_job"
    rpa_process: str = ""
    rpa_queue: str = ""
    rpa_wait: bool = True
    rpa_timeout_seconds: int = 120
    rpa_fail_on_error: bool = True

    # sql_query / sql_execute — via a Connector (type "database"; app/crud/
    # sql_providers.py). `sql` is ONE statement using :name placeholders,
    # each bound from the declared input of that name as a real driver
    # parameter (never pasted into the text). sql_query is always read-only
    # and returns {rows, row_count, truncated} plus the first row's columns
    # by name; sql_execute needs a connector that allows writes and returns
    # {affected_rows, rows (from RETURNING / OUTPUT), …first row}.
    sql: str = ""
    sql_max_rows: int = 1000
    sql_timeout_seconds: int = 30

    # ftp_transfer — via a Connector (type "file_transfer"; app/crud/
    # file_transfer.py). remote_path / ftp_target_path are templates rendered
    # against every declared input, relative to the connector's base folder.
    #   list     — files in remote_path (optionally matching ftp_pattern):
    #              {files, names, count, folder}
    #   download — remote_path is a file, or a folder + ftp_pattern (up to 20
    #              files); each becomes an uploaded file: {file_id, file_ids,
    #              file_name, count} — feed file_id to Excel/PDF Reader
    #   upload   — needs a required string input "file_id" (an uploaded
    #              file); saved into remote_path (a folder, or a full path),
    #              named ftp_target_path or the file's own name
    #   move     — remote_path (+ ftp_pattern) to ftp_target_path (a folder,
    #              or a new name for one file)
    #   delete   — remote_path (+ ftp_pattern)
    ftp_operation: FtpOperation = "list"
    remote_path: str = ""
    ftp_pattern: str = ""
    ftp_target_path: str = ""

    # data_get / data_list / data_update / data_delete — against a Master or
    # a Form's submissions, per datasource_type (defaults to "master" so
    # every already-persisted node keeps working unchanged). data_get/
    # data_delete look for a required string input named "match_value";
    # data_list looks for an optional one named "filter_value" (omit it to
    # list every record); data_update writes every OTHER declared input onto
    # the record's field_values under its own name, using "match_value" (if
    # declared) to find an existing record to update instead of always
    # creating a new one. match_field/filter_field is normally a field name,
    # but "id" is reserved to mean the record's own primary key instead —
    # the only way to target one exact record when the value you're writing
    # (e.g. a freshly generated application number) can't itself be used to
    # find the record, and the only way at all for a Form submission, which
    # has no author-defined unique field of its own. A Form's on_submit
    # event exposes the just-created submission's id as context key
    # "recordId" (see FormEventAction docs in app/schemas/workflow.py) — an
    # input_param with source_field "recordId" feeds it into "match_value"
    # here. A Form submission is always created against its current
    # (is_current) version. Deleting a Form submission soft-deletes it
    # (is_deleted — same column the Form Submissions list's own delete path
    # uses); deleting a Master record removes it outright, unchanged from
    # before Forms were supported here.
    datasource_type: TaskDatasourceType = "master"
    master_id: uuid.UUID | None = None
    form_id: uuid.UUID | None = None
    match_field: str | None = None  # field "match_value"/"filter_value" is compared against
    filter_field: str | None = None  # data_list only

    # excel_read / pdf_read — look for a required string input named
    # "file_id" (an UploadedFile id — e.g. a Form's file_upload field value,
    # already an id by the time it reaches context; see FormFieldRenderer.tsx)
    sheet_name: str | None = None  # excel_read only; None = first sheet

    # pdf_generate — template rendered against every declared input, one PDF
    # line per template line
    template: str = ""

    # api_rest — url/headers/body_template are templates rendered against
    # every declared input
    method: str = "GET"
    url: str = ""
    headers: dict[str, str] = {}
    body_template: str = ""  # templated, then parsed as JSON if non-empty

    # api_soap (app/crud/soap_client.py) — soap_mode "wsdl": every declared
    # input is passed to the operation as an argument by name, on top of the
    # optional soap_args_template (JSON, for nested structures); "raw":
    # soap_envelope is posted to soap_endpoint as written, its {{placeholders}}
    # XML-escaped. A node saved without soap_mode is the first version of this
    # node and keeps its whole-result-as-text output. `headers` are HTTP
    # headers; soap_password (and any field) may hold {{secret.NAME}}.
    wsdl_url: str = ""
    operation: str = ""
    soap_mode: SoapMode | None = None
    soap_endpoint: str = ""  # wsdl: overrides the WSDL's address; raw: where to post
    soap_service: str = ""  # wsdl: which service/port, when the WSDL has several
    soap_port: str = ""
    soap_args_template: str = ""
    soap_envelope: str = ""  # raw
    soap_action: str = ""  # raw
    soap_version: SoapVersion = "1.1"  # raw
    soap_header_xml: str = ""  # one or more SOAP header elements, wsdl mode
    soap_auth: SoapAuth = "none"
    soap_username: str = ""
    soap_password: str = ""
    soap_timeout_seconds: int = Field(default=30, ge=1, le=300)
    soap_verify_tls: bool = True
    soap_fail_on_fault: bool = True

    # google_sheets / google_docs / google_slides / google_drive /
    # google_calendar / google_gmail / google_forms — via a Connector (type
    # "google"; app/crud/google_apps.py); firebase_push / firebase_topic — via
    # a Connector (type "firebase"; app/crud/firebase_push.py).
    # service_operation names what to do, service_config holds that operation's
    # settings by name — templates, each rendered against the run. A setting
    # that is nothing but one {{placeholder}} keeps the value's real type, so a
    # list of rows from a Data — List node can be passed whole.
    service_operation: str = ""
    service_config: dict[str, Any] = {}

    # api_mqtt — publish only (a synchronous /run has no persistent listener
    # for subscribe); payload_template is a template rendered against every
    # declared input
    broker_host: str = ""
    broker_port: int = 1883
    topic: str = ""
    payload_template: str = ""

    # custom_code — runs in a separate OS process (subprocess.run, not
    # in-process eval) with a hard timeout and a stripped environment; NOT a
    # true sandbox (no network/filesystem jail) — see app/crud/tasks.py's
    # module docstring. Only the resolved `inputs` dict (not the whole
    # shared context) is piped to the script's stdin as JSON.
    language: TaskCodeLanguage | None = None
    code: str = ""

    # navigate — once the whole run has finished successfully, the caller
    # that started it (a Form or Page event, an embedded form, …) opens
    # nav_target_type / nav_target_id, with every declared input added as a
    # query parameter under its own name — e.g. an input "applicationNo"
    # read from a Generate Sequence output arrives as ?applicationNo=…,
    # which the target Page reads as {{applicationNo}} and a Form's prefill
    # reads as a URL parameter. nav_target_type "url" opens `url` (a
    # template, rendered like any other node's) in a new tab instead, the
    # inputs appended the same way. The walk carries on past this node, and
    # the last Navigate node reached wins. Nothing opens if the run fails,
    # or when the task runs on a schedule or at sign-in — there's no
    # browser to send anywhere.
    nav_target_type: TaskNavigateTarget = "page"
    nav_target_id: uuid.UUID | None = None


# Where a task variable's starting value comes from: "value" — text, where
# {{name}} placeholders are filled from the run (a fixed value, or e.g.
# "{{recordId}}-{{session.office_id}}"); "input" — a value the caller passes
# (a page parameter, a form field, recordId…), by name; "session" — a
# session value (Users › Session data), worked out on the server; "user" —
# the signed-in user's id / name / email / vendor_id / tenant_id.
TaskVariableSource = Literal["value", "input", "session", "user"]
TASK_VARIABLE_USER_FIELDS = ("id", "name", "email", "vendor_id", "tenant_id")
# Names the run itself uses.
_RESERVED_VARIABLES = {"secret", "session", "current_user", "currentUserId", "item", "index"}


class TaskVariableDef(BaseModel):
    """One task variable — set before the first node runs, in order (a later
    one can use an earlier one), and readable by every node by its name: as
    an input's source field, in a rule, or as {{name}} in a template. A Set
    variable node can change it later in the run."""

    name: str = Field(min_length=1, max_length=60, pattern=r"^[A-Za-z_][A-Za-z0-9_]*$")
    type: ParamType = "string"
    source: TaskVariableSource = "value"
    # value: the text/template; input: the caller's value name; session: the
    # session key; user: one of TASK_VARIABLE_USER_FIELDS.
    value: str = Field(default="", max_length=2000)
    description: str | None = Field(default=None, max_length=300)

    @model_validator(mode="after")
    def _check(self) -> "TaskVariableDef":
        if self.name in _RESERVED_VARIABLES:
            raise ValueError(f"'{self.name}' is used by every run — choose another variable name")
        if self.source in ("input", "session") and not self.value.strip():
            raise ValueError(f"variable '{self.name}': name the {'value the caller passes' if self.source == 'input' else 'session value'}")
        if self.source == "user" and self.value not in TASK_VARIABLE_USER_FIELDS:
            raise ValueError(f"variable '{self.name}': the signed-in user's {', '.join(TASK_VARIABLE_USER_FIELDS)}")
        return self


def _unique_variables(variables: list[TaskVariableDef] | None) -> list[TaskVariableDef] | None:
    if variables:
        names = [v.name for v in variables]
        dupes = sorted({n for n in names if names.count(n) > 1})
        if dupes:
            raise ValueError(f"variable names must be unique: {', '.join(dupes)}")
    return variables


class TaskCreate(BaseModel):
    name: str
    nodes: list[TaskNodeDef] = []
    variables: list[TaskVariableDef] = []

    @model_validator(mode="after")
    def _check_variables(self) -> "TaskCreate":
        _unique_variables(self.variables)
        return self


class TaskUpdate(BaseModel):
    name: str | None = None
    nodes: list[TaskNodeDef] | None = None
    variables: list[TaskVariableDef] | None = None

    @model_validator(mode="after")
    def _check_variables(self) -> "TaskUpdate":
        _unique_variables(self.variables)
        return self


class TaskDuplicate(BaseModel):
    # Blank/omitted -> "<source name> (copy)".
    name: str | None = Field(default=None, max_length=200)


class TaskRead(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: uuid.UUID
    tenant_id: uuid.UUID
    name: str
    nodes: list[dict]
    variables: list[dict] = []
    status: PublishStatus
    created_by: uuid.UUID | None
    created_at: datetime
    updated_at: datetime


class TaskRunSource(BaseModel):
    kind: Literal["page", "form", "portal", "app"] = "app"
    id: str | None = Field(default=None, max_length=64)


class TaskRunRequest(BaseModel):
    context: dict = {}
    # Set by the Task Designer's own "Test run" button — lets a draft run
    # for real without publishing first. Any other caller (e.g. a Form's
    # OnSubmit event) omits this, so an unpublished task still can't fire
    # from a real trigger.
    allow_draft: bool = False
    # Where the browser asked from, for the run's history. Only a label: it
    # grants nothing, and who ran it comes from the session.
    source: TaskRunSource | None = None


class TaskNodeRunResult(BaseModel):
    node_key: str
    node_type: TaskNodeType
    ok: bool
    detail: str
    # Inside a loop: which iteration, e.g. "Each invoice 2/10" (nested loops
    # joined with " › "). None outside loops.
    iteration: str | None = None


class TaskNavigation(BaseModel):
    """Where a Navigate node asked the caller to go (see TaskNodeDef)."""

    target_type: TaskNavigateTarget
    target_id: uuid.UUID | None = None  # every target_type except "url"
    url: str | None = None  # "url" only, already rendered
    params: dict[str, str] = {}  # query parameters, by name


class TaskRunResult(BaseModel):
    ok: bool  # false if execution stopped early because a node failed
    trace: list[TaskNodeRunResult]
    context: dict  # final accumulated context, for debugging in the designer
    # Set when the run reached a Navigate node and finished successfully.
    navigation: TaskNavigation | None = None
    # Set variable nodes' values to keep: session_data — the caller's
    # browser merges these into the session ({{session.<name>}}); params —
    # the page/form that ran the task takes them as its own values.
    session_data: dict = {}
    params: dict = {}
    # This run in the task's history (GET /tasks/{id}/runs/{run_id}).
    run_id: uuid.UUID | None = None


class TaskRunSummary(BaseModel):
    """One run in a task's history list."""

    model_config = ConfigDict(from_attributes=True)

    id: uuid.UUID
    task_id: uuid.UUID
    task_name: str
    task_status: str
    task_updated_at: datetime | None = None
    trigger: str
    trigger_ref: str | None = None
    trigger_name: str | None = None
    started_by: uuid.UUID | None = None
    started_by_name: str | None = None
    status: str
    detail: str | None = None
    step_count: int
    failed_steps: int
    lost_steps: int
    started_at: datetime
    finished_at: datetime | None = None


class TaskRunPage(BaseModel):
    items: list[TaskRunSummary]
    has_more: bool


class TaskRunDetail(TaskRunSummary):
    input: object | None = None
    output: object | None = None


class TaskRunLevelItem(BaseModel):
    """A row of a run's timeline: a step (values fetched on opening it), or
    one loop iteration, summarised — `path` asks for what's inside it."""

    kind: Literal["step", "iteration"]
    seq: int
    started_at: datetime
    # a step
    node_key: str | None = None
    node_label: str | None = None
    node_type: str | None = None
    iteration: str | None = None
    ok: bool | None = None
    detail: str | None = None
    truncated: bool = False
    duration_ms: int | None = None
    # an iteration
    label: str | None = None
    path: str | None = None
    last_seq: int | None = None
    steps: int | None = None
    failed: int | None = None


class TaskRunLevelPage(BaseModel):
    items: list[TaskRunLevelItem]
    has_more: bool


class TaskRunStepRead(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    seq: int
    node_key: str
    node_label: str | None = None
    node_type: str
    iteration: str | None = None
    ok: bool
    detail: str
    inputs: object | None = None
    outputs: object | None = None
    saved: object | None = None
    truncated: bool
    started_at: datetime
    duration_ms: int


class TaskHistorySettings(BaseModel):
    retention_days: int = Field(ge=1, le=365)
    max_runs_per_task: int | None = None  # read-only: the cap is fixed


class SoapInspectRequest(BaseModel):
    """What it takes to read a WSDL — from the SOAP node's own settings."""

    wsdl_url: str = Field(min_length=1, max_length=2000)
    soap_auth: SoapAuth = "none"
    soap_username: str = ""
    soap_password: str = ""
    headers: dict[str, str] = {}
    soap_timeout_seconds: int = Field(default=30, ge=1, le=300)
    soap_verify_tls: bool = True


class SoapOperationInput(BaseModel):
    name: str
    type: str = ""
    required: bool = True


class SoapOperationInfo(BaseModel):
    name: str
    service: str = ""
    port: str = ""
    address: str = ""
    signature: str = ""
    inputs: list[SoapOperationInput] = []


class SoapInspectResult(BaseModel):
    operations: list[SoapOperationInfo]


class CodeRunRequest(BaseModel):
    """A Custom Node's script run from the Task Designer's code editor —
    the same run a task does, against test values. `context` stands in for
    the running task's variables: the node's input_params are resolved from
    it exactly as in a real run, and its output_params are checked against
    the result."""

    language: TaskCodeLanguage = "python"
    code: str = Field(default="", max_length=200_000)
    context: dict = {}
    input_params: list[NodeInputParam] = []
    output_params: list[NodeOutputParam] = []


class CodeProblem(BaseModel):
    severity: Literal["error", "warning", "info"]
    message: str
    fix: str
    line: int | None = None
    column: int | None = None


class CodeRunResult(BaseModel):
    ok: bool  # ran (or would run) with no errors
    ran: bool  # false when stopped before running (syntax error, missing input)
    detail: str
    inputs: dict = {}  # what the script was given on stdin
    result: object | None = None
    outputs: dict | None = None  # what the node's outputs would be set to
    logs: list[str] = []
    stderr: str = ""
    exit_code: int | None = None
    duration_ms: int = 0
    problems: list[CodeProblem] = []