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:



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

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:



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

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) 


157
158
159
160
161
 
# File 'lib/riffer/agent/session.rb', line 157

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:



176
177
178
179
 
# File 'lib/riffer/agent/session.rb', line 176

def final_assistant_message
  # TODO: Replace with rfind when minimum Ruby is 4.0+
  @messages.reverse_each.find { |m| m.is_a?(Riffer::Messages::Assistant) } #: Riffer::Messages::Assistant?
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
38
 
# 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]) 


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

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 do |m|
    next [] unless m.is_a?(Riffer::Messages::Assistant)

    m.tool_calls.reject { |tc| result_ids.include?(tc.call_id) }.map(&:call_id)
  end
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:



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

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 do |m|
    m.tool_call_id if m.is_a?(Riffer::Messages::Tool)
  end

  [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:



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
228
 
# File 'lib/riffer/agent/session.rb', line 197

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:



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

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) 


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

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) 


167
168
169
 
# File 'lib/riffer/agent/session.rb', line 167

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

#unset ⇒ self

Clears the session.

: () -> self

Returns:

  • (self) 


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

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:



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

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