tapir/internal/adapters/transcribe/ensemble_progress_test.go

package transcribe_test

import (
	"context"
	"errors"
	"strings"
	"sync"
	"testing"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"

	"git.d-ma.be/mathias/tapir/internal/adapters/transcribe"
	"git.d-ma.be/mathias/tapir/internal/domain"
	"git.d-ma.be/mathias/tapir/internal/progress"
	"git.d-ma.be/mathias/tapir/internal/reconcile"
)

// What the Ensemble reports to the page and the log (ADR-035). The reporter here is a
// recording double carried in the context, exactly as the worker's Tracker is.

type recReporter struct {
	mu     sync.Mutex
	events []string
	notes  []map[string]any
}

func (r *recReporter) add(e string) {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.events = append(r.events, e)
}

func (r *recReporter) Begin(key string, _ int)   { r.add("begin:" + key) }
func (r *recReporter) Advance(key string, _ int) { r.add("advance:" + key) }
func (r *recReporter) Done(key string)           { r.add("done:" + key) }
func (r *recReporter) Fail(key string, _ error)  { r.add("fail:" + key) }
func (r *recReporter) Note(msg string, kv ...any) {
	m := map[string]any{"msg": msg}
	for i := 0; i+1 < len(kv); i += 2 {
		m[kv[i].(string)] = kv[i+1]
	}
	r.mu.Lock()
	r.notes = append(r.notes, m)
	r.mu.Unlock()
	r.add("note:" + msg)
}

func (r *recReporter) joined() string {
	r.mu.Lock()
	defer r.mu.Unlock()
	return strings.Join(r.events, ",")
}

func TestEnsemblePlan_DeclaresBothEnginesAndTheReconcileStepsByName(t *testing.T) {
	r := newEnsemble("a", "a", nil, nil, false, &verdictAdj{})

	plan := r.e.Plan()

	require.Len(t, plan, 4)
	assert.Equal(t, []string{"engine_a", "engine_b", "compare", "decide"}, []string{plan[0].Key, plan[1].Key, plan[2].Key, plan[3].Key})
	assert.Contains(t, plan[0].Label, "iguana/pianissimo", "the page says which engine is running")
	assert.Contains(t, plan[1].Label, "iguana/kb-whisper")
}

func TestEnsemblePlan_TheCloudFallbackIsAStepOnlyWhenOneIsConfigured(t *testing.T) {
	with := newEnsemble("a", "a", nil, nil, true, &verdictAdj{}).e.Plan()
	without := newEnsemble("a", "a", nil, nil, false, &verdictAdj{}).e.Plan()

	assert.Equal(t, "fallback", with[len(with)-1].Key)
	assert.Len(t, without, 4, "no fallback configured: the page must not promise one")
}

func TestEnsembleProgress_BothEnginesAreReportedInTheOrderTheyRun(t *testing.T) {
	rep := &recReporter{}
	r := newEnsemble("Totalt 1015 kronor idag.", "Totalt 1050 kronor idag.", nil, nil, false, &verdictAdj{v: reconcile.VerdictAlt})

	_, err := r.e.Transcribe(progress.With(context.Background(), rep), "/a.wav")

	require.NoError(t, err)
	got := rep.joined()
	a, aDone := strings.Index(got, "begin:engine_a"), strings.Index(got, "done:engine_a")
	b, bDone := strings.Index(got, "begin:engine_b"), strings.Index(got, "done:engine_b")
	require.True(t, a >= 0 && aDone > a && b > aDone && bDone > b, "events out of order: %s", got)
	assert.NotContains(t, got, "fail:")
}

func TestEnsembleProgress_AnEngineThatFailsIsShownFailedAndTheOtherStillRuns(t *testing.T) {
	rep := &recReporter{}
	r := newEnsemble("text", "text", errors.New("pianissimo down"), nil, false, &verdictAdj{})

	got, err := r.e.Transcribe(progress.With(context.Background(), rep), "/a.wav")

	require.NoError(t, err, "one engine down is a degraded result, not a failed job")
	assert.Equal(t, domain.ReconcileSingle, got.Reconciliation.Status)
	ev := rep.joined()
	assert.Contains(t, ev, "fail:engine_a")
	assert.Contains(t, ev, "done:engine_b")
	assert.NotContains(t, ev, "begin:compare", "nothing to compare with one engine")
}

func TestEnsembleProgress_TheFallbackIsReportedOnlyWhenBothEnginesFailed(t *testing.T) {
	rep := &recReporter{}
	r := newEnsemble("t", "t", errors.New("a down"), errors.New("b down"), true, &verdictAdj{})

	_, err := r.e.Transcribe(progress.With(context.Background(), rep), "/a.wav")

	require.NoError(t, err)
	ev := rep.joined()
	assert.Contains(t, ev, "fail:engine_a")
	assert.Contains(t, ev, "fail:engine_b")
	assert.Contains(t, ev, "begin:fallback")
	assert.Contains(t, ev, "done:fallback")

	rep2 := &recReporter{}
	ok := newEnsemble("x", "x", nil, nil, true, &verdictAdj{})
	_, err = ok.e.Transcribe(progress.With(context.Background(), rep2), "/a.wav")
	require.NoError(t, err)
	assert.NotContains(t, rep2.joined(), "fallback", "a healthy run never touches the cloud step")
}

// cancellingBackend ends the job in the middle of its own call, like a shutdown.
type cancellingBackend struct {
	name   string
	cancel context.CancelFunc
}

func (c cancellingBackend) Name() string { return c.name }
func (c cancellingBackend) Transcribe(ctx context.Context, _ string) (domain.Transcription, error) {
	c.cancel()
	return domain.Transcription{}, ctx.Err()
}

func TestEnsembleProgress_AShutdownMidEngineIsNotReportedAsAFailedEngine(t *testing.T) {
	rep := &recReporter{}
	ctx, cancel := context.WithCancel(progress.With(context.Background(), rep))
	calls := &[]string{}
	alt := &orderedBackend{name: "iguana/kb-whisper", text: "x", calls: calls}
	e := transcribe.NewEnsemble(cancellingBackend{name: "iguana/pianissimo", cancel: cancel}, alt, nil, &verdictAdj{}, ensembleOpts)

	_, err := e.Transcribe(ctx, "/a.wav")

	require.ErrorIs(t, err, context.Canceled)
	assert.NotContains(t, rep.joined(), "fail:", "a stopped job is not a broken engine: the page must not blame one")
	assert.Empty(t, *calls, "and the second engine never starts")
}

func TestEnsembleProgress_TheReconcileCountsAreNotedForTheOperator(t *testing.T) {
	rep := &recReporter{}
	r := newEnsemble("Totalt 1015 kronor idag.", "Totalt 1050 kronor idag.", nil, nil, false, &verdictAdj{v: reconcile.VerdictAlt})

	_, err := r.e.Transcribe(progress.With(context.Background(), rep), "/a.wav")

	require.NoError(t, err)
	require.Len(t, rep.notes, 1)
	n := rep.notes[0]
	assert.Equal(t, "reconciled", n["msg"])
	assert.Equal(t, 1, n["disputes"])
	assert.Equal(t, 1, n["changed"])
	assert.Equal(t, 0, n["unsure"])
	assert.Equal(t, 0, n["errors"])
}

// An auto-language upload never runs the base engine, the comparison or the decisions, so the
// page must not list them as coming (tapir#61, PR #62 review).
func TestEnsemblePlanFor_AnAutoLanguageListsOnlyThePhasesThatCanRun(t *testing.T) {
	with := newEnsemble("a", "a", nil, nil, true, &verdictAdj{}).e
	without := newEnsemble("a", "a", nil, nil, false, &verdictAdj{}).e
	keys := func(p []progress.Planned) []string {
		var k []string
		for _, s := range p {
			k = append(k, s.Key)
		}
		return k
	}

	assert.Equal(t, []string{"engine_b", "fallback"}, keys(with.PlanFor(domain.LangAuto)))
	assert.Equal(t, []string{"engine_b"}, keys(without.PlanFor(domain.LangAuto)), "no fallback configured: none promised")
	assert.Equal(t, with.Plan(), with.PlanFor(domain.LangSwedish), "Swedish is the plan it always was")
	assert.Contains(t, with.PlanFor(domain.LangAuto)[0].Label, "iguana/kb-whisper")
	assert.Equal(t, with.Plan(), with.PlanFor(""), "a job queued before the choice existed is Swedish")
	assert.Equal(t, with.PlanFor(domain.LangAuto), with.PlanFor(domain.Language("xx")),
		"an unknown language is planned as it is run: not the Swedish phases")
}