From 8f83054c274db888f8019750798917958095766b Mon Sep 17 00:00:00 2001
From: Aurash Karimi <a.karimi@imperial.ac.uk>
Date: Fri, 13 Mar 2026 15:44:03 +0000
Subject: [PATCH 01/10] correct flush command behaviour

---
 src/drunc/process_manager/process_manager.py  |  89 +++-----
 .../process_manager/ssh_process_manager.py    | 201 ++++++++++++------
 .../ssh_process_lifetime_manager_shell.py     |  13 ++
 3 files changed, 185 insertions(+), 118 deletions(-)

diff --git a/src/drunc/process_manager/process_manager.py b/src/drunc/process_manager/process_manager.py
index f997ef808..8fc32e1d8 100644
--- a/src/drunc/process_manager/process_manager.py
+++ b/src/drunc/process_manager/process_manager.py
@@ -13,12 +13,9 @@
     BootRequest,
     LogLines,
     LogRequest,
-    ProcessDescription,
     ProcessInstance,
     ProcessInstanceList,
     ProcessQuery,
-    ProcessRestriction,
-    ProcessUUID,
 )
 from druncschema.process_manager_pb2_grpc import ProcessManagerServicer
 from druncschema.request_response_pb2 import (
@@ -448,6 +445,10 @@ def ps(
 
         return response
 
+    @abc.abstractmethod
+    def _flush_impl(self, query: ProcessQuery) -> ProcessInstanceList:
+        raise NotImplementedError
+
     # ORDER MATTERS!
     @broadcasted  # outer most wrapper 1st step
     @authentified_and_authorised(
@@ -456,64 +457,38 @@ def ps(
     def flush(
         self, request: ProcessQuery, context: ServicerContext
     ) -> ProcessInstanceList:
+        """Remove dead processes from tracking so they no longer appear in ps.
+
+        Dead processes that were killed externally (e.g. via kill -9) will remain
+        visible in ps until flushed. This command clears them from internal state
+        so they cannot be restarted and will not appear in subsequent ps output.
+
+        Args:
+            request: ProcessQuery specifying which processes to flush.
+            context: gRPC servicer context (unused directly).
+
+        Returns:
+            ProcessInstanceList containing the processes that were flushed.
+        """
         self.log.debug(f"{self.name} running flush")
 
-        ret = []
-        for uuid in self._get_process_uid(request):
-            # Some unknown process was found, assume it is dead and move on
-            if uuid not in self.boot_request:
-                pu = ProcessUUID(uuid=uuid)
-                pi = ProcessInstance(
-                    process_description=ProcessDescription(),
-                    process_restriction=ProcessRestriction(),
-                    status_code=ProcessInstance.StatusCode.DEAD,
-                    return_code=None,
-                    uuid=pu,
-                )
-                ret += [pi]
-                continue
-
-            pd = ProcessDescription()
-            pd.CopyFrom(self.boot_request[uuid].process_description)
-            pr = ProcessRestriction()
-            pr.CopyFrom(self.boot_request[uuid].process_restriction)
-            pu = ProcessUUID(uuid=uuid)
-
-            return_code = None
-            try:
-                if not self.process_store[
-                    uuid
-                ].is_alive():  # OMG!! remove this implementation code
-                    return_code = self.process_store[uuid].exit_code
-            except Exception:
-                pass
-
-            # If a process is already dead, remove it from the process store
-            if not self.process_store[uuid].is_alive():
-                pi = ProcessInstance(
-                    process_description=pd,
-                    process_restriction=pr,
-                    status_code=(
-                        ProcessInstance.StatusCode.RUNNING
-                        if self.process_store[uuid].is_alive()
-                        else ProcessInstance.StatusCode.DEAD
-                    ),
-                    return_code=return_code,
-                    uuid=pu,
-                )
-                # If we know that this process has died intentionally, remove it from
-                # tracking
-                self.remove_process_from_expected_dead_processes(uuid)
+        try:
+            response = self._flush_impl(request)
+        except NotImplementedError:
+            raise DruncNotImplementedException(
+                message="Implementation missing",
+                domain="ProcessManager.flush",
+            )
+        except Exception as e:
+            context_msg = f"Unhandled exception in ProcessManager.flush: {e}"
+            self.log.exception(context_msg)
 
-                del self.process_store[uuid]
-                ret += [pi]
+            raise DruncCommandException(
+                message=context_msg,
+                domain="ProcessManager.flush",
+            )
 
-        return ProcessInstanceList(
-            name=self.name,
-            token=None,
-            values=ret,
-            flag=ResponseFlag.EXECUTED_SUCCESSFULLY,
-        )
+        return response
 
     # ORDER MATTERS!
     @broadcasted  # outer most wrapper 1st step
diff --git a/src/drunc/process_manager/ssh_process_manager.py b/src/drunc/process_manager/ssh_process_manager.py
index c3370cda5..29b7eea8a 100644
--- a/src/drunc/process_manager/ssh_process_manager.py
+++ b/src/drunc/process_manager/ssh_process_manager.py
@@ -1,4 +1,5 @@
 import getpass
+import threading
 import uuid
 from typing import List, Optional
 
@@ -25,6 +26,8 @@ class SSHProcessManager(ProcessManager):
     def __init__(
         self, configuration, LifetimeManagerClass: ProcessLifetimeManager, **kwargs
     ):
+        # Used to prevent races between process exit callbacks and ps/kill/flush queries
+        self.boot_request_lock = threading.Lock()
         self.ssh_lifetime_manager: Optional[ProcessLifetimeManager] = None
         self.session = getpass.getuser()  # unfortunate
 
@@ -103,22 +106,10 @@ def _get_process_timeouts(self, uuids: List[str]) -> dict[str, float]:
     def _on_ssh_process_exit(
         self, uuid: str, exit_code: Optional[int], exception: Optional[Exception]
     ) -> None:
-        """
-        Callback invoked when an SSH process exits.
-
-        Args:
-            uuid: Process UUID that exited
-            exit_code: Exit code from process (None if still running)
-            exception: Exception if process failed abnormally
-        """
         if uuid not in self.boot_request:
             return
 
         if exception is not None:
-            # TODO disabled error logging so the integration tests pass
-            # self.log.error(
-            #     f"Process with UUID {uuid} threw an exception when we tried to kill it: {exception!s}"
-            # )
             self.log.debug(
                 f"Process with UUID {uuid} threw an exception when we tried to kill it: {exception!s}"
             )
@@ -131,6 +122,11 @@ def _on_ssh_process_exit(
         else:
             self.log.debug(f"Process with UUID {uuid} exited with code {exit_code}.")
 
+        if uuid not in self.archived_exit_codes:
+            self.archived_exit_codes[uuid] = exit_code
+        if uuid not in self.expected_dead_applications:
+            self.add_process_to_expected_dead_processes(uuid)
+
         boot_req = self.boot_request[uuid]
         name = boot_req.process_description.metadata.name
         session = boot_req.process_description.metadata.session
@@ -408,61 +404,62 @@ def _ps_impl(self, query: ProcessQuery) -> ProcessInstanceList:
         Returns:
             ProcessInstanceList containing status information for matching processes
         """
-        ret = []
+        with self.boot_request_lock:
+            ret = []
 
-        process_uuids = ProcessManager._match_processes_against_query(
-            query=query,
-            available_uuids=list(self._get_active_process_keys()),
-            boot_request_dict=self.boot_request,
-            order_by="random",
-        )
+            process_uuids = ProcessManager._match_processes_against_query(
+                query=query,
+                available_uuids=list(self._get_active_process_keys()),
+                boot_request_dict=self.boot_request,
+                order_by="random",
+            )
 
-        # Iterate through all processes matching the query
-        for proc_uuid in process_uuids:
-            # Handle case where process UUID exists in boot_request but not in SSH manager
-            # This can occur if process failed to start or has been cleaned up
-            if proc_uuid not in self.boot_request:
-                pu = ProcessUUID(uuid=proc_uuid)
-                pi = ProcessInstance(
-                    process_description=ProcessDescription(),
-                    process_restriction=ProcessRestriction(),
-                    status_code=ProcessInstance.StatusCode.DEAD,
-                    return_code=None,
-                    uuid=pu,
+            # Iterate through all processes matching the query
+            for proc_uuid in process_uuids:
+                # Handle case where process UUID exists in boot_request but not in SSH manager
+                # This can occur if process failed to start or has been cleaned up
+                if proc_uuid not in self.boot_request:
+                    pu = ProcessUUID(uuid=proc_uuid)
+                    pi = ProcessInstance(
+                        process_description=ProcessDescription(),
+                        process_restriction=ProcessRestriction(),
+                        status_code=ProcessInstance.StatusCode.DEAD,
+                        return_code=None,
+                        uuid=pu,
+                    )
+                    ret += [pi]
+                    continue
+
+                # Query SSH manager for current process status
+                alive = self.ssh_lifetime_manager.is_process_alive(proc_uuid)
+
+                # Retrieve archived exit code if process is dead
+                return_code = (
+                    self.archived_exit_codes.get(proc_uuid, None) if not alive else None
                 )
-                ret += [pi]
-                continue
-
-            # Query SSH manager for current process status
-            alive = self.ssh_lifetime_manager.is_process_alive(proc_uuid)
-
-            # Retrieve archived exit code if process is dead
-            return_code = (
-                self.archived_exit_codes.get(proc_uuid, None) if not alive else None
-            )
-            if not alive:
-                self.log.debug(
-                    f"Process {proc_uuid} is dead with exit code: {return_code}"
+                if not alive:
+                    self.log.debug(
+                        f"Process {proc_uuid} is dead with exit code: {return_code}"
+                    )
+
+                # Build ProcessInstance with current status
+                pi = self._build_process_instance(
+                    uuid=proc_uuid,
+                    status_code=(
+                        ProcessInstance.StatusCode.RUNNING
+                        if alive
+                        else ProcessInstance.StatusCode.DEAD
+                    ),
+                    return_code=return_code,
                 )
+                ret += [pi]
 
-            # Build ProcessInstance with current status
-            pi = self._build_process_instance(
-                uuid=proc_uuid,
-                status_code=(
-                    ProcessInstance.StatusCode.RUNNING
-                    if alive
-                    else ProcessInstance.StatusCode.DEAD
-                ),
-                return_code=return_code,
+            return ProcessInstanceList(
+                name=self.name,
+                token=None,
+                values=ret,
+                flag=ResponseFlag.EXECUTED_SUCCESSFULLY,
             )
-            ret += [pi]
-
-        return ProcessInstanceList(
-            name=self.name,
-            token=None,
-            values=ret,
-            flag=ResponseFlag.EXECUTED_SUCCESSFULLY,
-        )
 
     def _boot_impl(self, boot_request: BootRequest) -> ProcessInstanceList:
         self.log.debug(f"{self.name} running boot command")
@@ -545,3 +542,85 @@ def _kill_impl(self, query: ProcessQuery) -> ProcessInstanceList:
             values=[],
             flag=ResponseFlag.EXECUTED_SUCCESSFULLY,
         )
+
+    def _flush_impl(self, query: ProcessQuery) -> ProcessInstanceList:
+        """Remove dead processes from tracking so they no longer appear in ps.
+
+        Matches processes against the query, checks each for liveness via the
+        SSH lifetime manager, and removes any dead ones from boot_request and
+        archived_exit_codes. Only dead processes are affected by this command.
+
+        Args:
+            query: ProcessQuery specifying which processes to consider for flushing.
+
+        Returns:
+            ProcessInstanceList containing the ProcessInstance objects that were
+            successfully flushed (i.e. removed from internal tracking).
+        """
+        self.log.info(f"{self.name} flushing dead processes matching {query.names}")
+
+        with self.boot_request_lock:
+            candidate_uuids = ProcessManager._match_processes_against_query(
+                query=query,
+                available_uuids=list(self.boot_request.keys()),
+                boot_request_dict=self.boot_request,
+                order_by="random",
+            )
+
+        # Perform liveness checks outside the lock — these may involve SSH calls
+        # and must not block the publish thread for extended periods.
+        dead_uuids = []
+        for proc_uuid in candidate_uuids:
+            if not self.ssh_lifetime_manager.is_process_alive(proc_uuid):
+                dead_uuids.append(proc_uuid)
+            else:
+                self.log.debug(
+                    f"Process {proc_uuid} is still running — skipping flush."
+                )
+
+        flushed = []
+
+        # Perform all mutations to boot_request under the lock so ps command always sees a
+        # consistent boot_request
+        with self.boot_request_lock:
+            for proc_uuid in dead_uuids:
+                # Guard against the process having been removed between the
+                # liveness check above and acquiring the lock here.
+                if proc_uuid not in self.boot_request:
+                    self.log.debug(
+                        f"Process {proc_uuid} was already removed before flush lock acquired — skipping."
+                    )
+                    continue
+
+                return_code = self.archived_exit_codes.pop(proc_uuid, None)
+
+                pi = self._build_process_instance(
+                    uuid=proc_uuid,
+                    status_code=ProcessInstance.StatusCode.DEAD,
+                    return_code=return_code,
+                )
+
+                del self.boot_request[proc_uuid]
+                # Clean data associated with the process from the lifetime manager
+                self.ssh_lifetime_manager.kill_process(
+                    proc_uuid, self.configuration.data.kill_timeout
+                )
+
+                self.log.info(
+                    f"Flushed dead process {proc_uuid} "
+                    f"(name: {pi.process_description.metadata.name}, "
+                    f"exit code: {return_code})."
+                )
+                flushed.append(pi)
+
+        for pi in flushed:
+            proc_uuid = pi.uuid.uuid
+            if proc_uuid in self.expected_dead_applications:
+                self.remove_process_from_expected_dead_processes(proc_uuid)
+
+        return ProcessInstanceList(
+            name=self.name,
+            token=None,
+            values=flushed,
+            flag=ResponseFlag.EXECUTED_SUCCESSFULLY,
+        )
diff --git a/src/drunc/processes/ssh_process_lifetime_manager_shell.py b/src/drunc/processes/ssh_process_lifetime_manager_shell.py
index e8de4da4d..f2cf44e74 100644
--- a/src/drunc/processes/ssh_process_lifetime_manager_shell.py
+++ b/src/drunc/processes/ssh_process_lifetime_manager_shell.py
@@ -78,6 +78,19 @@ def run(self):
                 with self.manager.lock:
                     self.manager.metadata[self.uuid] = metadata
                 self.logger.debug(f"Metadata retrieved for process {self.uuid}")
+
+                # Log the terminal commands used to manually SIGKILL this process
+                # from outside the process manager which can be useful for debugging
+                # unexpected process deaths
+                if metadata.pid is not None:
+                    self.logger.debug(
+                        f"To manually kill remote process '{metadata.name}' (UUID: {self.uuid}), run: "
+                        f"ssh {self.user}@{self.hostname} kill -9 {metadata.pid}"
+                    )
+                self.logger.debug(
+                    f"To manually kill the local SSH client for '{metadata.name}' (UUID: {self.uuid}), run: "
+                    f"kill -9 {self.process.pid}"
+                )
             else:
                 # If metadata could not be read, fall back to monitoring SSH client
                 self.logger.warning(

From d0955e77b4cca2869ee28da14bc93bac4b456276 Mon Sep 17 00:00:00 2001
From: Aurash Karimi <a.karimi@imperial.ac.uk>
Date: Tue, 17 Mar 2026 17:14:55 +0000
Subject: [PATCH 02/10] add flush endpoint unit test

---
 .../process_manager_mock_impls.py             |  3 ++
 .../test_process_manager_endpoints.py         | 33 +++++++++++++++++++
 2 files changed, 36 insertions(+)

diff --git a/tests/process_manager/process_manager_mock_impls.py b/tests/process_manager/process_manager_mock_impls.py
index c45159269..b9b84cf23 100644
--- a/tests/process_manager/process_manager_mock_impls.py
+++ b/tests/process_manager/process_manager_mock_impls.py
@@ -99,3 +99,6 @@ def _logs_impl(self, log_request: LogRequest) -> LogLines:
             lines=[],
             flag=ResponseFlag.NOT_EXECUTED_NOT_IMPLEMENTED,
         )
+
+    def _flush_impl(self, query: ProcessQuery) -> ProcessInstanceList:
+        return self._not_implemented_response()
diff --git a/tests/process_manager/test_process_manager_endpoints.py b/tests/process_manager/test_process_manager_endpoints.py
index e81266e85..5848bdfc7 100644
--- a/tests/process_manager/test_process_manager_endpoints.py
+++ b/tests/process_manager/test_process_manager_endpoints.py
@@ -277,6 +277,39 @@ def test_ps_endpoint(grpc_test_server_factory, process_query_request, ps_respons
     assert expected_response == response
 
 
+def test_flush_endpoint(
+    grpc_test_server_factory, process_query_request, flush_response
+):
+    """
+    Test that invoking the flush method gives the expected response.
+
+    Validates that the flush endpoint correctly processes ProcessQuery requests
+    and returns the expected ProcessInstanceList response format.
+    """
+    grpc_test_server, expected_response = grpc_test_server_factory(
+        "flush", flush_response
+    )
+
+    # Invoke the flush method via gRPC testing framework
+    flush_method = grpc_test_server.invoke_unary_unary(
+        method_descriptor=(
+            DESCRIPTOR.services_by_name["ProcessManager"].methods_by_name["flush"]
+        ),
+        invocation_metadata={},
+        request=process_query_request,
+        timeout=1,
+    )
+
+    # Block until response is ready and extract all response components
+    response, metadata, code, details = flush_method.termination()
+
+    # Verify the RPC completed successfully without errors
+    assert code == grpc.StatusCode.OK
+
+    # Verify all response fields match expected values
+    assert expected_response == response
+
+
 def test_logs_endpoint(grpc_test_server_factory, log_request, logs_response):
     """
     Test that invoking the logs method gives the expected response.

From 5bf50f733c1b8a480136ee8db3001457ed68cf01 Mon Sep 17 00:00:00 2001
From: Aurash Karimi <a.karimi@imperial.ac.uk>
Date: Fri, 20 Mar 2026 12:51:44 +0000
Subject: [PATCH 03/10] fix flushing behaviour of kill command

---
 .../process_manager/ssh_process_manager.py    | 47 ++++++++++---------
 .../ssh_process_lifetime_manager_shell.py     |  4 +-
 2 files changed, 28 insertions(+), 23 deletions(-)

diff --git a/src/drunc/process_manager/ssh_process_manager.py b/src/drunc/process_manager/ssh_process_manager.py
index 29b7eea8a..4a467a45f 100644
--- a/src/drunc/process_manager/ssh_process_manager.py
+++ b/src/drunc/process_manager/ssh_process_manager.py
@@ -407,17 +407,26 @@ def _ps_impl(self, query: ProcessQuery) -> ProcessInstanceList:
         with self.boot_request_lock:
             ret = []
 
+            # check from alive/active processes in the lifetime manager and from archived exit codes for dead processes
+            available_uuids = (
+                list(self._get_active_process_keys())
+                + list(self.archived_exit_codes.keys())
+                if hasattr(self, "archived_exit_codes")
+                else []
+            )
+
             process_uuids = ProcessManager._match_processes_against_query(
                 query=query,
-                available_uuids=list(self._get_active_process_keys()),
+                available_uuids=available_uuids,
                 boot_request_dict=self.boot_request,
                 order_by="random",
             )
 
             # Iterate through all processes matching the query
             for proc_uuid in process_uuids:
-                # Handle case where process UUID exists in boot_request but not in SSH manager
-                # This can occur if process failed to start or has been cleaned up
+                # Handle case where process UUID does not exist in the boot_request but is active in SSH manager
+                # This can occur if process has been cleaned up in the process manager but is still alive in the
+                # lifetime manager
                 if proc_uuid not in self.boot_request:
                     pu = ProcessUUID(uuid=proc_uuid)
                     pi = ProcessInstance(
@@ -430,28 +439,21 @@ def _ps_impl(self, query: ProcessQuery) -> ProcessInstanceList:
                     ret += [pi]
                     continue
 
-                # Query SSH manager for current process status
-                alive = self.ssh_lifetime_manager.is_process_alive(proc_uuid)
+                exit_code = self.archived_exit_codes.get(proc_uuid, None)
 
-                # Retrieve archived exit code if process is dead
-                return_code = (
-                    self.archived_exit_codes.get(proc_uuid, None) if not alive else None
-                )
-                if not alive:
-                    self.log.debug(
-                        f"Process {proc_uuid} is dead with exit code: {return_code}"
+                if exit_code is not None:
+                    pi = self._build_process_instance(
+                        uuid=proc_uuid,
+                        status_code=ProcessInstance.StatusCode.DEAD,
+                        return_code=exit_code,
+                    )
+                else:
+                    pi = self._build_process_instance(
+                        uuid=proc_uuid,
+                        status_code=ProcessInstance.StatusCode.RUNNING,
+                        return_code=None,
                     )
 
-                # Build ProcessInstance with current status
-                pi = self._build_process_instance(
-                    uuid=proc_uuid,
-                    status_code=(
-                        ProcessInstance.StatusCode.RUNNING
-                        if alive
-                        else ProcessInstance.StatusCode.DEAD
-                    ),
-                    return_code=return_code,
-                )
                 ret += [pi]
 
             return ProcessInstanceList(
@@ -501,6 +503,7 @@ def _restart_impl(self, query: ProcessQuery) -> ProcessInstanceList:
         # Remove the application from the list of dead applications
         self.remove_process_from_expected_dead_processes(uuid)
 
+        del self.archived_exit_codes[uuid]
         del uuid
         del same_uuid_br
         del same_uuid
diff --git a/src/drunc/processes/ssh_process_lifetime_manager_shell.py b/src/drunc/processes/ssh_process_lifetime_manager_shell.py
index f2cf44e74..4606c6e29 100644
--- a/src/drunc/processes/ssh_process_lifetime_manager_shell.py
+++ b/src/drunc/processes/ssh_process_lifetime_manager_shell.py
@@ -975,7 +975,7 @@ def kill_process(
         self,
         uuid: str,
         timeout: float = ProcessLifetimeManager.DEFAULT_TIMEOUT_FOR_KILLING_PROCESS,
-    ) -> Optional[int]:
+    ) -> int | None:
         """
         Kill a remote process and clean up all associated resources.
 
@@ -1065,6 +1065,8 @@ def kill_process(
             self.log.error(f"Error terminating remote process {uuid}: {e}")
             return None
 
+        return None
+
     def _cleanup_process_resources(self, uuid: str) -> None:
         """Remove all resources associated with a process UUID."""
         with self.lock:

From 2a455c988cf16ac7c4c96b83ca257948ed2d90f0 Mon Sep 17 00:00:00 2001
From: Aurash Karimi <a.karimi@imperial.ac.uk>
Date: Fri, 20 Mar 2026 13:22:11 +0000
Subject: [PATCH 04/10] fix unexpectedly killed process behaviour

---
 src/drunc/process_manager/ssh_process_manager.py | 11 ++++++++---
 1 file changed, 8 insertions(+), 3 deletions(-)

diff --git a/src/drunc/process_manager/ssh_process_manager.py b/src/drunc/process_manager/ssh_process_manager.py
index 4a467a45f..99c17da04 100644
--- a/src/drunc/process_manager/ssh_process_manager.py
+++ b/src/drunc/process_manager/ssh_process_manager.py
@@ -408,12 +408,17 @@ def _ps_impl(self, query: ProcessQuery) -> ProcessInstanceList:
             ret = []
 
             # check from alive/active processes in the lifetime manager and from archived exit codes for dead processes
-            available_uuids = (
-                list(self._get_active_process_keys())
-                + list(self.archived_exit_codes.keys())
+            dead_processes = (
+                list(self.archived_exit_codes.keys())
                 if hasattr(self, "archived_exit_codes")
                 else []
             )
+            alive_processes = [
+                uuid
+                for uuid in self._get_active_process_keys()
+                if self.ssh_lifetime_manager.is_process_alive(uuid) == True
+            ]
+            available_uuids = alive_processes + dead_processes
 
             process_uuids = ProcessManager._match_processes_against_query(
                 query=query,

From 1f4119a00ce82b8adf890cfb6106d708e11a7137 Mon Sep 17 00:00:00 2001
From: Aurash Karimi <a.karimi@imperial.ac.uk>
Date: Fri, 20 Mar 2026 14:35:13 +0000
Subject: [PATCH 05/10] patch ps_impl to not break publish() method in PM which
 working out intended behaviour

---
 .../process_manager/ssh_process_manager.py    | 39 +++++++++++++------
 1 file changed, 28 insertions(+), 11 deletions(-)

diff --git a/src/drunc/process_manager/ssh_process_manager.py b/src/drunc/process_manager/ssh_process_manager.py
index 99c17da04..f74dbc6e1 100644
--- a/src/drunc/process_manager/ssh_process_manager.py
+++ b/src/drunc/process_manager/ssh_process_manager.py
@@ -398,6 +398,20 @@ def _ps_impl(self, query: ProcessQuery) -> ProcessInstanceList:
         Returns process details including running status, exit codes, and metadata
         for all processes that match the provided query criteria.
 
+        Args:
+            query: ProcessQuery object containing process selection criteria
+
+        Returns:
+            ProcessInstanceList containing status information for matching processes
+        """
+        return self.__ps_impl(query=query, show_dead=False)
+
+    def __ps_impl(self, query: ProcessQuery, show_dead: bool) -> ProcessInstanceList:
+        """
+        Retrieve process status information for processes matching the query.
+        Returns process details including running status, exit codes, and metadata
+        for all processes that match the provided query criteria.
+
         Args:
             query: ProcessQuery object containing process selection criteria
 
@@ -408,17 +422,20 @@ def _ps_impl(self, query: ProcessQuery) -> ProcessInstanceList:
             ret = []
 
             # check from alive/active processes in the lifetime manager and from archived exit codes for dead processes
-            dead_processes = (
-                list(self.archived_exit_codes.keys())
-                if hasattr(self, "archived_exit_codes")
-                else []
-            )
-            alive_processes = [
-                uuid
-                for uuid in self._get_active_process_keys()
-                if self.ssh_lifetime_manager.is_process_alive(uuid) == True
-            ]
-            available_uuids = alive_processes + dead_processes
+            if show_dead:
+                dead_processes = (
+                    list(self.archived_exit_codes.keys())
+                    if hasattr(self, "archived_exit_codes")
+                    else []
+                )
+                alive_processes = [
+                    uuid
+                    for uuid in self._get_active_process_keys()
+                    if self.ssh_lifetime_manager.is_process_alive(uuid) == True
+                ]
+                available_uuids = alive_processes + dead_processes
+            else:
+                available_uuids = self._get_active_process_keys()
 
             process_uuids = ProcessManager._match_processes_against_query(
                 query=query,

From 044adf919f00db4244c8050b009bcd8739ed3081 Mon Sep 17 00:00:00 2001
From: Aurash Karimi <a.karimi@imperial.ac.uk>
Date: Mon, 23 Mar 2026 12:29:45 +0000
Subject: [PATCH 06/10] simplify ps command

---
 .../process_manager/ssh_process_manager.py    | 31 ++-----------------
 1 file changed, 2 insertions(+), 29 deletions(-)

diff --git a/src/drunc/process_manager/ssh_process_manager.py b/src/drunc/process_manager/ssh_process_manager.py
index f74dbc6e1..667839682 100644
--- a/src/drunc/process_manager/ssh_process_manager.py
+++ b/src/drunc/process_manager/ssh_process_manager.py
@@ -398,20 +398,6 @@ def _ps_impl(self, query: ProcessQuery) -> ProcessInstanceList:
         Returns process details including running status, exit codes, and metadata
         for all processes that match the provided query criteria.
 
-        Args:
-            query: ProcessQuery object containing process selection criteria
-
-        Returns:
-            ProcessInstanceList containing status information for matching processes
-        """
-        return self.__ps_impl(query=query, show_dead=False)
-
-    def __ps_impl(self, query: ProcessQuery, show_dead: bool) -> ProcessInstanceList:
-        """
-        Retrieve process status information for processes matching the query.
-        Returns process details including running status, exit codes, and metadata
-        for all processes that match the provided query criteria.
-
         Args:
             query: ProcessQuery object containing process selection criteria
 
@@ -421,21 +407,8 @@ def __ps_impl(self, query: ProcessQuery, show_dead: bool) -> ProcessInstanceList
         with self.boot_request_lock:
             ret = []
 
-            # check from alive/active processes in the lifetime manager and from archived exit codes for dead processes
-            if show_dead:
-                dead_processes = (
-                    list(self.archived_exit_codes.keys())
-                    if hasattr(self, "archived_exit_codes")
-                    else []
-                )
-                alive_processes = [
-                    uuid
-                    for uuid in self._get_active_process_keys()
-                    if self.ssh_lifetime_manager.is_process_alive(uuid) == True
-                ]
-                available_uuids = alive_processes + dead_processes
-            else:
-                available_uuids = self._get_active_process_keys()
+            # Check through all processes that the lifetime manager knows about
+            available_uuids = self._get_active_process_keys()
 
             process_uuids = ProcessManager._match_processes_against_query(
                 query=query,

From 37620b030dd5b23866ebbe8addd3c69b3607a97d Mon Sep 17 00:00:00 2001
From: Aurash Karimi <a.karimi@imperial.ac.uk>
Date: Mon, 23 Mar 2026 13:19:18 +0000
Subject: [PATCH 07/10] add --crash option to kill command

---
 .../process_manager/interface/cli_argument.py |  8 ++-
 .../process_manager/ssh_process_manager.py    | 49 ++++++++++++++++++-
 src/drunc/process_manager/utils.py            |  1 +
 .../processes/ssh_process_lifetime_manager.py | 16 ++++++
 ...ss_lifetime_manager_from_forked_process.py | 13 +++++
 .../ssh_process_lifetime_manager_shell.py     | 35 +++++++++++++
 6 files changed, 119 insertions(+), 3 deletions(-)

diff --git a/src/drunc/process_manager/interface/cli_argument.py b/src/drunc/process_manager/interface/cli_argument.py
index 27e84872f..2612e591c 100644
--- a/src/drunc/process_manager/interface/cli_argument.py
+++ b/src/drunc/process_manager/interface/cli_argument.py
@@ -38,6 +38,12 @@ def wrapper(f0):
             multiple=True,
             help="Select the process of a particular UUIDs",
         )(f3)
-        return generate_process_query(f4, at_least_one, all_processes_by_default)
+        f5 = click.option(
+            "--crash",
+            is_flag=True,
+            default=False,
+            help="Simulate a crash: send SIGKILL without any cleanup, leaving the process manager in an unexpected-death state.",
+        )(f4)
+        return generate_process_query(f5, at_least_one, all_processes_by_default)
 
     return wrapper
diff --git a/src/drunc/process_manager/ssh_process_manager.py b/src/drunc/process_manager/ssh_process_manager.py
index 667839682..42cdac531 100644
--- a/src/drunc/process_manager/ssh_process_manager.py
+++ b/src/drunc/process_manager/ssh_process_manager.py
@@ -515,12 +515,15 @@ def _kill_impl(self, query: ProcessQuery) -> ProcessInstanceList:
         Kill processes matching the query.
 
         Terminates all processes that match the provided query criteria.
+        If query.crash is True, sends SIGKILL without any cleanup to simulate
+        an unexpected process crash.
 
         Args:
-            query: ProcessQuery object containing process selection criteria
+            query: ProcessQuery object containing process selection criteria.
+                   Set query.crash=True to simulate a crash instead of a clean kill.
 
         Returns:
-            ProcessInstanceList containing status of killed processes
+            ProcessInstanceList containing status of killed/crashed processes
         """
         self.log.info(f"{self.name} killing {query.names} in session {self.session}")
 
@@ -531,6 +534,10 @@ def _kill_impl(self, query: ProcessQuery) -> ProcessInstanceList:
                 boot_request_dict=self.boot_request,
                 order_by="leaf_first",
             )
+
+            if query.crash:
+                return self._crash_processes(uuids)
+
             return self.kill_processes(uuids)
 
         self.log.info("No known process to kill")
@@ -541,6 +548,44 @@ def _kill_impl(self, query: ProcessQuery) -> ProcessInstanceList:
             flag=ResponseFlag.EXECUTED_SUCCESSFULLY,
         )
 
+    def _crash_processes(self, uuids: list) -> ProcessInstanceList:
+        """
+        Simulate crashes for processes identified by their UUIDs.
+
+        Sends SIGKILL to each process via the lifetime manager's crash_process
+        method without performing any cleanup. This deliberately avoids marking
+        processes as expected-dead so that the subsequent unexpected process
+        deaths trigger crash-recovery handling.
+
+        Args:
+            uuids: List of process UUIDs to crash
+
+        Returns:
+            ProcessInstanceList containing the ProcessInstances for each
+            crashed process with DEAD status and no return code.
+        """
+        for this_uuid in uuids:
+            self.log.info(
+                f"Simulating crash of process {this_uuid} (sending SIGKILL, no cleanup)."
+            )
+            self.ssh_lifetime_manager.crash_process(this_uuid)
+
+        ret = [
+            self._build_process_instance(
+                uuid=uuid,
+                status_code=ProcessInstance.StatusCode.DEAD,
+                return_code=None,
+            )
+            for uuid in uuids
+        ]
+
+        return ProcessInstanceList(
+            name=self.name,
+            token=None,
+            values=ret,
+            flag=ResponseFlag.EXECUTED_SUCCESSFULLY,
+        )
+
     def _flush_impl(self, query: ProcessQuery) -> ProcessInstanceList:
         """Remove dead processes from tracking so they no longer appear in ps.
 
diff --git a/src/drunc/process_manager/utils.py b/src/drunc/process_manager/utils.py
index 82c4720fa..243b6914f 100644
--- a/src/drunc/process_manager/utils.py
+++ b/src/drunc/process_manager/utils.py
@@ -49,6 +49,7 @@ def new_func(ctx, session, name, user, uuid, **kwargs):
             names=name,
             user=user,
             uuids=uuids,
+            crash=kwargs.pop("crash", False),
         )
         # print(query)
         return ctx.invoke(f, query=query, **kwargs)
diff --git a/src/drunc/processes/ssh_process_lifetime_manager.py b/src/drunc/processes/ssh_process_lifetime_manager.py
index 7b849dcd4..1932aad58 100644
--- a/src/drunc/processes/ssh_process_lifetime_manager.py
+++ b/src/drunc/processes/ssh_process_lifetime_manager.py
@@ -133,6 +133,22 @@ def kill_process(
         """
         pass
 
+    @abstractmethod
+    def crash_process(self, uuid: str) -> None:
+        """
+        Simulate a process crash by sending SIGKILL without performing any cleanup.
+
+        Unlike kill_process, this method only sends the kill signal to the remote
+        process without waiting for termination or cleaning up associated resources
+        (metadata files, internal tracking structures, etc.). This is intended for
+        testing failure scenarios where the process manager should observe an
+        unexpected process death.
+
+        Args:
+            uuid: Process UUID to crash
+        """
+        pass
+
     @abstractmethod
     def kill_processes(
         self, uuids: List[str], process_timeouts: Optional[Dict[str, float]] = None
diff --git a/src/drunc/processes/ssh_process_lifetime_manager_from_forked_process.py b/src/drunc/processes/ssh_process_lifetime_manager_from_forked_process.py
index 32633a381..760cc10e5 100644
--- a/src/drunc/processes/ssh_process_lifetime_manager_from_forked_process.py
+++ b/src/drunc/processes/ssh_process_lifetime_manager_from_forked_process.py
@@ -525,6 +525,19 @@ def kill_process(
         """
         return self._call("kill_process", uuid, timeout)
 
+    def crash_process(self, uuid: str) -> None:
+        """
+        Simulate a process crash by sending SIGKILL without performing any cleanup.
+
+        Delegates to the underlying SSHProcessLifetimeManagerShell running in
+        the forked worker process. Sends SIGKILL to the remote process without
+        cleaning up any associated resources, simulating an unexpected crash.
+
+        Args:
+            uuid: Process UUID to crash
+        """
+        self._call("crash_process", uuid)
+
     def kill_processes(
         self,
         uuids: List[str],
diff --git a/src/drunc/processes/ssh_process_lifetime_manager_shell.py b/src/drunc/processes/ssh_process_lifetime_manager_shell.py
index 4606c6e29..03613f177 100644
--- a/src/drunc/processes/ssh_process_lifetime_manager_shell.py
+++ b/src/drunc/processes/ssh_process_lifetime_manager_shell.py
@@ -1067,6 +1067,41 @@ def kill_process(
 
         return None
 
+    def crash_process(self, uuid: str) -> None:
+        """
+        Simulate a process crash by sending SIGKILL without performing any cleanup.
+
+        Sends SIGKILL to the remote process identified by uuid but deliberately
+        skips all cleanup steps (metadata file removal, internal tracking cleanup,
+        SSH client termination). This leaves the process manager in the same state
+        as if the process had crashed unexpectedly, allowing crash-recovery logic
+        to be exercised in tests.
+
+        Args:
+            uuid: Process UUID to crash
+        """
+        if uuid not in self.process_store:
+            self.log.warning(f"crash_process called for unknown UUID {uuid}")
+            return
+
+        process_info = self.process_store[uuid]
+        hostname = process_info["hostname"]
+        user = process_info["user"]
+
+        metadata = self.metadata.get(uuid, None)
+        if metadata is None or metadata.pid is None:
+            self.log.warning(
+                f"No remote PID for {uuid}, cannot send SIGKILL to simulate crash."
+            )
+            return
+
+        remote_pid = metadata.pid
+        self.log.debug(
+            f"Simulating crash of process {uuid} (PID {remote_pid}): "
+            f"sending SIGKILL without cleanup."
+        )
+        self._send_remote_signal(hostname, user, remote_pid, "KILL")
+
     def _cleanup_process_resources(self, uuid: str) -> None:
         """Remove all resources associated with a process UUID."""
         with self.lock:

From 9e6bfa80cce907f1a20ca12f3dac4b3f2bb0966b Mon Sep 17 00:00:00 2001
From: Aurash Karimi <a.karimi@imperial.ac.uk>
Date: Mon, 23 Mar 2026 14:14:14 +0000
Subject: [PATCH 08/10] add remote process pids to long info ps command

---
 .../process_manager/ssh_process_manager.py    |  12 ++
 src/drunc/process_manager/utils.py            |  17 ++-
 .../processes/ssh_process_lifetime_manager.py |  32 ++++
 ...ss_lifetime_manager_from_forked_process.py |  22 ++-
 .../ssh_process_lifetime_manager_shell.py     |  23 ++-
 tests/process_manager/test_utils.py           | 140 ++++++++++++++++++
 6 files changed, 242 insertions(+), 4 deletions(-)

diff --git a/src/drunc/process_manager/ssh_process_manager.py b/src/drunc/process_manager/ssh_process_manager.py
index 42cdac531..14af9c86b 100644
--- a/src/drunc/process_manager/ssh_process_manager.py
+++ b/src/drunc/process_manager/ssh_process_manager.py
@@ -431,6 +431,13 @@ def _ps_impl(self, query: ProcessQuery) -> ProcessInstanceList:
                         return_code=None,
                         uuid=pu,
                     )
+                    remote_pid_result = self.ssh_lifetime_manager.get_remote_pid(
+                        proc_uuid
+                    )
+                    if remote_pid_result.successful:
+                        pi.remote_pid = str(remote_pid_result.pid)
+                    else:
+                        pi.remote_pid = remote_pid_result.reason
                     ret += [pi]
                     continue
 
@@ -449,6 +456,11 @@ def _ps_impl(self, query: ProcessQuery) -> ProcessInstanceList:
                         return_code=None,
                     )
 
+                remote_pid_result = self.ssh_lifetime_manager.get_remote_pid(proc_uuid)
+                if remote_pid_result.successful:
+                    pi.remote_pid = str(remote_pid_result.pid)
+                else:
+                    pi.remote_pid = remote_pid_result.reason
                 ret += [pi]
 
             return ProcessInstanceList(
diff --git a/src/drunc/process_manager/utils.py b/src/drunc/process_manager/utils.py
index 243b6914f..896f261f2 100644
--- a/src/drunc/process_manager/utils.py
+++ b/src/drunc/process_manager/utils.py
@@ -128,10 +128,17 @@ def tabulate_process_instance_list(
     t.add_column("uuid")
     t.add_column("alive")
     t.add_column("exit-code")
+
+    sorted_pil = order_process_by_name(pil.values)
+
+    show_remote_pid = long and any(
+        process.HasField("remote_pid") for process in sorted_pil
+    )
+    if show_remote_pid:
+        t.add_column("remote-pid")
     if long:
         t.add_column("executable")
 
-    sorted_pil = order_process_by_name(pil.values)
     tree_str = make_tree(sorted_pil)
     try:
         for process, line in zip(sorted_pil, tree_str):
@@ -142,12 +149,18 @@ def tabulate_process_instance_list(
                 else "[danger]False[/danger]"
             )
             row = [m.session, line, m.user, m.hostname, process.uuid.uuid]
+            row += [alive, f"{process.return_code}"]
+            if show_remote_pid:
+                row += [
+                    process.remote_pid
+                    if process.HasField("remote_pid")
+                    else "no metadata"
+                ]
             if long:
                 executables = [
                     e.exec for e in process.process_description.executable_and_arguments
                 ]
                 row += ["; ".join(executables)]
-            row += [alive, f"{process.return_code}"]
             t.add_row(*row)
     except TypeError:
         raise DruncCommandException(
diff --git a/src/drunc/processes/ssh_process_lifetime_manager.py b/src/drunc/processes/ssh_process_lifetime_manager.py
index 1932aad58..a61bcd0a5 100644
--- a/src/drunc/processes/ssh_process_lifetime_manager.py
+++ b/src/drunc/processes/ssh_process_lifetime_manager.py
@@ -6,6 +6,7 @@
 """
 
 from abc import ABC, abstractmethod
+from dataclasses import dataclass
 from typing import Any, Dict, List, Optional
 
 from druncschema.process_manager_pb2 import BootRequest
@@ -13,6 +14,22 @@
 from drunc.processes.connection_utils import wait_for
 
 
+@dataclass
+class RemotePidResult:
+    """
+    Result of a remote PID query.
+
+    Either ``pid`` is set (success) or ``reason`` explains why it is unavailable.
+    """
+
+    pid: Optional[int] = None
+    reason: Optional[str] = None
+
+    @property
+    def successful(self) -> bool:
+        return self.pid is not None
+
+
 class ProcessLifetimeManager(ABC):
     """
     Abstract base class for process lifetime management.
@@ -287,3 +304,18 @@ def validate_host_connection(
             RuntimeError: If SSH connection or command execution fails
         """
         pass
+
+    @abstractmethod
+    def get_remote_pid(self, uuid: str) -> "RemotePidResult":
+        """
+        Return the remote PID for the process, if available.
+
+        Args:
+            uuid: Process UUID to query.
+
+        Returns:
+            RemotePidResult with ``pid`` set on success, or ``reason`` describing
+            why the PID is unavailable (e.g. ``"no metadata"`` when the metadata
+            file has not yet been written by the remote shell wrapper).
+        """
+        pass
diff --git a/src/drunc/processes/ssh_process_lifetime_manager_from_forked_process.py b/src/drunc/processes/ssh_process_lifetime_manager_from_forked_process.py
index 760cc10e5..8bce21ff9 100644
--- a/src/drunc/processes/ssh_process_lifetime_manager_from_forked_process.py
+++ b/src/drunc/processes/ssh_process_lifetime_manager_from_forked_process.py
@@ -16,7 +16,10 @@
 
 from druncschema.process_manager_pb2 import BootRequest
 
-from drunc.processes.ssh_process_lifetime_manager import ProcessLifetimeManager
+from drunc.processes.ssh_process_lifetime_manager import (
+    ProcessLifetimeManager,
+    RemotePidResult,
+)
 from drunc.processes.ssh_process_lifetime_manager_shell import (
     SSHProcessLifetimeManagerShell,
 )
@@ -654,3 +657,20 @@ def validate_host_connection(
             RuntimeError: If the SSH connection validation fails.
         """
         self._call("validate_host_connection", host, auth_method, user)
+
+    def get_remote_pid(self, uuid: str) -> RemotePidResult:
+        """
+        Retrieve the remote PID for a managed process via the child process.
+
+        Delegates to the underlying SSHProcessLifetimeManagerShell running in the
+        forked worker process. The returned RemotePidResult dataclass is picklable
+        and transmitted across the process boundary without special handling.
+
+        Args:
+            uuid: Process UUID to query.
+
+        Returns:
+            RemotePidResult with ``pid`` set on success, or ``reason`` explaining
+            why the PID is unavailable (e.g. metadata not yet written).
+        """
+        return self._call("get_remote_pid", uuid)
diff --git a/src/drunc/processes/ssh_process_lifetime_manager_shell.py b/src/drunc/processes/ssh_process_lifetime_manager_shell.py
index 03613f177..f9d28e506 100644
--- a/src/drunc/processes/ssh_process_lifetime_manager_shell.py
+++ b/src/drunc/processes/ssh_process_lifetime_manager_shell.py
@@ -19,7 +19,10 @@
 from drunc.process_manager.utils import on_parent_exit
 from drunc.processes.connection_utils import wait_for
 from drunc.processes.process_metadata import ProcessMetadata
-from drunc.processes.ssh_process_lifetime_manager import ProcessLifetimeManager
+from drunc.processes.ssh_process_lifetime_manager import (
+    ProcessLifetimeManager,
+    RemotePidResult,
+)
 from drunc.utils.utils import get_logger
 
 
@@ -273,6 +276,24 @@ def get_active_process_keys(self) -> List[str]:
         with self.lock:
             return list(self.process_store.keys())
 
+    def get_remote_pid(self, uuid: str) -> RemotePidResult:
+        """
+        Return the remote PID for the process identified by *uuid*.
+
+        Args:
+            uuid: Process UUID to query.
+
+        Returns:
+            RemotePidResult with ``pid`` set on success, or ``reason``
+            set to ``"no metadata"`` when the metadata file has not yet
+            been written or could not be read.
+        """
+        with self.lock:
+            metadata = self.metadata.get(uuid)
+        if metadata is None or metadata.pid is None:
+            return RemotePidResult(reason="no metadata")
+        return RemotePidResult(pid=metadata.pid)
+
     def start_process(self, uuid: str, boot_request: BootRequest) -> None:
         """
         Start a remote process via SSH using the boot request configuration.
diff --git a/tests/process_manager/test_utils.py b/tests/process_manager/test_utils.py
index e69de29bb..828851e4f 100644
--- a/tests/process_manager/test_utils.py
+++ b/tests/process_manager/test_utils.py
@@ -0,0 +1,140 @@
+from druncschema.process_manager_pb2 import (
+    ProcessDescription,
+    ProcessInstance,
+    ProcessInstanceList,
+    ProcessMetadata,
+    ProcessRestriction,
+    ProcessUUID,
+)
+from druncschema.request_response_pb2 import ResponseFlag
+from rich.console import Console
+
+from drunc.process_manager.utils import tabulate_process_instance_list
+
+
+def _make_process_instance(
+    name: str,
+    uuid: str,
+    remote_pid: str | None = None,
+) -> ProcessInstance:
+    pi = ProcessInstance(
+        process_description=ProcessDescription(
+            metadata=ProcessMetadata(
+                session="session-1",
+                name=name,
+                user="user-1",
+                hostname="host-1",
+                tree_id="0",
+            ),
+            executable_and_arguments=[
+                ProcessDescription.ExecAndArgs(exec="/bin/sleep", args=["10"])
+            ],
+        ),
+        process_restriction=ProcessRestriction(),
+        status_code=ProcessInstance.StatusCode.RUNNING,
+        return_code=0,
+        uuid=ProcessUUID(uuid=uuid),
+    )
+    if remote_pid is not None:
+        pi.remote_pid = remote_pid
+    return pi
+
+
+def test_tabulate_long_format_shows_remote_pid_column():
+    """When at least one ProcessInstance has remote_pid set, the remote-pid column should appear."""
+    process_list = ProcessInstanceList(
+        name="pm",
+        values=[
+            _make_process_instance("app-with-pid", "uuid-1", remote_pid="4242"),
+        ],
+        flag=ResponseFlag.EXECUTED_SUCCESSFULLY,
+    )
+
+    table = tabulate_process_instance_list(process_list, title="test", long=True)
+
+    console = Console(record=True, width=200)
+    console.print(table)
+    rendered = console.export_text()
+
+    assert "remote-pid" in rendered
+    assert "4242" in rendered
+
+
+def test_tabulate_long_format_shows_no_metadata_when_pid_is_reason():
+    """When remote_pid contains a reason string it should appear as-is."""
+    process_list = ProcessInstanceList(
+        name="pm",
+        values=[
+            _make_process_instance("app-no-meta", "uuid-2", remote_pid="no metadata"),
+        ],
+        flag=ResponseFlag.EXECUTED_SUCCESSFULLY,
+    )
+
+    table = tabulate_process_instance_list(process_list, title="test", long=True)
+
+    console = Console(record=True, width=200)
+    console.print(table)
+    rendered = console.export_text()
+
+    assert "remote-pid" in rendered
+    assert "no metadata" in rendered
+
+
+def test_tabulate_long_format_no_remote_pid_column_when_field_absent():
+    """When no ProcessInstance has remote_pid set, the remote-pid column must not appear."""
+    process_list = ProcessInstanceList(
+        name="pm",
+        values=[
+            _make_process_instance("app-1", "uuid-1"),
+        ],
+        flag=ResponseFlag.EXECUTED_SUCCESSFULLY,
+    )
+
+    table = tabulate_process_instance_list(process_list, title="test", long=True)
+
+    console = Console(record=True, width=200)
+    console.print(table)
+    rendered = console.export_text()
+
+    assert "remote-pid" not in rendered
+
+
+def test_tabulate_short_format_never_shows_remote_pid_column():
+    """The remote-pid column must not appear when long=False even if remote_pid is set."""
+    process_list = ProcessInstanceList(
+        name="pm",
+        values=[
+            _make_process_instance("app-1", "uuid-1", remote_pid="1234"),
+        ],
+        flag=ResponseFlag.EXECUTED_SUCCESSFULLY,
+    )
+
+    table = tabulate_process_instance_list(process_list, title="test", long=False)
+
+    console = Console(record=True, width=200)
+    console.print(table)
+    rendered = console.export_text()
+
+    assert "remote-pid" not in rendered
+
+
+def test_tabulate_long_format_executable_column_also_shown():
+    """Executable column should still appear alongside remote-pid column."""
+    process_list = ProcessInstanceList(
+        name="pm",
+        values=[
+            _make_process_instance("app-1", "uuid-1", remote_pid="7777"),
+        ],
+        flag=ResponseFlag.EXECUTED_SUCCESSFULLY,
+    )
+
+    table = tabulate_process_instance_list(process_list, title="test", long=True)
+
+    console = Console(record=True, width=200)
+    console.print(table)
+    rendered = console.export_text()
+
+    assert "executable" in rendered
+    assert "/bin/sleep" in rendered
+    assert "remote-pid" in rendered
+    assert "7777" in rendered

From 04b7e15c1894af0f13d15e843323a25353d30446 Mon Sep 17 00:00:00 2001
From: Aurash Karimi <a.karimi@imperial.ac.uk>
Date: Mon, 23 Mar 2026 14:18:04 +0000
Subject: [PATCH 09/10] handle crash not in process query as it is optional

---
 src/drunc/process_manager/ssh_process_manager.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/drunc/process_manager/ssh_process_manager.py b/src/drunc/process_manager/ssh_process_manager.py
index 14af9c86b..bba65e6de 100644
--- a/src/drunc/process_manager/ssh_process_manager.py
+++ b/src/drunc/process_manager/ssh_process_manager.py
@@ -547,7 +547,7 @@ def _kill_impl(self, query: ProcessQuery) -> ProcessInstanceList:
                 order_by="leaf_first",
             )
 
-            if query.crash:
+            if hasattr(query, "crash") and query.crash:
                 return self._crash_processes(uuids)
 
             return self.kill_processes(uuids)

From f3fbc04227197ac0ad6773cd077f8281a4537826 Mon Sep 17 00:00:00 2001
From: Aurash Karimi <a.karimi@imperial.ac.uk>
Date: Mon, 23 Mar 2026 14:35:01 +0000
Subject: [PATCH 10/10] cleanup

---
 src/drunc/process_manager/ssh_process_manager.py | 2 ++
 src/drunc/process_manager/utils.py               | 2 +-
 tests/process_manager/test_utils.py              | 3 +--
 3 files changed, 4 insertions(+), 3 deletions(-)

diff --git a/src/drunc/process_manager/ssh_process_manager.py b/src/drunc/process_manager/ssh_process_manager.py
index bba65e6de..e80d421b7 100644
--- a/src/drunc/process_manager/ssh_process_manager.py
+++ b/src/drunc/process_manager/ssh_process_manager.py
@@ -95,6 +95,7 @@ def _build_process_instance(
             status_code=status_code,
             return_code=return_code,
             uuid=pu,
+            remote_pid="not available",
         )
 
     def _get_process_timeouts(self, uuids: List[str]) -> dict[str, float]:
@@ -430,6 +431,7 @@ def _ps_impl(self, query: ProcessQuery) -> ProcessInstanceList:
                         status_code=ProcessInstance.StatusCode.DEAD,
                         return_code=None,
                         uuid=pu,
+                        remote_pid="not available",
                     )
                     remote_pid_result = self.ssh_lifetime_manager.get_remote_pid(
                         proc_uuid
diff --git a/src/drunc/process_manager/utils.py b/src/drunc/process_manager/utils.py
index 896f261f2..602a2dab2 100644
--- a/src/drunc/process_manager/utils.py
+++ b/src/drunc/process_manager/utils.py
@@ -154,7 +154,7 @@ def tabulate_process_instance_list(
                 row += [
                     process.remote_pid
                     if process.HasField("remote_pid")
-                    else "no metadata"
+                    else "Not available"
                 ]
             if long:
                 executables = [
diff --git a/tests/process_manager/test_utils.py b/tests/process_manager/test_utils.py
index 828851e4f..4bce5ffce 100644
--- a/tests/process_manager/test_utils.py
+++ b/tests/process_manager/test_utils.py
@@ -34,9 +34,8 @@ def _make_process_instance(
         status_code=ProcessInstance.StatusCode.RUNNING,
         return_code=0,
         uuid=ProcessUUID(uuid=uuid),
+        remote_pid=remote_pid if remote_pid is not None else None,
     )
-    if remote_pid is not None:
-        pi.remote_pid = remote_pid
     return pi