FEAT(exec,.TC): +TASK_CONTEXT (not-nested), XFail ...

parallel+marshal.
Tried but din't fixed it:
    self.result = copy_context().run(self.op.compute, self.sol)

Neglible burden on TCs executions ~2.5s:
  pytest -k 'not site and not directive and test_graphtik'

Author: ankostis

docs/source/operations.rst

@@ -50,10 +50,10 @@ positional parameters are not supported:
       >>> add_op(a=3, b=4)
       {'a_plus_b': 7}
 
-   Calling an operation, it delegates to :meth:`.Operation.compute()` method,
-   which checks the inputs, match the *needs*/*provides* to function arguments,
-   calls the function,  and finally "zip" the function results with the operation's
-   *provides*. (read more on :ref:`graph-computations`).
+.. tip::
+   In case your function needs to access the :mod:`.execution` machinery
+   or its wrapping operation, it can do that through the :data:`.task_context`
+   (unstable API).
 
 
 Builder pattern

graphtik/execution.py

@@ -5,6 +5,7 @@
 import random
 import time
 from collections import ChainMap, abc, defaultdict, namedtuple
+from contextvars import ContextVar
 from functools import partial
 from itertools import chain
 from typing import Any, Collection, List, Mapping, Optional, Tuple, Union
@@ -17,6 +18,9 @@
     AbortedException,
     IncompleteExecutionError,
     Items,
+    Operation,
+    PlotArgs,
+    Plottable,
     aslist,
     astuple,
     first_solid,
@@ -39,7 +43,6 @@
     yield_node_names,
     yield_ops,
 )
-from .base import Operation, PlotArgs, Plottable
 
 #: If this logger is *eventually* DEBUG-enabled,
 #: the string-representation of network-objects (network, plan, solution)
@@ -302,7 +305,13 @@ def __call__(self):
             self.result = None
             log = logging.getLogger(self.logname)
             log.debug("+++ (%s) Executing %s...", self.solid, self)
-            self.result = self.op.compute(self.sol)
+            token = task_context.set(self)
+            try:
+                ## Not really needed ...
+                # self.result = copy_context().run(self.op.compute, self.sol)
+                self.result = self.op.compute(self.sol)
+            finally:
+                task_context.reset(token)
 
         return self.result
 
@@ -316,6 +325,11 @@ def __repr__(self):
         return f"OpTask({self.op}, sol_keys={sol_items!r})"
 
 
+#: Populated with the :class:`_OpTask` for the currently executing operation.
+#: It does not work for :term:`parallel execution`.
+task_context: ContextVar[_OpTask] = ContextVar("task_context")
+
+
 def _do_task(task):
     """
     Un-dill the *simpler* :class:`_OpTask` & Dill the results, to pass through pool-processes.

test/test_graphtik.py

@@ -34,7 +34,7 @@
     operations_reschedullled,
     tasks_marshalled,
 )
-from graphtik.execution import Solution
+from graphtik.execution import Solution, task_context
 from graphtik.base import Operation
 from graphtik.modifiers import dep_renamed, optional, sfx, sfxed, vararg
 from graphtik.op import NO_RESULT, NO_RESULT_BUT_SFX, operation
@@ -285,6 +285,27 @@ def powers_in_range(a, exponent):
     assert sol == exp
 
 
+def test_task_context(exemethod):
+    def check_task_context():
+        assert task_context.get().op == next(iop)
+
+    n_ops = 30
+    pipe = compose(
+        "t",
+        *(operation(check_task_context, f"op{i}", provides="a") for i in range(n_ops)),
+        parallel=exemethod,
+    )
+    iop = iter(pipe.ops)
+    if exemethod and is_marshal_tasks():
+        with pytest.raises(AssertionError, match="^assert FunctionalOperation"):
+            pipe.compute()
+        raise pytest.xfail("Cannot marshal `task_context` :-(.")
+    else:
+        pipe.compute()
+        with pytest.raises(StopIteration):
+            next(iop)
+
+
 def test_compose_rename_dict(caplog):
     pip = compose(
         "t",