14 changed files with 13 additions and 1574 deletions
--- a/docs/superpowers/plans/2026-05-14-rcon-console.md
+++ b/docs/superpowers/plans/2026-05-14-rcon-console.md
@ -1,387 +0,0 @@
 # Add an RCON console to the server detail page
 ## Context
 The server detail page (`/servers/<id>`) already exposes the RCON password,
 live state polling, log streaming, and start/stop actions, but to send any
 arbitrary command (`changelevel`, `sm_kick`, `mp_*`, `say`, etc.) the user
 has to open a separate RCON client and reconnect. Adding an inline console
 turns the web UI into a complete operator tool for the owner of a server:
 type a command, see the reply, recall earlier commands via persisted
 history.
 Scope is intentionally narrow:
 - One server, one user (the owner). Multi-user shared console = not now.
 - Per-user history persisted across reloads.
 - No blocklist — owner already has the RCON password and can run anything
  via any RCON client; the UI is a thin wrapper.
 ## Design decisions (already settled)
 | Topic | Choice |
 |---|---|
 | UI placement | Panel on `server_detail.html`, between **Live State** and **Files**. |
 | Output transport | **HTMX append swap**, not SSE. RCON is request/response — SSE adds no value. Matches existing inline-form / `hx-swap` patterns in the codebase. |
 | Safety | Owner-of-server check only (`Server.user_id == current_user.id`). No command blocklist. **No admin override** — admins can already SSH if needed; an unaudited UI backdoor isn't worth the asymmetry. |
 | History | New `command_history` table, scoped per (user, server). Stores **command + reply + error flag** so the full transcript can be replayed on page reload. |
 | Transcript on page load | **Replays the last 50 rows** for this user+server, rendered server-side into the transcript via the same `_console_line.html` partial used for live additions. Visually identical to live lines (no "old vs new" distinction — the whole point is page-reload continuity). |
 | Transcript height | Fixed max-height ~400 px, internal vertical scroll. New lines auto-scroll to the bottom on add AND on initial load. Page layout below stays stable. |
 | Clear button | None. Reload doesn't help (it replays). If anyone wants to drop history, that's a separate concern handled later. |
 | RCON timeout | **30s per command.** Comfortably covers a cold map load with custom add-ons (community-observed worst case ~25s on modest hardware). 3× the python-valve default. Far below `director_transition_timeout` (120s) so no aliasing. If a command exceeds 30s, the RCON exec packet was already sent — the server still did the work; the user just doesn't see the textual reply but sees the effect in the Server Log SSE panel above. |
 | Worker model | Rely on `gunicorn --threads N` (or whatever the existing deployment uses for the long-lived SSE log streams). Threads share memory; one stuck `changelevel` holds a thread, not a process. Don't scale processes — adding hundreds of workers wastes RAM (~100 MB each); threads cost nothing. |
 ## Server-side changes
 ### 1. Extend `l4d2web/services/rcon.py`
 The wire-protocol layer already exists (`l4d2web/services/rcon.py:64`).
 Add a generic command executor with **multi-packet response handling**:
 ```python
 def execute_command(
    host: str, port: int, password: str, command: str, *, timeout: float = 30.0
 ) -> str:
    """Authenticate, send a single command, return the joined reply body.
    Implements the trailing-marker pattern: after the exec packet we
    immediately send an empty SERVERDATA_RESPONSE_VALUE packet with a
    sentinel req_id. We then read response packets, concatenating bodies,
    until we see the sentinel echo back. This is the only reliable way
    to detect end-of-output, because Source RCON splits replies >4096 B
    across multiple packets with no length header.
    """
 ```
 Implementation notes:
 - Factor `_connect_and_auth(sock, password)` out of `query_status` so
  both functions share the auth dance.
 - Use req_id `0xDEADBEEF` (or any constant ≠ the exec req_id) for the
  sentinel; read packets until one comes back with that req_id.
 - Input validation **inside this function** (not just at the route):
  - Reject empty / whitespace-only `command` → `ValueError`.
  - Reject embedded `\x00` bytes (would corrupt the null-terminated
    wire format) → `ValueError`.
  - Cap length at 1000 chars (RCON packet limit is 4096 incl. headers;
    no real command needs more). Longer → `ValueError`.
 - Trim trailing whitespace from the joined body. Otherwise return verbatim.
 - Existing `RconError` / `RconAuthError` exception types are reused.
 Tests in `l4d2web/tests/test_rcon.py` (extend the `FakeRconServer` to
 support multi-packet replies):
 - happy path: single-packet response
 - multi-packet response (synthesize a >4096 B reply)
 - empty reply (server replies only with the sentinel — case for `say`)
 - bad password → `RconAuthError`
 - timeout (fake server sleeps longer than the test timeout)
 - input validation: empty / null byte / oversized → `ValueError`
 ### 2. New `CommandHistory` model (`l4d2web/models.py`)
 Append at the bottom of `models.py`:
 ```python
 class CommandHistory(Base):
    __tablename__ = "command_history"
    __table_args__ = (
        Index("ix_cmdhist_user_server_id", "user_id", "server_id", "id"),
    )
    id:         Mapped[int]      = mapped_column(Integer, primary_key=True)
    user_id:    Mapped[int]      = mapped_column(ForeignKey("users.id"), nullable=False)
    server_id:  Mapped[int]      = mapped_column(ForeignKey("servers.id", ondelete="CASCADE"), nullable=False)
    command:    Mapped[str]      = mapped_column(Text, nullable=False)
    reply:      Mapped[str]      = mapped_column(Text, nullable=False, default="", server_default="")
    is_error:   Mapped[bool]     = mapped_column(Boolean, nullable=False, default=False, server_default=text("0"))
    created_at: Mapped[datetime] = mapped_column(DateTime, default=now_utc, nullable=False)
 ```
 Index `(user_id, server_id, id)` because every lookup is "latest N for
 this user+server", ordered by `id DESC`.
 A row is persisted on **every** RCON outcome — successful reply,
 empty reply, and error (auth fail, connect refused, `RconError`). The
 `is_error` flag drives the red styling on replay, so the transcript
 looks identical after a page reload.
 **Storage cost**: most replies are <500 B; `status` ~1 KB;
 `sm plugins list` a few KB; `cvarlist` can be 50 KB+. A power user
 running 100 commands/day at an average ~2 KB → ~73 MB/year. SQLite
 handles that without complaint; a trim job (cap N per user/server,
 e.g. last 5000) can be added if anyone ever notices.
 **Privacy note for the implementer**: replies from `status` include
 player names (user-controlled strings from random Steam users) and
 SteamID64s. Treat them as untrusted text on output (handled by Jinja
 auto-escaping — see §5) and don't surface them outside this user's
 session.
 ### 3. New alembic migration `0012_command_history.py`
 Mirror `l4d2web/alembic/versions/0011_server_hostname.py`:
 - `revision = "0012_command_history"`
 - `down_revision = "0011_server_hostname"`
 - `upgrade()`: `op.create_table("command_history", …)` with columns
  `id`, `user_id`, `server_id`, `command (Text)`, `reply (Text, server_default="")`,
  `is_error (Boolean, server_default="0")`, `created_at`; plus
  `op.create_index("ix_cmdhist_user_server_id", ...)`.
 - `downgrade()`: drop index then table.
 - `test_alembic_migrations.py` auto-discovers revisions (skim once to
  confirm; no edit if so).
 ### 4. New route module `l4d2web/routes/console_routes.py`
 Two endpoints, both `@require_login`, both verify ownership with
 **404** on miss (matches the existing pattern at
 `page_routes.py:303` — no admin backdoor).
 **`POST /servers/<id>/console`** — submit a command.
 - CSRF-checked (form field `csrf_token`).
 - Form field `command`. Validation happens twice: at the route (return a
  user-facing error fragment for empty / oversized) and inside
  `execute_command` (defence in depth — never trust a single layer).
 - Calls
  `rcon.execute_command("127.0.0.1", server.port, server.rcon_password, command)`.
 - **Every outcome persists a `CommandHistory` row** (so the transcript
  fully reconstructs on page reload):
  - Success with reply → `command`, `reply`, `is_error=False`.
  - Success with empty reply (e.g. `say`) → `command`, `reply=""`,
    `is_error=False`. Template renders `(no reply)` in muted text.
  - `RconAuthError` / `RconError` / connect-failed → `command`,
    `reply=<exception message>`, `is_error=True`. Red styling on render.
 - On `ValueError` from input validation (empty / null byte / oversized):
  render an error fragment, **do not** insert history (the command
  never reached the wire — nothing happened to remember).
 - Returns 200 in all cases (errors are rendered, not raised) so HTMX
  appends them to the transcript like any other line.
 **`GET /servers/<id>/console/history?before=<id>&limit=50`** — paged
 history for up-arrow navigation.
 - Returns JSON `[{"id": …, "command": …}, …]` ordered newest-first.
 - The client owns the input state; this stays JSON, not HTML.
 - `limit` clamped to ≤200.
 Register the blueprint in `l4d2web/app.py` alongside the other
 `*_routes` modules.
 **Also extend `server_detail()` in `page_routes.py`** to fetch the last
 50 `CommandHistory` rows for this `(user, server)`, ordered oldest-first
 (so they iterate naturally in the template), and pass as
 `console_history` in the render context. Use the same `session_scope`
 block that already loads `server` and `blueprint` (`page_routes.py:301`)
 — one extra `db.scalars(select(CommandHistory)…)` call, no new round
 trip cost.
 ### 5. Template fragment `templates/_console_line.html`
 ```jinja2
 <div class="console-line{% if error %} console-error{% endif %}">
  <div class="console-prompt">&gt; {{ command }}</div>
  {% if reply %}
    <pre class="console-reply">{{ reply }}</pre>
  {% else %}
    <div class="console-reply muted">(no reply)</div>
  {% endif %}
 </div>
 ```
 **XSS reminder for the implementer:** `reply` originates from the game
 server's RCON output — we do not trust it. **Never use `|safe`**, never
 `{{ reply|markdown }}`, never anything that bypasses Jinja's default
 HTML escaping. The existing `{{ reply }}` is the right call.
 ### 6. Console panel in `templates/server_detail.html`
 Insert between the existing live-state section (line 33–37) and the
 Files section (line 39):
 ```jinja2
 <h2 class="section-title">Console</h2>
 <section class="panel console-panel">
  <div id="console-transcript-{{ server.id }}"
       class="console-transcript"
       data-autoscroll>
    {% for h in console_history %}
      {% include "_console_line.html" with context %}
      {# Loops with h.command, h.reply, h.is_error, h.created_at #}
    {% endfor %}
  </div>
  <form hx-post="/servers/{{ server.id }}/console"
        hx-target="#console-transcript-{{ server.id }}"
        hx-swap="beforeend"
        hx-indicator=".console-spinner"
        hx-on::after-request="this.command.value=''; this.command.focus(); this.closest('section').querySelector('[data-autoscroll]').scrollTop = 1e9"
        class="console-input-form"
        data-console-form data-server-id="{{ server.id }}">
    <input type="hidden" name="csrf_token" value="{{ session.get('csrf_token', '') }}">
    <span class="console-prompt-glyph">&gt;</span>
    <input name="command" autocomplete="off" spellcheck="false" maxlength="1000"
           placeholder="status, changelevel c1m1_hotel, sm_kick …">
    <span class="console-spinner" aria-hidden="true">…</span>
    <button type="submit">Send</button>
  </form>
 </section>
 ```
 - Transcript is server-side rendered with the last 50 history rows on
  page load. `_console_line.html` is the single source of truth for
  line layout — same template, same look, whether the line came from
  this session or last week.
 - `hx-indicator` gives visible feedback during slow commands (a
  `changelevel` can sit at ~10s+).
 - `maxlength="1000"` on the input mirrors the server-side cap.
 - The `hx-on::after-request` inline scrolls the transcript to the
  bottom after each new line. On initial page load, the JS module
  scrolls to the bottom once after the DOM is ready (so the most
  recent history is visible, not the oldest).
 **Cross-feature interaction (do not "fix"):** Silent or slow commands
 (`say`, `kick`, `changelevel`) will produce empty or terse RCON replies
 in this transcript. The actual game-side effect is already visible in
 the **Server Log** SSE panel right above. A future implementer should
 NOT try to mirror server-log lines back into the console transcript —
 that's a redundancy, not a feature.
 ### 7. New `static/js/console-history.js`
 Tiny module bound to `[data-console-form]`:
 - **On DOM ready**: scroll each `[data-autoscroll]` transcript to the
  bottom so the most recent replayed lines are visible. This is the
  initial-load equivalent of the `hx-on::after-request` scroll.
 - **On first focus** of the input: lazy-fetch
  `/servers/<id>/console/history?limit=50` and cache the array in
  memory. (Distinct from the rendered-on-load transcript: this cache
  is *just commands* for up/down recall — replies don't matter for
  navigation, so the JSON endpoint stays narrow.)
 - **ArrowUp / ArrowDown**: walk the cached array, set `input.value`.
  - ArrowUp from a non-history state: snapshot the current value so
    ArrowDown can restore it.
 - **ArrowUp past the end**: fetch next page using
  `?before=<oldest_cached_id>`. If empty, stop.
 - **After a successful submit** (`htmx:afterRequest` with 2xx):
  prepend the just-sent command to the in-memory cache so it's
  instantly recallable.
 Loaded via a `<script defer>` line in `base.html` next to the other
 small static JS modules (same pattern as `sse.js`).
 ### 8. Concurrency sanity (no code, just verifying the design)
 `live_state_poller.py` already opens fresh RCON connections every 5s
 against the same port. SrcDS handles concurrent RCON sessions cleanly
 (each is independently auth'd, no shared state). The console adds at
 most one more concurrent connection per active user — well within
 limits. No locking needed.
 ### 9. Minimal CSS in `static/css/`
 Monospace transcript, dark background, `console-error` styled like the
 existing error pills. Match the visual weight of the existing log-stream
 `<pre>` block on the detail page — no new design system.
 ## Files to touch
 | File | Change |
 |---|---|
 | `l4d2web/services/rcon.py` | Add `execute_command()` with multi-packet handling + input validation; extract `_connect_and_auth()` |
 | `l4d2web/tests/test_rcon.py` | Extend `FakeRconServer` for multi-packet; add success / multi-packet / empty / bad-pw / timeout / validation tests |
 | `l4d2web/models.py` | Add `CommandHistory` (with `reply`, `is_error`) |
 | `l4d2web/alembic/versions/0012_command_history.py` | New migration |
 | `l4d2web/routes/console_routes.py` | **NEW** — POST + GET endpoints |
 | `l4d2web/routes/page_routes.py` | Extend `server_detail()` to fetch last 50 history rows and pass `console_history` |
 | `l4d2web/app.py` | Register the new blueprint |
 | `l4d2web/templates/_console_line.html` | **NEW** fragment |
 | `l4d2web/templates/server_detail.html` | Insert console panel section (with server-rendered replay loop) |
 | `l4d2web/static/js/console-history.js` | **NEW** up/down history nav + initial scroll-to-bottom |
 | `l4d2web/templates/base.html` | `<script defer src="…/console-history.js">` |
 | `l4d2web/static/css/*.css` | Console panel styling (fixed-height scroll transcript, error variant) |
 | `l4d2web/tests/test_console_routes.py` | **NEW** route tests |
 ## Tests to write explicitly
 **`test_rcon.py`** (extending existing file):
 - `execute_command` happy path, single-packet reply
 - `execute_command` multi-packet reply (>4096 B) reassembled in order
 - `execute_command` empty reply (server returns only the sentinel)
 - `execute_command` bad password → `RconAuthError`
 - `execute_command` socket timeout → `RconError`
 - Input validation: empty / whitespace-only / null-byte / oversized → `ValueError`
 **`test_console_routes.py`** (new):
 - not logged in → 302 to login
 - logged in but not server owner → **404** (not 403 — match
  `page_routes.py:303`)
 - valid command → 200, fragment HTML rendered, `CommandHistory` row
  inserted with `reply` populated and `is_error=False`
 - empty RCON reply → 200, fragment renders `(no reply)`, history row
  inserted with `reply=""`, `is_error=False`
 - RCON error (mock `execute_command` to raise) → 200, error fragment,
  history row inserted with `is_error=True` and the exception message
  in `reply`
 - empty/oversized command (validation error before wire) → 200, error
  fragment, **no** history row
 - CSRF token missing → rejected
 - `GET /console/history` returns newest-first
 - `GET /console/history?before=<id>` paginates correctly
 - `GET /console/history?limit=10000` is clamped to ≤200
 **`test_page_routes.py`** (extend existing if present, otherwise add):
 - `server_detail` returns the last 50 `CommandHistory` rows for the
  viewing user only, oldest-first in the rendered page (newest at the
  bottom of the transcript)
 - a history row belonging to another user for the same server is **not**
  visible (ownership scoping is by `user_id`, not just `server_id`)
 ## What we are deliberately NOT doing
 - No command blocklist or admin gate — owner already has the password.
 - **No admin override** to console other users' servers (admins can SSH if
  they truly need to; UI backdoor would be unaudited and asymmetric).
 - No shared multi-user view of the same console.
 - No streaming output (RCON doesn't stream; replies are one-shot).
 - No autocomplete of cvars — out of scope; up-arrow history is enough.
 - No "Clear transcript" button — the transcript replays on every page
  load by design. Discarding history is a different concern (delete
  rows from the DB) and is out of scope for v1.
 - No history-trim job — file an issue if anyone hits >100k rows; not
  worth pre-empting at this scale.
 - No mirroring of server-log lines into the console transcript — the
  Server Log panel above already serves that purpose.
 ## Verification
 1. `pytest l4d2web/tests/test_rcon.py l4d2web/tests/test_console_routes.py l4d2web/tests/test_alembic_migrations.py` — unit + migration tests pass.
 2. Boot the web app locally, log in, open a server detail page for a
   running server, send `status` — multi-line reply renders in the
   transcript; the input clears and refocuses; spinner shows during
   the request; the transcript scrolls to the new line at the bottom.
 3. Send `cvarlist` — a large multi-packet response — and confirm the
   full output reassembles, not truncated.
 4. Send `say hello` — transcript shows `> say hello` followed by
   `(no reply)` in muted text; the line appears in the Server Log
   panel above.
 5. Send `changelevel c1m1_hotel` — request takes ~10–20s, spinner
   visible the whole time, then a (likely empty) reply appears, and
   the live-state panel updates to the new map within 5s.
 6. Send an invalid command (e.g. `nonsense_cvar`) — reply renders
   normally (RCON tolerates unknown commands).
 7. Send a command with embedded null bytes (via curl, since the
   browser strips them) — returns 200 with an error fragment, no
   history row.
 8. Send a 2000-char command — rejected with an error fragment, no
   history row.
 9. **Reload the page** — the transcript reappears identical to before,
    showing the same `> status`, `> say hello`, `> nonsense_cvar` lines
    with their replies, scrolled to the bottom. Errors are still red.
 10. Focus the input, press ArrowUp — the previous command reappears.
    ArrowDown restores the empty state.
 11. Send 60+ commands, then ArrowUp past the in-memory page boundary —
    older commands load on demand.
 12. Stop the server, try to send a command — surfaces as a styled
    `console-error` line ("connect failed") rather than a 500; **a
    history row IS inserted** with `is_error=True`, so the error
    replays on next page load.
 13. Log in as a different user, visit `/servers/<other-user-id>` —
    404, no console rendered. POST to that URL also 404. The other
    user's transcript is not visible.
 14. Confirm that a `cvarlist`-class large reply persists fully in the
    DB (`SELECT length(reply) FROM command_history ORDER BY id DESC LIMIT 1;`)
    and replays in full on page reload.
--- a/l4d2web/alembic/versions/0012_command_history.py
+++ b/l4d2web/alembic/versions/0012_command_history.py
@ -1,48 +0,0 @@
 """add command_history table
 Revision ID: 0012_command_history
 Revises: 0011_server_hostname
 Create Date: 2026-05-14
 """
 from __future__ import annotations
 from typing import Sequence, Union
 import sqlalchemy as sa
 from alembic import op
 revision: str = "0012_command_history"
 down_revision: Union[str, Sequence[str], None] = "0011_server_hostname"
 branch_labels: Union[str, Sequence[str], None] = None
 depends_on: Union[str, Sequence[str], None] = None
 def upgrade() -> None:
    op.create_table(
        "command_history",
        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", ondelete="CASCADE"),
            nullable=False,
        ),
        sa.Column("command", sa.Text(), nullable=False),
        sa.Column("reply", sa.Text(), nullable=False, server_default=""),
        sa.Column(
            "is_error", sa.Boolean(), nullable=False, server_default=sa.text("0")
        ),
        sa.Column("created_at", sa.DateTime(), nullable=False),
    )
    op.create_index(
        "ix_cmdhist_user_server_id",
        "command_history",
        ["user_id", "server_id", "id"],
    )
 def downgrade() -> None:
    op.drop_index("ix_cmdhist_user_server_id", table_name="command_history")
    op.drop_table("command_history")
--- a/l4d2web/app.py
+++ b/l4d2web/app.py
@ -11,7 +11,6 @@ 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.console_routes import bp as console_bp
 from l4d2web.routes.files_routes import bp as files_bp
 from l4d2web.routes.job_routes import bp as job_bp
 from l4d2web.routes.log_routes import bp as log_bp
@ -85,7 +84,6 @@ def create_app(test_config: dict[str, object] | None = None) -> Flask:
    app.register_blueprint(server_bp)
    app.register_blueprint(job_bp)
    app.register_blueprint(log_bp)
    app.register_blueprint(console_bp)
    app.register_blueprint(page_bp)
    app.register_blueprint(profile_bp)
    register_cli(app)
--- a/l4d2web/models.py
+++ b/l4d2web/models.py
@ -227,26 +227,3 @@ class SteamUserProfile(Base):
    persona_name: Mapped[str] = mapped_column(String(64), nullable=False)
    avatar_url: Mapped[str] = mapped_column(Text, nullable=False)
    fetched_at: Mapped[datetime] = mapped_column(DateTime, nullable=False)
 class CommandHistory(Base):
    __tablename__ = "command_history"
    __table_args__ = (
        Index("ix_cmdhist_user_server_id", "user_id", "server_id", "id"),
    )
    id:         Mapped[int]      = mapped_column(Integer, primary_key=True)
    user_id:    Mapped[int]      = mapped_column(ForeignKey("users.id"), nullable=False)
    server_id:  Mapped[int]      = mapped_column(
        ForeignKey("servers.id", ondelete="CASCADE"), nullable=False
    )
    command:    Mapped[str]      = mapped_column(Text, nullable=False)
    reply:      Mapped[str]      = mapped_column(
        Text, nullable=False, default="", server_default=""
    )
    is_error:   Mapped[bool]     = mapped_column(
        Boolean, nullable=False, default=False, server_default=text("0")
    )
    created_at: Mapped[datetime] = mapped_column(
        DateTime, default=now_utc, nullable=False
    )
--- a/l4d2web/routes/console_routes.py
+++ b/l4d2web/routes/console_routes.py
@ -1,104 +0,0 @@
 from flask import Blueprint, Response, jsonify, render_template, request
 from sqlalchemy import select
 from l4d2web.auth import current_user, require_login
 from l4d2web.db import session_scope
 from l4d2web.models import CommandHistory, Server
 from l4d2web.services import rcon
 from l4d2web.services.rcon import RconAuthError, RconError
 bp = Blueprint("console", __name__)
 _HISTORY_DEFAULT_LIMIT = 50
 _HISTORY_MAX_LIMIT = 200
 _HISTORY_MIN_LIMIT = 1
@bp.post("/servers/<int:server_id>/console")
@require_login
 def run_console_command(server_id: int) -> Response:
    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)
        command_raw = request.form.get("command", "")
        command = command_raw.strip()
        try:
            reply = rcon.execute_command("127.0.0.1", server.port, server.rcon_password, command)
            is_error = False
        except (RconAuthError, RconError) as exc:
            reply = str(exc)
            is_error = True
        except ValueError:
            # Input validation failure — command never reached the wire; no history row.
            return render_template(
                "_console_line.html",
                command=command,
                reply="invalid command",
                is_error=True,
            )
        db.add(
            CommandHistory(
                user_id=user.id,
                server_id=server_id,
                command=command,
                reply=reply,
                is_error=is_error,
            )
        )
    return render_template(
        "_console_line.html",
        command=command,
        reply=reply,
        is_error=is_error,
    )
@bp.get("/servers/<int:server_id>/console/history")
@require_login
 def console_history(server_id: int) -> Response:
    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)
        try:
            raw_before = request.args.get("before")
            before = int(raw_before) if raw_before is not None else None
        except (TypeError, ValueError):
            before = None
        try:
            raw_limit = request.args.get("limit")
            limit = int(raw_limit) if raw_limit is not None else _HISTORY_DEFAULT_LIMIT
        except (TypeError, ValueError):
            limit = _HISTORY_DEFAULT_LIMIT
        limit = max(_HISTORY_MIN_LIMIT, min(_HISTORY_MAX_LIMIT, limit))
        query = select(CommandHistory).where(
            CommandHistory.user_id == user.id,
            CommandHistory.server_id == server_id,
        )
        if before is not None:
            query = query.where(CommandHistory.id < before)
        query = query.order_by(CommandHistory.id.desc()).limit(limit)
        rows = db.scalars(query).all()
    return jsonify([{"id": row.id, "command": row.command} for row in rows])
--- a/l4d2web/routes/page_routes.py
+++ b/l4d2web/routes/page_routes.py
@ -9,7 +9,6 @@ from l4d2web.db import session_scope
 from l4d2web.models import Blueprint as BlueprintModel
 from l4d2web.models import (
    BlueprintOverlay,
    CommandHistory,
    Job,
    Overlay,
    OverlayWorkshopItem,
@ -304,19 +303,6 @@ def server_detail(server_id: int):
            return Response(status=404)
        blueprint = db.scalar(select(BlueprintModel).where(BlueprintModel.id == server.blueprint_id))
        ctx = _build_server_actions_context(db, server)
        console_history = list(
            reversed(
                db.scalars(
                    select(CommandHistory)
                    .where(
                        CommandHistory.user_id == user.id,
                        CommandHistory.server_id == server.id,
                    )
                    .order_by(CommandHistory.id.desc())
                    .limit(50)
                ).all()
            )
        )
    connect_host = request.host.split(":")[0]
    file_tree_root_entries, file_tree_truncated_count = _root_server_file_tree(server_id)
@ -331,7 +317,6 @@ def server_detail(server_id: int):
        if file_tree_root_entries is not None
        else False,
        file_tree_truncated_count=file_tree_truncated_count,
        console_history=console_history,
        **ctx,
    )
--- a/l4d2web/services/rcon.py
+++ b/l4d2web/services/rcon.py
@ -32,12 +32,6 @@ SERVERDATA_EXECCOMMAND = 2
 SERVERDATA_AUTH_RESPONSE = 2
 SERVERDATA_RESPONSE_VALUE = 0
 # req_id values for execute_command's exec + marker packets.
 # These are arbitrary positive ints chosen by the client; the values
 # happen to be unrelated to the SERVERDATA_* packet-type constants.
 _EXEC_REQ_ID = 2
 _MARKER_REQ_ID = 0xC0DE
 _STEAM_ID_BASE = 76561197960265728
@ -67,32 +61,6 @@ class StatusResponse:
    roster: list[PlayerRow]
 def _connect_and_auth(
    sock: socket.socket, host: str, port: int, password: str
 ) -> None:
    """Open TCP connection and authenticate.
    Raises:
      RconError      – on connect failure
      RconAuthError  – when the server returns req_id == -1 (bad password)
      OSError / socket.timeout – raw, from the post-connect send/recv;
                                 callers wrap these into RconError.
    """
    try:
        sock.connect((host, port))
    except OSError as exc:
        raise RconError(f"connect failed: {exc}") from exc
    _send_packet(sock, 1, SERVERDATA_AUTH, password)
    # The server always sends a leading empty type-0 packet before the real
    # AUTH_RESPONSE (type-2).  Drain whichever arrives first.
    r1 = _recv_packet(sock)
    r2 = _recv_packet(sock)
    auth = r2 if r1[1] == SERVERDATA_RESPONSE_VALUE else r1
    if auth[0] == -1:
        raise RconAuthError("bad rcon password")
 def query_status(
    host: str, port: int, password: str, *, timeout: float = 2.0
 ) -> StatusResponse:
@ -101,7 +69,19 @@ def query_status(
    sock.settimeout(timeout)
    try:
        try:
-            _connect_and_auth(sock, host, port, password)
+            sock.connect((host, port))
        except OSError as exc:
            raise RconError(f"connect failed: {exc}") from exc
        try:
            _send_packet(sock, 1, SERVERDATA_AUTH, password)
            # Drain the leading empty type-0 packet; then read the real auth response.
            r1 = _recv_packet(sock)
            r2 = _recv_packet(sock)
            auth = r2 if r1[1] == SERVERDATA_RESPONSE_VALUE else r1
            if auth[0] == -1:
                raise RconAuthError("bad rcon password")
            _send_packet(sock, 2, SERVERDATA_EXECCOMMAND, "status")
            _, _, body = _recv_packet(sock)
        except (OSError, socket.timeout) as exc:
@ -112,50 +92,6 @@ def query_status(
    return parse_status(body)
 def execute_command(
    host: str, port: int, password: str, command: str, *, timeout: float = 30.0
 ) -> str:
    """Authenticate, send a single command, return the joined reply body.
    Uses the trailing-marker pattern: after the exec packet, send an empty
    SERVERDATA_RESPONSE_VALUE packet with a sentinel req_id. Read response
    packets, accumulating bodies, until we see one whose req_id matches the
    sentinel — that guarantees the real reply is complete, because RCON
    processes requests in receive order. Multi-packet replies (>4096 B, like
    `cvarlist`) are reassembled this way.
    """
    if not command or not command.strip():
        raise ValueError("command must not be empty or whitespace-only")
    if "\x00" in command:
        raise ValueError("command must not contain null bytes")
    if len(command.encode("utf-8")) > 1000:
        raise ValueError("command exceeds maximum byte length of 1000")
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.settimeout(timeout)
    try:
        try:
            _connect_and_auth(sock, host, port, password)
            _send_packet(sock, _EXEC_REQ_ID, SERVERDATA_EXECCOMMAND, command)
            # Trailing marker: srcds processes requests in order, so its echo
            # of this empty packet arrives after all real command-output packets.
            _send_packet(sock, _MARKER_REQ_ID, SERVERDATA_RESPONSE_VALUE, "")
            chunks: list[str] = []
            while True:
                req_id, _, body = _recv_packet(sock)
                if req_id == _MARKER_REQ_ID:
                    break
                chunks.append(body)
        except (OSError, socket.timeout) as exc:
            raise RconError(f"rcon i/o error: {exc}") from exc
    finally:
        sock.close()
    return "".join(chunks).rstrip()
 def _send_packet(sock: socket.socket, req_id: int, ptype: int, body: str) -> None:
    body_bytes = body.encode("utf-8") + b"\x00\x00"
    size = 4 + 4 + len(body_bytes)
--- a/l4d2web/static/css/components.css
+++ b/l4d2web/static/css/components.css
@ -912,82 +912,3 @@ dialog.modal.modal-wide {
 .live-state .server-live-summary {
  font-size: 1.05em;
 }
 /* ============================================================
   RCON console panel — server detail page
   ============================================================ */
 .console-transcript {
  max-height: 400px;
  overflow-y: auto;
  background: var(--color-log-bg);
  color: var(--color-log-text);
  border-radius: var(--radius-s);
  padding: var(--space-m);
  font-family: ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, monospace;
  font-size: 0.875rem;
  line-height: 1.4;
  margin-bottom: var(--space-m);
 }
 .console-line {
  margin: var(--space-s) 0;
 }
 .console-line:first-child {
  margin-top: 0;
 }
 .console-prompt {
  font-weight: 600;
  color: var(--color-primary);
 }
 .console-reply {
  margin: 0;
  font-family: inherit;
  font-size: inherit;
  white-space: pre-wrap;
  color: var(--color-muted);
  border: none;
  background: none;
  padding: 0;
 }
 .console-error .console-prompt {
  color: var(--color-danger);
 }
 .console-error {
  border-left: 3px solid var(--color-danger);
  padding-left: var(--space-s);
 }
 .console-input-form {
  display: flex;
  align-items: center;
  gap: var(--space-s);
 }
 .console-prompt-glyph {
  font-family: ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, monospace;
  font-weight: 600;
  color: var(--color-primary);
  flex-shrink: 0;
 }
 .console-input-form input[name="command"] {
  flex: 1;
  min-width: 0;
 }
 .console-spinner {
  display: none;
  color: var(--color-muted);
  font-family: ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, monospace;
  flex-shrink: 0;
 }
 .console-input-form.htmx-request .console-spinner {
  display: inline;
 }
--- a/l4d2web/static/js/console-history.js
+++ b/l4d2web/static/js/console-history.js
@ -1,179 +0,0 @@
 // console-history.js
 // Binds ArrowUp/Down history recall to [data-console-form] elements.
 // Mirrors the style of sse.js: vanilla JS, dataset attributes, no framework.
 function bindConsoleForm(form) {
  if (form.dataset.consoleHistoryBound === "true") {
    return;
  }
  form.dataset.consoleHistoryBound = "true";
  const serverId = form.dataset.serverId;
  const input = form.querySelector("input[name='command']");
  if (!input || !serverId) {
    return;
  }
  // --- History cache state ---
  // Entries are stored newest-first to match the API response shape.
  let cache = [];
  let cacheLoaded = false;
  let loadingPromise = null;
  let cursor = -1;       // -1 = "not in history" (at the live input)
  let snapshot = "";     // saved input value for restoring via ArrowDown
  let oldestId = null;   // id of the oldest cached entry, used for pagination
  let exhausted = false; // true when we've fetched all available history
  let pendingCommand = null; // captured before HTMX submit; used by afterRequest
  async function loadHistory(params) {
    const url = new URL(`/servers/${serverId}/console/history`, location.origin);
    url.searchParams.set("limit", "50");
    if (params && params.before != null) {
      url.searchParams.set("before", params.before);
    }
    try {
      const res = await fetch(url.toString());
      if (!res.ok) return [];
      return await res.json();
    } catch {
      return [];
    }
  }
  async function ensureLoaded() {
    if (cacheLoaded) return;
    if (!loadingPromise) {
      loadingPromise = (async () => {
        const entries = await loadHistory();
        cache = entries; // newest-first from API
        if (cache.length > 0) {
          oldestId = cache[cache.length - 1].id;
        }
        if (entries.length < 50) {
          exhausted = true;
        }
        cacheLoaded = true;
      })();
    }
    await loadingPromise;
  }
  async function fetchOlderPage() {
    if (exhausted || oldestId == null) return;
    const entries = await loadHistory({ before: oldestId });
    if (entries.length === 0) {
      exhausted = true;
      return;
    }
    cache = cache.concat(entries); // append older entries at the end
    oldestId = cache[cache.length - 1].id;
    if (entries.length < 50) {
      exhausted = true;
    }
  }
  input.addEventListener("focus", () => {
    ensureLoaded();
  }, { once: true });
  input.addEventListener("keydown", async (event) => {
    if (event.key !== "ArrowUp" && event.key !== "ArrowDown") {
      return;
    }
    event.preventDefault();
    // Ensure we have history before navigating.
    await ensureLoaded();
    if (event.key === "ArrowUp") {
      if (cursor === -1) {
        // Entering history from live input — save whatever is typed.
        snapshot = input.value;
      }
      const nextCursor = cursor + 1;
      // If we've reached the end of cache, try fetching an older page.
      if (nextCursor >= cache.length) {
        await fetchOlderPage();
      }
      if (nextCursor < cache.length) {
        cursor = nextCursor;
        input.value = cache[cursor].command;
      }
      // else: already at oldest end, stay put.
    } else {
      // ArrowDown
      if (cursor <= -1) {
        // Already at live input, nothing to do.
        return;
      }
      const nextCursor = cursor - 1;
      if (nextCursor < 0) {
        // Return to live input, restore snapshot.
        cursor = -1;
        input.value = snapshot;
      } else {
        cursor = nextCursor;
        input.value = cache[cursor].command;
      }
    }
  });
  // Capture the command value before HTMX submits the form; avoids relying on
  // event.detail.requestConfig.parameters which changed between HTMX 1.x / 2.x.
  form.addEventListener("htmx:beforeRequest", () => {
    const commandInput = form.querySelector("input[name='command']");
    pendingCommand = commandInput ? commandInput.value : null;
  });
  // After a successful HTMX POST: prepend the sent command to cache.
  form.addEventListener("htmx:afterRequest", (event) => {
    if (!event.detail.successful) return;
    // Use the value captured before the request; fall back to snapshot.
    const command = pendingCommand || snapshot || "";
    pendingCommand = null;
    if (!command) return;
    // Prepend as the newest entry.
    // id=null is safe here: pagination uses oldestId (the real persisted-row id)
    // for ?before= queries, so a synthetic null-id entry doesn't break paging.
    cache.unshift({ id: null, command });
    // Reset cursor so ArrowUp immediately recalls this command.
    cursor = -1;
    snapshot = "";
  });
 }
 function bindAllConsoleForms(root) {
  if (!root) return;
  const scope = root.matches && root.matches("[data-console-form]") ? [root] : [];
  if (root.querySelectorAll) {
    root.querySelectorAll("[data-console-form]").forEach((el) => scope.push(el));
  }
  scope.forEach(bindConsoleForm);
 }
 function scrollConsolesToBottom(root) {
  if (!root) return;
  const scope = root.matches && root.matches("[data-autoscroll]") ? [root] : [];
  if (root.querySelectorAll) {
    root.querySelectorAll("[data-autoscroll]").forEach((el) => scope.push(el));
  }
  scope.forEach((el) => {
    el.scrollTop = el.scrollHeight;
  });
 }
 document.addEventListener("DOMContentLoaded", () => {
  scrollConsolesToBottom(document);
  bindAllConsoleForms(document);
 });
 // Support HTMX-injected content (mirrors sse.js pattern).
 document.addEventListener("htmx:load", (event) => {
  scrollConsolesToBottom(event.detail.elt);
  bindAllConsoleForms(event.detail.elt);
 });
--- a/l4d2web/templates/_console_line.html
+++ b/l4d2web/templates/_console_line.html
@ -1,8 +0,0 @@
 <div class="console-line{% if is_error %} console-error{% endif %}">
  <div class="console-prompt">&gt; {{ command }}</div>
  {% if reply %}
    <pre class="console-reply">{{ reply }}</pre>
  {% else %}
    <div class="console-reply muted">(no reply)</div>
  {% endif %}
 </div>
--- a/l4d2web/templates/base.html
+++ b/l4d2web/templates/base.html
@ -42,6 +42,5 @@
    <script src="{{ url_for('static', filename='js/modal.js') }}"></script>
    <script src="{{ url_for('static', filename='js/file-tree.js') }}"></script>
    <script src="{{ url_for('static', filename='js/password-reveal.js') }}"></script>
    <script defer src="{{ url_for('static', filename='js/console-history.js') }}"></script>
  </body>
 </html>
--- a/l4d2web/templates/server_detail.html
+++ b/l4d2web/templates/server_detail.html
@ -36,33 +36,6 @@
           hx-swap="innerHTML">
  </section>
  <h2 class="section-title">Console</h2>
  <section class="panel console-panel">
    <div id="console-transcript-{{ server.id }}"
         class="console-transcript"
         data-autoscroll>
      {% for h in console_history %}
        {% with command=h.command, reply=h.reply, is_error=h.is_error %}
          {% include "_console_line.html" %}
        {% endwith %}
      {% endfor %}
    </div>
    <form hx-post="/servers/{{ server.id }}/console"
          hx-target="#console-transcript-{{ server.id }}"
          hx-swap="beforeend"
          hx-indicator=".console-spinner"
          hx-on::after-request="this.command.value=''; this.command.focus(); this.closest('section').querySelector('[data-autoscroll]').scrollTop = 1e9"
          class="console-input-form"
          data-console-form data-server-id="{{ server.id }}">
      <input type="hidden" name="csrf_token" value="{{ session.get('csrf_token', '') }}">
      <span class="console-prompt-glyph">&gt;</span>
      <input name="command" autocomplete="off" spellcheck="false" maxlength="1000"
             placeholder="status, changelevel c1m1_hotel, sm_kick …">
      <span class="console-spinner" aria-hidden="true">…</span>
      <button type="submit">Send</button>
    </form>
  </section>
  <h2 class="section-title">Files</h2>
  {% if not file_tree_root_entries %}
    <p class="muted">No files yet — start the server to mount its runtime.</p>
--- a/l4d2web/tests/test_console_routes.py
+++ b/l4d2web/tests/test_console_routes.py
@ -1,476 +0,0 @@
 import json
 from datetime import UTC, datetime
 from unittest.mock import patch
 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, CommandHistory, Server, User
 from l4d2web.services.rcon import RconAuthError, RconError
 # ---------------------------------------------------------------------------
 # Helpers / fixtures
 # ---------------------------------------------------------------------------
 def _make_app(tmp_path, monkeypatch, db_suffix="console.db"):
    db_url = f"sqlite:///{tmp_path / db_suffix}"
    monkeypatch.setenv("DATABASE_URL", db_url)
    app = create_app({"TESTING": True, "DATABASE_URL": db_url, "SECRET_KEY": "test"})
    init_db()
    return app
 def _seed(app, *, owner_username="alice", other_username=None):
    """Seed owner + blueprint + server (port 27015). Optionally seed a second user."""
    with session_scope() as s:
        owner = User(username=owner_username, password_digest=hash_password("x"), admin=False)
        s.add(owner)
        if other_username:
            other = User(username=other_username, password_digest=hash_password("x"), admin=False)
            s.add(other)
        s.flush()
        bp = Blueprint(user_id=owner.id, name="bp", arguments="[]", config="[]")
        s.add(bp)
        s.flush()
        server = Server(user_id=owner.id, blueprint_id=bp.id, name="srv", port=27015)
        s.add(server)
        s.flush()
        result = {"owner_id": owner.id, "server_id": server.id}
        if other_username:
            result["other_id"] = other.id
    return result
 def _login(client, user_id):
    with client.session_transaction() as sess:
        sess["user_id"] = user_id
        sess["pw_changed_at"] = datetime.now(UTC).isoformat()
        sess["csrf_token"] = "test-token"
 def _post_command(client, server_id, command, *, csrf="test-token"):
    return client.post(
        f"/servers/{server_id}/console",
        data={"command": command, "csrf_token": csrf},
        headers={"X-CSRF-Token": csrf},
    )
 def _get_history(client, server_id, **params):
    return client.get(f"/servers/{server_id}/console/history", query_string=params)
 def _history_count():
    with session_scope() as s:
        return s.query(CommandHistory).count()
 # ---------------------------------------------------------------------------
 # 1. Not logged in → redirect to login
 # ---------------------------------------------------------------------------
 def test_not_logged_in_post_returns_csrf_error_or_redirect(tmp_path, monkeypatch):
    # The CSRF before_request runs before load_current_user. An anonymous POST
    # without a session CSRF token will be rejected with 400 (CSRF mismatch)
    # before the require_login decorator can redirect. This is the correct
    # project behavior — anonymous users cannot reach the endpoint at all.
    app = _make_app(tmp_path, monkeypatch, "anon_post.db")
    data = _seed(app)
    client = app.test_client()
    resp = _post_command(client, data["server_id"], "status")
    # 400 (CSRF) or 302 (auth redirect) are both valid rejection responses.
    assert resp.status_code in (302, 400)
 def test_not_logged_in_get_history_redirects(tmp_path, monkeypatch):
    app = _make_app(tmp_path, monkeypatch, "anon_get.db")
    data = _seed(app)
    client = app.test_client()
    resp = _get_history(client, data["server_id"])
    assert resp.status_code == 302
    assert "/login" in resp.headers["Location"]
 # ---------------------------------------------------------------------------
 # 2. Logged in but not the owner → 404
 # ---------------------------------------------------------------------------
 def test_non_owner_post_returns_404(tmp_path, monkeypatch):
    app = _make_app(tmp_path, monkeypatch, "nonowner_post.db")
    data = _seed(app, other_username="bob")
    client = app.test_client()
    _login(client, data["other_id"])
    resp = _post_command(client, data["server_id"], "status")
    assert resp.status_code == 404
 def test_non_owner_get_history_returns_404(tmp_path, monkeypatch):
    app = _make_app(tmp_path, monkeypatch, "nonowner_get.db")
    data = _seed(app, other_username="bob")
    client = app.test_client()
    _login(client, data["other_id"])
    resp = _get_history(client, data["server_id"])
    assert resp.status_code == 404
 def test_nonexistent_server_post_returns_404(tmp_path, monkeypatch):
    app = _make_app(tmp_path, monkeypatch, "noserver_post.db")
    data = _seed(app)
    client = app.test_client()
    _login(client, data["owner_id"])
    resp = _post_command(client, 9999, "status")
    assert resp.status_code == 404
 # ---------------------------------------------------------------------------
 # 3. Valid command, RCON returns reply
 # ---------------------------------------------------------------------------
 def test_valid_command_inserts_history_row(tmp_path, monkeypatch):
    app = _make_app(tmp_path, monkeypatch, "ok.db")
    data = _seed(app)
    client = app.test_client()
    _login(client, data["owner_id"])
    with patch("l4d2web.routes.console_routes.rcon.execute_command", return_value="pong") as mock_exec:
        resp = _post_command(client, data["server_id"], "ping")
    assert resp.status_code == 200
    text = resp.get_data(as_text=True)
    assert "ping" in text
    assert "pong" in text
    assert "[ERROR]" not in text
    mock_exec.assert_called_once_with("127.0.0.1", 27015, mock_exec.call_args[0][2], "ping")
    with session_scope() as s:
        row = s.query(CommandHistory).one()
        assert row.command == "ping"
        assert row.reply == "pong"
        assert row.is_error is False
        assert row.user_id == data["owner_id"]
        assert row.server_id == data["server_id"]
 # ---------------------------------------------------------------------------
 # 4. Valid command, RCON returns empty string
 # ---------------------------------------------------------------------------
 def test_valid_command_empty_reply(tmp_path, monkeypatch):
    app = _make_app(tmp_path, monkeypatch, "empty_reply.db")
    data = _seed(app)
    client = app.test_client()
    _login(client, data["owner_id"])
    with patch("l4d2web.routes.console_routes.rcon.execute_command", return_value=""):
        resp = _post_command(client, data["server_id"], "noop")
    assert resp.status_code == 200
    assert "[ERROR]" not in resp.get_data(as_text=True)
    with session_scope() as s:
        row = s.query(CommandHistory).one()
        assert row.reply == ""
        assert row.is_error is False
 # ---------------------------------------------------------------------------
 # 5. RCON raises RconError
 # ---------------------------------------------------------------------------
 def test_rcon_error_inserts_error_row(tmp_path, monkeypatch):
    app = _make_app(tmp_path, monkeypatch, "rconerr.db")
    data = _seed(app)
    client = app.test_client()
    _login(client, data["owner_id"])
    with patch(
        "l4d2web.routes.console_routes.rcon.execute_command",
        side_effect=RconError("connection refused"),
    ):
        resp = _post_command(client, data["server_id"], "status")
    assert resp.status_code == 200
    assert "connection refused" in resp.get_data(as_text=True)
    assert "console-error" in resp.get_data(as_text=True)
    with session_scope() as s:
        row = s.query(CommandHistory).one()
        assert row.is_error is True
        assert "connection refused" in row.reply
 # ---------------------------------------------------------------------------
 # 6. RCON raises RconAuthError
 # ---------------------------------------------------------------------------
 def test_rcon_auth_error_inserts_error_row(tmp_path, monkeypatch):
    app = _make_app(tmp_path, monkeypatch, "autherr.db")
    data = _seed(app)
    client = app.test_client()
    _login(client, data["owner_id"])
    with patch(
        "l4d2web.routes.console_routes.rcon.execute_command",
        side_effect=RconAuthError("bad rcon password"),
    ):
        resp = _post_command(client, data["server_id"], "status")
    assert resp.status_code == 200
    assert "bad rcon password" in resp.get_data(as_text=True)
    assert "console-error" in resp.get_data(as_text=True)
    with session_scope() as s:
        row = s.query(CommandHistory).one()
        assert row.is_error is True
        assert "bad rcon password" in row.reply
 # ---------------------------------------------------------------------------
 # 7. Input validation: empty command → no history row
 # ---------------------------------------------------------------------------
 def test_empty_command_no_history_row(tmp_path, monkeypatch):
    app = _make_app(tmp_path, monkeypatch, "empty_cmd.db")
    data = _seed(app)
    client = app.test_client()
    _login(client, data["owner_id"])
    # execute_command raises ValueError for empty commands; the route catches it
    # before hitting rcon, but we patch anyway to be explicit.
    with patch(
        "l4d2web.routes.console_routes.rcon.execute_command",
        side_effect=ValueError("command must not be empty or whitespace-only"),
    ):
        resp = _post_command(client, data["server_id"], "")
    assert resp.status_code == 200
    assert "console-error" in resp.get_data(as_text=True)
    assert _history_count() == 0
 # ---------------------------------------------------------------------------
 # 8. Input validation: oversized command → no history row
 # ---------------------------------------------------------------------------
 def test_oversized_command_no_history_row(tmp_path, monkeypatch):
    app = _make_app(tmp_path, monkeypatch, "oversized.db")
    data = _seed(app)
    client = app.test_client()
    _login(client, data["owner_id"])
    with patch(
        "l4d2web.routes.console_routes.rcon.execute_command",
        side_effect=ValueError("command exceeds maximum byte length of 1000"),
    ):
        resp = _post_command(client, data["server_id"], "x" * 1001)
    assert resp.status_code == 200
    assert "console-error" in resp.get_data(as_text=True)
    assert _history_count() == 0
 # ---------------------------------------------------------------------------
 # 9. CSRF token missing/invalid → 400
 # ---------------------------------------------------------------------------
 def test_missing_csrf_token_rejected(tmp_path, monkeypatch):
    app = _make_app(tmp_path, monkeypatch, "csrf.db")
    data = _seed(app)
    client = app.test_client()
    _login(client, data["owner_id"])
    # Send with a wrong CSRF token (session has "test-token" from _login).
    resp = client.post(
        f"/servers/{data['server_id']}/console",
        data={"command": "status"},
        # No X-CSRF-Token header and no csrf_token form field.
    )
    assert resp.status_code == 400
 # ---------------------------------------------------------------------------
 # 10. GET /console/history with rows in DB → newest-first JSON list
 # ---------------------------------------------------------------------------
 def test_get_history_returns_newest_first(tmp_path, monkeypatch):
    app = _make_app(tmp_path, monkeypatch, "hist_order.db")
    data = _seed(app)
    client = app.test_client()
    _login(client, data["owner_id"])
    with session_scope() as s:
        for cmd in ["alpha", "beta", "gamma"]:
            s.add(
                CommandHistory(
                    user_id=data["owner_id"],
                    server_id=data["server_id"],
                    command=cmd,
                    reply="ok",
                    is_error=False,
                )
            )
    resp = _get_history(client, data["server_id"])
    assert resp.status_code == 200
    rows = json.loads(resp.get_data(as_text=True))
    assert [r["command"] for r in rows] == ["gamma", "beta", "alpha"]
    for r in rows:
        assert "id" in r
        assert "command" in r
        # reply should NOT be included
        assert "reply" not in r
 # ---------------------------------------------------------------------------
 # 11. GET /console/history?before=<id> → only rows with id < before
 # ---------------------------------------------------------------------------
 def test_get_history_before_pagination(tmp_path, monkeypatch):
    app = _make_app(tmp_path, monkeypatch, "hist_before.db")
    data = _seed(app)
    client = app.test_client()
    _login(client, data["owner_id"])
    row_ids = []
    with session_scope() as s:
        for cmd in ["one", "two", "three"]:
            row = CommandHistory(
                user_id=data["owner_id"],
                server_id=data["server_id"],
                command=cmd,
                reply="",
                is_error=False,
            )
            s.add(row)
            s.flush()
            row_ids.append(row.id)
    pivot = row_ids[2]  # id of "three"
    resp = _get_history(client, data["server_id"], before=pivot)
    assert resp.status_code == 200
    rows = json.loads(resp.get_data(as_text=True))
    # Should only have "two" and "one" (ids < pivot), newest first.
    assert len(rows) == 2
    assert rows[0]["command"] == "two"
    assert rows[1]["command"] == "one"
 # ---------------------------------------------------------------------------
 # 12. GET /console/history?limit=10000 → clamped to ≤200
 # ---------------------------------------------------------------------------
 def test_get_history_limit_clamped(tmp_path, monkeypatch):
    app = _make_app(tmp_path, monkeypatch, "hist_clamp.db")
    data = _seed(app)
    client = app.test_client()
    _login(client, data["owner_id"])
    # Monkeypatch the max-limit constant to 3 so we can verify the clamp
    # without inserting 200+ rows.
    monkeypatch.setattr("l4d2web.routes.console_routes._HISTORY_MAX_LIMIT", 3)
    # Insert 5 rows — more than the patched max of 3.
    with session_scope() as s:
        for i in range(5):
            s.add(
                CommandHistory(
                    user_id=data["owner_id"],
                    server_id=data["server_id"],
                    command=f"cmd{i}",
                    reply="",
                    is_error=False,
                )
            )
    # Request with limit=100 (far above the patched max of 3).
    resp = _get_history(client, data["server_id"], limit=100)
    assert resp.status_code == 200
    rows = json.loads(resp.get_data(as_text=True))
    # Clamp must cap at _HISTORY_MAX_LIMIT (3), not at the requested 100.
    assert len(rows) == 3
 # ---------------------------------------------------------------------------
 # 13 (in test_pages.py style, placed here): server_detail renders console_history
 #     scoped to current user, oldest-first
 # ---------------------------------------------------------------------------
 def test_server_detail_console_history_scoped_and_ordered(tmp_path, monkeypatch):
    """Verify server_detail loads console_history scoped to the owner, oldest-first.
    The server_detail template does not render console_history items yet (the
    UI subagent will add that). We therefore verify the data contract by:
      1. Checking the page loads (200).
      2. Directly querying the DB to confirm only the owner's rows exist.
      3. Testing ordering by verifying row IDs ascend (oldest-first) after the
         route's list(reversed(...)) logic — done via a separate call to the
         history endpoint which already has ordering tested above.
    """
    app = _make_app(tmp_path, monkeypatch, "detail_hist.db")
    data = _seed(app, other_username="bob")
    client = app.test_client()
    _login(client, data["owner_id"])
    with session_scope() as s:
        # Insert rows for alice (owner) — intentionally inserted newest first
        # so that ordering logic is meaningful.
        row_second = CommandHistory(
            user_id=data["owner_id"], server_id=data["server_id"],
            command="second", reply="r2", is_error=False,
        )
        s.add(row_second)
        s.flush()
        second_id = row_second.id
        row_first = CommandHistory(
            user_id=data["owner_id"], server_id=data["server_id"],
            command="first", reply="r1", is_error=False,
        )
        s.add(row_first)
        s.flush()
        # Insert a row for the other user — must NOT be included.
        s.add(CommandHistory(
            user_id=data["other_id"], server_id=data["server_id"],
            command="other_cmd", reply="x", is_error=False,
        ))
    # The detail page must load successfully.
    resp = client.get(f"/servers/{data['server_id']}")
    assert resp.status_code == 200
    # Verify scoping: the history endpoint is already tested for ordering above.
    # Here we confirm via GET /console/history that only alice's rows are returned.
    hist_resp = _get_history(client, data["server_id"])
    assert hist_resp.status_code == 200
    rows = json.loads(hist_resp.get_data(as_text=True))
    commands = [r["command"] for r in rows]
    # Only alice's commands appear.
    assert "other_cmd" not in commands
    assert "first" in commands
    assert "second" in commands
    # Newest-first from history endpoint; IDs must descend.
    ids = [r["id"] for r in rows]
    assert ids == sorted(ids, reverse=True)
--- a/l4d2web/tests/test_rcon.py
+++ b/l4d2web/tests/test_rcon.py
@ -19,14 +19,9 @@ import pytest
 from l4d2web.services.rcon import (
    RconAuthError,
    RconError,
    execute_command,
    query_status,
 )
 # req_id constants (must match rcon.py)
 _EXEC_REQ_ID = 2
 _MARKER_REQ_ID = 0xC0DE
 def _pack(req_id: int, ptype: int, body: str) -> bytes:
    body_bytes = body.encode("utf-8") + b"\x00\x00"
@ -82,14 +77,6 @@ def fake_rcon_server(handler) -> Iterator[int]:
        raise AssertionError("fake_rcon_server handler raised") from handler_error[0]
 def _do_auth_handshake(conn: socket.socket) -> None:
    """Receive AUTH packet and respond with the standard two-packet handshake."""
    req_id, ptype, body = _unpack_one(conn)
    assert ptype == 3  # SERVERDATA_AUTH
    conn.sendall(_pack(req_id, 0, ""))   # leading empty type-0
    conn.sendall(_pack(req_id, 2, ""))   # auth response — req_id != -1 means OK
 def test_auth_success_then_status(monkeypatch: pytest.MonkeyPatch) -> None:
    response_body = (
        "hostname: Left 4 Dead 2\n"
@ -166,138 +153,3 @@ def test_parse_duration_rejects_malformed_as_rcon_error() -> None:
    for bad in ["", ":", "abc", "1:", ":5", "1:2:3:4"]:
        with pytest.raises(RconError):
            _parse_duration(bad)
 # ---------------------------------------------------------------------------
 # execute_command tests
 # ---------------------------------------------------------------------------
 def test_execute_command_single_packet_reply() -> None:
    """Handler sends one body packet then the marker echo; client returns body."""
    def handler(conn: socket.socket) -> None:
        _do_auth_handshake(conn)
        # Receive EXECCOMMAND
        exec_id, exec_type, cmd = _unpack_one(conn)
        assert exec_type == 2  # SERVERDATA_EXECCOMMAND
        assert cmd == "echo hello"
        # Receive the marker packet
        marker_id, marker_type, marker_body = _unpack_one(conn)
        assert marker_type == 0  # SERVERDATA_RESPONSE_VALUE
        assert marker_body == ""
        # Send reply body, then marker echo
        conn.sendall(_pack(exec_id, 0, "hello"))
        conn.sendall(_pack(marker_id, 0, ""))
    with fake_rcon_server(handler) as port:
        result = execute_command("127.0.0.1", port, "letmein", "echo hello", timeout=2.0)
    assert result == "hello"
 def test_execute_command_multi_packet_reply() -> None:
    """Multi-packet replies are concatenated in order."""
    chunk1 = "A" * 3000
    chunk2 = "B" * 3000
    def handler(conn: socket.socket) -> None:
        _do_auth_handshake(conn)
        exec_id, _, _ = _unpack_one(conn)
        marker_id, _, _ = _unpack_one(conn)
        conn.sendall(_pack(exec_id, 0, chunk1))
        conn.sendall(_pack(exec_id, 0, chunk2))
        conn.sendall(_pack(marker_id, 0, ""))
    with fake_rcon_server(handler) as port:
        result = execute_command("127.0.0.1", port, "letmein", "cvarlist", timeout=2.0)
    assert result == chunk1 + chunk2
 def test_execute_command_empty_reply() -> None:
    """Server echoes only the marker (e.g. for `say`); result is empty string."""
    def handler(conn: socket.socket) -> None:
        _do_auth_handshake(conn)
        _unpack_one(conn)  # exec packet
        marker_id, _, _ = _unpack_one(conn)
        conn.sendall(_pack(marker_id, 0, ""))
    with fake_rcon_server(handler) as port:
        result = execute_command("127.0.0.1", port, "letmein", "say hi", timeout=2.0)
    assert result == ""
 def test_execute_command_bad_password() -> None:
    """Bad password on auth raises RconAuthError."""
    def handler(conn: socket.socket) -> None:
        req_id, _, _ = _unpack_one(conn)
        conn.sendall(_pack(req_id, 0, ""))
        conn.sendall(_pack(-1, 2, ""))  # bad password sentinel
    with fake_rcon_server(handler) as port:
        with pytest.raises(RconAuthError):
            execute_command("127.0.0.1", port, "wrong", "status", timeout=2.0)
 def test_execute_command_timeout() -> None:
    """Server hangs; client raises RconError."""
    def handler(conn: socket.socket) -> None:
        import time
        time.sleep(3.0)
    with fake_rcon_server(handler) as port:
        with pytest.raises(RconError):
            execute_command("127.0.0.1", port, "x", "status", timeout=0.3)
 def test_execute_command_rejects_empty_command() -> None:
    with pytest.raises(ValueError):
        execute_command("127.0.0.1", 27015, "pw", "")
 def test_execute_command_rejects_whitespace_only_command() -> None:
    with pytest.raises(ValueError):
        execute_command("127.0.0.1", 27015, "pw", "   ")
 def test_execute_command_rejects_null_byte_in_command() -> None:
    with pytest.raises(ValueError):
        execute_command("127.0.0.1", 27015, "pw", "echo\x00hello")
 def test_execute_command_rejects_oversized_command() -> None:
    with pytest.raises(ValueError):
        execute_command("127.0.0.1", 27015, "pw", "x" * 1001)
 def test_execute_command_drains_marker_after_response() -> None:
    """Client stops reading exactly after the marker — does not block on a 5th packet."""
    def handler(conn: socket.socket) -> None:
        _do_auth_handshake(conn)
        exec_id, _, _ = _unpack_one(conn)
        marker_id, _, _ = _unpack_one(conn)
        # Send one body packet then the marker — nothing else.
        conn.sendall(_pack(exec_id, 0, "done"))
        conn.sendall(_pack(marker_id, 0, ""))
        # Handler returns immediately; if client reads past the marker it would
        # block (no more data) and hit a timeout — the 2 s timeout guards us.
    with fake_rcon_server(handler) as port:
        result = execute_command("127.0.0.1", port, "letmein", "status", timeout=2.0)
    assert result == "done"