_background_cleanup silently drops callbacks when config=None

[FernandoCelmer]

Description

In Manager.__init__ (core/workflow.py, lines 155-163), the background cleanup thread is only started when self.config is truthy (elif self.config:). When no config is provided (config=None), the _background_cleanup function is never executed.

This means _callback_workflow is never called, and on_success/on_failure callbacks are silently dropped in background mode without config.

Location

dotflow/core/workflow.py — lines 155-163

Current Behavior

if mode != TypeExecution.BACKGROUND:
    self._callback_workflow(tasks=self.tasks)
elif self.config:
    def _background_cleanup():
        self.thread.join()
        self._callback_workflow(tasks=self.tasks)
    threading.Thread(target=_background_cleanup, daemon=True).start()

When config=None, neither branch fires _callback_workflow for background mode.

Expected Behavior

if mode != TypeExecution.BACKGROUND:
    self._callback_workflow(tasks=self.tasks)
else:
    def _background_cleanup():
        self.thread.join()
        self._callback_workflow(tasks=self.tasks)
    threading.Thread(target=_background_cleanup, daemon=True).start()

Always run _background_cleanup for background mode, regardless of config.

Impact

• Severity: Critical
• Symptom: on_success/on_failure callbacks silently never execute in background mode without config
• Tracer, metrics, and server finalization hooks are also skipped

Priority

Critical

[FernandoCelmer]

Adjacent gap: daemon thread can drop work on interpreter exit

While fixing _background_cleanup for config=None, the same code path has a separate durability problem. dotflow/core/workflow.py:391-395:

def run(self):
    self.thread = threading.Thread(target=self._run_sequential, daemon=True)
    self.thread.start()

daemon=True means the thread is killed when the main interpreter exits. In a script that calls workflow.start(mode="background") and returns, the thread is killed mid-task — checkpoints, logs, and metrics never flush.

Suggested complement

Register an atexit handler that joins with a configurable timeout:

import atexit

def run(self):
    self.thread = threading.Thread(target=self._run_sequential, daemon=True)
    self.thread.start()
    atexit.register(self._wait_completion, timeout=30)

def _wait_completion(self, timeout):
    if self.thread.is_alive():
        self.thread.join(timeout=timeout)

Or expose a daemon=False option for users who want the strict guarantee at the cost of blocking interpreter exit.

This is closely related to the cooperative cancellation work in #292 and the bug being fixed here.