Class: Phronomy::Runtime

Inherits:
Object
  • Object
show all
Defined in:
lib/phronomy/engine/runtime.rb,
lib/phronomy/engine/runtime/scheduler.rb,
lib/phronomy/engine/runtime/timer_queue.rb,
lib/phronomy/engine/runtime/task_registry.rb,
lib/phronomy/engine/runtime/timer_service.rb,
lib/phronomy/engine/runtime/fake_scheduler.rb,
lib/phronomy/engine/runtime/runtime_metrics.rb,
lib/phronomy/engine/runtime/shutdown_result.rb,
lib/phronomy/engine/runtime/thread_scheduler.rb,
lib/phronomy/engine/runtime/deterministic_scheduler.rb,
lib/phronomy/engine/runtime/scheduler_timer_adapter.rb

Overview

Central authority for concurrent primitives.

Runtime is the single place that creates Tasks, TaskGroups, and manages the lifecycle of all concurrency in Phronomy. It owns:

In production, use the process-wide singleton via Runtime.instance. In tests, construct a Runtime with a FakeScheduler to run tasks synchronously without spawning additional threads:

Examples:

Production usage

group = Phronomy::Runtime.instance.task_group(limit: 4)
tools.each { |t| group.spawn { t.call } }
results = group.await_all

Test usage — no extra threads

runtime = Phronomy::Runtime.new(scheduler: Phronomy::Runtime::FakeScheduler.new)
task = runtime.spawn { 42 }
expect(task.wait_result).to eq(42)

Defined Under Namespace

Classes: DeterministicScheduler, FakeScheduler, RuntimeMetrics, Scheduler, SchedulerTimerAdapter, ShutdownResult, TaskRegistry, ThreadScheduler, TimerQueue, TimerService

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(scheduler: ThreadScheduler.new) ⇒ Runtime

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns a new instance of Runtime.

Parameters:



182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
 
# File 'lib/phronomy/engine/runtime.rb', line 182

def initialize(scheduler: ThreadScheduler.new)
  @scheduler = scheduler
  @event_loop_scheduler = ThreadScheduler.new
  @task_registry = TaskRegistry.new
  @metrics = RuntimeMetrics.new
  @timer_service = TimerService.new(scheduler)
  @pool_registry = Phronomy::Concurrency::PoolRegistry.new(
    timer_queue_provider: -> { @timer_service.timer_queue }
  )
  @lifecycle_mutex = Mutex.new
  @shutdown_mutex = Mutex.new
  @state = :running
  @event_loop = nil
  @failure = nil
  @shutdown_result = nil
end

Instance Attribute Details

#scheduler ⇒ Scheduler (readonly)

The scheduler backing this runtime instance.

Returns:



172
173
174
 
# File 'lib/phronomy/engine/runtime.rb', line 172

def scheduler
  @scheduler
end

Class Method Details

.default_if_initialized_for_test ⇒ Object

Test-only, non-creating access to the default Runtime.



66
67
68
 
# File 'lib/phronomy/engine/runtime.rb', line 66

def default_if_initialized_for_test
  instance_mutex.synchronize { @instance }
end

.in_event_loop_context? ⇒ Boolean

Does not create a Runtime or EventLoop.

Returns:

  • (Boolean) 


101
102
103
104
 
# File 'lib/phronomy/engine/runtime.rb', line 101

def in_event_loop_context?
  runtime = instance_mutex.synchronize { @instance }
  runtime&.event_loop_current? || false
end

.in_scheduler_context? ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns true when the calling thread is executing inside an active scheduler task (i.e. Task.current is non-nil). Code running inside a #spawn block is always in a scheduler context.

Use this to detect potential scheduler-blocking calls:

if Phronomy::Runtime.in_scheduler_context?
Phronomy.configuration.logger&.warn("blocking call inside scheduler task")
end

Returns:

  • (Boolean) 


149
150
151
 
# File 'lib/phronomy/engine/runtime.rb', line 149

def self.in_scheduler_context?
  !Task.current.nil?
end

.instance ⇒ Object 



54
55
56
57
58
 
# File 'lib/phronomy/engine/runtime.rb', line 54

def instance
  instance_mutex.synchronize do
    @instance ||= build_default_runtime
  end
end

.instance=(runtime) ⇒ Object

Compatibility setter retained for existing tests.



61
62
63
 
# File 'lib/phronomy/engine/runtime.rb', line 61

def instance=(runtime)
  replace_default_for_test(runtime)
end

.measure_ms { ... } ⇒ Array(Object, Integer)

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Executes block and returns [result, elapsed_ms] where elapsed_ms is the wall-clock duration in milliseconds (Integer, rounded).

Isolates all direct references to Process.clock_gettime / Process::CLOCK_MONOTONIC in one place so that callers stay at the framework abstraction level.

Yields:

  • block to time

Returns:

  • (Array(Object, Integer)) —

    [block_return_value, elapsed_ms]



163
164
165
166
167
168
 
# File 'lib/phronomy/engine/runtime.rb', line 163

def self.measure_ms
  t0 = Process.clock_gettime(Process::CLOCK_MONOTONIC)
  result = yield
  elapsed_ms = ((Process.clock_gettime(Process::CLOCK_MONOTONIC) - t0) * 1000).round
  [result, elapsed_ms]
end

.replace_default_for_test(runtime) ⇒ Object

Test-only replacement. The caller owns both Runtime lifecycles.



71
72
73
74
75
76
77
 
# File 'lib/phronomy/engine/runtime.rb', line 71

def replace_default_for_test(runtime)
  instance_mutex.synchronize do
    previous = @instance
    @instance = runtime
    previous
  end
end

.reset_default!(timeout: Phronomy.configuration.event_loop_stop_grace_seconds) ⇒ Object 



84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
 
# File 'lib/phronomy/engine/runtime.rb', line 84

def reset_default!(timeout: Phronomy.configuration.event_loop_stop_grace_seconds)
  runtime = instance_mutex.synchronize { @instance }
  return ShutdownResult.not_started unless runtime

  result = runtime.shutdown(timeout: timeout)
  unless result.cleanup_complete?
    raise Phronomy::RuntimeShutdownError,
      "Runtime cleanup is incomplete; default Runtime was retained"
  end

  instance_mutex.synchronize do
    @instance = nil if @instance.equal?(runtime)
  end
  result
end

.restore_default_for_test(runtime) ⇒ Object

Test-only restoration of a previously captured Runtime.



80
81
82
 
# File 'lib/phronomy/engine/runtime.rb', line 80

def restore_default_for_test(runtime)
  instance_mutex.synchronize { @instance = runtime }
end

Instance Method Details

#__event_loop_failed(error) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Called only for an unexpected dispatcher failure.



441
442
443
444
445
446
447
448
 
# File 'lib/phronomy/engine/runtime.rb', line 441

def __event_loop_failed(error)
  @lifecycle_mutex.synchronize do
    return if @shutdown_result || @state == :terminated

    @failure ||= error
    @state = :failed
  end
end

#__spawn_event_loop_service(&block) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Internal EventLoop service spawn. Always uses a real OS thread and is deliberately excluded from the normal TaskRegistry drain.



435
436
437
 
# File 'lib/phronomy/engine/runtime.rb', line 435

def __spawn_event_loop_service(&block)
  @event_loop_scheduler.spawn(name: "event-loop", parent: nil, &block)
end

#blocking_io(pool_size: Phronomy.configuration.blocking_io_pool_size, queue_size: Phronomy.configuration.blocking_io_queue_size) ⇒ BlockingAdapterPool

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns the shared BlockingAdapterPool for this Runtime. All blocking I/O (LLM HTTP, MCP, ActiveRecord, Redis) should be submitted through this pool.

Pool settings default to 10 workers / 100-deep queue. Override by constructing a Runtime with custom pool options or by replacing the shared Runtime via instance= in tests.

Parameters:

Returns:

  • (BlockingAdapterPool) 


357
358
359
360
361
 
# File 'lib/phronomy/engine/runtime.rb', line 357

def blocking_io(pool_size: Phronomy.configuration.blocking_io_pool_size,
  queue_size: Phronomy.configuration.blocking_io_queue_size)
  ensure_accepting_work!
  @pool_registry.default_pool(pool_size: pool_size, queue_size: queue_size)
end

#event_loop ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns the Runtime-owned EventLoop, creating it once on first use. During draining an existing loop remains available, but an unused loop is never created after shutdown begins.



408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
 
# File 'lib/phronomy/engine/runtime.rb', line 408

def event_loop
  @lifecycle_mutex.synchronize do
    case @state
    when :running
      @event_loop ||= EventLoop.new(runtime: self)
    when :draining
      return @event_loop if @event_loop

      raise Phronomy::RuntimeShutdownError,
        "EventLoop was not initialized before Runtime shutdown began"
    else
      raise Phronomy::RuntimeShutdownError,
        "Runtime is #{@state}; EventLoop is unavailable"
    end
  end
end

#event_loop_current? ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Does not create an EventLoop.

Returns:

  • (Boolean) 


427
428
429
430
 
# File 'lib/phronomy/engine/runtime.rb', line 427

def event_loop_current?
  event_loop = @lifecycle_mutex.synchronize { @event_loop }
  event_loop&.current? || false
end

#non_yield_threshold_violation_count ⇒ Integer

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Number of times a task has exceeded the CPU-bound detection threshold (i.e. ran longer than blocking_detect_threshold_ms without yielding). Resets to 0 when the Runtime is recreated.

Returns:

  • (Integer) 


240
241
242
 
# File 'lib/phronomy/engine/runtime.rb', line 240

def non_yield_threshold_violation_count
  @metrics.starvation_count
end

#pool(name, size: 10, queue_size: 100) ⇒ BlockingAdapterPool

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns (or lazily creates) a named BlockingAdapterPool.

Named pools allow per-subsystem thread-budget control and observability. Recommended pool names: :llm, :mcp, :db, :redis, :tool. Each pool gets its own dedicated worker threads labelled with the pool name.

Examples:

runtime.pool(:llm)            # default size (10 workers)
runtime.pool(:db, size: 20)   # custom size

Parameters:

  • name (Symbol, String) —

    pool identifier

  • size (Integer) (defaults to: 10) —

    worker thread count (default: 10)

  • queue_size (Integer) (defaults to: 100) —

    max pending operations (default: 100)

Returns:

  • (BlockingAdapterPool) 


378
379
380
381
 
# File 'lib/phronomy/engine/runtime.rb', line 378

def pool(name, size: 10, queue_size: 100)
  ensure_accepting_work!
  @pool_registry.named_pool(name, size: size, queue_size: queue_size)
end

#shutdown(timeout: Phronomy.configuration.event_loop_stop_grace_seconds, cancel_grace: timeout) ⇒ Runtime::ShutdownResult

Synchronous, bounded Runtime shutdown. Must be invoked from an external management thread, lifecycle hook, or test teardown—not a Phronomy Task.

timeout bounds TaskRegistry and EventLoop graceful shutdown. Existing pool and timer shutdown contracts are unchanged by this proposal.

Returns:



457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
 
# File 'lib/phronomy/engine/runtime.rb', line 457

def shutdown(
  timeout: Phronomy.configuration.event_loop_stop_grace_seconds,
  cancel_grace: timeout
)
  if Phronomy::Task.current
    raise Phronomy::RuntimeShutdownReentrancyError,
      "Runtime#shutdown must be called from an external management thread"
  end
  validate_timeout!(timeout, :timeout)
  validate_timeout!(cancel_grace, :cancel_grace)

  @shutdown_mutex.synchronize do
    return @shutdown_result if @shutdown_result

    deadline = monotonic_now + timeout
    event_loop = @lifecycle_mutex.synchronize do
      @state = :draining unless @state == :failed
      @event_loop
    end
    event_loop&.begin_draining

    task_status = drain_runtime_work(event_loop, deadline)

    @lifecycle_mutex.synchronize do
      @state = :stopping unless @state == :failed
    end

    event_loop_status = if event_loop
      event_loop.shutdown(deadline: deadline, cancel_grace: cancel_grace)
    else
      :not_started
    end

    subsystem_error = shutdown_pools_and_timer
    final_task_status = @task_registry.empty? ? :empty : task_status
    cleanup_complete = final_task_status == :empty &&
      (!event_loop || !event_loop.task_alive?) &&
      event_loop_status != :cancel_timeout &&
      subsystem_error.nil?

    failure = @lifecycle_mutex.synchronize { @failure } || subsystem_error
    runtime_outcome = if failure || event_loop_status == :failed
      :failed
    else
      :terminated
    end
    result = ShutdownResult.new(
      runtime_outcome: runtime_outcome,
      cleanup_status: cleanup_complete ? :complete : :incomplete,
      event_loop_status: event_loop_status,
      task_registry_status: final_task_status,
      error: failure
    )

    @lifecycle_mutex.synchronize do
      @state = cleanup_complete ? runtime_outcome : :failed
      @shutdown_result = result
    end
    result
  end
end

#spawn(name: nil) { ... } ⇒ Task

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Spawns a single Task using the runtime's scheduler.

The spawned task is registered in the task registry so #shutdown can wait for it to complete. The task is automatically deregistered from the registry when it finishes (success, failure, or cancellation) so long-lived runtimes do not accumulate stale references.

Task names beginning with a recognised type prefix are counted in the task-centric metrics returned by #task_snapshot. Recognised prefixes: agent-, tool-, workflow-, rag-, llm-, vector-.

Parameters:

  • name (String, nil) (defaults to: nil) —

    optional label for debugging

Yields:

  • block to execute (concurrently or synchronously, depending on the configured scheduler)

Returns:



293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
 
# File 'lib/phronomy/engine/runtime.rb', line 293

def spawn(name: nil, &block)
  ensure_accepting_work!
  type = _task_type(name)
  spawn_at = Process.clock_gettime(Process::CLOCK_MONOTONIC, :millisecond)
  @metrics.record_start(type)

  task = @scheduler.spawn(name: name, parent: Task.current) do
    run_start = Process.clock_gettime(Process::CLOCK_MONOTONIC, :millisecond)
    @metrics.record_wait(run_start - spawn_at)
    begin
      result = block.call
      @metrics.record_end(type, :completed, run_start)
      result
    rescue CancellationError
      @metrics.record_end(type, :cancelled, run_start)
      raise
    rescue => e
      @metrics.record_end(type, :failed, run_start)
      raise e
    ensure
      current = Task.current
      @task_registry.deregister(current) if current
    end
  end
  @task_registry.register(task)
  task
end

#state ⇒ Symbol

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns current Runtime lifecycle state.

Returns:

  • (Symbol) —

    current Runtime lifecycle state



176
177
178
 
# File 'lib/phronomy/engine/runtime.rb', line 176

def state
  @lifecycle_mutex.synchronize { @state }
end

#task_group(limit: Float::INFINITY, failure_policy: :fail_fast) ⇒ TaskGroup

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Creates a new TaskGroup with an optional concurrency cap.

Parameters:

  • limit (Integer, Float::INFINITY) (defaults to: Float::INFINITY) —

    max simultaneous tasks

  • failure_policy (Symbol) (defaults to: :fail_fast) —

    one of :fail_fast, :collect_all, :skip_failed (default :fail_fast)

Returns:



272
273
274
275
 
# File 'lib/phronomy/engine/runtime.rb', line 272

def task_group(limit: Float::INFINITY, failure_policy: :fail_fast)
  ensure_accepting_work!
  TaskGroup.new(limit: limit, failure_policy: failure_policy, runtime: self)
end

#task_snapshot ⇒ Hash{Symbol => Numeric}

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns a snapshot of task-centric metrics for the current Runtime.

Key Description
active_agent_tasks currently running agent spawns
active_tool_tasks currently running tool spawns
active_workflow_tasks currently running workflow spawns
active_llm_tasks currently running LLM calls
task_wait_time_p50_ms p50 spawn-to-start latency (ms)
task_wait_time_p95_ms p95 spawn-to-start latency (ms)
task_run_time_p50_ms p50 execution duration (ms)
task_run_time_p95_ms p95 execution duration (ms)
cancelled_tasks total cancelled task count
failed_tasks total failed task count
non_yield_threshold_violation_count cumulative count of tasks that ran past blocking_detect_threshold_ms without yielding

Returns:

  • (Hash{Symbol => Numeric}) 


339
340
341
 
# File 'lib/phronomy/engine/runtime.rb', line 339

def task_snapshot
  @metrics.snapshot
end

#timer_queue ⇒ TimerQueue, SchedulerTimerAdapter

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns the shared timer queue for this Runtime.

When the scheduler is a DeterministicScheduler (e.g. the :fiber runtime backend), returns a SchedulerTimerAdapter that integrates with the scheduler's tick cycle instead of spawning a background OS thread. This is the first concrete step of the TimerQueue scheduler-tick integration described in ADR-010 (Issue #331).

For all other schedulers, returns a TimerQueue backed by a single background thread.

All deadline-based cancellation should be registered here instead of spawning one-off sleep threads. Lazily created on first access.

Returns:



399
400
401
402
 
# File 'lib/phronomy/engine/runtime.rb', line 399

def timer_queue
  ensure_accepting_work!
  @timer_service.timer_queue
end

#yield ⇒ void

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

This method returns an undefined value.

Cooperative yield point.

Signals the scheduler that the current task is willing to give up CPU time so that other ready tasks can run. On the default ThreadScheduler this calls Thread.pass. On a future fiber-based scheduler this would switch to the next runnable fiber.

When blocking_detect_threshold_ms is configured, checks whether the current task has exceeded that threshold without yielding; if so, emits a warning via the configured logger and increments non_yield_threshold_violation_count.

Call this inside tight loops or CPU-intensive sections of tool execute methods and Workflow actions to keep the scheduler responsive.



216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
 
# File 'lib/phronomy/engine/runtime.rb', line 216

def yield
  if (threshold = Phronomy.configuration.blocking_detect_threshold_ms)
    slice_start = Task.current_cpu_slice_start_ms
    if slice_start
      elapsed = Process.clock_gettime(Process::CLOCK_MONOTONIC, :millisecond) - slice_start
      if elapsed > threshold
        name = Task.current&.name || "unknown"
        Phronomy.configuration.logger&.warn(
          "[Phronomy] CPU-bound task detected: '#{name}' ran #{elapsed.round}ms " \
          "without yielding (threshold: #{threshold}ms)"
        )
        @metrics.increment_starvation
      end
    end
  end
  Task.record_yield!
  @scheduler.yield
end

#yield_if_needed(every: 1000) ⇒ void

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

This method returns an undefined value.

Cooperative yield point with a call-count gate.

Increments a per-thread counter and calls #yield when the counter reaches a multiple of every. The counter is thread-local so concurrent tasks each maintain their own independent loop counter without requiring a mutex.

Examples:

data.each_with_index do |row, i|
  process(row)
  Phronomy::Runtime.instance.yield_if_needed(every: 500)
end

Parameters:

  • every (Integer) (defaults to: 1000) —

    yield once every N calls (default: 1000)



260
261
262
263
264
 
# File 'lib/phronomy/engine/runtime.rb', line 260

def yield_if_needed(every: 1000)
  # Delegate Thread.current access to Task so that runtime.rb stays outside
  # the Thread.current allowlist (Issue #302).
  self.yield if (Task.increment_yield_counter! % every).zero?
end