The Plan Server

The Plan Server: a server is three disciplines wearing one process - accept concurrently, share resources safely, and above all SHUT DOWN WELL. This serves plan executions over a real socket with a thread pool, a shared (mutexed) rate limit across all request threads, and the part everyone skips: a graceful drain where in-flight requests finish, new ones are refused, and the process exits clean.

source on github

a real captured run

THE PLAN SERVER (loopback:42981, 3 worker threads, shared quota)

  burst of 8 concurrent requests, 3 workers:
    processed 8 words  (summarize ticket number 0 for ...)
    processed 8 words  (summarize ticket number 1 for ...)
    processed 8 words  (summarize ticket number 2 for ...)
    ... 8 of 8 answered

  graceful drain with one request in flight:
    in-flight request completed: "processed 7 words"
    drain took 15ms; total served: 9; refused after: connection refused

  the order of operations IS the grace: close the LISTENER first
  (the OS starts refusing for you - no accept race), let workers
  finish what they hold, join, exit. kill -9 has none of these
  steps, which is why deploys under it drop the request that was
  42 seconds into a 43-second plan. the shared quota is the other
  server lesson: request threads are REAL threads, and the
  windowed limiter holds because round 12 gave its bookkeeping a
  real Mutex - a server is where every thread-safety promise in
  your dependency tree gets called at once.

source

# frozen_string_literal: true

# The Plan Server: a server is three disciplines wearing one process -
# accept concurrently, share resources safely, and above all SHUT DOWN
# WELL. This serves plan executions over a real socket with a thread
# pool, a shared (mutexed) rate limit across all request threads, and
# the part everyone skips: a graceful drain where in-flight requests
# finish, new ones are refused, and the process exits clean.
#
#   bundle exec ruby examples/plan_server.rb
#
# Runs offline; the socket is loopback, the clients are threads.

require class="s">"bundler/setup"
require class="s">"agentic"
require class="s">"socket"
require class="s">"json"

Agentic.logger.level = class="y">:fatal

class PlanServer
  def initialize(workers: 3)
    @server = TCPServer.new(class="s">"127.0.0.1", 0) # ephemeral port
    @workers = workers
    @quota = Agentic:class="y">:RateLimit.new(100, per: 60) # shared across ALL request threads
    @draining = false
    @in_flight = 0
    @served = 0
    @lock = Mutex.new
  end

  def port = @server.addr[1]

  attr_reader class="y">:served

  def start
    @threads = @workers.times.map do
      Thread.new do
        loop do
          socket = begin
            @server.accept
          rescue IOError
            break # listener closed: drain mode
          end
          handle(socket)
        end
      end
    end
  end

  # The graceful drain: stop the LISTENER first (new connections get
  # refused by the OS), then wait for in-flight work, then exit
  def drain
    @lock.synchronize { @draining = true }
    @server.close
    @threads.each(&class="y">:join)
  end

  private

  def handle(socket)
    @lock.synchronize { @in_flight += 1 }
    goal = socket.gets&.strip

    unless @quota.try_acquire
      socket.puts JSON.generate({error: class="s">"quota exhausted", retry_after: 60})
      return
    end

    orchestrator = Agentic:class="y">:PlanOrchestrator.new
    fetch = Agentic:class="y">:Task.new(description: class="s">"fetch", agent_spec: {class="s">"name" => class="s">"f", class="s">"instructions" => class="s">"w"})
    answer = Agentic:class="y">:Task.new(description: class="s">"answer", agent_spec: {class="s">"name" => class="s">"a", class="s">"instructions" => class="s">"w"})
    orchestrator.add_task(fetch, agent: ->(_t) {
      sleep(0.02)
      goal.to_s.split.size
    })
    orchestrator.add_task(answer, [fetch], agent: ->(t) { class="s">"processed #{t.previous_output} words" })
    result = orchestrator.execute_plan

    socket.puts JSON.generate({goal: goal, answer: result.task_result(answer.id).output})
    @lock.synchronize { @served += 1 }
  ensure
    @lock.synchronize { @in_flight -= 1 }
    socket.close
  end
end

server = PlanServer.new(workers: 3)
server.start

puts class="s">"THE PLAN SERVER (loopback:#{server.port}, 3 worker threads, shared quota)"
puts

# --- clients: a burst of concurrent requests -------------------------------------
responses = 8.times.map { |i|
  Thread.new do
    TCPSocket.open(class="s">"127.0.0.1", server.port) do |s|
      s.puts class="s">"summarize ticket number #{i} for the weekly report"
      JSON.parse(s.gets, symbolize_names: true)
    end
  end
}.map(&class="y">:value)

puts class="s">"  burst of 8 concurrent requests, 3 workers:"
responses.first(3).each { |r| puts class="s">"    #{r[class="y">:answer]}  (#{r[class="y">:goal][0, 30]}...)" }
puts class="s">"    ... #{responses.count { |r| r[class="y">:answer] }} of 8 answered"
puts

# --- the drain: one slow request in flight when the order comes ----------------
slow_client = Thread.new do
  TCPSocket.open(class="s">"127.0.0.1", server.port) do |s|
    s.puts class="s">"one last long report before the deploy"
    JSON.parse(s.gets, symbolize_names: true)
  end
end
sleep(0.01) # let it get in the door
drained_at = Process.clock_gettime(Process:class="y">:CLOCK_MONOTONIC)
server.drain
drain_ms = ((Process.clock_gettime(Process:class="y">:CLOCK_MONOTONIC) - drained_at) * 1000).round
last = slow_client.value

puts class="s">"  graceful drain with one request in flight:"
puts class="s">"    in-flight request completed: #{last[class="y">:answer].inspect}"
puts class="s">"    drain took #{drain_ms}ms; total served: #{server.served}; refused after: connection refused"
puts
puts class="s">"  the order of operations IS the grace: close the LISTENER first"
puts class="s">"  (the OS starts refusing for you - no accept race), let workers"
puts class="s">"  finish what they hold, join, exit. kill -9 has none of these"
puts class="s">"  steps, which is why deploys under it drop the request that was"
puts class="s">"  42 seconds into a 43-second plan. the shared quota is the other"
puts class="s">"  server lesson: request threads are REAL threads, and the"
puts class="s">"  windowed limiter holds because round 12 gave its bookkeeping a"
puts class="s">"  real Mutex - a server is where every thread-safety promise in"
puts class="s">"  your dependency tree gets called at once."