Class: PromptObjects::Runtime

Inherits:
Object
  • Object
show all
Defined in:
lib/prompt_objects/environment.rb

Overview

The runtime environment that holds all capabilities and coordinates execution. Can be initialized either from:

  • A user environment directory (with manifest.yml, git integration)
  • A simple objects directory backed by an in-memory runtime store

Constant Summary collapse

DEFAULT_SCHEDULER_OPTIONS = 
{
  worker_limit: 4,
  provider_limit: 4,
  tool_limit: 8,
  cpu_limit: 2,
  max_active_runs: 64,
  max_active_per_po: 4,
  max_queued_per_thread: 20,
  max_delegation_depth: 8,
  max_descendants: 64,
  run_event_retention: Operations::Retention::DEFAULT_RUN_EVENT_LIMIT,
  workspace_event_retention: Operations::Retention::DEFAULT_WORKSPACE_EVENT_LIMIT
}.freeze

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(env_path: nil, objects_dir: nil, primitives_dir: nil, llm: nil, provider: nil, model: nil, auto_commit: nil, runtime_model: nil, scheduler_options: nil) ⇒ Runtime

Initialize from an environment path (with manifest) or objects directory.

Parameters:

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

    Path to environment directory (preferred)

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

    Legacy: path to objects directory

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

    Path to primitives directory

  • llm (Object, nil) (defaults to: nil) —

    LLM adapter (deprecated, use provider/model instead)

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

    LLM provider (openai, anthropic, gemini)

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

    Model name

  • auto_commit (Boolean) (defaults to: nil) —

    Auto-commit changes to git (default: true for env_path)



84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
 
# File 'lib/prompt_objects/environment.rb', line 84

def initialize(env_path: nil, objects_dir: nil, primitives_dir: nil, llm: nil, provider: nil, model: nil,
               auto_commit: nil, runtime_model: nil, scheduler_options: nil)
  if env_path
    # Environment-based initialization
    @env_path = env_path
    @objects_dir = File.join(env_path, "objects")
    @primitives_dir = primitives_dir || File.join(env_path, "primitives")
    @services_dir = File.join(env_path, "services")
    @connectors_dir = File.join(env_path, "connectors")
    @manifest = Env::Manifest.load_from_dir(env_path)
    @auto_commit = auto_commit.nil? ? true : auto_commit
    @manifest.touch_opened!
    @manifest.save_to_dir(env_path)

    # Initialize session store
    db_path = File.join(env_path, "sessions.db")
    @session_store = Session::Store.new(db_path)
  else
    # Legacy objects_dir initialization
    @env_path = nil
    @objects_dir = objects_dir || "objects"
    @primitives_dir = primitives_dir || File.join(File.dirname(@objects_dir), "primitives")
    @services_dir = nil
    @connectors_dir = nil
    @manifest = nil
    @auto_commit = auto_commit || false
    @session_store = Session::Store.new(":memory:")
  end

  @runtime_model = "runs"
  if @manifest && @manifest.runtime_model != "runs"
    @manifest.runtime_model = "runs"
    @manifest.save_to_dir(@env_path)
  end
  @scheduler_options = DEFAULT_SCHEDULER_OPTIONS.merge(scheduler_options || {})
  @child_admission_mutex = Mutex.new

  # Initialize LLM - prefer explicit llm, then use factory with provider/model
  if llm
    @llm = llm
    @current_provider = nil
    @current_model = nil
  else
    manifest_llm = @manifest&.llm || {}
    @current_provider = provider || manifest_llm["provider"] || LLM::Factory::DEFAULT_PROVIDER
    manifest_model = manifest_llm["model"] unless provider
    @current_model = model || manifest_model || LLM::Factory.default_model(@current_provider)
    @llm = LLM::Factory.create(provider: @current_provider, model: @current_model)
  end

  @registry = Registry.new
  @bus = MessageBus.new(session_store: @session_store)
  @human_queue = HumanQueue.new
  # In-memory boot-cycle activity. Autonomous POs can opt to stay dormant
  # until a human has initialized them during this server run; persisted
  # sessions from a previous run intentionally do not arm the timer.
  @human_message_received = {}
  @autonomous_schedule_states = {}
  @autonomous_schedule_mutex = Mutex.new

  # Paths the app just wrote itself (via PO#save / #update_source). The
  # FileWatcher consults this so an in-app save doesn't trigger a redundant,
  # disruptive reload of a PO that's mid-conversation — the in-memory PO is
  # already up to date. Manual/external edits aren't registered, so they
  # still hot-reload normally.
  @suppressed_reloads = {}
  @suppressed_reloads_mutex = Mutex.new

  validate_dependencies if @manifest&.dependencies&.any?

  register_primitives
  register_universal_capabilities

  @services = []
  load_custom_services

  @connectors = []
  load_custom_connectors

  recover_interrupted_runs
end

Instance Attribute Details

#active_session_id ⇒ Object (readonly)

Returns the value of attribute active_session_id.



66
67
68
 
# File 'lib/prompt_objects/environment.rb', line 66

def active_session_id
  @active_session_id
end

#auto_commit ⇒ Object (readonly)

Returns the value of attribute auto_commit.



66
67
68
 
# File 'lib/prompt_objects/environment.rb', line 66

def auto_commit
  @auto_commit
end

#bus ⇒ Object (readonly)

Returns the value of attribute bus.



66
67
68
 
# File 'lib/prompt_objects/environment.rb', line 66

def bus
  @bus
end

#connectors ⇒ Object (readonly)

Returns the value of attribute connectors.



66
67
68
 
# File 'lib/prompt_objects/environment.rb', line 66

def connectors
  @connectors
end

#current_model ⇒ Object (readonly)

Returns the value of attribute current_model.



66
67
68
 
# File 'lib/prompt_objects/environment.rb', line 66

def current_model
  @current_model
end

#current_provider ⇒ Object (readonly)

Returns the value of attribute current_provider.



66
67
68
 
# File 'lib/prompt_objects/environment.rb', line 66

def current_provider
  @current_provider
end

#env_path ⇒ Object (readonly)

Returns the value of attribute env_path.



66
67
68
 
# File 'lib/prompt_objects/environment.rb', line 66

def env_path
  @env_path
end

#human_queue ⇒ Object (readonly)

Returns the value of attribute human_queue.



66
67
68
 
# File 'lib/prompt_objects/environment.rb', line 66

def human_queue
  @human_queue
end

#llm ⇒ Object (readonly)

Returns the value of attribute llm.



66
67
68
 
# File 'lib/prompt_objects/environment.rb', line 66

def llm
  @llm
end

#manifest ⇒ Object (readonly)

Returns the value of attribute manifest.



66
67
68
 
# File 'lib/prompt_objects/environment.rb', line 66

def manifest
  @manifest
end

#objects_dir ⇒ Object (readonly)

Returns the value of attribute objects_dir.



66
67
68
 
# File 'lib/prompt_objects/environment.rb', line 66

def objects_dir
  @objects_dir
end

#on_broadcast ⇒ Object

Callback for services to push messages to all WebSocket clients



74
75
76
 
# File 'lib/prompt_objects/environment.rb', line 74

def on_broadcast
  @on_broadcast
end

#on_delegation_event ⇒ Object

Callback for PO-to-PO delegation start/complete



72
73
74
 
# File 'lib/prompt_objects/environment.rb', line 72

def on_delegation_event
  @on_delegation_event
end

#on_env_data_changed ⇒ Object

Callback for env data store/update/delete



73
74
75
 
# File 'lib/prompt_objects/environment.rb', line 73

def on_env_data_changed
  @on_env_data_changed
end

#on_po_modified ⇒ Object

Callback for when a PO is modified (capabilities changed, etc.)



71
72
73
 
# File 'lib/prompt_objects/environment.rb', line 71

def on_po_modified
  @on_po_modified
end

#on_po_registered ⇒ Object

Callback for when a PO is registered



70
71
72
 
# File 'lib/prompt_objects/environment.rb', line 70

def on_po_registered
  @on_po_registered
end

#primitives_dir ⇒ Object (readonly)

Returns the value of attribute primitives_dir.



66
67
68
 
# File 'lib/prompt_objects/environment.rb', line 66

def primitives_dir
  @primitives_dir
end

#registry ⇒ Object (readonly)

Returns the value of attribute registry.



66
67
68
 
# File 'lib/prompt_objects/environment.rb', line 66

def registry
  @registry
end

#runtime_model ⇒ Object (readonly)

Returns the value of attribute runtime_model.



66
67
68
 
# File 'lib/prompt_objects/environment.rb', line 66

def runtime_model
  @runtime_model
end

#services ⇒ Object (readonly)

Returns the value of attribute services.



66
67
68
 
# File 'lib/prompt_objects/environment.rb', line 66

def services
  @services
end

#session_store ⇒ Object (readonly)

Returns the value of attribute session_store.



66
67
68
 
# File 'lib/prompt_objects/environment.rb', line 66

def session_store
  @session_store
end

Class Method Details

.from_environment(name, manager: nil) ⇒ Runtime

Create runtime from a user environment by name.

Parameters:

  • name (String) —

    Environment name

  • manager (Env::Manager, nil) (defaults to: nil) —

    Manager instance

Returns:

Raises:



170
171
172
173
174
175
 
# File 'lib/prompt_objects/environment.rb', line 170

def self.from_environment(name, manager: nil)
  manager ||= Env::Manager.new
  raise Error, "Environment '#{name}' not found" unless manager.environment_exists?(name)

  new(env_path: manager.environment_path(name))
end

Instance Method Details

#active_session(initiated_by: "human") ⇒ Hash

Get or create the active environment session.

Parameters:

  • initiated_by (String) (defaults to: "human") —

    Who initiated the session (default: "human")

Returns:

  • (Hash) —

    Session data



383
384
385
386
 
# File 'lib/prompt_objects/environment.rb', line 383

def active_session(initiated_by: "human")
  @active_session_id ||= find_or_create_environment_session(initiated_by: initiated_by)
  @session_store&.get_session(@active_session_id)
end

#archive_runtime_thread(thread_id) ⇒ Object 



787
788
789
790
791
792
 
# File 'lib/prompt_objects/environment.rb', line 787

def archive_runtime_thread(thread_id)
  thread = required_runtime_thread(thread_id)
  archived = thread_repository.archive(thread.id)
  broadcast_workspace_snapshot(thread.workspace_session_id)
  archived
end

#autonomous_schedule_states ⇒ Object 



985
986
987
988
989
 
# File 'lib/prompt_objects/environment.rb', line 985

def autonomous_schedule_states
  @autonomous_schedule_mutex.synchronize do
    @autonomous_schedule_states.transform_values(&:dup)
  end
end

#broadcast(**message) ⇒ Object

Broadcast a message to all connected WebSocket clients. No-op if on_broadcast callback isn't set (e.g., in tests or CLI mode).

Parameters:

  • message (Hash) —

    WebSocket message to broadcast



994
995
996
997
998
 
# File 'lib/prompt_objects/environment.rb', line 994

def broadcast(**message)
  @on_broadcast&.call(message)
rescue => e
  # Don't crash if broadcast fails
end

#cancel_run(run_id, reason: "cancelled") ⇒ Object 



591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
 
# File 'lib/prompt_objects/environment.rb', line 591

def cancel_run(run_id, reason: "cancelled")
  run = run_repository.get(run_id)
  return false unless run

  cancelled = @run_scheduler&.cancel(run_id, reason: reason)
  unless cancelled
    return false unless run.status == "queued"

    cancel_queued_run(run, reason)
    cancelled = true
  end
  run_repository.active_children(parent_run_id: run_id).each do |child|
    next if metadata_value(child, :child_lifetime) == "detached"

    cancel_run(child.id, reason: "parent cancelled: #{reason}")
  end
  cancelled
end

#capability_policy_version_for_config(config) ⇒ Object 



555
556
557
558
559
560
561
562
 
# File 'lib/prompt_objects/environment.rb', line 555

def capability_policy_version_for_config(config)
  names = Array(config["capabilities"] || config[:capabilities]) + UNIVERSAL_CAPABILITIES
  signatures = names.uniq.sort.map do |name|
    capability = registry.get(name)
    [name, capability&.execution_policy_signature || "missing"]
  end
  Digest::SHA256.hexdigest(JSON.generate(signatures))
end

#context(tui_mode: false) ⇒ Context

Create a context for capability execution.

Parameters:

  • tui_mode (Boolean) (defaults to: false) —

    Whether running in TUI mode

Returns:



262
263
264
265
266
 
# File 'lib/prompt_objects/environment.rb', line 262

def context(tui_mode: false)
  ctx = Context.new(env: self, bus: @bus, human_queue: @human_queue)
  ctx.tui_mode = tui_mode
  ctx
end

#create_environment_session(name: nil, initiated_by: "human") ⇒ Hash

Create a new environment session, making it the active one.

Parameters:

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

    Optional session name

  • initiated_by (String) (defaults to: "human") —

    Who initiated (human, timer, api, etc.)

Returns:

  • (Hash) —

    Session data



404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
 
# File 'lib/prompt_objects/environment.rb', line 404

def create_environment_session(name: nil, initiated_by: "human")
  return nil unless @session_store

  id = SecureRandom.uuid
  workspace_repository.create(id: id, name: name)
  id = @session_store.create_session(
    id: id,
    po_name: nil,
    name: name,
    source: "web",
    thread_type: "root",
    workspace_session_id: id
  )
  @active_session_id = id
  @session_store.get_session(id)
end

#create_runtime_thread(workspace_session_id:, po_name:, thread_type: "human", name: nil, parent_thread_id: nil, parent_run_id: nil) ⇒ Object

Raises:



760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
 
# File 'lib/prompt_objects/environment.rb', line 760

def create_runtime_thread(workspace_session_id:, po_name:, thread_type: "human", name: nil,
                          parent_thread_id: nil, parent_run_id: nil)
  workspace_id = resolve_workspace_id(workspace_session_id, initiated_by: "human")
  po = @registry.get(po_name)
  raise Error, "Unknown prompt object: #{po_name}" unless po.is_a?(PromptObject)

  thread = thread_repository.create(
    workspace_session_id: workspace_id,
    po_name: po_name,
    thread_type: thread_type,
    name: name,
    parent_thread_id: parent_thread_id,
    parent_run_id: parent_run_id,
    retention_policy: thread_type == "autonomous" ? "summary" : "full"
  )
  broadcast_workspace_snapshot(workspace_id)
  broadcast(**thread_snapshot_envelope(thread))
  thread
end

#delete_environment_data(key:, context:) ⇒ Object 



923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
 
# File 'lib/prompt_objects/environment.rb', line 923

def delete_environment_data(key:, context:)
  workspace_id = environment_data_workspace_id(context)
  stored_by = context.calling_po || (context.respond_to?(:target_po) ? context.target_po : nil) || "unknown"
  unless workspace_repository.get(workspace_id)
    deleted = @session_store.delete_env_data(root_thread_id: workspace_id, key: key)
    if deleted
      notify_env_data_changed(
        action: "delete", root_thread_id: workspace_id, key: key, stored_by: stored_by
      )
    end
    return deleted
  end
  removal = env_data_repository.delete_with_cursor(
    workspace_session_id: workspace_id,
    key: key,
    stored_by: stored_by
  )
  return false unless removal.deleted

  publish_env_data_snapshot(workspace_id, cursor: removal.cursor)
  publish_environment_data_run_event(context, action: "delete", key: key, stored_by: stored_by)
  notify_env_data_changed(
    action: "delete",
    root_thread_id: workspace_id,
    key: key,
    stored_by: stored_by
  )
  true
end

#delete_environment_session(session_id) ⇒ Boolean

Delete an environment session (and its messages). If it was active, the next active_session call will find or create another.

Returns:

  • (Boolean) 


460
461
462
463
464
465
466
 
# File 'lib/prompt_objects/environment.rb', line 460

def delete_environment_session(session_id)
  return false unless @session_store && @session_store.get_session(session_id)

  @session_store.delete_session(session_id)
  @active_session_id = nil if @active_session_id == session_id
  true
end

#dirty? ⇒ Boolean

Check if there are unsaved changes.

Returns:

  • (Boolean) 


200
201
202
203
204
 
# File 'lib/prompt_objects/environment.rb', line 200

def dirty?
  return false unless environment?

  Env::Git.dirty?(@env_path)
end

#dispatch_run(target_po:, content:, context:, thread_id: nil, lifetime: "attached", on_complete: nil) ⇒ Object

Raises:



633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
 
# File 'lib/prompt_objects/environment.rb', line 633

def dispatch_run(target_po:, content:, context:, thread_id: nil, lifetime: "attached", on_complete: nil)
  raise Error, "Dispatch requires a run context" unless context.run_id
  raise Error, "lifetime must be attached or detached" unless %w[attached detached].include?(lifetime)

  parent = run_repository.get(context.run_id)
  raise Error, "Unknown parent run: #{context.run_id}" unless parent
  validate_dispatch_target!(parent.po_name, target_po)
  target = registry.get(target_po)
  definition = execution_definition_for(target)
  thread = nil
  child = nil
  with_child_run_admission(parent) do |depth|
    thread = if thread_id
               required_dispatch_thread(thread_id, parent, target_po)
             else
               thread_repository.create(
                 workspace_session_id: parent.workspace_session_id,
                 po_name: target_po,
                 thread_type: "delegation",
                 parent_thread_id: parent.thread_id,
                 parent_run_id: parent.id,
                 retention_policy: "full"
               )
             end
    child = run_repository.enqueue(
      workspace_session_id: parent.workspace_session_id,
      thread_id: thread.id,
      po_name: target_po,
      input_content: content,
      definition_version: definition.version,
      provider: definition.provider,
      model: definition.model,
      capability_policy_version: definition.capability_policy_version,
      source_type: "delegation",
      source_name: parent.po_name,
      parent_run_id: parent.id,
      priority: 80,
      metadata: {
        dispatch_mode: "dispatch",
        child_lifetime: lifetime,
        delegation_depth: depth,
        deadline: context.deadline&.iso8601(6),
        on_complete: on_complete
      }
    )
  end
  definition_repository.attach(run_id: child.id, definition: definition)
  publish_workspace_snapshot(parent.workspace_session_id)
  publish_run_event(event_repository.append_run_event(
    run_id: child.id,
    kind: "run_queued",
    data: { source_type: child.source_type, turn_ordinal: child.turn_ordinal, mode: "dispatch" }
  ))
  publish_run_event(event_repository.append_run_event(
    run_id: parent.id,
    kind: "delegation_started",
    data: {
      child_run_id: child.id,
      child_thread_id: thread.id,
      target_po: target_po,
      mode: "dispatch",
      lifetime: lifetime
    }
  ))
  schedule_run(child.id, tui_mode: context.tui_mode)
  { run_id: child.id, thread_id: thread.id, status: child.status, lifetime: lifetime }
end

#enforce_retention ⇒ Object 



753
754
755
756
757
758
 
# File 'lib/prompt_objects/environment.rb', line 753

def enforce_retention
  Operations::Retention.new(@session_store).prune_events!(
    run_event_limit: @scheduler_options.fetch(:run_event_retention),
    workspace_event_limit: @scheduler_options.fetch(:workspace_event_retention)
  )
end

#enqueue_run(target_po:, content:, from_source: "human", workspace_session_id: nil, thread_id: nil, client_request_id: nil, deadline: nil) ⇒ Object

Raises:



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
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
 
# File 'lib/prompt_objects/environment.rb', line 491

def enqueue_run(target_po:, content:, from_source: "human", workspace_session_id: nil,
                thread_id: nil, client_request_id: nil, deadline: nil)
  po = @registry.get(target_po)
  raise Error, "Unknown prompt object: #{target_po}" unless po.is_a?(PromptObject)

  workspace_id = resolve_workspace_id(workspace_session_id, initiated_by: from_source)
  thread = resolve_run_thread(
    workspace_id: workspace_id,
    po: po,
    from_source: from_source,
    thread_id: thread_id
  )
  # enqueue_run is the canonical entry point used by WebSocket, REST, CLI,
  # connectors, and inject_message. Arm wait-for-human schedules here so a
  # valid human run wakes autonomous POs regardless of which surface sent it.
  @human_message_received[target_po] = true if from_source == "human"
  definition = execution_definition_for(po)
  source_type = source_type_for(from_source)
  if source_type == "autonomous"
    active = run_repository.active_for_source(
      workspace_session_id: workspace_id,
      po_name: po.name,
      source_type: source_type,
      source_name: from_source
    )
    return active if active
  end
  normalized_deadline = normalize_deadline(deadline)
  run, created = run_repository.enqueue_with_result(
    workspace_session_id: workspace_id,
    thread_id: thread.id,
    po_name: po.name,
    input_content: content,
    definition_version: definition.version,
    provider: definition.provider,
    model: definition.model,
    capability_policy_version: definition.capability_policy_version,
    source_type: source_type,
    source_name: from_source,
    client_request_id: client_request_id,
    priority: priority_for(from_source),
    metadata: { deadline: normalized_deadline&.iso8601(6) }.compact
  )
  definition_repository.attach(run_id: run.id, definition: definition) unless definition_repository.get(run.id)
  if created
    broadcast_workspace_snapshot(workspace_id)
    publish_run_event(event_repository.append_run_event(
      run_id: run.id,
      kind: "run_queued",
      data: { source_type: run.source_type, turn_ordinal: run.turn_ordinal }
    ))
    broadcast_po_runtime_status(run)
  end
  run
end

#env_data_scope(session_id = nil) ⇒ String?

Resolve the env_data scope id for a given session, preferring the environment-level session (v8). Used by server-side readers (REST API, WebSocket, broadcasts) so they read from the same scope writers use.

Parameters:

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

    A session to fall back to if no env session

Returns:

  • (String, nil) —

    Scope id



393
394
395
396
397
398
 
# File 'lib/prompt_objects/environment.rb', line 393

def env_data_scope(session_id = nil)
  return @active_session_id if @active_session_id
  return nil unless @session_store && session_id

  @session_store.resolve_root_thread(session_id)
end

#environment? ⇒ Boolean

Check if this runtime has a persisted environment directory.

Returns:

  • (Boolean) 


185
186
187
 
# File 'lib/prompt_objects/environment.rb', line 185

def environment?
  !@env_path.nil?
end

#environment_sessions ⇒ Array<Hash>

List the environment-level sessions (the shared-context "jams") with a flag marking which one is active. Side-effect-free: reads the already- resolved @active_session_id rather than calling active_session (which would create a session on read), and queries the store once.

Returns:

  • (Array<Hash>) 


426
427
428
429
430
431
432
433
 
# File 'lib/prompt_objects/environment.rb', line 426

def environment_sessions
  return [] unless @session_store

  active = @active_session_id
  @session_store.list_environment_sessions.map do |s|
    s.merge(active: s[:id] == active)
  end
end

#execute_message_run(target_po:, content:, from_source:, workspace_session_id: nil, thread_id: nil, client_request_id: nil, deadline: nil, tui_mode: false) ⇒ Object 



576
577
578
579
580
581
582
583
584
585
586
587
588
589
 
# File 'lib/prompt_objects/environment.rb', line 576

def execute_message_run(target_po:, content:, from_source:, workspace_session_id: nil,
                        thread_id: nil, client_request_id: nil, deadline: nil, tui_mode: false)
  run = enqueue_run(
    target_po: target_po,
    content: content,
    from_source: from_source,
    workspace_session_id: workspace_session_id,
    thread_id: thread_id,
    client_request_id: client_request_id,
    deadline: deadline
  )
  response = execute_run(run.id, tui_mode: tui_mode)
  [run_repository.get(run.id), response]
end

#execute_run(run_id, tui_mode: true) ⇒ Object 



564
565
566
567
568
569
570
571
572
573
574
 
# File 'lib/prompt_objects/environment.rb', line 564

def execute_run(run_id, tui_mode: true)
  run = run_repository.get(run_id)
  raise Error, "Unknown run: #{run_id}" unless run
  return run.final_response if run.status == "completed"
  return :queued unless run.status == "queued"

  run_scheduler.submit(run, deadline: run_deadline(run), tui_mode: tui_mode).wait
rescue Execution::Scheduler::RejectedError => e
  reject_queued_run(run, e.message)
  raise Error, e.message
end

#execution_definition_for(po) ⇒ Object 



547
548
549
550
551
552
553
 
# File 'lib/prompt_objects/environment.rb', line 547

def execution_definition_for(po)
  po.execution_definition(
    provider: current_provider,
    model: current_model,
    capability_policy_version: capability_policy_version_for_config(po.config)
  )
end

#get(name) ⇒ Capability?

Get a capability by name (prompt object or primitive).

Parameters:

  • name (String) —

    The capability name

Returns:



362
363
364
 
# File 'lib/prompt_objects/environment.rb', line 362

def get(name)
  @registry.get(name)
end

#get_environment_data(key:, context:) ⇒ Object 



905
906
907
908
909
910
911
912
 
# File 'lib/prompt_objects/environment.rb', line 905

def get_environment_data(key:, context:)
  workspace_id = environment_data_workspace_id(context)
  if workspace_repository.get(workspace_id)
    env_data_repository.get(workspace_session_id: workspace_id, key: key)
  else
    @session_store.get_env_data(root_thread_id: workspace_id, key: key)
  end
end

#human_message_received?(po_name) ⇒ Boolean

Whether a PO has received a new human message during this runtime boot. Used by opt-in autonomous schedules that must not preempt initialization.

Returns:

  • (Boolean) 


969
970
971
 
# File 'lib/prompt_objects/environment.rb', line 969

def human_message_received?(po_name)
  !!@human_message_received[po_name]
end

#inject_message(target_po:, content:, from_source: "human", workspace_session_id: nil, thread_id: nil, client_request_id: nil, deadline: nil) ⇒ String

Inject a message through the run scheduler. All sources share this entry point and differ only in their provenance and source-specific thread.

Parameters:

  • target_po (String) —

    Name of the PO to receive the message

  • content (String) —

    Message content

  • from_source (String) (defaults to: "human") —

    Message source (human, timer:name, service:name, api)

Returns:

  • (String) —

    The PO's response

Raises:



475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
 
# File 'lib/prompt_objects/environment.rb', line 475

def inject_message(target_po:, content:, from_source: "human",
                   workspace_session_id: nil, thread_id: nil, client_request_id: nil, deadline: nil)
  po = @registry.get(target_po)
  raise Error, "Unknown prompt object: #{target_po}" unless po.is_a?(PromptObject)

  inject_run_message(
    po: po,
    content: content,
    from_source: from_source,
    workspace_session_id: workspace_session_id,
    thread_id: thread_id,
    client_request_id: client_request_id,
    deadline: deadline
  )
end

#inject_run_message(po:, content:, from_source:, workspace_session_id: nil, thread_id: nil, client_request_id: nil, deadline: nil) ⇒ Object 



953
954
955
956
957
958
959
960
961
962
963
964
965
 
# File 'lib/prompt_objects/environment.rb', line 953

def inject_run_message(po:, content:, from_source:, workspace_session_id: nil,
                       thread_id: nil, client_request_id: nil, deadline: nil)
  run = enqueue_run(
    target_po: po.name,
    content: content,
    from_source: from_source,
    workspace_session_id: workspace_session_id,
    thread_id: thread_id,
    client_request_id: client_request_id,
    deadline: deadline
  )
  execute_run(run.id)
end

#join_run(run_id, context:, timeout: nil) ⇒ Object 



710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
 
# File 'lib/prompt_objects/environment.rb', line 710

def join_run(run_id, context:, timeout: nil)
  run = authorized_related_run(run_id, context)
  unless terminal_run_status?(run.status)
    ticket = run_scheduler.ticket(run.id)
    ticket ||= schedule_run(run.id, tui_mode: context.tui_mode)
    ticket = run_scheduler.help(run.id) || ticket
    ticket&.wait(timeout: timeout)
    run = run_repository.get(run.id)
  end
  {
    run_id: run.id,
    thread_id: run.thread_id,
    status: run.status,
    response: run.final_response,
    error: run.error_message
  }
rescue Timeout::Error
  { run_id: run_id, status: "waiting", error: "join timed out" }
end

#join_runs(run_ids, context:, timeout: nil) ⇒ Object 



701
702
703
 
# File 'lib/prompt_objects/environment.rb', line 701

def join_runs(run_ids, context:, timeout: nil)
  Array(run_ids).map { |run_id| join_run(run_id, context: context, timeout: timeout) }
end

#list_environment_data(context:) ⇒ Object 



914
915
916
917
918
919
920
921
 
# File 'lib/prompt_objects/environment.rb', line 914

def list_environment_data(context:)
  workspace_id = environment_data_workspace_id(context)
  if workspace_repository.get(workspace_id)
    env_data_repository.list(workspace_session_id: workspace_id)
  else
    @session_store.list_env_data_full(root_thread_id: workspace_id)
  end
end

#llm_config ⇒ Hash

Get current LLM configuration.

Returns:

  • (Hash) 


234
235
236
237
238
239
240
241
 
# File 'lib/prompt_objects/environment.rb', line 234

def llm_config
  {
    provider: @current_provider,
    model: @current_model,
    providers: LLM::Factory.providers,
    available: LLM::Factory.available_providers
  }
end

#load_by_name(name) ⇒ PromptObject

Load a prompt object by name from the objects directory.

Parameters:

  • name (String) —

    Name of the prompt object (without .md extension)

Returns:



337
338
339
340
 
# File 'lib/prompt_objects/environment.rb', line 337

def load_by_name(name)
  path = File.join(@objects_dir, "#{name}.md")
  load_prompt_object(path)
end

#load_custom_connectors ⇒ Object

Load connectors from the environment's connectors/ directory. A connector is a Connectors::Base subclass that bridges an external system (Discord, Slack, a webhook listener, …) to POs by calling inject_message. Loaded like services; started/stopped with the server.



1620
1621
1622
1623
1624
1625
1626
1627
1628
1629
1630
1631
1632
1633
 
# File 'lib/prompt_objects/environment.rb', line 1620

def load_custom_connectors
  return unless @connectors_dir && Dir.exist?(@connectors_dir)

  Dir.glob(File.join(@connectors_dir, "*.rb")).sort.each do |path|
    begin
      class_name = File.basename(path, ".rb").split("_").map(&:capitalize).join
      load(path)
      klass = PromptObjects::Connectors.const_get(class_name)
      @connectors << klass.new(runtime: self)
    rescue SyntaxError, LoadError, StandardError => e
      warn "Warning: Failed to load connector #{path}: #{e.message}"
    end
  end
end

#load_dependencies(capability) ⇒ Object

Load all prompt objects that a capability depends on.

Parameters:

  • capability (PromptObject) —

    The capability to load dependencies for



344
345
346
347
348
349
350
351
352
353
354
355
356
357
 
# File 'lib/prompt_objects/environment.rb', line 344

def load_dependencies(capability)
  return unless capability.is_a?(PromptObject)

  deps = capability.config["capabilities"] || []
  deps.each do |dep_name|
    next if @registry.exists?(dep_name)

    # Try to load as a prompt object
    path = File.join(@objects_dir, "#{dep_name}.md")
    if File.exist?(path)
      load_prompt_object(path)
    end
  end
end

#load_prompt_object(path) ⇒ PromptObject

Load a prompt object from a file path.

Parameters:

  • path (String) —

    Path to the .md file

Returns:



271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
 
# File 'lib/prompt_objects/environment.rb', line 271

def load_prompt_object(path)
  data = Loader.load(path)

  po = PromptObject.new(
    config: data[:config],
    body: data[:body],
    env: self,
    llm: @llm,
    path: data[:path]
  )

  @registry.register(po)

  # Notify callback if registered (for live updates in web UI)
  @on_po_registered&.call(po)

  po
end

#loaded_objects ⇒ Array<String>

List all loaded prompt objects.

Returns:

  • (Array<String>) 


368
369
370
 
# File 'lib/prompt_objects/environment.rb', line 368

def loaded_objects
  @registry.prompt_objects.map(&:name)
end

#name ⇒ String

Name of the environment (from manifest or directory name).

Returns:

  • (String) 


179
180
181
 
# File 'lib/prompt_objects/environment.rb', line 179

def name
  @manifest&.name || File.basename(@objects_dir)
end

#notify_delegation(event_type, payload) ⇒ Object

Notify that a PO-to-PO delegation has started or completed.

Parameters:

  • event_type (Symbol) —

    :started or :completed

  • payload (Hash) —

    { target:, caller:, thread_id:, tool_call_id: }



321
322
323
 
# File 'lib/prompt_objects/environment.rb', line 321

def notify_delegation(event_type, payload)
  @on_delegation_event&.call(event_type, payload)
end

#notify_env_data_changed(action:, root_thread_id:, key:, stored_by:) ⇒ Object

Notify that environment data has changed (stored, updated, or deleted).

Parameters:

  • action (String) —

    "store", "update", or "delete"

  • root_thread_id (String) —

    Root thread scope

  • key (String) —

    The data key that changed

  • stored_by (String) —

    PO name that made the change



330
331
332
 
# File 'lib/prompt_objects/environment.rb', line 330

def notify_env_data_changed(action:, root_thread_id:, key:, stored_by:)
  @on_env_data_changed&.call(action: action, root_thread_id: root_thread_id, key: key, stored_by: stored_by)
end

#notify_po_modified(po) ⇒ Object

Notify that a PO has been modified (for live updates in web UI). Call this after modifying a PO's config/capabilities programmatically.

Parameters:



293
294
295
 
# File 'lib/prompt_objects/environment.rb', line 293

def notify_po_modified(po)
  @on_po_modified&.call(po)
end

#operational_diagnostics ⇒ Object 



735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
 
# File 'lib/prompt_objects/environment.rb', line 735

def operational_diagnostics
  Operations::Diagnostics.new(@session_store).snapshot(
    scheduler: {
      queued: @run_scheduler&.queued_count || 0,
      running: @run_scheduler&.running_count || 0,
      provider_permits_available: provider_permits.available,
      tool_permits_available: tool_permits.available,
      cpu_permits_available: cpu_permits.available,
      locked_resource_count: resource_lock_manager.locked_keys.length
    },
    limits: @scheduler_options.slice(
      :worker_limit, :provider_limit, :tool_limit, :cpu_limit, :max_active_runs,
      :max_active_per_po, :max_queued_per_thread, :max_delegation_depth, :max_descendants,
      :run_event_retention, :workspace_event_retention
    )
  )
end

#persist_artifact(title:, html:, mode:, context:) ⇒ Object

Persist an artifact before exposing it to clients. The artifact repository is authoritative for identity, provenance, and revision history.



830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
 
# File 'lib/prompt_objects/environment.rb', line 830

def persist_artifact(title:, html:, mode:, context:)
  workspace_id = artifact_workspace_id(context)
  owner_po = context.calling_po || (context.respond_to?(:target_po) ? context.target_po : nil) || "unknown"
  owner_thread_id = artifact_owner_thread_id(context, owner_po)
  producing_run_id = context.respond_to?(:run_id) ? context.run_id : nil
  mutation = artifact_repository.upsert_with_cursor(
    workspace_session_id: workspace_id,
    owner_po: owner_po,
    owner_thread_id: owner_thread_id,
    producing_run_id: producing_run_id,
    title: title,
    html: html,
    mode: mode,
    metadata: {}
  )
  artifact = mutation.artifact

  broadcast(
    version: Execution::Contracts::PROTOCOL_VERSION,
    type: "artifact_upserted",
    payload: {
      cursor: mutation.cursor,
      artifact: serialize_runtime_value(artifact.to_h)
    }
  )
  if producing_run_id
    publish_run_event(event_repository.append_run_event(
      run_id: producing_run_id,
      kind: "artifact_upserted",
      data: {
        artifact_id: artifact.id,
        title: artifact.title,
        revision: artifact.revision,
        owner_thread_id: artifact.owner_thread_id,
        mode: artifact.mode
      }
    ))
  end
  artifact
end

#primitives ⇒ Array<String>

List all registered primitives.

Returns:

  • (Array<String>) 


374
375
376
 
# File 'lib/prompt_objects/environment.rb', line 374

def primitives
  @registry.primitives.map(&:name)
end

#publish_run_event(event) ⇒ Object 



1000
1001
1002
1003
1004
1005
1006
 
# File 'lib/prompt_objects/environment.rb', line 1000

def publish_run_event(event)
  broadcast(
    version: Execution::Contracts::PROTOCOL_VERSION,
    type: "run_event",
    payload: serialize_runtime_value(event.to_h)
  )
end

#publish_thread_snapshot(thread_id, po_name: nil) ⇒ Object

Raises:



821
822
823
824
825
826
 
# File 'lib/prompt_objects/environment.rb', line 821

def publish_thread_snapshot(thread_id, po_name: nil)
  thread = thread_repository.get(thread_id)
  raise Error, "Unknown thread: #{thread_id}" unless thread

  broadcast_thread_snapshot(thread, po_name || thread.po_name)
end

#publish_workspace_snapshot(workspace_session_id) ⇒ Object 



817
818
819
 
# File 'lib/prompt_objects/environment.rb', line 817

def publish_workspace_snapshot(workspace_session_id)
  broadcast_workspace_snapshot(workspace_session_id)
end

#reload_suppressed?(path) ⇒ Boolean

Whether a pending in-app save should suppress the next reload of this path. Consumes the suppression (returns true at most once per registration).

Parameters:

  • path (String)

Returns:

  • (Boolean) 


311
312
313
314
315
316
 
# File 'lib/prompt_objects/environment.rb', line 311

def reload_suppressed?(path)
  return false unless path

  key = File.expand_path(path)
  @suppressed_reloads_mutex.synchronize { !!@suppressed_reloads.delete(key) }
end

#rename_runtime_thread(thread_id, name:) ⇒ Object 



780
781
782
783
784
785
 
# File 'lib/prompt_objects/environment.rb', line 780

def rename_runtime_thread(thread_id, name:)
  thread = required_runtime_thread(thread_id)
  renamed = thread_repository.rename(thread.id, name)
  broadcast_workspace_snapshot(thread.workspace_session_id)
  renamed
end

#rename_session(session_id, name) ⇒ Hash?

Rename an environment session.

Returns:

  • (Hash, nil) —

    The updated session



450
451
452
453
454
455
 
# File 'lib/prompt_objects/environment.rb', line 450

def rename_session(session_id, name)
  return nil unless @session_store && @session_store.get_session(session_id)

  @session_store.update_session(session_id, name: name)
  @session_store.get_session(session_id)
end

#run_status(run_id, context:) ⇒ Object 



705
706
707
708
 
# File 'lib/prompt_objects/environment.rb', line 705

def run_status(run_id, context:)
  run = authorized_related_run(run_id, context)
  serialize_runtime_value(run.to_h)
end

#runtime_snapshot_envelopes(workspace_session_id: nil) ⇒ Object 



794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
 
# File 'lib/prompt_objects/environment.rb', line 794

def runtime_snapshot_envelopes(workspace_session_id: nil)
  workspace_id = resolve_workspace_id(workspace_session_id, initiated_by: "reconnect")
  threads = thread_repository.list(workspace_session_id: workspace_id, include_archived: true)
  active_runs = run_repository.list(
    workspace_session_id: workspace_id,
    statuses: %w[queued running waiting],
    limit: 500
  )
  events = active_runs.flat_map do |run|
    event_repository.run_events(run_id: run.id).map do |event|
      {
        version: Execution::Contracts::PROTOCOL_VERSION,
        type: "run_event",
        payload: serialize_runtime_value(event.to_h)
      }
    end
  end
  [workspace_snapshot_envelope(workspace_id)] +
    threads.map { |thread| thread_snapshot_envelope(thread) } +
    events +
    [env_data_snapshot_envelope(workspace_id)]
end

#save(message = "Save changes") ⇒ Boolean

Save a commit with all current changes.

Parameters:

  • message (String) (defaults to: "Save changes")

Returns:

  • (Boolean) —

    True if committed



192
193
194
195
196
 
# File 'lib/prompt_objects/environment.rb', line 192

def save(message = "Save changes")
  return false unless environment? && @auto_commit

  Env::Git.auto_commit(@env_path, message)
end

#schedule_run(run_id, tui_mode: true) ⇒ Object 



610
611
612
613
614
615
616
617
618
619
 
# File 'lib/prompt_objects/environment.rb', line 610

def schedule_run(run_id, tui_mode: true)
  run = run_repository.get(run_id)
  raise Error, "Unknown run: #{run_id}" unless run
  return nil unless run.status == "queued"

  run_scheduler.submit(run, deadline: run_deadline(run), tui_mode: tui_mode)
rescue Execution::Scheduler::RejectedError => e
  reject_queued_run(run, e.message)
  raise Error, e.message
end

#set_environment_data(key:, short_description:, value:, context:) ⇒ Object 



871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
 
# File 'lib/prompt_objects/environment.rb', line 871

def set_environment_data(key:, short_description:, value:, context:)
  workspace_id = environment_data_workspace_id(context)
  stored_by = context.calling_po || (context.respond_to?(:target_po) ? context.target_po : nil) || "unknown"
  unless workspace_repository.get(workspace_id)
    @session_store.store_env_data(
      root_thread_id: workspace_id,
      key: key,
      short_description: short_description,
      value: value,
      stored_by: stored_by
    )
    notify_env_data_changed(
      action: "set", root_thread_id: workspace_id, key: key, stored_by: stored_by
    )
    return @session_store.get_env_data(root_thread_id: workspace_id, key: key)
  end
  mutation = env_data_repository.set_with_cursor(
    workspace_session_id: workspace_id,
    key: key,
    short_description: short_description,
    value: value,
    stored_by: stored_by
  )
  publish_env_data_snapshot(workspace_id, cursor: mutation.cursor)
  publish_environment_data_run_event(context, action: "set", key: key, stored_by: stored_by)
  notify_env_data_changed(
    action: "set",
    root_thread_id: workspace_id,
    key: key,
    stored_by: stored_by
  )
  mutation.entry
end

#shutdown_execution(wait: true) ⇒ Object 



730
731
732
733
 
# File 'lib/prompt_objects/environment.rb', line 730

def shutdown_execution(wait: true)
  @run_scheduler&.shutdown(wait: wait)
  @run_scheduler = nil
end

#start_connectors ⇒ Object

Start all loaded connectors. Each connector manages its own loop/thread.



1636
1637
1638
1639
1640
1641
1642
 
# File 'lib/prompt_objects/environment.rb', line 1636

def start_connectors
  @connectors.each do |connector|
    connector.start
  rescue => e
    warn "Warning: Failed to start connector #{connector.source_name}: #{e.message}"
  end
end

#start_services ⇒ Object

Start all loaded services by calling their start method. Does NOT schedule tick loops — that's the server's responsibility using Async tasks (see Server.start).



1597
1598
1599
1600
1601
1602
1603
1604
1605
 
# File 'lib/prompt_objects/environment.rb', line 1597

def start_services
  @services.each do |service|
    begin
      service.start
    rescue => e
      warn "Warning: Failed to start service #{service.name}: #{e.message}"
    end
  end
end

#stop_connectors ⇒ Object

Stop all running connectors.



1645
1646
1647
1648
1649
1650
1651
 
# File 'lib/prompt_objects/environment.rb', line 1645

def stop_connectors
  @connectors.each do |connector|
    connector.stop
  rescue => e
    warn "Warning: Failed to stop connector #{connector.source_name}: #{e.message}"
  end
end

#stop_services ⇒ Object

Stop all running services.



1608
1609
1610
1611
1612
1613
1614
 
# File 'lib/prompt_objects/environment.rb', line 1608

def stop_services
  @services.each do |service|
    service.stop
  rescue => e
    warn "Warning: Failed to stop service #{service.name}: #{e.message}"
  end
end

#suppress_next_reload(path) ⇒ Object

Mark a file path as just-written-by-the-app, so the FileWatcher skips the next reload event for it. Idempotent; absolute paths recommended.

Parameters:

  • path (String) 


300
301
302
303
304
305
 
# File 'lib/prompt_objects/environment.rb', line 300

def suppress_next_reload(path)
  return unless path

  key = File.expand_path(path)
  @suppressed_reloads_mutex.synchronize { @suppressed_reloads[key] = true }
end

#switch_active_session(session_id) ⇒ Hash?

Switch the active environment session (changes the shared env_data scope).

Parameters:

  • session_id (String)

Returns:

  • (Hash, nil) —

    The now-active session, or nil if not found



438
439
440
441
442
443
444
445
446
 
# File 'lib/prompt_objects/environment.rb', line 438

def switch_active_session(session_id)
  return nil unless @session_store

  session = @session_store.get_session(session_id)
  return nil unless session

  @active_session_id = session_id
  session
end

#switch_llm(provider:, model: nil) ⇒ Hash

Switch to a different LLM provider/model.

Parameters:

  • provider (String) —

    Provider name (openai, anthropic, gemini)

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

    Model name (defaults to provider's default)

Returns:

  • (Hash) —

    New provider/model info



210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
 
# File 'lib/prompt_objects/environment.rb', line 210

def switch_llm(provider:, model: nil)
  new_provider = provider
  new_model = model || LLM::Factory.default_model(provider)
  new_llm = LLM::Factory.create(provider: new_provider, model: new_model)

  @current_provider = new_provider
  @current_model = new_model
  @llm = new_llm

  # Update all loaded POs to use the new LLM
  @registry.prompt_objects.each do |po|
    po.instance_variable_set(:@llm, @llm)
  end

  if @manifest
    @manifest.llm = { "provider" => @current_provider, "model" => @current_model }
    @manifest.save_to_dir(@env_path)
  end

  { provider: @current_provider, model: @current_model }
end

#update_autonomous_schedule(po_name, **state) ⇒ Object

Publish truthful autonomous scheduling state for reconnectable UI countdowns. The scheduler owns these transitions; clients only render the server-provided next_run_at instead of guessing from page-load time.



976
977
978
979
980
981
982
983
 
# File 'lib/prompt_objects/environment.rb', line 976

def update_autonomous_schedule(po_name, **state)
  payload = @autonomous_schedule_mutex.synchronize do
    current = @autonomous_schedule_states[po_name] || { po_name: po_name }
    @autonomous_schedule_states[po_name] = current.merge(state).freeze
  end
  broadcast(type: "autonomous_schedule", payload: payload)
  payload
end

#update_manifest_stats! ⇒ Object

Update manifest stats from current state.



244
245
246
247
248
249
250
251
252
253
254
255
256
257
 
# File 'lib/prompt_objects/environment.rb', line 244

def update_manifest_stats!
  return unless @manifest

  stats = { po_count: @registry.prompt_objects.size }

  # Add session stats if available
  if @session_store
    stats[:total_sessions] = @session_store.total_sessions
    stats[:total_messages] = @session_store.total_messages
  end

  @manifest.update_stats(**stats)
  @manifest.save_to_dir(@env_path)
end

#with_child_run_admission(parent) ⇒ Object 



621
622
623
624
625
626
627
628
629
630
631
 
# File 'lib/prompt_objects/environment.rb', line 621

def with_child_run_admission(parent)
  @child_admission_mutex.synchronize do
    depth = run_depth(parent) + 1
    raise Error, "Maximum delegation depth exceeded" if depth > @scheduler_options.fetch(:max_delegation_depth)
    if run_repository.descendant_count(root_run_id: parent.root_run_id) >= @scheduler_options.fetch(:max_descendants)
      raise Error, "Maximum run descendant count exceeded"
    end

    yield depth
  end
end