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

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