1import os
2import platform
3import socket
4import subprocess
5import sys
6import tomllib
7from importlib.metadata import entry_points
8from importlib.util import find_spec
9from pathlib import Path
10
11import click
12from rich.columns import Columns
13from rich.console import Console
14from rich.text import Text
15
16from plain.runtime import APP_PATH, PLAIN_TEMP_PATH
17
18from .mkcert import MkcertManager
19from .process import ProcessManager
20from .utils import has_pyproject_toml
21
22ENTRYPOINT_GROUP = "plain.dev"
23
24
25class DevProcess(ProcessManager):
26 pidfile = PLAIN_TEMP_PATH / "dev" / "dev.pid"
27 log_dir = PLAIN_TEMP_PATH / "dev" / "logs" / "run"
28
29 def setup(
30 self, *, port: int | None, hostname: str | None, log_level: str | None
31 ) -> None:
32 if not hostname:
33 project_name = os.path.basename(
34 os.getcwd()
35 ) # Use directory name by default
36
37 if has_pyproject_toml(APP_PATH.parent):
38 with open(Path(APP_PATH.parent, "pyproject.toml"), "rb") as f:
39 pyproject = tomllib.load(f)
40 project_name = pyproject.get("project", {}).get(
41 "name", project_name
42 )
43
44 hostname = f"{project_name.lower()}.localhost"
45
46 self.hostname = hostname
47 self.log_level = log_level
48
49 self.pid_value = self.pid
50 self.prepare_log()
51
52 if port:
53 self.port = int(port)
54 if not self._port_available(self.port):
55 click.secho(f"Port {self.port} in use", fg="red")
56 raise SystemExit(1)
57 else:
58 self.port = self._find_open_port(8443)
59 if self.port != 8443:
60 click.secho(f"Port 8443 in use, using {self.port}", fg="yellow")
61
62 self.ssl_key_path = None
63 self.ssl_cert_path = None
64
65 self.url = f"https://{self.hostname}:{self.port}"
66 self.tunnel_url = os.environ.get("PLAIN_DEV_TUNNEL_URL", "")
67
68 self.plain_env = {
69 "PYTHONUNBUFFERED": "true",
70 "PLAIN_DEV": "true",
71 "FORCE_COLOR": "1",
72 **os.environ,
73 }
74
75 if log_level:
76 self.plain_env["PLAIN_FRAMEWORK_LOG_LEVEL"] = log_level.upper()
77 self.plain_env["PLAIN_LOG_LEVEL"] = log_level.upper()
78
79 self.custom_process_env = {
80 **self.plain_env,
81 "PORT": str(self.port),
82 "PLAIN_DEV_URL": self.url,
83 }
84
85 if self.tunnel_url:
86 status_bar = Columns(
87 [
88 Text.from_markup(
89 f"[bold]Tunnel[/bold] [underline][link={self.tunnel_url}]{self.tunnel_url}[/link][/underline]"
90 ),
91 Text.from_markup(
92 f"[dim][bold]Server[/bold] [link={self.url}]{self.url}[/link][/dim]"
93 ),
94 Text.from_markup(
95 "[dim][bold]Ctrl+C[/bold] to stop[/dim]",
96 justify="right",
97 ),
98 ],
99 expand=True,
100 )
101 else:
102 status_bar = Columns(
103 [
104 Text.from_markup(
105 f"[bold]Server[/bold] [underline][link={self.url}]{self.url}[/link][/underline]"
106 ),
107 Text.from_markup(
108 "[dim][bold]Ctrl+C[/bold] to stop[/dim]", justify="right"
109 ),
110 ],
111 expand=True,
112 )
113 self.console = Console(markup=False, highlight=False)
114 self.console_status = self.console.status(status_bar)
115
116 self.init_poncho(self.console.out)
117
118 def _find_open_port(self, start_port: int) -> int:
119 port = start_port
120 while not self._port_available(port):
121 port += 1
122 return port
123
124 def _port_available(self, port: int) -> bool:
125 with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
126 sock.settimeout(0.5)
127 result = sock.connect_ex(("127.0.0.1", port))
128 return result != 0
129
130 def run(self) -> int:
131 self.write_pidfile()
132 mkcert_manager = MkcertManager()
133 mkcert_manager.setup_mkcert(install_path=Path.home() / ".plain" / "dev")
134 self.ssl_cert_path, self.ssl_key_path = mkcert_manager.generate_certs(
135 domain=self.hostname,
136 storage_path=Path(PLAIN_TEMP_PATH) / "dev" / "certs",
137 )
138
139 self.symlink_plain_src()
140 self.generate_agents_md()
141 self.modify_hosts_file()
142
143 click.secho("→ Running preflight checks... ", dim=True, nl=False)
144 self.run_preflight()
145
146 # if ServicesProcess.running_pid():
147 # self.poncho.add_process(
148 # "services",
149 # f"{sys.executable} -m plain dev logs --services --follow",
150 # )
151
152 if find_spec("plain.models"):
153 click.secho("→ Waiting for database... ", dim=True, nl=False)
154 subprocess.run(
155 [sys.executable, "-m", "plain", "db", "wait"],
156 env=self.plain_env,
157 check=True,
158 )
159 click.secho("→ Running migrations...", dim=True)
160 subprocess.run(
161 [sys.executable, "-m", "plain", "migrate", "--backup"],
162 env=self.plain_env,
163 check=True,
164 )
165
166 click.secho("\n→ Starting app...", dim=True)
167
168 # Manually start the status bar now so it isn't bungled by
169 # another thread checking db stuff...
170 self.console_status.start()
171
172 assert self.poncho is not None, "poncho should be initialized"
173
174 self.add_server()
175 self.add_entrypoints()
176 self.add_pyproject_run()
177
178 try:
179 # Start processes we know about and block the main thread
180 self.poncho.loop()
181
182 # Remove the status bar
183 self.console_status.stop()
184 finally:
185 self.rm_pidfile()
186 self.close()
187
188 assert self.poncho.returncode is not None, "returncode should be set after loop"
189 return self.poncho.returncode
190
191 def symlink_plain_src(self) -> None:
192 """Symlink the plain package into .plain so we can look at it easily"""
193 spec = find_spec("plain.runtime")
194 if spec is None or spec.origin is None:
195 return None
196 plain_path = Path(spec.origin).parent.parent
197 if not PLAIN_TEMP_PATH.exists():
198 PLAIN_TEMP_PATH.mkdir()
199
200 symlink_path = PLAIN_TEMP_PATH / "src"
201
202 # The symlink is broken
203 if symlink_path.is_symlink() and not symlink_path.exists():
204 symlink_path.unlink()
205
206 # The symlink exists but points to the wrong place
207 if (
208 symlink_path.is_symlink()
209 and symlink_path.exists()
210 and symlink_path.resolve() != plain_path
211 ):
212 symlink_path.unlink()
213
214 if plain_path.exists() and not symlink_path.exists():
215 symlink_path.symlink_to(plain_path)
216
217 def generate_agents_md(self) -> None:
218 """Generate .plain/AGENTS.md from installed packages with AGENTS.md files."""
219 try:
220 result = subprocess.run(
221 [sys.executable, "-m", "plain", "agent", "md", "--save"],
222 check=False,
223 capture_output=True,
224 text=True,
225 )
226 if result.returncode != 0 and result.stderr:
227 click.secho(
228 f"Warning: Failed to generate .plain/AGENTS.md: {result.stderr}",
229 fg="yellow",
230 err=True,
231 )
232 except Exception as e:
233 click.secho(
234 f"Warning: Failed to generate .plain/AGENTS.md: {e}",
235 fg="yellow",
236 err=True,
237 )
238
239 def modify_hosts_file(self) -> None:
240 """Modify the hosts file to map the custom domain to 127.0.0.1."""
241 entry_identifier = "# Added by plain"
242 hosts_entry = f"127.0.0.1 {self.hostname} {entry_identifier}"
243
244 if platform.system() == "Windows":
245 hosts_path = Path(r"C:\Windows\System32\drivers\etc\hosts")
246 try:
247 with hosts_path.open("r") as f:
248 content = f.read()
249
250 if hosts_entry in content:
251 return # Entry already exists; no action needed
252
253 # Entry does not exist; add it
254 with hosts_path.open("a") as f:
255 f.write(f"{hosts_entry}\n")
256 click.secho(f"Added {self.hostname} to {hosts_path}", bold=True)
257 except PermissionError:
258 click.secho(
259 "Permission denied while modifying hosts file. Please run the script as an administrator.",
260 fg="red",
261 )
262 sys.exit(1)
263 else:
264 # For macOS and Linux
265 hosts_path = Path("/etc/hosts")
266 try:
267 with hosts_path.open("r") as f:
268 content = f.read()
269
270 if hosts_entry in content:
271 return # Entry already exists; no action needed
272
273 # Entry does not exist; append it using sudo
274 click.secho(
275 f"Adding {self.hostname} to /etc/hosts file. You may be prompted for your password.\n",
276 bold=True,
277 )
278 cmd = f"echo '{hosts_entry}' | sudo tee -a {hosts_path} >/dev/null"
279 subprocess.run(cmd, shell=True, check=True)
280 click.secho(f"Added {self.hostname} to {hosts_path}\n", bold=True)
281 except PermissionError:
282 click.secho(
283 "Permission denied while accessing hosts file.",
284 fg="red",
285 )
286 sys.exit(1)
287 except subprocess.CalledProcessError:
288 click.secho(
289 "Failed to modify hosts file. Please ensure you have sudo privileges.",
290 fg="red",
291 )
292 sys.exit(1)
293
294 def run_preflight(self) -> None:
295 if subprocess.run(
296 ["plain", "preflight", "--quiet"], env=self.plain_env
297 ).returncode:
298 click.secho("Preflight check failed!", fg="red")
299 sys.exit(1)
300
301 def add_server(self) -> None:
302 """Add the Plain HTTP server process."""
303 assert self.poncho is not None
304 server_cmd = [
305 sys.executable,
306 "-m",
307 "plain",
308 "server",
309 "--bind",
310 f"{self.hostname}:{self.port}",
311 "--certfile",
312 str(self.ssl_cert_path),
313 "--keyfile",
314 str(self.ssl_key_path),
315 "--threads",
316 "4",
317 "--timeout",
318 "60",
319 "--log-level",
320 self.log_level or "info",
321 "--log-format",
322 "'[%(levelname)s] %(message)s'",
323 "--access-log-format",
324 "'\"%(r)s\" status=%(s)s length=%(b)s time=%(M)sms'",
325 "--reload", # Enable auto-reload for development
326 ]
327
328 server = " ".join(server_cmd)
329 self.poncho.add_process("plain", server, env=self.plain_env)
330
331 def add_entrypoints(self) -> None:
332 assert self.poncho is not None
333 for entry_point in entry_points().select(group=ENTRYPOINT_GROUP):
334 self.poncho.add_process(
335 entry_point.name,
336 f"plain dev entrypoint {entry_point.name}",
337 env=self.plain_env,
338 )
339
340 def add_pyproject_run(self) -> None:
341 """Additional processes that only run during `plain dev`."""
342 assert self.poncho is not None
343 if not has_pyproject_toml(APP_PATH.parent):
344 return
345
346 with open(Path(APP_PATH.parent, "pyproject.toml"), "rb") as f:
347 pyproject = tomllib.load(f)
348
349 run_commands = (
350 pyproject.get("tool", {}).get("plain", {}).get("dev", {}).get("run", {})
351 )
352 for name, data in run_commands.items():
353 env = {
354 **self.custom_process_env,
355 **data.get("env", {}),
356 }
357 self.poncho.add_process(name, data["cmd"], env=env)