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