Abort forced streams without semantic progress
ober
48434b8113b3f901f9f872ccb741c0626dd13ce7
--- a/src/jcode/provider/provider.ss +++ b/src/jcode/provider/provider.ss @@ -1024,11 +1024,16 @@ "JCODE_INTERNAL_FORCE_FIRST_EDIT_RETRY") (def *forced-tool-no-call-default-char-limit* 2048) +(def *forced-tool-semantic-idle-default-secs* 60) (def (forced-tool-no-call-char-limit) (or (positive-int-value (getenv "JCODE_FORCED_TOOL_NO_CALL_CHAR_LIMIT")) *forced-tool-no-call-default-char-limit*)) +(def (forced-tool-semantic-idle-secs) + (or (positive-int-value (getenv "JCODE_FORCED_TOOL_SEMANTIC_IDLE_SECS")) + *forced-tool-semantic-idle-default-secs*)) + (def (forced-tool-choice-retry-content expected detail) (string-append forced-tool-choice-retry-marker @@ -2427,7 +2432,29 @@ (openai-tool-choice-name (current-tool-choice-override))) (forced-tool-mismatch-box (box #f)) (forced-tool-pre-call-chars-box (box 0)) - (forced-tool-pre-call-char-limit (forced-tool-no-call-char-limit))) + (forced-tool-pre-call-char-limit (forced-tool-no-call-char-limit)) + (forced-tool-semantic-idle-limit (forced-tool-semantic-idle-secs)) + (forced-tool-last-progress-box (box (time-second (current-time))))) + (def (forced-tool-progress!) + (set-box! forced-tool-last-progress-box (time-second (current-time)))) + (def (maybe-abort-forced-tool-semantic-idle!) + (when (and forced-tool-name + (not (unbox saw-done-box)) + (not (unbox finish-reason-box)) + (null? (hash-keys tc-table))) + (let ((idle (- (time-second (current-time)) + (unbox forced-tool-last-progress-box)))) + (when (>= idle forced-tool-semantic-idle-limit) + (set-box! forced-tool-mismatch-box + (cons forced-tool-name + (format "provider stream made no semantic progress for ~as" + idle))) + (log-warn logger "forced-tool-stream-semantic-idle" + `((expected . ,forced-tool-name) + (idle-secs . ,idle) + (limit . ,forced-tool-semantic-idle-limit))) + (error 'openai-stream-chat + "provider stream made no semantic progress during forced tool turn"))))) (let* ((body-json (json-object->string body)) (dummy (begin (log-info logger "stream-request" @@ -2443,126 +2470,136 @@ 200 (raise e))]) (jcode-http-post-stream url headers body-json - (lambda (event-str) - (when event-str - (log-debug logger "sse-event" `((data . ,(if (> (string-length event-str) 120) - (substring event-str 0 120) - event-str)))) - (when (tracing?) - (log-trace logger "openai-sse-event" `((data . ,event-str)))) - ;; Strip "data: " prefix - (let* ((data (if (string-prefix? "data: " event-str) - (substring event-str 6 (string-length event-str)) - event-str))) - (cond - ((equal? data "[DONE]") (set-box! saw-done-box #t)) - (else - (let ((json (guard (e [(error? e) #f]) - (string->json-object data)))) - (when (and json (hash-table? json)) - ;; Capture usage from final chunk - (let ((usage (hash-get json "usage"))) - (when (and usage (hash-table? usage)) - (hash-for-each (lambda (k v) (hash-put! usage-acc k v)) usage))) - (let* ((choices (hash-get json "choices")) - (choice (and (pair? choices) (car choices))) - (delta (and choice (hash-get choice "delta")))) - ;; Capture finish_reason whenever a choice carries one - (when choice - (let ((fr (hash-get choice "finish_reason"))) - (when (and fr (string? fr) (not (string=? fr ""))) - (set-box! finish-reason-box fr))) - ;; Accumulate per-token logprobs/entropy when requested. - ;; logprobs sit at the choice level (sibling of delta), - ;; so this runs on `choice`, not gated on `delta`. - (accumulate-logprobs! choice logprob-box entropy-box)) - (when delta - ;; Text content token - (let ((content (hash-get delta "content"))) - (when (and content (string? content) (> (string-length content) 0)) - (put-string text-acc content) - (when (and forced-tool-name - (null? (hash-keys tc-table))) - (set-box! forced-tool-pre-call-chars-box - (+ (unbox forced-tool-pre-call-chars-box) - (string-length content))) - (when (> (unbox forced-tool-pre-call-chars-box) - forced-tool-pre-call-char-limit) - (set-box! forced-tool-mismatch-box - (cons forced-tool-name - (format "provider streamed ~a text/reasoning characters without calling a tool" - (unbox forced-tool-pre-call-chars-box)))) - (log-warn logger "forced-tool-stream-no-call" - `((expected . ,forced-tool-name) - (chars . ,(unbox forced-tool-pre-call-chars-box)) - (limit . ,forced-tool-pre-call-char-limit))) - (error 'openai-stream-chat - "provider streamed too much text during forced tool turn"))) - (token-cb content))) - ;; Reasoning token (mlx-omni `reasoning`, R1 - ;; `reasoning_content`). Stream to UI so the user - ;; sees thinking, and accumulate so an otherwise - ;; empty turn still records something downstream. - (let ((r (or (hash-get delta "reasoning") - (hash-get delta "reasoning_content")))) - (when (and r (string? r) (> (string-length r) 0)) - (put-string reasoning-acc r) - (when (and forced-tool-name - (null? (hash-keys tc-table))) - (set-box! forced-tool-pre-call-chars-box - (+ (unbox forced-tool-pre-call-chars-box) - (string-length r))) - (when (> (unbox forced-tool-pre-call-chars-box) - forced-tool-pre-call-char-limit) - (set-box! forced-tool-mismatch-box - (cons forced-tool-name - (format "provider streamed ~a text/reasoning characters without calling a tool" - (unbox forced-tool-pre-call-chars-box)))) - (log-warn logger "forced-tool-stream-no-call" - `((expected . ,forced-tool-name) - (chars . ,(unbox forced-tool-pre-call-chars-box)) - (limit . ,forced-tool-pre-call-char-limit))) - (error 'openai-stream-chat - "provider streamed too much reasoning during forced tool turn"))) - (token-cb r))) - ;; Tool call fragments - (let ((tcs (hash-get delta "tool_calls"))) - (when (and tcs (list? tcs)) - (for-each - (lambda (tc) - (let* ((idx (or (hash-get tc "index") 0)) - (acc (or (hash-get tc-table idx) - (let ((a (make-hash-table))) - (hash-put! tc-table idx a) - a))) - (id (hash-get tc "id")) - (fn (hash-get tc "function"))) - (when id (hash-put! acc "id" id)) - (when fn - (let ((name (hash-get fn "name")) - (args (hash-get fn "arguments"))) - (when name - (when (and forced-tool-name - (string? name) - (not (forced-tool-name-accepted? - forced-tool-name name))) - (set-box! forced-tool-mismatch-box - (cons forced-tool-name - (string-append - "provider streamed hidden tool " - name))) - (log-warn logger "forced-tool-stream-mismatch" - `((expected . ,forced-tool-name) - (actual . ,name))) - (error 'openai-stream-chat - "provider streamed hidden tool during forced tool turn")) - (hash-put! acc "name" name)) - (when args - (hash-put! acc "args" - (string-append - (or (hash-get acc "args") "") - args))))))) - tcs)))))))))))) + (lambda (event-str) + (when event-str + (log-debug logger "sse-event" `((data . ,(if (> (string-length event-str) 120) + (substring event-str 0 120) + event-str)))) + (when (tracing?) + (log-trace logger "openai-sse-event" `((data . ,event-str)))) + ;; Strip "data: " prefix. SSE comments such as ": prefill" + ;; intentionally fall through as non-JSON events below. + (let* ((data (if (string-prefix? "data: " event-str) + (substring event-str 6 (string-length event-str)) + event-str))) + (cond + ((equal? data "[DONE]") + (forced-tool-progress!) + (set-box! saw-done-box #t)) + (else + (let ((json (guard (e [(error? e) #f]) + (string->json-object data)))) + (when (and json (hash-table? json)) + ;; Capture usage from final chunk + (let ((usage (hash-get json "usage"))) + (when (and usage (hash-table? usage)) + (hash-for-each + (lambda (k v) (hash-put! usage-acc k v)) + usage))) + (let* ((choices (hash-get json "choices")) + (choice (and (pair? choices) (car choices))) + (delta (and choice (hash-get choice "delta")))) + ;; Capture finish_reason whenever a choice carries one. + (when choice + (let ((fr (hash-get choice "finish_reason"))) + (when (and fr (string? fr) (not (string=? fr ""))) + (forced-tool-progress!) + (set-box! finish-reason-box fr))) + ;; Accumulate per-token logprobs/entropy when requested. + ;; logprobs sit at the choice level (sibling of delta), + ;; so this runs on `choice`, not gated on `delta`. + (accumulate-logprobs! choice logprob-box entropy-box)) + (when delta + ;; Text content token. + (let ((content (hash-get delta "content"))) + (when (and content (string? content) (> (string-length content) 0)) + (forced-tool-progress!) + (put-string text-acc content) + (when (and forced-tool-name + (null? (hash-keys tc-table))) + (set-box! forced-tool-pre-call-chars-box + (+ (unbox forced-tool-pre-call-chars-box) + (string-length content))) + (when (> (unbox forced-tool-pre-call-chars-box) + forced-tool-pre-call-char-limit) + (set-box! forced-tool-mismatch-box + (cons forced-tool-name + (format "provider streamed ~a text/reasoning characters without calling a tool" + (unbox forced-tool-pre-call-chars-box)))) + (log-warn logger "forced-tool-stream-no-call" + `((expected . ,forced-tool-name) + (chars . ,(unbox forced-tool-pre-call-chars-box)) + (limit . ,forced-tool-pre-call-char-limit))) + (error 'openai-stream-chat + "provider streamed too much text during forced tool turn"))) + (token-cb content))) + ;; Reasoning token (mlx-omni `reasoning`, R1 + ;; `reasoning_content`). Stream to UI so the user + ;; sees thinking, and accumulate so an otherwise + ;; empty turn still records something downstream. + (let ((r (or (hash-get delta "reasoning") + (hash-get delta "reasoning_content")))) + (when (and r (string? r) (> (string-length r) 0)) + (forced-tool-progress!) + (put-string reasoning-acc r) + (when (and forced-tool-name + (null? (hash-keys tc-table))) + (set-box! forced-tool-pre-call-chars-box + (+ (unbox forced-tool-pre-call-chars-box) + (string-length r))) + (when (> (unbox forced-tool-pre-call-chars-box) + forced-tool-pre-call-char-limit) + (set-box! forced-tool-mismatch-box + (cons forced-tool-name + (format "provider streamed ~a text/reasoning characters without calling a tool" + (unbox forced-tool-pre-call-chars-box)))) + (log-warn logger "forced-tool-stream-no-call" + `((expected . ,forced-tool-name) + (chars . ,(unbox forced-tool-pre-call-chars-box)) + (limit . ,forced-tool-pre-call-char-limit))) + (error 'openai-stream-chat + "provider streamed too much reasoning during forced tool turn"))) + (token-cb r))) + ;; Tool call fragments. + (let ((tcs (hash-get delta "tool_calls"))) + (when (and tcs (list? tcs)) + (forced-tool-progress!) + (for-each + (lambda (tc) + (let* ((idx (or (hash-get tc "index") 0)) + (acc (or (hash-get tc-table idx) + (let ((a (make-hash-table))) + (hash-put! tc-table idx a) + a))) + (id (hash-get tc "id")) + (fn (hash-get tc "function"))) + (when id (hash-put! acc "id" id)) + (when fn + (let ((name (hash-get fn "name")) + (args (hash-get fn "arguments"))) + (when name + (when (and forced-tool-name + (string? name) + (not (forced-tool-name-accepted? + forced-tool-name name))) + (set-box! forced-tool-mismatch-box + (cons forced-tool-name + (string-append + "provider streamed hidden tool " + name))) + (log-warn logger "forced-tool-stream-mismatch" + `((expected . ,forced-tool-name) + (actual . ,name))) + (error 'openai-stream-chat + "provider streamed hidden tool during forced tool turn")) + (hash-put! acc "name" name)) + (when args + (hash-put! acc "args" + (string-append + (or (hash-get acc "args") "") + args))))))) + tcs)))))))))) + (maybe-abort-forced-tool-semantic-idle!))) (lambda (status response-body) (raise-provider-api-error (provider-name provider) status response-body)))))) --- a/test/run.ss +++ b/test/run.ss @@ -259,6 +259,29 @@ (close-port out) (close-port in))))))) +(define (serve-one-captured-sse-body-after-delay! srv captured-body body delay-secs) + (fork-thread + (lambda () + (let-values ([(in out) (tcp-accept srv)]) + (dynamic-wind + (lambda () (void)) + (lambda () + (vector-set! captured-body 0 (read-test-http-request in)) + (put-string out + (string-append + "HTTP/1.1 200 OK\r\n" + "Content-Type: text/event-stream\r\n" + "Content-Length: " (number->string (string-length body)) "\r\n" + "Connection: close\r\n" + "\r\n")) + (flush-output-port out) + (thread-sleep! delay-secs) + (put-string out body) + (flush-output-port out)) + (lambda () + (close-port out) + (close-port in))))))) + (define (serve-one-captured-json! srv captured-body status body) (fork-thread (lambda () @@ -13289,6 +13312,51 @@ (putenv "JCODE_FORCED_TOOL_NO_CALL_CHAR_LIMIT" (or old-limit "")) (tcp-close srv)))) +(let* ([sse-body ": prefill\n\n"] + [srv (tcp-listen "127.0.0.1" 0)] + [base-url (format "http://127.0.0.1:~a/v1" (tcp-server-port srv))] + [captured (vector #f)] + [edit-tool (uuid-recovery-tool "edit" (args "path" (args "type" "string")) '("path"))] + [old-max-tokens (getenv "JCODE_MAX_TOKENS")] + [old-idle (getenv "JCODE_FORCED_TOOL_SEMANTIC_IDLE_SECS")]) + (dynamic-wind + (lambda () + (putenv "JCODE_MAX_TOKENS" "") + (putenv "JCODE_FORCED_TOOL_SEMANTIC_IDLE_SECS" "1")) + (lambda () + (serve-one-captured-sse-body-after-delay! srv captured sse-body 2) + (let* ([p (make-provider "mlx" "" "deepseek-v4-pro" base-url)] + [result (call-with-values + (lambda () + (parameterize ([current-tool-choice-override + (openai-function-tool-choice "edit")]) + (provider-stream-chat + p + (list (make-user-message "hi")) + (list edit-tool) + (lambda (_token) #f)))) + (lambda (content tcs usage) (list content tcs usage)))] + [req (vector-ref captured 0)]) + (check-pred! "mlx forced streaming prefill idle returns retry marker" + (car result) + (lambda (s) + (and (string? s) + (str-contains? s + "JCODE_INTERNAL_FORCE_FIRST_EDIT_RETRY") + (str-contains? s + "no semantic progress") + (str-contains? s "required edit")))) + (check! "mlx forced streaming prefill idle has no tool call" + (list-ref result 1) '()) + (check! "mlx forced streaming prefill idle request required edit" + (and req + (str-contains? req "\"tool_choice\"") + (str-contains? req "\"name\":\"edit\"")) #t))) + (lambda () + (putenv "JCODE_MAX_TOKENS" (or old-max-tokens "")) + (putenv "JCODE_FORCED_TOOL_SEMANTIC_IDLE_SECS" (or old-idle "")) + (tcp-close srv)))) + (let* ([sse-body "data: {\"choices\":[{\"index\":0,\"delta\":{\"reasoning_content\":\"partial\"},\"finish_reason\":null}]}\n\n"] [srv (tcp-listen "127.0.0.1" 0)]