#include "session_manager.h"
#include "crypto.h"
#include "hydra_log.h"
#include "telnet_stream.h"
#include "telnet_utils.h"
#include "utf8_utils.h"
#include <color_ops.h>
#include <nlohmann/json.hpp>
#ifdef GRPC_ENABLED
#include "base64.h"
#include "grpc_web.h"
#include "hydra.pb.h"
#endif
#include <cerrno>
#include <chrono>
#include <cstring>
#include <algorithm>
#include <cctype>
#include <new>
#include <sstream>
#include <stdexcept>

#ifndef MSG_NOSIGNAL
#define MSG_NOSIGNAL 0
#endif

// Context for scroll-back replay callback — renders PUA through the bridge.
struct ReplayContext {
    ganl::ConnectionHandle handle;
    ganl::EncodingType encoding;
    ColorDepth colorDepth;
    TelnetBridge* bridge;
    SessionManager* sessionMgr;
};

// buildGmcpFrame() is now in telnet_utils.h (shared with grpc_server.cpp)

// Send Hydra.Links GMCP to all GMCP-enabled front-doors of a session.
// Called whenever link state changes (connect, disconnect, reconnect).
static void sendHydraLinksGmcp(HydraSession& session,
                               std::map<ganl::ConnectionHandle, FrontDoorState>& frontDoors,
                               std::function<void(ganl::ConnectionHandle, const std::string&)> writeFn) {
    nlohmann::json linksArr = nlohmann::json::array();
    for (size_t i = 0; i < session.links.size(); i++) {
        const auto& l = session.links[i];
        const char* state = "unknown";
        switch (l.state) {
        case LinkState::Connecting:     state = "connecting"; break;
        case LinkState::TlsHandshaking: state = "tls"; break;
        case LinkState::Negotiating:    state = "negotiating"; break;
        case LinkState::AutoLoggingIn:  state = "logging-in"; break;
        case LinkState::Active:         state = "active"; break;
        case LinkState::Reconnecting:   state = "reconnecting"; break;
        case LinkState::Suspended:      state = "suspended"; break;
        case LinkState::Dead:           state = "dead"; break;
        }
        linksArr.push_back({
            {"number", static_cast<int>(i) + 1},
            {"game", l.gameName},
            {"state", state},
            {"active", i == session.activeLink},
        });
    }
    std::string gmcpPayload = "Hydra.Links " + linksArr.dump();
    std::string frame = buildGmcpFrame(gmcpPayload);

    for (auto h : session.frontDoors) {
        auto fdIt = frontDoors.find(h);
        if (fdIt != frontDoors.end() && fdIt->second.gmcpEnabled) {
            writeFn(h, frame);
        }
    }
}

static const char* linkStateName(LinkState s) {
    switch (s) {
    case LinkState::Connecting:     return "connecting";
    case LinkState::TlsHandshaking: return "tls";
    case LinkState::Negotiating:    return "negotiating";
    case LinkState::AutoLoggingIn:  return "logging-in";
    case LinkState::Active:         return "active";
    case LinkState::Reconnecting:   return "reconnecting";
    case LinkState::Suspended:      return "suspended";
    case LinkState::Dead:           return "dead";
    }
    return "?";
}

static std::string hydraCharsetName(ganl::EncodingType encoding) {
    switch (encoding) {
    case ganl::EncodingType::Utf8:
        return "UTF-8";
    case ganl::EncodingType::Latin1:
        return "ISO-8859-1";
    case ganl::EncodingType::Cp437:
        return "CP437";
    case ganl::EncodingType::Cp1252:
        return "CP1252";
    case ganl::EncodingType::Ascii:
    default:
        return "US-ASCII";
    }
}

static ganl::EncodingType parseNegotiatedCharset(const std::string& name,
                                                 ganl::EncodingType fallback) {
    std::string upper;
    upper.reserve(name.size());
    for (unsigned char ch : name) {
        upper.push_back(static_cast<char>(std::toupper(ch)));
    }
    if (upper == "UTF-8" || upper == "UTF8") return ganl::EncodingType::Utf8;
    if (upper == "ISO-8859-1" || upper == "ISO8859-1" || upper == "LATIN-1" || upper == "LATIN1") {
        return ganl::EncodingType::Latin1;
    }
    if (upper == "CP437") return ganl::EncodingType::Cp437;
    if (upper == "CP1252" || upper == "WINDOWS-1252") return ganl::EncodingType::Cp1252;
    if (upper == "US-ASCII" || upper == "ASCII") return ganl::EncodingType::Ascii;
    return fallback;
}

static const char* frontDoorProtoName(FrontDoorProto proto) {
    switch (proto) {
    case FrontDoorProto::Telnet:
        return "telnet";
    case FrontDoorProto::WebSocket:
        return "websocket";
    case FrontDoorProto::GrpcWeb:
        return "grpc_web";
    case FrontDoorProto::WsGameSession:
        return "ws_game_session";
    }
    return "unknown";
}

static const char* linkStateMetricName(LinkState state) {
    switch (state) {
    case LinkState::Connecting:
        return "connecting";
    case LinkState::TlsHandshaking:
        return "tls_handshaking";
    case LinkState::Negotiating:
        return "negotiating";
    case LinkState::AutoLoggingIn:
        return "auto_logging_in";
    case LinkState::Active:
        return "active";
    case LinkState::Reconnecting:
        return "reconnecting";
    case LinkState::Suspended:
        return "suspended";
    case LinkState::Dead:
        return "dead";
    }
    return "unknown";
}

static const char* sessionStateMetricName(SessionState state) {
    switch (state) {
    case SessionState::Login:
        return "login";
    case SessionState::Active:
        return "active";
    case SessionState::Detached:
        return "detached";
    }
    return "unknown";
}

static bool charsetOffered(const std::string& payload, const std::string& wanted) {
    if (payload.empty()) return false;
    char sep = payload[0];
    std::string wantedUpper;
    wantedUpper.reserve(wanted.size());
    for (unsigned char ch : wanted) {
        wantedUpper.push_back(static_cast<char>(std::toupper(ch)));
    }

    size_t start = 1;
    while (start <= payload.size()) {
        size_t end = payload.find(sep, start);
        if (end == std::string::npos) end = payload.size();
        if (end > start) {
            std::string candidate = payload.substr(start, end - start);
            std::string candidateUpper;
            candidateUpper.reserve(candidate.size());
            for (unsigned char ch : candidate) {
                candidateUpper.push_back(static_cast<char>(std::toupper(ch)));
            }
            if (candidateUpper == wantedUpper) return true;
        }
        if (end == payload.size()) break;
        start = end + 1;
    }
    return false;
}

std::atomic<size_t> SessionManager::globalScrollbackBytes_{0};

SessionManager::SessionManager(ganl::NetworkEngine& engine,
                               AccountManager& accounts,
                               const HydraConfig& config)
    : engine_(engine), accounts_(accounts), config_(config) {
}

SessionManager::~SessionManager() {
}

void SessionManager::sendToClient(ganl::ConnectionHandle handle,
                                  const std::string& text) {
    int sockfd = static_cast<int>(handle);

    auto it = frontDoors_.find(handle);
    if (it != frontDoors_.end()) {
        if (it->second.proto == FrontDoorProto::WebSocket) {
            std::string frame = wsEncodeFrame(text);
            safeWrite(handle, frame);
            return;
        }
#ifdef GRPC_ENABLED
        if (it->second.grpcWebSubscribed) {
            // Send as a grpc-web data frame with the text as GameOutput
            hydra::GameOutput go;
            go.set_text(sanitizeProtoTextForLog(
                text, "grpc-web system output", "hydra", 0));
            go.set_source("hydra");
            go.set_timestamp(static_cast<int64_t>(time(nullptr)));

            std::string gwFrame = grpcWebEncodeDataFrame(
                go.SerializeAsString());
            if (it->second.grpcWebTextMode) gwFrame = base64Encode(gwFrame);

            std::string chunk = std::to_string(gwFrame.size())
                + "\r\n" + gwFrame + "\r\n";
            safeWrite(handle, chunk);
            return;
        }
#endif
    }

    safeWrite(handle, text);
}

void SessionManager::safeWrite(ganl::ConnectionHandle handle,
                               const std::string& data) {
    safeWrite(handle, data.data(), data.size());
}

void SessionManager::safeWrite(ganl::ConnectionHandle handle,
                               const char* data, size_t len) {
    if (len == 0) return;

    auto fdIt = frontDoors_.find(handle);
    if (fdIt != frontDoors_.end() && fdIt->second.tlsTransport) {
        // #1847: TLS branch used to append unbounded to tlsPlainOut and
        // bypass #1096's MAX_WRITE_BUFFER.  A peer that stops reading can
        // grow plain + ciphertext until OOM / IoBuffer throw, and hydra
        // has no top-level exception barrier.
        //
        FrontDoorState& fd = fdIt->second;
        auto tlsPending = [](const FrontDoorState& f) -> size_t {
            size_t n = 0;
            if (f.tlsPlainOut) {
                n += f.tlsPlainOut->readableBytes();
            }
            if (f.tlsEncryptedOut) {
                n += f.tlsEncryptedOut->readableBytes();
            }
            return n;
        };
        auto closeSlowTls = [&]() {
            LOG_WARN("safeWrite: TLS write buffer cap (%zu) exceeded on fd %lu — closing",
                     MAX_WRITE_BUFFER, (unsigned long)handle);
            if (fd.tlsPlainOut) {
                fd.tlsPlainOut->clear();
            }
            if (fd.tlsEncryptedOut) {
                fd.tlsEncryptedOut->clear();
            }
            engine_.closeConnection(handle);
        };

        // Small headroom for TLS record framing beyond plaintext length.
        //
        constexpr size_t kTlsRecordOverhead = 64;
        if (tlsPending(fd) + len + kTlsRecordOverhead > MAX_WRITE_BUFFER) {
            closeSlowTls();
            return;
        }

        try {
            if (!fd.tlsPlainOut) {
                fd.tlsPlainOut = std::make_unique<ganl::IoBuffer>(4096);
            }
            fd.tlsPlainOut->append(data, len);
            if (!flushFrontDoorTlsOutgoing(fd)) {
                return;
            }
            // Encrypt can grow ciphertext past the pre-append estimate.
            //
            if (tlsPending(fd) > MAX_WRITE_BUFFER) {
                closeSlowTls();
            }
        } catch (const std::bad_alloc&) {
            LOG_WARN("safeWrite: TLS allocation failure on fd %lu — closing",
                     (unsigned long)handle);
            if (fd.tlsPlainOut) {
                fd.tlsPlainOut->clear();
            }
            if (fd.tlsEncryptedOut) {
                fd.tlsEncryptedOut->clear();
            }
            engine_.closeConnection(handle);
        } catch (const std::length_error&) {
            LOG_WARN("safeWrite: TLS IoBuffer ceiling on fd %lu — closing",
                     (unsigned long)handle);
            if (fd.tlsPlainOut) {
                fd.tlsPlainOut->clear();
            }
            if (fd.tlsEncryptedOut) {
                fd.tlsEncryptedOut->clear();
            }
            engine_.closeConnection(handle);
        }
        return;
    }

    // #1096: helper — append at most up to MAX_WRITE_BUFFER; on overflow
    // close the slow front-door (session itself is kept).
    auto appendCapped = [&](WriteBuffer& wb, const char* p, size_t n) -> bool {
        if (wb.remaining() + n > MAX_WRITE_BUFFER) {
            LOG_WARN("safeWrite: write buffer cap (%zu) exceeded on fd %lu — closing",
                     MAX_WRITE_BUFFER, (unsigned long)handle);
            writeBuffers_.erase(handle);
            engine_.closeConnection(handle);
            return false;
        }
        wb.append(p, n);
        return true;
    };

    // If there is already buffered data for this connection, append and
    // let drainWriteBuffer send it in order.
    auto it = writeBuffers_.find(handle);
    if (it != writeBuffers_.end() && !it->second.empty()) {
        (void)appendCapped(it->second, data, len);
        return;
    }

    // Try a non-blocking send immediately.
    int fd = static_cast<int>(handle);
    ssize_t sent = ::send(fd, data, len, MSG_DONTWAIT | MSG_NOSIGNAL);

    if (sent < 0) {
        if (errno == EAGAIN || errno == EWOULDBLOCK) {
            // Socket not ready — buffer everything and arm EPOLLOUT.
            auto& wb = writeBuffers_[handle];
            wb.reset();
            if (!appendCapped(wb, data, len)) {
                return;
            }
            ganl::ErrorCode err = 0;
            engine_.postWrite(handle, nullptr, 0, err);
        } else {
            LOG_DEBUG("safeWrite send failed for fd %d: %s", fd, strerror(errno));
        }
        return;
    }

    size_t n = static_cast<size_t>(sent);
    if (n < len) {
        // Partial write — buffer the remainder and arm EPOLLOUT.
        auto& wb = writeBuffers_[handle];
        wb.reset();
        if (!appendCapped(wb, data + n, len - n)) {
            return;
        }
        ganl::ErrorCode err = 0;
        engine_.postWrite(handle, nullptr, 0, err);
    }
}

void SessionManager::drainWriteBuffer(ganl::ConnectionHandle handle) {
    auto fdIt = frontDoors_.find(handle);
    if (fdIt != frontDoors_.end() && fdIt->second.tlsTransport) {
        flushFrontDoorTlsOutgoing(fdIt->second);
        return;
    }

    auto it = writeBuffers_.find(handle);
    if (it == writeBuffers_.end() || it->second.empty()) return;

    WriteBuffer& wb = it->second;
    int fd = static_cast<int>(handle);
    ssize_t sent = ::send(fd, wb.ptr(), wb.remaining(),
                          MSG_DONTWAIT | MSG_NOSIGNAL);

    if (sent < 0) {
        if (errno == EAGAIN || errno == EWOULDBLOCK) {
            // Still blocked — re-arm EPOLLOUT.
            ganl::ErrorCode err = 0;
            engine_.postWrite(handle, nullptr, 0, err);
        } else {
            LOG_DEBUG("drainWriteBuffer send failed for fd %d: %s",
                      fd, strerror(errno));
            writeBuffers_.erase(it);
        }
        return;
    }

    wb.advance(static_cast<size_t>(sent));
    if (!wb.empty()) {
        // More data remains — re-arm.
        ganl::ErrorCode err = 0;
        engine_.postWrite(handle, nullptr, 0, err);
    } else {
        // Fully drained.
        writeBuffers_.erase(it);
    }
}

bool SessionManager::drainTlsCiphertext(FrontDoorState& fd) {
    if (!fd.tlsEncryptedOut || fd.tlsEncryptedOut->readableBytes() == 0) {
        return true;
    }

    int sock = static_cast<int>(fd.handle);
    while (fd.tlsEncryptedOut->readableBytes() > 0) {
        ssize_t sent = ::send(sock, fd.tlsEncryptedOut->readPtr(),
                              fd.tlsEncryptedOut->readableBytes(),
                              MSG_DONTWAIT | MSG_NOSIGNAL);
        if (sent < 0) {
            if (errno == EAGAIN || errno == EWOULDBLOCK) {
                ganl::ErrorCode err = 0;
                engine_.postWrite(fd.handle, nullptr, 0, err);
                return true;
            }
            LOG_DEBUG("TLS send failed for fd %d: %s", sock, strerror(errno));
            engine_.closeConnection(fd.handle);
            return false;
        }
        fd.tlsEncryptedOut->consumeRead(static_cast<size_t>(sent));
    }

    return true;
}

bool SessionManager::flushFrontDoorTlsOutgoing(FrontDoorState& fd) {
    if (!fd.tlsTransport) return true;

    if (!fd.tlsPlainOut) {
        fd.tlsPlainOut = std::make_unique<ganl::IoBuffer>(4096);
    }
    if (!fd.tlsEncryptedOut) {
        fd.tlsEncryptedOut = std::make_unique<ganl::IoBuffer>(4096);
    }

    while (fd.tlsPlainOut->readableBytes() > 0) {
        ganl::TlsResult res = fd.tlsTransport->processOutgoing(
            fd.handle, *fd.tlsPlainOut, *fd.tlsEncryptedOut, true);
        if (fd.tlsTransport->isEstablished(fd.handle)) {
            fd.tlsEstablished = true;
        }
        if (res == ganl::TlsResult::Error || res == ganl::TlsResult::Closed) {
            LOG_WARN("TLS outgoing failed for fd %lu: %s",
                     (unsigned long)fd.handle,
                     fd.tlsTransport->getLastTlsErrorString(fd.handle).c_str());
            engine_.closeConnection(fd.handle);
            return false;
        }
        if (res == ganl::TlsResult::WantRead || res == ganl::TlsResult::WantWrite) {
            break;
        }
    }

    return drainTlsCiphertext(fd);
}

void SessionManager::showBanner(ganl::ConnectionHandle handle) {
    bool bootstrap = accounts_.isEmpty();
    std::string banner = "\r\nHydra v0.2\r\n";
    if (bootstrap) {
        banner += "First connection — create the admin account.\r\n";
        banner += "Or type 'create <username> <password>' to create an account.\r\n";
    }
    banner += "\r\nUsername: ";
    sendToClient(handle, banner);
}

void SessionManager::showGameMenu(HydraSession& session,
                                  ganl::ConnectionHandle fdHandle) {
    std::string menu = "\r\n--- Available Games ---\r\n";
    for (size_t i = 0; i < config_.games.size(); i++) {
        menu += "  " + config_.games[i].name + " ("
              + config_.games[i].host + ":"
              + std::to_string(config_.games[i].port) + ")\r\n";
    }
    menu += "\r\nUse /connect <game> to connect.\r\n";
    if (!session.links.empty()) {
        BackDoorLink* active = session.getActiveLink();
        if (active && active->state == LinkState::Active) {
            menu += "[Active link: " + active->gameName + "]\r\n";
        }
    }
    sendToClient(fdHandle, menu);

    // Replay cached GMCP state to this front-door (if GMCP-enabled)
    auto fdIt = frontDoors_.find(fdHandle);
    if (fdIt != frontDoors_.end() && fdIt->second.gmcpEnabled) {
        for (const auto& [pkg, json] : session.gmcpCache) {
            safeWrite(fdHandle, buildGmcpFrame(pkg + " " + json));
        }
    }
}

void SessionManager::replayGmcpCache(HydraSession& session,
                                      std::shared_ptr<HydraSession::SubscriberQueue> sq) {
    // Replay cached GMCP state into a gRPC subscriber queue
    for (const auto& [pkg, json] : session.gmcpCache) {
        HydraSession::GmcpItem item;
        item.package = pkg;
        item.json = json;
        item.linkNumber = session.activeLink < session.links.size()
            ? static_cast<int>(session.activeLink) + 1 : 0;
        sq->gmcp.push(std::move(item));
    }
}

// ---- Reverse map helper ----

bool SessionManager::findByBackDoor(ganl::ConnectionHandle bdHandle,
                                    HydraSession*& session,
                                    BackDoorLink*& link,
                                    size_t& linkIdx) {
    auto it = backDoorMap_.find(bdHandle);
    if (it == backDoorMap_.end()) return false;

    auto sit = sessions_.find(it->second.internalSessionId);
    if (sit == sessions_.end()) return false;

    session = &sit->second;
    linkIdx = it->second.linkIndex;
    if (linkIdx >= session->links.size()) return false;
    link = &session->links[linkIdx];
    return true;
}

// ---- Front-door lifecycle ----

void SessionManager::onAccept(ganl::ConnectionHandle handle,
                              const std::string& clientIp,
                              bool deferPrompt) {
    if (!checkConnectionLimits(clientIp)) {
        engine_.closeConnection(handle);
        return;
    }
    recordConnect(clientIp);

    FrontDoorState fd;
    fd.handle = handle;
    fd.loginPhase = FrontDoorState::AwaitUsername;
    fd.proto = FrontDoorProto::Telnet;
    fd.clientIp = clientIp;
    frontDoors_[handle] = std::move(fd);

    LOG_INFO("Session manager: new front-door %lu (telnet) from %s",
             (unsigned long)handle, clientIp.c_str());
    // #1846: deferPrompt means TLS transport is not attached yet
    // (setFrontDoorTls runs after onAccept).  Any safeWrite here is
    // plaintext on the wire — including WILL EOR.  Defer negotiation and
    // banner until the handshake is established (onFrontDoorData).
    //
    if (!deferPrompt) {
        safeWrite(handle, buildTelnetCommandFrame(telnet::WILL, telnet::EOR_OPT));
        showBanner(handle);
        frontDoors_[handle].initialPromptSent = true;
    }
}

void SessionManager::onAcceptWebSocket(ganl::ConnectionHandle handle,
                                       const std::string& clientIp) {
    if (!checkConnectionLimits(clientIp)) {
        engine_.closeConnection(handle);
        return;
    }
    recordConnect(clientIp);

    FrontDoorState fd;
    fd.handle = handle;
    fd.loginPhase = FrontDoorState::AwaitUsername;
    fd.proto = FrontDoorProto::WebSocket;
    fd.encoding = ganl::EncodingType::Utf8;
    fd.colorDepth = ColorDepth::Ansi256;
    fd.clientIp = clientIp;
    frontDoors_[handle] = std::move(fd);

    LOG_INFO("Session manager: new front-door %lu (websocket) from %s",
             (unsigned long)handle, clientIp.c_str());
}

void SessionManager::onAcceptGrpcWeb(ganl::ConnectionHandle handle,
                                     const std::string& clientIp) {
    if (!checkConnectionLimits(clientIp)) {
        engine_.closeConnection(handle);
        return;
    }
    recordConnect(clientIp);

    FrontDoorState fd;
    fd.handle = handle;
    fd.loginPhase = FrontDoorState::AwaitUsername;
    fd.proto = FrontDoorProto::GrpcWeb;
    fd.encoding = ganl::EncodingType::Utf8;
    fd.colorDepth = ColorDepth::Ansi256;
    fd.clientIp = clientIp;
    frontDoors_[handle] = std::move(fd);

    LOG_INFO("Session manager: new front-door %lu (grpc-web) from %s",
             (unsigned long)handle, clientIp.c_str());
}

void SessionManager::handleFrontDoorPlainData(FrontDoorState& fd,
                                              const char* data, size_t len) {
    ganl::ConnectionHandle handle = fd.handle;

    // ---- WebSocket path (also handles WsGameSession during handshake) ----
    if (fd.proto == FrontDoorProto::WebSocket) {
        if (!fd.wsState.handshakeComplete) {
            // Still in HTTP upgrade handshake
            std::string response = wsProcessHandshake(fd.wsState, data, len);
            if (!response.empty()) {
                int sockfd = static_cast<int>(handle);
                safeWrite(handle, response);

                if (!fd.wsState.handshakeOk) {
                    // Handshake failed — close
                    engine_.closeConnection(handle);
                    return;
                }

                // Check if this is a GameSession WebSocket (hydra-gamesession subprotocol)
#ifdef GRPC_ENABLED
                if (fd.wsState.isGameSession) {
                    fd.proto = FrontDoorProto::WsGameSession;
                    LOG_INFO("WebSocket GameSession handshake complete for fd %lu",
                             (unsigned long)handle);
                    // No banner, no session.frontDoors — wait for first-message auth.
                    // Process any trailing data as WsGameSession frames.
                    if (!fd.wsState.handshakeBuf.empty()) {
                        std::string trailing = fd.wsState.handshakeBuf;
                        fd.wsState.handshakeBuf.clear();
                        handleWsGameSessionData(fd, trailing.data(), trailing.size());
                    }
                    return;
                }
#endif

                LOG_INFO("WebSocket handshake complete for fd %lu",
                         (unsigned long)handle);
                showBanner(handle);

                // Process any trailing data after the handshake
                if (!fd.wsState.handshakeBuf.empty()) {
                    std::string trailing = fd.wsState.handshakeBuf;
                    fd.wsState.handshakeBuf.clear();
                    std::string responses;
                    auto msgs = wsDecodeFrames(fd.wsState,
                        trailing.data(), trailing.size(), responses);
                    if (!responses.empty()) {
                        safeWrite(handle, responses);
                    }
                    for (const auto& msg : msgs) {
                        if (msg.opcode == WS_OP_TEXT) {
                            // Process as lines
                            for (char ch : msg.payload) {
                                if (ch == '\n') {
                                    if (!fd.lineBuf.empty() && fd.lineBuf.back() == '\r')
                                        fd.lineBuf.pop_back();
                                    processLine(fd, fd.lineBuf);
                                    fd.lineBuf.clear();
                                } else {
                                    fd.lineBuf += ch;
                                }
                            }
                            // WS messages often don't end with \n
                            if (!fd.lineBuf.empty()) {
                                processLine(fd, fd.lineBuf);
                                fd.lineBuf.clear();
                            }
                        }
                    }
                }
            }
            return;
        }

        // Decode WebSocket frames
        std::string responses;
        auto msgs = wsDecodeFrames(fd.wsState, data, len, responses);
        if (!responses.empty()) {
            safeWrite(handle, responses);
        }
        for (const auto& msg : msgs) {
            if (msg.opcode == WS_OP_CLOSE) {
                engine_.closeConnection(handle);
                return;
            }
            if (msg.opcode == WS_OP_TEXT) {
                // Each WS text message is one line (or contains newlines)
                for (char ch : msg.payload) {
                    if (ch == '\n') {
                        if (!fd.lineBuf.empty() && fd.lineBuf.back() == '\r')
                            fd.lineBuf.pop_back();
                        processLine(fd, fd.lineBuf);
                        fd.lineBuf.clear();
                    } else {
                        fd.lineBuf += ch;
                    }
                }
                if (!fd.lineBuf.empty()) {
                    processLine(fd, fd.lineBuf);
                    fd.lineBuf.clear();
                }
            }
        }
        return;
    }

    // ---- WebSocket GameSession path ----
#ifdef GRPC_ENABLED
    if (fd.proto == FrontDoorProto::WsGameSession) {
        handleWsGameSessionData(fd, data, len);
        return;
    }
#endif

    // ---- gRPC-Web path ----
#ifdef GRPC_ENABLED
    if (fd.proto == FrontDoorProto::GrpcWeb) {
        if (fd.grpcWebSubscribed) {
            // This fd is a Subscribe stream — ignore incoming data.
            // The client shouldn't be sending anything.
            return;
        }
        fd.httpBuf.append(data, len);
        if (fd.httpBuf.size() > 1024 * 1024) {  // 1MB limit
            LOG_WARN("Front-door %lu: HTTP buffer too large (%zu bytes), closing",
                     (unsigned long)fd.handle, fd.httpBuf.size());
            std::string resp = "HTTP/1.1 413 Payload Too Large\r\n"
                               "Content-Length: 0\r\n\r\n";
            safeWrite(fd.handle, resp);
            engine_.closeConnection(fd.handle);
            return;
        }
        handleGrpcWebRequest(fd);
        return;
    }
#endif

    // ---- Telnet path ----

    // Split GMCP sub-negotiations from regular data
    std::string regular;
    std::vector<TelnetGmcpMessage> gmcpMsgs;
    TelnetSignals signals;
    splitTelnetStream(data, len, fd.telnetState, regular, gmcpMsgs, signals);

    // #1101: an IAC SB reassembly that overruns TELNET_SB_MAX trips sbOverflow
    // (the parser was already reset to stop unbounded growth).  Tear the
    // connection down — leaving it open lets the post-reset flood bytes desync
    // into the regular stream and lets the abuser repeat.  Do not touch fd
    // after closeConnection (its close callback may free it).
    if (fd.telnetState.sbOverflow) {
        LOG_WARN("front-door telnet SB reassembly overflow; closing connection");
        engine_.closeConnection(fd.handle);
        return;
    }

    // Track client GMCP capability
    if (signals.sawWillGmcp || signals.sawDoGmcp) {
        fd.gmcpEnabled = true;
    }
    if (signals.sawDoEor) {
        fd.eorEnabled = true;
    }
    if (signals.sawDontEor) {
        fd.eorEnabled = false;
    }

    // Forward client GMCP to the active back-door link
    if (!gmcpMsgs.empty() && fd.internalSessionId != InvalidHydraSessionId) {
        auto sit = sessions_.find(fd.internalSessionId);
        if (sit != sessions_.end()) {
            BackDoorLink* active = sit->second.getActiveLink();
            if (active && active->handle != ganl::InvalidConnectionHandle &&
                active->gmcpEnabled) {
                for (const auto& gm : gmcpMsgs) {
                    std::string frame = buildGmcpFrame(gm.payload);
                    safeWrite(active->handle, frame);
                }
            }
        }
    }

    // Process regular (non-GMCP) data as lines
    for (size_t i = 0; i < regular.size(); i++) {
        char ch = regular[i];
        if (ch == '\n') {
            if (!fd.lineBuf.empty() && fd.lineBuf.back() == '\r') {
                fd.lineBuf.pop_back();
            }
            processLine(fd, fd.lineBuf);
            fd.lineBuf.clear();
        } else if (ch == '\0') {
            // Telnet NUL after CR — ignore
        } else if (fd.lineBuf.size() < FrontDoorState::MAX_LINE_LENGTH) {
            fd.lineBuf += ch;
        } else {
            // Line too long — drop the connection
            LOG_WARN("Front-door %lu: line too long (%zu bytes), closing",
                     (unsigned long)fd.handle, fd.lineBuf.size());
            engine_.closeConnection(fd.handle);
            return;
        }
    }
}

void SessionManager::onFrontDoorData(ganl::ConnectionHandle handle,
                                     const char* data, size_t len) {
    auto it = frontDoors_.find(handle);
    if (it == frontDoors_.end()) return;

    FrontDoorState& fd = it->second;
    if (!fd.initialPromptSent && !fd.tlsTransport && fd.proto == FrontDoorProto::Telnet) {
        showBanner(handle);
        fd.initialPromptSent = true;
    }

    if (!fd.tlsTransport) {
        handleFrontDoorPlainData(fd, data, len);
        return;
    }

    if (!fd.tlsEncryptedIn) fd.tlsEncryptedIn = std::make_unique<ganl::IoBuffer>(4096);
    if (!fd.tlsDecryptedIn) fd.tlsDecryptedIn = std::make_unique<ganl::IoBuffer>(4096);
    if (!fd.tlsEncryptedOut) fd.tlsEncryptedOut = std::make_unique<ganl::IoBuffer>(4096);
    if (!fd.tlsPlainOut) fd.tlsPlainOut = std::make_unique<ganl::IoBuffer>(4096);

    fd.tlsEncryptedIn->append(data, len);

    while (true) {
        size_t encBefore = fd.tlsEncryptedIn->readableBytes();
        size_t decBefore = fd.tlsDecryptedIn->readableBytes();
        size_t outBefore = fd.tlsEncryptedOut->readableBytes();

        ganl::TlsResult res = fd.tlsTransport->processIncoming(
            handle, *fd.tlsEncryptedIn, *fd.tlsDecryptedIn, *fd.tlsEncryptedOut, true);

        if (fd.tlsTransport->isEstablished(handle) && !fd.tlsEstablished) {
            fd.tlsEstablished = true;
            // #1846: first post-handshake output — WILL EOR was deferred
            // from onAccept so it is not cleartext before TLS attachment.
            //
            if (!fd.initialPromptSent && fd.proto == FrontDoorProto::Telnet) {
                safeWrite(handle,
                    buildTelnetCommandFrame(telnet::WILL, telnet::EOR_OPT));
                showBanner(handle);
                fd.initialPromptSent = true;
            }
        }

        if (fd.tlsEncryptedOut->readableBytes() > 0) {
            if (!drainTlsCiphertext(fd)) return;
        }

        if (fd.tlsDecryptedIn->readableBytes() > 0) {
            std::string plain = fd.tlsDecryptedIn->consumeReadAllAsString();
            handleFrontDoorPlainData(fd, plain.data(), plain.size());
        }

        if (res == ganl::TlsResult::Error || res == ganl::TlsResult::Closed) {
            LOG_WARN("TLS incoming failed for fd %lu: %s",
                     (unsigned long)handle,
                     fd.tlsTransport->getLastTlsErrorString(handle).c_str());
            engine_.closeConnection(handle);
            return;
        }

        bool progressed =
            fd.tlsEncryptedIn->readableBytes() != encBefore ||
            fd.tlsDecryptedIn->readableBytes() != decBefore ||
            fd.tlsEncryptedOut->readableBytes() != outBefore;

        if (!progressed || res == ganl::TlsResult::WantRead) {
            break;
        }
    }
}

void SessionManager::onFrontDoorClose(ganl::ConnectionHandle handle) {
    writeBuffers_.erase(handle);
    auto it = frontDoors_.find(handle);
    if (it == frontDoors_.end()) return;

    FrontDoorState& fd = it->second;
    if (fd.tlsTransport) {
        fd.tlsTransport->destroySessionContext(handle);
    }
    HydraSessionId sid = fd.internalSessionId;
    std::string ip = fd.clientIp;

#ifdef GRPC_ENABLED
    // WsGameSession: unsubscribe from output queue (not in session.frontDoors)
    if (fd.proto == FrontDoorProto::WsGameSession && fd.wsGameSessionOQ) {
        std::lock_guard<std::mutex> lock(fd.wsGameSessionOQ->mutex);
        fd.wsGameSessionOQ->removeSubscriber(fd.wsGameSessionSubId);
        LOG_INFO("WsGameSession fd %lu: unsubscribed (sub %d)",
                 (unsigned long)handle, fd.wsGameSessionSubId);
    }
#endif

    frontDoors_.erase(it);
    recordDisconnect(ip);

    if (sid != InvalidHydraSessionId) {
        auto sit = sessions_.find(sid);
        if (sit != sessions_.end()) {
            HydraSession& session = sit->second;
            auto& fds = session.frontDoors;
            fds.erase(std::remove(fds.begin(), fds.end(), handle),
                      fds.end());

            if (fds.empty() && !session.outputQueue->hasSubscribers()) {
                LOG_INFO("Session %lu: no front-doors or subscribers, detaching",
                         (unsigned long)sid);
                session.state = SessionState::Detached;
                flushSession(session);
            }
        }
    }

    LOG_INFO("Front-door %lu closed", (unsigned long)handle);
}

#ifdef GRPC_ENABLED
void SessionManager::handleWsGameSessionData(FrontDoorState& fd,
                                              const char* data, size_t len) {
    // Decode WebSocket frames
    std::string responses;
    auto msgs = wsDecodeFrames(fd.wsState, data, len, responses);
    if (!responses.empty()) {
        safeWrite(fd.handle, responses);
    }

    for (const auto& msg : msgs) {
        if (msg.opcode == WS_OP_CLOSE) {
            engine_.closeConnection(fd.handle);
            return;
        }
        if (msg.opcode != WS_OP_BINARY) continue;

        hydra::ClientMessage cmsg;
        if (!cmsg.ParseFromString(msg.payload)) {
            LOG_WARN("WsGameSession fd %lu: invalid ClientMessage",
                     (unsigned long)fd.handle);
            continue;
        }

        // First-message auth: expect SetPreferences with session_id
        if (!fd.wsGameSessionOQ) {
            if (!cmsg.has_preferences() ||
                cmsg.preferences().session_id().empty()) {
                // Send error and close
                hydra::ServerMessage errMsg;
                auto* notice = errMsg.mutable_notice();
                notice->set_text("First message must be SetPreferences with session_id");
                notice->set_severity(hydra::SEVERITY_ERROR);
                std::string frame = wsEncodeFrame(errMsg.SerializeAsString(),
                                                  WS_OP_BINARY);
                safeWrite(fd.handle, frame);
                safeWrite(fd.handle, wsCloseFrame(1008));
                engine_.closeConnection(fd.handle);
                return;
            }

            const auto& prefs = cmsg.preferences();
            std::string sid = prefs.session_id();

            HydraSession* session = findByPersistId(sid);
            if (!session) {
                hydra::ServerMessage errMsg;
                auto* notice = errMsg.mutable_notice();
                notice->set_text("Invalid session_id");
                notice->set_severity(hydra::SEVERITY_ERROR);
                std::string frame = wsEncodeFrame(errMsg.SerializeAsString(),
                                                  WS_OP_BINARY);
                safeWrite(fd.handle, frame);
                safeWrite(fd.handle, wsCloseFrame(1008));
                engine_.closeConnection(fd.handle);
                return;
            }

            // Register subscriber (#1266: reject when at cap).
            {
                std::lock_guard<std::mutex> lock(session->outputQueue->mutex);
                auto [id, sq] = session->outputQueue->addSubscriber(true, true);
                fd.wsGameSessionSubId = id;
                fd.wsGameSessionQueue = sq;
            }
            if (!fd.wsGameSessionQueue) {
                hydra::ServerMessage errMsg;
                auto* notice = errMsg.mutable_notice();
                notice->set_text("Too many subscribers on this session");
                notice->set_severity(hydra::SEVERITY_ERROR);
                std::string frame = wsEncodeFrame(errMsg.SerializeAsString(),
                                                  WS_OP_BINARY);
                safeWrite(fd.handle, frame);
                safeWrite(fd.handle, wsCloseFrame(1008));
                engine_.closeConnection(fd.handle);
                return;
            }
            fd.wsGameSessionOQ = session->outputQueue;
            fd.wsGameSessionPersistId = sid;
            fd.internalSessionId = session->internalId;

            session->state = SessionState::Active;
            session->lastActivity = time(nullptr);

            // Replay GMCP cache
            replayGmcpCache(*session, fd.wsGameSessionQueue);

            // Apply preferences (color format, NAWS, ttype)
            if (prefs.color_format() != hydra::COLOR_UNSPECIFIED) {
                std::lock_guard<std::mutex> lock(fd.wsGameSessionOQ->mutex);
                fd.wsGameSessionQueue->renderFormat =
                    static_cast<HydraSession::RenderFormat>(prefs.color_format());
            }
            if (prefs.terminal_width() > 0 || prefs.terminal_height() > 0) {
                BackDoorLink* active = session->getActiveLink();
                if (active && active->handle != ganl::InvalidConnectionHandle) {
                    uint16_t w = prefs.terminal_width() ? static_cast<uint16_t>(prefs.terminal_width()) : 80;
                    uint16_t h = prefs.terminal_height() ? static_cast<uint16_t>(prefs.terminal_height()) : 24;
                    safeWrite(active->handle, buildNawsFrame(w, h));
                }
            }
            if (!prefs.terminal_type().empty()) {
                session->terminalType = prefs.terminal_type();
            }

            LOG_INFO("WsGameSession fd %lu: authenticated session %s (sub %d)",
                     (unsigned long)fd.handle, sid.c_str(), fd.wsGameSessionSubId);
            return;
        }

        // Authenticated — dispatch ClientMessage
        HydraSession* session = findByPersistId(fd.wsGameSessionPersistId);
        if (!session) continue;

        session->lastActivity = time(nullptr);

        if (cmsg.has_input_line()) {
            // #1268: match front-door telnet line limit.
            if (cmsg.input_line().size() > HydraSession::MAX_INPUT_LINE_LENGTH) {
                LOG_WARN("WsGameSession fd %lu: input_line too long (%zu), dropped",
                         (unsigned long)fd.handle, cmsg.input_line().size());
            } else {
                BackDoorLink* active = session->getActiveLink();
                if (active && active->state == LinkState::Active) {
                    std::string line = bridge_.convertInput(
                        ganl::EncodingType::Utf8,
                        active->protoState.encoding,
                        cmsg.input_line());
                    line += "\r\n";
                    safeWrite(active->handle, line);
                }
            }
        } else if (cmsg.has_ping()) {
            // Queue pong sentinel for the drain loop
            HydraSession::OutputItem pongItem;
            pongItem.puaText = "\x01PONG\x01" +
                std::to_string(cmsg.ping().client_timestamp());
            pongItem.source = "__pong__";
            pongItem.timestamp = 0;
            pongItem.linkNumber = 0;
            {
                std::lock_guard<std::mutex> lock(fd.wsGameSessionOQ->mutex);
                fd.wsGameSessionQueue->output.push(std::move(pongItem));
            }
            fd.wsGameSessionOQ->cv.notify_all();
        } else if (cmsg.has_preferences()) {
            const auto& prefs = cmsg.preferences();
            if (prefs.color_format() != hydra::COLOR_UNSPECIFIED) {
                std::lock_guard<std::mutex> lock(fd.wsGameSessionOQ->mutex);
                fd.wsGameSessionQueue->renderFormat =
                    static_cast<HydraSession::RenderFormat>(prefs.color_format());
            }
            if (prefs.terminal_width() > 0 || prefs.terminal_height() > 0) {
                BackDoorLink* active = session->getActiveLink();
                if (active && active->handle != ganl::InvalidConnectionHandle) {
                    uint16_t w = prefs.terminal_width() ? static_cast<uint16_t>(prefs.terminal_width()) : 80;
                    uint16_t h = prefs.terminal_height() ? static_cast<uint16_t>(prefs.terminal_height()) : 24;
                    safeWrite(active->handle, buildNawsFrame(w, h));
                }
            }
            if (!prefs.terminal_type().empty()) {
                session->terminalType = prefs.terminal_type();
            }
        } else if (cmsg.has_gmcp()) {
            BackDoorLink* active = session->getActiveLink();
            if (active && active->handle != ganl::InvalidConnectionHandle
                && active->gmcpEnabled) {
                std::string payload = cmsg.gmcp().package() + " " + cmsg.gmcp().json();
                safeWrite(active->handle, buildGmcpFrame(payload));
            }
        }
    }
}

void SessionManager::drainWsGameSessions() {
    for (auto& [handle, fd] : frontDoors_) {
        if (fd.proto != FrontDoorProto::WsGameSession) continue;
        if (!fd.wsGameSessionQueue) continue;

        auto& oq = fd.wsGameSessionOQ;
        auto& sq = fd.wsGameSessionQueue;

        HydraSession::RenderFormat fmt;
        std::vector<HydraSession::OutputItem> outputItems;
        std::vector<HydraSession::GmcpItem> gmcpItems;

        {
            std::lock_guard<std::mutex> lock(oq->mutex);
            if (sq->output.empty() && sq->gmcp.empty()) continue;
            fmt = sq->renderFormat;
            while (!sq->output.empty()) {
                outputItems.push_back(std::move(sq->output.front()));
                sq->output.pop();
            }
            while (!sq->gmcp.empty()) {
                gmcpItems.push_back(std::move(sq->gmcp.front()));
                sq->gmcp.pop();
            }
        }

        for (const auto& item : outputItems) {
            hydra::ServerMessage smsg;
            if (item.source == "__pong__") {
                auto* pong = smsg.mutable_pong();
                int64_t clientTs = 0;
                try { clientTs = std::stoll(item.puaText.substr(6)); }
                catch (...) {}
                pong->set_client_timestamp(clientTs);
                pong->set_server_timestamp(
                    std::chrono::duration_cast<std::chrono::milliseconds>(
                        std::chrono::system_clock::now().time_since_epoch()).count());
            } else {
                auto* go = smsg.mutable_game_output();
                go->set_text(sanitizeProtoTextForLog(
                    item.render(fmt), "ws gamesession output",
                    item.source, item.linkNumber));
                go->set_source(item.source);
                go->set_timestamp(static_cast<int64_t>(item.timestamp));
                go->set_link_number(item.linkNumber);
                go->set_end_of_record(item.endOfRecord);
            }
            std::string frame = wsEncodeFrame(smsg.SerializeAsString(),
                                              WS_OP_BINARY);
            safeWrite(handle, frame);
        }

        for (const auto& item : gmcpItems) {
            hydra::ServerMessage smsg;
            auto* gm = smsg.mutable_gmcp();
            gm->set_package(item.package);
            gm->set_json(item.json);
            gm->set_link_number(item.linkNumber);
            std::string frame = wsEncodeFrame(smsg.SerializeAsString(),
                                              WS_OP_BINARY);
            safeWrite(handle, frame);
        }
    }
}
#endif

void SessionManager::processLine(FrontDoorState& fd,
                                 const std::string& line) {
    if (fd.loginPhase != FrontDoorState::Authenticated) {
        handleLogin(fd, line);
        return;
    }

    auto sit = sessions_.find(fd.internalSessionId);
    if (sit == sessions_.end()) return;
    HydraSession& session = sit->second;

    session.lastActivity = time(nullptr);

    if (!line.empty() && line[0] == '/') {
        if (line.size() > 1 && line[1] == '/') {
            forwardToGame(session, fd.handle, line.substr(1));
            return;
        }
        dispatchCommand(session, fd.handle, line);
        return;
    }

    forwardToGame(session, fd.handle, line);
}

void SessionManager::handleLogin(FrontDoorState& fd,
                                 const std::string& line) {
    switch (fd.loginPhase) {
    case FrontDoorState::AwaitUsername: {
        if (line.substr(0, 7) == "create " && line.size() > 7) {
            // Credential commands require TLS regardless of allow_plaintext
            if (!fd.tlsTransport) {
                sendToClient(fd.handle,
                    "Account creation requires a TLS connection.\r\n"
                    "Username: ");
                return;
            }
            // Rate-limit account creation per IP
            if (!checkAccountCreateRate(fd.clientIp)) {
                sendToClient(fd.handle,
                    "Too many accounts created from this address. "
                    "Try again later.\r\nUsername: ");
                return;
            }
            size_t space = line.find(' ', 7);
            if (space != std::string::npos && space + 1 < line.size()) {
                std::string username = line.substr(7, space - 7);
                std::string password = line.substr(space + 1);
                bool admin = accounts_.isEmpty();

                uint32_t accountId = 0;
                std::string errorMsg;
                if (accounts_.createAccount(username, password, admin,
                                            accountId, errorMsg)) {
                    recordAccountCreate(fd.clientIp);
                    sendToClient(fd.handle,
                        "Account '" + username + "' created"
                        + (admin ? " (admin)" : "") + ".\r\n"
                        "Logging in...\r\n");

                    std::vector<uint8_t> sbKey;
                    uint32_t authId = accounts_.authenticate(
                        username, password, sbKey);
                    if (authId > 0) {
                        fd.pendingUsername = username;
                        fd.loginPhase = FrontDoorState::Authenticated;

                        HydraSession session;
                        session.internalId = nextSessionId_++;
                        session.accountId = authId;
                        session.username = username;
                        session.created = time(nullptr);
                        session.lastActivity = session.created;
                        session.scrollbackKey = sbKey;
                        session.persistId = generatePersistId();
                        session.dbPersistId = session.persistId;
                        session.tokenCreated = session.created;
                        session.scrollback = makeScrollback();
                        session.frontDoors.push_back(fd.handle);

                        fd.internalSessionId = session.internalId;
                        sessions_[session.internalId] = std::move(session);
                        indexSession(sessions_[fd.internalSessionId]);

                        LOG_INFO("Session %lu (%s) created for '%s'",
                                 (unsigned long)fd.internalSessionId,
                                 sessions_[fd.internalSessionId].persistId.c_str(),
                                 username.c_str());

                        showGameMenu(sessions_[fd.internalSessionId], fd.handle);
                    }
                } else {
                    sendToClient(fd.handle,
                        "Account creation failed: " + errorMsg + "\r\n"
                        "Username: ");
                }
                return;
            }
        }

        fd.pendingUsername = line;
        fd.loginPhase = FrontDoorState::AwaitPassword;
        sendToClient(fd.handle, "Password: ");
    } break;

    case FrontDoorState::AwaitPassword: {
        if (isLockedOut(fd.clientIp)) {
            sendToClient(fd.handle,
                "\r\nAccount temporarily locked. Try again later.\r\n");
            engine_.closeConnection(fd.handle);
            return;
        }

        std::vector<uint8_t> sbKey;
        uint32_t accountId = accounts_.authenticate(
            fd.pendingUsername, line, sbKey);

        if (accountId == 0) {
            LOG_INFO("Login failed for '%s' from fd %lu",
                     fd.pendingUsername.c_str(),
                     (unsigned long)fd.handle);
            authFailuresTotal_.fetch_add(1);
            recordLoginFailure(fd.clientIp);
            if (isLockedOut(fd.clientIp)) {
                sendToClient(fd.handle,
                    "\r\nToo many failed attempts. Try again later.\r\n");
                engine_.closeConnection(fd.handle);
                return;
            }
            sendToClient(fd.handle,
                "\r\nLogin failed.\r\n\r\nUsername: ");
            fd.loginPhase = FrontDoorState::AwaitUsername;
            fd.pendingUsername.clear();
            return;
        }

        clearLoginFailures(fd.clientIp);

        fd.loginPhase = FrontDoorState::Authenticated;

        // Check for existing in-memory session
        HydraSession* existing = nullptr;
        for (auto& [sid, sess] : sessions_) {
            if (sess.accountId == accountId) {
                existing = &sess;
                break;
            }
        }

        if (existing) {
            if (static_cast<int>(existing->frontDoors.size()) >= config_.maxFrontDoorsPerSession) {
                sendToClient(fd.handle,
                    "\r\nMaximum connections to this session reached.\r\n");
                engine_.closeConnection(fd.handle);
                return;
            }
            existing->frontDoors.push_back(fd.handle);
            existing->state = SessionState::Active;
            existing->lastActivity = time(nullptr);
            fd.internalSessionId = existing->internalId;

            // Provide the scrollback key if the session was restored without one
            if (existing->scrollbackKey.empty() && !sbKey.empty()) {
                existing->scrollbackKey = sbKey;
                existing->username = fd.pendingUsername;

                // Load persisted scroll-back now that we have the key
                // Use dbPersistId — persistId may have been rotated by gRPC re-auth
                const std::string& loadId = existing->dbPersistId.empty()
                    ? existing->persistId : existing->dbPersistId;
                existing->scrollback.loadFromDb(
                    accounts_.db(), loadId,
                    existing->accountId, sbKey);
            }

            // Reconnect deferred links (from eager restore without key)
            if (!existing->pendingLinksJson.empty()) {
                std::vector<SavedLinkInfo> savedLinks;
                size_t savedActiveLink = 0;
                parseLinksJson(existing->pendingLinksJson, savedLinks, savedActiveLink);
                existing->activeLink = savedActiveLink;
                restoreSessionLinks(*existing, savedLinks);
                existing->pendingLinksJson.clear();

                sendToClient(fd.handle,
                    "\r\nReconnecting " + std::to_string(savedLinks.size())
                    + " saved link(s)...\r\n");
            }

            LOG_INFO("Session %lu resumed for '%s'",
                     (unsigned long)existing->internalId,
                     fd.pendingUsername.c_str());

            sendToClient(fd.handle, "\r\nResuming session...\r\n");

            if (existing->scrollback.count() > 0) {
                sendToClient(fd.handle,
                    "-- Scroll-back ("
                    + std::to_string(existing->scrollback.count())
                    + " lines) --\r\n");

                ReplayContext rctx{fd.handle, fd.encoding, fd.colorDepth, &bridge_, this};
                existing->scrollback.replay(
                    existing->scrollback.count(),
                    [](const std::string& text,
                       const std::string& source,
                       time_t timestamp, void* ctx) {
                        auto* rc = static_cast<ReplayContext*>(ctx);
                        std::string rendered = rc->bridge->renderForClient(
                            rc->encoding, rc->colorDepth, text);
                        rendered += "\r\n";
                        rc->sessionMgr->safeWrite(rc->handle, rendered);
                    },
                    &rctx);

                sendToClient(fd.handle, "-- End scroll-back --\r\n");
            }

            if (!existing->links.empty()) {
                // Show link status
                for (size_t i = 0; i < existing->links.size(); i++) {
                    const auto& lnk = existing->links[i];
                    if (lnk.state == LinkState::Active ||
                        lnk.state == LinkState::Connecting) {
                        std::string marker = (i == existing->activeLink) ? "*" : " ";
                        sendToClient(fd.handle,
                            "[" + marker + std::to_string(i + 1) + "] "
                            + lnk.gameName + ": " + linkStateName(lnk.state)
                            + "\r\n");
                    }
                }
            } else {
                showGameMenu(*existing, fd.handle);
            }
        } else {
            // Check for saved session in SQLite (after restart)
            AccountManager::SavedSession saved;
            if (accounts_.loadSession(accountId, saved)) {
                resumeSavedSession(fd, accountId, sbKey);
                return;
            }

            HydraSession session;
            session.internalId = nextSessionId_++;
            session.accountId = accountId;
            session.username = fd.pendingUsername;
            session.created = time(nullptr);
            session.lastActivity = session.created;
            session.scrollbackKey = sbKey;
            session.persistId = generatePersistId();
            session.dbPersistId = session.persistId;
            session.tokenCreated = session.created;
            session.scrollback = makeScrollback();
            session.frontDoors.push_back(fd.handle);

            fd.internalSessionId = session.internalId;
            sessions_[session.internalId] = std::move(session);
            indexSession(sessions_[fd.internalSessionId]);

            LOG_INFO("Session %lu (%s) created for '%s'",
                     (unsigned long)fd.internalSessionId,
                     sessions_[fd.internalSessionId].persistId.c_str(),
                     fd.pendingUsername.c_str());

            sendToClient(fd.handle, "\r\nWelcome, "
                + fd.pendingUsername + ".\r\n");
            showGameMenu(sessions_[fd.internalSessionId], fd.handle);
        }
    } break;

    case FrontDoorState::Authenticated:
        break;
    }
}

// ---- Session commands ----

void SessionManager::dispatchCommand(HydraSession& session,
                                     ganl::ConnectionHandle fdHandle,
                                     const std::string& line) {
    std::string cmd = line.substr(1);
    size_t space = cmd.find(' ');
    std::string verb = cmd.substr(0, space);
    std::string args = (space != std::string::npos)
        ? cmd.substr(space + 1) : "";

    std::transform(verb.begin(), verb.end(), verb.begin(),
        [](unsigned char c) { return std::tolower(c); });

    if (verb == "games") {
        showGameMenu(session, fdHandle);
    } else if (verb == "connect") {
        if (args.empty()) {
            sendToClient(fdHandle, "Usage: /connect <game>\r\n");
            return;
        }
        connectToGame(session, args);
    } else if (verb == "switch") {
        if (args.empty()) {
            sendToClient(fdHandle, "Usage: /switch <link#>\r\n");
            return;
        }
        size_t idx = 0;
        try { idx = std::stoul(args); } catch (...) {}
        if (idx < 1 || idx > session.links.size()) {
            sendToClient(fdHandle, "Invalid link number. Use /links to see available links.\r\n");
            return;
        }
        session.activeLink = idx - 1;
        BackDoorLink& lnk = session.links[session.activeLink];
        sendToClient(fdHandle,
            "Switched to link " + std::to_string(idx) + " ("
            + lnk.gameName + ": " + linkStateName(lnk.state) + ")\r\n");
    } else if (verb == "disconnect") {
        if (args.empty()) {
            sendToClient(fdHandle, "Usage: /disconnect <link#>\r\n");
            return;
        }
        size_t idx = 0;
        try { idx = std::stoul(args); } catch (...) {}
        if (idx < 1 || idx > session.links.size()) {
            sendToClient(fdHandle, "Invalid link number.\r\n");
            return;
        }
        size_t li = idx - 1;
        closeLink(session, li);
        sendToClient(fdHandle,
            "Link " + std::to_string(idx) + " disconnected.\r\n");
    } else if (verb == "quit") {
        // Close all links
        for (size_t i = 0; i < session.links.size(); i++) {
            closeLink(session, i);
        }
        // Delete persisted session and scroll-back (use dbPersistId for FK)
        const std::string& delId = session.dbPersistId.empty()
            ? session.persistId : session.dbPersistId;
        if (!delId.empty()) {
            accounts_.deleteSession(delId);
        }
        for (auto h : session.frontDoors) {
            sendToClient(h, "Session destroyed. Goodbye.\r\n");
            engine_.closeConnection(h);
            frontDoors_.erase(h);
        }
        unindexSession(session);
        sessions_.erase(session.internalId);
    } else if (verb == "detach") {
        sendToClient(fdHandle, "Session detached. Reconnect to resume.\r\n");
        engine_.closeConnection(fdHandle);
    } else if (verb == "scroll") {
        size_t n = 20;
        if (!args.empty()) {
            try { n = std::stoul(args); } catch (...) {}
        }
        if (session.scrollback.count() == 0) {
            sendToClient(fdHandle, "[No scroll-back]\r\n");
        } else {
            sendToClient(fdHandle,
                "-- Last " + std::to_string(n) + " lines --\r\n");
            auto fdIt = frontDoors_.find(fdHandle);
            ganl::EncodingType enc = ganl::EncodingType::Utf8;
            ColorDepth depth = ColorDepth::Ansi256;
            if (fdIt != frontDoors_.end()) {
                enc = fdIt->second.encoding;
                depth = fdIt->second.colorDepth;
            }
            ReplayContext rctx{fdHandle, enc, depth, &bridge_, this};
            session.scrollback.replay(n,
                [](const std::string& text,
                   const std::string& source,
                   time_t timestamp, void* ctx) {
                    auto* rc = static_cast<ReplayContext*>(ctx);
                    std::string rendered = rc->bridge->renderForClient(
                        rc->encoding, rc->colorDepth, text);
                    rendered += "\r\n";
                    rc->sessionMgr->safeWrite(rc->handle, rendered);
                },
                &rctx);
            sendToClient(fdHandle, "-- End --\r\n");
        }
    } else if (verb == "links") {
        if (session.links.empty()) {
            sendToClient(fdHandle, "  No links.\r\n");
        } else {
            std::string out;
            for (size_t i = 0; i < session.links.size(); i++) {
                const BackDoorLink& lnk = session.links[i];
                std::string marker = (i == session.activeLink) ? "*" : " ";
                out += "  [" + marker + std::to_string(i + 1) + "] "
                     + lnk.gameName;
                if (!lnk.character.empty()) {
                    out += " (" + lnk.character + ")";
                }
                out += " — " + std::string(linkStateName(lnk.state));
                if (lnk.state == LinkState::Reconnecting) {
                    time_t now = time(nullptr);
                    if (lnk.nextRetry > now) {
                        out += " (retry in "
                             + std::to_string(lnk.nextRetry - now) + "s)";
                    }
                }
                out += "\r\n";
            }
            sendToClient(fdHandle, out);
        }
    } else if (verb == "addcred") {
        // Credential commands require TLS regardless of allow_plaintext
        auto fdIt2 = frontDoors_.find(fdHandle);
        if (fdIt2 != frontDoors_.end() && !fdIt2->second.tlsTransport) {
            sendToClient(fdHandle,
                "Credential storage requires a TLS connection.\r\n");
            return;
        }
        std::vector<std::string> parts;
        size_t pos = 0;
        std::string tmp = args;
        while (parts.size() < 4 && (pos = tmp.find(' ')) != std::string::npos) {
            parts.push_back(tmp.substr(0, pos));
            tmp = tmp.substr(pos + 1);
        }
        if (!tmp.empty()) parts.push_back(tmp);

        if (parts.size() < 5) {
            sendToClient(fdHandle,
                "Usage: /addcred <game> <character> <verb> <name> <secret>\r\n"
                "Example: /addcred mygame player1 connect player1 mypassword\r\n");
            return;
        }

        std::string errorMsg;
        if (accounts_.storeCredential(session.accountId,
                                      parts[0], parts[1], parts[2],
                                      parts[3], parts[4], errorMsg)) {
            sendToClient(fdHandle,
                "Credential stored for " + parts[0] + "/" + parts[1] + "\r\n");
        } else {
            sendToClient(fdHandle, "Failed: " + errorMsg + "\r\n");
        }
    } else if (verb == "delcred") {
        if (args.empty()) {
            sendToClient(fdHandle, "Usage: /delcred <game> [character]\r\n");
            return;
        }
        std::string game = args;
        std::string character;
        size_t sp = args.find(' ');
        if (sp != std::string::npos) {
            game = args.substr(0, sp);
            character = args.substr(sp + 1);
        }
        if (accounts_.deleteCredential(session.accountId, game, character)) {
            sendToClient(fdHandle, "Credential(s) deleted.\r\n");
        } else {
            sendToClient(fdHandle, "Delete failed.\r\n");
        }
    } else if (verb == "creds") {
        auto creds = accounts_.listCredentials(session.accountId);
        if (creds.empty()) {
            sendToClient(fdHandle, "No stored credentials.\r\n");
        } else {
            std::string out = "--- Stored Credentials ---\r\n";
            for (const auto& c : creds) {
                out += "  " + c.game + "/" + c.character
                     + "  verb=" + c.verb + " name=" + c.name
                     + (c.autoLogin ? " [auto]" : "") + "\r\n";
            }
            sendToClient(fdHandle, out);
        }
    } else if (verb == "start") {
        if (!accounts_.isAdmin(session.accountId)) {
            sendToClient(fdHandle, "Permission denied (admin required).\r\n");
            return;
        }
        if (args.empty()) {
            sendToClient(fdHandle, "Usage: /start <game>\r\n");
            return;
        }
        const GameConfig* game = nullptr;
        for (const auto& g : config_.games) {
            if (g.name == args) { game = &g; break; }
        }
        if (!game) {
            sendToClient(fdHandle, "Unknown game: " + args + "\r\n");
        } else if (game->type != GameType::Local) {
            sendToClient(fdHandle, args + " is a remote game (cannot start).\r\n");
        } else {
            std::string errorMsg;
            if (procMgr_.startGame(*game, errorMsg)) {
                sendToClient(fdHandle,
                    "Started " + args + " (pid "
                    + std::to_string(procMgr_.getPid(args)) + ")\r\n");
            } else {
                sendToClient(fdHandle, "Start failed: " + errorMsg + "\r\n");
            }
        }
    } else if (verb == "stop") {
        if (!accounts_.isAdmin(session.accountId)) {
            sendToClient(fdHandle, "Permission denied (admin required).\r\n");
            return;
        }
        if (args.empty()) {
            sendToClient(fdHandle, "Usage: /stop <game>\r\n");
            return;
        }
        if (procMgr_.stopGame(args)) {
            sendToClient(fdHandle, "Stopping " + args + "...\r\n");
        } else {
            sendToClient(fdHandle, args + " is not running.\r\n");
        }
    } else if (verb == "restart") {
        if (!accounts_.isAdmin(session.accountId)) {
            sendToClient(fdHandle, "Permission denied (admin required).\r\n");
            return;
        }
        if (args.empty()) {
            sendToClient(fdHandle, "Usage: /restart <game>\r\n");
            return;
        }
        const GameConfig* game = nullptr;
        for (const auto& g : config_.games) {
            if (g.name == args) { game = &g; break; }
        }
        if (!game) {
            sendToClient(fdHandle, "Unknown game: " + args + "\r\n");
        } else if (game->type != GameType::Local) {
            sendToClient(fdHandle, args + " is a remote game.\r\n");
        } else {
            std::string errorMsg;
            if (procMgr_.restartGame(*game, errorMsg)) {
                sendToClient(fdHandle,
                    "Restarted " + args + " (pid "
                    + std::to_string(procMgr_.getPid(args)) + ")\r\n");
            } else {
                sendToClient(fdHandle, "Restart failed: " + errorMsg + "\r\n");
            }
        }
    } else if (verb == "passwd") {
        // Require TLS for password changes.
        auto fdIt2 = frontDoors_.find(fdHandle);
        if (fdIt2 != frontDoors_.end() && !fdIt2->second.tlsTransport) {
            sendToClient(fdHandle,
                "Password changes require a TLS connection.\r\n");
            return;
        }
        // #1097: require current password so a stolen session cannot take over.
        size_t sp = args.find(' ');
        if (args.empty() || sp == std::string::npos || sp == 0 ||
            sp + 1 >= args.size()) {
            sendToClient(fdHandle,
                "Usage: /passwd <oldpassword> <newpassword>\r\n");
            return;
        }
        std::string oldPassword = args.substr(0, sp);
        std::string newPassword = args.substr(sp + 1);
        // Trim leading spaces on new password only (allow spaces in new? no — first word)
        while (!newPassword.empty() && newPassword[0] == ' ') {
            newPassword.erase(newPassword.begin());
        }
        // If new password has more spaces, keep rest as part of password.
        std::vector<uint8_t> checkKey;
        uint32_t authId = accounts_.authenticate(
            session.username, oldPassword, checkKey);
        if (authId == 0 || authId != session.accountId) {
            sendToClient(fdHandle, "Password change failed: wrong password.\r\n");
            return;
        }
        std::vector<uint8_t> newKey;
        std::string errorMsg;
        if (accounts_.changePassword(session.accountId, newPassword,
                                      session.scrollbackKey,
                                      newKey, errorMsg)) {
            session.scrollbackKey = newKey;
            sendToClient(fdHandle, "Password changed.\r\n");
        } else {
            sendToClient(fdHandle,
                "Password change failed: " + errorMsg + "\r\n");
        }
    } else {
        sendToClient(fdHandle,
            "Unknown command: /" + verb + "\r\n"
            "Commands: /games /connect /switch /disconnect /links\r\n"
            "          /scroll /detach /quit /passwd\r\n"
            "          /start /stop /restart\r\n"
            "          /addcred /delcred /creds\r\n");
    }
}

void SessionManager::connectToGame(HydraSession& session,
                                   const std::string& gameName) {
    // Find game config
    const GameConfig* game = nullptr;
    for (const auto& g : config_.games) {
        if (g.name == gameName) {
            game = &g;
            break;
        }
    }

    if (!game) {
        for (auto h : session.frontDoors) {
            sendToClient(h, "Unknown game: " + gameName + "\r\n");
        }
        return;
    }

    // Enforce back-door TLS policy: if tls_required is set but the game
    // doesn't have TLS enabled, reject the connection.
    if (game->tlsRequired && !game->tls) {
        for (auto h : session.frontDoors) {
            sendToClient(h,
                "[" + gameName + ": TLS is required but not configured. "
                "Set 'tls_required no' in the game block to allow "
                "plaintext back-door connections.]\r\n");
        }
        return;
    }

    // Check for existing link to this game.
    // If active/connecting/reconnecting, reject the duplicate.
    // If dead, reuse the slot.
    size_t reuseIdx = SIZE_MAX;
    for (size_t i = 0; i < session.links.size(); i++) {
        if (session.links[i].gameName == gameName) {
            if (session.links[i].state != LinkState::Dead) {
                for (auto h : session.frontDoors) {
                    sendToClient(h,
                        "[Already connected to " + gameName + " (link "
                        + std::to_string(i + 1) + ")]\r\n");
                }
                return;
            }
            if (reuseIdx == SIZE_MAX) {
                reuseIdx = i;
            }
        }
    }

    // Check max links limit (dead slots being reused don't count)
    if (reuseIdx == SIZE_MAX &&
        session.links.size() >= static_cast<size_t>(config_.maxLinksPerSession)) {
        for (auto h : session.frontDoors) {
            sendToClient(h, "Maximum links reached ("
                + std::to_string(config_.maxLinksPerSession) + ").\r\n");
        }
        return;
    }

    // Autostart local game if configured and not running
    if (game->type == GameType::Local && game->autostart &&
        !procMgr_.isRunning(game->name)) {
        std::string startErr;
        if (procMgr_.startGame(*game, startErr)) {
            LOG_INFO("Autostarted game '%s'", game->name.c_str());
            for (auto h : session.frontDoors) {
                sendToClient(h,
                    "[" + game->name + ": autostarted (pid "
                    + std::to_string(procMgr_.getPid(game->name)) + ")]\r\n");
            }
        } else {
            LOG_WARN("Autostart failed for '%s': %s",
                     game->name.c_str(), startErr.c_str());
        }
    }

    ganl::ErrorCode err = 0;
    ganl::ConnectionHandle bdHandle;

    if (game->transport == GameTransport::Unix) {
        bdHandle = engine_.initiateUnixConnect(
            game->socketPath, nullptr, err);
    } else {
        bdHandle = engine_.initiateConnect(
            game->host, game->port, nullptr, err);
    }

    if (bdHandle == ganl::InvalidConnectionHandle) {
        for (auto h : session.frontDoors) {
            sendToClient(h,
                "[" + game->name + ": connection failed: "
                + strerror(err) + "]\r\n");
        }
        return;
    }

    // Reuse a dead slot or create a new link.
    size_t linkIdx;
    if (reuseIdx != SIZE_MAX) {
        linkIdx = reuseIdx;
        BackDoorLink& reused = session.links[linkIdx];
        reused.handle = bdHandle;
        reused.gameName = game->name;
        reused.state = LinkState::Connecting;
        reused.protoState.encoding = game->charset;
        reused.gameConfig = game;
        reused.retryCount = 0;
        reused.nextRetry = 0;
        reused.character.clear();
        reused.gmcpEnabled = false;
        reused.sentWillTtype = false;
        reused.sentWillNaws = false;
        reused.sentWillCharset = false;
        reused.sentWillEor = false;
        reused.utf8Carry.clear();
        reused.telnetState = TelnetParseState{};
    } else {
        BackDoorLink link;
        link.handle = bdHandle;
        link.gameName = game->name;
        link.state = LinkState::Connecting;
        link.protoState.encoding = game->charset;
        link.gameConfig = game;
        link.utf8Carry.clear();
        linkIdx = session.links.size();
        session.links.push_back(std::move(link));
    }
    session.activeLink = linkIdx;

    backDoorMap_[bdHandle] = {session.internalId, linkIdx};

    for (auto h : session.frontDoors) {
        sendToClient(h,
            "[" + game->name + ": connecting... (link "
            + std::to_string(linkIdx + 1) + ")]\r\n");
    }

    LOG_INFO("Session %lu: connecting link %zu to %s (%s:%u)",
             (unsigned long)session.internalId, linkIdx + 1,
             game->name.c_str(), game->host.c_str(), game->port);
}

void SessionManager::forwardToGame(HydraSession& session,
                                   ganl::ConnectionHandle fdHandle,
                                   const std::string& line) {
    BackDoorLink* active = session.getActiveLink();
    if (!active || active->handle == ganl::InvalidConnectionHandle ||
        active->state != LinkState::Active) {
        for (auto h : session.frontDoors) {
            sendToClient(h,
                "[No active game link. Use /connect <game>]\r\n");
        }
        return;
    }

    ganl::EncodingType clientEnc = ganl::EncodingType::Utf8;
    auto fdIt = frontDoors_.find(fdHandle);
    if (fdIt != frontDoors_.end()) {
        clientEnc = fdIt->second.encoding;
    }

    std::string converted = bridge_.convertInput(
        clientEnc, active->protoState.encoding, line);
    converted += "\r\n";

    safeWrite(active->handle, converted);
}

void SessionManager::closeLink(HydraSession& session, size_t linkIdx) {
    if (linkIdx >= session.links.size()) return;
    BackDoorLink& link = session.links[linkIdx];

    if (link.handle != ganl::InvalidConnectionHandle) {
        backDoorMap_.erase(link.handle);
        engine_.closeConnection(link.handle);
        link.handle = ganl::InvalidConnectionHandle;
    }
    link.state = LinkState::Dead;
    link.retryCount = 0;
    link.nextRetry = 0;
    link.sentWillTtype = false;
    link.sentWillNaws = false;
    link.sentWillCharset = false;
    link.sentWillEor = false;
    link.utf8Carry.clear();
    link.telnetState = TelnetParseState{};
}

// ---- Back-door lifecycle ----

void SessionManager::onBackDoorConnect(ganl::ConnectionHandle bdHandle) {
    HydraSession* session = nullptr;
    BackDoorLink* link = nullptr;
    size_t linkIdx = 0;
    if (!findByBackDoor(bdHandle, session, link, linkIdx)) return;

    link->state = LinkState::Active;
    link->retryCount = 0;

    LOG_INFO("Session %lu: link %zu connected to %s",
             (unsigned long)session->internalId, linkIdx + 1,
             link->gameName.c_str());

    for (auto h : session->frontDoors) {
        sendToClient(h,
            "[" + link->gameName + " (link "
            + std::to_string(linkIdx + 1) + "): connected]\r\n");
    }

    // Advertise the back-door capabilities Hydra can actually drive.
    // This makes later GMCP/TTYPE/NAWS traffic standards-compliant instead of
    // relying on tolerant servers to accept unsolicited subnegotiations.
    safeWrite(link->handle, buildTelnetCommandFrame(telnet::DO, telnet::GMCP));
    safeWrite(link->handle, buildTelnetCommandFrame(telnet::DO, telnet::EOR_OPT));
    safeWrite(link->handle, buildTelnetCommandFrame(telnet::WILL, telnet::TTYPE));
    safeWrite(link->handle, buildTelnetCommandFrame(telnet::WILL, telnet::NAWS));
    safeWrite(link->handle, buildTelnetCommandFrame(telnet::WILL, telnet::CHARSET));
    link->sentWillTtype = true;
    link->sentWillNaws = true;
    link->sentWillCharset = true;
    link->sentWillEor = false;

    // Send Core.Hello to the game if GMCP is enabled on this link
    if (link->gmcpEnabled) {
        std::string hello = "Core.Hello {\"client\":\"Hydra\",\"version\":\"0.2\"}";
        safeWrite(link->handle, buildGmcpFrame(hello));
    }

    // Send Hydra.Links GMCP to all GMCP-enabled front-doors
    sendHydraLinksGmcp(*session, frontDoors_,
        [this](ganl::ConnectionHandle h, const std::string& data) {
            safeWrite(h, data);
        });

    // Auto-login
    std::string verb, loginName, secret;
    if (accounts_.getLoginSecret(session->accountId, link->gameName,
                                 verb, loginName, secret)) {
        std::string loginCmd = verb + " " + loginName + " " + secret + "\r\n";
        safeWrite(link->handle, loginCmd);

        link->character = loginName;

        LOG_INFO("Session %lu: auto-login sent for link %zu (%s)",
                 (unsigned long)session->internalId, linkIdx + 1,
                 link->gameName.c_str());

        for (auto h : session->frontDoors) {
            sendToClient(h,
                "[" + link->gameName + ": auto-login sent]\r\n");
        }
    }
}

void SessionManager::onBackDoorConnectFail(ganl::ConnectionHandle bdHandle,
                                           int error) {
    backendConnectFailuresTotal_.fetch_add(1);
    HydraSession* session = nullptr;
    BackDoorLink* link = nullptr;
    size_t linkIdx = 0;
    if (!findByBackDoor(bdHandle, session, link, linkIdx)) {
        backDoorMap_.erase(bdHandle);
        return;
    }

    link->state = LinkState::Dead;
    link->handle = ganl::InvalidConnectionHandle;
    backDoorMap_.erase(bdHandle);

    LOG_ERROR("Session %lu: link %zu connect failed: %s",
              (unsigned long)session->internalId, linkIdx + 1, strerror(error));

    for (auto h : session->frontDoors) {
        sendToClient(h,
            "[" + link->gameName + " (link "
            + std::to_string(linkIdx + 1) + "): connection failed: "
            + strerror(error) + "]\r\n");
    }
}

void SessionManager::onBackDoorData(ganl::ConnectionHandle bdHandle,
                                    const char* data, size_t len) {
    HydraSession* session = nullptr;
    BackDoorLink* link = nullptr;
    size_t linkIdx = 0;
    if (!findByBackDoor(bdHandle, session, link, linkIdx)) return;

    // Split GMCP sub-negotiations from regular data.
    // Strip all telnet sequences — back-door data should be clean text.
    std::string regular;
    std::vector<TelnetGmcpMessage> gmcpMsgs;
    TelnetSignals signals;
    splitTelnetStream(data, len, link->telnetState, regular, gmcpMsgs, signals, true);

    // #1101: same SB-overflow guard as the front-door — drop the misbehaving
    // back-door link rather than desync its stream.  Do not touch link/session
    // after closeConnection (onBackDoorClose may invalidate them).
    if (link->telnetState.sbOverflow) {
        LOG_WARN("back-door telnet SB reassembly overflow; closing link");
        engine_.closeConnection(bdHandle);
        return;
    }

    // Track GMCP capability on the back-door link
    if (signals.sawWillGmcp || signals.sawDoGmcp) {
        link->gmcpEnabled = true;
    }

    if (signals.sawDoTtype && !link->sentWillTtype) {
        link->sentWillTtype = true;
        safeWrite(link->handle,
                  buildTelnetCommandFrame(telnet::WILL, telnet::TTYPE));
    }
    if (signals.sawDontTtype) {
        link->sentWillTtype = false;
    }
    if (signals.sawDoCharset && !link->sentWillCharset) {
        link->sentWillCharset = true;
        safeWrite(link->handle,
                  buildTelnetCommandFrame(telnet::WILL, telnet::CHARSET));
    }
    if (signals.sawDontCharset) {
        link->sentWillCharset = false;
    }
    if (signals.sawDoEor && !link->sentWillEor) {
        link->sentWillEor = true;
        safeWrite(link->handle,
                  buildTelnetCommandFrame(telnet::WILL, telnet::EOR_OPT));
    }
    if (signals.sawDontEor) {
        link->sentWillEor = false;
    }
    if (signals.sawDontNaws) {
        link->sentWillNaws = false;
    }
    if (signals.sawTtypeSend) {
        safeWrite(link->handle,
                  buildTtypeIsFrame(session->terminalType.empty()
                      ? std::string("Hydra")
                      : session->terminalType));
    }
    if (signals.sawCharsetRequest) {
        std::string wanted = hydraCharsetName(link->protoState.encoding);
        if (charsetOffered(signals.charsetRequestPayload, wanted)) {
            safeWrite(link->handle, buildCharsetAcceptedFrame(wanted));
        } else {
            safeWrite(link->handle, buildCharsetRejectedFrame());
        }
    }
    if (signals.sawCharsetAccepted) {
        link->protoState.encoding =
            parseNegotiatedCharset(signals.charsetAcceptedPayload, link->protoState.encoding);
    }
    if (signals.sawCharsetRejected) {
        LOG_INFO("Session %lu: link %zu charset negotiation rejected by %s",
                 (unsigned long)session->internalId, linkIdx + 1,
                 link->gameName.c_str());
    }

    // Forward GMCP messages to all GMCP-enabled front-doors
    for (const auto& gm : gmcpMsgs) {
        // Cache GMCP state for replay on reconnect
        {
            std::string pkg;
            size_t sp = gm.payload.find(' ');
            if (sp != std::string::npos) {
                pkg = gm.payload.substr(0, sp);
                // Cap cache size — update existing keys freely, but
                // reject new keys once the limit is reached.
                if (session->gmcpCache.count(pkg) ||
                    session->gmcpCache.size() < HydraSession::MAX_GMCP_CACHE_ENTRIES) {
                    session->gmcpCache[pkg] = gm.payload.substr(sp + 1);
                }
            }
        }

        std::string frame = buildGmcpFrame(gm.payload);
        for (auto h : session->frontDoors) {
            auto fdIt = frontDoors_.find(h);
            if (fdIt == frontDoors_.end()) continue;
            if (!fdIt->second.gmcpEnabled) continue;
            safeWrite(h, frame);
        }

        // Push to gRPC GMCP queue — replicated to all subscribers
        {
            std::lock_guard<std::mutex> lock(session->outputQueue->mutex);
            if (session->outputQueue->hasGmcpSubscribers()) {
                // Parse GMCP payload: "Package.Name json_data"
                std::string pkg, json;
                size_t sp = gm.payload.find(' ');
                if (sp != std::string::npos) {
                    pkg = gm.payload.substr(0, sp);
                    json = gm.payload.substr(sp + 1);
                } else {
                    pkg = gm.payload;
                }

                HydraSession::GmcpItem item;
                item.package = pkg;
                item.json = json;
                item.linkNumber = static_cast<int>(linkIdx) + 1;

                session->outputQueue->pushGmcp(std::move(item));
            }
        }
    }

    // Process regular (non-GMCP) data through the color/charset bridge
    if (!regular.empty()) {
        std::string puaText = bridge_.ingestGameOutput(
            link->protoState, regular.data(), regular.size(), &link->utf8Carry);

        // Track global scrollback memory via atomic counter (O(1) per append).
        size_t oldBytes = session->scrollback.memoryBytes();
        session->scrollback.append(puaText, link->gameName);
        size_t newBytes = session->scrollback.memoryBytes();
        int64_t delta = static_cast<int64_t>(newBytes) - static_cast<int64_t>(oldBytes);
        if (delta > 0) globalScrollbackBytes_.fetch_add(static_cast<size_t>(delta));
        else if (delta < 0) globalScrollbackBytes_.fetch_sub(static_cast<size_t>(-delta));

        if (config_.maxScrollbackMemoryMb > 0) {
            size_t limitBytes = config_.maxScrollbackMemoryMb * 1024 * 1024;
            if (globalScrollbackBytes_.load() >= limitBytes && !scrollbackLimitWarned_) {
                LOG_WARN("Global scrollback memory limit reached (%zu MB)",
                         config_.maxScrollbackMemoryMb);
                scrollbackLimitWarned_ = true;
            } else if (globalScrollbackBytes_.load() < limitBytes) {
                scrollbackLimitWarned_ = false;
            }
        }

        for (auto h : session->frontDoors) {
            auto fdIt = frontDoors_.find(h);
            if (fdIt == frontDoors_.end()) continue;
            const FrontDoorState& fd = fdIt->second;

            std::string rendered = bridge_.renderForClient(
                fd.encoding, fd.colorDepth, puaText);

            if (fd.proto == FrontDoorProto::WebSocket) {
                std::string frame = wsEncodeFrame(rendered);
                safeWrite(h, frame);
#ifdef GRPC_ENABLED
            } else if (fd.grpcWebSubscribed) {
                // Send as chunked grpc-web data frame
                hydra::GameOutput go;
                go.set_text(sanitizeProtoTextForLog(
                    rendered, "grpc-web live output", link->gameName,
                    static_cast<int>(linkIdx) + 1));
                go.set_source(link->gameName);
                go.set_timestamp(static_cast<int64_t>(time(nullptr)));
                go.set_link_number(static_cast<int>(linkIdx) + 1);

                std::string gwFrame = grpcWebEncodeDataFrame(
                    go.SerializeAsString());
                if (fd.grpcWebTextMode) gwFrame = base64Encode(gwFrame);

                std::string chunk = std::to_string(gwFrame.size())
                    + "\r\n" + gwFrame + "\r\n";
                safeWrite(h, chunk);
#endif
            } else {
                safeWrite(h, rendered);
            }
        }

        // Push PUA-encoded text to gRPC output queue — each subscriber
        // renders at their preferred color format when reading.
        {
            std::lock_guard<std::mutex> lock(session->outputQueue->mutex);
            if (session->outputQueue->hasOutputSubscribers()) {
                HydraSession::OutputItem item;
                item.puaText = puaText;
                item.source = link->gameName;
                item.timestamp = time(nullptr);
                item.linkNumber = static_cast<int>(linkIdx) + 1;

                session->outputQueue->pushOutput(std::move(item));
            }
        }
    }

    if (signals.sawEor) {
        for (auto h : session->frontDoors) {
            auto fdIt = frontDoors_.find(h);
            if (fdIt == frontDoors_.end()) continue;
            const FrontDoorState& fd = fdIt->second;

            if (fd.proto == FrontDoorProto::WebSocket) {
                continue;
#ifdef GRPC_ENABLED
            } else if (fd.grpcWebSubscribed) {
                hydra::GameOutput go;
                go.set_text("");
                go.set_source(link->gameName);
                go.set_timestamp(static_cast<int64_t>(time(nullptr)));
                go.set_link_number(static_cast<int>(linkIdx) + 1);
                go.set_end_of_record(true);

                std::string gwFrame = grpcWebEncodeDataFrame(go.SerializeAsString());
                if (fd.grpcWebTextMode) gwFrame = base64Encode(gwFrame);
                std::string chunk = std::to_string(gwFrame.size())
                    + "\r\n" + gwFrame + "\r\n";
                safeWrite(h, chunk);
#endif
            } else if (fd.eorEnabled) {
                safeWrite(h, buildTelnetTwoByteCommand(telnet::EOR_CMD));
            }
        }

        {
            std::lock_guard<std::mutex> lock(session->outputQueue->mutex);
            if (session->outputQueue->hasOutputSubscribers()) {
                HydraSession::OutputItem item;
                item.source = link->gameName;
                item.timestamp = time(nullptr);
                item.linkNumber = static_cast<int>(linkIdx) + 1;
                item.endOfRecord = true;
                session->outputQueue->pushOutput(std::move(item));
            }
        }
    }
}

void SessionManager::onBackDoorClose(ganl::ConnectionHandle bdHandle) {
    writeBuffers_.erase(bdHandle);
    backendDisconnectsTotal_.fetch_add(1);
    HydraSession* session = nullptr;
    BackDoorLink* link = nullptr;
    size_t linkIdx = 0;
    if (!findByBackDoor(bdHandle, session, link, linkIdx)) {
        backDoorMap_.erase(bdHandle);
        return;
    }

    backDoorMap_.erase(bdHandle);
    link->handle = ganl::InvalidConnectionHandle;

    // Check for reconnect
    if (link->gameConfig && link->gameConfig->reconnect &&
        (link->state == LinkState::Active ||
         link->state == LinkState::Reconnecting)) {

        const auto& schedule = link->gameConfig->retrySchedule;
        if (link->retryCount < static_cast<int>(schedule.size())) {
            int delay = schedule[static_cast<size_t>(link->retryCount)];
            link->state = LinkState::Reconnecting;
            link->nextRetry = time(nullptr) + delay;
            link->retryCount++;

            LOG_INFO("Session %lu: link %zu (%s) lost, reconnecting in %ds",
                     (unsigned long)session->internalId, linkIdx + 1,
                     link->gameName.c_str(), delay);

            for (auto h : session->frontDoors) {
                sendToClient(h,
                    "[" + link->gameName + ": disconnected, retrying in "
                    + std::to_string(delay) + "s]\r\n");
            }
            sendHydraLinksGmcp(*session, frontDoors_,
                [this](ganl::ConnectionHandle h, const std::string& d) { safeWrite(h, d); });
            return;
        }
    }

    link->state = LinkState::Dead;

    LOG_INFO("Session %lu: link %zu (%s) lost",
             (unsigned long)session->internalId, linkIdx + 1,
             link->gameName.c_str());

    for (auto h : session->frontDoors) {
        sendToClient(h,
            "[" + link->gameName + " (link "
            + std::to_string(linkIdx + 1) + "): disconnected]\r\n");
    }
    sendHydraLinksGmcp(*session, frontDoors_,
        [this](ganl::ConnectionHandle h, const std::string& d) { safeWrite(h, d); });
}

bool SessionManager::isBackDoor(ganl::ConnectionHandle handle) const {
    return backDoorMap_.find(handle) != backDoorMap_.end();
}

void SessionManager::setFrontDoorTls(ganl::ConnectionHandle handle,
                                      ganl::SecureTransport* transport) {
    auto it = frontDoors_.find(handle);
    if (it != frontDoors_.end()) {
        it->second.tlsTransport = transport;
        it->second.tlsEstablished = false;
        it->second.initialPromptSent = false;
        it->second.tlsEncryptedIn = std::make_unique<ganl::IoBuffer>(4096);
        it->second.tlsDecryptedIn = std::make_unique<ganl::IoBuffer>(4096);
        it->second.tlsPlainOut = std::make_unique<ganl::IoBuffer>(4096);
        it->second.tlsEncryptedOut = std::make_unique<ganl::IoBuffer>(4096);
    }
}

void SessionManager::runTimers() {
    time_t now = time(nullptr);

    // Periodic scroll-back flush.
    // Sessions with active gRPC subscribers flush every 15 seconds
    // (narrower crash window).  Others flush every 60 seconds.
    for (auto& [sid, session] : sessions_) {
        if (session.scrollback.dirtyCount() == 0) continue;

        bool hasSubscribers = false;
        {
            std::lock_guard<std::mutex> lock(session.outputQueue->mutex);
            hasSubscribers = !session.outputQueue->subscribers.empty();
        }

        int interval = hasSubscribers ? 15 : 60;
        if (now - lastFlush_ >= interval) {
            flushSession(session);
        }
    }
    bool ranFlushTick = false;
    if (now - lastFlush_ >= 15) {
        lastFlush_ = now;
        ranFlushTick = true;
    }

    // GMCP Core.KeepAlive to active game links (prevents idle disconnects).
    // Send once per flush tick rather than in a short burst window.
    if (ranFlushTick) {
        for (auto& [sid, session] : sessions_) {
            for (auto& link : session.links) {
                if (link.state == LinkState::Active && link.gmcpEnabled &&
                    link.handle != ganl::InvalidConnectionHandle) {
                    safeWrite(link.handle,
                        buildGmcpFrame("Core.KeepAlive"));
                }
            }
        }
    }

    // Reap child processes
    procMgr_.reapChildren();

    // Reconnect backoff
    for (auto& [sid, session] : sessions_) {
        for (size_t i = 0; i < session.links.size(); i++) {
            BackDoorLink& link = session.links[i];
            if (link.state != LinkState::Reconnecting) continue;
            if (now < link.nextRetry) continue;

            // Attempt reconnect
            if (!link.gameConfig) {
                link.state = LinkState::Dead;
                continue;
            }

            const GameConfig* game = link.gameConfig;
            ganl::ErrorCode err = 0;
            ganl::ConnectionHandle bdHandle;
            reconnectAttemptsTotal_.fetch_add(1);

            if (game->transport == GameTransport::Unix) {
                bdHandle = engine_.initiateUnixConnect(
                    game->socketPath, nullptr, err);
            } else {
                bdHandle = engine_.initiateConnect(
                    game->host, game->port, nullptr, err);
            }

            if (bdHandle == ganl::InvalidConnectionHandle) {
                reconnectFailuresTotal_.fetch_add(1);
                // Schedule next retry or give up
                const auto& schedule = game->retrySchedule;
                if (link.retryCount < static_cast<int>(schedule.size())) {
                    int delay = schedule[static_cast<size_t>(link.retryCount)];
                    link.nextRetry = now + delay;
                    link.retryCount++;

                    LOG_INFO("Session %lu: link %zu reconnect failed, retry in %ds",
                             (unsigned long)sid, i + 1, delay);
                } else {
                    link.state = LinkState::Dead;
                    LOG_INFO("Session %lu: link %zu reconnect exhausted",
                             (unsigned long)sid, i + 1);

                    for (auto h : session.frontDoors) {
                        sendToClient(h,
                            "[" + link.gameName + " (link "
                            + std::to_string(i + 1)
                            + "): reconnect failed, giving up]\r\n");
                    }
                }
                continue;
            }

            link.handle = bdHandle;
            link.state = LinkState::Connecting;
            backDoorMap_[bdHandle] = {sid, i};

            LOG_INFO("Session %lu: link %zu reconnecting to %s",
                     (unsigned long)sid, i + 1, link.gameName.c_str());

            for (auto h : session.frontDoors) {
                sendToClient(h,
                    "[" + link.gameName + " (link "
                    + std::to_string(i + 1) + "): reconnecting...]\r\n");
            }
        }
    }

    // Prune stale IP tracker entries every 5 minutes
    if (now - lastIpPrune_ >= 300) {
        lastIpPrune_ = now;
        time_t rateWindow = now - 3600;  // 1-hour window for account creation
        for (auto it = ipTrackers_.begin(); it != ipTrackers_.end(); ) {
            auto& tracker = it->second;

            // Prune expired account-creation timestamps first
            tracker.accountCreateTimes.erase(
                std::remove_if(tracker.accountCreateTimes.begin(),
                               tracker.accountCreateTimes.end(),
                               [rateWindow](time_t t) { return t < rateWindow; }),
                tracker.accountCreateTimes.end());

            // Only erase the entry if truly idle: no connections, no lockout,
            // and no recent account creations within the rate window.
            if (tracker.connectionCount <= 0 &&
                tracker.lockoutUntil <= now &&
                tracker.accountCreateTimes.empty()) {
                it = ipTrackers_.erase(it);
            } else {
                ++it;
            }
        }
    }

    // Reap idle and detached sessions
    std::vector<HydraSessionId> toReap;
    for (auto& [sid, session] : sessions_) {
        int timeout = 0;
        if (session.state == SessionState::Detached) {
            timeout = config_.detachedSessionTimeout;
        } else if (session.frontDoors.empty() &&
                   !session.outputQueue->hasSubscribers()) {
            // No front-doors and no gRPC subscribers — effectively detached
            timeout = config_.detachedSessionTimeout;
        } else {
            timeout = config_.sessionIdleTimeout;
        }

        if (timeout > 0 && (now - session.lastActivity) >= timeout) {
            toReap.push_back(sid);
        }
    }
    for (auto sid : toReap) {
        auto sit = sessions_.find(sid);
        if (sit == sessions_.end()) continue;
        HydraSession& session = sit->second;

        LOG_INFO("Session %lu (%s) timed out, destroying",
                 (unsigned long)sid, session.persistId.c_str());

        // Close all links
        for (size_t i = 0; i < session.links.size(); i++) {
            closeLink(session, i);
        }
        // Flush and delete persisted state (use dbPersistId for FK)
        flushSession(session);
        const std::string& reapDelId = session.dbPersistId.empty()
            ? session.persistId : session.dbPersistId;
        if (!reapDelId.empty()) {
            accounts_.deleteSession(reapDelId);
        }
        // Close any remaining front-doors
        for (auto h : session.frontDoors) {
            sendToClient(h, "\r\n[Session timed out. Goodbye.]\r\n");
            engine_.closeConnection(h);
            frontDoors_.erase(h);
        }
        unindexSession(session);
        sessions_.erase(sid);
    }
}

// ---- Persistence helpers ----

std::string SessionManager::generatePersistId() {
    uint8_t buf[16];
    randomBytes(buf, sizeof(buf));
    static const char hex[] = "0123456789abcdef";
    std::string id;
    id.reserve(32);
    for (int i = 0; i < 16; i++) {
        id.push_back(hex[buf[i] >> 4]);
        id.push_back(hex[buf[i] & 0x0F]);
    }
    return id;
}

ScrollBack SessionManager::makeScrollback() const {
    // #1100: honor scrollback_lines from config (default 10000).  Clamp both
    // ends: 0 -> default, and anything above MAX_SCROLLBACK_LINES down to it,
    // since cap feeds ScrollBack's up-front buffer_.resize(cap) — a negative
    // scrollback_lines parses to SIZE_MAX and would OOM one vector per session.
    size_t cap = config_.scrollbackLines;
    if (cap == 0) {
        cap = 10000;
    }
    if (cap > MAX_SCROLLBACK_LINES) {
        LOG_WARN("scrollback_lines %zu exceeds max %zu; clamping",
                 cap, MAX_SCROLLBACK_LINES);
        cap = MAX_SCROLLBACK_LINES;
    }
    return ScrollBack(cap);
}

size_t SessionManager::countSessionsForAccount(uint32_t accountId) const {
    size_t n = 0;
    for (const auto& [id, sess] : sessions_) {
        (void)id;
        if (sess.accountId == accountId) {
            ++n;
        }
    }
    return n;
}

void SessionManager::flushSession(HydraSession& session) {
    // Use dbPersistId for SQLite operations — persistId may have been
    // rotated in-memory without updating the database row.
    const std::string& dbId = session.dbPersistId.empty()
        ? session.persistId : session.dbPersistId;
    if (dbId.empty()) return;

    std::string created = std::to_string(session.created);
    std::string lastActive = std::to_string(session.lastActivity);

    // Serialize links via nlohmann/json (handles escaping properly).
    // Dead links are compacted out — remap activeLink to match.
    nlohmann::json linksArr = nlohmann::json::array();
    size_t remappedActive = 0;
    size_t compactIdx = 0;
    for (size_t i = 0; i < session.links.size(); i++) {
        if (session.links[i].state == LinkState::Dead) continue;
        if (i == session.activeLink) remappedActive = compactIdx;
        linksArr.push_back({{"game", session.links[i].gameName},
                            {"character", session.links[i].character}});
        compactIdx++;
    }
    nlohmann::json linksDoc;
    linksDoc["activeLink"] = remappedActive;
    linksDoc["links"] = linksArr;
    std::string linksJson = linksDoc.dump();

    std::string errorMsg;
    if (!accounts_.saveSession(dbId, session.accountId,
                               created, lastActive, linksJson, errorMsg)) {
        LOG_ERROR("Failed to save session %s: %s",
                  dbId.c_str(), errorMsg.c_str());
        return;
    }

    // Flush scroll-back only if we have the encryption key
    if (!session.scrollbackKey.empty()) {
        int n = session.scrollback.flushToDb(
            accounts_.db(), dbId,
            session.accountId, session.scrollbackKey);
        if (n < 0) {
            LOG_ERROR("Scroll-back flush failed for session %s",
                      dbId.c_str());
        }
    }
}

void SessionManager::resumeSavedSession(FrontDoorState& fd,
                                        uint32_t accountId,
                                        const std::vector<uint8_t>& sbKey) {
    AccountManager::SavedSession saved;
    if (!accounts_.loadSession(accountId, saved)) return;

    HydraSession session;
    session.internalId = nextSessionId_++;
    session.accountId = accountId;
    session.username = fd.pendingUsername;
    // Preserve original timestamps from saved data
    session.created = static_cast<time_t>(std::atoll(saved.created.c_str()));
    session.lastActivity = static_cast<time_t>(std::atoll(saved.lastActive.c_str()));
    if (session.created == 0) session.created = time(nullptr);
    if (session.lastActivity == 0) session.lastActivity = time(nullptr);
    session.scrollbackKey = sbKey;
    session.persistId = saved.persistId;
    session.dbPersistId = session.persistId;
    session.tokenCreated = time(nullptr);
    session.scrollback = makeScrollback();
    session.frontDoors.push_back(fd.handle);

    int loaded = session.scrollback.loadFromDb(
        accounts_.db(), saved.persistId, accountId, sbKey);

    // Restore links from JSON
    std::vector<SavedLinkInfo> savedLinks;
    size_t savedActiveLink = 0;
    parseLinksJson(saved.linksJson, savedLinks, savedActiveLink);
    session.activeLink = savedActiveLink;

    fd.internalSessionId = session.internalId;
    HydraSessionId sessId = session.internalId;
    sessions_[sessId] = std::move(session);
    indexSession(sessions_[sessId]);
    HydraSession& sess = sessions_[fd.internalSessionId];

    // Reconnect saved links
    restoreSessionLinks(sess, savedLinks);

    LOG_INFO("Session %lu (%s) restored from SQLite for '%s' (%d lines, %zu links)",
             (unsigned long)fd.internalSessionId, saved.persistId.c_str(),
             fd.pendingUsername.c_str(), loaded > 0 ? loaded : 0,
             savedLinks.size());

    sendToClient(fd.handle, "\r\nRestoring saved session...\r\n");

    if (sess.scrollback.count() > 0) {
        sendToClient(fd.handle,
            "-- Scroll-back ("
            + std::to_string(sess.scrollback.count())
            + " lines) --\r\n");

        ReplayContext rctx{fd.handle, fd.encoding, fd.colorDepth, &bridge_, this};
        sess.scrollback.replay(
            sess.scrollback.count(),
            [](const std::string& text,
               const std::string& source,
               time_t timestamp, void* ctx) {
                auto* rc = static_cast<ReplayContext*>(ctx);
                std::string rendered = rc->bridge->renderForClient(
                    rc->encoding, rc->colorDepth, text);
                rendered += "\r\n";
                rc->sessionMgr->safeWrite(rc->handle, rendered);
            },
            &rctx);

        sendToClient(fd.handle, "-- End scroll-back --\r\n");
    }

    showGameMenu(sess, fd.handle);
}

void SessionManager::shutdownSessions() {
    LOG_INFO("Flushing all sessions before shutdown");
    for (auto& [sid, session] : sessions_) {
        // Notify connected clients
        for (auto h : session.frontDoors) {
            sendToClient(h, "\r\n[Hydra shutting down — reconnect to resume]\r\n");
        }
        // Flush session state and scroll-back to SQLite
        flushSession(session);
        // Wake any blocked gRPC subscribers so they can exit
        session.outputQueue->cv.notify_all();
    }
}

// ---- gRPC support ----

void SessionManager::indexSession(const HydraSession& session) {
    if (!session.persistId.empty()) {
        persistIdIndex_[session.persistId] = session.internalId;
    }
}

void SessionManager::unindexSession(const HydraSession& session) {
    if (!session.persistId.empty()) {
        persistIdIndex_.erase(session.persistId);
    }
}

HydraSession* SessionManager::findByPersistId(const std::string& persistId) {
    auto it = persistIdIndex_.find(persistId);
    if (it == persistIdIndex_.end()) return nullptr;

    auto sit = sessions_.find(it->second);
    if (sit == sessions_.end()) {
        // Stale index entry — clean up
        persistIdIndex_.erase(it);
        return nullptr;
    }

    HydraSession& session = sit->second;

    // Check token TTL
    if (config_.sessionTokenTtl > 0 && session.tokenCreated > 0) {
        time_t now = time(nullptr);
        if ((now - session.tokenCreated) >= config_.sessionTokenTtl) {
            LOG_INFO("Session token expired for '%s' (age %lds)",
                     session.username.c_str(),
                     (long)(now - session.tokenCreated));
            return nullptr;
        }
    }
    return &session;
}

std::string SessionManager::authenticateAndGetSession(
    const std::string& username, const std::string& password,
    const std::string& clientIp) {
    // #1097: apply the same failed-login lockout as the telnet front-door.
    if (!clientIp.empty() && isLockedOut(clientIp)) {
        authFailuresTotal_.fetch_add(1);
        return "";
    }

    std::vector<uint8_t> sbKey;
    uint32_t accountId = accounts_.authenticate(username, password, sbKey);
    if (accountId == 0) {
        authFailuresTotal_.fetch_add(1);
        if (!clientIp.empty()) {
            recordLoginFailure(clientIp);
        }
        return "";
    }
    if (!clientIp.empty()) {
        clearLoginFailures(clientIp);
    }

    // Check for existing in-memory session
    for (auto& [sid, sess] : sessions_) {
        if (sess.accountId == accountId) {
            sess.state = SessionState::Active;
            sess.lastActivity = time(nullptr);

            // Provide scrollback key and reconnect deferred links
            if (sess.scrollbackKey.empty() && !sbKey.empty()) {
                sess.scrollbackKey = sbKey;
                sess.username = username;
                const std::string& loadId = sess.dbPersistId.empty()
                    ? sess.persistId : sess.dbPersistId;
                sess.scrollback.loadFromDb(
                    accounts_.db(), loadId, sess.accountId, sbKey);
            }
            if (!sess.pendingLinksJson.empty()) {
                std::vector<SavedLinkInfo> savedLinks;
                size_t savedActiveLink = 0;
                parseLinksJson(sess.pendingLinksJson, savedLinks, savedActiveLink);
                sess.activeLink = savedActiveLink;
                restoreSessionLinks(sess, savedLinks);
                sess.pendingLinksJson.clear();
                LOG_INFO("Session %s: deferred links reconnected via gRPC auth",
                         sess.persistId.c_str());
            }

            // Rotate session token on re-authentication (in-memory only).
            // The SQLite saved_sessions row keeps the original persistId
            // for crash recovery — the user re-authenticates post-crash
            // and gets a fresh token then.  We do NOT delete+reinsert
            // because ON DELETE CASCADE would destroy scrollback rows.
            unindexSession(sess);
            sess.persistId = generatePersistId();
            sess.tokenCreated = time(nullptr);
            indexSession(sess);

            LOG_INFO("Session token rotated for '%s' (new=%s)",
                     username.c_str(), sess.persistId.c_str());
            return sess.persistId;
        }
    }

    // Check for saved session in SQLite (after restart)
    AccountManager::SavedSession saved;
    if (accounts_.loadSession(accountId, saved)) {
        HydraSession session;
        session.internalId = nextSessionId_++;
        session.accountId = accountId;
        session.username = username;
        session.created = time(nullptr);
        session.lastActivity = session.created;
        session.scrollbackKey = sbKey;
        session.persistId = saved.persistId;
        session.dbPersistId = session.persistId;
        session.tokenCreated = session.created;
        session.scrollback = makeScrollback();

        session.scrollback.loadFromDb(
            accounts_.db(), saved.persistId, accountId, sbKey);

        // Restore links from JSON
        std::vector<SavedLinkInfo> savedLinks;
        size_t savedActiveLink = 0;
        parseLinksJson(saved.linksJson, savedLinks, savedActiveLink);
        session.activeLink = savedActiveLink;

        HydraSessionId sessId = session.internalId;
        sessions_[sessId] = std::move(session);
        indexSession(sessions_[sessId]);
        restoreSessionLinks(sessions_[sessId], savedLinks);

        LOG_INFO("Session %lu (%s) restored via gRPC for '%s' (%zu links)",
                 (unsigned long)sessId, saved.persistId.c_str(),
                 username.c_str(), savedLinks.size());
        return sessions_[sessId].persistId;
    }

    // #1100: one session per account is STRUCTURAL.  The reuse-existing loop
    // and the restore-saved-session path above collapse any re-login for an
    // account onto its single session, so countSessionsForAccount(accountId)
    // is always 0 here.  This guard is therefore a defense-in-depth backstop
    // (fires only if a duplicate ever slips past reuse), NOT a tunable policy:
    // max_sessions_per_account > 1 cannot produce multiple concurrent sessions
    // under the current model — multiple game *connections* per account are
    // links within the one session, not separate sessions.  (Same invariant
    // holds on the telnet front-door path, which reuses identically.)
    if (config_.maxSessionsPerAccount > 0 &&
        countSessionsForAccount(accountId) >=
            static_cast<size_t>(config_.maxSessionsPerAccount)) {
        LOG_WARN("max_sessions_per_account reached for account %u", accountId);
        return "";
    }

    // New session
    HydraSession session;
    session.internalId = nextSessionId_++;
    session.accountId = accountId;
    session.username = username;
    session.created = time(nullptr);
    session.lastActivity = session.created;
    session.scrollbackKey = sbKey;
    session.persistId = generatePersistId();
    session.dbPersistId = session.persistId;
    session.tokenCreated = session.created;
    session.scrollback = makeScrollback();

    std::string pid = session.persistId;
    HydraSessionId sessId = session.internalId;
    sessions_[sessId] = std::move(session);
    indexSession(sessions_[sessId]);

    LOG_INFO("Session %lu (%s) created via gRPC for '%s'",
             (unsigned long)sessId, pid.c_str(),
             username.c_str());
    return pid;
}

std::string SessionManager::createAccountAndGetSession(
    const std::string& username, const std::string& password,
    const std::string& clientIp, std::string& errorOut) {
    // Rate-limit account creation per IP
    if (!clientIp.empty() && !checkAccountCreateRate(clientIp)) {
        errorOut = "too many accounts created from this address";
        return "";
    }

    bool admin = accounts_.isEmpty();
    uint32_t accountId = 0;
    if (!accounts_.createAccount(username, password, admin,
                                 accountId, errorOut)) {
        return "";
    }

    if (!clientIp.empty()) {
        recordAccountCreate(clientIp);
    }

    // Auto-login (pass clientIp so lockout bookkeeping stays consistent)
    return authenticateAndGetSession(username, password, clientIp);
}

// ---- OutputItem rendering ----

std::string HydraSession::OutputItem::render(RenderFormat fmt) const {
    if (endOfRecord) return "";
    if (puaText.empty()) return puaText;
    if (fmt == RenderFormat::PuaUtf8) return puaText;

    const unsigned char* src =
        reinterpret_cast<const unsigned char*>(puaText.data());
    size_t srcLen = puaText.size();

    // Truecolor SGR expansion can be ~4x; allocate generous heap buffer.
    size_t bufSize = srcLen * 4 + 256;
    std::vector<unsigned char> buf(bufSize);
    size_t n = 0;

    switch (fmt) {
    case RenderFormat::Unspecified:  // fall through to TrueColor default
    case RenderFormat::TrueColor:
        n = co_render_truecolor(buf.data(), src, srcLen, 0, 0);
        break;
    case RenderFormat::Ansi256:
        n = co_render_ansi256(buf.data(), src, srcLen, 0, 0);
        break;
    case RenderFormat::Ansi16:
        n = co_render_ansi16(buf.data(), src, srcLen, 0, 0);
        break;
    case RenderFormat::PuaUtf8:
        return puaText;  // unreachable, handled above
    case RenderFormat::Plain:
        n = co_strip_color(buf.data(), src, srcLen);
        break;
    }

    return std::string(reinterpret_cast<char*>(buf.data()), n);
}

// ---- gRPC-Web request handler ----

#ifdef GRPC_ENABLED
std::string SessionManager::renderPrometheusMetrics() {
    std::map<FrontDoorProto, size_t> frontDoorByProto;
    std::map<SessionState, size_t> sessionByState;
    std::map<LinkState, size_t> backDoorByState;

    size_t tlsFrontDoors = 0;
    size_t grpcWebSubscribers = 0;
    size_t lineBufferBytes = 0;
    size_t httpBufferBytes = 0;
    size_t tlsPlaintextBufferBytes = 0;
    size_t tlsCiphertextBufferBytes = 0;
    for (const auto& [handle, fd] : frontDoors_) {
        (void)handle;
        frontDoorByProto[fd.proto]++;
        if (fd.tlsTransport) tlsFrontDoors++;
        if (fd.grpcWebSubscribed) grpcWebSubscribers++;
        lineBufferBytes += fd.lineBuf.size();
        httpBufferBytes += fd.httpBuf.size();
        if (fd.tlsPlainOut) tlsPlaintextBufferBytes += fd.tlsPlainOut->readableBytes();
        if (fd.tlsEncryptedOut) tlsCiphertextBufferBytes += fd.tlsEncryptedOut->readableBytes();
    }

    size_t totalLinks = 0;
    size_t scrollbackLines = 0;
    size_t scrollbackDirtyLines = 0;
    size_t outputSubscribers = 0;
    size_t gmcpSubscribers = 0;
    for (auto& [sid, session] : sessions_) {
        (void)sid;
        sessionByState[session.state]++;
        totalLinks += session.links.size();
        scrollbackLines += session.scrollback.count();
        scrollbackDirtyLines += session.scrollback.dirtyCount();
        for (const auto& link : session.links) {
            backDoorByState[link.state]++;
        }

        if (session.outputQueue) {
            std::lock_guard<std::mutex> guard(session.outputQueue->mutex);
            for (const auto& [subId, sq] : session.outputQueue->subscribers) {
                (void)subId;
                outputSubscribers++;
                if (sq->wantsGmcp) gmcpSubscribers++;
            }
        }
    }

    size_t writeBufferBytes = 0;
    for (const auto& [handle, wb] : writeBuffers_) {
        (void)handle;
        writeBufferBytes += wb.remaining();
    }

    size_t ipLockedOut = 0;
    size_t ipActiveConnections = 0;
    size_t ipRateTrackedCreates = 0;
    time_t now = time(nullptr);
    for (const auto& [ip, tracker] : ipTrackers_) {
        (void)ip;
        if (tracker.lockoutUntil > now) ipLockedOut++;
        ipActiveConnections += static_cast<size_t>(std::max(tracker.connectionCount, 0));
        ipRateTrackedCreates += tracker.accountCreateTimes.size();
    }

    std::ostringstream out;
    out << "# HELP hydra_sessions Current Hydra sessions.\n";
    out << "# TYPE hydra_sessions gauge\n";
    out << "hydra_sessions " << sessions_.size() << "\n";
    out << "# HELP hydra_sessions_by_state Current Hydra sessions by session state.\n";
    out << "# TYPE hydra_sessions_by_state gauge\n";
    for (SessionState state : {SessionState::Login, SessionState::Active, SessionState::Detached}) {
        out << "hydra_sessions_by_state{state=\"" << sessionStateMetricName(state)
            << "\"} " << sessionByState[state] << "\n";
    }

    out << "# HELP hydra_front_doors Current connected front-door clients.\n";
    out << "# TYPE hydra_front_doors gauge\n";
    out << "hydra_front_doors " << frontDoors_.size() << "\n";
    out << "# HELP hydra_front_doors_by_proto Current front-door clients by protocol.\n";
    out << "# TYPE hydra_front_doors_by_proto gauge\n";
    for (FrontDoorProto proto : {FrontDoorProto::Telnet, FrontDoorProto::WebSocket,
                                 FrontDoorProto::GrpcWeb, FrontDoorProto::WsGameSession}) {
        out << "hydra_front_doors_by_proto{proto=\"" << frontDoorProtoName(proto)
            << "\"} " << frontDoorByProto[proto] << "\n";
    }
    out << "# HELP hydra_front_doors_tls Current TLS-enabled front-door clients.\n";
    out << "# TYPE hydra_front_doors_tls gauge\n";
    out << "hydra_front_doors_tls " << tlsFrontDoors << "\n";
    out << "# HELP hydra_grpc_web_subscribers Current grpc-web Subscribe streams.\n";
    out << "# TYPE hydra_grpc_web_subscribers gauge\n";
    out << "hydra_grpc_web_subscribers " << grpcWebSubscribers << "\n";

    out << "# HELP hydra_back_doors Current back-door game links.\n";
    out << "# TYPE hydra_back_doors gauge\n";
    out << "hydra_back_doors " << totalLinks << "\n";
    out << "# HELP hydra_back_doors_by_state Current back-door links by state.\n";
    out << "# TYPE hydra_back_doors_by_state gauge\n";
    for (LinkState state : {LinkState::Connecting, LinkState::TlsHandshaking,
                            LinkState::Negotiating, LinkState::AutoLoggingIn,
                            LinkState::Active, LinkState::Reconnecting,
                            LinkState::Suspended, LinkState::Dead}) {
        out << "hydra_back_doors_by_state{state=\"" << linkStateMetricName(state)
            << "\"} " << backDoorByState[state] << "\n";
    }

    out << "# HELP hydra_scrollback_bytes Current in-memory scrollback bytes.\n";
    out << "# TYPE hydra_scrollback_bytes gauge\n";
    out << "hydra_scrollback_bytes " << globalScrollbackBytes_.load() << "\n";
    out << "# HELP hydra_scrollback_lines Current in-memory scrollback lines.\n";
    out << "# TYPE hydra_scrollback_lines gauge\n";
    out << "hydra_scrollback_lines " << scrollbackLines << "\n";
    out << "# HELP hydra_scrollback_dirty_lines Current scrollback lines pending flush.\n";
    out << "# TYPE hydra_scrollback_dirty_lines gauge\n";
    out << "hydra_scrollback_dirty_lines " << scrollbackDirtyLines << "\n";

    out << "# HELP hydra_write_buffer_connections Connections with pending non-blocking writes.\n";
    out << "# TYPE hydra_write_buffer_connections gauge\n";
    out << "hydra_write_buffer_connections " << writeBuffers_.size() << "\n";
    out << "# HELP hydra_write_buffer_bytes Buffered bytes awaiting socket write.\n";
    out << "# TYPE hydra_write_buffer_bytes gauge\n";
    out << "hydra_write_buffer_bytes " << writeBufferBytes << "\n";
    out << "# HELP hydra_front_door_line_buffer_bytes Buffered telnet line assembly bytes.\n";
    out << "# TYPE hydra_front_door_line_buffer_bytes gauge\n";
    out << "hydra_front_door_line_buffer_bytes " << lineBufferBytes << "\n";
    out << "# HELP hydra_front_door_http_buffer_bytes Buffered grpc-web HTTP request bytes.\n";
    out << "# TYPE hydra_front_door_http_buffer_bytes gauge\n";
    out << "hydra_front_door_http_buffer_bytes " << httpBufferBytes << "\n";
    out << "# HELP hydra_tls_plaintext_buffer_bytes Buffered plaintext bytes waiting for TLS framing.\n";
    out << "# TYPE hydra_tls_plaintext_buffer_bytes gauge\n";
    out << "hydra_tls_plaintext_buffer_bytes " << tlsPlaintextBufferBytes << "\n";
    out << "# HELP hydra_tls_ciphertext_buffer_bytes Buffered TLS ciphertext bytes waiting for socket write.\n";
    out << "# TYPE hydra_tls_ciphertext_buffer_bytes gauge\n";
    out << "hydra_tls_ciphertext_buffer_bytes " << tlsCiphertextBufferBytes << "\n";

    out << "# HELP hydra_output_subscribers Current streaming output subscribers.\n";
    out << "# TYPE hydra_output_subscribers gauge\n";
    out << "hydra_output_subscribers " << outputSubscribers << "\n";
    out << "# HELP hydra_gmcp_subscribers Current GMCP subscribers.\n";
    out << "# TYPE hydra_gmcp_subscribers gauge\n";
    out << "hydra_gmcp_subscribers " << gmcpSubscribers << "\n";

    out << "# HELP hydra_ip_tracker_entries IP tracker entries retained for rate limiting.\n";
    out << "# TYPE hydra_ip_tracker_entries gauge\n";
    out << "hydra_ip_tracker_entries " << ipTrackers_.size() << "\n";
    out << "# HELP hydra_ip_locked_out Current IPs under failed-login lockout.\n";
    out << "# TYPE hydra_ip_locked_out gauge\n";
    out << "hydra_ip_locked_out " << ipLockedOut << "\n";
    out << "# HELP hydra_ip_active_connections Current total active IP-scoped connection count.\n";
    out << "# TYPE hydra_ip_active_connections gauge\n";
    out << "hydra_ip_active_connections " << ipActiveConnections << "\n";
    out << "# HELP hydra_account_create_window_events Account-creation timestamps currently retained for rate limiting.\n";
    out << "# TYPE hydra_account_create_window_events gauge\n";
    out << "hydra_account_create_window_events " << ipRateTrackedCreates << "\n";
    out << "# HELP hydra_auth_failures_total Total failed authentication attempts.\n";
    out << "# TYPE hydra_auth_failures_total counter\n";
    out << "hydra_auth_failures_total " << authFailuresTotal_.load() << "\n";
    out << "# HELP hydra_reconnect_attempts_total Total scheduled back-door reconnect attempts executed.\n";
    out << "# TYPE hydra_reconnect_attempts_total counter\n";
    out << "hydra_reconnect_attempts_total " << reconnectAttemptsTotal_.load() << "\n";
    out << "# HELP hydra_reconnect_failures_total Total back-door reconnect attempts that failed before a socket was established.\n";
    out << "# TYPE hydra_reconnect_failures_total counter\n";
    out << "hydra_reconnect_failures_total " << reconnectFailuresTotal_.load() << "\n";
    out << "# HELP hydra_backend_disconnects_total Total back-door disconnect events observed after a live socket existed.\n";
    out << "# TYPE hydra_backend_disconnects_total counter\n";
    out << "hydra_backend_disconnects_total " << backendDisconnectsTotal_.load() << "\n";
    out << "# HELP hydra_backend_connect_failures_total Total initial back-door connection failures before a live socket was established.\n";
    out << "# TYPE hydra_backend_connect_failures_total counter\n";
    out << "hydra_backend_connect_failures_total " << backendConnectFailuresTotal_.load() << "\n";

    return out.str();
}

// #1102: clientIp is "host:port" (IPv4) or "[host]:port" (IPv6) from
// NetworkAddress::toString(), or a bare host from the gRPC peer parse.  A
// naive `ip == "127.0.0.1"` never matches (the ":port" suffix), and IPv6
// "[::1]:port" matched nothing at all.  Extract the host and test loopback.
static bool isLoopbackClientIp(const std::string& clientIp) {
    std::string host = clientIp;
    if (!host.empty() && host.front() == '[') {
        // "[v6]:port" -> "v6"
        const auto rb = host.find(']');
        host = (rb == std::string::npos) ? host.substr(1)
                                         : host.substr(1, rb - 1);
    } else {
        // Strip a trailing ":port" only when there is exactly one colon; a
        // bare, unbracketed IPv6 literal has several colons and no port.
        const auto first = host.find(':');
        if (first != std::string::npos && first == host.rfind(':')) {
            host = host.substr(0, first);
        }
    }
    return host == "127.0.0.1" || host == "::1" || host == "localhost"
        || host.rfind("127.", 0) == 0                 // 127.0.0.0/8
        || host == "::ffff:127.0.0.1"
        || host.rfind("::ffff:127.", 0) == 0;         // v4-mapped loopback
}

void SessionManager::handleGrpcWebRequest(FrontDoorState& fd) {
    HttpRequest req;
    if (!parseHttpRequest(fd.httpBuf, req)) return;  // incomplete

    // Health check endpoint — responds to any method on /healthz
    if (req.path == "/healthz") {
        std::string resp = "HTTP/1.1 200 OK\r\n"
                           "Content-Type: text/plain\r\n"
                           "Content-Length: 2\r\n"
                           "\r\nok";
        safeWrite(fd.handle, resp);
        fd.httpBuf.clear();
        return;
    }

    if (req.path == "/metrics") {
        // #1102: metrics are recon-useful — only serve to loopback peers.
        if (!isLoopbackClientIp(fd.clientIp)) {
            std::string resp = "HTTP/1.1 403 Forbidden\r\n"
                               "Content-Type: text/plain\r\n"
                               "Content-Length: 10\r\n"
                               "\r\nForbidden\n";
            safeWrite(fd.handle, resp);
            fd.httpBuf.clear();
            return;
        }
        std::string body = renderPrometheusMetrics();
        std::string resp = "HTTP/1.1 200 OK\r\n"
                           "Content-Type: text/plain; version=0.0.4\r\n"
                           "Cache-Control: no-store\r\n"
                           "Content-Length: " + std::to_string(body.size()) + "\r\n"
                           "\r\n" + body;
        safeWrite(fd.handle, resp);
        fd.httpBuf.clear();
        return;
    }

    // Extract Origin header for CORS
    std::string requestOrigin;
    auto originIt = req.headers.find("origin");
    if (originIt != req.headers.end()) requestOrigin = originIt->second;

    // CORS preflight
    if (req.method == "OPTIONS") {
        std::string resp = corsPreflightResponse(requestOrigin,
                                                  config_.corsOrigins);
        safeWrite(fd.handle, resp);
        fd.httpBuf.clear();
        return;
    }

    if (req.method != "POST") {
        std::string resp = "HTTP/1.1 405 Method Not Allowed\r\n"
                           "Content-Length: 0\r\n\r\n";
        safeWrite(fd.handle, resp);
        fd.httpBuf.clear();
        return;
    }

    // Decode grpc-web body
    auto ctIt = req.headers.find("content-type");
    bool isText = ctIt != req.headers.end() && isGrpcWebTextContentType(ctIt->second);

    std::string protoBody;
    grpcWebDecodeRequest(req.body, isText, protoBody);

    // Extract auth token from headers
    std::string authToken;
    auto authIt = req.headers.find("authorization");
    if (authIt != req.headers.end()) authToken = authIt->second;

    // Route by path: /hydra.HydraService/MethodName
    std::string method;
    auto slash = req.path.rfind('/');
    if (slash != std::string::npos) {
        method = req.path.substr(slash + 1);
    }

    // Response content type matches request
    std::string respContentType = isText ? "application/grpc-web-text+proto"
                                         : "application/grpc-web+proto";

    // Helper to send a unary grpc-web response
    auto sendUnaryResponse = [&](const std::string& respProto, int status,
                                  const std::string& msg = "") {
        std::string body = grpcWebEncodeUnaryResponse(respProto, status, msg);
        if (isText) body = base64Encode(body);

        std::string http = "HTTP/1.1 200 OK\r\n"
            "Content-Type: " + respContentType + "\r\n"
            + corsHeaders(requestOrigin, config_.corsOrigins)
            + "Content-Length: " + std::to_string(body.size()) + "\r\n"
            "\r\n" + body;
        safeWrite(fd.handle, http);
    };

    // #1887: never treat a failed ParseFromString as a default-valued request.
    // INVALID_ARGUMENT is gRPC status code 3.
    auto parseRpc = [&](google::protobuf::MessageLite& rpcReq) -> bool {
        if (!rpcReq.ParseFromString(protoBody)) {
            sendUnaryResponse("", 3, "invalid request");
            return false;
        }
        return true;
    };

    // ---- Dispatch RPCs ----

    if (method == "ListGames") {
        // #1102: require a valid session token (auth header).
        if (authToken.empty() || !findByPersistId(authToken)) {
            sendUnaryResponse("", 16, "unauthenticated");  // UNAUTHENTICATED
        } else {
            hydra::GameList resp;
            for (const auto& game : config_.games) {
                auto* gi = resp.add_games();
                gi->set_name(game.name);
                gi->set_host(game.host);
                gi->set_port(game.port);
                gi->set_type(game.type == GameType::Local
                    ? hydra::GAME_LOCAL : hydra::GAME_REMOTE);
                gi->set_autostart(game.autostart);
            }
            sendUnaryResponse(resp.SerializeAsString(), 0);
        }

    } else if (method == "Authenticate") {
        hydra::AuthRequest rpcReq;
        if (parseRpc(rpcReq)) {
            // #1097: lockout applies on grpc-web too (peer IP on front-door).
            std::string pid = authenticateAndGetSession(
                rpcReq.username(), rpcReq.password(), fd.clientIp);

            hydra::AuthResponse resp;
            if (pid.empty()) {
                resp.set_success(false);
                resp.set_error("authentication failed");
            } else {
                resp.set_success(true);
                resp.set_session_id(pid);
            }
            sendUnaryResponse(resp.SerializeAsString(), 0);
        }

    } else if (method == "CreateAccount") {
        hydra::CreateAccountRequest rpcReq;
        if (parseRpc(rpcReq)) {
            std::string errorMsg;
            std::string pid = createAccountAndGetSession(
                rpcReq.username(), rpcReq.password(), fd.clientIp, errorMsg);

            hydra::CreateAccountResponse resp;
            if (pid.empty()) {
                resp.set_success(false);
                resp.set_error(errorMsg.empty() ? "failed" : errorMsg);
            } else {
                resp.set_success(true);
                resp.set_session_id(pid);
            }
            sendUnaryResponse(resp.SerializeAsString(), 0);
        }

    } else if (method == "GetSession") {
        hydra::SessionRequest rpcReq;
        if (parseRpc(rpcReq)) {
            std::string sid = authToken.empty() ? rpcReq.session_id() : authToken;

            HydraSession* s = findByPersistId(sid);
            if (!s) {
                sendUnaryResponse("", 5, "session not found");  // NOT_FOUND
            } else {
                hydra::SessionInfo resp;
                resp.set_session_id(s->persistId);
                resp.set_username(s->username);
                resp.set_active_link(s->links.empty() ? 0
                    : static_cast<int>(s->activeLink) + 1);
                resp.set_scrollback_lines(static_cast<int>(s->scrollback.count()));
                resp.set_state(s->state == SessionState::Active
                    ? hydra::SESSION_ACTIVE : hydra::SESSION_DETACHED);
                resp.set_created(static_cast<int64_t>(s->created));
                resp.set_last_activity(static_cast<int64_t>(s->lastActivity));
                sendUnaryResponse(resp.SerializeAsString(), 0);
            }
        }

    } else if (method == "SendInput") {
        hydra::InputRequest rpcReq;
        if (parseRpc(rpcReq)) {
            std::string sid = authToken.empty() ? rpcReq.session_id() : authToken;

            HydraSession* s = findByPersistId(sid);
            hydra::InputResponse resp;
            if (!s) {
                resp.set_error("session not found");
            } else if (rpcReq.line().size() > HydraSession::MAX_INPUT_LINE_LENGTH) {
                // #1268
                resp.set_error("line too long");
            } else {
                BackDoorLink* active = s->getActiveLink();
                if (active && active->state == LinkState::Active) {
                    // #1267: same charset conversion as native gRPC / WS GameSession.
                    std::string data = bridge_.convertInput(
                        ganl::EncodingType::Utf8,
                        active->protoState.encoding,
                        rpcReq.line());
                    data += "\r\n";
                    safeWrite(active->handle, data);
                    resp.set_success(true);
                } else {
                    resp.set_error("no active link");
                }
            }
            sendUnaryResponse(resp.SerializeAsString(), 0);
        }

    } else if (method == "Connect") {
        hydra::ConnectRequest rpcReq;
        if (parseRpc(rpcReq)) {
            std::string sid = authToken.empty() ? rpcReq.session_id() : authToken;

            HydraSession* s = findByPersistId(sid);
            hydra::ConnectResponse resp;
            if (!s) {
                resp.set_error("session not found");
            } else {
                size_t before = s->links.size();
                connectToGame(*s, rpcReq.game_name());
                if (s->links.size() > before) {
                    resp.set_success(true);
                    resp.set_link_number(static_cast<int>(s->links.size()));
                } else {
                    resp.set_error("connect failed");
                }
            }
            sendUnaryResponse(resp.SerializeAsString(), 0);
        }

    } else if (method == "ListLinks") {
        hydra::SessionRequest rpcReq;
        if (parseRpc(rpcReq)) {
            std::string sid = authToken.empty() ? rpcReq.session_id() : authToken;

            HydraSession* s = findByPersistId(sid);
            if (!s) {
                sendUnaryResponse("", 5, "session not found");
            } else {
                hydra::LinkList resp;
                for (size_t i = 0; i < s->links.size(); i++) {
                    auto* li = resp.add_links();
                    li->set_number(static_cast<int>(i) + 1);
                    li->set_game_name(s->links[i].gameName);
                    li->set_character(s->links[i].character);
                    li->set_active(i == s->activeLink);
                    li->set_gmcp_enabled(s->links[i].gmcpEnabled);
                }
                sendUnaryResponse(resp.SerializeAsString(), 0);
            }
        }

    } else if (method == "Ping") {
        hydra::SessionRequest rpcReq;
        if (parseRpc(rpcReq)) {
            std::string sid = authToken.empty() ? rpcReq.session_id() : authToken;

            HydraSession* s = findByPersistId(sid);
            if (s) s->lastActivity = time(nullptr);
            hydra::Empty resp;
            sendUnaryResponse(resp.SerializeAsString(), s ? 0 : 5);
        }

    } else if (method == "Subscribe") {
        // Server-streaming: register this connection as a live subscriber.
        // It stays open and receives chunked grpc-web data frames as
        // game output arrives in onBackDoorData().
        hydra::SessionRequest rpcReq;
        if (!parseRpc(rpcReq)) {
            // INVALID_ARGUMENT already sent
        } else {
            std::string sid = authToken.empty() ? rpcReq.session_id() : authToken;

            HydraSession* s = findByPersistId(sid);
            if (!s) {
                sendUnaryResponse("", 5, "session not found");
            } else {
                // Send HTTP headers for chunked streaming
                std::string httpHdr = "HTTP/1.1 200 OK\r\n"
                    "Content-Type: " + respContentType + "\r\n"
                    + corsHeaders(requestOrigin, config_.corsOrigins)
                    + "Transfer-Encoding: chunked\r\n"
                    "\r\n";
                safeWrite(fd.handle, httpHdr);

                // Mark this FrontDoorState as a live grpc-web subscriber.
                // onBackDoorData() will send output to it as chunked frames.
                fd.grpcWebSubscribed = true;
                fd.grpcWebSessionId = s->persistId;
                fd.grpcWebTextMode = isText;
                fd.internalSessionId = s->internalId;

                // Add to session's front-door list so it receives output
                s->frontDoors.push_back(fd.handle);
                s->state = SessionState::Active;

                // Forward terminal size to the active game link if provided
                if (rpcReq.terminal_width() > 0 || rpcReq.terminal_height() > 0) {
                    BackDoorLink* active = s->getActiveLink();
                    if (active && active->handle != ganl::InvalidConnectionHandle) {
                        uint16_t w = rpcReq.terminal_width() ? static_cast<uint16_t>(rpcReq.terminal_width()) : 80;
                        uint16_t h = rpcReq.terminal_height() ? static_cast<uint16_t>(rpcReq.terminal_height()) : 24;
                        safeWrite(active->handle, buildNawsFrame(w, h));
                    }
                }

                LOG_INFO("grpc-web Subscribe: fd %lu subscribed to session %s",
                         (unsigned long)fd.handle, s->persistId.c_str());

                // Don't clear httpBuf — we won't process further HTTP
                // requests on this fd (it's now a streaming connection)
                return;
            }
        }

    } else if (method == "GetScrollBack") {
        hydra::ScrollBackRequest rpcReq;
        if (parseRpc(rpcReq)) {
            std::string sid = authToken.empty() ? rpcReq.session_id() : authToken;

            HydraSession* s = findByPersistId(sid);
            if (!s) {
                sendUnaryResponse("", 5, "session not found");
            } else {
                hydra::ScrollBackResponse resp;
                size_t n = rpcReq.max_lines() > 0
                    ? static_cast<size_t>(rpcReq.max_lines())
                    : s->scrollback.count();
                struct Ctx { hydra::ScrollBackResponse* resp; };
                Ctx ctx{&resp};
                s->scrollback.replay(n,
                    [](const std::string& text, const std::string& source,
                       time_t timestamp, void* c) {
                        auto* rc = static_cast<Ctx*>(c);
                        auto* line = rc->resp->add_lines();
                        line->set_text(sanitizeProtoTextForLog(
                            text, "grpc-web scrollback", source, 0));
                        line->set_source(source);
                        line->set_timestamp(static_cast<int64_t>(timestamp));
                    },
                    &ctx);
                sendUnaryResponse(resp.SerializeAsString(), 0);
            }
        }

    } else if (method == "SwitchLink") {
        hydra::SwitchRequest rpcReq;
        if (parseRpc(rpcReq)) {
            std::string sid = authToken.empty() ? rpcReq.session_id() : authToken;

            HydraSession* s = findByPersistId(sid);
            hydra::SwitchResponse resp;
            if (!s) {
                resp.set_error("session not found");
            } else {
                size_t idx = static_cast<size_t>(rpcReq.link_number() - 1);
                if (idx >= s->links.size()) {
                    resp.set_error("invalid link number");
                } else {
                    s->activeLink = idx;
                    resp.set_success(true);
                }
            }
            sendUnaryResponse(resp.SerializeAsString(), 0);
        }

    } else if (method == "DisconnectLink") {
        hydra::DisconnectRequest rpcReq;
        if (parseRpc(rpcReq)) {
            std::string sid = authToken.empty() ? rpcReq.session_id() : authToken;

            HydraSession* s = findByPersistId(sid);
            hydra::DisconnectResponse resp;
            if (!s) {
                resp.set_error("session not found");
            } else {
                size_t idx = static_cast<size_t>(rpcReq.link_number() - 1);
                if (idx >= s->links.size()) {
                    resp.set_error("invalid link number");
                } else {
                    closeLink(*s, idx);
                    resp.set_success(true);
                }
            }
            sendUnaryResponse(resp.SerializeAsString(), 0);
        }

    } else {
        // Unimplemented method
        sendUnaryResponse("", 12, "method not found: " + method);
    }

    fd.httpBuf.clear();
}
#else
void SessionManager::handleGrpcWebRequest(FrontDoorState&) {}
void SessionManager::handleWsGameSessionData(FrontDoorState&, const char*, size_t) {}
void SessionManager::drainWsGameSessions() {}
#endif

// ---- Rate limiting ----

std::string SessionManager::getClientIp(ganl::ConnectionHandle) const {
    return "";  // Placeholder — IP is passed via onAccept now
}

bool SessionManager::checkConnectionLimits(const std::string& ip) {
    if (ip.empty()) return true;  // can't enforce without IP

    auto& tracker = ipTrackers_[ip];

    // Check per-IP connection limit
    if (tracker.connectionCount >= config_.maxConnectionsPerIp) {
        LOG_WARN("Rate limit: %s has %d connections (max %d)",
                 ip.c_str(), tracker.connectionCount, config_.maxConnectionsPerIp);
        return false;
    }

    // Check connection rate limit (per minute)
    time_t now = time(nullptr);
    time_t oneMinuteAgo = now - 60;

    // Prune old timestamps
    auto& times = tracker.connectTimes;
    times.erase(
        std::remove_if(times.begin(), times.end(),
            [oneMinuteAgo](time_t t) { return t < oneMinuteAgo; }),
        times.end());

    if (static_cast<int>(times.size()) >= config_.connectRateLimit) {
        LOG_WARN("Rate limit: %s exceeded %d connections/minute",
                 ip.c_str(), config_.connectRateLimit);
        return false;
    }

    return true;
}

void SessionManager::recordConnect(const std::string& ip) {
    if (ip.empty()) return;
    auto& tracker = ipTrackers_[ip];
    tracker.connectionCount++;
    tracker.connectTimes.push_back(time(nullptr));
}

void SessionManager::recordDisconnect(const std::string& ip) {
    if (ip.empty()) return;
    auto it = ipTrackers_.find(ip);
    if (it != ipTrackers_.end() && it->second.connectionCount > 0) {
        it->second.connectionCount--;
    }
}

void SessionManager::recordLoginFailure(const std::string& ip) {
    if (ip.empty()) return;
    auto& tracker = ipTrackers_[ip];
    tracker.failedLogins++;
    if (tracker.failedLogins >= config_.failedLoginLockout) {
        tracker.lockoutUntil = time(nullptr) + config_.failedLoginLockoutMinutes * 60;
        LOG_WARN("Rate limit: %s locked out for %d minutes after %d failed logins",
                 ip.c_str(), config_.failedLoginLockoutMinutes, tracker.failedLogins);
    }
}

void SessionManager::clearLoginFailures(const std::string& ip) {
    if (ip.empty()) return;
    auto it = ipTrackers_.find(ip);
    if (it != ipTrackers_.end()) {
        it->second.failedLogins = 0;
        it->second.lockoutUntil = 0;
    }
}

bool SessionManager::isLockedOut(const std::string& ip) {
    if (ip.empty()) return false;
    auto it = ipTrackers_.find(ip);
    if (it == ipTrackers_.end()) return false;
    if (it->second.lockoutUntil == 0) return false;
    if (time(nullptr) >= it->second.lockoutUntil) {
        // Lockout expired
        it->second.lockoutUntil = 0;
        it->second.failedLogins = 0;
        return false;
    }
    return true;
}

bool SessionManager::checkAccountCreateRate(const std::string& ip) {
    if (ip.empty()) return true;
    auto it = ipTrackers_.find(ip);
    if (it == ipTrackers_.end()) return true;

    // Allow max 2 account creations per hour per IP
    time_t now = time(nullptr);
    time_t window = now - 3600;
    auto& times = it->second.accountCreateTimes;

    // Prune old entries
    times.erase(std::remove_if(times.begin(), times.end(),
        [window](time_t t) { return t < window; }),
        times.end());

    return times.size() < 2;
}

void SessionManager::recordAccountCreate(const std::string& ip) {
    if (ip.empty()) return;
    ipTrackers_[ip].accountCreateTimes.push_back(time(nullptr));
}

// ---- Phase 4: Session serialization ----

bool SessionManager::parseLinksJson(const std::string& jsonStr,
                                    std::vector<SavedLinkInfo>& links,
                                    size_t& activeLink) {
    links.clear();
    activeLink = 0;

    if (jsonStr.empty()) return false;

    try {
        auto doc = nlohmann::json::parse(jsonStr);

        if (doc.contains("activeLink") && doc["activeLink"].is_number()) {
            activeLink = doc["activeLink"].get<size_t>();
        }

        if (doc.contains("links") && doc["links"].is_array()) {
            for (const auto& entry : doc["links"]) {
                SavedLinkInfo li;
                if (entry.contains("game") && entry["game"].is_string()) {
                    li.game = entry["game"].get<std::string>();
                }
                if (entry.contains("character") && entry["character"].is_string()) {
                    li.character = entry["character"].get<std::string>();
                }
                if (!li.game.empty()) {
                    links.push_back(std::move(li));
                }
            }
        }
    } catch (const nlohmann::json::exception& e) {
        LOG_WARN("Failed to parse links_json: %s", e.what());
        return false;
    }

    return !links.empty();
}

void SessionManager::restoreSessionLinks(HydraSession& session,
                                         const std::vector<SavedLinkInfo>& savedLinks) {
    for (const auto& sl : savedLinks) {
        // Resolve game config by name
        const GameConfig* game = nullptr;
        for (const auto& g : config_.games) {
            if (g.name == sl.game) { game = &g; break; }
        }
        if (!game) {
            LOG_WARN("Session %s: saved link for unknown game '%s', skipping",
                     session.persistId.c_str(), sl.game.c_str());
            continue;
        }

        // Track link count to verify connectToGame actually added one
        size_t before = session.links.size();
        connectToGame(session, sl.game);

        if (session.links.size() > before) {
            session.links.back().character = sl.character;
        } else {
            LOG_WARN("Session %s: connectToGame('%s') did not create a link",
                     session.persistId.c_str(), sl.game.c_str());
        }
    }

    // Restore activeLink index (clamp to valid range)
    if (session.activeLink >= session.links.size() && !session.links.empty()) {
        session.activeLink = 0;
    }
}

void SessionManager::restoreAllSessions() {
    auto saved = accounts_.loadAllSessions();
    if (saved.empty()) {
        LOG_INFO("No saved sessions to restore");
        return;
    }

    LOG_INFO("Restoring %zu saved sessions", saved.size());

    for (const auto& s : saved) {
        // Deduplication: skip if this account already has an in-memory session
        bool alreadyLoaded = false;
        for (const auto& [id, sess] : sessions_) {
            if (sess.accountId == s.accountId) {
                alreadyLoaded = true;
                break;
            }
        }
        if (alreadyLoaded) {
            LOG_INFO("Session %s (account %u) already in memory, skipping restore",
                     s.persistId.c_str(), s.accountId);
            continue;
        }

        // Create in-memory session (detached — no front-door yet)
        HydraSession session;
        session.internalId = nextSessionId_++;
        session.accountId = s.accountId;
        // Preserve original timestamps from saved data
        session.created = static_cast<time_t>(std::atoll(s.created.c_str()));
        session.lastActivity = static_cast<time_t>(std::atoll(s.lastActive.c_str()));
        if (session.created == 0) session.created = time(nullptr);
        if (session.lastActivity == 0) session.lastActivity = session.created;
        session.persistId = s.persistId;
        session.dbPersistId = s.persistId;
        session.state = SessionState::Detached;
        session.scrollback = makeScrollback();
        // scrollbackKey is empty — can't decrypt scroll-back until player logs in.
        // Defer link reconnection: back-door links will reconnect when the
        // player authenticates, providing the scrollback key.  This prevents
        // accumulating unflushable output if Hydra crashes again.
        session.pendingLinksJson = s.linksJson;

        HydraSessionId sessId = session.internalId;
        sessions_[sessId] = std::move(session);
        indexSession(sessions_[sessId]);

        LOG_INFO("Restored session %s (account %u, links deferred until login)",
                 s.persistId.c_str(), s.accountId);
    }
}

void SessionManager::dumpStatus() const {
    LOG_INFO("=== Hydra Status Dump ===");
    LOG_INFO("Sessions: %zu  Front-doors: %zu  Back-doors: %zu",
             sessions_.size(), frontDoors_.size(), backDoorMap_.size());
    LOG_INFO("Global scrollback memory: %zu bytes",
             globalScrollbackBytes_.load());

    for (const auto& kv : sessions_) {
        const HydraSession& s = kv.second;
        const char* stateStr = "?";
        switch (s.state) {
            case SessionState::Login:    stateStr = "Login";    break;
            case SessionState::Active:   stateStr = "Active";   break;
            case SessionState::Detached: stateStr = "Detached"; break;
        }

        LOG_INFO("  Session %s [%s] user=%s fd=%zu links=%zu "
                 "scrollback=%zu/%zu dirty=%zu",
                 s.persistId.c_str(), stateStr, s.username.c_str(),
                 s.frontDoors.size(), s.links.size(),
                 s.scrollback.count(), s.scrollback.memoryBytes(),
                 s.scrollback.dirtyCount());

        for (size_t i = 0; i < s.links.size(); i++) {
            const BackDoorLink& link = s.links[i];
            const char* linkStr = "?";
            switch (link.state) {
                case LinkState::Connecting:     linkStr = "Connecting";     break;
                case LinkState::TlsHandshaking: linkStr = "TlsHandshake";  break;
                case LinkState::Negotiating:    linkStr = "Negotiating";    break;
                case LinkState::AutoLoggingIn:  linkStr = "AutoLogin";      break;
                case LinkState::Active:         linkStr = "Active";         break;
                case LinkState::Reconnecting:   linkStr = "Reconnecting";   break;
                case LinkState::Suspended:      linkStr = "Suspended";      break;
                case LinkState::Dead:           linkStr = "Dead";           break;
            }
            LOG_INFO("    Link %zu: game=%s char=%s [%s]%s",
                     i + 1, link.gameName.c_str(), link.character.c_str(),
                     linkStr, (i == s.activeLink) ? " *active*" : "");
        }
    }
    LOG_INFO("=== End Status Dump ===");
}