fix(minors): clean-code, perf, safety, feature gating

- watchdog skips terminal jobs (#89); wire retryable + Unknown arm (#90)
- client observes cancellation tokens (#98); reject re-hello (#106)
- session-scoped LastEventSeq via Interlocked; subscribed_from in subscriber space (#111)
- enforce negotiated features on submit (#114); duplicate-currency budget subset sum (#116)
- readEnvelope rejects null/empty envelopes (#117); ArcpResult avoids shadowing FSharp.Core Result (#118)
- StaticBearerVerifier constant-time compare (#119); DevMode rejects whitespace (#54)
- glob regex cache (#44); literalPrefix/isSubset cleanup (#43,#45); array retryDelays (#55)
- single idempotency lookup (#52); credential revoke failure observable (#53)
- dead-code removal (#51,#67); volatile stdio closed (#68); typed pending error (#69)
- Connected member (#70); optional jobId mapping (#71); lazy Otel source from Version.Sdk (#42)

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

Author: Nick Ficano

src/Arcp.Client/ArcpClient.fs

@@ -40,18 +40,31 @@ type ArcpClient(transport: ITransport, options: ArcpClientOptions) =
         | Some s when s.NegotiatedFeatures.Contains flag -> Ok()
         | _ -> Error(ARCPError.InvalidRequest(sprintf "Feature %s was not negotiated" flag, None))
 
-    let sendEnvelope (env: Envelope) : Task =
+    let sendEnvelopeCt (env: Envelope) (ct: CancellationToken) : Task =
         let env =
             match sessionCtx with
             | Some s -> Envelope.withSessionId s.SessionId env
             | None -> env
 
-        transport.SendAsync(env, receiveLoopCts.Token)
+        transport.SendAsync(env, ct)
+
+    let sendEnvelope (env: Envelope) : Task = sendEnvelopeCt env receiveLoopCts.Token
 
     let sendMessage (msg: Message) : Task =
         let env = Codec.toEnvelope msg
         sendEnvelope env
 
+    /// Await a correlated response while honoring the caller's token; on
+    /// cancellation drop the pending entry so it does not leak (#98).
+    let awaitResponse (requestId: string) (waiter: Task<Envelope>) (ct: CancellationToken) : Task<Envelope> =
+        task {
+            try
+                return! waiter.WaitAsync(ct)
+            with :? OperationCanceledException as ex ->
+                pending.Remove requestId
+                return raise ex
+        }
+
     // Job-addressed envelopes can arrive before `SubmitAsync`/
     // `SubscribeAsync` register the handle (the receive loop completes
     // the request waiter and races ahead). Buffer such envelopes per
@@ -83,7 +96,10 @@ type ArcpClient(transport: ITransport, options: ArcpClientOptions) =
             w.ResultSetter.TrySetResult(Ok payload) |> ignore
         | Message.JobError payload ->
             handles.TryRemove jid |> ignore
-            let err = JobErrorMapper.ofWire payload.Code payload.Message payload.Details jid
+
+            let err =
+                JobErrorMapper.ofWireWith payload.Code payload.Message payload.Details payload.Retryable (Some jid)
+
             w.Channel.Writer.TryComplete() |> ignore
             w.ResultSetter.TrySetResult(Error err) |> ignore
         | _ -> ()
@@ -236,7 +252,7 @@ type ArcpClient(transport: ITransport, options: ArcpClientOptions) =
             let env = Codec.toEnvelope (Message.SessionHello(buildHello ()))
             let waiter = pending.Register env.Id
             do! transport.SendAsync(env, ct)
-            let! welcomeEnv = waiter
+            let! welcomeEnv = awaitResponse env.Id waiter ct
 
             match Codec.toMessage welcomeEnv with
             | Ok(Message.SessionWelcome w) -> return acceptWelcome welcomeEnv w
@@ -269,8 +285,8 @@ type ArcpClient(transport: ITransport, options: ArcpClientOptions) =
             | Ok() ->
                 let env = Codec.toEnvelope (Message.JobSubmit payload)
                 let waiter = pending.Register env.Id
-                do! sendEnvelope env
-                let! acceptedEnv = waiter
+                do! sendEnvelopeCt env ct
+                let! acceptedEnv = awaitResponse env.Id waiter ct
 
                 match Codec.toMessage acceptedEnv with
                 | Ok(Message.JobAccepted accepted) ->
@@ -293,8 +309,14 @@ type ArcpClient(transport: ITransport, options: ArcpClientOptions) =
                     registerHandle accepted.JobId writer
                     return handle
                 | Ok(Message.JobError errPayload) ->
+                    // §71: no job id context here — pass None.
                     let err =
-                        JobErrorMapper.ofWire errPayload.Code errPayload.Message errPayload.Details ""
+                        JobErrorMapper.ofWireWith
+                            errPayload.Code
+                            errPayload.Message
+                            errPayload.Details
+                            errPayload.Retryable
+                            None
 
                     return raise (ArcpException err)
                 | _ -> return raise (ArcpException(ARCPError.InvalidRequest("Expected job.accepted", None)))
@@ -315,8 +337,8 @@ type ArcpClient(transport: ITransport, options: ArcpClientOptions) =
 
                 let env = Codec.toEnvelope (Message.JobSubscribe payload)
                 let waiter = pending.Register env.Id
-                do! sendEnvelope env
-                let! subscribedEnv = waiter
+                do! sendEnvelopeCt env ct
+                let! subscribedEnv = awaitResponse env.Id waiter ct
 
                 // §7.6 / #96: surface subscription denials instead of
                 // returning a live-looking handle.
@@ -329,7 +351,10 @@ type ArcpClient(transport: ITransport, options: ArcpClientOptions) =
                     registerHandle jobId.Value writer
                     return handle
                 | Ok(Message.SessionError e) ->
-                    return raise (ArcpException(JobErrorMapper.ofWire e.Code e.Message e.Details jobId.Value))
+                    return
+                        raise (
+                            ArcpException(JobErrorMapper.ofWireWith e.Code e.Message e.Details e.Retryable (Some jobId.Value))
+                        )
                 | _ -> return raise (ArcpException(ARCPError.InvalidRequest("Expected job.subscribed", None)))
         }
 
@@ -365,8 +390,8 @@ type ArcpClient(transport: ITransport, options: ArcpClientOptions) =
 
                 let env = Codec.toEnvelope (Message.SessionListJobs payload)
                 let waiter = pending.Register env.Id
-                do! sendEnvelope env
-                let! respEnv = waiter
+                do! sendEnvelopeCt env ct
+                let! respEnv = awaitResponse env.Id waiter ct
 
                 match Codec.toMessage respEnv with
                 | Ok(Message.SessionJobs jobs) -> return jobs
@@ -398,6 +423,11 @@ type ArcpClient(transport: ITransport, options: ArcpClientOptions) =
     /// or fault). Lets callers observe that the client stopped pumping (#61).
     member _.Completion: Task = receiveLoopTask
 
+    /// Resolves with the negotiated session once `session.welcome` is
+    /// received; faults if the handshake fails. A separate handle to the
+    /// connect result for callers that did not await `ConnectAsync` (#70).
+    member _.Connected: Task<SessionContext> = connectedTcs.Task
+
     /// Close the session cleanly with an optional reason.
     member this.CloseAsync(reason: string option, ct: CancellationToken) : Task =
         task {

src/Arcp.Client/Internal/JobErrorMapper.fs

@@ -45,16 +45,23 @@ module internal JobErrorMapper =
             | _ -> None)
         |> Option.defaultValue fallback
 
-    let ofWire
+    /// Map a wire error. `jobId` is the job context when known (used for
+    /// `JOB_NOT_FOUND`); `None` for non-job dispatch sites (#71).
+    /// `retryable` is the wire flag, honored for unknown codes (#90).
+    let ofWireWith
         (code: string)
         (message: string)
         (details: System.Text.Json.JsonElement option)
-        (jobId: string)
+        (retryable: bool)
+        (jobId: string option)
         : ARCPError =
         match code with
         | "PERMISSION_DENIED" -> ARCPError.PermissionDenied(message, details)
         | "LEASE_SUBSET_VIOLATION" -> ARCPError.LeaseSubsetViolation(message, details)
-        | "JOB_NOT_FOUND" -> ARCPError.JobNotFound jobId
+        | "JOB_NOT_FOUND" ->
+            match jobId with
+            | Some j -> ARCPError.JobNotFound j
+            | None -> ARCPError.InvalidRequest(message, details)
         | "DUPLICATE_KEY" -> ARCPError.DuplicateKey message
         | "AGENT_NOT_AVAILABLE" -> ARCPError.AgentNotAvailable message
         | "AGENT_VERSION_NOT_AVAILABLE" -> ARCPError.AgentVersionNotAvailable(message, strField details "version" "")
@@ -67,4 +74,15 @@ module internal JobErrorMapper =
         | "UNAUTHENTICATED" -> ARCPError.Unauthenticated message
         | "RESUME_WINDOW_EXPIRED" ->
             ARCPError.ResumeWindowExpired(int64Field details "from_seq" 0L, intField details "window_sec" 0)
-        | _ -> ARCPError.InternalError message
+        | other -> ARCPError.Unknown(other, message, retryable)
+
+    /// Backwards-compatible entry point: jobId as a plain string and a
+    /// default retryable of false for unknown codes.
+    let ofWire
+        (code: string)
+        (message: string)
+        (details: System.Text.Json.JsonElement option)
+        (jobId: string)
+        : ARCPError =
+        let jid = if String.IsNullOrEmpty jobId then None else Some jobId
+        ofWireWith code message details false jid

src/Arcp.Client/Internal/Pending.fs

@@ -19,7 +19,8 @@ type internal PendingRegistry() =
             TaskCompletionSource<Envelope>(TaskCreationOptions.RunContinuationsAsynchronously)
 
         if not (pending.TryAdd(requestId, tcs)) then
-            failwithf "Duplicate pending request id: %s" requestId
+            // §69: surface as a typed ARCP error rather than a bare exn.
+            raise (ArcpException(ARCPError.InternalError(sprintf "Duplicate pending request id: %s" requestId)))
 
         tcs.Task
 
@@ -32,6 +33,10 @@ type internal PendingRegistry() =
             true
         | _ -> false
 
+    /// Drop a pending request without completing it (e.g. when the
+    /// caller's cancellation token fires).
+    member _.Remove(requestId: string) : unit = pending.TryRemove requestId |> ignore
+
     /// Fail all pending operations (e.g. on transport close).
     member _.FailAll(error: exn) : unit =
         for kvp in pending do

src/Arcp.Client/Transport/Memory.fs

@@ -17,12 +17,6 @@ type MemoryTransport private (outgoing: Channel<Envelope>, incoming: Channel<Env
             task { do! outgoing.Writer.WriteAsync(env, ct).AsTask() } :> Task
 
         member _.Receive(ct) =
-            let enumerable =
-                seq {
-                    while not ct.IsCancellationRequested do
-                        yield incoming
-                }
-            // Convert the channel reader into IAsyncEnumerable via a helper.
             let reader = incoming.Reader
 
             { new IAsyncEnumerable<Envelope> with

src/Arcp.Client/Transport/Stdio.fs

@@ -11,7 +11,12 @@ open ARCP.Client
 /// Newline-delimited JSON over a pair of streams. Used for stdio
 /// child processes (spec §4: stdio mandatory for in-process children).
 type StdioTransport(input: TextReader, output: TextWriter, ownsStreams: bool) =
+    // §68: `closed` is written by CloseAsync (possibly another thread)
+    // and read in the receive loop; mark it volatile so the write is
+    // observed promptly on weak memory models.
+    [<VolatileField>]
     let mutable closed = false
+
     let writeLock = obj ()
 
     interface ITransport with

src/Arcp.Core/Codec.fs

@@ -81,6 +81,15 @@ module Codec =
     /// Parse a JSON string from the wire into an envelope.
     let readEnvelope (json: string) : Result<Envelope, ARCPError> =
         try
-            Ok(Json.deserialize<Envelope> json)
+            let env = Json.deserialize<Envelope> json
+
+            // The JSON literal `null` deserializes to a null record; reject
+            // it (and missing type/id) so the session loop never NREs (§5, §12).
+            if obj.ReferenceEquals(env, null) then
+                Error(ARCPError.InvalidRequest("Envelope must be a JSON object", None))
+            elif String.IsNullOrEmpty env.Type || String.IsNullOrEmpty env.Id then
+                Error(ARCPError.InvalidRequest("Envelope is missing required type/id", None))
+            else
+                Ok env
         with :? JsonException as ex ->
             Error(ARCPError.InvalidRequest(sprintf "Malformed envelope: %s" ex.Message, None))

src/Arcp.Core/Errors.fs

@@ -31,6 +31,9 @@ type ARCPError =
     | InvalidRequest of message: string * details: JsonElement option
     | Unauthenticated of message: string
     | InternalError of message: string
+    /// Forward-compatible arm for wire error codes this SDK version
+    /// does not model, carrying the wire `retryable` flag verbatim (§12).
+    | Unknown of code: string * message: string * retryable: bool
 
 [<RequireQualifiedAccess>]
 module ARCPError =
@@ -52,6 +55,7 @@ module ARCPError =
         | ARCPError.InvalidRequest _ -> "INVALID_REQUEST"
         | ARCPError.Unauthenticated _ -> "UNAUTHENTICATED"
         | ARCPError.InternalError _ -> "INTERNAL_ERROR"
+        | ARCPError.Unknown(c, _, _) -> c
 
     /// Human-readable message; suitable for `error.message` on the wire.
     let message (e: ARCPError) : string =
@@ -73,13 +77,15 @@ module ARCPError =
         | ARCPError.InvalidRequest(m, _) -> m
         | ARCPError.Unauthenticated m -> m
         | ARCPError.InternalError m -> m
+        | ARCPError.Unknown(_, m, _) -> m
 
     /// Spec §12: retryable iff a different attempt could succeed.
     let retryable (e: ARCPError) : bool =
         match e with
         | ARCPError.Timeout _
         | ARCPError.HeartbeatLost
         | ARCPError.InternalError _ -> true
+        | ARCPError.Unknown(_, _, r) -> r
         | _ -> false
 
     let details (e: ARCPError) : JsonElement option =
@@ -105,8 +111,11 @@ type ArcpException(error: ARCPError, ?inner: exn) =
     member _.Code = ARCPError.code error
     member _.Retryable = ARCPError.retryable error
 
+/// Helpers bridging `Result<_, ARCPError>` and the throwing C#-style
+/// surface. Named `ArcpResult` (not `Result`) so it does not shadow
+/// FSharp.Core's `Result` for consumers that `open ARCP.Core` (#118).
 [<RequireQualifiedAccess>]
-module Result =
+module ArcpResult =
     /// Throw `ArcpException` on `Error`, return the value on `Ok`.
     /// The seam between internal `Result<_, ARCPError>` and the
     /// public C#-friendly throwing API.