refactor: lean on F# idioms in transport, server, and event-log internals

- WebSocket.receiveOne: drop the malformed-envelope fallthrough and
  recurse instead of fabricating a sentinel from a re-decoded payload.
  Frame reassembly stays imperative (BCL ReceiveAsync gives no
  functional aggregator) but is now isolated behind a Choice.
- Glob.compile: replace mutable StringBuilder + index walk with a tail
  recursive translator over the char list.
- EventLog.EvictExpired: switch the per-session prune to a recursive
  drop inside the lock and aggregate counts via Seq.sumBy.
- ArcpServer.registerHandler: replace the silent default-of JsonElement
  placeholder with one that raises if the inventory adapter is ever
  routed through, plus a comment naming the cycle the ref cell breaks.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: Nick Ficano

src/Arcp.Client/Transport/WebSocket.fs

@@ -32,30 +32,38 @@ type WebSocketClientTransport(socket: WebSocket, ownsSocket: bool) =
                     socket.SendAsync(ArraySegment<byte>(bytes), WebSocketMessageType.Text, true, ct))
         } :> Task
 
-    let receiveOne (ct: CancellationToken) : Task<Envelope option> =
+    let rec receiveOne (ct: CancellationToken) : Task<Envelope option> =
         task {
             let buffer = ArrayPool<byte>.Shared.Rent(8192)
-            try
-                use ms = new MemoryStream()
-                let mutable endOfMessage = false
-                let mutable closedRemotely = false
-                while not endOfMessage && not closedRemotely do
-                    let! result = socket.ReceiveAsync(ArraySegment<byte>(buffer), ct)
-                    if result.MessageType = WebSocketMessageType.Close then
-                        closedRemotely <- true
-                    else
-                        ms.Write(buffer, 0, result.Count)
-                        endOfMessage <- result.EndOfMessage
-                if closedRemotely then return None
-                else
-                    let text = Encoding.UTF8.GetString(ms.ToArray())
-                    match Codec.readEnvelope text with
-                    | Ok env -> return Some env
-                    | Error _ -> return Some (Codec.readEnvelope text |> function Ok e -> e | _ -> Unchecked.defaultof<_>)
-                    // Fall back to ignoring malformed messages by returning a sentinel; we'll
-                    // re-loop on the caller side. Simpler: re-call receiveOne when malformed.
-            finally
-                ArrayPool<byte>.Shared.Return(buffer)
+            let! frame =
+                task {
+                    try
+                        use ms = new MemoryStream()
+                        // Mutation here drives WebSocket frame reassembly:
+                        // BCL `ReceiveAsync` returns one frame at a time and
+                        // there is no functional aggregator for it.
+                        let mutable endOfMessage = false
+                        let mutable closedRemotely = false
+                        while not endOfMessage && not closedRemotely do
+                            let! result = socket.ReceiveAsync(ArraySegment<byte>(buffer), ct)
+                            if result.MessageType = WebSocketMessageType.Close then
+                                closedRemotely <- true
+                            else
+                                ms.Write(buffer, 0, result.Count)
+                                endOfMessage <- result.EndOfMessage
+                        if closedRemotely then return Choice1Of2 ()
+                        else return Choice2Of2 (Encoding.UTF8.GetString(ms.ToArray()))
+                    finally
+                        ArrayPool<byte>.Shared.Return(buffer)
+                }
+            match frame with
+            | Choice1Of2 () -> return None
+            | Choice2Of2 text ->
+                match Codec.readEnvelope text with
+                | Ok env -> return Some env
+                // Malformed envelopes are dropped at the transport boundary
+                // per spec §4 — clients must not surface them to dispatch.
+                | Error _ -> return! receiveOne ct
         }
 
     interface ITransport with

src/Arcp.Core/Lease.fs

@@ -46,26 +46,15 @@ module Glob =
     /// `**` matches any path segment including `/`; `*` matches
     /// any character except `/`; `?` matches a single non-`/` char.
     let compile (pattern: string) : Regex =
-        let sb = System.Text.StringBuilder()
-        sb.Append "^" |> ignore
-        let mutable i = 0
-        let n = pattern.Length
-        while i < n do
-            let c = pattern.[i]
-            if c = '*' && i + 1 < n && pattern.[i + 1] = '*' then
-                sb.Append ".*" |> ignore
-                i <- i + 2
-            elif c = '*' then
-                sb.Append "[^/]*" |> ignore
-                i <- i + 1
-            elif c = '?' then
-                sb.Append "[^/]" |> ignore
-                i <- i + 1
-            else
-                sb.Append(Regex.Escape(string c)) |> ignore
-                i <- i + 1
-        sb.Append "$" |> ignore
-        Regex(sb.ToString(), RegexOptions.Compiled ||| RegexOptions.CultureInvariant)
+        let rec translate (chars: char list) (acc: string list) : string list =
+            match chars with
+            | [] -> List.rev acc
+            | '*' :: '*' :: rest -> translate rest (".*" :: acc)
+            | '*' :: rest -> translate rest ("[^/]*" :: acc)
+            | '?' :: rest -> translate rest ("[^/]" :: acc)
+            | c :: rest -> translate rest (Regex.Escape(string c) :: acc)
+        let body = pattern |> List.ofSeq |> (fun cs -> translate cs []) |> String.concat ""
+        Regex("^" + body + "$", RegexOptions.Compiled ||| RegexOptions.CultureInvariant)
 
     let isMatch (pattern: string) (target: string) : bool =
         // Amount strings used in `cost.budget` and any non-glob

src/Arcp.Runtime/ArcpServer.fs

@@ -49,9 +49,10 @@ type ArcpServer(options: ArcpServerOptions) =
     let sessions = ConcurrentDictionary<string, ServerSessionContext>()
     let agentHandlers = ConcurrentDictionary<string, ArcpAgentHandler>()
 
-    // The outbox is built per-session and stored in a ref cell so
-    // `JobManager` can route emits back through it without circular
-    // construction.
+    // `JobManager` and the real outbox are mutually dependent: the
+    // outbox needs `jobs` (for Subscribers / Terminate) and `jobs`
+    // needs the outbox. We break the cycle with a ref cell that
+    // `BuildOutbox` assigns into; mutation is unavoidable here.
     let outbox : IJobOutbox ref = ref Unchecked.defaultof<IJobOutbox>
     let jobs =
         JobManager(
@@ -63,8 +64,16 @@ type ArcpServer(options: ArcpServerOptions) =
 
     let registerHandler (name: string) (version: string) (h: ArcpAgentHandler) =
         agentHandlers.[name + "@" + version] <- h
-        let adapter : AgentHandler = fun _ -> task { return Unchecked.defaultof<JsonElement> }
-        inventory.Register(name, version, adapter)
+        // The inventory stores an `AgentHandler` purely as a presence
+        // marker — `JobSubmitFlow` discards it and dispatches via
+        // `agentHandlers` keyed by `name@version`. The placeholder
+        // raises so any regression that routes through it surfaces
+        // loudly instead of returning a garbage JsonElement.
+        let placeholder : AgentHandler =
+            fun _ ->
+                raise (InvalidOperationException
+                    (sprintf "ARCP runtime invariant: inventory placeholder invoked for %s@%s" name version))
+        inventory.Register(name, version, placeholder)
 
     /// Register an agent under the default version (`default`).
     member _.RegisterAgent(name: string, handler: ArcpAgentHandler) : unit =

src/Arcp.Runtime/Store/EventLog.fs

@@ -105,11 +105,16 @@ type internal EventLog(options: EventLogOptions) =
     member _.EvictExpired() : int =
         let now = options.TimeProvider.GetUtcNow()
         let cutoff = now.AddSeconds(-float options.ResumeWindowSec)
-        let mutable evicted = 0
-        for kvp in perSession do
-            lock kvp.Value (fun () ->
-                let mutable i = 0
-                while i < kvp.Value.Count && kvp.Value.[i].Timestamp < cutoff do
-                    kvp.Value.RemoveAt 0
-                    evicted <- evicted + 1)
-        evicted
+        // The per-session buffer is a mutable `List<T>` from the BCL
+        // (chosen for O(1) Add + RemoveAt 0 amortised semantics under
+        // a lock); eviction has to mutate it in place.
+        let evictOne (list: List<EventLogEntry>) : int =
+            lock list (fun () ->
+                let rec drop removed =
+                    if list.Count > 0 && list.[0].Timestamp < cutoff then
+                        list.RemoveAt 0
+                        drop (removed + 1)
+                    else removed
+                drop 0)
+        perSession
+        |> Seq.sumBy (fun kvp -> evictOne kvp.Value)