Provider hardening: retries, tools guard, stream parser, silent catches
ober
02fba8c22fa416c9d637ff3a07e79b91e51d02f2
--- a/src/jcode/core/config.ss +++ b/src/jcode/core/config.ss @@ -13,9 +13,12 @@ (import :std/text/json :std/os/path + :jcode/core/log :jcode/core/models :jcode/core/secrets) +(def logger (make-logger "config")) + (def *config* (make-parameter #f)) (def *version* "0.1.1") @@ -90,7 +93,9 @@ (hash-put! providers name p)))))) '("openrouter" "deepseek" "google" "openai" "anthropic" "xai" "groq" "mistral" "together" "cerebras" "perplexity"))) - (catch (e) (void)))))) + (catch (e) + (log-warn logger "auth-parse-failed" + `((file . ,auth-file) (error . ,(err->string e))))))))) (def (config-ref . keys) (let loop ((obj (*config*)) (keys keys)) --- a/src/jcode/core/models.ss +++ b/src/jcode/core/models.ss @@ -18,7 +18,10 @@ (import :std/text/json :std/os/path - :std/misc/string) + :std/misc/string + :jcode/core/log) + +(def logger (make-logger "models")) ;; ---- Provider metadata ---- @@ -114,6 +117,8 @@ (*models-cache* (if (hash-table? data) data (make-hash-table))) (*models-cache*)) (catch (e) + (log-warn logger "cache-parse-failed" + `((path . ,path) (error . ,(err->string e)))) (*models-cache* (make-hash-table)) (*models-cache*))) (begin --- a/src/jcode/provider/provider.ss +++ b/src/jcode/provider/provider.ss @@ -56,13 +56,20 @@ (def (retryable-error? e) ;; Retry on transient HTTP statuses AND on streaming-layer flakiness: ;; mid-stream silence (watchdog timeout) and pre-status connection - ;; close (e.g. provider RST after handshake). Both have been observed - ;; intermittently from real providers and reliably succeed on retry. + ;; close (e.g. provider RST after handshake). 408 timeout, 429 rate + ;; limit, 5xx server errors, 520-524 Cloudflare edge, 529 overloaded. (let ((msg (with-output-to-string (lambda () (display-condition e))))) - (or (string-contains msg "429") + (or (string-contains msg "408") + (string-contains msg "429") (string-contains msg "500") (string-contains msg "502") (string-contains msg "503") + (string-contains msg "504") + (string-contains msg "520") + (string-contains msg "521") + (string-contains msg "522") + (string-contains msg "523") + (string-contains msg "524") (string-contains msg "529") (string-contains msg "connection closed before HTTP status") (string-contains msg "stream read timed out")))) @@ -315,6 +322,25 @@ (or (string->number (substring line 9 12)) 0) 0)) +;; A chunked-transfer size line is bare hex digits (optionally with ";ext"), +;; nothing else. SSE lines always contain ":" or text, never bare hex. +(def (chunk-size-line? line) + (and (string? line) + (> (string-length line) 0) + (let* ((semi (string-contains line ";")) + (hex (if semi (substring line 0 semi) line))) + (and (> (string-length hex) 0) + (let loop ((i 0)) + (cond + ((>= i (string-length hex)) #t) + ((hex-char? (string-ref hex i)) (loop (+ i 1))) + (else #f))))))) + +(def (hex-char? c) + (or (and (char>=? c #\0) (char<=? c #\9)) + (and (char>=? c #\a) (char<=? c #\f)) + (and (char>=? c #\A) (char<=? c #\F)))) + ;; Parse headers until blank line, returns alist (def (read-tls-headers conn) (let loop ((headers '())) @@ -481,14 +507,18 @@ (if (= status 0) (format "connection closed before HTTP status received (host: ~a)" host) (format "API error ~a: ~a" status body))))) - ;; Read SSE lines until EOF or chunked terminator + ;; Read SSE lines until EOF or chunked terminator. + ;; HTTP chunked transfer-encoding interleaves hex size lines + ;; (e.g. "27a") between data chunks — filter those out so they + ;; never reach the SSE parser. (let loop () (let ((line (tls-read-line conn))) (touch!) (when line - (unless (equal? line "0") ;; chunked transfer end - (line-cb line) - (loop))))) + (cond + ((equal? line "0") (void)) ;; chunked end + ((chunk-size-line? line) (loop)) + (else (line-cb line) (loop)))))) (when (vector-ref timed-out? 0) (error 'jcode-http-post-stream (format "stream read timed out after ~as of silence (host: ~a)" @@ -512,14 +542,17 @@ (if (= status 0) (format "connection closed before HTTP status received (host: ~a)" host) (format "API error ~a: ~a" status body))))) - ;; Read SSE lines until EOF or chunked terminator + ;; Read SSE lines until EOF or chunked terminator. + ;; Filter HTTP chunked transfer-encoding size lines — see + ;; above (TLS branch) for details. (let loop () (let ((c (peek-char in))) (unless (eof-object? c) (let ((line (port-read-line in))) - (unless (equal? line "0") - (line-cb line) - (loop)))))) + (cond + ((equal? line "0") (void)) + ((chunk-size-line? line) (loop)) + (else (line-cb line) (loop))))))) status)) (lambda () (close-port in) @@ -631,11 +664,18 @@ (apply-mlx-sampling! provider body) (apply-logprobs! body) (hash-put! body "messages" (map message->json messages)) - (when (and tools (not (null? tools))) + (when (and tools (not (null? tools)) + (not (model-rejects-tools? (provider-model provider)))) (hash-put! body "tools" tools) (hash-put! body "tool_choice" "auto")) body)) +;; Some models reject the `tools` parameter entirely (e.g. DeepSeek's +;; reasoner returns 400 "does not support Function Calling"). +(def (model-rejects-tools? model) + (or (equal? model "deepseek-reasoner") + (and (string? model) (string-contains model "reasoner")))) + (def (openai-parse-response json) (let* ((choices (hash-ref json "choices" '())) (choice (if (null? choices) #f (car choices))) @@ -986,7 +1026,8 @@ (hash-put! opts "include_usage" #t) (hash-put! body "stream_options" opts)) (hash-put! body "messages" (map message->json messages)) - (when (and tools (not (null? tools))) + (when (and tools (not (null? tools)) + (not (model-rejects-tools? (provider-model provider)))) (hash-put! body "tools" tools) (hash-put! body "tool_choice" "auto")) body))