Extract task execution to TaskRunner

At its core a process is a function that accepts an input, executes a
task and returns an asynchronous result. TaskProcessor implements this
functionality together with the dataflow interface of the process, so
that a task can only be executed by feeding values through the process
input channels.

Move the execution of a task to a separate `TaskRunner` class:

- the resolution of the task inputs and outputs
- the store directory and cache lookup
- the delegation to the executor, including task arrays
- the task error strategy and retries

TaskProcessor keeps the dataflow interface -- the operator, the process
state, the output binding and the publishing -- as well as the process
scoped services shared by all task executions, such as the name, the
config, the executor and the task environment.

The processor creates a runner with `taskCompleted` as the completion
handler, so that the runner does not depend on the dataflow interface:
a task can be executed by invoking `TaskRunner.submit()` directly,
without the process being part of a dataflow network (see the test
'should execute a task without a dataflow network').

Note that `resumeOrDie` is now synchronized on the runner rather than on
the processor, so it no longer excludes `bindOutputs` and
`sendPoisonPill`. These methods do not share mutable state, the coarse
lock was incidental.

Assisted-by: Claude Opus 5 (1M context)
Signed-off-by: Ben Sherman <bentshermann@gmail.com>

Author: bentsherman

docs/developer/diagrams/nextflow.processor.mmd

@@ -12,8 +12,13 @@ classDiagram
         operator : DataflowProcessor
         taskBody : BodyDef
     }
-    TaskProcessor --> TaskRun : invokeTask
-    TaskProcessor --> PublishDir : finalizeTask
+    TaskProcessor --* TaskRunner
+    TaskProcessor --> PublishDir : taskCompleted
+
+    class TaskRunner {
+        processor : TaskProcessor
+    }
+    TaskRunner --> TaskRun : submit
 
     class TaskRun {
         config : TaskConfig

docs/developer/nextflow.processor.mdx

@@ -20,8 +20,13 @@ classDiagram
         operator : DataflowProcessor
         taskBody : BodyDef
     }
-    TaskProcessor --> TaskRun : invokeTask
-    TaskProcessor --> PublishDir : finalizeTask
+    TaskProcessor --* TaskRunner
+    TaskProcessor --> PublishDir : taskCompleted
+
+    class TaskRunner {
+        processor : TaskProcessor
+    }
+    TaskRunner --> TaskRun : submit
 
     class TaskRun {
         config : TaskConfig

modules/nextflow/src/main/groovy/nextflow/processor/TaskPollingMonitor.groovy

@@ -677,7 +677,7 @@ class TaskPollingMonitor implements TaskMonitor {
             if (evict(handler)) {
                 handler.decProcessForks()
             }
-            fault = handler.task.processor.resumeOrDie(handler?.task, error, handler.getTraceRecord())
+            fault = handler.task.processor.getRunner().resumeOrDie(handler?.task, error, handler.getTraceRecord())
             log.trace "Task fault (1): $fault"
         }
         finally {
@@ -743,7 +743,7 @@ class TaskPollingMonitor implements TaskMonitor {
 
     protected void finalizeTask( TaskHandler handler ) {
         // finalize the task execution
-        final fault = handler.task.processor.finalizeTask(handler)
+        final fault = handler.task.processor.getRunner().finalizeTask(handler)
 
         // notify task completion
         session.notifyTaskComplete(handler)