feat: add session, endpoint, pubsub, channel modules

- Session plug with fiber-local storage and kcas for concurrency safety
- Token plug for HMAC signing and AEAD encryption
- Negotiate plug for content negotiation (Accept header handling)
- Endpoint module for application bootstrap (router + plugs + server)
- PubSub module for lock-free topic-based messaging
- Channel module for WebSocket topic subscriptions

dune-project

···
   (digestif (>= 1.2))
   (base64 (>= 3.5))
   (bigstringaf (>= 0.10))
-  (faraday (>= 0.8))
-   (climate (>= 0.9))
+   (faraday (>= 0.8))
+   (kcas (>= 0.7))
+    (climate (>= 0.9))
    (zlib (>= 0.8))
    (zstd (>= 0.4))
    (alcotest (and (>= 1.9) :with-test))

lib/channel.ml

···
+type event = {
+  topic : string;
+  event : string;
+  payload : string;
+  ref : int option;
+}
+
+type t = {
+  ws : Websocket.t;
+  pubsub : Pubsub.t;
+  subscriptions : (string, Pubsub.subscription) Kcas_data.Hashtbl.t;
+  handlers : (string, event -> unit) Kcas_data.Hashtbl.t;
+  mutable next_ref : int;
+}
+
+let parse_event s =
+  let len = String.length s in
+  if len < 2 || s.[0] <> '{' then None
+  else
+    let find_string_field key =
+      let pattern = "\"" ^ key ^ "\":\"" in
+      let plen = String.length pattern in
+      let rec search i =
+        if i + plen >= len then None
+        else if String.sub s i plen = pattern then
+          let start = i + plen in
+          let rec find_end j =
+            if j >= len then None
+            else if s.[j] = '"' && (j = start || s.[j - 1] <> '\\') then
+              Some (String.sub s start (j - start))
+            else find_end (j + 1)
+          in
+          find_end start
+        else search (i + 1)
+      in
+      search 0
+    in
+    let find_int_field key =
+      let pattern = "\"" ^ key ^ "\":" in
+      let plen = String.length pattern in
+      let rec search i =
+        if i + plen >= len then None
+        else if String.sub s i plen = pattern then
+          let start = i + plen in
+          let rec find_end j =
+            if j >= len then j
+            else match s.[j] with '0' .. '9' -> find_end (j + 1) | _ -> j
+          in
+          let end_pos = find_end start in
+          if end_pos > start then
+            int_of_string_opt (String.sub s start (end_pos - start))
+          else None
+        else search (i + 1)
+      in
+      search 0
+    in
+    match (find_string_field "t", find_string_field "e") with
+    | Some topic, Some event ->
+        let payload = Option.value ~default:"" (find_string_field "p") in
+        let ref = find_int_field "r" in
+        Some { topic; event; payload; ref }
+    | _ -> None
+
+let serialize_event ev =
+  let ref_part =
+    match ev.ref with Some r -> Printf.sprintf ",\"r\":%d" r | None -> ""
+  in
+  Printf.sprintf "{\"t\":\"%s\",\"e\":\"%s\",\"p\":\"%s\"%s}" ev.topic ev.event
+    ev.payload ref_part
+
+let create ~ws ~pubsub () =
+  {
+    ws;
+    pubsub;
+    subscriptions = Kcas_data.Hashtbl.create ();
+    handlers = Kcas_data.Hashtbl.create ();
+    next_ref = 0;
+  }
+
+let join t ~topic ~params:_ =
+  if Kcas_data.Hashtbl.mem t.subscriptions topic then Ok ()
+  else
+    let sub =
+      Pubsub.subscribe t.pubsub topic (fun msg ->
+          let ev = { topic; event = "msg"; payload = msg; ref = None } in
+          ignore (Websocket.send_text t.ws (serialize_event ev)))
+    in
+    Kcas_data.Hashtbl.replace t.subscriptions topic sub;
+    Ok ()
+
+let leave t ~topic =
+  match Kcas_data.Hashtbl.find_opt t.subscriptions topic with
+  | None -> ()
+  | Some sub ->
+      Pubsub.unsubscribe t.pubsub sub;
+      Kcas_data.Hashtbl.remove t.subscriptions topic
+
+let push t ~topic ~event ~payload =
+  let ref = t.next_ref in
+  t.next_ref <- t.next_ref + 1;
+  let ev = { topic; event; payload; ref = Some ref } in
+  ignore (Websocket.send_text t.ws (serialize_event ev))
+
+let broadcast t ~topic ~payload = Pubsub.broadcast t.pubsub topic payload
+let on t ~event ~handler = Kcas_data.Hashtbl.replace t.handlers event handler
+
+let handle_message t msg =
+  match parse_event msg with
+  | None -> ()
+  | Some ev -> (
+      match ev.event with
+      | "join" -> ignore (join t ~topic:ev.topic ~params:ev.payload)
+      | "leave" -> leave t ~topic:ev.topic
+      | _ -> (
+          match Kcas_data.Hashtbl.find_opt t.handlers ev.event with
+          | Some handler -> handler ev
+          | None -> ()))
+
+let close t =
+  Kcas_data.Hashtbl.iter
+    (fun _ sub -> Pubsub.unsubscribe t.pubsub sub)
+    t.subscriptions;
+  Kcas_data.Hashtbl.clear t.subscriptions;
+  Websocket.close t.ws
+
+let run t =
+  let rec loop () =
+    match Websocket.recv_message t.ws with
+    | Ok (Websocket.Opcode.Text, msg) ->
+        handle_message t msg;
+        loop ()
+    | Ok (Websocket.Opcode.Binary, msg) ->
+        handle_message t msg;
+        loop ()
+    | Ok _ -> loop ()
+    | Error Websocket.Connection_closed -> close t
+    | Error _ -> close t
+  in
+  loop ()

lib/dune

···
 (library
  (name hcs)
  (public_name hcs)
- (libraries eio eio.unix h1 h2 tls-eio tls ca-certs x509 ptime ptime.clock.os cstruct uri digestif base64 bigstringaf faraday bytesrw bytesrw.zlib bytesrw.zstd))
+ (libraries eio eio.unix h1 h2 tls-eio tls ca-certs x509 ptime ptime.clock.os cstruct uri digestif base64 bigstringaf faraday bytesrw bytesrw.zlib bytesrw.zstd mirage-crypto mirage-crypto-rng kcas kcas_data unix))

lib/endpoint.ml

···
+type config = {
+  port : int;
+  bind : string;
+  domains : int;
+  secret_key_base : string;
+  protocol : Server.protocol;
+  tls : Tls_config.Server.t option;
+  health_check : bool;
+}
+
+let default_config =
+  {
+    port = 8080;
+    bind = "0.0.0.0";
+    domains = 1;
+    secret_key_base = "";
+    protocol = Server.Http1_only;
+    tls = None;
+    health_check = true;
+  }
+
+type params_handler = Router.params -> Server.request -> Server.response
+
+type t = {
+  config : config;
+  plugs : Plug.t list;
+  router : params_handler Router.t option;
+  ws_handler : Server.ws_handler option;
+}
+
+let create config = { config; plugs = []; router = None; ws_handler = None }
+let plug t p = { t with plugs = p :: t.plugs }
+let router t r = { t with router = Some r }
+let websocket t handler = { t with ws_handler = Some handler }
+let not_found_handler _req = Server.respond ~status:`Not_found "Not Found"
+
+let health_handler _req =
+  Server.respond ~status:`OK ~headers:[ ("Content-Type", "text/plain") ] "ok"
+
+let build_handler t =
+  let base_handler =
+    match t.router with
+    | None -> not_found_handler
+    | Some r -> (
+        fun req ->
+          let path =
+            match String.index_opt req.Server.target '?' with
+            | Some i -> String.sub req.target 0 i
+            | None -> req.target
+          in
+          match Router.lookup r ~method_:req.meth ~path with
+          | Some (handler, params) -> handler params req
+          | None -> not_found_handler req)
+  in
+  let with_health =
+    if t.config.health_check then fun req ->
+      if req.Server.target = "/_health" || req.Server.target = "/health" then
+        health_handler req
+      else base_handler req
+    else base_handler
+  in
+  let pipeline = Plug.compose_all (List.rev t.plugs) in
+  Plug.apply pipeline with_health
+
+let start t ~env =
+  let handler = build_handler t in
+  let server_config =
+    {
+      Server.default_config with
+      port = t.config.port;
+      host = t.config.bind;
+      domain_count = t.config.domains;
+      protocol = t.config.protocol;
+      tls = t.config.tls;
+    }
+  in
+  Eio.Switch.run @@ fun sw ->
+  let net = Eio.Stdenv.net env in
+  if t.config.domains > 1 then
+    let domain_mgr = Eio.Stdenv.domain_mgr env in
+    Server.run_parallel ~sw ~net ~domain_mgr ~config:server_config
+      ?ws_handler:t.ws_handler handler
+  else
+    Server.run ~sw ~net ~config:server_config ?ws_handler:t.ws_handler handler

lib/hcs.ml

···
 module Plug = Plug
 (** Plug-based middleware system (Phoenix-style) *)
 
+module Endpoint = Endpoint
+(** Application bootstrap tying router, plugs, and server *)
+
 module Middleware = Middleware
 (** Generic middleware composition (for non-HTTP use cases) *)
 
···
 
 module Websocket = Websocket
 (** WebSocket support *)
+
+module Pubsub = Pubsub
+(** Lock-free topic-based pub/sub messaging *)
+
+module Channel = Channel
+(** WebSocket channel abstraction with topic subscriptions *)
 
 module Codec = Codec
 (** Codec system for serialization/deserialization *)

lib/plug.ml

···
 module Retry = Retry
 module Basic_auth = Basic_auth
 module Csrf = Csrf
+module Negotiate = Negotiate
+module Token = Token
+module Session = Session

lib/plug/negotiate.ml

···
+(** Content negotiation plug.
+
+    Parses Accept headers and selects response format based on client
+    preferences and server capabilities. Follows RFC 7231 content negotiation
+    semantics.
+
+    {1 Usage}
+
+    {[
+      (* In router pipeline *)
+      let pipeline =
+        Plug.Negotiate.create ~formats:[ Json; Html ] () @> Plug.identity
+
+      (* In handler *)
+      let handler req =
+        match get_format req with
+        | Some Json -> Response.json {|{"message": "hello"}|}
+        | Some Html -> Response.html "<h1>Hello</h1>"
+        | _ -> Response.text "hello"
+
+      (* Or use respond helper *)
+      let handler req =
+        respond req
+          ~json:(fun () -> {|{"message": "hello"}|})
+          ~html:(fun () -> "<h1>Hello</h1>")
+    ]} *)
+
+(** {1 Types} *)
+
+type media_type = {
+  type_ : string;  (** e.g., "application" *)
+  subtype : string;  (** e.g., "json" *)
+  quality : float;  (** 0.0 - 1.0, default 1.0 *)
+  params : (string * string) list;  (** Additional parameters *)
+}
+(** Parsed media type from Accept header *)
+
+(** Response format - common formats plus custom *)
+type format =
+  | Json
+  | Html
+  | Text
+  | Xml
+  | Csv
+  | Custom of string  (** Custom MIME type *)
+
+(** Content negotiation errors *)
+type error = Not_acceptable
+
+exception Not_acceptable_exn
+
+(** {1 Format Utilities} *)
+
+(** Get MIME type string for a format *)
+let mime_type_of_format = function
+  | Json -> "application/json"
+  | Html -> "text/html"
+  | Text -> "text/plain"
+  | Xml -> "application/xml"
+  | Csv -> "text/csv"
+  | Custom s -> s
+
+(** Get format from MIME type string *)
+let format_of_mime_type s =
+  match String.lowercase_ascii s with
+  | "application/json" -> Some Json
+  | "text/json" -> Some Json
+  | "text/html" -> Some Html
+  | "text/plain" -> Some Text
+  | "application/xml" -> Some Xml
+  | "text/xml" -> Some Xml
+  | "text/csv" -> Some Csv
+  | _ -> Some (Custom s)
+
+(** Get short name for format (for internal use) *)
+let format_name = function
+  | Json -> "json"
+  | Html -> "html"
+  | Text -> "text"
+  | Xml -> "xml"
+  | Csv -> "csv"
+  | Custom s -> s
+
+(** {1 Accept Header Parsing} *)
+
+(** Skip whitespace in string starting at position i *)
+let skip_ws s i =
+  let len = String.length s in
+  let rec loop i =
+    if i < len && (s.[i] = ' ' || s.[i] = '\t') then loop (i + 1) else i
+  in
+  loop i
+
+(** Parse a token (sequence of non-separator chars) *)
+let parse_token s i =
+  let len = String.length s in
+  let rec loop j =
+    if j >= len then j
+    else
+      match s.[j] with
+      | ' ' | '\t' | ',' | ';' | '=' | '/' -> j
+      | _ -> loop (j + 1)
+  in
+  let j = loop i in
+  (String.sub s i (j - i), j)
+
+(** Parse a quoted string *)
+let parse_quoted s i =
+  let len = String.length s in
+  if i >= len || s.[i] <> '"' then ("", i)
+  else
+    let buf = Buffer.create 32 in
+    let rec loop j =
+      if j >= len then j
+      else if s.[j] = '"' then j + 1
+      else if s.[j] = '\\' && j + 1 < len then begin
+        Buffer.add_char buf s.[j + 1];
+        loop (j + 2)
+      end
+      else begin
+        Buffer.add_char buf s.[j];
+        loop (j + 1)
+      end
+    in
+    let j = loop (i + 1) in
+    (Buffer.contents buf, j)
+
+(** Parse a parameter value (token or quoted-string) *)
+let parse_value s i =
+  let i = skip_ws s i in
+  if i < String.length s && s.[i] = '"' then parse_quoted s i
+  else parse_token s i
+
+(** Parse parameters after media type: ;name=value pairs *)
+let parse_params s i =
+  let len = String.length s in
+  let rec loop i acc quality =
+    let i = skip_ws s i in
+    if i >= len || s.[i] = ',' then (List.rev acc, quality, i)
+    else if s.[i] = ';' then begin
+      let i = skip_ws s (i + 1) in
+      let name, i = parse_token s i in
+      let name = String.lowercase_ascii name in
+      let i = skip_ws s i in
+      if i < len && s.[i] = '=' then begin
+        let value, i = parse_value s (i + 1) in
+        if name = "q" then
+          let q = try Float.of_string value with _ -> 1.0 in
+          let q = Float.max 0.0 (Float.min 1.0 q) in
+          loop i acc q
+        else loop i ((name, value) :: acc) quality
+      end
+      else loop i acc quality
+    end
+    else (List.rev acc, quality, i)
+  in
+  loop i [] 1.0
+
+(** Parse a single media type entry *)
+let parse_media_type s i =
+  let i = skip_ws s i in
+  let type_, i = parse_token s i in
+  let type_ = String.lowercase_ascii type_ in
+  let i = skip_ws s i in
+  if i >= String.length s || s.[i] <> '/' then None
+  else begin
+    let subtype, i = parse_token s (i + 1) in
+    let subtype = String.lowercase_ascii subtype in
+    let params, quality, i = parse_params s i in
+    Some ({ type_; subtype; quality; params }, i)
+  end
+
+(** Parse Accept header into list of media types, sorted by quality (highest
+    first).
+
+    Handles:
+    - Multiple types separated by commas
+    - Quality values (q=0.8)
+    - Parameters (charset=utf-8)
+    - Wildcards (* / * and type/ *)
+
+    @param header Accept header value
+    @return List of media types sorted by quality descending *)
+let parse_accept (header : string) : media_type list =
+  let len = String.length header in
+  let rec loop i acc =
+    if i >= len then acc
+    else
+      match parse_media_type header i with
+      | None ->
+          (* Skip to next comma *)
+          let rec skip j =
+            if j >= len || header.[j] = ',' then j + 1 else skip (j + 1)
+          in
+          loop (skip i) acc
+      | Some (mt, i) ->
+          let i = skip_ws header i in
+          let i = if i < len && header.[i] = ',' then i + 1 else i in
+          loop i (mt :: acc)
+  in
+  let types = loop 0 [] in
+  (* Sort by quality descending, then by specificity (more specific first) *)
+  List.sort
+    (fun a b ->
+      let cmp = Float.compare b.quality a.quality in
+      if cmp <> 0 then cmp
+      else
+        (* More specific types have higher priority *)
+        let specificity mt =
+          if mt.type_ = "*" then 0
+          else if mt.subtype = "*" then 1
+          else 2 + List.length mt.params
+        in
+        Int.compare (specificity b) (specificity a))
+    types
+
+(** {1 Negotiation} *)
+
+(** Check if a media type matches a format *)
+let media_type_matches (mt : media_type) (format : format) : bool =
+  if mt.type_ = "*" && mt.subtype = "*" then true
+  else
+    let mime = mime_type_of_format format in
+    match String.split_on_char '/' mime with
+    | [ t; s ] ->
+        (mt.type_ = "*" || mt.type_ = t) && (mt.subtype = "*" || mt.subtype = s)
+    | _ -> false
+
+(** Negotiate best format from Accept header.
+
+    @param accept Accept header value
+    @param available List of formats the server can produce
+    @return Best matching format or None if no match *)
+let negotiate ~accept ~available : format option =
+  if available = [] then None
+  else if accept = "" || accept = "*/*" then Some (List.hd available)
+  else
+    let types = parse_accept accept in
+    (* Find first accepted type that matches an available format *)
+    let rec find_match types =
+      match types with
+      | [] -> None
+      | mt :: rest -> (
+          if mt.quality <= 0.0 then find_match rest
+          else
+            (* Find first available format matching this media type *)
+            let matching =
+              List.find_opt (fun fmt -> media_type_matches mt fmt) available
+            in
+            match matching with
+            | Some _ as result -> result
+            | None -> find_match rest)
+    in
+    find_match types
+
+(** Negotiate best format, raising exception if no match.
+
+    @param accept Accept header value
+    @param available List of formats the server can produce
+    @return Best matching format
+    @raise Not_acceptable_exn if no format matches *)
+let negotiate_exn ~accept ~available : format =
+  match negotiate ~accept ~available with
+  | Some fmt -> fmt
+  | None -> raise Not_acceptable_exn
+
+(** {1 Request Format Storage} *)
+
+(** Key for storing negotiated format in request. We store format as a header
+    since requests are immutable records. *)
+let format_header = "x-hcs-negotiated-format"
+
+(** Get negotiated format from request (set by Negotiate plug) *)
+let get_format (req : Server.request) : format option =
+  match
+    List.find_opt
+      (fun (n, _) -> String.lowercase_ascii n = format_header)
+      req.headers
+  with
+  | None -> None
+  | Some (_, v) -> (
+      match v with
+      | "json" -> Some Json
+      | "html" -> Some Html
+      | "text" -> Some Text
+      | "xml" -> Some Xml
+      | "csv" -> Some Csv
+      | s -> Some (Custom s))
+
+(** Set negotiated format on request *)
+let set_format (req : Server.request) (fmt : format) : Server.request =
+  let headers =
+    (format_header, format_name fmt)
+    :: List.filter
+         (fun (n, _) -> String.lowercase_ascii n <> format_header)
+         req.headers
+  in
+  { req with headers }
+
+(** {1 Response Helpers} *)
+
+(** Create response with correct Content-Type for format *)
+let respond_format (fmt : format) ~(body : string) : Server.response =
+  let content_type =
+    match fmt with
+    | Json -> "application/json; charset=utf-8"
+    | Html -> "text/html; charset=utf-8"
+    | Text -> "text/plain; charset=utf-8"
+    | Xml -> "application/xml; charset=utf-8"
+    | Csv -> "text/csv; charset=utf-8"
+    | Custom s -> s
+  in
+  {
+    status = `OK;
+    headers =
+      [
+        ("Content-Type", content_type);
+        ("Content-Length", string_of_int (String.length body));
+      ];
+    body = Server.Body_string body;
+  }
+
+(** Respond based on negotiated format with lazy body generation. Only evaluates
+    the body function for the selected format.
+
+    @param req Request with negotiated format
+    @param json Function to generate JSON body
+    @param html Function to generate HTML body
+    @param text Optional function for plain text (defaults to json)
+    @param xml Optional function for XML
+    @return Response with appropriate body and Content-Type *)
+let respond (req : Server.request) ?json ?html ?text ?xml () : Server.response =
+  let fmt = get_format req in
+  match fmt with
+  | Some Json -> (
+      match json with
+      | Some f -> respond_format Json ~body:(f ())
+      | None -> respond_format Text ~body:"")
+  | Some Html -> (
+      match html with
+      | Some f -> respond_format Html ~body:(f ())
+      | None -> (
+          match json with
+          | Some f -> respond_format Json ~body:(f ())
+          | None -> respond_format Text ~body:""))
+  | Some Text -> (
+      match text with
+      | Some f -> respond_format Text ~body:(f ())
+      | None -> (
+          match json with
+          | Some f -> respond_format Json ~body:(f ())
+          | None -> respond_format Text ~body:""))
+  | Some Xml -> (
+      match xml with
+      | Some f -> respond_format Xml ~body:(f ())
+      | None -> respond_format Text ~body:"")
+  | Some Csv -> respond_format Csv ~body:""
+  | Some (Custom _) | None -> (
+      (* Default to first available *)
+      match json with
+      | Some f -> respond_format Json ~body:(f ())
+      | None -> (
+          match html with
+          | Some f -> respond_format Html ~body:(f ())
+          | None -> respond_format Text ~body:""))
+
+(** {1 Plug} *)
+
+(** Create content negotiation plug.
+
+    Parses the Accept header and determines the best response format. If no
+    acceptable format is found, returns 406 Not Acceptable.
+
+    @param formats List of formats this endpoint can produce
+    @return Plug that sets negotiated format on request *)
+let create ~(formats : format list) () : Core.t =
+ fun handler req ->
+  let accept =
+    List.find_opt
+      (fun (n, _) -> String.lowercase_ascii n = "accept")
+      req.headers
+    |> Option.map snd
+    |> Option.value ~default:"*/*"
+  in
+  match negotiate ~accept ~available:formats with
+  | None ->
+      (* 406 Not Acceptable *)
+      let acceptable =
+        String.concat ", " (List.map mime_type_of_format formats)
+      in
+      {
+        status = `Not_acceptable;
+        headers =
+          [
+            ("Content-Type", "text/plain; charset=utf-8"); ("Accept", acceptable);
+          ];
+        body = Server.Body_string "Not Acceptable";
+      }
+  | Some fmt ->
+      let req = set_format req fmt in
+      handler req

lib/plug/session.ml

···
+(** Cookie-based session management with fiber-local isolation.
+
+    {[
+      let store = Session.Memory_store.create () in
+      let session_plug = Session.create ~store () in
+
+      let handler req =
+        match Session.get "user_id" with
+        | Some uid -> Session.put "last_seen" (string_of_float now); ...
+        | None -> ...
+    ]} *)
+
+type session = {
+  id : string;
+  data : (string * string) list;
+  created_at : float;
+}
+
+type store =
+  | Memory of (string, session) Kcas_data.Hashtbl.t
+  | Cookie of { secret : string; max_age : float }
+
+type session_ctx = {
+  mutable data : (string * string) list;
+  mutable modified : bool;
+}
+
+let session_key : session_ctx Eio.Fiber.key = Eio.Fiber.create_key ()
+
+let get_ctx () =
+  match Eio.Fiber.get session_key with
+  | Some ctx -> ctx
+  | None -> failwith "Session: not in session scope (missing Session plug)"
+
+module Memory_store = struct
+  let create () = Memory (Kcas_data.Hashtbl.create ())
+
+  let cleanup ?(max_age = 86400.0) store =
+    match store with
+    | Memory tbl ->
+        let now = Unix.gettimeofday () in
+        Kcas_data.Hashtbl.filter_map_inplace
+          (fun _ sess ->
+            if now -. sess.created_at > max_age then None else Some sess)
+          tbl
+    | Cookie _ -> ()
+end
+
+module Cookie_store = struct
+  let create ~secret ?(max_age = 86400.0) () = Cookie { secret; max_age }
+
+  let serialize data =
+    List.map (fun (k, v) -> Token.b64_encode k ^ "=" ^ Token.b64_encode v) data
+    |> String.concat "&"
+
+  let deserialize s =
+    if s = "" then []
+    else
+      String.split_on_char '&' s
+      |> List.filter_map (fun pair ->
+          match String.index_opt pair '=' with
+          | Some i -> (
+              let k_b64 = String.sub pair 0 i in
+              let v_b64 =
+                String.sub pair (i + 1) (String.length pair - i - 1)
+              in
+              match (Token.b64_decode k_b64, Token.b64_decode v_b64) with
+              | Ok k, Ok v -> Some (k, v)
+              | _ -> None)
+          | None -> None)
+
+  let encrypt ~secret ~max_age data =
+    Token.encrypt ~secret ~data:(serialize data) ~max_age
+
+  let decrypt ~secret ~max_age token =
+    match Token.decrypt ~secret ~token ~max_age with
+    | Ok data -> Some (deserialize data)
+    | Error _ -> None
+end
+
+let get key =
+  let ctx = get_ctx () in
+  List.assoc_opt key ctx.data
+
+let get_with_req _req key = get key
+
+let put key value =
+  let ctx = get_ctx () in
+  ctx.modified <- true;
+  ctx.data <- (key, value) :: List.filter (fun (k, _) -> k <> key) ctx.data
+
+let put_with_req _req key value = put key value
+
+let delete key =
+  let ctx = get_ctx () in
+  ctx.modified <- true;
+  ctx.data <- List.filter (fun (k, _) -> k <> key) ctx.data
+
+let delete_with_req _req key = delete key
+
+let clear () =
+  let ctx = get_ctx () in
+  ctx.modified <- true;
+  ctx.data <- []
+
+let clear_with_req _req = clear ()
+
+let get_all () =
+  let ctx = get_ctx () in
+  ctx.data
+
+let get_all_with_req _req = get_all ()
+
+type config = {
+  cookie_name : string;
+  store : store;
+  secure : bool;
+  same_site : [ `Strict | `Lax | `None ];
+  http_only : bool;
+  path : string;
+  max_age : float;
+}
+
+let parse_cookie ~cookie_name headers =
+  match
+    List.find_opt (fun (n, _) -> String.lowercase_ascii n = "cookie") headers
+  with
+  | None -> None
+  | Some (_, cookies) ->
+      String.split_on_char ';' cookies
+      |> List.find_map (fun pair ->
+          let pair = String.trim pair in
+          match String.index_opt pair '=' with
+          | None -> None
+          | Some i ->
+              let name = String.sub pair 0 i in
+              if name = cookie_name then
+                Some (String.sub pair (i + 1) (String.length pair - i - 1))
+              else None)
+
+let make_cookie ~config value =
+  let same_site =
+    match config.same_site with
+    | `Strict -> "; SameSite=Strict"
+    | `Lax -> "; SameSite=Lax"
+    | `None -> "; SameSite=None"
+  in
+  Printf.sprintf "%s=%s; Path=%s; Max-Age=%.0f%s%s%s" config.cookie_name value
+    config.path config.max_age
+    (if config.secure then "; Secure" else "")
+    (if config.http_only then "; HttpOnly" else "")
+    same_site
+
+let generate_id () = Token.b64_encode (Mirage_crypto_rng.generate 16)
+
+(** Create session plug with configurable storage and cookie options. *)
+let create ~store ?(cookie_name = "_session") ?(secure = true)
+    ?(same_site = `Lax) ?(http_only = true) ?(path = "/") ?(max_age = 86400.0)
+    () : Core.t =
+  let config =
+    { cookie_name; store; secure; same_site; http_only; path; max_age }
+  in
+  fun handler req ->
+    let existing_id =
+      parse_cookie ~cookie_name:config.cookie_name req.Server.headers
+    in
+    let initial_data =
+      match config.store with
+      | Memory tbl -> (
+          match existing_id with
+          | Some id -> (
+              match Kcas_data.Hashtbl.find_opt tbl id with
+              | Some sess -> sess.data
+              | None -> [])
+          | None -> [])
+      | Cookie { secret; max_age } -> (
+          match existing_id with
+          | Some token -> (
+              match Cookie_store.decrypt ~secret ~max_age token with
+              | Some data -> data
+              | None -> [])
+          | None -> [])
+    in
+    let ctx = { data = initial_data; modified = false } in
+    let resp = Eio.Fiber.with_binding session_key ctx (fun () -> handler req) in
+    if not ctx.modified then resp
+    else
+      match config.store with
+      | Memory tbl ->
+          let session_id =
+            match existing_id with
+            | Some id when Kcas_data.Hashtbl.mem tbl id -> id
+            | _ -> generate_id ()
+          in
+          if ctx.data <> [] then begin
+            Kcas_data.Hashtbl.replace tbl session_id
+              {
+                id = session_id;
+                data = ctx.data;
+                created_at = Unix.gettimeofday ();
+              };
+            let cookie = make_cookie ~config session_id in
+            {
+              resp with
+              Server.headers = ("Set-Cookie", cookie) :: resp.headers;
+            }
+          end
+          else begin
+            (match existing_id with
+            | Some id -> Kcas_data.Hashtbl.remove tbl id
+            | None -> ());
+            resp
+          end
+      | Cookie { secret; max_age } ->
+          if ctx.data <> [] then begin
+            let token = Cookie_store.encrypt ~secret ~max_age ctx.data in
+            let cookie = make_cookie ~config token in
+            {
+              resp with
+              Server.headers = ("Set-Cookie", cookie) :: resp.headers;
+            }
+          end
+          else resp

lib/plug/token.ml

···
+(** Signed and encrypted tokens for authentication.
+
+    Provides secure token generation and verification for auth tokens, password
+    reset, email verification, etc.
+
+    Token format:
+    - signed: base64(data) . timestamp . base64(hmac)
+    - encrypted: base64(nonce + ciphertext + tag) . timestamp . base64(hmac) *)
+
+(** Error types for token operations *)
+type error =
+  | Expired  (** Token has exceeded max_age *)
+  | Invalid  (** Token format invalid or signature mismatch *)
+  | Tampered  (** Token data has been modified *)
+
+let error_to_string = function
+  | Expired -> "token expired"
+  | Invalid -> "invalid token"
+  | Tampered -> "token tampered"
+
+(** {1 Internal helpers} *)
+
+(** Get current Unix timestamp *)
+let now () = Unix.gettimeofday ()
+
+(** Encode timestamp as compact varint-like format (8 bytes max) *)
+let encode_timestamp ts =
+  let ts_int = Int64.of_float ts in
+  let buf = Buffer.create 8 in
+  let rec encode v =
+    if Int64.compare v 0L = 0 then ()
+    else begin
+      Buffer.add_char buf (Char.chr (Int64.to_int (Int64.logand v 0xFFL)));
+      encode (Int64.shift_right_logical v 8)
+    end
+  in
+  if Int64.compare ts_int 0L = 0 then Buffer.add_char buf '\000'
+  else encode ts_int;
+  Buffer.contents buf
+
+(** Decode timestamp from bytes *)
+let decode_timestamp s =
+  let len = String.length s in
+  let rec decode acc shift i =
+    if i >= len then acc
+    else
+      let byte = Int64.of_int (Char.code s.[i]) in
+      decode (Int64.logor acc (Int64.shift_left byte shift)) (shift + 8) (i + 1)
+  in
+  Int64.to_float (decode 0L 0 0)
+
+(** Constant-time string comparison to prevent timing attacks *)
+let secure_compare a b =
+  let len_a = String.length a in
+  let len_b = String.length b in
+  if len_a <> len_b then false
+  else
+    let result = ref 0 in
+    for i = 0 to len_a - 1 do
+      result := !result lor (Char.code a.[i] lxor Char.code b.[i])
+    done;
+    !result = 0
+
+(** Compute HMAC-SHA256 *)
+let hmac ~key data =
+  Digestif.SHA256.hmac_string ~key data |> Digestif.SHA256.to_raw_string
+
+(** Base64 URL-safe encoding (no padding) *)
+let b64_encode s =
+  Base64.encode_string ~pad:false ~alphabet:Base64.uri_safe_alphabet s
+
+(** Base64 URL-safe decoding *)
+let b64_decode s =
+  (* Add padding if needed *)
+  let len = String.length s in
+  let padded = match len mod 4 with 2 -> s ^ "==" | 3 -> s ^ "=" | _ -> s in
+  Base64.decode ~alphabet:Base64.uri_safe_alphabet padded
+
+(** {1 Signed tokens} *)
+
+(** Sign data with HMAC-SHA256.
+
+    Format: base64(data).base64(timestamp).base64(hmac)
+
+    The HMAC covers both data and timestamp to prevent replay attacks with
+    modified timestamps.
+
+    @param secret Shared secret key (should be at least 32 bytes)
+    @param data Payload to sign (arbitrary string)
+    @param max_age Maximum token age in seconds
+    @return URL-safe signed token string *)
+let sign ~secret ~data ~max_age:_ =
+  let ts = now () in
+  let ts_bytes = encode_timestamp ts in
+  let data_b64 = b64_encode data in
+  let ts_b64 = b64_encode ts_bytes in
+  (* HMAC covers data and timestamp *)
+  let to_sign = data_b64 ^ "." ^ ts_b64 in
+  let signature = hmac ~key:secret to_sign in
+  let sig_b64 = b64_encode signature in
+  data_b64 ^ "." ^ ts_b64 ^ "." ^ sig_b64
+
+(** Verify a signed token.
+
+    @param secret Shared secret key (must match signing key)
+    @param token The signed token to verify
+    @param max_age Maximum allowed age in seconds
+    @return Ok(data) if valid, Error(reason) otherwise *)
+let verify ~secret ~token ~max_age =
+  match String.split_on_char '.' token with
+  | [ data_b64; ts_b64; sig_b64 ] -> (
+      (* Decode signature and verify first (fail fast on tampering) *)
+      match (b64_decode sig_b64, b64_decode data_b64, b64_decode ts_b64) with
+      | Ok signature, Ok data, Ok ts_bytes ->
+          let to_sign = data_b64 ^ "." ^ ts_b64 in
+          let expected = hmac ~key:secret to_sign in
+          if not (secure_compare expected signature) then Error Tampered
+          else
+            let ts = decode_timestamp ts_bytes in
+            let age = now () -. ts in
+            if age > max_age then Error Expired
+            else if age < -60.0 then
+              (* Allow 60s clock skew, but reject future tokens *)
+              Error Invalid
+            else Ok data
+      | _ -> Error Invalid)
+  | _ -> Error Invalid
+
+(** {1 Encrypted tokens} *)
+
+(** Encrypt and sign data using AES-256-GCM.
+
+    Format: base64(nonce + ciphertext + tag).base64(timestamp).base64(hmac)
+
+    Encryption uses AES-256-GCM for authenticated encryption. The HMAC provides
+    an additional layer of authentication over the encrypted blob and timestamp.
+
+    @param secret Shared secret key (first 32 bytes for AES, full key for HMAC)
+    @param data Payload to encrypt
+    @param max_age Maximum token age in seconds
+    @return URL-safe encrypted token string *)
+let encrypt ~secret ~data ~max_age:_ =
+  let ts = now () in
+  let ts_bytes = encode_timestamp ts in
+  let aes_key =
+    if String.length secret >= 32 then String.sub secret 0 32
+    else Digestif.SHA256.(digest_string secret |> to_raw_string)
+  in
+  let nonce = Mirage_crypto_rng.generate 12 in
+  let key = Mirage_crypto.AES.GCM.of_secret aes_key in
+  let ciphertext =
+    Mirage_crypto.AES.GCM.authenticate_encrypt ~key ~nonce data
+  in
+  let encrypted = nonce ^ ciphertext in
+  let encrypted_b64 = b64_encode encrypted in
+  let ts_b64 = b64_encode ts_bytes in
+  let to_sign = encrypted_b64 ^ "." ^ ts_b64 in
+  let signature = hmac ~key:secret to_sign in
+  let sig_b64 = b64_encode signature in
+  encrypted_b64 ^ "." ^ ts_b64 ^ "." ^ sig_b64
+
+(** Decrypt and verify an encrypted token.
+
+    @param secret Shared secret key (must match encryption key)
+    @param token The encrypted token to decrypt
+    @param max_age Maximum allowed age in seconds
+    @return Ok(data) if valid, Error(reason) otherwise *)
+let decrypt ~secret ~token ~max_age =
+  match String.split_on_char '.' token with
+  | [ encrypted_b64; ts_b64; sig_b64 ] -> (
+      match
+        (b64_decode sig_b64, b64_decode encrypted_b64, b64_decode ts_b64)
+      with
+      | Ok signature, Ok encrypted, Ok ts_bytes -> (
+          let to_sign = encrypted_b64 ^ "." ^ ts_b64 in
+          let expected = hmac ~key:secret to_sign in
+          if not (secure_compare expected signature) then Error Tampered
+          else
+            let ts = decode_timestamp ts_bytes in
+            let age = now () -. ts in
+            if age > max_age then Error Expired
+            else if age < -60.0 then Error Invalid
+            else if String.length encrypted < 28 then Error Invalid
+            else
+              let nonce = String.sub encrypted 0 12 in
+              let ciphertext =
+                String.sub encrypted 12 (String.length encrypted - 12)
+              in
+              let aes_key =
+                if String.length secret >= 32 then String.sub secret 0 32
+                else Digestif.SHA256.(digest_string secret |> to_raw_string)
+              in
+              let key = Mirage_crypto.AES.GCM.of_secret aes_key in
+              match
+                Mirage_crypto.AES.GCM.authenticate_decrypt ~key ~nonce
+                  ciphertext
+              with
+              | Some plaintext -> Ok plaintext
+              | None -> Error Tampered)
+      | _ -> Error Invalid)
+  | _ -> Error Invalid
+
+(** {1 Convenience functions} *)
+
+(** Create a signed auth token with user data. Default max_age is 24 hours
+    (86400 seconds). *)
+let auth_token ~secret ~user_id = sign ~secret ~data:user_id ~max_age:86400.0
+
+(** Create an encrypted password reset token. Default max_age is 1 hour (3600
+    seconds). Encrypted so user cannot see/modify the payload. *)
+let reset_token ~secret ~user_id = encrypt ~secret ~data:user_id ~max_age:3600.0
+
+(** Verify an auth token, returning the user_id if valid. *)
+let verify_auth ~secret ~token ~max_age = verify ~secret ~token ~max_age
+
+(** Verify a reset token, returning the user_id if valid. *)
+let verify_reset ~secret ~token ~max_age = decrypt ~secret ~token ~max_age

lib/pubsub.ml

···
+type topic = string
+type message = string
+type subscription = { id : int; topic : topic; callback : message -> unit }
+
+type t = {
+  topics : (topic, subscription list) Kcas_data.Hashtbl.t;
+  next_id : int Atomic.t;
+}
+
+let create () =
+  { topics = Kcas_data.Hashtbl.create (); next_id = Atomic.make 0 }
+
+let subscribe t topic callback =
+  let id = Atomic.fetch_and_add t.next_id 1 in
+  let sub = { id; topic; callback } in
+  let subs =
+    match Kcas_data.Hashtbl.find_opt t.topics topic with
+    | Some existing -> sub :: existing
+    | None -> [ sub ]
+  in
+  Kcas_data.Hashtbl.replace t.topics topic subs;
+  sub
+
+let unsubscribe t sub =
+  match Kcas_data.Hashtbl.find_opt t.topics sub.topic with
+  | None -> ()
+  | Some subs ->
+      let remaining = List.filter (fun s -> s.id <> sub.id) subs in
+      if remaining = [] then Kcas_data.Hashtbl.remove t.topics sub.topic
+      else Kcas_data.Hashtbl.replace t.topics sub.topic remaining
+
+let broadcast t topic msg =
+  match Kcas_data.Hashtbl.find_opt t.topics topic with
+  | None -> ()
+  | Some subs -> List.iter (fun sub -> sub.callback msg) subs
+
+let subscriber_count t topic =
+  match Kcas_data.Hashtbl.find_opt t.topics topic with
+  | None -> 0
+  | Some subs -> List.length subs
+
+let topics t = Kcas_data.Hashtbl.to_seq_keys t.topics |> List.of_seq