diff --git a/app/controllers/messages_controller.rb b/app/controllers/messages_controller.rb index f5e98ae41..71c2107f3 100644 --- a/app/controllers/messages_controller.rb +++ b/app/controllers/messages_controller.rb @@ -17,6 +17,17 @@ class MessagesController < ApplicationController end end + # Called by the chat watchdog when an assistant "Thinking…" bubble has waited + # too long for a response that never arrived. Idempotent and scoped to the + # current user's chat. + def report_timeout + message = @chat.messages.find(params[:id]) + @chat.handle_undelivered_response!(message) + head :ok + rescue ActiveRecord::RecordNotFound + head :not_found + end + private def set_chat @chat = Current.user.chats.find(params[:chat_id]) diff --git a/app/javascript/controllers/chat_controller.js b/app/javascript/controllers/chat_controller.js index 25020f35e..dfb750238 100644 --- a/app/javascript/controllers/chat_controller.js +++ b/app/javascript/controllers/chat_controller.js @@ -1,17 +1,31 @@ import { Controller } from "@hotwired/stimulus"; export default class extends Controller { - static targets = ["messages", "form", "input", "submit"]; + static targets = ["messages", "form", "input", "submit", "pendingResponse"]; + static values = { + // How long a pending "Thinking…" bubble may wait before we assume the + // background worker never delivered a response. Generous so slow models or + // tool calls don't trip it. + responseTimeout: { type: Number, default: 90000 }, + // How often to re-check pending bubbles. + pollInterval: { type: Number, default: 5000 }, + }; connect() { + this.reportedUrls = new Set(); + this.inFlightUrls = new Set(); this.#configureAutoScroll(); this.#updateSubmitState(); + this.#startUndeliveredWatchdog(); } disconnect() { if (this.messagesObserver) { this.messagesObserver.disconnect(); } + if (this.watchdogTimer) { + clearInterval(this.watchdogTimer); + } } autoResize() { @@ -72,4 +86,66 @@ export default class extends Controller { #scrollToBottom = () => { this.messagesTarget.scrollTop = this.messagesTarget.scrollHeight; }; + + // Watchdog: a "Thinking…" bubble only resolves when the background worker + // streams a response over Turbo. If the worker is down — or the job dies + // before it can broadcast an error — the bubble would otherwise spin forever + // with no feedback. We detect a pending bubble that has waited past the + // threshold and ask the server to mark it failed, so the user gets an error + // message + Retry instead of a dead spinner. + // + // We key off the pending marker itself (it only exists while pending and + // disappears the instant a real response renders) rather than a status flag, + // so a response that starts streaming can never be falsely timed out. + #startUndeliveredWatchdog() { + this.#checkUndeliveredResponses(); + this.watchdogTimer = setInterval(() => { + this.#checkUndeliveredResponses(); + }, this.pollIntervalValue); + } + + #checkUndeliveredResponses() { + if (!this.hasPendingResponseTarget) return; + + const now = Date.now(); + + this.pendingResponseTargets.forEach((el) => { + const url = el.dataset.pendingResponseTimeoutUrl; + // Skip if already reported (succeeded) or a report is in flight. + if (!url || this.reportedUrls.has(url) || this.inFlightUrls.has(url)) return; + + const createdAt = Date.parse(el.dataset.pendingResponseCreatedAt); + if (Number.isNaN(createdAt)) return; + if (now - createdAt < this.responseTimeoutValue) return; + + this.#reportUndelivered(url); + }); + } + + #reportUndelivered(url) { + const token = document.querySelector('meta[name="csrf-token"]')?.content; + + this.inFlightUrls.add(url); + + fetch(url, { + method: "POST", + headers: { + "X-CSRF-Token": token || "", + Accept: "text/vnd.turbo-stream.html, text/html", + }, + credentials: "same-origin", + }) + .then((response) => { + // Only mark as reported on success. fetch resolves on HTTP 4xx/5xx + // (it rejects only on network errors), so without this check a failed + // POST would permanently suppress retries and strand the bubble. + if (response.ok) this.reportedUrls.add(url); + }) + .catch(() => { + // Best-effort. Leave the URL un-reported so the next tick can retry. + }) + .finally(() => { + this.inFlightUrls.delete(url); + }); + } } diff --git a/app/models/background_job_health.rb b/app/models/background_job_health.rb new file mode 100644 index 000000000..fe3b1bc48 --- /dev/null +++ b/app/models/background_job_health.rb @@ -0,0 +1,60 @@ +require "sidekiq/api" + +# Reads Sidekiq's runtime state (live worker processes + queue latency) directly +# from Redis in the *web* process, so a down or stuck worker can be detected even +# when the worker itself is the thing that's broken (a worker-side healthcheck +# can't report that it's dead). +# +# Fails open: if Sidekiq/Redis is unreachable we assume healthy rather than block +# or alarm the UI on an unrelated infrastructure blip. +class BackgroundJobHealth + # The queue chat (AssistantResponseJob), syncs and imports run on. If nothing + # is polling it, user-facing background work silently never runs. + CRITICAL_QUEUE = "high_priority".freeze + LATENCY_WARN_SECONDS = 60 + CACHE_KEY = "background_job_health".freeze + CACHE_TTL = 15.seconds + + class << self + def current + Rails.cache.fetch(CACHE_KEY, expires_in: CACHE_TTL) { compute } + rescue => e + Rails.logger.warn("BackgroundJobHealth check failed: #{e.class}: #{e.message}") + fail_open + end + + def healthy? + current[:healthy] + end + + def snapshot + current + end + + def summary + h = current + "workers=#{h[:workers].inspect} polls_#{CRITICAL_QUEUE}=#{h[:polls_critical_queue].inspect} latency=#{h[:latency].inspect}s" + end + + private + def compute + processes = Sidekiq::ProcessSet.new + workers = processes.size + polled_queues = processes.flat_map { |p| Array(p["queues"]) }.uniq + polls_critical = polled_queues.include?(CRITICAL_QUEUE) + latency = Sidekiq::Queue.new(CRITICAL_QUEUE).latency + + { + healthy: workers.positive? && polls_critical && latency < LATENCY_WARN_SECONDS, + workers: workers, + polls_critical_queue: polls_critical, + latency: latency.round(1), + checked_at: Time.current + } + end + + def fail_open + { healthy: true, workers: nil, polls_critical_queue: nil, latency: nil, checked_at: Time.current, error: true } + end + end +end diff --git a/app/models/chat.rb b/app/models/chat.rb index f8935e427..6b76f4c95 100644 --- a/app/models/chat.rb +++ b/app/models/chat.rb @@ -132,8 +132,74 @@ class Chat < ApplicationRecord assistant.respond_to(message, assistant_message: assistant_message) end + # Minimum age before the server will treat a still-pending response as + # undelivered. The browser watchdog waits longer (default 90s) before it even + # asks, but the client clock is untrusted, so the server enforces its own floor. + UNDELIVERED_RESPONSE_TIMEOUT = 60.seconds + + # Handles the case where an assistant response was never delivered — the + # background worker never ran `AssistantResponseJob` (or it died before it + # could broadcast an error), leaving a `pending` "Thinking…" bubble forever. + # Mirrors `Assistant::Builtin`'s rescue: clears the dead bubble, records a + # friendly error + a debug log entry, and broadcasts the error/Retry UI. + # + # Driven by an untrusted client watchdog, so the state change is gated behind a + # row lock + a server-side age check: we re-read the row under lock and only act + # if it is *still* pending and has genuinely waited past the timeout, so we never + # race a worker that is finishing a legitimate (slow) response. + def handle_undelivered_response!(assistant_message) + return false unless assistant_message.is_a?(AssistantMessage) + + resolved = assistant_message.with_lock do + next false unless assistant_message.pending? + next false if assistant_message.created_at > UNDELIVERED_RESPONSE_TIMEOUT.ago + + if assistant_message.content.blank? + assistant_message.destroy! + else + # Demote partially-streamed turns to `failed` so history builders exclude them. + assistant_message.update_columns(status: "failed") + end + true + end + + return false unless resolved + + capture_undelivered_response!(assistant_message) + update!(error: undelivered_error_payload(assistant_message).to_json) + broadcast_append target: messages_target, partial: "chats/error", locals: { chat: self } + true + end + private + def undelivered_error_payload(assistant_message) + { + message: I18n.t("chat.errors.no_response"), + technical_message: "Assistant response was never delivered. The background worker did not process " \ + "AssistantResponseJob for message ##{assistant_message.id} (model #{assistant_message.ai_model}). " \ + "#{BackgroundJobHealth.summary}", + type: "DeliveryTimeout" + } + end + + def capture_undelivered_response!(assistant_message) + DebugLogEntry.capture( + category: "assistant", + level: "error", + message: "Assistant response not delivered — background worker likely down or not polling the high_priority queue", + source: "chat.delivery_timeout", + metadata: { + chat_id: id, + message_id: assistant_message.id, + ai_model: assistant_message.ai_model, + waited_seconds: (Time.current - assistant_message.created_at).round, + background_jobs: BackgroundJobHealth.snapshot + }, + family: user&.family + ) + end + def build_error_payload(error) technical_message = error_message_for(error) diff --git a/app/views/assistant_messages/_assistant_message.html.erb b/app/views/assistant_messages/_assistant_message.html.erb index 6b00a3c2d..85768996c 100644 --- a/app/views/assistant_messages/_assistant_message.html.erb +++ b/app/views/assistant_messages/_assistant_message.html.erb @@ -2,7 +2,10 @@
<% if assistant_message.pending? %> -
+
<%= render "chats/ai_avatar" %>

<%= t("chats.thinking") %>

diff --git a/app/views/chats/_worker_health_warning.html.erb b/app/views/chats/_worker_health_warning.html.erb new file mode 100644 index 000000000..0d36176e8 --- /dev/null +++ b/app/views/chats/_worker_health_warning.html.erb @@ -0,0 +1,10 @@ +<%# locals: () %> +<%# Warns when the background worker that delivers AI responses is down or + backed up. Self-hosted only — managed instances run their own workers. %> +<% if Rails.application.config.app_mode.self_hosted? && Current.user.ai_enabled? && !BackgroundJobHealth.healthy? %> + <%= render DS::Alert.new( + variant: "warning", + message: t("chats.worker_unhealthy_warning"), + live: :polite + ) %> +<% end %> diff --git a/app/views/chats/show.html.erb b/app/views/chats/show.html.erb index 2b69c1f0d..743f7169b 100644 --- a/app/views/chats/show.html.erb +++ b/app/views/chats/show.html.erb @@ -8,6 +8,8 @@
<%= render "chats/chat_nav", chat: @chat %> + <%= render "chats/worker_health_warning" %> + <% if show_demo_warning? %> <%= render "shared/demo_warning", title: t("chats.demo_banner_title"), diff --git a/config/locales/models/chat/en.yml b/config/locales/models/chat/en.yml index 705495c78..d0a69e892 100644 --- a/config/locales/models/chat/en.yml +++ b/config/locales/models/chat/en.yml @@ -5,4 +5,5 @@ en: rate_limited: "The AI provider is rate limited right now. Please try again in a few minutes." temporarily_unavailable: "The AI provider is temporarily unavailable right now. Please try again in a few minutes." misconfigured: "The AI provider is not configured correctly. Please contact your administrator." + no_response: "The assistant didn't respond. The background worker may be down or AI may not be fully configured. Please try again." default: "Failed to generate a response. Please try again." diff --git a/config/locales/views/chats/en.yml b/config/locales/views/chats/en.yml index df4fe0fde..261b41d4f 100644 --- a/config/locales/views/chats/en.yml +++ b/config/locales/views/chats/en.yml @@ -4,6 +4,7 @@ en: demo_banner_title: "Demo Mode Active" demo_banner_message: "You are using LLMs via credits provided by Cloudflare Workers AI. Results may vary since the codebase was tested on `gpt-4.1` but your tokens don't go anywhere else to be trained with! 🤖" thinking: "Thinking ..." + worker_unhealthy_warning: "AI responses may not be delivered right now — the background worker appears to be down or backed up. Check that your Sidekiq worker is running." ai_greeting: greeting: "Hey %{name}! I'm an AI/large-language-model that can help with your finances. I have access to the web and your account data." there: "there" diff --git a/config/routes.rb b/config/routes.rb index af0e04bc4..3570ce392 100644 --- a/config/routes.rb +++ b/config/routes.rb @@ -185,7 +185,13 @@ Rails.application.routes.draw do # AI chats resources :chats do - resources :messages, only: :create + resources :messages, only: :create do + member do + # Client-side watchdog reports a "Thinking…" bubble that never received + # a response (e.g. the background worker is down) so it can be failed. + post :report_timeout + end + end member do post :retry diff --git a/test/controllers/messages_controller_test.rb b/test/controllers/messages_controller_test.rb index 5f6d21e56..d97fd3f96 100644 --- a/test/controllers/messages_controller_test.rb +++ b/test/controllers/messages_controller_test.rb @@ -19,4 +19,27 @@ class MessagesControllerTest < ActionDispatch::IntegrationTest assert_response :forbidden end + + test "report_timeout fails an undelivered assistant message" do + BackgroundJobHealth.stubs(:snapshot).returns({}) + BackgroundJobHealth.stubs(:summary).returns("") + + pending = @chat.messages.create!(type: "AssistantMessage", content: "", ai_model: "gpt-4.1", status: :pending, created_at: 5.minutes.ago) + + post report_timeout_chat_message_url(@chat, pending) + + assert_response :ok + assert_not Message.exists?(pending.id) + assert @chat.reload.error.present? + end + + test "report_timeout cannot touch another user's chat" do + other_chat = users(:family_member).chats.first + pending = other_chat.messages.create!(type: "AssistantMessage", content: "", ai_model: "gpt-4.1", status: :pending) + + post report_timeout_chat_message_url(other_chat, pending) + + assert_response :not_found + assert Message.exists?(pending.id) + end end diff --git a/test/models/background_job_health_test.rb b/test/models/background_job_health_test.rb new file mode 100644 index 000000000..85cc6a1ea --- /dev/null +++ b/test/models/background_job_health_test.rb @@ -0,0 +1,50 @@ +require "test_helper" + +class BackgroundJobHealthTest < ActiveSupport::TestCase + setup { Rails.cache.delete(BackgroundJobHealth::CACHE_KEY) } + teardown { Rails.cache.delete(BackgroundJobHealth::CACHE_KEY) } + + test "healthy when a worker polls the critical queue with low latency" do + stub_sidekiq(processes: [ { "queues" => [ "high_priority", "default" ] } ], latency: 1.0) + + assert BackgroundJobHealth.healthy? + assert_equal 1, BackgroundJobHealth.snapshot[:workers] + end + + test "unhealthy when no workers are online" do + stub_sidekiq(processes: [], latency: 0.0) + + assert_not BackgroundJobHealth.healthy? + end + + test "unhealthy when the critical queue is not polled by any worker" do + stub_sidekiq(processes: [ { "queues" => [ "default" ] } ], latency: 0.0) + + assert_not BackgroundJobHealth.healthy? + end + + test "unhealthy when the critical queue is badly backed up" do + stub_sidekiq(processes: [ { "queues" => [ "high_priority" ] } ], latency: 9_999) + + assert_not BackgroundJobHealth.healthy? + end + + test "fails open when Sidekiq/Redis is unreachable" do + Sidekiq::ProcessSet.stubs(:new).raises(StandardError.new("no redis")) + + assert BackgroundJobHealth.healthy? + assert BackgroundJobHealth.snapshot[:error] + end + + private + def stub_sidekiq(processes:, latency:) + process_set = mock("ProcessSet") + process_set.stubs(:size).returns(processes.size) + process_set.stubs(:flat_map).returns(processes.flat_map { |p| p["queues"] }) + Sidekiq::ProcessSet.stubs(:new).returns(process_set) + + queue = mock("Queue") + queue.stubs(:latency).returns(latency) + Sidekiq::Queue.stubs(:new).with("high_priority").returns(queue) + end +end diff --git a/test/models/chat_test.rb b/test/models/chat_test.rb index accf5e0cb..7466ba82b 100644 --- a/test/models/chat_test.rb +++ b/test/models/chat_test.rb @@ -150,4 +150,53 @@ class ChatTest < ActiveSupport::TestCase assert_equal I18n.t("chat.errors.rate_limited"), chat.presentable_error_message assert_equal "OpenAI API error 429: rate limit exceeded", chat.technical_error_message end + + test "handle_undelivered_response! clears a blank pending bubble and records an error + debug log" do + BackgroundJobHealth.stubs(:snapshot).returns({ healthy: false, workers: 0 }) + BackgroundJobHealth.stubs(:summary).returns("workers=0") + + chat = chats(:two) + pending = chat.messages.create!(type: "AssistantMessage", content: "", ai_model: "gpt-4.1", status: :pending, created_at: 5.minutes.ago) + + assert_difference -> { DebugLogEntry.count } => 1, -> { Message.count } => -1 do + assert chat.handle_undelivered_response!(pending) + end + + assert_not Message.exists?(pending.id) + assert_equal I18n.t("chat.errors.no_response"), chat.reload.presentable_error_message + end + + test "handle_undelivered_response! demotes a partially-streamed pending bubble to failed" do + BackgroundJobHealth.stubs(:snapshot).returns({}) + BackgroundJobHealth.stubs(:summary).returns("") + + chat = chats(:two) + pending = chat.messages.create!(type: "AssistantMessage", content: "partial answer", ai_model: "gpt-4.1", status: :pending, created_at: 5.minutes.ago) + + assert_no_difference -> { Message.count } do + assert chat.handle_undelivered_response!(pending) + end + + assert_equal "failed", pending.reload.status + end + + test "handle_undelivered_response! ignores a pending bubble younger than the server timeout" do + chat = chats(:two) + fresh = chat.messages.create!(type: "AssistantMessage", content: "", ai_model: "gpt-4.1", status: :pending, created_at: 5.seconds.ago) + + assert_no_difference [ "DebugLogEntry.count", "Message.count" ] do + assert_not chat.handle_undelivered_response!(fresh) + end + + assert fresh.reload.pending? + end + + test "handle_undelivered_response! is a no-op for non-pending messages" do + chat = chats(:one) + complete = chat.messages.create!(type: "AssistantMessage", content: "done", ai_model: "gpt-4.1", status: :complete) + + assert_no_difference [ "DebugLogEntry.count", "Message.count" ] do + assert_not chat.handle_undelivered_response!(complete) + end + end end