fix(bugs): crypto trace ids, error mapping, caps, lifecycle, transports

- crypto RNG for trace/span ids (#41)
- handler resolved before job.accepted; shared unwind (#47)
- lease/runtime watchdog callbacks observe exceptions (#48)
- 3-state RevocationOutcome; permanent failures keep credential outstanding (#49)
- auto-ack/receive-loop/dispose tasks observed; Completion exposed (#60,#61,#62)
- JobErrorMapper parses details fields (#63)
- AutoAck docs corrected to event-driven (#64)
- ChunkAssembler + WebSocket message size/chunk caps (#65,#66)
- Giraffe short-circuits ws/400 branches (#40)
- CLI awaits enumerator dispose and honors --stdio (#38,#39)

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: Nick Ficano

samples/LiteLLM/Program.fs

@@ -89,7 +89,7 @@ type LiteLLMProvisioner(baseUrl: Uri, adminKey: string, http: HttpClient) =
             task {
                 let body = {| key = credentialId |} :> obj
                 let! _ = postJsonAsync "/key/delete" body ct
-                return true
+                return RevocationOutcome.Revoked
             }
 
 [<EntryPoint>]

samples/ProvisionedCredentials/Program.fs

@@ -31,7 +31,7 @@ type StaticProvisioner(revoked: HashSet<string>) =
 
         member _.RevokeAsync(credentialId, _ct) =
             lock revoked (fun () -> revoked.Add credentialId |> ignore)
-            Task.FromResult true
+            Task.FromResult RevocationOutcome.Revoked
 
 [<EntryPoint>]
 let main _argv =

src/Arcp.Cli/Program.fs

@@ -88,7 +88,11 @@ let private streamEventsAsync (handle: JobHandle) : Task =
                 else
                     writeLine (sprintf "event: %s" (JobEventBody.kind enumerator.Current))
         finally
-            ignore (enumerator.DisposeAsync().AsTask())
+            ()
+
+        // §38: await disposal so teardown errors surface and complete
+        // before this function returns.
+        do! enumerator.DisposeAsync()
     }
     :> Task
 
@@ -162,7 +166,12 @@ let main argv =
                         let env = Environment.GetEnvironmentVariable "ARCP_TOKEN"
                         if String.IsNullOrEmpty env then None else Some env)
 
-                (serveStdio token).GetAwaiter().GetResult()
+                // §39: honor the --stdio flag instead of ignoring it.
+                if sub.Contains ServeArgs.Stdio then
+                    (serveStdio token).GetAwaiter().GetResult()
+                else
+                    errorLine "serve requires a transport flag; only --stdio is currently supported (pass --stdio)"
+                    2
             | Send sub :: _ ->
                 let url = sub.GetResult SendArgs.Url
 

src/Arcp.Client/ArcpClient.fs

@@ -18,8 +18,20 @@ type ArcpClient(transport: ITransport, options: ArcpClientOptions) =
     let handles = ConcurrentDictionary<string, JobHandleWriter>()
     let mutable sessionCtx: SessionContext option = None
     let mutable autoAck: AutoAckScheduler option = None
+    let mutable receiveLoopTask: Task = Task.CompletedTask
     let receiveLoopCts = new CancellationTokenSource()
 
+    /// Attach a faulted-state observer so fire-and-forget sends do not
+    /// become unobserved task exceptions (#60).
+    let observeTask (t: Task) : unit =
+        t.ContinueWith(
+            (fun (tt: Task) ->
+                if tt.IsFaulted then
+                    eprintfn "[ARCP] background send failed: %O" tt.Exception),
+            TaskContinuationOptions.OnlyOnFaulted
+        )
+        |> ignore
+
     let connectedTcs =
         TaskCompletionSource<SessionContext>(TaskCreationOptions.RunContinuationsAsynchronously)
 
@@ -120,7 +132,7 @@ type ArcpClient(transport: ITransport, options: ArcpClientOptions) =
             match sched.OnEvent seq with
             | Some toAck ->
                 let ack: SessionAckPayload = { LastProcessedSeq = toAck }
-                ignore (sendMessage (Message.SessionAck ack))
+                observeTask (sendMessage (Message.SessionAck ack))
             | None -> ()
         | _ -> ()
 
@@ -172,7 +184,9 @@ type ArcpClient(transport: ITransport, options: ArcpClientOptions) =
                     handles.Clear()
                     orphans.Clear())
 
-                ignore (enumerator.DisposeAsync().AsTask())
+            // §62: await enumerator disposal so transport teardown errors
+            // surface rather than being swallowed on a background thread.
+            do! enumerator.DisposeAsync()
         }
         :> Task
 
@@ -215,7 +229,10 @@ type ArcpClient(transport: ITransport, options: ArcpClientOptions) =
     /// receive loop, then resolves with the negotiated session context.
     member this.ConnectAsync(ct: CancellationToken) : Task<SessionContext> =
         task {
-            ignore (runReceiveLoop ())
+            // §61: retain the receive-loop task so its completion (and any
+            // fault) is observable via `Completion`.
+            receiveLoopTask <- runReceiveLoop ()
+            observeTask receiveLoopTask
             let env = Codec.toEnvelope (Message.SessionHello(buildHello ()))
             let waiter = pending.Register env.Id
             do! transport.SendAsync(env, ct)
@@ -377,6 +394,10 @@ type ArcpClient(transport: ITransport, options: ArcpClientOptions) =
 
     member _.Session = sessionCtx
 
+    /// Completes when the receive loop terminates (clean EOF, cancellation,
+    /// or fault). Lets callers observe that the client stopped pumping (#61).
+    member _.Completion: Task = receiveLoopTask
+
     /// Close the session cleanly with an optional reason.
     member this.CloseAsync(reason: string option, ct: CancellationToken) : Task =
         task {

src/Arcp.Client/Internal/AutoAck.fs

@@ -10,8 +10,11 @@ type AutoAckOptions =
     {
         /// Maximum number of events to receive before forcing an `ack`.
         EveryEvents: int
-        /// Maximum time between acks. The scheduler flushes if either
-        /// threshold is reached.
+        /// Minimum elapsed time before the next event triggers an ack.
+        /// NOTE: the scheduler is event-driven — it is evaluated only on
+        /// `OnEvent`, so this interval gates acks on the next event
+        /// arrival rather than firing on its own timer. `session.ack` is
+        /// advisory (spec §6.5), so no standalone timer is used.
         Interval: TimeSpan
     }
 
@@ -24,12 +27,14 @@ module AutoAckOptions =
         }
 
 /// Tracks `last_processed_seq` and decides when to emit a
-/// `session.ack` based on event count and elapsed time.
+/// `session.ack`. Evaluation is event-driven: `OnEvent` returns the
+/// seq to ack when the event-count threshold is hit, or when the
+/// configured interval has elapsed *as observed on the current event*.
+/// There is no background timer — if events stop arriving, no ack is
+/// produced (acks are advisory per spec §6.5).
 ///
 /// The scheduler does NOT send the ack itself — it returns the seq
-/// to send so the client can build/send the envelope. Spec §6.5
-/// notes ack is purely advisory; this implementation matches the
-/// TS SDK's behaviour (ack every 32 events / 250 ms by default).
+/// to send so the client can build/send the envelope.
 type internal AutoAckScheduler(options: AutoAckOptions, timeProvider: TimeProvider) =
     let lockObj = obj ()
     let mutable lastSeq: int64 = 0L

src/Arcp.Client/Internal/ChunkAssembler.fs

@@ -11,20 +11,35 @@ open ARCP.Core
 /// One assembler instance per `result_id`. Chunks MUST arrive in
 /// `chunk_seq` order per spec §8.4; out-of-order arrivals raise
 /// `InvalidRequest` and the caller is expected to terminate the job.
-type internal ChunkAssembler() =
+/// Default caps guarding against unbounded result streams (DoS).
+[<RequireQualifiedAccess>]
+module internal ChunkLimits =
+    [<Literal>]
+    let DefaultMaxBytes: int64 = 256L * 1024L * 1024L // 256 MiB
+
+    [<Literal>]
+    let DefaultMaxChunks: int = 1_000_000
+
+type internal ChunkAssembler(maxBytes: int64, maxChunks: int) =
     let buffer = ResizeArray<byte[]>()
     let mutable expectedSeq: int64 = 0L
+    let mutable totalBytes: int64 = 0L
     let mutable closed = false
 
+    new() = ChunkAssembler(ChunkLimits.DefaultMaxBytes, ChunkLimits.DefaultMaxChunks)
+
     /// Append a chunk. Returns `Ok finished` where `finished` is
-    /// `true` once a `more = false` chunk has arrived.
+    /// `true` once a `more = false` chunk has arrived. A stream that
+    /// exceeds the byte/chunk cap is rejected to bound memory (§8.4).
     member _.Append(chunkSeq: int64, data: string, encoding: ChunkEncoding, more: bool) : Result<bool, ARCPError> =
         if closed then
             Error(ARCPError.InvalidRequest("Chunk arrived after stream closed", None))
         elif chunkSeq <> expectedSeq then
             Error(
                 ARCPError.InvalidRequest(sprintf "Out-of-order chunk: expected %d, got %d" expectedSeq chunkSeq, None)
             )
+        elif int64 buffer.Count >= int64 maxChunks then
+            Error(ARCPError.InvalidRequest(sprintf "Result stream exceeded max chunk count (%d)" maxChunks, None))
         else
             let bytesResult =
                 try
@@ -42,8 +57,11 @@ type internal ChunkAssembler() =
 
             match bytesResult with
             | Error e -> Error e
+            | Ok bytes when totalBytes + int64 bytes.Length > maxBytes ->
+                Error(ARCPError.InvalidRequest(sprintf "Result stream exceeded max byte budget (%d)" maxBytes, None))
             | Ok bytes ->
                 buffer.Add bytes
+                totalBytes <- totalBytes + int64 bytes.Length
                 expectedSeq <- expectedSeq + 1L
 
                 if not more then

src/Arcp.Client/Internal/JobErrorMapper.fs

@@ -1,13 +1,50 @@
 namespace ARCP.Client.Internal
 
 open System
+open System.Text.Json
 open ARCP.Core
 
 /// Map a wire `job.error` / `session.error` code string back to an
-/// `ARCPError` DU. The reverse direction is `ARCPError.code`. Out
-/// of scope: details payloads beyond `message`.
+/// `ARCPError` DU, parsing structured fields out of the `details`
+/// payload where the DU carries them. The reverse direction is
+/// `ARCPError.code`.
 [<RequireQualifiedAccess>]
 module internal JobErrorMapper =
+    let private prop (details: JsonElement option) (name: string) : JsonElement option =
+        match details with
+        | Some d when d.ValueKind = JsonValueKind.Object ->
+            match d.TryGetProperty name with
+            | true, v when v.ValueKind <> JsonValueKind.Null -> Some v
+            | _ -> None
+        | _ -> None
+
+    let private strField details name fallback =
+        prop details name |> Option.map (fun v -> v.GetString()) |> Option.defaultValue fallback
+
+    let private intField details name fallback =
+        prop details name
+        |> Option.bind (fun v ->
+            match v.TryGetInt32() with
+            | true, n -> Some n
+            | _ -> None)
+        |> Option.defaultValue fallback
+
+    let private int64Field details name fallback =
+        prop details name
+        |> Option.bind (fun v ->
+            match v.TryGetInt64() with
+            | true, n -> Some n
+            | _ -> None)
+        |> Option.defaultValue fallback
+
+    let private dateField details name fallback =
+        prop details name
+        |> Option.bind (fun v ->
+            match v.TryGetDateTimeOffset() with
+            | true, d -> Some d
+            | _ -> None)
+        |> Option.defaultValue fallback
+
     let ofWire
         (code: string)
         (message: string)
@@ -20,13 +57,14 @@ module internal JobErrorMapper =
         | "JOB_NOT_FOUND" -> ARCPError.JobNotFound jobId
         | "DUPLICATE_KEY" -> ARCPError.DuplicateKey message
         | "AGENT_NOT_AVAILABLE" -> ARCPError.AgentNotAvailable message
-        | "AGENT_VERSION_NOT_AVAILABLE" -> ARCPError.AgentVersionNotAvailable(message, "")
+        | "AGENT_VERSION_NOT_AVAILABLE" -> ARCPError.AgentVersionNotAvailable(message, strField details "version" "")
         | "CANCELLED" -> ARCPError.Cancelled(Some message)
-        | "TIMEOUT" -> ARCPError.Timeout 0
+        | "TIMEOUT" -> ARCPError.Timeout(intField details "timeout_sec" 0)
         | "HEARTBEAT_LOST" -> ARCPError.HeartbeatLost
-        | "LEASE_EXPIRED" -> ARCPError.LeaseExpired DateTimeOffset.MinValue
-        | "BUDGET_EXHAUSTED" -> ARCPError.BudgetExhausted "USD"
+        | "LEASE_EXPIRED" -> ARCPError.LeaseExpired(dateField details "expires_at" DateTimeOffset.MinValue)
+        | "BUDGET_EXHAUSTED" -> ARCPError.BudgetExhausted(strField details "currency" "USD")
         | "INVALID_REQUEST" -> ARCPError.InvalidRequest(message, details)
         | "UNAUTHENTICATED" -> ARCPError.Unauthenticated message
-        | "RESUME_WINDOW_EXPIRED" -> ARCPError.ResumeWindowExpired(0L, 0)
+        | "RESUME_WINDOW_EXPIRED" ->
+            ARCPError.ResumeWindowExpired(int64Field details "from_seq" 0L, intField details "window_sec" 0)
         | _ -> ARCPError.InternalError message

src/Arcp.Client/Transport/WebSocket.fs

@@ -17,7 +17,7 @@ open ARCP.Client
 ///
 /// One text frame per envelope. The receive loop reassembles
 /// continuation frames into a single message.
-type WebSocketClientTransport(socket: WebSocket, ownsSocket: bool) =
+type WebSocketClientTransport(socket: WebSocket, ownsSocket: bool, maxMessageBytes: int64) =
     // The BCL WebSocket allows only one outstanding send at a time,
     // so we serialise concurrent callers (auto-ack, pong, submit,
     // cancel, close) through an async-aware semaphore — a plain
@@ -59,6 +59,16 @@ type WebSocketClientTransport(socket: WebSocket, ownsSocket: bool) =
                             if result.MessageType = WebSocketMessageType.Close then
                                 closedRemotely <- true
                             else
+                                if ms.Length + int64 result.Count > maxMessageBytes then
+                                    // §66: reject oversized messages before the
+                                    // codec ever sees them.
+                                    raise (
+                                        WebSocketException(
+                                            WebSocketError.HeaderError,
+                                            sprintf "Inbound message exceeded %d bytes" maxMessageBytes
+                                        )
+                                    )
+
                                 ms.Write(buffer, 0, result.Count)
                                 endOfMessage <- result.EndOfMessage
 
@@ -80,6 +90,11 @@ type WebSocketClientTransport(socket: WebSocket, ownsSocket: bool) =
                 | Error _ -> return! receiveOne ct
         }
 
+    /// Defaults to a 256 MiB cap on a single reassembled message to
+    /// bound memory against a peer streaming continuation frames
+    /// indefinitely.
+    new(socket: WebSocket, ownsSocket: bool) = WebSocketClientTransport(socket, ownsSocket, 256L * 1024L * 1024L)
+
     interface ITransport with
         member _.SendAsync(env, ct) = sendOne env ct
 

src/Arcp.Core/Trace.fs

@@ -1,6 +1,7 @@
 namespace ARCP.Core
 
 open System
+open System.Security.Cryptography
 
 /// W3C Trace Context identifiers. Trace ids are 32 hex chars,
 /// span ids 16 hex chars; the strings are kept opaque on the wire.
@@ -19,22 +20,21 @@ type SpanId =
     member this.Value = let (SpanId v) = this in v
     override this.ToString() = this.Value
 
-[<RequireQualifiedAccess>]
-module TraceId =
-    let private hex (count: int) =
+/// Cryptographically random hex string. Trace/span ids cross trust
+/// boundaries, so guess-resistance matters (W3C Trace Context).
+[<AutoOpen>]
+module private TraceRandom =
+    let cryptoHex (count: int) =
         let buf = Array.zeroCreate<byte> count
-        Random.Shared.NextBytes(buf)
+        RandomNumberGenerator.Fill(Span<byte>(buf))
         Convert.ToHexString(buf).ToLowerInvariant()
 
-    let newId () : TraceId = TraceId(hex 16)
+[<RequireQualifiedAccess>]
+module TraceId =
+    let newId () : TraceId = TraceId(cryptoHex 16)
     let ofString (s: string) : TraceId = TraceId s
 
 [<RequireQualifiedAccess>]
 module SpanId =
-    let private hex (count: int) =
-        let buf = Array.zeroCreate<byte> count
-        Random.Shared.NextBytes(buf)
-        Convert.ToHexString(buf).ToLowerInvariant()
-
-    let newId () : SpanId = SpanId(hex 8)
+    let newId () : SpanId = SpanId(cryptoHex 8)
     let ofString (s: string) : SpanId = SpanId s

src/Arcp.Giraffe/GiraffeEndpoint.fs

@@ -19,16 +19,21 @@ module ArcpGiraffe =
         fun (next: HttpFunc) (ctx: HttpContext) ->
             task {
                 if ctx.Request.Path.Value <> path then
+                    // Only fall through to the next handler on a path miss.
                     return! next ctx
                 elif not ctx.WebSockets.IsWebSocketRequest then
+                    // §40: short-circuit so downstream handlers cannot
+                    // overwrite the 400 status.
                     ctx.Response.StatusCode <- StatusCodes.Status400BadRequest
-                    return! next ctx
+                    return Some ctx
                 else
                     let! socket = ctx.WebSockets.AcceptWebSocketAsync()
 
                     let transport =
                         new WebSocketClientTransport(socket, ownsSocket = true) :> ITransport
 
                     do! server.HandleSessionAsync(transport, ctx.RequestAborted)
-                    return! next ctx
+                    // §40: the response is hijacked by the upgrade; do not
+                    // continue the pipeline.
+                    return Some ctx
             }