// Explicit remote SQL API. SQLite's db command is deliberately unchanged.
static void remoteSqlValidate(const CodeLine& c) {
    count(c,2,6); const auto& a=c.arg; const auto& s=c.source;
    if(a[1].text=="open") { count(c,5,5); destination(a[2],s); }
    else if(a[1].text=="query" || a[1].text=="exec") { count(c,6,6); destination(a[2],s); destination(a[5],s); }
    else if(a[1].text=="begin" || a[1].text=="commit" || a[1].text=="rollback" || a[1].text=="close") count(c,3,3);
    else fail(s,"unknown sql operation");
}
static Json remoteSqlProfile(const std::string& path,const std::string& name,const Source& s) {
    int fd=::open(path.c_str(),O_RDONLY|O_CLOEXEC|O_NOFOLLOW|O_NONBLOCK);
    if(fd<0) fail(s,"SQL profile file unavailable");
    struct Guard { int fd; ~Guard(){::close(fd);} } guard{fd};
    struct stat st{};
    if(fstat(fd,&st) || !S_ISREG(st.st_mode) || st.st_uid!=geteuid() || (st.st_mode&077)!=0 || st.st_size>65536)
        fail(s,"SQL profile file must be owned by current user, private (0600), regular, <=64 KiB");
    std::string data; char buf[4096];
    for(;;) { auto n=::read(fd,buf,sizeof(buf)); if(n<0 && errno==EINTR) continue; if(n<0) fail(s,"SQL profile read failed"); if(!n) break; data.append(buf,static_cast<std::size_t>(n)); if(data.size()>65536) failLimit(s,"SQL profile size limit"); }
    Json config;
    try { config=Json::parse(data); } catch(...) { fail(s,"invalid SQL profile JSON (details redacted)"); }
    if(!config.is_object() || !config.contains(name) || !config[name].is_object()) fail(s,"SQL profile not found");
    return config[name];
}
struct vScript::RemoteDatabase {
    MYSQL* mysql=nullptr;
    SQLHENV env=SQL_NULL_HENV;
    SQLHDBC dbc=SQL_NULL_HDBC;
    bool transaction=false;
    unsigned timeout=2;
    ~RemoteDatabase() {
        if(mysql) mysql_close(mysql);
        if(dbc!=SQL_NULL_HDBC) { if(transaction) SQLEndTran(SQL_HANDLE_DBC,dbc,SQL_ROLLBACK); SQLDisconnect(dbc); SQLFreeHandle(SQL_HANDLE_DBC,dbc); }
        if(env!=SQL_NULL_HENV) SQLFreeHandle(SQL_HANDLE_ENV,env);
    }
    static void od(SQLRETURN rc,SQLSMALLINT kind,SQLHANDLE handle,const Source& s) {
        if(rc==SQL_SUCCESS) return;
        // Do not expose driver messages: they may echo credentials, SQL or data.
        SQLCHAR state[6]{}; SQLINTEGER native=0; SQLCHAR message[2]{}; SQLSMALLINT size=0;
        SQLGetDiagRec(kind,handle,1,state,&native,message,2,&size);
        fail(s,"remote SQL ODBC error SQLSTATE="+std::string(reinterpret_cast<char*>(state),5)+" code="+std::to_string(native));
    }
    static std::string brace(const std::string& x) {
        std::string r="{"; for(char ch:x) { r+=ch; if(ch=='}') r+=ch; } return r+"}";
    }
    void connect(const std::string& driver,const Json& p,const Source& s) {
        auto text=[&](const char* key) { if(!p.contains(key) || !p[key].is_string()) fail(s,"SQL profile missing string field"); auto x=p[key].get<std::string>(); if(x.size()>4096 || x.find('\0')!=std::string::npos) fail(s,"invalid SQL profile field"); return x; };
        if(text("driver")!=driver) fail(s,"SQL profile driver mismatch");
        auto host=text("host"), user=text("user"), pass=text("password"), database=text("database");
        if(!p.contains("port") || !p["port"].is_number_unsigned()) fail(s,"SQL profile port must be positive integer");
        auto port=p["port"].get<unsigned long long>(); if(port<1 || port>65535) fail(s,"SQL profile port out of range");
        if(p.contains("timeout")) { if(!p["timeout"].is_number_unsigned() || p["timeout"].get<unsigned long long>()<1 || p["timeout"].get<unsigned long long>()>30) fail(s,"SQL timeout range is 1..30 seconds"); timeout=p["timeout"].get<unsigned>(); }
        std::string tls=p.contains("tls") ? text("tls") : "verify";
        if(tls!="verify" && tls!="insecure-loopback") fail(s,"SQL TLS mode must be verify or insecure-loopback");
        if(tls=="insecure-loopback" && host!="127.0.0.1" && host!="::1") fail(s,"insecure SQL permitted only on numeric loopback");
        if(driver=="mysql" || driver=="mariadb") {
            mysql=mysql_init(nullptr); if(!mysql) fail(s,"remote SQL allocation failed");
            auto opt=[&](enum mysql_option key,const void* v) { if(mysql_optionsv(mysql,key,v)) fail(s,"remote SQL option failed"); };
            opt(MYSQL_OPT_CONNECT_TIMEOUT,&timeout); opt(MYSQL_OPT_READ_TIMEOUT,&timeout); opt(MYSQL_OPT_WRITE_TIMEOUT,&timeout);
            unsigned int local=0; opt(MYSQL_OPT_LOCAL_INFILE,&local);
            my_bool reconnect=0; opt(MYSQL_OPT_RECONNECT,&reconnect);
            opt(MYSQL_SET_CHARSET_NAME,"utf8mb4");
            unsigned int protocol=MYSQL_PROTOCOL_TCP; opt(MYSQL_OPT_PROTOCOL,&protocol);
            std::string ca;
            if(tls=="verify") { ca=text("ca"); opt(MYSQL_OPT_SSL_CA,ca.c_str()); my_bool yes=1; opt(MYSQL_OPT_SSL_ENFORCE,&yes); opt(MYSQL_OPT_SSL_VERIFY_SERVER_CERT,&yes); }
            if(!mysql_real_connect(mysql,host.c_str(),user.c_str(),pass.c_str(),database.c_str(),static_cast<unsigned>(port),nullptr,0))
                fail(s,"remote SQL connect failed code="+std::to_string(mysql_errno(mysql)));
            if(tls=="verify" && !mysql_get_ssl_cipher(mysql)) fail(s,"remote SQL verified TLS required");
        } else {
            od(SQLAllocHandle(SQL_HANDLE_ENV,SQL_NULL_HANDLE,&env),SQL_HANDLE_ENV,env,s);
            od(SQLSetEnvAttr(env,SQL_ATTR_ODBC_VERSION,reinterpret_cast<SQLPOINTER>(SQL_OV_ODBC3),0),SQL_HANDLE_ENV,env,s);
            od(SQLAllocHandle(SQL_HANDLE_DBC,env,&dbc),SQL_HANDLE_ENV,env,s);
            od(SQLSetConnectAttr(dbc,SQL_ATTR_LOGIN_TIMEOUT,reinterpret_cast<SQLPOINTER>(static_cast<uintptr_t>(timeout)),0),SQL_HANDLE_DBC,dbc,s);
            auto server="tcp:"+host+","+std::to_string(port);
            std::string cs="Driver={ODBC Driver 18 for SQL Server};Server="+brace(server)+";UID="+brace(user)+";PWD="+brace(pass)+";Database="+brace(database)+";Encrypt=yes;TrustServerCertificate="+(tls=="verify" ? "no" : "yes")+";";
            auto rc=SQLDriverConnect(dbc,nullptr,reinterpret_cast<SQLCHAR*>(cs.data()),SQL_NTS,nullptr,0,nullptr,SQL_DRIVER_NOPROMPT);
            if(rc==SQL_SUCCESS_WITH_INFO) {
                // SQL Server reports database/language selection as connect info.
                for(SQLSMALLINT i=1;;++i) { SQLCHAR state[6]{}; SQLINTEGER native=0; SQLCHAR message[2]{}; SQLSMALLINT size=0;
                    auto diag=SQLGetDiagRec(SQL_HANDLE_DBC,dbc,i,state,&native,message,2,&size); if(diag==SQL_NO_DATA) break;
                    if(!SQL_SUCCEEDED(diag) || std::memcmp(state,"01000",5)!=0 || (native!=5701 && native!=5703)) od(rc,SQL_HANDLE_DBC,dbc,s);
                }
            } else od(rc,SQL_HANDLE_DBC,dbc,s);
        }
    }
    void control(const std::string& op,const Source& s) {
        if(op=="begin" ? transaction : !transaction) fail(s,"remote SQL transaction state mismatch");
        if(mysql) {
            const char* sql=op=="begin" ? "START TRANSACTION" : op=="commit" ? "COMMIT" : "ROLLBACK";
            if(mysql_real_query(mysql,sql,static_cast<unsigned long>(std::strlen(sql)))) fail(s,"remote SQL transaction failed code="+std::to_string(mysql_errno(mysql)));
        } else if(op=="begin") od(SQLSetConnectAttr(dbc,SQL_ATTR_AUTOCOMMIT,reinterpret_cast<SQLPOINTER>(SQL_AUTOCOMMIT_OFF),0),SQL_HANDLE_DBC,dbc,s);
        else { od(SQLEndTran(SQL_HANDLE_DBC,dbc,op=="commit" ? SQL_COMMIT : SQL_ROLLBACK),SQL_HANDLE_DBC,dbc,s); od(SQLSetConnectAttr(dbc,SQL_ATTR_AUTOCOMMIT,reinterpret_cast<SQLPOINTER>(SQL_AUTOCOMMIT_ON),0),SQL_HANDLE_DBC,dbc,s); }
        transaction=op=="begin";
    }
    struct Result {
        Json rows=Json::array(); long long changed=0; std::size_t bytes=128, nodes=1;
        void add(Json row,const Source& s) {
            if(rows.size()>=1024) failLimit(s,"remote SQL row limit (1024)");
            nodes+=row.size()+1; if(nodes>itemLimit) failLimit(s,"remote SQL node limit");
            bytes+=jsonSize(row,s)+sizeof(Json); if(bytes>valueLimit) failLimit(s,"remote SQL result byte limit (1 MiB)");
            rows.push_back(std::move(row));
        }
    };
    static void name(std::vector<std::string>& names,std::string n,const Source& s) {
        if(n.empty() || n.size()>256 || std::find(names.begin(),names.end(),n)!=names.end()) fail(s,"remote SQL column names must be unique, nonempty and <=256 bytes; use aliases");
        names.push_back(std::move(n));
    }
    static Json cell(const std::string& text,bool integral,bool real,const Source& s) {
        try {
            if(integral) return integer(text,-9007199254740991LL,9007199254740991LL,s);
            if(real) return number(text,s);
        } catch(const Error&) { fail(s,"remote SQL numeric result outside supported range"); }
        return text;
    }
    Result run(const std::string& sql,const Json& params,bool query,const Source& s) {
        if(sql.empty() || sql.size()>65536 || sql.find('\0')!=std::string::npos) fail(s,"remote SQL requires nonempty SQL <=64 KiB without NUL");
        if(!params.is_array() || params.size()>256) fail(s,"remote SQL parameters require array of <=256 scalars");
        for(const auto& p:params) if(!(p.is_null() || p.is_boolean() || p.is_number() || p.is_string())) fail(s,"remote SQL parameters must be scalar JSON");
        // Conservative lexical policy; not a SQL parser or a security sandbox.
        // Quoted semicolons are data. Comments/batches/control SQL are unsupported.
        std::string word; char quote=0;
        auto token=[&]() { if(word=="BEGIN" || word=="COMMIT" || word=="ROLLBACK" || word=="SAVEPOINT" || word=="TRANSACTION" || word=="AUTOCOMMIT" || word=="IMPLICIT_TRANSACTIONS" || word=="CALL" || word=="EXEC" || word=="EXECUTE") fail(s,"remote SQL transaction/control SQL unsupported; use sql commands"); word.clear(); };
        for(std::size_t i=0;i<sql.size();++i) { char ch=sql[i]; if(quote) { if(ch=='\\') fail(s,"remote SQL backslash quoted escapes unsupported; bind values"); if(ch==quote) { if(i+1<sql.size() && sql[i+1]==quote) ++i; else quote=0; } continue; }
            if(ch=='\'' || ch=='"' || ch=='`' || ch=='[') { token(); quote=ch=='[' ? ']' : ch; }
            else if(ch==';' || ch=='#' || (ch=='-' && i+1<sql.size() && sql[i+1]=='-') || (ch=='/' && i+1<sql.size() && sql[i+1]=='*')) fail(s,"remote SQL comments and semicolon batches unsupported");
            else if(std::isalnum(static_cast<unsigned char>(ch)) || ch=='_') word+=static_cast<char>(std::toupper(static_cast<unsigned char>(ch))); else token(); }
        token(); if(quote) fail(s,"remote SQL unterminated quoted token");
        return mysql ? runMysql(sql,params,query,s) : runOdbc(sql,params,query,s);
    }
    Result runMysql(const std::string& sql,const Json& params,bool query,const Source& s) {
        std::unique_ptr<MYSQL_STMT,decltype(&mysql_stmt_close)> stmt(mysql_stmt_init(mysql),mysql_stmt_close);
        if(!stmt) fail(s,"remote SQL statement allocation failed");
        auto raw=stmt.get();
        auto check=[&](int rc) { if(rc) fail(s,"remote SQL statement failed code="+std::to_string(mysql_stmt_errno(raw))); };
        check(mysql_stmt_prepare(raw,sql.data(),static_cast<unsigned long>(sql.size())));
        if(mysql_stmt_param_count(raw)!=params.size()) fail(s,"remote SQL parameter count mismatch");
        std::vector<MYSQL_BIND> binds(params.size()); std::vector<long long> ints(params.size()); std::vector<double> doubles(params.size()); std::vector<unsigned long> lengths(params.size());
        for(std::size_t i=0;i<params.size();++i) { const auto& p=params[i]; auto& b=binds[i];
            if(p.is_null()) b.buffer_type=MYSQL_TYPE_NULL;
            else if(p.is_boolean() || p.is_number_integer()) { ints[i]=p.is_boolean() ? (p.get<bool>()?1:0) : p.get<long long>(); b.buffer_type=MYSQL_TYPE_LONGLONG; b.buffer=&ints[i]; }
            else if(p.is_number()) { doubles[i]=p.get<double>(); b.buffer_type=MYSQL_TYPE_DOUBLE; b.buffer=&doubles[i]; }
            else { auto& str=p.get_ref<const std::string&>(); lengths[i]=static_cast<unsigned long>(str.size()); b.buffer_type=MYSQL_TYPE_STRING; b.buffer=const_cast<char*>(str.data()); b.buffer_length=lengths[i]; b.length=&lengths[i]; }
        }
        if(!binds.empty()) check(mysql_stmt_bind_param(raw,binds.data()));
        auto columns=mysql_stmt_field_count(raw); if(columns>128) failLimit(s,"remote SQL column limit (128)");
        if(query ? !columns : columns!=0) fail(s,"remote SQL query requires rows; exec cannot return rows");
        check(mysql_stmt_execute(raw)); Result result;
        if(!query) { auto n=mysql_stmt_affected_rows(raw); if(n>9007199254740991ULL) fail(s,"remote SQL affected rows out of safe range"); result.changed=static_cast<long long>(n); return result; }
        std::unique_ptr<MYSQL_RES,decltype(&mysql_free_result)> metadata(mysql_stmt_result_metadata(raw),mysql_free_result);
        if(!metadata) fail(s,"remote SQL result metadata failed");
        auto fields=mysql_fetch_fields(metadata.get());
        std::vector<std::string> names; std::vector<MYSQL_BIND> out(columns); std::vector<unsigned long> sizes(columns); std::vector<double> reals(columns); std::unique_ptr<my_bool[]> nulls(new my_bool[columns]{});
        for(unsigned i=0;i<columns;++i) { name(names,std::string(fields[i].name,fields[i].name_length),s); out[i].buffer_type=MYSQL_TYPE_STRING; out[i].length=&sizes[i]; out[i].is_null=&nulls[i];
            if(fields[i].type==MYSQL_TYPE_DOUBLE || fields[i].type==MYSQL_TYPE_FLOAT) { out[i].buffer_type=MYSQL_TYPE_DOUBLE; out[i].buffer=&reals[i]; out[i].buffer_length=sizeof(double); }
        }
        check(mysql_stmt_bind_result(raw,out.data()));
        for(;;) { auto rc=mysql_stmt_fetch(raw); if(rc==MYSQL_NO_DATA) break; if(rc!=0 && rc!=MYSQL_DATA_TRUNCATED) check(rc); Json row=Json::object();
            for(unsigned i=0;i<columns;++i) { if(nulls[i]) { row[names[i]]=nullptr; continue; } if(sizes[i]>65536) failLimit(s,"remote SQL cell limit (64 KiB)");
                auto type=fields[i].type;
                if(type==MYSQL_TYPE_DOUBLE || type==MYSQL_TYPE_FLOAT) { if(!std::isfinite(reals[i])) fail(s,"remote SQL nonfinite number"); row[names[i]]=reals[i]; continue; }
                if(fields[i].charsetnr==63 && (type==MYSQL_TYPE_BLOB || type==MYSQL_TYPE_LONG_BLOB || type==MYSQL_TYPE_MEDIUM_BLOB || type==MYSQL_TYPE_TINY_BLOB || type==MYSQL_TYPE_STRING || type==MYSQL_TYPE_VAR_STRING)) fail(s,"remote SQL binary results unsupported");
                std::string text(sizes[i],'\0'); MYSQL_BIND b{}; b.buffer_type=MYSQL_TYPE_STRING; b.buffer=text.data(); b.buffer_length=sizes[i]; check(mysql_stmt_fetch_column(raw,&b,i,0));
                bool integral=type==MYSQL_TYPE_TINY || type==MYSQL_TYPE_SHORT || type==MYSQL_TYPE_LONG || type==MYSQL_TYPE_LONGLONG || type==MYSQL_TYPE_INT24 || type==MYSQL_TYPE_YEAR;
                row[names[i]]=cell(text,integral,type==MYSQL_TYPE_FLOAT || type==MYSQL_TYPE_DOUBLE,s);
            } result.add(std::move(row),s);
        } return result;
    }
    Result runOdbc(const std::string& sql,const Json& params,bool query,const Source& s) {
        SQLHSTMT stmt=SQL_NULL_HSTMT; od(SQLAllocHandle(SQL_HANDLE_STMT,dbc,&stmt),SQL_HANDLE_DBC,dbc,s);
        struct Guard { SQLHSTMT h; ~Guard(){SQLFreeHandle(SQL_HANDLE_STMT,h);} } guard{stmt};
        auto check=[&](SQLRETURN rc) { od(rc,SQL_HANDLE_STMT,stmt,s); };
        check(SQLSetStmtAttr(stmt,SQL_ATTR_QUERY_TIMEOUT,reinterpret_cast<SQLPOINTER>(static_cast<uintptr_t>(timeout)),0));
        check(SQLPrepare(stmt,reinterpret_cast<SQLCHAR*>(const_cast<char*>(sql.data())),static_cast<SQLINTEGER>(sql.size())));
        SQLSMALLINT countParams=0; check(SQLNumParams(stmt,&countParams)); if(static_cast<std::size_t>(countParams)!=params.size()) fail(s,"remote SQL parameter count mismatch");
        std::vector<long long> ints(params.size()); std::vector<double> doubles(params.size()); std::vector<SQLLEN> lengths(params.size());
        for(std::size_t i=0;i<params.size();++i) { const auto& p=params[i]; SQLSMALLINT ctype=SQL_C_CHAR, type=SQL_VARCHAR; SQLPOINTER ptr=nullptr; SQLULEN size=1; SQLLEN capacity=0;
            if(p.is_null()) { lengths[i]=SQL_NULL_DATA; ptr=&ints[i]; }
            else if(p.is_boolean() || p.is_number_integer()) { ints[i]=p.is_boolean() ? (p.get<bool>()?1:0) : p.get<long long>(); ctype=SQL_C_SBIGINT; type=SQL_BIGINT; ptr=&ints[i]; size=19; capacity=sizeof(long long); }
            else if(p.is_number()) { doubles[i]=p.get<double>(); ctype=SQL_C_DOUBLE; type=SQL_DOUBLE; ptr=&doubles[i]; size=15; capacity=sizeof(double); }
            else { const auto& str=p.get_ref<const std::string&>(); type=SQL_WVARCHAR; ptr=const_cast<char*>(str.data()); size=std::max<std::size_t>(1,str.size()); lengths[i]=static_cast<SQLLEN>(str.size()); capacity=lengths[i]; }
            check(SQLBindParameter(stmt,static_cast<SQLUSMALLINT>(i+1),SQL_PARAM_INPUT,ctype,type,size,0,ptr,capacity,&lengths[i]));
        }
        auto executed=SQLExecute(stmt); if(executed!=SQL_NO_DATA) check(executed);
        SQLSMALLINT columns=0; check(SQLNumResultCols(stmt,&columns));
        if(columns>128) failLimit(s,"remote SQL column limit (128)");
        if(query ? columns==0 : columns!=0) fail(s,"remote SQL query requires rows; exec cannot return rows");
        Result result;
        if(!query) { SQLLEN n=0; check(SQLRowCount(stmt,&n)); result.changed=static_cast<long long>(n); }
        else {
            std::vector<std::string> names; std::vector<SQLSMALLINT> types;
            for(SQLUSMALLINT i=1;i<=static_cast<SQLUSMALLINT>(columns);++i) { SQLCHAR buf[257]{}; SQLSMALLINT len=0,type=0,dec=0,nullable=0; SQLULEN size=0;
                check(SQLDescribeCol(stmt,i,buf,sizeof(buf),&len,&type,&size,&dec,&nullable)); if(len<0 || len>256) failLimit(s,"remote SQL column name limit"); name(names,std::string(reinterpret_cast<char*>(buf),static_cast<std::size_t>(len)),s); types.push_back(type); }
            for(;;) { auto rc=SQLFetch(stmt); if(rc==SQL_NO_DATA) break; check(rc); Json row=Json::object();
                for(SQLUSMALLINT i=1;i<=static_cast<SQLUSMALLINT>(columns);++i) { auto type=types[i-1]; if(type==SQL_BINARY || type==SQL_VARBINARY || type==SQL_LONGVARBINARY) fail(s,"remote SQL binary results unsupported");
                    std::vector<char> buf(65537); SQLLEN len=0; auto got=SQLGetData(stmt,i,SQL_C_CHAR,buf.data(),static_cast<SQLLEN>(buf.size()),&len);
                    if(len==SQL_NULL_DATA) { check(got); row[names[i-1]]=nullptr; continue; }
                    if(len==SQL_NO_TOTAL || len>65536) failLimit(s,"remote SQL cell limit (64 KiB)");
                    check(got); if(len<0) fail(s,"remote SQL invalid column length");
                    bool integral=type==SQL_BIT || type==SQL_TINYINT || type==SQL_SMALLINT || type==SQL_INTEGER || type==SQL_BIGINT;
                    row[names[i-1]]=cell(std::string(buf.data(),static_cast<std::size_t>(len)),integral,type==SQL_REAL || type==SQL_FLOAT || type==SQL_DOUBLE,s);
                } result.add(std::move(row),s);
            }
        }
        auto more=SQLMoreResults(stmt); if(more!=SQL_NO_DATA) { check(more); fail(s,"remote SQL multiple results unsupported; effects may have occurred"); }
        return result;
    }
};
bool vScript::remoteSql(Frame& frame,const CodeLine& c) {
    const auto& a=c.arg; if(a[0].text!="sql") return false; const auto& s=c.source; const auto& op=a[1].text;
    if(!options.allowDb || !options.allowNetwork) fail(s,"remote SQL requires --allow-db and --allow-network (not a sandbox)");
    auto val=[&](std::size_t i){return value(frame,a[i],s);};
    if(op=="open") {
        lookup(frame,a[2],s);
        if(options.sqlConfig.empty()) fail(s,"remote SQL requires operator --sql-config file");
        if(remoteDatabases.size()>=8) failLimit(s,"remote SQL handle limit (8)");
        auto driver=val(3); if(driver!="mysql" && driver!="mariadb" && driver!="mssql") fail(s,"remote SQL driver must be mysql, mariadb or mssql");
        auto db=std::make_shared<RemoteDatabase>(); db->connect(driver,remoteSqlProfile(options.sqlConfig,val(4),s),s);
        auto id="sql:"+std::to_string(nextRemoteDatabase++); assign(frame,a[2],id,s); remoteDatabases.emplace(id,std::move(db));
    } else {
        bool statement=op=="exec" || op=="query"; if(statement) lookup(frame,a[2],s);
        auto it=remoteDatabases.find(val(statement?3:2)); if(it==remoteDatabases.end()) fail(s,"invalid or closed remote SQL handle");
        if(op=="close") { remoteDatabases.erase(it); return true; }
        try {
            if(++remoteSqlCalls>1024) failLimit(s,"remote SQL call limit (1024)");
            if(statement) { auto r=it->second->run(val(4),jsonValue(frame,a[5],s),op=="query",s);
                if(op=="exec") assign(frame,a[2],std::to_string(r.changed),s);
                else { if(r.bytes>16*valueLimit-remoteSqlBytes) failLimit(s,"remote SQL cumulative result limit (16 MiB)"); remoteSqlBytes+=r.bytes; putJson(frame,a[2],std::move(r.rows),s); }
            } else it->second->control(op,s);
        } catch(...) { remoteDatabases.erase(it); throw; } // no reuse after ambiguous state
    }
    return true;
}