meerschaum.utils.daemon.Daemon

   1#! /usr/bin/env python3
   2# -*- coding: utf-8 -*-
   3# vim:fenc=utf-8
   4
   5"""
   6Manage running daemons via the Daemon class.
   7"""
   8
   9from __future__ import annotations
  10import os
  11import importlib
  12import pathlib
  13import json
  14import shutil
  15import signal
  16import time
  17import traceback
  18from functools import partial
  19from datetime import datetime, timezone
  20
  21import meerschaum as mrsm
  22from meerschaum.utils.typing import (
  23    Optional, Dict, Any, SuccessTuple, Callable, List, Union,
  24    is_success_tuple, Tuple,
  25)
  26from meerschaum.config import get_config
  27from meerschaum._internal.static import STATIC_CONFIG
  28from meerschaum.config._patch import apply_patch_to_config
  29from meerschaum.utils.warnings import warn, error
  30from meerschaum.utils.packages import attempt_import
  31from meerschaum.utils.venv import venv_exec
  32from meerschaum.utils.daemon._names import get_new_daemon_name
  33from meerschaum.utils.daemon.RotatingFile import RotatingFile
  34from meerschaum.utils.daemon.StdinFile import StdinFile
  35from meerschaum.utils.threading import RepeatTimer
  36from meerschaum.__main__ import _close_pools
  37
  38_daemons = []
  39_results = {}
  40
  41class Daemon:
  42    """
  43    Daemonize Python functions into background processes.
  44
  45    Examples
  46    --------
  47    >>> import meerschaum as mrsm
  48    >>> from meerschaum.utils.daemons import Daemon
  49    >>> daemon = Daemon(print, ('hi',))
  50    >>> success, msg = daemon.run()
  51    >>> print(daemon.log_text)
  52
  53    2024-07-29 18:03 | hi
  54    2024-07-29 18:03 |
  55    >>> daemon.run(allow_dirty_run=True)
  56    >>> print(daemon.log_text)
  57
  58    2024-07-29 18:03 | hi
  59    2024-07-29 18:03 |
  60    2024-07-29 18:05 | hi
  61    2024-07-29 18:05 |
  62    >>> mrsm.pprint(daemon.properties)
  63    {
  64        'label': 'print',
  65        'target': {'name': 'print', 'module': 'builtins', 'args': ['hi'], 'kw': {}},
  66        'result': None,
  67        'process': {'ended': '2024-07-29T18:03:33.752806'}
  68    }
  69
  70    """
  71
  72    def __new__(
  73        cls,
  74        *args,
  75        daemon_id: Optional[str] = None,
  76        **kw
  77    ):
  78        """
  79        If a daemon_id is provided and already exists, read from its pickle file.
  80        """
  81        instance = super(Daemon, cls).__new__(cls)
  82        if daemon_id is not None:
  83            instance.daemon_id = daemon_id
  84            if instance.pickle_path.exists():
  85                instance = instance.read_pickle()
  86        return instance
  87
  88    @classmethod
  89    def from_properties_file(cls, daemon_id: str) -> Daemon:
  90        """
  91        Return a Daemon from a properties dictionary.
  92        """
  93        properties_path = cls._get_properties_path_from_daemon_id(daemon_id)
  94        if not properties_path.exists():
  95            raise OSError(f"Properties file '{properties_path}' does not exist.")
  96
  97        try:
  98            with open(properties_path, 'r', encoding='utf-8') as f:
  99                properties = json.load(f)
 100        except Exception:
 101            properties = {}
 102
 103        if not properties:
 104            raise ValueError(f"No properties could be read for daemon '{daemon_id}'.")
 105
 106        daemon_id = properties_path.parent.name
 107        target_cf = properties.get('target', {})
 108        target_module_name = target_cf.get('module', None)
 109        target_function_name = target_cf.get('name', None)
 110        target_args = target_cf.get('args', None)
 111        target_kw = target_cf.get('kw', None)
 112        label = properties.get('label', None)
 113
 114        if None in [
 115            target_module_name,
 116            target_function_name,
 117            target_args,
 118            target_kw,
 119        ]:
 120            raise ValueError("Missing target function information.")
 121
 122        target_module = importlib.import_module(target_module_name)
 123        target_function = getattr(target_module, target_function_name)
 124
 125        return Daemon(
 126            daemon_id=daemon_id,
 127            target=target_function,
 128            target_args=target_args,
 129            target_kw=target_kw,
 130            properties=properties,
 131            label=label,
 132        )
 133
 134
 135    def __init__(
 136        self,
 137        target: Optional[Callable[[Any], Any]] = None,
 138        target_args: Union[List[Any], Tuple[Any], None] = None,
 139        target_kw: Optional[Dict[str, Any]] = None,
 140        env: Optional[Dict[str, str]] = None,
 141        daemon_id: Optional[str] = None,
 142        label: Optional[str] = None,
 143        properties: Optional[Dict[str, Any]] = None,
 144        pickle: bool = True,
 145    ):
 146        """
 147        Parameters
 148        ----------
 149        target: Optional[Callable[[Any], Any]], default None,
 150            The function to execute in a child process.
 151
 152        target_args: Union[List[Any], Tuple[Any], None], default None
 153            Positional arguments to pass to the target function.
 154
 155        target_kw: Optional[Dict[str, Any]], default None
 156            Keyword arguments to pass to the target function.
 157
 158        env: Optional[Dict[str, str]], default None
 159            If provided, set these environment variables in the daemon process.
 160
 161        daemon_id: Optional[str], default None
 162            Build a `Daemon` from an existing `daemon_id`.
 163            If `daemon_id` is provided, other arguments are ignored and are derived
 164            from the existing pickled `Daemon`.
 165
 166        label: Optional[str], default None
 167            Label string to help identifiy a daemon.
 168            If `None`, use the function name instead.
 169
 170        properties: Optional[Dict[str, Any]], default None
 171            Override reading from the properties JSON by providing an existing dictionary.
 172        """
 173        _pickle = self.__dict__.get('_pickle', False)
 174        if daemon_id is not None:
 175            self.daemon_id = daemon_id
 176            if not self.pickle_path.exists() and not target and ('target' not in self.__dict__):
 177
 178                if not self.properties_path.exists():
 179                    raise Exception(
 180                        f"Daemon '{self.daemon_id}' does not exist. "
 181                        + "Pass a target to create a new Daemon."
 182                    )
 183
 184                try:
 185                    new_daemon = self.from_properties_file(daemon_id)
 186                except Exception:
 187                    new_daemon = None
 188
 189                if new_daemon is not None:
 190                    new_daemon.write_pickle()
 191                    target = new_daemon.target
 192                    target_args = new_daemon.target_args
 193                    target_kw = new_daemon.target_kw
 194                    label = new_daemon.label
 195                    self._properties = new_daemon.properties
 196                else:
 197                    try:
 198                        self.properties_path.unlink()
 199                    except Exception:
 200                        pass
 201
 202                    raise Exception(
 203                        f"Could not recover daemon '{self.daemon_id}' "
 204                        + "from its properties file."
 205                    )
 206
 207        if 'target' not in self.__dict__:
 208            if target is None:
 209                error("Cannot create a Daemon without a target.")
 210            self.target = target
 211
 212        self.pickle = pickle
 213
 214        ### NOTE: We have to check self.__dict__ in case we un-pickling.
 215        if '_target_args' not in self.__dict__:
 216            self._target_args = target_args
 217        if '_target_kw' not in self.__dict__:
 218            self._target_kw = target_kw
 219
 220        if 'label' not in self.__dict__:
 221            if label is None:
 222                label = (
 223                    self.target.__name__ if '__name__' in self.target.__dir__()
 224                        else str(self.target)
 225                )
 226            self.label = label
 227        elif label is not None:
 228            self.label = label
 229
 230        if 'daemon_id' not in self.__dict__:
 231            self.daemon_id = get_new_daemon_name()
 232        if '_properties' not in self.__dict__:
 233            self._properties = properties
 234        elif properties:
 235            if self._properties is None:
 236                self._properties = {}
 237            self._properties.update(properties)
 238        if self._properties is None:
 239            self._properties = {}
 240
 241        self._properties.update({'label': self.label})
 242        if env:
 243            self._properties.update({'env': env})
 244
 245        ### Instantiate the process and if it doesn't exist, make sure the PID is removed.
 246        _ = self.process
 247
 248
 249    def _run_exit(
 250        self,
 251        keep_daemon_output: bool = True,
 252        allow_dirty_run: bool = False,
 253    ) -> Any:
 254        """Run the daemon's target function.
 255        NOTE: This WILL EXIT the parent process!
 256
 257        Parameters
 258        ----------
 259        keep_daemon_output: bool, default True
 260            If `False`, delete the daemon's output directory upon exiting.
 261
 262        allow_dirty_run, bool, default False:
 263            If `True`, run the daemon, even if the `daemon_id` directory exists.
 264            This option is dangerous because if the same `daemon_id` runs twice,
 265            the last to finish will overwrite the output of the first.
 266
 267        Returns
 268        -------
 269        Nothing — this will exit the parent process.
 270        """
 271        import platform
 272        import sys
 273        import os
 274        import traceback
 275        from meerschaum.utils.warnings import warn
 276        from meerschaum.config import get_config
 277        daemon = attempt_import('daemon')
 278        lines = get_config('jobs', 'terminal', 'lines')
 279        columns = get_config('jobs', 'terminal', 'columns')
 280
 281        if platform.system() == 'Windows':
 282            return False, "Windows is no longer supported."
 283
 284        self._setup(allow_dirty_run)
 285
 286        _daemons.append(self)
 287
 288        logs_cf = self.properties.get('logs', {})
 289        log_refresh_seconds = logs_cf.get('refresh_files_seconds', None)
 290        if log_refresh_seconds is None:
 291            log_refresh_seconds = get_config('jobs', 'logs', 'refresh_files_seconds')
 292        write_timestamps = logs_cf.get('write_timestamps', None)
 293        if write_timestamps is None:
 294            write_timestamps = get_config('jobs', 'logs', 'timestamps', 'enabled')
 295
 296        self._log_refresh_timer = RepeatTimer(
 297            log_refresh_seconds,
 298            partial(self.rotating_log.refresh_files, start_interception=write_timestamps),
 299        )
 300
 301        capture_stdin = logs_cf.get('stdin', True)
 302        cwd = self.properties.get('cwd', os.getcwd())
 303
 304        ### NOTE: The SIGINT handler has been removed so that child processes may handle
 305        ###       KeyboardInterrupts themselves.
 306        ###       The previous aggressive approach was redundant because of the SIGTERM handler.
 307        self._daemon_context = daemon.DaemonContext(
 308            pidfile=self.pid_lock,
 309            stdout=self.rotating_log,
 310            stderr=self.rotating_log,
 311            stdin=(self.stdin_file if capture_stdin else None),
 312            working_directory=cwd,
 313            detach_process=True,
 314            files_preserve=list(self.rotating_log.subfile_objects.values()),
 315            signal_map={
 316                signal.SIGTERM: self._handle_sigterm,
 317            },
 318        )
 319
 320        if capture_stdin and sys.stdin is None:
 321            raise OSError("Cannot daemonize without stdin.")
 322
 323        try:
 324            os.environ['LINES'], os.environ['COLUMNS'] = str(int(lines)), str(int(columns))
 325            with self._daemon_context:
 326                if capture_stdin:
 327                    sys.stdin = self.stdin_file
 328                _ = os.environ.pop(STATIC_CONFIG['environment']['systemd_stdin_path'], None)
 329                os.environ[STATIC_CONFIG['environment']['daemon_id']] = self.daemon_id
 330                os.environ['PYTHONUNBUFFERED'] = '1'
 331
 332                ### Allow the user to override environment variables.
 333                env = self.properties.get('env', {})
 334                if env and isinstance(env, dict):
 335                    os.environ.update({str(k): str(v) for k, v in env.items()})
 336
 337                self.rotating_log.refresh_files(start_interception=True)
 338                result = None
 339                try:
 340                    with open(self.pid_path, 'w+', encoding='utf-8') as f:
 341                        f.write(str(os.getpid()))
 342
 343                    ### NOTE: The timer fails to start for remote actions to localhost.
 344                    try:
 345                        if not self._log_refresh_timer.is_running():
 346                            self._log_refresh_timer.start()
 347                    except Exception:
 348                        pass
 349
 350                    self.properties['result'] = None
 351                    self._capture_process_timestamp('began')
 352                    result = self.target(*self.target_args, **self.target_kw)
 353                    self.properties['result'] = result
 354                except SystemExit as e:
 355                    result = (
 356                        (True, f"Stopped daemon '{self.daemon_id}'.")
 357                        if e.code is None or e.code == 0
 358                        else (False, traceback.format_exc())
 359                    )
 360                except (BrokenPipeError, KeyboardInterrupt):
 361                    result = False, traceback.format_exc()
 362                except Exception as e:
 363                    warn(
 364                        f"Exception in daemon target function: {traceback.format_exc()}",
 365                    )
 366                    result = False, str(e)
 367                finally:
 368                    _results[self.daemon_id] = result
 369                    self.properties['result'] = result
 370
 371                    if keep_daemon_output:
 372                        self._capture_process_timestamp('ended')
 373                    else:
 374                        self.cleanup()
 375
 376                    self._log_refresh_timer.cancel()
 377                    try:
 378                        if self.pid is None and self.pid_path.exists():
 379                            self.pid_path.unlink()
 380                    except Exception:
 381                        pass
 382
 383                    if is_success_tuple(result):
 384                        try:
 385                            mrsm.pprint(result)
 386                        except BrokenPipeError:
 387                            pass
 388
 389        except Exception:
 390            daemon_error = traceback.format_exc()
 391            import meerschaum.config.paths as paths
 392            with open(paths.DAEMON_ERROR_LOG_PATH, 'a+', encoding='utf-8') as f:
 393                f.write(
 394                    f"Error in Daemon '{self}':\n\n"
 395                    f"{sys.stdin=}\n"
 396                    f"{self.stdin_file_path=}\n"
 397                    f"{self.stdin_file_path.exists()=}\n\n"
 398                    f"{daemon_error}\n\n"
 399                )
 400            warn(f"Encountered an error while running the daemon '{self}':\n{daemon_error}")
 401
 402    def _capture_process_timestamp(
 403        self,
 404        process_key: str,
 405        write_properties: bool = True,
 406    ) -> None:
 407        """
 408        Record the current timestamp to the parameters `process:<process_key>`.
 409
 410        Parameters
 411        ----------
 412        process_key: str
 413            Under which key to store the timestamp.
 414
 415        write_properties: bool, default True
 416            If `True` persist the properties to disk immediately after capturing the timestamp.
 417        """
 418        if 'process' not in self.properties:
 419            self.properties['process'] = {}
 420
 421        if process_key not in ('began', 'ended', 'paused', 'stopped'):
 422            raise ValueError(f"Invalid key '{process_key}'.")
 423
 424        self.properties['process'][process_key] = (
 425            datetime.now(timezone.utc).replace(tzinfo=None).isoformat()
 426        )
 427        if write_properties:
 428            self.write_properties()
 429
 430    def run(
 431        self,
 432        keep_daemon_output: bool = True,
 433        allow_dirty_run: bool = False,
 434        wait: bool = False,
 435        timeout: Union[int, float] = 4,
 436        debug: bool = False,
 437    ) -> SuccessTuple:
 438        """Run the daemon as a child process and continue executing the parent.
 439
 440        Parameters
 441        ----------
 442        keep_daemon_output: bool, default True
 443            If `False`, delete the daemon's output directory upon exiting.
 444
 445        allow_dirty_run: bool, default False
 446            If `True`, run the daemon, even if the `daemon_id` directory exists.
 447            This option is dangerous because if the same `daemon_id` runs concurrently,
 448            the last to finish will overwrite the output of the first.
 449
 450        wait: bool, default True
 451            If `True`, block until `Daemon.status` is running (or the timeout expires).
 452
 453        timeout: Union[int, float], default 4
 454            If `wait` is `True`, block for up to `timeout` seconds before returning a failure.
 455
 456        Returns
 457        -------
 458        A SuccessTuple indicating success.
 459
 460        """
 461        import platform
 462        if platform.system() == 'Windows':
 463            return False, "Cannot run background jobs on Windows."
 464
 465        ### The daemon might exist and be paused.
 466        if self.status == 'paused':
 467            return self.resume()
 468
 469        self._remove_stop_file()
 470        if self.status == 'running':
 471            return True, f"Daemon '{self}' is already running."
 472
 473        self.mkdir_if_not_exists(allow_dirty_run)
 474        _write_pickle_success_tuple = self.write_pickle()
 475        if not _write_pickle_success_tuple[0]:
 476            return _write_pickle_success_tuple
 477
 478        _launch_daemon_code = (
 479            "from meerschaum.utils.daemon import Daemon, _daemons; "
 480            f"daemon = Daemon(daemon_id='{self.daemon_id}'); "
 481            f"_daemons['{self.daemon_id}'] = daemon; "
 482            f"daemon._run_exit(keep_daemon_output={keep_daemon_output}, "
 483            "allow_dirty_run=True)"
 484        )
 485        env = dict(os.environ)
 486        _launch_success_bool = venv_exec(_launch_daemon_code, debug=debug, venv=None, env=env)
 487        msg = (
 488            "Success"
 489            if _launch_success_bool
 490            else f"Failed to start daemon '{self.daemon_id}'."
 491        )
 492        if not wait or not _launch_success_bool:
 493            return _launch_success_bool, msg
 494
 495        timeout = self.get_timeout_seconds(timeout)
 496        check_timeout_interval = self.get_check_timeout_interval_seconds()
 497
 498        if not timeout:
 499            success = self.status == 'running'
 500            msg = "Success" if success else f"Failed to run daemon '{self.daemon_id}'."
 501            if success:
 502                self._capture_process_timestamp('began')
 503            return success, msg
 504
 505        begin = time.perf_counter()
 506        while (time.perf_counter() - begin) < timeout:
 507            if self.status == 'running':
 508                self._capture_process_timestamp('began')
 509                return True, "Success"
 510            time.sleep(check_timeout_interval)
 511
 512        return False, (
 513            f"Failed to start daemon '{self.daemon_id}' within {timeout} second"
 514            + ('s' if timeout != 1 else '') + '.'
 515        )
 516
 517
 518    def kill(self, timeout: Union[int, float, None] = 8) -> SuccessTuple:
 519        """
 520        Forcibly terminate a running daemon.
 521        Sends a SIGTERM signal to the process.
 522
 523        Parameters
 524        ----------
 525        timeout: Optional[int], default 3
 526            How many seconds to wait for the process to terminate.
 527
 528        Returns
 529        -------
 530        A SuccessTuple indicating success.
 531        """
 532        ### A lost PID file means a detached/orphaned daemon — possibly SEVERAL processes
 533        ### sharing this daemon_id (e.g. accumulated across crashed restarts). Reap them
 534        ### ALL by daemon_id rather than reporting a false "stopped" and stranding them
 535        ### (which forces a manual `pkill meerschaum` or hunting the daemon_id in htop).
 536        if not self.pid_path.exists():
 537            reaped = self._kill_detached_processes(timeout)
 538            self._write_stop_file('kill')
 539            self.stdin_file.close()
 540            self._remove_blocking_stdin_file()
 541            return True, (
 542                (f"Reaped {reaped} detached process" + ('es' if reaped != 1 else '') + '.')
 543                if reaped
 544                else "Process has already stopped."
 545            )
 546
 547        if self.status != 'paused':
 548            success, msg = self._send_signal(signal.SIGTERM, timeout=timeout)
 549            if success:
 550                ### Sweep any sibling/detached processes left over for this daemon_id.
 551                self._kill_detached_processes(timeout)
 552                self._write_stop_file('kill')
 553                self.stdin_file.close()
 554                self._remove_blocking_stdin_file()
 555                return success, msg
 556
 557        if self.status == 'stopped':
 558            reaped = self._kill_detached_processes(timeout)
 559            self._write_stop_file('kill')
 560            self.stdin_file.close()
 561            self._remove_blocking_stdin_file()
 562            return True, (
 563                (f"Reaped {reaped} detached process" + ('es' if reaped != 1 else '') + '.')
 564                if reaped
 565                else "Process has already stopped."
 566            )
 567
 568        psutil = attempt_import('psutil')
 569        process = self.process
 570        try:
 571            process.terminate()
 572            process.kill()
 573            process.wait(timeout=timeout)
 574        except Exception as e:
 575            return False, f"Failed to kill job {self} ({process}) with exception: {e}"
 576
 577        try:
 578            if process.status():
 579                return False, "Failed to stop daemon '{self}' ({process})."
 580        except psutil.NoSuchProcess:
 581            pass
 582
 583        if self.pid_path.exists():
 584            try:
 585                self.pid_path.unlink()
 586            except Exception:
 587                pass
 588
 589        self._write_stop_file('kill')
 590        self.stdin_file.close()
 591        self._remove_blocking_stdin_file()
 592        return True, "Success"
 593
 594    def quit(self, timeout: Union[int, float, None] = None) -> SuccessTuple:
 595        """Gracefully quit a running daemon."""
 596        if self.status == 'paused':
 597            return self.kill(timeout)
 598
 599        ### A lost PID file means a detached/orphaned daemon (possibly several processes);
 600        ### the normal signal path can't see them, so reap them all by daemon_id instead
 601        ### of returning a false "not running".
 602        if not self.pid_path.exists():
 603            reaped = self._kill_detached_processes(timeout)
 604            self._write_stop_file('quit')
 605            self.stdin_file.close()
 606            self._remove_blocking_stdin_file()
 607            return True, (
 608                (f"Reaped {reaped} detached process" + ('es' if reaped != 1 else '') + '.')
 609                if reaped
 610                else "Process is not running."
 611            )
 612
 613        signal_success, signal_msg = self._send_signal(signal.SIGINT, timeout=timeout)
 614        if signal_success:
 615            self._write_stop_file('quit')
 616            self.stdin_file.close()
 617            self._remove_blocking_stdin_file()
 618        return signal_success, signal_msg
 619
 620    def pause(
 621        self,
 622        timeout: Union[int, float, None] = None,
 623        check_timeout_interval: Union[float, int, None] = None,
 624    ) -> SuccessTuple:
 625        """
 626        Pause the daemon if it is running.
 627
 628        Parameters
 629        ----------
 630        timeout: Union[float, int, None], default None
 631            The maximum number of seconds to wait for a process to suspend.
 632
 633        check_timeout_interval: Union[float, int, None], default None
 634            The number of seconds to wait between checking if the process is still running.
 635
 636        Returns
 637        -------
 638        A `SuccessTuple` indicating whether the `Daemon` process was successfully suspended.
 639        """
 640        self._remove_blocking_stdin_file()
 641
 642        if self.process is None:
 643            return False, f"Daemon '{self.daemon_id}' is not running and cannot be paused."
 644
 645        if self.status == 'paused':
 646            return True, f"Daemon '{self.daemon_id}' is already paused."
 647
 648        self._write_stop_file('pause')
 649        self.stdin_file.close()
 650        self._remove_blocking_stdin_file()
 651        try:
 652            self.process.suspend()
 653        except Exception as e:
 654            return False, f"Failed to pause daemon '{self.daemon_id}':\n{e}"
 655
 656        timeout = self.get_timeout_seconds(timeout)
 657        check_timeout_interval = self.get_check_timeout_interval_seconds(
 658            check_timeout_interval
 659        )
 660
 661        psutil = attempt_import('psutil')
 662
 663        if not timeout:
 664            try:
 665                success = self.process.status() == 'stopped'
 666            except psutil.NoSuchProcess:
 667                success = True
 668            msg = "Success" if success else f"Failed to suspend daemon '{self.daemon_id}'."
 669            if success:
 670                self._capture_process_timestamp('paused')
 671            return success, msg
 672
 673        begin = time.perf_counter()
 674        while (time.perf_counter() - begin) < timeout:
 675            try:
 676                if self.process.status() == 'stopped':
 677                    self._capture_process_timestamp('paused')
 678                    return True, "Success"
 679            except psutil.NoSuchProcess as e:
 680                return False, f"Process exited unexpectedly. Was it killed?\n{e}"
 681            time.sleep(check_timeout_interval)
 682
 683        return False, (
 684            f"Failed to pause daemon '{self.daemon_id}' within {timeout} second"
 685            + ('s' if timeout != 1 else '') + '.'
 686        )
 687
 688    def resume(
 689        self,
 690        timeout: Union[int, float, None] = None,
 691        check_timeout_interval: Union[float, int, None] = None,
 692    ) -> SuccessTuple:
 693        """
 694        Resume the daemon if it is paused.
 695
 696        Parameters
 697        ----------
 698        timeout: Union[float, int, None], default None
 699            The maximum number of seconds to wait for a process to resume.
 700
 701        check_timeout_interval: Union[float, int, None], default None
 702            The number of seconds to wait between checking if the process is still stopped.
 703
 704        Returns
 705        -------
 706        A `SuccessTuple` indicating whether the `Daemon` process was successfully resumed.
 707        """
 708        if self.status == 'running':
 709            return True, f"Daemon '{self.daemon_id}' is already running."
 710
 711        if self.status == 'stopped':
 712            return False, f"Daemon '{self.daemon_id}' is stopped and cannot be resumed."
 713
 714        self._remove_stop_file()
 715        try:
 716            if self.process is None:
 717                return False, f"Cannot resume daemon '{self.daemon_id}'."
 718
 719            self.process.resume()
 720        except Exception as e:
 721            return False, f"Failed to resume daemon '{self.daemon_id}':\n{e}"
 722
 723        timeout = self.get_timeout_seconds(timeout)
 724        check_timeout_interval = self.get_check_timeout_interval_seconds(
 725            check_timeout_interval
 726        )
 727
 728        if not timeout:
 729            success = self.status == 'running'
 730            msg = "Success" if success else f"Failed to resume daemon '{self.daemon_id}'."
 731            if success:
 732                self._capture_process_timestamp('began')
 733            return success, msg
 734
 735        begin = time.perf_counter()
 736        while (time.perf_counter() - begin) < timeout:
 737            if self.status == 'running':
 738                self._capture_process_timestamp('began')
 739                return True, "Success"
 740            time.sleep(check_timeout_interval)
 741
 742        return False, (
 743            f"Failed to resume daemon '{self.daemon_id}' within {timeout} second"
 744            + ('s' if timeout != 1 else '') + '.'
 745        )
 746
 747    def _write_stop_file(self, action: str) -> SuccessTuple:
 748        """Write the stop file timestamp and action."""
 749        if action not in ('quit', 'kill', 'pause'):
 750            return False, f"Unsupported action '{action}'."
 751
 752        if not self.stop_path.parent.exists():
 753            self.stop_path.parent.mkdir(parents=True, exist_ok=True)
 754
 755        with open(self.stop_path, 'w+', encoding='utf-8') as f:
 756            json.dump(
 757                {
 758                    'stop_time': datetime.now(timezone.utc).isoformat(),
 759                    'action': action,
 760                },
 761                f
 762            )
 763
 764        return True, "Success"
 765
 766    def _remove_stop_file(self) -> SuccessTuple:
 767        """Remove the stop file"""
 768        if not self.stop_path.exists():
 769            return True, "Stop file does not exist."
 770
 771        try:
 772            self.stop_path.unlink()
 773        except Exception as e:
 774            return False, f"Failed to remove stop file:\n{e}"
 775
 776        return True, "Success"
 777
 778    def _read_stop_file(self) -> Dict[str, Any]:
 779        """
 780        Read the stop file if it exists.
 781        """
 782        if not self.stop_path.exists():
 783            return {}
 784
 785        try:
 786            with open(self.stop_path, 'r', encoding='utf-8') as f:
 787                data = json.load(f)
 788            return data
 789        except Exception:
 790            return {}
 791
 792    def _remove_blocking_stdin_file(self) -> mrsm.SuccessTuple:
 793        """
 794        Remove the blocking STDIN file if it exists.
 795        """
 796        try:
 797            if self.blocking_stdin_file_path.exists():
 798                self.blocking_stdin_file_path.unlink()
 799        except Exception as e:
 800            return False, str(e)
 801
 802        return True, "Success"
 803
 804    def _handle_sigterm(self, signal_number: int, stack_frame: 'frame') -> None:
 805        """
 806        Handle `SIGTERM` within the `Daemon` context.
 807        This method is injected into the `DaemonContext`.
 808        """
 809        from meerschaum.utils.process import signal_handler
 810        from meerschaum.utils.threading import request_stop, interrupt_threads
 811        signal_handler(signal_number, stack_frame)
 812
 813        ### Tell cooperative loops (e.g. `sync pipes`) to stop, then actively unwind
 814        ### any worker threads so they cannot keep the process alive as a zombie.
 815        request_stop()
 816        interrupt_threads(SystemExit)
 817
 818        timer = self.__dict__.get('_log_refresh_timer', None)
 819        if timer is not None:
 820            timer.cancel()
 821
 822        daemon_context = self.__dict__.get('_daemon_context', None)
 823        if daemon_context is not None:
 824            daemon_context.close()
 825
 826        _close_pools()
 827        raise SystemExit(0)
 828
 829    def _send_signal(
 830        self,
 831        signal_to_send,
 832        timeout: Union[float, int, None] = None,
 833        check_timeout_interval: Union[float, int, None] = None,
 834    ) -> SuccessTuple:
 835        """Send a signal to the daemon process.
 836
 837        Parameters
 838        ----------
 839        signal_to_send:
 840            The signal the send to the daemon, e.g. `signals.SIGINT`.
 841
 842        timeout: Union[float, int, None], default None
 843            The maximum number of seconds to wait for a process to terminate.
 844
 845        check_timeout_interval: Union[float, int, None], default None
 846            The number of seconds to wait between checking if the process is still running.
 847
 848        Returns
 849        -------
 850        A SuccessTuple indicating success.
 851        """
 852        try:
 853            pid = self.pid
 854            if pid is None:
 855                return (
 856                    False,
 857                    f"Daemon '{self.daemon_id}' is not running, "
 858                    + f"cannot send signal '{signal_to_send}'."
 859                )
 860            
 861            os.kill(pid, signal_to_send)
 862        except Exception:
 863            return False, f"Failed to send signal {signal_to_send}:\n{traceback.format_exc()}"
 864
 865        timeout = self.get_timeout_seconds(timeout)
 866        check_timeout_interval = self.get_check_timeout_interval_seconds(
 867            check_timeout_interval
 868        )
 869
 870        if not timeout:
 871            return True, f"Successfully sent '{signal_to_send}' to daemon '{self.daemon_id}'."
 872
 873        begin = time.perf_counter()
 874        while (time.perf_counter() - begin) < timeout:
 875            if not self.status == 'running':
 876                return True, "Success"
 877            time.sleep(check_timeout_interval)
 878
 879        return False, (
 880            f"Failed to stop daemon '{self.daemon_id}' (PID: {pid}) within {timeout} second"
 881            + ('s' if timeout != 1 else '') + '.'
 882        )
 883
 884    def _find_detached_pids(self) -> List[int]:
 885        """
 886        Return the PIDs of any processes whose command line references this daemon_id.
 887
 888        A daemon is launched via `venv_exec` with `Daemon(daemon_id='<id>')` embedded in
 889        the executed code, so the daemon_id stays visible in the process's command line.
 890        When the PID file is lost (e.g. the launcher exited without cleanup, leaving an
 891        orphaned/detached daemon), this is the only reliable way to find the process —
 892        otherwise stopping the job reports a false "already stopped" and the user must
 893        resort to `pkill meerschaum` or hunting the daemon_id in `htop`.
 894        """
 895        psutil = attempt_import('psutil')
 896        marker = f"daemon_id='{self.daemon_id}'"
 897        my_pid = os.getpid()
 898        pids = []
 899        for proc in psutil.process_iter(['pid', 'cmdline']):
 900            try:
 901                if proc.info['pid'] == my_pid:
 902                    continue
 903                cmdline = proc.info.get('cmdline') or []
 904                if any(marker in (part or '') for part in cmdline):
 905                    pids.append(int(proc.info['pid']))
 906            except Exception:
 907                continue
 908        return pids
 909
 910    def _kill_detached_processes(self, timeout: Union[int, float, None] = None) -> int:
 911        """
 912        SIGTERM then SIGKILL any detached processes referencing this daemon_id.
 913
 914        Fallback for when the PID file is gone but the daemon (or its threads) are still
 915        alive. Returns the number of processes that were targeted.
 916        """
 917        psutil = attempt_import('psutil')
 918        pids = self._find_detached_pids()
 919        if not pids:
 920            return 0
 921
 922        procs = []
 923        for pid in pids:
 924            try:
 925                procs.append(psutil.Process(pid))
 926            except Exception:
 927                continue
 928        for proc in procs:
 929            try:
 930                proc.terminate()
 931            except Exception:
 932                pass
 933
 934        timeout = self.get_timeout_seconds(timeout)
 935        try:
 936            _, alive = psutil.wait_procs(procs, timeout=(timeout or 8))
 937        except Exception:
 938            alive = procs
 939        for proc in alive:
 940            try:
 941                proc.kill()
 942            except Exception:
 943                pass
 944        return len(procs)
 945
 946    def mkdir_if_not_exists(self, allow_dirty_run: bool = False):
 947        """Create the Daemon's directory.
 948        If `allow_dirty_run` is `False` and the directory already exists,
 949        raise a `FileExistsError`.
 950        """
 951        try:
 952            self.path.mkdir(parents=True, exist_ok=True)
 953            _already_exists = any(os.scandir(self.path))
 954        except FileExistsError:
 955            _already_exists = True
 956
 957        if _already_exists and not allow_dirty_run:
 958            error(
 959                f"Daemon '{self.daemon_id}' already exists. " +
 960                "To allow this daemon to run, do one of the following:\n"
 961                + "  - Execute `daemon.cleanup()`.\n"
 962                + f"  - Delete the directory '{self.path}'.\n"
 963                + "  - Pass `allow_dirty_run=True` to `daemon.run()`.\n",
 964                FileExistsError,
 965            )
 966
 967    @property
 968    def process(self) -> Union['psutil.Process', None]:
 969        """
 970        Return the psutil process for the Daemon.
 971        """
 972        psutil = attempt_import('psutil')
 973        pid = self.pid
 974        if pid is None:
 975            return None
 976        if '_process' not in self.__dict__ or self.__dict__['_process'].pid != int(pid):
 977            try:
 978                self._process = psutil.Process(int(pid))
 979                process_exists = True
 980            except Exception:
 981                process_exists = False
 982            if not process_exists:
 983                _ = self.__dict__.pop('_process', None)
 984                try:
 985                    if self.pid_path.exists():
 986                        self.pid_path.unlink()
 987                except Exception:
 988                    pass
 989                return None
 990        return self._process
 991
 992    @property
 993    def status(self) -> str:
 994        """
 995        Return the running status of this Daemon.
 996        """
 997        if self.process is None:
 998            return 'stopped'
 999
1000        psutil = attempt_import('psutil', lazy=False)
1001        try:
1002            if self.process.status() == 'stopped':
1003                return 'paused'
1004            if self.process.status() == 'zombie':
1005                raise psutil.NoSuchProcess(self.process.pid)
1006        except (psutil.NoSuchProcess, AttributeError):
1007            if self.pid_path.exists():
1008                try:
1009                    self.pid_path.unlink()
1010                except Exception:
1011                    pass
1012            return 'stopped'
1013
1014        return 'running'
1015
1016    @classmethod
1017    def _get_path_from_daemon_id(cls, daemon_id: str) -> pathlib.Path:
1018        """
1019        Return a Daemon's path from its `daemon_id`.
1020        """
1021        import meerschaum.config.paths as paths
1022        return paths.DAEMON_RESOURCES_PATH / daemon_id
1023
1024    @property
1025    def path(self) -> pathlib.Path:
1026        """
1027        Return the path for this Daemon's directory.
1028        """
1029        return self._get_path_from_daemon_id(self.daemon_id)
1030
1031    @classmethod
1032    def _get_properties_path_from_daemon_id(cls, daemon_id: str) -> pathlib.Path:
1033        """
1034        Return the `properties.json` path for a given `daemon_id`.
1035        """
1036        return cls._get_path_from_daemon_id(daemon_id) / 'properties.json'
1037
1038    @property
1039    def properties_path(self) -> pathlib.Path:
1040        """
1041        Return the `propterties.json` path for this Daemon.
1042        """
1043        return self._get_properties_path_from_daemon_id(self.daemon_id)
1044
1045    @property
1046    def stop_path(self) -> pathlib.Path:
1047        """
1048        Return the path for the stop file (created when manually stopped).
1049        """
1050        return self.path / '.stop.json'
1051
1052    @property
1053    def log_path(self) -> pathlib.Path:
1054        """
1055        Return the log path.
1056        """
1057        logs_cf = self.properties.get('logs', None) or {}
1058        if 'path' not in logs_cf:
1059            import meerschaum.config.paths as paths
1060            return paths.LOGS_RESOURCES_PATH / (self.daemon_id + '.log')
1061
1062        return pathlib.Path(logs_cf['path'])
1063
1064    @property
1065    def stdin_file_path(self) -> pathlib.Path:
1066        """
1067        Return the stdin file path.
1068        """
1069        return self.path / 'input.stdin'
1070
1071    @property
1072    def blocking_stdin_file_path(self) -> pathlib.Path:
1073        """
1074        Return the stdin file path.
1075        """
1076        if '_blocking_stdin_file_path' in self.__dict__:
1077            return self._blocking_stdin_file_path
1078
1079        return self.path / 'input.stdin.block'
1080
1081    @property
1082    def prompt_kwargs_file_path(self) -> pathlib.Path:
1083        """
1084        Return the file path to the kwargs for the invoking `prompt()`.
1085        """
1086        return self.path / 'prompt_kwargs.json'
1087
1088    @property
1089    def log_offset_path(self) -> pathlib.Path:
1090        """
1091        Return the log offset file path.
1092        """
1093        import meerschaum.config.paths as paths
1094        return paths.LOGS_RESOURCES_PATH / ('.' + self.daemon_id + '.log.offset')
1095
1096    @property
1097    def log_offset_lock(self) -> 'InterProcessLock':
1098        """
1099        Return the process lock context manager.
1100        """
1101        if '_log_offset_lock' in self.__dict__:
1102            return self._log_offset_lock
1103
1104        from meerschaum.utils.locks import InterProcessLock
1105        self._log_offset_lock = InterProcessLock(self.log_offset_path)
1106        return self._log_offset_lock
1107
1108    @property
1109    def rotating_log(self) -> RotatingFile:
1110        """
1111        The rotating log file for the daemon's output.
1112        """
1113        if '_rotating_log' in self.__dict__:
1114            return self._rotating_log
1115
1116        logs_cf = self.properties.get('logs', None) or {}
1117        write_timestamps = logs_cf.get('write_timestamps', None)
1118        if write_timestamps is None:
1119            write_timestamps = get_config('jobs', 'logs', 'timestamps', 'enabled')
1120
1121        timestamp_format = logs_cf.get('timestamp_format', None)
1122        if timestamp_format is None:
1123            timestamp_format = get_config('jobs', 'logs', 'timestamps', 'format')
1124
1125        num_files_to_keep = logs_cf.get('num_files_to_keep', None)
1126        if num_files_to_keep is None:
1127            num_files_to_keep = get_config('jobs', 'logs', 'num_files_to_keep')
1128
1129        max_file_size = logs_cf.get('max_file_size', None)
1130        if max_file_size is None:
1131            max_file_size = get_config('jobs', 'logs', 'max_file_size')
1132
1133        redirect_streams = logs_cf.get('redirect_streams', True)
1134
1135        self._rotating_log = RotatingFile(
1136            self.log_path,
1137            redirect_streams=redirect_streams,
1138            write_timestamps=write_timestamps,
1139            timestamp_format=timestamp_format,
1140            num_files_to_keep=num_files_to_keep,
1141            max_file_size=max_file_size,
1142        )
1143        return self._rotating_log
1144
1145    @property
1146    def stdin_file(self):
1147        """
1148        Return the file handler for the stdin file.
1149        """
1150        if (_stdin_file := self.__dict__.get('_stdin_file', None)):
1151            return _stdin_file
1152
1153        self._stdin_file = StdinFile(
1154            self.stdin_file_path,
1155            lock_file_path=self.blocking_stdin_file_path,
1156        )
1157        return self._stdin_file
1158
1159    @property
1160    def log_text(self) -> Union[str, None]:
1161        """
1162        Read the log files and return their contents.
1163        Returns `None` if the log file does not exist.
1164        """
1165        logs_cf = self.properties.get('logs', None) or {}
1166        write_timestamps = logs_cf.get('write_timestamps', None)
1167        if write_timestamps is None:
1168            write_timestamps = get_config('jobs', 'logs', 'timestamps', 'enabled')
1169
1170        timestamp_format = logs_cf.get('timestamp_format', None)
1171        if timestamp_format is None:
1172            timestamp_format = get_config('jobs', 'logs', 'timestamps', 'format')
1173
1174        num_files_to_keep = logs_cf.get('num_files_to_keep', None)
1175        if num_files_to_keep is None:
1176            num_files_to_keep = get_config('jobs', 'logs', 'num_files_to_keep')
1177
1178        max_file_size = logs_cf.get('max_file_size', None)
1179        if max_file_size is None:
1180            max_file_size = get_config('jobs', 'logs', 'max_file_size')
1181
1182        new_rotating_log = RotatingFile(
1183            self.rotating_log.file_path,
1184            num_files_to_keep=num_files_to_keep,
1185            max_file_size=max_file_size,
1186            write_timestamps=write_timestamps,
1187            timestamp_format=timestamp_format,
1188        )
1189        return new_rotating_log.read()
1190
1191    def readlines(self) -> List[str]:
1192        """
1193        Read the next log lines, persisting the cursor for later use.
1194        Note this will alter the cursor of `self.rotating_log`.
1195        """
1196        self.rotating_log._cursor = self._read_log_offset()
1197        lines = self.rotating_log.readlines()
1198        self._write_log_offset()
1199        return lines
1200
1201    def _read_log_offset(self) -> Tuple[int, int]:
1202        """
1203        Return the current log offset cursor.
1204
1205        Returns
1206        -------
1207        A tuple of the form (`subfile_index`, `position`).
1208        """
1209        if not self.log_offset_path.exists():
1210            return 0, 0
1211
1212        try:
1213            with open(self.log_offset_path, 'r', encoding='utf-8') as f:
1214                cursor_text = f.read()
1215            cursor_parts = cursor_text.split(' ')
1216            subfile_index, subfile_position = int(cursor_parts[0]), int(cursor_parts[1])
1217            return subfile_index, subfile_position
1218        except Exception as e:
1219            warn(f"Failed to read cursor:\n{e}")
1220        return 0, 0
1221
1222    def _write_log_offset(self) -> None:
1223        """
1224        Write the current log offset file.
1225        """
1226        with self.log_offset_lock:
1227            with open(self.log_offset_path, 'w+', encoding='utf-8') as f:
1228                subfile_index = self.rotating_log._cursor[0]
1229                subfile_position = self.rotating_log._cursor[1]
1230                f.write(f"{subfile_index} {subfile_position}")
1231
1232    @property
1233    def pid(self) -> Union[int, None]:
1234        """
1235        Read the PID file and return its contents.
1236        Returns `None` if the PID file does not exist.
1237        """
1238        if not self.pid_path.exists():
1239            ### The PID file can be lost while a detached daemon keeps running (e.g. the
1240            ### launcher exited without cleanup). Recover the PID by finding the process
1241            ### whose command line embeds this daemon_id, so `status`/`stop`/`kill` see the
1242            ### live process instead of reporting a false "stopped" (which would strand the
1243            ### orphan, forcing a manual `pkill meerschaum`).
1244            detached_pids = self._find_detached_pids()
1245            return detached_pids[0] if detached_pids else None
1246        try:
1247            with open(self.pid_path, 'r', encoding='utf-8') as f:
1248                text = f.read()
1249            if len(text) == 0:
1250                return None
1251            pid = int(text.rstrip())
1252        except Exception as e:
1253            warn(e)
1254            text = None
1255            pid = None
1256        return pid
1257
1258    @property
1259    def pid_path(self) -> pathlib.Path:
1260        """
1261        Return the path to a file containing the PID for this Daemon.
1262        """
1263        return self.path / 'process.pid'
1264
1265    @property
1266    def pid_lock(self) -> 'InterProcessLock':
1267        """
1268        Return the process lock context manager.
1269        """
1270        if '_pid_lock' in self.__dict__:
1271            return self._pid_lock
1272
1273        from meerschaum.utils.locks import InterProcessLock
1274        self._pid_lock = InterProcessLock(self.pid_path)
1275        return self._pid_lock
1276
1277    @property
1278    def pickle_path(self) -> pathlib.Path:
1279        """
1280        Return the path for the pickle file.
1281        """
1282        return self.path / 'pickle.pkl'
1283
1284    def read_properties(self) -> Optional[Dict[str, Any]]:
1285        """Read the properties JSON file and return the dictionary."""
1286        if not self.properties_path.exists():
1287            return None
1288        try:
1289            with open(self.properties_path, 'r', encoding='utf-8') as file:
1290                properties = json.load(file)
1291        except Exception:
1292            properties = {}
1293        
1294        return properties or {}
1295
1296    def read_pickle(self) -> Daemon:
1297        """Read a Daemon's pickle file and return the `Daemon`."""
1298        import pickle
1299        import traceback
1300        if not self.pickle_path.exists():
1301            error(f"Pickle file does not exist for daemon '{self.daemon_id}'.")
1302
1303        if self.pickle_path.stat().st_size == 0:
1304            error(f"Pickle was empty for daemon '{self.daemon_id}'.")
1305
1306        try:
1307            with open(self.pickle_path, 'rb') as pickle_file:
1308                daemon = pickle.load(pickle_file)
1309            success, msg = True, 'Success'
1310        except Exception as e:
1311            success, msg = False, str(e)
1312            daemon = None
1313            traceback.print_exception(type(e), e, e.__traceback__)
1314        if not success:
1315            error(msg)
1316        return daemon
1317
1318    @property
1319    def properties(self) -> Dict[str, Any]:
1320        """
1321        Return the contents of the properties JSON file.
1322        """
1323        try:
1324            _file_properties = self.read_properties() or {}
1325        except Exception:
1326            traceback.print_exc()
1327            _file_properties = {}
1328
1329        if not self._properties:
1330            self._properties = _file_properties
1331
1332        if self._properties is None:
1333            self._properties = {}
1334
1335        if (
1336            self._properties.get('result', None) is None
1337            and _file_properties.get('result', None) is not None
1338        ):
1339            _ = self._properties.pop('result', None)
1340
1341        if _file_properties is not None:
1342            self._properties = apply_patch_to_config(
1343                _file_properties,
1344                self._properties,
1345            )
1346
1347        return self._properties
1348
1349    @property
1350    def hidden(self) -> bool:
1351        """
1352        Return a bool indicating whether this Daemon should be displayed.
1353        """
1354        return self.daemon_id.startswith('_') or self.daemon_id.startswith('.')
1355
1356    def write_properties(self) -> SuccessTuple:
1357        """Write the properties dictionary to the properties JSON file
1358        (only if self.properties exists).
1359        """
1360        from meerschaum.utils.misc import generate_password
1361        success, msg = (
1362            False,
1363            f"No properties to write for daemon '{self.daemon_id}'."
1364        )
1365        backup_path = self.properties_path.parent / (generate_password(8) + '.json')
1366        props = self.properties
1367        if props is not None:
1368            try:
1369                self.path.mkdir(parents=True, exist_ok=True)
1370                if self.properties_path.exists():
1371                    self.properties_path.rename(backup_path)
1372                with open(self.properties_path, 'w+', encoding='utf-8') as properties_file:
1373                    json.dump(props, properties_file)
1374                success, msg = True, 'Success'
1375            except Exception as e:
1376                success, msg = False, str(e)
1377
1378        try:
1379            if backup_path.exists():
1380                if not success:
1381                    backup_path.rename(self.properties_path)
1382                else:
1383                    backup_path.unlink()
1384        except Exception as e:
1385            success, msg = False, str(e)
1386
1387        return success, msg
1388
1389    def write_pickle(self) -> SuccessTuple:
1390        """Write the pickle file for the daemon."""
1391        import pickle
1392        import traceback
1393        from meerschaum.utils.misc import generate_password
1394
1395        if not self.pickle:
1396            return True, "Success"
1397
1398        from meerschaum._internal.entry import _shells
1399        if _shells:
1400            from meerschaum._internal.shell.Shell import revert_input
1401            revert_input()
1402
1403        backup_path = self.pickle_path.parent / (generate_password(7) + '.pkl')
1404        try:
1405            self.path.mkdir(parents=True, exist_ok=True)
1406            if self.pickle_path.exists():
1407                self.pickle_path.rename(backup_path)
1408            with open(self.pickle_path, 'wb+') as pickle_file:
1409                pickle.dump(self, pickle_file)
1410            success, msg = True, "Success"
1411        except Exception as e:
1412            success, msg = False, str(e)
1413            traceback.print_exception(type(e), e, e.__traceback__)
1414        try:
1415            if backup_path.exists():
1416                if not success:
1417                    backup_path.rename(self.pickle_path)
1418                else:
1419                    backup_path.unlink()
1420        except Exception as e:
1421            success, msg = False, str(e)
1422        return success, msg
1423
1424
1425    def _setup(
1426        self,
1427        allow_dirty_run: bool = False,
1428    ) -> None:
1429        """
1430        Update properties before starting the Daemon.
1431        """
1432        if self.properties is None:
1433            self._properties = {}
1434
1435        self._properties.update({
1436            'target': {
1437                'name': self.target.__name__,
1438                'module': self.target.__module__,
1439                'args': self.target_args,
1440                'kw': self.target_kw,
1441            },
1442        })
1443        self.mkdir_if_not_exists(allow_dirty_run)
1444        _write_properties_success_tuple = self.write_properties()
1445        if not _write_properties_success_tuple[0]:
1446            error(_write_properties_success_tuple[1])
1447
1448        _write_pickle_success_tuple = self.write_pickle()
1449        if not _write_pickle_success_tuple[0]:
1450            error(_write_pickle_success_tuple[1])
1451
1452    def cleanup(self, keep_logs: bool = False) -> SuccessTuple:
1453        """
1454        Remove a daemon's directory after execution.
1455
1456        Parameters
1457        ----------
1458        keep_logs: bool, default False
1459            If `True`, skip deleting the daemon's log files.
1460
1461        Returns
1462        -------
1463        A `SuccessTuple` indicating success.
1464        """
1465        if self.path.exists():
1466            try:
1467                shutil.rmtree(self.path)
1468            except Exception as e:
1469                msg = f"Failed to clean up '{self.daemon_id}':\n{e}"
1470                warn(msg)
1471                return False, msg
1472        if not keep_logs:
1473            self.rotating_log.delete()
1474            try:
1475                if self.log_offset_path.exists():
1476                    self.log_offset_path.unlink()
1477            except Exception as e:
1478                msg = f"Failed to remove offset file for '{self.daemon_id}':\n{e}"
1479                warn(msg)
1480                return False, msg
1481        return True, "Success"
1482
1483
1484    def get_timeout_seconds(self, timeout: Union[int, float, None] = None) -> Union[int, float]:
1485        """
1486        Return the timeout value to use. Use `--timeout-seconds` if provided,
1487        else the configured default (8).
1488        """
1489        if isinstance(timeout, (int, float)):
1490            return timeout
1491        return get_config('jobs', 'timeout_seconds')
1492
1493
1494    def get_check_timeout_interval_seconds(
1495        self,
1496        check_timeout_interval: Union[int, float, None] = None,
1497    ) -> Union[int, float]:
1498        """
1499        Return the interval value to check the status of timeouts.
1500        """
1501        if isinstance(check_timeout_interval, (int, float)):
1502            return check_timeout_interval
1503        return get_config('jobs', 'check_timeout_interval_seconds')
1504
1505    @property
1506    def target_args(self) -> Union[Tuple[Any], None]:
1507        """
1508        Return the positional arguments to pass to the target function.
1509        """
1510        target_args = (
1511            self.__dict__.get('_target_args', None)
1512            or self.properties.get('target', {}).get('args', None)
1513        )
1514        if target_args is None:
1515            return tuple([])
1516
1517        return tuple(target_args)
1518
1519    @property
1520    def target_kw(self) -> Union[Dict[str, Any], None]:
1521        """
1522        Return the keyword arguments to pass to the target function.
1523        """
1524        target_kw = (
1525            self.__dict__.get('_target_kw', None)
1526            or self.properties.get('target', {}).get('kw', None)
1527        )
1528        if target_kw is None:
1529            return {}
1530
1531        return {key: val for key, val in target_kw.items()}
1532
1533    @staticmethod
1534    def _get_target_reference(target) -> Union[Dict[str, str], None]:
1535        """
1536        If `target` is an importable, top-level function (e.g. `entry`), return a
1537        ``{'module': ..., 'qualname': ...}`` reference so it can be re-imported in the
1538        daemon process instead of pickled by value.
1539
1540        Pickling such a target by value serializes its entire global graph, which can
1541        reach unpicklable live state (e.g. a connector caching an `_asyncio.Task`) and
1542        raise `TypeError: cannot pickle '_asyncio.Task' object`. dill also falls back to
1543        by-value pickling whenever it cannot match the function by identity — which
1544        happens when two copies of a package are importable (a stale install shadowing
1545        a venv), so `byref=True` alone is not enough. Returns `None` for closures,
1546        lambdas, and anything not importable, which fall back to dill.
1547        """
1548        module = getattr(target, '__module__', None)
1549        qualname = getattr(target, '__qualname__', None)
1550        if not module or not qualname:
1551            return None
1552        if '<locals>' in qualname or '<lambda>' in qualname:
1553            return None
1554        return {'module': module, 'qualname': qualname}
1555
1556    @staticmethod
1557    def _load_target_reference(target_ref: Dict[str, str]):
1558        """
1559        Re-import a target from a `{'module': ..., 'qualname': ...}` reference.
1560        """
1561        import importlib
1562        obj = importlib.import_module(target_ref['module'])
1563        for part in target_ref['qualname'].split('.'):
1564            obj = getattr(obj, part)
1565        return obj
1566
1567    def __getstate__(self):
1568        """
1569        Pickle this Daemon.
1570        """
1571        dill = attempt_import('dill')
1572        state = {
1573            'target_args': self.target_args,
1574            'target_kw': self.target_kw,
1575            'daemon_id': self.daemon_id,
1576            'label': self.label,
1577            'properties': self.properties,
1578        }
1579        target_ref = self._get_target_reference(self.target)
1580        if target_ref is not None:
1581            ### Store by reference (re-imported in the daemon process), not by value.
1582            state['target'] = None
1583            state['target_ref'] = target_ref
1584        else:
1585            state['target'] = dill.dumps(self.target, byref=True)
1586        return state
1587
1588    def __setstate__(self, _state: Dict[str, Any]):
1589        """
1590        Restore this Daemon from a pickled state.
1591        If the properties file exists, skip the old pickled version.
1592        """
1593        dill = attempt_import('dill')
1594        target_ref = _state.pop('target_ref', None)
1595        if target_ref is not None and _state.get('target', None) is None:
1596            _state['target'] = self._load_target_reference(target_ref)
1597        else:
1598            _state['target'] = dill.loads(_state['target'])
1599        self._pickle = True
1600        daemon_id = _state.get('daemon_id', None)
1601        if not daemon_id:
1602            raise ValueError("Need a daemon_id to un-pickle a Daemon.")
1603
1604        properties_path = self._get_properties_path_from_daemon_id(daemon_id)
1605        ignore_properties = properties_path.exists()
1606        if ignore_properties:
1607            _state = {
1608                key: val
1609                for key, val in _state.items()
1610                if key != 'properties'
1611            }
1612        self.__init__(**_state)
1613
1614
1615    def __repr__(self):
1616        return str(self)
1617
1618    def __str__(self):
1619        return self.daemon_id
1620
1621    def __eq__(self, other):
1622        if not isinstance(other, Daemon):
1623            return False
1624        return self.daemon_id == other.daemon_id
1625
1626    def __hash__(self):
1627        return hash(self.daemon_id)