Class: Riffer::Agent::Session

Inherits:
Object
  • Object
show all
Includes:
Enumerable, Enumerable[Riffer::Messages::Base]
Defined in:
lib/riffer/agent/session.rb,
sig/generated/riffer/agent/session.rbs

Overview

Owns the conversation handle for an agent: the message array, the on_message callbacks, and the tool_use ↔ tool_result invariant that keeps tool calls and their results consistent.

agent.session.add(msg)                  # append + fire callbacks
agent.session.set([msg1, msg2])         # bulk replace (silent)
agent.session.unset                     # clear (silent)
agent.session.remove(id: "a_1")
agent.session.update(id: "a_1", content: "...")
agent.session.find { |m| m.id == "a_1" }

Defined Under Namespace

Modules: Repair

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(messages: []) ⇒ Session

-- : (?messages: Array) -> void

Parameters:



25
26
27
28
 
# File 'lib/riffer/agent/session.rb', line 25

def initialize(messages: [])
  @messages = messages
  @callbacks = [] #: Array[^(Riffer::Messages::Base) -> void]
end

Instance Attribute Details

#messages ⇒ Array[Riffer::Messages::Base] (readonly)

The message history.

Returns:



21
22
23
 
# File 'lib/riffer/agent/session.rb', line 21

def messages
  @messages
end

Instance Method Details

#add(message, silent: false) ⇒ Riffer::Messages::Base

Appends message and fires every registered callback once with it. Pass silent: true to skip callbacks — used for non-inference inputs like user messages that subscribers don't expect on the callback channel.

: (Riffer::Messages::Base, ?silent: bool) -> Riffer::Messages::Base

Parameters:

Returns:



44
45
46
47
48
 
# File 'lib/riffer/agent/session.rb', line 44

def add(message, silent: false)
  @messages << message
  @callbacks.each { |callback| callback.call(message) } unless silent
  message
end

#cascade_dropped_tool_calls(old, replacement) ⇒ void

This method returns an undefined value.

-- : (Riffer::Messages::Base, Riffer::Messages::Base) -> void

Parameters:



184
185
186
187
188
189
190
191
192
 
# File 'lib/riffer/agent/session.rb', line 184

def cascade_dropped_tool_calls(old, replacement)
  return unless old.is_a?(Riffer::Messages::Assistant)
  return unless replacement.is_a?(Riffer::Messages::Assistant)

  removed_ids = old.tool_calls.map(&:call_id) - replacement.tool_calls.map(&:call_id)
  return if removed_ids.empty?

  @messages.reject! { |m| m.is_a?(Riffer::Messages::Tool) && removed_ids.include?(m.tool_call_id) }
end

#each ⇒ Enumerator[Riffer::Messages::Base, self] #each ⇒ Object

Yields each message in order, or returns an Enumerator without a block.

: () -> Enumerator[Riffer::Messages::Base, self] : () { (Riffer::Messages::Base) -> void } -> untyped

Overloads:

Yields:

Yield Parameters:

Yield Returns:

  • (void) 


155
156
157
158
 
# File 'lib/riffer/agent/session.rb', line 155

def each(&block)
  return @messages.each unless block
  @messages.each(&block)
end

#final_assistant_message ⇒ Riffer::Messages::Assistant?

The most recent Riffer::Messages::Assistant in the session, or nil when none exists.

-- : () -> Riffer::Messages::Assistant?

Returns:



173
174
175
176
177
178
 
# File 'lib/riffer/agent/session.rb', line 173

def final_assistant_message
  # TODO: Replace with rfind when minimum Ruby is 4.0+
  # rubocop:disable Style/ReverseFind
  @messages.reverse_each.find { |m| m.is_a?(Riffer::Messages::Assistant) } #: Riffer::Messages::Assistant?
  # rubocop:enable Style/ReverseFind
end

#on_message {|arg0| ... } ⇒ self

Registers a callback invoked once per message appended via #add.

: () { (Riffer::Messages::Base) -> void } -> self

Yields:

Yield Parameters:

Yield Returns:

  • (void)

Returns:

  • (self) 


33
34
35
36
37
 
# File 'lib/riffer/agent/session.rb', line 33

def on_message(&block)
  raise Riffer::ArgumentError, "on_message requires a block" unless block_given?
  @callbacks << block
  self
end

#orphaned_tool_call_ids ⇒ Array[String]

Returns the call_ids of every tool_call with no matching Riffer::Messages::Tool result anywhere in history — a hook for checking the tool_use ↔ tool_result invariant before mutating or persisting.

: () -> Array

Returns:

  • (Array[String]) 


125
126
127
128
129
130
131
 
# File 'lib/riffer/agent/session.rb', line 125

def orphaned_tool_call_ids
  result_ids = @messages.filter_map { |m| m.tool_call_id if m.is_a?(Riffer::Messages::Tool) }
  @messages.flat_map { |m|
    next [] unless m.is_a?(Riffer::Messages::Assistant)
    m.tool_calls.reject { |tc| result_ids.include?(tc.call_id) }.map(&:call_id)
  }
end

#pending_tool_calls ⇒ [ Riffer::Messages::Assistant?, Array[Riffer::Messages::Assistant::ToolCall] ]

Returns [last_assistant, pending_tool_calls]; the second element is empty when there's no assistant message or no pending calls.

: () -> [Riffer::Messages::Assistant?, Array[Riffer::Messages::Assistant::ToolCall]]

Returns:



137
138
139
140
141
142
143
144
145
146
147
148
149
 
# File 'lib/riffer/agent/session.rb', line 137

def pending_tool_calls
  last_assistant_idx = @messages.rindex { |m| m.is_a?(Riffer::Messages::Assistant) }
  return [nil, []] unless last_assistant_idx

  assistant = @messages[last_assistant_idx] #: Riffer::Messages::Assistant
  return [assistant, []] if assistant.tool_calls.empty?

  executed_ids = (@messages[(last_assistant_idx + 1)..] || []).filter_map { |m|
    m.tool_call_id if m.is_a?(Riffer::Messages::Tool)
  }

  [assistant, assistant.tool_calls.reject { |tc| executed_ids.include?(tc.call_id) }]
end

#rebuild_message(old, attrs) ⇒ Riffer::Messages::Base

-- : (Riffer::Messages::Base, Hash[Symbol, untyped]) -> Riffer::Messages::Base

Parameters:

Returns:



196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
 
# File 'lib/riffer/agent/session.rb', line 196

def rebuild_message(old, attrs)
  case old
  when Riffer::Messages::Assistant
    Riffer::Messages::Assistant.new(
      attrs.fetch(:content, old.content),
      id: attrs.fetch(:id, old.id),
      tool_calls: attrs.fetch(:tool_calls, old.tool_calls),
      token_usage: attrs.fetch(:token_usage, old.token_usage),
      structured_output: attrs.fetch(:structured_output, old.structured_output)
    )
  when Riffer::Messages::Tool
    Riffer::Messages::Tool.new(
      attrs.fetch(:content, old.content),
      tool_call_id: old.tool_call_id,
      name: attrs.fetch(:name, old.name),
      id: attrs.fetch(:id, old.id),
      error: attrs.fetch(:error, old.error),
      error_type: attrs.fetch(:error_type, old.error_type)
    )
  when Riffer::Messages::User
    Riffer::Messages::User.new(
      attrs.fetch(:content, old.content),
      id: attrs.fetch(:id, old.id),
      files: attrs.fetch(:files, old.files)
    )
  else
    old.class.new(
      attrs.fetch(:content, old.content),
      id: attrs.fetch(:id, old.id)
    )
  end
end

#remove(id:) ⇒ Riffer::Messages::Base?

Removes a message by id, cascading to drop the Tool results of a removed assistant's tool_calls so the tool_use ↔ tool_result invariant holds. Raises on a Tool message — that would orphan its parent; use #update instead. Returns nil if no message matches.

: (id: String) -> Riffer::Messages::Base?

Parameters:

  • id: (String)

Returns:



72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
 
# File 'lib/riffer/agent/session.rb', line 72

def remove(id:)
  idx = @messages.index { |m| m.id == id }
  return nil unless idx

  target = @messages[idx]
  if target.is_a?(Riffer::Messages::Tool)
    raise Riffer::ArgumentError,
      "remove cannot drop a Tool message (would orphan the parent's tool_use); use #update instead"
  end

  if target.is_a?(Riffer::Messages::Assistant) && !target.tool_calls.empty?
    child_ids = target.tool_calls.map(&:call_id)
    @messages.reject! { |m| m.is_a?(Riffer::Messages::Tool) && child_ids.include?(m.tool_call_id) }
    @messages.delete(target)
  else
    @messages.delete_at(idx)
  end
  target
end

#set(messages) ⇒ self

Replaces the message history wholesale

: (Array) -> self

Parameters:

Returns:

  • (self) 


53
54
55
56
 
# File 'lib/riffer/agent/session.rb', line 53

def set(messages)
  @messages = messages
  self
end

#steps ⇒ Integer

The number of LLM steps completed, used by the agent loop to enforce max_steps on resume.

: () -> Integer

Returns:

  • (Integer) 


164
165
166
 
# File 'lib/riffer/agent/session.rb', line 164

def steps
  @messages.count { |m| m.is_a?(Riffer::Messages::Assistant) }
end

#unset ⇒ self

Clears the session.

: () -> self

Returns:

  • (self) 


61
62
63
64
 
# File 'lib/riffer/agent/session.rb', line 61

def unset
  @messages = []
  self
end

#update(id: nil, tool_call_id: nil, **attrs) ⇒ Riffer::Messages::Base

Partial in-place update: looks up a message by id: or tool_call_id: (exactly one), overlays attrs onto a same-type replacement, and swaps it in. Dropping tool_calls from an assistant cascades to remove their Tool results, preserving the invariant. Raises on neither/both keys or no match.

: (?id: String?, ?tool_call_id: String?, **untyped) -> Riffer::Messages::Base

Parameters:

  • id: (String, nil) (defaults to: nil)
  • tool_call_id: (String, nil) (defaults to: nil)
  • (Object)

Returns:



98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
 
# File 'lib/riffer/agent/session.rb', line 98

def update(id: nil, tool_call_id: nil, **attrs)
  raise Riffer::ArgumentError, "update requires either id: or tool_call_id:" if id.nil? && tool_call_id.nil?
  raise Riffer::ArgumentError, "update accepts id: or tool_call_id:, not both" if id && tool_call_id

  idx = if id
    @messages.index { |m| m.id == id }
  else
    @messages.index { |m| m.is_a?(Riffer::Messages::Tool) && m.tool_call_id == tool_call_id }
  end

  unless idx
    key = id ? "id #{id.inspect}" : "tool_call_id #{tool_call_id.inspect}"
    raise Riffer::ArgumentError, "no message found for #{key}"
  end

  old = @messages[idx] #: Riffer::Messages::Base
  replacement = rebuild_message(old, attrs)
  @messages[idx] = replacement
  cascade_dropped_tool_calls(old, replacement)
  replacement
end