diff --git a/backend/app/controllers/admin/stream_nodes_controller.rb b/backend/app/controllers/admin/stream_nodes_controller.rb
index 46e0283..c69d52a 100644
--- a/backend/app/controllers/admin/stream_nodes_controller.rb
+++ b/backend/app/controllers/admin/stream_nodes_controller.rb
@@ -44,6 +44,16 @@ module Admin
       redirect_to admin_stream_nodes_path, alert: e.message
     end
 
+    def kill_switch
+      Streams::Autoscaler.engage_kill_switch!
+      redirect_to admin_stream_nodes_path, notice: t("admin.flash.autoscale_kill_on")
+    end
+
+    def clear_kill_switch
+      Streams::Autoscaler.clear_kill_switch!
+      redirect_to admin_stream_nodes_path, notice: t("admin.flash.autoscale_kill_off")
+    end
+
     private
 
     def lab_dns_snippet
diff --git a/backend/app/models/ops/incident.rb b/backend/app/models/ops/incident.rb
index de88ec5..44f8a51 100644
--- a/backend/app/models/ops/incident.rb
+++ b/backend/app/models/ops/incident.rb
@@ -4,7 +4,7 @@ module Ops
 
     KINDS = %w[
       disk_space recordings_size service_down http_public http_rails rails_latency
-      sidekiq_stale sidekiq_dead log_pattern garage_storage
+      sidekiq_stale sidekiq_dead log_pattern garage_storage stream_overflow
     ].freeze
     SEVERITIES = %w[critical warning info].freeze
     STATUSES = %w[open acknowledged resolved].freeze
diff --git a/backend/app/services/ops/health_checks.rb b/backend/app/services/ops/health_checks.rb
index 3c8c552..8194b01 100644
--- a/backend/app/services/ops/health_checks.rb
+++ b/backend/app/services/ops/health_checks.rb
@@ -44,7 +44,8 @@ module Ops
         check_sidekiq_heartbeat,
         check_sidekiq_dead,
         check_http_rails,
-        check_rails_latency
+        check_rails_latency,
+        check_stream_overflow
       ]
       findings << check_http_public if public_check_due?
       findings
@@ -161,6 +162,58 @@ module Ops
       fail_finding("garage_storage", "warning", "garage_storage:head", "Garage storage non raggiungibile", e.message)
     end
 
+    def check_stream_overflow
+      return ok_finding("stream_overflow:skip", "Nodi stream non migrati") unless ActiveRecord::Base.connection.data_source_exists?("stream_nodes")
+
+      orphan_hours = ENV.fetch("STREAM_OVERFLOW_ORPHAN_HOURS", "3").to_i
+      orphans = StreamNode.where.not(slug: "home").where(status: %w[ready draining]).select do |n|
+        n.active_publishers.zero? && n.created_at < orphan_hours.hours.ago
+      end
+      metrics = Streams::Autoscaler.metrics
+      over_budget = !metrics[:within_budget]
+      at_max = metrics[:overflow_nodes] >= metrics[:max_overflow_nodes] && metrics[:free_slots] <= metrics[:soft_free_slots]
+
+      if orphans.any?
+        return Finding.new(
+          kind: "stream_overflow",
+          severity: "warning",
+          healthy: false,
+          title: "Nodi stream overflow idle",
+          message: "#{orphans.size} nodo/i idle da >#{orphan_hours}h: #{orphans.map(&:slug).join(', ')}",
+          metadata: { "slugs" => orphans.map(&:slug) },
+          fingerprint: "stream_overflow:orphan_idle"
+        )
+      end
+
+      if over_budget
+        return Finding.new(
+          kind: "stream_overflow",
+          severity: "warning",
+          healthy: false,
+          title: "Budget overflow streaming",
+          message: "Stima €#{metrics[:estimated_monthly_eur]}/mese > budget €#{metrics[:monthly_budget_eur]}",
+          metadata: metrics.transform_keys(&:to_s),
+          fingerprint: "stream_overflow:budget"
+        )
+      end
+
+      if at_max
+        return Finding.new(
+          kind: "stream_overflow",
+          severity: "warning",
+          healthy: false,
+          title: "Capacità stream al massimo",
+          message: "overflow=#{metrics[:overflow_nodes]}/#{metrics[:max_overflow_nodes]} free_slots=#{metrics[:free_slots]}",
+          metadata: metrics.transform_keys(&:to_s),
+          fingerprint: "stream_overflow:at_max"
+        )
+      end
+
+      ok_finding("stream_overflow:ok", "Overflow streaming OK")
+    rescue StandardError => e
+      fail_finding("stream_overflow", "warning", "stream_overflow:error", "Check overflow fallito", e.message)
+    end
+
     def check_sidekiq_heartbeat
       redis = Redis.new(url: ENV.fetch("REDIS_URL", "redis://localhost:6379/0"))
       last = redis.get(Ops::HealthMonitorJob::HEARTBEAT_KEY).to_i
diff --git a/backend/app/services/streams/autoscaler.rb b/backend/app/services/streams/autoscaler.rb
index 07f6d62..9076db8 100644
--- a/backend/app/services/streams/autoscaler.rb
+++ b/backend/app/services/streams/autoscaler.rb
@@ -2,15 +2,31 @@
 
 module Streams
   # Scale-out / warm spare / scale-in dei nodi overflow (lab o Hetzner).
-  # Kill-switch: STREAM_AUTOSCALE_ENABLED!=1 → no-op.
+  # Kill-switch: STREAM_AUTOSCALE_ENABLED!=1 OPPURE Redis streams:autoscaler:kill_switch=1.
   class Autoscaler
     Result = Struct.new(:actions, :metrics, :skipped, :error, keyword_init: true)
 
     LOCK_KEY = "streams:autoscaler:lock"
+    KILL_SWITCH_KEY = "streams:autoscaler:kill_switch"
 
     class << self
       def enabled?
-        ENV["STREAM_AUTOSCALE_ENABLED"] == "1"
+        return false if kill_switch_engaged?
+        return false unless ENV["STREAM_AUTOSCALE_ENABLED"] == "1"
+
+        true
+      end
+
+      def kill_switch_engaged?
+        redis_get(KILL_SWITCH_KEY) == "1"
+      end
+
+      def engage_kill_switch!
+        redis_set(KILL_SWITCH_KEY, "1")
+      end
+
+      def clear_kill_switch!
+        redis_del(KILL_SWITCH_KEY)
       end
 
       def soft_free_slots
@@ -33,6 +49,28 @@ module Streams
         ENV.fetch("STREAM_AUTOSCALE_KIND", "lab") # lab|cloud
       end
 
+      def allow_cloud?
+        ENV["STREAM_AUTOSCALE_ALLOW_CLOUD"] == "1" && ENV["HCLOUD_TOKEN"].present?
+      end
+
+      def node_eur_per_hour
+        ENV.fetch("STREAM_AUTOSCALE_NODE_EUR_PER_HOUR", "0.015").to_f
+      end
+
+      def monthly_budget_eur
+        ENV.fetch("STREAM_AUTOSCALE_MONTHLY_BUDGET_EUR", "40").to_f
+      end
+
+      def estimated_monthly_eur(overflow_count = nil)
+        count = overflow_count || metrics[:overflow_nodes]
+        # Worst case: nodi sempre accesi 24/7
+        (count * node_eur_per_hour * 24 * 30).round(2)
+      end
+
+      def within_budget?(overflow_count = nil)
+        estimated_monthly_eur(overflow_count) <= monthly_budget_eur
+      end
+
       def reconcile!(provisioner: nil)
         return Result.new(skipped: true, actions: [], metrics: metrics) unless enabled?
 
@@ -62,13 +100,33 @@ module Streams
           warm_spare_min: warm_spare_min,
           max_overflow_nodes: max_overflow_nodes,
           enabled: enabled?,
-          kind: kind
+          env_enabled: ENV["STREAM_AUTOSCALE_ENABLED"] == "1",
+          kill_switch: kill_switch_engaged?,
+          kind: kind,
+          allow_cloud: allow_cloud?,
+          estimated_monthly_eur: estimated_monthly_eur(overflow.size),
+          monthly_budget_eur: monthly_budget_eur,
+          within_budget: within_budget?(overflow.size)
         }
       end
 
       def worker_id
         ENV.fetch("HOSTNAME", "autoscaler")
       end
+
+      def redis_get(key)
+        Redis.new(url: ENV.fetch("REDIS_URL", "redis://localhost:6379/0")).get(key)
+      rescue Redis::BaseError
+        nil
+      end
+
+      def redis_set(key, value)
+        Redis.new(url: ENV.fetch("REDIS_URL", "redis://localhost:6379/0")).set(key, value)
+      end
+
+      def redis_del(key)
+        Redis.new(url: ENV.fetch("REDIS_URL", "redis://localhost:6379/0")).del(key)
+      end
     end
 
     def initialize(provisioner: nil)
@@ -83,6 +141,9 @@ module Streams
         provision_overflow!
         actions << :scale_out
         m = self.class.metrics
+      elsif need_capacity?(m) && !can_provision?(m)
+        actions << :blocked_capacity
+        Rails.logger.warn("[Streams::Autoscaler] capacity needed but blocked metrics=#{m.inspect}")
       end
 
       if warm_spare_desired?(m) && m[:spare_ready] < self.class.warm_spare_min && can_provision?(m)
@@ -94,7 +155,7 @@ module Streams
       scale_in_candidates.each do |node|
         next if keep_as_warm_spare?(node)
 
-        @provisioner.decommission!(node)
+        safe_scale_in!(node)
         actions << :"scale_in_#{node.slug}"
         m = self.class.metrics
       rescue NodeProvisioner::BusyError, NodeProvisioner::Error => e
@@ -122,16 +183,33 @@ module Streams
     end
 
     def can_provision?(m)
-      m[:overflow_nodes] < self.class.max_overflow_nodes
+      return false if m[:overflow_nodes] >= self.class.max_overflow_nodes
+      return false unless self.class.within_budget?(m[:overflow_nodes] + 1)
+      return false if self.class.kind == "cloud" && !self.class.allow_cloud?
+
+      true
     end
 
     def provision_overflow!
       case self.class.kind
-      when "cloud" then @provisioner.provision_cloud!
-      else @provisioner.provision_lab!
+      when "cloud"
+        raise NodeProvisioner::Error, "Cloud autoscale disabilitato (STREAM_AUTOSCALE_ALLOW_CLOUD / HCLOUD_TOKEN)" unless self.class.allow_cloud?
+
+        @provisioner.provision_cloud!
+      else
+        @provisioner.provision_lab!
       end
     end
 
+    def safe_scale_in!(node)
+      @provisioner.drain!(node) unless node.status == "draining"
+      node.reload
+      raise NodeProvisioner::BusyError, "sessioni ancora attive" if node.occupying_sessions.exists?
+      raise NodeProvisioner::Error, "idle insufficiente" unless idle_long_enough?(node)
+
+      @provisioner.decommission!(node)
+    end
+
     def scale_in_candidates
       StreamNode.where.not(slug: NodeRegistry::HOME_SLUG)
         .where(status: %w[ready draining])
diff --git a/backend/app/views/admin/stream_nodes/index.html.erb b/backend/app/views/admin/stream_nodes/index.html.erb
index d45b759..10e8277 100644
--- a/backend/app/views/admin/stream_nodes/index.html.erb
+++ b/backend/app/views/admin/stream_nodes/index.html.erb
@@ -16,6 +16,15 @@
         max: @autoscale_metrics[:max_overflow_nodes],
         kind: @autoscale_metrics[:kind]
       ) %>
+  · <%= t(
+        "admin.stream_nodes.autoscale_budget",
+        eur: @autoscale_metrics[:estimated_monthly_eur],
+        budget: @autoscale_metrics[:monthly_budget_eur],
+        ok: (@autoscale_metrics[:within_budget] ? "OK" : "OVER")
+      ) %>
+  <% if @autoscale_metrics[:kill_switch] %>
+    · <strong><%= t("admin.stream_nodes.kill_switch_active") %></strong>
+  <% end %>
 </p>
 
 <p>
@@ -26,6 +35,12 @@
   <% else %>
     <span class="admin-muted"><%= t("admin.stream_nodes.hetzner_token_missing") %></span>
   <% end %>
+  <% if @autoscale_metrics[:kill_switch] %>
+    <%= button_to t("admin.stream_nodes.clear_kill_switch"), clear_kill_switch_admin_stream_nodes_path, method: :delete, class: "admin-btn admin-btn-secondary" %>
+  <% else %>
+    <%= button_to t("admin.stream_nodes.engage_kill_switch"), kill_switch_admin_stream_nodes_path, method: :post, class: "admin-btn admin-btn-danger",
+          form: { data: { confirm: t("admin.stream_nodes.kill_switch_confirm") } } %>
+  <% end %>
 </p>
 
 <table class="admin-table">
diff --git a/backend/config/locales/admin.de.yml b/backend/config/locales/admin.de.yml
index 8346fcb..1144f6b 100644
--- a/backend/config/locales/admin.de.yml
+++ b/backend/config/locales/admin.de.yml
@@ -30,6 +30,8 @@ de:
       stream_node_created: "Lab-Knoten %{slug} bereitgestellt."
       stream_node_destroyed: "Knoten %{slug} entfernt."
       stream_node_draining: "Knoten %{slug} im Drain-Modus."
+      autoscale_kill_on: "Autoscaler-Kill-Switch aktiv. Kein automatisches Scale-out."
+      autoscale_kill_off: "Autoscaler-Kill-Switch aus (STREAM_AUTOSCALE_ENABLED=1 weiterhin nötig)."
       comped_granted: "Kostenloses Abonnement %{plan} für %{club} aktiviert."
       comped_revoked: "Kostenloses Abonnement für %{club} widerrufen."
       session_already_terminated: "Sitzung bereits beendet (%{status})."
@@ -118,6 +120,12 @@ de:
     stream_nodes:
       title: Stream-Knoten
       providers: "Cloud-Provider: %{cloud} · DNS-Provider: %{dns}"
+      autoscale: "Autoscaler %{enabled} · free_slots=%{free} (soft≤%{soft}) · spare=%{spare}/%{warm} · overflow=%{overflow}/%{max} · kind=%{kind}"
+      autoscale_budget: "Budget €%{eur}/€%{budget} (%{ok})"
+      kill_switch_active: "KILL-SWITCH AKTIV"
+      engage_kill_switch: "Kill-switch ON"
+      clear_kill_switch: "Kill-switch OFF"
+      kill_switch_confirm: "Autoscaler sofort blockieren?"
       provision_lab: Lab-Knoten bereitstellen
       drain: Drain
       destroy: Löschen
diff --git a/backend/config/locales/admin.en.yml b/backend/config/locales/admin.en.yml
index ded5346..5aa7676 100644
--- a/backend/config/locales/admin.en.yml
+++ b/backend/config/locales/admin.en.yml
@@ -30,6 +30,8 @@ en:
       stream_node_created: "Lab node %{slug} provisioned."
       stream_node_destroyed: "Node %{slug} removed."
       stream_node_draining: "Node %{slug} is draining (no new sessions)."
+      autoscale_kill_on: "Autoscaler kill-switch engaged. No automatic scale-out."
+      autoscale_kill_off: "Autoscaler kill-switch cleared (still needs STREAM_AUTOSCALE_ENABLED=1)."
       comped_granted: "%{plan} complimentary subscription activated for %{club}."
       comped_revoked: "Complimentary subscription revoked for %{club}."
       session_already_terminated: "Session already ended (%{status})."
@@ -119,6 +121,11 @@ en:
       title: Streaming nodes
       providers: "Cloud provider: %{cloud} · DNS provider: %{dns}"
       autoscale: "Autoscaler %{enabled} · free_slots=%{free} (soft≤%{soft}) · spare=%{spare}/%{warm} · overflow=%{overflow}/%{max} · kind=%{kind}"
+      autoscale_budget: "budget €%{eur}/€%{budget} (%{ok})"
+      kill_switch_active: "KILL-SWITCH ACTIVE"
+      engage_kill_switch: "Kill-switch ON (block autoscaler)"
+      clear_kill_switch: "Kill-switch OFF"
+      kill_switch_confirm: "Immediately block the autoscaler? Existing nodes stay up."
       provision_lab: Provision lab node
       provision_cloud: Provision Hetzner node
       provision_cloud_confirm: "Create a Hetzner Cloud server + DNS record on mltv-stream.net? Billing applies until destroyed."
diff --git a/backend/config/locales/admin.es.yml b/backend/config/locales/admin.es.yml
index 667d3ba..7d622e2 100644
--- a/backend/config/locales/admin.es.yml
+++ b/backend/config/locales/admin.es.yml
@@ -30,6 +30,8 @@ es:
       stream_node_created: "Nodo lab %{slug} provisionado."
       stream_node_destroyed: "Nodo %{slug} eliminado."
       stream_node_draining: "Nodo %{slug} en drain."
+      autoscale_kill_on: "Kill-switch del autoscaler activado. Sin scale-out automático."
+      autoscale_kill_off: "Kill-switch del autoscaler desactivado (hace falta STREAM_AUTOSCALE_ENABLED=1)."
       comped_granted: "Suscripción de cortesía %{plan} activada para %{club}."
       comped_revoked: "Suscripción de cortesía revocada para %{club}."
       session_already_terminated: "La sesión ya ha finalizado (%{status})."
@@ -118,6 +120,12 @@ es:
     stream_nodes:
       title: Nodos streaming
       providers: "Cloud provider: %{cloud} · DNS provider: %{dns}"
+      autoscale: "Autoscaler %{enabled} · free_slots=%{free} (soft≤%{soft}) · spare=%{spare}/%{warm} · overflow=%{overflow}/%{max} · kind=%{kind}"
+      autoscale_budget: "presupuesto €%{eur}/€%{budget} (%{ok})"
+      kill_switch_active: "KILL-SWITCH ACTIVO"
+      engage_kill_switch: "Kill-switch ON"
+      clear_kill_switch: "Kill-switch OFF"
+      kill_switch_confirm: "¿Bloquear el autoscaler inmediatamente?"
       provision_lab: Provisionar nodo lab
       drain: Drain
       destroy: Eliminar
diff --git a/backend/config/locales/admin.fr.yml b/backend/config/locales/admin.fr.yml
index 0c8ab75..19ea9cd 100644
--- a/backend/config/locales/admin.fr.yml
+++ b/backend/config/locales/admin.fr.yml
@@ -30,6 +30,8 @@ fr:
       stream_node_created: "Nœud lab %{slug} provisionné."
       stream_node_destroyed: "Nœud %{slug} supprimé."
       stream_node_draining: "Nœud %{slug} en drain."
+      autoscale_kill_on: "Kill-switch autoscaler activé. Pas de scale-out automatique."
+      autoscale_kill_off: "Kill-switch autoscaler désactivé (nécessite aussi STREAM_AUTOSCALE_ENABLED=1)."
       comped_granted: "Abonnement offert %{plan} activé pour %{club}."
       comped_revoked: "Abonnement offert révoqué pour %{club}."
       session_already_terminated: "Session déjà terminée (%{status})."
@@ -118,6 +120,12 @@ fr:
     stream_nodes:
       title: Nœuds streaming
       providers: "Cloud provider: %{cloud} · DNS provider: %{dns}"
+      autoscale: "Autoscaler %{enabled} · free_slots=%{free} (soft≤%{soft}) · spare=%{spare}/%{warm} · overflow=%{overflow}/%{max} · kind=%{kind}"
+      autoscale_budget: "budget €%{eur}/€%{budget} (%{ok})"
+      kill_switch_active: "KILL-SWITCH ACTIF"
+      engage_kill_switch: "Kill-switch ON"
+      clear_kill_switch: "Kill-switch OFF"
+      kill_switch_confirm: "Bloquer immédiatement l'autoscaler ?"
       provision_lab: Provisionner un nœud lab
       drain: Drain
       destroy: Supprimer
diff --git a/backend/config/locales/admin.it.yml b/backend/config/locales/admin.it.yml
index 1b03118..98cab44 100644
--- a/backend/config/locales/admin.it.yml
+++ b/backend/config/locales/admin.it.yml
@@ -30,6 +30,8 @@ it:
       stream_node_created: "Nodo lab %{slug} provisionato."
       stream_node_destroyed: "Nodo %{slug} rimosso."
       stream_node_draining: "Nodo %{slug} in drain (niente nuove sessioni)."
+      autoscale_kill_on: "Kill-switch autoscaler attivato. Nessun scale-out automatico."
+      autoscale_kill_off: "Kill-switch autoscaler disattivato (serve comunque STREAM_AUTOSCALE_ENABLED=1)."
       comped_granted: "Abbonamento omaggio %{plan} attivato per %{club}."
       comped_revoked: "Abbonamento omaggio revocato per %{club}."
       session_already_terminated: "Sessione già terminata (%{status})."
@@ -119,6 +121,11 @@ it:
       title: Nodi streaming
       providers: "Cloud provider: %{cloud} · DNS provider: %{dns}"
       autoscale: "Autoscaler %{enabled} · free_slots=%{free} (soft≤%{soft}) · spare=%{spare}/%{warm} · overflow=%{overflow}/%{max} · kind=%{kind}"
+      autoscale_budget: "budget €%{eur}/€%{budget} (%{ok})"
+      kill_switch_active: "KILL-SWITCH ATTIVO"
+      engage_kill_switch: "Kill-switch ON (blocca autoscaler)"
+      clear_kill_switch: "Kill-switch OFF"
+      kill_switch_confirm: "Bloccare immediatamente l'autoscaler? I nodi esistenti restano accesi."
       provision_lab: Provisiona nodo lab
       provision_cloud: Provisiona nodo Hetzner
       provision_cloud_confirm: "Creare un server Hetzner Cloud + record DNS su mltv-stream.net? Verrà addebitato fino allo spegnimento."
diff --git a/backend/config/routes.rb b/backend/config/routes.rb
index 2d09cc3..181bce7 100644
--- a/backend/config/routes.rb
+++ b/backend/config/routes.rb
@@ -114,6 +114,10 @@ Rails.application.routes.draw do
       member do
         post :drain
       end
+      collection do
+        post :kill_switch
+        delete :clear_kill_switch
+      end
     end
     get "youtube/platform", to: "youtube#platform", as: :youtube_platform
   end
diff --git a/backend/spec/services/ops/health_checks_spec.rb b/backend/spec/services/ops/health_checks_spec.rb
index e4f7f62..cbd7471 100644
--- a/backend/spec/services/ops/health_checks_spec.rb
+++ b/backend/spec/services/ops/health_checks_spec.rb
@@ -25,6 +25,7 @@ RSpec.describe Ops::HealthChecks do
       %i[
         check_recordings_size check_postgres check_redis check_mediamtx check_garage
         check_sidekiq_heartbeat check_sidekiq_dead check_http_rails check_rails_latency
+        check_stream_overflow
       ].each do |method|
         allow_any_instance_of(described_class).to receive(method).and_return(
           described_class::Finding.new(
@@ -52,6 +53,7 @@ RSpec.describe Ops::HealthChecks do
       %i[
         check_recordings_size check_postgres check_redis check_mediamtx check_garage
         check_sidekiq_heartbeat check_sidekiq_dead check_http_rails check_rails_latency
+        check_stream_overflow
       ].each do |method|
         allow_any_instance_of(described_class).to receive(method).and_return(
           described_class::Finding.new(
@@ -65,7 +67,7 @@ RSpec.describe Ops::HealthChecks do
       summary = described_class.new.summary
 
       expect(summary[:status]).to eq("ok")
-      expect(summary[:checks].size).to eq(10)
+      expect(summary[:checks].size).to eq(11)
     end
 
     it "degraded quando la latenza p95 supera la soglia warning" do
diff --git a/backend/spec/services/streams/autoscaler_spec.rb b/backend/spec/services/streams/autoscaler_spec.rb
index c795652..551e858 100644
--- a/backend/spec/services/streams/autoscaler_spec.rb
+++ b/backend/spec/services/streams/autoscaler_spec.rb
@@ -15,6 +15,7 @@ RSpec.describe Streams::Autoscaler do
 
   before do
     redis.del(Streams::Autoscaler::LOCK_KEY)
+    redis.del(Streams::Autoscaler::KILL_SWITCH_KEY)
     redis.del(Streams::DnsProviders::Lab::REDIS_KEY)
   end
 
@@ -116,10 +117,55 @@ RSpec.describe Streams::Autoscaler do
       )
 
       provisioner = instance_double(Streams::NodeProvisioner)
+      allow(provisioner).to receive(:drain!)
       expect(provisioner).to receive(:decommission!).with(idle)
 
       result = described_class.reconcile!(provisioner: provisioner)
       expect(result.actions.map(&:to_s)).to include("scale_in_ingest-lab-99")
     end
   end
+
+  it "skips when Redis kill-switch is engaged even if ENV is enabled" do
+    redis.del(described_class::KILL_SWITCH_KEY)
+    with_env("STREAM_AUTOSCALE_ENABLED" => "1") do
+      described_class.engage_kill_switch!
+      expect(described_class.enabled?).to eq(false)
+      result = described_class.reconcile!
+      expect(result.skipped).to eq(true)
+    ensure
+      described_class.clear_kill_switch!
+    end
+  end
+
+  it "blocks scale-out when next node would exceed monthly budget" do
+    with_env(
+      "STREAM_AUTOSCALE_ENABLED" => "1",
+      "STREAM_AUTOSCALE_SOFT_FREE_SLOTS" => "2",
+      "STREAM_AUTOSCALE_WARM_SPARE" => "0",
+      "STREAM_AUTOSCALE_KIND" => "lab",
+      "STREAM_AUTOSCALE_MAX_NODES" => "5",
+      "STREAM_AUTOSCALE_NODE_EUR_PER_HOUR" => "1",
+      "STREAM_AUTOSCALE_MONTHLY_BUDGET_EUR" => "10",
+      "STREAM_CLOUD_PROVIDER" => "local_lab",
+      "STREAM_DNS_PROVIDER" => "lab",
+      "STREAM_NODE_HOME_MAX_PUBLISHERS" => "1",
+      "MEDIAMTX_API_URL" => "http://mtx-home:9997",
+      "MEDIAMTX_RTMP_URL" => "rtmp://home.example:1935",
+      "HLS_PUBLIC_URL" => "https://home.example/hls"
+    ) do
+      home = Streams::NodeRegistry.ensure_home_from_env!
+      user = User.create!(email: "budget@example.com", name: "B", password: "Password123", role: "coach")
+      club = Club.create!(name: "Budget", sport: "volleyball")
+      team = club.teams.create!(name: "T", sport: "volleyball", slug: "budget-t")
+      match = team.matches.create!(opponent_name: "X", scheduled_at: 1.hour.from_now)
+      StreamSession.create!(match: match, user: user, platform: "matchlivetv", status: "live", stream_node: home)
+
+      provisioner = instance_double(Streams::NodeProvisioner)
+      expect(provisioner).not_to receive(:provision_lab!)
+
+      result = described_class.reconcile!(provisioner: provisioner)
+      expect(result.actions).to include(:blocked_capacity)
+      expect(described_class.metrics[:within_budget]).to eq(true)
+    end
+  end
 end
diff --git a/docs/infrastructure/STREAMING_AUTOSCALE.md b/docs/infrastructure/STREAMING_AUTOSCALE.md
index fe14507..3dcfedf 100644
--- a/docs/infrastructure/STREAMING_AUTOSCALE.md
+++ b/docs/infrastructure/STREAMING_AUTOSCALE.md
@@ -1,8 +1,8 @@
 # Autoscale streaming — Proxmox attuale + Hetzner Cloud
 
-**Stato:** design approvato (decisioni chiuse) — pronto per implementazione  
+**Stato:** fasi 0–4 implementate sul branch — **solo test lab; nessun deploy produzione**  
 **Branch:** `feature/streaming-autoscale-hetzner`  
-**Ultimo aggiornamento:** 2026-08-09 (dominio ingest: `mltv-stream.net`)  
+**Ultimo aggiornamento:** 2026-08-09 (Fase 4 hardening + runbook)  
 
 **Sostituisce / estende:** il piano overflow-only [`HETZNER_CLOUD_RELAY_OVERFLOW.md`](HETZNER_CLOUD_RELAY_OVERFLOW.md). Questo documento copre **MediaMTX + ffmpeg**, warm spare, DNS, storage, disco e lab sul Proxmox di produzione.
 
@@ -211,7 +211,7 @@ Cosa il lab **non** replica al 100%: tempi boot Hetzner, vSwitch, RTMP 4G multi-
 | **1 — Hetzner Cloud + DNS** | `HetznerCloudProvider` + `HetznerDnsProvider`, `mltv-stream.net`, cloud-init, WireGuard (§16) | **implementata sul branch** (WG ops manuale) |
 | **2 — Routing relay** | Coda `youtube_relay`, sticky owner, cap `RELAY_MAX_CONCURRENT` | **implementata sul branch** |
 | **3 — Autoscaler** | Soglie + warm spare (+1), kill-switch `STREAM_AUTOSCALE_ENABLED` | **implementata sul branch** |
-| **4 — Hardening** | Drain, budget, runbook, kill-switch | Picchi weekend |
+| **4 — Hardening** | Drain sicuro, budget, kill-switch Redis/admin, alert overflow, runbook | **implementata sul branch** (solo test lab — **non in prod**) |
 | **A — Auction (futuro)** | Migrazione control plane + vSwitch al posto di WireGuard | Hardware dedicato Hetzner |
 | **B2 (opz.)** | Switch object storage | Quando misurato |
 
@@ -260,6 +260,28 @@ Cosa il lab **non** replica al 100%: tempi boot Hetzner, vSwitch, RTMP 4G multi-
 - Cap `STREAM_AUTOSCALE_MAX_NODES`; kind `lab|cloud`
 - Metriche in admin Nodi stream
 
+### Fase 4 — hardening (implementata sul branch)
+
+> **NON rilasciare in produzione** finché lab + smoke Cloud non sono verdi. Su questo branch restano `STREAM_AUTOSCALE_ENABLED=0` e `STREAM_AUTOSCALE_ALLOW_CLOUD=0`.
+
+- Scale-in **sicuro**: `drain!` → attesa sessioni zero → `decommission!`
+- Budget soft: `STREAM_AUTOSCALE_MONTHLY_BUDGET_EUR` + `STREAM_AUTOSCALE_NODE_EUR_PER_HOUR` (stima 24/7); scale-out bloccato se sforerebbe
+- Cloud autoscale solo con `STREAM_AUTOSCALE_ALLOW_CLOUD=1` **e** `HCLOUD_TOKEN`
+- Kill-switch runtime Redis `streams:autoscaler:kill_switch` (bottoni admin ON/OFF) oltre all’ENV
+- Health check `stream_overflow` → incidente Ops/ntfy su nodi idle orfani, over-budget, capacità al max
+- Admin: metriche budget + stato kill-switch
+
+#### Runbook operativo (lab / pre-prod)
+
+| Situazione | Azione |
+|------------|--------|
+| Dubbio / picco anomalo | Admin → **Kill-switch ON** (o `STREAM_AUTOSCALE_ENABLED=0`) |
+| Nodo spillato | Drain dal admin; dopo fine live → Destroy |
+| Budget alert Ops | Abbassa `MAX_NODES` / alza budget solo dopo review costi Hetzner |
+| Smoke Cloud | Provision **manuale** admin (kind cloud), non autoscaler; verifica WG + RTMP 1935 |
+| Scale-out lab OK | Solo dopo: considerare `ENABLED=1` + `KIND=lab` su staging/lab Proxmox |
+| Produzione overflow | Solo dopo checklist §12 punto 6 |
+
 Ordine di lavoro consigliato sul branch: **0 → L → 1 → 2 → 3 → 4**; **A** quando si decide di lasciare il Proxmox casa.
 
 ---
diff --git a/infra/.env.production.example b/infra/.env.production.example
index bc61072..92c9ca6 100644
--- a/infra/.env.production.example
+++ b/infra/.env.production.example
@@ -124,11 +124,15 @@ STREAM_CLOUD_PROVIDER=local_lab
 RELAY_MAX_CONCURRENT=4
 YOUTUBE_RELAY_WORKER=1
 
-# Autoscaler (kill-switch: 0 finché WireGuard/smoke Cloud non sono OK)
+# Autoscaler (kill-switch: 0 finché WireGuard/smoke Cloud non sono OK — NON abilitare in prod senza lab)
 STREAM_AUTOSCALE_ENABLED=0
 STREAM_AUTOSCALE_KIND=lab
+STREAM_AUTOSCALE_ALLOW_CLOUD=0
 STREAM_AUTOSCALE_SOFT_FREE_SLOTS=2
 STREAM_AUTOSCALE_WARM_SPARE=1
 STREAM_AUTOSCALE_IDLE_MINUTES=30
 STREAM_AUTOSCALE_MAX_NODES=5
 STREAM_AUTOSCALE_INTERVAL_SECS=60
+STREAM_AUTOSCALE_NODE_EUR_PER_HOUR=0.015
+STREAM_AUTOSCALE_MONTHLY_BUDGET_EUR=40
+STREAM_OVERFLOW_ORPHAN_HOURS=3