narratio/internal/app/whisperx_wiring_test.go

package app

import (
	"context"
	"io"
	"net/http"
	"net/http/httptest"
	"os"
	"path/filepath"
	"slices"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"gitea.maximumdirect.net/eric/narratio/internal/adapters/whisperx"
	"gitea.maximumdirect.net/eric/narratio/internal/config"
	"gitea.maximumdirect.net/eric/narratio/internal/manifest"
	"gitea.maximumdirect.net/eric/narratio/internal/stage"
)

func TestExecuteStagesDefaultWiringUsesWhisperXHTTPClient(t *testing.T) {
	var calls atomic.Int32
	var gotLanguage string
	var gotFileBytes int

	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		calls.Add(1)
		if r.Method != http.MethodPost {
			t.Fatalf("method = %s, want POST", r.Method)
		}
		if err := r.ParseMultipartForm(8 << 20); err != nil {
			t.Fatalf("ParseMultipartForm() error = %v", err)
		}
		gotLanguage = r.FormValue("language")
		file, _, err := r.FormFile("file")
		if err != nil {
			t.Fatalf("FormFile(file) error = %v", err)
		}
		defer file.Close()
		data, err := io.ReadAll(file)
		if err != nil {
			t.Fatalf("ReadAll(file) error = %v", err)
		}
		gotFileBytes = len(data)
		w.Header().Set("Content-Type", "application/json")
		_, _ = w.Write([]byte(`{"source":"httptest","segments":[{"speaker":"alice"}]}`))
	}))
	defer srv.Close()

	cfg := testConfig(t)
	setWhisperXConfig(cfg, srv.URL, 0)

	summary, err := executeStages(context.Background(), cfg, transcribePipelineStages(), RunOptions{})
	if err != nil {
		t.Fatalf("executeStages() error = %v", err)
	}
	if calls.Load() != 1 {
		t.Fatalf("whisperx calls = %d, want 1", calls.Load())
	}
	if gotLanguage != "en" {
		t.Fatalf("language = %q, want en", gotLanguage)
	}
	if gotFileBytes == 0 {
		t.Fatal("audio file payload was empty")
	}

	outPath := filepath.Join(cfg.Pipeline.Workspace.Root, "work", cfg.Session.Campaign, cfg.Session.SessionID, "transcripts", "raw", "alice.json")
	data, err := os.ReadFile(outPath)
	if err != nil {
		t.Fatalf("ReadFile(%q) error = %v", outPath, err)
	}
	got := strings.TrimSpace(string(data))
	if got == `{"schema":"speaker_transcript.v1","segments":[]}` {
		t.Fatalf("got noop transcript output in default wiring: %q", got)
	}
	if !strings.Contains(got, `"source":"httptest"`) {
		t.Fatalf("output = %q, want server response json", got)
	}

	store := &manifest.LocalStore{}
	m, err := store.Load(context.Background(), summary.ManifestPath)
	if err != nil {
		t.Fatalf("Load manifest error = %v", err)
	}
	if m.Stages["transcribe"] == nil || m.Stages["transcribe"].Status != manifest.StatusSucceeded {
		t.Fatalf("transcribe stage = %#v, want succeeded", m.Stages["transcribe"])
	}
}

func TestExecuteStagesDefaultWiringWhisperXFailureMarksManifestFailed(t *testing.T) {
	var calls atomic.Int32
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		calls.Add(1)
		http.Error(w, "temporary", http.StatusInternalServerError)
	}))
	defer srv.Close()

	cfg := testConfig(t)
	setWhisperXConfig(cfg, srv.URL, 1)

	summary, err := executeStages(context.Background(), cfg, transcribePipelineStages(), RunOptions{})
	if err == nil {
		t.Fatal("expected error, got nil")
	}
	if summary != nil {
		t.Fatalf("summary = %#v, want nil on failure", summary)
	}
	if calls.Load() != 2 {
		t.Fatalf("whisperx calls = %d, want 2 (retry + initial)", calls.Load())
	}

	store := &manifest.LocalStore{}
	m, loadErr := store.Load(context.Background(), manifestPathFor(cfg))
	if loadErr != nil {
		t.Fatalf("Load manifest error = %v", loadErr)
	}
	if m.Stages["transcribe"] == nil || m.Stages["transcribe"].Status != manifest.StatusFailed {
		t.Fatalf("transcribe stage = %#v, want failed", m.Stages["transcribe"])
	}
}

func TestExhaustedWhisperXFailurePreservesEarlierSpeakerCheckpoint(t *testing.T) {
	var successful atomic.Bool
	var requestMu sync.Mutex
	var requested []string
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		file, header, err := r.FormFile("file")
		if err != nil {
			t.Errorf("FormFile(file) error = %v", err)
			http.Error(w, "invalid multipart", http.StatusBadRequest)
			return
		}
		_, readErr := io.Copy(io.Discard, file)
		_ = file.Close()
		if readErr != nil {
			t.Errorf("read uploaded file: %v", readErr)
			return
		}
		speaker := strings.TrimSuffix(header.Filename, filepath.Ext(header.Filename))
		requestMu.Lock()
		requested = append(requested, speaker)
		requestMu.Unlock()
		if speaker == "b" && !successful.Load() {
			http.Error(w, "temporary", http.StatusInternalServerError)
			return
		}
		_, _ = w.Write([]byte(`{"segments":[]}`))
	}))
	defer srv.Close()

	cfg, store := setupTranscribeRecoveryFixture(t, []string{"a", "b", "c"})
	setWhisperXConfig(cfg, srv.URL, 1)
	concurrency := 1
	cfg.Pipeline.WhisperX.Concurrency = &concurrency
	httpClient, err := whisperx.NewHTTPClient(whisperx.HTTPClientConfig{
		TranscribeURL: srv.URL,
		Language:      "en",
		Timeout:       2 * time.Second,
		Retries:       1,
		RetryDelay:    time.Millisecond,
	})
	if err != nil {
		t.Fatalf("NewHTTPClient() error = %v", err)
	}
	transcribe := mustSelectStage(t, "transcribe")

	if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
		Env: &Env{ManifestStore: store, WhisperX: httpClient},
	}); err == nil {
		t.Fatal("executeStages() error = nil, want exhausted HTTP failure")
	}
	failed := loadTranscribeRecoveryManifest(t, store, cfg)
	if got := transcribeSpeakerKeys(failed); !slices.Equal(got, []string{"a"}) {
		t.Fatalf("checkpointed speakers = %#v, want a", got)
	}
	requestMu.Lock()
	firstRequests := append([]string(nil), requested...)
	requested = nil
	requestMu.Unlock()
	if !slices.Equal(firstRequests, []string{"a", "b", "b"}) {
		t.Fatalf("initial HTTP speakers = %#v, want a then two attempts for b", firstRequests)
	}

	successful.Store(true)
	if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
		Env: &Env{ManifestStore: store, WhisperX: httpClient},
	}); err != nil {
		t.Fatalf("retry executeStages() error = %v", err)
	}
	requestMu.Lock()
	retryRequests := append([]string(nil), requested...)
	requestMu.Unlock()
	if !slices.Equal(retryRequests, []string{"b", "c"}) {
		t.Fatalf("retry HTTP speakers = %#v, want only b and c", retryRequests)
	}
	completed := loadTranscribeRecoveryManifest(t, store, cfg)
	if got := transcribeSpeakerKeys(completed); !slices.Equal(got, []string{"a", "b", "c"}) {
		t.Fatalf("completed speakers = %#v, want all speakers", got)
	}
}

func TestExecuteStagesExplicitWhisperXInjectionOverridesDefaultWiring(t *testing.T) {
	cfg := testConfig(t)
	setWhisperXConfig(cfg, "https://127.0.0.1:1/transcribe", 0)

	fake := &whisperx.FakeClient{}
	_, err := executeStages(
		context.Background(),
		cfg,
		transcribePipelineStages(),
		RunOptions{Env: &Env{WhisperX: fake}},
	)
	if err != nil {
		t.Fatalf("executeStages() error = %v", err)
	}
	requests := fake.RequestsSnapshot()
	if len(requests) != 1 {
		t.Fatalf("fake whisperx requests = %d, want 1", len(requests))
	}
}

func setWhisperXConfig(cfg *config.Config, url string, retries int) {
	concurrency := 2
	retryDelay := "1ms"
	timeout := "2s"
	cfg.Pipeline.WhisperX = config.WhisperXConfig{
		TranscribeURL: url,
		Language:      "en",
		Timeout:       timeout,
		Retries:       &retries,
		RetryDelay:    retryDelay,
		Concurrency:   &concurrency,
	}
}

func transcribePipelineStages() []stage.Stage {
	all := BuildFullPlan()
	return []stage.Stage{all[0], all[1]}
}