internal/adapters/transcribe/ensemble.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 
 	"git.d-ma.be/mathias/tapir/internal/domain"
+	"git.d-ma.be/mathias/tapir/internal/progress"
 	"git.d-ma.be/mathias/tapir/internal/reconcile"
 )
 
@@ -30,13 +31,30 @@ func NewEnsemble(base, alt, fallback Backend, adj reconcile.Adjudicator, opts re
 	return &Ensemble{base: base, alt: alt, fallback: fallback, adj: adj, opts: opts}
 }
 
-// Transcribe implements uploads.Transcriber.
+// Plan implements uploads.Planner: the phases this ensemble reports, so the page can
+// show them before they run. The fallback is a phase only when one is configured.
+func (e *Ensemble) Plan() []progress.Planned {
+	plan := []progress.Planned{
+		{Key: "engine_a", Label: "Transcribing with " + e.base.Name()},
+		{Key: "engine_b", Label: "Transcribing with " + e.alt.Name()},
+		{Key: "compare", Label: "Comparing the two transcripts"},
+		{Key: "decide", Label: "Deciding the differences"},
+	}
+	if e.fallback != nil {
+		plan = append(plan, progress.Planned{Key: "fallback", Label: "Cloud fallback (" + e.fallback.Name() + ")"})
+	}
+	return plan
+}
+
+// Transcribe implements uploads.Transcriber. Each phase is reported through the
+// progress.Reporter in ctx (a no-op when nobody listens).
 func (e *Ensemble) Transcribe(ctx context.Context, wavPath string) (domain.Transcription, error) {
-	b, berr := e.base.Transcribe(ctx, wavPath)
+	rep := progress.From(ctx)
+	b, berr := e.run(ctx, rep, "engine_a", e.base, wavPath)
 	if err := ctx.Err(); err != nil {
 		return domain.Transcription{}, err
 	}
-	a, aerr := e.alt.Transcribe(ctx, wavPath)
+	a, aerr := e.run(ctx, rep, "engine_b", e.alt, wavPath)
 	if err := ctx.Err(); err != nil {
 		return domain.Transcription{}, err
 	}
@@ -48,6 +66,10 @@ func (e *Ensemble) Transcribe(ctx context.Context, wavPath string) (domain.Trans
 			return domain.Transcription{}, err
 		}
 		out.Endpoint = fmt.Sprintf("%s + %s, reconciled by %s", e.base.Name(), e.alt.Name(), e.opts.Reconciler)
+		if rc := out.Reconciliation; rc != nil {
+			rep.Note("reconciled", "status", string(rc.Status), "disputes", rc.Disputes, "auto_resolved", rc.AutoResolved,
+				"adjudicated", rc.Adjudicated, "changed", rc.Changed, "unsure", rc.Unsure, "errors", rc.Errors)
+		}
 		return out, nil
 	case berr == nil:
 		return e.single(b, e.base.Name(), fmt.Sprintf("%s failed: %v", e.alt.Name(), aerr)), nil
@@ -59,13 +81,27 @@ func (e *Ensemble) Transcribe(ctx context.Context, wavPath string) (domain.Trans
 	if e.fallback == nil {
 		return domain.Transcription{}, both
 	}
-	f, ferr := e.fallback.Transcribe(ctx, wavPath)
+	f, ferr := e.run(ctx, rep, "fallback", e.fallback, wavPath)
 	if ferr != nil {
 		return domain.Transcription{}, errors.Join(both, fmt.Errorf("%s: %w", e.fallback.Name(), ferr))
 	}
 	return e.single(f, e.fallback.Name(), "both local engines failed"), nil
 }
 
+// run is one engine call, reported as a phase. A job that was stopped is not a broken
+// engine, so a cancelled context is never reported as a failed phase.
+func (e *Ensemble) run(ctx context.Context, rep progress.Reporter, key string, be Backend, wav string) (domain.Transcription, error) {
+	rep.Begin(key, 0)
+	t, err := be.Transcribe(ctx, wav)
+	switch {
+	case err == nil:
+		rep.Done(key)
+	case ctx.Err() == nil:
+		rep.Fail(key, err)
+	}
+	return t, err
+}
+
 // single is a transcript from one engine only: nothing to reconcile, and the result
 // says why, because a reader should know the second opinion was missing.
 func (e *Ensemble) single(t domain.Transcription, engine, reason string) domain.Transcription {