# 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 fixed write commands (`install`, `initialize`, `start`, `stop`, `delete`) plus read commands/APIs needed by the 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 write behavior remains raw/stderr-first, and CLI read commands expose status/log output for the web/host boundary while library internals keep callback-based streaming and read APIs.

**Tech Stack:** Python 3.12+, Typer, PyYAML, pytest, subprocess, systemd user units, fuse-overlayfs.

---

## Scope and Contracts

- Write 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.
- Read commands are allowed for web/host boundary consistency:
  - `l4d2ctl status <name> --json`
  - `l4d2ctl logs <name> --lines <n> --follow/--no-follow`
- Additional host-local read APIs:
  - `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

- `l4d2host/pyproject.toml`
- `l4d2host/cli.py`
- `l4d2host/spec.py`
- `l4d2host/process.py`
- `l4d2host/steam_install.py`
- `l4d2host/systemd_user.py`
- `l4d2host/fs/base.py`
- `l4d2host/fs/fuse_overlayfs.py`
- `l4d2host/instances.py`
- `l4d2host/status.py`
- `l4d2host/logs.py`
- `l4d2host/templates/l4d2@.service`
- `l4d2host/tests/test_cli.py`
- `l4d2host/tests/test_spec.py`
- `l4d2host/tests/test_process.py`
- `l4d2host/tests/test_install.py`
- `l4d2host/tests/test_initialize.py`
- `l4d2host/tests/test_lifecycle.py`
- `l4d2host/tests/test_status.py`
- `l4d2host/tests/test_logs.py`
- `l4d2host/README.md`

### Task 1: Scaffold package and CLI entrypoint

**Files:**
- Create: `l4d2host/pyproject.toml`
- Create: `l4d2host/__init__.py`
- Create: `l4d2host/cli.py`
- Test: `l4d2host/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 l4d2host/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 l4d2host/tests/test_cli.py -q`
Expected: PASS.

- [ ] **Step 5: Commit scaffold**

```bash
git add l4d2host
git commit -m "feat(l4d2): scaffold package and v1 CLI entrypoint"
```

### Task 2: Implement YAML spec parser

**Files:**
- Create: `l4d2host/spec.py`
- Test: `l4d2host/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 l4d2host/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 l4d2host/tests/test_spec.py -q`
Expected: PASS.

- [ ] **Step 5: Commit parser**

```bash
git add l4d2host/spec.py l4d2host/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: `l4d2host/process.py`
- Test: `l4d2host/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 l4d2host/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 l4d2host/tests/test_process.py -q`
Expected: PASS.

- [ ] **Step 5: Commit process runner**

```bash
git add l4d2host/process.py l4d2host/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: `l4d2host/steam_install.py`
- Test: `l4d2host/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 l4d2host/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 l4d2host/tests/test_install.py -q`
Expected: PASS.

- [ ] **Step 5: Commit install command**

```bash
git add l4d2host/steam_install.py l4d2host/tests/test_install.py
git commit -m "feat(l4d2): implement callback-aware install command"
```

### Task 5: Implement initialize and systemd template management

**Files:**
- Create: `l4d2host/systemd_user.py`
- Create: `l4d2host/instances.py`
- Create: `l4d2host/templates/l4d2@.service`
- Test: `l4d2host/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 l4d2host/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 l4d2host/tests/test_initialize.py -q`
Expected: PASS.

- [ ] **Step 5: Commit initialize logic**

```bash
git add l4d2host/{instances.py,systemd_user.py,templates/l4d2@.service} l4d2host/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: `l4d2host/fs/base.py`
- Create: `l4d2host/fs/fuse_overlayfs.py`
- Modify: `l4d2host/instances.py`
- Test: `l4d2host/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 l4d2host/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 l4d2host/tests/test_lifecycle.py -q`
Expected: PASS.

- [ ] **Step 5: Commit lifecycle implementation**

```bash
git add l4d2host/{fs/base.py,fs/fuse_overlayfs.py,instances.py} l4d2host/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: `l4d2host/status.py`
- Create: `l4d2host/logs.py`
- Test: `l4d2host/tests/test_status.py`
- Test: `l4d2host/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 l4d2host/tests/test_status.py l4d2host/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 l4d2host/tests/test_status.py l4d2host/tests/test_logs.py -q`
Expected: PASS.

- [ ] **Step 5: Commit read APIs**

```bash
git add l4d2host/{status.py,logs.py} l4d2host/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: `l4d2host/cli.py`
- Create: `l4d2host/README.md`
- Modify: tests under `l4d2host/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 l4d2host/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 l4d2host/tests -q`
Expected: PASS.

- [ ] **Step 5: Commit finalization**

```bash
git add l4d2host
git commit -m "docs(l4d2): finalize v1 CLI contracts and web-facing read APIs"
```

---

## Self-Review

- [ ] Spec coverage: write command surface fixed, read commands allowed, 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.
