.gitignore

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

components/l4d2-host-lib/README.md

@@ -1,31 +0,0 @@
-# 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

@@ -1,23 +0,0 @@
-[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

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

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

@@ -1,56 +0,0 @@
-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

@@ -1,30 +0,0 @@
-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

@@ -1,45 +0,0 @@
-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

@@ -1,149 +0,0 @@
-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

@@ -1,34 +0,0 @@
-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

@@ -1,79 +0,0 @@
-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

@@ -1,22 +0,0 @@
-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

@@ -1,56 +0,0 @@
-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

@@ -1,37 +0,0 @@
-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

@@ -1,29 +0,0 @@
-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

@@ -1,14 +0,0 @@
-[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

@@ -1,26 +0,0 @@
-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

@@ -1,22 +0,0 @@
-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

@@ -1,35 +0,0 @@
-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

@@ -1,33 +0,0 @@
-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

@@ -1,42 +0,0 @@
-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

@@ -1,22 +0,0 @@
-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

@@ -1,36 +0,0 @@
-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

@@ -1,8 +0,0 @@
-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

@@ -1,28 +0,0 @@
-# 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

@@ -1,35 +0,0 @@
-[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

@@ -1,53 +0,0 @@
-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

@@ -1,100 +0,0 @@
-"""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

@@ -1,22 +0,0 @@
-[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

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

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

@@ -1,72 +0,0 @@
-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

@@ -1,64 +0,0 @@
-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

@@ -1,19 +0,0 @@
-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

@@ -1,8 +0,0 @@
-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

@@ -1,55 +0,0 @@
-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

@@ -1,98 +0,0 @@
-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

@@ -1,81 +0,0 @@
-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

@@ -1,75 +0,0 @@
-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

@@ -1,39 +0,0 @@
-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

@@ -1,33 +0,0 @@
-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

@@ -1,32 +0,0 @@
-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

@@ -1,78 +0,0 @@
-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

@@ -1,69 +0,0 @@
-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

@@ -1,64 +0,0 @@
-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

@@ -1,77 +0,0 @@
-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

@@ -1,11 +0,0 @@
-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

@@ -1,13 +0,0 @@
-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

@@ -1,14 +0,0 @@
-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

@@ -1,45 +0,0 @@
-.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

@@ -1,38 +0,0 @@
-* {
-  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

@@ -1,10 +0,0 @@
-.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

@@ -1,17 +0,0 @@
-: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

@@ -1,10 +0,0 @@
-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

@@ -1,19 +0,0 @@
-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

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

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

@@ -1,30 +0,0 @@
-{% 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

@@ -1,30 +0,0 @@
-<!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

@@ -1,21 +0,0 @@
-{% 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

@@ -1,28 +0,0 @@
-{% 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

@@ -1,16 +0,0 @@
-{% 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

@@ -1,46 +0,0 @@
-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

@@ -1,85 +0,0 @@
-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

@@ -1,11 +0,0 @@
-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

@@ -1,69 +0,0 @@
-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

@@ -1,65 +0,0 @@
-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

@@ -1,62 +0,0 @@
-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

@@ -1,20 +0,0 @@
-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

@@ -1,44 +0,0 @@
-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

@@ -1,72 +0,0 @@
-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

@@ -1,30 +0,0 @@
-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

@@ -1,71 +0,0 @@
-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

@@ -1,54 +0,0 @@
-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"