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