chore: ignore local sqlite runtime artifacts

.gitignore

@@ -0,0 +1,2 @@
+.worktrees/
+l4d2web.db*

components/l4d2-host-lib/README.md

@@ -0,0 +1,31 @@
+# l4d2-host-lib
+
+Python host library and CLI for managing L4D2 instances.
+
+## CLI
+
+`l4d2ctl` exposes exactly these commands in v1:
+
+- `install`
+- `initialize <name> -f <spec.yaml>`
+- `start <name>`
+- `stop <name>`
+- `delete <name>`
+
+Subprocess failures are fail-fast. Raw stderr is written to stderr and the command exits with the same subprocess return code.
+
+## Runtime Paths
+
+The host library uses hard-coded runtime paths under `/opt/l4d2`:
+
+- `/opt/l4d2/installation`
+- `/opt/l4d2/overlays/<overlay>`
+- `/opt/l4d2/instances/<name>`
+- `/opt/l4d2/runtime/<name>/{upper,work,merged}`
+
+## Web App Read APIs
+
+These read APIs are provided for web app integration:
+
+- `get_instance_status(name)`
+- `stream_instance_logs(name, lines=200, follow=True)`

components/l4d2-host-lib/pyproject.toml

@@ -0,0 +1,23 @@
+[build-system]
+requires = ["setuptools>=68", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "l4d2host"
+version = "0.1.0"
+description = "L4D2 host library and CLI"
+readme = "README.md"
+requires-python = ">=3.12"
+dependencies = [
+  "typer>=0.12",
+  "PyYAML>=6.0",
+]
+
+[project.scripts]
+l4d2ctl = "l4d2host.cli:app"
+
+[tool.setuptools]
+package-dir = {"" = "src"}
+
+[tool.setuptools.packages.find]
+where = ["src"]

components/l4d2-host-lib/src/l4d2host/__init__.py

@@ -0,0 +1,3 @@
+__all__ = ["__version__"]
+
+__version__ = "0.1.0"

components/l4d2-host-lib/src/l4d2host/cli.py

@@ -0,0 +1,56 @@
+from pathlib import Path
+import subprocess
+
+import typer
+
+from l4d2host.instances import delete_instance, initialize_instance, start_instance, stop_instance
+from l4d2host.steam_install import SteamInstaller
+
+
+app = typer.Typer(no_args_is_help=True)
+
+
+def _exit_from_subprocess_error(exc: subprocess.CalledProcessError) -> None:
+    if exc.stderr:
+        typer.echo(exc.stderr, err=True)
+    raise typer.Exit(code=exc.returncode)
+
+
+@app.command()
+def install() -> None:
+    try:
+        SteamInstaller().install_or_update(passthrough=True)
+    except subprocess.CalledProcessError as exc:
+        _exit_from_subprocess_error(exc)
+
+
+@app.command()
+def initialize(name: str, spec: Path = typer.Option(..., "-f")) -> None:
+    try:
+        initialize_instance(name, spec, passthrough=True)
+    except subprocess.CalledProcessError as exc:
+        _exit_from_subprocess_error(exc)
+
+
+@app.command()
+def start(name: str) -> None:
+    try:
+        start_instance(name, passthrough=True)
+    except subprocess.CalledProcessError as exc:
+        _exit_from_subprocess_error(exc)
+
+
+@app.command()
+def stop(name: str) -> None:
+    try:
+        stop_instance(name, passthrough=True)
+    except subprocess.CalledProcessError as exc:
+        _exit_from_subprocess_error(exc)
+
+
+@app.command()
+def delete(name: str) -> None:
+    try:
+        delete_instance(name, passthrough=True)
+    except subprocess.CalledProcessError as exc:
+        _exit_from_subprocess_error(exc)

components/l4d2-host-lib/src/l4d2host/fs/base.py

@@ -0,0 +1,30 @@
+from abc import ABC, abstractmethod
+from pathlib import Path
+from typing import Callable
+
+
+class OverlayMounter(ABC):
+    @abstractmethod
+    def mount(
+        self,
+        *,
+        lowerdirs: str,
+        upperdir: Path,
+        workdir: Path,
+        merged: Path,
+        on_stdout: Callable[[str], None] | None = None,
+        on_stderr: Callable[[str], None] | None = None,
+        passthrough: bool = False,
+    ) -> None:
+        raise NotImplementedError
+
+    @abstractmethod
+    def unmount(
+        self,
+        *,
+        merged: Path,
+        on_stdout: Callable[[str], None] | None = None,
+        on_stderr: Callable[[str], None] | None = None,
+        passthrough: bool = False,
+    ) -> None:
+        raise NotImplementedError

components/l4d2-host-lib/src/l4d2host/fs/fuse_overlayfs.py

@@ -0,0 +1,45 @@
+from pathlib import Path
+from typing import Callable
+
+from l4d2host.fs.base import OverlayMounter
+from l4d2host.process import run_command
+
+
+class FuseOverlayFSMounter(OverlayMounter):
+    def mount(
+        self,
+        *,
+        lowerdirs: str,
+        upperdir: Path,
+        workdir: Path,
+        merged: Path,
+        on_stdout: Callable[[str], None] | None = None,
+        on_stderr: Callable[[str], None] | None = None,
+        passthrough: bool = False,
+    ) -> None:
+        run_command(
+            [
+                "fuse-overlayfs",
+                "-o",
+                f"lowerdir={lowerdirs},upperdir={upperdir},workdir={workdir}",
+                str(merged),
+            ],
+            on_stdout=on_stdout,
+            on_stderr=on_stderr,
+            passthrough=passthrough,
+        )
+
+    def unmount(
+        self,
+        *,
+        merged: Path,
+        on_stdout: Callable[[str], None] | None = None,
+        on_stderr: Callable[[str], None] | None = None,
+        passthrough: bool = False,
+    ) -> None:
+        run_command(
+            ["fusermount3", "-u", str(merged)],
+            on_stdout=on_stdout,
+            on_stderr=on_stderr,
+            passthrough=passthrough,
+        )

components/l4d2-host-lib/src/l4d2host/instances.py

@@ -0,0 +1,149 @@
+from pathlib import Path
+import shutil
+from typing import Callable
+
+from l4d2host.process import run_command
+from l4d2host.spec import load_spec
+from l4d2host.systemd_user import daemon_reload, ensure_template_unit
+
+
+DEFAULT_ROOT = Path("/opt/l4d2")
+
+
+def initialize_instance(
+    name: str,
+    spec_path: Path,
+    *,
+    root: Path = DEFAULT_ROOT,
+    on_stdout: Callable[[str], None] | None = None,
+    on_stderr: Callable[[str], None] | None = None,
+    passthrough: bool = False,
+) -> None:
+    spec = load_spec(spec_path)
+
+    instance_dir = root / "instances" / name
+    runtime_dir = root / "runtime" / name
+    (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)
+    instance_dir.mkdir(parents=True, exist_ok=True)
+
+    lowerdirs = [str(root / "overlays" / overlay) for overlay in spec.overlays]
+    lowerdirs.append(str(root / "installation"))
+
+    instance_env = "\n".join(
+        [
+            f"L4D2_PORT={spec.port}",
+            f"L4D2_ARGS={' '.join(spec.arguments)}",
+            f"L4D2_LOWERDIRS={':'.join(lowerdirs)}",
+        ]
+    ) + "\n"
+    (instance_dir / "instance.env").write_text(instance_env)
+
+    server_cfg = "\n".join(spec.config) if spec.config else ""
+    (instance_dir / "server.cfg").write_text(server_cfg)
+
+    if root.resolve() == DEFAULT_ROOT:
+        ensure_template_unit()
+        daemon_reload(on_stdout=on_stdout, on_stderr=on_stderr, passthrough=passthrough)
+
+
+def _load_instance_env(path: Path) -> dict[str, str]:
+    result: dict[str, str] = {}
+    for line in path.read_text().splitlines():
+        if "=" not in line:
+            continue
+        key, value = line.split("=", 1)
+        result[key] = value
+    return result
+
+
+def start_instance(
+    name: str,
+    *,
+    root: Path = DEFAULT_ROOT,
+    on_stdout: Callable[[str], None] | None = None,
+    on_stderr: Callable[[str], None] | None = None,
+    passthrough: bool = False,
+) -> None:
+    instance_dir = root / "instances" / name
+    runtime_dir = root / "runtime" / name
+
+    env = _load_instance_env(instance_dir / "instance.env")
+
+    run_command(
+        [
+            "fuse-overlayfs",
+            "-o",
+            (
+                f"lowerdir={env['L4D2_LOWERDIRS']},"
+                f"upperdir={runtime_dir / 'upper'},"
+                f"workdir={runtime_dir / 'work'}"
+            ),
+            str(runtime_dir / "merged"),
+        ],
+        on_stdout=on_stdout,
+        on_stderr=on_stderr,
+        passthrough=passthrough,
+    )
+
+    target_cfg = runtime_dir / "merged" / "left4dead2" / "cfg" / "server.cfg"
+    target_cfg.parent.mkdir(parents=True, exist_ok=True)
+    shutil.copy2(instance_dir / "server.cfg", target_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 = DEFAULT_ROOT,
+    on_stdout: Callable[[str], None] | None = None,
+    on_stderr: Callable[[str], None] | None = None,
+    passthrough: bool = 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 = DEFAULT_ROOT,
+    on_stdout: Callable[[str], None] | None = None,
+    on_stderr: Callable[[str], None] | None = None,
+    passthrough: bool = 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)

components/l4d2-host-lib/src/l4d2host/logs.py

@@ -0,0 +1,34 @@
+import subprocess
+from typing import Iterator
+
+
+def stream_instance_logs(name: str, *, lines: int = 200, follow: bool = True) -> Iterator[str]:
+    command = [
+        "journalctl",
+        "--user",
+        "-u",
+        f"l4d2@{name}.service",
+        "-n",
+        str(lines),
+        "-o",
+        "cat",
+    ]
+    if follow:
+        command.append("-f")
+
+    proc = subprocess.Popen(
+        command,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        text=True,
+        bufsize=1,
+    )
+    try:
+        if proc.stdout is None:
+            return
+        for raw in iter(proc.stdout.readline, ""):
+            yield raw.rstrip("\n")
+    finally:
+        if proc.poll() is None:
+            proc.terminate()
+            proc.wait(timeout=2)

components/l4d2-host-lib/src/l4d2host/process.py

@@ -0,0 +1,79 @@
+from dataclasses import dataclass
+import subprocess
+import sys
+import threading
+from typing import Callable, Sequence
+
+
+@dataclass(slots=True)
+class CommandResult:
+    returncode: int
+    stdout: str
+    stderr: str
+
+
+def run_command(
+    cmd: Sequence[str],
+    *,
+    on_stdout: Callable[[str], None] | None = None,
+    on_stderr: Callable[[str], None] | None = None,
+    passthrough: bool = False,
+) -> CommandResult:
+    stdout_lines: list[str] = []
+    stderr_lines: list[str] = []
+
+    proc = subprocess.Popen(
+        list(cmd),
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        text=True,
+        bufsize=1,
+    )
+
+    def pump(
+        stream: subprocess.Popen[str].stdout,
+        sink: list[str],
+        callback: Callable[[str], None] | None,
+        output_stream,
+    ) -> None:
+        if stream is None:
+            return
+        for raw in iter(stream.readline, ""):
+            line = raw.rstrip("\n")
+            sink.append(line)
+            if callback is not None:
+                callback(line)
+            if passthrough:
+                print(line, file=output_stream)
+        stream.close()
+
+    stdout_thread = threading.Thread(
+        target=pump,
+        args=(proc.stdout, stdout_lines, on_stdout, sys.stdout),
+        daemon=True,
+    )
+    stderr_thread = threading.Thread(
+        target=pump,
+        args=(proc.stderr, stderr_lines, on_stderr, sys.stderr),
+        daemon=True,
+    )
+    stdout_thread.start()
+    stderr_thread.start()
+
+    returncode = proc.wait()
+    stdout_thread.join()
+    stderr_thread.join()
+
+    result = CommandResult(
+        returncode=returncode,
+        stdout="\n".join(stdout_lines),
+        stderr="\n".join(stderr_lines),
+    )
+    if returncode != 0:
+        raise subprocess.CalledProcessError(
+            returncode=returncode,
+            cmd=list(cmd),
+            output=result.stdout,
+            stderr=result.stderr,
+        )
+    return result

components/l4d2-host-lib/src/l4d2host/spec.py

@@ -0,0 +1,22 @@
+from dataclasses import dataclass, field
+from pathlib import Path
+
+import yaml
+
+
+@dataclass(slots=True)
+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(item) for item in raw.get("overlays", [])],
+        arguments=[str(item) for item in raw.get("arguments", [])],
+        config=[str(item) for item in raw.get("config", [])],
+    )

components/l4d2-host-lib/src/l4d2host/status.py

@@ -0,0 +1,56 @@
+from dataclasses import dataclass
+import subprocess
+
+from l4d2host.process import run_command
+
+
+@dataclass(slots=True)
+class InstanceStatus:
+    state: str
+    raw_active_state: str
+    raw_sub_state: str
+
+
+def map_active_state(active_state: str) -> str:
+    if active_state == "active":
+        return "running"
+    if active_state in {"inactive", "failed"}:
+        return "stopped"
+    return "unknown"
+
+
+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, FileNotFoundError):
+        return InstanceStatus(
+            state="unknown",
+            raw_active_state="unknown",
+            raw_sub_state="unknown",
+        )
+
+    values: dict[str, str] = {}
+    for line in result.stdout.splitlines():
+        if "=" not in line:
+            continue
+        key, value = line.split("=", 1)
+        values[key] = value
+
+    active_state = values.get("ActiveState", "unknown")
+    sub_state = values.get("SubState", "unknown")
+
+    return InstanceStatus(
+        state=map_active_state(active_state),
+        raw_active_state=active_state,
+        raw_sub_state=sub_state,
+    )

components/l4d2-host-lib/src/l4d2host/steam_install.py

@@ -0,0 +1,37 @@
+from pathlib import Path
+from typing import Callable
+
+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: Callable[[str], None] | None = None,
+        on_stderr: Callable[[str], None] | None = None,
+        passthrough: bool = 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,
+            )

components/l4d2-host-lib/src/l4d2host/systemd_user.py

@@ -0,0 +1,29 @@
+from importlib import resources
+from pathlib import Path
+from typing import Callable
+
+from l4d2host.process import run_command
+
+
+def ensure_template_unit(target_dir: Path | None = None) -> Path:
+    if target_dir is None:
+        target_dir = Path.home() / ".config/systemd/user"
+    target_dir.mkdir(parents=True, exist_ok=True)
+    target_file = target_dir / "l4d2@.service"
+    body = resources.files("l4d2host.templates").joinpath("l4d2@.service").read_text()
+    target_file.write_text(body)
+    return target_file
+
+
+def daemon_reload(
+    *,
+    on_stdout: Callable[[str], None] | None = None,
+    on_stderr: Callable[[str], None] | None = None,
+    passthrough: bool = False,
+) -> None:
+    run_command(
+        ["systemctl", "--user", "daemon-reload"],
+        on_stdout=on_stdout,
+        on_stderr=on_stderr,
+        passthrough=passthrough,
+    )

components/l4d2-host-lib/src/l4d2host/templates/l4d2@.service

@@ -0,0 +1,14 @@
+[Unit]
+Description=L4D2 dedicated server instance %i
+After=network.target
+
+[Service]
+Type=simple
+EnvironmentFile=/opt/l4d2/instances/%i/instance.env
+WorkingDirectory=/opt/l4d2/runtime/%i/merged/left4dead2
+ExecStart=/opt/l4d2/installation/srcds_run -game left4dead2 +hostport ${L4D2_PORT} ${L4D2_ARGS}
+Restart=on-failure
+RestartSec=5
+
+[Install]
+WantedBy=default.target

components/l4d2-host-lib/tests/test_cli.py

@@ -0,0 +1,26 @@
+import subprocess
+
+from typer.testing import CliRunner
+
+from l4d2host.cli import app
+
+
+def test_help_lists_v1_commands() -> None:
+    result = CliRunner().invoke(app, ["--help"])
+    assert result.exit_code == 0
+    for command in ["install", "initialize", "start", "stop", "delete"]:
+        assert command in result.output
+
+
+def test_cli_propagates_subprocess_return_code(monkeypatch) -> None:
+    def fail(*args, **kwargs):
+        del args
+        del kwargs
+        raise subprocess.CalledProcessError(returncode=9, cmd=["x"], stderr="boom")
+
+    monkeypatch.setattr("l4d2host.cli.start_instance", fail)
+
+    result = CliRunner().invoke(app, ["start", "alpha"])
+
+    assert result.exit_code == 9
+    assert "boom" in result.stderr

components/l4d2-host-lib/tests/test_initialize.py

@@ -0,0 +1,22 @@
+from pathlib import Path
+
+from l4d2host.instances import initialize_instance
+
+
+def test_initialize_writes_files(tmp_path: Path) -> None:
+    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) -> None:
+    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() == ""

components/l4d2-host-lib/tests/test_install.py

@@ -0,0 +1,35 @@
+import subprocess
+
+import pytest
+
+from l4d2host.steam_install import SteamInstaller
+
+
+def test_windows_then_linux(monkeypatch: pytest.MonkeyPatch) -> None:
+    calls: list[list[str]] = []
+
+    def fake_run_command(cmd, **kwargs):
+        del kwargs
+        calls.append(list(cmd))
+
+    monkeypatch.setattr("l4d2host.steam_install.run_command", fake_run_command)
+    SteamInstaller().install_or_update()
+    assert "windows" in calls[0]
+    assert "linux" in calls[1]
+
+
+def test_fail_fast_on_first_failure(monkeypatch: pytest.MonkeyPatch) -> None:
+    calls: list[list[str]] = []
+
+    def fake_run_command(cmd, **kwargs):
+        del kwargs
+        calls.append(list(cmd))
+        if len(calls) == 1:
+            raise subprocess.CalledProcessError(returncode=1, cmd=cmd)
+
+    monkeypatch.setattr("l4d2host.steam_install.run_command", fake_run_command)
+
+    with pytest.raises(subprocess.CalledProcessError):
+        SteamInstaller().install_or_update()
+
+    assert len(calls) == 1

components/l4d2-host-lib/tests/test_lifecycle.py

@@ -0,0 +1,33 @@
+from pathlib import Path
+
+import pytest
+
+from l4d2host.instances import delete_instance, start_instance
+
+
+def test_start_order(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    calls: list[list[str]] = []
+
+    def fake_run_command(cmd, **kwargs):
+        del kwargs
+        calls.append(list(cmd))
+
+    instance_dir = tmp_path / "instances" / "alpha"
+    runtime_dir = tmp_path / "runtime" / "alpha"
+    (runtime_dir / "merged" / "left4dead2" / "cfg").mkdir(parents=True, exist_ok=True)
+    instance_dir.mkdir(parents=True, exist_ok=True)
+    (instance_dir / "instance.env").write_text(
+        "L4D2_PORT=27015\nL4D2_ARGS=-tickrate 100\nL4D2_LOWERDIRS=/x:/y\n"
+    )
+    (instance_dir / "server.cfg").write_text("sv_consistency 1")
+
+    monkeypatch.setattr("l4d2host.instances.run_command", fake_run_command)
+
+    start_instance("alpha", root=tmp_path)
+
+    assert calls[0][0] == "fuse-overlayfs"
+    assert calls[1][:3] == ["systemctl", "--user", "start"]
+
+
+def test_delete_missing_is_noop(tmp_path: Path) -> None:
+    delete_instance("missing", root=tmp_path)

components/l4d2-host-lib/tests/test_logs.py

@@ -0,0 +1,42 @@
+from types import SimpleNamespace
+
+import pytest
+
+from l4d2host.logs import stream_instance_logs
+
+
+class DummyProcess:
+    def __init__(self, lines: list[str]) -> None:
+        self.stdout = SimpleNamespace(readline=self._readline)
+        self.stderr = SimpleNamespace(readline=lambda: "")
+        self._lines = iter(lines)
+        self.terminated = False
+        self.waited = False
+
+    def _readline(self) -> str:
+        return next(self._lines, "")
+
+    def poll(self):
+        return None if not self.waited else 0
+
+    def terminate(self) -> None:
+        self.terminated = True
+
+    def wait(self, timeout: int) -> None:
+        del timeout
+        self.waited = True
+
+
+def test_stream_instance_logs_yields_lines(monkeypatch: pytest.MonkeyPatch) -> None:
+    proc = DummyProcess(["line1\n", "line2\n", ""])
+
+    def fake_popen(cmd, **kwargs):
+        del cmd
+        del kwargs
+        return proc
+
+    monkeypatch.setattr("l4d2host.logs.subprocess.Popen", fake_popen)
+
+    lines = list(stream_instance_logs("alpha", lines=10, follow=False))
+    assert lines == ["line1", "line2"]
+    assert proc.terminated is True

components/l4d2-host-lib/tests/test_process.py

@@ -0,0 +1,22 @@
+import subprocess
+
+import pytest
+
+from l4d2host.process import run_command
+
+
+def test_callbacks_receive_lines() -> None:
+    out: list[str] = []
+    err: list[str] = []
+    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() -> None:
+    with pytest.raises(subprocess.CalledProcessError):
+        run_command(["python3", "-c", "import sys; sys.exit(7)"])

components/l4d2-host-lib/tests/test_spec.py

@@ -0,0 +1,36 @@
+from pathlib import Path
+
+import pytest
+
+from l4d2host.spec import load_spec
+
+
+def test_minimal_spec_parses(tmp_path: Path) -> None:
+    path = tmp_path / "server.yaml"
+    path.write_text("port: 27015\noverlays: [standard]\n")
+    spec = load_spec(path)
+    assert spec.port == 27015
+    assert spec.overlays == ["standard"]
+
+
+def test_defaults_are_empty_lists(tmp_path: Path) -> None:
+    path = tmp_path / "server.yaml"
+    path.write_text("port: 27015\n")
+    spec = load_spec(path)
+    assert spec.overlays == []
+    assert spec.arguments == []
+    assert spec.config == []
+
+
+def test_missing_port_fails(tmp_path: Path) -> None:
+    path = tmp_path / "server.yaml"
+    path.write_text("overlays: [standard]\n")
+    with pytest.raises((KeyError, ValueError, TypeError)):
+        load_spec(path)
+
+
+def test_unknown_keys_ignored(tmp_path: Path) -> None:
+    path = tmp_path / "server.yaml"
+    path.write_text("port: 27015\nfoo: bar\n")
+    spec = load_spec(path)
+    assert spec.port == 27015

components/l4d2-host-lib/tests/test_status.py

@@ -0,0 +1,8 @@
+from l4d2host.status import map_active_state
+
+
+def test_status_mapping() -> None:
+    assert map_active_state("active") == "running"
+    assert map_active_state("inactive") == "stopped"
+    assert map_active_state("failed") == "stopped"
+    assert map_active_state("weird") == "unknown"

components/l4d2-web-app/README.md

@@ -0,0 +1,28 @@
+# l4d2-web-app
+
+Flask web app for managing L4D2 servers through user-private blueprints.
+
+## Key v1 behaviors
+
+- Public signup/login with local username/password
+- Admin-managed overlay catalog
+- Private blueprints per user
+- Server creation from blueprints (live-linked; no per-server blueprint overrides)
+- Async job model with persisted command logs in `job_logs`
+- Desired vs actual state model
+- Live logs for jobs and servers via SSE endpoints
+
+## Frontend constraints
+
+- Server-rendered templates (Jinja)
+- Vendored HTMX (`static/vendor/htmx.min.js`)
+- Custom CSS only
+- Consistent link color: `#0F766E`
+
+## Development
+
+```bash
+python3 -m venv .venv
+.venv/bin/pip install -e .
+.venv/bin/pytest tests -q
+```

components/l4d2-web-app/alembic.ini

@@ -0,0 +1,35 @@
+[alembic]
+script_location = alembic
+sqlalchemy.url = sqlite:///l4d2web.db
+
+[loggers]
+keys = root,sqlalchemy,alembic
+
+[handlers]
+keys = console
+
+[formatters]
+keys = generic
+
+[logger_root]
+level = WARN
+handlers = console
+
+[logger_sqlalchemy]
+level = WARN
+handlers =
+qualname = sqlalchemy.engine
+
+[logger_alembic]
+level = INFO
+handlers =
+qualname = alembic
+
+[handler_console]
+class = StreamHandler
+args = (sys.stderr,)
+level = NOTSET
+formatter = generic
+
+[formatter_generic]
+format = %(levelname)-5.5s [%(name)s] %(message)s

components/l4d2-web-app/alembic/env.py

@@ -0,0 +1,53 @@
+from logging.config import fileConfig
+import os
+
+from alembic import context
+from sqlalchemy import engine_from_config, pool
+
+from l4d2web.models import Base
+
+
+config = context.config
+
+if config.config_file_name is not None:
+    fileConfig(config.config_file_name)
+
+target_metadata = Base.metadata
+
+
+def _database_url() -> str:
+    return os.getenv("DATABASE_URL", "sqlite:///l4d2web.db")
+
+
+def run_migrations_offline() -> None:
+    context.configure(
+        url=_database_url(),
+        target_metadata=target_metadata,
+        literal_binds=True,
+        dialect_opts={"paramstyle": "named"},
+    )
+
+    with context.begin_transaction():
+        context.run_migrations()
+
+
+def run_migrations_online() -> None:
+    configuration = config.get_section(config.config_ini_section) or {}
+    configuration["sqlalchemy.url"] = _database_url()
+    connectable = engine_from_config(
+        configuration,
+        prefix="sqlalchemy.",
+        poolclass=pool.NullPool,
+    )
+
+    with connectable.connect() as connection:
+        context.configure(connection=connection, target_metadata=target_metadata)
+
+        with context.begin_transaction():
+            context.run_migrations()
+
+
+if context.is_offline_mode():
+    run_migrations_offline()
+else:
+    run_migrations_online()

components/l4d2-web-app/alembic/versions/0001_initial.py

@@ -0,0 +1,100 @@
+"""initial schema
+
+Revision ID: 0001_initial
+Revises:
+Create Date: 2026-04-23
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "0001_initial"
+down_revision = None
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "users",
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column("username", sa.String(length=64), nullable=False, unique=True),
+        sa.Column("password_digest", sa.String(length=255), nullable=False),
+        sa.Column("admin", sa.Boolean(), nullable=False, server_default=sa.false()),
+        sa.Column("created_at", sa.DateTime(), nullable=False),
+        sa.Column("updated_at", sa.DateTime(), nullable=False),
+    )
+    op.create_table(
+        "overlays",
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column("name", sa.String(length=128), nullable=False, unique=True),
+        sa.Column("path", sa.String(length=512), nullable=False),
+        sa.Column("created_at", sa.DateTime(), nullable=False),
+        sa.Column("updated_at", sa.DateTime(), nullable=False),
+    )
+    op.create_table(
+        "blueprints",
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column("user_id", sa.Integer(), sa.ForeignKey("users.id"), nullable=False),
+        sa.Column("name", sa.String(length=128), nullable=False),
+        sa.Column("arguments", sa.Text(), nullable=False),
+        sa.Column("config", sa.Text(), nullable=False),
+        sa.Column("created_at", sa.DateTime(), nullable=False),
+        sa.Column("updated_at", sa.DateTime(), nullable=False),
+    )
+    op.create_table(
+        "blueprint_overlays",
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column("blueprint_id", sa.Integer(), sa.ForeignKey("blueprints.id"), nullable=False),
+        sa.Column("overlay_id", sa.Integer(), sa.ForeignKey("overlays.id"), nullable=False),
+        sa.Column("position", sa.Integer(), nullable=False),
+        sa.Column("created_at", sa.DateTime(), nullable=False),
+        sa.Column("updated_at", sa.DateTime(), nullable=False),
+    )
+    op.create_table(
+        "servers",
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column("user_id", sa.Integer(), sa.ForeignKey("users.id"), nullable=False),
+        sa.Column("blueprint_id", sa.Integer(), sa.ForeignKey("blueprints.id"), nullable=False),
+        sa.Column("name", sa.String(length=128), nullable=False, unique=True),
+        sa.Column("port", sa.Integer(), nullable=False),
+        sa.Column("desired_state", sa.String(length=16), nullable=False),
+        sa.Column("actual_state", sa.String(length=16), nullable=False),
+        sa.Column("actual_state_updated_at", sa.DateTime(), nullable=True),
+        sa.Column("last_error", sa.Text(), nullable=False),
+        sa.Column("created_at", sa.DateTime(), nullable=False),
+        sa.Column("updated_at", sa.DateTime(), nullable=False),
+    )
+    op.create_table(
+        "jobs",
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column("user_id", sa.Integer(), sa.ForeignKey("users.id"), nullable=False),
+        sa.Column("server_id", sa.Integer(), sa.ForeignKey("servers.id"), nullable=True),
+        sa.Column("operation", sa.String(length=32), nullable=False),
+        sa.Column("state", sa.String(length=16), nullable=False),
+        sa.Column("exit_code", sa.Integer(), nullable=True),
+        sa.Column("started_at", sa.DateTime(), nullable=True),
+        sa.Column("finished_at", sa.DateTime(), nullable=True),
+        sa.Column("created_at", sa.DateTime(), nullable=False),
+        sa.Column("updated_at", sa.DateTime(), nullable=False),
+    )
+    op.create_table(
+        "job_logs",
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column("job_id", sa.Integer(), sa.ForeignKey("jobs.id"), nullable=False),
+        sa.Column("seq", sa.Integer(), nullable=False),
+        sa.Column("stream", sa.String(length=8), nullable=False),
+        sa.Column("line", sa.Text(), nullable=False),
+        sa.Column("created_at", sa.DateTime(), nullable=False),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("job_logs")
+    op.drop_table("jobs")
+    op.drop_table("servers")
+    op.drop_table("blueprint_overlays")
+    op.drop_table("blueprints")
+    op.drop_table("overlays")
+    op.drop_table("users")

components/l4d2-web-app/pyproject.toml

@@ -0,0 +1,22 @@
+[build-system]
+requires = ["setuptools>=68", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "l4d2web"
+version = "0.1.0"
+description = "L4D2 web app"
+readme = "README.md"
+requires-python = ">=3.12"
+dependencies = [
+  "Flask>=3.0",
+  "SQLAlchemy>=2.0",
+  "alembic>=1.13",
+  "PyYAML>=6.0",
+]
+
+[tool.setuptools]
+package-dir = {"" = "src"}
+
+[tool.setuptools.packages.find]
+where = ["src"]

components/l4d2-web-app/src/l4d2web/__init__.py

@@ -0,0 +1,3 @@
+__all__ = ["__version__"]
+
+__version__ = "0.1.0"

components/l4d2-web-app/src/l4d2web/app.py

@@ -0,0 +1,72 @@
+import os
+import secrets
+
+from flask import Flask, Response, jsonify, request, session
+
+from l4d2web.auth import load_current_user
+from l4d2web.cli import register_cli
+from l4d2web.config import DEFAULT_CONFIG
+from l4d2web.db import init_db
+from l4d2web.routes.blueprint_routes import bp as blueprint_bp
+from l4d2web.routes.auth_routes import bp as auth_bp
+from l4d2web.routes.auth_routes import reset_login_rate_limits
+from l4d2web.routes.job_routes import bp as job_bp
+from l4d2web.routes.log_routes import bp as log_bp
+from l4d2web.routes.overlay_routes import bp as overlay_bp
+from l4d2web.routes.page_routes import bp as page_bp
+from l4d2web.routes.server_routes import bp as server_bp
+from l4d2web.services.job_worker import recover_stale_jobs
+
+
+def create_app(test_config: dict[str, object] | None = None) -> Flask:
+    app = Flask(__name__)
+    app.config.from_mapping(DEFAULT_CONFIG)
+    app.config.update(
+        SESSION_COOKIE_HTTPONLY=True,
+        SESSION_COOKIE_SAMESITE="Lax",
+        CSRF_EXEMPT_PATHS={"/login", "/signup", "/health"},
+    )
+    if test_config is not None:
+        app.config.update(test_config)
+
+    os.environ["DATABASE_URL"] = str(app.config["DATABASE_URL"])
+    init_db()
+
+    @app.before_request
+    def csrf_protect() -> Response | None:
+        if "csrf_token" not in session:
+            session["csrf_token"] = secrets.token_hex(16)
+
+        if request.method not in {"POST", "PUT", "PATCH", "DELETE"}:
+            return None
+
+        if request.path in app.config["CSRF_EXEMPT_PATHS"]:
+            return None
+
+        token = request.headers.get("X-CSRF-Token") or request.form.get("csrf_token")
+        if token != session.get("csrf_token"):
+            return Response("csrf token missing or invalid", status=400)
+        return None
+
+    app.before_request(load_current_user)
+    app.register_blueprint(auth_bp)
+    app.register_blueprint(overlay_bp)
+    app.register_blueprint(blueprint_bp)
+    app.register_blueprint(server_bp)
+    app.register_blueprint(job_bp)
+    app.register_blueprint(log_bp)
+    app.register_blueprint(page_bp)
+    register_cli(app)
+    if app.config.get("TESTING"):
+        reset_login_rate_limits()
+    recover_stale_jobs()
+
+    @app.get("/health")
+    def health():
+        return jsonify({"status": "ok"})
+
+    @app.get("/")
+    def root():
+        return jsonify({"status": "ok"})
+
+    return app

components/l4d2-web-app/src/l4d2web/auth.py

@@ -0,0 +1,64 @@
+from functools import wraps
+from typing import Callable, TypeVar
+
+from flask import abort, g, redirect, session
+from sqlalchemy import select
+from werkzeug.security import check_password_hash, generate_password_hash
+
+from l4d2web.db import session_scope
+from l4d2web.models import User
+
+
+F = TypeVar("F", bound=Callable)
+
+
+def hash_password(raw: str) -> str:
+    return generate_password_hash(raw)
+
+
+def verify_password(raw: str, digest: str) -> bool:
+    return check_password_hash(digest, raw)
+
+
+def load_current_user() -> None:
+    user_id = session.get("user_id")
+    if user_id is None:
+        g.user = None
+        return
+    with session_scope() as db:
+        g.user = db.scalar(select(User).where(User.id == int(user_id)))
+
+
+def current_user() -> User | None:
+    return getattr(g, "user", None)
+
+
+def login_user(user_id: int) -> None:
+    session["user_id"] = user_id
+
+
+def logout_user() -> None:
+    session.pop("user_id", None)
+
+
+def require_login(func: F) -> F:
+    @wraps(func)
+    def wrapper(*args, **kwargs):
+        if current_user() is None:
+            return redirect("/login")
+        return func(*args, **kwargs)
+
+    return wrapper  # type: ignore[return-value]
+
+
+def require_admin(func: F) -> F:
+    @wraps(func)
+    def wrapper(*args, **kwargs):
+        user = current_user()
+        if user is None:
+            return redirect("/login")
+        if not user.admin:
+            abort(403)
+        return func(*args, **kwargs)
+
+    return wrapper  # type: ignore[return-value]

components/l4d2-web-app/src/l4d2web/cli.py

@@ -0,0 +1,19 @@
+import click
+from sqlalchemy import select
+
+from l4d2web.db import session_scope
+from l4d2web.models import User
+
+
+@click.command("promote-admin")
+@click.argument("username")
+def promote_admin(username: str) -> None:
+    with session_scope() as db:
+        user = db.scalar(select(User).where(User.username == username))
+        if user is None:
+            raise click.ClickException("user not found")
+        user.admin = True
+
+
+def register_cli(app) -> None:
+    app.cli.add_command(promote_admin)

components/l4d2-web-app/src/l4d2web/config.py

@@ -0,0 +1,8 @@
+DEFAULT_CONFIG: dict[str, object] = {
+    "SECRET_KEY": "dev",
+    "DATABASE_URL": "sqlite:///l4d2web.db",
+    "STATUS_REFRESH_SECONDS": 8,
+    "JOB_WORKER_THREADS": 4,
+    "JOB_LOG_REPLAY_LIMIT": 2000,
+    "JOB_LOG_LINE_MAX_CHARS": 4096,
+}

components/l4d2-web-app/src/l4d2web/db.py

@@ -0,0 +1,55 @@
+from contextlib import contextmanager
+import os
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session, sessionmaker
+
+
+_engine = None
+_engine_url = None
+_Session = None
+
+
+def get_database_url() -> str:
+    return os.getenv("DATABASE_URL", "sqlite:///l4d2web.db")
+
+
+def get_engine():
+    global _engine
+    global _engine_url
+    global _Session
+
+    db_url = get_database_url()
+    if _engine is None or _engine_url != db_url:
+        connect_args = {"check_same_thread": False} if db_url.startswith("sqlite") else {}
+        _engine = create_engine(db_url, connect_args=connect_args)
+        if db_url.startswith("sqlite"):
+            with _engine.connect() as conn:
+                conn.exec_driver_sql("PRAGMA journal_mode=WAL;")
+                conn.exec_driver_sql("PRAGMA busy_timeout=5000;")
+        _engine_url = db_url
+        _Session = sessionmaker(bind=_engine, expire_on_commit=False)
+    return _engine
+
+
+def init_db() -> None:
+    from l4d2web.models import Base
+
+    Base.metadata.create_all(bind=get_engine())
+
+
+@contextmanager
+def session_scope() -> Session:
+    global _Session
+    if _Session is None:
+        get_engine()
+    assert _Session is not None
+    session = _Session()
+    try:
+        yield session
+        session.commit()
+    except Exception:
+        session.rollback()
+        raise
+    finally:
+        session.close()

components/l4d2-web-app/src/l4d2web/models.py

@@ -0,0 +1,98 @@
+from datetime import UTC, datetime
+
+from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, String, Text
+from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
+
+
+class Base(DeclarativeBase):
+    pass
+
+
+def now_utc() -> datetime:
+    return datetime.now(UTC)
+
+
+class User(Base):
+    __tablename__ = "users"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    username: Mapped[str] = mapped_column(String(64), unique=True, nullable=False)
+    password_digest: Mapped[str] = mapped_column(String(255), nullable=False)
+    admin: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
+    created_at: Mapped[datetime] = mapped_column(DateTime, default=now_utc, nullable=False)
+    updated_at: Mapped[datetime] = mapped_column(DateTime, default=now_utc, nullable=False)
+
+
+class Overlay(Base):
+    __tablename__ = "overlays"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    name: Mapped[str] = mapped_column(String(128), unique=True, nullable=False)
+    path: Mapped[str] = mapped_column(String(512), nullable=False)
+    created_at: Mapped[datetime] = mapped_column(DateTime, default=now_utc, nullable=False)
+    updated_at: Mapped[datetime] = mapped_column(DateTime, default=now_utc, nullable=False)
+
+
+class Blueprint(Base):
+    __tablename__ = "blueprints"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), nullable=False)
+    name: Mapped[str] = mapped_column(String(128), nullable=False)
+    arguments: Mapped[str] = mapped_column(Text, default="[]", nullable=False)
+    config: Mapped[str] = mapped_column(Text, default="[]", nullable=False)
+    created_at: Mapped[datetime] = mapped_column(DateTime, default=now_utc, nullable=False)
+    updated_at: Mapped[datetime] = mapped_column(DateTime, default=now_utc, nullable=False)
+
+
+class BlueprintOverlay(Base):
+    __tablename__ = "blueprint_overlays"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    blueprint_id: Mapped[int] = mapped_column(ForeignKey("blueprints.id"), nullable=False)
+    overlay_id: Mapped[int] = mapped_column(ForeignKey("overlays.id"), nullable=False)
+    position: Mapped[int] = mapped_column(Integer, nullable=False)
+    created_at: Mapped[datetime] = mapped_column(DateTime, default=now_utc, nullable=False)
+    updated_at: Mapped[datetime] = mapped_column(DateTime, default=now_utc, nullable=False)
+
+
+class Server(Base):
+    __tablename__ = "servers"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), nullable=False)
+    blueprint_id: Mapped[int] = mapped_column(ForeignKey("blueprints.id"), nullable=False)
+    name: Mapped[str] = mapped_column(String(128), unique=True, nullable=False)
+    port: Mapped[int] = mapped_column(Integer, nullable=False)
+    desired_state: Mapped[str] = mapped_column(String(16), default="stopped", nullable=False)
+    actual_state: Mapped[str] = mapped_column(String(16), default="unknown", nullable=False)
+    actual_state_updated_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
+    last_error: Mapped[str] = mapped_column(Text, default="", nullable=False)
+    created_at: Mapped[datetime] = mapped_column(DateTime, default=now_utc, nullable=False)
+    updated_at: Mapped[datetime] = mapped_column(DateTime, default=now_utc, nullable=False)
+
+
+class Job(Base):
+    __tablename__ = "jobs"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), nullable=False)
+    server_id: Mapped[int | None] = mapped_column(ForeignKey("servers.id"), nullable=True)
+    operation: Mapped[str] = mapped_column(String(32), nullable=False)
+    state: Mapped[str] = mapped_column(String(16), default="queued", nullable=False)
+    exit_code: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    started_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
+    finished_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
+    created_at: Mapped[datetime] = mapped_column(DateTime, default=now_utc, nullable=False)
+    updated_at: Mapped[datetime] = mapped_column(DateTime, default=now_utc, nullable=False)
+
+
+class JobLog(Base):
+    __tablename__ = "job_logs"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    job_id: Mapped[int] = mapped_column(ForeignKey("jobs.id"), nullable=False)
+    seq: Mapped[int] = mapped_column(Integer, nullable=False)
+    stream: Mapped[str] = mapped_column(String(8), nullable=False)
+    line: Mapped[str] = mapped_column(Text, nullable=False)
+    created_at: Mapped[datetime] = mapped_column(DateTime, default=now_utc, nullable=False)

components/l4d2-web-app/src/l4d2web/routes/auth_routes.py

@@ -0,0 +1,81 @@
+import time
+
+from flask import Blueprint, Response, request, redirect
+from sqlalchemy import select
+
+from l4d2web.auth import hash_password, login_user, logout_user, verify_password
+from l4d2web.db import session_scope
+from l4d2web.models import User
+
+
+bp = Blueprint("auth", __name__)
+LOGIN_RATE_LIMIT_WINDOW_SECONDS = 60
+LOGIN_RATE_LIMIT_MAX_ATTEMPTS = 20
+LOGIN_ATTEMPTS_BY_IP: dict[str, list[float]] = {}
+
+
+def reset_login_rate_limits() -> None:
+    LOGIN_ATTEMPTS_BY_IP.clear()
+
+
+def is_login_rate_limited(remote_addr: str) -> bool:
+    now = time.time()
+    attempts = LOGIN_ATTEMPTS_BY_IP.setdefault(remote_addr, [])
+    cutoff = now - LOGIN_RATE_LIMIT_WINDOW_SECONDS
+    attempts[:] = [ts for ts in attempts if ts >= cutoff]
+    if len(attempts) >= LOGIN_RATE_LIMIT_MAX_ATTEMPTS:
+        return True
+    attempts.append(now)
+    return False
+
+
+@bp.get("/signup")
+def signup_form() -> Response:
+    return Response("signup", mimetype="text/plain")
+
+
+@bp.post("/signup")
+def signup() -> Response:
+    username = request.form.get("username", "").strip()
+    password = request.form.get("password", "")
+    if not username or not password:
+        return Response("missing credentials", status=400)
+
+    with session_scope() as db:
+        existing = db.scalar(select(User).where(User.username == username))
+        if existing is not None:
+            return Response("username already exists", status=409)
+        user = User(username=username, password_digest=hash_password(password), admin=False)
+        db.add(user)
+        db.flush()
+        login_user(user.id)
+
+    return redirect("/dashboard")
+
+
+@bp.get("/login")
+def login_form() -> Response:
+    return Response("login", mimetype="text/plain")
+
+
+@bp.post("/login")
+def login() -> Response:
+    remote_addr = request.remote_addr or "unknown"
+    if is_login_rate_limited(remote_addr):
+        return Response("too many login attempts", status=429)
+
+    username = request.form.get("username", "").strip()
+    password = request.form.get("password", "")
+    with session_scope() as db:
+        user = db.scalar(select(User).where(User.username == username))
+        if user is None or not verify_password(password, user.password_digest):
+            return Response("invalid credentials", status=401)
+        login_user(user.id)
+        LOGIN_ATTEMPTS_BY_IP.pop(remote_addr, None)
+    return redirect("/dashboard")
+
+
+@bp.post("/logout")
+def logout() -> Response:
+    logout_user()
+    return redirect("/login")

components/l4d2-web-app/src/l4d2web/routes/blueprint_routes.py

@@ -0,0 +1,75 @@
+import json
+
+from flask import Blueprint, Response, jsonify, request
+from sqlalchemy import delete, func, select
+
+from l4d2web.auth import current_user, require_login
+from l4d2web.db import session_scope
+from l4d2web.models import Blueprint as BlueprintModel
+from l4d2web.models import BlueprintOverlay, Server
+
+
+bp = Blueprint("blueprint", __name__)
+
+
+@bp.post("/blueprints")
+@require_login
+def create_blueprint() -> Response:
+    payload = request.get_json(silent=True) or {}
+    user = current_user()
+    assert user is not None
+
+    name = str(payload.get("name", "")).strip()
+    if not name:
+        return Response("name is required", status=400)
+
+    with session_scope() as db:
+        blueprint = BlueprintModel(
+            user_id=user.id,
+            name=name,
+            arguments=json.dumps(payload.get("arguments", [])),
+            config=json.dumps(payload.get("config", [])),
+        )
+        db.add(blueprint)
+        db.flush()
+
+        for position, overlay_id in enumerate(payload.get("overlay_ids", [])):
+            db.add(
+                BlueprintOverlay(
+                    blueprint_id=blueprint.id,
+                    overlay_id=int(overlay_id),
+                    position=position,
+                )
+            )
+
+        blueprint_id = blueprint.id
+
+    return jsonify({"id": blueprint_id}), 201
+
+
+@bp.delete("/blueprints/<int:blueprint_id>")
+@require_login
+def delete_blueprint(blueprint_id: int) -> Response:
+    user = current_user()
+    assert user is not None
+
+    with session_scope() as db:
+        blueprint = db.scalar(
+            select(BlueprintModel).where(
+                BlueprintModel.id == blueprint_id,
+                BlueprintModel.user_id == user.id,
+            )
+        )
+        if blueprint is None:
+            return Response(status=404)
+
+        linked_count = db.scalar(
+            select(func.count(Server.id)).where(Server.blueprint_id == blueprint.id)
+        ) or 0
+        if linked_count > 0:
+            return Response("blueprint is in use", status=409)
+
+        db.execute(delete(BlueprintOverlay).where(BlueprintOverlay.blueprint_id == blueprint.id))
+        db.delete(blueprint)
+
+    return Response(status=204)

components/l4d2-web-app/src/l4d2web/routes/job_routes.py

@@ -0,0 +1,39 @@
+from flask import Blueprint, Response, current_app, request
+from sqlalchemy import select
+
+from l4d2web.auth import current_user, require_login
+from l4d2web.db import session_scope
+from l4d2web.models import Job, JobLog
+
+
+bp = Blueprint("job", __name__)
+
+
+@bp.get("/jobs/<int:job_id>/stream")
+@require_login
+def stream_job(job_id: int) -> Response:
+    user = current_user()
+    assert user is not None
+
+    last_seq = int(request.args.get("last_seq", "0"))
+    limit = int(current_app.config["JOB_LOG_REPLAY_LIMIT"])
+
+    with session_scope() as db:
+        job = db.scalar(select(Job).where(Job.id == job_id, Job.user_id == user.id))
+        if job is None:
+            return Response(status=404)
+
+    def generate():
+        with session_scope() as db:
+            rows = db.scalars(
+                select(JobLog)
+                .where(JobLog.job_id == job_id, JobLog.seq > last_seq)
+                .order_by(JobLog.seq)
+                .limit(limit)
+            ).all()
+            for row in rows:
+                yield f"id: {row.seq}\n"
+                yield f"event: {row.stream}\n"
+                yield f"data: {row.line}\n\n"
+
+    return Response(generate(), mimetype="text/event-stream")

components/l4d2-web-app/src/l4d2web/routes/log_routes.py

@@ -0,0 +1,33 @@
+from flask import Blueprint, Response
+from sqlalchemy import select
+
+from l4d2web.auth import current_user, require_login
+from l4d2web.db import session_scope
+from l4d2web.models import Server
+from l4d2web.services import l4d2_facade as facade
+
+
+bp = Blueprint("logs", __name__)
+
+
+def load_authorized_server(server_id: int) -> Server | None:
+    user = current_user()
+    if user is None:
+        return None
+    with session_scope() as db:
+        server = db.scalar(select(Server).where(Server.id == server_id, Server.user_id == user.id))
+        return server
+
+
+@bp.get("/servers/<int:server_id>/logs/stream")
+@require_login
+def stream_server_logs(server_id: int) -> Response:
+    server = load_authorized_server(server_id)
+    if server is None:
+        return Response(status=404)
+
+    def generate():
+        for line in facade.stream_server_logs(server.name, lines=200, follow=True):
+            yield f"data: {line}\n\n"
+
+    return Response(generate(), mimetype="text/event-stream")

components/l4d2-web-app/src/l4d2web/routes/overlay_routes.py

@@ -0,0 +1,32 @@
+from flask import Blueprint, Response, redirect, request
+from sqlalchemy import select
+
+from l4d2web.auth import require_admin
+from l4d2web.db import session_scope
+from l4d2web.models import Overlay
+from l4d2web.services.security import validate_overlay_path
+
+
+bp = Blueprint("overlay", __name__)
+
+
+@bp.post("/admin/overlays")
+@require_admin
+def create_overlay() -> Response:
+    name = request.form.get("name", "").strip()
+    raw_path = request.form.get("path", "").strip()
+    if not name or not raw_path:
+        return Response("missing fields", status=400)
+
+    try:
+        validated_path = validate_overlay_path(raw_path)
+    except ValueError as exc:
+        return Response(str(exc), status=400)
+
+    with session_scope() as db:
+        existing = db.scalar(select(Overlay).where(Overlay.name == name))
+        if existing is not None:
+            return Response("overlay already exists", status=409)
+        db.add(Overlay(name=name, path=str(validated_path)))
+
+    return redirect("/admin/overlays")

components/l4d2-web-app/src/l4d2web/routes/page_routes.py

@@ -0,0 +1,78 @@
+import json
+
+from flask import Blueprint, Response, current_app, render_template
+from sqlalchemy import select
+
+from l4d2web.auth import current_user, require_admin, require_login
+from l4d2web.db import session_scope
+from l4d2web.models import Blueprint as BlueprintModel
+from l4d2web.models import BlueprintOverlay, Overlay, Server
+
+
+bp = Blueprint("pages", __name__)
+
+
+@bp.get("/dashboard")
+@require_login
+def dashboard() -> str:
+    user = current_user()
+    assert user is not None
+
+    with session_scope() as db:
+        servers = db.scalars(select(Server).where(Server.user_id == user.id).order_by(Server.name)).all()
+    return render_template(
+        "dashboard.html",
+        servers=servers,
+        refresh_seconds=current_app.config["STATUS_REFRESH_SECONDS"],
+    )
+
+
+@bp.get("/blueprints/<int:blueprint_id>")
+@require_login
+def blueprint_page(blueprint_id: int):
+    user = current_user()
+    assert user is not None
+
+    with session_scope() as db:
+        blueprint = db.scalar(select(BlueprintModel).where(BlueprintModel.id == blueprint_id))
+        if blueprint is None:
+            return Response(status=404)
+        if blueprint.user_id != user.id:
+            return Response(status=403)
+
+        overlay_rows = db.execute(
+            select(Overlay.name)
+            .join(BlueprintOverlay, BlueprintOverlay.overlay_id == Overlay.id)
+            .where(BlueprintOverlay.blueprint_id == blueprint.id)
+            .order_by(BlueprintOverlay.position)
+        ).all()
+
+    return render_template(
+        "blueprints.html",
+        blueprint=blueprint,
+        overlay_names=[row[0] for row in overlay_rows],
+        arguments=json.loads(blueprint.arguments),
+        config_lines=json.loads(blueprint.config),
+    )
+
+
+@bp.get("/servers/<int:server_id>")
+@require_login
+def server_detail(server_id: int):
+    user = current_user()
+    assert user is not None
+
+    with session_scope() as db:
+        server = db.scalar(select(Server).where(Server.id == server_id, Server.user_id == user.id))
+        if server is None:
+            return Response(status=404)
+
+    return render_template("server_detail.html", server=server)
+
+
+@bp.get("/admin/overlays")
+@require_admin
+def admin_overlays() -> str:
+    with session_scope() as db:
+        overlays = db.scalars(select(Overlay).order_by(Overlay.name)).all()
+    return render_template("admin_overlays.html", overlays=overlays)

components/l4d2-web-app/src/l4d2web/routes/server_routes.py

@@ -0,0 +1,69 @@
+from flask import Blueprint, Response, jsonify, request
+from sqlalchemy import select
+
+from l4d2web.auth import current_user, require_login
+from l4d2web.db import session_scope
+from l4d2web.models import Blueprint as BlueprintModel
+from l4d2web.models import Server
+
+
+bp = Blueprint("server", __name__)
+
+
+@bp.post("/servers")
+@require_login
+def create_server() -> Response:
+    user = current_user()
+    assert user is not None
+    payload = request.get_json(silent=True) or {}
+
+    with session_scope() as db:
+        blueprint = db.scalar(
+            select(BlueprintModel).where(
+                BlueprintModel.id == int(payload["blueprint_id"]),
+                BlueprintModel.user_id == user.id,
+            )
+        )
+        if blueprint is None:
+            return Response("blueprint not found", status=404)
+
+        server = Server(
+            user_id=user.id,
+            blueprint_id=blueprint.id,
+            name=str(payload["name"]),
+            port=int(payload["port"]),
+            desired_state="stopped",
+            actual_state="unknown",
+            last_error="",
+        )
+        db.add(server)
+        db.flush()
+        server_id = server.id
+
+    return jsonify({"id": server_id}), 201
+
+
+@bp.patch("/servers/<int:server_id>")
+@require_login
+def update_server(server_id: int) -> Response:
+    user = current_user()
+    assert user is not None
+    payload = request.get_json(silent=True) or {}
+
+    with session_scope() as db:
+        server = db.scalar(select(Server).where(Server.id == server_id, Server.user_id == user.id))
+        if server is None:
+            return Response(status=404)
+
+        blueprint = db.scalar(
+            select(BlueprintModel).where(
+                BlueprintModel.id == int(payload["blueprint_id"]),
+                BlueprintModel.user_id == user.id,
+            )
+        )
+        if blueprint is None:
+            return Response("blueprint not found", status=404)
+
+        server.blueprint_id = blueprint.id
+
+    return jsonify({"id": server_id}), 200

components/l4d2-web-app/src/l4d2web/services/job_worker.py

@@ -0,0 +1,64 @@
+from dataclasses import dataclass, field
+from datetime import UTC, datetime
+
+from sqlalchemy import func, select
+from sqlalchemy.orm import Session
+
+from l4d2web.db import session_scope
+from l4d2web.models import Job, JobLog, Server
+
+
+@dataclass
+class SchedulerState:
+    install_running: bool = False
+    running_servers: set[int] = field(default_factory=set)
+
+
+def can_start(job, state: SchedulerState) -> bool:
+    if job.operation == "install":
+        return (not state.install_running) and (len(state.running_servers) == 0)
+    if state.install_running:
+        return False
+    if job.server_id is None:
+        return False
+    return job.server_id not in state.running_servers
+
+
+def recover_stale_jobs() -> int:
+    now = datetime.now(UTC)
+    with session_scope() as db:
+        jobs = db.scalars(select(Job).where(Job.state == "running")).all()
+        for job in jobs:
+            job.state = "failed"
+            job.finished_at = now
+            job.updated_at = now
+        return len(jobs)
+
+
+def append_job_log(
+    session: Session,
+    job_id: int,
+    stream: str,
+    line: str,
+    max_chars: int = 4096,
+) -> int:
+    last_seq = session.scalar(select(func.max(JobLog.seq)).where(JobLog.job_id == job_id)) or 0
+    next_seq = int(last_seq) + 1
+    session.add(JobLog(job_id=job_id, seq=next_seq, stream=stream, line=line[:max_chars]))
+    session.flush()
+    return next_seq
+
+
+def refresh_server_actual_state(server_id: int) -> str:
+    from l4d2web.services import l4d2_facade
+
+    now = datetime.now(UTC)
+    with session_scope() as db:
+        server = db.scalar(select(Server).where(Server.id == server_id))
+        if server is None:
+            return "unknown"
+        status = l4d2_facade.server_status(server.name)
+        server.actual_state = status.state
+        server.actual_state_updated_at = now
+        server.updated_at = now
+        return server.actual_state

components/l4d2-web-app/src/l4d2web/services/l4d2_facade.py

@@ -0,0 +1,77 @@
+import json
+from pathlib import Path
+
+from sqlalchemy import select
+
+from l4d2host.instances import delete_instance, initialize_instance, start_instance, stop_instance
+from l4d2host.logs import stream_instance_logs
+from l4d2host.status import get_instance_status
+from l4d2host.steam_install import SteamInstaller
+from l4d2web.db import session_scope
+from l4d2web.models import Blueprint, BlueprintOverlay, Overlay, Server
+from l4d2web.services.spec_yaml import write_temp_spec
+
+
+def build_server_spec_payload(server: Server, blueprint: Blueprint, overlay_names: list[str]) -> dict:
+    return {
+        "port": server.port,
+        "overlays": overlay_names,
+        "arguments": json.loads(blueprint.arguments),
+        "config": json.loads(blueprint.config),
+    }
+
+
+def load_server_blueprint_bundle(server_id: int) -> tuple[Server, Blueprint, list[str]]:
+    with session_scope() as db:
+        server = db.scalar(select(Server).where(Server.id == server_id))
+        if server is None:
+            raise ValueError("server not found")
+
+        blueprint = db.scalar(select(Blueprint).where(Blueprint.id == server.blueprint_id))
+        if blueprint is None:
+            raise ValueError("blueprint not found")
+
+        rows = db.execute(
+            select(Overlay.name)
+            .join(BlueprintOverlay, BlueprintOverlay.overlay_id == Overlay.id)
+            .where(BlueprintOverlay.blueprint_id == blueprint.id)
+            .order_by(BlueprintOverlay.position)
+        ).all()
+        overlay_names = [row[0] for row in rows]
+        return server, blueprint, overlay_names
+
+
+def install_runtime(on_stdout=None, on_stderr=None) -> None:
+    SteamInstaller().install_or_update(on_stdout=on_stdout, on_stderr=on_stderr)
+
+
+def initialize_server(server_id: int, on_stdout=None, on_stderr=None) -> None:
+    server, blueprint, overlay_names = load_server_blueprint_bundle(server_id)
+    spec_path = write_temp_spec(build_server_spec_payload(server, blueprint, overlay_names))
+    try:
+        initialize_instance(server.name, spec_path, on_stdout=on_stdout, on_stderr=on_stderr)
+    finally:
+        spec_path.unlink(missing_ok=True)
+
+
+def start_server(server_id: int, on_stdout=None, on_stderr=None) -> None:
+    server, _, _ = load_server_blueprint_bundle(server_id)
+    start_instance(server.name, on_stdout=on_stdout, on_stderr=on_stderr)
+
+
+def stop_server(server_id: int, on_stdout=None, on_stderr=None) -> None:
+    server, _, _ = load_server_blueprint_bundle(server_id)
+    stop_instance(server.name, on_stdout=on_stdout, on_stderr=on_stderr)
+
+
+def delete_server(server_id: int, on_stdout=None, on_stderr=None) -> None:
+    server, _, _ = load_server_blueprint_bundle(server_id)
+    delete_instance(server.name, on_stdout=on_stdout, on_stderr=on_stderr)
+
+
+def server_status(server_name: str):
+    return get_instance_status(server_name)
+
+
+def stream_server_logs(server_name: str, *, lines: int = 200, follow: bool = True):
+    return stream_instance_logs(server_name, lines=lines, follow=follow)

components/l4d2-web-app/src/l4d2web/services/security.py

@@ -0,0 +1,11 @@
+from pathlib import Path
+
+
+OVERLAY_ROOT = Path("/opt/l4d2/overlays").resolve()
+
+
+def validate_overlay_path(raw: str) -> Path:
+    path = Path(raw).resolve()
+    if OVERLAY_ROOT not in path.parents and path != OVERLAY_ROOT:
+        raise ValueError("overlay path must be under /opt/l4d2/overlays")
+    return path

components/l4d2-web-app/src/l4d2web/services/spec_yaml.py

@@ -0,0 +1,13 @@
+import os
+import tempfile
+from pathlib import Path
+
+import yaml
+
+
+def write_temp_spec(payload: dict) -> Path:
+    handle, filename = tempfile.mkstemp(prefix="l4d2-spec-", suffix=".yaml")
+    path = Path(filename)
+    with os.fdopen(handle, "w", encoding="utf-8") as file_obj:
+        file_obj.write(yaml.safe_dump(payload, sort_keys=False))
+    return path

components/l4d2-web-app/src/l4d2web/services/status.py

@@ -0,0 +1,14 @@
+def compute_display_state(active_operation: str | None, actual_state: str) -> str:
+    if active_operation == "delete":
+        return "deleting"
+    if active_operation == "start":
+        return "starting"
+    if active_operation == "stop":
+        return "stopping"
+    if active_operation == "initialize":
+        return "initializing"
+    return actual_state
+
+
+def has_drift(desired_state: str, actual_state: str, has_active_job: bool) -> bool:
+    return (not has_active_job) and desired_state != actual_state

components/l4d2-web-app/src/l4d2web/static/css/components.css

@@ -0,0 +1,45 @@
+.card {
+  background: var(--color-card);
+  border: 1px solid var(--color-border);
+  border-radius: var(--radius);
+  padding: var(--space-4);
+  margin-bottom: var(--space-4);
+  box-shadow: 0 8px 20px #0F766E12;
+}
+
+.table {
+  width: 100%;
+  border-collapse: collapse;
+}
+
+.table th,
+.table td {
+  text-align: left;
+  padding: var(--space-2) var(--space-3);
+  border-bottom: 1px solid var(--color-border);
+}
+
+.muted {
+  color: var(--color-muted);
+}
+
+.stack {
+  display: grid;
+  gap: var(--space-3);
+}
+
+input,
+button,
+select,
+textarea {
+  font: inherit;
+}
+
+button {
+  background: var(--color-link);
+  border: none;
+  border-radius: 8px;
+  color: #fff;
+  padding: var(--space-2) var(--space-4);
+  cursor: pointer;
+}

components/l4d2-web-app/src/l4d2web/static/css/layout.css

@@ -0,0 +1,38 @@
+* {
+  box-sizing: border-box;
+}
+
+body {
+  margin: 0;
+  font-family: "Segoe UI", "Helvetica Neue", sans-serif;
+  background: radial-gradient(circle at top right, #DDEFEA, #F3F7F6 45%);
+  color: var(--color-text);
+}
+
+.site-header {
+  background: #FFFFFFD9;
+  border-bottom: 1px solid var(--color-border);
+  position: sticky;
+  top: 0;
+  backdrop-filter: blur(6px);
+}
+
+.site-header-inner {
+  max-width: 960px;
+  margin: 0 auto;
+  padding: var(--space-4);
+  display: flex;
+  justify-content: space-between;
+  align-items: center;
+}
+
+.brand {
+  font-weight: 700;
+  text-decoration: none;
+}
+
+.container {
+  max-width: 960px;
+  margin: 0 auto;
+  padding: var(--space-6) var(--space-4) var(--space-6);
+}

components/l4d2-web-app/src/l4d2web/static/css/logs.css

@@ -0,0 +1,10 @@
+.log-stream {
+  min-height: 180px;
+  max-height: 360px;
+  overflow: auto;
+  background: #0A1412;
+  color: #CCE9E1;
+  border-radius: 8px;
+  padding: var(--space-3);
+  font-family: "SFMono-Regular", Menlo, Consolas, monospace;
+}

components/l4d2-web-app/src/l4d2web/static/css/tokens.css

@@ -0,0 +1,17 @@
+:root {
+  --color-link: #0F766E;
+  --color-bg: #F3F7F6;
+  --color-text: #11201D;
+  --color-card: #FFFFFF;
+  --color-border: #D4E4DF;
+  --color-muted: #4A6A63;
+  --radius: 10px;
+  --space-2: 0.5rem;
+  --space-3: 0.75rem;
+  --space-4: 1rem;
+  --space-6: 1.5rem;
+}
+
+a {
+  color: var(--color-link);
+}

components/l4d2-web-app/src/l4d2web/static/js/csrf.js

@@ -0,0 +1,10 @@
+document.addEventListener("DOMContentLoaded", () => {
+  const token = document.querySelector("meta[name='csrf-token']")?.getAttribute("content");
+  if (!token || !window.htmx || !window.htmx.on) {
+    return;
+  }
+
+  window.htmx.on("htmx:configRequest", (event) => {
+    event.detail.headers["X-CSRF-Token"] = token;
+  });
+});

components/l4d2-web-app/src/l4d2web/static/js/sse.js

@@ -0,0 +1,19 @@
+function streamTextToElement(url, elementId) {
+  const target = document.getElementById(elementId);
+  if (!target) {
+    return;
+  }
+
+  const source = new EventSource(url);
+  source.onmessage = (event) => {
+    target.textContent += `${event.data}\n`;
+    target.scrollTop = target.scrollHeight;
+  };
+}
+
+document.addEventListener("DOMContentLoaded", () => {
+  const serverLog = document.getElementById("server-log-stream");
+  if (serverLog) {
+    streamTextToElement(serverLog.dataset.serverLogUrl, "server-log-stream");
+  }
+});

components/l4d2-web-app/src/l4d2web/static/vendor/htmx.min.js

@@ -0,0 +1 @@
+window.htmx = window.htmx || {};

components/l4d2-web-app/src/l4d2web/templates/admin_overlays.html

@@ -0,0 +1,30 @@
+{% extends "base.html" %}
+
+{% block title %}Admin Overlays | left4me{% endblock %}
+
+{% block content %}
+<section class="card">
+  <h1>Overlay Catalog</h1>
+  <form method="post" action="/admin/overlays" class="stack">
+    <input type="hidden" name="csrf_token" value="{{ session.get('csrf_token', '') }}">
+    <label>
+      Name
+      <input name="name" required>
+    </label>
+    <label>
+      Path
+      <input name="path" required placeholder="/opt/l4d2/overlays/example">
+    </label>
+    <button type="submit">Add Overlay</button>
+  </form>
+
+  <h2>Known overlays</h2>
+  <ul>
+    {% for overlay in overlays %}
+    <li><strong>{{ overlay.name }}</strong> <span class="muted">{{ overlay.path }}</span></li>
+    {% else %}
+    <li class="muted">No overlays configured.</li>
+    {% endfor %}
+  </ul>
+</section>
+{% endblock %}

components/l4d2-web-app/src/l4d2web/templates/base.html

@@ -0,0 +1,30 @@
+<!doctype html>
+<html lang="en">
+  <head>
+    <meta charset="utf-8">
+    <meta name="viewport" content="width=device-width, initial-scale=1">
+    <meta name="csrf-token" content="{{ session.get('csrf_token', '') }}">
+    <title>{% block title %}left4me{% endblock %}</title>
+    <link rel="stylesheet" href="{{ url_for('static', filename='css/tokens.css') }}">
+    <link rel="stylesheet" href="{{ url_for('static', filename='css/layout.css') }}">
+    <link rel="stylesheet" href="{{ url_for('static', filename='css/components.css') }}">
+    <link rel="stylesheet" href="{{ url_for('static', filename='css/logs.css') }}">
+  </head>
+  <body>
+    <header class="site-header">
+      <div class="site-header-inner">
+        <a class="brand" href="/dashboard">left4me</a>
+        <nav>
+          <a href="/dashboard">Dashboard</a>
+          <a href="/admin/overlays">Overlays</a>
+        </nav>
+      </div>
+    </header>
+    <main class="container">
+      {% block content %}{% endblock %}
+    </main>
+    <script src="{{ url_for('static', filename='vendor/htmx.min.js') }}"></script>
+    <script src="{{ url_for('static', filename='js/csrf.js') }}"></script>
+    <script src="{{ url_for('static', filename='js/sse.js') }}"></script>
+  </body>
+</html>

components/l4d2-web-app/src/l4d2web/templates/blueprints.html

@@ -0,0 +1,21 @@
+{% extends "base.html" %}
+
+{% block title %}Blueprint | left4me{% endblock %}
+
+{% block content %}
+<section class="card">
+  <h1>Blueprint: {{ blueprint.name }}</h1>
+  <h2>Overlays</h2>
+  <ul>
+    {% for name in overlay_names %}
+    <li>{{ name }}</li>
+    {% else %}
+    <li class="muted">No overlays configured.</li>
+    {% endfor %}
+  </ul>
+  <h2>Arguments</h2>
+  <pre>{{ arguments | join('\n') }}</pre>
+  <h2>Config</h2>
+  <pre>{{ config_lines | join('\n') }}</pre>
+</section>
+{% endblock %}

components/l4d2-web-app/src/l4d2web/templates/dashboard.html

@@ -0,0 +1,28 @@
+{% extends "base.html" %}
+
+{% block title %}Dashboard | left4me{% endblock %}
+
+{% block content %}
+<section class="card">
+  <h1>Dashboard</h1>
+  <p class="muted">Status refresh every {{ refresh_seconds }}s.</p>
+  <table class="table">
+    <thead>
+      <tr><th>Name</th><th>Port</th><th>Desired</th><th>Actual</th><th></th></tr>
+    </thead>
+    <tbody>
+      {% for server in servers %}
+      <tr>
+        <td>{{ server.name }}</td>
+        <td>{{ server.port }}</td>
+        <td>{{ server.desired_state }}</td>
+        <td>{{ server.actual_state }}</td>
+        <td><a href="/servers/{{ server.id }}">View</a></td>
+      </tr>
+      {% else %}
+      <tr><td colspan="5" class="muted">No servers yet.</td></tr>
+      {% endfor %}
+    </tbody>
+  </table>
+</section>
+{% endblock %}

components/l4d2-web-app/src/l4d2web/templates/server_detail.html

@@ -0,0 +1,16 @@
+{% extends "base.html" %}
+
+{% block title %}Server {{ server.name }} | left4me{% endblock %}
+
+{% block content %}
+<section class="card">
+  <h1>Server: {{ server.name }}</h1>
+  <p><strong>Port:</strong> {{ server.port }}</p>
+  <p><strong>Desired:</strong> {{ server.desired_state }} | <strong>Actual:</strong> {{ server.actual_state }}</p>
+</section>
+
+<section class="card">
+  <h2>Live Logs</h2>
+  <pre id="server-log-stream" class="log-stream" data-server-log-url="/servers/{{ server.id }}/logs/stream"></pre>
+</section>
+{% endblock %}

components/l4d2-web-app/tests/test_auth.py

@@ -0,0 +1,46 @@
+import pytest
+
+from l4d2web.app import create_app
+from l4d2web.auth import hash_password
+from l4d2web.db import init_db, session_scope
+from l4d2web.models import User
+
+
+@pytest.fixture
+def client(tmp_path, monkeypatch):
+    db_url = f"sqlite:///{tmp_path/'auth.db'}"
+    monkeypatch.setenv("DATABASE_URL", db_url)
+    app = create_app({"TESTING": True, "DATABASE_URL": db_url, "SECRET_KEY": "test"})
+    init_db()
+    return app.test_client()
+
+
+@pytest.fixture
+def seed_user(tmp_path, monkeypatch):
+    db_url = f"sqlite:///{tmp_path/'seed.db'}"
+    monkeypatch.setenv("DATABASE_URL", db_url)
+    app = create_app({"TESTING": True, "DATABASE_URL": db_url, "SECRET_KEY": "test"})
+    init_db()
+    with app.app_context():
+        with session_scope() as session:
+            user = User(username="alice", password_digest=hash_password("secret"), admin=False)
+            session.add(user)
+            session.flush()
+            user_id = user.id
+    return user_id
+
+
+def test_public_signup(client) -> None:
+    response = client.post("/signup", data={"username": "alice", "password": "secret"})
+    assert response.status_code == 302
+
+
+def test_login_sets_session(client) -> None:
+    with session_scope() as session:
+        session.add(User(username="alice", password_digest=hash_password("secret"), admin=False))
+
+    response = client.post("/login", data={"username": "alice", "password": "secret"})
+    assert response.status_code == 302
+
+    with client.session_transaction() as sess:
+        assert sess.get("user_id") is not None

components/l4d2-web-app/tests/test_blueprints.py

@@ -0,0 +1,85 @@
+import json
+
+import pytest
+
+from l4d2web.app import create_app
+from l4d2web.auth import hash_password
+from l4d2web.db import init_db, session_scope
+from l4d2web.models import Blueprint, BlueprintOverlay, Overlay, Server, User
+
+
+@pytest.fixture
+def user_client(tmp_path, monkeypatch):
+    db_url = f"sqlite:///{tmp_path/'blueprint.db'}"
+    monkeypatch.setenv("DATABASE_URL", db_url)
+    app = create_app({"TESTING": True, "DATABASE_URL": db_url, "SECRET_KEY": "test"})
+    init_db()
+
+    with session_scope() as session:
+        user = User(username="alice", password_digest=hash_password("secret"), admin=False)
+        session.add(user)
+        session.flush()
+        user_id = user.id
+        session.add_all(
+            [
+                Overlay(name="o1", path="/opt/l4d2/overlays/o1"),
+                Overlay(name="o2", path="/opt/l4d2/overlays/o2"),
+            ]
+        )
+
+    client = app.test_client()
+    with client.session_transaction() as sess:
+        sess["user_id"] = user_id
+        sess["csrf_token"] = "test-token"
+    return client
+
+
+@pytest.fixture
+def linked_blueprint(tmp_path, monkeypatch):
+    db_url = f"sqlite:///{tmp_path/'linked.db'}"
+    monkeypatch.setenv("DATABASE_URL", db_url)
+    app = create_app({"TESTING": True, "DATABASE_URL": db_url, "SECRET_KEY": "test"})
+    init_db()
+
+    with session_scope() as session:
+        user = User(username="bob", password_digest=hash_password("secret"), admin=False)
+        session.add(user)
+        session.flush()
+
+        blueprint = Blueprint(user_id=user.id, name="linked", arguments="[]", config="[]")
+        session.add(blueprint)
+        session.flush()
+
+        session.add(Server(user_id=user.id, blueprint_id=blueprint.id, name="alpha", port=27015))
+        blueprint_id = blueprint.id
+        user_id = user.id
+
+    client = app.test_client()
+    with client.session_transaction() as sess:
+        sess["user_id"] = user_id
+        sess["csrf_token"] = "test-token"
+
+    return client, blueprint_id
+
+
+def test_user_can_create_private_blueprint(user_client) -> None:
+    payload = {
+        "name": "comp",
+        "arguments": ["-tickrate 100"],
+        "config": ["sv_consistency 1"],
+        "overlay_ids": [1, 2],
+    }
+
+    response = user_client.post(
+        "/blueprints",
+        data=json.dumps(payload),
+        content_type="application/json",
+        headers={"X-CSRF-Token": "test-token"},
+    )
+    assert response.status_code == 201
+
+
+def test_delete_blueprint_blocked_when_in_use(linked_blueprint) -> None:
+    client, blueprint_id = linked_blueprint
+    response = client.delete(f"/blueprints/{blueprint_id}", headers={"X-CSRF-Token": "test-token"})
+    assert response.status_code == 409

components/l4d2-web-app/tests/test_health.py

@@ -0,0 +1,11 @@
+from l4d2web.app import create_app
+
+
+def test_health_endpoint() -> None:
+    app = create_app({"TESTING": True})
+    client = app.test_client()
+
+    response = client.get("/health")
+
+    assert response.status_code == 200
+    assert response.get_json() == {"status": "ok"}

components/l4d2-web-app/tests/test_job_logs.py

@@ -0,0 +1,69 @@
+from sqlalchemy import text
+
+import pytest
+
+from l4d2web.app import create_app
+from l4d2web.auth import hash_password
+from l4d2web.db import get_engine, init_db, session_scope
+from l4d2web.models import Job, JobLog, User
+from l4d2web.services.job_worker import append_job_log
+
+
+@pytest.fixture
+def seeded_job_logs(tmp_path, monkeypatch):
+    db_url = f"sqlite:///{tmp_path/'joblogs.db'}"
+    monkeypatch.setenv("DATABASE_URL", db_url)
+    app = create_app({"TESTING": True, "DATABASE_URL": db_url, "SECRET_KEY": "test"})
+    init_db()
+
+    with session_scope() as session:
+        user = User(username="alice", password_digest=hash_password("secret"), admin=False)
+        session.add(user)
+        session.flush()
+
+        job = Job(user_id=user.id, server_id=None, operation="install", state="queued")
+        session.add(job)
+        session.flush()
+
+        for idx in range(1, 8):
+            session.add(JobLog(job_id=job.id, seq=idx, stream="stdout", line=f"line-{idx}"))
+
+        job_id = job.id
+        user_id = user.id
+
+    return app, job_id, user_id
+
+
+def test_job_logs_seq_monotonic(seeded_job_logs) -> None:
+    app, job_id, _ = seeded_job_logs
+    with app.app_context():
+        with get_engine().connect() as conn:
+            rows = conn.execute(
+                text("select seq from job_logs where job_id=:id order by seq"),
+                {"id": job_id},
+            ).all()
+
+    values = [row[0] for row in rows]
+    assert values == sorted(values)
+
+
+def test_append_job_log_increments_seq(seeded_job_logs) -> None:
+    app, job_id, _ = seeded_job_logs
+
+    with app.app_context():
+        with session_scope() as session:
+            append_job_log(session, job_id=job_id, stream="stdout", line="new line")
+        with session_scope() as session:
+            last = session.query(JobLog).filter(JobLog.job_id == job_id).order_by(JobLog.seq.desc()).first()
+            assert last is not None
+            assert last.seq == 8
+
+
+def test_sse_resume_from_last_seq(seeded_job_logs) -> None:
+    app, job_id, user_id = seeded_job_logs
+    client = app.test_client()
+    with client.session_transaction() as sess:
+        sess["user_id"] = user_id
+
+    response = client.get(f"/jobs/{job_id}/stream?last_seq=5")
+    assert response.status_code == 200

components/l4d2-web-app/tests/test_job_worker.py

@@ -0,0 +1,65 @@
+from dataclasses import dataclass
+
+import pytest
+
+from l4d2web.services.job_worker import SchedulerState, can_start, recover_stale_jobs
+
+
+@dataclass
+class DummyJob:
+    operation: str
+    server_id: int | None = None
+
+
+@pytest.fixture
+def worker_fixture(tmp_path, monkeypatch):
+    from l4d2web.app import create_app
+    from l4d2web.db import init_db
+
+    db_url = f"sqlite:///{tmp_path/'worker.db'}"
+    monkeypatch.setenv("DATABASE_URL", db_url)
+    app = create_app({"TESTING": True, "DATABASE_URL": db_url, "SECRET_KEY": "test"})
+    init_db()
+
+    class WorkerFixture:
+        def run_once(self):
+            state = SchedulerState()
+
+            state.running_servers.add(1)
+            same_server_parallel = can_start(DummyJob(operation="start", server_id=1), state)
+
+            different_servers_parallel = can_start(DummyJob(operation="start", server_id=2), state)
+
+            install_parallel = can_start(DummyJob(operation="install", server_id=None), state)
+
+            return {
+                "same_server_parallel": same_server_parallel,
+                "different_servers_parallel": different_servers_parallel,
+                "install_parallel": install_parallel,
+            }
+
+        def recover_stale_jobs(self):
+            with app.app_context():
+                return recover_stale_jobs()
+
+    return WorkerFixture()
+
+
+def test_same_server_jobs_serialized(worker_fixture) -> None:
+    result = worker_fixture.run_once()
+    assert result["same_server_parallel"] is False
+
+
+def test_different_servers_can_run_parallel(worker_fixture) -> None:
+    result = worker_fixture.run_once()
+    assert result["different_servers_parallel"] is True
+
+
+def test_install_global_exclusive(worker_fixture) -> None:
+    result = worker_fixture.run_once()
+    assert result["install_parallel"] is False
+
+
+def test_recover_stale_running_jobs(worker_fixture) -> None:
+    recovered = worker_fixture.recover_stale_jobs()
+    assert recovered >= 0

components/l4d2-web-app/tests/test_l4d2_facade.py

@@ -0,0 +1,62 @@
+from pathlib import Path
+
+import pytest
+
+from l4d2web.app import create_app
+from l4d2web.auth import hash_password
+from l4d2web.db import init_db, session_scope
+from l4d2web.models import Blueprint, BlueprintOverlay, Overlay, Server, User
+
+
+@pytest.fixture
+def server_with_blueprint(tmp_path, monkeypatch):
+    db_url = f"sqlite:///{tmp_path/'facade.db'}"
+    monkeypatch.setenv("DATABASE_URL", db_url)
+    app = create_app({"TESTING": True, "DATABASE_URL": db_url, "SECRET_KEY": "test"})
+    init_db()
+
+    with app.app_context():
+        with session_scope() as session:
+            user = User(username="alice", password_digest=hash_password("secret"), admin=False)
+            session.add(user)
+            session.flush()
+
+            overlay = Overlay(name="standard", path="/opt/l4d2/overlays/standard")
+            session.add(overlay)
+            session.flush()
+
+            blueprint = Blueprint(
+                user_id=user.id,
+                name="default",
+                arguments='["-tickrate 100"]',
+                config='["sv_consistency 1"]',
+            )
+            session.add(blueprint)
+            session.flush()
+
+            session.add(BlueprintOverlay(blueprint_id=blueprint.id, overlay_id=overlay.id, position=0))
+
+            server = Server(user_id=user.id, blueprint_id=blueprint.id, name="alpha", port=27015)
+            session.add(server)
+            session.flush()
+            server_id = server.id
+
+    return server_id
+
+
+def test_initialize_uses_latest_blueprint_data(monkeypatch: pytest.MonkeyPatch, server_with_blueprint) -> None:
+    called: dict[str, str] = {}
+
+    def fake_initialize(name, spec_path, **kwargs):
+        del kwargs
+        called["name"] = name
+        called["spec"] = Path(spec_path).read_text()
+
+    monkeypatch.setattr("l4d2web.services.l4d2_facade.initialize_instance", fake_initialize)
+
+    from l4d2web.services.l4d2_facade import initialize_server
+
+    initialize_server(server_with_blueprint)
+
+    assert called["name"] == "alpha"
+    assert "sv_consistency 1" in called["spec"]

components/l4d2-web-app/tests/test_models.py

@@ -0,0 +1,20 @@
+from l4d2web.db import init_db, session_scope
+from l4d2web.models import Blueprint, User
+
+
+def test_create_user_and_blueprint(tmp_path, monkeypatch) -> None:
+    monkeypatch.setenv("DATABASE_URL", f"sqlite:///{tmp_path/'app.db'}")
+
+    init_db()
+
+    with session_scope() as session:
+        user = User(username="alice", password_digest="digest", admin=False)
+        session.add(user)
+        session.flush()
+
+        blueprint = Blueprint(user_id=user.id, name="default", arguments="[]", config="[]")
+        session.add(blueprint)
+        session.flush()
+
+        assert user.id is not None
+        assert blueprint.id is not None

components/l4d2-web-app/tests/test_overlays.py

@@ -0,0 +1,44 @@
+import pytest
+
+from l4d2web.app import create_app
+from l4d2web.auth import hash_password
+from l4d2web.db import init_db, session_scope
+from l4d2web.models import User
+
+
+@pytest.fixture
+def admin_client(tmp_path, monkeypatch):
+    db_url = f"sqlite:///{tmp_path/'overlay.db'}"
+    monkeypatch.setenv("DATABASE_URL", db_url)
+    app = create_app({"TESTING": True, "DATABASE_URL": db_url, "SECRET_KEY": "test"})
+    init_db()
+
+    with session_scope() as session:
+        admin = User(username="admin", password_digest=hash_password("secret"), admin=True)
+        session.add(admin)
+        session.flush()
+        admin_id = admin.id
+
+    client = app.test_client()
+    with client.session_transaction() as sess:
+        sess["user_id"] = admin_id
+        sess["csrf_token"] = "test-token"
+    return client
+
+
+def test_admin_can_create_overlay(admin_client) -> None:
+    response = admin_client.post(
+        "/admin/overlays",
+        data={"name": "standard", "path": "/opt/l4d2/overlays/standard"},
+        headers={"X-CSRF-Token": "test-token"},
+    )
+    assert response.status_code == 302
+
+
+def test_overlay_path_must_be_under_root(admin_client) -> None:
+    response = admin_client.post(
+        "/admin/overlays",
+        data={"name": "bad", "path": "/tmp/bad"},
+        headers={"X-CSRF-Token": "test-token"},
+    )
+    assert response.status_code == 400

components/l4d2-web-app/tests/test_pages.py

@@ -0,0 +1,72 @@
+import pytest
+
+from l4d2web.app import create_app
+from l4d2web.auth import hash_password
+from l4d2web.db import init_db, session_scope
+from l4d2web.models import Blueprint, Server, User
+
+
+@pytest.fixture
+def auth_client_with_server(tmp_path, monkeypatch):
+    db_url = f"sqlite:///{tmp_path/'pages.db'}"
+    monkeypatch.setenv("DATABASE_URL", db_url)
+    app = create_app({"TESTING": True, "DATABASE_URL": db_url, "SECRET_KEY": "test"})
+    init_db()
+
+    with session_scope() as session:
+        user = User(username="alice", password_digest=hash_password("secret"), admin=False)
+        session.add(user)
+        session.flush()
+
+        blueprint = Blueprint(user_id=user.id, name="default", arguments="[]", config="[]")
+        session.add(blueprint)
+        session.flush()
+
+        session.add(Server(user_id=user.id, blueprint_id=blueprint.id, name="alpha", port=27015))
+        session.flush()
+        user_id = user.id
+
+    client = app.test_client()
+    with client.session_transaction() as sess:
+        sess["user_id"] = user_id
+    return client
+
+
+@pytest.fixture
+def user_client_and_other_blueprint(tmp_path, monkeypatch):
+    db_url = f"sqlite:///{tmp_path/'otherbp.db'}"
+    monkeypatch.setenv("DATABASE_URL", db_url)
+    app = create_app({"TESTING": True, "DATABASE_URL": db_url, "SECRET_KEY": "test"})
+    init_db()
+
+    with session_scope() as session:
+        owner = User(username="owner", password_digest=hash_password("secret"), admin=False)
+        other = User(username="other", password_digest=hash_password("secret"), admin=False)
+        session.add_all([owner, other])
+        session.flush()
+
+        blueprint = Blueprint(user_id=other.id, name="private", arguments="[]", config="[]")
+        session.add(blueprint)
+        session.flush()
+
+        owner_id = owner.id
+        blueprint_id = blueprint.id
+
+    client = app.test_client()
+    with client.session_transaction() as sess:
+        sess["user_id"] = owner_id
+
+    return client, blueprint_id
+
+
+def test_dashboard_renders_server_and_status(auth_client_with_server) -> None:
+    response = auth_client_with_server.get("/dashboard")
+    text = response.get_data(as_text=True)
+    assert response.status_code == 200
+    assert "alpha" in text
+
+
+def test_blueprint_page_private_visibility(user_client_and_other_blueprint) -> None:
+    client, blueprint_id = user_client_and_other_blueprint
+    response = client.get(f"/blueprints/{blueprint_id}")
+    assert response.status_code == 403

components/l4d2-web-app/tests/test_security.py

@@ -0,0 +1,30 @@
+import pytest
+
+from l4d2web.app import create_app
+from l4d2web.auth import hash_password
+from l4d2web.db import init_db, session_scope
+from l4d2web.models import User
+
+
+@pytest.fixture
+def client(tmp_path, monkeypatch):
+    db_url = f"sqlite:///{tmp_path/'security.db'}"
+    monkeypatch.setenv("DATABASE_URL", db_url)
+    app = create_app({"TESTING": True, "DATABASE_URL": db_url, "SECRET_KEY": "test"})
+    init_db()
+    with session_scope() as session:
+        session.add(User(username="alice", password_digest=hash_password("secret"), admin=False))
+    return app.test_client()
+
+
+def test_csrf_required(client) -> None:
+    response = client.post("/servers", data={"name": "x"})
+    assert response.status_code == 400
+
+
+def test_login_rate_limit(client) -> None:
+    for _ in range(20):
+        client.post("/login", data={"username": "x", "password": "y"})
+
+    response = client.post("/login", data={"username": "x", "password": "y"})
+    assert response.status_code == 429

components/l4d2-web-app/tests/test_servers.py

@@ -0,0 +1,71 @@
+import json
+
+import pytest
+
+from l4d2web.app import create_app
+from l4d2web.auth import hash_password
+from l4d2web.db import init_db, session_scope
+from l4d2web.models import Blueprint, User
+
+
+@pytest.fixture
+def user_client_with_blueprints(tmp_path, monkeypatch):
+    db_url = f"sqlite:///{tmp_path/'servers.db'}"
+    monkeypatch.setenv("DATABASE_URL", db_url)
+    app = create_app({"TESTING": True, "DATABASE_URL": db_url, "SECRET_KEY": "test"})
+    init_db()
+
+    with session_scope() as session:
+        user = User(username="alice", password_digest=hash_password("secret"), admin=False)
+        session.add(user)
+        session.flush()
+        blueprint_one = Blueprint(user_id=user.id, name="bp1", arguments="[]", config="[]")
+        blueprint_two = Blueprint(user_id=user.id, name="bp2", arguments="[]", config="[]")
+        session.add_all([blueprint_one, blueprint_two])
+        session.flush()
+        payload = {
+            "user_id": user.id,
+            "blueprint_id": blueprint_one.id,
+            "other_blueprint_id": blueprint_two.id,
+        }
+
+    client = app.test_client()
+    with client.session_transaction() as sess:
+        sess["user_id"] = payload["user_id"]
+        sess["csrf_token"] = "test-token"
+
+    return client, payload
+
+
+def test_create_server_from_blueprint(user_client_with_blueprints) -> None:
+    client, data = user_client_with_blueprints
+    payload = {"name": "alpha", "port": 27015, "blueprint_id": data["blueprint_id"]}
+    response = client.post(
+        "/servers",
+        data=json.dumps(payload),
+        content_type="application/json",
+        headers={"X-CSRF-Token": "test-token"},
+    )
+    assert response.status_code == 201
+
+
+def test_reassign_blueprint_anytime(user_client_with_blueprints) -> None:
+    client, data = user_client_with_blueprints
+
+    create_payload = {"name": "alpha", "port": 27015, "blueprint_id": data["blueprint_id"]}
+    create_response = client.post(
+        "/servers",
+        data=json.dumps(create_payload),
+        content_type="application/json",
+        headers={"X-CSRF-Token": "test-token"},
+    )
+    server_id = create_response.get_json()["id"]
+
+    patch_payload = {"blueprint_id": data["other_blueprint_id"]}
+    response = client.patch(
+        f"/servers/{server_id}",
+        data=json.dumps(patch_payload),
+        content_type="application/json",
+        headers={"X-CSRF-Token": "test-token"},
+    )
+    assert response.status_code == 200

components/l4d2-web-app/tests/test_status_and_server_logs.py

@@ -0,0 +1,54 @@
+import pytest
+
+from l4d2web.app import create_app
+from l4d2web.auth import hash_password
+from l4d2web.db import init_db, session_scope
+from l4d2web.models import Blueprint, Server, User
+
+
+@pytest.fixture
+def owner_client_with_server(tmp_path, monkeypatch):
+    db_url = f"sqlite:///{tmp_path/'status.db'}"
+    monkeypatch.setenv("DATABASE_URL", db_url)
+    app = create_app({"TESTING": True, "DATABASE_URL": db_url, "SECRET_KEY": "test"})
+    init_db()
+
+    with session_scope() as session:
+        user = User(username="alice", password_digest=hash_password("secret"), admin=False)
+        session.add(user)
+        session.flush()
+
+        blueprint = Blueprint(user_id=user.id, name="default", arguments="[]", config="[]")
+        session.add(blueprint)
+        session.flush()
+
+        server = Server(user_id=user.id, blueprint_id=blueprint.id, name="alpha", port=27015)
+        session.add(server)
+        session.flush()
+
+        user_id = user.id
+        server_id = server.id
+
+    client = app.test_client()
+    with client.session_transaction() as sess:
+        sess["user_id"] = user_id
+
+    return client, server_id
+
+
+def test_owner_can_stream_server_logs(owner_client_with_server, monkeypatch) -> None:
+    client, server_id = owner_client_with_server
+
+    monkeypatch.setattr(
+        "l4d2web.services.l4d2_facade.stream_server_logs",
+        lambda name, lines=200, follow=True: iter(["first", "second"]),
+    )
+
+    response = client.get(f"/servers/{server_id}/logs/stream")
+    assert response.status_code == 200
+
+
+def test_status_precedence() -> None:
+    from l4d2web.services.status import compute_display_state
+
+    assert compute_display_state("start", "stopped") == "starting"