// ===== rpc:call 元协议纯核(rex 线形——rpc.erl:130 真源对位)=====
// 真件③收口(2026-11-07 真网门 VERDICT PASS——probea@dev 实测背书):
// rpc:call 线形(同包单头——真 OTP dist 一包,len|cachehdr|ctl|msg):
//   ctl 4 元(dist.c DOP_REG_SEND 臂:tuple_arity 必须为 4):
//     {6, FromPid, Cookie='', ToName}——R9B+ cookie 恒空 atom
//     (3 元=arity≠4→Invalid→erts_kill_dist_connection→FIN——probea
//      报告 {6,,rex} 即 3 元 ctl 本身的直证,茗篇层 2)
//   载荷 {'$gen_call', From, Request}——From=调用方 pid 结构(VPid 真形,
//     ALIVE2 注册 creation——编造 creation=shutdown 铁律坑账条 8)
//   Request 五元 {call, Mod, Fun, Args, Gleader}——Gleader=真 pid
//     (rpc.erl:130 set_group_leader(Gleader) pid 域;裸 atom=function_clause
//      EXIT 堆栈直证;'send_stdout_to_caller 自补'仅 handle_info 路径
//      rpc.erl:220 专属——$gen_call 路径不自补,茗篇层 6)
//   交集无段头位时传输面组帧='p'(PASS_THROUGH 0x70)+131+ctl+131+msg
//     (external.c:6420/919/500——PASS_THROUGH 前缀形,probea dump 铁证;
//      "payload 分离"系逻辑域分离非物理帧分离——词表注记勘误)
// 应答真形:rex 经 gen_server:reply → {Ref, Reply} 投递(SEND_SENDER 22
//   形,dist.h:155——{rex, Result} 老形勘误退役,probe dump 直证:
//   回流帧 msg={ref,'probea@dev'} 即 erlang:node() 回执)
// 依赖倒置:EtfCodec trait(moonbit-eetf 组合位——核心不 import eetf);
//   传输组装('p' 前缀+131 连排)住调用方/probe(dist_frame 形),核心保持
//   值构造纯核。

///|
pub(all) suberror RpcError {
  Codec(String) // EETF 编解码错(调用方 codec 抛出转译)
  Transport(String) // 收发错(RpcPort 抛出转译)
  BadReply(String) // ref 不匹配/形不符
}

///|
/// EETF 编解码注入位(moonbit-eetf 组合根实现——双实现安放律)
pub(open) trait EtfCodec {
  fn encode(self : Self, terms : Array[EtfVal]) -> Bytes raise RpcError
  fn decode(self : Self, data : Bytes) -> EtfVal raise RpcError
}

///|
/// EtfVal:rpc 面项值(eetf EtfTerm 的 rpc 面映射——组合根 eetf_codec 全谱)
pub(all) enum EtfVal {
  VAtom(String)
  VInt(Int32)
  VFloat(Double)
  VTuple(Array[EtfVal])
  VList(Array[EtfVal])
  VNil
  VBinary(Bytes)
  VString(String) // 107 latin1 串(整数字节列快捷形)
  VMap(Array[(EtfVal, EtfVal)]) // 116 键值对保序
  VPid(String, Int32, Int32, Int32) // node, id, serial, creation
  VPort(String, Int32, Int32) // node, id, creation
  VRef(String, Int32, Int32, Int32) // node, id, serial(骨架恒 0), creation
  VOther(String) // 未知直通占位(eetf 14 变体全谱落——encode 拒,真形候真帧)
}

///|
/// RpcPort(传输注入位):send 发请求字节;recv 收应答字节(阻塞/超时由实现定;
/// 线形组帧——同包单头 'p'+131+ctl+131+msg——住调用方/probe 的 dist_frame 形,
/// 核心保持值构造纯核)
pub(open) trait RpcPort {
  async fn send(self : Self, node : String, data : Bytes) -> Unit raise RpcError
  async fn recv(self : Self) -> Bytes raise RpcError
}

///|
/// group leader 处理形(rpc.erl 真源三路径勘定 2026-11-07——EXIT 堆栈直证):
///Pid 真形=标准形(rpc.erl:130 set_group_leader(Gleader) pid 域;
///   $gen_call/handle_call 路径**不自补**——裸 atom=function_clause);
///StdoutToCaller atom=仅 handle_info({From,{call,...}}) 隐藏 C 节点路径
///   自补形(rpc.erl:220 {send_stdout_to_caller, From})——gen_call 路径禁用
pub(all) enum Gleader {
  StdoutToCaller // 仅 handle_info 路径合法——$gen_call 路径禁用(锚 6)
  GPid(String) // pid 文本句柄——$gen_call 路径真形
}

///|

///|
/// rpc:call 请求组装(rex 元协议——Request 五元真形):
/// {call, Mod, Fun, Args, Gleader}(rpc.erl handle_call 首参铁证;
/// **Gleader=真 pid**($gen_call 路径)——GPid 形传 caller pid 文本→
/// 组合根 VPid 化;StdoutToCaller 裸 atom 在 $gen_call 路径=function_clause)
pub fn build_call(
  mod_ : String,
  fun : String,
  args : Array[EtfVal],
  gleader~ : EtfVal,
) -> EtfVal {
  VTuple([VAtom("call"), VAtom(mod_), VAtom(fun), VList(args), gleader])
}

///|
/// reg_send 控制元组 4 元(dist.c DOP_REG_SEND 臂真源:tuple_arity 必须为 4):
/// {6, FromPid, Cookie='', ToName}——R9B+ cookie 恒空 atom('');
/// 3 元=arity≠4→Invalid→erts_kill_dist_connection→FIN(茗篇层 2)
pub fn reg_send_ctl(from : EtfVal, to_name : EtfVal) -> EtfVal {
  VTuple([VInt(6), from, VAtom(""), to_name])
}

///|
/// $gen_call 包裹(gen_server 元协议真形——probea rex 实测对拍):
/// {'$gen_call', {FromPid, Ref}, Request}——From={caller pid, ref} 二元
/// (ref=NEW_REFERENCE 结构真形;rex reply 经 {Ref, Reply} 回投——ref 匹配
/// 由 parse_reply 判)
pub fn wrap_gen_call(
  from_pid : EtfVal,
  ref_ : EtfVal,
  request : EtfVal,
) -> EtfVal {
  VTuple([VAtom("$gen_call"), VTuple([from_pid, ref_]), request])
}

///|
/// 应答解真形(probea rex 实测直证 2026-11-07):rex 经 gen_server:reply
/// 回投 {Ref, Reply}(SEND_SENDER 22 形投递)——**{rex, Result} 老形勘误
/// 退役**(真 Erlang rex 从不发 {rex,...};旧罐随勘误退役)。
/// expect_ref=我方发的 ref(回显匹配核——不匹配=串包,拒)
pub fn parse_reply(
  data : Bytes,
  codec : &EtfCodec,
  expect_ref~ : EtfVal,
) -> EtfVal raise RpcError {
  let t = codec.decode(data)
  match t {
    VTuple(items) => {
      if items.length() == 2 {
        // ref 回显匹配核(值相等——VPid/VRef 结构对拍)
        if show_val(items[0]) == show_val(expect_ref) {
          return items[1]
        }
        raise RpcError::BadReply(
          "ref 不匹配(串包?): got \{show_val(items[0])} want \{show_val(expect_ref)}",
        )
      }
      raise RpcError::BadReply("reply 非二元: \{show_val(t)}")
    }
    _ => raise RpcError::BadReply("reply 非元组")
  }
}

///|
/// rpc:call 全程:组装→send→recv→parse(组合根一行装配;
/// caller/refv=真 ALIVE2 注册 pid/ref(编造 creation=shutdown 铁律);
/// 传输面组帧=同包单头 'p'+131+ctl+131+msg(dist_frame 形,probe 实证))
pub async fn call(
  codec : &EtfCodec,
  port : &RpcPort,
  node : String,
  caller~ : EtfVal,
  ref_~ : EtfVal,
  mod_ : String,
  fun : String,
  args : Array[EtfVal],
) -> EtfVal raise RpcError {
  let frame = codec.encode([
    reg_send_ctl(caller, VAtom("rex")),
    wrap_gen_call(caller, ref_, build_call(mod_, fun, args, gleader=caller)),
  ])
  port.send(node, frame)
  let reply = port.recv()
  parse_reply(reply, codec, expect_ref=ref_)
}

///|
/// 词表面(数据化骨架——:erlang/:ets/:application/:supervisor 常用调用形;
/// 大 gen 思想:词表 TOML 化挂账,现显式穷尽;Gleader 由 call 装配注入)
pub fn erlang_nodes() -> EtfVal {
  build_call("erlang", "nodes", [], gleader=VAtom("user"))
}

///|
pub fn ets_all(node_name : String) -> EtfVal {
  build_call("ets", "all", [VAtom(node_name)], gleader=VAtom("user"))
}

///|
pub fn application_which_applications() -> EtfVal {
  build_call("application", "which_applications", [], gleader=VAtom("user"))
}

///|
/// observer FetchPort 词表(真件④ v1——零参面四件;真源勘定 2026-11-07:
/// observer_backend 无常驻进程(etop.erl:184 remote spawn 一次性形)——
/// 跨节点数据面走 rex apply 观察键勘定 39 键谱(uptime/schedulers/
/// process_count/内存七项等);procs_info/1 需 opts record 参数候勘形)
pub fn observer_sys_info() -> EtfVal {
  build_call("observer_backend", "sys_info", [], gleader=VAtom("user"))
}

///|
pub fn observer_socket_info() -> EtfVal {
  build_call("observer_backend", "socket_info", [], gleader=VAtom("user"))
}

///|
pub fn observer_port_list() -> EtfVal {
  build_call("observer_backend", "get_port_list", [], gleader=VAtom("user"))
}

///|
pub fn observer_socket_list() -> EtfVal {
  build_call("observer_backend", "get_socket_list", [], gleader=VAtom("user"))
}

///|
/// 进程详情 fetch(FetchPort v2 真件——简化路 v2 per-pid 五键:
/// memory/reductions/message_queue_len/registered_name/current_function;
/// 键映射 etop_proc_info 七字段谱:pid/mem/reds/name/runtime/cf/mq——
/// runtime 侧两拍差分自算;真网勘形回执全解通 2026-11-07)
pub fn observer_process_info(pid : EtfVal) -> EtfVal {
  build_call(
    "erlang",
    "process_info",
    [
      pid,
      VList([
        VAtom("memory"),
        VAtom("reductions"),
        VAtom("message_queue_len"),
        VAtom("registered_name"),
        VAtom("current_function"),
      ]),
    ],
    gleader=VAtom("user"),
  )
}

///|
pub fn supervisor_which_children(sup_name : String) -> EtfVal {
  build_call(
    "supervisor",
    "which_children",
    [VAtom(sup_name)],
    gleader=VAtom("user"),
  )
}

///|
/// 错域文本面(手写防 0079 晋升+零依赖)
pub fn RpcError::show(self : RpcError) -> String {
  match self {
    Codec(m) => "Codec(\{m})"
    Transport(m) => "Transport(\{m})"
    BadReply(m) => "BadReply(\{m})"
  }
}

///|
/// 项值印面(勘定/诊断)
pub fn show_val(v : EtfVal) -> String {
  match v {
    VAtom(s) => "Atom(\{s})"
    VInt(n) => "Int(\{n})"
    VFloat(d) => "Float(\{d})"
    VTuple(items) => {
      let inner = items.map(x => show_val(x)).join(",")
      "Tuple(\{inner})"
    }
    VList(items) => {
      let inner = items.map(x => show_val(x)).join(",")
      "List(\{inner})"
    }
    VNil => "Nil"
    VBinary(b) => "Bin[\{b.length()}B]"
    VString(s) => "Str(\{s})"
    VMap(pairs) => {
      let inner = pairs.map(p => "\{show_val(p.0)}:\{show_val(p.1)}").join(",")
      "Map(\{inner})"
    }
    VPid(node, id, serial, creation) =>
      "Pid(<\{node}.\{id}.\{serial}>\{creation})"
    VPort(node, id, creation) => "Port(<\{node}.\{id}>\{creation})"
    VRef(node, id, serial, creation) =>
      "Ref(<\{node}.\{id}.\{serial}>\{creation})"
    VOther(s) => "Other(\{s})"
  }
}