feat(l4d2-web): add live server logs and desired-vs-actual status model

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

@@ -9,6 +9,7 @@ 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.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.server_routes import bp as server_bp
 from l4d2web.services.job_worker import recover_stale_jobs
@@ -29,6 +30,7 @@ def create_app(test_config: dict[str, object] | None = None) -> Flask:
     app.register_blueprint(blueprint_bp)
     app.register_blueprint(server_bp)
     app.register_blueprint(job_bp)
+    app.register_blueprint(log_bp)
     register_cli(app)
     recover_stale_jobs()
 

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

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

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

@@ -5,7 +5,7 @@ from sqlalchemy import func, select
 from sqlalchemy.orm import Session
 
 from l4d2web.db import session_scope
-from l4d2web.models import Job, JobLog
+from l4d2web.models import Job, JobLog, Server
 
 
 @dataclass
@@ -47,3 +47,18 @@ def append_job_log(
     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/status.py

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

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

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