Class: Insika::TaskActor

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

Overview

Actor model: one Async fiber per Task + a mailbox. The message enum is cancel/user_message/approval/pause/resume/timeout/heartbeat, with await as the cooperative SUSPENSION primitive. Cancellation/suspension only at stage boundaries — never in the middle of an operation.

Constant Summary collapse

MESSAGES =

user_message is posted by Executor#steer_into_running and consumed by SteerInjector at a tool-batch boundary. pause/resume (operator), approval (human-in-the-loop), timeout/heartbeat (watchdog/liveness, observation).

%i[cancel user_message approval pause resume timeout heartbeat].freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(task_id:, parent: Async::Task.current) ⇒ TaskActor

Returns a new instance of TaskActor.



24
25
26
27
28
29
30
31
32
 
# File 'lib/insika/task_actor.rb', line 24

def initialize(task_id:, parent: Async::Task.current)
  @task_id = task_id
  @parent = parent
  @mailbox = Async::Queue.new
  @pending_user_messages = []
  @pause_requested = false
  @heartbeats = 0
  @user_messages_posted = 0
end

Instance Attribute Details

#heartbeats ⇒ Object (readonly)

Returns the value of attribute heartbeats.



18
19
20
 
# File 'lib/insika/task_actor.rb', line 18

def heartbeats
  @heartbeats
end

#pending_user_messages ⇒ Object (readonly)

Returns the value of attribute pending_user_messages.



18
19
20
 
# File 'lib/insika/task_actor.rb', line 18

def pending_user_messages
  @pending_user_messages
end

#task_id ⇒ Object (readonly)

Returns the value of attribute task_id.



18
19
20
 
# File 'lib/insika/task_actor.rb', line 18

def task_id
  @task_id
end

#user_messages_posted ⇒ Object (readonly)

How many messages this run was ASKED to absorb, ever — including the ones already injected and cleared. It is what bounds steering (steer_max_messages), so it counts posts and never decreases.



22
23
24
 
# File 'lib/insika/task_actor.rb', line 22

def user_messages_posted
  @user_messages_posted
end

Instance Method Details

#await(reason:) ⇒ Object

BLOCKS the turn's fiber until a RESOLUTION (yields the reactor — no spin). Used by the Executor in :paused (waits for :resume) and by the ToolEnvelope in :waiting (waits for :approval). Returns [:resume, nil] or [:approval, data]. :cancel -> CancelledError; :timeout -> TimeoutError. A legitimate resolution only arrives WITH the fiber already blocked here (the operator only resumes/approves what is suspended), so it is consumed by this dequeue — there is no race requiring a buffer. Non-resolution messages received during the wait are ABSORBED without changing the suspension state (a redundant :pause does not re-arm the pause).



92
93
94
95
96
97
98
99
100
101
102
103
104
105
 
# File 'lib/insika/task_actor.rb', line 92

def await(reason:)
  @pause_requested = false # the pause/wait is being handled now
  loop do
    message, data = @mailbox.dequeue
    case message
    when :cancel        then raise CancelledError, "task #{@task_id} cancelled"
    when :timeout       then raise Insika::TimeoutError.new("wait (#{reason}) exceeded", stage: data || reason)
    when :resume, :approval then return [message, data]
    when :heartbeat     then @heartbeats += 1
    when :user_message  then @pending_user_messages << data
    # :pause during the wait: already suspended, ignore (does not re-arm pause_requested)
    end
  end
end

#drain! ⇒ Object

Drains the mailbox WITHOUT blocking (boundaries). :cancel raises (the top of the fiber maps to :cancelled). :pause arms the suspension (the Executor checks pause_requested?). Resolutions (:resume/:approval/:timeout) that arrive here WITH no pending suspension are DISCARDED (idempotent, no-op).



53
54
55
56
57
58
 
# File 'lib/insika/task_actor.rb', line 53

def drain!
  until @mailbox.empty?
    route_boundary(*@mailbox.dequeue)
  end
  nil
end

#pause_requested? ⇒ Boolean

Did the operator request a pause? (consumed by the Executor; await clears the flag).

Returns:

  • (Boolean) 


62
 
# File 'lib/insika/task_actor.rb', line 62

def pause_requested? = @pause_requested

#post(message, data = nil) ⇒ Object

Non-blocking. A message outside the enum is a caller bug.

Raises:

  • (ArgumentError) 


35
36
37
38
39
40
41
 
# File 'lib/insika/task_actor.rb', line 35

def post(message, data = nil)
  raise ArgumentError, "unknown message: #{message}" unless MESSAGES.include?(message)

  @user_messages_posted += 1 if message == :user_message
  @mailbox.enqueue([message, data])
  nil
end

#run(&turn_block) ⇒ Object

Runs the block on an Async fiber CHILD of the parent. Returns the Task.



44
45
46
 
# File 'lib/insika/task_actor.rb', line 44

def run(&turn_block)
  @async_task = @parent.async { turn_block.call(self) }
end

#take_user_messages! ⇒ Object

takes the steered messages and clears the buffer, WITHOUT observing anything else in the mailbox: whatever is not a :user_message is put back, in the order it arrived.

Why not drain!: this runs INSIDE RubyLLM's tool loop (a after_message callback), and drain! raises on :cancel. Cancellation is only ever observed at the Executor's own stage boundaries — that is what keeps a tool batch one unit of work (and for the same rule under turn_timeout). Injecting a message must not quietly become a new place a turn can die.



73
74
75
76
77
78
79
80
81
 
# File 'lib/insika/task_actor.rb', line 73

def take_user_messages!
  @mailbox.size.times do
    message, data = @mailbox.dequeue
    message == :user_message ? @pending_user_messages << data : @mailbox.enqueue([message, data])
  end
  taken = @pending_user_messages.dup
  @pending_user_messages.clear
  taken
end

#wait ⇒ Object

specs/boot await the fiber's completion.



108
 
# File 'lib/insika/task_actor.rb', line 108

def wait = @async_task&.wait