meerschaum.utils.daemon

109def daemon_action(**kw) -> SuccessTuple:
110    """Execute a Meerschaum action as a daemon."""
111    from meerschaum.utils.packages import run_python_package
112    from meerschaum.utils.threading import Thread
113    from meerschaum._internal.arguments._parse_arguments import parse_dict_to_sysargs
114    from meerschaum.actions import get_action
115
116    kw['daemon'] = True
117    kw['shell'] = False
118
119    action = kw.get('action', None)
120    if action and get_action(action) is None:
121        if not kw.get('allow_shell_job') and not kw.get('force'):
122            return False, (
123                f"Action '{action}' isn't recognized.\n\n"
124                + "  Include `--allow-shell-job`, `--force`, or `-f`\n  " 
125                + "to enable shell commands to run as Meerschaum jobs."
126            )
127
128    sysargs = parse_dict_to_sysargs(kw)
129    rc = run_python_package('meerschaum', sysargs, venv=None, debug=False)
130    msg = "Success" if rc == 0 else f"Daemon returned code: {rc}"
131    return rc == 0, msg

 43def daemon_entry(sysargs: Optional[List[str]] = None) -> SuccessTuple:
 44    """Parse sysargs and execute a Meerschaum action as a daemon.
 45
 46    Parameters
 47    ----------
 48    sysargs: Optional[List[str]], default None
 49        The command line arguments used in a Meerschaum action.
 50
 51    Returns
 52    -------
 53    A SuccessTuple.
 54    """
 55    from meerschaum._internal.entry import entry
 56    _args = {}
 57    if '--name' in sysargs or '--job-name' in sysargs:
 58        from meerschaum._internal.arguments._parse_arguments import parse_arguments
 59        _args = parse_arguments(sysargs)
 60    filtered_sysargs = [arg for arg in sysargs if arg not in ('-d', '--daemon')]
 61    try:
 62        label = shlex.join(filtered_sysargs) if sysargs else None
 63    except Exception:
 64        label = ' '.join(filtered_sysargs) if sysargs else None
 65
 66    name = _args.get('name', None)
 67    daemon = None
 68    if name:
 69        try:
 70            daemon = Daemon(daemon_id=name)
 71        except Exception:
 72            daemon = None
 73
 74    if daemon is not None:
 75        existing_sysargs = daemon.properties['target']['args'][0]
 76        existing_kwargs = parse_arguments(existing_sysargs)
 77
 78        ### Remove sysargs because flags are aliased.
 79        _ = _args.pop('daemon', None)
 80        _ = _args.pop('sysargs', None)
 81        _ = _args.pop('filtered_sysargs', None)
 82        debug = _args.pop('debug', None)
 83        _args['sub_args'] = sorted(_args.get('sub_args', []))
 84        _ = existing_kwargs.pop('daemon', None)
 85        _ = existing_kwargs.pop('sysargs', None)
 86        _ = existing_kwargs.pop('filtered_sysargs', None)
 87        _ = existing_kwargs.pop('debug', None)
 88        existing_kwargs['sub_args'] = sorted(existing_kwargs.get('sub_args', []))
 89
 90        ### Only run if the kwargs equal or no actions are provided.
 91        if existing_kwargs == _args or not _args.get('action', []):
 92            if daemon.status == 'running':
 93                return True, f"Daemon '{daemon}' is already running."
 94            return daemon.run(
 95                debug=debug,
 96                allow_dirty_run=True,
 97            )
 98
 99    success_tuple = run_daemon(
100        entry,
101        filtered_sysargs,
102        daemon_id=_args.get('name', None) if _args else None,
103        label=label,
104        keep_daemon_output=('--rm' not in (sysargs or [])),
105    )
106    return success_tuple

157def get_daemons() -> List[Daemon]:
158    """
159    Return all existing Daemons, sorted by end time.
160    """
161    daemons = [Daemon(daemon_id=d_id) for d_id in get_daemon_ids()]
162    daemons_status = {daemon: daemon.status for daemon in daemons}
163    running_daemons = {
164        daemon: daemons_status[daemon]
165        for daemon in daemons
166        if daemons_status[daemon] == 'running'
167    }
168    paused_daemons = {
169        daemon: daemons_status[daemon]
170        for daemon in daemons
171        if daemons_status[daemon] == 'paused'
172    }
173    stopped_daemons = {
174        daemon: daemons_status[daemon]
175        for daemon in daemons
176        if daemons_status[daemon] == 'stopped'
177    }
178    daemons_began = {
179        daemon: daemon.properties.get('process', {}).get('began', '9999')
180        for daemon in daemons
181    }
182    daemons_paused = {
183        daemon: daemon.properties.get('process', {}).get('paused', '9999')
184        for daemon in daemons
185    }
186    daemons_ended = {
187        daemon: daemon.properties.get('process', {}).get('ended', '9999')
188        for daemon in daemons
189    }
190    sorted_stopped_daemons = [
191        daemon
192        for daemon in sorted(stopped_daemons, key=lambda x: daemons_ended[x])
193    ]
194    sorted_paused_daemons = [
195        daemon
196        for daemon in sorted(paused_daemons, key=lambda x: daemons_paused[x])
197    ]
198    sorted_running_daemons = [
199        daemon
200        for daemon in sorted(running_daemons, key=lambda x: daemons_began[x])
201    ]
202    return sorted_stopped_daemons + sorted_paused_daemons + sorted_running_daemons

205def get_daemon_ids() -> List[str]:
206    """
207    Return the IDs of all daemons on disk.
208    """
209    import meerschaum.config.paths as paths
210    return [
211        daemon_dir
212        for daemon_dir in sorted(os.listdir(paths.DAEMON_RESOURCES_PATH))
213        if (paths.DAEMON_RESOURCES_PATH / daemon_dir / 'properties.json').exists()
214    ]

217def get_running_daemons(daemons: Optional[List[Daemon]] = None) -> List[Daemon]:
218    """
219    Return a list of currently running daemons.
220    """
221    if daemons is None:
222        daemons = get_daemons()
223    return [
224        d
225        for d in daemons
226        if d.status == 'running'
227    ]

243def get_stopped_daemons(daemons: Optional[List[Daemon]] = None) -> List[Daemon]:
244    """
245    Return a list of stopped daemons.
246    """
247    if daemons is None:
248        daemons = get_daemons()
249
250    return [
251        d
252        for d in daemons
253        if d.status == 'stopped'
254    ]

230def get_paused_daemons(daemons: Optional[List[Daemon]] = None) -> List[Daemon]:
231    """
232    Return a list of active but paused daemons.
233    """
234    if daemons is None:
235        daemons = get_daemons()
236    return [
237        d
238        for d in daemons
239        if d.status == 'paused'
240    ]

257def get_filtered_daemons(
258    filter_list: Optional[List[str]] = None,
259    warn: bool = False,
260) -> List[Daemon]:
261    """
262    Return a list of `Daemons` filtered by a list of `daemon_ids`.
263    Only `Daemons` that exist are returned.
264    
265    If `filter_list` is `None` or empty, return all `Daemons` (from `get_daemons()`).
266
267    Parameters
268    ----------
269    filter_list: Optional[List[str]], default None
270        List of `daemon_ids` to include. If `daemon_ids` is `None` or empty,
271        return all `Daemons`.
272
273    warn: bool, default False
274        If `True`, raise warnings for non-existent `daemon_ids`.
275
276    Returns
277    -------
278    A list of Daemon objects.
279
280    """
281    if not filter_list:
282        daemons = get_daemons()
283        return [d for d in daemons if not d.hidden]
284
285    from meerschaum.utils.warnings import warn as _warn
286    daemons = []
287    for d_id in filter_list:
288        try:
289            d = Daemon(daemon_id=d_id)
290            _exists = d.path.exists()
291        except Exception:
292            _exists = False
293        if not _exists:
294            if warn:
295                _warn(f"Daemon '{d_id}' does not exist.", stack=False)
296            continue
297        daemons.append(d)
298    return daemons

117def get_new_daemon_name() -> str:
118    """
119    Generate a new random name until a unique one is found
120    (up to ~6000 maximum possibilities).
121    """
122    import meerschaum.config.paths as paths
123    existing_names = (
124        os.listdir(paths.DAEMON_RESOURCES_PATH)
125        if paths.DAEMON_RESOURCES_PATH.exists()
126        else []
127    )
128    while True:
129        _name = generate_random_name()
130        if _name in existing_names:
131            continue
132        break
133    return _name

134def run_daemon(
135    func: Callable[[Any], Any],
136    *args,
137    daemon_id: Optional[str] = None,
138    keep_daemon_output: bool = True,
139    allow_dirty_run: bool = False,
140    label: Optional[str] = None,
141    **kw
142) -> Any:
143    """Execute a function as a daemon."""
144    daemon = Daemon(
145        func,
146        daemon_id=daemon_id,
147        target_args=[arg for arg in args],
148        target_kw=kw,
149        label=label,
150    )
151    return daemon.run(
152        keep_daemon_output=keep_daemon_output,
153        allow_dirty_run=allow_dirty_run,
154    )

301def running_in_daemon() -> bool:
302    """
303    Return whether the current thread is running in a Daemon context.
304    """
305    from meerschaum._internal.static import STATIC_CONFIG
306    daemon_env_var = STATIC_CONFIG['environment']['daemon_id']
307    return daemon_env_var in os.environ

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