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, contact_store: nil, followup_store: nil, store: 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.

contact_store/followup_store : the contact-state hook — a customer message reopens the contact cell and a cancellation keyword revokes it + falls the pending records. nil = the hook is off (parity). store is the SHARED backend the two stores ride — the keyword revoke commits in ONE transaction (D2). nil = best-effort separate writes.



29
30
31
32
33
34
35
36
37
38
39
 
# File 'lib/insika/commands/send_message.rb', line 29

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

Instance Method Details

#call(command) ⇒ Object

Raises:



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
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
 
# File 'lib/insika/commands/send_message.rb', line 41

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")

  # A turn is text OR media. The media half (WS9) is what a voice note
  # with no caption looks like on the wire — `{ parts: [{type: "audio",
  # url: …}] }` and nothing else — and demanding a message here made the
  # anchor use case unreachable end to end: the audio becomes the message
  # at the :media stage, one step later.
  message = p[:message]
  if message.to_s.strip.empty? && !media?(p[:parts])
    raise Insika::ValidationError, "message is required and non-empty (or a media part)"
  end

  # 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.
  origin = Insika::MessageOrigin.parse!(p[:origin])
  # `scheduled` is ENGINE-RESERVED — the FollowupEngine's
  # synthetic turn stamps it, and the edge must not let a consumer
  # impersonate the engine's kick (a spoofed follow-up is the spam bug).
  if origin == Insika::MessageOrigin::SCHEDULED
    raise Insika::ValidationError,
          "origin 'scheduled' is engine-reserved: it is stamped by the follow-up engine only"
  end
  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

  # the contact-state hook — the ONLY path that sees every
  # customer message. A real customer message reopens the contact cell; a
  # cancellation keyword revokes it and falls the pending records in ONE
  # transaction. Runs AFTER validation and dedup: a refused or duplicated
  # message must not touch contact state. Nil-safe and policy-gated: no
  # profile declaration = the hook is off (parity).
  touch_contact(p, profile, command, origin)

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