"""The server-run workflow engine (app/crud/workflow_engine.py): routing,
parallel branches, timers, sub-workflows, Task steps, voting, claiming and
delegating, SLA escalation, out-of-office stand-ins, an administrator's
controls, and the checks before publishing."""

import uuid
from datetime import datetime, timedelta, timezone

from sqlalchemy import select

from app.core import jobs
from app.core.auth import create_access_token, hash_password
from app.crud import workflow_engine as engine
from app.models.identity import Role, User, UserRole
from app.models.notification import Notification
from app.models.workflow import WorkflowInstance
from tests.conftest import AsyncSessionLocal


async def _role(tenant_id) -> uuid.UUID:
    async with AsyncSessionLocal() as db:
        async with db.begin():
            role = Role(tenant_id=tenant_id, name=f"Role {uuid.uuid4().hex[:6]}", action_matrix={})
            db.add(role)
            await db.flush()
            return role.id


async def _person(tenant_id, role_id=None):
    async with AsyncSessionLocal() as db:
        async with db.begin():
            user = User(tenant_id=tenant_id, name=f"P {uuid.uuid4().hex[:4]}", email=f"p-{uuid.uuid4().hex[:8]}@example.com", password_hash=hash_password("pytest-password-123"), status="active")
            db.add(user)
            await db.flush()
            if role_id:
                db.add(UserRole(user_id=user.id, role_id=role_id))
            user_id = user.id
    return {"Authorization": f"Bearer {create_access_token(tenant_id=tenant_id, user_id=user_id, vendor_id=None)}"}, user_id


def _step(key, type_, label=None, to=(), **extra):
    nexts = [t if isinstance(t, dict) else {"to": t} for t in to]
    return {"key": key, "type": type_, "label": label or key.title(), "next_steps": nexts, **extra}


def _when(field, op, value):
    return {"driver_field": field, "operator": op, "value": value}


async def _publish(client, headers, steps, settings=None) -> str:
    made = await client.post("/workflows", headers=headers, json={"name": f"WF {uuid.uuid4().hex[:6]}", "step_definitions": steps, "settings": settings or {}})
    assert made.status_code == 201, made.text
    wf_id = made.json()["workflow_id"]
    pub = await client.post(f"/workflows/{wf_id}/publish", headers=headers)
    assert pub.status_code == 200, pub.text
    return wf_id


async def _start(client, headers, wf_id, context=None) -> dict:
    resp = await client.post("/workflow-instances", headers=headers, json={"workflow_id": wf_id, "initial_step": "start", "context": context or {}})
    assert resp.status_code == 201, resp.text
    return resp.json()


async def _get(client, headers, instance_id) -> dict:
    return (await client.get(f"/workflow-instances/{instance_id}", headers=headers)).json()


async def _tasks(client, headers, instance_id) -> list[dict]:
    return [t for t in (await client.get("/approvals", headers=headers)).json() if t["instance_id"] == instance_id]


async def _act(client, headers, instance_id, to, token_id=None, **extra):
    return await client.post(f"/approvals/{instance_id}/act", headers=headers, json={"to": to, "token_id": token_id, **extra})


async def _backdate(instance_id, key: str):
    """Makes `key` (wake_at / due_at / remind_at) of every token already past."""
    past = (datetime.now(timezone.utc) - timedelta(minutes=1)).isoformat()
    async with AsyncSessionLocal() as db:
        async with db.begin():
            inst = await db.get(WorkflowInstance, uuid.UUID(instance_id))
            inst.tokens = [{**t, key: past} if t.get(key) else t for t in inst.tokens]
            inst.next_due_at = datetime.now(timezone.utc) - timedelta(minutes=1)


async def _due(tenant_id):
    async with jobs.tenant_session(tenant_id) as db:
        return await engine.due(db, tenant_id)


async def test_decisions_route_themselves_and_a_dead_end_is_an_incident(client, auth_headers, test_tenant):
    role = await _role(test_tenant["tenant_id"])
    steps = [
        _step("start", "start", to=["check"]),
        _step("check", "decision", to=[{"to": "big", "condition": _when("amount", "gt", 1000)}, {"to": "small", "condition": _when("amount", "lte", 1000)}]),
        _step("big", "approval", to=["done"], approver_roles=[str(role)]),
        _step("small", "end"),
        _step("done", "end"),
    ]
    wf = await _publish(client, auth_headers, steps)
    small = await _start(client, auth_headers, wf, {"amount": 10})
    assert small["status"] == "completed" and small["current_step"] == "small"
    big = await _start(client, auth_headers, wf, {"amount": 5000})
    assert big["status"] == "in_progress" and big["current_step"] == "big"
    history = (await client.get(f"/workflow-instances/{big['id']}/history", headers=auth_headers)).json()
    # Nobody holds the role yet: the step is flagged as unassigned.
    assert [h["action"] for h in history] == ["started", "submitted", "routed", "unassigned"]

    # No branch matches a missing amount: the branch stops as an incident.
    stuck = await _start(client, auth_headers, wf, {})
    assert stuck["has_incident"] is True and stuck["tokens"][0]["state"] == "failed"
    assert "No branch of “Check” matched" in stuck["tokens"][0]["error"]
    listed = (await client.get("/workflow-instances?incident=true", headers=auth_headers)).json()
    assert stuck["id"] in [i["id"] for i in listed]
    # An administrator moves it on by hand.
    moved = await client.post(f"/workflow-instances/{stuck['id']}/move", headers=auth_headers, json={"token_id": stuck["tokens"][0]["id"], "to": "small"})
    assert moved.status_code == 200, moved.text
    assert moved.json()["status"] == "completed" and moved.json()["has_incident"] is False


async def test_parallel_branches_join_when_all_arrive(client, auth_headers, test_tenant):
    tid = test_tenant["tenant_id"]
    legal_role, finance_role = await _role(tid), await _role(tid)
    legal, _ = await _person(tid, legal_role)
    finance, _ = await _person(tid, finance_role)
    steps = [
        _step("start", "start", to=["split"]),
        _step("split", "parallel", to=["legal", "finance", {"to": "extra", "condition": _when("international", "equals", True)}]),
        _step("legal", "approval", to=["join"], approver_roles=[str(legal_role)]),
        _step("finance", "approval", to=["join"], approver_roles=[str(finance_role)]),
        _step("extra", "approval", to=["join"], approver_roles=[str(legal_role)]),
        _step("join", "join", to=["done"]),
        _step("done", "end"),
    ]
    wf = await _publish(client, auth_headers, steps)
    inst = await _start(client, auth_headers, wf, {"international": False})
    # The conditional branch doesn't apply: two branches, both waiting.
    assert sorted(t["step"] for t in inst["tokens"]) == ["finance", "legal"]

    [legal_task] = await _tasks(client, legal, inst["id"])
    assert (await _act(client, legal, inst["id"], "join", legal_task["token_id"])).status_code == 200
    after_one = await _get(client, auth_headers, inst["id"])
    assert after_one["status"] == "in_progress" and [t["step"] for t in after_one["tokens"]] == ["finance"]

    [finance_task] = await _tasks(client, finance, inst["id"])
    done = await _act(client, finance, inst["id"], "join", finance_task["token_id"])
    assert done.json()["status"] == "completed", done.json()["tokens"]
    actions = [h["action"] for h in (await client.get(f"/workflow-instances/{inst['id']}/history", headers=auth_headers)).json()]
    assert "split" in actions and "joined" in actions and actions[-1] == "completed"


async def test_a_first_wins_join_stops_the_other_branches(client, auth_headers, test_tenant):
    tid = test_tenant["tenant_id"]
    role = await _role(tid)
    person, _ = await _person(tid, role)
    steps = [
        _step("start", "start", to=["split"]),
        _step("split", "parallel", to=["a", "b"]),
        _step("a", "approval", to=["join"], approver_roles=[str(role)]),
        _step("b", "approval", to=["join"], approver_roles=[str(role)]),
        _step("join", "join", to=["done"], join_mode="any"),
        _step("done", "end"),
    ]
    wf = await _publish(client, auth_headers, steps)
    inst = await _start(client, auth_headers, wf)
    tasks = await _tasks(client, person, inst["id"])
    assert len(tasks) == 2
    first = next(t for t in tasks if t["step"] == "a")
    resp = await _act(client, person, inst["id"], "join", first["token_id"])
    assert resp.json()["status"] == "completed" and resp.json()["tokens"] == []
    assert await _tasks(client, person, inst["id"]) == []


async def test_all_must_approve_and_one_rejection_decides(client, auth_headers, test_tenant):
    tid = test_tenant["tenant_id"]
    role = await _role(tid)
    a, _ = await _person(tid, role)
    b, _ = await _person(tid, role)
    steps = [
        _step("start", "start", to=["board"]),
        _step("board", "approval", to=["approved", "rejected"], approver_roles=[str(role)], approval_mode="all"),
        _step("approved", "end"),
        _step("rejected", "end"),
    ]
    wf = await _publish(client, auth_headers, steps)

    yes = await _start(client, auth_headers, wf)
    [task] = await _tasks(client, a, yes["id"])
    assert task["approval_mode"] == "all" and task["voters"] == 2
    first = await _act(client, a, yes["id"], "approved", task["token_id"], comment="Fine by me")
    assert first.json()["status"] == "in_progress"
    # Still waiting on the second voter, who sees the first vote.
    [task_b] = await _tasks(client, b, yes["id"])
    assert [(v["to"], v["comment"]) for v in task_b["votes"]] == [("approved", "Fine by me")]
    second = await _act(client, b, yes["id"], "approved", task_b["token_id"])
    assert second.json()["status"] == "completed" and second.json()["current_step"] == "approved"

    no = await _start(client, auth_headers, wf)
    [task] = await _tasks(client, a, no["id"])
    rejected = await _act(client, a, no["id"], "rejected", task["token_id"])
    assert rejected.json()["status"] == "completed" and rejected.json()["current_step"] == "rejected"


async def test_majority_vote(client, auth_headers, test_tenant):
    tid = test_tenant["tenant_id"]
    role = await _role(tid)
    voters = [await _person(tid, role) for _ in range(3)]
    steps = [
        _step("start", "start", to=["panel"]),
        _step("panel", "approval", to=["yes", "no"], approver_roles=[str(role)], approval_mode="majority"),
        _step("yes", "end"),
        _step("no", "end"),
    ]
    wf = await _publish(client, auth_headers, steps)
    inst = await _start(client, auth_headers, wf)
    [t0] = await _tasks(client, voters[0][0], inst["id"])
    assert (await _act(client, voters[0][0], inst["id"], "no", t0["token_id"])).json()["status"] == "in_progress"
    [t1] = await _tasks(client, voters[1][0], inst["id"])
    assert (await _act(client, voters[1][0], inst["id"], "yes", t1["token_id"])).json()["status"] == "in_progress"
    [t2] = await _tasks(client, voters[2][0], inst["id"])
    final = await _act(client, voters[2][0], inst["id"], "yes", t2["token_id"])
    assert final.json()["status"] == "completed" and final.json()["current_step"] == "yes"


async def test_claiming_hides_it_from_the_group_and_handing_it_on(client, auth_headers, test_tenant):
    tid = test_tenant["tenant_id"]
    role = await _role(tid)
    a, _ = await _person(tid, role)
    b, _ = await _person(tid, role)
    c, c_id = await _person(tid)
    steps = [_step("start", "start", to=["review"]), _step("review", "task", to=["done"], approver_roles=[str(role)]), _step("done", "end")]
    wf = await _publish(client, auth_headers, steps)
    inst = await _start(client, auth_headers, wf)
    [task] = await _tasks(client, a, inst["id"])
    claimed = await client.post(f"/approvals/{inst['id']}/claim", headers=a, json={"token_id": task["token_id"]})
    assert claimed.status_code == 200, claimed.text
    assert (await _tasks(client, a, inst["id"]))[0]["claimed_by_me"] is True
    assert await _tasks(client, b, inst["id"]) == []
    assert (await _act(client, b, inst["id"], "done", task["token_id"])).status_code == 409
    released = await client.post(f"/approvals/{inst['id']}/release", headers=a, json={"token_id": task["token_id"]})
    assert released.status_code == 200
    assert len(await _tasks(client, b, inst["id"])) == 1

    # Handing it to someone outside the role: now only they see it.
    handed = await client.post(f"/approvals/{inst['id']}/delegate", headers=b, json={"token_id": task["token_id"], "user_id": str(c_id), "comment": "You know this one"})
    assert handed.status_code == 200, handed.text
    assert await _tasks(client, a, inst["id"]) == [] and await _tasks(client, b, inst["id"]) == []
    [mine] = await _tasks(client, c, inst["id"])
    assert (await _act(client, c, inst["id"], "done", mine["token_id"])).json()["status"] == "completed"
    async with AsyncSessionLocal() as db:
        note = (await db.execute(select(Notification).where(Notification.recipient_user_id == c_id, Notification.title.like("Handed to you%")))).scalars().first()
    assert note is not None


async def test_out_of_office_stand_in_acts_on_behalf(client, auth_headers, test_tenant):
    tid = test_tenant["tenant_id"]
    manager, manager_id = await _person(tid)
    deputy, deputy_id = await _person(tid)
    stranger, _ = await _person(tid)
    steps = [
        _step("start", "start", to=["manager"]),
        _step("manager", "approval", to=["ok"], assignee_field="manager_id"),
        _step("ok", "end"),
    ]
    wf = await _publish(client, auth_headers, steps)
    inst = await _start(client, auth_headers, wf, {"manager_id": str(manager_id)})
    assert len(await _tasks(client, manager, inst["id"])) == 1
    assert await _tasks(client, deputy, inst["id"]) == []

    now = datetime.now(timezone.utc)
    away = {"delegate_id": str(deputy_id), "starts_at": (now - timedelta(hours=1)).isoformat(), "ends_at": (now + timedelta(days=3)).isoformat(), "reason": "Leave"}
    made = await client.post("/workflow-delegations", headers=manager, json=away)
    assert made.status_code == 201, made.text
    assert made.json()["delegate_name"] and made.json()["user_id"] == str(manager_id)
    # Only one stand-in at a time, and only your own unless you manage workflows.
    assert (await client.post("/workflow-delegations", headers=manager, json=away)).status_code == 409
    assert (await client.post("/workflow-delegations", headers=stranger, json={**away, "user_id": str(manager_id)})).status_code == 403

    [task] = await _tasks(client, deputy, inst["id"])
    assert task["on_behalf_of"] == str(manager_id) and task["on_behalf_of_name"]
    done = await _act(client, deputy, inst["id"], "ok", task["token_id"])
    assert done.json()["status"] == "completed"
    history = (await client.get(f"/workflow-instances/{inst['id']}/history", headers=auth_headers)).json()
    assert any(h["comment"] and h["comment"].startswith("On behalf of") for h in history)

    mine = (await client.get("/workflow-delegations", headers=deputy)).json()
    assert [d["id"] for d in mine] == [made.json()["id"]]
    assert (await client.delete(f"/workflow-delegations/{made.json()['id']}", headers=deputy)).status_code == 403
    assert (await client.delete(f"/workflow-delegations/{made.json()['id']}", headers=manager)).status_code == 204


async def test_timers_wait_then_move_on(client, auth_headers, test_tenant):
    tid = test_tenant["tenant_id"]
    steps = [_step("start", "start", to=["cool_off"]), _step("cool_off", "timer", to=["done"], timer={"mode": "duration", "hours": 48}), _step("done", "end")]
    wf = await _publish(client, auth_headers, steps)
    inst = await _start(client, auth_headers, wf)
    token = inst["tokens"][0]
    assert token["state"] == "waiting" and token["wake_at"] and inst["next_due_at"]
    assert await _due(tid) == 0  # nothing due yet
    await _backdate(inst["id"], "wake_at")
    assert await _due(tid) == 1
    after = await _get(client, auth_headers, inst["id"])
    assert after["status"] == "completed"

    # Waiting until a date in the context — already past: straight through.
    steps2 = [_step("start", "start", to=["until"]), _step("until", "timer", to=["done"], timer={"mode": "until_field", "field": "go_live"}), _step("done", "end")]
    wf2 = await _publish(client, auth_headers, steps2)
    assert (await _start(client, auth_headers, wf2, {"go_live": "2020-01-01"}))["status"] == "completed"
    assert (await _start(client, auth_headers, wf2, {"go_live": "2999-01-01"}))["tokens"][0]["state"] == "waiting"


async def test_sla_reminder_and_escalation(client, auth_headers, test_tenant):
    tid = test_tenant["tenant_id"]
    clerk_role, boss_role = await _role(tid), await _role(tid)
    clerk, clerk_id = await _person(tid, clerk_role)
    boss, boss_id = await _person(tid, boss_role)
    steps = [
        _step("start", "start", to=["review"]),
        _step("review", "approval", to=["done"], approver_roles=[str(clerk_role)], sla_hours=24, remind_before_hours=4,
              escalation={"action": "reassign", "role_ids": [str(boss_role)]}),
        _step("done", "end"),
    ]
    wf = await _publish(client, auth_headers, steps)
    inst = await _start(client, auth_headers, wf)
    token = inst["tokens"][0]
    assert token["due_at"] and token["remind_at"]

    await _backdate(inst["id"], "remind_at")
    assert await _due(tid) == 1
    async with AsyncSessionLocal() as db:
        reminded = (await db.execute(select(Notification).where(Notification.recipient_user_id == clerk_id, Notification.title.like("Reminder:%")))).scalars().first()
    assert reminded is not None

    await _backdate(inst["id"], "due_at")
    assert await _due(tid) == 1
    after = await _get(client, auth_headers, inst["id"])
    assert after["has_escalation"] is True
    # Reassigned to the boss's role: the clerk no longer sees it.
    assert await _tasks(client, clerk, inst["id"]) == []
    [task] = await _tasks(client, boss, inst["id"])
    assert task["escalated"] is True
    escalated = (await client.get("/workflow-instances?escalated=true", headers=auth_headers)).json()
    assert inst["id"] in [i["id"] for i in escalated]
    assert (await client.get("/workflow-instances/summary", headers=auth_headers)).json()["escalated"] >= 1

    # Route on timeout instead: it moves on by itself.
    steps2 = [
        _step("start", "start", to=["review"]),
        _step("review", "approval", to=["done"], approver_roles=[str(clerk_role)], sla_hours=1, escalation={"action": "route", "route_to": "auto"}),
        _step("auto", "end", label="Auto-approved"),
        _step("done", "end"),
    ]
    wf2 = await _publish(client, auth_headers, steps2)
    inst2 = await _start(client, auth_headers, wf2)
    await _backdate(inst2["id"], "due_at")
    await _due(tid)
    after2 = await _get(client, auth_headers, inst2["id"])
    assert after2["status"] == "completed" and after2["current_step"] == "auto"


async def test_sub_workflows_wait_and_map_values_back(client, auth_headers, test_tenant):
    tid = test_tenant["tenant_id"]
    role = await _role(tid)
    office, _ = await _person(tid, role)
    child_steps = [
        _step("start", "start", to=["check"]),
        _step("check", "task", to=["done"], approver_roles=[str(role)], output_params=[{"name": "verdict", "type": "string", "source": "verdict"}]),
        _step("done", "end"),
    ]
    child_wf = await _publish(client, auth_headers, child_steps)
    parent_steps = [
        _step("start", "start", to=["each"]),
        _step("each", "subprocess", to=["done"], subprocess={"workflow_id": child_wf, "for_each": "offices", "input_map": {"office": "item"}, "output_map": {"verdicts": "verdict"}}),
        _step("done", "end"),
    ]
    parent_wf = await _publish(client, auth_headers, parent_steps)
    parent = await _start(client, auth_headers, parent_wf, {"offices": ["Aluva", "Kochi"]})
    assert parent["tokens"][0]["state"] == "waiting" and len(parent["tokens"][0]["child_ids"]) == 2
    children = (await client.get(f"/workflow-instances?parent_instance_id={parent['id']}", headers=auth_headers)).json()
    assert len(children) == 2

    tasks = (await client.get("/approvals", headers=office)).json()
    child_tasks = [t for t in tasks if t["instance_id"] in parent["tokens"][0]["child_ids"]]
    assert len(child_tasks) == 2
    for t, verdict in zip(sorted(child_tasks, key=lambda t: t["context"]["item_index"]), ["ok", "late"]):
        resp = await _act(client, office, t["instance_id"], "done", t["token_id"], context={"verdict": verdict})
        assert resp.status_code == 200, resp.text
    after = await _get(client, auth_headers, parent["id"])
    assert after["status"] == "completed"
    assert after["context"]["verdicts"] == ["ok", "late"]

    # Cancelling a parent cancels the children it waits on.
    again = await _start(client, auth_headers, parent_wf, {"offices": ["Thrissur"]})
    cancelled = await client.post(f"/workflow-instances/{again['id']}/cancel", headers=auth_headers, json={"reason": "Withdrawn"})
    assert cancelled.status_code == 200 and cancelled.json()["status"] == "cancelled" and cancelled.json()["cancel_reason"] == "Withdrawn"
    child = await _get(client, auth_headers, again["tokens"][0]["child_ids"][0])
    assert child["status"] == "cancelled"


async def test_service_steps_run_a_task_and_failures_become_incidents(client, auth_headers, test_tenant):
    tid = test_tenant["tenant_id"]
    good_nodes = [
        {"key": "start", "type": "start", "label": "Start", "next_nodes": [{"to": "score"}]},
        {"key": "score", "type": "set_variable", "label": "Score", "input_params": [{"name": "score", "type": "number", "value": "42"}]},
    ]
    good = (await client.post("/tasks", headers=auth_headers, json={"name": f"Score {uuid.uuid4().hex[:4]}", "nodes": good_nodes})).json()
    assert (await client.post(f"/tasks/{good['id']}/publish", headers=auth_headers)).status_code == 200
    bad = (await client.post("/tasks", headers=auth_headers, json={
        "name": f"Bad {uuid.uuid4().hex[:4]}",
        "nodes": [{"key": "start", "type": "start", "label": "Start", "next_nodes": []}],
        "variables": [{"name": "fee", "type": "int", "source": "input", "value": "fee"}],
    })).json()
    assert (await client.post(f"/tasks/{bad['id']}/publish", headers=auth_headers)).status_code == 200

    steps = [
        _step("start", "start", to=["svc"]),
        _step("svc", "service", to=["route"], service={"task_id": good["id"]}),
        _step("route", "decision", to=[{"to": "high", "condition": _when("score", "gt", 10)}, {"to": "low"}]),
        _step("high", "end"),
        _step("low", "end"),
    ]
    wf = await _publish(client, auth_headers, steps)
    inst = await _start(client, auth_headers, wf)
    assert inst["tokens"][0]["state"] == "waiting" and inst["tokens"][0]["step"] == "svc"
    await jobs.run_due([engine.SERVICE_JOB], tenant_id=tid)
    after = await _get(client, auth_headers, inst["id"])
    assert after["status"] == "completed" and after["current_step"] == "high" and after["context"]["score"] == 42

    steps_bad = [_step("start", "start", to=["svc"]), _step("svc", "service", to=["done"], service={"task_id": bad["id"], "retries": 0}), _step("done", "end")]
    wf_bad = await _publish(client, auth_headers, steps_bad)
    failing = await _start(client, auth_headers, wf_bad, {"fee": "ten"})
    await jobs.run_due([engine.SERVICE_JOB], tenant_id=tid)
    broken = await _get(client, auth_headers, failing["id"])
    assert broken["has_incident"] is True and "fee" in broken["tokens"][0]["error"]
    # Skipping it moves on as if the Task had worked.
    skipped = await client.post(f"/workflow-instances/{failing['id']}/skip", headers=auth_headers, json={"token_id": broken["tokens"][0]["id"], "comment": "Done by hand"})
    assert skipped.status_code == 200, skipped.text
    assert skipped.json()["status"] == "completed"


async def test_suspend_reassign_and_admin_permissions(client, auth_headers, test_tenant):
    tid = test_tenant["tenant_id"]
    role = await _role(tid)
    a, _ = await _person(tid, role)
    b, b_id = await _person(tid)
    steps = [_step("start", "start", to=["review"]), _step("review", "approval", to=["done"], approver_roles=[str(role)]), _step("done", "end")]
    wf = await _publish(client, auth_headers, steps)
    inst = await _start(client, auth_headers, wf)
    token_id = inst["tokens"][0]["id"]

    # Approvers aren't administrators.
    assert (await client.post(f"/workflow-instances/{inst['id']}/suspend", headers=a)).status_code == 403
    assert (await client.post(f"/workflow-instances/{inst['id']}/suspend", headers=auth_headers)).status_code == 200
    assert await _tasks(client, a, inst["id"]) == []
    assert (await _act(client, a, inst["id"], "done", token_id)).status_code in (404, 409)
    assert (await client.get("/workflow-instances?status=suspended", headers=auth_headers)).json()[0]["id"] == inst["id"]
    assert (await client.post(f"/workflow-instances/{inst['id']}/resume", headers=auth_headers)).status_code == 200

    moved = await client.post(f"/workflow-instances/{inst['id']}/reassign", headers=auth_headers, json={"token_id": token_id, "user_ids": [str(b_id)], "comment": "Covering"})
    assert moved.status_code == 200, moved.text
    assert await _tasks(client, a, inst["id"]) == []
    [task] = await _tasks(client, b, inst["id"])
    assert (await _act(client, b, inst["id"], "done", task["token_id"])).json()["status"] == "completed"
    actions = [h["action"] for h in (await client.get(f"/workflow-instances/{inst['id']}/history", headers=auth_headers)).json()]
    assert actions == ["started", "submitted", "suspended", "resumed", "reassigned", "done", "completed"]


async def test_publishing_checks_the_definition(client, auth_headers):
    steps = [
        _step("start", "start", to=["review"]),
        _step("review", "approval", to=["ghost"]),
        _step("split", "parallel", to=["review"]),
        _step("wait", "timer", to=["review"]),
        _step("svc", "service", to=["review"]),
        _step("sub", "subprocess", to=["review"]),
    ]
    made = (await client.post("/workflows", headers=auth_headers, json={"name": "Broken", "step_definitions": steps})).json()
    resp = await client.post(f"/workflows/{made['workflow_id']}/publish", headers=auth_headers)
    assert resp.status_code == 422
    detail = resp.json()["detail"]
    for expected in (
        "Add an End step",
        "“Review” connects to a step that no longer exists",
        "nobody is assigned",
        "“Split” splits into parallel branches",
        "“Wait” needs a duration",
        "“Svc” needs a Task to run",
        "“Sub” needs a workflow to start",
    ):
        assert expected in detail, expected