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, media: nil, media_output: nil, grounding_enforcer: nil, cache_series_store: nil, contact_store: nil, followup_store: nil, model_visible_trace_store: 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
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
 
# 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, media: nil, media_output: nil,
                grounding_enforcer: nil, cache_series_store: nil,
                contact_store: nil, followup_store: nil, model_visible_trace_store: 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
  # the model-visible trace — what the provider received per
  # (task, turn), captured at the chat boundary. nil = off (no record,
  # zero overhead — parity).
  @model_visible_trace_store = model_visible_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
  # WS9 media seam (nil = default built on first audio turn): ->(url) { text }.
  @media = media
  # WS9 (saída) media-generation seams: { image: ->(prompt, cfg) [part,
  # usage], tts: ->(text, cfg) [part, usage] }. nil = the defaults (built
  # lazily on first generation — RubyLLM + Net::HTTP behind lazy requires,
  # the core stays gem-free at load). Injected by specs; a production graph
  # that wants a non-RubyLLM backend injects its own lambdas.
  @media_output = media_output
  # the :enforce boundary step, called between stages 6 and 8.
  # Defaults to a REAL enforcer (inert unless the profile's grounding.mode is
  # :enforce — zero behavior change for parity) so an embedder that builds
  # the Executor directly still gets the cut; `nil` stays injectable for
  # stubs that want none.
  @grounding_enforcer = grounding_enforcer || Insika::Safety::GroundingEnforcer.new
  # the per-AGENT cache-hit series. nil = no series recorded
  # (parity — the trace store still gets the per-turn entry when wired).
  @cache_series_store = cache_series_store
  # 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 follow-up stores the ChatBuilder gates the
  # schedule/cancel_followup tools on (nil = never wired — parity).
  @contact_store = contact_store
  @followup_store = followup_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,
    # WS9 (saída): the media-generation runner, same shape — the Executor
    # owns the seams + usage accounting, the builder only wires the tools
    # the turn's gates allow.
    media_runner: self,
    # the builder wires the briefing-write system tools gated by
    # @session_store + profile.briefing_fields. nil = never wired (parity).
    session_store: session_store,
    # the builder wires the schedule/cancel_followup system
    # tools gated by a parsed policy AND both stores present. nil = never
    # wired (parity).
    contact_store: contact_store,
    followup_store: followup_store
  )
  # 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.



164
165
166
 
# File 'lib/insika/executor.rb', line 164

def alert_dispatcher
  @alert_dispatcher
end

#distill_engine ⇒ Object

the distillation engine — the tick-duty that finds idle customer sessions and distills them on its own worker fiber (a child of the turn supervisor, like the tick). nil = distillation off (parity — nothing scans, nothing distills).



170
171
172
 
# File 'lib/insika/executor.rb', line 170

def distill_engine
  @distill_engine
end

#harvest_engine ⇒ Object

the harvest engine — the tick-duty that finds idle, unmined sessions and mines them on its own worker fiber (a child of the turn supervisor, like the tick). nil = harvest off (parity — nothing scans, nothing mines).



176
177
178
 
# File 'lib/insika/executor.rb', line 176

def harvest_engine
  @harvest_engine
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).



152
153
154
 
# File 'lib/insika/executor.rb', line 152

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).



410
411
412
 
# File 'lib/insika/executor.rb', line 410

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).



159
160
161
 
# File 'lib/insika/executor.rb', line 159

def tick
  @tick
end

Instance Method Details

#account_media_usage(state, part, usage) ⇒ Object

Accounts a generated part in the turn's usage: the provider's token counts (images — the merge keeps what the classifier already banked), plus an honest media call counter per part (the speech API reports no tokens; the part itself carries the model for consumer-side pricing).



660
661
662
663
664
665
666
667
668
669
670
671
 
# File 'lib/insika/executor.rb', line 660

def (state, part, usage)
  usage ||= {}
  tokens = {}
  tokens[:input_tokens] = usage[:input_tokens].to_i if usage[:input_tokens]
  tokens[:output_tokens] = usage[:output_tokens].to_i if usage[:output_tokens]
  tokens[:total_tokens] = tokens[:input_tokens].to_i + tokens[:output_tokens].to_i if tokens.any?
  unless tokens.empty?
    tokens[:model] = part["model"] if part["model"]
    state.usage = merge_usage(tokens, state.usage)
  end
  state.usage = (state.usage || {}).merge(media: state.usage&.fetch(:media, 0).to_i + 1)
end

#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.



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

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.



183
184
185
 
# File 'lib/insika/executor.rb', line 183

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.



198
199
200
201
202
 
# File 'lib/insika/executor.rb', line 198

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.



325
326
327
328
329
330
331
332
333
334
335
 
# File 'lib/insika/executor.rb', line 325

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) 


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

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.



416
417
418
 
# File 'lib/insika/executor.rb', line 416

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, timing: nil) ⇒ Object

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



566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
 
# File 'lib/insika/executor.rb', line 566

def execute(task, profile:, actor:, resume_from: nil, timing: 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, timing)
# 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::RoutingError => e
  # WS4: a route's delegate is missing or its turn failed — an operator
  # config error, staged so the envelope names routing, never :unknown.
  fail_task(task, e, stage: :routing)
rescue Insika::MediaError => e
  # WS9: a voice message that could not be fetched/transcribed (or a media
  # URL the egress guard refused) — heard-loud, never a silent drop.
  fail_task(task, e, stage: :media)
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

#generate_media_output(kind, content, config) ⇒ Object

WS9 (saída), the RUNNER side the generate_image/tts system tools call (public like run_subagent — a tool reaches back into the Executor):

-> [part, usage]: resolve the seam (injected or the lazy default) and run it. The default seams are built on FIRST use, when the turn already has a chat (ruby_llm loaded), so the load-guard holds.

Raises:



649
650
651
652
653
654
 
# File 'lib/insika/executor.rb', line 649

def generate_media_output(kind, content, config)
  seam = @media_output&.fetch(kind, nil) || Insika::Media::Output.defaults(context: @llm)[kind]
  raise Insika::MediaError, "no #{kind} output seam" unless seam

  seam.call(content, config)
end

#in_flight ⇒ Object

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



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

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).



389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
 
# File 'lib/insika/executor.rb', line 389

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.



207
208
209
210
211
 
# File 'lib/insika/executor.rb', line 207

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] }



559
560
561
562
563
 
# File 'lib/insika/executor.rb', line 559

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] }



540
541
542
543
544
545
546
547
548
549
550
551
552
 
# File 'lib/insika/executor.rb', line 540

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.



428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
 
# File 'lib/insika/executor.rb', line 428

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".



244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
 
# File 'lib/insika/executor.rb', line 244

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.



217
218
219
220
221
 
# File 'lib/insika/executor.rb', line 217

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

#run_serial(task, profile:, resume_from: nil, timing: 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.



465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
 
# File 'lib/insika/executor.rb', line 465

def run_serial(task, profile:, resume_from: nil, timing: 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, timing: timing)
  @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).



503
504
505
506
507
508
509
510
511
512
 
# File 'lib/insika/executor.rb', line 503

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: }



521
522
523
524
525
526
527
528
529
530
 
# File 'lib/insika/executor.rb', line 521

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) 


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

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

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

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

timing is the channel clock a channel turn allocated at 202 acceptance, already carrying :inbound; nil means the pipeline allocates its own (resume, engine-initiated, non-channel). mark is first-write-wins, so re-marking :inbound in the pipeline is a no-op on a threaded clock.

Raises:



284
285
286
287
288
289
290
291
 
# File 'lib/insika/executor.rb', line 284

def spawn(task, profile:, resume_from: nil, timing: 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, timing: timing) }
  task.id
end

#spawn_in_session(task, profile:, resume_from: nil, timing: 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).



297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
 
# File 'lib/insika/executor.rb', line 297

def spawn_in_session(task, profile:, resume_from: nil, timing: 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, timing: timing)
  end

  session_actor(task.session_id).enqueue(task, profile: profile, resume_from: resume_from,
                                               policy: queue_policy(profile, task.session_id),
                                               timing: timing)
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).



346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
 
# File 'lib/insika/executor.rb', line 346

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 is the collect door's other window — a steer agent with a
  # debounce merges there instead , so no message waits on either.
  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).



485
486
487
488
489
 
# File 'lib/insika/executor.rb', line 485

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