"""Executes a Task automation's node graph (app/models/automation.py::Task).

POST /tasks/{id}/run starts at the "start" node and walks next_nodes — the
first transition whose condition evaluates true against the running shared
context (or an unconditional one) — until a node has no matching transition
or a node reports failure. Two nodes steer the walk instead of doing work
(see the engine at the bottom): an If / Else ("condition") node follows its
"true" or "false" exit, and a Loop node runs its "body" path once per item
(for each), N times (repeat) or while its rules hold, then continues along
"done". Step guards catch accidental cycles. A Navigate node does no work
either: it records where the browser that started the run should go once
it finishes (run_task_with_navigation).

Every node's data contract is generic and runtime-enforced (app/crud/
params.py): before a node runs, its declared input_params are resolved out
of the shared context and type-checked into a local `inputs` dict — that's
how parameters actually pass from one node to the next, since an upstream
node's output_params write into the very same context keys a downstream
node's input_params read from. After a node runs, its declared
output_params are extracted from its result and type-checked back into
context. A node type still hardcodes which *named* inputs it looks for
(e.g. email_send wants one called "to") — see each _exec_* function's
docstring — declaring generic params doesn't mean a node can be configured
with arbitrary unrelated data, just that the values it does use are named,
typed, and validated instead of raw "read this context key" fields.

Deliberately not sharing dispatch_step_notifications with
app/crud/notifications.py: that helper is shaped around a workflow step's
notification config (connector + workflow/step/instance), not a node's
flatter config/context. The create_notification node instead calls that
module's separate create_task_notification — a plain "write one row" helper
with no workflow-shaped assumptions, so it's fine for both to use it.

custom_code nodes run via subprocess.run (a separate OS process, a hard
timeout, and a stripped environment) — NOT a real sandbox. It still has
whatever filesystem/network access the backend's own OS user has; the
isolation here is "not in-process" and "can't hang the request forever",
nothing stronger. Only the resolved `inputs` dict (not the whole shared
context) is piped to the script, so a script only ever sees what its node
declared.
"""

import asyncio
import email
import imaplib
import io
import json
import math
import mimetypes
import os
import re
import smtplib
import time
import uuid
from datetime import datetime, timezone
from email.mime.text import MIMEText
from pathlib import Path
from typing import Any, Awaitable, Callable
from xml.sax.saxutils import escape as xml_escape

import httpx
import openpyxl
import paho.mqtt.client as mqtt
import pypdf
from reportlab.lib.pagesizes import letter
from reportlab.pdfgen import canvas
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.crud import firebase_push, google_apps, record_sql, soap_client
from app.crud.code_runner import run_code
from app.crud.connectors import get_connector as _get_connector
from app.crud.secrets import MASK, SECRET_NODE_TYPES, Scrubber, connector_config, load_values, referenced_names
from app.crud.notifications import create_task_notification
from app.crud.params import _coerce_type, apply_output_params, get_path, render_value, resolve_input_params
from app.crud.task_history import RunHistory, RunTrigger, too_short_to_scrub
from app.crud.task_logic import condition_passes, describe, rules_pass
from app.crud.payments import PaymentGatewayError, initiate_payment
from app.crud.file_transfer import FileTransferError, run_operation as run_ftp_operation
from app.crud.google_auth import GoogleError
from app.crud.sql_providers import SqlError, run_sql
from app.crud.rpa_providers import FAILED, STOPPED, Job, RpaError, adapter_for as rpa_adapter_for, run_process, wait_for_job
from app.crud.sequence import next_sequence_value
from app.models.automation import Task
from app.models.identity import User
from app.models.master import Master, MasterRecord
from app.models.upload import UploadedFile
from app.models.workflow import Form, FormSubmission

UPLOAD_ROOT = Path(__file__).resolve().parent.parent.parent / "uploads"

NodeResult = tuple[bool, str, Any]
NodeHandler = Callable[[AsyncSession, uuid.UUID, dict, dict], Awaitable[NodeResult]]


_PLACEHOLDER = re.compile(r"\{\{\s*([\w.\[\]-]+)\s*\}\}")


def _template_value(value: Any) -> str:
    if isinstance(value, (dict, list)):
        return json.dumps(value, default=str)
    return str(value)


def _render(template: str, inputs: dict) -> str:
    """{{name}} = a declared input; any other {{variable}} or {{a.b.0.c}}
    path is looked up in the whole running context (see NodeInputs). An
    unknown placeholder is left as written."""
    text = template
    for key, value in inputs.items():
        text = text.replace("{{" + key + "}}", _template_value(value))
    context = getattr(inputs, "context", None)
    if not context or "{{" not in text:
        return text

    def lookup(m: re.Match) -> str:
        value = get_path(context, m.group(1), default=_UNSET)
        return m.group(0) if value is _UNSET else _template_value("" if value is None else value)

    return _PLACEHOLDER.sub(lookup, text)


_UNSET = object()


def _wrap_email_html_style(body: str, node: dict) -> str:
    """email_send, body_format "html" only — wraps the rendered body in one
    inline-styled <div> built from email_font_family/email_font_size/
    email_font_color/email_background_color. Inline, not a <style> block or
    CSS class, since most email clients strip both of those."""
    declarations = []
    if node.get("email_font_family"):
        declarations.append(f"font-family:{node['email_font_family']}")
    if node.get("email_font_size"):
        declarations.append(f"font-size:{node['email_font_size']}px")
    if node.get("email_font_color"):
        declarations.append(f"color:{node['email_font_color']}")
    if node.get("email_background_color"):
        declarations.append(f"background-color:{node['email_background_color']}")
    if not declarations:
        return body
    return f'<div style="{";".join(declarations)}">{body}</div>'


def _write_file_sync(path: Path, data: bytes) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_bytes(data)


async def _get_uploaded_file(db: AsyncSession, tenant_id: uuid.UUID, file_id: object) -> UploadedFile | None:
    try:
        fid = uuid.UUID(str(file_id))
    except ValueError:
        return None
    return (
        await db.execute(select(UploadedFile).where(UploadedFile.tenant_id == tenant_id, UploadedFile.id == fid))
    ).scalar_one_or_none()


# ---- start ------------------------------------------------------------------------------------


async def _exec_start(db: AsyncSession, tenant_id: uuid.UUID, node: dict, inputs: dict) -> NodeResult:
    return True, "started", None


async def _exec_set_variable(db: AsyncSession, tenant_id: uuid.UUID, node: dict, inputs: dict) -> NodeResult:
    """Each declared input is a variable to set, to its resolved value — the
    run's context is updated in _run_node, so every later node sees it."""
    values = dict(inputs)
    if not values:
        return True, "nothing to set — every value was empty in this run", {}
    return True, "set " + ", ".join(f"{k} = {_short(v)}" for k, v in values.items()), values


# A Set variable row's name says where its value goes: "session.<key>" — the
# signed-in session (an existing session value or a new one), readable as
# {{session.<key>}} by later nodes and, once the run succeeds, on every page,
# form and task; "param.<name>" — a context parameter handed back to the
# page/form that ran the task ({{name}} there, and to later nodes here); any
# other name — a task variable. A row with no prefix follows the node's own
# variable_scope (how the first version of this node saved its target).
_SESSION_PREFIX, _PARAM_PREFIX = "session.", "param."


def _apply_set_variable(run: "_Run", node: dict, values: dict, context: dict) -> None:
    scope = node.get("variable_scope") or "task"
    for name, value in values.items():
        if name.startswith(_SESSION_PREFIX) and len(name) > len(_SESSION_PREFIX):
            key = name[len(_SESSION_PREFIX):]
            context[name] = value
            run.session_data[key] = value
        elif name.startswith(_PARAM_PREFIX) and len(name) > len(_PARAM_PREFIX):
            key = name[len(_PARAM_PREFIX):]
            context[key] = value
            run.params[key] = value
        else:
            context[name] = value
            if scope == "session":
                context[f"{_SESSION_PREFIX}{name}"] = value
                run.session_data[name] = value
            elif scope == "context":
                run.params[name] = value


def _resolve_in_order(params: list[dict], context: dict) -> tuple[bool, str, dict]:
    """A Set variable node's rows, top to bottom — each can use the ones
    above it, under the name later nodes will read it by."""
    resolved: dict = {}
    scratch = dict(context)
    for p in params:
        ok, detail, one = resolve_input_params([p], scratch)
        if not ok:
            return False, detail, {}
        resolved.update(one)
        for name, value in one.items():
            scratch[name] = value
            if name.startswith(_PARAM_PREFIX):
                scratch[name[len(_PARAM_PREFIX):]] = value
    return True, "", resolved


def _short(value: Any) -> str:
    text = repr(value) if isinstance(value, str) else json.dumps(value, default=str)
    return text if len(text) <= 60 else text[:57] + "…"


# ---- email ------------------------------------------------------------------------------------


def _send_email_sync(config: dict, to_addr: str, subject: str, body: str, body_format: str = "text") -> None:
    host = config.get("host")
    from_addr = config.get("from") or config.get("username")
    if not host or not from_addr:
        raise ValueError("connector is missing host/from configuration")
    port = int(config.get("port") or 587)
    username, password = config.get("username"), config.get("password")

    msg = MIMEText(body, "html" if body_format == "html" else "plain", "utf-8")
    msg["Subject"] = subject
    msg["From"] = from_addr
    msg["To"] = to_addr

    with smtplib.SMTP(host, port, timeout=10) as smtp:
        smtp.starttls()
        if username and password:
            smtp.login(username, password)
        smtp.sendmail(from_addr, [to_addr], msg.as_string())


async def _exec_email_send(db: AsyncSession, tenant_id: uuid.UUID, node: dict, inputs: dict) -> NodeResult:
    """Looks for a required string input named "to"."""
    connector = await _get_connector(db, tenant_id, node.get("connector_id"), "email_smtp")
    if connector is None:
        return False, "connector not found, disabled, or not an email_smtp connector", None
    recipient = inputs.get("to")
    if not recipient:
        return False, "no recipient — declare a required input named 'to'", None
    subject = _render(node.get("subject") or "", inputs)
    body = _render(node.get("body") or "", inputs)
    body_format = node.get("body_format") or "text"
    if body_format == "html":
        body = _wrap_email_html_style(body, node)
    try:
        await asyncio.to_thread(_send_email_sync, await connector_config(db, tenant_id, connector), str(recipient), subject, body, body_format)
        return True, f"emailed to {recipient}", None
    except (smtplib.SMTPException, OSError, ValueError) as exc:
        return False, f"email failed: {exc}", None


async def _exec_sms_send(db: AsyncSession, tenant_id: uuid.UUID, node: dict, inputs: dict) -> NodeResult:
    """Looks for a required string input named "to". Generic REST gateway —
    the connector's configuration.api_url is POSTed a JSON body of
    {to, message, sender_id}; most SMS providers' HTTP APIs are close enough
    to this shape to work as-is, or the connector can point at a thin relay
    in front of the real provider."""
    connector = await _get_connector(db, tenant_id, node.get("connector_id"), "sms")
    if connector is None:
        return False, "connector not found, disabled, or not an sms connector", None
    recipient = inputs.get("to")
    if not recipient:
        return False, "no recipient — declare a required input named 'to'", None
    try:
        status_code = await send_sms(await connector_config(db, tenant_id, connector), str(recipient), _render(node.get("body") or "", inputs))
    except ValueError as exc:
        return False, str(exc), None
    except httpx.HTTPError as exc:
        return False, f"SMS send failed: {exc}", None
    return status_code < 400, f"HTTP {status_code}", {"status": status_code}


async def send_sms(config: dict, to: str, message: str) -> int:
    """POSTs {to, message, sender_id} to an "sms" Connector's
    configuration.api_url (Bearer api_key if set) and returns the HTTP
    status. Shared with account verification (app/crud/user_verification.py).
    Raises ValueError when the connector isn't configured, httpx.HTTPError
    when the gateway can't be reached."""
    api_url = config.get("api_url")
    if not api_url:
        raise ValueError("connector is missing configuration.api_url")
    payload = {"to": to, "message": message, "sender_id": config.get("sender_id")}
    headers = {"Authorization": f"Bearer {config['api_key']}"} if config.get("api_key") else {}
    async with httpx.AsyncClient(timeout=10.0) as client:
        resp = await client.post(api_url, json=payload, headers=headers)
    return resp.status_code


def _imap_fetch_sync(config: dict, folder: str, criteria: str, limit: int) -> list[dict]:
    host = config.get("host")
    if not host:
        raise ValueError("connector is missing host configuration")
    port = int(config.get("port") or 993)
    username, password = config.get("username"), config.get("password")
    if not username or not password:
        raise ValueError("connector is missing username/password configuration")

    conn = imaplib.IMAP4_SSL(host, port) if config.get("use_ssl", True) else imaplib.IMAP4(host, port)
    try:
        conn.login(username, password)
        conn.select(folder or "INBOX")
        status, data = conn.search(None, criteria or "ALL")
        if status != "OK":
            raise ValueError(f"IMAP search failed: {status}")
        ids = data[0].split()
        if limit:
            ids = ids[-limit:]
        messages: list[dict] = []
        for msg_id in ids:
            status, msg_data = conn.fetch(msg_id, "(RFC822)")
            if status != "OK" or not msg_data or msg_data[0] is None:
                continue
            msg = email.message_from_bytes(msg_data[0][1])
            body = ""
            if msg.is_multipart():
                for part in msg.walk():
                    if part.get_content_type() == "text/plain":
                        payload = part.get_payload(decode=True)
                        body = payload.decode(part.get_content_charset() or "utf-8", errors="replace") if payload else ""
                        break
            else:
                payload = msg.get_payload(decode=True)
                body = payload.decode(msg.get_content_charset() or "utf-8", errors="replace") if payload else ""
            messages.append({"subject": msg.get("Subject", ""), "from_addr": msg.get("From", ""), "date": msg.get("Date", ""), "body": body[:2000]})
        return messages
    finally:
        try:
            conn.logout()
        except (imaplib.IMAP4.error, OSError):
            pass


async def _exec_email_read(db: AsyncSession, tenant_id: uuid.UUID, node: dict, inputs: dict) -> NodeResult:
    """No inputs required — declare one "list"-typed output to capture the
    fetched messages."""
    connector = await _get_connector(db, tenant_id, node.get("connector_id"), "email_imap")
    if connector is None:
        return False, "connector not found, disabled, or not an email_imap connector", None
    try:
        messages = await asyncio.to_thread(
            _imap_fetch_sync,
            await connector_config(db, tenant_id, connector),
            node.get("folder") or "INBOX",
            node.get("search_criteria") or "ALL",
            int(node.get("limit") or 10),
        )
        return True, f"fetched {len(messages)} message(s)", messages
    except (imaplib.IMAP4.error, OSError, ValueError) as exc:
        return False, f"IMAP fetch failed: {exc}", None


# ---- data (Master or Form CRUD) ------------------------------------------------------------------

# Both MasterRecord and FormSubmission expose the same `field_values` JSONB
# column, so data_get/data_list/data_delete's matching/filtering logic below
# is identical regardless of which one a node points at — only the query
# (and, for data_delete, hard- vs. soft-delete) differs.


def _match_field_value(record: object, field: str) -> object:
    """"id" is a reserved match_field value meaning "the record's own
    primary key" rather than a field_values entry — 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. Every other value looks up
    field_values as before."""
    if field == "id":
        return str(record.id)
    return record.field_values.get(field)


def _match_clause(model: Any, field: str, value: Any) -> Any:
    """SQL that finds the candidates for str(_match_field_value(record,
    field)) == str(value) without reading the whole datasource — the Python
    check still makes the final call on what it returns."""
    wanted = str(value)
    if field == "id":
        try:
            return model.id == uuid.UUID(wanted)
        except ValueError:
            return model.id.is_(None)
    return record_sql.str_equals_clause(model.field_values, field, wanted).sql


def _node_model(node: dict) -> Any:
    return FormSubmission if node.get("datasource_type") == "form" else MasterRecord


def _data_candidates_stmt(tenant_id: uuid.UUID, node: dict):
    """The SELECT to run for data_get/data_list/data_delete, or None if the
    node isn't configured yet — datasource_type picks Master vs. Form,
    defaulting to "master" so every already-persisted node (from before
    Forms were supported here) keeps working unchanged."""
    if node.get("datasource_type") == "form":
        form_id = node.get("form_id")
        if not form_id:
            return None
        return (
            select(FormSubmission)
            .where(
                FormSubmission.tenant_id == tenant_id,
                FormSubmission.form_id == uuid.UUID(str(form_id)),
                FormSubmission.is_deleted.is_(False),
                FormSubmission.status == "submitted",
            )
            .order_by(FormSubmission.submitted_at, FormSubmission.id)
        )
    master_id = node.get("master_id")
    if not master_id:
        return None
    # Oldest first — a data_list loops over records in the order they were added.
    return (
        select(MasterRecord)
        .where(MasterRecord.tenant_id == tenant_id, MasterRecord.master_id == uuid.UUID(str(master_id)), MasterRecord.is_deleted.is_(False))
        .order_by(MasterRecord.created_at, MasterRecord.id)
    )


# A data node's result is the record's own field values plus its id (a real
# column, not a field) — so a task can output it, e.g. the id of the payment
# row it just created, for a later step to update that same row.
def _record_result(record: Any) -> dict:
    return {**record.field_values, "id": str(record.id)}


async def _exec_data_get(db: AsyncSession, tenant_id: uuid.UUID, node: dict, inputs: dict) -> NodeResult:
    """Looks for a required string input named "match_value"."""
    match_field = node.get("match_field")
    stmt = _data_candidates_stmt(tenant_id, node)
    if stmt is None or not match_field:
        return False, "datasource or match field not configured", None
    match_value = inputs.get("match_value")
    if match_value is None:
        return False, "declare a required input named 'match_value'", None
    stmt = stmt.where(_match_clause(_node_model(node), match_field, match_value))
    candidates = (await db.execute(stmt)).scalars().all()
    record = next((r for r in candidates if str(_match_field_value(r, match_field)) == str(match_value)), None)
    if record is None:
        return True, "no matching record found", None
    return True, f"found record {record.id}", _record_result(record)


async def _exec_data_list(db: AsyncSession, tenant_id: uuid.UUID, node: dict, inputs: dict) -> NodeResult:
    """Looks for an optional string input named "filter_value" — omit it to
    list every record. Declare a "list"-typed output to capture the rows."""
    stmt = _data_candidates_stmt(tenant_id, node)
    if stmt is None:
        return False, "no datasource configured", None
    filter_field, filter_value = node.get("filter_field"), inputs.get("filter_value")
    if filter_field and filter_value is not None:
        model = _node_model(node)
        stmt = stmt.where(record_sql.str_equals_clause(model.field_values, filter_field, str(filter_value)).sql)
    records = list((await db.execute(stmt.limit(record_sql.MAX_PYTHON_ROWS + 1))).scalars().all())
    if len(records) > record_sql.MAX_PYTHON_ROWS:
        return False, f"more than {record_sql.MAX_PYTHON_ROWS:,} records match — set a filter on this step", None
    if filter_field and filter_value is not None:
        records = [r for r in records if str(r.field_values.get(filter_field)) == str(filter_value)]
    rows = [_record_result(r) for r in records]
    return True, f"{len(rows)} record(s)", rows


# Who a data write is recorded against (created_by / updated_by): the user the
# run is for — "currentUserId", which POST /tasks/{id}/run always sets to the
# real caller. A scheduled run carries whatever its job saved, so the id must
# be a user of this tenant; anything else records no one rather than failing.
async def _acting_user(db: AsyncSession, tenant_id: uuid.UUID, inputs: dict) -> uuid.UUID | None:
    raw = (getattr(inputs, "context", None) or {}).get("currentUserId")
    try:
        user_id = uuid.UUID(str(raw)) if raw else None
    except ValueError:
        return None
    if user_id is None:
        return None
    found = await db.execute(select(User.id).where(User.tenant_id == tenant_id, User.id == user_id))
    return user_id if found.scalar_one_or_none() else None


# Task values often arrive as text — "10" shared by a page Label, a query
# parameter, a template — but a datasource or form stored as a table only
# takes a number in a number field and a yes/no in a yes/no field
# (app/crud/entity_storage.py). So text that clearly is one is written as
# one; anything else is left as it is, for the usual checks to report.
_NUMBER_TYPES = {"number", "numeric", "real", "double_precision", "currency", "percentage", "slider", "rating"}
_WHOLE_TYPES = {"smallint", "integer", "bigint"}
_YES_NO_TYPES = {"boolean", "checkbox", "toggle", "consent"}
_YES, _NO = {"true", "yes", "y", "1", "on"}, {"false", "no", "n", "0", "off"}


def _typed(field_type: str | None, value: Any) -> Any:
    if not isinstance(value, str) or not field_type:
        return value
    text = value.strip()
    if field_type in _NUMBER_TYPES or field_type in _WHOLE_TYPES:
        if text == "":
            return None
        try:
            number = float(text)
        except ValueError:
            return value
        if not math.isfinite(number):
            return value
        if number.is_integer():
            return int(number)
        return value if field_type in _WHOLE_TYPES else number
    if field_type in _YES_NO_TYPES:
        if text.lower() in _YES:
            return True
        if text.lower() in _NO:
            return False
    return value


def _typed_values(fields: list[dict] | None, values: dict) -> dict:
    types = {f.get("name"): f.get("type") for f in fields or [] if isinstance(f, dict)}
    return {k: _typed(types.get(k), v) for k, v in values.items()}


async def _flush_write(db: AsyncSession, tenant_id: uuid.UUID, *new: Any) -> str | None:
    """Saves a Data — Update write in its own savepoint. A value the stored
    table refuses comes back as words for the run's trace (e.g. "payment_amount
    must be a valid number.") instead of failing the whole request; the rest
    of the run's work is kept."""
    from sqlalchemy.exc import DBAPIError

    from app.core.jobs import tenant_session
    from app.crud.entity_storage import explain_violation

    try:
        async with db.begin_nested():
            db.add_all(new)
            await db.flush()
    except DBAPIError as exc:
        violation = await explain_violation(lambda: tenant_session(tenant_id), exc)
        if violation is None:
            raise
        return violation.message
    return None


async def _exec_data_update(db: AsyncSession, tenant_id: uuid.UUID, node: dict, inputs: dict) -> NodeResult:
    """Every declared input EXCEPT one named "match_value" is written onto
    the record's field_values under its own name. "match_value" (if
    declared) finds an existing record to update instead of always
    creating a new one. A new Form submission is always created against
    that Form's live (latest published) version. The running user is recorded
    as the row's created_by (new rows) and updated_by, like a save from the UI."""
    match_value = inputs.get("match_value")
    values = {k: v for k, v in inputs.items() if k != "match_value"}
    if not values:
        # Optional inputs whose variable wasn't set are skipped (params.py), so
        # "nothing to set" usually means they were all empty in this run — say
        # which, rather than implying none were declared.
        empty = [
            f"'{p.get('name')}' ({p['value'] if p.get('value') is not None else p.get('source_field')})"
            for p in node.get("input_params") or []
            if p.get("name") and p.get("name") != "match_value" and (p.get("source_field") or p.get("value") is not None)
        ]
        if not empty:
            return False, "no fields to set — declare at least one input other than 'match_value'", None
        detail = f"no fields to set — every input was empty in this run: {', '.join(empty)}"
        if any(str(p.get("source_field") or "").startswith("payment_") for p in node.get("input_params") or []):
            detail += ". payment_* values are filled only when the run has a payment_reference"
        return False, detail, None
    match_field = node.get("match_field")

    if node.get("datasource_type") == "form":
        form_id = node.get("form_id")
        if not form_id:
            return False, "no datasource configured", None
        from app.crud.form_versions import live_form

        # A new submission is made against the live (latest published)
        # version, not a draft being edited in the Form Designer.
        form = await live_form(db, tenant_id, uuid.UUID(str(form_id)))
        if form is None:
            return False, "target form not found", None

        submission = None
        if match_field and match_value is not None:
            stmt = select(FormSubmission).where(
                FormSubmission.tenant_id == tenant_id,
                FormSubmission.form_id == form.form_id,
                FormSubmission.is_deleted.is_(False),
                FormSubmission.status == "submitted",
                _match_clause(FormSubmission, match_field, match_value),
            )
            candidates = (await db.execute(stmt)).scalars().all()
            submission = next((r for r in candidates if str(_match_field_value(r, match_field)) == str(match_value)), None)

        user_id = await _acting_user(db, tenant_id, inputs)
        values = _typed_values(form.field_definitions, values)
        if submission is not None:
            submission.field_values = {**submission.field_values, **values}
            if user_id:
                submission.updated_by = user_id
            problem = await _flush_write(db, tenant_id)
            if problem:
                return False, problem, None
            return True, f"updated submission {submission.id}", _record_result(submission)

        submission = FormSubmission(
            tenant_id=tenant_id, form_id=form.form_id, form_version=form.version, field_values=values, submitted_by=user_id, updated_by=user_id
        )
        problem = await _flush_write(db, tenant_id, submission)
        if problem:
            return False, problem, None
        return True, f"created submission {submission.id}", _record_result(submission)

    master_id = node.get("master_id")
    if not master_id:
        return False, "no datasource configured", None
    master = (
        await db.execute(select(Master).where(Master.tenant_id == tenant_id, Master.id == uuid.UUID(str(master_id))))
    ).scalar_one_or_none()
    if master is None:
        return False, "target datasource not found", None

    # A deleted record no longer counts as a match, so the write creates a new
    # one, as for a value nothing matches. The deleted row stays as it was.
    record = None
    if match_field and match_value is not None:
        stmt = select(MasterRecord).where(
            MasterRecord.tenant_id == tenant_id,
            MasterRecord.master_id == master.id,
            MasterRecord.is_deleted.is_(False),
            _match_clause(MasterRecord, match_field, match_value),
        )
        candidates = (await db.execute(stmt)).scalars().all()
        record = next((r for r in candidates if str(_match_field_value(r, match_field)) == str(match_value)), None)

    user_id = await _acting_user(db, tenant_id, inputs)
    values = _typed_values(master.field_schema, values)
    from app.crud import field_constraints

    if record is not None:
        merged = field_constraints.keep_autonumbers(master, {**record.field_values, **values}, record.field_values)
        clash = await field_constraints.unique_errors(db, tenant_id, master, merged, record_id=record.id)
        if clash:
            return False, "; ".join(clash), None
        record.field_values = merged
        if user_id:
            record.updated_by = user_id
        problem = await _flush_write(db, tenant_id)
        if problem:
            return False, problem, None
        return True, f"updated record {record.id}", _record_result(record)

    values = await field_constraints.prepare_create(db, tenant_id, user_id, master, values)
    clash = await field_constraints.unique_errors(db, tenant_id, master, values)
    if clash:
        return False, "; ".join(clash), None
    values = await field_constraints.finish_create(db, tenant_id, master, values)
    record = MasterRecord(tenant_id=tenant_id, master_id=master.id, field_values=values, created_by=user_id, updated_by=user_id)
    problem = await _flush_write(db, tenant_id, record)
    if problem:
        return False, problem, None
    return True, f"created record {record.id}", _record_result(record)


async def _exec_data_delete(db: AsyncSession, tenant_id: uuid.UUID, node: dict, inputs: dict) -> NodeResult:
    """Looks for a required string input named "match_value". A matching
    Form submission is soft-deleted (is_deleted — same column the Form
    Submissions list's own delete path uses); a matching Master record is
    removed outright, unchanged from before Forms were supported here."""
    match_field = node.get("match_field")
    stmt = _data_candidates_stmt(tenant_id, node)
    if stmt is None or not match_field:
        return False, "datasource or match field not configured", None
    match_value = inputs.get("match_value")
    if match_value is None:
        return False, "declare a required input named 'match_value'", None
    stmt = stmt.where(_match_clause(_node_model(node), match_field, match_value))
    candidates = (await db.execute(stmt)).scalars().all()
    matches = [r for r in candidates if str(_match_field_value(r, match_field)) == str(match_value)]
    if node.get("datasource_type") == "form":
        for r in matches:
            r.is_deleted = True
    else:
        for r in matches:
            await db.delete(r)
    await db.flush()
    return True, f"deleted {len(matches)} record(s)", len(matches)


# ---- files (Excel / PDF) ------------------------------------------------------------------------


def _excel_read_sync(path: Path, sheet_name: str | None) -> list[dict]:
    wb = openpyxl.load_workbook(path, data_only=True)
    ws = wb[sheet_name] if sheet_name and sheet_name in wb.sheetnames else wb.active
    rows = list(ws.iter_rows(values_only=True))
    if not rows:
        return []
    header = [str(h) if h is not None else f"col_{i}" for i, h in enumerate(rows[0])]
    return [dict(zip(header, row)) for row in rows[1:]]


async def _exec_excel_read(db: AsyncSession, tenant_id: uuid.UUID, node: dict, inputs: dict) -> NodeResult:
    """Looks for a required string input named "file_id" (an UploadedFile
    id). Declare a "list"-typed output to capture the parsed rows."""
    file_id = inputs.get("file_id")
    if not file_id:
        return False, "declare a required input named 'file_id'", None
    file = await _get_uploaded_file(db, tenant_id, file_id)
    if file is None:
        return False, "referenced file not found", None
    try:
        rows = await asyncio.to_thread(_excel_read_sync, Path(file.storage_path), node.get("sheet_name"))
        return True, f"{len(rows)} row(s)", rows
    except (OSError, ValueError, KeyError) as exc:
        return False, f"excel read failed: {exc}", None


def _pdf_read_sync(path: Path) -> str:
    reader = pypdf.PdfReader(str(path))
    return "\n".join((page.extract_text() or "") for page in reader.pages)


async def _exec_pdf_read(db: AsyncSession, tenant_id: uuid.UUID, node: dict, inputs: dict) -> NodeResult:
    """Looks for a required string input named "file_id" (an UploadedFile id)."""
    file_id = inputs.get("file_id")
    if not file_id:
        return False, "declare a required input named 'file_id'", None
    file = await _get_uploaded_file(db, tenant_id, file_id)
    if file is None:
        return False, "referenced file not found", None
    try:
        text = await asyncio.to_thread(_pdf_read_sync, Path(file.storage_path))
        return True, f"extracted {len(text)} char(s)", text
    except (OSError, ValueError) as exc:
        return False, f"pdf read failed: {exc}", None


def _pdf_generate_sync(text: str) -> bytes:
    buffer = io.BytesIO()
    c = canvas.Canvas(buffer, pagesize=letter)
    width, height = letter
    y = height - 50
    for line in text.splitlines() or [""]:
        if y < 50:
            c.showPage()
            y = height - 50
        c.drawString(50, y, line[:110])
        y -= 16
    c.save()
    return buffer.getvalue()


async def _exec_pdf_generate(db: AsyncSession, tenant_id: uuid.UUID, node: dict, inputs: dict) -> NodeResult:
    rendered = _render(node.get("template") or "", inputs)
    data = await asyncio.to_thread(_pdf_generate_sync, rendered)
    file_id = uuid.uuid4()
    storage_path = UPLOAD_ROOT / str(tenant_id) / f"{file_id}_generated.pdf"
    await asyncio.to_thread(_write_file_sync, storage_path, data)
    db.add(
        UploadedFile(
            id=file_id,
            tenant_id=tenant_id,
            filename="generated.pdf",
            content_type="application/pdf",
            size_bytes=len(data),
            storage_path=str(storage_path),
        )
    )
    await db.flush()
    return True, f"generated file {file_id}", str(file_id)


# ---- APIs (REST / SOAP / MQTT) -------------------------------------------------------------------


async def _exec_api_rest(db: AsyncSession, tenant_id: uuid.UUID, node: dict, inputs: dict) -> NodeResult:
    url = _render(node.get("url") or "", inputs)
    if not url:
        return False, "no URL configured", None
    method = node.get("method") or "GET"
    headers = {k: _render(v, inputs) for k, v in (node.get("headers") or {}).items()}
    body_template = (node.get("body_template") or "").strip()
    json_body = None
    if body_template:
        try:
            json_body = json.loads(_render(body_template, inputs))
        except json.JSONDecodeError:
            return False, "body_template did not render to valid JSON", None
    try:
        async with httpx.AsyncClient(timeout=15.0) as client:
            resp = await client.request(method, url, headers=headers, json=json_body)
        try:
            parsed_body: object = resp.json()
        except ValueError:
            parsed_body = resp.text
        return resp.status_code < 400, f"HTTP {resp.status_code}", {"status": resp.status_code, "body": parsed_body}
    except httpx.HTTPError as exc:
        return False, f"request failed: {exc}", None


def _render_xml(template: str, inputs: dict) -> str:
    """_render for an XML template: filled-in values are escaped, so a value
    from a form can't add elements to the envelope."""
    context = getattr(inputs, "context", None) or {}

    def lookup(m: re.Match) -> str:
        name = m.group(1)
        value = inputs[name] if name in inputs else get_path(context, name, default=_UNSET)
        return m.group(0) if value is _UNSET else xml_escape(_template_value("" if value is None else value), {'"': "&quot;", "'": "&apos;"})

    return _PLACEHOLDER.sub(lookup, template)


def soap_spec(node: dict, inputs: dict) -> soap_client.SoapSpec:
    return soap_client.SoapSpec(
        wsdl_url=_render(node.get("wsdl_url") or "", inputs).strip(),
        operation=_render(node.get("operation") or "", inputs).strip(),
        endpoint=_render(node.get("soap_endpoint") or "", inputs).strip(),
        service=node.get("soap_service") or "",
        port=node.get("soap_port") or "",
        auth=node.get("soap_auth") or "none",
        username=_render(node.get("soap_username") or "", inputs),
        password=_render(node.get("soap_password") or "", inputs),
        headers={k: _render(str(v), inputs) for k, v in (node.get("headers") or {}).items()},
        header_xml=_render_xml(node.get("soap_header_xml") or "", inputs),
        timeout=int(node.get("soap_timeout_seconds") or 30),
        verify_tls=bool(node.get("soap_verify_tls", True)),
        envelope=_render_xml(node.get("soap_envelope") or "", inputs),
        soap_action=_render(node.get("soap_action") or "", inputs),
        version=node.get("soap_version") or "1.1",
    )


def _soap_arguments(node: dict, inputs: dict) -> dict:
    """The operation's arguments: the JSON template (for nested structures)
    with each declared input on top, by name."""
    args: dict = {}
    template = (node.get("soap_args_template") or "").strip()
    if template:
        try:
            parsed = json.loads(_render(template, inputs))
        except json.JSONDecodeError as exc:
            raise soap_client.SoapError("the arguments template did not render to valid JSON") from exc
        if not isinstance(parsed, dict):
            raise soap_client.SoapError("the arguments template must be a JSON object")
        args.update(parsed)
    args.update(inputs)
    return args


async def _exec_api_soap(db: AsyncSession, tenant_id: uuid.UUID, node: dict, inputs: dict) -> NodeResult:
    """Every declared input is passed to the operation as an argument, by name
    (mode "wsdl"), or fills {{name}} in the envelope (mode "raw"). Result:
    {result, fault: null, …the response's own top-level fields} — or, for an
    operation returning one plain value, that value. A SOAP fault fails the
    node unless soap_fail_on_fault is off (then branch on "fault")."""
    mode = node.get("soap_mode")
    legacy = mode is None  # saved before modes existed: its outputs read a text result
    try:
        spec = soap_spec(node, inputs)
        if mode == "raw":
            answer = await asyncio.to_thread(soap_client.call_raw, spec)
        else:
            if not spec.wsdl_url or not spec.operation:
                return False, "wsdl_url and operation are required", None
            answer = await asyncio.to_thread(soap_client.call_wsdl, spec, _soap_arguments(node, inputs))
    except soap_client.SoapError as exc:
        return False, str(exc), None
    except Exception as exc:  # noqa: BLE001 - zeep/requests/lxml raise a wide variety of their own exception types
        return False, f"SOAP call failed: {str(exc).strip()[:500] or type(exc).__name__}", None

    fault = answer.get("fault")
    if fault:
        detail = f"SOAP fault {fault['code']}: {fault['message']}".strip()[:600]
        result = {"fault": fault, "fault_code": fault["code"], "fault_message": fault["message"], "result": None, "xml": answer.get("xml")}
        return not node.get("soap_fail_on_fault", True), detail, result

    value = answer.get("value")
    if legacy:
        return True, "call succeeded", str(value)
    detail = f"HTTP {answer['status']}" if answer.get("status") else "call succeeded"
    if not isinstance(value, dict) and mode != "raw":
        return True, detail, value
    fields = value if isinstance(value, dict) else {}
    return True, detail, {**fields, "result": value, "fault": None, "xml": answer.get("xml")}


# ---- Google Workspace / Firebase ------------------------------------------------------------------


def _setting_value(template: Any, inputs: dict) -> Any:
    """A service node's setting: a value that is only one {{placeholder}} keeps
    its real type (a list of rows stays a list); any other text is rendered."""
    if not isinstance(template, str):
        return template
    whole = _PLACEHOLDER.fullmatch(template.strip())
    if whole:
        name = whole.group(1)
        value = inputs[name] if name in inputs else get_path(getattr(inputs, "context", None) or {}, name, default=_UNSET)
        if value is not _UNSET:
            return value
    return _render(template, inputs)


def _service_settings(node: dict, inputs: dict) -> dict:
    return {k: _setting_value(v, inputs) for k, v in (node.get("service_config") or {}).items()}


async def _save_file(db: AsyncSession, tenant_id: uuid.UUID, name: str, content_type: str, data: bytes) -> str:
    """Stores bytes as an UploadedFile, for Excel/PDF Reader, Email or File
    Transfer nodes further along. The id is what comes back."""
    file_id = uuid.uuid4()
    safe = "".join(c if c.isalnum() or c in "._-" else "_" for c in name)[:120] or "file"
    storage_path = UPLOAD_ROOT / str(tenant_id) / f"{file_id}_{safe}"
    await asyncio.to_thread(_write_file_sync, storage_path, data)
    db.add(UploadedFile(id=file_id, tenant_id=tenant_id, filename=name[:255], content_type=content_type, size_bytes=len(data), storage_path=str(storage_path)))
    await db.flush()
    return str(file_id)


async def _exec_google(db: AsyncSession, tenant_id: uuid.UUID, node: dict, inputs: dict) -> NodeResult:
    """Sheets, Docs, Slides, Drive, Calendar, Gmail and Forms, through a
    "google" Connector — see app/crud/google_apps.py for each operation. The
    settings are templates; on Docs/Slides each declared input also replaces
    {{its name}} inside the document."""
    connector = await _get_connector(db, tenant_id, node.get("connector_id"), "google")
    if connector is None:
        return False, "connector not found, disabled, or not a Google connector", None

    async def load(file_id: str) -> tuple[str, str, bytes] | None:
        file = await _get_uploaded_file(db, tenant_id, file_id)
        if file is None:
            return None
        return file.filename, file.content_type or "application/octet-stream", await asyncio.to_thread(Path(file.storage_path).read_bytes)

    async def save(name: str, content_type: str, data: bytes) -> str:
        return await _save_file(db, tenant_id, name, content_type, data)

    call = google_apps.Call(
        config=await connector_config(db, tenant_id, connector),
        settings=_service_settings(node, inputs),
        inputs=dict(inputs),
        load_file=load,
        save_file=save,
    )
    try:
        detail, result = await google_apps.run(node["type"], node.get("service_operation") or "", call)
    except GoogleError as exc:
        return False, str(exc), None
    except (httpx.HTTPError, OSError) as exc:
        return False, f"couldn't reach Google: {exc}", None
    return True, detail, result


async def _exec_firebase_push(db: AsyncSession, tenant_id: uuid.UUID, node: dict, inputs: dict) -> NodeResult:
    """Web, Android and iOS through FCM; native Windows through WNS — see
    app/crud/firebase_push.py. Result: {sent, failed, invalid_tokens,
    expired_channels, message_id, …}."""
    connector = await _get_connector(db, tenant_id, node.get("connector_id"), "firebase")
    if connector is None:
        return False, "connector not found, disabled, or not a Firebase connector", None
    try:
        return await firebase_push.send(await connector_config(db, tenant_id, connector), _service_settings(node, inputs))
    except GoogleError as exc:
        return False, str(exc), None
    except httpx.HTTPError as exc:
        return False, f"couldn't reach the push service: {exc}", None


async def _exec_firebase_topic(db: AsyncSession, tenant_id: uuid.UUID, node: dict, inputs: dict) -> NodeResult:
    """Subscribes devices to a topic, unsubscribes them, or lists one's topics."""
    connector = await _get_connector(db, tenant_id, node.get("connector_id"), "firebase")
    if connector is None:
        return False, "connector not found, disabled, or not a Firebase connector", None
    settings = {**_service_settings(node, inputs), "operation": node.get("service_operation") or "subscribe"}
    try:
        return await firebase_push.manage_topic(await connector_config(db, tenant_id, connector), settings)
    except GoogleError as exc:
        return False, str(exc), None
    except httpx.HTTPError as exc:
        return False, f"couldn't reach Firebase: {exc}", None


def _mqtt_publish_sync(host: str, port: int, topic: str, payload: str) -> None:
    client = mqtt.Client(callback_api_version=mqtt.CallbackAPIVersion.VERSION2)
    client.connect(host, port, keepalive=10)
    client.loop_start()
    try:
        info = client.publish(topic, payload, qos=1)
        info.wait_for_publish(timeout=5)
    finally:
        client.loop_stop()
        client.disconnect()


async def _exec_api_mqtt(db: AsyncSession, tenant_id: uuid.UUID, node: dict, inputs: dict) -> NodeResult:
    host, topic = node.get("broker_host") or "", node.get("topic") or ""
    if not host:
        return False, "broker_host is required", None
    if not topic:
        return False, "topic is required", None
    payload = _render(node.get("payload_template") or "", inputs)
    try:
        await asyncio.to_thread(_mqtt_publish_sync, host, int(node.get("broker_port") or 1883), topic, payload)
        return True, f"published to {topic}", payload
    except (OSError, ValueError) as exc:
        return False, f"publish failed: {exc}", None


# ---- custom code ----------------------------------------------------------------------------------


def _run_custom_code_sync(language: str, code: str, inputs: dict) -> NodeResult:
    # Shared with the Task Designer's code editor (POST /tasks/code/run), so a
    # script behaves the same in a test run as in the task.
    run = run_code(language, code, inputs)
    return run.ok, run.detail, run.result

async def _exec_custom_code(db: AsyncSession, tenant_id: uuid.UUID, node: dict, inputs: dict) -> NodeResult:
    code = node.get("code") or ""
    if not code.strip():
        return False, "no code configured", None
    return await asyncio.to_thread(_run_custom_code_sync, node.get("language") or "python", code, inputs)


async def _exec_create_notification(db: AsyncSession, tenant_id: uuid.UUID, node: dict, inputs: dict) -> NodeResult:
    """No Connector — writes directly to the in-app notification list.
    Reuses `subject`/`body` (both rendered as templates) as the
    notification's title/message. Recipients are notify_role_ids +
    notify_user_ids + an optional "recipient_user_id" input (one id or a
    list); none of the three = the tenant-wide internal broadcast."""
    title = _render(node.get("subject") or "", inputs) or "Notification"
    message = _render(node.get("body") or "", inputs)
    user_ids = list(node.get("notify_user_ids") or [])
    dynamic = inputs.get("recipient_user_id")
    if isinstance(dynamic, list):
        user_ids.extend(dynamic)
    elif dynamic:
        user_ids.append(dynamic)
    role_ids = node.get("notify_role_ids") or []
    targeted = bool(role_ids) or bool(user_ids) or "recipient_user_id" in inputs
    if not targeted:
        await create_task_notification(db, tenant_id, title, message)
        return True, "notification created (all internal users)", None
    if not role_ids and not user_ids:
        # Declared a recipient_user_id input that came through empty — a
        # targeted notification with nobody to target, not a broadcast.
        return True, "no matching active recipients — nothing sent", None
    count = await create_task_notification(db, tenant_id, title, message, role_ids=role_ids, user_ids=user_ids)
    if count == 0:
        return True, "no matching active recipients — nothing sent", None
    return True, f"notification sent to {count} user{'s' if count != 1 else ''}", None


def _sequence_effective_key(base_key: str, reset_period: str) -> str:
    if reset_period == "yearly":
        return f"{base_key}:{datetime.now(timezone.utc).year}"
    if reset_period == "monthly":
        now = datetime.now(timezone.utc)
        return f"{base_key}:{now.year}-{now.month:02d}"
    return base_key


async def _exec_generate_sequence(db: AsyncSession, tenant_id: uuid.UUID, node: dict, inputs: dict) -> NodeResult:
    """Declare ONE output param to capture the formatted result, e.g.
    "RTI-2026-0007". No inputs required."""
    base_key = (node.get("sequence_key") or "").strip()
    if not base_key:
        return False, "no sequence_key configured", None
    effective_key = _sequence_effective_key(base_key, node.get("sequence_reset_period") or "never")
    value = await next_sequence_value(db, tenant_id, effective_key)
    padding = int(node.get("sequence_padding") or 1)
    formatted = f"{node.get('sequence_prefix') or ''}{str(value).zfill(padding)}"
    return True, f"generated {formatted}", formatted


async def _exec_payment_initiate(db: AsyncSession, tenant_id: uuid.UUID, node: dict, inputs: dict) -> NodeResult:
    """Looks for a required int input named "amount" (minor currency units)
    and optional string inputs "currency"/"reference". Delegates the actual
    gateway call to app/crud/payments.py::initiate_payment — the same
    generic Connector contract the Page Designer "Payment Initiator" block
    (app/routers/payments.py) calls directly, without a Task in between."""
    amount = inputs.get("amount")
    if amount is None:
        return False, "declare a required int input named 'amount' (minor currency units)", None
    try:
        result = await initiate_payment(
            db,
            tenant_id,
            node.get("connector_id"),
            amount,
            currency=inputs.get("currency") or "INR",
            reference=inputs.get("reference") or "",
            checkout_path=node.get("checkout_path") or "/checkout",
        )
    except PaymentGatewayError as exc:
        return False, str(exc), None
    return True, "checkout session created", result


# ---- RPA (UiPath / Automation Anywhere / Power Automate / Blue Prism / REST) ----------------------

_RPA_MAX_TIMEOUT = 600


def _rpa_result(node: dict, job: Job, started: float) -> NodeResult:
    """{job_id, status, provider_status, info} plus every output argument
    by name. A failed/stopped job fails the node unless rpa_fail_on_error is
    off, so a graph can branch on "status" instead."""
    result = {**job.outputs, "job_id": job.job_id, "status": job.status, "provider_status": job.provider_status or "", "info": job.info or ""}
    took = f"{time.monotonic() - started:.0f}s"
    detail = f"job {job.job_id} {job.status}" + (f" ({job.provider_status})" if job.provider_status and job.provider_status.lower() != job.status else "")
    if job.info:
        detail += f": {job.info[:300]}"
    if not job.done and node.get("rpa_wait", True):
        detail += f" — still {job.status} after {took}; check it later with an RPA — Job Status node"
    elif job.done:
        detail += f" after {took}"
    if job.status in (FAILED, STOPPED) and node.get("rpa_fail_on_error", True):
        return False, detail, result
    return True, detail, result


def _rpa_timeout(node: dict) -> float:
    return float(max(5, min(int(node.get("rpa_timeout_seconds") or 120), _RPA_MAX_TIMEOUT)))


async def _exec_rpa_run(db: AsyncSession, tenant_id: uuid.UUID, node: dict, inputs: dict) -> NodeResult:
    """Every declared input is passed to the process/bot as an input
    argument by name (for add_queue_item: the item's specific content, with
    an input named "reference" as its reference)."""
    connector = await _get_connector(db, tenant_id, node.get("connector_id"), "rpa")
    if connector is None:
        return False, "connector not found, disabled, or not an RPA connector", None
    started = time.monotonic()
    try:
        job = await run_process(
            await connector_config(db, tenant_id, connector),
            action=node.get("rpa_action") or "start_job",
            process=(node.get("rpa_process") or "").strip(),
            inputs=inputs,
            queue=(node.get("rpa_queue") or "").strip(),
            wait=bool(node.get("rpa_wait", True)),
            timeout=_rpa_timeout(node),
        )
    except RpaError as exc:
        return False, str(exc), None
    except (httpx.HTTPError, OSError, asyncio.TimeoutError) as exc:
        return False, f"couldn't reach the RPA platform: {exc}", None
    return _rpa_result(node, job, started)


async def _exec_rpa_job_status(db: AsyncSession, tenant_id: uuid.UUID, node: dict, inputs: dict) -> NodeResult:
    """Looks for a required string input named "job_id"."""
    connector = await _get_connector(db, tenant_id, node.get("connector_id"), "rpa")
    if connector is None:
        return False, "connector not found, disabled, or not an RPA connector", None
    job_id = inputs.get("job_id")
    if not job_id:
        return False, "declare a required string input named 'job_id'", None
    config = await connector_config(db, tenant_id, connector)
    started = time.monotonic()
    try:
        job = await rpa_adapter_for(config).job_status(config, str(job_id))
        if node.get("rpa_wait", True):
            job = await wait_for_job(config, job, _rpa_timeout(node))
    except RpaError as exc:
        return False, str(exc), None
    except (httpx.HTTPError, OSError, asyncio.TimeoutError) as exc:
        return False, f"couldn't reach the RPA platform: {exc}", None
    return _rpa_result(node, job, started)


# ---- SQL databases ------------------------------------------------------------------------------


async def _exec_sql(db: AsyncSession, tenant_id: uuid.UUID, node: dict, inputs: dict, *, write: bool) -> NodeResult:
    connector = await _get_connector(db, tenant_id, node.get("connector_id"), "database")
    if connector is None:
        return False, "connector not found, disabled, or not a database connector", None
    try:
        result = await run_sql(
            await connector_config(db, tenant_id, connector),
            node.get("sql") or "",
            inputs,
            write=write,
            max_rows=int(node.get("sql_max_rows") or 1000),
            timeout=int(node.get("sql_timeout_seconds") or 30),
        )
    except SqlError as exc:
        return False, str(exc), None
    except (asyncio.TimeoutError, TimeoutError):
        return False, "the database didn't answer in time — raise the node's timeout or speed up the query", None
    except Exception as exc:  # noqa: BLE001 - every driver raises its own error types; the message is what the author needs
        return False, f"database error: {str(exc).strip()[:400]}", None
    first = result.rows[0] if result.rows else {}
    if write:
        count = result.affected_rows if result.affected_rows is not None else len(result.rows)
        return True, f"{count} row(s) affected", {**first, "affected_rows": count, "rows": result.rows}
    detail = f"{len(result.rows)} row(s)" + (f" — stopped at the {len(result.rows)}-row limit" if result.truncated else "")
    return True, detail, {**first, "rows": result.rows, "row_count": len(result.rows), "truncated": "yes" if result.truncated else "no"}


async def _exec_sql_query(db: AsyncSession, tenant_id: uuid.UUID, node: dict, inputs: dict) -> NodeResult:
    """Every :name in the SQL is bound from the declared input of that name.
    Read-only. Declare a "list" output named "rows" for every row, or
    outputs named after columns to take the first row's values."""
    return await _exec_sql(db, tenant_id, node, inputs, write=False)


async def _exec_sql_execute(db: AsyncSession, tenant_id: uuid.UUID, node: dict, inputs: dict) -> NodeResult:
    """INSERT / UPDATE / DELETE / stored procedures, on a connector that
    allows writes. Output "affected_rows"; RETURNING/OUTPUT rows come back
    like a query's."""
    return await _exec_sql(db, tenant_id, node, inputs, write=True)


# ---- File transfer (SFTP / FTP / FTPS) ----------------------------------------------------------


async def _exec_ftp_transfer(db: AsyncSession, tenant_id: uuid.UUID, node: dict, inputs: dict) -> NodeResult:
    """remote_path / ftp_target_path are templates rendered against every
    declared input. "upload" looks for a required string input named
    "file_id" (an uploaded file)."""
    connector = await _get_connector(db, tenant_id, node.get("connector_id"), "file_transfer")
    if connector is None:
        return False, "connector not found, disabled, or not a file transfer connector", None
    operation = node.get("ftp_operation") or "list"
    remote_path = _render(node.get("remote_path") or "", inputs)
    target = _render(node.get("ftp_target_path") or "", inputs)
    upload = None
    if operation == "upload":
        file = await _get_uploaded_file(db, tenant_id, inputs.get("file_id")) if inputs.get("file_id") else None
        if file is None:
            return False, "declare a required string input named 'file_id' pointing at an uploaded or generated file", None
        try:
            upload = (file.filename, await asyncio.to_thread(Path(file.storage_path).read_bytes))
        except OSError as exc:
            return False, f"couldn't read {file.filename}: {exc}", None
    try:
        result = await run_ftp_operation(await connector_config(db, tenant_id, connector), operation, remote_path, pattern=node.get("ftp_pattern") or "", target=target, upload=upload)
    except FileTransferError as exc:
        return False, str(exc), None
    except (asyncio.TimeoutError, TimeoutError):
        return False, "the file server didn't answer in time", None
    except Exception as exc:  # noqa: BLE001 - paramiko/ftplib/socket errors; the message is what the author needs
        return False, f"file transfer failed: {str(exc).strip()[:400] or type(exc).__name__}", None

    if operation == "download":
        downloads = result["downloads"]
        if not downloads:
            return True, f"no files to download in {result['folder']}", {"file_ids": [], "count": 0}
        ids: list[str] = []
        for name, remote, data in downloads:
            file_id = uuid.uuid4()
            safe = "".join(c if c.isalnum() or c in "._-" else "_" for c in name)[:120] or "download"
            storage_path = UPLOAD_ROOT / str(tenant_id) / f"{file_id}_{safe}"
            await asyncio.to_thread(_write_file_sync, storage_path, data)
            db.add(
                UploadedFile(
                    id=file_id,
                    tenant_id=tenant_id,
                    filename=name[:255],
                    content_type=mimetypes.guess_type(name)[0] or "application/octet-stream",
                    size_bytes=len(data),
                    storage_path=str(storage_path),
                )
            )
            ids.append(str(file_id))
        await db.flush()
        names = [d[0] for d in downloads]
        return True, f"downloaded {', '.join(names[:3])}{' …' if len(names) > 3 else ''}", {"file_id": ids[0], "file_ids": ids, "file_name": names[0], "file_names": names, "count": len(ids)}
    if operation == "list":
        return True, f"{result['count']} item(s) in {result['folder']}", result
    if operation == "upload":
        return True, f"uploaded to {result['remote_path']}", result
    if operation == "move":
        return True, f"moved {result['count']} file(s)", result
    return True, f"deleted {result['count']} file(s)", result


_HANDLERS: dict[str, NodeHandler] = {
    "start": _exec_start,
    "set_variable": _exec_set_variable,
    "email_read": _exec_email_read,
    "email_send": _exec_email_send,
    "sms_send": _exec_sms_send,
    "data_get": _exec_data_get,
    "data_list": _exec_data_list,
    "data_update": _exec_data_update,
    "data_delete": _exec_data_delete,
    "excel_read": _exec_excel_read,
    "pdf_read": _exec_pdf_read,
    "pdf_generate": _exec_pdf_generate,
    "api_rest": _exec_api_rest,
    "api_soap": _exec_api_soap,
    "api_mqtt": _exec_api_mqtt,
    "create_notification": _exec_create_notification,
    "generate_sequence": _exec_generate_sequence,
    "payment_initiate": _exec_payment_initiate,
    "rpa_run": _exec_rpa_run,
    "rpa_job_status": _exec_rpa_job_status,
    "sql_query": _exec_sql_query,
    "sql_execute": _exec_sql_execute,
    "ftp_transfer": _exec_ftp_transfer,
    "custom_code": _exec_custom_code,
    **{node_type: _exec_google for node_type in google_apps.NODE_TYPES},
    "firebase_push": _exec_firebase_push,
    "firebase_topic": _exec_firebase_topic,
}


# ---- Engine: walking the graph, If / Else, loops ------------------------------------------------

# A single path segment (the top level, or one loop iteration) visiting more
# than this many nodes is a cycle, not work; the whole run is capped too, so
# nested loops over big lists can't run away.
_MAX_SEGMENT_STEPS = 200
_MAX_TOTAL_STEPS = 20_000
_MAX_LOOP_ITERATIONS = 10_000
_MAX_TRACE = 1_000


class NodeInputs(dict):
    """A node's resolved, declared inputs — plus the whole running context
    (`.context`), so a template's {{placeholder}} can name any variable
    without declaring it. Handlers that pass inputs on as a set (SOAP call
    arguments, Data — Update fields, bot arguments, SQL parameters) still
    only ever see the declared ones."""

    context: dict

    def __init__(self, values: dict, context: dict):
        super().__init__(values)
        self.context = context


class _Run:
    def __init__(self, db: AsyncSession, tenant_id: uuid.UUID, nodes: list[dict], history: RunHistory):
        self.db, self.tenant_id = db, tenant_id
        self.history = history  # every step, with its values (app/crud/task_history.py)
        self.step_began = time.monotonic()
        self.by_key = {n["key"]: n for n in nodes}
        self.trace: list[dict] = []
        self.dropped = 0
        self.total_steps = 0
        self.frames: list[str] = []  # "Each invoice 2/10" per active loop
        self.navigation: dict | None = None  # the last Navigate node reached
        # Set variable nodes saving to the session / the caller (variable_scope).
        self.session_data: dict = {}
        self.params: dict = {}
        # The tenant's secrets (app/crud/secrets.py), loaded once per run:
        # handed only to the outbound nodes that reference them, and masked
        # out of everything the run produces.
        self.secrets: dict[str, str] = {}
        self.scrub = Scrubber([])

    def record(self, node: dict, ok: bool, detail: str, *, inputs: Any = None, outputs: Any = None, saved: Any = None, began: float | None = None, also_mask: list[str] | None = None) -> None:
        """Adds a step to the trace (capped, handed back to the caller) and to
        the run's history (every step, with `inputs`/`outputs`/`saved`, which
        the history masks and cuts down). `began` defaults to when the walk
        reached this step."""
        took = int((time.monotonic() - (self.step_began if began is None else began)) * 1000)
        self.history.add(node, ok, detail, " › ".join(self.frames) or None, took, inputs=inputs, outputs=outputs, saved=saved, also_mask=also_mask)
        if len(self.trace) >= _MAX_TRACE and ok:
            self.dropped += 1
            return
        self.trace.append({"node_key": node.get("key", "?"), "node_type": node.get("type", "?"), "ok": ok, "detail": self.scrub(detail), "iteration": " › ".join(self.frames) or None})


_SECRET_NODE_LABELS = "REST, SOAP, MQTT, SQL, File Transfer, RPA, Payment or Custom code"


def _secret_check(run: _Run, node: dict) -> tuple[str | None, set[str]]:
    """(error, names) — the secrets a node references, or why it can't."""
    names = referenced_names({k: v for k, v in node.items() if k not in ("label", "next_nodes")})
    if not names:
        return None, names
    if node.get("type") not in SECRET_NODE_TYPES:
        return (
            f"secrets can only be used in nodes that send them to another system ({_SECRET_NODE_LABELS}) — "
            f"here it could end up in a record, message or page people can read: {", ".join(sorted(names))}"
        ), names
    missing = sorted(names - run.secrets.keys())
    if missing:
        return f"unknown secret {", ".join(missing)} — add it under Settings › Secrets (names are case-sensitive)", names
    return None, names


def _branch_of(node: dict, index: int, transition: dict) -> str | None:
    """A transition's exit. Unlabelled ones on an If / Else or Loop (older
    or hand-written graphs) are read in order: If / Else — first true,
    second false; Loop — first body, the rest done."""
    if transition.get("branch"):
        return transition["branch"]
    if node.get("type") == "condition":
        return "true" if index == 0 else "false"
    if node.get("type") == "loop":
        return "body" if index == 0 else "done"
    return None


def _next_key(node: dict, context: dict, branch: str | None = None) -> str | None:
    """The first transition on `branch` whose own condition (if any) holds."""
    for i, t in enumerate(node.get("next_nodes") or []):
        if _branch_of(node, i, t) == branch and condition_passes(t.get("condition"), context):
            return t.get("to")
    return None


async def _walk(run: _Run, start: dict | None, context: dict, loop_key: str | None = None) -> bool:
    """Runs from `start` until the path ends. Inside a loop body, reaching
    the loop node again ends the iteration. False = a node failed."""
    current, steps = start, 0
    while current is not None:
        if loop_key and current.get("key") == loop_key:
            return True
        steps += 1
        run.total_steps += 1
        await run.history.checkpoint()
        run.step_began = time.monotonic()
        if steps > _MAX_SEGMENT_STEPS:
            run.record(current, False, f"stopped after {_MAX_SEGMENT_STEPS} steps without finishing (possible cycle — use a Loop node to repeat work)")
            return False
        if run.total_steps > _MAX_TOTAL_STEPS:
            run.record(current, False, f"stopped: the run passed {_MAX_TOTAL_STEPS} steps")
            return False

        node_type = current.get("type")
        if node_type in ("condition", "loop", "navigate"):
            error, _ = _secret_check(run, current)
            if error:
                run.record(current, False, error)
                return False
        if node_type == "condition":
            passed = rules_pass(current.get("rules") or [], current.get("rules_match") or "all", context)
            run.record(
                current, True, f"{'true' if passed else 'false'} — {describe(current.get('rules') or [], current.get('rules_match') or 'all')}",
                inputs=_rule_values(current, context), outputs={"result": passed},
            )
            next_key = _next_key(current, context, "true" if passed else "false")
        elif node_type == "loop":
            if not await _run_loop(run, current, context):
                return False
            next_key = _next_key(current, context, "done")
        elif node_type == "navigate":
            if not _run_navigate(run, current, context):
                return False
            next_key = _next_key(current, context)
        else:
            if not await _run_node(run, current, context):
                return False
            next_key = _next_key(current, context)
        current = run.by_key.get(next_key) if next_key else None
    return True


# What a step's history shows as sent: its templates, filled in. Rendered from
# the masked inputs and a masked secret.*, so this can't hold a secret's value.
_SENT_FIELDS = {
    "api_rest": ("method", "url", "headers", "body_template"),
    "email_send": ("subject", "body"),
    "sms_send": ("body",),
    "api_mqtt": ("topic", "payload_template"),
    "pdf_generate": ("template",),
    "create_notification": ("subject", "body"),
    "ftp_transfer": ("remote_path", "ftp_target_path"),
    "api_soap": ("wsdl_url", "soap_endpoint", "operation", "soap_action", "headers", "soap_args_template", "soap_envelope", "soap_header_xml"),
    **{node_type: ("service_operation", "service_config") for node_type in google_apps.NODE_TYPES},
    "firebase_push": ("service_config",),
    "firebase_topic": ("service_operation", "service_config"),
}


def _rule_values(node: dict, context: dict) -> dict:
    """The variables an If / Else or While node's rules read, as they stood."""
    return {r["driver_field"]: get_path(context, r["driver_field"]) for r in node.get("rules") or [] if r.get("driver_field")}


def _step_inputs(node: dict, resolved: dict, context: dict, secret_names: set[str]) -> dict:
    """A step's inputs for its history. One that read a secret is masked whole,
    whatever its value looks like — a short secret is otherwise not masked."""
    inputs = dict(resolved)
    if secret_names:
        for p in node.get("input_params") or []:
            if p.get("name") in inputs and referenced_names(p):
                inputs[p["name"]] = MASK
    fields = _SENT_FIELDS.get(node.get("type"))
    if fields:
        masked = NodeInputs(inputs, {**context, "secret": {n: MASK for n in secret_names}} if secret_names else context)
        sent = {}
        for name in fields:
            value = node.get(name)
            if isinstance(value, dict):
                value = {k: _render(str(v), masked) for k, v in value.items()} or None
            elif isinstance(value, str):
                value = _render(value, masked).strip() or None
            if value:
                sent[name] = value
        if sent:
            inputs = {**inputs, "$sent": sent}
    return inputs


async def _run_node(run: _Run, node: dict, context: dict) -> bool:
    handler = _HANDLERS.get(node.get("type"))
    if handler is None:
        run.record(node, False, "unknown node type")
        return False
    error, secret_names = _secret_check(run, node)
    if error:
        run.record(node, False, error)
        return False
    # Only this node sees the secrets it names — never the shared context.
    node_context = {**context, "secret": {n: run.secrets[n] for n in secret_names}} if secret_names else context
    if node.get("type") == "set_variable":
        in_ok, in_detail, resolved = _resolve_in_order(node.get("input_params") or [], node_context)
    else:
        in_ok, in_detail, resolved = resolve_input_params(node.get("input_params") or [], node_context)
    short = too_short_to_scrub([run.secrets[n] for n in secret_names])
    if not in_ok:
        run.record(node, False, in_detail, also_mask=short)
        return False
    inputs = _step_inputs(node, resolved, context, secret_names)
    try:
        ok, detail, result = await handler(run.db, run.tenant_id, node, NodeInputs(resolved, node_context))
    except Exception as exc:
        # It still fails the run as before, but the history shows where and on what.
        run.record(node, False, f"{type(exc).__name__}: {exc}", inputs=inputs, also_mask=short)
        raise
    # Whatever came back (an API echoing a header, a script printing a key)
    # is masked before it can reach the context — and so records, pages or
    # the designer.
    result, detail = run.scrub(result), run.scrub(detail)
    if ok and node.get("type") == "set_variable":
        _apply_set_variable(run, node, result or {}, context)
    saved = None
    if ok and node.get("output_params"):
        out_ok, out_detail, written = apply_output_params(node["output_params"], result)
        if not out_ok:
            run.record(node, False, f"{detail} — {out_detail}", inputs=inputs, outputs=result, also_mask=short)
            return False
        context.update(written)
        saved = written
    run.record(node, ok, detail, inputs=inputs, outputs=result, saved=saved, also_mask=short)
    return ok


_NAV_TARGET_LABELS = {"page": "page", "form": "form", "dashboard": "dashboard", "master": "datasource", "url": "URL"}


def _query_value(value: Any) -> str:
    if isinstance(value, bool):
        return "true" if value else "false"
    return _template_value(value)


def _run_navigate(run: _Run, node: dict, context: dict) -> bool:
    """Records where the caller should go once the run finishes; the
    browser-side caller does the actual navigating (TaskRunResult.navigation).
    Every declared input becomes a query parameter under its own name."""
    in_ok, in_detail, inputs = resolve_input_params(node.get("input_params") or [], context)
    if not in_ok:
        run.record(node, False, in_detail)
        return False
    target = node.get("nav_target_type") or "page"
    label = _NAV_TARGET_LABELS.get(target, target)
    params = {name: _query_value(value) for name, value in inputs.items() if value is not None}
    if target == "url":
        url = _render(node.get("url") or "", NodeInputs(inputs, context)).strip()
        if not url:
            run.record(node, False, "set the URL to open")
            return False
        run.navigation = {"target_type": "url", "target_id": None, "url": url, "params": params}
    else:
        if not node.get("nav_target_id"):
            run.record(node, False, f"pick the {label} to open")
            return False
        run.navigation = {"target_type": target, "target_id": str(node["nav_target_id"]), "url": None, "params": params}
    passing = f"with {', '.join(params)}" if params else "with no parameters"
    run.record(node, True, f"will open the {label} {passing} when the task finishes", inputs=inputs, outputs=run.navigation)
    return True


def _loop_plan(node: dict, inputs: dict, context: dict) -> tuple[str | None, list | None, int | None]:
    """(error, items, count) — for_each iterates items, repeat/while count
    iterations (while: an upper bound)."""
    limit = max(1, min(int(node.get("loop_max_iterations") or 1000), _MAX_LOOP_ITERATIONS))
    mode = node.get("loop_mode") or "for_each"
    if mode == "for_each":
        items = inputs.get("items")
        if items is None:
            return "declare a required list input named 'items' (e.g. a Data — List node's rows)", None, None
        if isinstance(items, dict):
            items = [items]
        if not isinstance(items, list):
            return f"'items' must be a list, got {type(items).__name__}", None, None
        if len(items) > limit:
            return f"{len(items)} items is over this loop's limit of {limit} — raise it or filter the list first", None, None
        return None, items, len(items)
    if mode == "repeat":
        count = inputs.get("count")
        if not isinstance(count, int) or isinstance(count, bool):
            return "declare a required int input named 'count'", None, None
        if count > limit:
            return f"{count} repetitions is over this loop's limit of {limit}", None, None
        return None, None, max(0, count)
    if not node.get("rules"):
        return "a While loop needs at least one rule", None, None
    return None, None, limit


async def _run_loop(run: _Run, node: dict, context: dict) -> bool:
    in_ok, in_detail, inputs = resolve_input_params(node.get("input_params") or [], context)
    if not in_ok:
        run.record(node, False, in_detail)
        return False
    error, items, count = _loop_plan(node, inputs, context)
    if error:
        run.record(node, False, error, inputs=inputs)
        return False
    began = run.step_began

    mode = node.get("loop_mode") or "for_each"
    item_var = (node.get("loop_item_var") or "item").strip() or "item"
    index_var = (node.get("loop_index_var") or "index").strip() or "index"
    collect = [c for c in (node.get("loop_collect") or []) if c]
    body_key = _next_key(node, context, "body")
    body = run.by_key.get(body_key) if body_key else None
    label = node.get("label") or node.get("key")
    rules, match = node.get("rules") or [], node.get("rules_match") or "all"
    run.record(
        node, True, {"for_each": f"looping over {count} item(s)", "repeat": f"repeating {count} time(s)", "while": f"repeating while {describe(rules, match)}"}[mode] + ("" if body else " — nothing is connected to its body"),
        inputs={**inputs, **_rule_values(node, context)},
    )

    saved = {k: context[k] for k in (item_var, index_var) if k in context}
    results: list = []
    iterations = failed = 0
    try:
        for i in range(count or 0):
            if mode == "while" and not rules_pass(rules, match, context):
                break
            context[index_var] = i
            if items is not None:
                context[item_var] = items[i]
            run.frames.append(f"{label} {i + 1}" + (f"/{count}" if mode != "while" else ""))
            try:
                ok = await _walk(run, body, context, loop_key=node["key"]) if body else True
            finally:
                run.frames.pop()
            iterations += 1
            if collect:
                results.append(get_path(context, collect[0]) if len(collect) == 1 else {c: get_path(context, c) for c in collect})
            if not ok:
                failed += 1
                if not node.get("loop_continue_on_error"):
                    run.record(node, False, f"stopped at iteration {i + 1} — a node inside the loop failed", began=began)
                    return False
        else:
            if mode == "while" and count and rules_pass(rules, match, context):
                run.record(node, False, f"still repeating after {count} iterations — the While rules never became false (raise the limit if that's expected)", began=began)
                return False
    finally:
        # The loop's own variables don't leak past it (an outer loop's are restored).
        for k in (item_var, index_var):
            context.pop(k, None)
        context.update(saved)

    result = {"results": results, "count": iterations, "failed": failed}
    saved = None
    if node.get("output_params"):
        out_ok, out_detail, written = apply_output_params(node["output_params"], result)
        if not out_ok:
            run.record(node, False, out_detail, outputs=result, began=began)
            return False
        context.update(written)
        saved = written
    run.record(node, True, f"loop finished — {iterations} iteration(s)" + (f", {failed} failed" if failed else ""), outputs=result, saved=saved, began=began)
    return True


async def _trusted_identity(db: AsyncSession, tenant_id: uuid.UUID, task: Task, context: dict) -> None:
    """current_user.* and session.* as the server knows them for the user the
    run is for ("currentUserId", which POST /tasks/{id}/run always sets to the
    real caller) — replacing whatever a caller claimed. Session values are
    the tenant's sign-in lookups (Users › Session data), worked out only when
    the task reads one; values the sign-in task added are left as passed."""
    raw = context.get("currentUserId")
    try:
        user_id = uuid.UUID(str(raw)) if raw else None
    except ValueError:
        user_id = None
    user = (await db.execute(select(User).where(User.tenant_id == tenant_id, User.id == user_id))).scalar_one_or_none() if user_id else None
    if user is None:
        return
    context.update(
        {
            "current_user.id": str(user.id),
            "current_user.name": user.name,
            "current_user.email": user.email,
            "current_user.vendor_id": str(user.vendor_id) if user.vendor_id else "",
            "current_user.tenant_id": str(user.tenant_id),
        }
    )
    wants_session = any(v.get("source") == "session" for v in task.variables or []) or "session." in json.dumps(task.nodes or [], default=str)
    if not wants_session:
        return
    from app.crud.login_session import build_login_session_data
    from app.models.identity import Tenant
    from app.schemas.identity import LoginSessionSettings

    tenant = await db.get(Tenant, tenant_id)
    try:
        # The configured lookups only — never the sign-in task, which may have side effects.
        settings = LoginSessionSettings.model_validate((tenant.login_session_settings if tenant else None) or {}).model_copy(update={"task_id": None})
        session = await build_login_session_data(db, tenant, user, settings) if tenant is not None else {}
    except ValueError:
        session = {}
    context.update({f"session.{k}": v for k, v in session.items()})


_USER_VARIABLE_KEYS = {
    "id": "current_user.id",
    "name": "current_user.name",
    "email": "current_user.email",
    "vendor_id": "current_user.vendor_id",
    "tenant_id": "current_user.tenant_id",
}


def _set_variables(run: _Run, start: dict, variables: list[dict], context: dict) -> bool:
    """The task's variables (Task Designer › Variables), set in order before
    the first node runs — each readable by every node under its own name.
    Recorded on the Start node's trace; a value that doesn't fit its type
    stops the run there."""
    if not variables:
        return True
    names = []
    for v in variables:
        name, source, spec = v.get("name"), v.get("source") or "value", v.get("value") or ""
        if not name:
            continue
        if source == "input":
            raw = get_path(context, spec)
        elif source == "session":
            raw = context.get(f"session.{spec.removeprefix('session.')}")
        elif source == "user":
            raw = context.get(_USER_VARIABLE_KEYS.get(spec, ""))
        else:
            raw = render_value(spec, context)
        ptype = v.get("type") or "string"
        if raw is None or raw == "":
            context[name] = "" if ptype == "string" else None
        else:
            ok, coerced = _coerce_type(raw, ptype)
            if not ok:
                run.record(start, False, f"variable '{name}' should be {ptype}, got {raw!r}")
                return False
            context[name] = coerced
        names.append(name)
    run.record(start, True, "variables: " + ", ".join(f"{n} = {_short(context[n])}" for n in names), outputs={n: context[n] for n in names})
    return True


async def run_task_with_navigation(
    db: AsyncSession, tenant_id: uuid.UUID, task: Task, initial_context: dict, trigger: RunTrigger | None = None
) -> tuple[bool, list[dict], dict, dict | None, dict]:
    """run_task plus where a Navigate node asked the caller to go — only when
    the whole run succeeded, so a failure never sends anyone onward — and
    what Set variable nodes saved for the caller: {"session_data", "params"}.
    Every run is kept in the task's history; `trigger` says what started it."""
    history = RunHistory(tenant_id, task, trigger, initial_context)
    try:
        return await _run_recorded(db, tenant_id, task, initial_context, history)
    except BaseException as exc:
        # Before the error reaches whatever rolls back: this is the run people will want to read.
        await asyncio.shield(history.finish("cancelled" if isinstance(exc, asyncio.CancelledError) else "error", f"{type(exc).__name__}: {exc}"))
        raise


async def _run_recorded(
    db: AsyncSession, tenant_id: uuid.UUID, task: Task, initial_context: dict, history: RunHistory
) -> tuple[bool, list[dict], dict, dict | None, dict]:
    nodes = task.nodes or []
    start = next((n for n in nodes if n.get("type") == "start"), None)
    context = dict(initial_context)
    if start is None:
        await history.finish("failed", "This task has no Start node.")
        return False, [], context, None, {"session_data": {}, "params": {}}
    run = _Run(db, tenant_id, nodes, history)
    run.secrets = await load_values(db, tenant_id)
    run.scrub = Scrubber(run.secrets)
    history.set_scrubber(run.scrub, context)
    await _trusted_identity(db, tenant_id, task, context)
    ok = _set_variables(run, start, task.variables or [], context) and await _walk(run, start, context)
    await history.finish("succeeded" if ok else "failed", output=run.scrub(context))
    if run.dropped:
        run.trace.append({"node_key": "…", "node_type": "start", "ok": True, "detail": f"{run.dropped} more successful step(s) not shown", "iteration": None})
    # Belt and braces: nothing the run hands back carries a secret value.
    kept = {"session_data": run.scrub(run.session_data), "params": run.scrub(run.params)} if ok else {"session_data": {}, "params": {}}
    return ok, run.scrub(run.trace), run.scrub(context), run.scrub(run.navigation) if ok else None, kept


async def run_task(db: AsyncSession, tenant_id: uuid.UUID, task: Task, initial_context: dict, trigger: RunTrigger | None = None) -> tuple[bool, list[dict], dict]:
    """For callers with no browser to navigate (schedules, sign-in)."""
    ok, trace, context, _, _ = await run_task_with_navigation(db, tenant_id, task, initial_context, trigger)
    return ok, trace, context