# L4D2 Host Library v1 Implementation Plan > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** Build a Python `l4d2` host library and `l4d2ctl` CLI with exactly five commands (`install`, `initialize`, `start`, `stop`, `delete`) plus read APIs needed by the local web app. **Architecture:** Runtime paths are hard-coded under `/opt/l4d2`. Write operations are imperative and fail-fast with no lock manager, no rollback, and no preflight checks. CLI behavior remains raw/stderr-first, while library internals additionally expose callback-based streaming and read APIs (`get_instance_status`, `stream_instance_logs`) for the web app. **Tech Stack:** Python 3.12+, Typer, PyYAML, pytest, subprocess, systemd user units, fuse-overlayfs. --- ## Scope and Contracts - Command surface is fixed in v1: - `l4d2ctl install` - `l4d2ctl initialize -f ` - `l4d2ctl start ` - `l4d2ctl stop ` - `l4d2ctl delete ` - CLI `` is the server identity source (YAML does not carry identity semantics). - Hard-coded runtime paths: - `/opt/l4d2/installation` - `/opt/l4d2/overlays/` - `/opt/l4d2/instances/` - `/opt/l4d2/runtime//{upper,work,merged}` - Spec fields: - required: `port` - optional: `overlays`, `arguments`, `config` (all default to `[]`) - unknown keys ignored - Overlay order is preserved from YAML, and the first overlay has highest precedence. - `initialize` always writes `server.cfg`; if `config` is empty/missing, `server.cfg` is empty. - `delete` is no-op success when instance/runtime directories are already missing. - CLI errors: print raw subprocess stderr and exit with subprocess return code. - Additional read APIs for web app (no extra CLI commands): - `get_instance_status(name)` - `stream_instance_logs(name, lines=200, follow=True)` - Blueprints are intentionally out of scope for this library; callers must resolve any blueprint linkage to a concrete YAML spec before calling `initialize`. ## Planned File Layout - `components/l4d2-host-lib/pyproject.toml` - `components/l4d2-host-lib/src/l4d2host/cli.py` - `components/l4d2-host-lib/src/l4d2host/spec.py` - `components/l4d2-host-lib/src/l4d2host/process.py` - `components/l4d2-host-lib/src/l4d2host/steam_install.py` - `components/l4d2-host-lib/src/l4d2host/systemd_user.py` - `components/l4d2-host-lib/src/l4d2host/fs/base.py` - `components/l4d2-host-lib/src/l4d2host/fs/fuse_overlayfs.py` - `components/l4d2-host-lib/src/l4d2host/instances.py` - `components/l4d2-host-lib/src/l4d2host/status.py` - `components/l4d2-host-lib/src/l4d2host/logs.py` - `components/l4d2-host-lib/src/l4d2host/templates/l4d2@.service` - `components/l4d2-host-lib/tests/test_cli.py` - `components/l4d2-host-lib/tests/test_spec.py` - `components/l4d2-host-lib/tests/test_process.py` - `components/l4d2-host-lib/tests/test_install.py` - `components/l4d2-host-lib/tests/test_initialize.py` - `components/l4d2-host-lib/tests/test_lifecycle.py` - `components/l4d2-host-lib/tests/test_status.py` - `components/l4d2-host-lib/tests/test_logs.py` - `components/l4d2-host-lib/README.md` ### Task 1: Scaffold package and CLI entrypoint **Files:** - Create: `components/l4d2-host-lib/pyproject.toml` - Create: `components/l4d2-host-lib/src/l4d2host/__init__.py` - Create: `components/l4d2-host-lib/src/l4d2host/cli.py` - Test: `components/l4d2-host-lib/tests/test_cli.py` - [ ] **Step 1: Write failing CLI help test** ```python from typer.testing import CliRunner from l4d2host.cli import app def test_help_lists_v1_commands(): result = CliRunner().invoke(app, ["--help"]) assert result.exit_code == 0 for command in ["install", "initialize", "start", "stop", "delete"]: assert command in result.output ``` - [ ] **Step 2: Run test and verify failure** Run: `pytest components/l4d2-host-lib/tests/test_cli.py -q` Expected: FAIL with missing package/app. - [ ] **Step 3: Implement pyproject script and minimal CLI app** ```toml [project.scripts] l4d2ctl = "l4d2host.cli:app" ``` ```python import typer app = typer.Typer(no_args_is_help=True) def _todo() -> None: raise typer.Exit(code=1) @app.command() def install() -> None: _todo() @app.command() def initialize(name: str, spec: str = typer.Option(..., "--spec", "-f")) -> None: _todo() @app.command() def start(name: str) -> None: _todo() @app.command() def stop(name: str) -> None: _todo() @app.command() def delete(name: str) -> None: _todo() ``` - [ ] **Step 4: Run tests and verify pass** Run: `pytest components/l4d2-host-lib/tests/test_cli.py -q` Expected: PASS. - [ ] **Step 5: Commit scaffold** ```bash git add components/l4d2-host-lib git commit -m "feat(l4d2): scaffold package and v1 CLI entrypoint" ``` ### Task 2: Implement YAML spec parser **Files:** - Create: `components/l4d2-host-lib/src/l4d2host/spec.py` - Test: `components/l4d2-host-lib/tests/test_spec.py` - [ ] **Step 1: Write failing parser tests** ```python import pytest from pathlib import Path from l4d2host.spec import load_spec def test_minimal_spec_parses(tmp_path: Path): p = tmp_path / "server.yaml" p.write_text("port: 27015\noverlays: [standard]\n") spec = load_spec(p) assert spec.port == 27015 assert spec.overlays == ["standard"] def test_defaults_are_empty_lists(tmp_path: Path): p = tmp_path / "server.yaml" p.write_text("port: 27015\n") spec = load_spec(p) assert spec.overlays == [] assert spec.arguments == [] assert spec.config == [] def test_missing_port_fails(tmp_path: Path): p = tmp_path / "server.yaml" p.write_text("overlays: [standard]\n") with pytest.raises((KeyError, ValueError)): load_spec(p) def test_unknown_keys_ignored(tmp_path: Path): p = tmp_path / "server.yaml" p.write_text("port: 27015\nfoo: bar\n") spec = load_spec(p) assert spec.port == 27015 ``` - [ ] **Step 2: Run tests and verify failure** Run: `pytest components/l4d2-host-lib/tests/test_spec.py -q` Expected: FAIL. - [ ] **Step 3: Implement parser and dataclass** ```python from dataclasses import dataclass, field from pathlib import Path import yaml @dataclass class InstanceSpec: port: int overlays: list[str] = field(default_factory=list) arguments: list[str] = field(default_factory=list) config: list[str] = field(default_factory=list) def load_spec(path: Path) -> InstanceSpec: raw = yaml.safe_load(path.read_text()) or {} return InstanceSpec( port=int(raw["port"]), overlays=[str(x) for x in raw.get("overlays", [])], arguments=[str(x) for x in raw.get("arguments", [])], config=[str(x) for x in raw.get("config", [])], ) ``` - [ ] **Step 4: Run tests and verify pass** Run: `pytest components/l4d2-host-lib/tests/test_spec.py -q` Expected: PASS. - [ ] **Step 5: Commit parser** ```bash git add components/l4d2-host-lib/src/l4d2host/spec.py components/l4d2-host-lib/tests/test_spec.py git commit -m "feat(l4d2): add spec parser with required port and permissive fields" ``` ### Task 3: Build streaming process runner with callbacks **Files:** - Create: `components/l4d2-host-lib/src/l4d2host/process.py` - Test: `components/l4d2-host-lib/tests/test_process.py` - [ ] **Step 1: Write failing process tests** ```python import pytest from l4d2host.process import run_command def test_callbacks_receive_lines(): out, err = [], [] run_command(["python3", "-c", "import sys; print('ok'); print('warn', file=sys.stderr)"], on_stdout=out.append, on_stderr=err.append) assert out == ["ok"] assert err == ["warn"] def test_nonzero_exit_raises(): with pytest.raises(Exception): run_command(["python3", "-c", "import sys; sys.exit(7)"]) ``` - [ ] **Step 2: Run tests and verify failure** Run: `pytest components/l4d2-host-lib/tests/test_process.py -q` Expected: FAIL. - [ ] **Step 3: Implement process runner** ```python from dataclasses import dataclass import subprocess import threading import sys @dataclass class CommandResult: returncode: int stdout: str stderr: str def run_command(cmd, *, on_stdout=None, on_stderr=None, passthrough=False): stdout_lines, stderr_lines = [], [] proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1) def pump(stream, sink, callback, out_stream): for raw in iter(stream.readline, ""): line = raw.rstrip("\n") sink.append(line) if callback: callback(line) if passthrough: print(line, file=out_stream) t_out = threading.Thread(target=pump, args=(proc.stdout, stdout_lines, on_stdout, sys.stdout), daemon=True) t_err = threading.Thread(target=pump, args=(proc.stderr, stderr_lines, on_stderr, sys.stderr), daemon=True) t_out.start(); t_err.start() returncode = proc.wait() t_out.join(); t_err.join() result = CommandResult(returncode=returncode, stdout="\n".join(stdout_lines), stderr="\n".join(stderr_lines)) if returncode != 0: raise subprocess.CalledProcessError(returncode=returncode, cmd=cmd, output=result.stdout, stderr=result.stderr) return result ``` - [ ] **Step 4: Run tests and verify pass** Run: `pytest components/l4d2-host-lib/tests/test_process.py -q` Expected: PASS. - [ ] **Step 5: Commit process runner** ```bash git add components/l4d2-host-lib/src/l4d2host/process.py components/l4d2-host-lib/tests/test_process.py git commit -m "feat(l4d2): add callback-capable streaming process runner" ``` ### Task 4: Implement install command with callback passthrough **Files:** - Create: `components/l4d2-host-lib/src/l4d2host/steam_install.py` - Test: `components/l4d2-host-lib/tests/test_install.py` - [ ] **Step 1: Write failing install tests** ```python import subprocess import pytest from l4d2host.steam_install import SteamInstaller def test_windows_then_linux(monkeypatch): calls = [] monkeypatch.setattr("l4d2host.steam_install.run_command", lambda cmd, **kwargs: calls.append(cmd)) SteamInstaller().install_or_update() assert "windows" in calls[0] assert "linux" in calls[1] def test_fail_fast_on_first_failure(monkeypatch): calls = [] def fake(cmd, **kwargs): calls.append(cmd) if len(calls) == 1: raise subprocess.CalledProcessError(returncode=1, cmd=cmd) monkeypatch.setattr("l4d2host.steam_install.run_command", fake) with pytest.raises(subprocess.CalledProcessError): SteamInstaller().install_or_update() assert len(calls) == 1 ``` - [ ] **Step 2: Run tests and verify failure** Run: `pytest components/l4d2-host-lib/tests/test_install.py -q` Expected: FAIL. - [ ] **Step 3: Implement installer** ```python from pathlib import Path from l4d2host.process import run_command class SteamInstaller: def __init__(self, install_dir: Path = Path("/opt/l4d2/installation"), steamcmd: str = "steamcmd"): self.install_dir = install_dir self.steamcmd = steamcmd def install_or_update(self, *, on_stdout=None, on_stderr=None, passthrough=False) -> None: for platform in ("windows", "linux"): run_command([ self.steamcmd, "+force_install_dir", str(self.install_dir), "+login", "anonymous", "+@sSteamCmdForcePlatformType", platform, "+app_update", "222860", "validate", "+quit", ], on_stdout=on_stdout, on_stderr=on_stderr, passthrough=passthrough) ``` - [ ] **Step 4: Run tests and verify pass** Run: `pytest components/l4d2-host-lib/tests/test_install.py -q` Expected: PASS. - [ ] **Step 5: Commit install command** ```bash git add components/l4d2-host-lib/src/l4d2host/steam_install.py components/l4d2-host-lib/tests/test_install.py git commit -m "feat(l4d2): implement callback-aware install command" ``` ### Task 5: Implement initialize and systemd template management **Files:** - Create: `components/l4d2-host-lib/src/l4d2host/systemd_user.py` - Create: `components/l4d2-host-lib/src/l4d2host/instances.py` - Create: `components/l4d2-host-lib/src/l4d2host/templates/l4d2@.service` - Test: `components/l4d2-host-lib/tests/test_initialize.py` - [ ] **Step 1: Write failing initialize tests** ```python from pathlib import Path from l4d2host.instances import initialize_instance def test_initialize_writes_files(tmp_path: Path): spec = tmp_path / "spec.yaml" spec.write_text("port: 27015\noverlays: [a,b]\nconfig: [\"sv_consistency 1\"]\n") initialize_instance("alpha", spec, root=tmp_path) assert (tmp_path / "instances/alpha/instance.env").exists() assert (tmp_path / "instances/alpha/server.cfg").exists() def test_empty_config_writes_empty_server_cfg(tmp_path: Path): spec = tmp_path / "spec.yaml" spec.write_text("port: 27015\n") initialize_instance("alpha", spec, root=tmp_path) assert (tmp_path / "instances/alpha/server.cfg").read_text() == "" ``` - [ ] **Step 2: Run tests and verify failure** Run: `pytest components/l4d2-host-lib/tests/test_initialize.py -q` Expected: FAIL. - [ ] **Step 3: Implement initialize flow** ```python from pathlib import Path from l4d2host.spec import load_spec from l4d2host.systemd_user import ensure_template_unit, daemon_reload def initialize_instance(name: str, spec_path: Path, *, root: Path = Path("/opt/l4d2"), on_stdout=None, on_stderr=None) -> None: spec = load_spec(spec_path) instance_dir = root / "instances" / name runtime_dir = root / "runtime" / name instance_dir.mkdir(parents=True, exist_ok=True) (runtime_dir / "upper").mkdir(parents=True, exist_ok=True) (runtime_dir / "work").mkdir(parents=True, exist_ok=True) (runtime_dir / "merged").mkdir(parents=True, exist_ok=True) lowerdirs = [str(root / "overlays" / overlay) for overlay in spec.overlays] + [str(root / "installation")] (instance_dir / "instance.env").write_text( f"L4D2_PORT={spec.port}\n" f"L4D2_ARGS={' '.join(spec.arguments)}\n" f"L4D2_LOWERDIRS={':'.join(lowerdirs)}\n" ) (instance_dir / "server.cfg").write_text("\n".join(spec.config) + ("\n" if spec.config else "")) ensure_template_unit() daemon_reload(on_stdout=on_stdout, on_stderr=on_stderr) ``` - [ ] **Step 4: Run tests and verify pass** Run: `pytest components/l4d2-host-lib/tests/test_initialize.py -q` Expected: PASS. - [ ] **Step 5: Commit initialize logic** ```bash git add components/l4d2-host-lib/src/l4d2host/{instances.py,systemd_user.py,templates/l4d2@.service} components/l4d2-host-lib/tests/test_initialize.py git commit -m "feat(l4d2): implement initialize flow and systemd user template management" ``` ### Task 6: Implement start/stop/delete lifecycle commands **Files:** - Create: `components/l4d2-host-lib/src/l4d2host/fs/base.py` - Create: `components/l4d2-host-lib/src/l4d2host/fs/fuse_overlayfs.py` - Modify: `components/l4d2-host-lib/src/l4d2host/instances.py` - Test: `components/l4d2-host-lib/tests/test_lifecycle.py` - [ ] **Step 1: Write failing lifecycle tests** ```python from pathlib import Path from l4d2host.instances import start_instance, stop_instance, delete_instance def test_start_order(monkeypatch): calls = [] monkeypatch.setattr("l4d2host.instances.run_command", lambda cmd, **kwargs: calls.append(cmd)) start_instance("alpha") assert calls[0][0] == "fuse-overlayfs" assert calls[1][:3] == ["systemctl", "--user", "start"] def test_delete_missing_is_noop(tmp_path: Path): delete_instance("missing", root=tmp_path) assert True ``` - [ ] **Step 2: Run tests and verify failure** Run: `pytest components/l4d2-host-lib/tests/test_lifecycle.py -q` Expected: FAIL. - [ ] **Step 3: Implement lifecycle methods with callbacks** ```python def start_instance(name: str, *, root: Path = Path("/opt/l4d2"), on_stdout=None, on_stderr=None, passthrough=False) -> None: instance_dir = root / "instances" / name runtime_dir = root / "runtime" / name env = { line.split("=", 1)[0]: line.split("=", 1)[1] for line in (instance_dir / "instance.env").read_text().splitlines() if "=" in line } run_command([ "fuse-overlayfs", "-o", f"lowerdir={env['L4D2_LOWERDIRS']},upperdir={runtime_dir/'upper'},workdir={runtime_dir/'work'}", str(runtime_dir / "merged"), ], on_stdout=on_stdout, on_stderr=on_stderr, passthrough=passthrough) shutil.copy2(instance_dir / "server.cfg", runtime_dir / "merged" / "left4dead2/cfg/server.cfg") run_command([ "systemctl", "--user", "start", f"l4d2@{name}.service" ], on_stdout=on_stdout, on_stderr=on_stderr, passthrough=passthrough) def stop_instance(name: str, *, root: Path = Path("/opt/l4d2"), on_stdout=None, on_stderr=None, passthrough=False) -> None: run_command([ "systemctl", "--user", "stop", f"l4d2@{name}.service" ], on_stdout=on_stdout, on_stderr=on_stderr, passthrough=passthrough) run_command([ "fusermount3", "-u", str(root / "runtime" / name / "merged") ], on_stdout=on_stdout, on_stderr=on_stderr, passthrough=passthrough) def delete_instance(name: str, *, root: Path = Path("/opt/l4d2"), on_stdout=None, on_stderr=None, passthrough=False) -> None: instance_dir = root / "instances" / name runtime_dir = root / "runtime" / name if not instance_dir.exists() and not runtime_dir.exists(): return stop_instance(name, root=root, on_stdout=on_stdout, on_stderr=on_stderr, passthrough=passthrough) if instance_dir.exists(): shutil.rmtree(instance_dir) if runtime_dir.exists(): shutil.rmtree(runtime_dir) ``` - [ ] **Step 4: Run tests and verify pass** Run: `pytest components/l4d2-host-lib/tests/test_lifecycle.py -q` Expected: PASS. - [ ] **Step 5: Commit lifecycle implementation** ```bash git add components/l4d2-host-lib/src/l4d2host/{fs/base.py,fs/fuse_overlayfs.py,instances.py} components/l4d2-host-lib/tests/test_lifecycle.py git commit -m "feat(l4d2): implement start stop delete lifecycle with callback support" ``` ### Task 7: Add status and log read APIs **Files:** - Create: `components/l4d2-host-lib/src/l4d2host/status.py` - Create: `components/l4d2-host-lib/src/l4d2host/logs.py` - Test: `components/l4d2-host-lib/tests/test_status.py` - Test: `components/l4d2-host-lib/tests/test_logs.py` - [ ] **Step 1: Write failing read-API tests** ```python from l4d2host.status import map_active_state def test_status_mapping(): assert map_active_state("active") == "running" assert map_active_state("inactive") == "stopped" assert map_active_state("weird") == "unknown" ``` - [ ] **Step 2: Run tests and verify failure** Run: `pytest components/l4d2-host-lib/tests/test_status.py components/l4d2-host-lib/tests/test_logs.py -q` Expected: FAIL. - [ ] **Step 3: Implement status/log readers** ```python from dataclasses import dataclass import subprocess from l4d2host.process import run_command @dataclass class InstanceStatus: state: str raw_active_state: str raw_sub_state: str def get_instance_status(name: str) -> InstanceStatus: try: result = run_command([ "systemctl", "--user", "show", f"l4d2@{name}.service", "--property=ActiveState", "--property=SubState", "--no-pager", ]) except subprocess.CalledProcessError: return InstanceStatus(state="unknown", raw_active_state="unknown", raw_sub_state="unknown") pairs = {} for line in result.stdout.splitlines(): if "=" in line: k, v = line.split("=", 1) pairs[k] = v active = pairs.get("ActiveState", "unknown") sub = pairs.get("SubState", "unknown") if active == "active": mapped = "running" elif active in {"inactive", "failed"}: mapped = "stopped" else: mapped = "unknown" return InstanceStatus(state=mapped, raw_active_state=active, raw_sub_state=sub) def stream_instance_logs(name: str, *, lines: int = 200, follow: bool = True): cmd = [ "journalctl", "--user", "-u", f"l4d2@{name}.service", "-n", str(lines), "-o", "cat", ] if follow: cmd.append("-f") proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1) try: for raw in iter(proc.stdout.readline, ""): yield raw.rstrip("\n") finally: if proc.poll() is None: proc.terminate() proc.wait(timeout=2) ``` - [ ] **Step 4: Run tests and verify pass** Run: `pytest components/l4d2-host-lib/tests/test_status.py components/l4d2-host-lib/tests/test_logs.py -q` Expected: PASS. - [ ] **Step 5: Commit read APIs** ```bash git add components/l4d2-host-lib/src/l4d2host/{status.py,logs.py} components/l4d2-host-lib/tests/test_{status,logs}.py git commit -m "feat(l4d2): add status and journald log read APIs" ``` ### Task 8: Wire CLI to implementations and finalize docs **Files:** - Modify: `components/l4d2-host-lib/src/l4d2host/cli.py` - Create: `components/l4d2-host-lib/README.md` - Modify: tests under `components/l4d2-host-lib/tests/` - [ ] **Step 1: Write failing CLI exit-code test** ```python from typer.testing import CliRunner from l4d2host.cli import app def test_cli_propagates_subprocess_return_code(monkeypatch): def fail(*args, **kwargs): import subprocess raise subprocess.CalledProcessError(returncode=9, cmd=["x"]) monkeypatch.setattr("l4d2host.cli.start_instance", fail) r = CliRunner().invoke(app, ["start", "alpha"]) assert r.exit_code == 9 ``` - [ ] **Step 2: Run tests and verify failure** Run: `pytest components/l4d2-host-lib/tests/test_cli.py -q` Expected: FAIL. - [ ] **Step 3: Implement CLI handlers** ```python import subprocess import typer def _exit_from_subprocess_error(exc: subprocess.CalledProcessError) -> None: raise typer.Exit(code=exc.returncode) @app.command() def start(name: str) -> None: try: start_instance(name, passthrough=True) except subprocess.CalledProcessError as exc: _exit_from_subprocess_error(exc) ``` - [ ] **Step 4: Final test run** Run: `pytest components/l4d2-host-lib/tests -q` Expected: PASS. - [ ] **Step 5: Commit finalization** ```bash git add components/l4d2-host-lib git commit -m "docs(l4d2): finalize v1 CLI contracts and web-facing read APIs" ``` --- ## Self-Review - [ ] Spec coverage: command surface fixed, hard-coded paths, config semantics, delete no-op, callback streaming, read APIs. - [ ] Placeholder scan: no TODO/TBD placeholders. - [ ] Consistency: argument names (`on_stdout`, `on_stderr`, `passthrough`) are consistent across tasks. - [ ] Verification: each task contains exact test commands and expected outcomes.