Class: Insika::QueuePolicy

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

Overview

what happens to an inbound message for a session that is ALREADY busy. Today the engine has exactly one answer, "it waits in line"; this names that answer followup and adds three others:

collect   — the message arrived BEFORE the turn started: merge the fragments
          into one turn (the whole mechanism is a timer at the door).
steer     — the turn is ALREADY running tools: append the message to the run in
          flight, at a tool-batch boundary, so the customer's correction lands
          before the model's next step instead of after the whole run.
interrupt — the turn is running and is now answering the wrong question: abandon
          it at its next boundary and let the new message be its own turn.

They never compete for the same message: collect only ever touches a turn that has not started; steer and interrupt only a turn that has, and they differ in whether the run in flight is still worth finishing.

Resolution per message, the order EdgeLimiter already documents (edge_limiter.rb:17) — configuration over convention:

session vars["queue_mode"]   — one conversation pinned by an operator
profile.limits[:<key>]       — per-agent (a PRESENT key wins, incl. nil/0 = off)
settings["queue"][<key>]     — platform default, editable in the Studio
DEFAULTS[<key>]              — = today's behavior

Every default is off: a bare wiring behaves exactly as it did before this existed. The knobs live in profile.limits next to tool_concurrency and turn_timeout because they are bounds on the same thing — how much work one turn is allowed to absorb.

Constant Summary collapse

MODES =

All four of are delivered, so there is no "specified but unshipped" tier any more — a mode outside this set is a typo, and it is refused rather than approximated. Treating an unknown mode as followup would look exactly like a mode that never fires.

%i[followup collect steer interrupt].freeze
DEFAULTS = 
{
  queue_mode: :followup,
  debounce_ms: 0,          # 0 = no window: dequeue immediately, today's path
  debounce_max_ms: 10_000, # ceiling on the TOTAL deferral (see #debounce_deadline_ms)
  steer_max_messages: 5,   # how many messages ONE run may absorb; overflow = followup
  steer_join: nil          # nil = the raw text; a template frames it (see #frame)
}.freeze
JOIN_PLACEHOLDER =

The placeholder steer_join must carry, so a template that would silently drop the customer's message is a config error and not a lost message.

"%{message}"

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(mode:, debounce_ms:, debounce_max_ms:, steer_max_messages: DEFAULTS[:steer_max_messages], steer_join: nil) ⇒ QueuePolicy

Returns a new instance of QueuePolicy.



70
71
72
73
74
75
76
77
78
79
80
81
82
 
# File 'lib/insika/queue_policy.rb', line 70

def initialize(mode:, debounce_ms:, debounce_max_ms:,
               steer_max_messages: DEFAULTS[:steer_max_messages], steer_join: nil)
  @mode = mode
  @debounce_ms = [debounce_ms.to_i, 0].max
  # A non-positive ceiling would mean "defer forever", which nobody wants and
  # which a stray 0 in a config would silently buy. Fall back to the default.
  max = debounce_max_ms.to_i
  @debounce_max_ms = max.positive? ? max : DEFAULTS[:debounce_max_ms]
  # 0 (or a negative) is a legitimate "never steer on this agent" — unlike the
  # ceiling above, it forbids rather than defers forever, so it is honored.
  @steer_max_messages = [steer_max_messages.to_i, 0].max
  @steer_join = join!(steer_join)
end

Instance Attribute Details

#debounce_max_ms ⇒ Object (readonly)

Returns the value of attribute debounce_max_ms.



53
54
55
 
# File 'lib/insika/queue_policy.rb', line 53

def debounce_max_ms
  @debounce_max_ms
end

#debounce_ms ⇒ Object (readonly)

Returns the value of attribute debounce_ms.



53
54
55
 
# File 'lib/insika/queue_policy.rb', line 53

def debounce_ms
  @debounce_ms
end

#mode ⇒ Object (readonly)

Returns the value of attribute mode.



53
54
55
 
# File 'lib/insika/queue_policy.rb', line 53

def mode
  @mode
end

#steer_join ⇒ Object (readonly)

Returns the value of attribute steer_join.



53
54
55
 
# File 'lib/insika/queue_policy.rb', line 53

def steer_join
  @steer_join
end

#steer_max_messages ⇒ Object (readonly)

Returns the value of attribute steer_max_messages.



53
54
55
 
# File 'lib/insika/queue_policy.rb', line 53

def steer_max_messages
  @steer_max_messages
end

Class Method Details

.mode!(value) ⇒ Object

A mode name -> Symbol, or raise. Blank = the default (an absent key is not an error; a WRONG key is).

Raises:



112
113
114
115
116
117
118
119
120
 
# File 'lib/insika/queue_policy.rb', line 112

def self.mode!(value)
  return DEFAULTS[:queue_mode] if Coercion.blank?(value)

  name = value.to_s.strip.downcase.to_sym
  return name if MODES.include?(name)

  raise Insika::ValidationError,
        "unknown queue_mode: #{value.inspect} (expected #{MODES.join(', ')})"
end

.resolve(profile, settings_store: nil, vars: nil) ⇒ Object

profile: an AgentProfile (or nil); settings_store: nil = no platform layer; vars: the session's vars Hash (string keys, as the SessionStore returns).



57
58
59
60
61
62
63
64
65
66
67
68
 
# File 'lib/insika/queue_policy.rb', line 57

def self.resolve(profile, settings_store: nil, vars: nil)
  platform = ((settings_store&.get || {})["queue"] || {})
  limits = profile.respond_to?(:limits) ? (profile.limits || {}) : {}

  new(
    mode: mode!(pick_mode(vars, limits, platform)),
    debounce_ms: pick(:debounce_ms, limits, platform),
    debounce_max_ms: pick(:debounce_max_ms, limits, platform),
    steer_max_messages: pick(:steer_max_messages, limits, platform),
    steer_join: pick_text(:steer_join, limits, platform)
  )
end

Instance Method Details

#collect? ⇒ Boolean

Does this policy merge into a turn that has not started yet?

Returns:

  • (Boolean) 


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

def collect? = @mode == :collect

#debounce? ⇒ Boolean

Does this policy want messages held at the door at all?

Returns:

  • (Boolean) 


85
 
# File 'lib/insika/queue_policy.rb', line 85

def debounce? = @debounce_ms.positive?

#frame(text) ⇒ Object

The content of the injected message. steer_join frames it when an agent needs the model to know this text arrived mid-run ("the customer just added: %message"); nil — the default — appends exactly what the person typed. A plain gsub, not format: the text is a customer's, and a stray % in it must not raise.



104
105
106
107
108
 
# File 'lib/insika/queue_policy.rb', line 104

def frame(text)
  return text.to_s if @steer_join.nil?

  @steer_join.gsub(JOIN_PLACEHOLDER, text.to_s)
end

#interrupt? ⇒ Boolean

Does this policy abandon the turn in flight? The new message then becomes an ordinary turn of its own — which is why interrupt, unlike the two joining modes, needs no verdict field and works on every surface.

Returns:

  • (Boolean) 


98
 
# File 'lib/insika/queue_policy.rb', line 98

def interrupt? = @mode == :interrupt

#steer? ⇒ Boolean

Does this policy append into a turn that is already running? steer_max_messages of 0 is the agent saying no, so it answers false rather than steering once and then refusing.

Returns:

  • (Boolean) 


93
 
# File 'lib/insika/queue_policy.rb', line 93

def steer? = @mode == :steer && @steer_max_messages.positive?