Initialize the process manager. Args: project_name: Name of the project project_dir: Absolute path to the project directory root_dir: Root directory of the autonomous-coding-ui project
(
self,
project_name: str,
project_dir: Path,
root_dir: Path,
)
| 60 | """ |
| 61 | |
| 62 | def __init__( |
| 63 | self, |
| 64 | project_name: str, |
| 65 | project_dir: Path, |
| 66 | root_dir: Path, |
| 67 | ): |
| 68 | """ |
| 69 | Initialize the process manager. |
| 70 | |
| 71 | Args: |
| 72 | project_name: Name of the project |
| 73 | project_dir: Absolute path to the project directory |
| 74 | root_dir: Root directory of the autonomous-coding-ui project |
| 75 | """ |
| 76 | self.project_name = project_name |
| 77 | self.project_dir = project_dir |
| 78 | self.root_dir = root_dir |
| 79 | self.process: subprocess.Popen | None = None |
| 80 | self._status: Literal["stopped", "running", "paused", "crashed"] = "stopped" |
| 81 | self.started_at: datetime | None = None |
| 82 | self._output_task: asyncio.Task | None = None |
| 83 | self.yolo_mode: bool = False # YOLO mode for rapid prototyping |
| 84 | self.model: str | None = None # Model being used |
| 85 | self.parallel_mode: bool = False # Parallel execution mode |
| 86 | self.max_concurrency: int | None = None # Max concurrent agents |
| 87 | self.testing_agent_ratio: int = 1 # Regression testing agents (0-3) |
| 88 | |
| 89 | # Support multiple callbacks (for multiple WebSocket clients) |
| 90 | self._output_callbacks: Set[Callable[[str], Awaitable[None]]] = set() |
| 91 | self._status_callbacks: Set[Callable[[str], Awaitable[None]]] = set() |
| 92 | self._callbacks_lock = threading.Lock() |
| 93 | |
| 94 | # Lock file to prevent multiple instances (stored in project directory) |
| 95 | from autoforge_paths import get_agent_lock_path |
| 96 | self.lock_file = get_agent_lock_path(self.project_dir) |
| 97 | |
| 98 | @property |
| 99 | def status(self) -> Literal["stopped", "running", "paused", "crashed"]: |
nothing calls this directly
no test coverage detected