Add external AI assistant with Pipelock security proxy

.github/workflows/pipelock.yml

@@ -20,5 +20,8 @@ jobs:
         uses: luckyPipewrench/pipelock@v1
         with:
           scan-diff: 'true'
-          fail-on-findings: 'false'
+          fail-on-findings: 'true'
           test-vectors: 'false'
+          exclude-paths: |
+            config/locales/views/reports/
+            app/models/assistant/external/client.rb

app/controllers/settings/hostings_controller.rb

@@ -3,7 +3,7 @@ class Settings::HostingsController < ApplicationController
 
   guard_feature unless: -> { self_hosted? }
 
-  before_action :ensure_admin, only: [ :update, :clear_cache ]
+  before_action :ensure_admin, only: [ :update, :clear_cache, :disconnect_external_assistant ]
 
   def show
     @breadcrumbs = [
@@ -118,6 +118,23 @@ def update
       Setting.openai_json_mode = hosting_params[:openai_json_mode].presence
     end
 
+    if hosting_params.key?(:external_assistant_url)
+      Setting.external_assistant_url = hosting_params[:external_assistant_url]
+    end
+
+    if hosting_params.key?(:external_assistant_token)
+      token_param = hosting_params[:external_assistant_token].to_s.strip
+      unless token_param.blank? || token_param == "********"
+        Setting.external_assistant_token = token_param
+      end
+    end
+
+    if hosting_params.key?(:external_assistant_agent_id)
+      Setting.external_assistant_agent_id = hosting_params[:external_assistant_agent_id]
+    end
+
+    update_assistant_type
+
     redirect_to settings_hosting_path, notice: t(".success")
   rescue Setting::ValidationError => error
     flash.now[:alert] = error.message
@@ -129,9 +146,29 @@ def clear_cache
     redirect_to settings_hosting_path, notice: t(".cache_cleared")
   end
 
+  def disconnect_external_assistant
+    Setting.external_assistant_url = nil
+    Setting.external_assistant_token = nil
+    Setting.external_assistant_agent_id = nil
+    Current.family.update!(assistant_type: "builtin") unless ENV["ASSISTANT_TYPE"].present?
+    redirect_to settings_hosting_path, notice: t(".external_assistant_disconnected")
+  rescue => e
+    Rails.logger.error("[External Assistant] Disconnect failed: #{e.message}")
+    redirect_to settings_hosting_path, alert: t("settings.hostings.update.failure")
+  end
+
   private
     def hosting_params
-      params.require(:setting).permit(:onboarding_state, :require_email_confirmation, :brand_fetch_client_id, :brand_fetch_high_res_logos, :twelve_data_api_key, :openai_access_token, :openai_uri_base, :openai_model, :openai_json_mode, :exchange_rate_provider, :securities_provider, :syncs_include_pending, :auto_sync_enabled, :auto_sync_time)
+      return ActionController::Parameters.new unless params.key?(:setting)
+      params.require(:setting).permit(:onboarding_state, :require_email_confirmation, :brand_fetch_client_id, :brand_fetch_high_res_logos, :twelve_data_api_key, :openai_access_token, :openai_uri_base, :openai_model, :openai_json_mode, :exchange_rate_provider, :securities_provider, :syncs_include_pending, :auto_sync_enabled, :auto_sync_time, :external_assistant_url, :external_assistant_token, :external_assistant_agent_id)
+    end
+
+    def update_assistant_type
+      return unless params[:family].present? && params[:family][:assistant_type].present?
+      return if ENV["ASSISTANT_TYPE"].present?
+
+      assistant_type = params[:family][:assistant_type]
+      Current.family.update!(assistant_type: assistant_type) if Family::ASSISTANT_TYPES.include?(assistant_type)
     end
 
     def ensure_admin

app/models/assistant.rb

@@ -36,7 +36,7 @@ def function_classes
 
       def implementation_for(chat)
         raise Error, "chat is required" if chat.blank?
-        type = chat.user&.family&.assistant_type.presence || "builtin"
+        type = ENV["ASSISTANT_TYPE"].presence || chat.user&.family&.assistant_type.presence || "builtin"
         REGISTRY.fetch(type) { REGISTRY["builtin"] }
       end
   end

app/models/assistant/external.rb

@@ -1,14 +1,106 @@
 class Assistant::External < Assistant::Base
+  Config = Struct.new(:url, :token, :agent_id, :session_key, keyword_init: true)
+
   class << self
     def for_chat(chat)
       new(chat)
     end
+
+    def configured?
+      config.url.present? && config.token.present?
+    end
+
+    def available_for?(user)
+      configured? && allowed_user?(user)
+    end
+
+    def allowed_user?(user)
+      allowed = ENV["EXTERNAL_ASSISTANT_ALLOWED_EMAILS"]
+      return true if allowed.blank?
+      allowed.split(",").map(&:strip).include?(user.email)
+    end
+
+    def config
+      Config.new(
+        url: ENV["EXTERNAL_ASSISTANT_URL"].presence || Setting.external_assistant_url,
+        token: ENV["EXTERNAL_ASSISTANT_TOKEN"].presence || Setting.external_assistant_token,
+        agent_id: ENV["EXTERNAL_ASSISTANT_AGENT_ID"].presence || Setting.external_assistant_agent_id.presence || "main",
+        session_key: ENV.fetch("EXTERNAL_ASSISTANT_SESSION_KEY", "agent:main:main")
+      )
+    end
   end
 
   def respond_to(message)
-    stop_thinking
-    chat.add_error(
-      StandardError.new("External assistant (OpenClaw/WebSocket) is not yet implemented.")
+    unless self.class.configured?
+      raise Assistant::Error,
+        "External assistant is not configured. Set the URL and token in Settings > Self-Hosting or via environment variables."
+    end
+
+    unless self.class.allowed_user?(chat.user)
+      raise Assistant::Error, "Your account is not authorized to use the external assistant."
+    end
+
+    assistant_message = AssistantMessage.new(
+      chat: chat,
+      content: "",
+      ai_model: "external-agent"
     )
+
+    client = build_client
+    messages = build_conversation_messages
+
+    model = client.chat(
+      messages: messages,
+      user: "sure-family-#{chat.user.family_id}"
+    ) do |text|
+      if assistant_message.content.blank?
+        stop_thinking
+        assistant_message.content = text
+        assistant_message.save!
+      else
+        assistant_message.append_text!(text)
+      end
+    end
+
+    if assistant_message.new_record?
+      stop_thinking
+      raise Assistant::Error, "External assistant returned an empty response."
+    end
+
+    assistant_message.update!(ai_model: model) if model.present?
+  rescue Assistant::Error, ActiveRecord::ActiveRecordError => e
+    cleanup_partial_response(assistant_message)
+    stop_thinking
+    chat.add_error(e)
+  rescue => e
+    Rails.logger.error("[Assistant::External] Unexpected error: #{e.class} - #{e.message}")
+    cleanup_partial_response(assistant_message)
+    stop_thinking
+    chat.add_error(Assistant::Error.new("Something went wrong with the external assistant. Check server logs for details."))
   end
+
+  private
+
+    def cleanup_partial_response(assistant_message)
+      assistant_message&.destroy! if assistant_message&.persisted?
+    rescue ActiveRecord::ActiveRecordError => e
+      Rails.logger.warn("[Assistant::External] Failed to clean up partial response: #{e.message}")
+    end
+
+    def build_client
+      Assistant::External::Client.new(
+        url: self.class.config.url,
+        token: self.class.config.token,
+        agent_id: self.class.config.agent_id,
+        session_key: self.class.config.session_key
+      )
+    end
+
+    MAX_CONVERSATION_MESSAGES = 20
+
+    def build_conversation_messages
+      chat.conversation_messages.ordered.last(MAX_CONVERSATION_MESSAGES).map do |msg|
+        { role: msg.role, content: msg.content }
+      end
+    end
 end

app/models/assistant/external/client.rb

@@ -0,0 +1,156 @@
+require "net/http"
+require "uri"
+require "json"
+
+class Assistant::External::Client
+  TIMEOUT_CONNECT = 10   # seconds
+  TIMEOUT_READ    = 120  # seconds (agent may take time to reason + call tools)
+  MAX_RETRIES     = 2
+  RETRY_DELAY     = 1    # seconds (doubles each retry)
+
+  TRANSIENT_ERRORS = [
+    Net::OpenTimeout,
+    Net::ReadTimeout,
+    Errno::ECONNREFUSED,
+    Errno::ECONNRESET,
+    Errno::EHOSTUNREACH,
+    SocketError
+  ].freeze
+
+  def initialize(url:, token:, agent_id: "main", session_key: "agent:main:main")
+    @url = url
+    @token = token
+    @agent_id = agent_id
+    @session_key = session_key
+  end
+
+  # Streams text chunks from an OpenAI-compatible chat endpoint via SSE.
+  #
+  # messages - Array of {role:, content:} hashes (conversation history)
+  # user     - Optional user identifier for session persistence
+  # block    - Called with each text chunk as it arrives
+  #
+  # Returns the model identifier string from the response.
+  def chat(messages:, user: nil, &block)
+    uri = URI(@url)
+    request = build_request(uri, messages, user)
+    retries = 0
+    streaming_started = false
+
+    begin
+      http = build_http(uri)
+      model = stream_response(http, request) do |content|
+        streaming_started = true
+        block.call(content)
+      end
+      model
+    rescue *TRANSIENT_ERRORS => e
+      # Don't retry once streaming has started — caller already has partial data
+      raise Assistant::Error, "External assistant connection lost: #{e.message}" if streaming_started
+
+      retries += 1
+      if retries <= MAX_RETRIES
+        Rails.logger.warn("[External::Client] Transient error (attempt #{retries}/#{MAX_RETRIES}): #{e.class} - #{e.message}")
+        sleep(RETRY_DELAY * retries)
+        retry
+      end
+      raise Assistant::Error, "External assistant unreachable after #{MAX_RETRIES + 1} attempts: #{e.message}"
+    end
+  end
+
+  private
+
+    def stream_response(http, request, &block)
+      model = nil
+      buffer = ""
+
+      http.request(request) do |response|
+        unless response.is_a?(Net::HTTPSuccess)
+          raise Assistant::Error, "External assistant returned HTTP #{response.code}: #{response.body}"
+        end
+
+        response.read_body do |chunk|
+          buffer += chunk
+
+          while (line_end = buffer.index("\n"))
+            line = buffer.slice!(0..line_end).strip
+            next if line.empty?
+            next unless line.start_with?("data:")
+
+            data = line.delete_prefix("data:")
+            data = data.delete_prefix(" ") # SSE spec: strip one optional leading space
+            break if data == "[DONE]"
+
+            parsed = parse_sse_data(data)
+            next unless parsed
+
+            model ||= parsed["model"]
+            content = parsed.dig("choices", 0, "delta", "content")
+            block.call(content) if content.present?
+          end
+        end
+      end
+
+      model
+    end
+
+    def build_http(uri)
+      proxy_uri = resolve_proxy(uri)
+
+      if proxy_uri
+        http = Net::HTTP.new(uri.host, uri.port, proxy_uri.host, proxy_uri.port)
+      else
+        http = Net::HTTP.new(uri.host, uri.port)
+      end
+
+      http.use_ssl = (uri.scheme == "https")
+      http.open_timeout = TIMEOUT_CONNECT
+      http.read_timeout = TIMEOUT_READ
+      http
+    end
+
+    def resolve_proxy(uri)
+      proxy_env = (uri.scheme == "https") ? "HTTPS_PROXY" : "HTTP_PROXY"
+      proxy_url = ENV[proxy_env] || ENV[proxy_env.downcase]
+      return nil if proxy_url.blank?
+
+      no_proxy = ENV["NO_PROXY"] || ENV["no_proxy"]
+      return nil if host_bypasses_proxy?(uri.host, no_proxy)
+
+      URI(proxy_url)
+    rescue URI::InvalidURIError => e
+      Rails.logger.warn("[External::Client] Invalid proxy URL ignored: #{e.message}")
+      nil
+    end
+
+    def host_bypasses_proxy?(host, no_proxy)
+      return false if no_proxy.blank?
+      no_proxy.split(",").any? { |pattern| host.end_with?(pattern.strip) }
+    end
+
+    def build_request(uri, messages, user)
+      request = Net::HTTP::Post.new(uri.request_uri)
+      request["Content-Type"] = "application/json"
+      request["Authorization"] = "Bearer #{@token}"
+      request["Accept"] = "text/event-stream"
+      request["X-Agent-Id"] = @agent_id
+      request["X-Session-Key"] = @session_key
+
+      payload = {
+        model: @agent_id,
+        messages: messages,
+        stream: true
+      }
+      payload[:user] = user if user.present?
+
+      request.body = payload.to_json
+      request
+    end
+
+    def parse_sse_data(data)
+      JSON.parse(data)
+    rescue JSON::ParserError => e
+      Rails.logger.warn("[External::Client] Unparseable SSE data: #{e.message}")
+      nil
+    end
+end

app/models/setting.rb

@@ -10,6 +10,9 @@ class ValidationError < StandardError; end
   field :openai_uri_base, type: :string, default: ENV["OPENAI_URI_BASE"]
   field :openai_model, type: :string, default: ENV["OPENAI_MODEL"]
   field :openai_json_mode, type: :string, default: ENV["LLM_JSON_MODE"]
+  field :external_assistant_url, type: :string, default: ENV["EXTERNAL_ASSISTANT_URL"]
+  field :external_assistant_token, type: :string, default: ENV["EXTERNAL_ASSISTANT_TOKEN"]
+  field :external_assistant_agent_id, type: :string, default: ENV.fetch("EXTERNAL_ASSISTANT_AGENT_ID", "main")
   field :brand_fetch_client_id, type: :string, default: ENV["BRAND_FETCH_CLIENT_ID"]
   field :brand_fetch_high_res_logos, type: :boolean, default: ENV.fetch("BRAND_FETCH_HIGH_RES_LOGOS", "false") == "true"
 

app/models/user.rb

@@ -136,7 +136,16 @@ def show_ai_sidebar?
   end
 
   def ai_available?
-    !Rails.application.config.app_mode.self_hosted? || ENV["OPENAI_ACCESS_TOKEN"].present? || Setting.openai_access_token.present?
+    return true unless Rails.application.config.app_mode.self_hosted?
+
+    effective_type = ENV["ASSISTANT_TYPE"].presence || family&.assistant_type.presence || "builtin"
+
+    case effective_type
+    when "external"
+      Assistant::External.available_for?(self)
+    else
+      ENV["OPENAI_ACCESS_TOKEN"].present? || Setting.openai_access_token.present?
+    end
   end
 
   def ai_enabled?