Class: Insika::Executor

Inherits:
Object
  • Object
show all
Defined in:
lib/insika/executor.rb

Overview

Coordinates execution. It does not build context, decide policy, or talk to the provider. This file does NOT require ruby_llm at load-time — the require is lazy inside the chat methods.

Surrounds the stages: spawn, state lifecycle (always via TaskStore), mailbox drain at the boundaries, in-process registration of live fibers (running?), and the emitter with meta + seq.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(context_builder:, policy_engine:, middleware:, hooks:, tool_registry:, skill_catalog:, profiles:, session_store:, task_store:, checkpoint_store:, event_stream:, workflow_registry: nil, pending_action_store: nil, capability_registry: nil, tool_catalog: nil, memory_store: nil, tool_trace_store: nil, settings_store: nil, content_filter_factory: nil, delegation_store: nil, channel_delivery: nil, llm: nil, context_trace_store: nil, reliability: nil) ⇒ Executor

Returns a new instance of Executor.



16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
 
# File 'lib/insika/executor.rb', line 16

def initialize(context_builder:, policy_engine:, middleware:, hooks:,
               tool_registry:, skill_catalog:, profiles:,
               session_store:, task_store:, checkpoint_store:,
               event_stream:, workflow_registry: nil, pending_action_store: nil,
               capability_registry: nil, tool_catalog: nil, memory_store: nil,
               tool_trace_store: nil, settings_store: nil, content_filter_factory: nil,
                delegation_store: nil, channel_delivery: nil, llm: nil,
                context_trace_store: nil, reliability: nil)
  @context_builder = context_builder
  @policy_engine = policy_engine
  @middleware = middleware
  @hooks = hooks
  @tool_registry = tool_registry
  @skill_catalog = skill_catalog
  # Legacy Hash -> StaticProfileSource; a ProfileSource passes through.
  @profiles = ProfileSource.coerce(profiles)
  @session_store = session_store
  @task_store = task_store
  @checkpoint_store = checkpoint_store
  @event_stream = event_stream
  @workflow_registry = workflow_registry # stage 6 of trigger_workflow
  @pending_action_store = pending_action_store # approval gate
  @capability_registry = capability_registry # capability resolution (nil = off)
  @tool_trace_store = tool_trace_store # tool-call trace for Studio debugging (nil = off)
  # per-turn context breakdown (tokens by category + budget) for the
  # Studio session card. nil = off (no record, zero overhead — parity).
  @context_trace_store = context_trace_store
  # Guardrails output filter: ->(state) { OutputFilter | nil }.
  # Injected by the Safety::Factory; nil = off (parity — the stream is untouched).
  # The INPUT guardrail is a Middleware (in the stack, not here); this is the seam
  # for the stream-side redaction the Executor owns.
  @content_filter_factory = content_filter_factory
  # durable record of ASYNC delegations. nil = async
  # delegation OFF (only the synchronous spawn_subagent works — parity). When
  # present, run_subagent(async: true) dispatches + returns immediately and the
  # child's result is delivered to the parent session as a NEW turn on completion.
  @delegation_store = delegation_store
  # outbound delivery for Shape B channels. nil = no channel
  # delivers out of band (parity — every surface today answers on the request's
  # own connection). When present, a turn that CAME IN through a channel writes
  # its answer to the outbox at the terminal and the dispatcher POSTs it.
  @channel_delivery = channel_delivery
  # the chat FACTORY this executor asks — a RubyLLM::Context (an
  # isolated config dup) when the graph owns its credentials, nil = the
  # process-wide RubyLLM constant (the historic single-graph deployment).
  # Duck-typed: Context#chat and RubyLLM.chat take the same keywords.
  @llm = llm
  # stability for the turn's single agent interaction (WS3): retries /
  # fallback / circuit breaker, all DATA on AgentProfile#reliability.
  # nil = the plain single ask (parity).
  @reliability = reliability
  # LLM config v2: resolves the model at turn start (Chat > Agent >
  # platform default) + model_policy + fallback chain. settings_store nil =
  # no platform layer (pre-v2 behavior: the agent's own model is used as-is).
  @model_resolver = ModelResolver.new(settings_store: settings_store)
  # the platform layer of the queue policy (nil = per-agent and
  # defaults only, which is `followup` with no window — today's behavior).
  @settings_store = settings_store
  # RubyLLM glue (stages 5-7): chat assembly delegated to ChatBuilder. Its
  # optional deps tool_catalog (Tool Search) and memory_store (cross-session
  # memory) matter only to it — nil = parity (deferred
  # not partitioned; no system remember).
  @chat_builder = ChatBuilder.new(
    tool_registry: tool_registry, skill_catalog: skill_catalog,
    checkpoint_store: checkpoint_store, event_stream: event_stream, hooks: hooks,
    tool_catalog: tool_catalog, memory_store: memory_store,
    # load_skill is not enveloped, so it records its own trace entry.
    tool_trace_store: tool_trace_store,
    # the ChatBuilder wires the spawn_subagent system tool (gated by
    # profile.subagents) and hands it this Executor as the runner. `self` is not
    # yet fully built here, but the ChatBuilder only STORES it (used per-turn).
    subagent_runner: self
  )
  # Stage-3-tail tool assembly (capability resolution, instantiation,
  # injection, dedup join, ToolEnvelope wrap) — extracted collaborator.
  @tool_assembly = ToolAssembly.new(
    tool_registry: tool_registry, capability_registry: capability_registry,
    event_stream: event_stream, checkpoint_store: checkpoint_store,
    tool_trace_store: tool_trace_store
  )
  @running = {}            # task_id => TaskActor (live fibers in this process)
  @seqs = Hash.new(0)      # monotonic counter per task
  @supervised = false      # serving mode? — see #turn_parent
  @supervisor = nil        # lazy long-lived supervisor (created when serving)
  @session_actors = {}     # session_id => SessionActor (FIFO queue)
  @draining = false        # shutdown drain — see #begin_drain!
end

Instance Attribute Details

#alert_dispatcher ⇒ Object

The WS6 alert dispatcher: answers budget_warning / breaker_open / delivery_failed with a durable webhook delivery. Started as a child of the turn supervisor in serving mode (like the tick); nil = no alerts.



125
126
127
 
# File 'lib/insika/executor.rb', line 125

def alert_dispatcher
  @alert_dispatcher
end

#supervised ⇒ Object

Turns on SERVING mode: the composition root's serving arm (serve.rb / config.ru) sets true AFTER recovery. Under HTTP spawn runs on the request's EPHEMERAL fiber; without this the turn would be its child and the runtime would CANCEL it on disconnect (violates the contract: "execution belongs to the runtime, not the connection"). When on, the turn is born a child of a long-lived supervisor (sibling of the accept loop) and outlives the connection. false (default) = parents on the current fiber: at recovery/boot and in tests the owner WANTS to wait for the turn to finish (structured concurrency).



113
114
115
 
# File 'lib/insika/executor.rb', line 113

def supervised
  @supervised
end

#task_store ⇒ Object (readonly)

The SessionActor writes the merged fragment through this (it has no store of its own, by design — it owns scheduling, not persistence).



352
353
354
 
# File 'lib/insika/executor.rb', line 352

def task_store
  @task_store
end

#tick ⇒ Object

the periodic tick (outbox drain + stale recovery sweep), wired by the graph AFTER the bus exists (the tick's recovery half dispatches through it). nil = no tick (parity — recovery stays boot-only). When present and serving, it starts as a child of the turn supervisor (see #turn_parent).



120
121
122
 
# File 'lib/insika/executor.rb', line 120

def tick
  @tick
end

Instance Method Details

#approve(task_id) ⇒ Object

ApproveAction's access point: WAKES the turn suspended in :waiting by posting :approval on the live fiber. The decision was already written to the store by the handler BEFORE this post (request_approval re-reads it from the store). No-op if there is no live fiber (process crashed) — recovery re-executes and uses the durable decision. Returns whether there was a fiber.



178
179
180
181
182
 
# File 'lib/insika/executor.rb', line 178

def approve(task_id)
  actor = @running[task_id]
  actor&.post(:approval)
  !actor.nil?
end

#begin_drain! ⇒ Object

closes the TURN intake for shutdown. Armed by Insika::Shutdown when the process is asked to stop: from here on a new top-level turn is left :queued (durable — the next boot's recovery replays it) instead of spawning, while the in-flight turns run to their natural end. One-way by design: a draining process never takes work again.



132
133
134
 
# File 'lib/insika/executor.rb', line 132

def begin_drain!
  @draining = true
end

#cancel(task_id) ⇒ Object

CancelTask's access point: posts :cancel if there is a live fiber. Idempotent no-op if there is none (terminal/orphan). Returns whether there was a fiber.



147
148
149
150
151
 
# File 'lib/insika/executor.rb', line 147

def cancel(task_id)
  actor = @running[task_id]
  actor&.post(:cancel)
  !actor.nil?
end

#collect_into_pending(session_id, text, profile:) ⇒ Object

the collect door, asked BEFORE a task is created. -> the task id the fragment joined, or nil (create a task and spawn as usual).

Asking first is what keeps the store clean: creating a task and then discarding it would leave an orphan :queued record for every fragment, and queued is what Recovery replays at boot.



268
269
270
271
272
273
274
275
276
277
278
 
# File 'lib/insika/executor.rb', line 268

def collect_into_pending(session_id, text, profile:)
  return nil unless @supervised && session_id

  policy = queue_policy(profile, session_id)
  return nil unless policy.collect? && policy.debounce?

  actor = @session_actors[session_id]
  return nil unless actor&.alive?

  actor.collect(text)
end

#draining? ⇒ Boolean

Returns:

  • (Boolean) 


136
 
# File 'lib/insika/executor.rb', line 136

def draining? = @draining

#emit_coalesced(task, merged:, arrivals: []) ⇒ Object

Emitted by the SessionActor when a window closes having merged more than one fragment. arrivals are the ISO8601 times each fragment landed — the ONLY record that they were separate messages, since a merged fragment creates no task of its own. Ids and times, never content.



358
359
360
 
# File 'lib/insika/executor.rb', line 358

def emit_coalesced(task, merged:, arrivals: [])
  emit(:turn_coalesced, { task_id: task.id, merged: merged, arrivals: arrivals }, task: task)
end

#execute(task, profile:, actor:, resume_from: nil) ⇒ Object

Stages 2..9. Runs INSIDE the task's fiber.



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
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
 
# File 'lib/insika/executor.rb', line 508

def execute(task, profile:, actor:, resume_from: nil)
  # Resume of a crash orphan: the interrupted attempt's Execution was left OPEN
  # (the fiber died). The TaskStore forbids opening a second one while one is
  # open -> close the orphan as :interrupted before opening the N+1 (a new
  # entry, never overwrites).
  close_orphan_execution(task) if resume_from
  @task_store.begin_execution(task.id) # attempt N+1
  # queued (normal spawn) and paused/waiting (resume) -> running. An orphan is
  # already :running (running->running is invalid) -> no transition.
  status = @task_store.find(task.id).status
  @task_store.transition(task.id, to: :running) if %i[queued paused waiting].include?(status)
  emit(:task_started, started_data(task, profile), task: task)

  actor.drain!
  run_pipeline(task, profile, actor, resume_from)
# SINGLE capture at the top of the fiber: a single place maps
# error -> terminal state -> events. Stages do no rescue of their own
# (except tool, RubyLLM semantics). The fiber NEVER re-raises.
rescue CancelledError
  # cancel is not an error: transition WITHOUT error: (does not close the
  # Execution), then finish_execution closes it with outcome :cancelled.
  @task_store.transition(task.id, to: :cancelled)
  @task_store.finish_execution(task.id, outcome: :cancelled)
  emit(:task_cancelled, { task_id: task.id }, task: task)
rescue PolicyDenied => e
  emit(:policy_denied, { policy: e.policy, reason: e.reason }, task: task)
  fail_task(task, e, stage: :policy)
rescue BudgetExceeded => e
  # WS2 hard budget: a typed, retryable failure — the envelope reads
  # budget_exceeded + retry_after (window roll), never a silent drop.
  fail_task(task, e, stage: :budget)
rescue CircuitOpenError => e
  # WS3 breaker: the turn died BEFORE the provider call — the envelope
  # reads circuit_open + retry_after (cooldown remaining). Worth its own
  # stage: an open breaker is a reliability decision, not an error bug.
  fail_task(task, e, stage: :reliability)
rescue Insika::WorkflowSchemaError => e
  # a workflow OUTPUT that violates its output_schema. Distinct
  # stage so a contract breach is not conflated with an :unknown failure. (INPUT
  # is validated synchronously in TriggerWorkflow -> 422, never reaches here.)
  fail_task(task, e, stage: :workflow_schema)
rescue ContextError => e
  fail_task(task, e, stage: :context)
rescue CapabilityError => e
  fail_task(task, e, stage: :capability)
rescue ProviderError => e
  fail_task(task, e, stage: :ruby_llm)
rescue StoreError => e
  fail_task(task, e, stage: :persistence)
rescue TimeoutError => e
  fail_task(task, e, stage: e.stage)
rescue StandardError => e
  # A provider/transport failure is NOT an :unknown bug: wrap it with its
  # action classification (B9) so the envelope can quote retryable and the
  # provider's own retry_after (A8). The classifier is class-name based —
  # the :ruby_llm stage stays reachable even under the smoke-shim's fake.
  if ProviderErrorClassifier.provider_error?(e)
    fail_task(task, ProviderErrorClassifier.wrap(e), stage: :ruby_llm)
  else
    fail_task(task, e, stage: :unknown)
  end
ensure
  @running.delete(task.id) # ALWAYS deregister (a false-positive running? would break the resume)
  # Deregistered FIRST on purpose: from here on the `steer` door finds no actor for
  # this session and answers nil, so a message arriving during the release becomes
  # its own turn instead of a post into a mailbox nobody reads again.
  release_steered(task, profile, actor)
end

#in_flight ⇒ Object

The turns still running in THIS process — what a drain waits on.



139
 
# File 'lib/insika/executor.rb', line 139

def in_flight = @running.keys

#interrupt_running(session_id, profile:, replaced_by: nil) ⇒ Object

the interrupt door: the turn in flight is answering a question the customer has already replaced, so it is abandoned and the new message becomes an ordinary turn. -> the abandoned task's id, or nil (nothing was running).

Unlike collect/steer this one JOINS nothing: replaced_by already has its own task and its own reply, so there is no verdict to report and every surface can use it, /v1/responses included.

What "abandon" means here is Insika's existing cancellation semantics, unchanged: :cancel is observed only at a stage boundary, so a tool call in flight runs to completion and is recorded. Cancelling the not-yet-started calls of a batch would leave it half applied, and fabricating failure results would teach the model that tools failed when they did not (records the same boundary for turn_timeout).



331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
 
# File 'lib/insika/executor.rb', line 331

def interrupt_running(session_id, profile:, replaced_by: nil)
  return nil unless @supervised && session_id

  return nil unless queue_policy(profile, session_id).interrupt?

  session_actor = @session_actors[session_id]
  return nil unless session_actor&.alive?

  task = session_actor.current_task
  return nil if task.nil?

  actor = @running[task.id]
  return nil if actor.nil?

  actor.post(:cancel)
  emit(:turn_interrupted, { task_id: task.id, replaced_by: replaced_by }, task: task)
  task.id
end

#pause(task_id) ⇒ Object

PauseTask's access point: posts :pause if there is a live fiber; the turn suspends at the next boundary (drain_and_maybe_suspend). Idempotent no-op if there is none. Returns whether there was a fiber.



156
157
158
159
160
 
# File 'lib/insika/executor.rb', line 156

def pause(task_id)
  actor = @running[task_id]
  actor&.post(:pause)
  !actor.nil?
end

#recover_channel_deliveries ⇒ Object

(boot): re-drives the outbound replies a previous process recorded and never claimed. Records left delivering are NOT swept — that process may have POSTed before it died, and re-sending is the duplicate the claim exists to prevent. No-op without a channel_delivery. -> { dispatched: [ids] }



501
502
503
504
505
 
# File 'lib/insika/executor.rb', line 501

def recover_channel_deliveries
  return { dispatched: [] } unless @channel_delivery

  @channel_delivery.sweep
end

#recover_delegations ⇒ Object

(boot): reconciles ASYNC delegations after a crash so a completed child's result is never lost. For each undelivered Delegation:

· child TERMINAL, not captured -> capture + deliver.
· child TERMINAL, captured (completed) -> deliver (crash before delivery).
· child NOT terminal -> left as-is; the normal task Recovery resumes the
child and its terminal hook (finalize_delegation) delivers when it finishes.

Called once at boot AFTER the task Recovery. No-op without a delegation_store. -> { delivered: [ids] }



482
483
484
485
486
487
488
489
490
491
492
493
494
 
# File 'lib/insika/executor.rb', line 482

def recover_delegations
  return { delivered: [] } unless @delegation_store

  delivered = []
  @delegation_store.undelivered.each do |deleg|
    child = @task_store.find(deleg.child_task_id)
    next unless child && TERMINAL_STATUSES.include?(child.status)

    finalize_delegation(child) # capture (if needed) + claim + deliver
    delivered << deleg.id
  end
  { delivered: delivered }
end

#release_steered(task, profile, actor) ⇒ Object

a steered message the run could NOT absorb: no tool batch ever closed (a text-only turn), the batch ended in halt_when, the turn failed, or it was cancelled. The message is a person's and must not evaporate, so it is released as the next turn on this session — which is followup, arrived at late.

Runs in execute's ensure, on the dying turn's fiber: spawn_in_session only enqueues, and the session loop is still awaiting THIS turn, so the follow-up runs after it, in order.



370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
 
# File 'lib/insika/executor.rb', line 370

def release_steered(task, profile, actor)
  # Cheap gate first: a turn nobody steered pays one integer read and no store round
  # trip, which is every turn on every agent that never turned the mode on.
  return unless actor.user_messages_posted.positive?

  # Only a turn that actually FINISHED releases. A process going down (Async::Stop
  # through this ensure) leaves the task `:running` for Recovery to replay, and
  # spawning a follow-up during shutdown would parent a turn on a supervisor that is
  # already stopping. The honest limitation: a steered message lives in memory until a
  # boundary writes it to the transcript, so a hard stop in that window loses it.
  current = @task_store.find(task.id)
  return unless current && TERMINAL_STATUSES.include?(current.status)

  texts = actor.take_user_messages!
  return if texts.empty?

  command = Insika::Command.build(
    :send_message,
    # No `origin`: a person typed this, which is exactly what an absent origin means.
    { "agent" => profile.id, "message" => texts.join("\n"), "session_id" => task.session_id }
  ).to_h
  follow_up = @task_store.create(command: command, session_id: task.session_id)
  emit(:turn_steer_released, { task_id: task.id, released_as: follow_up.id, count: texts.size },
       task: task)
  spawn_in_session(follow_up, profile: profile)
rescue Insika::Error
  # Best-effort: the turn is already terminal and durable, and a store that refuses
  # here must not turn a committed turn into a failed one. The message is then lost,
  # and the ABSENCE of :turn_steer_released is what says so — there is no half state.
  nil
end

#request_approval(task:, turn:, tool:, args:, actor:) ⇒ Object

Approval gate, called by the ToolEnvelope at stage 6 when the tool requires approval. Creates/queries the PendingAction (deterministic id by task+turn+tool — per-tool correlation as with the side-effect), suspends the turn in :waiting and BLOCKS (await(:approval)) until the operator resolves it via ApproveAction. The AUTHORITATIVE decision comes from the durable store (crash-safe): on a post-crash re-execution, an already-resolved PendingAction is reused without re-suspending; a :pending one re-suspends. -> "approved" | "rejected".



193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
 
# File 'lib/insika/executor.rb', line 193

def request_approval(task:, turn:, tool:, args:, actor:)
  # Fail-closed: requiring approval with nowhere to persist/query the decision
  # is a misconfiguration — fail LOUD, never hang nor auto-approve.
  if @pending_action_store.nil?
    raise Insika::Error, "tool '#{tool}' requires approval but PendingActionStore is not configured"
  end

  id = pending_id(task.id, turn, tool)
  existing = @pending_action_store.find(id)
  return existing.status.to_s if existing && existing.status != :pending # re-execution: already resolved

  unless existing
    @pending_action_store.create(id: id, task_id: task.id, turn: turn, tool: tool, args: args || {})
    emit(:approval_requested, { pending_id: id, tool: tool.to_s, args: args }, task: task)
  end

  @task_store.transition(task.id, to: :waiting) if @task_store.find(task.id).status == :running

  # Awaits the resolution of THIS pending. A spurious :approval (duplicate or
  # from another pending of the same actor) that wakes up before resolution is
  # ignored (re-await) — fail-closed: only the real resolution of this id
  # unblocks.
  status = nil
  loop do
    actor.await(reason: :approval) # blocks (or raises on :cancel/:timeout)
    status = @pending_action_store.find(id)&.status
    break unless status == :pending
  end

  @task_store.transition(task.id, to: :running)
  (status || :rejected).to_s # fail-closed: missing record -> reject (defensive)
end

#resume_live(task_id) ⇒ Object

ResumeTask's access point for a paused IN-PROCESS task: posts :resume on the live fiber (which is blocked in await). Returns whether there was a live fiber — the handler decides between in-process resume and crash re-dispatch based on that.



166
167
168
169
170
 
# File 'lib/insika/executor.rb', line 166

def resume_live(task_id)
  actor = @running[task_id]
  actor&.post(:resume)
  !actor.nil?
end

#run_serial(task, profile:, resume_from: nil) ⇒ Object

Runs ONE turn serially (called by the SessionActor loop): spawns (the turn is born a child of the supervisor, non-blocking) and AWAITS its completion before returning — that is what serializes the session. A turn error is already mapped to a terminal state inside its own fiber (single capture); here we only ensure the session loop does not die.



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

def run_serial(task, profile:, resume_from: nil)
  # a drain that started with turns already queued behind this
  # session's current one must not keep feeding the loop — without this gate
  # the drain would only converge when the whole backlog ran out.
  return defer_turn(task) if @draining

  spawn(task, profile: profile, resume_from: resume_from)
  @running[task.id]&.wait
rescue Async::Stop
  raise # shutdown: propagate (ends the session loop)
rescue StandardError => e
  # A SYNCHRONOUS spawn error (before the fiber): execute's single capture
  # does not act (the turn never ran). Mark :failed here so as NOT to orphan
  # the task as :queued with no terminal state nor event (the client would
  # hang).
  fail_spawn(task, e)
end

#run_subagent(agent:, message:, parent_state:, async: false) ⇒ Object

runs a CHILD agent turn (called by Tools::Subagent during stage 6). Isolated context (fresh child session), capability NON-inheritance (child profile resolved fresh), environment inheritance (model/thinking seeded from the parent). NEVER raises: a bad agent/depth/child failure is a message to the model, not a turn-killer.

async:false (default) — SYNCHRONOUS: runs the child inside the parent's fiber and returns { text:, session_id: } (the child result is the tool result). async:true — DURABLE dispatch: spawns the child NON-blocking, persists a Delegation, returns { dispatched:, agent:, session_id: } immediately; the parent turn ends and the child's result is later delivered as a NEW turn on the parent session (needs a delegation_store — else falls back to sync).



445
446
447
448
449
450
451
452
453
454
 
# File 'lib/insika/executor.rb', line 445

def run_subagent(agent:, message:, parent_state:, async: false)
  plan = plan_subagent({ "agent" => agent, "message" => message }, parent_state)
  return { error: plan[:error] } if plan[:error]

  if async && @delegation_store
    dispatch_async_child(plan[:profile], plan[:message], plan[:depth], parent_state)
  else
    spawn_and_await_child(plan[:profile], plan[:message], plan[:depth], parent_state)
  end
end

#run_subagents(tasks:, parent_state:) ⇒ Object

(fan-out): runs SEVERAL child turns IN PARALLEL and returns all results together, in the requested order. This is the real latency win — the children overlap their provider waits on the reactor, so wall-clock ≈ the slowest child, not the sum. Always sync-join (a combined result in the parent's turn); the async deliver-as-new-turn mode is single-child only, by design. NEVER raises: per-task errors keep their slot; a bad envelope returns { error: }. -> { results: [text:, session_id: | error:, ...] } | { error: }



463
464
465
466
467
468
469
470
471
472
 
# File 'lib/insika/executor.rb', line 463

def run_subagents(tasks:, parent_state:)
  list = Array(tasks)
  return { error: "tasks must be a non-empty list of {agent, message}" } if list.empty?

  cap = SubagentGraph.fan_out_cap
  return { error: "too many subagents in one call: #{list.size} (max #{cap})" } if list.size > cap

  plans = list.map { |t| plan_subagent(t, parent_state) }
  { results: spawn_all_and_project(plans, parent_state) }
end

#running?(task_id) ⇒ Boolean

In-process registry of live fibers (ResumeTask's criterion).

Returns:

  • (Boolean) 


142
 
# File 'lib/insika/executor.rb', line 142

def running?(task_id) = @running.key?(task_id)

#spawn(task, profile:, resume_from: nil) ⇒ Object

Stage 1 (async part): creates the actor, registers it and fires the fiber. Called by the turn handlers (SendMessage/ResumeTask/TriggerWorkflow).

Raises:



228
229
230
231
232
233
234
235
 
# File 'lib/insika/executor.rb', line 228

def spawn(task, profile:, resume_from: nil)
  raise Insika::ValidationError, "task already running: #{task.id}" if running?(task.id)

  actor = TaskActor.new(task_id: task.id, parent: turn_parent)
  @running[task.id] = actor
  actor.run { execute(task, profile: profile, resume_from: resume_from, actor: actor) }
  task.id
end

#spawn_in_session(task, profile:, resume_from: nil) ⇒ Object

Turn entry point that RESPECTS the session: a turn with a session_id is SERIALIZED in that session's SessionActor queue (one at a time); without a session_id (one-shot/history) it goes straight to spawn (standalone).



241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
 
# File 'lib/insika/executor.rb', line 241

def spawn_in_session(task, profile:, resume_from: nil)
  # the intake is closed. The task is already durable (queued);
  # answering with its id and spawning NOTHING is what "stops accepting new
  # turns" means — the next boot's recovery replays it. Subagent turns are NOT
  # gated (they spawn directly): a child of an in-flight parent is part of the
  # work the drain is waiting FOR, and refusing it would wedge the parent.
  return defer_turn(task) if @draining

  # SessionActor only in SERVING mode (@supervised): serializes concurrent
  # REQUESTS. At boot/recovery (non-supervised) the replay is sequential and
  # the long-lived loop would hang the Boot's Sync — use direct spawn (the
  # owner awaits the turn). One-shot/history (no session_id)
  # never serialize.
  unless @supervised && task.session_id
    return spawn(task, profile: profile, resume_from: resume_from)
  end

  session_actor(task.session_id).enqueue(task, profile: profile, resume_from: resume_from,
                                               policy: queue_policy(profile, task.session_id))
end

#steer_into_running(session_id, text, profile:) ⇒ Object

the steer door: a message for a session whose turn is ALREADY running is appended to that run instead of becoming a turn of its own. -> the RUNNING task's id (the turn that will answer it), or nil (create a task and spawn as usual, which is followup).

Asked BEFORE a task is created, like the collect door, and for the same reason. The post lands in the turn's mailbox; SteerInjector reads it at the next tool-batch boundary. Nothing here touches the run in flight — a message that no boundary ever arrives for is released as a follow-up turn (#release_steered).



289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
 
# File 'lib/insika/executor.rb', line 289

def steer_into_running(session_id, text, profile:)
  return nil unless @supervised && session_id

  policy = queue_policy(profile, session_id)
  return nil unless policy.steer?

  session_actor = @session_actors[session_id]
  return nil unless session_actor&.alive?

  # No turn running (or one still at the door): there is nothing to steer INTO.
  # A turn at the door belongs to `collect`, which is a different mode.
  task = session_actor.current_task
  return nil if task.nil?
  # A workflow turn orchestrates RubyLLM itself and has no Insika chat to append to
  # (docs/WORKFLOWS.md says so). Refused at the door so the message becomes an
  # ordinary next turn instead of a phantom one.
  return nil if workflow_turn?(task)

  actor = @running[task.id]
  return nil if actor.nil?
  # The bound is on what ONE run may absorb, so it counts posts and not the buffer:
  # a message already injected still spent its slot. Overflow degrades to
  # `followup` rather than growing an unbounded tail.
  return nil if actor.user_messages_posted >= policy.steer_max_messages

  actor.post(:user_message, text)
  task.id
end

#stop_session_actors ⇒ Object

Shuts down all SessionActors (server shutdown / tests — the loop blocks forever on dequeue when idle).



427
428
429
430
431
 
# File 'lib/insika/executor.rb', line 427

def stop_session_actors
  @session_actors.each_value(&:stop)
  @session_actors.clear
  nil
end