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) ⇒ 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
 
# 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_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)
  # Guardrails output filter (RFC-0009 §3.2): ->(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
  # RFC-0010 Fase 2: 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
  # RFC-0011 §6.5: 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
  # RFC-0017 A2: 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
  # LLM config v2 (§10): 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)
  # RFC-0015 §4: 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,
    # RFC-0010: 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, D2
  # injection, dedup join, ToolEnvelope wrap) — extracted collaborator (§11 B5).
  @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 (RFC-0016 A3) — see #begin_drain!
end

Instance Attribute Details

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



103
104
105
 
# File 'lib/insika/executor.rb', line 103

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



330
331
332
 
# File 'lib/insika/executor.rb', line 330

def task_store
  @task_store
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.



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

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

#begin_drain! ⇒ Object

RFC-0016 A3: 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.



110
111
112
 
# File 'lib/insika/executor.rb', line 110

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.



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

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

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

RFC-0015 §5.3 — 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.



246
247
248
249
250
251
252
253
254
255
256
 
# File 'lib/insika/executor.rb', line 246

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) 


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

def draining? = @draining

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

RFC-0015 §8. 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.



336
337
338
 
# File 'lib/insika/executor.rb', line 336

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.



486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
 
# File 'lib/insika/executor.rb', line 486

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 Insika::WorkflowSchemaError => e
  # Item 22 / §4.4: 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
  fail_task(task, e, stage: :unknown)
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.



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

def in_flight = @running.keys

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

RFC-0015 §6.4 — 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 (D7 records the same boundary for turn_timeout).



309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
 
# File 'lib/insika/executor.rb', line 309

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.



134
135
136
137
138
 
# File 'lib/insika/executor.rb', line 134

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

#recover_channel_deliveries ⇒ Object

RFC-0011 §6.5 (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] }



479
480
481
482
483
 
# File 'lib/insika/executor.rb', line 479

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

  @channel_delivery.sweep
end

#recover_delegations ⇒ Object

RFC-0010 Fase 2 (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] }



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

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

RFC-0015 §5.2 — 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.



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
375
376
377
378
 
# File 'lib/insika/executor.rb', line 348

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



171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
 
# File 'lib/insika/executor.rb', line 171

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.



144
145
146
147
148
 
# File 'lib/insika/executor.rb', line 144

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.



385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
 
# File 'lib/insika/executor.rb', line 385

def run_serial(task, profile:, resume_from: nil)
  # RFC-0016 A3: 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

RFC-0010 (item 21): 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, Fase 1) — SYNCHRONOUS: runs the child inside the parent's fiber and returns { text:, session_id: } (the child result is the tool result). async:true (Fase 2) — 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).



423
424
425
426
427
428
429
430
431
432
 
# File 'lib/insika/executor.rb', line 423

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

RFC-0010 §A (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: }



441
442
443
444
445
446
447
448
449
450
 
# File 'lib/insika/executor.rb', line 441

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) 


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

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:



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

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



219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
 
# File 'lib/insika/executor.rb', line 219

def spawn_in_session(task, profile:, resume_from: nil)
  # RFC-0016 A3: 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

RFC-0015 §5.1 — 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).



267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
 
# File 'lib/insika/executor.rb', line 267

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



405
406
407
408
409
 
# File 'lib/insika/executor.rb', line 405

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