Prd/components/backend/test/test_pipeline.rb

# frozen_string_literal: true
# Zero-dependency offline tests for Pipeline#process. Run: ruby components/backend/test/test_pipeline.rb
# Covers two things: (1) the AI-context plumbing — the active ledger goes in, context fields come out;
# (2) the stage SEQUENCING + error handling — a failed transcription never reaches the summariser, and a
# failed summary keeps the transcript (the durable artifact). grep anchor: pipeline-sequencing.
require_relative "../pipeline"
require "json"

$pass = 0; $fail = 0
def ok(d); $pass += 1; puts "  ok  #{d}"; end
def bad(d, got = nil); $fail += 1; puts "  FAIL #{d}#{got.nil? ? '' : "  (got: #{got.inspect})"}"; end
def eq(a, b, d); a == b ? ok(d) : bad(d, a); end
def truthy(x, d); x ? ok(d) : bad(d, x); end

puts "pipeline threads ledger in + surfaces context fields out"
faket = Object.new
def faket.transcribe(_a, **_); { state: "transcribed", retryable: false, transcript: "reset ne" }; end
seen = {}
fakes = Object.new
fakes.define_singleton_method(:summarise) do |t, operator: nil, campaign: nil, wiki_context: nil, ledger: []|
  seen[:ledger] = ledger
  { ok: true, summary: "S", action_items: [], suggested_position: nil,
    context_digest: "D", caller_facts: ["F"],
    ledger_deltas: [{ "entry_id" => nil, "question" => "reset?", "decision" => "no", "quote" => "ne" }] }
end
pipe = Helpdesk::Pipeline.new(transcriber: faket, summariser: fakes)
out = pipe.process("/tmp/x", operator: "tmobile", ledger: [{ "id" => "led1", "question" => "reset?", "decision" => "no" }])
eq(out[:stage], "summarised", "reaches summarised stage")
eq(seen[:ledger].first["id"], "led1", "active ledger passed into the summariser")
eq(out[:context_digest], "D", "context_digest surfaced from the pipeline")
eq(out[:ledger_deltas].size, 1, "ledger_deltas surfaced from the pipeline")

# --- stage sequencing + error handling (a stub transcriber + a fake-transport summariser) ---
# stub transcriber: returns a canned result, ignoring the audio + kwargs
class StubTx
  def initialize(result) = (@r = result)
  def transcribe(_audio, **_kw) = @r
end
def env(content) = JSON.generate("choices" => [{ "message" => { "content" => content } }], "usage" => { "cost" => 0.0002 })
GOOD = JSON.generate("summary" => "Shrnutí hovoru.", "action_items" => [{ "text" => "Doplnit hint", "kind" => "tool_hint", "target" => "yamt/pinger" }])
def summariser(transport) = Helpdesk::Summariser.new(api_key: "k", transport: transport)

puts "full chain: transcript → summary"
tx = StubTx.new(state: "transcribed", transcript: "[Speaker 1]: Nejde napingat.", retryable: false, exit_code: 0)
r = Helpdesk::Pipeline.new(transcriber: tx, summariser: summariser(->(*_) { [200, env(GOOD)] })).process("a.wav", operator: "tmobile")
eq(r[:ok], true, "pipeline ok")
eq(r[:stage], "summarised", "reached summarised")
truthy(r[:transcript].include?("napingat"), "transcript carried through")
truthy(r[:summary].include?("Shrnutí"), "summary produced")
eq(r[:action_items].first["target"], "yamt/pinger", "action items produced")
eq(r[:cost], 0.0002, "cost captured")

puts "transcription failure → stops before summarising"
tx = StubTx.new(state: "failed", retryable: true, exit_code: 137)
called = false
r = Helpdesk::Pipeline.new(transcriber: tx, summariser: summariser(->(*_) { called = true; [200, env(GOOD)] })).process("a.wav")
eq(r[:ok], false, "not ok when transcription fails")
eq(r[:stage], "transcription", "stage = transcription")
eq(called, false, "summariser NOT called after a failed transcription")

puts "no speech → ok, no summary attempted"
tx = StubTx.new(state: "no_speech", transcript: "", retryable: false, exit_code: 0)
called = false
r = Helpdesk::Pipeline.new(transcriber: tx, summariser: summariser(->(*_) { called = true; [200, env(GOOD)] })).process("a.wav")
eq(r[:stage], "no_speech", "stage = no_speech")
eq(r[:ok], true, "no_speech is still ok")
eq(called, false, "summariser skipped on no speech")

puts "summary failure → transcript kept, pipeline still ok"
tx = StubTx.new(state: "transcribed", transcript: "[Speaker 1]: Ahoj.", retryable: false, exit_code: 0)
r = Helpdesk::Pipeline.new(transcriber: tx, summariser: summariser(->(*_) { [200, env("not json")] })).process("a.wav")
eq(r[:ok], true, "still ok — transcript is the durable artifact")
eq(r[:stage], "transcribed", "stage = transcribed (summary left for retry)")
truthy(r[:transcript].include?("Ahoj"), "transcript preserved")
eq(r[:summary], nil, "no summary")

# --- dual-channel: the phone recorded the two legs separately ------------------------------------
# The valuable behaviour is that a stereo recording is SPLIT and each channel transcribed on its own,
# and that every way this can go wrong still yields an ordinary transcript.
require "tmpdir"
def stereo_file(dir, frames: 8)
  pcm = +""
  frames.times { |i| pcm << [i].pack("v") << [-i].pack("v") }
  body = +"fmt " + [16].pack("V") + [1].pack("v") + [2].pack("v") +
         [16_000].pack("V") + [64_000].pack("V") + [4].pack("v") + [16].pack("v") +
         "data" + [pcm.bytesize].pack("V") + pcm
  path = File.join(dir, "stereo.wav")
  File.binwrite(path, "RIFF" + [4 + body.bytesize].pack("V") + "WAVE" + body)
  path
end

# Records what it was asked to transcribe, and answers differently per channel.
class LegTx
  attr_reader :calls
  def initialize(left:, right:, fail_right: false)
    @left = left; @right = right; @fail_right = fail_right; @calls = []
  end
  def transcribe(audio, **kw)
    @calls << { audio: File.basename(audio.to_s), kw: kw }
    return { state: "failed", retryable: true, transcript: nil } if @fail_right && audio.to_s.include?("caller")
    text = audio.to_s.include?("operator") ? @left : audio.to_s.include?("caller") ? @right : "MIXED"
    { state: "transcribed", retryable: false, transcript: text }
  end
end
NOSUM = Object.new
def NOSUM.summarise(*, **) = { ok: false, error: "skip" }

puts "dual-channel: a stereo recording is split and each leg transcribed alone"
Dir.mktmpdir do |dir|
  tx = LegTx.new(left: "[00:00:01] [Speaker 1]: Dobrý den, helpdesk.",
                 right: "[00:00:04] [Speaker 1]: Dobrý den, mám problém.")
  r = Helpdesk::Pipeline.new(transcriber: tx, summariser: NOSUM).process(stereo_file(dir))
  eq(tx.calls.length, 2, "transcribed twice, once per channel")
  eq(tx.calls.map { |c| c[:audio] }.sort, ["caller.wav", "operator.wav"], "each leg sent separately")
  eq(r[:channels], 2, "result records that it was a two-channel recording")
  truthy(r[:transcript].include?("[Operator]:"), "left channel labelled Operator")
  truthy(r[:transcript].include?("[Caller]:"), "right channel labelled Caller")
  truthy(r[:transcript].index("Operator") < r[:transcript].index("Caller"), "merged in time order")
  # Diarisation is used as a SEGMENTER here: its labels are thrown away, but its turn boundaries are
  # the only timestamps the service emits, and pinning a channel to one speaker collapses it to a
  # single block with nothing to interleave on. Measured: 1 speaker -> 1 segment, 4-8 -> 6, same words.
  truthy(tx.calls.all? { |c| c[:kw][:min_speakers] == Helpdesk::Pipeline::SEGMENT_MIN_SPEAKERS &&
                             c[:kw][:max_speakers] == Helpdesk::Pipeline::SEGMENT_MAX_SPEAKERS },
         "each channel is asked for fine segmentation, not for one speaker")
end

puts "dual-channel: every failure still produces a transcript"
Dir.mktmpdir do |dir|
  tx = LegTx.new(left: "[00:00:01] [Speaker 1]: a", right: "x", fail_right: true)
  r = Helpdesk::Pipeline.new(transcriber: tx, summariser: NOSUM).process(stereo_file(dir))
  eq(r[:transcript], "MIXED", "a failed leg falls back to the mixed file")
  eq(r[:channels], nil, "and is not reported as a two-channel result")
end
Dir.mktmpdir do |dir|
  same = "[00:00:01] [Speaker 1]: identical"
  tx = LegTx.new(left: same, right: same)
  r = Helpdesk::Pipeline.new(transcriber: tx, summariser: NOSUM).process(stereo_file(dir))
  eq(r[:transcript], "MIXED", "identical channels mean the split gained nothing -> mixed file")
end

puts "a mono recording is untouched by any of this"
Dir.mktmpdir do |dir|
  mono = File.join(dir, "m.wav")
  pcm = [1, 2, 3].pack("v*")
  body = +"fmt " + [16].pack("V") + [1].pack("v") + [1].pack("v") +
         [16_000].pack("V") + [32_000].pack("V") + [2].pack("v") + [16].pack("v") +
         "data" + [pcm.bytesize].pack("V") + pcm
  File.binwrite(mono, "RIFF" + [4 + body.bytesize].pack("V") + "WAVE" + body)
  tx = LegTx.new(left: "l", right: "r")
  r = Helpdesk::Pipeline.new(transcriber: tx, summariser: NOSUM).process(mono)
  eq(tx.calls.length, 1, "transcribed once, not split")
  eq(r[:transcript], "MIXED", "the file itself was transcribed")
end

puts
puts $fail == 0 ? "All #{$pass} assertions passed." : "#{$fail} FAILED, #{$pass} passed."
exit($fail == 0 ? 0 : 1)