diff --git a/baikai-openai.cabal b/baikai-openai.cabal --- a/baikai-openai.cabal +++ b/baikai-openai.cabal @@ -1,6 +1,6 @@ cabal-version: 3.4 name: baikai-openai -version: 0.1.1.0 +version: 0.2.0.0 synopsis: OpenAI providers for the baikai abstraction description: Wraps the openai Haskell package as a Baikai Provider for OpenAI's Chat Completions API. @@ -33,18 +33,21 @@ exposed-modules: Baikai.Provider.OpenAI.Api Baikai.Provider.OpenAI.Cli + Baikai.Provider.OpenAI.ErrorClass Baikai.Provider.OpenAI.Interactive build-depends: , aeson - , baikai ^>=0.1.1 + , baikai ^>=0.2.0 , base >=4.20 && <5 , base64-bytestring , bytestring + , case-insensitive , containers , generic-lens , http-client , http-client-tls + , http-types , lens ^>=5.3 , openai , process @@ -60,17 +63,22 @@ type: exitcode-stdio-1.0 hs-source-dirs: test main-is: Main.hs + other-modules: ErrorClassSpec build-depends: , aeson - , baikai ^>=0.1.1 + , baikai ^>=0.2.0 , baikai-openai - , base >=4.20 && <5 + , base >=4.20 && <5 , bytestring + , case-insensitive + , containers , generic-lens - , lens ^>=5.3 + , http-types + , lens ^>=5.3 , openai - , streamly-core >=0.3 && <0.5 + , servant-client + , streamly-core >=0.3 && <0.5 , tasty , tasty-hunit - , text ^>=2.1 + , text ^>=2.1 , vector diff --git a/src/Baikai/Provider/OpenAI/Api.hs b/src/Baikai/Provider/OpenAI/Api.hs --- a/src/Baikai/Provider/OpenAI/Api.hs +++ b/src/Baikai/Provider/OpenAI/Api.hs @@ -44,9 +44,11 @@ import Baikai.Context (Context (..)) import Baikai.Cost (_Cost) import Baikai.Cost.Pricing qualified as Pricing +import Baikai.Error (BaikaiError, invalidRequest) import Baikai.Message qualified as Msg import Baikai.Model (Model, openaiCompletionsCompatFor) import Baikai.Options (Options (..)) +import Baikai.Provider.OpenAI.ErrorClass (classifyErrorText, classifyException) import Baikai.Provider.Registry ( ApiProvider (..), ProviderRegistry, @@ -62,8 +64,9 @@ DeltaPayload (..), IndexPayload (..), StartPayload (..), - TerminalPayload (..), ToolCallEndPayload (..), + doneTerminal, + errorTerminal, ) import Baikai.ThinkingLevel ( ThinkingLevel (..), @@ -72,7 +75,7 @@ import Baikai.Usage qualified as Usage import Control.Concurrent (forkIO) import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan) -import Control.Exception (SomeException, displayException, try) +import Control.Exception (SomeException, try) import Control.Lens ((%~), (&), (.~), (^.)) import Data.Aeson (Value (..), (.:?)) import Data.Aeson qualified as Aeson @@ -137,7 +140,8 @@ Right call -> do ch <- newChan :: IO (Chan (Maybe RawChunk)) tref <- newIORef False - _ <- forkIO (worker call ch) + eref <- newIORef Nothing + _ <- forkIO (worker call ch eref) startTime <- getCurrentTime let initialState = ProducerState @@ -145,7 +149,8 @@ pending = [EventStart StartPayload {partial = skeletonStart m startTime}], assembler = emptyAssembler m startTime, finished = False, - terminalRef = tref + terminalRef = tref, + errInfoRef = eref } pure (Stream.unfoldrM step initialState) @@ -223,8 +228,8 @@ deriving stock (Show, Generic) worker :: - OpenAICall -> Chan (Maybe RawChunk) -> IO () -worker call ch = do + OpenAICall -> Chan (Maybe RawChunk) -> IORef (Maybe BaikaiError) -> IO () +worker call ch errInfoRef = do let OpenAI.Methods {OpenAI.createChatCompletionStream = stream'} = call ^. #methods r <- try @SomeException $ @@ -235,8 +240,10 @@ Right chunk -> writeChan ch (Just chunk) case r of Right () -> pure () - Left e -> - writeChan ch (Just (errorChunk (Text.pack (displayException e)))) + -- An HTTP-level exception carries an HTTP status; classify it and + -- stash the structured error for the end-of-stream recovery path + -- (see 'step' / 'closeOpenStream'). No terminal chunk is written. + Left e -> writeIORef errInfoRef (Just (classifyException e)) writeChan ch Nothing errorChunk :: Text -> RawChunk @@ -363,7 +370,10 @@ pending :: ![AssistantMessageEvent], assembler :: !Assembler, finished :: !Bool, - terminalRef :: !(IORef Bool) + terminalRef :: !(IORef Bool), + -- | Set by the worker when an HTTP-level exception is caught, so the + -- end-of-stream recovery path can surface a categorised error. + errInfoRef :: !(IORef (Maybe BaikaiError)) } deriving stock (Generic) @@ -389,7 +399,8 @@ then pure Nothing else do now <- getCurrentTime - let (events, ass') = closeOpenStream now (s ^. #assembler) + mErr <- readIORef (s ^. #errInfoRef) + let (events, ass') = closeOpenStream now mErr (s ^. #assembler) case events of [] -> pure Nothing (e : rest) -> do @@ -491,7 +502,7 @@ translate chunk ass now | Just errMsg <- chunk ^. #error = let msg = finalMessage ass now (Just errMsg) Stop.ErrorReason - in ( [EventError TerminalPayload {reason = Stop.ErrorReason, message = msg}], + in ( [EventError (errorTerminal Stop.ErrorReason msg (classifyErrorText errMsg))], ass & #errorMsg .~ Just errMsg ) | otherwise = @@ -644,28 +655,27 @@ ) closeOpenStream :: - UTCTime -> Assembler -> ([AssistantMessageEvent], Assembler) -closeOpenStream now ass + UTCTime -> Maybe BaikaiError -> Assembler -> ([AssistantMessageEvent], Assembler) +closeOpenStream now mErr ass | ass ^. #finishSeen = -- Channel closed cleanly after finish_reason. Emit -- EventDone with the accumulated content + usage. let reason = ass ^. #stopReason msg = finalMessage ass now Nothing reason - in ([EventDone TerminalPayload {reason = reason, message = msg}], ass) + in ([EventDone (doneTerminal reason msg)], ass) | otherwise = -- Channel closed without a finish_reason. Force-close any - -- still-open blocks and emit EventError with the accumulated - -- content. + -- still-open blocks and emit EventError. When the worker stored a + -- classified HTTP error ('Just be'), surface it structurally; + -- otherwise report the unexpected end of stream. let (closeText, ass1) = closeOpenText ass (closeTools, ass2) = closeOpenTools ass1 reason = Stop.ErrorReason - msg = - finalMessage - ass2 - now - (Just "openai stream ended without finish_reason") - reason - errEv = EventError TerminalPayload {reason = reason, message = msg} + errText = case mErr of + Just be -> be ^. #message + Nothing -> "openai stream ended without finish_reason" + msg = finalMessage ass2 now (Just errText) reason + errEv = EventError (errorTerminal reason msg mErr) in (closeText <> closeTools <> [errEv], ass2) finalMessage :: @@ -702,7 +712,7 @@ Msg.errorMessage = Just errText, Msg.timestamp = now } - pure (EventError TerminalPayload {reason = Stop.ErrorReason, message = msg}) + pure (EventError (errorTerminal Stop.ErrorReason msg (Just (invalidRequest errText)))) -- ============================================================ -- Request mapping (preserved from EP-2 with minor refactoring) diff --git a/src/Baikai/Provider/OpenAI/Cli.hs b/src/Baikai/Provider/OpenAI/Cli.hs --- a/src/Baikai/Provider/OpenAI/Cli.hs +++ b/src/Baikai/Provider/OpenAI/Cli.hs @@ -17,7 +17,7 @@ import Baikai.Api (Api (..)) import Baikai.Content (AssistantContent (..), TextContent (..)) import Baikai.Context (Context) -import Baikai.Error (BaikaiError (..)) +import Baikai.Error (processError, providerError) import Baikai.Message (AssistantPayload (..)) import Baikai.Model (Model) import Baikai.Options (Options) @@ -152,14 +152,14 @@ (Maybe Handle, Maybe Handle, Maybe Handle, P.ProcessHandle) -> IO Resp.Response consume start m (_, mOut, mErr, ph) = do - hOut <- maybe (throwIO (ProviderError "codex: stdout handle missing")) pure mOut - hErr <- maybe (throwIO (ProviderError "codex: stderr handle missing")) pure mErr + hOut <- maybe (throwIO (providerError "codex: stdout handle missing")) pure mOut + hErr <- maybe (throwIO (providerError "codex: stderr handle missing")) pure mErr body <- Internal.parseCodexJsonlStream (handleStream hOut) errBytes <- BS.hGetContents hErr exitCode <- P.waitForProcess ph end <- getCurrentTime case exitCode of - ExitFailure n -> throwIO (ProcessError n (Internal.decodeUtf8Lenient errBytes)) + ExitFailure n -> throwIO (processError n (Internal.decodeUtf8Lenient errBytes)) ExitSuccess -> pure Resp.Response @@ -176,7 +176,8 @@ Resp.api = OpenAICompletionsCli, Resp.provider = m ^. #provider, Resp.responseId = Nothing, - Resp.latencyMs = millisBetween start end + Resp.latencyMs = millisBetween start end, + Resp.errorInfo = Nothing } millisBetween :: UTCTime -> UTCTime -> Integer diff --git a/src/Baikai/Provider/OpenAI/ErrorClass.hs b/src/Baikai/Provider/OpenAI/ErrorClass.hs new file mode 100644 --- /dev/null +++ b/src/Baikai/Provider/OpenAI/ErrorClass.hs @@ -0,0 +1,104 @@ +-- | Map failures from the OpenAI SDK onto baikai's typed 'BaikaiError'. +-- 'classifyException' handles an exception thrown by the +-- @servant-client@ HTTP layer (carrying an HTTP status); +-- 'classifyErrorText' handles an error that arrives mid-stream as a +-- plain text message (the OpenAI streaming chunk's @error@ field). +module Baikai.Provider.OpenAI.ErrorClass + ( classifyException, + classifyErrorText, + -- | Exposed for testing the HTTP-status mapping without a live call. + responseToError, + ) +where + +import Baikai.Error + ( BaikaiError (..), + ErrorCategory (..), + bodyIndicatesOverflow, + classifyHttpStatusWithBody, + decodeError, + providerError, + ) +import Control.Exception (SomeException, displayException, fromException) +import Data.ByteString (ByteString) +import Data.ByteString.Lazy qualified as LBS +import Data.CaseInsensitive qualified as CI +import Data.Foldable (toList) +import Data.Sequence (Seq) +import Data.Text (Text) +import Data.Text qualified as Text +import Data.Text.Encoding qualified as Text +import Data.Text.Encoding.Error qualified as Text +import Network.HTTP.Types.Status (statusCode) +import Servant.Client (ClientError, ResponseF (..)) +import Servant.Client qualified as Servant +import Text.Read (readMaybe) + +-- | Convert any exception caught from the OpenAI SDK into a categorised +-- 'BaikaiError'. Recognises @servant-client@ 'ClientError'; anything +-- else degrades to a generic provider error carrying the displayed +-- exception text. +classifyException :: SomeException -> BaikaiError +classifyException ex = case fromException ex of + Just clientErr -> fromClientError clientErr + Nothing -> providerError (Text.pack (displayException ex)) + +fromClientError :: ClientError -> BaikaiError +fromClientError clientErr = case clientErr of + Servant.FailureResponse _req resp -> responseToError resp + Servant.DecodeFailure detail _ -> decodeError detail + Servant.UnsupportedContentType _ _ -> decodeError "unsupported content type in OpenAI response" + Servant.InvalidContentTypeHeader _ -> decodeError "invalid content-type header in OpenAI response" + Servant.ConnectionError exc -> + (providerError ("connection error: " <> Text.pack (displayException exc))) + { category = TransientError + } + +responseToError :: ResponseF LBS.ByteString -> BaikaiError +responseToError resp = + BaikaiError + { category = classifyHttpStatusWithBody status retryAfter body, + message = msg, + httpStatus = Just status, + retryAfterSeconds = retryAfter, + exitCode = Nothing + } + where + status = statusCode (responseStatusCode resp) + body = decodeLenient (LBS.toStrict (responseBody resp)) + retryAfter = parseRetryAfter (responseHeaders resp) + snippet = Text.take 300 (Text.strip body) + msg = + "HTTP " + <> Text.pack (show status) + <> (if Text.null snippet then "" else ": " <> snippet) + +parseRetryAfter :: Seq (CI.CI ByteString, ByteString) -> Maybe Int +parseRetryAfter headers = do + raw <- lookup (CI.mk "Retry-After") (toList headers) + readMaybe (Text.unpack (Text.strip (decodeLenient raw))) + +decodeLenient :: ByteString -> Text +decodeLenient = Text.decodeUtf8With Text.lenientDecode + +-- | Best-effort classification of an OpenAI streamed error message, +-- which arrives as plain text without an HTTP status. Returns 'Nothing' +-- for empty text (the caller keeps the raw message and 'errorInfo' +-- stays absent). +classifyErrorText :: Text -> Maybe BaikaiError +classifyErrorText t + | Text.null (Text.strip t) = Nothing + | otherwise = Just (providerError t) {category = cat} + where + lower = Text.toLower t + has needle = needle `Text.isInfixOf` lower + cat + | bodyIndicatesOverflow t = ContextOverflow + | has "rate limit" || has "rate_limit" = RateLimited + | has "overloaded" || has "server_error" || has "service unavailable" = TransientError + | has "insufficient_quota" + || has "invalid api key" + || has "incorrect api key" + || has "invalid_api_key" = + AuthError + | otherwise = OtherError diff --git a/test/ErrorClassSpec.hs b/test/ErrorClassSpec.hs new file mode 100644 --- /dev/null +++ b/test/ErrorClassSpec.hs @@ -0,0 +1,85 @@ +module ErrorClassSpec (tests) where + +import Baikai.Error (BaikaiError (..), ErrorCategory (..)) +import Baikai.Provider.OpenAI.ErrorClass + ( classifyErrorText, + classifyException, + responseToError, + ) +import Control.Exception (toException) +import Data.ByteString (ByteString) +import Data.ByteString.Lazy qualified as LBS +import Data.CaseInsensitive qualified as CI +import Data.Sequence qualified as Seq +import Data.Text qualified as Text +import Network.HTTP.Types.Status (mkStatus) +import Network.HTTP.Types.Version (http11) +import Servant.Client (ResponseF (..)) +import Test.Tasty (TestTree, testGroup) +import Test.Tasty.HUnit (assertBool, testCase, (@?=)) + +tests :: TestTree +tests = + testGroup + "Baikai.Provider.OpenAI.ErrorClass" + [ httpStatusTests, + streamedErrorTests, + fallbackTests + ] + +mkResp :: Int -> [(ByteString, ByteString)] -> LBS.ByteString -> ResponseF LBS.ByteString +mkResp status hdrs body = + Response + { responseStatusCode = mkStatus status "", + responseHeaders = Seq.fromList [(CI.mk n, v) | (n, v) <- hdrs], + responseHttpVersion = http11, + responseBody = body + } + +httpStatusTests :: TestTree +httpStatusTests = + testGroup + "responseToError (HTTP status)" + [ testCase "429 + Retry-After -> RateLimited with hint" $ do + let e = responseToError (mkResp 429 [("Retry-After", "12")] "slow down") + category e @?= RateLimited + httpStatus e @?= Just 429 + retryAfterSeconds e @?= Just 12, + testCase "401 -> AuthError" $ + category (responseToError (mkResp 401 [] "bad key")) @?= AuthError, + testCase "400 with overflow body -> ContextOverflow" $ + category (responseToError (mkResp 400 [] "maximum context length is 8192 tokens")) + @?= ContextOverflow, + testCase "400 ordinary -> InvalidRequest" $ + category (responseToError (mkResp 400 [] "unknown parameter")) @?= InvalidRequest, + testCase "500 -> TransientError" $ + category (responseToError (mkResp 500 [] "")) @?= TransientError + ] + +streamedErrorTests :: TestTree +streamedErrorTests = + testGroup + "classifyErrorText (mid-stream error text)" + [ testCase "rate limit text -> RateLimited" $ + fmap category (classifyErrorText "Rate limit reached for requests") @?= Just RateLimited, + testCase "context length text -> ContextOverflow" $ + fmap category (classifyErrorText "This model's maximum context length is 4096 tokens") + @?= Just ContextOverflow, + testCase "invalid api key text -> AuthError" $ + fmap category (classifyErrorText "Incorrect API key provided") @?= Just AuthError, + testCase "unknown text -> OtherError" $ + fmap category (classifyErrorText "something odd happened") @?= Just OtherError, + testCase "blank text -> Nothing" $ + classifyErrorText " " @?= Nothing + ] + +fallbackTests :: TestTree +fallbackTests = + testGroup + "classifyException fallback" + [ testCase "non-ClientError exception -> OtherError, text preserved" $ do + let e = classifyException (toException (userError "weird failure")) + category e @?= OtherError + assertBool "message keeps the original text" $ + "weird failure" `Text.isInfixOf` message e + ] diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -9,6 +9,7 @@ import Data.Generics.Labels () import Data.Text qualified as Text import Data.Vector qualified as Vector +import ErrorClassSpec qualified import OpenAI.V1.Chat.Completions qualified as Chat import OpenAI.V1.ResponseFormat qualified as RF import Streamly.Data.Stream qualified as Stream @@ -24,7 +25,8 @@ promptRenderingTest, compatDetectionTest, rejectsImageToolResultsTest, - responseFormatMappingTest + responseFormatMappingTest, + ErrorClassSpec.tests ] -- | A 'JsonSchema' on 'Options.responseFormat' maps onto the