"""RPA platform adapters behind the one "rpa" Connector type.

Each platform knows its own protocol — auth, how a process/bot is named and
started, how its job status and output arguments are read — and maps it onto
one small contract (run_process/wait_for_job below drive it):

    list_processes(config)                         -> [{id, name, description}]
    start_job(config, process, inputs, timeout)    -> Job
    job_status(config, job_id)                     -> Job
    add_queue_item(config, queue, inputs, ref)     -> Job   ("queues" capability only)
    test_credentials(config)                       -> (ok, detail)

`Job.status` is always normalised to queued | running | submitted |
succeeded | failed | stopped so Tasks never need to know platform-specific
values ("submitted" = the platform accepted the run but can't report on it,
e.g. a Power Automate flow without a Response action). PROVIDERS also
describes each platform's settings for the Connectors screen, so adding a
platform is one adapter plus one metadata entry — no UI work.
"""

import asyncio
import json
import re
import time
import uuid
from dataclasses import dataclass, field
from typing import Any
from urllib.parse import quote, urlsplit

import httpx

TIMEOUT = 30.0
QUEUED, RUNNING, SUBMITTED, SUCCEEDED, FAILED, STOPPED = "queued", "running", "submitted", "succeeded", "failed", "stopped"
DONE = {SUBMITTED, SUCCEEDED, FAILED, STOPPED}
_sleep = asyncio.sleep  # swapped out in tests


class RpaError(Exception):
    pass


@dataclass
class Job:
    job_id: str
    status: str
    provider_status: str | None = None
    outputs: dict = field(default_factory=dict)
    info: str = ""

    @property
    def done(self) -> bool:
        return self.status in DONE


def _client(timeout: float = TIMEOUT) -> httpx.AsyncClient:
    """One place every adapter gets its HTTP client — tests swap in a
    MockTransport here."""
    return httpx.AsyncClient(timeout=timeout)


def _dig(data: Any, path: str) -> Any:
    """"value.0.Id" style lookup; None when absent."""
    for part in [p for p in (path or "").split(".") if p]:
        if isinstance(data, dict):
            data = data.get(part)
        elif isinstance(data, list) and part.isdigit() and int(part) < len(data):
            data = data[int(part)]
        else:
            return None
    return data


def _json(resp: httpx.Response) -> Any:
    try:
        return resp.json()
    except ValueError:
        return {"text": resp.text[:500]}


def _error_text(data: Any) -> str:
    if isinstance(data, dict):
        for path in ("message", "error_description", "error.message", "details", "errorMessage", "text", "error"):
            value = _dig(data, path)
            if value and isinstance(value, (str, int)):
                return str(value)
    return str(data)[:300]


def _odata_quote(value: str) -> str:
    return value.replace("'", "''")


# Short-lived access tokens, reused until a minute before they expire.
_TOKENS: dict[tuple, tuple[str, float]] = {}


def _cached(key: tuple) -> str | None:
    hit = _TOKENS.get(key)
    return hit[0] if hit and hit[1] > time.monotonic() + 60 else None


# ---- UiPath Orchestrator (Automation Cloud or on-prem / Automation Suite) ----------------------

_UIPATH_STATES = {
    "Pending": QUEUED, "Running": RUNNING, "Stopping": RUNNING, "Terminating": RUNNING, "Resumed": RUNNING, "Suspended": RUNNING,
    "Successful": SUCCEEDED, "Faulted": FAILED, "Stopped": STOPPED,
}
_UIPATH_QUEUE_STATES = {
    "New": QUEUED, "InProgress": RUNNING, "Retried": RUNNING,
    "Successful": SUCCEEDED, "Failed": FAILED, "Abandoned": FAILED, "Deleted": STOPPED,
}


class UiPath:
    capabilities = ["jobs", "queues", "status"]
    _folders: dict[tuple, int] = {}

    @staticmethod
    def _base(config: dict) -> str:
        base = (config.get("orchestrator_url") or "").strip().rstrip("/")
        if not base:
            raise RpaError("set the Orchestrator URL")
        return base

    @staticmethod
    def identity_url(config: dict) -> str:
        if config.get("identity_url"):
            return config["identity_url"].strip()
        parts = urlsplit(UiPath._base(config))
        # Automation Cloud has one identity server for every org; on-prem
        # Orchestrator / Automation Suite serve it at /identity on the same host.
        suffix = "/identity_/connect/token" if parts.hostname and parts.hostname.endswith("uipath.com") else "/identity/connect/token"
        return f"{parts.scheme}://{parts.netloc}{suffix}"

    async def _token(self, config: dict) -> str:
        client_id, secret = (config.get("client_id") or "").strip(), config.get("client_secret") or ""
        if not client_id or not secret:
            raise RpaError("UiPath needs the App ID and App Secret of an external application")
        url = self.identity_url(config)
        scope = (config.get("scope") or "OR.Jobs OR.Execution OR.Folders.Read OR.Queues").strip()
        key = ("uipath", url, client_id, scope)
        if token := _cached(key):
            return token
        async with _client() as client:
            resp = await client.post(url, data={"grant_type": "client_credentials", "client_id": client_id, "client_secret": secret, "scope": scope})
        data = _json(resp)
        if resp.status_code >= 400 or not isinstance(data, dict) or not data.get("access_token"):
            raise RpaError(f"UiPath rejected the App ID / App Secret: {_error_text(data)}")
        _TOKENS[key] = (data["access_token"], time.monotonic() + int(data.get("expires_in") or 3600))
        return data["access_token"]

    async def _folder_id(self, config: dict, client: httpx.AsyncClient, token: str) -> int | None:
        folder = (config.get("folder") or "").strip()
        if not folder:
            return None
        if folder.isdigit():
            return int(folder)
        key = (self._base(config), folder)
        if key in self._folders:
            return self._folders[key]
        resp = await client.get(
            f"{self._base(config)}/odata/Folders",
            params={"$filter": f"FullyQualifiedName eq '{_odata_quote(folder)}'", "$select": "Id,FullyQualifiedName"},
            headers={"Authorization": f"Bearer {token}"},
        )
        data = _json(resp)
        if resp.status_code >= 400:
            raise RpaError(f"couldn't look up folder '{folder}': HTTP {resp.status_code} {_error_text(data)}")
        found = _dig(data, "value.0.Id")
        if found is None:
            raise RpaError(f"no Orchestrator folder named '{folder}' is visible to this app — use its full path, e.g. \"Finance/AP\"")
        self._folders[key] = int(found)
        return int(found)

    async def _call(self, config: dict, method: str, path: str, **kwargs) -> Any:
        token = await self._token(config)
        async with _client() as client:
            headers = {"Authorization": f"Bearer {token}"}
            folder_id = await self._folder_id(config, client, token)
            if folder_id is not None:
                headers["X-UIPATH-OrganizationUnitId"] = str(folder_id)
            resp = await client.request(method, f"{self._base(config)}{path}", headers=headers, **kwargs)
        data = _json(resp)
        if resp.status_code >= 400:
            raise RpaError(f"UiPath HTTP {resp.status_code}: {_error_text(data)}")
        return data

    async def list_processes(self, config: dict) -> list[dict]:
        data = await self._call(config, "GET", "/odata/Releases", params={"$select": "Key,Name,ProcessKey,Description", "$orderby": "Name", "$top": "500"})
        return [{"id": r["Name"], "name": r["Name"], "description": r.get("Description") or r.get("ProcessKey") or ""} for r in (data.get("value") or [])]

    async def _release_key(self, config: dict, process: str) -> str:
        try:
            return str(uuid.UUID(process))
        except ValueError:
            pass
        data = await self._call(config, "GET", "/odata/Releases", params={"$filter": f"Name eq '{_odata_quote(process)}'", "$select": "Key,Name"})
        key = _dig(data, "value.0.Key")
        if not key:
            raise RpaError(f"no process named '{process}' in this Orchestrator folder")
        return key

    @staticmethod
    def _job(row: dict) -> Job:
        state = row.get("State")
        outputs: dict = {}
        if row.get("OutputArguments"):
            try:
                outputs = json.loads(row["OutputArguments"]) or {}
            except ValueError:
                outputs = {}
        return Job(job_id=str(row.get("Id")), status=_UIPATH_STATES.get(state, RUNNING), provider_status=state, outputs=outputs, info=row.get("Info") or "")

    async def start_job(self, config: dict, process: str, inputs: dict, timeout: float) -> Job:
        key = await self._release_key(config, process)
        start_info = {"ReleaseKey": key, "Strategy": "ModernJobsCount", "JobsCount": 1, "InputArguments": json.dumps(inputs, default=str)}
        data = await self._call(config, "POST", "/odata/Jobs/UiPath.Server.Configuration.OData.StartJobs", json={"startInfo": start_info})
        row = _dig(data, "value.0")
        if not row:
            raise RpaError(f"UiPath didn't start a job: {_error_text(data)}")
        return self._job(row)

    async def add_queue_item(self, config: dict, queue: str, inputs: dict, reference: str | None) -> Job:
        if not queue:
            raise RpaError("choose the Orchestrator queue to add the item to")
        item = {"Name": queue, "Priority": "Normal", "SpecificContent": inputs}
        if reference:
            item["Reference"] = reference
        data = await self._call(config, "POST", "/odata/Queues/UiPathODataSvc.AddQueueItem", json={"itemData": item})
        return Job(job_id=f"queue:{data.get('Id')}", status=_UIPATH_QUEUE_STATES.get(data.get("Status"), QUEUED), provider_status=data.get("Status"))

    async def job_status(self, config: dict, job_id: str) -> Job:
        if job_id.startswith("queue:"):
            item_id = job_id.split(":", 1)[1]
            data = await self._call(config, "GET", f"/odata/QueueItems({int(item_id)})")
            status = data.get("Status")
            info = _dig(data, "ProcessingException.Reason") or ""
            return Job(job_id=job_id, status=_UIPATH_QUEUE_STATES.get(status, RUNNING), provider_status=status, outputs=data.get("Output") or {}, info=info)
        if not job_id.isdigit():
            raise RpaError(f"'{job_id}' isn't a UiPath job id")
        return self._job(await self._call(config, "GET", f"/odata/Jobs({int(job_id)})"))

    async def test_credentials(self, config: dict) -> tuple[bool, str]:
        processes = await self.list_processes(config)
        where = f"folder '{config['folder']}'" if config.get("folder") else "the default folder"
        return True, f"UiPath accepted the credentials — {len(processes)} process(es) visible in {where}"


# ---- Automation Anywhere (Automation 360 Control Room) ------------------------------------------

_AA_STATES = {
    "QUEUED": QUEUED, "PENDING_EXECUTION": QUEUED, "DEPLOYED": QUEUED, "UPDATE": RUNNING, "RUNNING": RUNNING, "RUN_PAUSED": RUNNING,
    "COMPLETED": SUCCEEDED, "RUN_FAILED": FAILED, "DEPLOY_FAILED": FAILED, "RUN_TIMED_OUT": FAILED, "UNKNOWN": FAILED, "RUN_ABORTED": STOPPED,
}


def _aa_value(value: Any) -> dict:
    if isinstance(value, bool):
        return {"type": "BOOLEAN", "boolean": str(value).lower()}
    if isinstance(value, (int, float)):
        return {"type": "NUMBER", "number": str(value)}
    if isinstance(value, list):
        return {"type": "LIST", "list": [_aa_value(v) for v in value]}
    return {"type": "STRING", "string": "" if value is None else str(value)}


def _aa_plain(value: Any) -> Any:
    if not isinstance(value, dict):
        return value
    kind = (value.get("type") or "").upper()
    if kind == "LIST":
        return [_aa_plain(v) for v in value.get("list") or []]
    if kind == "NUMBER":
        number = value.get("number")
        try:
            return int(number) if str(number).lstrip("-").isdigit() else float(number)
        except (TypeError, ValueError):
            return number
    if kind == "BOOLEAN":
        return str(value.get("boolean")).lower() == "true"
    return value.get("string", value.get("value"))


class AutomationAnywhere:
    capabilities = ["jobs", "status"]

    @staticmethod
    def _base(config: dict) -> str:
        base = (config.get("control_room_url") or "").strip().rstrip("/")
        if not base:
            raise RpaError("set the Control Room URL")
        return base

    async def _token(self, config: dict) -> str:
        user = (config.get("username") or "").strip()
        api_key, password = config.get("api_key") or "", config.get("password") or ""
        if not user or not (api_key or password):
            raise RpaError("Automation Anywhere needs a username plus an API key (or password)")
        key = ("aa", self._base(config), user)
        if token := _cached(key):
            return token
        body = {"username": user, "apiKey": api_key} if api_key else {"username": user, "password": password}
        async with _client() as client:
            resp = await client.post(f"{self._base(config)}/v2/authentication", json=body)
        data = _json(resp)
        if resp.status_code >= 400 or not isinstance(data, dict) or not data.get("token"):
            raise RpaError(f"the Control Room rejected the login: {_error_text(data)}")
        _TOKENS[key] = (data["token"], time.monotonic() + 15 * 60)  # A360 tokens last 20 minutes
        return data["token"]

    async def _post(self, config: dict, path: str, body: dict) -> Any:
        token = await self._token(config)
        async with _client() as client:
            resp = await client.post(f"{self._base(config)}{path}", json=body, headers={"X-Authorization": token})
        data = _json(resp)
        if resp.status_code >= 400:
            raise RpaError(f"Control Room HTTP {resp.status_code}: {_error_text(data)}")
        return data

    async def list_processes(self, config: dict) -> list[dict]:
        body = {
            "filter": {"operator": "eq", "field": "type", "value": "application/vnd.aa.taskbot"},
            "sort": [{"field": "name", "direction": "asc"}],
            "page": {"offset": 0, "length": 500},
        }
        data = await self._post(config, "/v2/repository/workspaces/public/files/list", body)
        return [{"id": str(f["id"]), "name": f.get("name") or str(f["id"]), "description": f.get("path") or ""} for f in data.get("list") or []]

    @staticmethod
    def _ids(value: Any) -> list[int]:
        return [int(v) for v in re.split(r"[,\s]+", str(value or "")) if v.strip().isdigit()]

    async def start_job(self, config: dict, process: str, inputs: dict, timeout: float) -> Job:
        bot_id = process.strip()
        if not bot_id.isdigit():
            match = next((p for p in await self.list_processes(config) if p["name"] == bot_id or p["description"] == bot_id), None)
            if not match:
                raise RpaError(f"no bot named '{process}' in the public workspace")
            bot_id = match["id"]
        run_as, pools = self._ids(config.get("run_as_user_ids")), self._ids(config.get("device_pool_ids"))
        if not run_as:
            raise RpaError("set the Run-as user id(s) on the connector — unattended bots need a Bot Runner to run as")
        unattended: dict = {"runAsUserIds": run_as, "numOfRunAsUsersToUse": 1}
        if pools:
            unattended["poolIds"] = pools
        body = {"botId": int(bot_id), "unattendedRequest": unattended, "botInput": {k: _aa_value(v) for k, v in inputs.items()}}
        data = await self._post(config, "/v3/automations/deploy", body)
        deployment = data.get("deploymentId")
        if not deployment:
            raise RpaError(f"the Control Room didn't deploy the bot: {_error_text(data)}")
        return Job(job_id=str(deployment), status=QUEUED, provider_status="DEPLOYED")

    async def job_status(self, config: dict, job_id: str) -> Job:
        data = await self._post(config, "/v3/activity/list", {"filter": {"operator": "eq", "field": "deploymentId", "value": job_id}})
        row = _dig(data, "list.0")
        if not row:
            return Job(job_id=job_id, status=QUEUED, provider_status=None, info="waiting for a Bot Runner to pick the job up")
        status = row.get("status")
        values = _dig(row, "botOutVariables.values") or {}
        outputs = {k: _aa_plain(v) for k, v in values.items()} if isinstance(values, dict) else {}
        return Job(job_id=job_id, status=_AA_STATES.get(status, RUNNING), provider_status=status, outputs=outputs, info=row.get("message") or "")

    async def test_credentials(self, config: dict) -> tuple[bool, str]:
        await self._token(config)
        bots = await self.list_processes(config)
        return True, f"Control Room accepted the login — {len(bots)} bot(s) in the public workspace"


# ---- Microsoft Power Automate (flows with an HTTP request trigger) ------------------------------


def power_automate_flows(config: dict) -> dict[str, str]:
    """The connector's "Flows" box — one "Name | URL" per line (a bare URL
    is named "Flow 1", "Flow 2", …)."""
    flows: dict[str, str] = {}
    for line in (config.get("flows") or "").splitlines():
        line = line.strip()
        if not line:
            continue
        name, _, url = line.rpartition("|") if "|" in line else ("", "", line)
        flows[name.strip() or f"Flow {len(flows) + 1}"] = url.strip()
    return flows


class PowerAutomate:
    capabilities = ["jobs", "status"]

    async def list_processes(self, config: dict) -> list[dict]:
        return [{"id": name, "name": name, "description": urlsplit(url).hostname or ""} for name, url in power_automate_flows(config).items()]

    @staticmethod
    def _outputs(resp: httpx.Response) -> dict:
        data = _json(resp) if resp.content else {}
        return data if isinstance(data, dict) else {"result": data}

    async def start_job(self, config: dict, process: str, inputs: dict, timeout: float) -> Job:
        flows = power_automate_flows(config)
        if not flows:
            raise RpaError("add at least one flow URL to the connector")
        url = flows.get(process) or (next(iter(flows.values())) if len(flows) == 1 and not process else None)
        if not url:
            raise RpaError(f"no flow named '{process}' on this connector — pick one of: {', '.join(flows)}")
        async with _client(timeout=max(timeout, 10)) as client:
            resp = await client.post(url, json=inputs)
        run_id = resp.headers.get("x-ms-workflow-run-id") or uuid.uuid4().hex
        if resp.status_code >= 400:
            return Job(job_id=run_id, status=FAILED, provider_status=f"HTTP {resp.status_code}", info=_error_text(_json(resp)))
        if resp.status_code == 202:
            # An asynchronous Response action hands back a Location to poll;
            # without one the flow ran, but nothing reports how it went.
            if location := resp.headers.get("location"):
                return Job(job_id=location, status=RUNNING, provider_status="Accepted")
            return Job(job_id=run_id, status=SUBMITTED, provider_status="Accepted", info="flow triggered — add a Response action to the flow to get its result back")
        return Job(job_id=run_id, status=SUCCEEDED, provider_status=f"HTTP {resp.status_code}", outputs=self._outputs(resp))

    async def job_status(self, config: dict, job_id: str) -> Job:
        if not job_id.startswith("https://"):
            raise RpaError("this flow run can't be checked — only flows whose Response action is asynchronous report status")
        async with _client() as client:
            resp = await client.get(job_id)
        if resp.status_code == 202:
            return Job(job_id=job_id, status=RUNNING, provider_status="Running")
        if resp.status_code >= 400:
            return Job(job_id=job_id, status=FAILED, provider_status=f"HTTP {resp.status_code}", info=_error_text(_json(resp)))
        return Job(job_id=job_id, status=SUCCEEDED, provider_status=f"HTTP {resp.status_code}", outputs=self._outputs(resp))

    async def test_credentials(self, config: dict) -> tuple[bool, str]:
        # Any request to a trigger URL runs the flow, so only check that each
        # URL is well-formed and its host answers — never call it.
        flows = power_automate_flows(config)
        if not flows:
            return False, "add at least one flow URL"
        for name, url in flows.items():
            parts = urlsplit(url)
            if parts.scheme != "https" or not parts.hostname or "sig=" not in (parts.query or ""):
                return False, f"'{name}' doesn't look like an HTTP-trigger URL — copy it from the trigger's \"HTTP POST URL\" (it includes sig=…)"
            _, writer = await asyncio.wait_for(asyncio.open_connection(parts.hostname, 443, ssl=True), timeout=5.0)
            writer.close()
        return True, f"Reached Power Automate for {len(flows)} flow(s) — flows aren't triggered by this test"


# ---- Blue Prism (processes exposed as web services) ---------------------------------------------


class BluePrism:
    capabilities = ["jobs"]

    @staticmethod
    def _base(config: dict) -> str:
        base = (config.get("runtime_url") or "").strip().rstrip("/")
        if not base:
            raise RpaError("set the Runtime Resource URL, e.g. http://bp-runtime:8181")
        return base

    @staticmethod
    def _auth(config: dict) -> tuple[str, str]:
        return (config.get("username") or "", config.get("password") or "")

    async def list_processes(self, config: dict) -> list[dict]:
        names = [n.strip() for n in re.split(r"[,\n]", config.get("processes") or "") if n.strip()]
        return [{"id": n, "name": n, "description": "exposed web service"} for n in names]

    def _call_sync(self, config: dict, process: str, inputs: dict, timeout: float) -> dict:
        import requests
        import zeep
        from zeep.helpers import serialize_object
        from zeep.transports import Transport

        session = requests.Session()
        session.auth = self._auth(config)
        client = zeep.Client(f"{self._base(config)}/ws/{quote(process)}?wsdl", transport=Transport(session=session, operation_timeout=timeout, timeout=15))
        operations = [op for service in client.wsdl.services.values() for port in service.ports.values() for op in port.binding._operations]
        name = next((op for op in operations if op.lower() == process.replace(" ", "").lower()), operations[0] if len(operations) == 1 else None)
        if not name:
            raise RpaError(f"'{process}' exposes several operations ({', '.join(operations)}) — expose one process per web service")
        result = serialize_object(getattr(client.service, name)(**inputs))
        return dict(result) if isinstance(result, dict) else ({} if result is None else {"result": result})

    async def start_job(self, config: dict, process: str, inputs: dict, timeout: float) -> Job:
        # A Blue Prism web-service call runs the process to the end, so the
        # job is already finished when this returns.
        try:
            outputs = await asyncio.to_thread(self._call_sync, config, process, inputs, timeout)
        except RpaError:
            raise
        except Exception as exc:  # noqa: BLE001 - zeep/requests raise many types; the fault text is what matters
            return Job(job_id=uuid.uuid4().hex, status=FAILED, provider_status="Fault", info=str(exc)[:500])
        return Job(job_id=uuid.uuid4().hex, status=SUCCEEDED, provider_status="Completed", outputs=json.loads(json.dumps(outputs, default=str)))

    async def job_status(self, config: dict, job_id: str) -> Job:
        raise RpaError("Blue Prism web-service runs finish within the call — there's no job to check afterwards")

    async def test_credentials(self, config: dict) -> tuple[bool, str]:
        processes = await self.list_processes(config)
        target = f"{self._base(config)}/ws/{quote(processes[0]['name'])}?wsdl" if processes else f"{self._base(config)}/ws/"
        async with _client() as client:
            resp = await client.get(target, auth=self._auth(config))
        if resp.status_code == 401:
            return False, "Blue Prism rejected the username / password"
        if resp.status_code >= 400:
            return False, f"HTTP {resp.status_code} from {target} — check the process is exposed as a web service"
        return True, f"Reached the Blue Prism runtime ({'WSDL for ' + processes[0]['name'] if processes else 'service list'} OK)"


# ---- Any other bot / orchestrator with a REST API ------------------------------------------------


def _render(template: str, values: dict) -> str:
    out = template
    for k, v in values.items():
        out = out.replace("{{" + k + "}}", "" if v is None else str(v))
    return out


def _value_set(raw: str | None, default: str) -> set[str]:
    return {v.strip().lower() for v in (raw or default).split(",") if v.strip()}


class CustomRest:
    capabilities = ["jobs", "status"]

    @staticmethod
    def _base(config: dict) -> str:
        base = (config.get("api_base_url") or "").strip().rstrip("/")
        if not base:
            raise RpaError("set the API base URL")
        return base

    @staticmethod
    def _auth(config: dict) -> tuple[dict, tuple[str, str] | None]:
        kind = config.get("auth_type") or "bearer"
        key = config.get("api_key") or ""
        if kind == "bearer" and key:
            return {"Authorization": f"Bearer {key}"}, None
        if kind == "basic":
            return {}, (key, config.get("api_secret") or "")
        if kind == "header" and key:
            return {(config.get("auth_header") or "X-API-Key"): key}, None
        return {}, None

    async def _request(self, config: dict, method: str, path: str, **kwargs) -> Any:
        headers, auth = self._auth(config)
        async with _client() as client:
            resp = await client.request(method, f"{self._base(config)}{path}", headers=headers, auth=auth, **kwargs)
        data = _json(resp)
        if resp.status_code >= 400:
            raise RpaError(f"HTTP {resp.status_code} from {path}: {_error_text(data)}")
        return data

    async def list_processes(self, config: dict) -> list[dict]:
        path = (config.get("processes_path") or "").strip()
        if not path:
            return []
        data = await self._request(config, "GET", path)
        rows = _dig(data, config.get("processes_field") or "") if config.get("processes_field") else data
        id_field, name_field = config.get("process_id_field") or "id", config.get("process_name_field") or "name"
        out = []
        for row in rows if isinstance(rows, list) else []:
            if isinstance(row, dict):
                out.append({"id": str(_dig(row, id_field) or _dig(row, name_field)), "name": str(_dig(row, name_field) or _dig(row, id_field)), "description": ""})
            else:
                out.append({"id": str(row), "name": str(row), "description": ""})
        return out

    def _job(self, config: dict, job_id: str, data: Any) -> Job:
        raw_status = _dig(data, config.get("status_field") or "status")
        value = str(raw_status or "").lower()
        if value in _value_set(config.get("succeeded_values"), "succeeded,success,successful,completed,done,finished"):
            status = SUCCEEDED
        elif value in _value_set(config.get("failed_values"), "failed,faulted,error,errored"):
            status = FAILED
        elif value in {"stopped", "cancelled", "canceled", "aborted"}:
            status = STOPPED
        elif value in {"queued", "pending", "new", "waiting"}:
            status = QUEUED
        else:
            status = RUNNING
        outputs = _dig(data, config.get("outputs_field") or "outputs")
        return Job(job_id=job_id, status=status, provider_status=None if raw_status is None else str(raw_status), outputs=outputs if isinstance(outputs, dict) else {}, info=str(_dig(data, "message") or ""))

    async def start_job(self, config: dict, process: str, inputs: dict, timeout: float) -> Job:
        template = (config.get("body_template") or "").strip() or '{"process": "{{process}}", "inputs": {{inputs_json}}}'
        rendered = _render(template, {"process": process, "inputs_json": json.dumps(inputs, default=str), **{k: v for k, v in inputs.items() if not isinstance(v, (dict, list))}})
        try:
            body = json.loads(rendered)
        except ValueError as exc:
            raise RpaError(f"the start-job body template isn't valid JSON after filling it in: {exc}") from exc
        data = await self._request(config, "POST", _render(config.get("start_path") or "/jobs", {"process": quote(process)}), json=body)
        job_id = _dig(data, config.get("job_id_field") or "id")
        if job_id is None:
            raise RpaError(f"the response has no '{config.get('job_id_field') or 'id'}' — set the job id field")
        job = self._job(config, str(job_id), data)
        # No status in the start response just means "started".
        return job if _dig(data, config.get("status_field") or "status") is not None else Job(job_id=str(job_id), status=RUNNING)

    async def job_status(self, config: dict, job_id: str) -> Job:
        path = (config.get("status_path") or "/jobs/{job_id}").replace("{job_id}", quote(job_id))
        return self._job(config, job_id, await self._request(config, "GET", path))

    async def test_credentials(self, config: dict) -> tuple[bool, str]:
        headers, auth = self._auth(config)
        target = f"{self._base(config)}{config.get('processes_path') or ''}"
        async with _client() as client:
            resp = await client.get(target, headers=headers, auth=auth)
        if resp.status_code in (401, 403):
            return False, f"HTTP {resp.status_code} — the API rejected the credentials"
        return resp.status_code < 500, f"Reached the API (HTTP {resp.status_code})"


ADAPTERS: dict[str, Any] = {
    "uipath": UiPath(),
    "automation_anywhere": AutomationAnywhere(),
    "power_automate": PowerAutomate(),
    "blue_prism": BluePrism(),
    "custom": CustomRest(),
}


def adapter_for(config: dict):
    provider = config.get("provider")
    if provider not in ADAPTERS:
        raise RpaError("choose the RPA platform on the connector")
    return ADAPTERS[provider]


async def wait_for_job(config: dict, job: Job, timeout: float, *, poll_interval: float = 2.0) -> Job:
    """Polls until the job finishes or `timeout` seconds pass (then returns
    it still queued/running)."""
    adapter = adapter_for(config)
    if "status" not in adapter.capabilities and not job.job_id.startswith("queue:"):
        return job
    deadline = time.monotonic() + timeout
    interval = poll_interval
    while not job.done:
        if time.monotonic() + interval > deadline:
            break
        await _sleep(interval)
        job = await adapter.job_status(config, job.job_id)
        interval = min(interval * 1.5, 10.0)
    return job


async def run_process(
    config: dict,
    *,
    action: str,
    process: str,
    inputs: dict,
    queue: str = "",
    wait: bool = True,
    timeout: float = 120,
) -> Job:
    adapter = adapter_for(config)
    if action == "add_queue_item":
        if "queues" not in adapter.capabilities:
            raise RpaError("this RPA platform doesn't support queues here — start a job instead")
        reference = inputs.get("reference")
        job = await adapter.add_queue_item(config, queue, {k: v for k, v in inputs.items() if k != "reference"}, None if reference is None else str(reference))
    else:
        if not process and not isinstance(adapter, PowerAutomate):
            raise RpaError("choose the process / bot to run")
        job = await adapter.start_job(config, process, inputs, timeout)
    return await wait_for_job(config, job, timeout) if wait else job


# What the Connectors screen shows per platform (GET /rpa/providers).
def _f(key: str, label: str, *, secret: bool = False, required: bool = True, placeholder: str = "", help: str = "", options: list | None = None, advanced: bool = False, multiline: bool = False) -> dict:
    return {"key": key, "label": label, "secret": secret, "required": required, "placeholder": placeholder, "help": help, "options": options, "advanced": advanced, "multiline": multiline}


def _p(pid: str, name: str, tagline: str, docs_url: str | None, fields: list[dict], notes: str) -> dict:
    return {"id": pid, "name": name, "tagline": tagline, "environments": [], "docs_url": docs_url, "fields": fields, "notes": notes, "capabilities": ADAPTERS[pid].capabilities}


PROVIDERS: list[dict] = [
    _p(
        "uipath", "UiPath Orchestrator", "Start unattended jobs or add queue items — Cloud or on-prem",
        "https://docs.uipath.com/orchestrator/automation-cloud/latest/api-guide/jobs-requests",
        [
            _f("orchestrator_url", "Orchestrator URL", placeholder="https://cloud.uipath.com/acme/DefaultTenant/orchestrator_", help="Automation Cloud: https://cloud.uipath.com/{organization}/{tenant}/orchestrator_. On-prem: your Orchestrator's address."),
            _f("client_id", "App ID", help="Admin → External Applications → Add application (confidential, application scopes OR.Jobs, OR.Execution, OR.Folders.Read, OR.Queues)."),
            _f("client_secret", "App Secret", secret=True),
            _f("folder", "Folder", required=False, placeholder="Shared", help='The Orchestrator folder the processes live in — full path for sub-folders, e.g. "Finance/AP". Blank = the default folder.'),
            _f("scope", "Scopes", required=False, placeholder="OR.Jobs OR.Execution OR.Folders.Read OR.Queues", advanced=True),
            _f("identity_url", "Identity server token URL", required=False, placeholder="worked out from the Orchestrator URL", advanced=True, help="Only for unusual on-prem setups."),
        ],
        "Inputs become the job's input arguments (or the queue item's specific content); output arguments come back when the job finishes.",
    ),
    _p(
        "automation_anywhere", "Automation Anywhere", "Deploy Automation 360 bots to unattended Bot Runners",
        "https://docs.automationanywhere.com/bundle/enterprise-v2019/page/automation-360-api-deploy-bot.html",
        [
            _f("control_room_url", "Control Room URL", placeholder="https://acme.my.automationanywhere.digital"),
            _f("username", "Username", help="A Control Room user allowed to deploy bots (API key generation enabled)."),
            _f("api_key", "API key", secret=True, required=False, help="My settings → Generate API key. Preferred over a password."),
            _f("password", "Password", secret=True, required=False, help="Only if you don't use an API key."),
            _f("run_as_user_ids", "Run-as user id(s)", placeholder="e.g. 12, 14", help="Numeric ids of the unattended Bot Runner users that run the bot (Administration → Users)."),
            _f("device_pool_ids", "Device pool id(s)", required=False, placeholder="e.g. 3", advanced=True),
        ],
        "Inputs become the bot's input variables (text, number, boolean or list); its output variables come back when it completes.",
    ),
    _p(
        "power_automate", "Power Automate", "Trigger cloud flows (and the desktop flows they run) over HTTP",
        "https://learn.microsoft.com/power-automate/oauth-authentication",
        [
            _f(
                "flows", "Flows", multiline=True,
                placeholder="Invoice entry | https://prod-12.westus.logic.azure.com:443/workflows/…/triggers/manual/paths/invoke?…&sig=…",
                help='One flow per line as "Name | HTTP POST URL" — copy the URL from the flow\'s "When an HTTP request is received" trigger. The URL is the credential, so keep it secret.',
            ),
        ],
        "Inputs are POSTed as the JSON body. Add a Response action to the flow to return results; make it asynchronous for long-running desktop flows so FlowFabric can wait for them.",
    ),
    _p(
        "blue_prism", "Blue Prism", "Run processes exposed as web services on a Runtime Resource",
        "https://bpdocs.blueprism.com/bp-7-1/en-us/helpSystemExposedWebServices.htm",
        [
            _f("runtime_url", "Runtime Resource URL", placeholder="http://bp-runtime-01:8181"),
            _f("username", "Username", help="A Blue Prism user allowed to run the exposed processes."),
            _f("password", "Password", secret=True),
            _f("processes", "Exposed processes", multiline=True, placeholder="InvoiceProcessing\nVendorOnboarding", help="System → Objects/Processes → Exposure: the web-service names, one per line."),
        ],
        "The process runs to completion inside the call; its startup parameters are the inputs and its output parameters come back.",
    ),
    _p(
        "custom", "Other RPA / bot API", "Any orchestrator with a REST API to start and check jobs",
        None,
        [
            _f("api_base_url", "API base URL", placeholder="https://rpa.example.com/api"),
            _f("auth_type", "Authentication", options=[["bearer", "Bearer token"], ["basic", "Basic (key + secret)"], ["header", "API key header"], ["none", "None"]]),
            _f("api_key", "API key / token", secret=True, required=False),
            _f("api_secret", "API secret (Basic auth)", secret=True, required=False),
            _f("auth_header", "Header name (API key header)", required=False, placeholder="X-API-Key"),
            _f("start_path", "Start-job path", required=False, placeholder="/jobs", advanced=True, help="POST. {process} = the chosen process."),
            _f("body_template", "Start-job body (JSON template)", required=False, multiline=True, advanced=True, placeholder='{"process": "{{process}}", "inputs": {{inputs_json}}}', help="Placeholders: {{process}}, {{inputs_json}} and each input by name."),
            _f("job_id_field", "Job id field in the response", required=False, placeholder="id", advanced=True),
            _f("status_path", "Job-status path", required=False, placeholder="/jobs/{job_id}", advanced=True),
            _f("status_field", "Status field", required=False, placeholder="status", advanced=True),
            _f("succeeded_values", "Values meaning succeeded", required=False, placeholder="succeeded,success,completed,done", advanced=True),
            _f("failed_values", "Values meaning failed", required=False, placeholder="failed,faulted,error", advanced=True),
            _f("outputs_field", "Outputs field", required=False, placeholder="outputs", advanced=True),
            _f("processes_path", "List-processes path", required=False, placeholder="/processes", advanced=True, help="Optional GET that lists processes, for the picker in Task Designer."),
            _f("processes_field", "Process list field", required=False, placeholder="(the whole response)", advanced=True),
            _f("process_id_field", "Process id field", required=False, placeholder="id", advanced=True),
            _f("process_name_field", "Process name field", required=False, placeholder="name", advanced=True),
        ],
        "FlowFabric POSTs to start a job, then polls the status path until it finishes.",
    ),
]