# L4D2 Web App v1 Implementation Plan > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** Build a local Flask web app where users manage L4D2 servers with auth, overlay selection, async jobs, desired-vs-actual status, and live logs. **Architecture:** The app runs as a single Flask process with Jinja templates and vendored local HTMX. It imports `l4d2host` directly (no subprocess boundary), persists state in SQLite, and executes lifecycle actions asynchronously through an in-process scheduler with lock rules: same server serialized, different servers parallel, install global-exclusive. **Tech Stack:** Python 3.12+, Flask, SQLAlchemy, Alembic, pytest, vendored HTMX, custom CSS, vanilla JS (SSE). --- ## Scope and Constraints - In scope: - public signup/login - admin role with global permissions - server CRUD with ordered overlays - async lifecycle actions (`install`, `initialize`, `start`, `stop`, `delete`) - live job logs and live server logs - desired-vs-actual state in server list - Out of scope (later phases): - workshop mod management - user-created overlays and web file manager - Frontend constraints: - no CSS/JS framework dependencies - HTMX vendored locally only (no CDN) - custom CSS only; links use consistent accent `#0F766E` - Runtime constraints: - single-process deployment for v1 - periodic status refresh every 8 seconds ## Planned File Layout - `components/l4d2-web-app/pyproject.toml` - `components/l4d2-web-app/src/l4d2web/app.py` - `components/l4d2-web-app/src/l4d2web/config.py` - `components/l4d2-web-app/src/l4d2web/db.py` - `components/l4d2-web-app/src/l4d2web/models.py` - `components/l4d2-web-app/src/l4d2web/auth.py` - `components/l4d2-web-app/src/l4d2web/cli.py` - `components/l4d2-web-app/src/l4d2web/services/l4d2_facade.py` - `components/l4d2-web-app/src/l4d2web/services/spec_yaml.py` - `components/l4d2-web-app/src/l4d2web/services/job_worker.py` - `components/l4d2-web-app/src/l4d2web/services/status.py` - `components/l4d2-web-app/src/l4d2web/routes/auth_routes.py` - `components/l4d2-web-app/src/l4d2web/routes/server_routes.py` - `components/l4d2-web-app/src/l4d2web/routes/overlay_routes.py` - `components/l4d2-web-app/src/l4d2web/routes/job_routes.py` - `components/l4d2-web-app/src/l4d2web/routes/log_routes.py` - `components/l4d2-web-app/src/l4d2web/templates/*.html` - `components/l4d2-web-app/src/l4d2web/static/vendor/htmx.min.js` - `components/l4d2-web-app/src/l4d2web/static/css/{tokens,layout,components,logs}.css` - `components/l4d2-web-app/src/l4d2web/static/js/sse.js` - `components/l4d2-web-app/alembic.ini` - `components/l4d2-web-app/alembic/env.py` - `components/l4d2-web-app/alembic/versions/0001_initial.py` - `components/l4d2-web-app/tests/*.py` - `components/l4d2-web-app/README.md` ### Task 1: Scaffold Flask app, config, and health endpoint **Files:** - Create: `components/l4d2-web-app/pyproject.toml` - Create: `components/l4d2-web-app/src/l4d2web/app.py` - Create: `components/l4d2-web-app/src/l4d2web/config.py` - Test: `components/l4d2-web-app/tests/test_health.py` - [ ] **Step 1: Write failing health test** ```python from l4d2web.app import create_app def test_health_endpoint(): app = create_app({"TESTING": True}) client = app.test_client() r = client.get("/health") assert r.status_code == 200 assert r.get_json() == {"status": "ok"} ``` - [ ] **Step 2: Run test and verify failure** Run: `pytest components/l4d2-web-app/tests/test_health.py -q` Expected: FAIL. - [ ] **Step 3: Implement app factory and route** ```python from flask import Flask, jsonify def create_app(test_config=None): app = Flask(__name__) app.config.from_mapping( SECRET_KEY="dev", DATABASE_URL="sqlite:///l4d2web.db", STATUS_REFRESH_SECONDS=8, ) if test_config: app.config.update(test_config) @app.get("/health") def health(): return jsonify({"status": "ok"}) return app ``` - [ ] **Step 4: Run test and verify pass** Run: `pytest components/l4d2-web-app/tests/test_health.py -q` Expected: PASS. - [ ] **Step 5: Commit scaffold** ```bash git add components/l4d2-web-app git commit -m "feat(l4d2-web): scaffold flask app and health endpoint" ``` ### Task 2: Add database models and migration baseline **Files:** - Create: `components/l4d2-web-app/src/l4d2web/db.py` - Create: `components/l4d2-web-app/src/l4d2web/models.py` - Create: `components/l4d2-web-app/alembic.ini` - Create: `components/l4d2-web-app/alembic/env.py` - Create: `components/l4d2-web-app/alembic/versions/0001_initial.py` - Test: `components/l4d2-web-app/tests/test_models.py` - [ ] **Step 1: Write failing model tests** ```python from l4d2web.db import init_db, session_scope from l4d2web.models import User def test_create_user(tmp_path, monkeypatch): monkeypatch.setenv("DATABASE_URL", f"sqlite:///{tmp_path/'app.db'}") init_db() with session_scope() as s: user = User(username="alice", password_digest="digest", admin=False) s.add(user) s.flush() assert user.id is not None ``` - [ ] **Step 2: Run tests and verify failure** Run: `pytest components/l4d2-web-app/tests/test_models.py -q` Expected: FAIL. - [ ] **Step 3: Implement schema with Rails-style names** ```python class User(Base): __tablename__ = "users" id = mapped_column(Integer, primary_key=True) username = mapped_column(String(64), unique=True, nullable=False) password_digest = mapped_column(String(255), nullable=False) admin = mapped_column(Boolean, default=False, nullable=False) class Server(Base): __tablename__ = "servers" id = mapped_column(Integer, primary_key=True) user_id = mapped_column(ForeignKey("users.id"), nullable=False) name = mapped_column(String(128), unique=True, nullable=False) port = mapped_column(Integer, nullable=False) desired_state = mapped_column(String(16), default="stopped", nullable=False) actual_state = mapped_column(String(16), default="unknown", nullable=False) ``` - [ ] **Step 4: Run tests and verify pass** Run: `pytest components/l4d2-web-app/tests/test_models.py -q` Expected: PASS. - [ ] **Step 5: Commit models and migration** ```bash git add components/l4d2-web-app/src/l4d2web/{db.py,models.py} components/l4d2-web-app/alembic* git commit -m "feat(l4d2-web): add sqlite schema and migration baseline" ``` ### Task 3: Implement auth flows and admin bootstrap CLI **Files:** - Create: `components/l4d2-web-app/src/l4d2web/auth.py` - Create: `components/l4d2-web-app/src/l4d2web/routes/auth_routes.py` - Create: `components/l4d2-web-app/src/l4d2web/cli.py` - Modify: `components/l4d2-web-app/src/l4d2web/app.py` - Test: `components/l4d2-web-app/tests/test_auth.py` - [ ] **Step 1: Write failing auth tests** ```python def test_signup_creates_user(client): r = client.post("/signup", data={"username": "alice", "password": "secret"}) assert r.status_code == 302 def test_login_sets_session(client, seed_user): r = client.post("/login", data={"username": "alice", "password": "secret"}) assert r.status_code == 302 with client.session_transaction() as sess: assert sess["user_id"] == seed_user.id ``` - [ ] **Step 2: Run tests and verify failure** Run: `pytest components/l4d2-web-app/tests/test_auth.py -q` Expected: FAIL. - [ ] **Step 3: Implement auth and promote-admin command** ```python from werkzeug.security import generate_password_hash, check_password_hash 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) ``` ```python @click.command("promote-admin") @click.argument("username") def promote_admin(username: str): with session_scope() as s: user = s.scalar(select(User).where(User.username == username)) if not user: raise click.ClickException("user not found") user.admin = True s.commit() ``` - [ ] **Step 4: Run tests and verify pass** Run: `pytest components/l4d2-web-app/tests/test_auth.py -q` Expected: PASS. - [ ] **Step 5: Commit auth** ```bash git add components/l4d2-web-app/src/l4d2web/{auth.py,cli.py,app.py} components/l4d2-web-app/src/l4d2web/routes/auth_routes.py components/l4d2-web-app/tests/test_auth.py git commit -m "feat(l4d2-web): add signup/login and promote-admin bootstrap command" ``` ### Task 4: Implement overlays and server spec CRUD **Files:** - Create: `components/l4d2-web-app/src/l4d2web/routes/overlay_routes.py` - Create: `components/l4d2-web-app/src/l4d2web/routes/server_routes.py` - Test: `components/l4d2-web-app/tests/test_servers_overlays.py` - [ ] **Step 1: Write failing CRUD tests** ```python def test_admin_can_create_overlay(admin_client): r = admin_client.post("/admin/overlays", data={"name": "standard", "path": "/opt/l4d2/overlays/standard"}) assert r.status_code == 302 def test_server_create_persists_overlay_order(user_client, seed_overlays): payload = {"name": "alpha", "port": 27015, "overlay_ids": [2, 1], "arguments": [], "config": []} r = user_client.post("/servers", json=payload) assert r.status_code == 201 ``` - [ ] **Step 2: Run tests and verify failure** Run: `pytest components/l4d2-web-app/tests/test_servers_overlays.py -q` Expected: FAIL. - [ ] **Step 3: Implement routes with ownership rules** ```python @bp.post("/servers") @require_login def create_server(): payload = request.get_json() with session_scope() as s: server = Server( user_id=current_user().id, name=payload["name"], port=int(payload["port"]), arguments=json.dumps(payload.get("arguments", [])), config=json.dumps(payload.get("config", [])), ) s.add(server) s.flush() for position, overlay_id in enumerate(payload.get("overlay_ids", [])): s.add(ServerOverlay(server_id=server.id, overlay_id=overlay_id, position=position)) s.commit() return {"id": server.id}, 201 ``` - [ ] **Step 4: Run tests and verify pass** Run: `pytest components/l4d2-web-app/tests/test_servers_overlays.py -q` Expected: PASS. - [ ] **Step 5: Commit CRUD routes** ```bash git add components/l4d2-web-app/src/l4d2web/routes/{overlay_routes.py,server_routes.py} components/l4d2-web-app/tests/test_servers_overlays.py git commit -m "feat(l4d2-web): add overlay admin CRUD and server spec CRUD with ordering" ``` ### Task 5: Integrate direct `l4d2host` facade and YAML generation **Files:** - Create: `components/l4d2-web-app/src/l4d2web/services/spec_yaml.py` - Create: `components/l4d2-web-app/src/l4d2web/services/l4d2_facade.py` - Test: `components/l4d2-web-app/tests/test_l4d2_facade.py` - [ ] **Step 1: Write failing facade tests** ```python def test_initialize_writes_yaml_and_calls_library(monkeypatch, server_record): called = {} def fake_initialize(name, spec_path, **kwargs): called["name"] = name called["spec_path"] = str(spec_path) monkeypatch.setattr("l4d2web.services.l4d2_facade.initialize_instance", fake_initialize) from l4d2web.services.l4d2_facade import initialize_server initialize_server(server_record) assert called["name"] == server_record.name assert called["spec_path"].endswith(".yaml") ``` - [ ] **Step 2: Run tests and verify failure** Run: `pytest components/l4d2-web-app/tests/test_l4d2_facade.py -q` Expected: FAIL. - [ ] **Step 3: Implement facade and YAML generator** ```python def write_temp_spec(server, overlays: list[str]) -> Path: payload = { "port": server.port, "overlays": overlays, "arguments": json.loads(server.arguments), "config": json.loads(server.config), } f = tempfile.NamedTemporaryFile("w", suffix=".yaml", delete=False) yaml.safe_dump(payload, f, sort_keys=False) f.close() return Path(f.name) ``` - [ ] **Step 4: Run tests and verify pass** Run: `pytest components/l4d2-web-app/tests/test_l4d2_facade.py -q` Expected: PASS. - [ ] **Step 5: Commit facade** ```bash git add components/l4d2-web-app/src/l4d2web/services/{spec_yaml.py,l4d2_facade.py} components/l4d2-web-app/tests/test_l4d2_facade.py git commit -m "feat(l4d2-web): add direct l4d2host facade with generated yaml specs" ``` ### Task 6: Build scheduler, in-process worker pool, and recovery **Files:** - Create: `components/l4d2-web-app/src/l4d2web/services/job_worker.py` - Modify: `components/l4d2-web-app/src/l4d2web/app.py` - Test: `components/l4d2-web-app/tests/test_job_worker.py` - [ ] **Step 1: Write failing worker tests** ```python def test_same_server_jobs_serialized(worker_fixture): result = worker_fixture.run_once() assert result["same_server_parallel"] is False def test_install_is_global_exclusive(worker_fixture): result = worker_fixture.run_once() assert result["install_parallel"] is False def test_startup_recovers_stale_running_jobs(app): from l4d2web.services.job_worker import recover_stale_jobs recovered = recover_stale_jobs() assert recovered >= 0 ``` - [ ] **Step 2: Run tests and verify failure** Run: `pytest components/l4d2-web-app/tests/test_job_worker.py -q` Expected: FAIL. - [ ] **Step 3: Implement scheduler and lock rules** ```python class SchedulerState: def __init__(self): self.install_running = False self.running_servers: set[int] = set() def can_start(job, state: SchedulerState) -> bool: if job.operation == "install": return (not state.install_running) and (not state.running_servers) if state.install_running: return False return job.server_id not in state.running_servers ``` - [ ] **Step 4: Add single-process guard and stale job recovery** Run: `pytest components/l4d2-web-app/tests/test_job_worker.py -q` Expected: PASS. - [ ] **Step 5: Commit worker engine** ```bash git add components/l4d2-web-app/src/l4d2web/services/job_worker.py components/l4d2-web-app/src/l4d2web/app.py components/l4d2-web-app/tests/test_job_worker.py git commit -m "feat(l4d2-web): implement async scheduler, lock rules, and startup recovery" ``` ### Task 7: Persist command logs and add SSE job stream **Files:** - Create: `components/l4d2-web-app/src/l4d2web/routes/job_routes.py` - Modify: `components/l4d2-web-app/src/l4d2web/services/job_worker.py` - Test: `components/l4d2-web-app/tests/test_job_logs.py` - [ ] **Step 1: Write failing log tests** ```python def test_job_logs_seq_monotonic(db_session, finished_job): rows = db_session.execute(text("select seq from job_logs where job_id=:id order by seq"), {"id": finished_job.id}).all() assert rows == sorted(rows) def test_sse_resume_from_last_seq(client, seeded_job_logs): r = client.get(f"/jobs/{seeded_job_logs.job_id}/stream?last_seq=3") assert r.status_code == 200 ``` - [ ] **Step 2: Run tests and verify failure** Run: `pytest components/l4d2-web-app/tests/test_job_logs.py -q` Expected: FAIL. - [ ] **Step 3: Implement log persistence + SSE** ```python def append_job_log(session, job_id: int, stream: str, line: str) -> None: last_seq = session.scalar(select(func.max(JobLog.seq)).where(JobLog.job_id == job_id)) or 0 clipped = line[:4096] session.add(JobLog(job_id=job_id, seq=last_seq + 1, stream=stream, line=clipped)) session.flush() ``` - [ ] **Step 4: Run tests and verify pass** Run: `pytest components/l4d2-web-app/tests/test_job_logs.py -q` Expected: PASS. - [ ] **Step 5: Commit job logs** ```bash git add components/l4d2-web-app/src/l4d2web/{routes/job_routes.py,services/job_worker.py} components/l4d2-web-app/tests/test_job_logs.py git commit -m "feat(l4d2-web): persist command logs and stream job output over sse" ``` ### Task 8: Add SSE server runtime logs **Files:** - Create: `components/l4d2-web-app/src/l4d2web/routes/log_routes.py` - Test: `components/l4d2-web-app/tests/test_server_logs.py` - [ ] **Step 1: Write failing server-log tests** ```python def test_owner_can_stream_server_logs(owner_client, owned_server): r = owner_client.get(f"/servers/{owned_server.id}/logs/stream") assert r.status_code == 200 def test_non_owner_forbidden(user_client, foreign_server): r = user_client.get(f"/servers/{foreign_server.id}/logs/stream") assert r.status_code == 403 ``` - [ ] **Step 2: Run tests and verify failure** Run: `pytest components/l4d2-web-app/tests/test_server_logs.py -q` Expected: FAIL. - [ ] **Step 3: Implement runtime log stream route** ```python @bp.get("/servers//logs/stream") @require_login def stream_server_logs(server_id: int): server = load_authorized_server(server_id) def gen(): for line in facade.stream_server_logs(server.name, lines=200, follow=True): yield f"data: {line}\n\n" return Response(gen(), mimetype="text/event-stream") ``` - [ ] **Step 4: Run tests and verify pass** Run: `pytest components/l4d2-web-app/tests/test_server_logs.py -q` Expected: PASS. - [ ] **Step 5: Commit server log streaming** ```bash git add components/l4d2-web-app/src/l4d2web/routes/log_routes.py components/l4d2-web-app/tests/test_server_logs.py git commit -m "feat(l4d2-web): add live server log streaming with ownership checks" ``` ### Task 9: Implement desired-vs-actual state model and refresh **Files:** - Create: `components/l4d2-web-app/src/l4d2web/services/status.py` - Modify: `components/l4d2-web-app/src/l4d2web/services/job_worker.py` - Modify: `components/l4d2-web-app/src/l4d2web/routes/server_routes.py` - Test: `components/l4d2-web-app/tests/test_status.py` - [ ] **Step 1: Write failing status tests** ```python def test_display_state_priority(status_service): assert status_service.display_state(active_job_operation="start", actual_state="stopped") == "starting" def test_drift_badge(status_service): assert status_service.is_drift(desired_state="running", actual_state="stopped", has_active_job=False) is True ``` - [ ] **Step 2: Run tests and verify failure** Run: `pytest components/l4d2-web-app/tests/test_status.py -q` Expected: FAIL. - [ ] **Step 3: Implement state logic and refresh hooks** ```python def compute_display_state(active_operation, actual_state): priority = { "delete": "deleting", "start": "starting", "stop": "stopping", "initialize": "initializing", } if active_operation in priority: return priority[active_operation] return actual_state ``` - [ ] **Step 4: Implement 8-second actual-state refresh loop** Run: `pytest components/l4d2-web-app/tests/test_status.py -q` Expected: PASS. - [ ] **Step 5: Commit state model** ```bash git add components/l4d2-web-app/src/l4d2web/{services/status.py,services/job_worker.py,routes/server_routes.py} components/l4d2-web-app/tests/test_status.py git commit -m "feat(l4d2-web): add desired vs actual status model with 8s refresh" ``` ### Task 10: Apply security and reliability hardening **Files:** - Modify: `components/l4d2-web-app/src/l4d2web/app.py` - Modify: `components/l4d2-web-app/src/l4d2web/routes/auth_routes.py` - Modify: `components/l4d2-web-app/src/l4d2web/routes/overlay_routes.py` - Modify: `components/l4d2-web-app/src/l4d2web/db.py` - Test: `components/l4d2-web-app/tests/test_security.py` - [ ] **Step 1: Write failing hardening tests** ```python def test_csrf_required(client, seed_user): r = client.post("/servers", data={"name": "x"}) assert r.status_code == 400 def test_overlay_path_must_be_under_opt_l4d2_overlays(admin_client): r = admin_client.post("/admin/overlays", data={"name": "bad", "path": "/tmp/bad"}) assert r.status_code == 400 ``` - [ ] **Step 2: Run tests and verify failure** Run: `pytest components/l4d2-web-app/tests/test_security.py -q` Expected: FAIL. - [ ] **Step 3: Implement hardening** ```python app.config.update( SESSION_COOKIE_HTTPONLY=True, SESSION_COOKIE_SAMESITE="Lax", ) ``` ```python engine = create_engine(db_url, connect_args={"check_same_thread": False}) with engine.connect() as conn: conn.exec_driver_sql("PRAGMA journal_mode=WAL;") conn.exec_driver_sql("PRAGMA busy_timeout=5000;") ``` - [ ] **Step 4: Run tests and verify pass** Run: `pytest components/l4d2-web-app/tests/test_security.py -q` Expected: PASS. - [ ] **Step 5: Commit hardening** ```bash git add components/l4d2-web-app/src/l4d2web/{app.py,db.py} components/l4d2-web-app/src/l4d2web/routes/{auth_routes.py,overlay_routes.py} components/l4d2-web-app/tests/test_security.py git commit -m "feat(l4d2-web): add csrf, rate limits, sqlite wal, and path safety checks" ``` ### Task 11: Build minimal UI with vendored HTMX and custom CSS **Files:** - Create: `components/l4d2-web-app/src/l4d2web/templates/base.html` - Create: `components/l4d2-web-app/src/l4d2web/templates/dashboard.html` - Create: `components/l4d2-web-app/src/l4d2web/templates/server_detail.html` - Create: `components/l4d2-web-app/src/l4d2web/templates/admin_overlays.html` - Create: `components/l4d2-web-app/src/l4d2web/static/vendor/htmx.min.js` - Create: `components/l4d2-web-app/src/l4d2web/static/css/tokens.css` - Create: `components/l4d2-web-app/src/l4d2web/static/css/layout.css` - Create: `components/l4d2-web-app/src/l4d2web/static/css/components.css` - Create: `components/l4d2-web-app/src/l4d2web/static/css/logs.css` - Create: `components/l4d2-web-app/src/l4d2web/static/js/sse.js` - Test: `components/l4d2-web-app/tests/test_pages.py` - [ ] **Step 1: Write failing page tests** ```python def test_dashboard_renders_server_name(auth_client_with_server): r = auth_client_with_server.get("/dashboard") assert r.status_code == 200 assert "alpha" in r.get_data(as_text=True) def test_non_admin_cannot_open_overlay_admin(auth_client): r = auth_client.get("/admin/overlays") assert r.status_code == 403 ``` - [ ] **Step 2: Run tests and verify failure** Run: `pytest components/l4d2-web-app/tests/test_pages.py -q` Expected: FAIL. - [ ] **Step 3: Implement templates and style system** ```css /* tokens.css */ :root { --color-link: #0F766E; --color-text: #132028; --color-bg: #f7fbfa; --space-2: 0.5rem; --space-4: 1rem; --radius: 10px; } a { color: var(--color-link); } ``` - [ ] **Step 4: Run tests and verify pass** Run: `pytest components/l4d2-web-app/tests/test_pages.py -q` Expected: PASS. - [ ] **Step 5: Commit UI** ```bash git add components/l4d2-web-app/src/l4d2web/{templates,static} components/l4d2-web-app/tests/test_pages.py git commit -m "feat(l4d2-web): add minimal htmx ui with vendored assets and custom css" ``` ### Task 12: Final integration, docs, and full verification **Files:** - Create: `components/l4d2-web-app/README.md` - Modify: `components/l4d2-web-app/src/l4d2web/app.py` - Test: all tests under `components/l4d2-web-app/tests` - [ ] **Step 1: Write failing integration smoke test** ```python def test_core_routes_registered(app): routes = {r.rule for r in app.url_map.iter_rules()} assert "/dashboard" in routes assert "/jobs//stream" in routes ``` - [ ] **Step 2: Run test and verify failure (if wiring incomplete)** Run: `pytest components/l4d2-web-app/tests -q` Expected: FAIL or partial PASS before final wiring. - [ ] **Step 3: Finalize app wiring and README** README sections required: - deployment model (single process) - auth/admin bootstrap (`l4d2web promote-admin `) - scheduler lock semantics - desired vs actual model and 8s refresh - log architecture (`job_logs` forever, server logs from journald) - frontend dependency policy (vendored HTMX only, custom CSS) - [ ] **Step 4: Run full test suite** Run: `pytest components/l4d2-web-app/tests -q` Expected: PASS. - [ ] **Step 5: Commit finalization** ```bash git add components/l4d2-web-app git commit -m "docs(l4d2-web): finalize v1 architecture contracts and verification" ``` --- ## Self-Review - [ ] Spec coverage complete for approved constraints (auth, admin bootstrap, overlays, async jobs, status model, logging, minimal UI). - [ ] No placeholders (`TODO`, `TBD`, vague directives). - [ ] Naming consistency (`user_id`, `server_id`, `overlay_id`, `job_id`, `password_digest`, `admin`). - [ ] Every task includes concrete test commands and expected outcomes.