Repository: ApeWorX/ape

Issue:
Add `--background` flag to `ape networks run`
### Describe

I take inspiration from ICP's `dfx` tool which has a `dfx start --background` and a `dfx stop` command for easily starting and stopping the dev node.

### Specification

when given background, background the process.
still track that is running so we can also add a

```
ape networks stop --network <choice>
```

command

### Dependencies

_No response_

Discussion:
@linear: <p><a href="https://linear.app/apeworx/issue/APE-1907/add-background-flag-to-ape-networks-run">APE-1907 Add `--background` flag to `ape networks run`</a></p>

Solution patch:
diff --git a/docs/userguides/networks.md b/docs/userguides/networks.md
index 77bb9f08d..7bf3a6257 100644
--- a/docs/userguides/networks.md
+++ b/docs/userguides/networks.md
@@ -525,6 +525,14 @@ To run a network with a process, use the `ape networks run` command:
 ape networks run
 ```
 
+This launches a development node in the current working terminal session.
+To continue developing, you will have to launch a new terminal session.
+Alternatively, you can use the `--background` flag to background the process:
+
+```shell
+ape networks run --background
+```
+
 By default, `ape networks run` runs a development Node (geth) process.
 To use a different network, such as `hardhat` or Anvil nodes, use the `--network` flag:
 
@@ -534,6 +542,23 @@ ape networks run --network ethereum:local:foundry
 
 To configure the network's block time, use the `--block-time` option.
 
+```shell
+ape networks run --network ethereum:local:foundry --block-time 10
+```
+
+Once you are done with your node, you can simply exit the process to tear it down.
+Or, if you used `--background` or lost the process some other way, you can stop the node using the `kill` command:
+
+```shell
+ape networks kill --all
+```
+
+To list all running networks, use the `list --running` command:
+
+```shell
+ape networks list --running
+```
+
 ## Provider Interaction
 
 Once you are connected to a network, you now have access to a `.provider`.
diff --git a/setup.py b/setup.py
index 40bf35194..e3f7c0a54 100644
--- a/setup.py
+++ b/setup.py
@@ -22,7 +22,7 @@ extras_require = {
         "hypothesis-jsonschema==0.19.0",  # JSON Schema fuzzer extension
     ],
     "lint": [
-        "ruff>=0.9.10",  # Unified linter and formatter
+        "ruff>=0.9.10,<0.10",  # Unified linter and formatter
         "mypy>=1.15.0,<1.16.0",  # Static type analyzer
         "types-PyYAML",  # Needed due to mypy typeshed
         "types-requests",  # Needed due to mypy typeshed
diff --git a/src/ape/api/providers.py b/src/ape/api/providers.py
index 34be0ebcf..922d4b96c 100644
--- a/src/ape/api/providers.py
+++ b/src/ape/api/providers.py
@@ -253,6 +253,13 @@ class ProviderAPI(BaseInterfaceModel):
         Disconnect from a provider, such as tear-down a process or quit an HTTP session.
         """
 
+    @property
+    def ipc_path(self) -> Optional[Path]:
+        """
+        Return the IPC path for the provider, if supported.
+        """
+        return None
+
     @property
     def http_uri(self) -> Optional[str]:
         """
@@ -984,7 +991,9 @@ class SubprocessProvider(ProviderAPI):
     """
 
     PROCESS_WAIT_TIMEOUT: int = 15
+    background: bool = False
     process: Optional[Popen] = None
+    allow_start: bool = True
     is_stopping: bool = False
 
     stdout_queue: Optional[JoinableQueue] = None
@@ -1059,7 +1068,7 @@ class SubprocessProvider(ProviderAPI):
             or self.config_manager.get_config("test").disconnect_providers_after
         )
         if disconnect_after:
-            atexit.register(self.disconnect)
+            atexit.register(self._disconnect_atexit)
 
         # Register handlers to ensure atexit handlers are called when Python dies.
         def _signal_handler(signum, frame):
@@ -1069,6 +1078,12 @@ class SubprocessProvider(ProviderAPI):
         signal(SIGINT, _signal_handler)
         signal(SIGTERM, _signal_handler)
 
+    def _disconnect_atexit(self):
+        if self.background:
+            return
+
+        self.disconnect()
+
     def disconnect(self):
         """
         Stop the process if it exists.
@@ -1078,25 +1093,38 @@ class SubprocessProvider(ProviderAPI):
         if self.process:
             self.stop()
 
+        # Delete entry from managed list of running nodes.
+        self.network_manager.running_nodes.remove_provider(self)
+
     def start(self, timeout: int = 20):
         """Start the process and wait for its RPC to be ready."""
 
         if self.is_connected:
             logger.info(f"Connecting to existing '{self.process_name}' process.")
             self.process = None  # Not managing the process.
-        else:
+
+        elif self.allow_start:
             logger.info(f"Starting '{self.process_name}' process.")
             pre_exec_fn = _linux_set_death_signal if platform.uname().system == "Linux" else None
             self.stderr_queue = JoinableQueue()
             self.stdout_queue = JoinableQueue()
-            out_file = PIPE if logger.level <= LogLevel.DEBUG else DEVNULL
+
+            if self.background or logger.level > LogLevel.DEBUG:
+                out_file = DEVNULL
+            else:
+                out_file = PIPE
+
             cmd = self.build_command()
-            self.process = Popen(cmd, preexec_fn=pre_exec_fn, stdout=out_file, stderr=out_file)
+            process = popen(cmd, preexec_fn=pre_exec_fn, stdout=out_file, stderr=out_file)
+            self.process = process
             spawn(self.produce_stdout_queue)
             spawn(self.produce_stderr_queue)
             spawn(self.consume_stdout_queue)
             spawn(self.consume_stderr_queue)
 
+            # Cache the process so we can manage it even if lost.
+            self.network_manager.running_nodes.cache_provider(self)
+
             with RPCTimeoutError(self, seconds=timeout) as _timeout:
                 while True:
                     if self.is_connected:
@@ -1105,6 +1133,9 @@ class SubprocessProvider(ProviderAPI):
                     time.sleep(0.1)
                     _timeout.check()
 
+        else:
+            raise ProviderError("Process not started and cannot connect to existing process.")
+
     def produce_stdout_queue(self):
         process = self.process
         if self.stdout_queue is None or process is None:
@@ -1250,3 +1281,8 @@ def _linux_set_death_signal():
     # the second argument is what signal to send to child subprocesses
     libc = ctypes.CDLL("libc.so.6")
     return libc.prctl(1, SIGTERM)
+
+
+def popen(cmd: list[str], **kwargs):
+    # Abstracted for testing purporses.
+    return Popen(cmd, **kwargs)
diff --git a/src/ape/managers/networks.py b/src/ape/managers/networks.py
index b0cf4a5c6..f0b72c9d4 100644
--- a/src/ape/managers/networks.py
+++ b/src/ape/managers/networks.py
@@ -1,27 +1,167 @@
+import os
+import signal
 from collections.abc import Collection, Iterator
 from functools import cached_property
-from typing import TYPE_CHECKING, Optional, Union
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, Optional, Union
 
 from evmchains import PUBLIC_CHAIN_META
+from pydantic import ValidationError
 
 from ape.api.networks import ProviderContextManager
 from ape.exceptions import EcosystemNotFoundError, NetworkError, NetworkNotFoundError
 from ape.managers.base import BaseManager
 from ape.utils.basemodel import (
+    BaseModel,
+    DiskCacheableModel,
     ExtraAttributesMixin,
     ExtraModelAttributes,
     get_attribute_with_extras,
     only_raise_attribute_error,
 )
 from ape.utils.misc import _dict_overlay, log_instead_of_fail
+from ape.utils.os import clean_path
 from ape_ethereum.provider import EthereumNodeProvider
 
 if TYPE_CHECKING:
     from ape.api.networks import EcosystemAPI, NetworkAPI
-    from ape.api.providers import ProviderAPI
+    from ape.api.providers import ProviderAPI, SubprocessProvider
     from ape.utils.rpc import RPCHeaders
 
 
+class NodeProcessData(BaseModel):
+    """
+    Cached data for node subprocesses managed by Ape.
+    """
+
+    network_choice: str
+    """
+    The network triple ``ecosystem:network:node``.
+    """
+
+    ipc_path: Optional[Path] = None
+    """
+    The IPC path this node process communicates on.
+    """
+
+    http_uri: Optional[str] = None
+    """
+    The HTTP URI this node process exposes.
+    """
+
+    ws_uri: Optional[str] = None
+    """
+    The websockets URI this node process exposes.
+    """
+
+    @log_instead_of_fail(default="<NodeProcessData>")
+    def __repr__(self) -> str:
+        if ipc := self.ipc_path:
+            return f"{self.network_choice} - {clean_path(ipc)}"
+
+        elif uri := (self.http_uri or self.ws_uri):
+            return f"{self.network_choice} - {uri}"
+
+        return self.network_choice
+
+    def matches_provider(self, provider: "SubprocessProvider") -> bool:
+        if self.network_choice != f"{provider.network.choice}:{provider.name}":
+            return False
+
+        # Skip if any of the connection paths (IPC, HTTP, WS) differ
+        for attr in ("ipc_path", "http_uri", "ws_uri"):
+            if getattr(provider, attr, None) != getattr(self, attr, None):
+                return False
+
+        return True
+
+
+class NodeProcessMap(DiskCacheableModel):
+    """
+    All managed running network subprocesses.
+    """
+
+    nodes: dict[int, NodeProcessData] = {}
+
+    @property
+    def path(self) -> Path:
+        return self._path
+
+    @property
+    def process_ids(self) -> tuple[int, ...]:
+        return tuple(self.nodes.keys())
+
+    def __bool__(self) -> bool:
+        """
+        True if there is at least one managed node process.
+        """
+        return bool(self.nodes)
+
+    def __contains__(self, pid_or_provider: Union[int, "SubprocessProvider"]) -> bool:
+        if isinstance(pid_or_provider, int):
+            return pid_or_provider in self.nodes
+
+        for data in self.nodes.values():
+            if data.matches_provider(pid_or_provider):
+                return True
+
+        return False
+
+    def get(self, pid: Union[int, str]) -> Optional[NodeProcessData]:
+        return self.nodes.get(int(pid))
+
+    def lookup_processes(self, provider: "SubprocessProvider") -> dict[int, NodeProcessData]:
+        return {pid: data for pid, data in self.nodes.items() if data.matches_provider(provider)}
+
+    def cache_provider(self, provider: "SubprocessProvider"):
+        # Don't use `provider.network_choice` here because we want to ensure the provider
+        # name is used and not the URI for the third part of the choice.
+        network_choice = f"{provider.network.choice}:{provider.name}"
+        data: dict[str, Any] = {
+            "network_choice": network_choice,
+            "ipc_path": provider.ipc_path,
+            "http_uri": provider.http_uri,
+            "ws_uri": provider.ws_uri,
+        }
+        data = {k: v for k, v in data.items() if v is not None}
+
+        if process := provider.process:
+            obj = NodeProcessData.model_validate(data)
+
+            # Ensure no duplicates (bad clean?)
+            self._delete_all_matching(obj)
+
+            self.nodes[process.pid] = obj
+            self.model_dump_file()
+
+        else:
+            raise NetworkError("Unable to cache subprocess-provider information: not connected.")
+
+    def _delete_all_matching(self, node: NodeProcessData):
+        pids_to_remove = set()
+        for pid, other_node in self.nodes.items():
+            if other_node == node:
+                # Exact match but different PID. Bad clean.
+                pids_to_remove.add(pid)
+
+        for pid in pids_to_remove:
+            del self.nodes[pid]
+
+    def remove_provider(self, provider: "SubprocessProvider"):
+        pids_to_remove = {
+            pid for pid, data in self.nodes.items() if data.matches_provider(provider)
+        }
+        self.nodes = {pid: node for pid, node in self.nodes.items() if pid not in pids_to_remove}
+        self.model_dump_file()
+
+    def remove_processes(self, *pids: int):
+        self.nodes = {k: v for k, v in self.nodes.items() if k not in pids}
+        self.model_dump_file()
+
+    def clean(self):
+        self._path.unlink(missing_ok=True)
+
+
 class NetworkManager(BaseManager, ExtraAttributesMixin):
     """
     The set of all blockchain network ecosystems registered from the plugin system.
@@ -97,6 +237,102 @@ class NetworkManager(BaseManager, ExtraAttributesMixin):
         """
         return self.network.ecosystem
 
+    @cached_property
+    def running_nodes(self) -> NodeProcessMap:
+        """
+        All running development nodes managed by Ape.
+        """
+        path = self.config_manager.DATA_FOLDER / "processes" / "nodes.json"
+        try:
+            return NodeProcessMap.model_validate_file(path)
+        except ValidationError:
+            path.unlink(missing_ok=True)
+            return NodeProcessMap.model_validate_file(path)
+
+    def get_running_node(self, pid: int) -> "SubprocessProvider":
+        """
+        Get a running subprocess provider for the given ``pid``.
+
+        Args:
+            pid (int): The process ID.
+
+        Returns:
+            class:`~ape.api.providers.SubprocessProvider`
+        """
+        if not (data := self.running_nodes.get(pid)):
+            raise NetworkError(f"No running node for pid '{pid}'.")
+
+        uri: Optional[Union[str, Path]] = None
+        if ipc := data.ipc_path:
+            if ipc.exists():
+                uri = ipc
+
+        else:
+            uri = data.http_uri or data.ws_uri
+
+        if uri is None:
+            NetworkError(f"Cannot connect to node on PID '{pid}': Missing URI data.")
+
+        # In this case, we want the more connectable network choice.
+        network_parts = data.network_choice.split(":")
+        network_choice = f"{':'.join(network_parts[:2])}:{uri}"
+
+        provider_settings: dict = {
+            network_parts[0]: {
+                network_parts[1]: {
+                    "ipc_path": data.ipc_path,
+                    "http_uri": data.http_uri,
+                    "ws_uri": data.ws_uri,
+                    "uri": None,
+                }
+            }
+        }
+        provider = self.get_provider_from_choice(
+            network_choice=network_choice, provider_settings=provider_settings or None
+        )
+
+        # If this is not a subprocess provider, it may be ok to proceed.
+        # However, the rest of Ape will assume it is.
+        return provider  # type: ignore[return-value]
+
+    def kill_node_process(self, *process_ids: int) -> dict[int, NodeProcessData]:
+        """
+        Kill a node process managed by Ape.
+
+        Args:
+            *process_ids (int): The process ID to kill.
+
+        Returns:
+            dict[str, :class:`~ape.managers.networks.NodeProcessData`]: The process data
+            of all terminated processes.
+        """
+        if not self.running_nodes:
+            return {}
+
+        pids_killed = {}
+        for pid in process_ids:
+            if not (data := self.running_nodes.nodes.get(pid)):
+                continue
+
+            try:
+                provider = self.get_running_node(pid)
+            except Exception:
+                # Still try to kill the process (below).
+                pass
+            else:
+                # Gracefully disconnect _before_ killing process.
+                provider.disconnect()
+
+            try:
+                os.kill(pid, signal.SIGTERM)
+            except Exception:
+                pass
+            else:
+                pids_killed[pid] = data
+
+        self.running_nodes.remove_processes(*process_ids)
+        return pids_killed
+
     def get_request_headers(
         self, ecosystem_name: str, network_name: str, provider_name: str
     ) -> "RPCHeaders":
@@ -169,7 +405,10 @@ class NetworkManager(BaseManager, ExtraAttributesMixin):
             {"fork": {self.ecosystem.name: {self.network.name: fork_settings}}},
         )
 
-        shared_kwargs: dict = {"provider_settings": provider_settings, "disconnect_after": True}
+        shared_kwargs: dict = {
+            "provider_settings": provider_settings,
+            "disconnect_after": True,
+        }
         return (
             forked_network.use_provider(provider_name, **shared_kwargs)
             if provider_name
@@ -557,6 +796,14 @@ class NetworkManager(BaseManager, ExtraAttributesMixin):
             default_network = self.default_ecosystem.default_network
             return default_network.get_provider(provider_settings=provider_settings)
 
+        elif network_choice.startswith("pid://"):
+            # Was given a process ID (already running node on local machine).
+            pid_str = network_choice[len("pid://") :]
+            if not pid_str.isdigit():
+                raise ValueError(f"Invalid PID: {pid_str}")
+
+            return self.get_running_node(int(pid_str))
+
         elif _is_adhoc_url(network_choice):
             # Custom network w/o ecosystem & network spec.
             return self.create_custom_provider(network_choice)
@@ -703,7 +950,9 @@ class NetworkManager(BaseManager, ExtraAttributesMixin):
                 continue
 
             ecosystem_data = self._get_ecosystem_data(
-                ecosystem_name, network_filter=network_filter, provider_filter=provider_filter
+                ecosystem_name,
+                network_filter=network_filter,
+                provider_filter=provider_filter,
             )
             data["ecosystems"].append(ecosystem_data)
 
diff --git a/src/ape/utils/basemodel.py b/src/ape/utils/basemodel.py
index ed231f854..d0c7eee51 100644
--- a/src/ape/utils/basemodel.py
+++ b/src/ape/utils/basemodel.py
@@ -3,6 +3,7 @@ TODO: In 0.9, move this module to `ape.types`.
 """
 
 import inspect
+import json
 from abc import ABC
 from collections.abc import Callable, Iterator, Mapping, Sequence
 from importlib import import_module
@@ -626,6 +627,25 @@ class DiskCacheableModel(BaseModel):
         super().__init__(*args, **kwargs)
         self._path = path
 
+    def model_read_file(self, path: Optional[Path] = None) -> dict:
+        """
+        Get the file's raw data. This is different from ``model_dump()`` because it
+        reads directly from the file without validation.
+        """
+        path = self._get_path(path=path)
+        return self._model_read_file(path)
+
+    @classmethod
+    def _model_read_file(cls, path: Path) -> dict:
+        """
+        Get the file's raw data. This is different from ``model_dump()`` because it
+        reads directly from the file without validation.
+        """
+        if json_str := path.read_text(encoding="utf8") if path.is_file() else "":
+            return json.loads(json_str)
+
+        return {}
+
     def model_dump_file(self, path: Optional[Path] = None, **kwargs):
         """
         Save this model to disk.
@@ -653,11 +673,8 @@ class DiskCacheableModel(BaseModel):
               if one wasn't declared at init time.
             **kwargs: Extra kwargs to pass to ``.model_validate_json()``.
         """
-        if json_str := path.read_text(encoding="utf8") if path.is_file() else "":
-            model = cls.model_validate_json(json_str, **kwargs)
-        else:
-            model = cls.model_validate({})
-
+        data = cls._model_read_file(path)
+        model = cls.model_validate(data, **kwargs)
         model._path = path
         return model
 
diff --git a/src/ape_ethereum/provider.py b/src/ape_ethereum/provider.py
index d5bfa1d8b..4e7611608 100644
--- a/src/ape_ethereum/provider.py
+++ b/src/ape_ethereum/provider.py
@@ -214,7 +214,7 @@ class Web3Provider(ProviderAPI, ABC):
 
     @property
     def _network_config(self) -> dict:
-        config: dict = self.config.get(self.network.ecosystem.name, None)
+        config: dict = self.settings.get(self.network.ecosystem.name, None)
         if config is None:
             return {}
 
@@ -226,13 +226,13 @@ class Web3Provider(ProviderAPI, ABC):
         result = None
         rpc: str
         if rpc := settings.get(key):
-            result = rpc
+            result = f"{rpc}"
 
         else:
             # See if it was configured for the network directly.
             config = self._network_config
             if rpc := config.get(key):
-                result = rpc
+                result = f"{rpc}"
 
         if result:
             if validator(result):
@@ -257,7 +257,7 @@ class Web3Provider(ProviderAPI, ABC):
 
     @property
     def _configured_uri(self) -> Optional[str]:
-        for key in ("uri", "url"):
+        for key in ("uri", "url", "ipc_path", "http_uri", "ws_uri"):
             if rpc := self._get_configured_rpc(key, _is_uri):
                 return rpc
 
@@ -1573,7 +1573,7 @@ class EthereumNodeProvider(Web3Provider, ABC):
 
     @property
     def _clean_uri(self) -> str:
-        uri = self.uri
+        uri = f"{self.uri}"
         return sanitize_url(uri) if _is_http_url(uri) or _is_ws_url(uri) else uri
 
     @property
@@ -1584,7 +1584,7 @@ class EthereumNodeProvider(Web3Provider, ABC):
         return _get_default_data_dir()
 
     @property
-    def ipc_path(self) -> Path:
+    def ipc_path(self) -> Optional[Path]:
         if path := super().ipc_path:
             return path
 
@@ -1692,9 +1692,10 @@ class EthereumNodeProvider(Web3Provider, ABC):
 
     def _log_connection(self, client_name: str):
         msg = f"Connecting to existing {client_name.strip()} node at"
+
         suffix = (
             self.ipc_path.as_posix().replace(Path.home().as_posix(), "$HOME")
-            if self.ipc_path.exists()
+            if self.ipc_path is not None and self.ipc_path.exists()
             else self._clean_uri
         )
         logger.info(f"{msg} {suffix}.")
diff --git a/src/ape_networks/_cli.py b/src/ape_networks/_cli.py
index 3a0770761..9bb74361c 100644
--- a/src/ape_networks/_cli.py
+++ b/src/ape_networks/_cli.py
@@ -51,10 +51,16 @@ def _lazy_get(name: str) -> Sequence:
 @_filter_option("ecosystem", lambda: _lazy_get("ecosystem"))
 @_filter_option("network", lambda: _lazy_get("network"))
 @_filter_option("provider", lambda: _lazy_get("provider"))
-def _list(cli_ctx, output_format, ecosystem_filter, network_filter, provider_filter):
+@click.option("--running", is_flag=True, help="List running networks")
+def _list(cli_ctx, output_format, ecosystem_filter, network_filter, provider_filter, running):
     """
     List all the registered ecosystems, networks, and providers.
     """
+    if running:
+        # TODO: Honor filter args.
+        _print_running_networks(cli_ctx)
+        return
+
     network_data = cli_ctx.network_manager.get_network_data(
         ecosystem_filter=ecosystem_filter,
         network_filter=network_filter,
@@ -109,11 +115,38 @@ def _list(cli_ctx, output_format, ecosystem_filter, network_filter, provider_fil
             ) from err
 
 
+def _print_running_networks(cli_ctx):
+    from ape.utils.os import clean_path
+
+    rows = [["PID", "NETWORK", "IPC", "HTTP", "WS"]]  # Store headers as a list
+    for pid, node in cli_ctx.network_manager.running_nodes.nodes.items():
+        rows.append(
+            [
+                pid,
+                node.network_choice,
+                str(clean_path(node.ipc_path) if node.ipc_path else None),
+                node.http_uri,
+                node.ws_uri,
+            ]
+        )
+
+    if len(rows) == 1:
+        # Only header row.
+        click.echo("Local node(s) not running.")
+
+    else:
+        col_widths = [max(len(str(row[i])) for row in rows) for i in range(len(rows[0]))]
+        for row in rows:
+            formatted_row = "  ".join(str(row[i]).ljust(col_widths[i]) for i in range(len(row)))
+            echo_rich_text(formatted_row)
+
+
 @cli.command(short_help="Start a node process")
 @ape_cli_context()
 @network_option(default="ethereum:local:node")
 @click.option("--block-time", default=None, type=int, help="Block time in seconds")
-def run(cli_ctx, provider, block_time):
+@click.option("--background", is_flag=True, help="Run in the background")
+def run(cli_ctx, provider, block_time, background):
     """
     Start a subprocess node as if running independently
     and stream stdout and stderr.
@@ -142,15 +175,22 @@ def run(cli_ctx, provider, block_time):
     cli_ctx.logger.format(fmt="%(message)s")
 
     try:
-        _run(cli_ctx, provider)
+        _run(cli_ctx, provider, background=background)
     finally:
         cli_ctx.logger.set_level(original_level)
         cli_ctx.logger.format(fmt=original_format)
 
 
-def _run(cli_ctx, provider: "SubprocessProvider"):
+def _run(cli_ctx, provider: "SubprocessProvider", background: bool = False):
+    provider.background = background
     provider.connect()
+
     if process := provider.process:
+        if background:
+            # End this process, letting the node continue running.
+            # This node can be killed later using the `ape networks kill` command.
+            return
+
         try:
             process.wait()
         finally:
@@ -163,3 +203,57 @@ def _run(cli_ctx, provider: "SubprocessProvider"):
     else:
         provider.disconnect()
         cli_ctx.abort("Process already running.")
+
+
+@cli.command(short_help="Stop node processes")
+@ape_cli_context()
+@click.argument("process_ids", nargs=-1, type=int)
+@click.option("--all", "kill_all", is_flag=True, help="Kill all running processes")
+@network_option(default=None)
+def kill(cli_ctx, process_ids, kill_all):
+    """
+    Stop node processes
+    """
+    if kill_all:
+        if process_ids:
+            raise click.BadOptionUsage("--all", "Cannot use `--all` with PID arguments.")
+
+        process_ids = cli_ctx.network_manager.running_nodes.process_ids
+
+    if not process_ids:
+        message = (
+            "No running nodes found."
+            if not cli_ctx.network_manager.running_nodes
+            else "No processes given. Use `--all` to kill all processes."
+        )
+        echo_rich_text(message)
+        return
+
+    elif processes_killed := cli_ctx.network_manager.kill_node_process(*process_ids):
+        # Killed 1 or more nodes.
+        click.echo("Stopped the following node(s):")
+        pids_stopped = set()
+        for pid, data in processes_killed.items():
+            echo_rich_text(f"\t{repr(data)}")
+            pids_stopped.add(pid)
+
+        if rest := [pid for pid in process_ids if pid not in pids_stopped]:
+            click.echo(f"The remaining process IDs were no longer valid: {','.join(rest)}.")
+
+    else:
+        # Terminated the process outside of Ape.
+        click.echo("No running nodes found, but cleaned up cache.")
+
+
+@cli.command(short_help="Check if a provider is available")
+@ape_cli_context()
+@network_option()
+def ping(cli_ctx, provider):
+    if hasattr(provider, "allow_start"):
+        # We don't want to allow starting processes; this is used for
+        # checking if processes are alive (as well as checking live URIs).
+        provider.allow_start = False
+
+    provider.connect()
+    status = "AVAILABLE" if provider.is_connected else "UNAVAILABLE"
+    click.echo(f"'{provider.network_choice}' connection status: {status}")
diff --git a/src/ape_node/provider.py b/src/ape_node/provider.py
index 04bb668c0..31ce6a599 100644
--- a/src/ape_node/provider.py
+++ b/src/ape_node/provider.py
@@ -113,6 +113,7 @@ class GethDevProcess(BaseGethProcess):
         block_time: Optional[int] = None,
         generate_accounts: bool = True,
         initialize_chain: bool = True,
+        background: bool = False,
     ):
         executable = executable or "geth"
         if not shutil.which(executable):
@@ -121,6 +122,7 @@ class GethDevProcess(BaseGethProcess):
         self._data_dir = data_dir
         self.is_running = False
         self._auto_disconnect = auto_disconnect
+        self.background = background
 
         kwargs_ctor: dict = {
             "data_dir": self.data_dir,
@@ -184,6 +186,7 @@ class GethDevProcess(BaseGethProcess):
 
         process_kwargs = {
             "auto_disconnect": kwargs.get("auto_disconnect", True),
+            "background": kwargs.get("background", False),
             "block_time": block_time,
             "executable": kwargs.get("executable"),
             "extra_funded_accounts": extra_accounts,
@@ -276,7 +279,12 @@ class GethDevProcess(BaseGethProcess):
             return
 
         self.is_running = True
-        out_file = PIPE if logger.level <= LogLevel.DEBUG else DEVNULL
+
+        if self.background or logger.level > LogLevel.DEBUG:
+            out_file = DEVNULL
+        else:
+            out_file = PIPE
+
         self.proc = Popen(
             self.command,
             stdin=PIPE,
@@ -447,9 +455,11 @@ class GethDev(EthereumNodeProvider, TestProviderAPI, SubprocessProvider):
         self._set_web3()
         if self.is_connected:
             self._complete_connect()
-        else:
+
+        elif self.allow_start:
             # Starting the process.
             self.start()
+            atexit.register(self._disconnect_atexit)
 
     def start(self, timeout: int = 20):
         geth_dev = self._create_process()
@@ -475,6 +485,8 @@ class GethDev(EthereumNodeProvider, TestProviderAPI, SubprocessProvider):
             spawn(self.consume_stdout_queue)
             spawn(self.consume_stderr_queue)
 
+            self.network_manager.running_nodes.cache_provider(self)
+
     def _create_process(self) -> GethDevProcess:
         # NOTE: Using JSON mode to ensure types can be passed as CLI args.
         test_config = self.config_manager.get_config("test").model_dump(mode="json")
@@ -484,9 +496,11 @@ class GethDev(EthereumNodeProvider, TestProviderAPI, SubprocessProvider):
             test_config["executable"] = self.settings.executable
 
         test_config["ipc_path"] = self.ipc_path
-        test_config["auto_disconnect"] = self._test_runner is None or test_config.get(
-            "disconnect_providers_after", True
-        )
+
+        # Let the provider handle disconnecting the process.
+        # This avoids multiple atexit handlers from being unnecessarily
+        # registered that do some of the same thing.
+        test_config["auto_disconnect"] = False
 
         # Include extra accounts to allocated funds to at genesis.
         extra_accounts = self.settings.ethereum.local.get("extra_funded_accounts", [])
@@ -494,9 +508,14 @@ class GethDev(EthereumNodeProvider, TestProviderAPI, SubprocessProvider):
         extra_accounts = list({a.lower() for a in extra_accounts})
         test_config["extra_funded_accounts"] = extra_accounts
         test_config["initial_balance"] = self.test_config.balance
+        test_config["background"] = self.background
         uri = self.ws_uri or self.uri
+
         return GethDevProcess.from_uri(
-            uri, self.data_dir, block_time=self.block_time, **test_config
+            uri,
+            self.data_dir,
+            block_time=self.block_time,
+            **test_config,
         )
 
     def disconnect(self):
@@ -505,6 +524,9 @@ class GethDev(EthereumNodeProvider, TestProviderAPI, SubprocessProvider):
             self._process.disconnect()
             self._process = None
 
+            # Remove self from managed-processes list.
+            self.network_manager.running_nodes.remove_provider(self)
+
         # Also unset the subprocess-provider reference.
         # NOTE: Type ignore is wrong; TODO: figure out why.
         self.process = None  # type: ignore[assignment]


Test patch:
diff --git a/tests/functional/geth/test_network_manager.py b/tests/functional/geth/test_network_manager.py
index 26fe5ed09..fad1ef4b7 100644
--- a/tests/functional/geth/test_network_manager.py
+++ b/tests/functional/geth/test_network_manager.py
@@ -64,3 +64,16 @@ def test_parse_network_choice_evmchains(networks, connection_str):
             else f"{connection_str}:node"
         )
         assert moon_provider.network_choice == expected_network_choice
+
+
+@geth_process_test
+def test_parse_network_choice_pid(geth_provider, networks):
+    if proc := geth_provider.process:
+        pid = proc.pid
+    else:
+        pid = next(networks.running_nodes.lookup_processes(geth_provider))
+
+    # Show we are able to connect to providers via PID URL.
+    with networks.parse_network_choice(f"pid://{pid}") as provider:
+        assert provider.is_connected
+        assert provider.network_choice == f"ethereum:local:{geth_provider.ipc_path}"
diff --git a/tests/functional/test_network_manager.py b/tests/functional/test_network_manager.py
index be5ebeb04..9c2b17708 100644
--- a/tests/functional/test_network_manager.py
+++ b/tests/functional/test_network_manager.py
@@ -5,8 +5,11 @@ import pytest
 
 import ape
 from ape.api.networks import EcosystemAPI
+from ape.api.providers import SubprocessProvider
 from ape.exceptions import NetworkError, ProviderNotFoundError
+from ape.managers.networks import NodeProcessData, NodeProcessMap
 from ape.utils.misc import LOCAL_NETWORK_NAME
+from ape.utils.os import create_tempdir
 from ape.utils.testing import DEFAULT_TEST_CHAIN_ID
 
 
@@ -301,7 +304,7 @@ def test_create_custom_provider_ws(networks, scheme):
 def test_create_custom_provider_ipc(networks):
     provider = networks.create_custom_provider("path/to/geth.ipc")
     assert provider.ipc_path == Path("path/to/geth.ipc")
-    assert provider.uri == provider.ipc_path
+    assert provider.uri == f"{provider.ipc_path}"
 
 
 def test_ecosystems(networks):
@@ -496,3 +499,40 @@ def test_get_ecosystem_from_evmchains(networks):
     moonbeam = networks.get_ecosystem("moonbeam")
     assert isinstance(moonbeam, EcosystemAPI)
     assert moonbeam.name == "moonbeam"
+
+
+class TestNodeProcessData:
+    def test_matches_provider(self, eth_tester_provider):
+        data = NodeProcessData(
+            network_choice=f"{eth_tester_provider.network.choice}:{eth_tester_provider.name}",
+            ipc_path="test.ipc",
+        )
+        assert not data.matches_provider(eth_tester_provider)
+        data.ipc_path = None
+        assert data.matches_provider(eth_tester_provider)
+
+
+class TestNodeProcessMap:
+    def test_cache_and_remove_provider(self, mocker, eth_tester_provider):
+        mock_process = mocker.MagicMock()
+        mock_process.pid = 12345678901234567890
+
+        class MyFakeProvider(SubprocessProvider):
+            @property
+            def is_connected(self):
+                return True
+
+        # Hack to allow abstract methods.
+        MyFakeProvider.__abstractmethods__ = set()  # type: ignore
+        provider = MyFakeProvider(name="fake", network=eth_tester_provider.network)  # type: ignore
+        provider.process = mock_process
+
+        with create_tempdir() as tmp:
+            file = tmp / "networks.json"
+            mapping = NodeProcessMap(path=file)
+            mapping.cache_provider(provider)
+            assert provider in mapping
+
+            # Not test removing it.
+            mapping.remove_provider(provider)
+            assert provider not in mapping
diff --git a/tests/functional/test_provider.py b/tests/functional/test_provider.py
index 4f270c50f..2e39cb2da 100644
--- a/tests/functional/test_provider.py
+++ b/tests/functional/test_provider.py
@@ -12,6 +12,7 @@ from requests import HTTPError
 from web3.exceptions import ContractPanicError, TimeExhausted
 
 from ape import convert
+from ape.api.providers import SubprocessProvider
 from ape.exceptions import (
     APINotImplementedError,
     BlockNotFoundError,
@@ -367,7 +368,9 @@ def test_set_timestamp_handle_same_time_race_condition(mocker, eth_tester_provid
     eth_tester_provider.set_timestamp(123)
 
 
-def test_get_virtual_machine_error_when_txn_failed_includes_base_error(eth_tester_provider):
+def test_get_virtual_machine_error_when_txn_failed_includes_base_error(
+    eth_tester_provider,
+):
     txn_failed = TransactionFailed()
     actual = eth_tester_provider.get_virtual_machine_error(txn_failed)
     assert actual.base_err == txn_failed
@@ -416,7 +419,12 @@ def test_no_comma_in_rpc_url():
 
 
 def test_send_transaction_when_no_error_and_receipt_fails(
-    mocker, mock_web3, mock_transaction, eth_tester_provider, owner, vyper_contract_instance
+    mocker,
+    mock_web3,
+    mock_transaction,
+    eth_tester_provider,
+    owner,
+    vyper_contract_instance,
 ):
     start_web3 = eth_tester_provider._web3
     eth_tester_provider._web3 = mock_web3
@@ -661,7 +669,11 @@ def test_account_balance_state(project, eth_tester_provider, owner):
 
 @pytest.mark.parametrize(
     "uri,key",
-    [("ws://example.com", "ws_uri"), ("wss://example.com", "ws_uri"), ("wss://example.com", "uri")],
+    [
+        ("ws://example.com", "ws_uri"),
+        ("wss://example.com", "ws_uri"),
+        ("wss://example.com", "uri"),
+    ],
 )
 def test_node_ws_uri(project, uri, key):
     node = project.network_manager.ethereum.sepolia.get_provider("node")
@@ -755,3 +767,55 @@ def test_connect_uses_cached_chain_id(mocker, mock_web3, ethereum, eth_tester_pr
     provider.connect()
     # It is still cached from the previous connection.
     assert chain_id_tracker.call_count == 1
+
+
+class TestSubprocessProvider:
+    FAKE_PID = 12345678901234567890
+
+    @pytest.fixture(autouse=True)
+    def mock_process(self, mocker):
+        mock_process = mocker.MagicMock()
+        mock_process.pid = self.FAKE_PID
+        return mock_process
+
+    @pytest.fixture(autouse=True)
+    def popen_patch(self, mocker, mock_process):
+        # Prevent actually creating new processes.
+        patch = mocker.patch("ape.api.providers.popen")
+        patch.return_value = mock_process
+        return patch
+
+    @pytest.fixture(autouse=True)
+    def spawn_patch(self, mocker):
+        # Prevent spawning process monitoring threads.
+        return mocker.patch("ape.api.providers.spawn")
+
+    @pytest.fixture
+    def subprocess_provider(self, popen_patch, eth_tester_provider):
+        class MockSubprocessProvider(SubprocessProvider):
+            @property
+            def is_connected(self):
+                # Once Popen is called once, we are "connected"
+                return popen_patch.call_count > 0
+
+            def build_command(self) -> list[str]:
+                return ["apemockprocess"]
+
+        # Hack to allow abstract methods anyway.
+        MockSubprocessProvider.__abstractmethods__ = set()  # type: ignore
+
+        return MockSubprocessProvider(name="apemockprocess", network=eth_tester_provider.network)  # type: ignore
+
+    def test_start(self, subprocess_provider):
+        assert not subprocess_provider.is_connected
+        subprocess_provider.start()
+        assert subprocess_provider.is_connected
+
+        # Show it gets tracked in network manager's managed nodes.
+        assert self.FAKE_PID in subprocess_provider.network_manager.running_nodes
+
+    def test_start_allow_start_false(self, subprocess_provider):
+        subprocess_provider.allow_start = False
+        expected = r"Process not started and cannot connect to existing process\."
+        with pytest.raises(ProviderError, match=expected):
+            subprocess_provider.start()
diff --git a/tests/integration/cli/test_networks.py b/tests/integration/cli/test_networks.py
index 45d33bc58..a9ee0e82e 100644
--- a/tests/integration/cli/test_networks.py
+++ b/tests/integration/cli/test_networks.py
@@ -152,6 +152,15 @@ def test_list_geth(ape_cli, runner, networks, project):
     assert actual_uri.startswith("http")
 
 
+@skip_projects_except("geth")
+def test_list_running(ape_cli, runner, geth_provider):
+    result = runner.invoke(ape_cli, ("networks", "list", "--running"))
+    assert result.exit_code == 0
+    assert geth_provider.ipc_path is not None, "any uri is needed for test"
+    actual = "".join(result.output.split("\n"))
+    assert f"{geth_provider.ipc_path}" in actual or "Local node(s) not running." in actual
+
+
 @run_once
 def test_list_filter_networks(ape_cli, runner):
     result = runner.invoke(ape_cli, ("networks", "list", "--network", "sepolia"))