Socket

Linear (QTT) Socket library — TCP/UDP via Python FFI. Every socket-consuming operation takes its socket parameter at multiplicity 1 (``(1 s: ...)``), so the type checker enforces the resource discipline at the binder. A ``let 1 s = stream_socket unit in body`` *must* consume ``s`` exactly once: passing it to ``stream_bind``, ``stream_close``, etc. discharges the linear obligation. Forgetting to close any handle, or using one after consumption, is rejected at type-check time. In addition to the linear discipline, refinements track the state-machine: ``stream_socket`` produces a fresh-not-bound-not- connected socket; ``stream_bind`` / ``stream_connect`` advance the state; ``stream_recv`` / ``stream_send`` require connected. Opaque Python wrappers (see native bodies below). Both are ``linear``: a socket handle is unique, so every binder holding one must be declared at multiplicity 1 and consume it exactly once.
Table of Contents

Types

StreamSocket

(type) linear
Opaque Python wrappers (see native bodies below). Both are ``linear``: a socket handle is unique, so every binder holding one must be declared at multiplicity 1 and consume it exactly once.
linear type StreamSocket

DatagramSocket

(type) linear
linear type DatagramSocket

SocketPayload

(type)
Raw payload (`bytes` at runtime).
type SocketPayload

SockAddr

(type)
Runtime: address tuple, e.g. `(host, port)` for `AF_INET`.
type SockAddr

TcpAccept

(type) linear
Runtime: `(socket, addr)` pair from `TCP accept`.
linear type TcpAccept

UdpRecvFrom

(type)
Runtime: `(data, addr)` pair from `UDP recvfrom`.
type UdpRecvFrom

Functions

socket

Python `socket` module (binds to the evaluator name `socket`).
def socket : Unit

ipv4_addr

def ipv4_addr (host : String) (port : Int | port ≥ 0 && port ≤ 65535) : {a : SockAddr | ipv4_sock_addr a = true}

stream_socket

def stream_socket (_ : Unit) : {s : StreamSocket | stream_bound s = false && stream_connected s = false}

dgram_socket

def dgram_socket (_ : Unit) : {s : DatagramSocket | dgram_bound s = false && dgram_connected s = false}

tcp_accept_socket

def tcp_accept_socket (1 r : TcpAccept) : {s : StreamSocket | stream_connected s = true}

tcp_accept_peer

def tcp_accept_peer (borrow r : TcpAccept) : {a : SockAddr | ipv4_sock_addr a = true}

stream_bind

def stream_bind (addr : SockAddr | ipv4_sock_addr addr = true) (1 s : StreamSocket | stream_bound s = false) : {out : StreamSocket | stream_bound out = true}

stream_listen

def stream_listen (backlog : Int | backlog > 0) (1 s : StreamSocket | stream_bound s = true) : {out : StreamSocket | stream_bound out = true}

stream_connect

def stream_connect (addr : SockAddr | ipv4_sock_addr addr = true) (1 s : StreamSocket | stream_connected s = false) : {out : StreamSocket | stream_connected out = true}

stream_accept

def stream_accept (1 s : StreamSocket | stream_bound s = true && stream_connected s = false) : TcpAccept

stream_send

def stream_send (data : SocketPayload) (1 s : StreamSocket | stream_connected s = true) : {sent : Int | sent ≥ 0 && sent ≤ payload_len data}

stream_recv

def stream_recv (bufsize : Int | bufsize > 0) (1 s : StreamSocket | stream_connected s = true) : {p : SocketPayload | payload_len p ≤ bufsize}

stream_close

``stream_close`` consumes the socket and returns ``Unit`` so the closed socket cannot be used again — it has nowhere to go.
def stream_close (1 s : StreamSocket) : Unit

dgram_bind

def dgram_bind (addr : SockAddr | ipv4_sock_addr addr = true) (1 s : DatagramSocket | dgram_bound s = false) : {out : DatagramSocket | dgram_bound out = true}

dgram_connect

def dgram_connect (addr : SockAddr | ipv4_sock_addr addr = true) (1 s : DatagramSocket) : {out : DatagramSocket | dgram_connected out = true}

dgram_sendto

def dgram_sendto (data : SocketPayload) (addr : SockAddr | ipv4_sock_addr addr = true) (1 s : DatagramSocket) : {sent : Int | sent ≥ 0 && sent ≤ payload_len data}

dgram_recvfrom

def dgram_recvfrom (bufsize : Int | bufsize > 0) (1 s : DatagramSocket | dgram_bound s = true) : {r : UdpRecvFrom | udp_payload_len r ≤ bufsize}

udp_recv_payload

def udp_recv_payload (r : UdpRecvFrom) : {p : SocketPayload | payload_len p = udp_payload_len r}

udp_recv_peer

def udp_recv_peer (r : UdpRecvFrom) : {a : SockAddr | ipv4_sock_addr a = true}

dgram_send

def dgram_send (data : SocketPayload) (1 s : DatagramSocket | dgram_connected s = true) : {sent : Int | sent ≥ 0 && sent ≤ payload_len data}

dgram_recv

def dgram_recv (bufsize : Int | bufsize > 0) (1 s : DatagramSocket | dgram_connected s = true) : {p : SocketPayload | payload_len p ≤ bufsize}

dgram_close

def dgram_close (1 s : DatagramSocket) : Unit

Uninterpreted

Functions declared as def f ... = uninterpreted: only their signature is known to the verifier; they have no body.

stream_bound

uninterpreted
def stream_bound : (s : StreamSocket) → Bool

stream_connected

uninterpreted
def stream_connected : (s : StreamSocket) → Bool

dgram_bound

uninterpreted
def dgram_bound : (s : DatagramSocket) → Bool

dgram_connected

uninterpreted
def dgram_connected : (s : DatagramSocket) → Bool

payload_len

uninterpreted
Byte length of a payload (`len(data)` at runtime).
def payload_len : (p : SocketPayload) → Int

udp_payload_len

uninterpreted
Length of the datagram payload in a `recvfrom` result.
def udp_payload_len : (r : UdpRecvFrom) → Int

ipv4_sock_addr

uninterpreted
Constructed IPv4 text endpoint (`(host, port)`).
def ipv4_sock_addr : (a : SockAddr) → Bool