|
#!/usr/bin/env python |
|
"""Spawn long-running tasks in a GUI terminal or tmux session. |
|
|
|
Purpose (agent + human): The agent must write the workload as a script under |
|
./scripts/ and launch it via a real terminal because the Codex harness hangs or |
|
times out on long-running commands. This CLI is the single, safe entrypoint. |
|
|
|
What it does: |
|
- loads .env early (without overriding existing env), |
|
- optionally activates a repo venv, |
|
- opens a terminal (gnome/konsole/kitty/xterm) or headless tmux session, |
|
- supports geometry hints, PID capture, tee logging, gnome dark theme, |
|
- keeps the shell open after the task finishes. |
|
|
|
Usage (shim): |
|
scripts/spawn_task_terminal.sh /abs/repo scripts/run_smokes.sh \ |
|
--title "Oracle Smokes" --geometry "160x40+3840+0" |
|
|
|
Headless: |
|
scripts/spawn_task_terminal.sh /abs/repo scripts/run_pipeline.sh --backend tmux |
|
|
|
Env helpers: |
|
PREFERRED_TERMINAL=gnome-terminal |
|
SPAWN_GEOMETRY="160x40+3840+0" |
|
GNOME_TERMINAL_THEME=dark |
|
""" |
|
|
|
from __future__ import annotations |
|
|
|
from dotenv import load_dotenv, find_dotenv # type: ignore |
|
|
|
# Load .env as early as possible for downstream env-dependent imports/config. |
|
load_dotenv(find_dotenv(usecwd=True), override=False) |
|
|
|
import os |
|
import shutil |
|
import subprocess |
|
import textwrap |
|
from enum import Enum |
|
from pathlib import Path |
|
from typing import Optional |
|
|
|
import typer |
|
|
|
|
|
class LogMode(str, Enum): |
|
tee = "tee" |
|
none = "none" |
|
|
|
|
|
class Backend(str, Enum): |
|
auto = "auto" |
|
terminal = "terminal" |
|
tmux = "tmux" |
|
|
|
|
|
class GeometryPreset(str, Enum): |
|
left = "left" |
|
right = "right" |
|
primary = "primary" |
|
|
|
|
|
class GnomeTheme(str, Enum): |
|
system = "system" |
|
dark = "dark" |
|
light = "light" |
|
|
|
|
|
def ensure_inside_repo(repo_abs: Path, target: Path) -> None: |
|
"""Ensure the script to run sits inside the repository root.""" |
|
|
|
repo_abs = repo_abs.resolve() |
|
target = target.resolve() |
|
try: |
|
is_rel = target.is_relative_to(repo_abs) # Python 3.9+ |
|
except AttributeError: |
|
is_rel = repo_abs == target or repo_abs in target.parents |
|
if not is_rel: |
|
raise typer.BadParameter(f"Script {target} is outside repo {repo_abs}") |
|
|
|
|
|
def pick_terminal() -> str: |
|
"""Prefer reliable geometry: gnome-terminal first, then common fallbacks.""" |
|
|
|
candidates = os.environ.get("PREFERRED_TERMINAL", "gnome-terminal konsole kitty xterm").split() |
|
for cand in candidates: |
|
if shutil.which(cand): |
|
return cand |
|
raise typer.BadParameter( |
|
"No supported GUI terminal found (tried gnome-terminal, konsole, kitty, xterm)." |
|
) |
|
|
|
|
|
DEFAULT_GEOM = "160x40+100+100" |
|
|
|
|
|
def main( |
|
repo_abs: Path = typer.Argument(..., help="Absolute path to repo root."), |
|
script_path: Path = typer.Argument(..., help="Script path (absolute or repo-relative)."), |
|
title: str = typer.Option("Task Runner", "--title", help="Window/tab title."), |
|
pid_file: Optional[Path] = typer.Option(None, "--pid-file", help="Write shell PID to this file."), |
|
geometry: str = typer.Option( |
|
DEFAULT_GEOM, |
|
"--geometry", |
|
help="Terminal geometry COLSxROWS+X+Y.", |
|
), |
|
geometry_preset: Optional[GeometryPreset] = typer.Option( |
|
None, |
|
"--geometry-preset", |
|
case_sensitive=False, |
|
help="left|right|primary. Uses env LEFT_MONITOR_GEOM/RIGHT_MONITOR_GEOM/PRIMARY_MONITOR_GEOM or falls back to --geometry/default.", |
|
), |
|
gnome_theme: GnomeTheme = typer.Option( |
|
GnomeTheme.dark, |
|
"--gnome-theme", |
|
case_sensitive=False, |
|
help="gnome-terminal theme variant (dark|light|system). Ignored for other terminals.", |
|
), |
|
backend: Backend = typer.Option( |
|
Backend.auto, |
|
"--backend", |
|
case_sensitive=False, |
|
help="auto (DISPLAY→terminal, else tmux), terminal, or tmux.", |
|
), |
|
log: LogMode = typer.Option(LogMode.tee, "--log", case_sensitive=False, help="tee or none."), |
|
no_log: bool = typer.Option(False, "--no-log", help="Shorthand for --log none."), |
|
venv: Optional[Path] = typer.Option( |
|
None, |
|
"--venv", |
|
help="Path to virtualenv activate script (default: <repo>/.venv/bin/activate if present).", |
|
), |
|
no_venv: bool = typer.Option(False, "--no-venv", help="Disable venv activation."), |
|
no_auto_exec: bool = typer.Option(False, "--no-auto-exec", help="Open shell but do not run the script."), |
|
dotenv: Optional[Path] = typer.Option( |
|
None, |
|
"--dotenv", |
|
help="Path to .env to load (default: <repo>/.env if present).", |
|
), |
|
no_dotenv: bool = typer.Option(False, "--no-dotenv", help="Do not load any .env file."), |
|
) -> None: |
|
""" |
|
Spawn a GUI terminal to run a repo-local script with optional venv/.env and logging. |
|
|
|
Designed so long-running tasks happen in a detached terminal instead of the Codex harness. |
|
""" |
|
|
|
repo_abs = repo_abs.expanduser().resolve() |
|
if not repo_abs.is_dir(): |
|
raise typer.BadParameter(f"Repo path {repo_abs} is not a directory") |
|
|
|
if script_path.is_absolute(): |
|
target_script = script_path.expanduser().resolve() |
|
else: |
|
target_script = (repo_abs / script_path).expanduser().resolve() |
|
|
|
if not target_script.is_file(): |
|
raise typer.BadParameter(f"Script {target_script} does not exist") |
|
|
|
ensure_inside_repo(repo_abs, target_script) |
|
|
|
# Make executable when possible. |
|
try: |
|
mode = target_script.stat().st_mode |
|
target_script.chmod(mode | 0o111) |
|
except PermissionError: |
|
pass |
|
|
|
if no_log: |
|
log = LogMode.none |
|
|
|
if no_venv: |
|
venv_path: Optional[Path] = None |
|
elif venv is not None: |
|
venv_path = venv.expanduser().resolve() |
|
else: |
|
default_venv = repo_abs / ".venv" / "bin" / "activate" |
|
venv_path = default_venv if default_venv.is_file() else None |
|
|
|
if pid_file is not None: |
|
pid_file = pid_file.expanduser().resolve() |
|
pid_file.parent.mkdir(parents=True, exist_ok=True) |
|
|
|
if no_dotenv: |
|
dotenv_path = None |
|
elif dotenv is not None: |
|
dotenv_path = dotenv.expanduser().resolve() |
|
else: |
|
candidate = repo_abs / ".env" |
|
dotenv_path = candidate if candidate.is_file() else None |
|
|
|
if dotenv_path is not None: |
|
load_dotenv(dotenv_path=dotenv_path, override=False) |
|
typer.echo(f"[spawn_task_terminal] loaded .env from {dotenv_path}") |
|
else: |
|
typer.echo("[spawn_task_terminal] no .env loaded") |
|
|
|
# Backend selection |
|
env_backend = os.environ.get("SPAWN_BACKEND") |
|
if env_backend: |
|
try: |
|
backend_env_val = Backend(env_backend) |
|
except ValueError: |
|
raise typer.BadParameter("SPAWN_BACKEND must be auto|terminal|tmux") |
|
else: |
|
backend_env_val = None |
|
|
|
env = os.environ.copy() |
|
env.update( |
|
{ |
|
"REPO_ABS": str(repo_abs), |
|
"TARGET_SCRIPT": str(target_script), |
|
"PID_FILE": str(pid_file) if pid_file is not None else "", |
|
"TITLE": title, |
|
"LOG_MODE": log.value, |
|
"AUTO_EXEC": "0" if no_auto_exec else "1", |
|
"VENV_PATH": str(venv_path) if venv_path is not None else "", |
|
} |
|
) |
|
|
|
inner_script = textwrap.dedent( |
|
r''' |
|
set -Eeuo pipefail |
|
|
|
if [[ -n "${PID_FILE:-}" ]]; then |
|
echo "$$" > "$PID_FILE" |
|
fi |
|
|
|
cd "$REPO_ABS" |
|
|
|
if [[ -n "${VENV_PATH:-}" && -f "$VENV_PATH" ]]; then |
|
# shellcheck disable=SC1090 |
|
source "$VENV_PATH" |
|
echo "[spawn_task_terminal] activated venv: $VENV_PATH" |
|
else |
|
echo "[spawn_task_terminal] no venv activated" |
|
fi |
|
|
|
chmod +x "$TARGET_SCRIPT" 2>/dev/null || true |
|
|
|
LOG_MODE="${LOG_MODE:-tee}" # tee | none |
|
AUTO_EXEC="${AUTO_EXEC:-1}" # 1 | 0 |
|
|
|
LOG_DIR="${LOG_DIR:-$REPO_ABS/logs}" |
|
mkdir -p "$LOG_DIR" 2>/dev/null || true |
|
ts=$(date +'%Y%m%d_%H%M%S') |
|
base=$(basename "$TARGET_SCRIPT") |
|
base_no_ext="${base%.*}" |
|
LOG_FILE="$LOG_DIR/${base_no_ext}_${ts}.log" |
|
|
|
if [[ "$AUTO_EXEC" != "1" ]]; then |
|
echo "[spawn_task_terminal] Opened shell in: $REPO_ABS" |
|
echo "[spawn_task_terminal] Target script : $TARGET_SCRIPT" |
|
if [[ "$LOG_MODE" != "none" ]]; then |
|
echo "[spawn_task_terminal] Suggested log: $LOG_FILE" |
|
fi |
|
echo |
|
echo "When ready, run:" |
|
echo " \"$TARGET_SCRIPT\"" |
|
if [[ "$LOG_MODE" != "none" ]]; then |
|
echo |
|
echo "For logging via tee, run:" |
|
echo " \"$TARGET_SCRIPT\" 2>&1 | tee -a \"$LOG_FILE\"" |
|
fi |
|
echo |
|
exec "${SHELL:-bash}" |
|
fi |
|
|
|
echo "[spawn_task_terminal] logging mode : $LOG_MODE" |
|
echo "[spawn_task_terminal] repo : $REPO_ABS" |
|
echo "[spawn_task_terminal] script : $TARGET_SCRIPT" |
|
if [[ "$LOG_MODE" != "none" ]]; then |
|
echo "[spawn_task_terminal] log file : $LOG_FILE" |
|
fi |
|
|
|
status=0 |
|
|
|
if [[ "$LOG_MODE" == "none" ]]; then |
|
"$TARGET_SCRIPT" || status=$? |
|
else |
|
"$TARGET_SCRIPT" 2>&1 | tee -a "$LOG_FILE" |
|
status=${PIPESTATUS[0]} |
|
fi |
|
|
|
echo |
|
echo "[$TITLE] finished with status $status." |
|
if [[ "$LOG_MODE" != "none" ]]; then |
|
echo "Log: $LOG_FILE" |
|
fi |
|
echo "You can close this window or press Ctrl+D." |
|
|
|
exec "${SHELL:-bash}" |
|
exit "$status" |
|
''' |
|
).strip() |
|
|
|
def maybe_set_gnome_theme(theme: GnomeTheme) -> None: |
|
if theme == GnomeTheme.system: |
|
return |
|
if shutil.which("gsettings") is None: |
|
typer.echo("[spawn_task_terminal] gsettings not found; cannot set gnome theme variant", err=True) |
|
return |
|
cmd = [ |
|
"gsettings", |
|
"set", |
|
"org.gnome.Terminal.Legacy.Settings", |
|
"theme-variant", |
|
theme.value, |
|
] |
|
subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) |
|
|
|
def launch_terminal(geom: str) -> None: |
|
terminal_bin = pick_terminal() |
|
|
|
if terminal_bin == "gnome-terminal": |
|
maybe_set_gnome_theme(gnome_theme) |
|
|
|
if terminal_bin == "konsole": |
|
cmd = [ |
|
terminal_bin, |
|
"--workdir", |
|
str(repo_abs), |
|
"--geometry", |
|
geom, |
|
"-p", |
|
f"tabtitle={title}", |
|
"-e", |
|
*shell_cmd, |
|
] |
|
elif terminal_bin in {"gnome-terminal", "mate-terminal"}: |
|
cmd = [ |
|
terminal_bin, |
|
"--title", |
|
title, |
|
"--geometry", |
|
geom, |
|
"--", |
|
*shell_cmd, |
|
] |
|
elif terminal_bin == "kitty": |
|
cmd = [terminal_bin, "-T", title, *shell_cmd] |
|
elif terminal_bin == "xterm": |
|
cmd = [terminal_bin, "-T", title, "-geometry", geom, "-e", *shell_cmd] |
|
else: |
|
raise typer.BadParameter(f"Unsupported terminal {terminal_bin!r}") |
|
|
|
subprocess.Popen(cmd, env=env) |
|
|
|
def launch_tmux(session_hint: str | None = None) -> None: |
|
session = session_hint or title.lower().replace(" ", "_") or "task" |
|
base_session = session |
|
counter = 1 |
|
while subprocess.run(["tmux", "has-session", "-t", session], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL).returncode == 0: |
|
session = f"{base_session}_{counter}" |
|
counter += 1 |
|
|
|
cmd = ["tmux", "new-session", "-d", "-s", session, *shell_cmd] |
|
subprocess.Popen(cmd, env=env) |
|
typer.echo(f"[spawn_task_terminal] tmux session: {session}") |
|
|
|
# Respect CLI backend override (Typer provides values now) |
|
# Re-evaluate backend with explicit parameter |
|
|
|
# Typer passes parameters after function definition; to keep minimal change, recompute here |
|
|
|
# pylint: disable=too-many-branches |
|
|
|
# Determine geometry: CLI beats env; env only overrides when CLI left at default |
|
def resolve_preset(preset: GeometryPreset | None) -> str | None: |
|
if preset is None: |
|
return None |
|
env_map = { |
|
GeometryPreset.left: os.environ.get("LEFT_MONITOR_GEOM"), |
|
GeometryPreset.right: os.environ.get("RIGHT_MONITOR_GEOM"), |
|
GeometryPreset.primary: os.environ.get("PRIMARY_MONITOR_GEOM"), |
|
} |
|
return env_map.get(preset) |
|
|
|
geometry_value = geometry |
|
preset_geom = resolve_preset(geometry_preset) |
|
if preset_geom: |
|
geometry_value = preset_geom |
|
elif geometry == DEFAULT_GEOM and os.environ.get("SPAWN_GEOMETRY"): |
|
geometry_value = os.environ["SPAWN_GEOMETRY"] |
|
|
|
# Allow env override for gnome theme when CLI kept default |
|
if gnome_theme == GnomeTheme.dark and os.environ.get("GNOME_TERMINAL_THEME"): |
|
try: |
|
gnome_theme = GnomeTheme(os.environ["GNOME_TERMINAL_THEME"]) |
|
except ValueError: |
|
raise typer.BadParameter("GNOME_TERMINAL_THEME must be system|dark|light") |
|
|
|
# Select user shell for execution; default to bash. For zsh, run interactive (-i) |
|
# so ~/.zshrc is sourced, keeping env consistent with user terminals. |
|
shell_bin = Path(os.environ.get("SHELL", "/bin/bash")).resolve() |
|
if shell_bin.name == "zsh": |
|
shell_cmd = [str(shell_bin), "-ic", inner_script] |
|
else: |
|
shell_cmd = ["bash", "-lc", inner_script] |
|
|
|
# Choose backend (CLI > env > auto) |
|
if backend_env_val is not None and backend == Backend.auto: |
|
backend_candidate = backend_env_val |
|
else: |
|
backend_candidate = backend |
|
|
|
if backend_candidate == Backend.auto: |
|
resolved_backend = Backend.terminal if os.environ.get("DISPLAY") else Backend.tmux |
|
else: |
|
resolved_backend = backend_candidate |
|
|
|
if resolved_backend == Backend.terminal and os.environ.get("DISPLAY") is None: |
|
raise typer.BadParameter("DISPLAY is not set; choose --backend tmux or set SPAWN_BACKEND=tmux for headless use.") |
|
|
|
if resolved_backend == Backend.tmux and shutil.which("tmux") is None: |
|
raise typer.BadParameter("tmux not found on PATH; install tmux or use --backend terminal with DISPLAY.") |
|
|
|
typer.echo("Spawning task runner:") |
|
typer.echo(f" backend: {resolved_backend.value}") |
|
typer.echo(f" repo : {repo_abs}") |
|
typer.echo(f" script : {target_script}") |
|
typer.echo(f" title : {title}") |
|
typer.echo(f" geom : {geometry_value}") |
|
typer.echo(f" log : {log.value}") |
|
typer.echo(f" auto : {0 if no_auto_exec else 1}") |
|
if venv_path is not None: |
|
typer.echo(f" venv : {venv_path}") |
|
if pid_file is not None: |
|
typer.echo(f" pidfile: {pid_file}") |
|
|
|
if resolved_backend == Backend.terminal: |
|
launch_terminal(geometry_value) |
|
else: |
|
launch_tmux() |
|
|
|
raise typer.Exit(code=0) |
|
|
|
|
|
if __name__ == "__main__": |
|
typer.run(main) |