tegwick
535836054a
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: sonnet Assistant-Process: 295952@bnt-lap001 Assistant-Session: e93f64ad-516c-46eb-9666-aad8d300c477
41 lines
1.3 KiB
Python
41 lines
1.3 KiB
Python
"""Launch a legacy-command pack: start the existing console and wait for it."""
|
|
from __future__ import annotations
|
|
|
|
import subprocess
|
|
import time
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
from .pack import Pack
|
|
|
|
|
|
def repo_root(pack: Pack) -> Path:
|
|
"""runbooks/<dir>/pack.yaml -> repository root."""
|
|
return pack.path.resolve().parents[2]
|
|
|
|
|
|
def wait_ready(url: str, timeout: float = 30.0) -> bool:
|
|
end = time.monotonic() + timeout
|
|
while time.monotonic() < end:
|
|
try:
|
|
with urllib.request.urlopen(url, timeout=2):
|
|
return True
|
|
except Exception:
|
|
time.sleep(0.3)
|
|
return False
|
|
|
|
|
|
def launch(pack: Pack, timeout: float = 30.0) -> subprocess.Popen:
|
|
"""Start the console in the owning repo. Caller owns the returned process."""
|
|
if pack.data.get("engine") != "legacy-command":
|
|
raise ValueError("not a legacy-command pack")
|
|
spec = pack.data["legacy"]
|
|
cwd = (repo_root(pack) / spec.get("cwd", ".")).resolve()
|
|
if repo_root(pack) not in [cwd, *cwd.parents]:
|
|
raise ValueError("legacy.cwd escapes the owning repository")
|
|
proc = subprocess.Popen(spec["command"], cwd=cwd)
|
|
url = spec.get("url")
|
|
if url and not wait_ready(url, timeout):
|
|
proc.terminate()
|
|
raise TimeoutError(f"{url} did not answer within {timeout}s")
|
|
return proc
|