Class: Insika::SessionActor

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

Overview

Sessions as Actors: one fiber per session with a FIFO queue of turns, executed ONE AT A TIME. Restores the "one owner at a time" invariant of the transcript that two concurrent send_message calls on the same session_id would break (read-modify-write on the Session Store). Turns from distinct sessions stay concurrent; one-shot/history (no session_id) do not go through here.

Lives in the SUPERVISED scope: the loop is a child of the supervisor, not of the request — it outlives the connection. The turn itself (spawned by the Executor) is also born on the supervisor; the SessionActor only AWAITS it to serialize.

it is also where an inbound message for a BUSY session is routed. That decision belongs here and nowhere else — this is already the object that owns "one turn at a time for this session". Putting it in the HTTP handler would duplicate the invariant; putting it in the Executor would mix turn execution with queue policy.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(session_id:, executor:, parent: Async::Task.current) ⇒ SessionActor

Returns a new instance of SessionActor.



26
27
28
29
30
31
32
33
34
35
36
 
# File 'lib/insika/session_actor.rb', line 26

def initialize(session_id:, executor:, parent: Async::Task.current)
  @session_id = session_id
  @executor = executor
  @queue = Async::Queue.new
  @running = false
  # The turn currently sitting at the door: created and :queued, but not yet
  # released to run. `collect` merges into THIS one. nil whenever there is
  # nothing mergeable — which is the common case and the safe default.
  @pending = nil
  @loop = parent.async { |t| t.annotate("session:#{session_id}"); run_loop }
end

Instance Attribute Details

#current_task ⇒ Object (readonly)

The turn this session is running RIGHT NOW, or nil when idle or still at the door. steer needs the Task itself and not just its id: whether a turn can absorb a message at all depends on what kind of turn it is (a workflow has no chat), and reading that off the object avoids a store round-trip on the request's path.



85
86
87
 
# File 'lib/insika/session_actor.rb', line 85

def current_task
  @current_task
end

Instance Method Details

#alive? ⇒ Boolean

Is the loop still alive? (the Executor revalidates before reusing from the cache — a dead loop would black-hole queued turns).

Returns:

  • (Boolean) 


92
 
# File 'lib/insika/session_actor.rb', line 92

def alive? = !!@loop&.running?

#collect(text) ⇒ Object

merge a fragment into the turn waiting at the door. -> the task id it joined, or nil when there is nothing to merge into (no pending turn, the window has closed, or the turn already started). nil is the caller's signal to create a task of its own.

Runs on the REQUEST's fiber, not the loop's; both are on the same reactor and neither yields between the check and the write below, so the "is it still mergeable" test and the append cannot interleave.



56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
 
# File 'lib/insika/session_actor.rb', line 56

def collect(text)
  pending = @pending
  return nil if pending.nil?

  @executor.task_store.append_message(pending[:task_id], text)
  pending[:count] += 1
  # A merged fragment leaves NO task of its own (see #hold_at_the_door), so this
  # is the only record that it arrived as a separate message. Kept as arrival
  # times — never the text — and shipped on :turn_coalesced, so "the customer
  # says they sent the order number" is answerable without the store carrying an
  # orphan task per fragment.
  pending[:arrivals] << Time.now.utc.iso8601
  pending[:version] += 1 # tells a sleeping debounce window that more arrived
  pending[:task_id]
rescue ArgumentError
  # The turn left :queued between the read of @pending and the append (it was
  # released while we were deciding). Not an error: the caller falls back to
  # creating its own task, which is exactly `followup`.
  nil
end

#collecting? ⇒ Boolean

Is there a turn at the door that collect could still merge into?

Returns:

  • (Boolean) 


88
 
# File 'lib/insika/session_actor.rb', line 88

def collecting? = !@pending.nil?

#depth ⇒ Object 



78
 
# File 'lib/insika/session_actor.rb', line 78

def depth = @queue.size

#enqueue(task, profile:, resume_from: nil, policy: nil) ⇒ Object

Enqueues a turn (FIFO). Non-blocking: the handler responds with an immediate task_id: even if the turn stays :queued behind another. -> task.id.

policy (a QueuePolicy) opens the debounce window for this turn; nil or a policy without a window behaves exactly as before — dequeued and run at once.



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

def enqueue(task, profile:, resume_from: nil, policy: nil)
  @queue.enqueue([task, profile, resume_from, policy])
  task.id
end

#running? ⇒ Boolean

Returns:

  • (Boolean) 


77
 
# File 'lib/insika/session_actor.rb', line 77

def running? = @running

#stop ⇒ Object

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



96
 
# File 'lib/insika/session_actor.rb', line 96

def stop = @loop&.stop