diff --git a/client/API.hs b/client/API.hs
--- a/client/API.hs
+++ b/client/API.hs
@@ -6,8 +6,10 @@
   , PublishOption
   , SubscribeOption
   , withSubscriptionExpiry
+  , withQueueGroup
   , withPayload
   , withReplyCallback
+  , withReplyTo
   , withHeaders
   ) where
 
@@ -29,6 +31,8 @@
                   -- ^ Subscribe with request semantics and auto-unsubscribe after a reply.
                 , unsubscribe :: SID -> IO ()
                   -- ^ Unsubscribe from a subscription by SID.
+                , newInbox :: IO Subject
+                  -- ^ Create a unique inbox subject for replies.
                 , ping :: IO () -> IO ()
                   -- ^ Send a ping and run the callback when a pong arrives.
                 , flush :: IO ()
@@ -71,6 +75,11 @@
 withSubscriptionExpiry :: NominalDiffTime -> SubscribeOption
 withSubscriptionExpiry expirySeconds cfg = cfg { expiry = Just expirySeconds }
 
+-- | withQueueGroup sets the queue group for a subscription.
+-- Default: no queue group.
+withQueueGroup :: Subject -> SubscribeOption
+withQueueGroup queueGroup cfg = cfg { subscribeQueueGroup = Just queueGroup }
+
 -- | withPayload is used to set the payload for a publish operation.
 -- Default: no payload.
 --
@@ -82,7 +91,8 @@
 -- publish client \"updates\" [withPayload \"hello\"]
 -- @
 withPayload :: Payload -> PublishOption
-withPayload payload (_, callback, headers) = (Just payload, callback, headers)
+withPayload payload (_, callback, headers, replyTo') =
+  (Just payload, callback, headers, replyTo')
 
 -- | withReplyCallback is used to set a callback for a reply to a publish operation.
 -- Default: no reply subscription; publishes are fire-and-forget.
@@ -95,15 +105,25 @@
 -- publish client \"service.echo\" [withReplyCallback print]
 -- @
 withReplyCallback :: (Maybe MsgView -> IO ()) -> PublishOption
-withReplyCallback callback (payload, _, headers) =
+withReplyCallback callback (payload, _, headers, replyTo') =
   (payload, Just (callback . fmap (\msg -> MsgView
     { subject = Msg.subject msg
     , sid = Msg.sid msg
     , replyTo = Msg.replyTo msg
     , payload = Msg.payload msg
     , headers = Msg.headers msg
-    })), headers)
+    })), headers, replyTo')
 
+-- | withReplyTo sets an explicit reply subject for a publish operation.
+-- Default: no reply subject unless a reply callback is configured.
+--
+-- This option only controls the outgoing reply subject. It does not create a
+-- subscription; callers that expect multiple replies should subscribe to the
+-- reply subject themselves.
+withReplyTo :: Subject -> PublishOption
+withReplyTo replySubject (payload, callback, headers, _) =
+  (payload, callback, headers, Just replySubject)
+
 -- | withHeaders is used to set headers for a publish operation.
 -- Default: no headers.
 --
@@ -115,4 +135,5 @@
 -- publish client \"updates\" [withHeaders [(\"source\", \"test\")]]
 -- @
 withHeaders :: Headers -> PublishOption
-withHeaders headers (payload, callback, _) = (payload, callback, Just headers)
+withHeaders headers (payload, callback, _, replyTo') =
+  (payload, callback, Just headers, replyTo')
diff --git a/client/Client.hs b/client/Client.hs
--- a/client/Client.hs
+++ b/client/Client.hs
@@ -101,7 +101,7 @@
     , unregister
     )
 import           Subscription.Types
-    ( SubscribeConfig (SubscribeConfig)
+    ( SubscribeConfig (..)
     , SubscriptionMeta (SubscriptionMeta)
     )
 import qualified Types.Connect            as Connect
@@ -201,6 +201,7 @@
         let cfg = applyCallOptions subscribeOptions defaultSubscribeConfig
         subscribeClient clientState store True subject cfg (toInternalCallback callback)
     , unsubscribe = unsubscribeClient clientState store
+    , newInbox = nextInbox clientState
     , ping = pingClient clientState
     , flush = flushClient clientState
     , reset = resetClient connectionApi clientState store
@@ -289,7 +290,7 @@
     }
 
 defaultSubscribeConfig :: SubscribeConfig
-defaultSubscribeConfig = SubscribeConfig Nothing
+defaultSubscribeConfig = SubscribeConfig Nothing Nothing
 
 selectAuth :: ClientAuth -> Auth
 selectAuth authSelection =
@@ -354,14 +355,14 @@
     }
 
 publishClient :: ClientState -> SubscriptionStore -> Msg.Subject -> PublishConfig -> IO ()
-publishClient client store subject (payload, callback, headers) = do
+publishClient client store subject (payload, callback, headers, configuredReplyTo) = do
   runClient client $
     logMessage Debug ("publishing to subject: " ++ show subject)
   replyTo <- case callback of
     Nothing ->
-      pure Nothing
+      pure configuredReplyTo
     Just replyCallback -> do
-      inbox <- nextInbox client
+      inbox <- maybe (nextInbox client) pure configuredReplyTo
       sid <- nextSid client
       let meta =
             SubscriptionMeta inbox Nothing True
@@ -394,14 +395,16 @@
   runClient client $
     logMessage Debug ("subscribing to subject: " ++ show subject)
   sid <- nextSid client
+  let queueGroup =
+        subscribeQueueGroup cfg
   let meta =
-        SubscriptionMeta subject Nothing isReply
+        SubscriptionMeta subject queueGroup isReply
   register store sid meta cfg callback
   enqueue client $
     QueueItem
       Sub.Sub
         { Sub.subject = subject
-        , Sub.queueGroup = Nothing
+        , Sub.queueGroup = queueGroup
         , Sub.sid = sid
         }
   when isReply $
diff --git a/internal/Plumbing/Parser/Attoparsec.hs b/internal/Plumbing/Parser/Attoparsec.hs
--- a/internal/Plumbing/Parser/Attoparsec.hs
+++ b/internal/Plumbing/Parser/Attoparsec.hs
@@ -183,9 +183,47 @@
 headerBlockParser :: A.Parser [(BS.ByteString, BS.ByteString)]
 headerBlockParser = do
   _ <- A.string "NATS/"
-  _ <- lineParser
-  headerPairsParser []
+  prelude <- lineParser
+  statusHeaders <- inlineStatusHeaders prelude
+  headerPairs <- headerPairsParser []
+  pure (statusHeaders ++ headerPairs)
 
+inlineStatusHeaders :: BS.ByteString -> A.Parser [(BS.ByteString, BS.ByteString)]
+inlineStatusHeaders prelude =
+  case BS.dropWhile isHorizontalSpaceByte rest of
+    statusBytes
+      | BS.null statusBytes ->
+          pure []
+      | otherwise ->
+          parseInlineStatus statusBytes
+  where
+    (_, rest) = BS.break isHorizontalSpaceByte prelude
+
+parseInlineStatus :: BS.ByteString -> A.Parser [(BS.ByteString, BS.ByteString)]
+parseInlineStatus bytes
+  | BS.length bytes < inlineStatusLength =
+      fail ("invalid HMSG status: " ++ B8.unpack bytes)
+  | not (isInlineStatusCode statusCode) =
+      fail ("invalid HMSG status: " ++ B8.unpack statusCode)
+  | BS.null remainder =
+      pure [("Status", statusCode)]
+  | isHorizontalSpaceByte (BS.head remainder) =
+      pure (("Status", statusCode) : descriptionHeader)
+  | otherwise =
+      fail ("invalid HMSG status: " ++ B8.unpack bytes)
+  where
+    (statusCode, remainder) = BS.splitAt inlineStatusLength bytes
+    description = stripHorizontalSpace remainder
+    descriptionHeader
+      | BS.null description =
+          []
+      | otherwise =
+          [("Description", description)]
+
+isInlineStatusCode :: BS.ByteString -> Bool
+isInlineStatusCode bytes =
+  BS.length bytes == inlineStatusLength && BS.all isDigitByte bytes
+
 headerPairsParser :: [(BS.ByteString, BS.ByteString)] -> A.Parser [(BS.ByteString, BS.ByteString)]
 headerPairsParser reversedPairs = do
   nextByte <- A.peekWord8'
@@ -276,6 +314,13 @@
 isHorizontalSpaceByte word8Value =
   word8Value == space || word8Value == horizontalTab
 
+isDigitByte :: Word8 -> Bool
+isDigitByte word8Value =
+  word8Value >= zero && word8Value <= nine
+
+inlineStatusLength :: Int
+inlineStatusLength = 3
+
 carriageReturn :: Word8
 carriageReturn = 13
 
@@ -290,3 +335,9 @@
 
 horizontalTab :: Word8
 horizontalTab = 9
+
+zero :: Word8
+zero = 48
+
+nine :: Word8
+nine = 57
diff --git a/internal/Policy/Publish.hs b/internal/Policy/Publish.hs
--- a/internal/Policy/Publish.hs
+++ b/internal/Policy/Publish.hs
@@ -5,4 +5,4 @@
 import           Publish.Config (PublishConfig)
 
 defaultPublishConfig :: PublishConfig
-defaultPublishConfig = (Nothing, Nothing, Nothing)
+defaultPublishConfig = (Nothing, Nothing, Nothing, Nothing)
diff --git a/internal/Policy/Publish/Config.hs b/internal/Policy/Publish/Config.hs
--- a/internal/Policy/Publish/Config.hs
+++ b/internal/Policy/Publish/Config.hs
@@ -5,6 +5,6 @@
   ) where
 
 import qualified Types.Msg as M
-import           Types.Msg (Headers, Payload)
+import           Types.Msg (Headers, Payload, Subject)
 
-type PublishConfig = (Maybe Payload, Maybe (Maybe M.Msg -> IO ()), Maybe Headers)
+type PublishConfig = (Maybe Payload, Maybe (Maybe M.Msg -> IO ()), Maybe Headers, Maybe Subject)
diff --git a/internal/Policy/Subscription/Types.hs b/internal/Policy/Subscription/Types.hs
--- a/internal/Policy/Subscription/Types.hs
+++ b/internal/Policy/Subscription/Types.hs
@@ -6,7 +6,10 @@
 import           Data.Time.Clock (NominalDiffTime)
 import           Types.Msg       (Subject)
 
-newtype SubscribeConfig = SubscribeConfig { expiry :: Maybe NominalDiffTime }
+data SubscribeConfig = SubscribeConfig
+                         { expiry              :: Maybe NominalDiffTime
+                         , subscribeQueueGroup :: Maybe Subject
+                         }
 
 data SubscriptionMeta = SubscriptionMeta
                           { subject    :: Subject
diff --git a/natskell.cabal b/natskell.cabal
--- a/natskell.cabal
+++ b/natskell.cabal
@@ -1,6 +1,6 @@
 cabal-version:          3.0
 name:                   natskell
-version:                0.0.0.1
+version:                0.1.0.0
 synopsis:               A NATS client library written in Haskell
 tested-with:
   GHC ==8.8,
@@ -50,7 +50,6 @@
     natskell-internal,
     stm  >= 2.5 && < 2.6,
     bytestring  >= 0.10 && < 0.13,
-    heap >= 0.6 && < 1.1,
     time >= 1.9 && < 1.16
 
   exposed-modules:
diff --git a/test/Integration/ClientSpec.hs b/test/Integration/ClientSpec.hs
--- a/test/Integration/ClientSpec.hs
+++ b/test/Integration/ClientSpec.hs
@@ -5,6 +5,9 @@
 import           API
     ( Client (..)
     , MsgView (..)
+    , withPayload
+    , withQueueGroup
+    , withReplyTo
     , withSubscriptionExpiry
     )
 import           Client
@@ -24,7 +27,7 @@
 import           Data.Word8
 import           Network.Socket            (Socket, accept, listen)
 import qualified Network.Socket
-import           Network.Socket.ByteString (sendAll)
+import           Network.Socket.ByteString (recv, sendAll)
 import           Network.Socket.Free
 import           System.Timeout            (timeout)
 import           Test.Hspec
@@ -74,6 +77,33 @@
     , "\r\n"
     ]
 
+recvUntil :: Socket -> (BS.ByteString -> Bool) -> IO BS.ByteString
+recvUntil sock done =
+  go BS.empty
+  where
+    go acc
+      | done acc = pure acc
+      | otherwise = do
+          chunk <- recv sock 4096
+          if BS.null chunk
+            then pure acc
+            else go (acc <> chunk)
+
+expectClientCommand :: Socket -> BS.ByteString -> IO ()
+expectClientCommand sock command = do
+  result <- timeout 1000000 $
+    recvUntil sock (BS.isInfixOf command)
+  case result of
+    Nothing ->
+      expectationFailure ("client did not send command: " ++ show command)
+    Just bytes ->
+      unless (BS.isInfixOf command bytes) $
+        expectationFailure ("unexpected client bytes: " ++ show bytes)
+
+expectQueuedSub :: Socket -> BS.ByteString -> BS.ByteString -> BS.ByteString -> IO ()
+expectQueuedSub sock subject queueGroup sid =
+  expectClientCommand sock (BS.concat ["SUB ", subject, " ", queueGroup, " ", sid, "\r\n"])
+
 startClientWith extraOptions = do
   (p, sock) <- openFreePort
   listen sock 1
@@ -145,6 +175,12 @@
         case result of
           Nothing -> expectationFailure "flush did not return after PONG"
           Just () -> pure ()
+      it "subscribes with a queue group" $ \(serv, client, _, _) -> do
+        sid <- subscribe client "JOBS" [withQueueGroup "WORKERS"] (const (pure ()))
+        expectQueuedSub serv "JOBS" "WORKERS" sid
+      it "publishes with an explicit reply subject" $ \(serv, client, _, _) -> do
+        publish client "REQUESTS" [withPayload "hello", withReplyTo "_INBOX.custom"]
+        expectClientCommand serv "PUB REQUESTS _INBOX.custom 5\r\nhello\r\n"
       it "PONG resolves one ping" $ \(serv, client, _, _) -> do
         first <- newEmptyMVar
         second <- newEmptyMVar
@@ -299,6 +335,33 @@
           ping client $ done wg
           sendAll secondConn "PONG\r\n"
           wait wg
+          close client
+          Network.Socket.close secondConn
+      Network.Socket.close sock
+    it "resubscribes with a queue group after reconnect" $ do
+      (p, sock) <- openFreePort
+      listen sock 2
+      clientVar <- newEmptyMVar
+      void . forkIO $ do
+        let configOptions =
+              [ withMinimumLogLevel Debug
+              , withConnectionAttempts 2
+              , withConnectName "test-client"
+              ]
+        c <- newClient [("127.0.0.1", p)] configOptions
+        putMVar clientVar c
+      (firstConn, _) <- accept sock
+      sendAll firstConn defaultINFO
+      clientResult <- timeout 1000000 (takeMVar clientVar)
+      case clientResult of
+        Nothing -> expectationFailure "client did not connect"
+        Just client -> do
+          sid <- subscribe client "JOBS" [withQueueGroup "WORKERS"] (const (pure ()))
+          expectQueuedSub firstConn "JOBS" "WORKERS" sid
+          Network.Socket.close firstConn
+          (secondConn, _) <- accept sock
+          sendAll secondConn defaultINFO
+          expectQueuedSub secondConn "JOBS" "WORKERS" sid
           close client
           Network.Socket.close secondConn
       Network.Socket.close sock
diff --git a/test/Unit/ParserSpec.hs b/test/Unit/ParserSpec.hs
--- a/test/Unit/ParserSpec.hs
+++ b/test/Unit/ParserSpec.hs
@@ -58,3 +58,43 @@
               expectationFailure ("unexpected parse result: " ++ show other)
         other ->
           expectationFailure ("unexpected parse result: " ++ show other)
+
+    it "exposes inline HMSG status and description as headers" $ do
+      let input =
+            "HMSG FOO 13 28 28\r\nNATS/1.0 404 No Messages\r\n\r\n\r\n"
+      case parse parserApi input of
+        Emit parsed rest -> do
+          rest `shouldBe` ""
+          case parsed of
+            ParsedMsg msg' -> do
+              Msg.headers msg' `shouldBe` Just [("Status", "404"), ("Description", "No Messages")]
+              Msg.payload msg' `shouldBe` Just ""
+            other ->
+              expectationFailure ("unexpected parse result: " ++ show other)
+        other ->
+          expectationFailure ("unexpected parse result: " ++ show other)
+
+    it "exposes inline HMSG status without a description" $ do
+      let input =
+            "HMSG FOO 13 16 16\r\nNATS/1.0 404\r\n\r\n\r\n"
+      case parse parserApi input of
+        Emit parsed rest -> do
+          rest `shouldBe` ""
+          case parsed of
+            ParsedMsg msg' -> do
+              Msg.headers msg' `shouldBe` Just [("Status", "404")]
+              Msg.payload msg' `shouldBe` Just ""
+            other ->
+              expectationFailure ("unexpected parse result: " ++ show other)
+        other ->
+          expectationFailure ("unexpected parse result: " ++ show other)
+
+    it "rejects malformed inline HMSG statuses" $ do
+      let input =
+            "HMSG FOO 13 17 17\r\nNATS/1.0 404x\r\n\r\n\r\n"
+      case parse parserApi input of
+        DropPrefix n reason -> do
+          n `shouldSatisfy` (> 0)
+          reason `shouldContain` "invalid HMSG status"
+        other ->
+          expectationFailure ("unexpected parse result: " ++ show other)
diff --git a/test/Unit/SubSpec.hs b/test/Unit/SubSpec.hs
--- a/test/Unit/SubSpec.hs
+++ b/test/Unit/SubSpec.hs
@@ -5,6 +5,7 @@
 import           Control.Monad
 import           Data.ByteString
 import qualified Data.ByteString.Lazy      as LBS
+import           Subscription.Types
 import           Test.Hspec
 import           Text.Printf
 import           Transformers.Transformers
@@ -13,8 +14,15 @@
 
 spec :: Spec
 spec = do
+  subscribeOptionCases
   transformerCases
   validateCases
+
+subscribeOptionCases = parallel $ do
+  describe "subscribe config" $ do
+    it "stores the queue group" $ do
+      subscribeQueueGroup (SubscribeConfig Nothing (Just "WORKERS"))
+        `shouldBe` Just "WORKERS"
 
 explicitTransformerCases :: [(Sub, ByteString)]
 explicitTransformerCases = [