fix: premature socket teardown on CLOSE + duplicate CLOSE echo

Two bugs when upstream closes:

1. Data loss: gatuna received CLOSE from gatunad and immediately
   RST'd the socket to curl while the drain pump still had queued
   payloads in _deliverChannel. Fix: OnRemoteClose() completes the
   channel writer and cancels the read pump, but lets the drain pump
   finish naturally. Drain pump no longer takes _cts.Token — it stops
   on channel completion and closes the socket gracefully (FIN, OS
   flushes in background).

2. Duplicate CLOSE: gatuna echoed CLOSE back to gatunad (server
   already knows — it initiated). SendClose() now guarded by
   _closeSent flag. OnRemoteClose() never sends CLOSE at all.

Server-side: remove session from store before sending CLOSE so the
retransmit timer stops before CLOSE is queued. Only log CLOSE if
the session actually existed (dedup).

.gitea/workflows/build.yml

@@ -0,0 +1,55 @@
+name: build
+on: [push, pull_request]
+
+jobs:
+  build:
+    runs-on: hugmaster
+    steps:
+      - name: Checkout
+        run: |
+          git clone "${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}.git" .
+          git checkout "$GITHUB_SHA"
+
+      - name: Build Rust
+        run: |
+          cd gatunad
+          cargo build --release
+
+      - name: Build .NET
+        run: |
+          cd gatuna-win
+          dotnet build
+
+      - name: Publish Windows (framework-dependent)
+        if: ${{ startsWith(github.ref, 'refs/tags/v') }}
+        run: |
+          cd gatuna-win
+          dotnet publish -c Release -r win-x64 --no-self-contained -o ../publish/win
+
+      - name: Package release assets
+        if: ${{ startsWith(github.ref, 'refs/tags/v') }}
+        run: |
+          mkdir -p release
+          cp gatunad/target/release/gatunad release/gatunad-linux-amd64
+          cd publish/win && zip -r ../../release/gatuna-win.zip . && cd ../..
+
+      - name: Create Gitea release
+        if: ${{ startsWith(github.ref, 'refs/tags/v') }}
+        run: |
+          TAG="${GITHUB_REF#refs/tags/}"
+          RESP=$(curl -sS -X POST \
+            -H "Authorization: token ${GITHUB_TOKEN}" \
+            -H "Content-Type: application/json" \
+            -d "{\"tag_name\":\"$TAG\",\"name\":\"$TAG\",\"body\":\"Automated build for $TAG\"}" \
+            "${GITHUB_SERVER_URL}/api/v1/repos/${GITHUB_REPOSITORY}/releases")
+          RID=$(echo "$RESP" | jq .id)
+          curl -sS -X POST \
+            -H "Authorization: token ${GITHUB_TOKEN}" \
+            -H "Content-Type: application/octet-stream" \
+            --data-binary @release/gatunad-linux-amd64 \
+            "${GITHUB_SERVER_URL}/api/v1/repos/${GITHUB_REPOSITORY}/releases/${RID}/assets?name=gatunad-linux-amd64"
+          curl -sS -X POST \
+            -H "Authorization: token ${GITHUB_TOKEN}" \
+            -H "Content-Type: application/octet-stream" \
+            --data-binary @release/gatuna-win.zip \
+            "${GITHUB_SERVER_URL}/api/v1/repos/${GITHUB_REPOSITORY}/releases/${RID}/assets?name=gatuna-win.zip"

PROTOCOL.md

@@ -3,7 +3,7 @@
 All frames ride Ethernet with ethertype `0x6969`. The ethertype is the sole
 discriminator; there is no magic number inside the payload.
 
-## Frame layout
+## Frame layout (non-DATA frames)
 
 ```
  0                   1                   2                   3
@@ -11,37 +11,69 @@ discriminator; there is no magic number inside the payload.
 +---------------+---------------+-------------------------------+
 | version       | type          | session_id                    |
 +---------------+---------------+-------------------------------+
-| payload ...                                                   |
+| payload_len (big-endian)      | payload ...                   |
++-------------------------------+                               +
+|                                                               |
 +---------------------------------------------------------------+
 ```
 
-- **version** (u8): protocol version. Currently `1`.
+- **version** (u8): protocol version. Currently `2`.
 - **type** (u8): frame type, see table below.
 - **session_id** (u32, big-endian): `0` for non-session frames; the
   server-assigned ID for session-scoped frames.
-- **payload** (variable): type-dependent. No length field is carried — the
+- **payload_len** (u16, big-endian): number of payload bytes that follow.
-  Ethernet frame length from the capture gives the payload extent.
+  The receiver reads exactly this many bytes and ignores any trailing
+  Ethernet padding (frames under 60 bytes are zero-padded by the NIC to
+  meet the minimum Ethernet frame size).
+- **payload** (`payload_len` bytes): type-dependent.
 
-No CRC, no retransmit, no ordering at this layer. TCP reliability is handled by
+## DATA frame layout
-the endpoints' TCP stacks; UDP (reserved) will rely on application-level
-mechanisms.
 
-Maximum payload: 1500 (Ethernet MTU) − 14 (eth header) − 6 (our header) =
+DATA frames carry two additional fields after the common header for L2-level
-**1480 bytes**. Larger payloads are not emitted in v1.
+reliability:
+
+```
+ 0                   1                   2                   3
+ 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
++---------------+---------------+-------------------------------+
+| version=2     | type=0x06     | session_id                    |
++---------------+---------------+-------------------------------+
+| payload_len (big-endian)      | seq (big-endian)              |
++-------------------------------+-------------------------------+
+| ack_seq (big-endian)          | payload ...                   |
++-------------------------------+                               +
+|                                                               |
++---------------------------------------------------------------+
+```
+
+- **seq** (u32, big-endian): monotonically increasing per-session sequence
+  number. Wraps at 2^32 (same as TCP). Identifies this DATA frame's position
+  in the byte stream.
+- **ack_seq** (u32, big-endian): cumulative acknowledgment — the highest
+  contiguous `seq` that the sender of this frame has delivered to its local
+  TCP socket. The receiver uses this to advance its retransmit window.
+- **payload** (`payload_len` bytes): raw application bytes (0–1480). A
+  `payload_len = 0` DATA frame is a **pure ACK** — it carries no data, just
+  an acknowledgment. This mirrors TCP's empty-segment ACK.
+
+Maximum payload: 1500 (Ethernet MTU) − 8 (common header) − 8 (seq + ack_seq)
+= **1484 bytes**. In practice we cap at **1480** to stay conservative.
 
 ## Frame types
 
 | Type | Name        | Direction     | session_id | Payload                          |
 |------|-------------|---------------|------------|----------------------------------|
 | 0x01 | DISCOVER    | C → broadcast | 0          | empty                            |
-| 0x02 | MANIFEST    | S → C         | 0          | `id:1, proto:1, port:2` × N      |
+| 0x02 | MANIFEST    | S → C         | 0          | `hostname_len:1, hostname:N, entries...` |
-| 0x03 | OPEN        | C → S         | 0          | `upstream_id:1`                  |
+| 0x03 | OPEN        | C → S         | 0          | `upstream_id:1, proto:1`        |
-| 0x04 | OPEN_ACK    | S → C         | assigned   | `upstream_id:1`                  |
+| 0x04 | OPEN_ACK    | S → C         | assigned   | `upstream_id:1, proto:1`        |
 | 0x05 | OPEN_NAK    | S → C         | 0          | `upstream_id:1, reason:1`        |
-| 0x06 | DATA        | both          | session    | raw bytes (≤1480)                |
+| 0x06 | DATA        | both          | session    | `seq:4, ack_seq:4, raw bytes`    |
 | 0x07 | CLOSE       | both          | session    | optional `reason:1`              |
+| 0x0B | PING        | C → S         | 0          | `nonce:8`                        |
+| 0x0C | PONG        | S → C         | 0          | `nonce:8` (echoed)               |
 
-Reserved (unimplemented in v1; parse returns Err, encode unimplemented):
+Reserved (unimplemented; parse returns Err, encode unimplemented):
 
 | Type | Name        |
 |------|-------------|
@@ -49,22 +81,116 @@ Reserved (unimplemented in v1; parse returns Err, encode unimplemented):
 | 0x09 | UDP_DATA    |
 | 0x0A | UDP_CLOSE   |
 
+## L2 reliability
+
+The reliability semantics of a session depend on its transport proto,
+which is carried end-to-end through OPEN/OPEN_ACK:
+
+- **TCP (`proto = 1`):** DATA frames are delivered reliably and in-order via
+  a TCP-like sequence + cumulative ACK + retransmit mechanism. This ensures
+  that a dropped Ethernet frame does not permanently corrupt a TCP session
+  (which would otherwise happen because the local kernel TCP stack ACKs data
+  before it is chunked into DATA frames).
+- **UDP (`proto = 2`, reserved):** DATA frames are delivered best-effort. The
+  `seq` and `ack_seq` fields are present in the frame layout for uniformity
+  but are ignored — no retransmit, no in-order buffering, no pure ACKs. Drops
+  are tolerated because stateless protocols either don't care or handle
+  recovery at the application layer.
+
+The following sections describe the TCP reliability mechanism.
+
+### Sender state (per session)
+
+- `send_seq`: next seq to assign (starts at 0, increments per DATA frame).
+- `retransmit_buffer`: map of `seq → (payload, timestamp)`, holding all sent
+  but unacked frames.
+- `acked_seq`: highest seq acknowledged by the peer (initially `None`).
+
+On sending DATA:
+1. Assign `seq = send_seq; send_seq += 1`.
+2. Set `ack_seq` to the highest contiguous seq we have received from the peer
+   (our receive side's `deliver_seq`).
+3. Store `(payload, now)` in `retransmit_buffer[seq]`.
+4. Transmit the frame.
+
+On receiving an `ack_seq` in any DATA frame (including pure ACKs):
+1. Advance `acked_seq` to `max(acked_seq, ack_seq)`.
+2. Remove all entries from `retransmit_buffer` with `seq <= ack_seq`.
+
+Retransmit timer (per session, checked periodically):
+1. For each entry in `retransmit_buffer` older than `RETRANSMIT_TIMEOUT`
+   (default 5 ms), retransmit the frame and reset its timestamp.
+2. If any entry has been retransmitted more than `MAX_RETRIES` times (default
+   10), send `CLOSE` and tear down the session.
+
+### Receiver state (per session)
+
+- `expected_seq`: next seq expected (starts at 0).
+- `receive_buffer`: map of `seq → payload`, holding out-of-order frames.
+- `deliver_seq`: highest seq delivered to the local TCP socket (starts at
+  `None`; reported as `ack_seq` in outgoing DATA frames).
+
+On receiving DATA with `seq`:
+1. If `seq < expected_seq`: duplicate (already delivered). Discard the payload,
+   but still process the `ack_seq` field to advance the send window. Send a
+   pure ACK so the sender can converge.
+2. If `seq == expected_seq`: deliver payload to the TCP socket. Increment
+   `expected_seq`. Then check `receive_buffer` for the next contiguous seq and
+   deliver those too (drain the buffer in order). Update `deliver_seq`.
+3. If `seq > expected_seq`: store in `receive_buffer[seq]`. Do not deliver yet.
+   Send a pure ACK (re-ACKing `deliver_seq`) to trigger retransmit of the gap.
+
+### Pure ACK frames
+
+When a side needs to ACK but has no data to send, it sends a DATA frame with
+`payload_len = 0`. The `seq` field is set to `send_seq` (consuming a seq
+number, same as TCP's empty segment) and `ack_seq` carries the cumulative
+acknowledgment. The receiver processes the `ack_seq` and discards the empty
+payload without delivering to the TCP socket.
+
+### Acknowledgment timing
+
+- When data is flowing in both directions, each DATA frame carries the latest
+  `ack_seq` — no separate ACK frames needed.
+- When data is one-directional, the receiver sends a pure ACK after each DATA
+  frame (or after a small batch, implementation-defined).
+- On receiving a duplicate or out-of-order frame, the receiver immediately
+  sends a pure ACK to help the sender converge.
+
+### Why this is not the Two Generals Problem
+
+We do not need mutual consensus. We need one-sided reliable delivery: the
+sender retransmits until it gets an ACK. If the ACK is lost, the sender
+retransmits the data; the receiver sees a duplicate, discards it, and re-ACKs.
+This converges in O(1) round trips. The only unsolvable case — the last frame
+before a permanent link death — is handled by `MAX_RETRIES` → `CLOSE`, which is
+correct: a dead link should kill the session.
+
 ## Payload field encodings
 
 ### MANIFEST payload
 
-Variable-length entries, parsed sequentially until the payload is consumed.
+A hostname prefix followed by variable-length entries, parsed sequentially
+until the payload is consumed.
 
 ```
  0                   1                   2                   3
  0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
-+---------------+---------------+-------------------------------+
++---------------+-----------------------------------------------+
+| hostname_len  | hostname (UTF-8, hostname_len bytes) ...       |
++---------------+-----------------------------------------------+
 | id            | proto         | port (big-endian)             |
 +---------------+---------------+-------------------------------+
 | label_len     | label (UTF-8, label_len bytes) ...           |
 +---------------+-----------------------------------------------+
+| ... repeated ...                                             |
++---------------------------------------------------------------+
 ```
 
+- **hostname_len** (u8): length in bytes of the server's hostname. `0` is valid
+  (unknown hostname).
+- **hostname** (`hostname_len` bytes, UTF-8): the server's hostname, read via
+  `gethostname(2)` at startup. Maximum 255 bytes.
 - **id** (u8): upstream identifier (1-based positional index from `gatunad`
   cmdline).
 - **proto** (u8): `1 = TCP`, `2 = UDP` (reserved; not emitted in v1).
@@ -74,29 +200,35 @@ Variable-length entries, parsed sequentially until the payload is consumed.
 - **label** (`label_len` bytes, UTF-8): human-readable name for the upstream,
   taken from the `PORT[:label]` cmdline argument. Maximum 255 bytes.
 
-To parse: read the 5-byte fixed prefix, then `label_len` bytes, and repeat until
+To parse: read `hostname_len`, then `hostname_len` bytes of hostname, then read
+5-byte fixed entry prefixes + `label_len` bytes of label each, repeating until
 the payload is exhausted. The number of entries is not carried explicitly.
 
 ### OPEN payload
 
 ```
-+---------------+
++---------------+---------------+
-| upstream_id   |
+| upstream_id   | proto         |
-+---------------+
++---------------+---------------+
 ```
 
 - **upstream_id** (u8): which MANIFEST entry to open.
+- **proto** (u8): the transport protocol of the upstream (`1 = TCP`,
+  `2 = UDP`). Carried end-to-end so both sides know which reliability
+  semantics apply to the session. Must match the proto advertised in the
+  MANIFEST for that upstream.
 
 ### OPEN_ACK payload
 
 ```
-+---------------+
++---------------+---------------+
-| upstream_id   |
+| upstream_id   | proto         |
-+---------------+
++---------------+---------------+
 ```
 
-- **upstream_id** (u8): echoes the requested upstream. The session is
+- **upstream_id** (u8): echoes the requested upstream.
-  identified by the `session_id` field in the header, not the payload.
+- **proto** (u8): echoes the requested proto. The session is identified by
+  the `session_id` field in the header, not the payload.
 
 ### OPEN_NAK payload
 
@@ -111,8 +243,10 @@ the payload is exhausted. The number of entries is not carried explicitly.
 
 ### DATA payload
 
-Raw application bytes. Up to 1480 bytes per frame. The `session_id` header
+`[ seq:4 ][ ack_seq:4 ][ raw bytes ]` — the `seq` and `ack_seq` fields are
-field identifies which session the bytes belong to.
+part of the DATA frame's extended header (between `payload_len` and the
+payload). The `payload_len` field counts only the raw bytes, not the seq/ack
+fields. Up to 1480 bytes of application data per frame.
 
 ### CLOSE payload
 
@@ -124,6 +258,18 @@ field identifies which session the bytes belong to.
 
 - **reason** (u8, optional): present iff payload length ≥ 1. See reason codes.
 
+### PING / PONG payload
+
+```
++                                                               +
+| nonce (big-endian, 8 bytes)                                   |
++                                                               +
+```
+
+- **nonce** (u64, big-endian): arbitrary value chosen by the client. The
+  server echoes it verbatim in the PONG reply. Used to correlate RTT
+  measurements.
+
 ## Reason codes
 
 | Value | Meaning           |
@@ -133,6 +279,7 @@ field identifies which session the bytes belong to.
 | 2     | connect_failed    |
 | 3     | oversize          |
 | 4     | unknown_session   |
+| 5     | max_retries       |
 
 ## Discovery flow
 
@@ -165,20 +312,58 @@ client                            server
   |   OPEN_NAK { reason }           |
   |<--------------------------------|   (on failure)
   |                                 |
-  | DATA { session_id, bytes }      |
+  | DATA { seq, ack_seq, bytes }    |
-  |<------------------------------->| DATA { session_id, bytes }
+  |<------------------------------->| DATA { seq, ack_seq, bytes }
   |                                 |
   | CLOSE { session_id, reason? }   |
-  |<------------------------------->|   (on EOF, RST, or error)
+  |<------------------------------->|   (on EOF, RST, or max_retries)
   |                                 |
 ```
 
 - `session_id` is allocated by the server as a monotonically increasing u32
   (starting at 1) from an atomic counter. Collision by wraparound is ignored.
+- `seq` starts at 0 on both sides of each session and increments per DATA
+  frame (including pure ACKs).
 - Either side may send `CLOSE`. The side receiving `CLOSE` tears down its half
   and stops emitting frames for that session.
 - The server's socket→tunnel pump reads `TcpStream` in 1480-byte chunks and
-  emits one `DATA` frame per chunk. On `read` returning 0 (FIN) or an error,
+  emits one DATA frame per chunk. On `read` returning 0 (FIN) or an error,
   it emits `CLOSE` and exits.
-- The server's tunnel→socket path writes `DATA` payloads to the `TcpStream`
+- The server's tunnel→socket path delivers DATA payloads in seq order to the
-  with `write_all`. On error it emits `CLOSE` and drops the session.
+  `TcpStream` with `write_all`. On error it emits `CLOSE` and drops the session.
+- A DATA frame is retransmitted if no ACK is received within 5 ms. After 10
+  failed retransmits, the session is closed with `reason = max_retries`.
+
+## Network test (PING/PONG)
+
+```
+client                            server
+  |                                 |
+  | PING { nonce }                  |
+  |-------------------------------->|
+  |                                 |
+  |        PONG { nonce }           |
+  |<--------------------------------|
+  |                                 |
+  | (repeated at random intervals)  |
+  |                                 |
+```
+
+- The client sends `PING` frames at random 10–100 ms intervals, each with a
+  unique `nonce`.
+- The server echoes the nonce verbatim in a `PONG` frame.
+- The client correlates `PONG` nonces with outstanding `PING` timestamps to
+  compute RTT, average latency, jitter (mean absolute delta of consecutive
+  RTTs), and drop rate (unanswered PINGs).
+- PING/PONG frames use `session_id = 0`; they are independent of TCP sessions.
+
+## Version compatibility
+
+- Version 1: no `seq`/`ack_seq` in DATA frames, no L2 reliability. Dropped
+  DATA frames kill the TCP session.
+- Version 2: DATA frames carry `seq` + `ack_seq`, L2 retransmit. Dropped
+  frames are recovered.
+
+Version 2 is the current version. A receiver that sees `version != 2` rejects
+the frame. Both sides of a session must speak the same version; there is no
+negotiation.

README.md

@@ -1,9 +1,16 @@
 # gatuna
 
-A raw-Ethernet tunnel for reaching services on a peer machine when a WFP
+**gata** (Spanish: *cat*) + **atún** (Spanish: *tuna*) — a cat fishing a
-killswitch on the local machine blocks normal IP traffic. Two programs carry
+tunnel. The "tun" root is also the commonplace shorthand for a network tunnel
-bytes between their respective loopbacks over a private L2 protocol, bypassing
+(as in `/dev/net/tun`, `wintun`, `tuntap`), so the name works in both languages:
-the IP stack (and therefore the killswitch) entirely.
+a Spanish portmanteau that sounds like "gatuna" (feline), and an English pun on
+"cat-tun" — a cat that tunnels.
+
+## Attribution
+
+Entirely vibe-coded with [GLM 5.2](https://github.com/zai-org/GLM-5.2). No
+human-typed source code; all implementation was directed through natural
+language and reviewed by the human in the loop.
 
 ## Why this works
 
@@ -16,33 +23,39 @@ pass unimpeded. See `wireguard-windows/tunnel/firewall/blocker.go:156`.
 
 ## Components
 
-- `gatunad` — Rust server. Runs on the peer (Linux) that owns the real
+- **`gatunad`** — Rust server (`gatunad/`). Runs on the peer (Linux) that owns
-  services. Announces upstreams and relays TCP between the tunnel and
+  the real services. Announces upstreams and relays TCP between the tunnel and
   `127.0.0.1:<port>`.
-- `gatuna` (planned) — .NET WinForms client. Runs on the killswitched Windows
+- **`gatuna`** — .NET 8 WinForms client (`gatuna-win/`). Runs on the
-  box. Discovers the server, presents its upstreams as local loopback
+  killswitched Windows box. Discovers the server, presents its upstreams as
-  listeners, and hauls bytes over the same L2 protocol.
+  local loopback listeners, and hauls bytes over the same L2 protocol. Includes
-
+  a network test (PING/PONG) for measuring latency, jitter, and loss.
-This repository builds `gatunad` first.
 
 ## Transport
 
 - **Medium:** raw Ethernet frames on a shared L2 segment.
 - **Ethertype:** `0x6969` (hardcoded).
-- **No IP stack involvement.** Frames carry only our 6-byte header + payload.
+- **No IP stack involvement.** Frames carry only our 8-byte header + payload.
-- **BPF:** the server attaches a classic BPF filter `ether proto 0x6969` to its
+- **BPF:** both sides filter on ethertype — the server via classic BPF on
-  `AF_PACKET` socket so it only wakes on our ethertype. No eBPF authoring.
+  `AF_PACKET` (`SO_ATTACH_FILTER`), the client via Npcap's compiled filter.
+  No eBPF authoring.
 
 ## Protocol
 
 See [`PROTOCOL.md`](PROTOCOL.md) for the full wire format. Summary:
 
-- 6-byte header, big-endian: `[ version:1 ][ type:1 ][ session_id:4 ][ payload:N ]`.
+- 8-byte header, big-endian:
-- `version` = `1`. No length field (frame length comes from the capture). No
+  `[ version:1 ][ type:1 ][ session_id:4 ][ payload_len:2 ][ payload:N ]`.
-  CRC. Ethertype discriminates our frames from everything else.
+- `version` = `2`. `payload_len` lets the receiver ignore Ethernet padding
-- Discovery: client broadcasts `DISCOVER`; server unicasts `MANIFEST` back.
+  (frames under 60 bytes are zero-padded by the NIC).
+- DATA frames carry an extended 16-byte header with `seq` and `ack_seq` fields
+  for L2-level reliability (retransmit + in-order delivery), preventing lost
+  Ethernet frames from corrupting TCP sessions.
+- Discovery: client broadcasts `DISCOVER`; server unicasts `MANIFEST` (with
+  hostname and upstream list) back.
 - Sessions: `OPEN` → `OPEN_ACK` (or `OPEN_NAK`) → `DATA`* ↔ `DATA`* → `CLOSE`.
-- v1 ships TCP only. UDP frame types are reserved but unimplemented.
+- Network test: `PING` (with 8-byte nonce) → `PONG` (nonce echoed).
+- TCP only. UDP frame types are reserved but unimplemented.
 
 ## `gatunad` usage
 
@@ -62,26 +75,136 @@ sudo ./gatunad eth0 22 8800:site-a 8080:site-b
 Exposes upstreams `1→127.0.0.1:22`, `2→127.0.0.1:8800` (label `site-a`),
 `3→127.0.0.1:8080` (label `site-b`).
 
+## `gatuna` usage
+
+1. Launch `gatuna.exe` (UAC prompt expected — Npcap requires admin).
+2. Select the network adapter connected to the same L2 segment as the server.
+3. Click **Discover**. The server's hostname and MAC appear; the upstream list
+   populates.
+4. Check the upstreams you want to use. The **Mirror** column shows the local
+   loopback port to connect to.
+5. Connect your app to `127.0.0.1:<mirror_port>`.
+6. Click **Test** to run a PING/PONG network test (latency, jitter, loss).
+7. Minimize to tray; close to exit.
+
+### Mirror ports
+
+Mirror ports are deterministic: `port ^ (mac[0]<<8 | mac[5])`, clamped to
+≥1024. The same server + upstream always yields the same local port, so you
+know where to connect without checking the UI. If the computed port is already
+in use, the client falls back to an OS-assigned port.
+
+## Building from source
+
+### Prerequisites
+
+- **Server:** Rust toolchain (edition 2021, MSRV 1.70+), Linux with `AF_PACKET`
+  support.
+- **Client:** .NET 8 SDK with Windows Desktop workload, Npcap installed
+  (bundled with Wireshark or standalone from <https://npcap.com>).
+
+### Build the server (Linux)
+
+```sh
+cd gatunad
+cargo build --release
+```
+
+The binary is at `gatunad/target/release/gatunad`.
+
+### Build the client (Windows)
+
+```powershell
+cd gatuna-win
+dotnet build -c Release
+```
+
+Or run directly:
+```powershell
+dotnet run --project gatuna-win -c Release
+```
+
+### Npcap
+
+The client requires Npcap's `wpcap.dll` and `Packet.dll` in the system path.
+These are installed by the Npcap installer (also bundled with Wireshark). No
+additional NuGet packages or driver installs are needed — SharpPcap talks to
+the existing Npcap installation.
+
 ## Privileges
 
-`AF_PACKET` requires `CAP_NET_RAW`. Run as root, or grant the binary the
+**Server (Linux):** `AF_PACKET` requires `CAP_NET_RAW`. Run as root, or grant
-capability once:
+the binary the capability once:
 ```
 sudo setcap cap_net_raw+ep ./target/release/gatunad
 ```
 
+**Client (Windows):** Npcap requires admin by default (`AdminOnly=1` in
+`HKLM\SYSTEM\CurrentControlSet\Services\npcap\Parameters`). The client's
+`app.manifest` requests `requireAdministrator`, so a UAC prompt appears on
+launch. See the [Npcap docs](https://npcap.com/guide/npcap-dev-guide-1.html)
+for details on the `AdminOnly` flag if you want to change this.
+
 ## Logging
 
-Errors only, to stdout. Normal lifecycle (DISCOVER/OPEN/CLOSE) is silent.
+**Server:** errors only, to stdout. Normal lifecycle (DISCOVER/OPEN/CLOSE) is
+silent.
 
-## v1 limitations
+**Client:** status line in the UI. Errors are not logged to disk.
+
+## Limitations
 
 - TCP only. UDP wire types reserved, code paths stubbed.
-- No retransmit at the L2 layer. A dropped `DATA` frame breaks the TCP session
-  irrecoverably because the localhost socket already ACKed the bytes. Acceptable
-  on a healthy switched link.
 - No auth/crypto. Anyone on the same L2 segment can `DISCOVER` and `OPEN`.
 - Single server instance per interface.
+- One outstanding `OPEN` at a time on the client (serialized via queue).
+- L2 retransmit caps at 10 retries × 5 ms = 50 ms. A permanently dead link
+  closes the session with `reason = max_retries`.
+
+## Repository layout
+
+```
+gatuna/
+├── README.md
+├── PROTOCOL.md
+├── LICENSE                  # CC0 1.0 Universal
+├── .gitignore
+├── gatunad/                 # Rust server
+│   ├── Cargo.toml
+│   └── src/
+│       ├── main.rs          # cmdline, rx dispatch, tx task
+│       ├── link.rs          # AF_PACKET, classic BPF, raw Ethernet I/O
+│       ├── frame.rs         # wire protocol encode/decode
+│       ├── upstream.rs      # cmdline parsing, hostname, upstream table
+│       └── session.rs       # session store, socket→tunnel pump
+└── gatuna-win/              # .NET 8 WinForms client
+    ├── gatuna.csproj
+    ├── app.manifest         # requireAdministrator
+    ├── Program.cs           # tray icon, entry point
+    ├── MainForm.cs          # UI: adapter picker, discover, test, upstream list
+    ├── TunnelLink.cs        # SharpPcap wrapper, raw Ethernet send/recv
+    ├── Frame.cs             # wire protocol encode/decode (mirrors Rust frame.rs)
+    ├── SessionManager.cs    # session lifecycle, listeners, OPEN serialization
+    └── PingTest.cs          # PING/PONG network test with stats
+```
+
+## Versioning
+
+Both programs use `MAJOR.MINOR.PATCH` where:
+
+- **MAJOR** matches the wire protocol version. A v2.x program rejects any frame
+  with `version != 2`. No cross-major compatibility.
+- **MINOR** is the main change indicator for features and fixes within a major
+  version.
+- **PATCH** is reserved for bug fixes; generally unused.
+
+Current version: **2.0.0** (wire protocol v2).
+
+Check versions:
+```sh
+gatunad --version          # Rust server
+```
+The Windows client shows its version in the title bar and tray tooltip.
 
 ## License
 

gatuna-win/Frame.cs

@@ -0,0 +1,247 @@
+namespace gatuna;
+
+using System.Text;
+
+static class Proto
+{
+    public const ushort EtherType = 0x6969;
+    public const int EthHeaderLen = 14;
+    public const byte Version = 2;
+    public const int HeaderLen = 8;
+    public const int DataHeaderLen = 16; // 8 common + 4 seq + 4 ack_seq
+    public const int MaxPayload = 1480;
+
+    public const byte TypeDiscover = 0x01;
+    public const byte TypeManifest = 0x02;
+    public const byte TypeOpen = 0x03;
+    public const byte TypeOpenAck = 0x04;
+    public const byte TypeOpenNak = 0x05;
+    public const byte TypeData = 0x06;
+    public const byte TypeClose = 0x07;
+    public const byte TypePing = 0x0B;
+    public const byte TypePong = 0x0C;
+
+    public const byte ProtoTcp = 1;
+    public const byte ProtoUdp = 2;
+
+    public const byte ReasonUnspecified = 0;
+    public const byte ReasonUnknownUpstream = 1;
+    public const byte ReasonConnectFailed = 2;
+    public const byte ReasonOversize = 3;
+    public const byte ReasonUnknownSession = 4;
+    public const byte ReasonMaxRetries = 5;
+}
+
+readonly record struct UpstreamEntry(
+    byte Id,
+    byte Protocol,
+    ushort Port,
+    string? Label)
+{
+    public string ProtoName => Protocol == Proto.ProtoTcp ? "tcp" : "udp";
+}
+
+abstract record Frame
+{
+    internal record Discover : Frame;
+    internal record Manifest(string Hostname, UpstreamEntry[] Entries) : Frame;
+    internal record Open(byte UpstreamId, byte Proto) : Frame;
+    internal record OpenAck(uint SessionId, byte UpstreamId, byte Proto) : Frame;
+    internal record OpenNak(byte UpstreamId, byte Reason) : Frame;
+    internal record Data(uint SessionId, uint Seq, uint AckSeq, byte[] Payload) : Frame;
+    internal record Close(uint SessionId, byte? Reason) : Frame;
+    internal record Ping(ulong Nonce) : Frame;
+    internal record Pong(ulong Nonce) : Frame;
+}
+
+static class FrameCodec
+{
+    /// Build a non-DATA frame: 8-byte common header + payload.
+    static byte[] Build(byte type, uint sessionId, byte[] payload)
+    {
+        var buf = new byte[Proto.HeaderLen + payload.Length];
+        buf[0] = Proto.Version;
+        buf[1] = type;
+        buf[2] = (byte)(sessionId >> 24);
+        buf[3] = (byte)(sessionId >> 16);
+        buf[4] = (byte)(sessionId >> 8);
+        buf[5] = (byte)(sessionId & 0xFF);
+        buf[6] = (byte)(payload.Length >> 8);
+        buf[7] = (byte)(payload.Length & 0xFF);
+        Buffer.BlockCopy(payload, 0, buf, Proto.HeaderLen, payload.Length);
+        return buf;
+    }
+
+    /// Build a DATA frame: 8-byte common header + seq + ack_seq + payload.
+    /// payload_len counts only the raw bytes, not seq/ack_seq.
+    static byte[] BuildData(uint sessionId, uint seq, uint ackSeq, byte[] payload)
+    {
+        var buf = new byte[Proto.DataHeaderLen + payload.Length];
+        buf[0] = Proto.Version;
+        buf[1] = Proto.TypeData;
+        buf[2] = (byte)(sessionId >> 24);
+        buf[3] = (byte)(sessionId >> 16);
+        buf[4] = (byte)(sessionId >> 8);
+        buf[5] = (byte)(sessionId & 0xFF);
+        buf[6] = (byte)(payload.Length >> 8);
+        buf[7] = (byte)(payload.Length & 0xFF);
+        buf[8] = (byte)(seq >> 24);
+        buf[9] = (byte)(seq >> 16);
+        buf[10] = (byte)(seq >> 8);
+        buf[11] = (byte)(seq & 0xFF);
+        buf[12] = (byte)(ackSeq >> 24);
+        buf[13] = (byte)(ackSeq >> 16);
+        buf[14] = (byte)(ackSeq >> 8);
+        buf[15] = (byte)(ackSeq & 0xFF);
+        Buffer.BlockCopy(payload, 0, buf, Proto.DataHeaderLen, payload.Length);
+        return buf;
+    }
+
+    public static byte[] Encode(Frame frame)
+    {
+        return frame switch
+        {
+            Frame.Discover =>
+                Build(Proto.TypeDiscover, 0, []),
+            Frame.Manifest manifest =>
+                Build(Proto.TypeManifest, 0, BuildManifestPayload(manifest.Hostname, manifest.Entries)),
+            Frame.Open open =>
+                Build(Proto.TypeOpen, 0, [open.UpstreamId, open.Proto]),
+            Frame.OpenAck ack =>
+                Build(Proto.TypeOpenAck, ack.SessionId, [ack.UpstreamId, ack.Proto]),
+            Frame.OpenNak nak =>
+                Build(Proto.TypeOpenNak, 0, [nak.UpstreamId, nak.Reason]),
+            Frame.Data data =>
+                BuildData(data.SessionId, data.Seq, data.AckSeq, data.Payload),
+            Frame.Close close =>
+                Build(Proto.TypeClose, close.SessionId,
+                    close.Reason.HasValue ? [close.Reason.Value] : []),
+            Frame.Ping ping =>
+                Build(Proto.TypePing, 0, EncodeNonce(ping.Nonce)),
+            Frame.Pong pong =>
+                Build(Proto.TypePong, 0, EncodeNonce(pong.Nonce)),
+            _ => throw new InvalidOperationException($"unknown frame type: {frame.GetType()}"),
+        };
+    }
+
+    static byte[] BuildManifestPayload(string hostname, UpstreamEntry[] entries)
+    {
+        using var ms = new MemoryStream();
+        var hnBytes = Encoding.UTF8.GetBytes(hostname);
+        var hnLen = (byte)Math.Min(hnBytes.Length, 255);
+        ms.WriteByte(hnLen);
+        if (hnLen > 0)
+            ms.Write(hnBytes, 0, hnLen);
+        foreach (var e in entries)
+        {
+            var labelBytes = Encoding.UTF8.GetBytes(e.Label ?? "");
+            var labelLen = (byte)Math.Min(labelBytes.Length, 255);
+            ms.WriteByte(e.Id);
+            ms.WriteByte(e.Protocol);
+            ms.WriteByte((byte)(e.Port >> 8));
+            ms.WriteByte((byte)(e.Port & 0xFF));
+            ms.WriteByte(labelLen);
+            if (labelLen > 0)
+                ms.Write(labelBytes, 0, labelLen);
+        }
+        return ms.ToArray();
+    }
+
+    public static Frame? Parse(ReadOnlySpan<byte> buf)
+    {
+        if (buf.Length < Proto.HeaderLen)
+            return null;
+        if (buf[0] != Proto.Version)
+            return null;
+        var type = buf[1];
+        var sessionId = (uint)(buf[2] << 24 | buf[3] << 16 | buf[4] << 8 | buf[5]);
+        var payloadLen = (ushort)(buf[6] << 8 | buf[7]);
+
+        // DATA frames have seq + ack_seq after the common header.
+        if (type == Proto.TypeData)
+        {
+            if (buf.Length < Proto.DataHeaderLen + payloadLen)
+                return null;
+            var seq = (uint)(buf[8] << 24 | buf[9] << 16 | buf[10] << 8 | buf[11]);
+            var ackSeq = (uint)(buf[12] << 24 | buf[13] << 16 | buf[14] << 8 | buf[15]);
+            var payload = buf.Slice(Proto.DataHeaderLen, payloadLen);
+            if (payload.Length > Proto.MaxPayload)
+                return null;
+            return new Frame.Data(sessionId, seq, ackSeq, payload.ToArray());
+        }
+
+        if (buf.Length < Proto.HeaderLen + payloadLen)
+            return null;
+        var payload2 = buf.Slice(Proto.HeaderLen, payloadLen);
+
+        return type switch
+        {
+            Proto.TypeManifest => ParseManifest(payload2),
+            Proto.TypeOpenAck when payload2.Length == 2 =>
+                new Frame.OpenAck(sessionId, payload2[0], payload2[1]),
+            Proto.TypeOpenNak when payload2.Length == 2 =>
+                new Frame.OpenNak(payload2[0], payload2[1]),
+            Proto.TypeClose when payload2.Length is 0 or 1 =>
+                new Frame.Close(sessionId, payload2.Length == 1 ? payload2[0] : null),
+            Proto.TypePong when payload2.Length == 8 =>
+                new Frame.Pong(ParseNonce(payload2)),
+            Proto.TypePing when payload2.Length == 8 =>
+                new Frame.Ping(ParseNonce(payload2)),
+            Proto.TypeDiscover when payload2.Length == 0 =>
+                new Frame.Discover(),
+            _ => null,
+        };
+    }
+
+    static ulong ParseNonce(ReadOnlySpan<byte> payload)
+    {
+        ulong nonce = 0;
+        for (int i = 0; i < 8; i++)
+            nonce = (nonce << 8) | payload[i];
+        return nonce;
+    }
+
+    static byte[] EncodeNonce(ulong nonce)
+    {
+        return [
+            (byte)(nonce >> 56), (byte)(nonce >> 48),
+            (byte)(nonce >> 40), (byte)(nonce >> 32),
+            (byte)(nonce >> 24), (byte)(nonce >> 16),
+            (byte)(nonce >> 8),  (byte)(nonce & 0xFF),
+        ];
+    }
+
+    static Frame.Manifest? ParseManifest(ReadOnlySpan<byte> payload)
+    {
+        if (payload.Length < 1)
+            return null;
+        var hnLen = payload[0];
+        if (1 + hnLen > payload.Length)
+            return null;
+        var hostname = hnLen == 0
+            ? ""
+            : Encoding.UTF8.GetString(payload[1..(1 + hnLen)]);
+        var rest = payload[(1 + hnLen)..];
+
+        var entries = new List<UpstreamEntry>();
+        int i = 0;
+        while (i < rest.Length)
+        {
+            if (i + 5 > rest.Length)
+                return null;
+            var id = rest[i];
+            var proto = rest[i + 1];
+            var port = (ushort)(rest[i + 2] << 8 | rest[i + 3]);
+            var labelLen = rest[i + 4];
+            i += 5;
+            if (i + labelLen > rest.Length)
+                return null;
+            string? label = labelLen == 0
+                ? null
+                : Encoding.UTF8.GetString(rest[i..(i + labelLen)]);
+            i += labelLen;
+            entries.Add(new UpstreamEntry(id, proto, port, label));
+        }
+        return new Frame.Manifest(hostname, entries.ToArray());
+    }
+}

gatuna-win/MainForm.cs

@@ -0,0 +1,214 @@
+using SharpPcap.LibPcap;
+
+namespace gatuna;
+
+public partial class MainForm : Form
+{
+    readonly SessionManager _sessions = new();
+    readonly ComboBox _deviceBox = new();
+    readonly Button _discoverBtn = new();
+    readonly Button _testBtn = new();
+    readonly Label _serverLabel = new();
+    readonly Label _pingStatsLabel = new();
+    readonly ListView _listView = new();
+    readonly Label _statusLabel = new();
+    TunnelLink? _link;
+    PingTest? _pingTest;
+
+    public MainForm()
+    {
+        var ver = System.Reflection.Assembly.GetExecutingAssembly().GetName().Version!;
+        Text = $"gatuna {ver.Major}.{ver.Minor}";
+        Width = 520;
+        Height = 420;
+        FormBorderStyle = FormBorderStyle.FixedSingle;
+        MaximizeBox = false;
+        StartPosition = FormStartPosition.CenterScreen;
+        Icon = SystemIcons.GetStockIcon(StockIconId.NetworkConnect, 32);
+        InitializeComponents();
+
+        _sessions.Log += msg => this.Invoke(() => _statusLabel.Text = msg);
+        _sessions.ManifestReceived += (hostname, mac, entries) =>
+            this.Invoke(() => PopulateList(hostname, mac, entries));
+
+        foreach (var d in TunnelLink.ListDevices())
+        {
+            var friendly = d.Interface?.FriendlyName ?? "";
+            var desc = d.Interface?.Description ?? "";
+            var mac = d.MacAddress?.GetAddressBytes();
+            var macStr = mac != null
+                ? string.Join(":", mac.Select(b => b.ToString("X2")))
+                : "??-??-??-??-??-??";
+            _deviceBox.Items.Add($"{friendly} — {desc} — {macStr}");
+        }
+        if (_deviceBox.Items.Count > 0)
+            _deviceBox.SelectedIndex = 0;
+    }
+
+    void InitializeComponents()
+    {
+        const int pad = 12;
+
+        _deviceBox.Left = pad; _deviceBox.Top = 12;
+        _deviceBox.Width = ClientSize.Width - pad * 2;
+        _deviceBox.DropDownStyle = ComboBoxStyle.DropDownList;
+        Controls.Add(_deviceBox);
+
+        _discoverBtn.Text = "Discover";
+        _discoverBtn.Left = pad; _discoverBtn.Top = _deviceBox.Bottom + 8;
+        _discoverBtn.Width = 80;
+        _discoverBtn.Click += OnDiscover;
+        Controls.Add(_discoverBtn);
+
+        _testBtn.Text = "Test";
+        _testBtn.Left = _discoverBtn.Right + 8; _testBtn.Top = _discoverBtn.Top;
+        _testBtn.Width = 60;
+        _testBtn.Click += OnTest;
+        Controls.Add(_testBtn);
+
+        _serverLabel.Left = pad; _serverLabel.Top = _discoverBtn.Bottom + 8;
+        _serverLabel.Width = ClientSize.Width - pad * 2;
+        _serverLabel.AutoEllipsis = true;
+        _serverLabel.Text = "Server: not connected";
+        Controls.Add(_serverLabel);
+
+        _pingStatsLabel.Left = pad; _pingStatsLabel.Top = _serverLabel.Bottom + 4;
+        _pingStatsLabel.Width = ClientSize.Width - pad * 2;
+        _pingStatsLabel.AutoEllipsis = true;
+        _pingStatsLabel.Text = "";
+        Controls.Add(_pingStatsLabel);
+
+        _listView.Left = pad; _listView.Top = _pingStatsLabel.Bottom + 8;
+        _listView.Width = ClientSize.Width - pad * 2;
+        _listView.Height = 180;
+        _listView.View = View.Details;
+        _listView.FullRowSelect = true;
+        _listView.CheckBoxes = true;
+        _listView.Columns.Add("ID", 36);
+        _listView.Columns.Add("Proto", 50);
+        _listView.Columns.Add("Port", 56);
+        _listView.Columns.Add("Label", 160);
+        _listView.Columns.Add("Mirror", 70);
+        _listView.ItemChecked += OnItemChecked;
+        Controls.Add(_listView);
+
+        _statusLabel.Left = pad; _statusLabel.Top = _listView.Bottom + 8;
+        _statusLabel.Width = ClientSize.Width - pad * 2;
+        _statusLabel.AutoEllipsis = true;
+        Controls.Add(_statusLabel);
+    }
+
+    void OnDiscover(object? s, EventArgs e)
+    {
+        if (_deviceBox.SelectedIndex < 0)
+        {
+            MessageBox.Show("Select a network adapter first.");
+            return;
+        }
+
+        if (_link != null)
+        {
+            _sessions.StopAll();
+            _sessions.DetachLink();
+            _link.Dispose();
+        }
+
+        var devices = TunnelLink.ListDevices();
+        var device = devices[_deviceBox.SelectedIndex];
+        _link = new TunnelLink(device);
+        _link.Log += msg => this.Invoke(() => _statusLabel.Text = msg);
+        _sessions.AttachLink(_link);
+        _link.Open();
+        _sessions.Discover();
+        _serverLabel.Text = "Server: discovering...";
+        _statusLabel.Text = "discovering...";
+    }
+
+    void OnTest(object? s, EventArgs e)
+    {
+        if (_pingTest != null && _pingTest.Running)
+        {
+            _sessions.StopPing();
+            _pingTest = null;
+            _testBtn.Text = "Test";
+            return;
+        }
+
+        _pingTest = _sessions.StartPing();
+        if (_pingTest == null)
+        {
+            MessageBox.Show("Discover a server first.");
+            return;
+        }
+        _pingTest.StatsUpdated += stats => this.Invoke(() =>
+        {
+            _pingStatsLabel.Text =
+                $"sent: {stats.Sent}  recv: {stats.Received}  " +
+                $"loss: {stats.LossPct:F1}%  " +
+                $"avg: {stats.AvgLatencyMs:F1}ms  " +
+                $"jitter: {stats.JitterMs:F1}ms";
+        });
+        _testBtn.Text = "Stop";
+        _pingStatsLabel.Text = "pinging...";
+    }
+
+    void PopulateList(string hostname, byte[] mac, UpstreamEntry[] entries)
+    {
+        var macStr = string.Join(":", mac.Select(b => b.ToString("X2")));
+        _serverLabel.Text = $"Server: {hostname} — {macStr}";
+
+        _listView.BeginUpdate();
+        _listView.Items.Clear();
+        foreach (var up in entries)
+        {
+            var item = new ListViewItem(up.Id.ToString());
+            item.SubItems.Add(up.ProtoName);
+            item.SubItems.Add(up.Port.ToString());
+            item.SubItems.Add(up.Label ?? "");
+            item.SubItems.Add("");
+            item.Tag = up;
+            _listView.Items.Add(item);
+        }
+        _listView.EndUpdate();
+    }
+
+    void OnItemChecked(object? s, ItemCheckedEventArgs e)
+    {
+        if (e.Item.Tag is not UpstreamEntry up) return;
+        if (e.Item.Checked)
+        {
+            var port = _sessions.StartListener(up);
+            e.Item.SubItems[4].Text = port.ToString();
+        }
+        else
+        {
+            if (int.TryParse(e.Item.SubItems[4].Text, out var port) && port > 0)
+            {
+                _sessions.StopListener(port);
+                e.Item.SubItems[4].Text = "";
+            }
+        }
+    }
+
+    protected override void OnResize(EventArgs e)
+    {
+        base.OnResize(e);
+        if (WindowState == FormWindowState.Minimized)
+        {
+            Hide();
+            WindowState = FormWindowState.Normal;
+        }
+    }
+
+    protected override void OnFormClosing(FormClosingEventArgs e)
+    {
+        Shutdown();
+        base.OnFormClosing(e);
+    }
+
+    public void Shutdown()
+    {
+        _sessions.Dispose();
+        _link?.Dispose();
+    }
+}

gatuna-win/PingTest.cs

@@ -0,0 +1,101 @@
+using System.Collections.Concurrent;
+
+namespace gatuna;
+
+sealed class PingTest
+{
+    readonly TunnelLink _link;
+    readonly byte[] _serverMac;
+    readonly CancellationTokenSource _cts = new();
+    readonly ConcurrentDictionary<ulong, long> _outstanding = new();
+    readonly ConcurrentQueue<double> _rtts = new();
+    long _sent;
+    long _received;
+    double _lastRttMs;
+    double _jitterSum;
+    long _jitterCount;
+
+    public event Action<PingStats>? StatsUpdated;
+    public event Action<string>? Log;
+
+    public bool Running { get; private set; }
+
+    public PingTest(TunnelLink link, byte[] serverMac)
+    {
+        _link = link;
+        _serverMac = serverMac;
+    }
+
+    public void Start()
+    {
+        Running = true;
+        _ = RunLoop();
+    }
+
+    public void Stop()
+    {
+        Running = false;
+        _cts.Cancel();
+    }
+
+    async Task RunLoop()
+    {
+        var rng = new Random();
+        var timer = new PeriodicTimer(TimeSpan.FromMilliseconds(10));
+        while (!_cts.IsCancellationRequested)
+        {
+            var nonce = (ulong)Interlocked.Increment(ref _sent);
+            var ticks = DateTime.UtcNow.Ticks;
+            _outstanding[nonce] = ticks;
+
+            _link.SendTo(_serverMac, new Frame.Ping(nonce));
+
+            var interval = rng.Next(10, 101);
+            try
+            {
+                await Task.Delay(interval, _cts.Token);
+            }
+            catch { break; }
+        }
+        Running = false;
+    }
+
+    public void HandlePong(ulong nonce)
+    {
+        if (_outstanding.TryRemove(nonce, out var sentTicks))
+        {
+            var rttMs = (DateTime.UtcNow.Ticks - sentTicks) / (double)TimeSpan.TicksPerMillisecond;
+            _rtts.Enqueue(rttMs);
+            Interlocked.Increment(ref _received);
+
+            if (_jitterCount > 0)
+            {
+                _jitterSum += Math.Abs(rttMs - _lastRttMs);
+            }
+            _lastRttMs = rttMs;
+            Interlocked.Increment(ref _jitterCount);
+
+            EmitStats();
+        }
+    }
+
+    void EmitStats()
+    {
+        var sent = Interlocked.Read(ref _sent);
+        var recv = Interlocked.Read(ref _received);
+        var loss = sent > 0 ? (1.0 - (double)recv / sent) * 100.0 : 0;
+
+        var rttList = _rtts.ToArray();
+        var avg = rttList.Length > 0 ? rttList.Average() : 0;
+        var jitter = _jitterCount > 1 ? _jitterSum / (_jitterCount - 1) : 0;
+
+        StatsUpdated?.Invoke(new PingStats(sent, recv, loss, avg, jitter));
+    }
+}
+
+readonly record struct PingStats(
+    long Sent,
+    long Received,
+    double LossPct,
+    double AvgLatencyMs,
+    double JitterMs);

gatuna-win/Program.cs

@@ -1,4 +1,4 @@
-namespace gatuna_client;
+namespace gatuna;
 
 static class Program
 {
@@ -9,10 +9,12 @@ static class Program
 
         var form = new MainForm();
 
+        var ver = System.Reflection.Assembly.GetExecutingAssembly().GetName().Version!;
+
         using var tray = new NotifyIcon
         {
-            Icon = SystemIcons.Application,
+            Icon = SystemIcons.GetStockIcon(StockIconId.NetworkConnect, 32),
-            Text = "gatuna",
+            Text = $"gatuna {ver.Major}.{ver.Minor}",
             Visible = true,
         };
 
@@ -20,18 +22,19 @@ static class Program
         tray.ContextMenuStrip.Items.Add("Show", null, (_, _) =>
         {
             form.Show();
+            form.WindowState = FormWindowState.Normal;
             form.Activate();
         });
         tray.ContextMenuStrip.Items.Add("-");
         tray.ContextMenuStrip.Items.Add("Exit", null, (_, _) =>
         {
-            form.Shutdown();
             tray.Visible = false;
-            Application.Exit();
+            form.Close();
         });
         tray.DoubleClick += (_, _) =>
         {
             form.Show();
+            form.WindowState = FormWindowState.Normal;
             form.Activate();
         };
 

gatuna-win/SessionManager.cs

@@ -0,0 +1,586 @@
+using System.Collections.Concurrent;
+using System.Net;
+using System.Net.Sockets;
+using System.Threading.Channels;
+
+namespace gatuna;
+
+sealed class SessionManager : IDisposable
+{
+    TunnelLink? _link;
+    readonly ConcurrentDictionary<uint, Session> _sessions = new();
+    readonly ConcurrentDictionary<int, ListenerState> _listeners = new();
+
+    byte[]? _serverMac;
+    string _serverHostname = "";
+    UpstreamEntry[] _upstreams = [];
+    PingTest? _pingTest;
+
+    // Serialized OPEN: only one outstanding at a time.
+    readonly object _openLock = new();
+    PendingOpen? _pending;
+    readonly Queue<PendingOpen> _openQueue = new();
+
+    public event Action<string>? Log;
+    public event Action<string, byte[], UpstreamEntry[]>? ManifestReceived;
+
+    public UpstreamEntry[] Upstreams => _upstreams;
+    public byte[]? ServerMac => _serverMac;
+    public string ServerHostname => _serverHostname;
+    public TunnelLink? Link => _link;
+
+    public void AttachLink(TunnelLink link)
+    {
+        _link = link;
+        link.FrameReceived += HandleFrame;
+    }
+
+    public void DetachLink()
+    {
+        if (_link != null)
+            _link.FrameReceived -= HandleFrame;
+        _link = null;
+    }
+
+    public void Discover()
+    {
+        if (_link == null) return;
+        _link.SendBroadcast(new Frame.Discover());
+    }
+
+    public PingTest? StartPing()
+    {
+        if (_link == null || _serverMac == null)
+            return null;
+        _pingTest?.Stop();
+        _pingTest = new PingTest(_link, _serverMac);
+        _pingTest.Start();
+        return _pingTest;
+    }
+
+    public void StopPing()
+    {
+        _pingTest?.Stop();
+        _pingTest = null;
+    }
+
+    public void HandleFrame(Frame frame, byte[] srcMac)
+    {
+        switch (frame)
+        {
+            case Frame.Manifest manifest:
+                _serverMac = srcMac;
+                _serverHostname = manifest.Hostname;
+                _upstreams = manifest.Entries;
+                Log?.Invoke($"manifest: {manifest.Entries.Length} upstreams from {manifest.Hostname} ({BitConverter.ToString(srcMac)})");
+                ManifestReceived?.Invoke(manifest.Hostname, srcMac, manifest.Entries);
+                break;
+
+            case Frame.OpenAck ack:
+                HandleOpenAck(ack, srcMac);
+                break;
+
+            case Frame.OpenNak nak:
+                Log?.Invoke($"OPEN_NAK upstream {nak.UpstreamId} reason {nak.Reason}");
+                lock (_openLock)
+                {
+                    if (_pending != null)
+                        NetUtil.RstClose(_pending.Client);
+                }
+                ProcessQueue();
+                break;
+
+            case Frame.Data data:
+                if (_sessions.TryGetValue(data.SessionId, out var session))
+                    session.HandleData(data);
+                else if (_link != null && _serverMac != null)
+                    _link.SendTo(_serverMac,
+                        new Frame.Close(data.SessionId, Proto.ReasonUnknownSession));
+                break;
+
+            case Frame.Close close:
+                if (_sessions.TryRemove(close.SessionId, out var s))
+                    s.OnRemoteClose();
+                break;
+
+            case Frame.Pong pong:
+                _pingTest?.HandlePong(pong.Nonce);
+                break;
+
+            case Frame.Ping _:
+                break;
+        }
+    }
+
+    /// <summary>
+    /// Compute a deterministic mirror port from the server MAC and the
+    /// upstream port. XOR the upstream port with (mac[0]<<8 | mac[5]),
+    /// then ensure the result is outside the privileged range.
+    /// </summary>
+    static ushort ComputeMirrorPort(byte[] serverMac, ushort upstreamPort)
+    {
+        var k = (ushort)((serverMac[0] << 8) | serverMac[5]);
+        var port = (ushort)(upstreamPort ^ k);
+        if (port < 1024)
+            port += 1024;
+        return port;
+    }
+
+    /// <summary>
+    /// Start a local TCP listener for the given upstream. Returns the mirror
+    /// port, or 0 on failure.
+    /// </summary>
+    public int StartListener(UpstreamEntry upstream)
+    {
+        if (_serverMac == null)
+            return 0;
+
+        var preferred = ComputeMirrorPort(_serverMac, upstream.Port);
+
+        // Try the deterministic port first; fall back to OS assignment.
+        TcpListener listener;
+        int port;
+        try
+        {
+            listener = new TcpListener(IPAddress.Loopback, preferred);
+            listener.Start();
+            port = ((IPEndPoint)listener.LocalEndpoint).Port;
+        }
+        catch
+        {
+            listener = new TcpListener(IPAddress.Loopback, 0);
+            listener.Start();
+            port = ((IPEndPoint)listener.LocalEndpoint).Port;
+            Log?.Invoke($"port {preferred} in use, fell back to {port}");
+        }
+
+        var state = new ListenerState(listener, upstream);
+        _listeners[port] = state;
+        _ = AcceptLoop(state);
+        Log?.Invoke($"listening 127.0.0.1:{port} -> upstream {upstream.Id} ({upstream.ProtoName}:{upstream.Port})");
+        return port;
+    }
+
+    public void StopListener(int port)
+    {
+        if (_listeners.TryRemove(port, out var state))
+        {
+            state.Listener.Stop();
+            Log?.Invoke($"stopped listener port {port}");
+        }
+    }
+
+    async Task AcceptLoop(ListenerState state)
+    {
+        while (true)
+        {
+            TcpClient client;
+            try
+            {
+                client = await state.Listener.AcceptTcpClientAsync();
+            }
+            catch { break; }
+            EnqueueOpen(client, state.Upstream.Id, state.Upstream.Protocol);
+        }
+    }
+
+    void EnqueueOpen(TcpClient client, byte upstreamId, byte proto)
+    {
+        lock (_openLock)
+        {
+            if (_pending == null)
+            {
+                _pending = new PendingOpen(client, upstreamId, proto);
+                SendOpen(_pending);
+            }
+            else
+            {
+                _openQueue.Enqueue(new PendingOpen(client, upstreamId, proto));
+            }
+        }
+    }
+
+    void SendOpen(PendingOpen po)
+    {
+        if (_link == null || _serverMac == null)
+        {
+            Log?.Invoke("no server; cannot OPEN");
+            po.Client.Dispose();
+            return;
+        }
+        _link.SendTo(_serverMac, new Frame.Open(po.UpstreamId, po.Proto));
+    }
+
+    void ProcessQueue()
+    {
+        lock (_openLock)
+        {
+            if (_openQueue.Count > 0)
+            {
+                _pending = _openQueue.Dequeue();
+                SendOpen(_pending);
+            }
+            else
+            {
+                _pending = null;
+            }
+        }
+    }
+
+    void HandleOpenAck(Frame.OpenAck ack, byte[] srcMac)
+    {
+        PendingOpen? po;
+        lock (_openLock)
+            po = _pending;
+
+        if (po == null || po.UpstreamId != ack.UpstreamId)
+        {
+            Log?.Invoke($"OPEN_ACK upstream {ack.UpstreamId} session {ack.SessionId} — no matching pending");
+            return;
+        }
+
+        var session = new Session(
+            ack.SessionId, ack.Proto, po.Client, srcMac, _link!,
+            () => _sessions.TryRemove(ack.SessionId, out _),
+            msg => Log?.Invoke(msg));
+        _sessions[ack.SessionId] = session;
+        session.Start();
+        Log?.Invoke($"session {ack.SessionId} upstream {ack.UpstreamId} established");
+        ProcessQueue();
+    }
+
+    public void StopAll()
+    {
+        StopPing();
+        foreach (var kv in _listeners)
+            kv.Value.Listener.Stop();
+        _listeners.Clear();
+        foreach (var s in _sessions.Values)
+            s.Dispose();
+        _sessions.Clear();
+    }
+
+    public void Dispose()
+    {
+        DetachLink();
+        StopAll();
+    }
+}
+
+sealed class ListenerState(TcpListener listener, UpstreamEntry upstream)
+{
+    public TcpListener Listener { get; } = listener;
+    public UpstreamEntry Upstream { get; } = upstream;
+}
+
+sealed class PendingOpen(TcpClient client, byte upstreamId, byte proto)
+{
+    public TcpClient Client { get; } = client;
+    public byte UpstreamId { get; } = upstreamId;
+    public byte Proto { get; } = proto;
+}
+
+/// L2 reliability: sender-side state.
+class SendState
+{
+    public uint SendSeq;
+    public uint AckedSeq;
+    // seq -> (frame_bytes, send_time_ticks, retry_count)
+    public readonly SortedList<uint, (byte[], long, uint)> RetransmitBuffer = new();
+
+    public uint NextSeq()
+    {
+        var s = SendSeq;
+        SendSeq++;
+        return s;
+    }
+
+    public void RecordSent(uint seq, byte[] frameBytes)
+    {
+        RetransmitBuffer[seq] = (frameBytes, Environment.TickCount64, 0);
+    }
+
+    public void ProcessAck(uint ackSeq)
+    {
+        var toRemove = RetransmitBuffer.Keys
+            .Where(k => k <= ackSeq)
+            .ToList();
+        foreach (var k in toRemove)
+            RetransmitBuffer.Remove(k);
+        if (ackSeq > AckedSeq)
+            AckedSeq = ackSeq;
+    }
+
+    /// Returns frames to retransmit and whether the session should close.
+    public (List<byte[]> resend, bool shouldClose) CheckRetransmit()
+    {
+        var resend = new List<byte[]>();
+        var shouldClose = false;
+        var now = Environment.TickCount64;
+        const int timeoutMs = 5;
+        const uint maxRetries = 10;
+
+        foreach (var kv in RetransmitBuffer.ToList())
+        {
+            if (now - kv.Value.Item2 > timeoutMs)
+            {
+                if (kv.Value.Item3 >= maxRetries)
+                {
+                    shouldClose = true;
+                    break;
+                }
+                RetransmitBuffer[kv.Key] = (kv.Value.Item1, now, kv.Value.Item3 + 1);
+                resend.Add(kv.Value.Item1);
+            }
+        }
+        return (resend, shouldClose);
+    }
+}
+
+/// L2 reliability: receiver-side state.
+class RecvState
+{
+    public uint ExpectedSeq;
+    public uint DeliverSeq;
+    // seq -> payload (out-of-order buffer)
+    public readonly SortedList<uint, byte[]> ReceiveBuffer = new();
+
+    /// Process an incoming DATA frame. Returns (payloads to deliver, needAck).
+    public (List<byte[]> deliver, bool needAck) ProcessData(uint seq, byte[] payload)
+    {
+        if (seq < ExpectedSeq)
+        {
+            // Duplicate.
+            return ([], true);
+        }
+        if (seq == ExpectedSeq)
+        {
+            // In-order: deliver and drain buffer.
+            var deliver = new List<byte[]> { payload };
+            ExpectedSeq++;
+            DeliverSeq = ExpectedSeq - 1;
+            while (ReceiveBuffer.Remove(ExpectedSeq, out var buffered))
+            {
+                deliver.Add(buffered);
+                ExpectedSeq++;
+                DeliverSeq = ExpectedSeq - 1;
+            }
+            return (deliver, false);
+        }
+        else
+        {
+            // Out-of-order: buffer.
+            ReceiveBuffer[seq] = payload;
+            return ([], true);
+        }
+    }
+
+    /// The ack_seq to report in outgoing DATA frames.
+    public uint CurrentAckSeq => ExpectedSeq == 0 ? uint.MaxValue : ExpectedSeq - 1;
+}
+
+sealed class Session(
+    uint sessionId,
+    byte proto,
+    TcpClient client,
+    byte[] serverMac,
+    TunnelLink link,
+    Action onClosed,
+    Action<string>? log) : IDisposable
+{
+    readonly CancellationTokenSource _cts = new();
+    readonly bool _isTcp = proto == Proto.ProtoTcp;
+    readonly SendState _send = new();
+    readonly RecvState _recv = new();
+    readonly object _sendLock = new();
+    readonly object _recvLock = new();
+    volatile bool _closeSent;
+
+    public void Start()
+    {
+        _ = PumpSocketToTunnel();
+        _ = PumpTunnelToSocket();
+        if (_isTcp)
+            _ = RetransmitTimer();
+    }
+
+    /// <summary>
+    /// Called when a CLOSE frame arrives from the server. The server has
+    /// already torn down its side — we just need to flush queued data to
+    /// the local socket and close gracefully. Do NOT echo CLOSE back.
+    /// </summary>
+    public void OnRemoteClose()
+    {
+        _cts.Cancel();
+        _deliverChannel.Writer.TryComplete();
+    }
+
+    /// Handle a DATA frame from the tunnel.
+    public void HandleData(Frame.Data data)
+    {
+        if (!_isTcp)
+        {
+            // Best-effort: deliver directly, ignore seq/ack.
+            if (data.Payload.Length > 0)
+                _deliverChannel.Writer.TryWrite(data.Payload);
+            return;
+        }
+
+        // TCP: process ack_seq to advance send window.
+        lock (_sendLock)
+            _send.ProcessAck(data.AckSeq);
+
+        // Process seq for in-order delivery.
+        List<byte[]> deliver;
+        bool needAck;
+        lock (_recvLock)
+            (deliver, needAck) = _recv.ProcessData(data.Seq, data.Payload);
+
+        // Deliver to the TCP socket via the channel.
+        foreach (var chunk in deliver)
+        {
+            if (!_deliverChannel.Writer.TryWrite(chunk))
+                log?.Invoke($"session {sessionId}: deliver channel full");
+        }
+
+        // Send pure ACK if duplicate or out-of-order.
+        if (needAck)
+            SendPureAck();
+    }
+
+    readonly Channel<byte[]> _deliverChannel = Channel.CreateBounded<byte[]>(256);
+
+    async Task PumpSocketToTunnel()
+    {
+        var buf = new byte[Proto.MaxPayload];
+        try
+        {
+            var stream = client.GetStream();
+            using var reg = _cts.Token.Register(() => client.Dispose());
+            while (!_cts.IsCancellationRequested)
+            {
+                var n = await stream.ReadAsync(buf, _cts.Token);
+                if (n == 0) break;
+
+                uint seq = 0, ackSeq = 0;
+                byte[]? frameBytes = null;
+
+                if (_isTcp)
+                {
+                    lock (_sendLock)
+                    {
+                        seq = _send.NextSeq();
+                        lock (_recvLock)
+                            ackSeq = _recv.CurrentAckSeq;
+                    }
+                }
+
+                var frame = new Frame.Data(sessionId, seq, ackSeq, buf[..n]);
+
+                if (_isTcp)
+                {
+                    frameBytes = FrameCodec.Encode(frame);
+                    lock (_sendLock)
+                        _send.RecordSent(seq, frameBytes);
+                }
+
+                link.SendTo(serverMac, frame);
+            }
+        }
+        catch { }
+        _cts.Cancel();
+        _deliverChannel.Writer.TryComplete();
+        SendClose();
+        onClosed();
+    }
+
+    async Task PumpTunnelToSocket()
+    {
+        try
+        {
+            var stream = client.GetStream();
+            await foreach (var payload in _deliverChannel.Reader.ReadAllAsync())
+                await stream.WriteAsync(payload);
+        }
+        catch { }
+        finally { client.Close(); }
+    }
+
+    async Task RetransmitTimer()
+    {
+        using var timer = new PeriodicTimer(TimeSpan.FromMilliseconds(1));
+        try
+        {
+            while (!_cts.IsCancellationRequested)
+            {
+                await timer.WaitForNextTickAsync(_cts.Token);
+
+                List<byte[]>? resend = null;
+                bool shouldClose = false;
+                lock (_sendLock)
+                    (resend, shouldClose) = _send.CheckRetransmit();
+
+                if (resend != null)
+                {
+                    foreach (var frameBytes in resend)
+                    {
+                        // Send raw bytes directly (already encoded).
+                        link.SendRaw(serverMac, frameBytes);
+                    }
+                }
+
+                if (shouldClose)
+                {
+                    link.SendTo(serverMac, new Frame.Close(sessionId, Proto.ReasonMaxRetries));
+                    onClosed();
+                    break;
+                }
+            }
+        }
+        catch { }
+    }
+
+    void SendPureAck()
+    {
+        if (!_isTcp) return;
+        uint seq, ackSeq;
+        lock (_sendLock)
+        {
+            seq = _send.NextSeq();
+            lock (_recvLock)
+                ackSeq = _recv.CurrentAckSeq;
+        }
+        // Pure ACK: empty payload. Not stored in retransmit buffer.
+        link.SendTo(serverMac, new Frame.Data(sessionId, seq, ackSeq, []));
+    }
+
+    void SendClose()
+    {
+        if (_closeSent) return;
+        _closeSent = true;
+        link.SendTo(serverMac, new Frame.Close(sessionId, null));
+    }
+
+    public void Dispose()
+    {
+        _cts.Cancel();
+        _deliverChannel.Writer.TryComplete();
+        SendClose();
+        NetUtil.RstClose(client);
+        _cts.Dispose();
+    }
+}
+
+/// Close a TcpClient with a TCP RST instead of a FIN.
+static partial class NetUtil
+{
+    public static void RstClose(TcpClient c)
+    {
+        try
+        {
+            c.LingerState = new LingerOption(true, 0);
+            c.Close();
+        }
+        catch { }
+    }
+}

gatuna-win/TunnelLink.cs

@@ -1,7 +1,7 @@
 using SharpPcap;
 using SharpPcap.LibPcap;
 
-namespace gatuna_client;
+namespace gatuna;
 
 sealed class TunnelLink : IDisposable
 {
@@ -50,6 +50,11 @@ sealed class TunnelLink : IDisposable
         var et = (ushort)(data[12] << 8 | data[13]);
         if (et != Proto.EtherType)
             return;
+        // Skip our own outgoing frames (Npcap loops them back in promiscuous mode).
+        if (data[6] == _ourMac[0] && data[7] == _ourMac[1]
+            && data[8] == _ourMac[2] && data[9] == _ourMac[3]
+            && data[10] == _ourMac[4] && data[11] == _ourMac[5])
+            return;
         var srcMac = new byte[6];
         Buffer.BlockCopy(data, 6, srcMac, 0, 6);
         var payload = data.AsSpan(Proto.EthHeaderLen);
@@ -67,7 +72,9 @@ sealed class TunnelLink : IDisposable
         SendRaw(dstMac, FrameCodec.Encode(frame));
     }
 
-    void SendRaw(byte[] dstMac, byte[] payload)
+    /// Send a pre-encoded protocol payload (for retransmit, where we already
+    /// have the exact bytes and want to avoid re-encoding).
+    public void SendRaw(byte[] dstMac, byte[] payload)
     {
         var frame = new byte[Proto.EthHeaderLen + payload.Length];
         Buffer.BlockCopy(dstMac, 0, frame, 0, 6);

gatuna-win/app.manifest

@@ -1,6 +1,6 @@
 <?xml version="1.0" encoding="utf-8"?>
 <assembly manifestVersion="1.0" xmlns="urn:schemas-microsoft-com:asm.v1">
-  <assemblyIdentity version="0.1.0.0" name="gatuna-client" />
+  <assemblyIdentity version="2.1.0.0" name="gatuna" />
   <trustInfo xmlns="urn:schemas-microsoft-com:asm.v2">
     <security>
       <requestedPrivileges xmlns="urn:schemas-microsoft-com:asm.v3">

gatuna-win/gatuna.csproj

@@ -3,7 +3,9 @@
   <PropertyGroup>
     <OutputType>WinExe</OutputType>
     <TargetFramework>net8.0-windows</TargetFramework>
-    <RootNamespace>gatuna_client</RootNamespace>
+    <AssemblyName>gatuna</AssemblyName>
+    <RootNamespace>gatuna</RootNamespace>
+    <Version>2.1.0</Version>
     <Nullable>enable</Nullable>
     <UseWindowsForms>true</UseWindowsForms>
     <ImplicitUsings>enable</ImplicitUsings>

gatunad/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "gatuna"
-version = "0.1.0"
+version = "2.1.0"
 edition = "2021"
 license = "CC0-1.0"
 
@@ -10,7 +10,7 @@ path = "src/main.rs"
 
 [dependencies]
 libc = "0.2"
-tokio = { version = "1", features = ["rt-multi-thread", "macros", "net", "sync", "io-util"] }
+tokio = { version = "1", features = ["rt-multi-thread", "macros", "net", "sync", "io-util", "time"] }
 clap = { version = "4", features = ["derive"] }
 pnet_datalink = "0.35"
 tracing = "0.1"

gatunad/src/frame.rs

@@ -1,8 +1,10 @@
-//! gatuna wire protocol frame encode/decode.
+//! gatuna wire protocol frame encode/decode (v2).
 
 pub const ETHERTYPE: u16 = 0x6969;
 pub const ETH_HEADER_LEN: usize = 14;
-pub const VERSION: u8 = 1;
+pub const VERSION: u8 = 2;
+pub const HEADER_LEN: usize = 8;
+pub const DATA_HEADER_LEN: usize = 16; // 8 common + 4 seq + 4 ack_seq
 pub const MAX_PAYLOAD: usize = 1480;
 
 pub const TYPE_DISCOVER: u8 = 0x01;
@@ -15,6 +17,8 @@ pub const TYPE_CLOSE: u8 = 0x07;
 pub const TYPE_UDP_OPEN: u8 = 0x08;
 pub const TYPE_UDP_DATA: u8 = 0x09;
 pub const TYPE_UDP_CLOSE: u8 = 0x0A;
+pub const TYPE_PING: u8 = 0x0B;
+pub const TYPE_PONG: u8 = 0x0C;
 
 pub const PROTO_TCP: u8 = 1;
 pub const PROTO_UDP: u8 = 2;
@@ -26,6 +30,7 @@ pub const REASON_CONNECT_FAILED: u8 = 2;
 #[allow(dead_code)]
 pub const REASON_OVERSIZE: u8 = 3;
 pub const REASON_UNKNOWN_SESSION: u8 = 4;
+pub const REASON_MAX_RETRIES: u8 = 5;
 
 #[derive(Clone, Debug)]
 pub struct UpstreamEntry {
@@ -38,12 +43,14 @@ pub struct UpstreamEntry {
 #[derive(Clone, Debug)]
 pub enum Frame {
     Discover,
-    Manifest(Vec<UpstreamEntry>),
+    Manifest { hostname: String, entries: Vec<UpstreamEntry> },
-    Open { upstream_id: u8 },
+    Open { upstream_id: u8, proto: u8 },
-    OpenAck { session_id: u32, upstream_id: u8 },
+    OpenAck { session_id: u32, upstream_id: u8, proto: u8 },
     OpenNak { upstream_id: u8, reason: u8 },
-    Data { session_id: u32, payload: Vec<u8> },
+    Data { session_id: u32, seq: u32, ack_seq: u32, payload: Vec<u8> },
     Close { session_id: u32, reason: Option<u8> },
+    Ping { nonce: u64 },
+    Pong { nonce: u64 },
 }
 
 #[derive(Debug)]
@@ -76,50 +83,72 @@ fn encode_entry(buf: &mut Vec<u8>, e: &UpstreamEntry) {
     buf.extend_from_slice(&label_bytes[..label_len as usize]);
 }
 
+/// Build a non-DATA frame: 8-byte common header + payload.
+fn build(type_byte: u8, session_id: u32, payload: Vec<u8>) -> Vec<u8> {
+    let len = payload.len() as u16;
+    let mut buf = Vec::with_capacity(HEADER_LEN + payload.len());
+    buf.push(VERSION);
+    buf.push(type_byte);
+    buf.extend_from_slice(&session_id.to_be_bytes());
+    buf.extend_from_slice(&len.to_be_bytes());
+    buf.extend_from_slice(&payload);
+    buf
+}
+
+/// Build a DATA frame: 8-byte common header + seq + ack_seq + payload.
+/// payload_len counts only the raw bytes, not seq/ack_seq.
+fn build_data(session_id: u32, seq: u32, ack_seq: u32, payload: Vec<u8>) -> Vec<u8> {
+    let len = payload.len() as u16;
+    let mut buf = Vec::with_capacity(DATA_HEADER_LEN + payload.len());
+    buf.push(VERSION);
+    buf.push(TYPE_DATA);
+    buf.extend_from_slice(&session_id.to_be_bytes());
+    buf.extend_from_slice(&len.to_be_bytes());
+    buf.extend_from_slice(&seq.to_be_bytes());
+    buf.extend_from_slice(&ack_seq.to_be_bytes());
+    buf.extend_from_slice(&payload);
+    buf
+}
+
 impl Frame {
     pub fn encode(&self) -> Vec<u8> {
-        let mut buf = Vec::new();
         match self {
-            Frame::Discover => {
+            Frame::Discover => build(TYPE_DISCOVER, 0, Vec::new()),
-                buf.extend_from_slice(&[VERSION, TYPE_DISCOVER, 0, 0, 0, 0]);
+            Frame::Manifest { hostname, entries } => {
-            }
+                let mut payload = Vec::new();
-            Frame::Manifest(entries) => {
+                let hn_bytes = hostname.as_bytes();
-                buf.extend_from_slice(&[VERSION, TYPE_MANIFEST, 0, 0, 0, 0]);
+                let hn_len = hn_bytes.len().min(255) as u8;
+                payload.push(hn_len);
+                payload.extend_from_slice(&hn_bytes[..hn_len as usize]);
                 for e in entries {
-                    encode_entry(&mut buf, e);
+                    encode_entry(&mut payload, e);
                 }
+                build(TYPE_MANIFEST, 0, payload)
             }
-            Frame::Open { upstream_id } => {
+            Frame::Open { upstream_id, proto } => build(TYPE_OPEN, 0, vec![*upstream_id, *proto]),
-                buf.extend_from_slice(&[VERSION, TYPE_OPEN, 0, 0, 0, 0, *upstream_id]);
+            Frame::OpenAck { session_id, upstream_id, proto } => {
-            }
+                build(TYPE_OPEN_ACK, *session_id, vec![*upstream_id, *proto])
-            Frame::OpenAck { session_id, upstream_id } => {
-                buf.extend_from_slice(&[VERSION, TYPE_OPEN_ACK]);
-                buf.extend_from_slice(&session_id.to_be_bytes());
-                buf.push(*upstream_id);
             }
             Frame::OpenNak { upstream_id, reason } => {
-                buf.extend_from_slice(&[VERSION, TYPE_OPEN_NAK, 0, 0, 0, 0]);
+                build(TYPE_OPEN_NAK, 0, vec![*upstream_id, *reason])
-                buf.push(*upstream_id);
-                buf.push(*reason);
             }
-            Frame::Data { session_id, payload } => {
+            Frame::Data { session_id, seq, ack_seq, payload } => {
-                buf.extend_from_slice(&[VERSION, TYPE_DATA]);
+                build_data(*session_id, *seq, *ack_seq, payload.clone())
-                buf.extend_from_slice(&session_id.to_be_bytes());
-                buf.extend_from_slice(payload);
             }
             Frame::Close { session_id, reason } => {
-                buf.extend_from_slice(&[VERSION, TYPE_CLOSE]);
+                let p = match reason {
-                buf.extend_from_slice(&session_id.to_be_bytes());
+                    Some(r) => vec![*r],
-                if let Some(r) = reason {
+                    None => Vec::new(),
-                    buf.push(*r);
+                };
-                }
+                build(TYPE_CLOSE, *session_id, p)
             }
+            Frame::Ping { nonce } => build(TYPE_PING, 0, nonce.to_be_bytes().to_vec()),
+            Frame::Pong { nonce } => build(TYPE_PONG, 0, nonce.to_be_bytes().to_vec()),
         }
-        buf
     }
 
     pub fn parse(buf: &[u8]) -> Result<Frame, DecodeError> {
-        if buf.len() < 6 {
+        if buf.len() < HEADER_LEN {
             return Err(DecodeError::Short);
         }
         let version = buf[0];
@@ -128,7 +157,32 @@ impl Frame {
         }
         let typ = buf[1];
         let session_id = u32::from_be_bytes([buf[2], buf[3], buf[4], buf[5]]);
-        let payload = &buf[6..];
+        let payload_len = u16::from_be_bytes([buf[6], buf[7]]) as usize;
+
+        // DATA frames have seq + ack_seq after the common header.
+        if typ == TYPE_DATA {
+            if buf.len() < DATA_HEADER_LEN + payload_len {
+                return Err(DecodeError::BadPayload("DATA: payload_len exceeds available data"));
+            }
+            let seq = u32::from_be_bytes([buf[8], buf[9], buf[10], buf[11]]);
+            let ack_seq = u32::from_be_bytes([buf[12], buf[13], buf[14], buf[15]]);
+            let payload = &buf[DATA_HEADER_LEN..DATA_HEADER_LEN + payload_len];
+            if payload.len() > MAX_PAYLOAD {
+                return Err(DecodeError::BadPayload("DATA payload exceeds max"));
+            }
+            return Ok(Frame::Data {
+                session_id,
+                seq,
+                ack_seq,
+                payload: payload.to_vec(),
+            });
+        }
+
+        if buf.len() < HEADER_LEN + payload_len {
+            return Err(DecodeError::BadPayload("payload_len exceeds available data"));
+        }
+        let payload = &buf[HEADER_LEN..HEADER_LEN + payload_len];
+
         match typ {
             TYPE_DISCOVER => {
                 if !payload.is_empty() {
@@ -137,41 +191,50 @@ impl Frame {
                 Ok(Frame::Discover)
             }
             TYPE_MANIFEST => {
+                if payload.is_empty() {
+                    return Err(DecodeError::BadPayload("MANIFEST missing hostname prefix"));
+                }
+                let hn_len = payload[0] as usize;
+                if 1 + hn_len > payload.len() {
+                    return Err(DecodeError::BadPayload("MANIFEST hostname truncated"));
+                }
+                let hostname = String::from_utf8_lossy(&payload[1..1 + hn_len]).into_owned();
+                let rest = &payload[1 + hn_len..];
                 let mut entries = Vec::new();
                 let mut i = 0;
-                while i < payload.len() {
+                while i < rest.len() {
-                    if i + 5 > payload.len() {
+                    if i + 5 > rest.len() {
                         return Err(DecodeError::BadPayload("MANIFEST entry truncated"));
                     }
-                    let id = payload[i];
+                    let id = rest[i];
-                    let proto = payload[i + 1];
+                    let proto = rest[i + 1];
-                    let port = u16::from_be_bytes([payload[i + 2], payload[i + 3]]);
+                    let port = u16::from_be_bytes([rest[i + 2], rest[i + 3]]);
-                    let label_len = payload[i + 4] as usize;
+                    let label_len = rest[i + 4] as usize;
                     i += 5;
-                    if i + label_len > payload.len() {
+                    if i + label_len > rest.len() {
                         return Err(DecodeError::BadPayload("MANIFEST label truncated"));
                     }
                     let label = if label_len == 0 {
                         None
                     } else {
-                        Some(String::from_utf8_lossy(&payload[i..i + label_len]).into_owned())
+                        Some(String::from_utf8_lossy(&rest[i..i + label_len]).into_owned())
                     };
                     i += label_len;
                     entries.push(UpstreamEntry { id, proto, port, label });
                 }
-                Ok(Frame::Manifest(entries))
+                Ok(Frame::Manifest { hostname, entries })
             }
             TYPE_OPEN => {
-                if payload.len() != 1 {
+                if payload.len() != 2 {
-                    return Err(DecodeError::BadPayload("OPEN payload must be 1 byte"));
+                    return Err(DecodeError::BadPayload("OPEN payload must be 2 bytes"));
                 }
-                Ok(Frame::Open { upstream_id: payload[0] })
+                Ok(Frame::Open { upstream_id: payload[0], proto: payload[1] })
             }
             TYPE_OPEN_ACK => {
-                if payload.len() != 1 {
+                if payload.len() != 2 {
-                    return Err(DecodeError::BadPayload("OPEN_ACK payload must be 1 byte"));
+                    return Err(DecodeError::BadPayload("OPEN_ACK payload must be 2 bytes"));
                 }
-                Ok(Frame::OpenAck { session_id, upstream_id: payload[0] })
+                Ok(Frame::OpenAck { session_id, upstream_id: payload[0], proto: payload[1] })
             }
             TYPE_OPEN_NAK => {
                 if payload.len() != 2 {
@@ -179,12 +242,6 @@ impl Frame {
                 }
                 Ok(Frame::OpenNak { upstream_id: payload[0], reason: payload[1] })
             }
-            TYPE_DATA => {
-                if payload.len() > MAX_PAYLOAD {
-                    return Err(DecodeError::BadPayload("DATA payload exceeds max"));
-                }
-                Ok(Frame::Data { session_id, payload: payload.to_vec() })
-            }
             TYPE_CLOSE => {
                 let reason = match payload.len() {
                     0 => None,
@@ -193,8 +250,22 @@ impl Frame {
                 };
                 Ok(Frame::Close { session_id, reason })
             }
+            TYPE_PING => {
+                if payload.len() != 8 {
+                    return Err(DecodeError::BadPayload("PING payload must be 8 bytes"));
+                }
+                let nonce = u64::from_be_bytes(payload.try_into().unwrap());
+                Ok(Frame::Ping { nonce })
+            }
+            TYPE_PONG => {
+                if payload.len() != 8 {
+                    return Err(DecodeError::BadPayload("PONG payload must be 8 bytes"));
+                }
+                let nonce = u64::from_be_bytes(payload.try_into().unwrap());
+                Ok(Frame::Pong { nonce })
+            }
             TYPE_UDP_OPEN | TYPE_UDP_DATA | TYPE_UDP_CLOSE => {
-                Err(DecodeError::BadPayload("UDP frame types not implemented in v1"))
+                Err(DecodeError::BadPayload("UDP frame types not implemented"))
             }
             other => Err(DecodeError::UnknownType(other)),
         }

gatunad/src/link.rs

@@ -12,6 +12,16 @@ use crate::frame::{ETHERTYPE, ETH_HEADER_LEN};
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub struct MacAddr(pub [u8; 6]);
 
+impl std::fmt::Display for MacAddr {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "{:02X}:{:02X}:{:02X}:{:02X}:{:02X}:{:02X}",
+            self.0[0], self.0[1], self.0[2], self.0[3], self.0[4], self.0[5]
+        )
+    }
+}
+
 impl MacAddr {
     #[allow(dead_code)]
     pub fn broadcast() -> Self {

gatunad/src/main.rs

@@ -16,13 +16,13 @@ use std::sync::{Arc, Mutex};
 use tokio::net::TcpStream;
 use tokio::sync::mpsc;
 use tokio::sync::Mutex as AsyncMutex;
-use tracing::{error, Level};
+use tracing::{error, info, Level};
 
 use crate::frame::{
     Frame, REASON_CONNECT_FAILED, REASON_UNKNOWN_SESSION, REASON_UNKNOWN_UPSTREAM,
 };
 use crate::link::Link;
-use crate::session::{spawn_pump, write_to_session, SessionHandle, SessionStore};
+use crate::session::{spawn_pump, handle_data, send_pure_ack, SessionHandle, SessionStore};
 use crate::upstream::build_table;
 
 #[derive(Parser)]
@@ -33,17 +33,21 @@ struct Args {
     /// One or more TCP upstreams as PORT[:label], relayed to 127.0.0.1:PORT.
     #[arg(num_args = 1..)]
     ports: Vec<String>,
+    /// Verbose logging (lifecycle events to stdout).
+    #[arg(short, long)]
+    verbose: bool,
 }
 
 #[tokio::main]
 async fn main() -> ExitCode {
+    let args = Args::parse();
+
+    let level = if args.verbose { Level::INFO } else { Level::ERROR };
     tracing_subscriber::fmt()
-        .with_max_level(Level::ERROR)
+        .with_max_level(level)
         .with_writer(|| std::io::stdout())
         .init();
 
-    let args = Args::parse();
-
     let table = match build_table(&args.ports) {
         Ok(t) => Arc::new(t),
         Err(e) => {
@@ -52,6 +56,8 @@ async fn main() -> ExitCode {
         }
     };
 
+    let hostname = Arc::new(upstream::get_hostname());
+
     let link = match Link::open(&args.iface) {
         Ok(l) => Arc::new(l),
         Err(e) => {
@@ -108,11 +114,11 @@ async fn main() -> ExitCode {
                 let frame = match Frame::parse(payload) {
                     Ok(f) => f,
                     Err(e) => {
-                        error!("decode from {src:?}: {e}");
+                        error!("decode from {src}: {e}");
                         continue;
                     }
                 };
-                handle_frame(frame, src, &tx, &store, &next_id, &table).await;
+                handle_frame(frame, src, &tx, &store, &next_id, &table, &hostname).await;
             }
             Ok(None) => {
                 // Ignorable frame (outgoing/short/mismatch) or transient; do not
@@ -135,16 +141,25 @@ async fn handle_frame(
     store: &SessionStore,
     next_id: &Arc<AtomicU32>,
     table: &Arc<crate::upstream::UpstreamTable>,
+    hostname: &Arc<String>,
 ) {
     match frame {
         Frame::Discover => {
-            let manifest = Frame::Manifest(table.entries());
+            info!("DISCOVER from {src}");
+            let manifest = Frame::Manifest {
+                hostname: (**hostname).clone(),
+                entries: table.entries(),
+            };
             let _ = tx.send((src, manifest.encode())).await;
+            info!("MANIFEST sent to {src} ({} upstreams)", table.0.len());
         }
-        Frame::Open { upstream_id } => {
+        Frame::Open { upstream_id, proto: _ } => {
-            let port = table.get(upstream_id).map(|u| u.port);
+            info!("OPEN upstream {upstream_id} from {src}");
-            match port {
+            let upstream = table.get(upstream_id);
-                Some(port) => {
+            match upstream {
+                Some(upstream) => {
+                    let port = upstream.port;
+                    let proto = upstream.proto.as_u8();
                     // Spawn so connect() doesn't block the rx loop.
                     let tx = tx.clone();
                     let store = Arc::clone(store);
@@ -159,15 +174,21 @@ async fn handle_frame(
                                     sid,
                                     SessionHandle {
                                         upstream_id,
+                                        proto,
                                         write: w,
+                                        send_state: Arc::new(Mutex::new(session::SendState::new())),
+                                        recv_state: Arc::new(Mutex::new(session::RecvState::new())),
+                                        peer_mac: src,
                                     },
                                 );
-                                let ack = Frame::OpenAck { session_id: sid, upstream_id };
+                                let ack = Frame::OpenAck { session_id: sid, upstream_id, proto };
                                 let _ = tx.send((src, ack.encode())).await;
-                                spawn_pump(r, sid, src, tx, store);
+                                info!("session {sid} established (upstream {upstream_id})");
+                                spawn_pump(r, sid, proto, src, tx, store);
                             }
                             Err(e) => {
                                 error!("connect 127.0.0.1:{port} failed: {e}");
+                                info!("OPEN_NAK upstream {upstream_id} (connect_failed) to {src}");
                                 let nak =
                                     Frame::OpenNak { upstream_id, reason: REASON_CONNECT_FAILED };
                                 let _ = tx.send((src, nak.encode())).await;
@@ -181,9 +202,13 @@ async fn handle_frame(
                 }
             }
         }
-        Frame::Data { session_id, payload } => {
+        Frame::Data { session_id, seq, ack_seq, payload } => {
-            match write_to_session(store, session_id, &payload).await {
+            match handle_data(store, session_id, seq, ack_seq, &payload, tx).await {
-                Ok(()) => {}
+                Ok(need_ack) => {
+                    if need_ack {
+                        send_pure_ack(store, session_id, tx);
+                    }
+                }
                 Err(session::WriteError::UnknownSession) => {
                     let close = Frame::Close {
                         session_id,
@@ -199,9 +224,16 @@ async fn handle_frame(
             }
         }
         Frame::Close { session_id, reason: _ } => {
-            store.lock().expect("store poisoned").remove(&session_id);
+            if store.lock().expect("store poisoned").remove(&session_id).is_some() {
+                info!("CLOSE session {session_id} from {src}");
+            }
+        }
+        Frame::Ping { nonce } => {
+            let pong = Frame::Pong { nonce };
+            let _ = tx.send((src, pong.encode())).await;
         }
         // Not expected from a client; ignore.
-        Frame::Manifest(_) | Frame::OpenAck { .. } | Frame::OpenNak { .. } => {}
+        Frame::Manifest { .. } | Frame::OpenAck { .. } | Frame::OpenNak { .. }
+        | Frame::Pong { .. } => {}
     }
 }

gatunad/src/session.rs

@@ -1,20 +1,142 @@
-//! Per-session state and the localhost→tunnel pump.
+//! Per-session state: L2 reliability layer + localhost→tunnel pump.
 
-use crate::frame::{Frame, MAX_PAYLOAD};
+use crate::frame::{Frame, MAX_PAYLOAD, REASON_MAX_RETRIES};
 use crate::link::MacAddr;
-use std::collections::HashMap;
+use std::collections::{HashMap, BTreeMap};
 use std::sync::{Arc, Mutex};
+use std::time::{Duration, Instant};
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
-use tokio::net::tcp::OwnedReadHalf;
+use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
 use tokio::sync::mpsc::Sender;
 use tokio::sync::Mutex as AsyncMutex;
 
 pub type TxChan = Sender<(MacAddr, Vec<u8>)>;
 
+const RETRANSMIT_TIMEOUT: Duration = Duration::from_millis(5);
+const MAX_RETRIES: u32 = 10;
+const RETRANSMIT_TICK: Duration = Duration::from_millis(1);
+
+/// Sender-side reliability state (per session).
+pub struct SendState {
+    send_seq: u32,
+    acked_seq: u32,
+    /// seq -> (frame_bytes, send_time, retry_count)
+    retransmit_buffer: BTreeMap<u32, (Vec<u8>, Instant, u32)>,
+}
+
+/// Receiver-side reliability state (per session).
+pub struct RecvState {
+    expected_seq: u32,
+    deliver_seq: u32,
+    /// seq -> payload (out-of-order buffer)
+    receive_buffer: BTreeMap<u32, Vec<u8>>,
+}
+
+impl SendState {
+    pub fn new() -> Self {
+        SendState {
+            send_seq: 0,
+            acked_seq: 0,
+            retransmit_buffer: BTreeMap::new(),
+        }
+    }
+
+    /// Record a sent DATA frame in the retransmit buffer.
+    fn record_sent(&mut self, seq: u32, frame_bytes: Vec<u8>) {
+        self.retransmit_buffer
+            .insert(seq, (frame_bytes, Instant::now(), 0));
+    }
+
+    /// Process an incoming ack_seq: advance window, remove acked frames.
+    fn process_ack(&mut self, ack_seq: u32) {
+        self.retransmit_buffer.retain(|&seq, _| seq > ack_seq);
+        if ack_seq > self.acked_seq || self.retransmit_buffer.is_empty() {
+            self.acked_seq = ack_seq.max(self.acked_seq);
+        }
+    }
+
+    /// Check for frames needing retransmit. Returns frames to resend and
+    /// whether the session should be closed (max retries exceeded).
+    fn check_retransmit(&mut self) -> (Vec<Vec<u8>>, bool) {
+        let mut resend = Vec::new();
+        let mut should_close = false;
+        let now = Instant::now();
+        for (_seq, (frame_bytes, send_time, retries)) in self.retransmit_buffer.iter_mut() {
+            if now.duration_since(*send_time) > RETRANSMIT_TIMEOUT {
+                if *retries >= MAX_RETRIES {
+                    should_close = true;
+                    break;
+                }
+                *retries += 1;
+                *send_time = now;
+                resend.push(frame_bytes.clone());
+            }
+        }
+        (resend, should_close)
+    }
+
+    fn next_seq(&mut self) -> u32 {
+        let s = self.send_seq;
+        self.send_seq = self.send_seq.wrapping_add(1);
+        s
+    }
+}
+
+impl RecvState {
+    pub fn new() -> Self {
+        RecvState {
+            expected_seq: 0,
+            deliver_seq: 0,
+            receive_buffer: BTreeMap::new(),
+        }
+    }
+
+    /// Process an incoming DATA frame's seq. Returns:
+    /// - `Ok((payloads, need_ack))` — payloads to deliver (may be empty if
+    ///   out-of-order), need_ack is true if the frame was duplicate or
+    ///   out-of-order and the caller should send a pure ACK.
+    /// - `Err(())` — should not happen with current callers.
+    fn process_data(&mut self, seq: u32, payload: Vec<u8>) -> (Vec<Vec<u8>>, bool) {
+        if seq < self.expected_seq {
+            // Duplicate — already delivered.
+            return (vec![], true);
+        }
+        if seq == self.expected_seq {
+            // In-order: deliver immediately, then drain the receive buffer.
+            let mut deliver = vec![payload];
+            self.expected_seq = self.expected_seq.wrapping_add(1);
+            self.deliver_seq = self.expected_seq.wrapping_sub(1);
+            // Drain contiguous buffered frames.
+            while let Some(payload) = self.receive_buffer.remove(&self.expected_seq) {
+                self.expected_seq = self.expected_seq.wrapping_add(1);
+                self.deliver_seq = self.expected_seq.wrapping_sub(1);
+                deliver.push(payload);
+            }
+            (deliver, false)
+        } else {
+            // Out-of-order: buffer it.
+            self.receive_buffer.insert(seq, payload);
+            (vec![], true)
+        }
+    }
+
+    /// The ack_seq to report in outgoing DATA frames (highest contiguous
+    /// delivered seq). If nothing delivered yet, report expected_seq - 1
+    /// (wrapping), which is the last seq we can cumulatively ACK.
+    fn current_ack_seq(&self) -> u32 {
+        self.expected_seq.wrapping_sub(1)
+    }
+}
+
+#[derive(Clone)]
 #[allow(dead_code)]
 pub struct SessionHandle {
     pub upstream_id: u8,
-    pub write: Arc<AsyncMutex<tokio::net::tcp::OwnedWriteHalf>>,
+    pub proto: u8,
+    pub write: Arc<AsyncMutex<OwnedWriteHalf>>,
+    pub send_state: Arc<Mutex<SendState>>,
+    pub recv_state: Arc<Mutex<RecvState>>,
+    pub peer_mac: MacAddr,
 }
 
 pub type SessionStore = Arc<Mutex<HashMap<u32, SessionHandle>>>;
@@ -25,54 +147,191 @@ pub enum WriteError {
     Io,
 }
 
-pub async fn write_to_session(
+/// Handle a DATA frame received from the tunnel. For TCP sessions, delivers
+/// in-order via the reliability layer. For stateless protos, delivers
+/// best-effort. Returns whether a pure ACK should be sent back.
+pub async fn handle_data(
     store: &SessionStore,
-    id: u32,
+    session_id: u32,
+    seq: u32,
+    ack_seq: u32,
     payload: &[u8],
-) -> Result<(), WriteError> {
+    _tx: &TxChan,
-    let write = {
+) -> Result<bool, WriteError> {
+    let handle = {
         let store = store.lock().expect("store lock poisoned");
-        store.get(&id).map(|h| h.write.clone())
+        store.get(&session_id).cloned()
     };
-    match write {
+
-        Some(w) => {
+    let Some(handle) = handle else {
-            let mut w = w.lock().await;
+        return Err(WriteError::UnknownSession);
-            w.write_all(payload).await.map_err(|_| WriteError::Io)?;
+    };
-            Ok(())
+
+    // Non-TCP protos: best-effort delivery, no reliability machinery.
+    if handle.proto != crate::frame::PROTO_TCP {
+        if !payload.is_empty() {
+            let mut w = handle.write.lock().await;
+            if w.write_all(payload).await.is_err() {
+                return Err(WriteError::Io);
+            }
         }
-        None => Err(WriteError::UnknownSession),
+        return Ok(false);
     }
+
+    // TCP: full reliability — process ack_seq to advance send window.
+    {
+        let mut ss = handle.send_state.lock().expect("send_state poisoned");
+        ss.process_ack(ack_seq);
+    }
+
+    // Process seq for in-order delivery.
+    let need_ack;
+    {
+        let mut rs = handle.recv_state.lock().expect("recv_state poisoned");
+        let (deliver, ack) = rs.process_data(seq, payload.to_vec());
+        if !deliver.is_empty() {
+            let mut w = handle.write.lock().await;
+            for chunk in deliver {
+                if w.write_all(&chunk).await.is_err() {
+                    return Err(WriteError::Io);
+                }
+            }
+        }
+        need_ack = ack;
+    }
+
+    Ok(need_ack)
+}
+
+/// Send a pure ACK (DATA frame with empty payload) for the given session.
+pub fn send_pure_ack(
+    store: &SessionStore,
+    session_id: u32,
+    tx: &TxChan,
+) {
+    let handle = {
+        let store = store.lock().expect("store lock poisoned");
+        store.get(&session_id).cloned()
+    };
+    let Some(handle) = handle else { return };
+
+    // Pure ACKs are only meaningful for reliable (TCP) sessions.
+    if handle.proto != crate::frame::PROTO_TCP {
+        return;
+    }
+
+    let (seq, ack_seq) = {
+        let mut ss = handle.send_state.lock().expect("send_state poisoned");
+        let rs = handle.recv_state.lock().expect("recv_state poisoned");
+        let seq = ss.next_seq();
+        (seq, rs.current_ack_seq())
+    };
+
+    // Pure ACKs are not stored in the retransmit buffer (no payload to lose).
+    let frame = Frame::Data {
+        session_id,
+        seq,
+        ack_seq,
+        payload: Vec::new(),
+    };
+    let _ = tx.try_send((handle.peer_mac, frame.encode()));
 }
 
 /// Spawn the socket→tunnel pump: reads from the localhost TCP stream in
-/// 1480-byte chunks and emits DATA frames. On EOF/error sends CLOSE and
+/// 1480-byte chunks, tags each with seq, and emits DATA frames. For TCP
-/// removes the session from the store.
+/// sessions, also stores in retransmit buffer and spawns the retransmit
+/// timer. For stateless protos, best-effort (no retransmit).
 pub fn spawn_pump(
     mut read: OwnedReadHalf,
     session_id: u32,
+    proto: u8,
     peer_mac: MacAddr,
     tx: TxChan,
     store: SessionStore,
 ) {
+    let is_tcp = proto == crate::frame::PROTO_TCP;
+
+    // Retransmit timer task (TCP only).
+    if is_tcp {
+        let store = Arc::clone(&store);
+        let tx = tx.clone();
+        tokio::spawn(async move {
+            let mut interval = tokio::time::interval(RETRANSMIT_TICK);
+            loop {
+                interval.tick().await;
+                let handle = {
+                    let store = store.lock().expect("store poisoned");
+                    store.get(&session_id).cloned()
+                };
+                let Some(handle) = handle else { break };
+
+                let (resend, should_close) = {
+                    let mut ss = handle.send_state.lock().expect("send_state poisoned");
+                    ss.check_retransmit()
+                };
+
+                for frame_bytes in resend {
+                    let _ = tx.try_send((peer_mac, frame_bytes));
+                }
+
+                if should_close {
+                    store.lock().expect("store poisoned").remove(&session_id);
+                    let close = Frame::Close {
+                        session_id,
+                        reason: Some(REASON_MAX_RETRIES),
+                    };
+                    let _ = tx.try_send((peer_mac, close.encode()));
+                    break;
+                }
+            }
+        });
+    }
+
+    // Socket read pump.
     tokio::spawn(async move {
         let mut buf = vec![0u8; MAX_PAYLOAD];
         loop {
-            match read.read(&mut buf).await {
+            let n = match read.read(&mut buf).await {
                 Ok(0) => break,
-                Ok(n) => {
+                Ok(n) => n,
-                    let frame = Frame::Data {
-                        session_id,
-                        payload: buf[..n].to_vec(),
-                    };
-                    if tx.send((peer_mac, frame.encode())).await.is_err() {
-                        break;
-                    }
-                }
                 Err(_) => break,
+            };
+
+            let handle = {
+                let store = store.lock().expect("store poisoned");
+                store.get(&session_id).cloned()
+            };
+            let Some(handle) = handle else { break };
+
+            let (seq, ack_seq) = if is_tcp {
+                let mut ss = handle.send_state.lock().expect("send_state poisoned");
+                let rs = handle.recv_state.lock().expect("recv_state poisoned");
+                let seq = ss.next_seq();
+                (seq, rs.current_ack_seq())
+            } else {
+                (0, 0)
+            };
+
+            let frame = Frame::Data {
+                session_id,
+                seq,
+                ack_seq,
+                payload: buf[..n].to_vec(),
+            };
+            let frame_bytes = frame.encode();
+
+            // Store in retransmit buffer before sending (TCP only).
+            if is_tcp {
+                let mut ss = handle.send_state.lock().expect("send_state poisoned");
+                ss.record_sent(seq, frame_bytes.clone());
+            }
+
+            if tx.send((peer_mac, frame_bytes)).await.is_err() {
+                break;
             }
         }
+        store.lock().expect("store poisoned").remove(&session_id);
         let close = Frame::Close { session_id, reason: None };
         let _ = tx.send((peer_mac, close.encode())).await;
-        store.lock().expect("store poisoned").remove(&session_id);
     });
 }

gatunad/src/upstream.rs

@@ -2,6 +2,20 @@
 
 use crate::frame::{UpstreamEntry, PROTO_TCP, PROTO_UDP};
 
+/// Read the system hostname via `gethostname(2)`.
+pub fn get_hostname() -> String {
+    let mut buf = [0u8; 256];
+    let ret = unsafe {
+        libc::gethostname(buf.as_mut_ptr() as *mut libc::c_char, buf.len())
+    };
+    if ret == 0 {
+        let len = buf.iter().position(|&b| b == 0).unwrap_or(buf.len());
+        String::from_utf8_lossy(&buf[..len]).into_owned()
+    } else {
+        String::new()
+    }
+}
+
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 #[allow(dead_code)]
 pub enum Proto {

win/gatuna-client/Frame.cs

@@ -1,156 +0,0 @@
-namespace gatuna_client;
-
-using System.Text;
-
-static class Proto
-{
-    public const ushort EtherType = 0x6969;
-    public const int EthHeaderLen = 14;
-    public const byte Version = 1;
-    public const int MaxPayload = 1480;
-
-    public const byte TypeDiscover = 0x01;
-    public const byte TypeManifest = 0x02;
-    public const byte TypeOpen = 0x03;
-    public const byte TypeOpenAck = 0x04;
-    public const byte TypeOpenNak = 0x05;
-    public const byte TypeData = 0x06;
-    public const byte TypeClose = 0x07;
-
-    public const byte ProtoTcp = 1;
-    public const byte ProtoUdp = 2;
-
-    public const byte ReasonUnspecified = 0;
-    public const byte ReasonUnknownUpstream = 1;
-    public const byte ReasonConnectFailed = 2;
-    public const byte ReasonOversize = 3;
-    public const byte ReasonUnknownSession = 4;
-}
-
-readonly record struct UpstreamEntry(
-    byte Id,
-    byte Protocol,
-    ushort Port,
-    string? Label)
-{
-    public string ProtoName => Protocol == Proto.ProtoTcp ? "tcp" : "udp";
-}
-
-abstract record Frame
-{
-    internal record Discover : Frame;
-    internal record Manifest(UpstreamEntry[] Entries) : Frame;
-    internal record Open(byte UpstreamId) : Frame;
-    internal record OpenAck(uint SessionId, byte UpstreamId) : Frame;
-    internal record OpenNak(byte UpstreamId, byte Reason) : Frame;
-    internal record Data(uint SessionId, byte[] Payload) : Frame;
-    internal record Close(uint SessionId, byte? Reason) : Frame;
-}
-
-static class FrameCodec
-{
-    public static byte[] Encode(Frame frame)
-    {
-        return frame switch
-        {
-            Frame.Discover =>
-                Header(Proto.TypeDiscover, 0),
-            Frame.Manifest manifest =>
-                BuildManifest(manifest.Entries),
-            Frame.Open open =>
-                [.. Header(Proto.TypeOpen, 0), open.UpstreamId],
-            Frame.OpenAck ack =>
-                [.. Header(Proto.TypeOpenAck, ack.SessionId), ack.UpstreamId],
-            Frame.OpenNak nak =>
-                [.. Header(Proto.TypeOpenNak, 0), nak.UpstreamId, nak.Reason],
-            Frame.Data data =>
-                [.. Header(Proto.TypeData, data.SessionId), .. data.Payload],
-            Frame.Close close =>
-                close.Reason.HasValue
-                    ? [.. Header(Proto.TypeClose, close.SessionId), close.Reason.Value]
-                    : Header(Proto.TypeClose, close.SessionId),
-            _ => throw new InvalidOperationException($"unknown frame type: {frame.GetType()}"),
-        };
-    }
-
-    static byte[] BuildManifest(UpstreamEntry[] entries)
-    {
-        using var ms = new MemoryStream();
-        ms.Write(Header(Proto.TypeManifest, 0));
-        foreach (var e in entries)
-        {
-            var labelBytes = Encoding.UTF8.GetBytes(e.Label ?? "");
-            var labelLen = (byte)Math.Min(labelBytes.Length, 255);
-            ms.WriteByte(e.Id);
-            ms.WriteByte(e.Protocol);
-            ms.WriteByte((byte)(e.Port >> 8));
-            ms.WriteByte((byte)(e.Port & 0xFF));
-            ms.WriteByte(labelLen);
-            if (labelLen > 0)
-                ms.Write(labelBytes, 0, labelLen);
-        }
-        return ms.ToArray();
-    }
-
-    static byte[] Header(byte type, uint sessionId)
-    {
-        return [
-            Proto.Version, type,
-            (byte)(sessionId >> 24),
-            (byte)(sessionId >> 16),
-            (byte)(sessionId >> 8),
-            (byte)(sessionId & 0xFF),
-        ];
-    }
-
-    public static Frame? Parse(ReadOnlySpan<byte> buf)
-    {
-        if (buf.Length < 6)
-            return null;
-        if (buf[0] != Proto.Version)
-            return null;
-        var type = buf[1];
-        var sessionId = (uint)(buf[2] << 24 | buf[3] << 16 | buf[4] << 8 | buf[5]);
-        var payload = buf[6..];
-
-        return type switch
-        {
-            Proto.TypeManifest => ParseManifest(payload),
-            Proto.TypeOpenAck when payload.Length == 1 =>
-                new Frame.OpenAck(sessionId, payload[0]),
-            Proto.TypeOpenNak when payload.Length == 2 =>
-                new Frame.OpenNak(payload[0], payload[1]),
-            Proto.TypeData when payload.Length <= Proto.MaxPayload =>
-                new Frame.Data(sessionId, payload.ToArray()),
-            Proto.TypeClose when payload.Length is 0 or 1 =>
-                new Frame.Close(sessionId, payload.Length == 1 ? payload[0] : null),
-            Proto.TypeDiscover when payload.Length == 0 =>
-                new Frame.Discover(),
-            _ => null,
-        };
-    }
-
-    static Frame.Manifest? ParseManifest(ReadOnlySpan<byte> payload)
-    {
-        var entries = new List<UpstreamEntry>();
-        int i = 0;
-        while (i < payload.Length)
-        {
-            if (i + 5 > payload.Length)
-                return null;
-            var id = payload[i];
-            var proto = payload[i + 1];
-            var port = (ushort)(payload[i + 2] << 8 | payload[i + 3]);
-            var labelLen = payload[i + 4];
-            i += 5;
-            if (i + labelLen > payload.Length)
-                return null;
-            string? label = labelLen == 0
-                ? null
-                : Encoding.UTF8.GetString(payload[i..(i + labelLen)]);
-            i += labelLen;
-            entries.Add(new UpstreamEntry(id, proto, port, label));
-        }
-        return new Frame.Manifest(entries.ToArray());
-    }
-}

win/gatuna-client/MainForm.cs

@@ -1,141 +0,0 @@
-using SharpPcap.LibPcap;
-
-namespace gatuna_client;
-
-public partial class MainForm : Form
-{
-    readonly SessionManager _sessions = new();
-    readonly ComboBox _deviceBox = new();
-    readonly Button _discoverBtn = new();
-    readonly ListView _listView = new();
-    readonly Label _statusLabel = new();
-    TunnelLink? _link;
-
-    public MainForm()
-    {
-        Text = "gatuna";
-        Width = 520;
-        Height = 380;
-        FormBorderStyle = FormBorderStyle.FixedSingle;
-        MaximizeBox = false;
-        MinimizeBox = false;
-        StartPosition = FormStartPosition.CenterScreen;
-        InitializeComponents();
-
-        _sessions.Log += msg => this.Invoke(() => _statusLabel.Text = msg);
-        _sessions.ManifestReceived += entries => this.Invoke(() => PopulateList(entries));
-
-        foreach (var d in TunnelLink.ListDevices())
-            _deviceBox.Items.Add($"{d.Name} — {d.Interface?.FriendlyName ?? d.Interface?.Description}");
-        if (_deviceBox.Items.Count > 0)
-            _deviceBox.SelectedIndex = 0;
-    }
-
-    void InitializeComponents()
-    {
-        Controls.Add(new Label { Text = "Adapter:", Left = 12, Top = 12, AutoSize = true });
-
-        _deviceBox.Left = 70; _deviceBox.Top = 9;
-        _deviceBox.Width = 330; _deviceBox.DropDownStyle = ComboBoxStyle.DropDownList;
-        Controls.Add(_deviceBox);
-
-        _discoverBtn.Text = "Discover";
-        _discoverBtn.Left = 410; _discoverBtn.Top = 8; _discoverBtn.Width = 80;
-        _discoverBtn.Click += OnDiscover;
-        Controls.Add(_discoverBtn);
-
-        _listView.Left = 12; _listView.Top = 40;
-        _listView.Width = 478; _listView.Height = 250;
-        _listView.View = View.Details;
-        _listView.FullRowSelect = true;
-        _listView.CheckBoxes = true;
-        _listView.Columns.Add("ID", 36);
-        _listView.Columns.Add("Proto", 50);
-        _listView.Columns.Add("Port", 56);
-        _listView.Columns.Add("Label", 160);
-        _listView.Columns.Add("Mirror", 70);
-        _listView.ItemChecked += OnItemChecked;
-        Controls.Add(_listView);
-
-        _statusLabel.Left = 12; _statusLabel.Top = 304;
-        _statusLabel.Width = 478; _statusLabel.AutoEllipsis = true;
-        Controls.Add(_statusLabel);
-    }
-
-    void OnDiscover(object? s, EventArgs e)
-    {
-        if (_deviceBox.SelectedIndex < 0)
-        {
-            MessageBox.Show("Select a network adapter first.");
-            return;
-        }
-
-        if (_link != null)
-        {
-            _sessions.StopAll();
-            _sessions.DetachLink();
-            _link.Dispose();
-        }
-
-        var devices = TunnelLink.ListDevices();
-        var device = devices[_deviceBox.SelectedIndex];
-        _link = new TunnelLink(device);
-        _link.Log += msg => this.Invoke(() => _statusLabel.Text = msg);
-        _sessions.AttachLink(_link);
-        _link.Open();
-        _sessions.Discover();
-        _statusLabel.Text = "discovering...";
-    }
-
-    void PopulateList(UpstreamEntry[] entries)
-    {
-        _listView.BeginUpdate();
-        _listView.Items.Clear();
-        foreach (var up in entries)
-        {
-            var item = new ListViewItem(up.Id.ToString());
-            item.SubItems.Add(up.ProtoName);
-            item.SubItems.Add(up.Port.ToString());
-            item.SubItems.Add(up.Label ?? "");
-            item.SubItems.Add("");
-            item.Tag = up;
-            _listView.Items.Add(item);
-        }
-        _listView.EndUpdate();
-    }
-
-    void OnItemChecked(object? s, ItemCheckedEventArgs e)
-    {
-        if (e.Item.Tag is not UpstreamEntry up) return;
-        if (e.Item.Checked)
-        {
-            var port = _sessions.StartListener(up);
-            e.Item.SubItems[4].Text = port.ToString();
-        }
-        else
-        {
-            if (int.TryParse(e.Item.SubItems[4].Text, out var port) && port > 0)
-            {
-                _sessions.StopListener(port);
-                e.Item.SubItems[4].Text = "";
-            }
-        }
-    }
-
-    protected override void OnFormClosing(FormClosingEventArgs e)
-    {
-        if (e.CloseReason == CloseReason.UserClosing)
-        {
-            e.Cancel = true;
-            Hide();
-            return;
-        }
-        base.OnFormClosing(e);
-    }
-
-    public void Shutdown()
-    {
-        _sessions.Dispose();
-        _link?.Dispose();
-    }
-}

win/gatuna-client/SessionManager.cs

@@ -1,284 +0,0 @@
-using System.Collections.Concurrent;
-using System.Net;
-using System.Net.Sockets;
-using System.Threading.Channels;
-
-namespace gatuna_client;
-
-sealed class SessionManager : IDisposable
-{
-    TunnelLink? _link;
-    readonly ConcurrentDictionary<uint, Session> _sessions = new();
-    readonly ConcurrentDictionary<int, ListenerState> _listeners = new();
-
-    byte[]? _serverMac;
-    UpstreamEntry[] _upstreams = [];
-
-    // Serialized OPEN: only one outstanding at a time.
-    readonly object _openLock = new();
-    PendingOpen? _pending;
-    readonly Queue<PendingOpen> _openQueue = new();
-
-    public event Action<string>? Log;
-    public event Action<UpstreamEntry[]>? ManifestReceived;
-
-    public UpstreamEntry[] Upstreams => _upstreams;
-    public byte[]? ServerMac => _serverMac;
-    public TunnelLink? Link => _link;
-
-    public void AttachLink(TunnelLink link)
-    {
-        _link = link;
-        link.FrameReceived += HandleFrame;
-    }
-
-    public void DetachLink()
-    {
-        if (_link != null)
-            _link.FrameReceived -= HandleFrame;
-        _link = null;
-    }
-
-    public void Discover()
-    {
-        if (_link == null) return;
-        _link.SendBroadcast(new Frame.Discover());
-    }
-
-    public void HandleFrame(Frame frame, byte[] srcMac)
-    {
-        switch (frame)
-        {
-            case Frame.Manifest manifest:
-                _serverMac = srcMac;
-                _upstreams = manifest.Entries;
-                Log?.Invoke($"manifest: {manifest.Entries.Length} upstreams from {BitConverter.ToString(srcMac)}");
-                ManifestReceived?.Invoke(manifest.Entries);
-                break;
-
-            case Frame.OpenAck ack:
-                HandleOpenAck(ack, srcMac);
-                break;
-
-            case Frame.OpenNak nak:
-                Log?.Invoke($"OPEN_NAK upstream {nak.UpstreamId} reason {nak.Reason}");
-                lock (_openLock)
-                {
-                    if (_pending != null)
-                        _pending.Client.Dispose();
-                }
-                ProcessQueue();
-                break;
-
-            case Frame.Data data:
-                if (_sessions.TryGetValue(data.SessionId, out var session))
-                    session.Deliver(data.Payload);
-                else if (_link != null && _serverMac != null)
-                    _link.SendTo(_serverMac,
-                        new Frame.Close(data.SessionId, Proto.ReasonUnknownSession));
-                break;
-
-            case Frame.Close close:
-                if (_sessions.TryRemove(close.SessionId, out var s))
-                    s.Dispose();
-                break;
-        }
-    }
-
-    /// <summary>
-    /// Start a local TCP listener for the given upstream. Returns the mirror
-    /// port, or 0 on failure.
-    /// </summary>
-    public int StartListener(UpstreamEntry upstream)
-    {
-        var listener = new TcpListener(IPAddress.Loopback, 0);
-        listener.Start();
-        var port = ((IPEndPoint)listener.LocalEndpoint).Port;
-        var state = new ListenerState(listener, upstream);
-        _listeners[port] = state;
-        _ = AcceptLoop(state);
-        Log?.Invoke($"listening 127.0.0.1:{port} -> upstream {upstream.Id} ({upstream.ProtoName}:{upstream.Port})");
-        return port;
-    }
-
-    public void StopListener(int port)
-    {
-        if (_listeners.TryRemove(port, out var state))
-        {
-            state.Listener.Stop();
-            Log?.Invoke($"stopped listener port {port}");
-        }
-    }
-
-    async Task AcceptLoop(ListenerState state)
-    {
-        while (true)
-        {
-            TcpClient client;
-            try
-            {
-                client = await state.Listener.AcceptTcpClientAsync();
-            }
-            catch { break; }
-            EnqueueOpen(client, state.Upstream.Id);
-        }
-    }
-
-    void EnqueueOpen(TcpClient client, byte upstreamId)
-    {
-        lock (_openLock)
-        {
-            if (_pending == null)
-            {
-                _pending = new PendingOpen(client, upstreamId);
-                SendOpen(_pending);
-            }
-            else
-            {
-                _openQueue.Enqueue(new PendingOpen(client, upstreamId));
-            }
-        }
-    }
-
-    void SendOpen(PendingOpen po)
-    {
-        if (_link == null || _serverMac == null)
-        {
-            Log?.Invoke("no server; cannot OPEN");
-            po.Client.Dispose();
-            return;
-        }
-        _link.SendTo(_serverMac, new Frame.Open(po.UpstreamId));
-    }
-
-    void ProcessQueue()
-    {
-        lock (_openLock)
-        {
-            if (_openQueue.Count > 0)
-            {
-                _pending = _openQueue.Dequeue();
-                SendOpen(_pending);
-            }
-            else
-            {
-                _pending = null;
-            }
-        }
-    }
-
-    void HandleOpenAck(Frame.OpenAck ack, byte[] srcMac)
-    {
-        PendingOpen? po;
-        lock (_openLock)
-            po = _pending;
-
-        if (po == null || po.UpstreamId != ack.UpstreamId)
-        {
-            Log?.Invoke($"OPEN_ACK upstream {ack.UpstreamId} session {ack.SessionId} — no matching pending");
-            return;
-        }
-
-        var session = new Session(
-            ack.SessionId, po.Client, srcMac, _link!,
-            () => _sessions.TryRemove(ack.SessionId, out _),
-            msg => Log?.Invoke(msg));
-        _sessions[ack.SessionId] = session;
-        session.Start();
-        Log?.Invoke($"session {ack.SessionId} upstream {ack.UpstreamId} established");
-        ProcessQueue();
-    }
-
-    public void StopAll()
-    {
-        foreach (var kv in _listeners)
-            kv.Value.Listener.Stop();
-        _listeners.Clear();
-        foreach (var s in _sessions.Values)
-            s.Dispose();
-        _sessions.Clear();
-    }
-
-    public void Dispose()
-    {
-        DetachLink();
-        StopAll();
-    }
-}
-
-sealed class ListenerState(TcpListener listener, UpstreamEntry upstream)
-{
-    public TcpListener Listener { get; } = listener;
-    public UpstreamEntry Upstream { get; } = upstream;
-}
-
-sealed class PendingOpen(TcpClient client, byte upstreamId)
-{
-    public TcpClient Client { get; } = client;
-    public byte UpstreamId { get; } = upstreamId;
-}
-
-sealed class Session(
-    uint sessionId,
-    TcpClient client,
-    byte[] serverMac,
-    TunnelLink link,
-    Action onClosed,
-    Action<string>? log) : IDisposable
-{
-    readonly CancellationTokenSource _cts = new();
-    readonly Channel<byte[]> _incoming = Channel.CreateBounded<byte[]>(256);
-
-    public void Start()
-    {
-        _ = PumpSocketToTunnel();
-        _ = PumpTunnelToSocket();
-    }
-
-    public void Deliver(byte[] payload)
-    {
-        if (!_incoming.Writer.TryWrite(payload))
-            log?.Invoke($"session {sessionId}: incoming channel full");
-    }
-
-    async Task PumpSocketToTunnel()
-    {
-        try
-        {
-            var stream = client.GetStream();
-            var buf = new byte[Proto.MaxPayload];
-            using var reg = _cts.Token.Register(() => client.Dispose());
-            while (!_cts.IsCancellationRequested)
-            {
-                var n = await stream.ReadAsync(buf, _cts.Token);
-                if (n == 0) break;
-                link.SendTo(serverMac, new Frame.Data(sessionId, buf[..n]));
-            }
-        }
-        catch { }
-        SendClose();
-        onClosed();
-    }
-
-    async Task PumpTunnelToSocket()
-    {
-        try
-        {
-            var stream = client.GetStream();
-            await foreach (var payload in _incoming.Reader.ReadAllAsync(_cts.Token))
-                await stream.WriteAsync(payload, _cts.Token);
-        }
-        catch { }
-    }
-
-    void SendClose() => link.SendTo(serverMac, new Frame.Close(sessionId, null));
-
-    public void Dispose()
-    {
-        _cts.Cancel();
-        _incoming.Writer.TryComplete();
-        SendClose();
-        try { client.Dispose(); } catch { }
-        _cts.Dispose();
-    }
-}