feat: 把 A 端的 Session/KCP/视频/控制 都收口到一个本地 daemon 进程里，Django 和输入发送端都改成通过本机 UDS HTTP 去访问它，同时补齐了观测、性能和可用性上的几个关键问题。

README.md

@@ -27,6 +27,30 @@ make python-ext
 make python-install
 ```
 
+## A-Side OmniDaemon
+
+The A-side daemon is configured from `config/a_side_omnidaemon.yaml` in this repo. The safest way to start it is to pass that file explicitly, because the installed Python package does not bundle the YAML config.
+
+Run from a source checkout:
+
+```bash
+python -m omnisocket_a_side.daemon --config "$(pwd)/config/a_side_omnidaemon.yaml"
+```
+
+Or, if you installed the console script:
+
+```bash
+OMNIDAEMON_CONFIG="$(pwd)/config/a_side_omnidaemon.yaml" \
+omnisocket-a-side-daemon
+```
+
+Optional overrides:
+
+- `OMNIDAEMON_SOCKET=/tmp/omnisocket-a-side.sock` selects the local UDS path.
+- `OMNIDAEMON_CONFIG=/abs/path/to/a_side_omnidaemon.yaml` overrides `--config`.
+
+For `robot-command-center` and the A-side senders, keep the daemon and its clients on the same Linux machine so they can share the Unix-domain socket.
+
 ## Run On Different Machines
 
 Server `D` runs the KCP hub on `0.0.0.0:10909`:

config/a_side_omnidaemon.yaml

@@ -0,0 +1,51 @@
+transport:
+  server_addr: "81.70.156.140:10909"
+  relay_via: "106.55.173.235:10909"
+  bind_ip: ""
+  bind_device: ""
+
+control_sender:
+  peer_id: "peer-a-ctrl"
+  target_peer: "peer-b-ctrl"
+  nodelay: 1
+  interval_ms: 5
+  resend: 2
+  nc: 1
+  sndwnd: 32
+  rcvwnd: 32
+  mtu: 1400
+  stats_interval_ms: 100
+
+video_receiver:
+  peer_id: "peer-a-video"
+  buffer_bytes: 1048576
+  nodelay: 1
+  interval_ms: 10
+  resend: 2
+  nc: 1
+  sndwnd: 256
+  rcvwnd: 256
+  mtu: 1400
+  stats_interval_ms: 100
+
+daemon:
+  socket_path: "/tmp/omnisocket-a-side.sock"
+  reconnect_delay_ms: 2000
+  telemetry_interval_ms: 100
+  analog_send_hz: 100
+  frame_stale_ms: 500
+
+policy:
+  health_window_ms: 2000
+  green_srtt_ms: 35
+  yellow_srtt_ms: 60
+  retrans_red_threshold: 10
+  profile_green:
+    fps: 15
+    max_frame_kb: 60
+  profile_yellow:
+    fps: 10
+    max_frame_kb: 40
+  profile_red:
+    fps: 5
+    max_frame_kb: 20

include/peer_kcp_client.h

@@ -27,6 +27,7 @@ int kcp_client_receive_timed(kcp_client_t *client, message_t *out_msg, int timeo
 int kcp_client_receive(kcp_client_t *client, message_t *out_msg);
 int kcp_client_receive_binary_into(kcp_client_t *client, void *buffer, size_t buffer_len, kcp_client_recv_meta_t *out_meta, int timeout_ms);
 int kcp_client_persist_message(kcp_client_t *client, const message_t *msg, const char *inbox_dir, char *out_path, size_t out_path_len);
+int kcp_client_metrics_snapshot(kcp_client_t *client, kcp_conn_metrics_t *out_metrics);
 int kcp_client_close(kcp_client_t *client);
 void kcp_client_free(kcp_client_t *client);
 

include/transport_kcp.h

@@ -43,6 +43,32 @@ extern "C" {
 
 typedef struct kcp_conn kcp_conn_t;
 typedef struct kcp_listener kcp_listener_t;
+typedef struct kcp_conn_metrics {
+    int connected;
+    int has_conv;
+    uint32_t conv;
+    char local_addr[OMNI_MAX_ADDR_TEXT];
+    char remote_addr[OMNI_MAX_ADDR_TEXT];
+    uint32_t rto_ms;
+    int32_t srtt_ms;
+    int32_t srttvar_ms;
+    uint64_t bytes_sent;
+    uint64_t bytes_received;
+    uint64_t in_pkts;
+    uint64_t out_pkts;
+    uint64_t in_segs;
+    uint64_t out_segs;
+    uint64_t retrans_segs;
+    uint64_t fast_retrans_segs;
+    uint64_t early_retrans_segs;
+    uint64_t lost_segs;
+    uint64_t repeat_segs;
+    uint64_t in_errs;
+    uint64_t kcp_in_errs;
+    uint64_t ring_buffer_snd_queue;
+    uint64_t ring_buffer_rcv_queue;
+    uint64_t ring_buffer_snd_buffer;
+} kcp_conn_metrics_t;
 typedef struct kcp_conn_options {
     int nodelay;
     int interval_ms;
@@ -68,6 +94,7 @@ int kcp_conn_close(kcp_conn_t *conn);
 void kcp_conn_free(kcp_conn_t *conn);
 uint32_t kcp_conn_conv(const kcp_conn_t *conn);
 int kcp_conn_local_addr(const kcp_conn_t *conn, struct sockaddr_storage *addr, socklen_t *addr_len);
+int kcp_conn_metrics_snapshot(kcp_conn_t *conn, kcp_conn_metrics_t *out_metrics);
 
 kcp_listener_t *kcp_listener_listen(const char *listen_addr, const char *bind_device, kcp_packet_debug_logger_t *packet_logger, const char *node_role, const char *node_id);
 kcp_conn_t *kcp_listener_accept(kcp_listener_t *listener);

python/omnisocket/_omnisocket.c

@@ -22,6 +22,13 @@ PyDoc_STRVAR(
     "current frame has already been consumed and is lost."
 );
 
+PyDoc_STRVAR(
+    PyOmniSession_kcp_metrics_doc,
+    "kcp_metrics() -> dict\n"
+    "\n"
+    "Return a snapshot of low-level KCP metrics for the current session."
+);
+
 static PyObject *PyOmniSession_new(PyTypeObject *type, PyObject *args, PyObject *kwargs) {
     PyOmniSession *self;
     (void) args;
@@ -281,6 +288,67 @@ static PyObject *PyOmniSession_stats(PyOmniSession *self, PyObject *Py_UNUSED(ig
     );
 }
 
+static PyObject *PyOmniSession_kcp_metrics(PyOmniSession *self, PyObject *Py_UNUSED(ignored)) {
+    omnisocket_session_kcp_metrics_t metrics;
+
+    memset(&metrics, 0, sizeof(metrics));
+    if (omnisocket_session_kcp_metrics_snapshot(&self->session, &metrics) != 0) {
+        return PyErr_SetFromErrno(PyExc_OSError);
+    }
+
+    return Py_BuildValue(
+        "{s:i,s:i,s:I,s:s,s:s,s:I,s:i,s:i,s:K,s:K,s:K,s:K,s:K,s:K,s:K,s:K,s:K,s:K,s:K,s:K,s:K,s:K,s:K}",
+        "connected",
+        metrics.connected,
+        "has_conv",
+        metrics.has_conv,
+        "conv",
+        metrics.conv,
+        "local_addr",
+        metrics.local_addr,
+        "remote_addr",
+        metrics.remote_addr,
+        "rto_ms",
+        metrics.rto_ms,
+        "srtt_ms",
+        metrics.srtt_ms,
+        "srttvar_ms",
+        metrics.srttvar_ms,
+        "bytes_sent",
+        (unsigned long long) metrics.bytes_sent,
+        "bytes_received",
+        (unsigned long long) metrics.bytes_received,
+        "in_pkts",
+        (unsigned long long) metrics.in_pkts,
+        "out_pkts",
+        (unsigned long long) metrics.out_pkts,
+        "in_segs",
+        (unsigned long long) metrics.in_segs,
+        "out_segs",
+        (unsigned long long) metrics.out_segs,
+        "retrans_segs",
+        (unsigned long long) metrics.retrans_segs,
+        "fast_retrans_segs",
+        (unsigned long long) metrics.fast_retrans_segs,
+        "early_retrans_segs",
+        (unsigned long long) metrics.early_retrans_segs,
+        "lost_segs",
+        (unsigned long long) metrics.lost_segs,
+        "repeat_segs",
+        (unsigned long long) metrics.repeat_segs,
+        "in_errs",
+        (unsigned long long) metrics.in_errs,
+        "kcp_in_errs",
+        (unsigned long long) metrics.kcp_in_errs,
+        "ring_buffer_snd_queue",
+        (unsigned long long) metrics.ring_buffer_snd_queue,
+        "ring_buffer_rcv_queue",
+        (unsigned long long) metrics.ring_buffer_rcv_queue,
+        "ring_buffer_snd_buffer",
+        (unsigned long long) metrics.ring_buffer_snd_buffer
+    );
+}
+
 static PyMethodDef PyOmniSession_methods[] = {
     {"connect", (PyCFunction) PyOmniSession_connect, METH_VARARGS | METH_KEYWORDS, NULL},
     {"close", (PyCFunction) PyOmniSession_close, METH_NOARGS, NULL},
@@ -288,6 +356,7 @@ static PyMethodDef PyOmniSession_methods[] = {
     {"recv", (PyCFunction) PyOmniSession_recv, METH_VARARGS | METH_KEYWORDS, PyOmniSession_recv_doc},
     {"recv_into", (PyCFunction) PyOmniSession_recv_into, METH_VARARGS | METH_KEYWORDS, PyOmniSession_recv_into_doc},
     {"stats", (PyCFunction) PyOmniSession_stats, METH_NOARGS, NULL},
+    {"kcp_metrics", (PyCFunction) PyOmniSession_kcp_metrics, METH_NOARGS, PyOmniSession_kcp_metrics_doc},
     {NULL, NULL, 0, NULL}
 };
 

python/omnisocket/omnisocket_client.c

@@ -246,3 +246,72 @@ void omnisocket_session_stats_snapshot(omnisocket_session_t *session, omnisocket
     *out_stats = session->stats;
     pthread_mutex_unlock(&session->mutex);
 }
+
+int omnisocket_session_kcp_metrics_snapshot(
+    omnisocket_session_t *session,
+    omnisocket_session_kcp_metrics_t *out_metrics
+) {
+    kcp_client_t *client = NULL;
+    kcp_conn_metrics_t metrics;
+    int rc = 0;
+
+    if (session == NULL || out_metrics == NULL) {
+        errno = EINVAL;
+        return -1;
+    }
+
+    memset(out_metrics, 0, sizeof(*out_metrics));
+
+    pthread_mutex_lock(&session->mutex);
+    if (session->client != NULL && !session->closing) {
+        client = session->client;
+        session->active_ops += 1;
+    }
+    pthread_mutex_unlock(&session->mutex);
+
+    if (client == NULL) {
+        return 0;
+    }
+
+    memset(&metrics, 0, sizeof(metrics));
+    rc = kcp_client_metrics_snapshot(client, &metrics);
+
+    pthread_mutex_lock(&session->mutex);
+    if (session->active_ops > 0) {
+        session->active_ops -= 1;
+    }
+    if (session->closing && session->active_ops == 0) {
+        pthread_cond_broadcast(&session->idle_cond);
+    }
+    pthread_mutex_unlock(&session->mutex);
+
+    if (rc != 0) {
+        return rc;
+    }
+
+    out_metrics->connected = metrics.connected;
+    out_metrics->has_conv = metrics.has_conv;
+    out_metrics->conv = metrics.conv;
+    snprintf(out_metrics->local_addr, sizeof(out_metrics->local_addr), "%s", metrics.local_addr);
+    snprintf(out_metrics->remote_addr, sizeof(out_metrics->remote_addr), "%s", metrics.remote_addr);
+    out_metrics->rto_ms = metrics.rto_ms;
+    out_metrics->srtt_ms = metrics.srtt_ms;
+    out_metrics->srttvar_ms = metrics.srttvar_ms;
+    out_metrics->bytes_sent = metrics.bytes_sent;
+    out_metrics->bytes_received = metrics.bytes_received;
+    out_metrics->in_pkts = metrics.in_pkts;
+    out_metrics->out_pkts = metrics.out_pkts;
+    out_metrics->in_segs = metrics.in_segs;
+    out_metrics->out_segs = metrics.out_segs;
+    out_metrics->retrans_segs = metrics.retrans_segs;
+    out_metrics->fast_retrans_segs = metrics.fast_retrans_segs;
+    out_metrics->early_retrans_segs = metrics.early_retrans_segs;
+    out_metrics->lost_segs = metrics.lost_segs;
+    out_metrics->repeat_segs = metrics.repeat_segs;
+    out_metrics->in_errs = metrics.in_errs;
+    out_metrics->kcp_in_errs = metrics.kcp_in_errs;
+    out_metrics->ring_buffer_snd_queue = metrics.ring_buffer_snd_queue;
+    out_metrics->ring_buffer_rcv_queue = metrics.ring_buffer_rcv_queue;
+    out_metrics->ring_buffer_snd_buffer = metrics.ring_buffer_snd_buffer;
+    return 0;
+}

python/omnisocket/omnisocket_client.h

@@ -14,6 +14,33 @@ typedef struct omnisocket_session_stats {
     int connected;
 } omnisocket_session_stats_t;
 
+typedef struct omnisocket_session_kcp_metrics {
+    int connected;
+    int has_conv;
+    uint32_t conv;
+    char local_addr[OMNI_MAX_ADDR_TEXT];
+    char remote_addr[OMNI_MAX_ADDR_TEXT];
+    uint32_t rto_ms;
+    int32_t srtt_ms;
+    int32_t srttvar_ms;
+    uint64_t bytes_sent;
+    uint64_t bytes_received;
+    uint64_t in_pkts;
+    uint64_t out_pkts;
+    uint64_t in_segs;
+    uint64_t out_segs;
+    uint64_t retrans_segs;
+    uint64_t fast_retrans_segs;
+    uint64_t early_retrans_segs;
+    uint64_t lost_segs;
+    uint64_t repeat_segs;
+    uint64_t in_errs;
+    uint64_t kcp_in_errs;
+    uint64_t ring_buffer_snd_queue;
+    uint64_t ring_buffer_rcv_queue;
+    uint64_t ring_buffer_snd_buffer;
+} omnisocket_session_kcp_metrics_t;
+
 typedef struct omnisocket_session {
     pthread_mutex_t mutex;
     pthread_cond_t idle_cond;
@@ -47,5 +74,9 @@ int omnisocket_session_recv_into(
     int timeout_ms
 );
 void omnisocket_session_stats_snapshot(omnisocket_session_t *session, omnisocket_session_stats_t *out_stats);
+int omnisocket_session_kcp_metrics_snapshot(
+    omnisocket_session_t *session,
+    omnisocket_session_kcp_metrics_t *out_metrics
+);
 
 #endif

python/omnisocket_a_side/__init__.py

@@ -0,0 +1,9 @@
+from pathlib import Path
+
+
+PACKAGE_ROOT = Path(__file__).resolve().parent
+PYTHON_ROOT = PACKAGE_ROOT.parent
+REPO_ROOT = PYTHON_ROOT.parent
+DEFAULT_SOCKET_PATH = "/tmp/omnisocket-a-side.sock"
+DEFAULT_CONFIG_PATH = REPO_ROOT / "config" / "a_side_omnidaemon.yaml"
+VERSION = "0.1.0"

python/omnisocket_a_side/__main__.py

@@ -0,0 +1,5 @@
+from .daemon import main
+
+
+if __name__ == "__main__":
+    main()

python/omnisocket_a_side/client.py

@@ -0,0 +1,149 @@
+"""Local Unix-domain HTTP client for the A-side OmniDaemon."""
+
+from __future__ import annotations
+
+import http.client
+import json
+import os
+import socket
+import threading
+from typing import Any
+
+from . import DEFAULT_SOCKET_PATH
+
+
+class OmniDaemonError(RuntimeError):
+    def __init__(self, message: str, status_code: int | None = None) -> None:
+        super().__init__(message)
+        self.status_code = status_code
+
+
+class UnixHTTPConnection(http.client.HTTPConnection):
+    def __init__(self, socket_path: str, timeout: float = 2.0) -> None:
+        super().__init__("localhost", timeout=timeout)
+        self.socket_path = socket_path
+
+    def connect(self) -> None:  # pragma: no cover - runtime depends on Linux socket support
+        if not hasattr(socket, "AF_UNIX"):
+            raise OSError("AF_UNIX sockets are not available on this platform")
+        self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+        self.sock.settimeout(self.timeout)
+        self.sock.connect(self.socket_path)
+
+
+class OmniDaemonClient:
+    def __init__(self, socket_path: str | None = None, timeout: float = 2.0) -> None:
+        self.socket_path = socket_path or os.getenv("OMNIDAEMON_SOCKET", DEFAULT_SOCKET_PATH)
+        self.timeout = timeout
+        self._local = threading.local()
+
+    def get_health(self) -> dict[str, Any]:
+        return self._request_json("GET", "/v1/health")
+
+    def get_state(self) -> dict[str, Any]:
+        return self._request_json("GET", "/v1/state")
+
+    def get_video_frame(self) -> bytes:
+        return self._request_bytes("GET", "/v1/video/frame")
+
+    def get_control_status(self) -> dict[str, Any]:
+        return self._request_json("GET", "/v1/control/status")
+
+    def send_control_event(
+        self,
+        *,
+        source: str,
+        event_code: str,
+        drive_value: float = 1.0,
+        client_time_ms: int | None = None,
+    ) -> dict[str, Any]:
+        payload = {
+            "source": source,
+            "event_code": event_code,
+            "drive_value": float(drive_value),
+            "client_time_ms": client_time_ms,
+        }
+        return self._request_json("POST", "/v1/control/event", payload)
+
+    def close(self) -> None:
+        self._reset_connection()
+
+    def _request_json(
+        self,
+        method: str,
+        path: str,
+        payload: dict[str, Any] | None = None,
+    ) -> dict[str, Any]:
+        raw = self._request_bytes(method, path, payload)
+        if not raw:
+            return {}
+        try:
+            return json.loads(raw.decode("utf-8"))
+        except json.JSONDecodeError as error:
+            raise OmniDaemonError(f"invalid daemon JSON response: {error}") from error
+
+    def _request_bytes(
+        self,
+        method: str,
+        path: str,
+        payload: dict[str, Any] | None = None,
+    ) -> bytes:
+        body = b""
+        headers: dict[str, str] = {}
+        if payload is not None:
+            body = json.dumps(payload).encode("utf-8")
+            headers["Content-Type"] = "application/json"
+            headers["Content-Length"] = str(len(body))
+        headers.setdefault("Connection", "keep-alive")
+
+        for attempt in range(2):
+            connection = self._get_connection()
+            try:
+                connection.request(method, path, body=body, headers=headers)
+                response = connection.getresponse()
+                raw = response.read()
+            except FileNotFoundError as error:
+                self._reset_connection()
+                raise OmniDaemonError(
+                    f"daemon socket not found: {self.socket_path}"
+                ) from error
+            except (OSError, http.client.HTTPException) as error:
+                self._reset_connection()
+                if attempt == 0:
+                    continue
+                raise OmniDaemonError(
+                    f"daemon request failed via {self.socket_path}: {error}"
+                ) from error
+
+            if getattr(response, "will_close", False):
+                self._reset_connection()
+
+            if response.status >= 400:
+                message = raw.decode("utf-8", errors="replace").strip() or response.reason
+                try:
+                    parsed = json.loads(message)
+                    if isinstance(parsed, dict) and "error" in parsed:
+                        message = str(parsed["error"])
+                except json.JSONDecodeError:
+                    pass
+                raise OmniDaemonError(message, status_code=response.status)
+            return raw
+
+        raise OmniDaemonError(f"daemon request failed via {self.socket_path}: retry exhausted")
+
+    def _get_connection(self) -> UnixHTTPConnection:
+        connection = getattr(self._local, "connection", None)
+        if connection is None:
+            connection = UnixHTTPConnection(self.socket_path, timeout=self.timeout)
+            self._local.connection = connection
+        return connection
+
+    def _reset_connection(self) -> None:
+        connection = getattr(self._local, "connection", None)
+        if connection is None:
+            return
+        try:
+            connection.close()
+        except OSError:
+            pass
+        self._local.connection = None

python/omnisocket_a_side/control_codec.py

@@ -0,0 +1,90 @@
+"""Binary control packet codec shared by the daemon and local clients."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+import struct
+import time
+
+
+CONTROL_PACKET_VERSION = 1
+CONTROL_PACKET_STRUCT = struct.Struct("!BBHIfQ")
+
+EVENT_NAME_TO_ID = {
+    "pose_home": 1,
+    "pose_hold": 2,
+    "mode_stride": 3,
+    "surge_up": 6,
+    "surge_down": 7,
+    "sway_left": 8,
+    "sway_right": 9,
+    "spin_left": 10,
+    "spin_right": 11,
+    "set_surge": 12,
+    "set_sway": 13,
+    "set_spin": 14,
+    "set_lift": 15,
+    "lift_up": 16,
+    "lift_down": 17,
+    "trim_reset": 18,
+    "session_quit": 19,
+}
+EVENT_ID_TO_NAME = {value: key for key, value in EVENT_NAME_TO_ID.items()}
+ANALOG_EVENT_CODES = {"set_surge", "set_sway", "set_spin"}
+
+
+@dataclass(slots=True)
+class ControlPacket:
+    seq_id: int
+    event_id: int
+    drive_value: float = 1.0
+    sent_at_ns: int = 0
+
+    @property
+    def event_name(self) -> str:
+        return EVENT_ID_TO_NAME.get(self.event_id, f"unknown_{self.event_id}")
+
+    def encode(self) -> bytes:
+        sent_at_ns = self.sent_at_ns or time.time_ns()
+        return CONTROL_PACKET_STRUCT.pack(
+            CONTROL_PACKET_VERSION,
+            self.event_id,
+            0,
+            int(self.seq_id),
+            float(self.drive_value),
+            int(sent_at_ns),
+        )
+
+    @classmethod
+    def decode(cls, payload: bytes) -> "ControlPacket":
+        if len(payload) != CONTROL_PACKET_STRUCT.size:
+            raise ValueError(
+                f"invalid control packet length {len(payload)}; "
+                f"want {CONTROL_PACKET_STRUCT.size}"
+            )
+        version, event_id, _reserved, seq_id, drive_value, sent_at_ns = (
+            CONTROL_PACKET_STRUCT.unpack(payload)
+        )
+        if version != CONTROL_PACKET_VERSION:
+            raise ValueError(f"unsupported control packet version {version}")
+        return cls(
+            seq_id=int(seq_id),
+            event_id=int(event_id),
+            drive_value=float(drive_value),
+            sent_at_ns=int(sent_at_ns),
+        )
+
+
+def make_control_packet(
+    seq_id: int,
+    event_name: str,
+    drive_value: float = 1.0,
+    sent_at_ns: int | None = None,
+) -> ControlPacket:
+    event_id = EVENT_NAME_TO_ID[event_name]
+    return ControlPacket(
+        seq_id=seq_id,
+        event_id=event_id,
+        drive_value=drive_value,
+        sent_at_ns=time.time_ns() if sent_at_ns is None else sent_at_ns,
+    )

python/setup.py

@@ -35,7 +35,12 @@ COMMON_SOURCES = [
 setup(
     name="omnisocket",
     version="0.1.0",
-    packages=["omnisocket"],
+    packages=["omnisocket", "omnisocket_a_side"],
+    entry_points={
+        "console_scripts": [
+            "omnisocket-a-side-daemon=omnisocket_a_side.daemon:main",
+        ]
+    },
     ext_modules=[
         Extension(
             "omnisocket._omnisocket",

src/peer_kcp_client.c

@@ -294,6 +294,14 @@ int kcp_client_persist_message(kcp_client_t *client, const message_t *msg, const
     return 0;
 }
 
+int kcp_client_metrics_snapshot(kcp_client_t *client, kcp_conn_metrics_t *out_metrics) {
+    if (client == NULL || out_metrics == NULL) {
+        errno = EINVAL;
+        return -1;
+    }
+    return kcp_conn_metrics_snapshot(client->conn, out_metrics);
+}
+
 int kcp_client_close(kcp_client_t *client) {
     return client == NULL ? 0 : kcp_conn_close(client->conn);
 }

src/transport_kcp.c

@@ -1788,6 +1788,83 @@ int kcp_conn_local_addr(const kcp_conn_t *conn, struct sockaddr_storage *addr, s
     return 0;
 }
 
+int kcp_conn_metrics_snapshot(kcp_conn_t *conn, kcp_conn_metrics_t *out_metrics) {
+    struct sockaddr_storage local_addr;
+    socklen_t local_len = sizeof(local_addr);
+
+    if (conn == NULL || out_metrics == NULL) {
+        errno = EINVAL;
+        return -1;
+    }
+
+    memset(out_metrics, 0, sizeof(*out_metrics));
+    out_metrics->connected = !atomic_load(&conn->closed);
+
+    if (conn->sock_state != NULL && !conn->socket_closed &&
+        getsockname(conn->sock_state->fd, (struct sockaddr *) &local_addr, &local_len) == 0) {
+        omni_sockaddr_to_string(
+            (const struct sockaddr *) &local_addr,
+            local_len,
+            out_metrics->local_addr,
+            sizeof(out_metrics->local_addr)
+        );
+    }
+    if (conn->remote_addr_len > 0) {
+        omni_sockaddr_to_string(
+            (const struct sockaddr *) &conn->remote_addr,
+            conn->remote_addr_len,
+            out_metrics->remote_addr,
+            sizeof(out_metrics->remote_addr)
+        );
+    }
+
+    if (conn->process_sampler != NULL) {
+        out_metrics->bytes_sent = atomic_load_explicit(&conn->process_sampler->bytes_sent, memory_order_relaxed);
+        out_metrics->bytes_received = atomic_load_explicit(&conn->process_sampler->bytes_received, memory_order_relaxed);
+        out_metrics->in_pkts = atomic_load_explicit(&conn->process_sampler->in_pkts, memory_order_relaxed);
+        out_metrics->out_pkts = atomic_load_explicit(&conn->process_sampler->out_pkts, memory_order_relaxed);
+        out_metrics->in_segs = atomic_load_explicit(&conn->process_sampler->in_segs, memory_order_relaxed);
+        out_metrics->out_segs = atomic_load_explicit(&conn->process_sampler->out_segs, memory_order_relaxed);
+        out_metrics->in_errs = atomic_load_explicit(&conn->process_sampler->in_errs, memory_order_relaxed);
+        out_metrics->kcp_in_errs = atomic_load_explicit(&conn->process_sampler->kcp_in_errs, memory_order_relaxed);
+    } else {
+        out_metrics->bytes_sent = conn->pending_bytes_sent;
+        out_metrics->bytes_received = conn->pending_bytes_received;
+        out_metrics->in_pkts = conn->pending_in_pkts;
+        out_metrics->out_pkts = conn->pending_out_pkts;
+        out_metrics->in_segs = conn->pending_in_segs;
+        out_metrics->out_segs = conn->pending_out_segs;
+        out_metrics->in_errs = conn->pending_in_errs;
+        out_metrics->kcp_in_errs = conn->pending_kcp_in_errs;
+    }
+
+    pthread_mutex_lock(&conn->kcp_mu);
+    if (conn->kcp != NULL) {
+        out_metrics->has_conv = 1;
+        out_metrics->conv = conn->kcp->conv;
+        out_metrics->rto_ms = conn->kcp->rx_rto;
+        out_metrics->srtt_ms = conn->kcp->rx_srtt;
+        out_metrics->srttvar_ms = conn->kcp->rx_rttval;
+        out_metrics->ring_buffer_snd_queue = conn->kcp->nsnd_que;
+        out_metrics->ring_buffer_rcv_queue = conn->kcp->nrcv_que;
+        out_metrics->ring_buffer_snd_buffer = conn->kcp->nsnd_buf;
+        out_metrics->fast_retrans_segs = conn->kcp->fast_retrans_xmit;
+        /* This KCP fork does not implement early retransmit, so the counter stays zero. */
+        out_metrics->early_retrans_segs = conn->kcp->early_retrans_xmit;
+        out_metrics->lost_segs = conn->kcp->lost_xmit;
+        out_metrics->repeat_segs = conn->kcp->repeat_xmit;
+        out_metrics->retrans_segs =
+            out_metrics->fast_retrans_segs +
+            out_metrics->early_retrans_segs +
+            out_metrics->lost_segs;
+    } else {
+        out_metrics->connected = 0;
+    }
+    pthread_mutex_unlock(&conn->kcp_mu);
+
+    return 0;
+}
+
 int kcp_conn_close(kcp_conn_t *conn) {
     if (conn == NULL) {
         return 0;

third_party/kcp/ikcp.c

@@ -298,6 +298,10 @@ ikcpcb *ikcp_create(IUINT32 conv, void *user)
     kcp->fastlimit = IKCP_FASTACK_LIMIT;
     kcp->nocwnd = 0;
     kcp->xmit = 0;
+    kcp->fast_retrans_xmit = 0;
+    kcp->early_retrans_xmit = 0;
+    kcp->lost_xmit = 0;
+    kcp->repeat_xmit = 0;
     kcp->dead_link = IKCP_DEADLINK;
     kcp->output = NULL;
     kcp->writelog = NULL;
@@ -788,6 +792,7 @@ void ikcp_parse_data(ikcpcb *kcp, IKCPSEG *newseg)
     }
     else
     {
+        kcp->repeat_xmit++;
         ikcp_segment_delete(kcp, newseg);
     }
 
@@ -1192,6 +1197,7 @@ void ikcp_flush(ikcpcb *kcp)
             needsend = 1;
             segment->xmit++;
             kcp->xmit++;
+            kcp->lost_xmit++;
             if (kcp->nodelay == 0)
             {
                 segment->rto += _imax_(segment->rto, (IUINT32)kcp->rx_rto);
@@ -1211,6 +1217,7 @@ void ikcp_flush(ikcpcb *kcp)
             {
                 needsend = 1;
                 segment->xmit++;
+                kcp->fast_retrans_xmit++;
                 segment->fastack = 0;
                 segment->resendts = current + segment->rto;
                 change++;

third_party/kcp/ikcp.h

@@ -300,6 +300,10 @@ struct IKCPCB
     IINT32 rx_rttval, rx_srtt, rx_rto, rx_minrto;
     IUINT32 snd_wnd, rcv_wnd, rmt_wnd, cwnd, probe;
     IUINT32 current, interval, ts_flush, xmit;
+    IUINT32 fast_retrans_xmit;
+    IUINT32 early_retrans_xmit;
+    IUINT32 lost_xmit;
+    IUINT32 repeat_xmit;
     IUINT32 nrcv_buf, nsnd_buf;
     IUINT32 nrcv_que, nsnd_que;
     IUINT32 nodelay, updated;