feat(companion): Add CompanionProcess runtime state and status helpers

docs/design/companion-process-manager.md

@@ -666,10 +666,10 @@ No per-companion logic in Arbiter.
 
 - [x] Add companion config settings in `gunicorn/config.py`.
 - [x] Add config validation for `companion_workers`.
-- [ ] Add `CompanionConfig` and config hash generation.
-- [ ] Add public process states.
-- [ ] Add `CompanionProcess` runtime state.
-- [ ] Add status description helpers.
+- [x] Add `CompanionConfig` and config hash generation.
+- [x] Add public process states.
+- [x] Add `CompanionProcess` runtime state.
+- [x] Add status description helpers.
 - [ ] Add `CompanionManager` skeleton.
 - [ ] Spawn one companion process from the manager.
 - [ ] Apply `cwd` and `env` before target.

gunicorn/companion/process.py

@@ -5,17 +5,25 @@
 
 import hashlib
 import json
+import time
+from enum import Enum
 
-# Public states, mimicking ``supervisorctl status``. The manager never
-# exposes EXITED/FATAL/UNKNOWN; an exited companion is either STOPPED (manual)
-# or BACKOFF (waiting to restart).
-STOPPED = "STOPPED"
-STARTING = "STARTING"
-RUNNING = "RUNNING"
-BACKOFF = "BACKOFF"
-STOPPING = "STOPPING"
+from gunicorn.util import format_uptime
 
-PUBLIC_STATES = (STOPPED, STARTING, RUNNING, BACKOFF, STOPPING)
+
+class State(str, Enum):
+    """Public states, mimicking ``supervisorctl status``.
+
+    The manager never exposes EXITED/FATAL/UNKNOWN; an exited companion is
+    either STOPPED (manual) or BACKOFF (waiting to restart). Members subclass
+    ``str`` so they compare and JSON-serialize as their plain value.
+    """
+
+    STOPPED = "STOPPED"
+    STARTING = "STARTING"
+    RUNNING = "RUNNING"
+    BACKOFF = "BACKOFF"
+    STOPPING = "STOPPING"
 
 
 class CompanionConfig:
@@ -83,3 +91,88 @@ class CompanionConfig:
 
     def __repr__(self):
         return "<CompanionConfig %s>" % self.name
+
+
+class CompanionProcess:
+    """Runtime state for one companion, separate from its static config.
+
+    Holds everything ``status`` needs: current public state, live pid, restart
+    and exit counters, last exit info, and the ``manual_stop`` flag that keeps a
+    user-stopped companion from auto-restarting.
+    """
+
+    def __init__(self, config):
+        self.config = config
+        self.state = State.STOPPED
+        self.pid = None
+        self.restart_delay = 5
+
+        self.started_at = None
+        self.exited_at = None
+        self.next_retry_at = None
+
+        self.restart_count = 0
+        self.exit_count = 0
+        self.kill_count = 0
+
+        self.last_exit_code = None
+        self.last_exit_signal = None
+
+        self.manual_stop = False
+
+    @property
+    def name(self):
+        return self.config.name
+
+    def uptime(self, now=None):
+        """Seconds since this companion last started, or ``None`` if not up."""
+        if self.state not in (State.RUNNING, State.STARTING) or self.started_at is None:
+            return None
+        return (now or time.time()) - self.started_at
+
+    def description(self, now=None):
+        """Human one-liner: state label plus runtime details."""
+        now = now or time.time()
+        label = self.state.lower()
+        detail = self._detail(now)
+        return "%s, %s" % (label, detail) if detail else label
+
+    def _detail(self, now):
+        if self.state == State.RUNNING:
+            return "pid %s, uptime %s" % (
+                self.pid,
+                format_uptime(self.uptime(now) or 0),
+            )
+        if self.state == State.BACKOFF and self.next_retry_at is not None:
+            left = max(0, int(self.next_retry_at - now))
+            return "exited with %s, retrying in %ds" % (self._exit_status(), left)
+        if self.state == State.BACKOFF:
+            return "exited with %s" % self._exit_status()
+        if self.state == State.STOPPED and self.manual_stop:
+            return "stopped manually"
+        if self.state == State.STOPPED and self.exited_at is not None:
+            return "exited with %s" % self._exit_status()
+        if self.state == State.STOPPED:
+            return "not started"
+        return ""
+
+    def _exit_status(self):
+        if self.last_exit_signal is not None:
+            return "signal %s" % self.last_exit_signal
+        return "status %s" % self.last_exit_code
+
+    def status_dict(self, now=None):
+        """Machine-readable status entry for the JSON control protocol."""
+        backoff = self.state == State.BACKOFF
+        return {
+            "name": self.name,
+            "state": self.state,
+            "pid": self.pid,
+            "description": self.description(now or time.time()),
+            "next_retry_at": self.next_retry_at if backoff else None,
+            "restart_delay": self.restart_delay if backoff else None,
+            "last_exit_code": self.last_exit_code if backoff else None,
+        }
+
+    def __repr__(self):
+        return "<CompanionProcess %s %s>" % (self.name, self.state)

gunicorn/util.py

@@ -647,3 +647,15 @@ def bytes_to_str(b):
 
 def unquote_to_wsgi_str(string):
     return urllib.parse.unquote_to_bytes(string).decode('latin-1')
+
+
+def format_uptime(seconds):
+    """Render a duration like supervisor: ``2 days, 03:12:44`` or ``0:05:12``."""
+    seconds = int(seconds)
+    days, rem = divmod(seconds, 86400)
+    hours, rem = divmod(rem, 3600)
+    minutes, secs = divmod(rem, 60)
+    if days:
+        unit = "day" if days == 1 else "days"
+        return "%d %s, %02d:%02d:%02d" % (days, unit, hours, minutes, secs)
+    return "%d:%02d:%02d" % (hours, minutes, secs)