Class: Protege::Message

Inherits:
ApplicationRecord show all
Includes:
BroadcastableMessage
Defined in:
app/models/protege/message.rb

Overview

Persisted record of a single email — inbound or outbound — in the Protege pipeline.

Message is the application's source of truth for everything the engine reasons about: the web UI, the inbox, and the ThreadHistoryResolver resolver all read from this model. The underlying ActionMailbox::InboundEmail is retained only as a raw delivery-pipeline artifact; application logic never queries it directly. Every message belongs to exactly one EmailThread and one Agent, and inbound messages link back to the InboundEmail they were ingested from.

Lifecycle (inbound)

pending     → received, InferenceJob not yet queued
processing  → InferenceJob is running
processed   → InferenceJob completed successfully
failed      → InferenceJob raised; error captured in +last_processing_error+
bounced     → no matching agent; handled in AgentMailbox before a job is queued

Lifecycle (outbound)

Outbound messages are persisted by MessageObserver once delivery succeeds and carry no processing_status — they are terminal records of what the agent sent.

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.build_from_mail(mail:, direction:, thread:, agent:, inbound_email: nil, attach_from_mail: true) ⇒ Protege::Message

Build (but do not save) a Message from a Mail::Message object.

Used for outbound messages constructed outside the mailbox ingress path. The caller owns any further field assignment and the eventual save!.

Parameters:

  • mail (Mail::Message) —

    the parsed email to copy attributes from

  • direction (Symbol, String) —

    :inbound or :outbound

  • thread (Protege::EmailThread) —

    the owning thread

  • agent (Protege::Agent) —

    the agent the message belongs to

  • inbound_email (ActionMailbox::InboundEmail, nil) (defaults to: nil) —

    the source record, if any

  • attach_from_mail (Boolean) (defaults to: true) —

    rebuild attachments from the mail's parts (true for inbound, whose files arrive as bytes); pass false for outbound, where the caller attaches known blobs.

Returns:



120
121
122
123
124
125
 
# File 'app/models/protege/message.rb', line 120

def build_from_mail(mail:, direction:, thread:, agent:, inbound_email: nil, attach_from_mail: true)
  new.tap do |record|
    record.assign_from_mail(mail:, thread:, agent:, inbound_email:, attach_from_mail:)
    record.direction = direction
  end
end

.ingest!(inbound_email:, agent:) ⇒ Protege::Message

Ingest an ActionMailbox::InboundEmail into a persisted, processing Message.

Finds or creates the owning EmailThread, copies all mail-derived attributes onto the record, and transitions it to processing. Keyed on inbound_email_id and agent, so a retry for the same agent updates its existing record, while a different agent addressed by the same email gets its own — one inbound email maps to one message per handling agent.

Parameters:

  • inbound_email (ActionMailbox::InboundEmail) —

    the raw received email

  • agent (Protege::Agent) —

    the agent the email routed to

Returns:



97
98
99
100
101
102
103
104
105
 
# File 'app/models/protege/message.rb', line 97

def ingest!(inbound_email:, agent:)
  mail   = inbound_email.mail
  thread = EmailThread.find_or_create_for(mail:, agent:)

  find_or_initialize_by(inbound_email_id: inbound_email.id, agent:).tap do |record|
    record.assign_from_mail(mail:, thread:, agent:, inbound_email:)
    record.start_processing!
  end
end

Instance Method Details

#assign_from_mail(mail:, thread:, agent:, inbound_email: nil, attach_from_mail: true) ⇒ void

This method returns an undefined value.

Copy every RFC 2822-derived attribute from a Mail::Message onto this record.

Shared by both ingest! and build_from_mail. Sets the thread/agent associations and then delegates the header, body, and threading field groups to private helpers so the work stays within Active Record's metrics budget. Does not save.

Parameters:

  • mail (Mail::Message) —

    the parsed email to read attributes from

  • thread (Protege::EmailThread) —

    the owning thread

  • agent (Protege::Agent) —

    the agent the message belongs to

  • inbound_email (ActionMailbox::InboundEmail, nil) (defaults to: nil) —

    the source record, if any

  • attach_from_mail (Boolean) (defaults to: true) —

    rebuild attachments from the mail's MIME parts; true for inbound (files arrive as bytes), false for outbound (the caller attaches the known blobs by reference).



252
253
254
255
256
257
258
259
260
261
 
# File 'app/models/protege/message.rb', line 252

def assign_from_mail(mail:, thread:, agent:, inbound_email: nil, attach_from_mail: true)
  self.email_thread  = thread
  self.agent         = agent
  self.inbound_email = inbound_email

  assign_envelope_from(mail)
  assign_threading_from(mail)
  assign_bodies_from(mail)
  assign_attachments_from(mail) if attach_from_mail
end

#attachment_parts_with_note ⇒ Array(Array<Object>, Protege::TextPart)

Walk this message's attachments once, returning [[vision_parts], note_part] — each viewable file rendered as its vision part (via StoredFile#to_part) and a single trailing TextPart note naming every attachment (unviewable ones included) with its blob id. Callers flatten both into content, or take just the note part (e.g. thread-history replay).

Returns:



226
227
228
229
230
231
232
233
234
235
236
237
 
# File 'app/models/protege/message.rb', line 226

def attachment_parts_with_note
  vision  = []
  entries = []

  attachments.each do |attachment|
    file = Protege::StoredFile.new(attachment.blob)
    vision << file.to_part if file.viewable?
    entries << file.summary
  end

  [vision, Protege::TextPart.new(text: attachment_note_text(entries))]
end

#content_header ⇒ String

The standard email-style header block for this message: From / To / Cc / Date / Subject, one line each, in that fixed order. Every line is always rendered — a field with no value is shown blank (e.g. Cc:), never omitted — so every turn has the same predictable shape for the model. Folded into #to_llm_text as plain text, deliberately kept in the content string rather than added as new message metadata.

Returns:

  • (String) —

    the five-line header block



184
185
186
187
188
189
190
191
192
 
# File 'app/models/protege/message.rb', line 184

def content_header
  [
    "From: #{from_address}",
    "To: #{Array(to_addresses).join(', ')}",
    "Cc: #{Array(cc_addresses).join(', ')}",
    "Date: #{sent_at&.utc&.strftime('%Y-%m-%d %H:%M UTC')}",
    "Subject: #{subject}"
  ].join("\n")
end

#content_parts ⇒ String, Array<Object>

This message expressed as provider content — the reply path's opening user turn. The text is #to_llm_text (header + body + attachment note); when the message carries viewable attachments (images/PDFs) their vision parts are appended so a multimodal model sees the files themselves. Returns a plain string when there are no attachments.

Returns:

  • (String, Array<Object>) —

    the serialized text, or that text plus attachment vision parts



213
214
215
216
217
218
 
# File 'app/models/protege/message.rb', line 213

def content_parts
  return to_llm_text unless attachments.attached?

  vision, note = attachment_parts_with_note
  [Protege::TextPart.new(text: compose_llm_text(note: note.text)), *vision]
end

#mark_failed!(error) ⇒ void

This method returns an undefined value.

Transition the message to failed and capture the error for later inspection.

Stores the error class, message, and the first five backtrace lines in last_processing_error so a failed job can be triaged without re-running it.

Parameters:

  • error (Exception) —

    the error raised during processing

Raises:

  • (ActiveRecord::RecordInvalid) —

    when the update fails validation



296
297
298
299
300
301
 
# File 'app/models/protege/message.rb', line 296

def mark_failed!(error)
  update!(
    processing_status:     :failed,
    last_processing_error: "#{error.class}: #{error.message}\n#{error.backtrace.first(5).join("\n")}"
  )
end

#mark_processed! ⇒ void

This method returns an undefined value.

Transition the message to processed and record the completion time.

Raises:

  • (ActiveRecord::RecordInvalid) —

    when the update fails validation



284
285
286
 
# File 'app/models/protege/message.rb', line 284

def mark_processed!
  update!(processing_status: :processed, processing_completed_at: Time.current)
end

#readable_body ⇒ String?

The message's best plain-text content for replay or display.

Prefers the plain-text body; when that is blank, falls back to the HTML body with its tags stripped, so an HTML-only email still yields readable text. Returns nil when the message carries neither — letting callers (e.g. the ThreadHistoryResolver resolver) skip empty turns.

Returns:

  • (String, nil) —

    the readable body, or nil when both bodies are blank



169
170
171
172
173
174
175
 
# File 'app/models/protege/message.rb', line 169

def readable_body
  text = text_body.to_s.strip
  return text unless text.empty?

  html = ActionController::Base.helpers.strip_tags(html_body.to_s).strip
  html unless html.empty?
end

#recursion_hops ⇒ Integer

How many agent hops this message has already taken, read from the linked raw inbound email's X-Protege-Recursion header. Zero for human mail (mail clients never echo the header back) and when the raw email is gone (Action Mailbox incinerates it eventually) — either way the chain resets, which is the intended behavior. A reply to this message stamps this count plus one (see Gateway::RECURSION_HEADER).

Returns:

  • (Integer) —

    the inbound hop count, or 0 when untracked



310
311
312
313
314
315
316
 
# File 'app/models/protege/message.rb', line 310

def recursion_hops
  raw = inbound_email&.mail
  return 0 unless raw

  field = raw.header[Gateway::RECURSION_HEADER]
  field ? field.value.to_i : 0
end

#start_processing! ⇒ void

This method returns an undefined value.

Mark this message as in-progress and persist it.

Called at the start of inference (by InferenceJob via ingest!, and by ConsoleInferenceJob directly). Defaults the direction to :inbound, stamps the start time, increments the attempt counter, and captures the current correlation ID for tracing.

Raises:

  • (ActiveRecord::RecordInvalid) —

    when the record fails validation



271
272
273
274
275
276
277
278
 
# File 'app/models/protege/message.rb', line 271

def start_processing!
  self.direction             = direction || :inbound
  self.processing_status     = :processing
  self.processing_started_at = Time.current
  self.processing_attempts   = processing_attempts.to_i + 1
  self.correlation_id        = Protege::Current.correlation_id
  save!
end

#to_llm_text ⇒ String

The canonical plain-text serialization of this message for the model: the standard #content_header block, the readable body, and — when the message has attachments — the note the app already builds for them (see #attachment_parts_with_note). This is the single "message as text" artifact both the reply opening turn and the ThreadHistoryResolver replay, so a turn is self-describing. Always includes the header; callers deciding whether to skip an empty turn test the message's body/attachments, not this.

Returns:

  • (String) —

    the header, body, and attachment note joined into one string



202
203
204
205
 
# File 'app/models/protege/message.rb', line 202

def to_llm_text
  note = attachment_parts_with_note.last.text if attachments.attached?
  compose_llm_text(note:)
end