# Copyright (c) 2020 Adam Souzis
# SPDX-License-Identifier: MIT
"""
A Job is generated by comparing a list of specs with the last known state of the system.
Job runs tasks, each of which has a configuration spec that is executed on the running system
Each task tracks and records its modifications to the system's state
"""

import collections
from datetime import datetime, timedelta
import itertools
import os
import json
from typing import (
    Any,
    Iterable,
    List,
    Dict,
    Mapping,
    MutableMapping,
    Optional,
    Sequence,
    Tuple,
    Union,
    cast,
    TYPE_CHECKING,
)
from .support import (
    Status,
    Priority,
    Defaults,
    AttributeManager,
    Reason,
    NodeState,
    AttributeChanges,
)
from .result import ResourceRef, serialize_value, ChangeRecord
from .util import UnfurlError, UnfurlTaskError, to_enum, change_cwd
from .merge import merge_dicts
from .runtime import EntityInstance, TopologyInstance, Operational, OperationalInstance
from .logs import end_collapsible, start_collapsible, getLogger
from .configurator import (
    TaskView,
    ConfiguratorResult,
    Configurator,
)
from .projectpaths import Folders, rmtree
from .planrequests import (
    ConfigurationSpec,
    PlanRequest,
    TaskRequest,
    TaskRequestGroup,
    SetStateRequest,
    JobRequest,
    do_render_requests,
    get_render_requests,
    set_fulfilled,
    set_fulfilled_stragglers,
    create_instance_from_spec,
    get_success_status,
)
from .plan import Plan
from .localenv import LocalEnv
from . import display
from . import configurators  # need to import configurators even though it is unused
from .reporting import JobReporter

from time import perf_counter

if TYPE_CHECKING:
    from unfurl.yamlmanifest import YamlManifest


logger = getLogger("unfurl")


class JobAborted(UnfurlError):
    pass


class ConfigChange(OperationalInstance, ChangeRecord):
    """
    Represents a configuration change made to the system.
    It has a operating status and a list of dependencies that contribute to its status.
    There are two kinds of dependencies:

    1. Live resource attributes that the configuration's inputs depend on.
    2. Other configurations and resources it relies on to function properly.
    """

    def __init__(
        self,
        parentJob: Optional["Job"] = None,
        startTime: Optional[datetime] = None,
        status: Optional[Union[OperationalInstance, int, str]] = None,
        previousId: Optional[str] = None,
        **kw: Any,
    ) -> None:
        OperationalInstance.__init__(self, status, **kw)
        if parentJob:  # use the parent's job id and startTime
            ChangeRecord.__init__(self, parentJob.changeId, parentJob.startTime)
        else:  # generate a new job id and use the given startTime
            ChangeRecord.__init__(self, startTime=startTime, previousId=previousId)

    def get_operational_dependencies(self) -> Iterable[Operational]:
        for d in self.dependencies:
            if d.target != self.target:  # type: ignore
                yield d


class JobOptions:
    """
    Options available to select which tasks are run, e.g. read-only

    does the config apply to the operation?
    is it out of date?
    is it in a ok state?
    """

    global_defaults = dict(
        # global options that we also want to apply to any child or external job
        verbose=0,
        masterJob=None,
        startTime=None,
        starttime=None,
        out=None,
        dryrun=False,
        planOnly=False,
        readonly=False,  # only run configurations that won't alter the system
        requiredOnly=False,
        commit=False,
        dirty="auto",  # run the job even if the repository has uncommitted changrs
        message=None,
    )

    parentJob = None
    masterJob: Optional["Job"] = None
    instance = None
    workflow = Defaults.workflow
    planOnly = False
    verbose = 0
    message: Optional[str] = None
    commit = False
    push = False
    template: Optional[str] = None
    add: bool = True
    skip_new: bool = False
    change_detection: str = "evaluate"
    repair: str = "error"
    force: bool = False
    check: bool = False
    prune: bool = False
    destroyunmanaged: bool = False
    upgrade: bool = False
    dryrun: bool = False

    defaults = dict(
        global_defaults,
        parentJob=parentJob,
        instance=instance,
        instances=None,
        template=None,
        # default options:
        add=True,  # add new templates
        skip_new=False,  # don't create newly defined instances (override add)
        update=True,  # run configurations that that have changed
        change_detection="evaluate",  # skip, spec, evaluate (skip sets update to False)
        repair="error",  # or 'degraded' or "none", run configurations that are not operational and/or degraded
        upgrade=False,  # run configurations with major version changes or whose spec has changed
        force=False,  # (re)run operation regardless of instance's status or state
        verify=False,  # XXX3 discover first and set status if it differs from expected state
        check=False,  # if new instances exist before deploying
        prune=False,
        destroyunmanaged=False,
        append=None,
        replace=None,
        workflow=workflow,
    )

    def __init__(self, **kw: Any) -> None:
        options = self.defaults.copy()
        if kw.get("starttime"):  # rename
            options["startTime"] = kw["starttime"]
        if kw.get("skip_new"):
            options["add"] = False
        if kw.get("change_detection") == "skip":
            options["update"] = False
        options.update(kw)
        self.__dict__.update(options)
        self.userConfig = kw

    def copy(self, **kw: Any) -> "JobOptions":
        # only copy global
        _copy = {
            k: self.userConfig[k]
            for k in set(self.userConfig) & set(self.global_defaults)
        }
        return JobOptions(**dict(_copy, **kw))

    def get_user_settings(self) -> dict:
        # only include settings different from the defaults
        return {
            k: self.userConfig[k]
            for k in set(self.userConfig) & set(self.defaults)
            if k not in ["out", "masterJob"] and self.userConfig[k] != self.defaults[k]
        }


class ConfigTask(TaskView, ConfigChange):
    """
    receives a configSpec and a target node instance
    instantiates and runs Configurator
    updates Configurator's target's status and lastConfigChange
    """

    def __init__(self, job, configSpec, target, reason=None):
        ConfigChange.__init__(self, job)
        TaskView.__init__(self, job.manifest, configSpec, target, reason)
        self.dry_run = job.dry_run
        self.verbose = job.verbose
        self._configurator = None
        self.generator = None
        self.job = job
        self.changeList: List[AttributeChanges] = []
        self.result = None
        self.outputs = None
        # for summary:
        self.modified_target = False
        self.target_status = target.status
        self.target_state = target.state

        # set the attribute manager on the root resource
        self._attributeManager = AttributeManager(self._manifest.yaml, self)
        self.target.root.set_attribute_manager(self._attributeManager)
        self._resolved_inputs = {}

    def _status(self, seen: Dict[int, Operational]) -> Status:
        status = self.local_status
        # if not status:
        #     return Status.unknown
        return status  # type: ignore

    def __priority():  # type: ignore
        doc = "The priority property."

        def fget(self):
            if self._priority is None:
                return self.configSpec.should_run()
            else:
                return self._priority

        def fset(self, value):
            self._priority = value

        def fdel(self):
            del self._priority

        return locals()

    priority: Priority = property(**__priority())  # type: ignore

    @property
    def configurator(self) -> Configurator:
        if self._configurator is None:
            self._configurator = self.configSpec.create()
        return self._configurator

    def start_run(self) -> None:
        self.generator = self.configurator.get_generator(self)

    def send(self, change):
        result = None
        assert self.generator
        try:
            result = self.generator.send(change)
        finally:
            # serialize configuration changes
            self.commit_changes()
        return result

    def start(self) -> None:
        self.start_run()
        self.target.root.set_attribute_manager(self._attributeManager)
        self.target_status = self.target.status
        self.target_state = self.target.state
        self.set_envvars()

    def _update_status(self, result) -> bool:
        """
        Update the instances status with the result of the operation.
        If status wasn't explicitly set but the operation changed the instance's configuration
        or state, choose a status based on the type of operation.
        """
        if result.status is not None:
            # status was explicitly set
            self.target.local_status = result.status
            if self.target.present and self.target.created is None:
                if self.configSpec.operation not in [
                    "check",
                    "discover",
                ]:
                    self.target.created = self.changeId
            self.logger.trace(
                "status was explicitly set for %s with local_status %s",
                self.target.name,
                self.target.local_status,
            )

            return True
        elif not result.success:
            # if any task failed and (maybe) modified, target.status will be set to error or unknown
            if result.modified:
                self.target.local_status = (
                    Status.error if self.required else Status.degraded
                )
                return True
            elif result.modified is None:
                self.target.local_status = Status.unknown
                return True
            # otherwise doesn't modify target status
        return False

    def _update_last_change(self, result: ConfiguratorResult) -> bool:
        """
        If the target's configuration or state has changed, set the instance's lastChange
        state to this tasks' changeid.
        """
        if self.target.last_change is None and self.target.status != Status.pending:
            # hacky but always save _lastConfigChange the first time to
            # distinguish this from a brand new resource
            self.target._lastConfigChange = self.changeId
            return True
        if result.modified or self._resourceChanges.get_attribute_changes(
            self.target.key
        ):
            if self.target.last_change != self.changeId:
                # save to create a linked list of tasks that modified the target
                self.previousId = self.target.last_change
            self.target._lastConfigChange = self.changeId
            return True
        return False

    def _finished_workflow(self, successStatus, workflow):
        instance = self.target
        self.modified_target = True
        instance.local_status = successStatus
        self.target_status = successStatus
        if instance.last_change != self.changeId:
            # save to create a linked list of tasks that modified the target
            self.previousId = instance.last_change
        instance._lastConfigChange = self.changeId
        if (
            workflow == "deploy"
            and successStatus == Status.ok
            and instance.created is None
        ):
            instance.created = self.changeId

    def finished(self, result: ConfiguratorResult):
        assert result
        self.restore_envvars()
        if self.generator:
            self.generator.close()
            self.generator = None

        self.outputs = result.outputs

        # don't set the changeId until we're finish so that we have a higher changeid
        # than nested tasks and jobs that ran
        # (task that never run will have the same changeId as its parent)
        self.set_task_id(self.job.increment_task_count())
        # XXX2 if attributes changed validate using attributesSchema
        # XXX2 Check that configuration provided the metadata that it declared (check postCondition)

        if self.changeList:
            # merge changes together (will be saved with changeset)
            changes = self.changeList
            accum = changes.pop(0)
            while changes:
                accum = cast(AttributeChanges, merge_dicts(accum, changes.pop(0)))

            # note: this might set _lastConfigChange on instances other than this target
            self._resourceChanges.update_changes(
                accum, self._attributeManager.statuses, self.target, self.changeId
            )
            # XXX implement:
            # if not result.applied:
            #    self._resourceChanges.rollback(self.target)

        # now that resourceChanges finalized:
        self._update_status(result)
        targetChanged = self._update_last_change(result)
        self.result = result
        self.local_status = Status.ok if result.success else Status.error
        self.modified_target = targetChanged or self.target_status != self.target.status
        self.target_status = self.target.status
        self.target_state = self.target.state
        return self

    @property
    def completed(self) -> bool:
        return bool(self.local_status and self.local_status > Status.unknown)

    def _reset(self):
        self._inputs = None
        self._environ = None

    def commit_changes(self):
        """
        This can be called multiple times if the configurator yields multiple times.
        Save the changes made each time.
        """
        changes, dependencies = self._attributeManager.commit_changes()
        self.changeList.append(changes)

        if self._inputs is not None:
            self._resolved_inputs.update(self.inputs.get_resolved())
        # need to reset because _attributeManager was reset in commit_changes()
        self._reset()

        # record the live attributes that we are dependent on
        # note: task start fresh with no dependencies so don't need to worry update or removing previous ones
        for key, (target, attributes) in dependencies.items():
            if key.startswith("::.requirements"):
                # hackish: exclude default connections (which are represented as root relationships)
                continue
            for name, (live, value) in attributes.items():
                # hackish: we only want the status of a dependency to reflect its target instance's status
                # when the attribute is live (as opposed to a static property set in the spec)
                # so don't set its target unless the attribute is live
                self.add_dependency(
                    key + "::" + name, value, target=target if live else None
                )
        return changes, dependencies

    # unused
    # def findLastConfigureOperation(self):
    #     if not self._manifest.changeSets:
    #         return None
    #     previousId = self.target.lastChange
    #     while previousId:
    #         previousChange = self._manifest.changeSets.get(previousId)
    #         if not previousChange:
    #             return None
    #         if previousChange.target != self.target.key:
    #             return (
    #                 None  # XXX handle case where lastChange was set by another target
    #             )
    #         if previousChange.operation == self.configSpec.operation:
    #             return previousChange
    #         previousId = previousChange.previousId
    #     return None

    def has_inputs_changed(self) -> bool:
        """
        Evaluate configuration spec's inputs and compare with the current inputs' values
        """
        changeset = cast("YamlManifest", self._manifest).find_last_operation(
            self.target.key, self.configSpec.operation
        )
        if not changeset:
            self.logger.debug(
                'Can\'t check for changes: could not find previous "%s" operation for "%s"',
                self.target.key,
                self.configSpec.operation,
            )
            return False

        return self.configurator.check_digest(self, changeset)

    def has_dependencies_changed(self):
        return False
        # XXX artifacts
        # XXX identity of requirements (how? what about imported nodes? instance keys?)
        # dynamic dependencies:
        # return any(d.hasChanged(self) for d in self.dependencies)

    # XXX unused
    # def refreshDependencies(self):
    #     for d in self.dependencies:
    #         d.refresh(self)

    def find_missing_dependencies(self):
        missing = []
        reason = ""
        # XXX this logic should be replaced with entry_state on check and delete operations
        if self.configSpec.operation not in ["check", "delete"] and (
            self.reason not in [Reason.force, Reason.run]
        ):
            # check if the live attributes this task depends on has been set
            missing, reason = _dependency_check(self)

            if not missing:
                reason = _dependency_check(self.target, self.configSpec.entry_state)
        return missing, reason

    @property
    def name(self):
        name = self.configSpec.name
        if self.configSpec.operation and self.configSpec.operation not in name:
            name = name + ":" + self.configSpec.operation
        if self.reason and self.reason not in name:
            return name + ":" + self.reason
        return name

    def summary(self, asJson=False, add_rendered=False):
        if self.target.name != self.target.template.name:
            rname = f"{self.target.name} ({self.target.template.name})"
        else:
            rname = self.target.name

        if self.configSpec.name != self.configSpec.className:
            cname = f"{self.configSpec.name} ({self.configSpec.className})"
        else:
            cname = self.configSpec.name

        if self._configurator:
            cClass = self._configurator.__class__
            configurator = f"{cClass.__module__}.{cClass.__name__}"
        else:
            configurator = self.configSpec.className

        status = None if self.status is None else self.status.name
        summary = dict(
            status=status,
            target=self.target.name,
            operation=self.configSpec.operation,
            template=self.target.template.name,
            type=self.target.template.type,
            targetStatus=None
            if self.target_status is None
            else self.target_status.name,
            targetState=self.target_state and self.target_state.name or None,
            changed=self.modified_target,
            configurator=configurator,
            priority=self.priority.name,
            reason=self.reason or "",
        )
        if add_rendered:
            summary["rendered_paths"] = [
                path
                for path in [wf.cwd for wf in self._workFolders.values()]
                + self._failed_paths
                if os.path.exists(path)
            ]

        if asJson:
            return summary
        else:
            return (
                "{operation} on instance {rname} (type {type}, status {targetStatus}) "
                + "using configurator {cname}, priority: {priority}, reason: {reason}"
            ).format(cname=cname, rname=rname, **summary)

    def __repr__(self):
        return f"ConfigTask({self.target}:{self.name})"


def _dependency_check(
    instance: Operational, state: Optional[NodeState] = None, operational=False
):
    missing = []
    for dep in instance.get_operational_dependencies():
        if dep.required:
            if dep.local_status == Status.error:
                # we only want to consider local_status especially for the case of subsequent runs
                # trying to repair a failed deployment.
                missing.append(dep)
            elif state and dep.state:
                if not dep.has_state(state):
                    missing.append(dep)
            elif operational and not dep.operational:  # operational == ok or degraded
                missing.append(dep)

    if missing:
        reason = "required dependencies not operational: %s" % ", ".join(
            [f"{dep.name} is {dep.status.name}" for dep in missing]
        )
    else:
        reason = ""
    return missing, reason


class Job(ConfigChange):
    """
    runs ConfigTasks and child Jobs
    """

    MAX_NESTED_SUBTASKS = 100

    def __init__(
        self,
        manifest: "YamlManifest",
        rootResource: TopologyInstance,
        jobOptions: JobOptions,
        previousId: Optional[str] = None,
    ) -> None:
        assert isinstance(jobOptions, JobOptions)
        self.__dict__.update(jobOptions.__dict__)
        super().__init__(self.parentJob, self.startTime, Status.ok, previousId)  # type: ignore
        self.dry_run = jobOptions.dryrun
        self.jobOptions = jobOptions
        self.manifest = manifest
        self.rootResource = rootResource
        self.jobRequestQueue: List[JobRequest] = []
        self.unexpectedAbort: Optional[UnfurlError] = None
        self.workDone: Dict[int, ConfigTask] = collections.OrderedDict()
        self.timeElapsed: float = 0
        self.plan_requests: Optional[List[TaskRequestGroup]] = None
        self.task_count = 0
        self.external_requests: Optional[List[Tuple[Any, List[JobRequest]]]] = None
        self.external_jobs: Optional[List["Job"]] = None

    def get_operational_dependencies(self) -> Iterable[ConfigTask]:
        # XXX3 this isn't right, root job might have too many and child job might not have enough
        # plus dynamic configurations probably shouldn't be included if yielded by a configurator
        for task in self.workDone.values():
            if task.status is not None and task.priority > Priority.ignore:
                yield task

    def get_outputs(self) -> Any:
        return self.rootResource.attributes["outputs"]  # type: ignore

    def is_filtered(self) -> bool:
        return self.instance or self.instances or self.template  # type: ignore

    def run_query(self, query: Union[str, Mapping], trace: int = 0) -> list:
        from .eval import eval_for_func, RefContext

        return eval_for_func(query, RefContext(self.rootResource, trace=trace))

    def create_task(
        self,
        configSpec: ConfigurationSpec,
        target: EntityInstance,
        reason: Optional[str] = None,
    ) -> ConfigTask:
        task = ConfigTask(self, configSpec, target, reason=reason)
        try:
            task.inputs
            task.configurator
        except Exception:
            UnfurlTaskError(task, "unable to create task")
        return task

    def validate_job_options(self) -> None:
        if (
            self.jobOptions.instance
            and not self.rootResource.find_resource(  # type: ignore  # mypy doesn't like the way JobOptions is defined
                self.jobOptions.instance  # type: ignore
            )
        ):
            logger.warning(
                'selected instance not found: "%s"', self.jobOptions.instance  # type: ignore
            )

    def render(self) -> Tuple[List[PlanRequest], List[PlanRequest], List[PlanRequest]]:
        if self.plan_requests is None:
            ready: Sequence[PlanRequest] = self.create_plan()
        else:
            ready = self.plan_requests[:]

        # run artifact job requests before render
        if self.external_requests:
            msg = "Running local installation tasks"
            plan, count = self._plan_summary([], self.external_requests)
            logger.info(msg + "\n%s", plan, extra=dict(truncate=0))

        # currently external jobs are just for installing artifacts
        # we want to run these even if we just generating a plan
        self.external_jobs = self.run_external(planOnly=False)
        if self.external_jobs and self.external_jobs[-1].status == Status.error:
            return [], [], []

        ready, notReady, errors = do_render_requests(self, ready)
        return ready, notReady, errors

    @staticmethod
    def _yield_serious_errors(errors: List[PlanRequest]) -> Iterable[UnfurlTaskError]:
        for req in errors:
            if not req.task or req.task.priority >= Priority.required:
                if req.render_errors:
                    yield from req.render_errors

    def _run_requests(
        self,
        rendered_requests: Optional[
            Tuple[List[PlanRequest], List[PlanRequest], List[PlanRequest]]
        ] = None,
    ) -> ResourceRef:
        if rendered_requests:
            ready, notReady, failed = rendered_requests
        else:
            ready, notReady, failed = self.render()

        self.workDone = collections.OrderedDict()

        serious_errors = list(self._yield_serious_errors(failed))
        if serious_errors:
            logger.error(
                "Aborting job: there were errors during rendering: %s", serious_errors
            )
            return self.rootResource

        # XXX update_plan(ready, unfulfilled) # try to reorder so we can add to ready
        while ready or notReady or self.jobRequestQueue:
            # XXX need to call self.run_external() here if update_plan() adds external job
            # create and run tasks for requests that have their dependencies fulfilled
            self.apply(ready)

            # the jobRequestQueue will have jobs that were added dynamically by a configurator
            # but were not yielding inside runTask
            while self.jobRequestQueue:
                jobRequest = self.jobRequestQueue[0]
                self.run_job_request(jobRequest)

            # remove requests from notReady if they've had all their dependencies fulfilled
            completed = ready
            if completed:
                ready, notReady = set_fulfilled(notReady, completed)
                check_target = ""
            else:
                if self.jobOptions.workflow == "deploy":
                    check_target = "operational"
                else:
                    check_target = "any"

                # we ran all the ready tasks we could so now we can run left-over tasks that depend on
                # live attributes that we no longer have to worry about being modified by another task
                ready, notReady = set_fulfilled_stragglers(notReady, check_target)
            logger.trace(
                "ready %s; not ready %s; completed: %s", ready, notReady, completed
            )

            # the first time we render them all, after that only re-render requests if their dependencies were fulfilled
            # the last time (when completed is empty) don't have render valid dependencies as unfulfilled
            ready, unfulfilled, errored = do_render_requests(
                self, ready, notReady, check_target
            )
            failed.extend(errored)
            if not ready and not completed:
                break  # none of the stragglers are ready, give up
            if unfulfilled:
                logger.trace("marking unfulfilled as not ready %s", unfulfilled)
                # XXX update_plan(ready, unfulfilled) # try to reorder so we can add to ready
                notReady.extend(unfulfilled)

        for req in failed:
            if req.render_errors:
                err_msg = str(req.render_errors[0])
            else:
                err_msg = "render failed"
            self._add_unrendered_task(req, err_msg)

        # if there were circular dependencies or errors then notReady won't be empty
        if notReady:
            for req in get_render_requests(notReady):
                if (
                    self.jobOptions.workflow == "deploy" and not req.include_in_plan()
                ):  # we don't want to run these
                    continue
                message = req.get_notready_message()
                self._add_unrendered_task(req, message)

        # force outputs to be evaluated now and commit the changes
        # so any attributes that were evaluated computing the outputs are saved in the manifest.
        outputs = serialize_value(self.get_outputs())
        if outputs:
            logger.info("Job outputs: %s", outputs)
        if self.rootResource.attributeManager:
            self.rootResource.attributeManager.commit_changes()
        return self.rootResource

    def _add_unrendered_task(self, req: PlanRequest, message: str):
        if req.task:
            req.task.logger.info(message)
            req.task.finished(ConfiguratorResult(False, False, result=message))
            self.add_work(req.task)

    def _run(self, ready: List[PlanRequest], notReady: List[PlanRequest], errors: List[PlanRequest]) -> None:
        jobOptions = self.jobOptions
        try:
            display.verbosity = jobOptions.verbose
            self._run_requests((ready, notReady, errors))
        except JobAborted as e:
            self.local_status = Status.error  # type: ignore
            logger.error("Aborting job: %s", e)
        except Exception:
            self.local_status = Status.error  # type: ignore
            self.unexpectedAbort = UnfurlError(
                "unexpected exception while running job", True, True
            )
        self._apply_workfolders()

    def run(self, rendered: Tuple[List[PlanRequest], List[PlanRequest], List[PlanRequest]]) -> None:
        manifest = self.manifest
        startTime = perf_counter()
        jobOptions = self.jobOptions
        with change_cwd(manifest.get_base_dir()):
            try:
                ready, notReady, errors = rendered
                if not jobOptions.out:  # type: ignore
                    # out is used by unit tests to avoid writing to disk
                    manifest.lock()
                if jobOptions.dirty == "auto":  # type: ignore  # default to false if committing
                    checkIfClean = jobOptions.commit  # type: ignore
                else:
                    checkIfClean = jobOptions.dirty == "abort"  # type: ignore
                if checkIfClean:
                    for repo in manifest.repositories.values():
                        if repo.is_dirty():
                            logger.error(
                                "aborting run: uncommitted files in %s (--dirty=ok to override)",
                                repo.working_dir,
                            )
                            return None
                self._run(ready, notReady, errors)
                manifest.commit_job(self)
            finally:
                self.timeElapsed = perf_counter() - startTime
                manifest.unlock()

    def _apply_workfolders(self) -> None:
        for task in self.workDone.values():
            task.apply_work_folders()

    def _update_joboption_instances(self) -> None:
        if not self.jobOptions.instances:  # type: ignore
            return
        # process any instances that are a full resource spec
        self.jobOptions.instances = [  # type: ignore
            resourceSpec
            if isinstance(resourceSpec, str)
            else create_instance_from_spec(
                self.manifest, self.rootResource, resourceSpec["name"], resourceSpec
            ).name
            for resourceSpec in self.jobOptions.instances  # type: ignore
        ]
        self.instances = self.jobOptions.instances  # type: ignore

    def create_plan(self) -> List[TaskRequestGroup]:
        self.validate_job_options()
        joboptions = self.jobOptions
        self._update_joboption_instances()
        self.plan_requests = []
        WorkflowPlan = Plan.get_plan_class_for_workflow(joboptions.workflow)
        if not WorkflowPlan:
            raise UnfurlError(f"unknown workflow: {joboptions.workflow}")

        if not self.jobOptions.parentJob:
            # don't do this when running a nested job
            # (planned was already removed and new tasks have already been rendered there)
            rmtree(os.path.join(self.rootResource.base_dir, Folders.Planned))
        plan = WorkflowPlan(self.rootResource, self.manifest.tosca, joboptions)
        plan_requests = list(plan.execute_plan())

        request_artifacts = []
        for r in plan_requests:
            artifacts = r.get_operation_artifacts()
            if artifacts:
                request_artifacts.extend(artifacts)

        # remove duplicates
        artifact_jobs = list({ajr.name: ajr for ajr in request_artifacts}.values())

        # create JobRequests for each external job we need to run by grouping requests by their manifest
        external_requests = []
        for key, reqs in itertools.groupby(artifact_jobs, lambda r: id(r.root)):
            # external manifest activating an instance via artifact reification
            # XXX or substitution mapping -- but unique inputs require dynamically creating ensembles??
            reqs = list(reqs)  # type: ignore
            externalManifest = self.manifest._importedManifests.get(key)
            if externalManifest:
                external_requests.append((externalManifest, reqs))
            else:
                # run artifact jobs as a seperate external job since we need to run them
                # before the render stage of this job
                external_requests.append((self.manifest, reqs))

        self.external_requests = external_requests  # type: ignore
        self.plan_requests = plan_requests
        return self.plan_requests[:]

    def apply(
        self,
        reqs: Sequence[Union[JobRequest, PlanRequest]],
        depth: int = 0,
    ) -> Optional[ConfigTask]:
        task = None
        for req in reqs:
            # if parent is set, stop processing requests once one fails
            if isinstance(req, JobRequest):
                self.jobRequestQueue.append(req)
                self.run_job_request(req)
                continue
            if req.group and req.group.has_errors():
                req.set_error("previous operation failed")
                logger.debug("Skipping task %s because previous operation failed", req)
                continue
            if isinstance(req, TaskRequestGroup):
                # XXX this shouldn't happen now
                task = self.apply(req.children, depth)
            elif isinstance(req, SetStateRequest):
                logger.debug("Setting state with %s", req)
                self._set_state(req)
            else:
                assert isinstance(req, TaskRequest), type(req)
                _task = None
                if req.task and req.task.completed:
                    # the task already ran (before the rest of its task group)
                    _task, success = req.task, req.task.local_status != Status.error
                else:
                    workflow = req.group.workflow if req.group else None
                    _task, success = self._run_operation(req, workflow, depth)
                if not _task:
                    if req.is_final_for_workflow:
                        # we didn't need to run this one, but we need to finalize the workflow
                        req = req.reassign_final_for_workflow()  # type: ignore
                        if req:
                            _task = req.task
                    if not _task:
                        continue
                task = _task
                if task.priority == Priority.critical:
                    if not success or (
                        task.result and task.result.status == Status.error
                    ):
                        raise JobAborted(
                            f"Critical task failed: {task.name} for {task.target.name}"
                        )
                # note: task won't be return from _run_operation is the target's workflow state was set
                if req and req.is_final_for_workflow and req.completed:
                    req.finish_workflow()

        return task  # return the last task executed

    def run_job_request(self, jobRequest: JobRequest) -> "Job":
        logger.debug("running jobrequest: %s", jobRequest)
        if self.jobRequestQueue:
            self.jobRequestQueue.remove(jobRequest)
        instance_specs = jobRequest.get_instance_specs()
        jobOptions = self.jobOptions.copy(parentJob=self, instances=instance_specs)
        childJob = create_job(self.manifest, jobOptions)
        childJob.set_task_id(self.increment_task_count())
        assert childJob.parentJob is self
        childJob._run_requests()
        return childJob

    def run_external(self, **opts: Any) -> List["Job"]:
        # note: manifest.lock() will raise error if there circular dependencies
        external_jobs = []
        external_requests = self.external_requests[:] if self.external_requests else []
        while external_requests:
            manifest, requests = external_requests.pop(0)
            instance_specs = []
            for request in requests:
                assert isinstance(
                    request, JobRequest
                ), "only JobRequest currently supported"
                instance_specs.extend(request.get_instance_specs())
            jobOptions = self.jobOptions.copy(
                instances=instance_specs,
                masterJob=self.jobOptions.masterJob or self,
                **opts,
            )
            # currently external jobs only install artifacts needed for running the job so always disable dryrun
            jobOptions.dryrun = False
            external_job = create_job(manifest, jobOptions)
            external_jobs.append(external_job)
            rendered, count = _render(external_job)
            if not jobOptions.planOnly and count:
                if manifest is not self.manifest:
                    external_job.run(rendered)
                else:
                    external_job._run(*rendered)
            if external_job.status == Status.error:
                break
        return external_jobs

    def should_run_task(self, task: ConfigTask) -> Tuple[bool, str]:
        """
        Checked at runtime right before each task is run
        """
        try:
            if task._configurator:
                priority = task.configurator.should_run(task)
            else:
                priority = task.priority
        except Exception:
            # unexpected error don't run this
            UnfurlTaskError(task, "shouldRun failed unexpectedly")
            return False, "unexpected failure"

        if isinstance(priority, bool):
            priority = priority and Priority.required or Priority.ignore
        else:
            priority = to_enum(Priority, priority)
        if priority != task.priority:
            logger.debug(
                "configurator changed task %s priority from %s to %s",
                task,
                task.priority,
                priority,
            )
            assert isinstance(priority, Priority)
            task.priority = priority
        if not priority > Priority.ignore:
            return False, "configurator cancelled"

        if task.reason == Reason.reconfigure:
            if task.has_inputs_changed() or task.has_dependencies_changed():
                return True, "change detected"
            else:
                return False, "no change detected"

        return True, "proceed"

    def can_run_task(self, task: ConfigTask) -> Tuple[bool, str]:
        """
        Checked at runtime right before each task is run

        * validate inputs
        * check pre-conditions to see if it can be run
        * check task if it can be run
        """
        can_run = False
        reason = ""
        try:
            if task._errors:
                can_run = False
                reason = "Error while creating task"  # XXX + str(_errors)
            elif task.dry_run and not task.configurator.can_dry_run(task):
                can_run = False
                reason = "dry run not supported"
            else:
                missing, reason = task.find_missing_dependencies()
                task.logger.debug(
                    "find missing %s %s %s",
                    missing,
                    task.configSpec.entry_state,
                    task.configSpec.entry_state > NodeState.initial,
                )
                if not missing:
                    errors = task.configSpec.find_invalidate_inputs(task.inputs)
                    if errors:
                        reason = f"invalid inputs: {str(errors)}"
                    else:
                        preErrors = task.configSpec.find_invalid_preconditions(
                            task.target
                        )
                        if preErrors:
                            reason = f"invalid preconditions: {str(preErrors)}"
                        else:
                            errors = task.configurator.can_run(task)
                            if not errors or not isinstance(errors, bool):
                                reason = f"configurator declined: {str(errors)}"
                            else:
                                can_run = True
        except Exception:
            UnfurlTaskError(task, "can_run_task failed unexpectedly")
            reason = "unexpected exception in can_run_task"
            can_run = False

        if can_run:
            return True, ""
        else:
            logger.info("could not run task %s: %s", task, reason)
            return False, "could not run: " + reason

    def _set_state(self, req: SetStateRequest) -> None:
        resource = req.target
        if "managed" in req.set_state:
            resource.created = False if req.set_state == "unmanaged" else self.changeId
        else:
            try:
                resource.state = req.set_state
            except KeyError:
                resource.local_status = to_enum(Status, req.set_state)
        req.completed = True

    def _entry_test(
        self, req: TaskRequest, workflow: Optional[str]
    ) -> Tuple[bool, str]:
        """
        Operations can dynamically advance the state of a instance so that an operation
        added by the plan no longer needs to run.
        For example, a check operation may determine a resource is already created
        or a create operation may also configure and start an instance.
        """
        resource = req.target
        assert resource
        logger.trace(
            "checking operation entry test on %s: current state %s start state %s op %s workflow %s",
            resource.key,
            resource.state,
            req.startState,
            req.configSpec.operation,
            workflow,
        )
        if req.configSpec.operation == "check":
            missing, reason = _dependency_check(resource, operational=True)
            if missing:
                return False, reason
            if not workflow:
                if (
                    resource.parent
                    and self.is_change_id(resource.parent.created)
                    and self.get_job_id(resource.parent.created) == self.changeId
                ):
                    # optimization:
                    # if parent was created during this job we don't need to run check operation
                    # we know we couldn't have existed
                    resource._localStatus = Status.pending
                    return False, "skipping check on a new instance"
                else:
                    return True, "passed"
        elif req.configSpec.operation == "restart" and not resource.present:
            return False, "instance can't be restart"

        if self.jobOptions.force:  # type: ignore  # always run
            return True, "passed"

        # Note: workflow is only set when request is a child of a TaskGroupRequest generated by the plan
        if get_success_status(workflow) == resource.status:
            return False, "instance already has desired status"

        if req.startState and resource.state and workflow == "deploy":
            # if we set a startState and force isn't set then only run
            # if we haven't already advanced to that state by another operation
            entryTest = NodeState(req.startState + 1)
            if resource.state > NodeState.started:
                # "started" is the final deploy state, target state must be stopped or deleted or error
                if req.configSpec.operation == "start":
                    # start operations can't handle deleted nodes
                    return (
                        resource.state <= NodeState.stopped,
                        "can't start a missing instance",
                    )
            elif resource.state > entryTest:
                return False, "instance already entered state"
        return True, "passed"

    def _run_operation(
        self, req: PlanRequest, workflow: Optional[str], depth: int
    ) -> Tuple[Optional[ConfigTask], bool]:
        assert isinstance(req, TaskRequest)
        if req.required is False:  # set after should_run() is called
            return None, True
        if req.error:
            return None, False

        test, msg = self._entry_test(req, workflow)
        if not test:
            req.completed = True
            logger.info(
                'Skipping operation "%s" on instance "%s" with state "%s" and status "%s": %s',
                req.configSpec.operation,
                req.target.name,
                req.target.state,
                req.target.status,
                msg,
            )
            return None, True

        task = req.task or self.create_task(
            req.configSpec, req.target, reason=req.reason
        )
        if task:
            start_collapsible(f"Task {req.target.name} ({req}", hash(req), True)
            task.logger.info(
                "started task.", extra=dict(json=task.summary(asJson=True))
            )

            resource = req.target
            startingStatus = resource._localStatus
            if req.startState is not None:
                resource.state = req.startState
            startingState = resource.state
            self.add_work(task)
            self.run_task(task, depth)

            # if # task succeeded but didn't update nodestate
            if task.result and task.result.success and resource.state == startingState:  # type: ignore
                if req.startState is not None:
                    # advance the state if a startState was set in the TaskRequest
                    resource.state = NodeState(req.startState + 1)
                elif (
                    req.configSpec.operation == "check"
                    and startingStatus != resource._localStatus
                ):
                    # if check operation set status but didn't update state, set a default state
                    state_map = {
                        Status.ok: NodeState.started,
                        Status.error: NodeState.error,
                        Status.absent: NodeState.deleted,
                        Status.pending: NodeState.initial,
                    }
                    if not resource.state or resource.state == NodeState.initial:
                        # it is a resource we didn't create
                        state_map[Status.absent] = NodeState.initial
                    else:
                        state_map[Status.absent] = NodeState.deleted
                    if resource._localStatus is None:
                        state = None
                    else:
                        state = state_map.get(resource._localStatus)
                    if state is not None:
                        resource.state = state
                task.target_state = resource.state

            # logger.debug(
            #     "changed %s to %s, wanted %s",
            #     startingState,
            #     task.target.state,
            #     req.startState,
            # )
            end_collapsible(hash(req))
            task_success = task.result and task.result.success
            status = task.target.status.name.upper()
            state_status = task.target.state.name if task.target.state else ""
            extra = dict(
                rich=dict(style=task.target.status.color),
                json=task.summary(asJson=True),
            )
            if task_success:
                task.logger.info(
                    "Task succeeded, Resource Status: %s State: %s",
                    status,
                    state_status,
                    extra=extra,
                )
            else:
                task.logger.error(
                    "Task failed, Resource Status: %s State: %s",
                    status,
                    state_status,
                    extra=extra,
                )
            return task, task_success
        return None, False

    def run_task(self, task: ConfigTask, depth: int = 0) -> ConfigTask:
        """
        During each task run:
        * Notification of metadata changes that reflect changes made to resources
        * Notification of add or removing dependency on a resource or properties of a resource
        * Notification of creation or deletion of a resource
        * Requests a resource with requested metadata, if it doesn't exist, a task is run to make it so
        (e.g. add a dns entry, install a package).

        Returns a task.
        """
        task.target.root.set_attribute_manager(task._attributeManager)
        errors: Optional[str] = None
        ok, errors = self.can_run_task(task)
        if not ok:
            return task.finished(ConfiguratorResult(False, False, result=errors))

        task.start()
        change = None
        while True:
            try:
                result = task.send(change)
            except Exception:
                UnfurlTaskError(task, "configurator.run failed")
                return task.finished(ConfiguratorResult(False, None, Status.error))
            if isinstance(result, PlanRequest):
                if depth >= self.MAX_NESTED_SUBTASKS:
                    UnfurlTaskError(task, "too many subtasks spawned")
                    change = task.finished(ConfiguratorResult(False, None))
                else:
                    ready, _, error_reqs = do_render_requests(self, [result])
                    if not ready:
                        err_msg = "render failed"
                        if error_reqs and error_reqs[0].render_errors:
                            err_msg = str(error_reqs[0].render_errors[0])
                        assert result.task
                        return result.task.finished(
                            ConfiguratorResult(False, False, result=err_msg)
                        )
                    change = self.apply(ready, depth + 1)
            elif isinstance(result, JobRequest):
                job = self.run_job_request(result)
                change = job
            elif isinstance(result, ConfiguratorResult):
                retVal = task.finished(result)
                logger.debug(
                    "completed task %s: %s; %s", task, task.target.status, result
                )
                return retVal
            else:
                UnfurlTaskError(task, "unexpected result from configurator")
                return task.finished(ConfiguratorResult(False, None, Status.error))

    def add_work(self, task: ConfigTask) -> None:
        key = id(task)
        self.workDone[key] = task
        if self.parentJob:  # type: ignore
            self.parentJob.add_work(task)  # type: ignore

    def increment_task_count(self) -> int:
        if self.parentJob:  # type: ignore
            return self.parentJob.increment_task_count()  # type: ignore
        else:
            self.task_count += 1
        return self.task_count

    @property
    def log_path(self) -> str:
        log_name = (
            self.startTime.strftime("%Y-%m-%d-%H-%M-%S") + "-" + self.changeId[:-4]
        )
        return self.manifest.get_job_log_path(log_name, ".log")

    ###########################################################################
    # Job Reporting methods
    ###########################################################################

    def stats(self, asMessage: bool = False) -> Union[Dict[str, int], str]:
        return JobReporter.stats(self.workDone.values(), asMessage)

    def get_end_time(self) -> str:
        return (self.startTime + timedelta(seconds=self.timeElapsed)).strftime(
            self.DateTimeFormat
        )

    def print_summary_table(self) -> str:
        try:
            return JobReporter.summary_table(self)
        except Exception:
            logger.error("Error while creating job summary", exc_info=True)
            return "Error while creating job summary"

    def _plan_summary(
        self,
        plan_requests: List[TaskRequest],
        external_requests: Iterable[Tuple[Any, Any]],
    ) -> Tuple[str, int]:
        return JobReporter.plan_summary(self, plan_requests, external_requests)

    def _json_plan_summary(
        self, pretty: bool = False, include_rendered: bool = True
    ) -> Union[str, list]:
        return JobReporter.json_plan_summary(self, pretty, include_rendered)

    def json_summary(
        self, pretty: bool = False, external: bool = False, add_rendered: bool = False
    ) -> Union[str, Dict[str, Any]]:
        job = dict(id=self.changeId, status=self.status.name)
        job.update(self.stats())  # type: ignore
        if not self.startTime:  # skip if startTime was explicitly set
            job["timeElapsed"] = self.timeElapsed  # type: ignore
        summary = dict(
            job=job,
            outputs=serialize_value(self.get_outputs()),
            tasks=[task.summary(True, add_rendered) for task in self.workDone.values()],
        )
        if external:
            summary["ensemble"] = self.manifest.path
        if self.external_jobs:
            summary["external_jobs"] = [
                external.json_summary(external=True) for external in self.external_jobs
            ]
        if pretty:
            return json.dumps(summary, indent=2)
        return summary

    def summary(self) -> str:
        outputString = ""
        outputs = self.get_outputs()
        if outputs:
            outputString = "\nOutputs:\n    " + "\n    ".join(
                f"{name}: {value}" for name, value in serialize_value(outputs).items()
            )

        if not self.workDone:
            return f"Job {self.changeId} completed: {self.status.name}. No tasks ran. {outputString}"

        def format(i: int, task: ConfigTask) -> str:
            return "%d. %s; %s" % (i, task.summary(), task.result or "skipped")

        line1 = "Job %s completed in %.3fs: %s. %s:\n    " % (
            self.changeId,
            self.timeElapsed,
            self.status.name,
            self.stats(asMessage=True),
        )

        tasks = "\n    ".join(
            format(i + 1, task) for i, task in enumerate(self.workDone.values())
        )
        return line1 + tasks + outputString


def create_job(manifest, joboptions, previousId=None):
    """
    Selects task to run based on the workflow and job options
    """
    root = manifest.get_root_resource()
    assert manifest.tosca
    job = Job(manifest, root, joboptions, previousId)

    if manifest.localEnv and not joboptions.parentJob and not joboptions.startTime:
        path = manifest.path
        if joboptions.planOnly:
            logger.info("creating %s plan for %s", joboptions.workflow, path)
        else:
            logger.info("starting %s job for %s", joboptions.workflow, path)

    return job


def _plan(manifest, jobOptions):
    assert jobOptions
    job = create_job(
        manifest,
        jobOptions,
        manifest.lastJob and manifest.lastJob["changeId"],
    )
    job.create_plan()
    if jobOptions.masterJob:
        msg = "Initial static plan for external job:"
    else:
        msg = "Initial static plan:"
    plan_msg, count = job._plan_summary(job.plan_requests, job.external_requests)
    logger.debug(msg + "\n%s", plan_msg, extra=dict(truncate=0))
    return job


def _render(job):
    # note: we need to call render() before lock because render might run this ensemble as an external_job
    with change_cwd(job.manifest.get_base_dir()):
        ready, notReady, errors = job.render()
        msg, count = job._plan_summary(ready + notReady, [])
        logger.info(msg, extra=dict(truncate=0))
    return (ready, notReady, errors), count


def start_job(manifestPath=None, _opts=None):
    _opts = _opts or {}
    localEnv = LocalEnv(
        manifestPath,
        _opts.get("home"),
        override_context=_opts.get("use_environment") or "",
        overrides=dict((n, v) for n, v in _opts.get("var", []))
        # XXX readonly=_opts.get("planOnly")
    )
    opts = JobOptions(**_opts)
    path = localEnv.manifestPath
    if not opts.planOnly:
        logger.info("creating %s job for %s", opts.workflow, path)
    try:
        manifest = localEnv.get_manifest()
    except Exception as e:
        logger.error(
            "failed to load manifest at %s: %s",
            path,
            str(e),
            exc_info=opts.verbose >= 2,
        )
        return None, None, False

    job = _plan(manifest, opts)
    rendered, count = _render(job)
    errors = list(job._yield_serious_errors(rendered[2]))
    if errors:
        logger.error("Aborting job: there were errors during rendering: %s", errors)
        job.local_status = Status.error  # type: ignore
    return job, rendered, count and not errors


def run_job(manifestPath: Optional[str] = None, _opts: Optional[dict] = None) -> "Job":
    """
    Loads the given Ensemble and creates and runs a job.

    Args:
        manifestPath (:obj:`str`, optional) The path the Ensemble manifest.
         If None, it will look for an ensemble in the current working directory.
        _opts (:obj:`dict`, optional) A dictionary of job options. Names and values should match
          the names of the command line options for creating jobs.

    Returns:
        (:obj:`Job`): The job that just ran or None if it couldn't be created.
    """
    job, rendered, proceed = start_job(manifestPath, _opts)
    if job:
        if not job.unexpectedAbort and not job.jobOptions.planOnly and proceed:
            job.run(rendered)
    return job


class Runner:
    # this class is only used by unit tests, application uses start_job() above

    def __init__(self, manifest):
        self.manifest = manifest
        assert self.manifest.tosca

    def run(self, jobOptions=None):
        if jobOptions is None:
            jobOptions = JobOptions()
        job = _plan(self.manifest, jobOptions)
        rendered, count = _render(job)
        if not jobOptions.planOnly:
            job.run(rendered)
        return job
