intellicrack.core.process_manager

Centralized process management for Intellicrack.

This module provides a singleton ProcessManager that tracks all spawned processes and ensures proper cleanup on application exit, signal handling, or exceptions.

pid_is_running(pid)[source]

Report whether a live OS process exists for pid.

Public wrapper over the module’s PID-liveness probe, for callers outside this module that need to check a process they track by PID rather than by Popen handle.

Parameters:: pid (int) – The process identifier to verify.
Returns:: True when a live process is detected for pid.
Return type:: bool

exception ProcessStateError[source]

Bases: RuntimeError

Raised when a tracked subprocess finishes in an unexpected state.

This error surfaces cases where a subprocess wrapped by ProcessManager leaves returncode unset after communicate returns, indicating the operating system failed to report the final exit status.

__init__(name, pid, message=None)[source]

Initialize the ProcessStateError.

Parameters:

name (str) – Human-readable name of the subprocess.
pid (int) – Process ID of the subprocess.
message (str | None) – Optional additional detail describing the failure.

Return type:

None

class ProcessType[source]

Bases: Enum

Type of process being tracked.

SUBPROCESS = 'subprocess'

ASYNC_SUBPROCESS = 'async_subprocess'

EXTERNAL_TOOL = 'external_tool'

SANDBOX = 'sandbox'

DEBUGGER = 'debugger'

class TrackedProcess[source]

Bases: object

Information about a tracked process.

process: Popen[bytes] | Process

process_type: ProcessType

name: str

registered_at: datetime

metadata: dict[str, Any]

cleanup_callback: Callable[[], Coroutine[Any, Any, None]] | None = None

property pid: int | None

Process ID if available.

Returns:: The process ID, or None if not available.
Return type:: int | None

property is_running: bool

Check if process is still running.

Returns:: True if the process is still running, False otherwise.
Return type:: bool

check_running()[source]

Check if process is still running (non-cached version).

This method exists to avoid mypy’s type narrowing on property access. Use this when checking running state after an operation that may have changed the process state.

Returns:: True if the process is still running, False otherwise.
Return type:: bool

__init__(process, process_type, name, registered_at=<factory>, metadata=<factory>, cleanup_callback=None)

Parameters:

process (Popen[bytes] | Process)
process_type (ProcessType)
name (str)
registered_at (datetime)
metadata (dict[str, Any])
cleanup_callback (Callable[[], Coroutine[Any, Any, None]] | None)

Return type:

None

class TrackedEntry[source]

Bases: object

Unified, serializable view of a tracked process or external PID.

Produced by ProcessManager.get_all_tracked_entries(), combining subprocess-backed TrackedProcess entries with PIDs registered via ProcessManager.register_external_pid() into a single shape so UI consumers (e.g. the Tracked tab) can render both categories without knowing which backing store an entry came from.

pid: int

name: str

process_type: ProcessType

registered_at: datetime

is_running: bool

metadata: dict[str, Any]

__init__(pid, name, process_type, registered_at, is_running, metadata=<factory>)

Parameters:

pid (int)
name (str)
process_type (ProcessType)
registered_at (datetime)
is_running (bool)
metadata (dict[str, Any])

Return type:

None

class ProcessManager[source]

Bases: object

Centralized manager for all spawned processes.

This singleton class tracks all processes spawned by Intellicrack and ensures proper cleanup on application exit. It handles: - Normal exit via atexit handlers - Signal-based termination (SIGINT, SIGTERM) - Graceful shutdown with timeout followed by forceful termination

Variables:

DEFAULT_GRACEFUL_TIMEOUT (float) – Default graceful shutdown timeout in seconds.
DEFAULT_FORCE_TIMEOUT (float) – Default forced termination timeout in seconds.

DEFAULT_GRACEFUL_TIMEOUT: float = 5.0

DEFAULT_FORCE_TIMEOUT: float = 3.0

static __new__(cls)[source]

Create or return the singleton instance.

Returns:: The singleton ProcessManager instance.
Return type:: Self

__init__()[source]

Initialize the ProcessManager singleton instance.

Return type:: None

classmethod get_instance()[source]

Get the singleton instance.

Returns:: The singleton ProcessManager instance.
Return type:: ProcessManager

classmethod reset_instance()[source]

Reset the singleton instance (for testing).

Return type:: None

prepare_for_teardown()[source]

Reset internal state in preparation for singleton teardown.

Return type:: None

install_handlers()[source]

Install signal handlers and atexit hook for cleanup.

This should be called once during application startup, typically in main.py before any processes are spawned. The atexit hook is registered at most once per Python interpreter — even when reset_instance() is invoked between calls — using a module-level guard so cleanup never executes twice on shutdown.

Return type:: None

uninstall_handlers()[source]

Uninstall signal handlers (restore original handlers).

Return type:: None

run_atexit_cleanup()[source]

Public entry point that runs the at-exit cleanup once.

Delegates to _atexit_cleanup(); provided so the global hook can invoke instance cleanup without violating member-access lint rules.

Return type:: None

static terminate_tree(pid, graceful_timeout=5.0, force_timeout=3.0)[source]

Terminate a process tree using psutil.

Kills the root process and all its descendants. First sends SIGTERM and waits for graceful_timeout, then sends SIGKILL to any survivors and waits for force_timeout.

Parameters:

pid (int) – Root process ID.
graceful_timeout (float) – Seconds to wait for SIGTERM.
force_timeout (float) – Seconds to wait for SIGKILL.

Return type:

None

register(process, name, process_type=ProcessType.SUBPROCESS, metadata=None, cleanup_callback=None)[source]

Register a process for tracking.

Parameters:

process (Popen[bytes] | Process) – The process to track.
name (str) – Human-readable name for the process.
process_type (ProcessType) – Type of process being tracked.
metadata (dict[str, Any] | None) – Optional metadata about the process.
cleanup_callback (Callable[[], Coroutine[Any, Any, None]] | None) – Optional async callback for custom cleanup.

Returns:

The process ID used as the tracking key.

Return type:

int

unregister(pid)[source]

Unregister a process from tracking.

Parameters:: pid (int) – The process ID to unregister.
Returns:: The tracked process info if found, None otherwise.
Return type:: TrackedProcess | None

get_tracked(pid)[source]

Get tracked process information.

Parameters:: pid (int) – The process ID to look up.
Returns:: The tracked process info if found, None otherwise.
Return type:: TrackedProcess | None

get_all_tracked()[source]

Get all tracked processes.

Returns:: List of all tracked processes.
Return type:: list[TrackedProcess]

get_all_tracked_entries()[source]

Get a unified view of all tracked processes and external PIDs.

Combines subprocess-backed entries from get_all_tracked() with PIDs registered via register_external_pid(), resolving live running state for each so UI consumers (e.g. the Tracked tab) can render both categories without querying two separate stores.

Returns:: Combined, serializable tracked-process entries.
Return type:: list[TrackedEntry]

get_running_processes()[source]

Get all currently running tracked processes.

Returns:: List of tracked processes that are still running.
Return type:: list[TrackedProcess]

async terminate_process(pid, graceful_timeout=None, force_timeout=None)[source]

Terminate a specific process.

Parameters:

pid (int) – The process ID to terminate.
graceful_timeout (float | None) – Timeout for graceful termination.
force_timeout (float | None) – Timeout for forceful termination.

Returns:

True if process was terminated, False if not found or already stopped.

Return type:

bool

async cleanup_all_async(graceful_timeout=None, force_timeout=None)[source]

Cleanup all tracked processes asynchronously.

Parameters:

graceful_timeout (float | None) – Timeout for graceful termination per process.
force_timeout (float | None) – Timeout for forceful termination per process.

Return type:

None

is_shutdown_requested()[source]

Check if shutdown has been requested via signal.

Returns:: True if a shutdown signal has been received, False otherwise.
Return type:: bool

clear_shutdown_request()[source]

Clear the shutdown request flag.

Return type:: None

request_shutdown()[source]

Request and perform an immediate shutdown of all tracked processes.

Sets the shutdown flag observed by is_shutdown_requested(), then synchronously runs _sync_cleanup(), which walks every tracked subprocess and external PID (including descendant processes) and terminates them. This gives callers – such as MainWindow.closeEvent – a guaranteed, bounded final sweep that reaps any process left behind by a bridge whose own detach/shutdown/stop teardown silently failed or timed out, rather than merely flipping a flag nothing else acts on.

Return type:: None

property process_count: int

Number of tracked processes.

Returns:: The total count of tracked processes.
Return type:: int

property running_count: int

Number of running tracked processes.

Returns:: The count of currently running tracked processes.
Return type:: int

__repr__()[source]

Return string representation.

Returns:: A string representation of the ProcessManager state.
Return type:: str

run_tracked(args, name, *, capture_output=True, text=True, timeout=None, cwd=None, env=None, check=False, creationflags=0)[source]

Execute a subprocess with ProcessManager tracking.

This method wraps subprocess execution to ensure the process is tracked and will be terminated during application shutdown.

Parameters:

args (list[str]) – Command and arguments to execute.
name (str) – Human-readable name for the process.
capture_output (bool) – Capture stdout and stderr.
text (bool) – Decode output as text (returns str); False returns bytes.
timeout (float | None) – Maximum time to wait for process.
cwd (str | None) – Working directory for the process.
env (dict[str, str] | None) – Environment variables for the process.
check (bool) – Raise CalledProcessError if process returns non-zero.
creationflags (int) – Windows process creation flags.

Returns:

Execution results (stdout/stderr as str if text=True).

Return type:

CompletedProcess[Any]

Raises:

TimeoutExpired – If timeout exceeded.
CalledProcessError – If check=True and process failed.
ProcessStateError – If the subprocess returns without an exit status.

async run_tracked_async(args, name, *, capture_output=True, text=True, process_timeout=None, cwd=None, env=None, check=False, creationflags=0)[source]

Execute a subprocess asynchronously with ProcessManager tracking.

This method wraps subprocess execution to ensure the process is tracked and will be terminated during application shutdown. It delegates to run_tracked via asyncio.to_thread.

Parameters:

args (list[str]) – Command and arguments to execute.
name (str) – Human-readable name for the process.
capture_output (bool) – Capture stdout and stderr.
text (bool) – Decode output as text.
process_timeout (float | None) – Maximum time to wait for process.
cwd (str | None) – Working directory for the process.
env (dict[str, str] | None) – Environment variables for the process.
check (bool) – Raise CalledProcessError if process returns non-zero.
creationflags (int) – Windows process creation flags.

Returns:

Execution results with captured stdout/stderr.

Return type:

CompletedProcess[Any]

register_external_pid(pid, name, process_type=ProcessType.EXTERNAL_TOOL, metadata=None)[source]

Register an external process by PID for cleanup tracking.

Use this for processes not directly spawned by subprocess (e.g., daemonized processes) that should be terminated when the application exits. Verifies the PID corresponds to a live OS process via _pid_exists() before registering.

Parameters:

pid (int) – The process ID to track.
name (str) – Human-readable name for the process.
process_type (ProcessType) – Type of process being tracked.
metadata (dict[str, Any] | None) – Optional metadata about the process.

Raises:

ValueError – If pid does not correspond to a live process.

Return type:

None

unregister_external_pid(pid)[source]

Unregister an external process from tracking.

Parameters:: pid (int) – The process ID to unregister.
Returns:: True if the PID was registered and removed, False otherwise.
Return type:: bool

terminate_external_pid(pid, *, force=False)[source]

Terminate an external process by PID using psutil (tree kill).

Parameters:

pid (int) – The process ID to terminate.
force (bool) – If True, skip graceful termination and kill immediately.

Returns:

True if process was terminated (or already gone), False on error.

Return type:

bool