"""The workflow engine: runs instances on the server.

An instance holds tokens (WorkflowInstance.tokens) — where it currently is,
one per active branch. Entering a step decides what the token does there
(app/schemas/workflow.py's WorkflowStepType has the list):

  * a person's step (task, approval, a start or decision with approvers)
    waits — state "active" — for one of its assignees to act (act() below),
    with an SLA, a reminder and an escalation the scheduler handles (due());
  * start, decision, parallel and join route themselves straight away;
  * timer, subprocess and service wait — state "waiting" — for the clock,
    the child instance(s), or the background job running the Task;
  * end consumes the token; the instance completes when none are left.

Anything that can't go on (no branch matched, a missing setting, a Task that
failed for good) turns its token into an incident — state "failed", with the
reason — for an administrator to retry, skip or move (the admin_* functions).

Every change to an instance happens under a row lock (FOR UPDATE), so two
people acting at once, or a person and the scheduler, can't both move it.
Each move writes a StepHistory row and fires the steps' on_enter/on_complete
notifications, as before. current_step mirrors the first token's step for
the readers that want a single one.
"""

from __future__ import annotations

import copy
import logging
import uuid
from dataclasses import dataclass, field
from datetime import date, datetime, timedelta, timezone
from typing import Any

from sqlalchemy import and_, or_, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm.attributes import flag_modified

from app.core import jobs
from app.crud.dataset_query import _eval_operator
from app.crud.notifications import dispatch_step_notifications
from app.crud.params import apply_output_params, resolve_input_params
from app.models.automation import Task
from app.models.identity import Role, User, UserRole
from app.models.notification import Notification
from app.models.workflow import StepHistory, Workflow, WorkflowDelegation, WorkflowInstance
from app.schemas.workflow import HUMAN_STEP_TYPES

log = logging.getLogger("flowfabric.workflow")

# Automatic moves one run may make before it's taken to be a loop.
MAX_AUTO_MOVES = 200
# Sub-workflows inside sub-workflows, at most.
MAX_DEPTH = 10
SERVICE_JOB = "workflow.service"
DUE_JOB = "workflow.due"
# A child started by a sub-workflow that doesn't wait for it.
DETACHED = "detached"
TERMINAL = ("completed", "cancelled")


class EngineError(Exception):
    """A request the engine refuses — the router turns it into an HTTP error."""

    def __init__(self, status_code: int, detail: str):
        super().__init__(detail)
        self.status_code = status_code
        self.detail = detail


def _now() -> datetime:
    return datetime.now(timezone.utc)


def _iso(value: datetime) -> str:
    return value.isoformat()


def _parse_dt(value: Any) -> datetime | None:
    if isinstance(value, datetime):
        return value if value.tzinfo else value.replace(tzinfo=timezone.utc)
    if isinstance(value, date):
        return datetime(value.year, value.month, value.day, tzinfo=timezone.utc)
    if isinstance(value, str) and value.strip():
        try:
            parsed = datetime.fromisoformat(value.strip().replace("Z", "+00:00"))
        except ValueError:
            return None
        return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)
    return None


def add_hours(start: datetime, hours: float, business_days: bool) -> datetime:
    """Calendar mode is a plain timedelta. Business-days mode spends the
    same hours only across Mon–Fri: a stretch that would land on a weekend
    jumps to the following Monday at the same time of day, as if Sat/Sun
    never happened. No holiday calendar — weekends only."""
    if not business_days:
        return start + timedelta(hours=hours)
    remaining = timedelta(hours=hours)
    current = start
    while remaining > timedelta(0):
        if current.weekday() >= 5:  # already Sat/Sun — jump straight to Monday
            days_to_monday = (7 - current.weekday()) % 7
            current = (current + timedelta(days=days_to_monday)).replace(hour=0, minute=0, second=0, microsecond=0)
            continue
        days_to_saturday = (5 - current.weekday()) % 7
        boundary = (current + timedelta(days=days_to_saturday)).replace(hour=0, minute=0, second=0, microsecond=0)
        available = boundary - current
        if available >= remaining:
            current += remaining
            remaining = timedelta(0)
        else:
            current = boundary
            remaining -= available
    return current


def find_step(workflow: Workflow | None, key: str | None) -> dict | None:
    if workflow is None or not key:
        return None
    return next((s for s in workflow.step_definitions or [] if s.get("key") == key), None)


def is_human(step: dict | None) -> bool:
    """A step that waits for a person: tasks and approvals, and a start or
    decision step that names approvers."""
    if not step:
        return False
    if step.get("type") in HUMAN_STEP_TYPES:
        return True
    return step.get("type") in ("start", "decision") and bool(step.get("approver_roles") or step.get("approver_users") or step.get("assignee_field"))


def _holds(transition: dict, context: dict) -> bool:
    cond = transition.get("condition")
    if not cond or not cond.get("driver_field"):
        return True
    return bool(_eval_operator(cond.get("operator") or "equals", context.get(cond.get("driver_field")), cond.get("value")))


def _conditional(transition: dict) -> bool:
    cond = transition.get("condition")
    return bool(cond and cond.get("driver_field"))


def open_choices(step: dict, context: dict) -> list[dict]:
    """The transitions a person may take from `step`: those whose condition holds."""
    return [t for t in step.get("next_steps") or [] if t.get("to") and _holds(t, context)]


def route(step: dict, context: dict) -> str | None:
    """Where a step that routes itself goes: the first conditional branch
    that holds, else the first branch with no condition (the "otherwise")."""
    transitions = [t for t in step.get("next_steps") or [] if t.get("to")]
    for t in transitions:
        if _conditional(t) and _holds(t, context):
            return t["to"]
    return next((t["to"] for t in transitions if not _conditional(t)), None)


def _ids(values: Any) -> list[str]:
    if values is None:
        return []
    if not isinstance(values, (list, tuple, set)):
        values = [values]
    out = []
    for v in values:
        try:
            out.append(str(uuid.UUID(str(v))))
        except (ValueError, TypeError):
            continue
    return out


# ---- the run: one instance, locked, being moved -------------------------------------------------


@dataclass
class _Run:
    db: AsyncSession
    tenant_id: uuid.UUID
    workflow: Workflow
    instance: WorkflowInstance
    actor_id: uuid.UUID | None
    tokens: list[dict] = field(default_factory=list)
    joins: dict = field(default_factory=dict)
    context: dict = field(default_factory=dict)
    moves: int = 0
    # Children whose parent token resumes once this run is saved.
    finished_children: list[WorkflowInstance] = field(default_factory=list)

    @property
    def business_days(self) -> bool:
        return bool((self.workflow.settings or {}).get("business_days"))

    def step(self, key: str | None) -> dict | None:
        return find_step(self.workflow, key)

    def token(self, token_id: str | None) -> dict | None:
        if token_id is None:
            return self.tokens[0] if self.tokens else None
        return next((t for t in self.tokens if t.get("id") == token_id), None)

    def history(self, step: str | None, action: str, comment: str | None = None, actor_id: uuid.UUID | None | bool = True) -> None:
        # Each row a moment after the last: the database's now() is the same
        # for the whole transaction, which would leave one run's rows unordered.
        at = max(_now(), self.db.info.get("workflow_history_at", _now()) + timedelta(microseconds=1))
        self.db.info["workflow_history_at"] = at
        self.db.add(
            StepHistory(
                tenant_id=self.tenant_id,
                instance_id=self.instance.id,
                step=(step or "")[:100],
                actor_id=self.actor_id if actor_id is True else (actor_id or None),
                action=action[:100],
                comment=comment,
                occurred_at=at,
            )
        )


def _busy(db: AsyncSession) -> set:
    return db.info.setdefault("workflow_engine_running", set())


async def _lock(db: AsyncSession, tenant_id: uuid.UUID, instance_id: uuid.UUID) -> WorkflowInstance:
    instance = (
        await db.execute(
            select(WorkflowInstance).where(WorkflowInstance.tenant_id == tenant_id, WorkflowInstance.id == instance_id).with_for_update()
        )
    ).scalar_one_or_none()
    if instance is None:
        raise EngineError(404, "Workflow instance not found")
    return instance


async def load_workflow(db: AsyncSession, tenant_id: uuid.UUID, instance: WorkflowInstance) -> Workflow | None:
    return (
        await db.execute(
            select(Workflow).where(
                Workflow.tenant_id == tenant_id, Workflow.workflow_id == instance.workflow_id, Workflow.version == instance.workflow_version
            )
        )
    ).scalar_one_or_none()


async def _open_run(db: AsyncSession, tenant_id: uuid.UUID, instance: WorkflowInstance, actor_id: uuid.UUID | None) -> _Run:
    workflow = await load_workflow(db, tenant_id, instance)
    if workflow is None:
        raise EngineError(409, "This instance's workflow version no longer exists")
    # Deep copies: the run changes them in place, and the loaded values must
    # stay as they were for the change to be seen and saved.
    run = _Run(
        db,
        tenant_id,
        workflow,
        instance,
        actor_id,
        tokens=copy.deepcopy(list(instance.tokens or [])),
        joins=copy.deepcopy(dict(instance.join_state or {})),
        context=copy.deepcopy(dict(instance.context or {})),
    )
    # An instance from before tokens existed: one at its current step.
    if not run.tokens and instance.status == "in_progress" and instance.current_step and instance.completed_at is None:
        step = run.step(instance.current_step)
        if step is not None and step.get("type") != "end":
            run.tokens.append(_new_token(instance.current_step, []))
            run.tokens[0]["state"] = "active" if is_human(step) or step.get("type") in ("start", "decision") else "waiting"
    return run


def _new_token(step_key: str, splits: list[str]) -> dict:
    return {"id": uuid.uuid4().hex, "step": step_key, "state": "active", "entered_at": _iso(_now()), "splits": list(splits)}


def _next_due(tokens: list[dict]) -> datetime | None:
    times: list[datetime] = []
    for t in tokens:
        if t.get("state") == "waiting" and t.get("wake_at"):
            times.append(_parse_dt(t["wake_at"]))
        if t.get("state") == "active":
            if t.get("remind_at") and not t.get("reminded"):
                times.append(_parse_dt(t["remind_at"]))
            if t.get("due_at") and not t.get("escalated"):
                times.append(_parse_dt(t["due_at"]))
    times = [x for x in times if x is not None]
    return min(times) if times else None


async def _save(run: _Run) -> None:
    inst = run.instance
    inst.tokens = copy.deepcopy(run.tokens)
    inst.join_state = copy.deepcopy(run.joins)
    inst.context = copy.deepcopy(run.context)
    for name in ("tokens", "join_state", "context"):
        flag_modified(inst, name)
    if run.tokens:
        inst.current_step = run.tokens[0]["step"]
    inst.next_due_at = _next_due(run.tokens) if inst.status == "in_progress" else None
    await run.db.flush()


# ---- people: who a step waits on ------------------------------------------------------------------


async def _role_members(db: AsyncSession, tenant_id: uuid.UUID, role_ids: list[str]) -> set[str]:
    if not role_ids:
        return set()
    rows = await db.execute(
        select(UserRole.user_id)
        .join(Role, Role.id == UserRole.role_id)
        .join(User, User.id == UserRole.user_id)
        .where(Role.tenant_id == tenant_id, Role.id.in_([uuid.UUID(r) for r in role_ids]), User.status == "active")
    )
    return {str(u) for (u,) in rows.all()}


def assignment(step: dict, token: dict, context: dict) -> tuple[list[str], list[str]]:
    """(user ids, role ids) a person's step waits on — the token's own
    (after a reassignment or escalation) or the step's approvers plus the
    user(s) named in its assignee_field."""
    override = token.get("assignees")
    if override:
        return _ids(override.get("user_ids")), _ids(override.get("role_ids"))
    users = _ids(step.get("approver_users"))
    if step.get("assignee_field"):
        users += [u for u in _ids(context.get(step["assignee_field"])) if u not in users]
    return users, _ids(step.get("approver_roles"))


async def resolve_assignees(db: AsyncSession, tenant_id: uuid.UUID, step: dict, token: dict, context: dict) -> list[str]:
    users, roles = assignment(step, token, context)
    members = await _role_members(db, tenant_id, roles)
    active = set()
    if users:
        rows = await db.execute(select(User.id).where(User.tenant_id == tenant_id, User.id.in_([uuid.UUID(u) for u in users]), User.status == "active"))
        active = {str(u) for (u,) in rows.all()}
    return sorted(active | members)


async def delegators_for(db: AsyncSession, tenant_id: uuid.UUID, delegate_id: uuid.UUID, workflow_id: uuid.UUID | None = None) -> dict[str, set]:
    """Who has handed their tasks to `delegate_id` right now:
    {user id: {workflow ids, or None for every workflow}}."""
    now = _now()
    stmt = select(WorkflowDelegation.user_id, WorkflowDelegation.workflow_id).where(
        WorkflowDelegation.tenant_id == tenant_id,
        WorkflowDelegation.delegate_id == delegate_id,
        WorkflowDelegation.starts_at <= now,
        WorkflowDelegation.ends_at > now,
    )
    if workflow_id is not None:
        stmt = stmt.where(or_(WorkflowDelegation.workflow_id.is_(None), WorkflowDelegation.workflow_id == workflow_id))
    out: dict[str, set] = {}
    for user_id, wf in (await db.execute(stmt)).all():
        out.setdefault(str(user_id), set()).add(str(wf) if wf else None)
    return out


async def delegates_of(db: AsyncSession, tenant_id: uuid.UUID, user_ids: list[str], workflow_id: uuid.UUID) -> dict[str, str]:
    """{user id: the delegate standing in for them right now, for this workflow}."""
    if not user_ids:
        return {}
    now = _now()
    rows = await db.execute(
        select(WorkflowDelegation.user_id, WorkflowDelegation.delegate_id)
        .where(
            WorkflowDelegation.tenant_id == tenant_id,
            WorkflowDelegation.user_id.in_([uuid.UUID(u) for u in user_ids]),
            WorkflowDelegation.starts_at <= now,
            WorkflowDelegation.ends_at > now,
            or_(WorkflowDelegation.workflow_id.is_(None), WorkflowDelegation.workflow_id == workflow_id),
        )
        .order_by(WorkflowDelegation.created_at.desc())
    )
    out: dict[str, str] = {}
    for user_id, delegate_id in rows.all():
        out.setdefault(str(user_id), str(delegate_id))
    return out


async def _notify(run: _Run, user_ids: list[str], title: str, message: str, source: str = "workflow_task") -> None:
    """In-app notifications to these people — and to whoever stands in for them."""
    recipients = set(user_ids)
    recipients |= set((await delegates_of(run.db, run.tenant_id, list(recipients), run.instance.workflow_id)).values())
    for uid in sorted(recipients):
        run.db.add(
            Notification(
                tenant_id=run.tenant_id,
                recipient_user_id=uuid.UUID(uid),
                title=title[:200],
                message=message[:1000],
                source=source,
                workflow_instance_id=run.instance.id,
            )
        )


def _title(run: _Run, step: dict) -> str:
    return f"{run.workflow.name} · {step.get('label') or step.get('key')}"


# ---- moving tokens --------------------------------------------------------------------------------


def _fail(run: _Run, token: dict, reason: str) -> None:
    token["state"] = "failed"
    token["error"] = reason[:1000]
    token["failed_at"] = _iso(_now())
    run.history(token["step"], "incident", reason[:2000], actor_id=None)


def _clear_step_fields(token: dict) -> None:
    for key in ("due_at", "remind_at", "reminded", "escalated", "voters", "votes", "claimed_by", "assignees", "wake_at", "child_ids", "error", "failed_at", "job_id", "attempts"):
        token.pop(key, None)


async def _enter(run: _Run, token: dict, key: str) -> None:
    """Puts `token` on step `key` and does what that step does."""
    run.moves += 1
    if run.moves > MAX_AUTO_MOVES:
        _fail(run, token, f"Stopped after {MAX_AUTO_MOVES} automatic moves — the workflow seems to loop without waiting anywhere.")
        return
    step = run.step(key)
    token["step"] = key
    token["entered_at"] = _iso(_now())
    token["state"] = "active"
    _clear_step_fields(token)
    if step is None:
        _fail(run, token, f"The step “{key}” doesn't exist in this version of the workflow.")
        return
    await dispatch_step_notifications(run.db, run.tenant_id, run.workflow, key, "on_enter", run.instance)
    kind = step.get("type")

    if is_human(step):
        await _wait_for_people(run, token, step)
    elif kind in ("start", "decision"):
        to = route(step, run.context)
        if to is None:
            _fail(run, token, f"No branch of “{step.get('label') or key}” matched, and it has no “otherwise” branch.")
            return
        await _leave(run, token, to, "routed" if kind == "decision" else "submitted", actor_id=None if kind == "decision" else True)
    elif kind == "parallel":
        await _split(run, token, step)
    elif kind == "join":
        await _arrive(run, token, step)
    elif kind == "timer":
        await _start_timer(run, token, step)
    elif kind == "subprocess":
        await _start_children(run, token, step)
    elif kind == "service":
        await _start_service(run, token, step)
    elif kind == "end":
        await _finish_token(run, token, step)
    else:
        _fail(run, token, f"Unknown step type “{kind}”.")


async def _leave(
    run: _Run,
    token: dict,
    to: str,
    action: str,
    *,
    comment: str | None = None,
    actor_id: uuid.UUID | None | bool = True,
    values: dict | None = None,
    strict: bool = False,
) -> None:
    """Moves `token` from its step to `to`: records it, takes the leaving
    step's outputs (from `values`, a person's input) into the context, checks
    the next step's inputs, fires on_complete, then enters `to`. `strict`
    (a person's move) refuses with 422 instead of raising an incident."""
    leaving = run.step(token["step"])
    entering = run.step(to)
    context = dict(run.context)
    if values:
        if leaving and leaving.get("output_params"):
            ok, detail, written = apply_output_params(leaving["output_params"], values)
            if not ok:
                if strict:
                    raise EngineError(422, detail)
                _fail(run, token, detail)
                return
            context.update(written)
        else:
            context.update(values)
    if entering and entering.get("input_params"):
        ok, detail, resolved = resolve_input_params(entering["input_params"], context)
        if not ok:
            if strict:
                raise EngineError(422, detail)
            _fail(run, token, f"Can't enter “{entering.get('label') or to}”: {detail}")
            return
        context.update(resolved)
    run.context = context
    run.history(token["step"], action, comment, actor_id=actor_id)
    await dispatch_step_notifications(run.db, run.tenant_id, run.workflow, token["step"], "on_complete", run.instance)
    await _enter(run, token, to)


async def _wait_for_people(run: _Run, token: dict, step: dict) -> None:
    token["state"] = "active"
    entered = _now()
    if step.get("sla_hours"):
        due = add_hours(entered, float(step["sla_hours"]), run.business_days)
        token["due_at"] = _iso(due)
        if step.get("remind_before_hours"):
            remind = due - timedelta(hours=float(step["remind_before_hours"]))
            if remind > entered:
                token["remind_at"] = _iso(remind)
    assignees = await resolve_assignees(run.db, run.tenant_id, step, token, run.context)
    if (step.get("approval_mode") or "any") != "any" and step.get("type") != "start":
        token["voters"] = assignees
        token["votes"] = []
    if not assignees:
        run.history(token["step"], "unassigned", "Nobody can act on this step — no active user holds its approver roles or is named for it.", actor_id=None)
        return
    due = f" Due {token['due_at'][:16].replace('T', ' ')} UTC." if token.get("due_at") else ""
    await _notify(run, assignees, f"Waiting on you: {_title(run, step)}", f"“{step.get('label') or step['key']}” in {run.workflow.name} is waiting on you.{due}")


async def _split(run: _Run, token: dict, step: dict) -> None:
    targets = [t["to"] for t in step.get("next_steps") or [] if t.get("to") and _holds(t, run.context)]
    if not targets:
        _fail(run, token, f"No branch of “{step.get('label') or step['key']}” applies.")
        return
    split_id = uuid.uuid4().hex[:12]
    # "outer": the splits the branches return to once they've joined.
    run.joins[split_id] = {"expected": len(targets), "arrived": 0, "fired": False, "split": step["key"], "outer": list(token.get("splits") or [])}
    run.history(step["key"], "split", f"{len(targets)} branches", actor_id=None)
    run.tokens.remove(token)
    branches = [_new_token(step["key"], [*token.get("splits", []), split_id]) for _ in targets]
    run.tokens.extend(branches)
    for branch, to in zip(branches, targets):
        if branch in run.tokens:
            await _enter(run, branch, to)


def _drop_branch(run: _Run, split_id: str) -> None:
    """A branch of `split_id` ended without reaching its join: expect one fewer."""
    j = run.joins.get(split_id)
    if j and not j.get("fired"):
        j["expected"] = max(0, int(j.get("expected", 1)) - 1)


async def _continue_after_join(run: _Run, split_id: str) -> None:
    j = run.joins[split_id]
    j["fired"] = True
    join_step = run.step(j.get("join"))
    token = _new_token(j["join"], list(j.get("outer") or []))
    run.tokens.append(token)
    to = route(join_step, run.context) if join_step else None
    if to is None:
        _fail(run, token, f"“{join_step.get('label') if join_step else j.get('join')}” has no next step.")
        return
    await _leave(run, token, to, "joined", actor_id=None)


async def _arrive(run: _Run, token: dict, step: dict) -> None:
    splits = list(token.get("splits") or [])
    j = run.joins.get(splits[-1]) if splits else None
    if j is None:
        # Not inside a parallel split: a join just passes through.
        to = route(step, run.context)
        if to is None:
            _fail(run, token, f"“{step.get('label') or step['key']}” has no next step.")
            return
        await _leave(run, token, to, "joined", actor_id=None)
        return
    split_id = splits[-1]
    run.tokens.remove(token)
    j["arrived"] = int(j.get("arrived", 0)) + 1
    j["join"] = step["key"]
    if j.get("fired"):
        run.history(step["key"], "branch_dropped", "A later branch arrived after the join had already moved on.", actor_id=None)
        return
    if (step.get("join_mode") or "all") == "any":
        # The first branch wins: the others stop where they are.
        for other in [t for t in run.tokens if split_id in (t.get("splits") or [])]:
            await _discard(run, other, "Another branch reached the join first.")
        await _continue_after_join(run, split_id)
    elif j["arrived"] >= int(j.get("expected", 0)):
        await _continue_after_join(run, split_id)
    else:
        run.history(step["key"], "waiting", f"{j['arrived']} of {j['expected']} branches have arrived", actor_id=None)


async def _discard(run: _Run, token: dict, reason: str) -> None:
    """Stops a branch: cancels the children it waits on and removes it."""
    for child_id in token.get("child_ids") or []:
        await _cancel_child(run, uuid.UUID(child_id), reason)
    if token in run.tokens:
        run.tokens.remove(token)
    run.history(token["step"], "branch_cancelled", reason, actor_id=None)


async def _start_timer(run: _Run, token: dict, step: dict) -> None:
    cfg = step.get("timer") or {}
    wake: datetime | None = None
    if (cfg.get("mode") or "duration") == "until_field":
        name = cfg.get("field")
        wake = _parse_dt(run.context.get(name)) if name else None
        if wake is None:
            _fail(run, token, f"The timer waits until the date in “{name or '?'}”, but the instance has no date there.")
            return
    else:
        hours = cfg.get("hours")
        if hours is None:
            _fail(run, token, "The timer has no duration.")
            return
        wake = add_hours(_now(), float(hours), run.business_days)
    if wake <= _now():
        to = route(step, run.context)
        if to is None:
            _fail(run, token, f"“{step.get('label') or step['key']}” has no next step.")
            return
        await _leave(run, token, to, "timer_elapsed", actor_id=None)
        return
    token["state"] = "waiting"
    token["wake_at"] = _iso(wake)


async def _depth(run: _Run) -> int:
    depth, parent_id = 0, run.instance.parent_instance_id
    while parent_id is not None and depth <= MAX_DEPTH:
        depth += 1
        parent_id = (await run.db.execute(select(WorkflowInstance.parent_instance_id).where(WorkflowInstance.id == parent_id))).scalar_one_or_none()
    return depth


async def _start_children(run: _Run, token: dict, step: dict) -> None:
    cfg = step.get("subprocess") or {}
    if not cfg.get("workflow_id"):
        _fail(run, token, "The sub-workflow step doesn't say which workflow to start.")
        return
    if await _depth(run) >= MAX_DEPTH:
        _fail(run, token, f"Sub-workflows are nested more than {MAX_DEPTH} deep.")
        return
    base = {child_key: run.context.get(parent_key or child_key) for child_key, parent_key in (cfg.get("input_map") or {}).items()}
    if cfg.get("for_each"):
        items = run.context.get(cfg["for_each"])
        if not isinstance(items, list):
            _fail(run, token, f"“{cfg['for_each']}” isn't a list, so there's nothing to start a sub-workflow for.")
            return
        contexts = [{**base, "item": item, "item_index": i} for i, item in enumerate(items)]
    else:
        contexts = [base]
    wait = cfg.get("wait", True)
    token["state"] = "waiting"
    token["child_ids"] = []
    for child_context in contexts:
        try:
            child = await start_instance(
                run.db,
                run.tenant_id,
                run.actor_id,
                workflow_id=uuid.UUID(str(cfg["workflow_id"])),
                vendor_id=run.instance.vendor_id,
                context=child_context,
                parent_instance_id=run.instance.id,
                parent_token_id=token["id"] if wait else DETACHED,
            )
        except EngineError as exc:
            _fail(run, token, f"Couldn't start the sub-workflow: {exc.detail}")
            return
        token["child_ids"].append(str(child.id))
    run.history(step["key"], "subprocess_started", f"{len(contexts)} sub-workflow instance(s)", actor_id=None)
    if not wait or not contexts:
        await _resume_from_children(run, token, step, [])
        return
    # Children that finished straight away (no step waited) count already.
    children = await _children(run, token)
    if all(c.status in TERMINAL for c in children):
        await _resume_from_children(run, token, step, children)


async def _children(run: _Run, token: dict) -> list[WorkflowInstance]:
    ids = [uuid.UUID(c) for c in token.get("child_ids") or []]
    if not ids:
        return []
    rows = (await run.db.execute(select(WorkflowInstance).where(WorkflowInstance.tenant_id == run.tenant_id, WorkflowInstance.id.in_(ids)))).scalars().all()
    order = {i: n for n, i in enumerate(ids)}
    return sorted(rows, key=lambda c: order[c.id])


async def _resume_from_children(run: _Run, token: dict, step: dict, children: list[WorkflowInstance]) -> None:
    cfg = step.get("subprocess") or {}
    cancelled = [c for c in children if c.status == "cancelled"]
    if cancelled:
        _fail(run, token, f"{len(cancelled)} sub-workflow instance(s) were cancelled.")
        return
    for parent_key, child_key in (cfg.get("output_map") or {}).items():
        values = [(c.context or {}).get(child_key or parent_key) for c in children]
        if cfg.get("for_each"):
            run.context[parent_key] = values
        elif values:
            run.context[parent_key] = values[0]
    to = route(step, run.context)
    if to is None:
        _fail(run, token, f"“{step.get('label') or step['key']}” has no next step.")
        return
    await _leave(run, token, to, "subprocess_completed", actor_id=None)


async def _start_service(run: _Run, token: dict, step: dict) -> None:
    cfg = step.get("service") or {}
    if not cfg.get("task_id"):
        _fail(run, token, "The service step doesn't say which Task to run.")
        return
    token["state"] = "waiting"
    token["job_id"] = str(
        await jobs.enqueue(
            run.db,
            SERVICE_JOB,
            {"instance_id": str(run.instance.id), "token_id": token["id"], "step": step["key"]},
            tenant_id=run.tenant_id,
            dedupe_key=f"{SERVICE_JOB}:{token['id']}",
            max_attempts=int(cfg.get("retries") or 0) + 1,
        )
    )


async def _finish_token(run: _Run, token: dict, step: dict) -> None:
    if token in run.tokens:
        run.tokens.remove(token)
    if step.get("terminate"):
        for other in list(run.tokens):
            await _discard(run, other, f"“{step.get('label') or step['key']}” ended the whole instance.")
    elif token.get("splits"):
        _drop_branch(run, token["splits"][-1])
        await _check_joins(run)
    if not run.tokens:
        await _complete(run, step["key"])


async def _check_joins(run: _Run) -> None:
    """After a branch ended early: a join now has every remaining branch."""
    for split_id, j in list(run.joins.items()):
        if not j.get("fired") and j.get("join") and int(j.get("arrived", 0)) > 0 and int(j["arrived"]) >= int(j.get("expected", 0)):
            await _continue_after_join(run, split_id)


async def _open_legacy_children(run: _Run) -> bool:
    """Children spawned by the old advance(spawn=…) — the instance waits for them to finish."""
    row = (
        await run.db.execute(
            select(WorkflowInstance.id)
            .where(
                WorkflowInstance.tenant_id == run.tenant_id,
                WorkflowInstance.parent_instance_id == run.instance.id,
                WorkflowInstance.parent_token_id.is_(None),
                WorkflowInstance.status.notin_(TERMINAL),
            )
            .limit(1)
        )
    ).first()
    return row is not None


async def _complete(run: _Run, end_key: str) -> None:
    run.instance.current_step = end_key
    if await _open_legacy_children(run):
        run.history(end_key, "waiting_for_children", "Every step is done; waiting for the child instances.", actor_id=None)
        return
    run.instance.status = "completed"
    run.instance.completed_at = _now()
    run.history(end_key, "completed", None, actor_id=None)
    if run.instance.parent_instance_id is not None:
        run.finished_children.append(run.instance)


async def _cancel_child(run: _Run, child_id: uuid.UUID, reason: str) -> None:
    child = await _lock(run.db, run.tenant_id, child_id)
    if child.status in TERMINAL:
        return
    child_run = await _open_run(run.db, run.tenant_id, child, run.actor_id)
    await _cancel(child_run, reason, notify_parent=False)
    await _save(child_run)


async def _cancel(run: _Run, reason: str, *, notify_parent: bool = True) -> None:
    for token in list(run.tokens):
        for child_id in token.get("child_ids") or []:
            await _cancel_child(run, uuid.UUID(child_id), reason)
    # Children spawned without a waiting token go too.
    open_children = (
        await run.db.execute(
            select(WorkflowInstance.id).where(
                WorkflowInstance.tenant_id == run.tenant_id,
                WorkflowInstance.parent_instance_id == run.instance.id,
                WorkflowInstance.status.notin_(TERMINAL),
            )
        )
    ).scalars().all()
    for child_id in open_children:
        await _cancel_child(run, child_id, reason)
    run.tokens = []
    run.instance.status = "cancelled"
    run.instance.completed_at = _now()
    run.instance.cancel_reason = reason
    run.history(run.instance.current_step, "cancelled", reason)
    if notify_parent and run.instance.parent_instance_id is not None:
        run.finished_children.append(run.instance)


async def _commit(run: _Run) -> None:
    """Saves the run, then lets any parent waiting on a child that just
    finished move on (each under its own lock)."""
    await _save(run)
    finished, run.finished_children = run.finished_children, []
    for child in finished:
        await _child_finished(run.db, run.tenant_id, child, run.actor_id)


async def _child_finished(db: AsyncSession, tenant_id: uuid.UUID, child: WorkflowInstance, actor_id: uuid.UUID | None) -> None:
    if child.parent_instance_id is None or child.parent_token_id == DETACHED:
        return
    if child.parent_instance_id in _busy(db):
        return  # the parent is mid-run (it started this child) and checks it itself
    parent = await _lock(db, tenant_id, child.parent_instance_id)
    if parent.status != "in_progress":
        return
    run = await _open_run(db, tenant_id, parent, actor_id)
    _busy(db).add(parent.id)
    try:
        if child.parent_token_id is None:
            # Spawned by the old advance(spawn=…): the parent completes once
            # its own path is done and every child is.
            step = run.step(parent.current_step)
            if not run.tokens and step and step.get("type") == "end" and not await _open_legacy_children(run):
                parent.status = "completed"
                parent.completed_at = _now()
                run.history(parent.current_step, "auto_completed", "All child instances completed")
                if parent.parent_instance_id is not None:
                    run.finished_children.append(parent)
        else:
            token = run.token(child.parent_token_id)
            if token and token.get("state") == "waiting":
                children = await _children(run, token)
                if all(c.status in TERMINAL for c in children):
                    await _resume_from_children(run, token, run.step(token["step"]) or {}, children)
    finally:
        _busy(db).discard(parent.id)
    await _commit(run)


async def _locked_run(db: AsyncSession, tenant_id: uuid.UUID, instance_id: uuid.UUID, actor_id: uuid.UUID | None) -> _Run:
    instance = await _lock(db, tenant_id, instance_id)
    return await _open_run(db, tenant_id, instance, actor_id)


class _Running:
    """Marks an instance as mid-run for the length of a block."""

    def __init__(self, run: _Run):
        self.run = run

    async def __aenter__(self) -> _Run:
        _busy(self.run.db).add(self.run.instance.id)
        return self.run

    async def __aexit__(self, *exc) -> None:
        _busy(self.run.db).discard(self.run.instance.id)


def _require_running(run: _Run) -> None:
    if run.instance.status != "in_progress":
        raise EngineError(409, f"This instance is {run.instance.status}.")
    if run.instance.suspended_at is not None:
        raise EngineError(409, "This instance is suspended — resume it first.")


# ---- starting ---------------------------------------------------------------------------------------


async def current_published(db: AsyncSession, tenant_id: uuid.UUID, workflow_id: uuid.UUID) -> Workflow:
    workflow = (
        await db.execute(select(Workflow).where(Workflow.tenant_id == tenant_id, Workflow.workflow_id == workflow_id, Workflow.is_current.is_(True)))
    ).scalar_one_or_none()
    if workflow is None:
        raise EngineError(404, "Workflow not found")
    if workflow.status != "published":
        raise EngineError(409, "Workflow's current version is not published")
    return workflow


async def start_instance(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    actor_id: uuid.UUID | None,
    *,
    workflow_id: uuid.UUID,
    initial_step: str | None = None,
    vendor_id: uuid.UUID | None = None,
    sla_due_at: datetime | None = None,
    context: dict | None = None,
    parent_instance_id: uuid.UUID | None = None,
    parent_token_id: str | None = None,
    sla_business_days: bool = False,
) -> WorkflowInstance:
    """Starts an instance of the workflow's current published version and
    runs it until every branch waits (for a person, a timer, a child, a
    Task) or ends. The Start step's input_params are checked first; its
    sla_hours, if any, sets the whole instance's deadline."""
    workflow = await current_published(db, tenant_id, workflow_id)
    start = find_step(workflow, initial_step) if initial_step else next((s for s in workflow.step_definitions or [] if s.get("type") == "start"), None)
    if start is None:
        raise EngineError(422, f"The workflow has no step “{initial_step}”." if initial_step else "The workflow has no Start step.")
    merged = dict(context or {})
    ok, detail, resolved = resolve_input_params(start.get("input_params") or [], merged)
    if not ok:
        raise EngineError(422, detail)
    merged.update(resolved)
    business = sla_business_days or bool((workflow.settings or {}).get("business_days"))
    if sla_due_at is None and start.get("sla_hours"):
        sla_due_at = add_hours(_now(), float(start["sla_hours"]), business)

    instance = WorkflowInstance(
        tenant_id=tenant_id,
        vendor_id=vendor_id,
        workflow_id=workflow.workflow_id,
        workflow_version=workflow.version,
        current_step=start["key"],
        status="in_progress",
        sla_due_at=sla_due_at,
        started_by=actor_id,
        context=merged,
        parent_instance_id=parent_instance_id,
        parent_token_id=parent_token_id,
        tokens=[],
        join_state={},
    )
    db.add(instance)
    await db.flush()
    run = _Run(db, tenant_id, workflow, instance, actor_id, context=merged)
    run.history(start["key"], "started")
    token = _new_token(start["key"], [])
    run.tokens.append(token)
    async with _Running(run):
        await _enter(run, token, start["key"])
    await _commit(run)
    await db.refresh(instance)
    return instance


# ---- people acting ------------------------------------------------------------------------------------


@dataclass
class Actor:
    user_id: uuid.UUID
    roles: set[str]
    vendor_id: uuid.UUID | None = None


async def acting_as(db: AsyncSession, tenant_id: uuid.UUID, run_or_instance, step: dict, token: dict, actor: Actor, context: dict) -> str | None:
    """Whose turn the actor takes on this token: their own (they're an
    assignee), or a colleague's they stand in for — or None."""
    instance = run_or_instance.instance if isinstance(run_or_instance, _Run) else run_or_instance
    users, roles = assignment(step, token, context)
    me = str(actor.user_id)
    claimed = token.get("claimed_by")

    def eligible(user: str, user_roles: set[str]) -> bool:
        return user in users or bool(user_roles & set(roles))

    # A Portal account acts through a role only within its own organisation.
    named = me in users
    if actor.vendor_id is not None and not named and instance.vendor_id != actor.vendor_id:
        return None
    if eligible(me, actor.roles) and (not claimed or claimed == me):
        return me
    for user_id, workflows in (await delegators_for(db, tenant_id, actor.user_id, instance.workflow_id)).items():
        if None not in workflows and str(instance.workflow_id) not in workflows:
            continue
        their_roles = {str(r) for (r,) in (await db.execute(select(UserRole.role_id).join(Role, Role.id == UserRole.role_id).where(UserRole.user_id == uuid.UUID(user_id), Role.tenant_id == tenant_id))).all()}
        if eligible(user_id, their_roles) and (not claimed or claimed == user_id):
            return user_id
    return None


async def act(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    actor: Actor,
    instance_id: uuid.UUID,
    *,
    to: str,
    token_id: str | None = None,
    comment: str | None = None,
    values: dict | None = None,
) -> WorkflowInstance:
    """A person's decision on a step waiting on them: one of its open
    choices. With "all" or "majority" approval it's a vote, and the step
    moves on once the votes decide it."""
    run = await _locked_run(db, tenant_id, instance_id, actor.user_id)
    _require_running(run)
    token = run.token(token_id) if token_id else next((t for t in run.tokens if t.get("state") == "active" and is_human(run.step(t["step"]))), None)
    if token is None or token.get("state") != "active":
        raise EngineError(404, "This item isn't waiting on you (anymore).")
    step = run.step(token["step"])
    if not is_human(step):
        raise EngineError(404, "This item isn't waiting on you (anymore).")
    as_user = await acting_as(db, tenant_id, run, step, token, actor, run.context)
    if as_user is None:
        if token.get("claimed_by") and token["claimed_by"] != str(actor.user_id):
            raise EngineError(409, "Someone else has claimed this item.")
        raise EngineError(404, "This item isn't waiting on you (anymore).")
    context = {**run.context, **(values or {})}
    choices = open_choices(step, context)
    if not any(t["to"] == to for t in choices):
        raise EngineError(422, "That isn't one of this step's next steps.")
    target = run.step(to)
    label = (target.get("label") if target else None) or to
    on_behalf = None if as_user == str(actor.user_id) else as_user
    note = comment
    if on_behalf:
        name = (await db.execute(select(User.name).where(User.id == uuid.UUID(on_behalf)))).scalar_one_or_none()
        note = f"{comment} (on behalf of {name})" if comment else f"On behalf of {name}"

    async with _Running(run):
        mode = step.get("approval_mode") or "any"
        voters = token.get("voters") or []
        if mode == "any" or not voters:
            await _leave(run, token, to, label.lower(), comment=note, values=values, strict=True)
        else:
            votes = [v for v in token.get("votes") or [] if v.get("user_id") != as_user]
            votes.append({"user_id": as_user, "by": str(actor.user_id), "to": to, "comment": comment, "at": _iso(_now())})
            token["votes"] = votes
            if values:
                run.context.update(values)
            run.history(token["step"], f"voted: {label.lower()}", note)
            decided = _tally(mode, [c["to"] for c in choices], voters, votes)
            if decided:
                decided_label = ((run.step(decided) or {}).get("label") or decided).lower()
                await _leave(run, token, decided, decided_label, comment=f"Decided by {len(votes)} of {len(voters)} vote(s)", actor_id=None)
    await _commit(run)
    await db.refresh(run.instance)
    return run.instance


def _tally(mode: str, choices: list[str], voters: list[str], votes: list[dict]) -> str | None:
    """Which choice the votes decide, if they do yet.
    all — the first choice needs everyone; any other choice wins at once.
    majority — a choice wins with more than half; a full tie goes to the last choice."""
    counted = [v for v in votes if v.get("user_id") in voters] or votes
    if not choices:
        return None
    if mode == "all":
        other = next((v["to"] for v in counted if v["to"] != choices[0]), None)
        if other:
            return other
        return choices[0] if len({v["user_id"] for v in counted}) >= len(voters) else None
    tally: dict[str, int] = {}
    for v in counted:
        tally[v["to"]] = tally.get(v["to"], 0) + 1
    for choice, n in tally.items():
        if n * 2 > len(voters):
            return choice
    if len({v["user_id"] for v in counted}) >= len(voters):
        return choices[-1]
    return None


async def claim(db: AsyncSession, tenant_id: uuid.UUID, actor: Actor, instance_id: uuid.UUID, token_id: str, release: bool = False) -> WorkflowInstance:
    """Takes a group task for yourself (others stop seeing it), or gives it back."""
    run = await _locked_run(db, tenant_id, instance_id, actor.user_id)
    _require_running(run)
    token = run.token(token_id)
    step = run.step(token["step"]) if token else None
    if token is None or token.get("state") != "active" or not is_human(step):
        raise EngineError(404, "This item isn't waiting on you (anymore).")
    me = str(actor.user_id)
    if release:
        if token.get("claimed_by") != me:
            raise EngineError(409, "You haven't claimed this item.")
        token.pop("claimed_by", None)
        run.history(token["step"], "released")
    else:
        if await acting_as(db, tenant_id, run, step, token, actor, run.context) is None:
            raise EngineError(409 if token.get("claimed_by") else 404, "Someone else has claimed this item." if token.get("claimed_by") else "This item isn't waiting on you (anymore).")
        token["claimed_by"] = me
        run.history(token["step"], "claimed")
    await _commit(run)
    await db.refresh(run.instance)
    return run.instance


async def delegate(
    db: AsyncSession, tenant_id: uuid.UUID, actor: Actor, instance_id: uuid.UUID, token_id: str, to_user: uuid.UUID, comment: str | None
) -> WorkflowInstance:
    """An assignee hands this one item to a colleague."""
    run = await _locked_run(db, tenant_id, instance_id, actor.user_id)
    _require_running(run)
    token = run.token(token_id)
    step = run.step(token["step"]) if token else None
    if token is None or token.get("state") != "active" or not is_human(step):
        raise EngineError(404, "This item isn't waiting on you (anymore).")
    if await acting_as(db, tenant_id, run, step, token, actor, run.context) is None:
        raise EngineError(404, "This item isn't waiting on you (anymore).")
    target = (await db.execute(select(User).where(User.tenant_id == tenant_id, User.id == to_user, User.status == "active"))).scalar_one_or_none()
    if target is None:
        raise EngineError(422, "Pick an active user to hand this to.")
    if target.id == actor.user_id:
        raise EngineError(422, "That's you.")
    _reassign_token(token, [str(target.id)], [])
    run.history(token["step"], "delegated", f"To {target.name}" + (f": {comment}" if comment else ""))
    await _notify(run, [str(target.id)], f"Handed to you: {_title(run, step)}", f"“{step.get('label') or step['key']}” in {run.workflow.name} was handed to you." + (f" “{comment}”" if comment else ""))
    await _commit(run)
    await db.refresh(run.instance)
    return run.instance


def _reassign_token(token: dict, user_ids: list[str], role_ids: list[str]) -> None:
    token["assignees"] = {"user_ids": user_ids, "role_ids": role_ids}
    token.pop("claimed_by", None)


# ---- administrators -----------------------------------------------------------------------------------


async def admin_cancel(db: AsyncSession, tenant_id: uuid.UUID, actor_id: uuid.UUID | None, instance_id: uuid.UUID, reason: str) -> WorkflowInstance:
    run = await _locked_run(db, tenant_id, instance_id, actor_id)
    if run.instance.status != "in_progress":
        raise EngineError(409, f"This instance is {run.instance.status}.")
    async with _Running(run):
        await _cancel(run, reason)
    await _commit(run)
    await db.refresh(run.instance)
    return run.instance


async def admin_suspend(db: AsyncSession, tenant_id: uuid.UUID, actor_id: uuid.UUID | None, instance_id: uuid.UUID, resume: bool, comment: str | None = None) -> WorkflowInstance:
    run = await _locked_run(db, tenant_id, instance_id, actor_id)
    if run.instance.status != "in_progress":
        raise EngineError(409, f"This instance is {run.instance.status}.")
    if resume:
        if run.instance.suspended_at is None:
            raise EngineError(409, "This instance isn't suspended.")
        run.instance.suspended_at = None
        run.history(run.instance.current_step, "resumed", comment)
    else:
        if run.instance.suspended_at is not None:
            raise EngineError(409, "This instance is already suspended.")
        run.instance.suspended_at = _now()
        run.history(run.instance.current_step, "suspended", comment)
    await _commit(run)
    await db.refresh(run.instance)
    return run.instance


def _admin_token(run: _Run, token_id: str) -> dict:
    _require_running(run)
    token = run.token(token_id)
    if token is None:
        raise EngineError(404, "That branch isn't on this instance (anymore).")
    return token


async def admin_reassign(
    db: AsyncSession, tenant_id: uuid.UUID, actor_id: uuid.UUID | None, instance_id: uuid.UUID, token_id: str, user_ids: list[uuid.UUID], role_ids: list[uuid.UUID], comment: str | None
) -> WorkflowInstance:
    run = await _locked_run(db, tenant_id, instance_id, actor_id)
    token = _admin_token(run, token_id)
    step = run.step(token["step"])
    if token.get("state") != "active" or not is_human(step):
        raise EngineError(409, "Only a step waiting for people can be reassigned.")
    if not user_ids and not role_ids:
        raise EngineError(422, "Pick at least one user or role.")
    _reassign_token(token, [str(u) for u in user_ids], [str(r) for r in role_ids])
    if token.get("voters") is not None:
        token["voters"] = await resolve_assignees(db, tenant_id, step, token, run.context)
    run.history(token["step"], "reassigned", comment)
    people = await resolve_assignees(db, tenant_id, step, token, run.context)
    await _notify(run, people, f"Assigned to you: {_title(run, step)}", f"“{step.get('label') or step['key']}” in {run.workflow.name} was assigned to you.")
    await _commit(run)
    await db.refresh(run.instance)
    return run.instance


async def admin_retry(db: AsyncSession, tenant_id: uuid.UUID, actor_id: uuid.UUID | None, instance_id: uuid.UUID, token_id: str, comment: str | None) -> WorkflowInstance:
    """Runs an incident's step again (after fixing what made it fail)."""
    run = await _locked_run(db, tenant_id, instance_id, actor_id)
    token = _admin_token(run, token_id)
    if token.get("state") != "failed":
        raise EngineError(409, "Only a failed branch can be retried.")
    run.history(token["step"], "retried", comment)
    async with _Running(run):
        await _enter(run, token, token["step"])
    await _commit(run)
    await db.refresh(run.instance)
    return run.instance


async def admin_skip(db: AsyncSession, tenant_id: uuid.UUID, actor_id: uuid.UUID | None, instance_id: uuid.UUID, token_id: str, comment: str | None) -> WorkflowInstance:
    """Moves a waiting or failed branch on as if its step had finished."""
    run = await _locked_run(db, tenant_id, instance_id, actor_id)
    token = _admin_token(run, token_id)
    step = run.step(token["step"]) or {}
    if token.get("state") == "active" and is_human(step):
        raise EngineError(409, "This step waits for a person — move or reassign it instead.")
    to = route(step, run.context) or next((t["to"] for t in step.get("next_steps") or [] if t.get("to")), None)
    async with _Running(run):
        for child_id in token.get("child_ids") or []:
            await _cancel_child(run, uuid.UUID(child_id), "Skipped by an administrator.")
        if to is None:
            run.history(token["step"], "skipped", comment)
            await _finish_token(run, token, {**step, "key": token["step"], "terminate": False})
        else:
            await _leave(run, token, to, "skipped", comment=comment)
    await _commit(run)
    await db.refresh(run.instance)
    return run.instance


async def admin_move(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    actor_id: uuid.UUID | None,
    instance_id: uuid.UUID,
    *,
    to: str,
    token_id: str | None = None,
    action: str = "moved",
    comment: str | None = None,
    values: dict | None = None,
    complete: bool = False,
    spawn: list | None = None,
) -> WorkflowInstance:
    """Moves a branch to any step of the workflow — an administrator's
    correction, and the Workflow Designer's test run. complete=True on a
    step that isn't an end finishes the branch there (the old "Mark
    complete"). spawn: the old way of starting child instances, which the
    instance then waits for before it completes."""
    run = await _locked_run(db, tenant_id, instance_id, actor_id)
    _require_running(run)
    token = run.token(token_id)
    if token is None:
        raise EngineError(409, "This instance has no active branch to move.")
    if run.step(to) is None:
        raise EngineError(422, f"The workflow has no step “{to}”.")
    async with _Running(run):
        for child in spawn or []:
            await start_instance(
                db,
                tenant_id,
                actor_id,
                workflow_id=child.workflow_id,
                initial_step=child.initial_step,
                vendor_id=run.instance.vendor_id,
                context=child.context,
                parent_instance_id=run.instance.id,
            )
        for child_id in token.get("child_ids") or []:
            await _cancel_child(run, uuid.UUID(child_id), "Moved on by an administrator.")
        target = run.step(to) or {}
        if complete and target.get("type") != "end":
            if values:
                leaving = run.step(token["step"])
                if leaving and leaving.get("output_params"):
                    ok, detail, written = apply_output_params(leaving["output_params"], values)
                    if not ok:
                        raise EngineError(422, detail)
                    run.context.update(written)
                else:
                    run.context.update(values)
            run.history(token["step"], action, comment)
            await _finish_token(run, token, {"key": to, "type": "end"})
        else:
            await _leave(run, token, to, action, comment=comment, values=values or None, strict=True)
    await _commit(run)
    await db.refresh(run.instance)
    return run.instance


# ---- background: Task steps, timers, SLAs -----------------------------------------------------------


@jobs.handler(SERVICE_JOB, max_attempts=1)
async def _run_service(job: jobs.Job) -> str:
    from app.crud.task_history import RunTrigger
    from app.crud.tasks import run_task

    payload = job.payload
    async with jobs.tenant_session(job.tenant_id) as db:
        run = await _locked_run(db, job.tenant_id, uuid.UUID(payload["instance_id"]), None)
        token = run.token(payload["token_id"])
        if run.instance.status != "in_progress" or token is None or token.get("state") != "waiting" or token.get("step") != payload.get("step"):
            return "skipped — the instance moved on"
        if run.instance.suspended_at is not None:
            raise jobs.Retry("the instance is suspended", 300)
        step = run.step(token["step"]) or {}
        task_id = (step.get("service") or {}).get("task_id")
        task = (await db.execute(select(Task).where(Task.tenant_id == job.tenant_id, Task.id == uuid.UUID(str(task_id))))).scalar_one_or_none() if task_id else None
        if task is None:
            raise jobs.Permanent("The Task this step runs no longer exists.")
        if task.status != "published":
            raise jobs.Permanent(f"The Task “{task.name}” isn't published.")
        ok, trace, result = await run_task(
            db, job.tenant_id, task, {**run.context, "workflow_instance_id": str(run.instance.id)},
            trigger=RunTrigger("workflow", ref=str(run.instance.id), name=step.get("label")),
        )
        if not ok:
            failed = next((t for t in reversed(trace) if not t.get("ok")), None)
            raise RuntimeError(f"{(failed or {}).get('node_key', '?')}: {(failed or {}).get('detail') or 'failed'}" if failed else f"The Task “{task.name}” failed.")
        if step.get("output_params"):
            good, detail, written = apply_output_params(step["output_params"], result)
            if not good:
                raise jobs.Permanent(detail)
        else:
            written = {k: v for k, v in result.items() if k != "workflow_instance_id" and run.context.get(k) != v}
        run.context.update(written)
        to = route(step, run.context)
        async with _Running(run):
            if to is None:
                _fail(run, token, f"“{step.get('label') or token['step']}” has no next step.")
            else:
                await _leave(run, token, to, "task_completed", comment=f"{task.name}: {len(trace)} node(s) ran", actor_id=None)
        await _commit(run)
    return f"ok — {task.name}"


@jobs.on_failure(SERVICE_JOB)
async def _service_failed(job: jobs.Job, exc: BaseException) -> None:
    async with jobs.tenant_session(job.tenant_id) as db:
        run = await _locked_run(db, job.tenant_id, uuid.UUID(job.payload["instance_id"]), None)
        token = run.token(job.payload["token_id"])
        if token is None or token.get("state") != "waiting" or run.instance.status != "in_progress":
            return
        _fail(run, token, f"The Task failed after {job.attempt} attempt(s): {exc}")
        await _commit(run)


async def due(db: AsyncSession, tenant_id: uuid.UUID, limit: int = 200) -> int:
    """Handles what's due across the tenant's running instances: timers
    that have elapsed, reminders, and SLAs passed (escalation). Returns how
    many instances it touched."""
    now = _now()
    ids = (
        await db.execute(
            select(WorkflowInstance.id)
            .where(
                WorkflowInstance.tenant_id == tenant_id,
                WorkflowInstance.status == "in_progress",
                WorkflowInstance.suspended_at.is_(None),
                WorkflowInstance.next_due_at <= now,
            )
            .order_by(WorkflowInstance.next_due_at)
            .limit(limit)
            .with_for_update(skip_locked=True)
        )
    ).scalars().all()
    for instance_id in ids:
        run = await _locked_run(db, tenant_id, instance_id, None)
        async with _Running(run):
            for token in list(run.tokens):
                if token not in run.tokens:
                    continue
                await _handle_due(run, token, now)
        await _commit(run)
    return len(ids)


async def _handle_due(run: _Run, token: dict, now: datetime) -> None:
    step = run.step(token["step"]) or {}
    label = step.get("label") or token["step"]
    if token.get("state") == "waiting" and token.get("wake_at") and _parse_dt(token["wake_at"]) <= now:
        to = route(step, run.context)
        if to is None:
            _fail(run, token, f"“{label}” has no next step.")
        else:
            await _leave(run, token, to, "timer_elapsed", actor_id=None)
        return
    if token.get("state") != "active":
        return
    if token.get("remind_at") and not token.get("reminded") and _parse_dt(token["remind_at"]) <= now:
        token["reminded"] = True
        people = await resolve_assignees(run.db, run.tenant_id, step, token, run.context)
        await _notify(run, people, f"Reminder: {_title(run, step)}", f"“{label}” in {run.workflow.name} is due {token.get('due_at', '')[:16].replace('T', ' ')} UTC.")
        run.history(token["step"], "reminded", None, actor_id=None)
    if token.get("due_at") and not token.get("escalated") and _parse_dt(token["due_at"]) <= now:
        token["escalated"] = True
        esc = step.get("escalation") or {}
        action = esc.get("action") or "notify"
        people = await resolve_assignees(run.db, run.tenant_id, step, token, run.context)
        extra = set(_ids(esc.get("user_ids"))) | await _role_members(run.db, run.tenant_id, _ids(esc.get("role_ids")))
        if action == "route" and esc.get("route_to") and run.step(esc["route_to"]):
            run.history(token["step"], "escalated", f"Past its SLA — moved on to “{(run.step(esc['route_to']) or {}).get('label') or esc['route_to']}”.", actor_id=None)
            await _notify(run, sorted(set(people) | extra), f"Overdue: {_title(run, step)}", f"“{label}” in {run.workflow.name} passed its SLA and moved on automatically.")
            await _leave(run, token, esc["route_to"], "timed_out", actor_id=None)
            return
        if action == "reassign" and (esc.get("user_ids") or esc.get("role_ids")):
            _reassign_token(token, _ids(esc.get("user_ids")), _ids(esc.get("role_ids")))
            if token.get("voters") is not None:
                token["voters"] = await resolve_assignees(run.db, run.tenant_id, step, token, run.context)
            run.history(token["step"], "escalated", "Past its SLA — reassigned.", actor_id=None)
            await _notify(run, sorted(extra), f"Escalated to you: {_title(run, step)}", f"“{label}” in {run.workflow.name} passed its SLA and is now waiting on you.")
            return
        run.history(token["step"], "escalated", "Past its SLA.", actor_id=None)
        await _notify(run, sorted(set(people) | extra), f"Overdue: {_title(run, step)}", f"“{label}” in {run.workflow.name} passed its SLA.")


@jobs.handler(DUE_JOB, max_attempts=3)
async def _run_due(job: jobs.Job) -> str:
    async with jobs.tenant_session(job.tenant_id) as db:
        n = await due(db, job.tenant_id)
    return f"{n} instance(s)"


async def queue_due(tenant_id: uuid.UUID) -> bool:
    """The scheduler's tick: queue a pass over the tenant's instances when anything is due."""
    async with jobs.tenant_session(tenant_id) as db:
        anything = (
            await db.execute(
                select(WorkflowInstance.id)
                .where(
                    WorkflowInstance.tenant_id == tenant_id,
                    WorkflowInstance.status == "in_progress",
                    WorkflowInstance.suspended_at.is_(None),
                    WorkflowInstance.next_due_at <= _now(),
                )
                .limit(1)
            )
        ).first()
        if anything is None:
            return False
        await jobs.enqueue(db, DUE_JOB, {}, tenant_id=tenant_id, dedupe_key=f"{DUE_JOB}:{tenant_id}")
    return True


# ---- checking a definition before it's published --------------------------------------------------


async def definition_problems(db: AsyncSession, tenant_id: uuid.UUID, workflow: Workflow) -> list[str]:
    """What would stop this workflow from running — empty when it can be published."""
    steps = workflow.step_definitions or []
    problems: list[str] = []
    keys = [s.get("key") for s in steps]
    by_key = {s.get("key"): s for s in steps}
    if len(set(keys)) != len(keys):
        problems.append("Two steps share a key.")
    starts = [s for s in steps if s.get("type") == "start"]
    if len(starts) != 1:
        problems.append("A workflow needs exactly one Start step." if not starts else "A workflow can have only one Start step.")
    if not any(s.get("type") == "end" for s in steps):
        problems.append("Add an End step.")
    for s in steps:
        name = f"“{s.get('label') or s.get('key')}”"
        kind = s.get("type")
        nexts = [t.get("to") for t in s.get("next_steps") or []]
        for to in nexts:
            if not to:
                problems.append(f"{name} has a connection that doesn't go anywhere.")
            elif to not in by_key:
                problems.append(f"{name} connects to a step that no longer exists.")
            elif to == s.get("key") and not is_human(s):
                problems.append(f"{name} connects to itself.")
        if kind != "end" and not nexts:
            problems.append(f"{name} has no next step.")
        if kind == "parallel" and len(nexts) < 2:
            problems.append(f"{name} splits into parallel branches — connect it to at least two steps.")
        if kind == "join" and len(nexts) != 1:
            problems.append(f"{name} should continue to exactly one step.")
        if kind in HUMAN_STEP_TYPES and not (s.get("approver_roles") or s.get("approver_users") or s.get("assignee_field")):
            problems.append(f"{name} waits for a person, but nobody is assigned — pick approver roles or users, or an assignee field.")
        if kind == "timer":
            t = s.get("timer") or {}
            if (t.get("mode") or "duration") == "until_field" and not t.get("field"):
                problems.append(f"{name} waits until a date — pick the field that holds it.")
            if (t.get("mode") or "duration") == "duration" and not t.get("hours"):
                problems.append(f"{name} needs a duration.")
        if kind == "subprocess":
            wf_id = (s.get("subprocess") or {}).get("workflow_id")
            if not wf_id:
                problems.append(f"{name} needs a workflow to start.")
            elif str(wf_id) == str(workflow.workflow_id):
                problems.append(f"{name} starts this same workflow — that would never end.")
            else:
                exists = (await db.execute(select(Workflow.workflow_id).where(Workflow.tenant_id == tenant_id, Workflow.workflow_id == uuid.UUID(str(wf_id))).limit(1))).first()
                if exists is None:
                    problems.append(f"{name} starts a workflow that no longer exists.")
        if kind == "service":
            task_id = (s.get("service") or {}).get("task_id")
            if not task_id:
                problems.append(f"{name} needs a Task to run.")
            else:
                exists = (await db.execute(select(Task.id).where(Task.tenant_id == tenant_id, Task.id == uuid.UUID(str(task_id))))).first()
                if exists is None:
                    problems.append(f"{name} runs a Task that no longer exists.")
        esc = s.get("escalation") or {}
        if esc.get("action") == "route" and esc.get("route_to") not in by_key:
            problems.append(f"{name} escalates to a step that doesn't exist.")
        if esc.get("action") == "reassign" and not (esc.get("user_ids") or esc.get("role_ids")):
            problems.append(f"{name} escalates by reassigning, but to nobody.")
        if esc.get("action") and not s.get("sla_hours") and kind in HUMAN_STEP_TYPES:
            problems.append(f"{name} has an escalation but no SLA to trigger it.")
    return list(dict.fromkeys(problems))


async def open_tasks_for(db: AsyncSession, tenant_id: uuid.UUID, actor: Actor, limit: int = 1000) -> list[tuple[WorkflowInstance, Workflow, dict, dict, str]]:
    """(instance, workflow, step, token, acting-as user id) for every step
    waiting on this person — directly, through a role, or standing in for a
    colleague who's away. Claimed by someone else: not listed."""
    rows = (
        await db.execute(
            select(WorkflowInstance, Workflow)
            .join(
                Workflow,
                and_(Workflow.tenant_id == WorkflowInstance.tenant_id, Workflow.workflow_id == WorkflowInstance.workflow_id, Workflow.version == WorkflowInstance.workflow_version),
            )
            .where(WorkflowInstance.tenant_id == tenant_id, WorkflowInstance.status == "in_progress", WorkflowInstance.suspended_at.is_(None))
            .order_by(WorkflowInstance.started_at.desc())
            .limit(limit)
        )
    ).all()
    delegated = await delegators_for(db, tenant_id, actor.user_id)
    role_cache: dict[str, set[str]] = {}

    async def roles_of(user_id: str) -> set[str]:
        if user_id not in role_cache:
            role_cache[user_id] = {str(r) for (r,) in (await db.execute(select(UserRole.role_id).join(Role, Role.id == UserRole.role_id).where(UserRole.user_id == uuid.UUID(user_id), Role.tenant_id == tenant_id))).all()}
        return role_cache[user_id]

    me = str(actor.user_id)
    out = []
    for instance, workflow in rows:
        tokens = list(instance.tokens or [])
        if not tokens and instance.current_step:
            tokens = [{"id": None, "step": instance.current_step, "state": "active"}]
        for token in tokens:
            step = find_step(workflow, token.get("step"))
            if token.get("state") != "active" or not is_human(step):
                continue
            users, roles = assignment(step, token, instance.context or {})
            claimed = token.get("claimed_by")
            named = me in users
            if actor.vendor_id is not None and not named and instance.vendor_id != actor.vendor_id:
                continue
            as_user = None
            if (named or set(roles) & actor.roles) and (not claimed or claimed == me):
                as_user = me
            else:
                for user_id, workflows in delegated.items():
                    if None not in workflows and str(instance.workflow_id) not in workflows:
                        continue
                    if (user_id in users or set(roles) & await roles_of(user_id)) and (not claimed or claimed == user_id):
                        as_user = user_id
                        break
            if as_user is not None:
                out.append((instance, workflow, step, token, as_user))
    return out