Class: Insika::Commands::SendMessage

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

Overview

Canonical turn command: validates everything synchronously, creates the Task, fires the fiber and responds {task_id:} immediately — the result flows through the Event Stream. Validations that fail do NOT create a Task (ValidationError/NotFoundError -> direct HTTP response).

Constant Summary collapse

COALESCABLE_TRANSPORTS =

surfaces whose response can carry the "you do not own the reply" verdict (merged for collect, steered for steer), and therefore the only ones where a message may join another turn. /v1/responses is NOT here: its body is OpenAI-shaped SSE with nowhere to put the field, and it is frozen because a live consumer speaks it. Joining a caller that cannot hear the verdict makes it deliver the same answer once per message, which is worse than not joining at all. Channels declare themselves by channel:<id> once lands.

%i[http:json].freeze

Instance Method Summary collapse

Constructor Details

#initialize(profiles:, session_store:, task_store:, executor:, inbound_log: nil) ⇒ SendMessage

inbound_log is the retry window for channel event ids. nil = no dedup (every surface that does not send an event_id, which is all of them today), and a caller that cannot supply a stable id gets at-least-once turns rather than a content hash pretending to be dedup.



23
24
25
26
27
28
29
 
# File 'lib/insika/commands/send_message.rb', line 23

def initialize(profiles:, session_store:, task_store:, executor:, inbound_log: nil)
  @profiles = ProfileSource.coerce(profiles)
  @session_store = session_store
  @task_store = task_store
  @executor = executor
  @inbound_log = inbound_log
end

Instance Method Details

#call(command) ⇒ Object

Raises:



31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
 
# File 'lib/insika/commands/send_message.rb', line 31

def call(command)
  p = normalize(command.payload) # accepts string and symbol keys

  agent = p[:agent].to_s
  raise Insika::ValidationError, "agent is required" if agent.empty?

  profile = @profiles[agent] ||
            (raise Insika::NotFoundError, "agent '#{agent}' not configured")

  message = p[:message]
  raise Insika::ValidationError, "message is required and non-empty" if message.to_s.strip.empty?

  # session_id XOR history (both -> error; neither -> one-shot).
  if p[:session_id] && p[:history]
    raise Insika::ValidationError, "session_id and history are mutually exclusive"
  end

  validate_history!(p[:history]) if p[:history]
  # WHO wrote this message (MessageOrigin). Absent = a customer typed it, which
  # is what a turn has always meant. Refused here rather than downstream: a
  # typo'd origin would read as absent, and a marker that silently means
  # "unmarked" is worse than none — it looks like the filtering is on.
  Insika::MessageOrigin.parse!(p[:origin])
  if p[:session_id]
    @session_store.find(p[:session_id]) ||
      (raise Insika::NotFoundError, "session '#{p[:session_id]}' not found")
  end

  # the platform retried a webhook it already delivered. Answer
  # with the turn it ALREADY produced and run nothing: without this, one flaky
  # ack costs a second LLM turn and sends the customer the same answer twice.
  # Checked before the queue doors on purpose — a duplicate is not a fragment to
  # merge and not a correction to steer with, it is the same message again.
  key = dedup_key(command, p)
  if key && (prior = @inbound_log.find(key))
    return { task_id: prior, duplicate: true }
  end

  result = start_turn(command, p, profile)
  @inbound_log.record(key, result[:task_id]) if key
  result
end