Manages dev server subprocess lifecycle for a single project. Provides start/stop with cross-platform support via psutil. Supports multiple output callbacks for WebSocket clients. Detects and tracks the server URL from output.
| 80 | |
| 81 | |
| 82 | class DevServerProcessManager: |
| 83 | """ |
| 84 | Manages dev server subprocess lifecycle for a single project. |
| 85 | |
| 86 | Provides start/stop with cross-platform support via psutil. |
| 87 | Supports multiple output callbacks for WebSocket clients. |
| 88 | Detects and tracks the server URL from output. |
| 89 | """ |
| 90 | |
| 91 | def __init__( |
| 92 | self, |
| 93 | project_name: str, |
| 94 | project_dir: Path, |
| 95 | ): |
| 96 | """ |
| 97 | Initialize the dev server process manager. |
| 98 | |
| 99 | Args: |
| 100 | project_name: Name of the project |
| 101 | project_dir: Absolute path to the project directory |
| 102 | """ |
| 103 | self.project_name = project_name |
| 104 | self.project_dir = project_dir |
| 105 | self.process: subprocess.Popen | None = None |
| 106 | self._status: Literal["stopped", "running", "crashed"] = "stopped" |
| 107 | self.started_at: datetime | None = None |
| 108 | self._output_task: asyncio.Task | None = None |
| 109 | self._detected_url: str | None = None |
| 110 | self._command: str | None = None # Store the command used to start |
| 111 | |
| 112 | # Support multiple callbacks (for multiple WebSocket clients) |
| 113 | self._output_callbacks: Set[Callable[[str], Awaitable[None]]] = set() |
| 114 | self._status_callbacks: Set[Callable[[str], Awaitable[None]]] = set() |
| 115 | self._callbacks_lock = threading.Lock() |
| 116 | |
| 117 | # Lock file to prevent multiple instances (stored in project directory) |
| 118 | from autoforge_paths import get_devserver_lock_path |
| 119 | self.lock_file = get_devserver_lock_path(self.project_dir) |
| 120 | |
| 121 | @property |
| 122 | def status(self) -> Literal["stopped", "running", "crashed"]: |
| 123 | """Current status of the dev server.""" |
| 124 | return self._status |
| 125 | |
| 126 | @status.setter |
| 127 | def status(self, value: Literal["stopped", "running", "crashed"]): |
| 128 | old_status = self._status |
| 129 | self._status = value |
| 130 | if old_status != value: |
| 131 | self._notify_status_change(value) |
| 132 | |
| 133 | @property |
| 134 | def detected_url(self) -> str | None: |
| 135 | """The URL detected from server output, if any.""" |
| 136 | return self._detected_url |
| 137 | |
| 138 | @property |
| 139 | def pid(self) -> int | None: |
no outgoing calls