Public cancellation API for running workflows

[FernandoCelmer]

Labels: enhancement, discovery

Context

A workflow started with mode="background" or mode="parallel" can run for hours. There is no public method to ask it to stop short of sending SIGKILL to the parent process. Users have asked for this in informal channels but no issue tracks it.

Cancellation is also a prerequisite for #233 (hard timeout via subprocess) and for any future scheduling story where a long task must yield to a higher-priority one.

Concept

Manager.cancel() for cooperative cancellation, Manager.terminate() for hard kill. Cancellation propagates to running steps via a cancelled flag on the task argument that steps already receive.

API sketch

manager = workflow.start(mode="background")
...
manager.cancel(timeout=10)   # cooperative; sets task.cancelled, joins
manager.terminate()          # SIGTERM the worker subprocess

@action
def long_running(task):
    for chunk in stream:
        if task.cancelled:
            return                 # graceful exit
        process(chunk)

Implementation sketch

• A shared multiprocessing.Event is created per workflow run.
• Each subprocess receives the event and exposes a cancelled
  property on task that proxies event.is_set().
• Manager.cancel(timeout=...) sets the event and joins each
  subprocess up to timeout seconds.
• Manager.terminate() sends SIGTERM via process.terminate(),
  then SIGKILL after a grace period.

Acceptance criteria

• Manager.cancel(timeout: float = 0) method
• Manager.terminate() method
• task.cancelled available inside steps
• Cancelled tasks land in TypeStatus.CANCELLED (new status)
• Test: a long sequential workflow stops within timeout after
  cancel()
• Test: terminate() kills a hung subprocess

Related

• Enhancement: Use multiprocessing for timeout to enable hard kill of stuck tasks #233 hard timeout
• Develop #1 (this batch) typed DotflowError adds Cancelled(DotflowError)

---