// Staged WebSocket operations over the Stage1 owned transport buffers.
bool vScript::websocketCommand(Frame& frame,const CodeLine& c) {
    const auto& a=c.arg; const auto& s=c.source; const auto& sub=a[1].text;
    auto val=[&](std::size_t i) { return value(frame,a[i],s); };
    if(!options.allowNetwork) fail(s,"websocket operations require --allow-network");
    if(++socketCalls>4096) failLimit(s,"socket call limit (4096)");
    bool stepping=sub=="step";
    if(sub!="abort") {
        auto& out=lookup(frame,a[2],s); if(out.array) fail(s,"websocket metadata destination must be scalar or JSON");
        if(stepping) { auto& data=lookup(frame,a[3],s); if(data.array||data.structured) fail(s,"websocket data destination must be scalar"); if(a[2].text==a[3].text) fail(s,"websocket destinations must differ"); }
        std::size_t total=frame.extraBytes+65536+32768;
        for(const auto& item:frame.vars) total+=variableSize(item.second)+item.first.capacity();
        if(total>frameLimit) failLimit(s,"websocket frame reservation limit");
    }
    auto key=val(sub=="abort" ? 2 : stepping ? 4 : 3); auto it=sockets.find(key);
    if(it==sockets.end()) fail(s,"invalid or closed socket handle");
    auto p=it->second;
    if(p->listener||p->datagram) fail(s,"websocket requires TCP connection");
    if(p->handshaking && SocketClock::now()>=p->handshakeDeadline) {
        sockets.erase(key); fail(s,"socket TLS handshake timeout");
    }
    auto bytesRoom=[&](std::size_t n) { if(n>16*valueLimit-socketBytes) failLimit(s,"socket cumulative byte limit (16 MiB)"); };
    auto storage=[&](std::size_t extra) {
        std::size_t n=0; for(const auto& e:sockets) n+=e.second->buffer.capacity()+e.second->pendingRead.capacity()+e.second->pendingWrite.capacity()+(e.second->ws ? e.second->ws->storage() : 0);
        if(n+extra>valueLimit) failLimit(s,"socket buffer storage limit (1 MiB)");
    };
    auto osError=[&](const char* operation) {
        int error=errno;
        if(error==ENOMEM || error==ENOBUFS || error==EMFILE || error==ENFILE) {
            for(auto& entry:sockets) { if(entry.second->fd>=0) ::close(entry.second->fd); entry.second->fd=-1; }
            sockets.clear(); failLimit(s,"websocket OS resource limit");
        }
        fail(s,std::string("websocket transport ")+operation+" failed: "+std::strerror(error));
    };
    auto reserveTime=[&](std::size_t n) { if(n>60000-socketTime) failLimit(s,"socket cumulative timeout reservation limit (60000 ms)"); socketTime+=n; };
    if(sub=="abort") { if(!p->ws) fail(s,"websocket handle required"); sockets.erase(key); return true; }
    std::size_t accepted=0,countWire=0,completed=0; std::string data,type,status="progress";
    auto result=[&]() {
        auto& w=*p->ws;
        bool pending=!p->pendingWrite.empty()||!w.queue.empty();
        std::string want;
        if(status!="closed") want=p->tls && !p->pendingRead.empty() && p->readWant==POLLOUT ? "w" : pending ? (p->want==POLLIN && !p->pendingWrite.empty() ? "r" : "w") : "r";
        Json out={{"status",status},{"want",want},{"accepted",accepted},{"count",countWire},{"completed",completed},{"pending",pending},{"type",type},{"code",w.closeCode}};
        if(stepping) assign(frame,a[3],data,s);
        putJson(frame,a[2],std::move(out),s);
    };
    if(sub=="client"||sub=="server") {
        if(p->ws) fail(s,"websocket already claimed");
        if(p->handshaking || !p->buffer.empty() || !p->pendingRead.empty() || !p->pendingWrite.empty() || p->eof) fail(s,"websocket requires idle established stream");
        if(sub=="server" && !options.allowListen) fail(s,"websocket server requires --allow-listen");
        auto config=jsonValue(frame,a[4],s);
        if(!config.is_object()||config.size()!=4) fail(s,"websocket config requires host/path/origin/protocol");
        for(const auto* name:{"host","path","origin","protocol"}) if(!config.contains(name)||!config[name].is_string()) fail(s,"websocket config fields must be strings");
        auto w=std::make_unique<SocketWebSocket>(); w->client=sub=="client";
        w->host=config["host"].get<std::string>(); w->path=config["path"].get<std::string>(); w->origin=config["origin"].get<std::string>(); w->protocol=config["protocol"].get<std::string>();
        for(const auto* field:{&w->host,&w->path,&w->origin,&w->protocol}) wsSafeField(*field,s);
        if(w->host.empty()||w->host.find_first_of("/\\@?#")!=std::string::npos || w->path.empty()||w->path[0]!='/'||w->path.find('#')!=std::string::npos || (!w->protocol.empty()&&!wsToken(w->protocol))) fail(s,"websocket invalid host/path/protocol");
        auto ms=static_cast<std::size_t>(integer(val(5),1,30000,s));
        storage(32768); bytesRoom(8192);
        if(w->client) {
            unsigned char random[16]; if(RAND_bytes(random,16)!=1) fail(s,"websocket random key failed");
            auto nonce=wsBase64(random,16); w->accept=wsAccept(nonce,s);
            SocketWebSocket::Output request; request.wire="GET "+w->path+" HTTP/1.1\r\nHost: "+w->host+"\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Version: 13\r\nSec-WebSocket-Key: "+nonce+"\r\n";
            if(!w->origin.empty()) request.wire+="Origin: "+w->origin+"\r\n";
            if(!w->protocol.empty()) request.wire+="Sec-WebSocket-Protocol: "+w->protocol+"\r\n";
            request.wire+="\r\n"; socketBytes+=request.wire.size(); w->queue.push_back(std::move(request)); w->phase=SocketWebSocket::sending;
        }
        reserveTime(ms+1000); w->deadline=SocketClock::now()+std::chrono::milliseconds(ms);
        p->ws=std::move(w); status="handshake"; result(); return true;
    }
    if(!p->ws) fail(s,"websocket handle required");
    auto& w=*p->ws;
    if(w.timed()&&SocketClock::now()>=w.deadline) { sockets.erase(key); fail(s,"websocket deadline expired"); }
    if(sub=="send"||sub=="close") {
        if(w.phase!=SocketWebSocket::open) fail(s,"websocket is not open");
        std::string payload; unsigned opcode;
        if(sub=="send") {
            auto kind=val(4); payload=val(5);
            if(kind!="text"&&kind!="binary"&&kind!="ping") fail(s,"websocket invalid send type");
            opcode=kind=="text" ? 1U : kind=="binary" ? 2U : 9U;
            if(payload.size()>(opcode==9 ? 125U : 65536U)) fail(s,"websocket send size limit");
            if(opcode==1&&!wsUTF8(payload)) fail(s,"websocket invalid UTF8 text");
            if(opcode!=9&&w.applicationPending) fail(s,"websocket application output pending");
        } else {
            auto code=static_cast<unsigned>(integer(val(4),1000,4999,s)); auto reason=val(5);
            if(!wsCode(code)||reason.size()>123||!wsUTF8(reason)) fail(s,"websocket invalid close code/reason");
            payload.push_back(static_cast<char>(code>>8)); payload.push_back(static_cast<char>(code)); payload+=reason; opcode=8;
        }
        if(w.queue.size()>=8) fail(s,"websocket output queue limit");
        storage(2*(payload.size()+64)+4096); bytesRoom(payload.size()+14);
        std::size_t ms=sub=="close" ? static_cast<std::size_t>(integer(val(6),1,30000,s)) : 0;
        if(ms>60000-socketTime) failLimit(s,"socket cumulative timeout reservation limit (60000 ms)");
        w.enqueue(opcode,payload,s); socketBytes+=w.queue.back().wire.size(); accepted=payload.size();
        if(opcode==1||opcode==2) w.applicationPending=true;
        if(sub=="close") { reserveTime(ms); w.phase=SocketWebSocket::closing; w.deadline=SocketClock::now()+std::chrono::milliseconds(ms); status="closing"; }
        result(); return true;
    }
    // Reserve temporary copies (frame decoding, reassembly, result and control).
    storage(2*(w.input.size()+w.message.size()+8192)+16384); bytesRoom(16384);
    try {
        bool readExclusive=p->tls&&!p->pendingRead.empty()&&p->readWant==POLLOUT;
        if(!readExclusive && (!p->pendingWrite.empty()||!w.queue.empty())) {
            if(p->pendingWrite.empty()) {
                auto& q=w.queue.front(); p->pendingWrite.swap(q.wire); w.activePayload=q.payload; w.activeOpcode=q.opcode; w.queue.erase(w.queue.begin());
            }
            auto n=p->tls ? p->tls->step(2,p->pendingWrite.data(),static_cast<int>(p->pendingWrite.size()),p->want,s) : ::send(p->fd,p->pendingWrite.data(),p->pendingWrite.size(),MSG_NOSIGNAL);
            if(n<0) { if(!p->tls&&!socketWouldBlock()) osError("send"); if(!p->tls) p->want=POLLOUT; status="wouldblock"; }
            else if(n==0) fail(s,"websocket transport write made no progress");
            else {
                countWire=static_cast<std::size_t>(n); p->pendingWrite=std::string(p->pendingWrite.data()+n,p->pendingWrite.size()-countWire);
                if(p->pendingWrite.empty()) {
                    completed=w.activePayload;
                    if(w.activeOpcode==1||w.activeOpcode==2) w.applicationPending=false;
                    if(w.activeOpcode==8) w.closeSent=true;
                    w.activePayload=0; w.activeOpcode=0; p->want=p->readWant;
                    if(w.phase==SocketWebSocket::sending) { w.phase=w.client ? SocketWebSocket::receiving : SocketWebSocket::open; status=w.client ? "handshake" : "open"; }
                }
            }
        } else {
            bool handled=false;
            auto parse=[&]() {
                if(w.phase==SocketWebSocket::receiving) { if(w.upgrade(s)) { status=w.phase==SocketWebSocket::open ? "open" : "handshake"; handled=true; if(!w.client) { socketBytes+=w.queue.back().wire.size(); } } }
                else if(w.phase==SocketWebSocket::open||w.phase==SocketWebSocket::closing) { status=w.parse(data,type,s); handled=!status.empty(); }
            };
            if(w.parseReady&&!readExclusive) parse();
            if(!handled) {
                bool attempt=true;
                if(p->tls&&p->pendingRead.empty()&&SSL_pending(p->tls->ssl)==0) {
                    pollfd probe{p->fd,POLLIN,0}; int rc=::poll(&probe,1,0); if(rc<0&&errno!=EINTR) osError("poll"); attempt=rc>0;
                }
                ssize_t n=-1;
                if(attempt) {
                    if(p->pendingRead.empty()) { p->pendingRead=std::string(8192,'\0'); p->readMax=8192; }
                    n=p->tls ? p->tls->step(1,p->pendingRead.data(),8192,p->want,s) : ::recv(p->fd,p->pendingRead.data(),8192,0);
                    if(p->tls) p->readWant=p->want;
                    if(n<0&&!p->tls&&!socketWouldBlock()) osError("receive");
                    if(n>0) {
                        socketBytes+=static_cast<std::size_t>(n);
                        if(w.input.size()+static_cast<std::size_t>(n)>73742) fail(s,"websocket input buffer limit");
                        std::string fresh=w.input; fresh.append(p->pendingRead.data(),static_cast<std::size_t>(n)); w.input.swap(fresh); w.parseReady=true;
                    }
                    if(n>=0||!p->tls) std::string().swap(p->pendingRead);
                }
                if(n==0) fail(s,"websocket truncated EOF before close exchange");
                if(n>0) parse();
                if(!handled) status=n<0 ? "wouldblock" : "progress";
            }
            if(status=="ping") { w.enqueue(10,data,s); socketBytes+=w.queue.back().wire.size(); }
            if(status=="closing" && w.closeReceived) {
                if(w.phase!=SocketWebSocket::closing) {
                    w.phase=SocketWebSocket::closing; w.deadline=SocketClock::now()+std::chrono::milliseconds(1000);
                    w.enqueue(8,data,s); socketBytes+=w.queue.back().wire.size();
                }
                data=data.empty() ? "" : data.substr(2);
            }
        }
        if(w.timed()&&SocketClock::now()>=w.deadline) fail(s,"websocket deadline expired");
        if(w.closeSent&&w.closeReceived) { status="closed"; sockets.erase(key); }
    } catch(...) { sockets.erase(key); throw; }
    result(); return true;
}