///|
#warnings("-alert_internal")
fn uuid_v4() -> @uuid.UUID raise {
  let bytes = @tls.rand_bytes(16)
  @uuid.from_bytes(bytes).as_version(@uuid.V4)
}

///|
fn build_command() -> @argparse.Command {
  @argparse.Command::new(
    "tun-server",
    about="TUN server that connects to a relay via WebSocket",
    options=[
      @argparse.OptionArg::new(
        "relay",
        short='r',
        long="relay",
        about="Relay host (optionally with :port, e.g. 127.0.0.1:8080)",
        default_values=["tun-poc.moonbitlang.com"],
      ),
    ],
    positionals=[
      @argparse.PositionArg::new(
        "child",
        num_args=@argparse.ValueRange::new(lower=0),
        allow_hyphen_values=true,
        about="Program to spawn (defaults to $SHELL or /bin/sh on Unix, %COMSPEC% or cmd.exe on Windows). Use `--` to pass flags to the program.",
      ),
    ],
  )
}

///|
/// True for hosts that should use plaintext `ws://` / `http://` instead of
/// TLS. Mirrors `isLocalHost` in `android/.../TerminalScreen.kt` so both
/// clients classify the same host identically.
fn is_local_host(host : String) -> Bool {
  host.has_prefix("127.") ||
  host.has_prefix("10.") ||
  host.has_prefix("192.168.") ||
  host.has_prefix("localhost")
}

///|
/// Build the relay WebSocket URL from a bare host (optionally with
/// `:port`, e.g. `127.0.0.1:8080`). Picks `ws://` for local-style hosts
/// and `wss://` otherwise, appends the fixed `/server` suffix, and parses
/// the result so downstream code gets a structured `@url.Url` and we fail
/// fast on a malformed `--relay` argument.
fn relay_url_from_host(host : String) -> @url.Url raise {
  let scheme = if is_local_host(host) { "ws" } else { "wss" }
  @url.Url::parse("\{scheme}://\{host}/server")
}

///|
/// Raised by `pty_reader` when the local shell exits. Propagates up through
/// the outer `with_task_group` in `run_bridge`, cancelling all bridge tasks
/// and the retry loop so `main` returns cleanly.
priv suberror ShellExited derive(Debug)

///|
/// Persistent PTY reader. Spawned once at startup and lives for the entire
/// session, independent of the relay. It tees output to the local terminal
/// **regardless of relay state** and broadcasts to remote clients only when
/// a relay session is active. This is what makes the local shell usable
/// even when the relay is down or unreachable.
///
/// On EOF (local shell exited / `exit` / `Ctrl-D`), it raises `ShellExited`,
/// which propagates out through the outer task group and shuts the server
/// down cleanly.
async fn pty_reader(
  pty : @pty.Pty,
  pty_ref : Ref[@pty.Pty?],
  clients : Map[String, Unit],
  send_queue_ref : Ref[@aqueue.Queue[String]?],
  shell_exited_ref : Ref[Bool],
  log : Array[(Int64, Bytes)],
  next_id : Ref[Int64],
) -> Unit {
  @log.info("pty_reader: starting, pid=\{pty.pid()}")
  let pid = pty.pid()
  // Use a task group so that when wait_pid fires (child exited), the
  // blocked read loop is cancelled — ConPTY on Windows does not deliver
  // EOF on the output pipe until ClosePseudoConsole, so we cannot rely
  // on read returning 0 to detect child exit.
  @async.with_task_group(async fn(group) {
    // Background task: wait for the child process to exit.
    group.spawn_bg(async fn() {
      let exit_code = @process.wait_pid(pid)
      @log.info("pty_reader: child exited(\{exit_code}), pid=\{pid}")
      pty_ref.val = None
      shell_exited_ref.val = true
      raise ShellExited
    })
    // Foreground: read loop.
    let reader = pty.reader()
    let buf : FixedArray[Byte] = FixedArray::make(4096, b'\x00')
    for ;; {
      let n = reader.read(buf)
      @log.info("pty_reader: read returned n=\{n}")
      if n > 0 {
        let bytes = Bytes::from_array(buf[:n])
        // Append to the id-indexed log for delta replay on reconnect.
        // Each PTY chunk gets a dense id starting at 1L. Callers that
        // reconnect send their highest-seen id as Connect.last_seen_id;
        // NewClient handling slices the tail of this log to replay.
        next_id.val = next_id.val + 1L
        let id = next_id.val
        log.push((id, bytes))
        // Local tee — always, even when no relay session is active.
        let written = @stdio.fwrite(bytes, 1, n.to_uint64(), @stdio.stdout)
        @log.info("pty_reader: fwrite n=\{n} -> written=\{written}")
        ignore(try? @stdio.fflush(@stdio.stdout))
        // Remote broadcast — only when a relay session is connected.
        if send_queue_ref.val is Some(send_queue) {
          let data = @base64.encode(bytes[:])
          for client_id, _ in clients {
            let msg : @proto.Message = Term(Data(client_id~, id~, data~))
            send_queue.put(msg.to_json().stringify()) catch {
              err if @async.is_being_cancelled() => raise err
              _ => ()
            }
          }
        }
      } else {
        // EOF on Unix — child exited and pipe closed.
        @log.info("pty_reader: read returned \{n} (EOF), pid=\{pid}")
        pty_ref.val = None
        shell_exited_ref.val = true
        raise ShellExited
      }
    }
  })
}

///|
/// Read from local stdin and forward every chunk to the PTY master. Exits
/// when stdin EOFs or when the task group cancels this task. Uses
/// `@async/stdio.stdin` directly so the same code works on Unix (epoll on
/// fd 0) and Windows (worker-thread blocking read on the console HANDLE).
async fn local_stdin_task(pty_ref : Ref[@pty.Pty?]) -> Unit {
  let buf : FixedArray[Byte] = FixedArray::make(4096, b'\x00')
  for ;; {
    let n = @async/stdio.stdin.read(buf)
    if n <= 0 {
      break
    }
    if pty_ref.val is Some(pty) {
      pty.write(Bytes::from_array(buf[:n])) catch {
        err if @async.is_being_cancelled() => raise err
        _ => @log.error("local stdin -> pty write failed")
      }
    }
  }
  @log.info("local_stdin_task: exited")
}

///|
/// Connect to the relay and send the initial Register message.
///
/// Returns an open WebSocket only after the relay has accepted the
/// registration with `Registered(...)`. Any other response is treated as an
/// error so the caller can retry or fail before touching tty/pty state.
async fn connect_and_register(
  url : @url.Url,
  headers : Map[String, String],
  task_uuid : @uuid.UUID,
  hostname : String,
  cwd : String,
) -> @ws.Conn {
  @log.info("Connecting to relay: \{url}")
  let ws = @ws.Conn::connect(url.to_string(), headers~)
  let register_msg : @proto.Message = Sys(Register(task_uuid~, hostname~, cwd~))
  ws.send_text(register_msg.to_json().stringify()) catch {
    err => {
      ws.close()
      raise err
    }
  }
  @log.info("Sent: register")
  let response = ws.recv().read_all().text() catch {
      err => {
        ws.close()
        raise err
      }
    }
  @log.info("Received: \{response}")
  let msg : @proto.Message = @json.from_json(@json.parse(response)) catch {
    err => {
      ws.close()
      raise err
    }
  }
  match msg {
    Sys(Registered(task_uuid=echoed_uuid)) => {
      if echoed_uuid != task_uuid {
        ws.close()
        fail(
          "Registered echoed wrong uuid: got \{echoed_uuid}, expected \{task_uuid}",
        )
      }
      @log.info("Registered task_uuid=\{task_uuid}")
      ws
    }
    _ => {
      ws.close()
      fail("unexpected register response: \{@debug.to_string(msg)}")
    }
  }
}

///|
fn is_auth_handshake_rejection(code : Int) -> Bool {
  code == 401 || code == 403
}

///|
/// Retry relay connection / registration until a session is established or a
/// fatal error occurs. This runs before tty/pty setup so early failures stay
/// visible on the terminal.
async fn connect_retry_loop(
  url : @url.Url,
  headers : Map[String, String],
  task_uuid : @uuid.UUID,
  hostname : String,
  cwd : String,
) -> @ws.Conn? {
  Some(
    @async.retry(
      ExponentialDelay(initial=1000, factor=2.0, maximum=5000),
      fatal_error=fn(err) {
        match err {
          @ws.HandshakeRejected(_, response) =>
            is_auth_handshake_rejection(response.code)
          _ => false
        }
      },
      () => connect_and_register(url, headers, task_uuid, hostname, cwd),
    ),
  ) catch {
    err if @async.is_being_cancelled() => raise err
    @ws.HandshakeRejected(_, response) if is_auth_handshake_rejection(
        response.code,
      ) => {
      remove_stale_credential(credential_path(url))
      @log.error(
        "relay rejected credential (this server may have been revoked); " +
        "removed stale credential, re-run tun-server to re-pair",
      )
      None
    }
    @ws.HandshakeRejected(msg, response) => {
      @log.error(
        "Fatal: relay handshake rejected: \{response.code} \{response.reason}: \{msg}",
      )
      None
    }
    @ws.InvalidHandshake(msg) => {
      @log.error("Fatal: relay handshake failed: \{msg}")
      None
    }
    err => {
      @log.error("Fatal: \{err}")
      None
    }
  }
}

///|
/// Run the relay retry loop. The shared `clients` / `send_queue_ref` are
/// owned by the caller and also read by `pty_reader` so the local tty
/// bridge keeps working across reconnects.
async fn run_retry_loop(
  url : @url.Url,
  headers : Map[String, String],
  task_uuid : @uuid.UUID,
  hostname : String,
  cwd : String,
  pty_ref : Ref[@pty.Pty?],
  clients : Map[String, Unit],
  send_queue_ref : Ref[@aqueue.Queue[String]?],
  shell_exited_ref : Ref[Bool],
  log : Array[(Int64, Bytes)],
  first_ws? : @ws.Conn,
) -> Unit {
  match first_ws {
    Some(ws) =>
      run_session(ws, pty_ref, clients, send_queue_ref, shell_exited_ref, log) catch {
        err if @async.is_being_cancelled() => raise err
        err => @log.info("relay session lost, will reconnect: \{err}")
      }
    None => ()
  }
  @async.retry(
    ExponentialDelay(initial=1000, factor=2.0, maximum=5000),
    fatal_error=fn(err) {
      match err {
        @ws.HandshakeRejected(_, response) =>
          is_auth_handshake_rejection(response.code)
        _ => false
      }
    },
    () => {
      let ws = connect_and_register(url, headers, task_uuid, hostname, cwd)
      run_session(ws, pty_ref, clients, send_queue_ref, shell_exited_ref, log)
    },
  ) catch {
    err if @async.is_being_cancelled() => raise err
    @ws.HandshakeRejected(_, response) if is_auth_handshake_rejection(
        response.code,
      ) => {
      remove_stale_credential(credential_path(url))
      @log.error(
        "relay rejected credential (this server may have been revoked); " +
        "removed stale credential, re-run tun-server to re-pair",
      )
    }
    @ws.HandshakeRejected(msg, response) =>
      @log.error(
        "Fatal: relay handshake rejected: \{response.code} \{response.reason}: \{msg}",
      )
    @ws.InvalidHandshake(msg) =>
      @log.error("Fatal: relay handshake failed: \{msg}")
    err => @log.error("Fatal: \{err}")
  }
}

///|
/// A single registered relay session. The caller owns connection setup and
/// passes an already-accepted WebSocket in; this function only runs the
/// session loop and raises on connection loss so the caller can retry.
async fn run_session(
  ws : @ws.Conn,
  pty_ref : Ref[@pty.Pty?],
  clients : Map[String, Unit],
  send_queue_ref : Ref[@aqueue.Queue[String]?],
  shell_exited_ref : Ref[Bool],
  log : Array[(Int64, Bytes)],
) -> Unit {
  defer ws.close()
  // Event loop with shared PTY
  @log.info("Entering message loop...")
  @async.with_task_group(async fn(group) {
    // Single send queue serializes all outgoing WebSocket messages.
    // Publish it via send_queue_ref so the persistent pty_reader can
    // broadcast pty bytes to remote clients during this session. The defer
    // clears it on disconnect so the reader knows there is no active session
    // and stops broadcasting.
    let send_queue : @aqueue.Queue[String] = @aqueue.Queue::new(kind=Unbounded)
    send_queue_ref.val = Some(send_queue)
    // Graceful-shutdown defer. pty_reader sets `shell_exited_ref.val = true`
    // when the local shell EOFs; by the time we reach here the surrounding
    // task group is already being cancelled (pty_reader raised ShellExited
    // up at run_bridge's outer group), but the ws is still open — outer
    // `defer ws.close()` only runs after run_session returns. We
    // `protect_from_cancel` the send so it survives the cancellation.
    group.add_defer(async fn() {
      if shell_exited_ref.val {
        @async.protect_from_cancel(async fn() {
          let exit_msg : @proto.Message = Sys(TaskExited(exit_code=0))
          ws.send_text(exit_msg.to_json().stringify()) catch {
            _ => ()
          }
        })
      }
      send_queue_ref.val = None
      clients.clear()
    })
    group.spawn_bg(async fn() {
      for ;; {
        let msg = send_queue.get() catch {
          err if @async.is_being_cancelled() => raise err
          _ => break
        }
        ws.send_text(msg) catch {
          err if @async.is_being_cancelled() => raise err
          _ => ()
        }
      }
    })
    for ;; {
      let text = ws.recv().read_all().text()
      let incoming : @proto.Message = @json.from_json(@json.parse(text)) catch {
        err => {
          @log.error(
            "Failed to decode incoming message: raw=\{@debug.to_string(text)}, error=\{@debug.to_string(err)}",
          )
          continue
        }
      }
      handle_message(incoming, pty_ref, clients, send_queue, log) catch {
        err if @async.is_being_cancelled() => raise err
        err =>
          @log.error(
            "Error processing message: message=\{incoming.to_log_string()}, raw=\{@debug.to_string(text)}, error=\{@debug.to_string(err)}",
          )
      }
    }
  })
}

///|
/// Move verbose runtime logs to the default file sink once the local tty has
/// entered raw mode so PTY traffic owns the terminal. Failures here are
/// non-fatal and continue on the current sink.
async fn maybe_redirect_log_to_file() -> Unit {
  let log_path = @log.default_log_path()
  try {
    @log.ensure_log_dir_for(log_path)
    @log.set_output_path(log_path, append=true)
  } catch {
    err if @async.is_being_cancelled() => raise err
    err =>
      @log.warn(
        "Failed to prepare log file: \{@debug.to_string(err)}; continuing with current log sink",
      )
  }
}

///|
/// Path to the long-lived credential file that holds the `tp_*` token
/// minted by the device-code flow. `$TUN_CREDENTIAL_PATH` overrides the
/// default for containers / systemd `LoadCredential=`.
///
/// The default path is scoped per relay (`//credential.key`)
/// so a dev relay at `localhost:12380` and production at
/// `tun-poc.moonbitlang.com` keep independent tokens. `:` in the host
/// segment is rewritten to `_` so the subdir name is also valid on Windows.
fn credential_path(relay_url : @url.Url) -> String {
  match @env.get_env_var("TUN_CREDENTIAL_PATH") {
    Some(path) if path != "" => path
    _ => {
      let host = relay_url.host().replace(old=":", new="_")
      "\{@xdg.data_dir()}/\{host}/credential.key"
    }
  }
}

///|
/// Convert a relay WebSocket URL (`ws://…/server` or `wss://…/server`) into
/// the HTTP base URL used for device-flow endpoints (`http://…/` /
/// `https://…/`). Re-parses `ws_url.href()` so the returned URL is a fresh
/// copy — the caller's `ws_url` is never mutated. The pathname is reset to
/// `/` because device-flow endpoints live at `/device/*`, not under
/// `/server`.
fn http_base_from_ws(ws_url : @url.Url) -> @url.Url raise {
  let http_base = @url.Url::parse(ws_url.href())
  let scheme = if ws_url.protocol() == "wss:" { "https:" } else { "http:" }
  http_base.set_protocol(scheme)
  http_base.set_pathname("/")
  http_base
}

///|
/// Read the credential file if it exists. Returns None on missing file or
/// any read error.
async fn read_credential_file(path : String) -> String? {
  let exists = @async/fs.exists(path) catch {
    err if @async.is_being_cancelled() => raise err
    _ => false
  }
  if !exists {
    return None
  }
  let data = @async/fs.read_file(path) catch {
    err if @async.is_being_cancelled() => raise err
    _ => return None
  }
  let text = data.text().trim(char_set=" \t\r\n").to_string()
  if text == "" {
    None
  } else {
    Some(text)
  }
}

///|
/// Recursively ensure a directory exists. Works around a bug in
/// `@async/fs.mkdir(recursive=true)` on Windows where `EEXIST` is not
/// handled during parent-directory recursion.
async fn ensure_dir(path : String) -> Unit {
  @log.info("ensure_dir: \{path}")
  @async/fs.mkdir(path, permission=0o700) catch {
    err if @async.is_being_cancelled() => raise err
    @os_error.OSError(_) as e if e.is_EEXIST() =>
      @log.info("ensure_dir: \{path} already exists")
    @os_error.OSError(_) as e if e.is_ENOENT() => {
      @log.info("ensure_dir: \{path} ENOENT, creating parent")
      let parent = match path.rev_find("/") {
        Some(i) => path.view(start_offset=0, end_offset=i).to_string()
        None => path
      }
      if parent == path {
        raise e
      }
      ensure_dir(parent)
      @async/fs.mkdir(path, permission=0o700) catch {
        err if @async.is_being_cancelled() => raise err
        @os_error.OSError(_) as e if e.is_EEXIST() => ()
        e => raise e
      }
    }
    e => {
      @log.error("ensure_dir: \{path} unexpected error: \{e}")
      raise e
    }
  }
}

///|
/// Best-effort removal of a stale credential file after the relay rejects
/// it. Missing file / permission errors are swallowed — the caller has
/// already logged the rejection and the user will simply be prompted for
/// a fresh device-flow pairing on next run.
async fn remove_stale_credential(path : String) -> Unit {
  @async/fs.remove(path) catch {
    err if @async.is_being_cancelled() => raise err
    _ => ()
  }
}

///|
/// Write the credential file with mode 0600. `@async/fs` doesn't expose
/// rename(2), so we write directly — the file is small and torn writes
/// would just be rejected by the relay on next connect, triggering a
/// re-pair on the next run.
async fn write_credential_file(path : String, token : String) -> Unit {
  // Make sure the parent directory exists (first run after install may
  // not have $XDG_DATA_HOME/tun-poc yet).
  // Note: we use rfind("/") instead of @path.Path::dirname() because
  // the win32 dirname only recognises "\", not "/".
  let parent = match path.rev_find("/") {
    Some(i) => path.view(start_offset=0, end_offset=i).to_string()
    None => path
  }
  ensure_dir(parent)
  @async/fs.write_file(path, @utf8.encode(token[:]), create=0o600)
  // Explicit chmod in case the file already existed with wider perms. POSIX
  // only — Windows file ACLs work differently and `@async/fs.chmod` isn't
  // exposed there. The `create=0o600` above already covers the common path
  // (file did not previously exist).
  chmod_credential_file(path)
}

///|
#cfg(not(platform="windows"))
async fn chmod_credential_file(path : String) -> Unit {
  @async/fs.chmod(path, 0o600) catch {
    err if @async.is_being_cancelled() => raise err
    _ => ()
  }
}

///|
#cfg(platform="windows")
fn chmod_credential_file(path : String) -> Unit {
  ignore(path)
}

///|
/// Raised by `getcwd` when `@env.current_dir()` returns `None`. In practice
/// a process without a valid cwd cannot run, so this is effectively
/// unreachable — the suberror exists so callers can depend on a `String`
/// rather than `String?`.
pub suberror GetcwdFailed derive(Debug)

///|
/// Returns the server process's current working directory. Thin wrapper
/// around `@env.current_dir()`, which reads the kernel's cwd via POSIX
/// `getcwd(2)`. The wrapper exists so callers can depend on a `String`
/// rather than `String?`.
pub fn getcwd() -> String raise GetcwdFailed {
  match @env.current_dir() {
    Some(p) => p
    None => raise GetcwdFailed
  }
}

///|
/// POST a JSON body to `url` and parse the response as JSON. Raises on
/// non-2xx status.
async fn http_post_json(url : String, body : Json) -> Json {
  let body_text = body.stringify()
  let headers : Map[String, String] = { "Content-Type": "application/json" }
  let (response, data) = @http.post(url, body_text, headers~)
  let text = data.text()
  if response.code < 200 || response.code >= 300 {
    fail("HTTP \{response.code}: \{text}")
  }
  @json.parse(text)
}

///|
/// Run the device-code flow: ask the relay for a pairing code, print the
/// verification URL (OSC 8 clickable if the terminal supports it), and
/// poll until the user approves or the code expires.
///
/// Returns the raw `tp_*` access token on success; raises on
/// denial/expiry/network error.
async fn run_device_flow(http_base : @url.Url) -> String {
  let auth_url = @url.Url::parse("/device/authorization", base=http_base).to_string()
  let token_url = @url.Url::parse("/device/token", base=http_base).to_string()
  let hostname = @os.gethostname()
  @log.info("device flow: requesting pairing code hostname=\{hostname}")
  let auth_req : @proto.DeviceAuthorizationRequest = { hostname, }
  let auth_resp_json = http_post_json(auth_url, auth_req.to_json()) catch {
    err if @async.is_being_cancelled() => raise err
    e => fail("device /authorization failed: \{e}")
  }
  let auth_resp : @proto.DeviceAuthorizationResponse = @json.from_json(
    auth_resp_json,
  ) catch {
    e => fail("device /authorization: malformed response: \{e}")
  }
  let device_code = auth_resp.device_code
  let user_code = auth_resp.user_code
  let verification_uri_complete = auth_resp.verification_uri_complete.unwrap_or(
    auth_resp.verification_uri + "?code=" + user_code,
  )
  let expires_in = auth_resp.expires_in
  let mut interval_ms = auth_resp.interval * 1000
  if interval_ms < 1000 {
    interval_ms = 1000
  }
  // OSC 8 hyperlink escape: ESC ] 8 ;;  ESC \  ESC ] 8 ;; ESC \
  // We print the raw URL separately for copy/paste and keep the OSC 8 label
  // short so terminals with partial hyperlink support stay readable.
  let esc = "\u{001B}"
  let st = "\u{001B}\\"
  let clickable = "\{esc}]8;;\{verification_uri_complete}\{st}Open pairing page\{esc}]8;;\{st}"
  println("")
  println("Pair this server by visiting:")
  println("  \{verification_uri_complete}")
  println("If your terminal supports hyperlinks, you can also click:")
  println("  \{clickable}")
  println("Waiting for approval (expires in \{expires_in}s)...")
  let deadline = @async.now() + expires_in.to_int64() * 1000L
  let token_req : @proto.DeviceTokenRequest = {
    grant_type: "urn:ietf:params:oauth:grant-type:device_code",
    device_code,
  }
  while @async.now() < deadline {
    @async.sleep(interval_ms)
    let resp_json = http_post_json(token_url, token_req.to_json()) catch {
      err if @async.is_being_cancelled() => raise err
      e => {
        @log.warn("device /token poll error: \{e}")
        continue
      }
    }
    let resp : @proto.DeviceTokenResponse = @json.from_json(resp_json) catch {
      e => {
        @log.warn("device /token malformed response: \{e}")
        continue
      }
    }
    match resp {
      Success(access_token~, token_type=_) => {
        println("Paired.")
        return access_token
      }
      Error(error=AuthorizationPending) => continue
      Error(error=SlowDown) => {
        interval_ms = interval_ms * 2
        continue
      }
      Error(error=ExpiredToken) => fail("pairing code expired")
      Error(error=AccessDenied) => fail("pairing denied by user")
      Error(error=InvalidRequest) => fail("device /token: invalid_request")
      Error(error=UnsupportedGrantType) =>
        fail("device /token: unsupported_grant_type")
      Error(error=Unknown(s)) => fail("device /token error: \{s}")
    }
  }
  fail("pairing code expired (no approval within \{expires_in}s)")
}

///|
/// Acquire a relay access token: read it from `credential.key` if present,
/// otherwise run the device flow and persist the result.
async fn acquire_token(relay_ws_url : @url.Url) -> String {
  let cred_path = credential_path(relay_ws_url)
  if read_credential_file(cred_path) is Some(existing) {
    @log.info("credential: loaded from \{cred_path}")
    return existing
  }
  @log.info("credential: missing at \{cred_path}, starting device flow")
  let http_base = http_base_from_ws(relay_ws_url)
  let token = run_device_flow(http_base)
  write_credential_file(cred_path, token)
  @log.info("credential: wrote \{cred_path}")
  token
}

///|
async fn main {
  let cmd = build_command()
  let matches = cmd.parse() catch {
    err => {
      @log.error(err.to_string())
      println(cmd.render_help())
      return
    }
  }
  let relay_host = matches.values["relay"][0]
  let url = relay_url_from_host(relay_host) catch {
    e => {
      @log.error("invalid --relay host '\{relay_host}': \{e}")
      return
    }
  }
  // Positional `child` is variadic: empty means "use $SHELL or /bin/sh",
  // non-empty means argv[0] is the program and argv[1..] are its arguments.
  let child_argv : Array[String] = match matches.values.get("child") {
    Some(args) if args.length() > 0 => args
    _ =>
      if @platform.win32 {
        [@env.get_env_var("COMSPEC").unwrap_or("cmd.exe")]
      } else {
        [@env.get_env_var("SHELL").unwrap_or("/bin/sh")]
      }
  }
  // tun-server bridges the launching terminal into the shared session, so
  // both stdin and stdout must be a tty. Backgrounded invocations
  // (nohup / systemd / `docker exec` without -t / `ssh -T`) are rejected up
  // front — wrap them in tmux / screen if you need to detach.
  let in_fd = @async/stdio.stdin.fd()
  let out_fd = @async/stdio.stdout.fd()
  if !@tty.isatty(in_fd) || !@tty.isatty(out_fd) {
    @log.error(
      "tun-server requires a tty on stdin and stdout; run it inside a terminal (tmux/screen if you need to background it)",
    )
    return
  }
  @log.info("Child argv: \{child_argv}")
  // Fetch the access token from `credential.key`, running the device flow
  // on first-run. A failure here is fatal — the device flow prints its
  // own errors and we just propagate.
  let token = acquire_token(url) catch {
    err if @async.is_being_cancelled() => raise err
    e => {
      @log.error("failed to acquire credential: \{e}")
      return
    }
  }
  // Process-lifetime identity for this tun-server invocation. The same
  // `task_uuid` is re-sent across reconnects so the relay's "newer wins"
  // eviction keeps reconnection semantics intact. Generated once here.
  let task_uuid = uuid_v4() catch {
    e => {
      @log.error("failed to generate task uuid: \{e}")
      return
    }
  }
  let hostname = @os.gethostname()
  let cwd = getcwd() catch {
    GetcwdFailed => {
      @log.error("failed to read current directory")
      return
    }
  }
  @log.info("task_uuid=\{task_uuid} hostname=\{hostname} cwd=\{cwd}")
  let pty_ref : Ref[@pty.Pty?] = { val: None }
  let headers : Map[String, String] = { "Authorization": "Bearer \{token}" }
  let first_ws = connect_retry_loop(url, headers, task_uuid, hostname, cwd) catch {
    err if @async.is_being_cancelled() => raise err
    e => {
      @log.error("failed to establish relay session: \{e}")
      return
    }
  }
  guard first_ws is Some(ws) else { return }
  // Reconnect forever with exponential backoff: 1s → 2s → 4s → ... → 30s max
  // Auth failures (401/403) are fatal — don't retry. Proxy errors (502/503)
  // or connection failures are transient and should be retried.
  run_bridge(url, headers, task_uuid, hostname, cwd, child_argv, pty_ref, ws)
}

///|
/// Bridge setup: now that the relay session is established, save termios,
/// set raw mode, redirect verbose logs to the file sink, spawn the PTY-backed
/// child, then run the active session and reconnect loop.
async fn run_bridge(
  url : @url.Url,
  headers : Map[String, String],
  task_uuid : @uuid.UUID,
  hostname : String,
  cwd : String,
  child_argv : Array[String],
  pty_ref : Ref[@pty.Pty?],
  first_ws : @ws.Conn,
) -> Unit {
  let in_fd = @async/stdio.stdin.fd()
  let out_fd = @async/stdio.stdout.fd()
  let saved = @tty.save_termios(in_fd, out_fd)
  @tty.set_raw(saved)
  maybe_redirect_log_to_file()
  @tty.arm_exit_restore(saved)
  defer (match @tty.restore(saved) {
    None => ()
    Some(@tty.TermiosSetFailed(errno~)) =>
      @log.warn("tty restore failed: \{errno.code()} (\{errno.strerror()})")
    Some(err) => @log.warn("tty restore failed: \{err.to_repr()}")
  })
  // The relay has already accepted the session, so it is now safe to take
  // over the local tty and spawn the PTY-backed child. The PTY is fixed at
  // 80x24 for the life of the session — no SIGWINCH handling, no
  // client-driven resize. Remote clients render this grid at whatever font
  // size / CSS scale fits their viewport.
  let pty = @pty.Pty::spawn(child_argv, cols=80, rows=24) catch {
    @pty.SpawnFailed(msg) => {
      @log.error("Failed to spawn child: \{msg}")
      return
    }
    err => {
      @log.error("Failed to spawn child: \{err.to_repr()}")
      return
    }
  }
  pty_ref.val = Some(pty)
  defer {
    pty_ref.val = None
    pty.close()
  }
  @log.info("spawned child pid=\{pty.pid()} size=80x24")
  // State shared across reconnects. pty_reader runs once for the entire
  // process lifetime; run_session sets/clears send_queue_ref per reconnect
  // via add_defer. clients accumulates during a session and is cleared on
  // disconnect by run_session's defer.
  let clients : Map[String, Unit] = {}
  let send_queue_ref : Ref[@aqueue.Queue[String]?] = { val: None }
  // Set to true by `pty_reader` on local shell EOF so `run_session`'s
  // group defer can send a best-effort `TaskExited` before the ws closes.
  let shell_exited_ref : Ref[Bool] = { val: false }
  // Id-indexed log — pty_reader appends (id, bytes) tuples with dense
  // ids starting at 1L; reconnecting clients send their last_seen_id in
  // Connect so the NewClient handler can slice the tail for delta
  // replay. Unbounded by design (no retention cap).
  let log : Array[(Int64, Bytes)] = []
  let next_id : Ref[Int64] = { val: 0L }
  @async.with_task_group(async fn(group) {
    group.spawn_bg(no_wait=true, async fn() { local_stdin_task(pty_ref) })
    // pty_reader runs persistently — tees to local stdout always, broadcasts
    // to remote clients only when a relay session is active. Spawning it
    // here (instead of inside run_session) means the local terminal works
    // the moment tun-server starts, regardless of relay state.
    group.spawn_bg(no_wait=true, async fn() {
      pty_reader(
        pty, pty_ref, clients, send_queue_ref, shell_exited_ref, log, next_id,
      )
    })
    run_retry_loop(
      url,
      headers,
      task_uuid,
      hostname,
      cwd,
      pty_ref,
      clients,
      send_queue_ref,
      shell_exited_ref,
      log,
      first_ws~,
    )
  }) catch {
    ShellExited => @log.info("shell exited, server shutting down")
    err => raise err
  }
}

///|
async fn handle_message(
  incoming : @proto.Message,
  pty_ref : Ref[@pty.Pty?],
  clients : Map[String, Unit],
  send_queue : @aqueue.Queue[String],
  log : Array[(Int64, Bytes)],
) -> Unit {
  match incoming {
    Sys(NewClient(client_id~, last_seen_id~)) => {
      @log.info(
        "NewClient: \{client_id} last_seen_id=\{@debug.to_string(last_seen_id)}",
      )
      // Ordering invariant — pty_reader runs in a separate task and
      // (a) appends (id, bytes) to `log`, (b) broadcasts Term(Data)
      // for those bytes to every entry in `clients` via send_queue.
      // A newly-registered client must:
      //   1. See every log entry with id > last_seen_id replayed in
      //      id-order, then
      //   2. See every live Term(Data) broadcast that pty_reader
      //      produces after this point.
      // The invariant holds iff we (i) register the client in
      // `clients`, (ii) enqueue the replay range, all inside one
      // synchronous block with no awaits. `@aqueue.Queue::put` is
      // synchronous, and pty_reader cannot interleave inside a sync
      // block — so its first Term(Data) for this client is ordered
      // after our replay puts.
      let threshold = match last_seen_id {
        Some(n) => n
        None => 0L
      }
      clients[client_id] = ()
      for entry in log {
        let (id, bytes) = entry
        if id > threshold {
          let data = @base64.encode(bytes[:])
          let msg : @proto.Message = Term(Data(client_id~, id~, data~))
          send_queue.put(msg.to_json().stringify()) catch {
            err if @async.is_being_cancelled() => raise err
            _ => ()
          }
        }
      }
    }
    Term(Data(data~, ..)) =>
      match pty_ref.val {
        Some(pty) =>
          pty.write(@utf8.encode(data)) catch {
            err if @async.is_being_cancelled() => raise err
            _ => @log.error("Write error")
          }
        None => ()
      }
    Sys(ClientDisconnected(client_id~)) => {
      @log.info("ClientDisconnected: \{client_id}")
      clients.remove(client_id)
    }
    Codex(..) =>
      @log.warn(
        "Codex message received but codex app-server spawning is disabled",
      )
    Fs(fs_msg) => {
      let response = @fs.handle_fs(fs_msg)
      send_queue.put(response.to_json().stringify()) catch {
        err if @async.is_being_cancelled() => raise err
        _ => ()
      }
    }
    other => @log.warn("Unhandled message: \{@debug.to_string(other)}")
  }
}