diff --git a/.DS_Store b/.DS_Store new file mode 100644 index 0000000..a84cbf7 Binary files /dev/null and b/.DS_Store differ diff --git a/.gitignore b/.gitignore index 6874f32..4d1ff75 100644 --- a/.gitignore +++ b/.gitignore @@ -27,6 +27,7 @@ go.work.sum # Compiled binaries and test configuration narratio +local-test pipeline.yml # ---> VisualStudioCode diff --git a/README.md b/README.md index aaea0db..07024e3 100644 --- a/README.md +++ b/README.md @@ -58,7 +58,7 @@ go run ./cmd/narratio plan --config examples/pipeline.minimal.yml --session exam The current `run` command executes `prepare` + real `transcribe` + placeholder downstream stages and records progress in `manifest.json`. -Note: default CLI wiring currently injects `whisperx.NoopClient` in the runner to avoid implicit network dependency; the real HTTP adapter is implemented under `internal/adapters/whisperx/http.go` and is used via explicit env wiring/tests. +Default CLI wiring builds and uses the real WhisperX HTTP adapter from `pipeline.whisperx` when `whisperx.transcribe_url` is configured. ```bash go run ./cmd/narratio run --config examples/pipeline.minimal.yml --session examples/session.minimal.yml diff --git a/internal/app/commands_test.go b/internal/app/commands_test.go index b36c8d4..846c9ea 100644 --- a/internal/app/commands_test.go +++ b/internal/app/commands_test.go @@ -3,6 +3,8 @@ package app import ( "bytes" "context" + "net/http" + "net/http/httptest" "os" "path/filepath" "strings" @@ -14,7 +16,13 @@ import ( func TestExecuteValidCommands(t *testing.T) { workspaceRoot := t.TempDir() - pipelinePath, sessionPath := writeValidConfigFiles(t, workspaceRoot) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"source":"commands-test","segments":[{"speaker":"alice"}]}`)) + })) + defer srv.Close() + + pipelinePath, sessionPath := writeValidConfigFiles(t, workspaceRoot, srv.URL) manifestPath := writeManifestPathForExecute(t) cases := []struct { @@ -83,7 +91,7 @@ func TestExecuteMissingRequiredFlags(t *testing.T) { func TestExecuteRunStageUnknownFails(t *testing.T) { workspaceRoot := t.TempDir() - pipelinePath, sessionPath := writeValidConfigFiles(t, workspaceRoot) + pipelinePath, sessionPath := writeValidConfigFiles(t, workspaceRoot, "https://example.com/transcribe") var stdout bytes.Buffer var stderr bytes.Buffer @@ -97,6 +105,50 @@ func TestExecuteRunStageUnknownFails(t *testing.T) { } } +func TestExecuteRunStageTranscribeUsesConfiguredWhisperXServer(t *testing.T) { + workspaceRoot := t.TempDir() + var serverCalls int + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + serverCalls++ + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"source":"run-stage-test","segments":[{"speaker":"alice"}]}`)) + })) + defer srv.Close() + + pipelinePath, sessionPath := writeValidConfigFiles(t, workspaceRoot, srv.URL) + + var stdout bytes.Buffer + var stderr bytes.Buffer + + code := Execute([]string{"run-stage", "--config", pipelinePath, "--session", sessionPath, "prepare"}, &stdout, &stderr) + if code != 0 { + t.Fatalf("prepare exit code = %d, want 0; stderr=%q", code, stderr.String()) + } + stdout.Reset() + stderr.Reset() + + code = Execute([]string{"run-stage", "--config", pipelinePath, "--session", sessionPath, "--force", "transcribe"}, &stdout, &stderr) + if code != 0 { + t.Fatalf("transcribe exit code = %d, want 0; stderr=%q", code, stderr.String()) + } + if serverCalls == 0 { + t.Fatal("expected whisperx server to be called at least once") + } + + outPath := filepath.Join(workspaceRoot, "work", "2026-05-03", "transcripts", "raw", "alice.json") + data, err := os.ReadFile(outPath) + if err != nil { + t.Fatalf("ReadFile(%q): %v", outPath, err) + } + got := strings.TrimSpace(string(data)) + if got == `{"schema":"speaker_transcript.v1","segments":[]}` { + t.Fatalf("got noop transcript output: %q", got) + } + if !strings.Contains(got, `"source":"run-stage-test"`) { + t.Fatalf("output = %q, want run-stage server json", got) + } +} + func TestExecuteInvalidCommand(t *testing.T) { var stdout bytes.Buffer var stderr bytes.Buffer @@ -133,19 +185,27 @@ func TestExecuteMissingCommand(t *testing.T) { } } -func writeValidConfigFiles(t *testing.T, workspaceRoot string) (string, string) { +func writeValidConfigFiles(t *testing.T, workspaceRoot string, transcribeURL ...string) (string, string) { t.Helper() dir := t.TempDir() pipelinePath := filepath.Join(dir, "pipeline.yml") sessionPath := filepath.Join(dir, "session.yml") + url := "https://example.com/transcribe" + if len(transcribeURL) > 0 && strings.TrimSpace(transcribeURL[0]) != "" { + url = transcribeURL[0] + } pipelineYAML := `workspace: root: ` + workspaceRoot + ` storage: backend: s3 whisperx: - transcribe_url: https://transcription.ai.rakestrawhome.com/transcribe + transcribe_url: ` + url + ` + timeout: 2s + retries: 0 + retry_delay: 1ms + concurrency: 1 seriatim: timeout: 30s audita: diff --git a/internal/app/resume_run_stage_test.go b/internal/app/resume_run_stage_test.go index 922614b..6630cf7 100644 --- a/internal/app/resume_run_stage_test.go +++ b/internal/app/resume_run_stage_test.go @@ -3,6 +3,8 @@ package app import ( "bytes" "context" + "net/http" + "net/http/httptest" "path/filepath" "strings" "testing" @@ -68,7 +70,12 @@ func TestResumeNoRemainingStages(t *testing.T) { func TestResumeForceRerunsSucceeded(t *testing.T) { workspaceRoot := t.TempDir() - pipelinePath, sessionPath := writeValidConfigFiles(t, workspaceRoot) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"source":"resume-force-test","segments":[{"speaker":"alice"}]}`)) + })) + defer srv.Close() + pipelinePath, sessionPath := writeValidConfigFiles(t, workspaceRoot, srv.URL) manifestPath := filepath.Join(workspaceRoot, "work", "2026-05-03", "manifest.json") store := &manifest.LocalStore{} diff --git a/internal/app/runner.go b/internal/app/runner.go index 27d137a..8e114a4 100644 --- a/internal/app/runner.go +++ b/internal/app/runner.go @@ -6,6 +6,7 @@ import ( "log/slog" "os" "path/filepath" + "strings" "gitea.maximumdirect.net/eric/narratio/internal/adapters/analyzer" "gitea.maximumdirect.net/eric/narratio/internal/adapters/audita" @@ -51,7 +52,11 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage env.Logger = logging.NewLogger(os.Stderr, slog.LevelInfo) } if env.WhisperX == nil { - env.WhisperX = &whisperx.NoopClient{} + client, err := buildDefaultWhisperXClient(env.Config) + if err != nil { + return nil, fmt.Errorf("initialize whisperx client: %w", err) + } + env.WhisperX = client } if env.Seriatim == nil { env.Seriatim = &seriatim.NoopRunner{} @@ -145,6 +150,35 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage }, nil } +func buildDefaultWhisperXClient(cfg *config.Config) (whisperx.Client, error) { + if cfg == nil || cfg.Pipeline == nil { + return &whisperx.NoopClient{}, nil + } + + wx := cfg.Pipeline.WhisperX + if strings.TrimSpace(wx.TranscribeURL) == "" { + // Compatibility fallback for tests or internal call paths that bypass config validation. + return &whisperx.NoopClient{}, nil + } + + retries := 0 + if wx.Retries != nil { + retries = *wx.Retries + } + + client, err := whisperx.NewHTTPClientFromConfigValues( + wx.TranscribeURL, + wx.Language, + wx.Timeout, + wx.RetryDelay, + retries, + ) + if err != nil { + return nil, fmt.Errorf("from pipeline.whisperx: %w", err) + } + return client, nil +} + func loadOrCreateManifest(ctx context.Context, store manifest.Store, path, sessionID string) (*manifest.Manifest, error) { exists, err := fileExists(path) if err != nil { diff --git a/internal/app/whisperx_wiring_test.go b/internal/app/whisperx_wiring_test.go new file mode 100644 index 0000000..4558d19 --- /dev/null +++ b/internal/app/whisperx_wiring_test.go @@ -0,0 +1,157 @@ +package app + +import ( + "context" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "sync/atomic" + "testing" + + "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.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 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) + } + if len(fake.Requests) != 1 { + t.Fatalf("fake whisperx requests = %d, want 1", len(fake.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]} +}