Fix stuck TUI busy state on stalled streams
ober
a7c3f8c815946b93af36fa6ec64ab3b8f0f61768
--- a/src/jcode/provider/provider.ss +++ b/src/jcode/provider/provider.ss @@ -29,6 +29,7 @@ :std/net/tls-rustls :std/net/tcp :std/misc/string + :std/misc/thread :std/misc/retry :jcode/core/config :jcode/core/log @@ -525,9 +526,11 @@ ;; Streaming read timeout: if no bytes arrive within this many seconds a ;; watchdog thread closes the connection so the read loop unblocks and we -;; raise a clear error instead of hanging indefinitely. Override at runtime -;; via (parameterize ((*stream-read-timeout-secs* N)) ...). -(def *stream-read-timeout-secs* (make-parameter 180)) +;; raise a clear error instead of hanging indefinitely. Local model servers +;; usually emit keepalive/progress bytes during long prefill; silence for this +;; long is treated as a dead stream. Override at runtime via +;; (parameterize ((*stream-read-timeout-secs* N)) ...). +(def *stream-read-timeout-secs* (make-parameter 45)) ;; Streaming HTTP POST: calls line-cb with each line of the response body. ;; Used for SSE (Server-Sent Events) streaming from LLM APIs. @@ -639,39 +642,86 @@ (lambda () (vector-set! done? 0 #t) (close-once!)))) - ;; Plain HTTP via tcp + ;; Plain HTTP via tcp -- same silence watchdog as the TLS branch. + ;; Local providers (mlx/ollama) can leave a streaming socket open with + ;; no further bytes; without a timeout the worker stays blocked in + ;; port-read-line/peek-char indefinitely. (let-values (((in out) (tcp-connect host port))) - (dynamic-wind - (lambda () (void)) - (lambda () - (port-write-string out req) - (let* ((status-line (port-read-line in)) - (status (parse-http-status status-line)) - (_headers (port-read-headers in))) - (unless (= status 200) - (let ((body (read-http-body-lines - (lambda () - (let ((c (peek-char in))) - (if (eof-object? c) #f (port-read-line in))))))) + (let ((last-activity (vector (time-second (current-time)))) + (timed-out? (vector #f)) + (done? (vector #f)) + (closed? (vector #f))) + (def (touch!) + (vector-set! last-activity 0 (time-second (current-time)))) + (def (close-once!) + (unless (vector-ref closed? 0) + (vector-set! closed? 0 #t) + (guard (e [(i/o-error? e) (void)]) (close-port in)) + (guard (e [(i/o-error? e) (void)]) (close-port out)))) + (fork-thread + (lambda () + (guard (e [#t (when (tracing?) + (log-trace logger "watchdog-died" + `((err . ,(err->string e)))))]) + (let loop () + (thread-sleep! 5) + (cond + ((vector-ref done? 0) (void)) + ((>= (- (time-second (current-time)) + (vector-ref last-activity 0)) + timeout-secs) + (vector-set! timed-out? 0 #t) + (when (tracing?) + (log-trace logger "stream-timeout" + `((url . ,(redact-url url)) + (idle-secs . ,(- (time-second (current-time)) + (vector-ref last-activity 0)))))) + (close-once!)) + (else (loop))))))) + (dynamic-wind + (lambda () (void)) + (lambda () + (port-write-string out req) + (touch!) + (let* ((status-line (port-read-line in)) + (status (parse-http-status status-line)) + (_headers (port-read-headers in))) + (touch!) + (unless (= status 200) + (let ((body (read-http-body-lines + (lambda () + (let ((c (peek-char in))) + (touch!) + (if (eof-object? c) + #f + (let ((line (port-read-line in))) + (touch!) + line))))))) + (error 'jcode-http-post-stream + (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. + ;; Filter HTTP chunked transfer-encoding size lines — see + ;; above (TLS branch) for details. + (let loop () + (let ((c (peek-char in))) + (touch!) + (unless (eof-object? c) + (let ((line (port-read-line in))) + (touch!) + (cond + ((equal? line "0") (void)) + ((chunk-size-line? line) (loop)) + (else (line-cb line) (loop))))))) + (when (vector-ref timed-out? 0) (error 'jcode-http-post-stream - (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. - ;; 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))) - (cond - ((equal? line "0") (void)) - ((chunk-size-line? line) (loop)) - (else (line-cb line) (loop))))))) - status)) - (lambda () - (close-port in) - (close-port out)))))))) + (format "stream read timed out after ~as of silence (host: ~a)" + timeout-secs host))) + status)) + (lambda () + (vector-set! done? 0 #t) + (close-once!))))))))) ;;; OpenAI-compatible API ;;; @@ -791,7 +841,7 @@ (else (loop (cdr markers) best best-idx)))))))) (def (parse-text-tool-call body) - (guard (e [#t #f]) + (guard (e [(error? e) #f]) (let* ((trimmed (string-trim body)) (json (string->json-object trimmed))) (and (hash-table? json) @@ -1340,7 +1390,7 @@ (cond ((equal? data "[DONE]") (void)) (else - (let ((json (guard (e [#t #f]) + (let ((json (guard (e [(error? e) #f]) (string->json-object data)))) (when (and json (hash-table? json)) ;; Capture usage from final chunk @@ -1507,7 +1557,7 @@ (set! data-str (substring line 6 (string-length line)))))) lines) (when (and event-type data-str) - (let ((json (guard (e [#t #f]) (string->json-object data-str)))) + (let ((json (guard (e [(error? e) #f]) (string->json-object data-str)))) (when (and json (hash-table? json)) (cond ;; Text delta @@ -1817,7 +1867,7 @@ (set! event-type (substring line 7 (string-length line)))) ((string-prefix? "data: " line) (let* ((data-str (substring line 6 (string-length line))) - (json (guard (e [#t #f]) (string->json-object data-str)))) + (json (guard (e [(error? e) #f]) (string->json-object data-str)))) (when (and json (hash-table? json)) (let ((etype (or (and (string? (hash-get json "type")) (hash-get json "type")) --- a/src/jcode/ui/tui.ss +++ b/src/jcode/ui/tui.ss @@ -330,6 +330,10 @@ (tui-log "event-loop: drained ~a agent events" n) (app-state-dirty?-set! state #t))) + ;; If the visible tab's worker died without a terminal event making it + ;; through the mailbox, do not leave the UI permanently "thinking". + (cleanup-stale-agent-run! state) + ;; Activity screen refreshes every tick: ages advance and background ;; runs keep logging even when this tab's agent is idle. (when (eq? (app-state-view state) 'activity) @@ -491,6 +495,7 @@ (set-car! *tui-stream-abort* #t) (bump-tui-run-gen!) (abort-agent-run! (app-state-session-id state)) + (clear-agent-run! (app-state-session-id state) #f) (app-state-agent-busy?-set! state #f) (add-message! state (msg-block-system "(interrupted)")) (app-state-dirty?-set! state #t)) @@ -1110,18 +1115,64 @@ ;; run flips its abort cell: the worker notices on its next callback, ;; and the tag check drops anything already in flight. -(def *agent-runs* (make-hash-table)) ;; sid -> (gen . abort-cell) +(def *agent-runs* (make-hash-table)) ;; sid -> #(gen abort-cell worker done-tick) + +(def (make-agent-run gen abort worker) + (vector gen abort worker #f)) + +(def (agent-run-gen r) (vector-ref r 0)) +(def (agent-run-abort r) (vector-ref r 1)) +(def (agent-run-worker r) (vector-ref r 2)) +(def (agent-run-done-tick r) (vector-ref r 3)) +(def (agent-run-worker-set! r worker) (vector-set! r 2 worker)) +(def (agent-run-done-tick-set! r tick) (vector-set! r 3 tick)) + +(def (register-agent-run! sid gen abort (worker #f)) + (hash-put! *agent-runs* sid (make-agent-run gen abort worker))) -(def (register-agent-run! sid gen abort) - (hash-put! *agent-runs* sid (cons gen abort))) +(def (agent-run-worker-set-for! sid gen worker) + (let ((r (hash-get *agent-runs* sid))) + (when (and r (= (agent-run-gen r) gen)) + (agent-run-worker-set! r worker)))) + +(def (clear-agent-run! sid gen) + (let ((r (hash-get *agent-runs* sid))) + (when (and r (or (not gen) (= (agent-run-gen r) gen))) + (hash-remove! *agent-runs* sid)))) (def (abort-agent-run! sid) (let ((r (hash-get *agent-runs* sid))) - (when r (set-car! (cdr r) #t)))) + (when r (set-car! (agent-run-abort r) #t)))) (def (agent-run-valid? sid gen) (let ((r (hash-get *agent-runs* sid))) - (and r (= (car r) gen) (not (car (cdr r)))))) + (and r (= (agent-run-gen r) gen) (not (car (agent-run-abort r)))))) + +(def (terminal-agent-event? ev) + (and (pair? ev) + (memq (car ev) '(agent-error agent-done agent-cancelled)))) + +(def (cleanup-stale-agent-run! state) + (when (app-state-agent-busy? state) + (let* ((sid (app-state-session-id state)) + (r (and sid (hash-get *agent-runs* sid))) + (worker (and r (agent-run-worker r)))) + (when (and worker (thread-done? worker)) + (let ((done-tick (agent-run-done-tick r))) + (cond + (done-tick + (when (>= (- (app-state-tick state) done-tick) 3) + (tui-log "cleanup-stale-agent-run: sid=~a gen=~a worker exited without terminal event" + sid (agent-run-gen r)) + (clear-agent-run! sid (agent-run-gen r)) + (app-state-agent-busy?-set! state #f) + (app-state-active-tools-set! state '()) + (app-state-stream-buf-set! state "") + (add-message! state + (msg-block-error "Agent worker exited without a final event.")) + (app-state-dirty?-set! state #t))) + (else + (agent-run-done-tick-set! r (app-state-tick state))))))))) (def (send-run-event! sid gen ev) (send-agent-event! (cons (cons sid gen) ev))) @@ -1201,9 +1252,13 @@ ((not (agent-run-valid? sid gen)) (tui-log "apply-agent-event: discarded dead run sid=~a gen=~a" sid gen)) ((equal? sid (app-state-session-id state)) - (apply-agent-event-body! state body)) + (apply-agent-event-body! state body) + (when (terminal-agent-event? body) + (clear-agent-run! sid gen))) (else - (apply-background-event! state sid body))))) + (apply-background-event! state sid body) + (when (terminal-agent-event? body) + (clear-agent-run! sid gen)))))) (else (void)))) (def (find-tab-by-session state sid) @@ -1409,49 +1464,51 @@ (err-port (current-error-port)) (log-lvl (current-log-level))) (register-agent-run! s-id gen abort) - (spawn - (lambda () - (tui-log "worker: entered gen=~a sid=~a main-thread=~a" gen s-id (*main-thread*)) - (try - (parameterize - ((current-error-port err-port) - (current-log-level log-lvl) - (current-provider-override p-override) - (current-model-override m-override) - (current-stream-cb - (lambda (token) - (when (car abort) - (error 'stream-aborted "interrupted by user")) - (send-run-event! s-id gen (list 'stream-token token)))) - (current-tool-cb - (lambda (event name args) - (when (car abort) - (error 'stream-aborted "interrupted by user")) - (tui-log "worker: tool-cb ~a ~a" event name) - (send-run-event! s-id gen (list 'tool-event event name args)))) - (current-usage-cb - (lambda (usage) - (send-run-event! s-id gen (list 'usage-update usage)))) - (current-expert-cb - (lambda (prov model reason sentinel?) - (when (car abort) - (error 'stream-aborted "interrupted by user")) - (send-run-event! s-id gen - (list 'escalation prov model reason sentinel?))))) - (tui-log "worker: calling agent-run") - (agent-run s-id text) - (tui-log "worker: agent-run returned, sending agent-done") - (send-run-event! s-id gen - (list 'agent-done (- (time-second (current-time)) t0))) - (tui-log "worker: agent-done sent, worker exiting normally")) - (catch (e) - (let ((msg (err->string e))) - (tui-log "worker: CAUGHT exception: ~a" msg) - (cond - ((string-contains msg "stream-aborted") - (send-run-event! s-id gen (list 'agent-cancelled))) - (else - (send-run-event! s-id gen (list 'agent-error msg)))))))))))) + (let ((worker + (spawn + (lambda () + (tui-log "worker: entered gen=~a sid=~a main-thread=~a" gen s-id (*main-thread*)) + (try + (parameterize + ((current-error-port err-port) + (current-log-level log-lvl) + (current-provider-override p-override) + (current-model-override m-override) + (current-stream-cb + (lambda (token) + (when (car abort) + (error 'stream-aborted "interrupted by user")) + (send-run-event! s-id gen (list 'stream-token token)))) + (current-tool-cb + (lambda (event name args) + (when (car abort) + (error 'stream-aborted "interrupted by user")) + (tui-log "worker: tool-cb ~a ~a" event name) + (send-run-event! s-id gen (list 'tool-event event name args)))) + (current-usage-cb + (lambda (usage) + (send-run-event! s-id gen (list 'usage-update usage)))) + (current-expert-cb + (lambda (prov model reason sentinel?) + (when (car abort) + (error 'stream-aborted "interrupted by user")) + (send-run-event! s-id gen + (list 'escalation prov model reason sentinel?))))) + (tui-log "worker: calling agent-run") + (agent-run s-id text) + (tui-log "worker: agent-run returned, sending agent-done") + (send-run-event! s-id gen + (list 'agent-done (- (time-second (current-time)) t0))) + (tui-log "worker: agent-done sent, worker exiting normally")) + (catch (e) + (let ((msg (err->string e))) + (tui-log "worker: CAUGHT exception: ~a" msg) + (cond + ((string-contains msg "stream-aborted") + (send-run-event! s-id gen (list 'agent-cancelled))) + (else + (send-run-event! s-id gen (list 'agent-error msg))))))))))) + (agent-run-worker-set-for! s-id gen worker))))) ;; ---- /ask-* second-opinion runners ---- @@ -2625,7 +2682,8 @@ (tab-cache-read t) (tab-cache-creation t) (+ (tab-cost t) cost) - (tab-tool-counts t)))) + (tab-tool-counts t) + #t))) ;; unread? (app-state-tabs-set! state (list-replace tabs idx updated)) (app-state-dirty?-set! state #t)))))) @@ -2656,6 +2714,7 @@ (tab-cache-read t) (tab-cache-creation t) (tab-cost t) - (tab-tool-counts t)))) + (tab-tool-counts t) + #t))) ;; unread? (app-state-tabs-set! state (list-replace tabs idx updated)) (app-state-dirty?-set! state #t)))))))