[Execution] Implement and use the Operation-based executor (#110)

Part of #99.

errors/errors.yml

@@ -142,6 +142,10 @@
   name: MaestroInstallError
   message: "An error occurred when installing Maestro in a remote environment: {error_message}"
 
+3006:
+  name: CombineOutputFileConflict
+  message: "The output file or directory '{output_file}' already exists and cannot be overwritten by the combine() task."
+
 
 # Archive and restore errors (error code 4xxx)
 4001:

src/conductor/cli/run.py

@@ -12,8 +12,8 @@
     AtLeastCommitNotAncestor,
 )
 from conductor.task_identifier import TaskIdentifier
-from conductor.execution.executor import Executor
-from conductor.execution.plan import ExecutionPlan
+from conductor.execution.executor2 import Executor2
+from conductor.execution.planning.planner import ExecutionPlanner
 from conductor.utils.user_code import cli_command
 from conductor.utils.colored_output import print_bold
 
@@ -151,8 +151,9 @@ def main(args):
         if not commit_is_ancestor:
             raise AtLeastCommitNotAncestor()
 
-    plan = ExecutionPlan.for_task(
-        task_identifier, ctx, run_again=args.again, at_least_commit=commit
+    ep = ExecutionPlanner(ctx)
+    plan = ep.create_plan_for(
+        task_identifier, run_again=args.again, at_least_commit=commit
     )
-    executor = Executor(execution_slots=num_jobs)
+    executor = Executor2(execution_slots=num_jobs)
     executor.run_plan(plan, ctx, stop_on_first_error=args.stop_early)

src/conductor/errors/generated.py

@@ -399,6 +399,19 @@ def _message(self):
         )
 
 
+class CombineOutputFileConflict(ConductorError):
+    error_code = 3006
+
+    def __init__(self, **kwargs):
+        super().__init__()
+        self.output_file = kwargs["output_file"]
+
+    def _message(self):
+        return "The output file or directory '{output_file}' already exists and cannot be overwritten by the combine() task.".format(
+            output_file=self.output_file,
+        )
+
+
 class OutputFileExists(ConductorError):
     error_code = 4001
 
@@ -650,6 +663,7 @@ def _message(self):
     "OutputDirTaken",
     "ConductorAbort",
     "MaestroInstallError",
+    "CombineOutputFileConflict",
     "OutputFileExists",
     "OutputPathDoesNotExist",
     "NoTaskOutputsToArchive",