left4me/docs/superpowers/plans/2026-04-22-l4d2-host-lib-v1.md

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 <name> -f <spec.yaml>
  • l4d2ctl start <name>
  • l4d2ctl stop <name>
  • l4d2ctl delete <name>
• CLI <name> is the server identity source (YAML does not carry identity semantics).
• Hard-coded runtime paths:
  • /opt/l4d2/installation
  • /opt/l4d2/overlays/<overlay>
  • /opt/l4d2/instances/<name>
  • /opt/l4d2/runtime/<name>/{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

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

[project.scripts]
l4d2ctl = "l4d2host.cli:app"

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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.