from __future__ import annotations

from dataclasses import replace
import unittest

from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker

from govoplan_core.auth import ApiPrincipal
from govoplan_core.core.access import (
    CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER,
    PrincipalRef,
)
from govoplan_core.core.automation import (
    ActionDefinition,
    ActionExecutionResult,
    ActionPreview,
    AutomationPrincipalResolution,
    EffectDefinition,
    EffectPreview,
    ObservedEffect,
)
from govoplan_core.core.dataflows import (
    CAPABILITY_DATAFLOW_RUN_LIFECYCLE,
    DataflowRunDescriptor,
)
from govoplan_core.db.base import Base
from govoplan_workflow_engine.backend.db.models import (
    WorkflowDefinition,
    WorkflowDefinitionRevision,
    WorkflowInstance,
    WorkflowInstanceEvent,
    WorkflowInstanceStep,
)
from govoplan_workflow_engine.backend.instance_service import (
    SqlWorkflowRuntimeWorker,
    cancel_instance,
    instance_response,
    reconcile_instance,
    resolve_step,
    start_instance,
)
from govoplan_workflow_engine.backend.schemas import (
    BpmnRevisionInput,
    WorkflowDefinitionCreateRequest,
    WorkflowEdge,
    WorkflowGraph,
    WorkflowInstanceStartRequest,
    WorkflowNode,
    WorkflowStepActionRequest,
)
from govoplan_workflow_engine.backend.bpmn_adapters import NATIVE_LINEAR_ADAPTER_ID
from govoplan_workflow_engine.backend.service import (
    WorkflowConflictError,
    activate_definition,
    create_definition,
)
try:
    from test_bpmn import NATIVE_BPMN
except ModuleNotFoundError as exc:
    if exc.name != "test_bpmn":
        raise
    from tests.test_bpmn import NATIVE_BPMN


def principal() -> ApiPrincipal:
    return ApiPrincipal(
        principal=PrincipalRef(
            account_id="account-1",
            membership_id="membership-1",
            tenant_id="tenant-1",
            scopes=frozenset(
                {
                    "workflow:definition:read",
                    "workflow:instance:read",
                    "workflow:instance:start",
                    "workflow:instance:transition",
                    "dataflow:pipeline:run",
                }
            ),
        ),
        account=object(),
        user=object(),
    )


def runtime_graph() -> WorkflowGraph:
    return WorkflowGraph(
        nodes=[
            WorkflowNode(
                id="start",
                type="workflow.start.manual",
                label="Start",
                config={"input_schema_ref": ""},
            ),
            WorkflowNode(
                id="flow",
                type="workflow.dataflow",
                label="Prepare evidence",
                config={
                    "pipeline_ref": "pipeline:pipeline-1",
                    "revision": 3,
                    "environment": "development",
                    "row_limit": 250,
                    "publication_target_ref": "",
                    "warning_policy": "review",
                    "input_mapping": {},
                    "view_surface_ids": [
                        "dataflow.module",
                        "dataflow.route.pipelines",
                    ],
                },
            ),
            WorkflowNode(
                id="complete",
                type="workflow.end.completed",
                label="Complete",
                config={"output_mapping": {}},
            ),
            WorkflowNode(
                id="cancelled",
                type="workflow.end.cancelled",
                label="Rejected",
                config={"reason": "Rejected during review"},
            ),
        ],
        edges=[
            WorkflowEdge(
                id="start-flow",
                source="start",
                target="flow",
            ),
            WorkflowEdge(
                id="flow-complete",
                source="flow",
                source_port="success",
                target="complete",
            ),
            WorkflowEdge(
                id="flow-warning",
                source="flow",
                source_port="warning",
                target="complete",
            ),
            WorkflowEdge(
                id="flow-review",
                source="flow",
                source_port="review_required",
                target="complete",
            ),
            WorkflowEdge(
                id="flow-failure",
                source="flow",
                source_port="failure",
                target="cancelled",
            ),
        ],
    )


def action_graph() -> WorkflowGraph:
    return WorkflowGraph(
        nodes=[
            WorkflowNode(
                id="start",
                type="workflow.start.manual",
                config={"input_schema_ref": ""},
            ),
            WorkflowNode(
                id="action",
                type="workflow.capability",
                config={
                    "capability": "test.actions",
                    "operation": "test.case.record",
                    "input_mapping": {"case_id": "$input.case_id"},
                    "idempotency_key": "$input.case_id",
                    "failure_policy": "manual",
                },
            ),
            WorkflowNode(
                id="complete",
                type="workflow.end.completed",
                config={"output_mapping": {}},
            ),
            WorkflowNode(
                id="failed",
                type="workflow.end.cancelled",
                config={"reason": "Action rejected"},
            ),
        ],
        edges=[
            WorkflowEdge(
                id="start-action",
                source="start",
                target="action",
            ),
            WorkflowEdge(
                id="action-complete",
                source="action",
                source_port="success",
                target="complete",
            ),
            WorkflowEdge(
                id="action-failed",
                source="action",
                source_port="failure",
                target="failed",
            ),
        ],
    )


class FakeDataflowLifecycle:
    def __init__(self) -> None:
        self.runs: dict[str, DataflowRunDescriptor] = {}
        self.requests = []
        self.cancelled: list[str] = []

    def start_run(self, _session, _principal, *, request):
        self.requests.append(request)
        run_ref = f"run:{len(self.requests)}"
        descriptor = DataflowRunDescriptor(
            ref=run_ref,
            pipeline_ref=request.pipeline_ref,
            revision=request.revision,
            status="queued",
            definition_hash="definition-hash",
            executor_version="test",
            metadata={"progress_percent": 0, "progress_phase": "queued"},
        )
        self.runs[run_ref] = descriptor
        return descriptor

    def get_run(self, _session, _principal, *, run_ref):
        return self.runs.get(run_ref)

    def cancel_run(self, _session, _principal, *, run_ref):
        descriptor = self.runs[run_ref]
        descriptor = replace(descriptor, status="cancelled")
        self.runs[run_ref] = descriptor
        self.cancelled.append(run_ref)
        return descriptor

    def finish(
        self,
        run_ref: str,
        *,
        diagnostics: list[dict[str, object]] | None = None,
    ) -> None:
        self.runs[run_ref] = replace(
            self.runs[run_ref],
            status="succeeded",
            output_publication_ref="publication:1",
            output_datasource_ref="datasource:1",
            output_materialization_ref="materialization:1",
            input_row_count=12,
            output_row_count=10,
            metadata={"diagnostics": diagnostics or []},
        )

    def fail(self, run_ref: str) -> None:
        self.runs[run_ref] = replace(
            self.runs[run_ref],
            status="failed",
            error="Data quality gate failed.",
        )


class FakeAutomationProvider:
    def __init__(self) -> None:
        self.requests = []

    def resolve_automation_principal(self, _session, *, request):
        self.requests.append(request)
        return AutomationPrincipalResolution(
            allowed=True,
            principal=principal(),
            granted_scopes=request.grant_scopes,
            provenance={"status": "rechecked"},
        )


class FakeActionProvider:
    action = ActionDefinition(
        action_key="test.case.record",
        owner_module="test",
        description="Record a test case.",
        input_schema_ref="schema:test.case.record@1",
        expected_effect_keys=("test.case.recorded",),
    )
    effect = EffectDefinition(
        effect_key="test.case.recorded",
        owner_module="test",
        operation="created",
        description="A test case was recorded.",
    )

    def __init__(self, *states: str) -> None:
        self.states = list(states or ("completed",))
        self.requests = []

    def action_definitions(self):
        return (self.action,)

    def effect_definitions(self):
        return (self.effect,)

    def preview_action(self, _session, _principal, *, request):
        return ActionPreview(
            action_key=request.action_key,
            allowed=True,
            summary="Record one case.",
            risk_level=self.action.risk_level,
            reversibility=self.action.reversibility,
            effects=(
                EffectPreview(
                    effect_key=self.effect.effect_key,
                    summary="Record case.",
                ),
            ),
            preview_ref="preview:test",
        )

    def execute_action(self, _session, _principal, *, request):
        self.requests.append(request)
        state = self.states.pop(0) if self.states else "completed"
        if state != "completed":
            return ActionExecutionResult(
                state=state,
                error="Temporary action failure.",
            )
        return ActionExecutionResult(
            state="completed",
            output={"case_ref": "case:1"},
            observed_effects=(
                ObservedEffect(
                    effect_key=self.effect.effect_key,
                    operation="created",
                    resource_ref="case:1",
                ),
            ),
            audit_event_refs=("audit:1",),
        )


class Registry:
    def __init__(
        self,
        dataflow: FakeDataflowLifecycle,
        automation: FakeAutomationProvider | None = None,
        action: FakeActionProvider | None = None,
    ) -> None:
        self.dataflow = dataflow
        self.automation = automation
        self.action = action

    def has_capability(self, name: str) -> bool:
        return name == CAPABILITY_DATAFLOW_RUN_LIFECYCLE or (
            name == CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER
            and self.automation is not None
        ) or (
            name == "test.actions"
            and self.action is not None
        )

    def capability(self, name: str):
        if name == CAPABILITY_DATAFLOW_RUN_LIFECYCLE:
            return self.dataflow
        if (
            name == CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER
            and self.automation is not None
        ):
            return self.automation
        if name == "test.actions" and self.action is not None:
            return self.action
        raise KeyError(name)


class WorkflowInstanceServiceTests(unittest.TestCase):
    def setUp(self) -> None:
        self.engine = create_engine("sqlite:///:memory:")
        Base.metadata.create_all(
            self.engine,
            tables=[
                WorkflowDefinition.__table__,
                WorkflowDefinitionRevision.__table__,
                WorkflowInstance.__table__,
                WorkflowInstanceStep.__table__,
                WorkflowInstanceEvent.__table__,
            ],
        )
        self.Session = sessionmaker(bind=self.engine)
        self.session: Session = self.Session()
        self.dataflow = FakeDataflowLifecycle()
        self.registry = Registry(self.dataflow)
        self.definition = create_definition(
            self.session,
            tenant_id="tenant-1",
            actor_id="account-1",
            payload=WorkflowDefinitionCreateRequest(
                name="Monthly governed processing",
                graph=runtime_graph(),
                execution_mode="hybrid",
                view_id="view-1",
                view_revision_id="view-revision-1",
            ),
        )
        activate_definition(
            self.session,
            tenant_id="tenant-1",
            definition_id=self.definition.id,
            actor_id="account-1",
            revision=1,
        )
        self.session.commit()

    def tearDown(self) -> None:
        self.session.close()
        Base.metadata.drop_all(
            self.engine,
            tables=[
                WorkflowInstanceEvent.__table__,
                WorkflowInstanceStep.__table__,
                WorkflowInstance.__table__,
                WorkflowDefinitionRevision.__table__,
                WorkflowDefinition.__table__,
            ],
        )
        self.engine.dispose()

    def _start(self, key: str = "request-1") -> WorkflowInstance:
        instance, replayed = start_instance(
            self.session,
            tenant_id="tenant-1",
            definition_id=self.definition.id,
            actor_id="account-1",
            principal=principal(),
            registry=self.registry,
            payload=WorkflowInstanceStartRequest(
                idempotency_key=key,
                input={"case_id": "case-1"},
                correlation_id="correlation-1",
            ),
        )
        self.assertFalse(replayed)
        return instance

    def test_start_pins_revision_and_replays_idempotently(self) -> None:
        instance = self._start()
        replayed, was_replayed = start_instance(
            self.session,
            tenant_id="tenant-1",
            definition_id=self.definition.id,
            actor_id="account-1",
            principal=principal(),
            registry=self.registry,
            payload=WorkflowInstanceStartRequest(
                idempotency_key="request-1",
                input={"case_id": "case-1"},
                correlation_id="correlation-1",
            ),
        )

        self.assertTrue(was_replayed)
        self.assertEqual(instance.id, replayed.id)
        self.assertEqual("waiting", instance.status)
        self.assertEqual("user", instance.start_origin)
        self.assertEqual(1, len(self.dataflow.requests))
        response = instance_response(self.session, instance)
        self.assertEqual("hybrid", response.execution_mode)
        self.assertEqual("user", response.start_origin)
        self.assertIsNotNone(response.view_context)
        self.assertEqual(
            [
                "dataflow.module",
                "dataflow.route.pipelines",
            ],
            response.view_context.visible_surface_ids,
        )
        self.assertEqual([1, 2], [step.sequence for step in response.steps])
        self.assertEqual("run:1", response.steps[-1].external_ref)
        self.assertGreaterEqual(len(response.events), 4)

        with self.assertRaises(WorkflowConflictError):
            start_instance(
                self.session,
                tenant_id="tenant-1",
                definition_id=self.definition.id,
                actor_id="account-1",
                principal=principal(),
                registry=self.registry,
                payload=WorkflowInstanceStartRequest(
                    idempotency_key="request-1",
                    input={"case_id": "another-case"},
                    correlation_id="correlation-1",
                ),
            )

    def test_guided_workflow_rejects_automated_start_origin(self) -> None:
        definition = create_definition(
            self.session,
            tenant_id="tenant-1",
            actor_id="account-1",
            payload=WorkflowDefinitionCreateRequest(
                name="Guided review",
                graph=WorkflowGraph(
                    nodes=[
                        WorkflowNode(
                            id="start",
                            type="workflow.start.api",
                            config={
                                "input_schema_ref": "schema:input",
                                "authorization_policy_ref": "policy:start",
                            },
                        ),
                        WorkflowNode(
                            id="activity",
                            type="workflow.activity",
                            config={"title": "Review"},
                        ),
                        WorkflowNode(
                            id="done",
                            type="workflow.end.completed",
                        ),
                    ],
                    edges=[
                        WorkflowEdge(
                            id="start-activity",
                            source="start",
                            target="activity",
                        ),
                        WorkflowEdge(
                            id="activity-done",
                            source="activity",
                            target="done",
                        ),
                    ],
                ),
                execution_mode="guided",
                allow_automation=True,
            ),
        )
        activate_definition(
            self.session,
            tenant_id="tenant-1",
            definition_id=definition.id,
            actor_id="account-1",
        )

        with self.assertRaisesRegex(
            WorkflowConflictError,
            "Guided workflows must be started by a user",
        ):
            start_instance(
                self.session,
                tenant_id="tenant-1",
                definition_id=definition.id,
                actor_id="account-1",
                principal=principal(),
                registry=self.registry,
                payload=WorkflowInstanceStartRequest(
                    idempotency_key="automated-guided",
                ),
                start_origin="api",
            )

    def test_module_action_records_effects_and_completes_idempotently(
        self,
    ) -> None:
        action = FakeActionProvider()
        registry = Registry(self.dataflow, action=action)
        definition = create_definition(
            self.session,
            tenant_id="tenant-1",
            actor_id="account-1",
            payload=WorkflowDefinitionCreateRequest(
                name="Action workflow",
                graph=action_graph(),
            ),
        )
        activate_definition(
            self.session,
            tenant_id="tenant-1",
            definition_id=definition.id,
            actor_id="account-1",
        )

        instance, replayed = start_instance(
            self.session,
            tenant_id="tenant-1",
            definition_id=definition.id,
            actor_id="account-1",
            principal=principal(),
            registry=registry,
            payload=WorkflowInstanceStartRequest(
                idempotency_key="action-instance",
                input={"case_id": "case-1"},
            ),
        )
        response = instance_response(self.session, instance)

        self.assertFalse(replayed)
        self.assertEqual("completed", response.status)
        self.assertEqual(1, len(action.requests))
        self.assertEqual(
            "test.actions:test.case.record:case-1",
            action.requests[0].idempotency_key,
        )
        self.assertEqual(
            "case:1",
            response.steps[1].output["execution"]["observed_effects"][0][
                "resource_ref"
            ],
        )
        self.assertIn(
            "workflow.action.completed",
            {event.kind for event in response.events},
        )

    def test_retryable_module_action_reuses_the_idempotency_key(self) -> None:
        action = FakeActionProvider("retryable", "completed")
        registry = Registry(self.dataflow, action=action)
        definition = create_definition(
            self.session,
            tenant_id="tenant-1",
            actor_id="account-1",
            payload=WorkflowDefinitionCreateRequest(
                name="Retry action",
                graph=action_graph(),
            ),
        )
        activate_definition(
            self.session,
            tenant_id="tenant-1",
            definition_id=definition.id,
            actor_id="account-1",
        )
        instance, _replayed = start_instance(
            self.session,
            tenant_id="tenant-1",
            definition_id=definition.id,
            actor_id="account-1",
            principal=principal(),
            registry=registry,
            payload=WorkflowInstanceStartRequest(
                idempotency_key="retry-action-instance",
                input={"case_id": "case-2"},
            ),
        )
        waiting = instance_response(self.session, instance)
        current = waiting.steps[-1]
        self.assertEqual("retryable", current.handoff["state"])

        resolved = resolve_step(
            self.session,
            tenant_id="tenant-1",
            instance_id=instance.id,
            step_id=current.id,
            actor_id="account-1",
            principal=principal(),
            registry=registry,
            payload=WorkflowStepActionRequest(action="retry"),
        )

        self.assertEqual("completed", resolved.status)
        self.assertEqual(
            action.requests[0].idempotency_key,
            action.requests[1].idempotency_key,
        )

    def test_automated_dataflow_failure_policy_fails_without_handoff(
        self,
    ) -> None:
        graph = runtime_graph()
        flow = next(node for node in graph.nodes if node.id == "flow")
        flow.config["warning_policy"] = "continue"
        flow.config["failure_policy"] = "fail"
        definition = create_definition(
            self.session,
            tenant_id="tenant-1",
            actor_id="account-1",
            payload=WorkflowDefinitionCreateRequest(
                name="Automated Dataflow",
                graph=graph,
                execution_mode="automated",
            ),
        )
        activate_definition(
            self.session,
            tenant_id="tenant-1",
            definition_id=definition.id,
            actor_id="account-1",
        )
        instance, _replayed = start_instance(
            self.session,
            tenant_id="tenant-1",
            definition_id=definition.id,
            actor_id="account-1",
            principal=principal(),
            registry=self.registry,
            payload=WorkflowInstanceStartRequest(
                idempotency_key="automated-dataflow",
            ),
        )
        self.dataflow.fail("run:1")

        changed = reconcile_instance(
            self.session,
            instance=instance,
            principal=principal(),
            registry=self.registry,
        )

        self.assertTrue(changed)
        self.assertEqual("failed", instance.status)
        self.assertIsNone(instance.current_step_id)
        self.assertFalse(
            any(
                step.handoff.get("kind") == "dataflow_failure"
                for step in instance.steps
            )
        )

    def test_native_bpmn_profile_runs_through_canonical_instance_state(self) -> None:
        definition = create_definition(
            self.session,
            tenant_id="tenant-1",
            actor_id="account-1",
            payload=WorkflowDefinitionCreateRequest(
                name="BPMN governed review",
                graph=runtime_graph(),
                bpmn=BpmnRevisionInput(
                    xml=NATIVE_BPMN,
                    adapter_id=NATIVE_LINEAR_ADAPTER_ID,
                ),
            ),
        )
        activate_definition(
            self.session,
            tenant_id="tenant-1",
            definition_id=definition.id,
            actor_id="account-1",
        )
        instance, replayed = start_instance(
            self.session,
            tenant_id="tenant-1",
            definition_id=definition.id,
            actor_id="account-1",
            principal=principal(),
            registry=self.registry,
            payload=WorkflowInstanceStartRequest(
                idempotency_key="bpmn-request-1",
                input={"case_id": "case-bpmn"},
            ),
        )

        self.assertFalse(replayed)
        self.assertEqual("waiting", instance.status)
        waiting = instance_response(self.session, instance).steps[-1]
        self.assertEqual("workflow.activity", waiting.node_type)
        self.assertEqual("Review request", waiting.handoff["title"])

        replay, was_replayed = start_instance(
            self.session,
            tenant_id="tenant-1",
            definition_id=definition.id,
            actor_id="account-1",
            principal=principal(),
            registry=self.registry,
            payload=WorkflowInstanceStartRequest(
                idempotency_key="bpmn-request-1",
                input={"case_id": "case-bpmn"},
            ),
        )
        self.assertTrue(was_replayed)
        self.assertEqual(instance.id, replay.id)

        resolved = resolve_step(
            self.session,
            tenant_id="tenant-1",
            instance_id=instance.id,
            step_id=waiting.id,
            actor_id="account-1",
            principal=principal(),
            registry=self.registry,
            payload=WorkflowStepActionRequest(
                action="complete",
                comment="Reviewed.",
                evidence=["case:case-bpmn"],
            ),
        )
        response = instance_response(self.session, resolved)
        self.assertEqual("completed", response.status)
        self.assertEqual(
            "workflow.instance.completed",
            response.events[-1].kind,
        )

    def test_reconcile_completes_with_stable_dataflow_output_refs(self) -> None:
        instance = self._start()
        self.dataflow.finish("run:1")

        changed = reconcile_instance(
            self.session,
            instance=instance,
            principal=principal(),
            registry=self.registry,
            actor_id="account-1",
        )
        response = instance_response(self.session, instance)

        self.assertTrue(changed)
        self.assertEqual("completed", response.status)
        flow_output = response.context["steps"]["flow"]
        self.assertEqual("publication:1", flow_output["output_publication_ref"])
        self.assertEqual("datasource:1", flow_output["output_datasource_ref"])
        self.assertEqual(
            "materialization:1",
            flow_output["output_materialization_ref"],
        )
        self.assertEqual("workflow.instance.completed", response.events[-1].kind)

    def test_warning_requires_review_and_approve_resumes(self) -> None:
        instance = self._start()
        self.dataflow.finish(
            "run:1",
            diagnostics=[
                {
                    "severity": "warning",
                    "code": "review.required",
                    "message": "Verify unmatched records.",
                }
            ],
        )
        reconcile_instance(
            self.session,
            instance=instance,
            principal=principal(),
            registry=self.registry,
        )
        step_id = str(instance.current_step_id)

        self.assertEqual(
            "review_required",
            instance_response(self.session, instance).steps[-1].handoff["state"],
        )
        resolved = resolve_step(
            self.session,
            tenant_id="tenant-1",
            instance_id=instance.id,
            step_id=step_id,
            actor_id="account-1",
            principal=principal(),
            registry=self.registry,
            payload=WorkflowStepActionRequest(
                action="approve",
                comment="Evidence verified.",
                evidence=["publication:1"],
            ),
        )

        self.assertEqual("completed", resolved.status)

    def test_failure_can_retry_and_reject_invalid_actions(self) -> None:
        instance = self._start()
        self.dataflow.fail("run:1")
        reconcile_instance(
            self.session,
            instance=instance,
            principal=principal(),
            registry=self.registry,
        )
        step_id = str(instance.current_step_id)

        with self.assertRaises(WorkflowConflictError):
            resolve_step(
                self.session,
                tenant_id="tenant-1",
                instance_id=instance.id,
                step_id=step_id,
                actor_id="account-1",
                principal=principal(),
                registry=self.registry,
                payload=WorkflowStepActionRequest(action="approve"),
            )
        retried = resolve_step(
            self.session,
            tenant_id="tenant-1",
            instance_id=instance.id,
            step_id=step_id,
            actor_id="account-1",
            principal=principal(),
            registry=self.registry,
            payload=WorkflowStepActionRequest(action="retry"),
        )

        self.assertEqual("waiting", retried.status)
        self.assertEqual(2, len(self.dataflow.requests))
        self.assertEqual("run:2", retried.steps[-1].external_ref)
        self.assertEqual("superseded", retried.steps[-2].status)

    def test_cancel_propagates_to_linked_dataflow(self) -> None:
        instance = self._start()

        cancelled = cancel_instance(
            self.session,
            tenant_id="tenant-1",
            instance_id=instance.id,
            actor_id="account-1",
            principal=principal(),
            registry=self.registry,
        )

        self.assertEqual("cancelled", cancelled.status)
        self.assertEqual(["run:1"], self.dataflow.cancelled)

    def test_worker_rechecks_authorization_before_reconciling(self) -> None:
        instance = self._start()
        self.session.commit()
        self.dataflow.finish("run:1")
        automation = FakeAutomationProvider()
        worker = SqlWorkflowRuntimeWorker(
            registry=Registry(self.dataflow, automation),
        )

        summary = worker.reconcile_pending(self.session)

        self.assertEqual(1, summary["advanced"])
        self.assertEqual("completed", instance.status)
        self.assertEqual(1, len(automation.requests))
        self.assertEqual(
            "rechecked",
            instance.authorization_["last_resolution"]["status"],
        )

    def test_worker_reconciles_pending_module_action(self) -> None:
        action = FakeActionProvider("pending", "completed")
        automation = FakeAutomationProvider()
        registry = Registry(
            self.dataflow,
            automation=automation,
            action=action,
        )
        definition = create_definition(
            self.session,
            tenant_id="tenant-1",
            actor_id="account-1",
            payload=WorkflowDefinitionCreateRequest(
                name="Asynchronous action",
                graph=action_graph(),
            ),
        )
        activate_definition(
            self.session,
            tenant_id="tenant-1",
            definition_id=definition.id,
            actor_id="account-1",
        )
        instance, _replayed = start_instance(
            self.session,
            tenant_id="tenant-1",
            definition_id=definition.id,
            actor_id="account-1",
            principal=principal(),
            registry=registry,
            payload=WorkflowInstanceStartRequest(
                idempotency_key="pending-action",
                input={"case_id": "case-3"},
            ),
        )
        self.session.commit()
        worker = SqlWorkflowRuntimeWorker(registry=registry)

        summary = worker.reconcile_pending(self.session)

        self.assertEqual(1, summary["advanced"])
        self.assertEqual("completed", instance.status)
        self.assertEqual(2, len(action.requests))
        self.assertEqual(
            action.requests[0].idempotency_key,
            action.requests[1].idempotency_key,
        )
        self.assertEqual(1, len(automation.requests))


if __name__ == "__main__":
    unittest.main()