Add remote executor count API with per-overseer support

Adds Api::ExecutorConfig for reading and writing executor count
overrides stored in the backend (global and per-overseer). The
overseer polls this value each heartbeat interval and auto-scales
its executor pool when it changes.

New configuration property `overseer_id` provides a stable,
user-chosen label for per-overseer config lookups (concurrency
limits and executor count).

New Api surface:
- Api.executor_count / Api.set_executor_count (global)
- Api.executor_count(overseer_id:) / Api.set_executor_count(count, overseer_id:)
- Api::ExecutorConfig with .resolve for precedence-aware lookups
- Configuration#overseer_id for stable per-overseer identity

https://claude.ai/code/session_01PrikEg4dJya2KYjLkkKAmL

Author: claude

spec/mosquito/api/executor_config_spec.cr

@@ -0,0 +1,142 @@
+require "../../spec_helper"
+
+describe "Mosquito::Api::ExecutorConfig" do
+  describe "global executor count" do
+    it "returns nil when no override is stored" do
+      clean_slate do
+        result = Mosquito::Api::ExecutorConfig.stored_executor_count
+        assert_nil result
+      end
+    end
+
+    it "round-trips a global executor count" do
+      clean_slate do
+        Mosquito::Api::ExecutorConfig.store_executor_count(8)
+        result = Mosquito::Api::ExecutorConfig.stored_executor_count
+        assert_equal 8, result
+      end
+    end
+
+    it "clears the global executor count" do
+      clean_slate do
+        Mosquito::Api::ExecutorConfig.store_executor_count(8)
+        Mosquito::Api::ExecutorConfig.clear_executor_count
+
+        result = Mosquito::Api::ExecutorConfig.stored_executor_count
+        assert_nil result
+      end
+    end
+  end
+
+  describe "per-overseer executor count" do
+    it "returns nil when no per-overseer override is stored" do
+      clean_slate do
+        result = Mosquito::Api::ExecutorConfig.stored_executor_count("gpu-worker-1")
+        assert_nil result
+      end
+    end
+
+    it "round-trips a per-overseer executor count" do
+      clean_slate do
+        Mosquito::Api::ExecutorConfig.store_executor_count(2, "gpu-worker-1")
+
+        result = Mosquito::Api::ExecutorConfig.stored_executor_count("gpu-worker-1")
+        assert_equal 2, result
+
+        # Global is unaffected.
+        global = Mosquito::Api::ExecutorConfig.stored_executor_count
+        assert_nil global
+      end
+    end
+
+    it "clears per-overseer without affecting global" do
+      clean_slate do
+        Mosquito::Api::ExecutorConfig.store_executor_count(8)
+        Mosquito::Api::ExecutorConfig.store_executor_count(2, "gpu-worker-1")
+
+        Mosquito::Api::ExecutorConfig.clear_executor_count("gpu-worker-1")
+
+        per_overseer = Mosquito::Api::ExecutorConfig.stored_executor_count("gpu-worker-1")
+        assert_nil per_overseer
+
+        global = Mosquito::Api::ExecutorConfig.stored_executor_count
+        assert_equal 8, global
+      end
+    end
+  end
+
+  describe ".resolve" do
+    it "returns nil when nothing is stored" do
+      clean_slate do
+        result = Mosquito::Api::ExecutorConfig.resolve
+        assert_nil result
+      end
+    end
+
+    it "returns the global count when no overseer_id is given" do
+      clean_slate do
+        Mosquito::Api::ExecutorConfig.store_executor_count(8)
+        result = Mosquito::Api::ExecutorConfig.resolve
+        assert_equal 8, result
+      end
+    end
+
+    it "prefers per-overseer over global" do
+      clean_slate do
+        Mosquito::Api::ExecutorConfig.store_executor_count(8)
+        Mosquito::Api::ExecutorConfig.store_executor_count(2, "gpu-worker-1")
+
+        result = Mosquito::Api::ExecutorConfig.resolve("gpu-worker-1")
+        assert_equal 2, result
+      end
+    end
+
+    it "falls back to global when per-overseer is not set" do
+      clean_slate do
+        Mosquito::Api::ExecutorConfig.store_executor_count(8)
+
+        result = Mosquito::Api::ExecutorConfig.resolve("gpu-worker-1")
+        assert_equal 8, result
+      end
+    end
+  end
+
+  describe "instance methods" do
+    it "delegates to class-level helpers" do
+      clean_slate do
+        config = Mosquito::Api::ExecutorConfig.instance
+
+        config.update(10)
+        assert_equal 10, config.executor_count
+
+        config.update(3, overseer_id: "worker-1")
+        assert_equal 3, config.executor_count(overseer_id: "worker-1")
+
+        config.clear(overseer_id: "worker-1")
+        assert_nil config.executor_count(overseer_id: "worker-1")
+
+        config.clear
+        assert_nil config.executor_count
+      end
+    end
+  end
+end
+
+describe "Mosquito::Api executor count convenience methods" do
+  it "reads and writes global executor count" do
+    clean_slate do
+      Mosquito::Api.set_executor_count(12)
+      assert_equal 12, Mosquito::Api.executor_count
+    end
+  end
+
+  it "reads and writes per-overseer executor count" do
+    clean_slate do
+      Mosquito::Api.set_executor_count(4, overseer_id: "gpu-worker-1")
+      assert_equal 4, Mosquito::Api.executor_count(overseer_id: "gpu-worker-1")
+
+      # Global unaffected.
+      assert_nil Mosquito::Api.executor_count
+    end
+  end
+end

spec/mosquito/dequeue_adapters/remote_config_dequeue_adapter_spec.cr

@@ -114,6 +114,30 @@ describe "Mosquito::RemoteConfigDequeueAdapter" do
     end
   end
 
+  it "preserves in-flight counts when limits are refreshed" do
+    clean_slate do
+      register QueuedTestJob
+      2.times { QueuedTestJob.new.enqueue }
+
+      adapter = Mosquito::RemoteConfigDequeueAdapter.new(
+        defaults: {"queued_test_job" => 1},
+        refresh_interval: 0.seconds,
+      )
+
+      result1 = adapter.dequeue(queue_list)
+      refute_nil result1
+      assert_equal 1, adapter.active_count("queued_test_job")
+
+      # Refresh with new limits — must not reset the in-flight counter.
+      Mosquito::RemoteConfigDequeueAdapter.store_limits({"queued_test_job" => 2})
+      adapter.refresh_limits
+      assert_equal 1, adapter.active_count("queued_test_job")
+
+      adapter.finished_with(result1.not_nil!.job_run, result1.not_nil!.queue)
+      assert_equal 0, adapter.active_count("queued_test_job")
+    end
+  end
+
   it "delegates finished_with to the inner adapter" do
     clean_slate do
       register QueuedTestJob
@@ -296,6 +320,38 @@ describe "Mosquito::RemoteConfigDequeueAdapter" do
       end
     end
 
+    it "store_limits overwrites rather than merges (stale entries are removed)" do
+      clean_slate do
+        Mosquito::RemoteConfigDequeueAdapter.store_limits({"queue_a" => 3, "queue_b" => 7})
+        Mosquito::RemoteConfigDequeueAdapter.store_limits({"queue_a" => 1})
+
+        retrieved = Mosquito::RemoteConfigDequeueAdapter.stored_limits
+        assert_equal 1, retrieved["queue_a"]
+        refute retrieved.has_key?("queue_b"), "queue_b should have been removed by the overwrite"
+      end
+    end
+
+    it "store_limits with overseer_id overwrites rather than merges" do
+      clean_slate do
+        Mosquito::RemoteConfigDequeueAdapter.store_limits({"queue_a" => 3, "queue_b" => 7}, overseer_id: "worker-1")
+        Mosquito::RemoteConfigDequeueAdapter.store_limits({"queue_a" => 1}, overseer_id: "worker-1")
+
+        retrieved = Mosquito::RemoteConfigDequeueAdapter.stored_limits("worker-1")
+        assert_equal 1, retrieved["queue_a"]
+        refute retrieved.has_key?("queue_b"), "queue_b should have been removed by the overwrite"
+      end
+    end
+
+    it "store_limits with an empty hash removes all stored limits" do
+      clean_slate do
+        Mosquito::RemoteConfigDequeueAdapter.store_limits({"queue_a" => 3})
+        Mosquito::RemoteConfigDequeueAdapter.store_limits({} of String => Int32)
+
+        retrieved = Mosquito::RemoteConfigDequeueAdapter.stored_limits
+        assert_equal({} of String => Int32, retrieved)
+      end
+    end
+
     it "returns an empty hash when no limits are stored" do
       clean_slate do
         retrieved = Mosquito::RemoteConfigDequeueAdapter.stored_limits

spec/mosquito/runners/overseer_spec.cr

@@ -103,6 +103,76 @@ describe "Mosquito::Runners::Overseer" do
     end
   end
 
+  describe "remote executor count" do
+    it "applies the remote executor count on each_run" do
+      clean_slate do
+        Mosquito.configuration.overseer_id = "test-worker"
+        Mosquito::Api.set_executor_count(3, overseer_id: "test-worker")
+
+        queue_list.state = Runnable::State::Working
+        overseer.each_run
+
+        assert_equal 3, overseer.executor_count
+      ensure
+        Mosquito.configuration.overseer_id = nil
+      end
+    end
+
+    it "prefers per-overseer count over global" do
+      clean_slate do
+        Mosquito.configuration.overseer_id = "test-worker"
+        Mosquito::Api.set_executor_count(10)
+        Mosquito::Api.set_executor_count(2, overseer_id: "test-worker")
+
+        queue_list.state = Runnable::State::Working
+        overseer.each_run
+
+        assert_equal 2, overseer.executor_count
+      ensure
+        Mosquito.configuration.overseer_id = nil
+      end
+    end
+
+    it "falls back to global when no per-overseer count is set" do
+      clean_slate do
+        Mosquito.configuration.overseer_id = "test-worker"
+        Mosquito::Api.set_executor_count(7)
+
+        queue_list.state = Runnable::State::Working
+        overseer.each_run
+
+        assert_equal 7, overseer.executor_count
+      ensure
+        Mosquito.configuration.overseer_id = nil
+      end
+    end
+
+    it "does not change executor_count when no remote value is set" do
+      clean_slate do
+        original_count = overseer.executor_count
+
+        queue_list.state = Runnable::State::Working
+        overseer.each_run
+
+        assert_equal original_count, overseer.executor_count
+      end
+    end
+
+    it "clamps an invalid remote executor count of 0 to 1" do
+      clean_slate do
+        Mosquito.configuration.overseer_id = "test-worker"
+        Mosquito::Api.set_executor_count(0, overseer_id: "test-worker")
+
+        queue_list.state = Runnable::State::Working
+        overseer.each_run
+
+        assert_equal 1, overseer.executor_count
+      ensure
+        Mosquito.configuration.overseer_id = nil
+      end
+    end
+  end
+
   describe "cleanup_orphaned_pending_jobs" do
     it "recovers a pending job whose overseer is dead" do
       clean_slate do

src/mosquito/api.cr

@@ -62,4 +62,30 @@ module Mosquito::Api
   def self.set_concurrency_limits(limits : Hash(String, Int32), overseer_id : String) : Nil
     concurrency_config.update(limits, overseer_id)
   end
+
+  # Returns an `ExecutorConfig` instance for reading and writing the
+  # remotely stored executor count.
+  def self.executor_config : ExecutorConfig
+    ExecutorConfig.instance
+  end
+
+  # Convenience reader for the global remote executor count.
+  def self.executor_count : Int32?
+    executor_config.executor_count
+  end
+
+  # Convenience reader for a specific overseer's executor count.
+  def self.executor_count(overseer_id : String) : Int32?
+    executor_config.executor_count(overseer_id)
+  end
+
+  # Convenience writer — sets the global executor count override.
+  def self.set_executor_count(count : Int32) : Nil
+    executor_config.update(count)
+  end
+
+  # Convenience writer — sets the executor count for a specific overseer.
+  def self.set_executor_count(count : Int32, overseer_id : String) : Nil
+    executor_config.update(count, overseer_id)
+  end
 end