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 (BrokenPipeError, KeyboardInterrupt, SystemExit):
 355                    result = False, traceback.format_exc()
 356                except Exception as e:
 357                    warn(
 358                        f"Exception in daemon target function: {traceback.format_exc()}",
 359                    )
 360                    result = False, str(e)
 361                finally:
 362                    _results[self.daemon_id] = result
 363                    self.properties['result'] = result
 364
 365                    if keep_daemon_output:
 366                        self._capture_process_timestamp('ended')
 367                    else:
 368                        self.cleanup()
 369
 370                    self._log_refresh_timer.cancel()
 371                    try:
 372                        if self.pid is None and self.pid_path.exists():
 373                            self.pid_path.unlink()
 374                    except Exception:
 375                        pass
 376
 377                    if is_success_tuple(result):
 378                        try:
 379                            mrsm.pprint(result)
 380                        except BrokenPipeError:
 381                            pass
 382
 383        except Exception:
 384            daemon_error = traceback.format_exc()
 385            import meerschaum.config.paths as paths
 386            with open(paths.DAEMON_ERROR_LOG_PATH, 'a+', encoding='utf-8') as f:
 387                f.write(
 388                    f"Error in Daemon '{self}':\n\n"
 389                    f"{sys.stdin=}\n"
 390                    f"{self.stdin_file_path=}\n"
 391                    f"{self.stdin_file_path.exists()=}\n\n"
 392                    f"{daemon_error}\n\n"
 393                )
 394            warn(f"Encountered an error while running the daemon '{self}':\n{daemon_error}")
 395
 396    def _capture_process_timestamp(
 397        self,
 398        process_key: str,
 399        write_properties: bool = True,
 400    ) -> None:
 401        """
 402        Record the current timestamp to the parameters `process:<process_key>`.
 403
 404        Parameters
 405        ----------
 406        process_key: str
 407            Under which key to store the timestamp.
 408
 409        write_properties: bool, default True
 410            If `True` persist the properties to disk immediately after capturing the timestamp.
 411        """
 412        if 'process' not in self.properties:
 413            self.properties['process'] = {}
 414
 415        if process_key not in ('began', 'ended', 'paused', 'stopped'):
 416            raise ValueError(f"Invalid key '{process_key}'.")
 417
 418        self.properties['process'][process_key] = (
 419            datetime.now(timezone.utc).replace(tzinfo=None).isoformat()
 420        )
 421        if write_properties:
 422            self.write_properties()
 423
 424    def run(
 425        self,
 426        keep_daemon_output: bool = True,
 427        allow_dirty_run: bool = False,
 428        wait: bool = False,
 429        timeout: Union[int, float] = 4,
 430        debug: bool = False,
 431    ) -> SuccessTuple:
 432        """Run the daemon as a child process and continue executing the parent.
 433
 434        Parameters
 435        ----------
 436        keep_daemon_output: bool, default True
 437            If `False`, delete the daemon's output directory upon exiting.
 438
 439        allow_dirty_run: bool, default False
 440            If `True`, run the daemon, even if the `daemon_id` directory exists.
 441            This option is dangerous because if the same `daemon_id` runs concurrently,
 442            the last to finish will overwrite the output of the first.
 443
 444        wait: bool, default True
 445            If `True`, block until `Daemon.status` is running (or the timeout expires).
 446
 447        timeout: Union[int, float], default 4
 448            If `wait` is `True`, block for up to `timeout` seconds before returning a failure.
 449
 450        Returns
 451        -------
 452        A SuccessTuple indicating success.
 453
 454        """
 455        import platform
 456        if platform.system() == 'Windows':
 457            return False, "Cannot run background jobs on Windows."
 458
 459        ### The daemon might exist and be paused.
 460        if self.status == 'paused':
 461            return self.resume()
 462
 463        self._remove_stop_file()
 464        if self.status == 'running':
 465            return True, f"Daemon '{self}' is already running."
 466
 467        self.mkdir_if_not_exists(allow_dirty_run)
 468        _write_pickle_success_tuple = self.write_pickle()
 469        if not _write_pickle_success_tuple[0]:
 470            return _write_pickle_success_tuple
 471
 472        _launch_daemon_code = (
 473            "from meerschaum.utils.daemon import Daemon, _daemons; "
 474            f"daemon = Daemon(daemon_id='{self.daemon_id}'); "
 475            f"_daemons['{self.daemon_id}'] = daemon; "
 476            f"daemon._run_exit(keep_daemon_output={keep_daemon_output}, "
 477            "allow_dirty_run=True)"
 478        )
 479        env = dict(os.environ)
 480        _launch_success_bool = venv_exec(_launch_daemon_code, debug=debug, venv=None, env=env)
 481        msg = (
 482            "Success"
 483            if _launch_success_bool
 484            else f"Failed to start daemon '{self.daemon_id}'."
 485        )
 486        if not wait or not _launch_success_bool:
 487            return _launch_success_bool, msg
 488
 489        timeout = self.get_timeout_seconds(timeout)
 490        check_timeout_interval = self.get_check_timeout_interval_seconds()
 491
 492        if not timeout:
 493            success = self.status == 'running'
 494            msg = "Success" if success else f"Failed to run daemon '{self.daemon_id}'."
 495            if success:
 496                self._capture_process_timestamp('began')
 497            return success, msg
 498
 499        begin = time.perf_counter()
 500        while (time.perf_counter() - begin) < timeout:
 501            if self.status == 'running':
 502                self._capture_process_timestamp('began')
 503                return True, "Success"
 504            time.sleep(check_timeout_interval)
 505
 506        return False, (
 507            f"Failed to start daemon '{self.daemon_id}' within {timeout} second"
 508            + ('s' if timeout != 1 else '') + '.'
 509        )
 510
 511
 512    def kill(self, timeout: Union[int, float, None] = 8) -> SuccessTuple:
 513        """
 514        Forcibly terminate a running daemon.
 515        Sends a SIGTERM signal to the process.
 516
 517        Parameters
 518        ----------
 519        timeout: Optional[int], default 3
 520            How many seconds to wait for the process to terminate.
 521
 522        Returns
 523        -------
 524        A SuccessTuple indicating success.
 525        """
 526        ### A lost PID file means a detached/orphaned daemon — possibly SEVERAL processes
 527        ### sharing this daemon_id (e.g. accumulated across crashed restarts). Reap them
 528        ### ALL by daemon_id rather than reporting a false "stopped" and stranding them
 529        ### (which forces a manual `pkill meerschaum` or hunting the daemon_id in htop).
 530        if not self.pid_path.exists():
 531            reaped = self._kill_detached_processes(timeout)
 532            self._write_stop_file('kill')
 533            self.stdin_file.close()
 534            self._remove_blocking_stdin_file()
 535            return True, (
 536                (f"Reaped {reaped} detached process" + ('es' if reaped != 1 else '') + '.')
 537                if reaped
 538                else "Process has already stopped."
 539            )
 540
 541        if self.status != 'paused':
 542            success, msg = self._send_signal(signal.SIGTERM, timeout=timeout)
 543            if success:
 544                ### Sweep any sibling/detached processes left over for this daemon_id.
 545                self._kill_detached_processes(timeout)
 546                self._write_stop_file('kill')
 547                self.stdin_file.close()
 548                self._remove_blocking_stdin_file()
 549                return success, msg
 550
 551        if self.status == 'stopped':
 552            reaped = self._kill_detached_processes(timeout)
 553            self._write_stop_file('kill')
 554            self.stdin_file.close()
 555            self._remove_blocking_stdin_file()
 556            return True, (
 557                (f"Reaped {reaped} detached process" + ('es' if reaped != 1 else '') + '.')
 558                if reaped
 559                else "Process has already stopped."
 560            )
 561
 562        psutil = attempt_import('psutil')
 563        process = self.process
 564        try:
 565            process.terminate()
 566            process.kill()
 567            process.wait(timeout=timeout)
 568        except Exception as e:
 569            return False, f"Failed to kill job {self} ({process}) with exception: {e}"
 570
 571        try:
 572            if process.status():
 573                return False, "Failed to stop daemon '{self}' ({process})."
 574        except psutil.NoSuchProcess:
 575            pass
 576
 577        if self.pid_path.exists():
 578            try:
 579                self.pid_path.unlink()
 580            except Exception:
 581                pass
 582
 583        self._write_stop_file('kill')
 584        self.stdin_file.close()
 585        self._remove_blocking_stdin_file()
 586        return True, "Success"
 587
 588    def quit(self, timeout: Union[int, float, None] = None) -> SuccessTuple:
 589        """Gracefully quit a running daemon."""
 590        if self.status == 'paused':
 591            return self.kill(timeout)
 592
 593        ### A lost PID file means a detached/orphaned daemon (possibly several processes);
 594        ### the normal signal path can't see them, so reap them all by daemon_id instead
 595        ### of returning a false "not running".
 596        if not self.pid_path.exists():
 597            reaped = self._kill_detached_processes(timeout)
 598            self._write_stop_file('quit')
 599            self.stdin_file.close()
 600            self._remove_blocking_stdin_file()
 601            return True, (
 602                (f"Reaped {reaped} detached process" + ('es' if reaped != 1 else '') + '.')
 603                if reaped
 604                else "Process is not running."
 605            )
 606
 607        signal_success, signal_msg = self._send_signal(signal.SIGINT, timeout=timeout)
 608        if signal_success:
 609            self._write_stop_file('quit')
 610            self.stdin_file.close()
 611            self._remove_blocking_stdin_file()
 612        return signal_success, signal_msg
 613
 614    def pause(
 615        self,
 616        timeout: Union[int, float, None] = None,
 617        check_timeout_interval: Union[float, int, None] = None,
 618    ) -> SuccessTuple:
 619        """
 620        Pause the daemon if it is running.
 621
 622        Parameters
 623        ----------
 624        timeout: Union[float, int, None], default None
 625            The maximum number of seconds to wait for a process to suspend.
 626
 627        check_timeout_interval: Union[float, int, None], default None
 628            The number of seconds to wait between checking if the process is still running.
 629
 630        Returns
 631        -------
 632        A `SuccessTuple` indicating whether the `Daemon` process was successfully suspended.
 633        """
 634        self._remove_blocking_stdin_file()
 635
 636        if self.process is None:
 637            return False, f"Daemon '{self.daemon_id}' is not running and cannot be paused."
 638
 639        if self.status == 'paused':
 640            return True, f"Daemon '{self.daemon_id}' is already paused."
 641
 642        self._write_stop_file('pause')
 643        self.stdin_file.close()
 644        self._remove_blocking_stdin_file()
 645        try:
 646            self.process.suspend()
 647        except Exception as e:
 648            return False, f"Failed to pause daemon '{self.daemon_id}':\n{e}"
 649
 650        timeout = self.get_timeout_seconds(timeout)
 651        check_timeout_interval = self.get_check_timeout_interval_seconds(
 652            check_timeout_interval
 653        )
 654
 655        psutil = attempt_import('psutil')
 656
 657        if not timeout:
 658            try:
 659                success = self.process.status() == 'stopped'
 660            except psutil.NoSuchProcess:
 661                success = True
 662            msg = "Success" if success else f"Failed to suspend daemon '{self.daemon_id}'."
 663            if success:
 664                self._capture_process_timestamp('paused')
 665            return success, msg
 666
 667        begin = time.perf_counter()
 668        while (time.perf_counter() - begin) < timeout:
 669            try:
 670                if self.process.status() == 'stopped':
 671                    self._capture_process_timestamp('paused')
 672                    return True, "Success"
 673            except psutil.NoSuchProcess as e:
 674                return False, f"Process exited unexpectedly. Was it killed?\n{e}"
 675            time.sleep(check_timeout_interval)
 676
 677        return False, (
 678            f"Failed to pause daemon '{self.daemon_id}' within {timeout} second"
 679            + ('s' if timeout != 1 else '') + '.'
 680        )
 681
 682    def resume(
 683        self,
 684        timeout: Union[int, float, None] = None,
 685        check_timeout_interval: Union[float, int, None] = None,
 686    ) -> SuccessTuple:
 687        """
 688        Resume the daemon if it is paused.
 689
 690        Parameters
 691        ----------
 692        timeout: Union[float, int, None], default None
 693            The maximum number of seconds to wait for a process to resume.
 694
 695        check_timeout_interval: Union[float, int, None], default None
 696            The number of seconds to wait between checking if the process is still stopped.
 697
 698        Returns
 699        -------
 700        A `SuccessTuple` indicating whether the `Daemon` process was successfully resumed.
 701        """
 702        if self.status == 'running':
 703            return True, f"Daemon '{self.daemon_id}' is already running."
 704
 705        if self.status == 'stopped':
 706            return False, f"Daemon '{self.daemon_id}' is stopped and cannot be resumed."
 707
 708        self._remove_stop_file()
 709        try:
 710            if self.process is None:
 711                return False, f"Cannot resume daemon '{self.daemon_id}'."
 712
 713            self.process.resume()
 714        except Exception as e:
 715            return False, f"Failed to resume daemon '{self.daemon_id}':\n{e}"
 716
 717        timeout = self.get_timeout_seconds(timeout)
 718        check_timeout_interval = self.get_check_timeout_interval_seconds(
 719            check_timeout_interval
 720        )
 721
 722        if not timeout:
 723            success = self.status == 'running'
 724            msg = "Success" if success else f"Failed to resume daemon '{self.daemon_id}'."
 725            if success:
 726                self._capture_process_timestamp('began')
 727            return success, msg
 728
 729        begin = time.perf_counter()
 730        while (time.perf_counter() - begin) < timeout:
 731            if self.status == 'running':
 732                self._capture_process_timestamp('began')
 733                return True, "Success"
 734            time.sleep(check_timeout_interval)
 735
 736        return False, (
 737            f"Failed to resume daemon '{self.daemon_id}' within {timeout} second"
 738            + ('s' if timeout != 1 else '') + '.'
 739        )
 740
 741    def _write_stop_file(self, action: str) -> SuccessTuple:
 742        """Write the stop file timestamp and action."""
 743        if action not in ('quit', 'kill', 'pause'):
 744            return False, f"Unsupported action '{action}'."
 745
 746        if not self.stop_path.parent.exists():
 747            self.stop_path.parent.mkdir(parents=True, exist_ok=True)
 748
 749        with open(self.stop_path, 'w+', encoding='utf-8') as f:
 750            json.dump(
 751                {
 752                    'stop_time': datetime.now(timezone.utc).isoformat(),
 753                    'action': action,
 754                },
 755                f
 756            )
 757
 758        return True, "Success"
 759
 760    def _remove_stop_file(self) -> SuccessTuple:
 761        """Remove the stop file"""
 762        if not self.stop_path.exists():
 763            return True, "Stop file does not exist."
 764
 765        try:
 766            self.stop_path.unlink()
 767        except Exception as e:
 768            return False, f"Failed to remove stop file:\n{e}"
 769
 770        return True, "Success"
 771
 772    def _read_stop_file(self) -> Dict[str, Any]:
 773        """
 774        Read the stop file if it exists.
 775        """
 776        if not self.stop_path.exists():
 777            return {}
 778
 779        try:
 780            with open(self.stop_path, 'r', encoding='utf-8') as f:
 781                data = json.load(f)
 782            return data
 783        except Exception:
 784            return {}
 785
 786    def _remove_blocking_stdin_file(self) -> mrsm.SuccessTuple:
 787        """
 788        Remove the blocking STDIN file if it exists.
 789        """
 790        try:
 791            if self.blocking_stdin_file_path.exists():
 792                self.blocking_stdin_file_path.unlink()
 793        except Exception as e:
 794            return False, str(e)
 795
 796        return True, "Success"
 797
 798    def _handle_sigterm(self, signal_number: int, stack_frame: 'frame') -> None:
 799        """
 800        Handle `SIGTERM` within the `Daemon` context.
 801        This method is injected into the `DaemonContext`.
 802        """
 803        from meerschaum.utils.process import signal_handler
 804        from meerschaum.utils.threading import request_stop, interrupt_threads
 805        signal_handler(signal_number, stack_frame)
 806
 807        ### Tell cooperative loops (e.g. `sync pipes`) to stop, then actively unwind
 808        ### any worker threads so they cannot keep the process alive as a zombie.
 809        request_stop()
 810        interrupt_threads(SystemExit)
 811
 812        timer = self.__dict__.get('_log_refresh_timer', None)
 813        if timer is not None:
 814            timer.cancel()
 815
 816        daemon_context = self.__dict__.get('_daemon_context', None)
 817        if daemon_context is not None:
 818            daemon_context.close()
 819
 820        _close_pools()
 821        raise SystemExit(0)
 822
 823    def _send_signal(
 824        self,
 825        signal_to_send,
 826        timeout: Union[float, int, None] = None,
 827        check_timeout_interval: Union[float, int, None] = None,
 828    ) -> SuccessTuple:
 829        """Send a signal to the daemon process.
 830
 831        Parameters
 832        ----------
 833        signal_to_send:
 834            The signal the send to the daemon, e.g. `signals.SIGINT`.
 835
 836        timeout: Union[float, int, None], default None
 837            The maximum number of seconds to wait for a process to terminate.
 838
 839        check_timeout_interval: Union[float, int, None], default None
 840            The number of seconds to wait between checking if the process is still running.
 841
 842        Returns
 843        -------
 844        A SuccessTuple indicating success.
 845        """
 846        try:
 847            pid = self.pid
 848            if pid is None:
 849                return (
 850                    False,
 851                    f"Daemon '{self.daemon_id}' is not running, "
 852                    + f"cannot send signal '{signal_to_send}'."
 853                )
 854            
 855            os.kill(pid, signal_to_send)
 856        except Exception:
 857            return False, f"Failed to send signal {signal_to_send}:\n{traceback.format_exc()}"
 858
 859        timeout = self.get_timeout_seconds(timeout)
 860        check_timeout_interval = self.get_check_timeout_interval_seconds(
 861            check_timeout_interval
 862        )
 863
 864        if not timeout:
 865            return True, f"Successfully sent '{signal_to_send}' to daemon '{self.daemon_id}'."
 866
 867        begin = time.perf_counter()
 868        while (time.perf_counter() - begin) < timeout:
 869            if not self.status == 'running':
 870                return True, "Success"
 871            time.sleep(check_timeout_interval)
 872
 873        return False, (
 874            f"Failed to stop daemon '{self.daemon_id}' (PID: {pid}) within {timeout} second"
 875            + ('s' if timeout != 1 else '') + '.'
 876        )
 877
 878    def _find_detached_pids(self) -> List[int]:
 879        """
 880        Return the PIDs of any processes whose command line references this daemon_id.
 881
 882        A daemon is launched via `venv_exec` with `Daemon(daemon_id='<id>')` embedded in
 883        the executed code, so the daemon_id stays visible in the process's command line.
 884        When the PID file is lost (e.g. the launcher exited without cleanup, leaving an
 885        orphaned/detached daemon), this is the only reliable way to find the process —
 886        otherwise stopping the job reports a false "already stopped" and the user must
 887        resort to `pkill meerschaum` or hunting the daemon_id in `htop`.
 888        """
 889        psutil = attempt_import('psutil')
 890        marker = f"daemon_id='{self.daemon_id}'"
 891        my_pid = os.getpid()
 892        pids = []
 893        for proc in psutil.process_iter(['pid', 'cmdline']):
 894            try:
 895                if proc.info['pid'] == my_pid:
 896                    continue
 897                cmdline = proc.info.get('cmdline') or []
 898                if any(marker in (part or '') for part in cmdline):
 899                    pids.append(int(proc.info['pid']))
 900            except Exception:
 901                continue
 902        return pids
 903
 904    def _kill_detached_processes(self, timeout: Union[int, float, None] = None) -> int:
 905        """
 906        SIGTERM then SIGKILL any detached processes referencing this daemon_id.
 907
 908        Fallback for when the PID file is gone but the daemon (or its threads) are still
 909        alive. Returns the number of processes that were targeted.
 910        """
 911        psutil = attempt_import('psutil')
 912        pids = self._find_detached_pids()
 913        if not pids:
 914            return 0
 915
 916        procs = []
 917        for pid in pids:
 918            try:
 919                procs.append(psutil.Process(pid))
 920            except Exception:
 921                continue
 922        for proc in procs:
 923            try:
 924                proc.terminate()
 925            except Exception:
 926                pass
 927
 928        timeout = self.get_timeout_seconds(timeout)
 929        try:
 930            _, alive = psutil.wait_procs(procs, timeout=(timeout or 8))
 931        except Exception:
 932            alive = procs
 933        for proc in alive:
 934            try:
 935                proc.kill()
 936            except Exception:
 937                pass
 938        return len(procs)
 939
 940    def mkdir_if_not_exists(self, allow_dirty_run: bool = False):
 941        """Create the Daemon's directory.
 942        If `allow_dirty_run` is `False` and the directory already exists,
 943        raise a `FileExistsError`.
 944        """
 945        try:
 946            self.path.mkdir(parents=True, exist_ok=True)
 947            _already_exists = any(os.scandir(self.path))
 948        except FileExistsError:
 949            _already_exists = True
 950
 951        if _already_exists and not allow_dirty_run:
 952            error(
 953                f"Daemon '{self.daemon_id}' already exists. " +
 954                "To allow this daemon to run, do one of the following:\n"
 955                + "  - Execute `daemon.cleanup()`.\n"
 956                + f"  - Delete the directory '{self.path}'.\n"
 957                + "  - Pass `allow_dirty_run=True` to `daemon.run()`.\n",
 958                FileExistsError,
 959            )
 960
 961    @property
 962    def process(self) -> Union['psutil.Process', None]:
 963        """
 964        Return the psutil process for the Daemon.
 965        """
 966        psutil = attempt_import('psutil')
 967        pid = self.pid
 968        if pid is None:
 969            return None
 970        if '_process' not in self.__dict__ or self.__dict__['_process'].pid != int(pid):
 971            try:
 972                self._process = psutil.Process(int(pid))
 973                process_exists = True
 974            except Exception:
 975                process_exists = False
 976            if not process_exists:
 977                _ = self.__dict__.pop('_process', None)
 978                try:
 979                    if self.pid_path.exists():
 980                        self.pid_path.unlink()
 981                except Exception:
 982                    pass
 983                return None
 984        return self._process
 985
 986    @property
 987    def status(self) -> str:
 988        """
 989        Return the running status of this Daemon.
 990        """
 991        if self.process is None:
 992            return 'stopped'
 993
 994        psutil = attempt_import('psutil', lazy=False)
 995        try:
 996            if self.process.status() == 'stopped':
 997                return 'paused'
 998            if self.process.status() == 'zombie':
 999                raise psutil.NoSuchProcess(self.process.pid)
1000        except (psutil.NoSuchProcess, AttributeError):
1001            if self.pid_path.exists():
1002                try:
1003                    self.pid_path.unlink()
1004                except Exception:
1005                    pass
1006            return 'stopped'
1007
1008        return 'running'
1009
1010    @classmethod
1011    def _get_path_from_daemon_id(cls, daemon_id: str) -> pathlib.Path:
1012        """
1013        Return a Daemon's path from its `daemon_id`.
1014        """
1015        import meerschaum.config.paths as paths
1016        return paths.DAEMON_RESOURCES_PATH / daemon_id
1017
1018    @property
1019    def path(self) -> pathlib.Path:
1020        """
1021        Return the path for this Daemon's directory.
1022        """
1023        return self._get_path_from_daemon_id(self.daemon_id)
1024
1025    @classmethod
1026    def _get_properties_path_from_daemon_id(cls, daemon_id: str) -> pathlib.Path:
1027        """
1028        Return the `properties.json` path for a given `daemon_id`.
1029        """
1030        return cls._get_path_from_daemon_id(daemon_id) / 'properties.json'
1031
1032    @property
1033    def properties_path(self) -> pathlib.Path:
1034        """
1035        Return the `propterties.json` path for this Daemon.
1036        """
1037        return self._get_properties_path_from_daemon_id(self.daemon_id)
1038
1039    @property
1040    def stop_path(self) -> pathlib.Path:
1041        """
1042        Return the path for the stop file (created when manually stopped).
1043        """
1044        return self.path / '.stop.json'
1045
1046    @property
1047    def log_path(self) -> pathlib.Path:
1048        """
1049        Return the log path.
1050        """
1051        logs_cf = self.properties.get('logs', None) or {}
1052        if 'path' not in logs_cf:
1053            import meerschaum.config.paths as paths
1054            return paths.LOGS_RESOURCES_PATH / (self.daemon_id + '.log')
1055
1056        return pathlib.Path(logs_cf['path'])
1057
1058    @property
1059    def stdin_file_path(self) -> pathlib.Path:
1060        """
1061        Return the stdin file path.
1062        """
1063        return self.path / 'input.stdin'
1064
1065    @property
1066    def blocking_stdin_file_path(self) -> pathlib.Path:
1067        """
1068        Return the stdin file path.
1069        """
1070        if '_blocking_stdin_file_path' in self.__dict__:
1071            return self._blocking_stdin_file_path
1072
1073        return self.path / 'input.stdin.block'
1074
1075    @property
1076    def prompt_kwargs_file_path(self) -> pathlib.Path:
1077        """
1078        Return the file path to the kwargs for the invoking `prompt()`.
1079        """
1080        return self.path / 'prompt_kwargs.json'
1081
1082    @property
1083    def log_offset_path(self) -> pathlib.Path:
1084        """
1085        Return the log offset file path.
1086        """
1087        import meerschaum.config.paths as paths
1088        return paths.LOGS_RESOURCES_PATH / ('.' + self.daemon_id + '.log.offset')
1089
1090    @property
1091    def log_offset_lock(self) -> 'fasteners.InterProcessLock':
1092        """
1093        Return the process lock context manager.
1094        """
1095        if '_log_offset_lock' in self.__dict__:
1096            return self._log_offset_lock
1097
1098        fasteners = attempt_import('fasteners')
1099        self._log_offset_lock = fasteners.InterProcessLock(self.log_offset_path)
1100        return self._log_offset_lock
1101
1102    @property
1103    def rotating_log(self) -> RotatingFile:
1104        """
1105        The rotating log file for the daemon's output.
1106        """
1107        if '_rotating_log' in self.__dict__:
1108            return self._rotating_log
1109
1110        logs_cf = self.properties.get('logs', None) or {}
1111        write_timestamps = logs_cf.get('write_timestamps', None)
1112        if write_timestamps is None:
1113            write_timestamps = get_config('jobs', 'logs', 'timestamps', 'enabled')
1114
1115        timestamp_format = logs_cf.get('timestamp_format', None)
1116        if timestamp_format is None:
1117            timestamp_format = get_config('jobs', 'logs', 'timestamps', 'format')
1118
1119        num_files_to_keep = logs_cf.get('num_files_to_keep', None)
1120        if num_files_to_keep is None:
1121            num_files_to_keep = get_config('jobs', 'logs', 'num_files_to_keep')
1122
1123        max_file_size = logs_cf.get('max_file_size', None)
1124        if max_file_size is None:
1125            max_file_size = get_config('jobs', 'logs', 'max_file_size')
1126
1127        redirect_streams = logs_cf.get('redirect_streams', True)
1128
1129        self._rotating_log = RotatingFile(
1130            self.log_path,
1131            redirect_streams=redirect_streams,
1132            write_timestamps=write_timestamps,
1133            timestamp_format=timestamp_format,
1134            num_files_to_keep=num_files_to_keep,
1135            max_file_size=max_file_size,
1136        )
1137        return self._rotating_log
1138
1139    @property
1140    def stdin_file(self):
1141        """
1142        Return the file handler for the stdin file.
1143        """
1144        if (_stdin_file := self.__dict__.get('_stdin_file', None)):
1145            return _stdin_file
1146
1147        self._stdin_file = StdinFile(
1148            self.stdin_file_path,
1149            lock_file_path=self.blocking_stdin_file_path,
1150        )
1151        return self._stdin_file
1152
1153    @property
1154    def log_text(self) -> Union[str, None]:
1155        """
1156        Read the log files and return their contents.
1157        Returns `None` if the log file does not exist.
1158        """
1159        logs_cf = self.properties.get('logs', None) or {}
1160        write_timestamps = logs_cf.get('write_timestamps', None)
1161        if write_timestamps is None:
1162            write_timestamps = get_config('jobs', 'logs', 'timestamps', 'enabled')
1163
1164        timestamp_format = logs_cf.get('timestamp_format', None)
1165        if timestamp_format is None:
1166            timestamp_format = get_config('jobs', 'logs', 'timestamps', 'format')
1167
1168        num_files_to_keep = logs_cf.get('num_files_to_keep', None)
1169        if num_files_to_keep is None:
1170            num_files_to_keep = get_config('jobs', 'logs', 'num_files_to_keep')
1171
1172        max_file_size = logs_cf.get('max_file_size', None)
1173        if max_file_size is None:
1174            max_file_size = get_config('jobs', 'logs', 'max_file_size')
1175
1176        new_rotating_log = RotatingFile(
1177            self.rotating_log.file_path,
1178            num_files_to_keep=num_files_to_keep,
1179            max_file_size=max_file_size,
1180            write_timestamps=write_timestamps,
1181            timestamp_format=timestamp_format,
1182        )
1183        return new_rotating_log.read()
1184
1185    def readlines(self) -> List[str]:
1186        """
1187        Read the next log lines, persisting the cursor for later use.
1188        Note this will alter the cursor of `self.rotating_log`.
1189        """
1190        self.rotating_log._cursor = self._read_log_offset()
1191        lines = self.rotating_log.readlines()
1192        self._write_log_offset()
1193        return lines
1194
1195    def _read_log_offset(self) -> Tuple[int, int]:
1196        """
1197        Return the current log offset cursor.
1198
1199        Returns
1200        -------
1201        A tuple of the form (`subfile_index`, `position`).
1202        """
1203        if not self.log_offset_path.exists():
1204            return 0, 0
1205
1206        try:
1207            with open(self.log_offset_path, 'r', encoding='utf-8') as f:
1208                cursor_text = f.read()
1209            cursor_parts = cursor_text.split(' ')
1210            subfile_index, subfile_position = int(cursor_parts[0]), int(cursor_parts[1])
1211            return subfile_index, subfile_position
1212        except Exception as e:
1213            warn(f"Failed to read cursor:\n{e}")
1214        return 0, 0
1215
1216    def _write_log_offset(self) -> None:
1217        """
1218        Write the current log offset file.
1219        """
1220        with self.log_offset_lock:
1221            with open(self.log_offset_path, 'w+', encoding='utf-8') as f:
1222                subfile_index = self.rotating_log._cursor[0]
1223                subfile_position = self.rotating_log._cursor[1]
1224                f.write(f"{subfile_index} {subfile_position}")
1225
1226    @property
1227    def pid(self) -> Union[int, None]:
1228        """
1229        Read the PID file and return its contents.
1230        Returns `None` if the PID file does not exist.
1231        """
1232        if not self.pid_path.exists():
1233            ### The PID file can be lost while a detached daemon keeps running (e.g. the
1234            ### launcher exited without cleanup). Recover the PID by finding the process
1235            ### whose command line embeds this daemon_id, so `status`/`stop`/`kill` see the
1236            ### live process instead of reporting a false "stopped" (which would strand the
1237            ### orphan, forcing a manual `pkill meerschaum`).
1238            detached_pids = self._find_detached_pids()
1239            return detached_pids[0] if detached_pids else None
1240        try:
1241            with open(self.pid_path, 'r', encoding='utf-8') as f:
1242                text = f.read()
1243            if len(text) == 0:
1244                return None
1245            pid = int(text.rstrip())
1246        except Exception as e:
1247            warn(e)
1248            text = None
1249            pid = None
1250        return pid
1251
1252    @property
1253    def pid_path(self) -> pathlib.Path:
1254        """
1255        Return the path to a file containing the PID for this Daemon.
1256        """
1257        return self.path / 'process.pid'
1258
1259    @property
1260    def pid_lock(self) -> 'fasteners.InterProcessLock':
1261        """
1262        Return the process lock context manager.
1263        """
1264        if '_pid_lock' in self.__dict__:
1265            return self._pid_lock
1266
1267        fasteners = attempt_import('fasteners')
1268        self._pid_lock = fasteners.InterProcessLock(self.pid_path)
1269        return self._pid_lock
1270
1271    @property
1272    def pickle_path(self) -> pathlib.Path:
1273        """
1274        Return the path for the pickle file.
1275        """
1276        return self.path / 'pickle.pkl'
1277
1278    def read_properties(self) -> Optional[Dict[str, Any]]:
1279        """Read the properties JSON file and return the dictionary."""
1280        if not self.properties_path.exists():
1281            return None
1282        try:
1283            with open(self.properties_path, 'r', encoding='utf-8') as file:
1284                properties = json.load(file)
1285        except Exception:
1286            properties = {}
1287        
1288        return properties or {}
1289
1290    def read_pickle(self) -> Daemon:
1291        """Read a Daemon's pickle file and return the `Daemon`."""
1292        import pickle
1293        import traceback
1294        if not self.pickle_path.exists():
1295            error(f"Pickle file does not exist for daemon '{self.daemon_id}'.")
1296
1297        if self.pickle_path.stat().st_size == 0:
1298            error(f"Pickle was empty for daemon '{self.daemon_id}'.")
1299
1300        try:
1301            with open(self.pickle_path, 'rb') as pickle_file:
1302                daemon = pickle.load(pickle_file)
1303            success, msg = True, 'Success'
1304        except Exception as e:
1305            success, msg = False, str(e)
1306            daemon = None
1307            traceback.print_exception(type(e), e, e.__traceback__)
1308        if not success:
1309            error(msg)
1310        return daemon
1311
1312    @property
1313    def properties(self) -> Dict[str, Any]:
1314        """
1315        Return the contents of the properties JSON file.
1316        """
1317        try:
1318            _file_properties = self.read_properties() or {}
1319        except Exception:
1320            traceback.print_exc()
1321            _file_properties = {}
1322
1323        if not self._properties:
1324            self._properties = _file_properties
1325
1326        if self._properties is None:
1327            self._properties = {}
1328
1329        if (
1330            self._properties.get('result', None) is None
1331            and _file_properties.get('result', None) is not None
1332        ):
1333            _ = self._properties.pop('result', None)
1334
1335        if _file_properties is not None:
1336            self._properties = apply_patch_to_config(
1337                _file_properties,
1338                self._properties,
1339            )
1340
1341        return self._properties
1342
1343    @property
1344    def hidden(self) -> bool:
1345        """
1346        Return a bool indicating whether this Daemon should be displayed.
1347        """
1348        return self.daemon_id.startswith('_') or self.daemon_id.startswith('.')
1349
1350    def write_properties(self) -> SuccessTuple:
1351        """Write the properties dictionary to the properties JSON file
1352        (only if self.properties exists).
1353        """
1354        from meerschaum.utils.misc import generate_password
1355        success, msg = (
1356            False,
1357            f"No properties to write for daemon '{self.daemon_id}'."
1358        )
1359        backup_path = self.properties_path.parent / (generate_password(8) + '.json')
1360        props = self.properties
1361        if props is not None:
1362            try:
1363                self.path.mkdir(parents=True, exist_ok=True)
1364                if self.properties_path.exists():
1365                    self.properties_path.rename(backup_path)
1366                with open(self.properties_path, 'w+', encoding='utf-8') as properties_file:
1367                    json.dump(props, properties_file)
1368                success, msg = True, 'Success'
1369            except Exception as e:
1370                success, msg = False, str(e)
1371
1372        try:
1373            if backup_path.exists():
1374                if not success:
1375                    backup_path.rename(self.properties_path)
1376                else:
1377                    backup_path.unlink()
1378        except Exception as e:
1379            success, msg = False, str(e)
1380
1381        return success, msg
1382
1383    def write_pickle(self) -> SuccessTuple:
1384        """Write the pickle file for the daemon."""
1385        import pickle
1386        import traceback
1387        from meerschaum.utils.misc import generate_password
1388
1389        if not self.pickle:
1390            return True, "Success"
1391
1392        from meerschaum._internal.entry import _shells
1393        if _shells:
1394            from meerschaum._internal.shell.Shell import revert_input
1395            revert_input()
1396
1397        backup_path = self.pickle_path.parent / (generate_password(7) + '.pkl')
1398        try:
1399            self.path.mkdir(parents=True, exist_ok=True)
1400            if self.pickle_path.exists():
1401                self.pickle_path.rename(backup_path)
1402            with open(self.pickle_path, 'wb+') as pickle_file:
1403                pickle.dump(self, pickle_file)
1404            success, msg = True, "Success"
1405        except Exception as e:
1406            success, msg = False, str(e)
1407            traceback.print_exception(type(e), e, e.__traceback__)
1408        try:
1409            if backup_path.exists():
1410                if not success:
1411                    backup_path.rename(self.pickle_path)
1412                else:
1413                    backup_path.unlink()
1414        except Exception as e:
1415            success, msg = False, str(e)
1416        return success, msg
1417
1418
1419    def _setup(
1420        self,
1421        allow_dirty_run: bool = False,
1422    ) -> None:
1423        """
1424        Update properties before starting the Daemon.
1425        """
1426        if self.properties is None:
1427            self._properties = {}
1428
1429        self._properties.update({
1430            'target': {
1431                'name': self.target.__name__,
1432                'module': self.target.__module__,
1433                'args': self.target_args,
1434                'kw': self.target_kw,
1435            },
1436        })
1437        self.mkdir_if_not_exists(allow_dirty_run)
1438        _write_properties_success_tuple = self.write_properties()
1439        if not _write_properties_success_tuple[0]:
1440            error(_write_properties_success_tuple[1])
1441
1442        _write_pickle_success_tuple = self.write_pickle()
1443        if not _write_pickle_success_tuple[0]:
1444            error(_write_pickle_success_tuple[1])
1445
1446    def cleanup(self, keep_logs: bool = False) -> SuccessTuple:
1447        """
1448        Remove a daemon's directory after execution.
1449
1450        Parameters
1451        ----------
1452        keep_logs: bool, default False
1453            If `True`, skip deleting the daemon's log files.
1454
1455        Returns
1456        -------
1457        A `SuccessTuple` indicating success.
1458        """
1459        if self.path.exists():
1460            try:
1461                shutil.rmtree(self.path)
1462            except Exception as e:
1463                msg = f"Failed to clean up '{self.daemon_id}':\n{e}"
1464                warn(msg)
1465                return False, msg
1466        if not keep_logs:
1467            self.rotating_log.delete()
1468            try:
1469                if self.log_offset_path.exists():
1470                    self.log_offset_path.unlink()
1471            except Exception as e:
1472                msg = f"Failed to remove offset file for '{self.daemon_id}':\n{e}"
1473                warn(msg)
1474                return False, msg
1475        return True, "Success"
1476
1477
1478    def get_timeout_seconds(self, timeout: Union[int, float, None] = None) -> Union[int, float]:
1479        """
1480        Return the timeout value to use. Use `--timeout-seconds` if provided,
1481        else the configured default (8).
1482        """
1483        if isinstance(timeout, (int, float)):
1484            return timeout
1485        return get_config('jobs', 'timeout_seconds')
1486
1487
1488    def get_check_timeout_interval_seconds(
1489        self,
1490        check_timeout_interval: Union[int, float, None] = None,
1491    ) -> Union[int, float]:
1492        """
1493        Return the interval value to check the status of timeouts.
1494        """
1495        if isinstance(check_timeout_interval, (int, float)):
1496            return check_timeout_interval
1497        return get_config('jobs', 'check_timeout_interval_seconds')
1498
1499    @property
1500    def target_args(self) -> Union[Tuple[Any], None]:
1501        """
1502        Return the positional arguments to pass to the target function.
1503        """
1504        target_args = (
1505            self.__dict__.get('_target_args', None)
1506            or self.properties.get('target', {}).get('args', None)
1507        )
1508        if target_args is None:
1509            return tuple([])
1510
1511        return tuple(target_args)
1512
1513    @property
1514    def target_kw(self) -> Union[Dict[str, Any], None]:
1515        """
1516        Return the keyword arguments to pass to the target function.
1517        """
1518        target_kw = (
1519            self.__dict__.get('_target_kw', None)
1520            or self.properties.get('target', {}).get('kw', None)
1521        )
1522        if target_kw is None:
1523            return {}
1524
1525        return {key: val for key, val in target_kw.items()}
1526
1527    @staticmethod
1528    def _get_target_reference(target) -> Union[Dict[str, str], None]:
1529        """
1530        If `target` is an importable, top-level function (e.g. `entry`), return a
1531        ``{'module': ..., 'qualname': ...}`` reference so it can be re-imported in the
1532        daemon process instead of pickled by value.
1533
1534        Pickling such a target by value serializes its entire global graph, which can
1535        reach unpicklable live state (e.g. a connector caching an `_asyncio.Task`) and
1536        raise `TypeError: cannot pickle '_asyncio.Task' object`. dill also falls back to
1537        by-value pickling whenever it cannot match the function by identity — which
1538        happens when two copies of a package are importable (a stale install shadowing
1539        a venv), so `byref=True` alone is not enough. Returns `None` for closures,
1540        lambdas, and anything not importable, which fall back to dill.
1541        """
1542        module = getattr(target, '__module__', None)
1543        qualname = getattr(target, '__qualname__', None)
1544        if not module or not qualname:
1545            return None
1546        if '<locals>' in qualname or '<lambda>' in qualname:
1547            return None
1548        return {'module': module, 'qualname': qualname}
1549
1550    @staticmethod
1551    def _load_target_reference(target_ref: Dict[str, str]):
1552        """
1553        Re-import a target from a `{'module': ..., 'qualname': ...}` reference.
1554        """
1555        import importlib
1556        obj = importlib.import_module(target_ref['module'])
1557        for part in target_ref['qualname'].split('.'):
1558            obj = getattr(obj, part)
1559        return obj
1560
1561    def __getstate__(self):
1562        """
1563        Pickle this Daemon.
1564        """
1565        dill = attempt_import('dill')
1566        state = {
1567            'target_args': self.target_args,
1568            'target_kw': self.target_kw,
1569            'daemon_id': self.daemon_id,
1570            'label': self.label,
1571            'properties': self.properties,
1572        }
1573        target_ref = self._get_target_reference(self.target)
1574        if target_ref is not None:
1575            ### Store by reference (re-imported in the daemon process), not by value.
1576            state['target'] = None
1577            state['target_ref'] = target_ref
1578        else:
1579            state['target'] = dill.dumps(self.target, byref=True)
1580        return state
1581
1582    def __setstate__(self, _state: Dict[str, Any]):
1583        """
1584        Restore this Daemon from a pickled state.
1585        If the properties file exists, skip the old pickled version.
1586        """
1587        dill = attempt_import('dill')
1588        target_ref = _state.pop('target_ref', None)
1589        if target_ref is not None and _state.get('target', None) is None:
1590            _state['target'] = self._load_target_reference(target_ref)
1591        else:
1592            _state['target'] = dill.loads(_state['target'])
1593        self._pickle = True
1594        daemon_id = _state.get('daemon_id', None)
1595        if not daemon_id:
1596            raise ValueError("Need a daemon_id to un-pickle a Daemon.")
1597
1598        properties_path = self._get_properties_path_from_daemon_id(daemon_id)
1599        ignore_properties = properties_path.exists()
1600        if ignore_properties:
1601            _state = {
1602                key: val
1603                for key, val in _state.items()
1604                if key != 'properties'
1605            }
1606        self.__init__(**_state)
1607
1608
1609    def __repr__(self):
1610        return str(self)
1611
1612    def __str__(self):
1613        return self.daemon_id
1614
1615    def __eq__(self, other):
1616        if not isinstance(other, Daemon):
1617            return False
1618        return self.daemon_id == other.daemon_id
1619
1620    def __hash__(self):
1621        return hash(self.daemon_id)