diff --git a/DECISIONS.md b/DECISIONS.md index 69786f3..eb976da 100644 --- a/DECISIONS.md +++ b/DECISIONS.md @@ -1434,7 +1434,12 @@ an hour of Swedish *was* an unknown here; it is now measured, see decision 8.) empty disables), recorded per row, and called out in Open items rather than defaulted silently. - Swedish only (`language=sv`). Language detection and routing (#28 decision 2) are not - touched. + touched. **Amended 2026-10-04 (#61, interim):** the language is now a per-upload choice on + the form, `sv` (preselected, and what an upload that names none gets) or `auto` ("Other": + `language` is omitted so the engine detects it), a closed set held by the form, the request + code and a CHECK constraint on `uploaded_transcripts.language` (migration 021). Automatic + detection and a general-purpose engine for other languages are still not built (#61); an + `auto` upload goes to the same configured engine, which is not tuned for it. - `-H 'Expect:'`, multipart field `file` — encoded in the adapter and its tests, not rediscovered. 7. **Jobs run in-process, one at a time (ADR-018 stands).** Transcription is minutes (~7.5 min @@ -1645,6 +1650,10 @@ on the slow numbers and is reopened: 31B scored 97% against 92%. job timeout rises from 30 to 60 minutes: ~4 min of kb-whisper and ~40 s of pianissimo per hour of audio, plus the adjudication. + **Amended 2026-10-04 (#61, interim):** reconcile is for Swedish. A recording whose language is + `auto` is never sent to the base engine (it is Swedish-only) and is not reconciled: the other + engine transcribes alone, then the cloud fallback if that fails and the job is live, and the + page does not claim a second opinion. 8. **Not decided here.** A cloud adjudicator (rejected: private content); tuning the prompt; making `gemma4-31b` the default; a "reconciling" progress state (the page says "transcribing" for the whole job); feeding the uncertain spots to chat; whether pianissimo should also be a middle tier in the diff --git a/cmd/tapir/uploads_test.go b/cmd/tapir/uploads_test.go index cbedef2..4d1cb36 100644 --- a/cmd/tapir/uploads_test.go +++ b/cmd/tapir/uploads_test.go @@ -25,7 +25,7 @@ type fakeUploadsStore struct { failedKinds []domain.UploadErrorKind } -func (f *fakeUploadsStore) CreateUpload(context.Context, string, string, bool) (string, error) { +func (f *fakeUploadsStore) CreateUpload(context.Context, string, string, domain.Language, bool) (string, error) { return "id", nil } diff --git a/docs/use-cases/audio_upload.feature b/docs/use-cases/audio_upload.feature index d591ed7..7db3219 100644 --- a/docs/use-cases/audio_upload.feature +++ b/docs/use-cases/audio_upload.feature @@ -274,3 +274,20 @@ Feature: Upload an audio recording, get a private Swedish transcript, discuss it Scenario: Input that cannot be right is a client error When the id is not an id, or a header or title holds control bytes Then the answer is a 4xx or the title is cleaned, never a server error + + Scenario: I choose the language of a recording + When I upload a recording and choose Swedish or Other on the form + Then the recording is transcribed as that language and its page says which + + Scenario: Swedish stays the default + When an upload names no language + Then it is transcribed as Swedish, as before the choice existed + + Scenario: A language outside the list is refused + When an upload names a language the form does not offer + Then it is answered 400 before anything is stored or sent to an engine + + Scenario: A recording in another language never reaches the Swedish-only engine + When I choose Other for a recording + Then the language is left for the engine to detect and the Swedish-only base engine and the reconcile are not used + And the engine that does run is the configured one, which is not tuned for other languages diff --git a/internal/adapters/store/migrations/021_uploaded_transcripts_language.down.sql b/internal/adapters/store/migrations/021_uploaded_transcripts_language.down.sql new file mode 100644 index 0000000..bcc9feb --- /dev/null +++ b/internal/adapters/store/migrations/021_uploaded_transcripts_language.down.sql @@ -0,0 +1,2 @@ +ALTER TABLE uploaded_transcripts + DROP COLUMN language; diff --git a/internal/adapters/store/migrations/021_uploaded_transcripts_language.up.sql b/internal/adapters/store/migrations/021_uploaded_transcripts_language.up.sql new file mode 100644 index 0000000..ef2503b --- /dev/null +++ b/internal/adapters/store/migrations/021_uploaded_transcripts_language.up.sql @@ -0,0 +1,13 @@ +-- Migration 021: the language an upload is transcribed as (tapir#61). +-- +-- 'sv' Swedish: the engines are sent language=sv. Every row before this migration was +-- transcribed that way (the request hard-coded it), so the default is true history. +-- 'auto' Other: the engine detects the language and the Swedish-only engine is not used. +-- +-- A closed set held by the table as well as by the form and the request code, so a second +-- writer or a bug cannot store a value the engines are never meant to receive. The column +-- sits on the already-FORCE'd, RLS-scoped uploaded_transcripts, so the exact-set isolation +-- test is unaffected. +ALTER TABLE uploaded_transcripts + ADD COLUMN language TEXT NOT NULL DEFAULT 'sv' + CONSTRAINT uploaded_transcripts_language_check CHECK (language IN ('sv', 'auto')); diff --git a/internal/adapters/store/uploaded_transcripts.go b/internal/adapters/store/uploaded_transcripts.go index 4d30e7a..9d1457e 100644 --- a/internal/adapters/store/uploaded_transcripts.go +++ b/internal/adapters/store/uploaded_transcripts.go @@ -31,12 +31,12 @@ var ErrUploadInProgress = errors.New("store: uploaded transcript is still being // CreateUpload records a new upload in the queued state and returns its id. // hadVideo records that the file carried a video track (only its audio is used); // it is kept for the life of the row so the page can say so. -func (s *Store) CreateUpload(ctx context.Context, userID, title string, hadVideo bool) (string, error) { +func (s *Store) CreateUpload(ctx context.Context, userID, title string, lang domain.Language, hadVideo bool) (string, error) { var id string if err := s.withUser(ctx, userID, func(tx pgx.Tx) error { return tx.QueryRow(ctx, - `INSERT INTO uploaded_transcripts (user_id, title, had_video) VALUES ($1, $2, $3) RETURNING id`, - userID, title, hadVideo).Scan(&id) + `INSERT INTO uploaded_transcripts (user_id, title, language, had_video) VALUES ($1, $2, $3, $4) RETURNING id`, + userID, title, string(lang), hadVideo).Scan(&id) }); err != nil { return "", fmt.Errorf("store: create upload: %w", err) } @@ -221,7 +221,7 @@ func (s *Store) FailInterruptedUploads(ctx context.Context) (int, error) { return total, nil } -const uploadSelect = `SELECT id, user_id, title, status, content, segments, endpoint, +const uploadSelect = `SELECT id, user_id, title, language, status, content, segments, endpoint, audio_seconds, error_kind, had_video, reconciliation, progress, progress_at, alive_at, created_at, finished_at FROM uploaded_transcripts` @@ -234,11 +234,13 @@ func scanUpload(row pgx.Row, u *domain.UploadedTranscript, withText bool) error seconds *float64 finished, progressAt, aliveAt *time.Time ) - if err := row.Scan(&u.ID, &u.UserID, &u.Title, &status, &content, &segments, &endpoint, + var language string + if err := row.Scan(&u.ID, &u.UserID, &u.Title, &language, &status, &content, &segments, &endpoint, &seconds, &errKind, &u.HadVideo, &reconciliation, &progress, &progressAt, &aliveAt, &u.CreatedAt, &finished); err != nil { return err } + u.Language = domain.Language(language) u.Status = domain.UploadStatus(deref(status)) u.Endpoint = deref(endpoint) u.ErrorKind = domain.UploadErrorKind(deref(errKind)) diff --git a/internal/adapters/store/uploaded_transcripts_delete_test.go b/internal/adapters/store/uploaded_transcripts_delete_test.go index fbabf9b..4358739 100644 --- a/internal/adapters/store/uploaded_transcripts_delete_test.go +++ b/internal/adapters/store/uploaded_transcripts_delete_test.go @@ -42,7 +42,7 @@ func TestDeleteUpload_RefusesAnUploadThatIsStillQueuedOrTranscribingAndKeepsTheR s := newStore(t) ctx := context.Background() queued := startedUpload(t, s) - running, err := s.CreateUpload(ctx, userA, "running", false) + running, err := s.CreateUpload(ctx, userA, "running", domain.LangSwedish, false) require.NoError(t, err) require.NoError(t, s.StartTranscribing(ctx, userA, running)) diff --git a/internal/adapters/store/uploaded_transcripts_failguard_test.go b/internal/adapters/store/uploaded_transcripts_failguard_test.go index c8ea345..75e355b 100644 --- a/internal/adapters/store/uploaded_transcripts_failguard_test.go +++ b/internal/adapters/store/uploaded_transcripts_failguard_test.go @@ -34,7 +34,7 @@ func TestFailUpload_StillFailsAQueuedOrTranscribingUpload(t *testing.T) { s := newStore(t) ctx := context.Background() queued := startedUpload(t, s) - running, err := s.CreateUpload(ctx, userA, "r", false) + running, err := s.CreateUpload(ctx, userA, "r", domain.LangSwedish, false) require.NoError(t, err) require.NoError(t, s.StartTranscribing(ctx, userA, running)) diff --git a/internal/adapters/store/uploaded_transcripts_progress_test.go b/internal/adapters/store/uploaded_transcripts_progress_test.go index 7fdf827..623435e 100644 --- a/internal/adapters/store/uploaded_transcripts_progress_test.go +++ b/internal/adapters/store/uploaded_transcripts_progress_test.go @@ -20,7 +20,7 @@ import ( func startedUpload(t *testing.T, s *store.Store) string { t.Helper() seedUsers(t, userA, userB) - id, err := s.CreateUpload(context.Background(), userA, "m", false) + id, err := s.CreateUpload(context.Background(), userA, "m", domain.LangSwedish, false) require.NoError(t, err) return id } diff --git a/internal/adapters/store/uploaded_transcripts_test.go b/internal/adapters/store/uploaded_transcripts_test.go index efa1948..11e49c0 100644 --- a/internal/adapters/store/uploaded_transcripts_test.go +++ b/internal/adapters/store/uploaded_transcripts_test.go @@ -33,7 +33,7 @@ func TestCreateUpload_StartsQueuedAndIsOwnedByTheCreator(t *testing.T) { seedUsers(t, userA) ctx := context.Background() - id, err := s.CreateUpload(ctx, userA, "Teams export", false) + id, err := s.CreateUpload(ctx, userA, "Teams export", domain.LangSwedish, false) require.NoError(t, err) got, err := s.GetUpload(ctx, userA, id) @@ -50,7 +50,7 @@ func TestUploadLifecycle_CompletedKeepsTextSegmentsEndpointAndDuration(t *testin s := newStore(t) seedUsers(t, userA) ctx := context.Background() - id, err := s.CreateUpload(ctx, userA, "m", false) + id, err := s.CreateUpload(ctx, userA, "m", domain.LangSwedish, false) require.NoError(t, err) require.NoError(t, s.StartTranscribing(ctx, userA, id)) @@ -84,7 +84,7 @@ func TestUploadLifecycle_FailedRecordsTypedKindAndNoContent(t *testing.T) { s := newStore(t) seedUsers(t, userA) ctx := context.Background() - id, err := s.CreateUpload(ctx, userA, "m", false) + id, err := s.CreateUpload(ctx, userA, "m", domain.LangSwedish, false) require.NoError(t, err) require.NoError(t, s.StartTranscribing(ctx, userA, id)) @@ -102,7 +102,7 @@ func TestGetUpload_AnotherUsersIDIsNotFound(t *testing.T) { s := newStore(t) seedUsers(t, userA, userB) ctx := context.Background() - id, err := s.CreateUpload(ctx, userA, "A's meeting", false) + id, err := s.CreateUpload(ctx, userA, "A's meeting", domain.LangSwedish, false) require.NoError(t, err) _, err = s.GetUpload(ctx, userB, id) @@ -115,7 +115,7 @@ func TestUploadWrites_ByAnotherUserChangeNothing(t *testing.T) { s := newStore(t) seedUsers(t, userA, userB) ctx := context.Background() - id, err := s.CreateUpload(ctx, userA, "A's meeting", false) + id, err := s.CreateUpload(ctx, userA, "A's meeting", domain.LangSwedish, false) require.NoError(t, err) assert.ErrorIs(t, s.StartTranscribing(ctx, userB, id), store.ErrUploadNotFound) @@ -132,11 +132,11 @@ func TestListUploads_ReturnsOnlyTheCallersNewestFirstWithoutTranscriptText(t *te s := newStore(t) seedUsers(t, userA, userB) ctx := context.Background() - first, err := s.CreateUpload(ctx, userA, "first", false) + first, err := s.CreateUpload(ctx, userA, "first", domain.LangSwedish, false) require.NoError(t, err) - second, err := s.CreateUpload(ctx, userA, "second", false) + second, err := s.CreateUpload(ctx, userA, "second", domain.LangSwedish, false) require.NoError(t, err) - _, err = s.CreateUpload(ctx, userB, "B's", false) + _, err = s.CreateUpload(ctx, userB, "B's", domain.LangSwedish, false) require.NoError(t, err) require.NoError(t, s.StartTranscribing(ctx, userA, first)) require.NoError(t, s.CompleteUpload(ctx, userA, first, domain.Transcription{Text: "long text body", Endpoint: "e"})) @@ -160,12 +160,12 @@ func TestFailInterruptedUploads_FailsOnlyUnfinishedRowsForEveryUser(t *testing.T _, err := super.Exec(ctx, `INSERT INTO user_identities (dex_subject, user_id) VALUES ($1, $2)`, subject, uid) require.NoError(t, err) } - queued, err := s.CreateUpload(ctx, userA, "queued", false) + queued, err := s.CreateUpload(ctx, userA, "queued", domain.LangSwedish, false) require.NoError(t, err) - running, err := s.CreateUpload(ctx, userB, "running", false) + running, err := s.CreateUpload(ctx, userB, "running", domain.LangSwedish, false) require.NoError(t, err) require.NoError(t, s.StartTranscribing(ctx, userB, running)) - done, err := s.CreateUpload(ctx, userA, "done", false) + done, err := s.CreateUpload(ctx, userA, "done", domain.LangSwedish, false) require.NoError(t, err) require.NoError(t, s.StartTranscribing(ctx, userA, done)) require.NoError(t, s.CompleteUpload(ctx, userA, done, domain.Transcription{Text: "kept", Endpoint: "e"})) @@ -191,9 +191,9 @@ func TestCreateUpload_RecordsWhetherTheFileHadAVideoTrack(t *testing.T) { s := newStore(t) seedUsers(t, userA) ctx := context.Background() - withVideo, err := s.CreateUpload(ctx, userA, "camera off", true) + withVideo, err := s.CreateUpload(ctx, userA, "camera off", domain.LangSwedish, true) require.NoError(t, err) - audioOnly, err := s.CreateUpload(ctx, userA, "voice only", false) + audioOnly, err := s.CreateUpload(ctx, userA, "voice only", domain.LangSwedish, false) require.NoError(t, err) gotVideo, err := s.GetUpload(ctx, userA, withVideo) @@ -216,9 +216,9 @@ func TestUploadHadVideo_SurvivesCompletionAndFailure(t *testing.T) { s := newStore(t) seedUsers(t, userA) ctx := context.Background() - done, err := s.CreateUpload(ctx, userA, "d", true) + done, err := s.CreateUpload(ctx, userA, "d", domain.LangSwedish, true) require.NoError(t, err) - failed, err := s.CreateUpload(ctx, userA, "f", true) + failed, err := s.CreateUpload(ctx, userA, "f", domain.LangSwedish, true) require.NoError(t, err) require.NoError(t, s.StartTranscribing(ctx, userA, done)) require.NoError(t, s.CompleteUpload(ctx, userA, done, domain.Transcription{Text: "x", Endpoint: "e"})) @@ -235,7 +235,7 @@ func TestUploadLifecycle_ReconciliationRoundTripsWithItsContestedSpots(t *testin s := newStore(t) seedUsers(t, userA) ctx := context.Background() - id, err := s.CreateUpload(ctx, userA, "m", false) + id, err := s.CreateUpload(ctx, userA, "m", domain.LangSwedish, false) require.NoError(t, err) require.NoError(t, s.StartTranscribing(ctx, userA, id)) want := &domain.Reconciliation{ @@ -258,7 +258,7 @@ func TestUploadLifecycle_ASingleEngineTranscriptHasNoReconciliation(t *testing.T s := newStore(t) seedUsers(t, userA) ctx := context.Background() - id, err := s.CreateUpload(ctx, userA, "m", false) + id, err := s.CreateUpload(ctx, userA, "m", domain.LangSwedish, false) require.NoError(t, err) require.NoError(t, s.StartTranscribing(ctx, userA, id)) require.NoError(t, s.CompleteUpload(ctx, userA, id, domain.Transcription{Text: "x", Endpoint: "e"})) @@ -273,7 +273,7 @@ func TestUploadLifecycle_ASkippedReconciliationKeepsItsReason(t *testing.T) { s := newStore(t) seedUsers(t, userA) ctx := context.Background() - id, err := s.CreateUpload(ctx, userA, "m", false) + id, err := s.CreateUpload(ctx, userA, "m", domain.LangSwedish, false) require.NoError(t, err) require.NoError(t, s.StartTranscribing(ctx, userA, id)) rec := &domain.Reconciliation{Status: domain.ReconcileSkipped, Reason: "the model failed 3 times in a row: gateway down", Errors: 3} @@ -284,3 +284,55 @@ func TestUploadLifecycle_ASkippedReconciliationKeepsItsReason(t *testing.T) { require.NoError(t, err) assert.Equal(t, rec, got.Reconciliation) } + +func TestCreateUpload_RecordsTheLanguageTheUserChoseAndTheListCarriesIt(t *testing.T) { + s := newStore(t) + seedUsers(t, userA) + ctx := context.Background() + swedish, err := s.CreateUpload(ctx, userA, "sv", domain.LangSwedish, false) + require.NoError(t, err) + auto, err := s.CreateUpload(ctx, userA, "auto", domain.LangAuto, false) + require.NoError(t, err) + + gotSv, err := s.GetUpload(ctx, userA, swedish) + require.NoError(t, err) + gotAuto, err := s.GetUpload(ctx, userA, auto) + require.NoError(t, err) + listed, err := s.ListUploads(ctx, userA) + require.NoError(t, err) + + assert.Equal(t, domain.LangSwedish, gotSv.Language) + assert.Equal(t, domain.LangAuto, gotAuto.Language) + byID := map[string]domain.Language{} + for _, u := range listed { + byID[u.ID] = u.Language + } + assert.Equal(t, map[string]domain.Language{swedish: domain.LangSwedish, auto: domain.LangAuto}, byID) +} + +// The closed set is also held by the table, so a bug or a second writer cannot store a +// value the engines are never meant to receive (tapir#61). +func TestUploadedTranscripts_TheTableRefusesALanguageOutsideTheClosedSet(t *testing.T) { + newStore(t) + seedUsers(t, userA) + + _, err := rawPool(t).Exec(context.Background(), + `INSERT INTO uploaded_transcripts (user_id, title, language) VALUES ($1, 'x', 'en')`, userA) + + require.Error(t, err) + assert.Contains(t, err.Error(), "language", "the CHECK constraint names the column") +} + +// A row from before the column existed was transcribed as Swedish; the default says so. +func TestUploadedTranscripts_ARowInsertedWithoutALanguageIsSwedish(t *testing.T) { + s := newStore(t) + seedUsers(t, userA) + var id string + require.NoError(t, rawPool(t).QueryRow(context.Background(), + `INSERT INTO uploaded_transcripts (user_id, title) VALUES ($1, 'old') RETURNING id`, userA).Scan(&id)) + + got, err := s.GetUpload(context.Background(), userA, id) + + require.NoError(t, err) + assert.Equal(t, domain.LangSwedish, got.Language) +} diff --git a/internal/adapters/transcribe/ensemble.go b/internal/adapters/transcribe/ensemble.go index 262e5fd..23854a1 100644 --- a/internal/adapters/transcribe/ensemble.go +++ b/internal/adapters/transcribe/ensemble.go @@ -32,14 +32,24 @@ func NewEnsemble(base, alt, fallback Backend, adj reconcile.Adjudicator, opts re return &Ensemble{base: base, alt: alt, fallback: fallback, adj: adj, opts: opts} } -// 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"}, +// Plan implements uploads.Planner: the phases of a Swedish job, the two-engine one. The +// fallback is a phase only when one is configured. +func (e *Ensemble) Plan() []progress.Planned { return e.PlanFor(domain.LangSwedish) } + +// PlanFor implements uploads.LanguagePlanner. A recording that is not Swedish never runs the +// base engine, the comparison or the decisions (transcribeOther), so the page lists only +// what can run: the other engine, then the fallback if one is configured. +func (e *Ensemble) PlanFor(lang domain.Language) []progress.Planned { + var plan []progress.Planned + if lang == domain.LangSwedish || lang == "" { + 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"}, + } + } else { + plan = []progress.Planned{{Key: "engine_b", Label: "Transcribing with " + e.alt.Name()}} } if e.fallback != nil { plan = append(plan, progress.Planned{Key: "fallback", Label: "Cloud fallback (" + e.fallback.Name() + ")"}) @@ -51,6 +61,9 @@ func (e *Ensemble) Plan() []progress.Planned { // progress.Reporter in ctx (a no-op when nobody listens). func (e *Ensemble) Transcribe(ctx context.Context, wavPath string) (domain.Transcription, error) { rep := progress.From(ctx) + if domain.LanguageFrom(ctx) != domain.LangSwedish { + return e.transcribeOther(ctx, rep, wavPath) + } b, berr := e.run(ctx, rep, "engine_a", e.base, wavPath) if err := ctx.Err(); err != nil { return domain.Transcription{}, err @@ -89,6 +102,28 @@ func (e *Ensemble) Transcribe(ctx context.Context, wavPath string) (domain.Trans return e.single(f, e.fallback.Name(), "both local engines failed"), nil } +// transcribeOther handles a recording that is not Swedish (tapir#61). The base engine is +// Swedish-only, so it never receives the audio, and with one reading there is nothing to +// reconcile: the other engine transcribes alone, then the cloud fallback if it fails and the +// job is still live. The phases that did not run are shown as skipped by the tracker. +func (e *Ensemble) transcribeOther(ctx context.Context, rep progress.Reporter, wavPath string) (domain.Transcription, error) { + t, err := e.run(ctx, rep, "engine_b", e.alt, wavPath) + if err == nil { + t.Endpoint = e.alt.Name() + return t, nil + } + if e.fallback == nil || ctx.Err() != nil { + return domain.Transcription{}, fmt.Errorf("%s: %w", e.alt.Name(), err) + } + f, ferr := e.run(ctx, rep, "fallback", e.fallback, wavPath) + if ferr != nil { + return domain.Transcription{}, errors.Join( + fmt.Errorf("%s: %w", e.alt.Name(), err), fmt.Errorf("%s: %w", e.fallback.Name(), ferr)) + } + f.Endpoint = e.fallback.Name() + return f, 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) { diff --git a/internal/adapters/transcribe/ensemble_progress_test.go b/internal/adapters/transcribe/ensemble_progress_test.go index 4c779cf..caa0cbc 100644 --- a/internal/adapters/transcribe/ensemble_progress_test.go +++ b/internal/adapters/transcribe/ensemble_progress_test.go @@ -160,3 +160,25 @@ func TestEnsembleProgress_TheReconcileCountsAreNotedForTheOperator(t *testing.T) 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") +} diff --git a/internal/adapters/transcribe/ensemble_test.go b/internal/adapters/transcribe/ensemble_test.go index f9feca4..a4dab38 100644 --- a/internal/adapters/transcribe/ensemble_test.go +++ b/internal/adapters/transcribe/ensemble_test.go @@ -204,3 +204,52 @@ func TestEnsemble_AShutdownDuringTheSecondEngineIsAnErrorNotASingleEngineSuccess assert.ErrorIs(t, err, context.Canceled, "a cancelled job must not be recorded as 'kb-whisper failed'") assert.Empty(t, got.Text) } + +// The base engine is Swedish-only. A recording that is not Swedish must never reach it, and +// there is no second opinion to reconcile: the other engine transcribes alone (tapir#61). + +func autoCtx() context.Context { return domain.WithLanguage(context.Background(), domain.LangAuto) } + +func TestEnsemble_AnAutoLanguageUploadNeverReachesTheBaseEngineAndIsNotReconciled(t *testing.T) { + adj := &verdictAdj{v: reconcile.VerdictAlt} + r := newEnsemble("base text", "alt text", nil, nil, true, adj) + + got, err := r.e.Transcribe(autoCtx(), "/a.wav") + + require.NoError(t, err) + assert.Equal(t, []string{"iguana/kb-whisper"}, *r.calls, "only the other engine is called") + assert.Zero(t, adj.calls, "nothing is adjudicated") + assert.Equal(t, "alt text", got.Text) + assert.Equal(t, "iguana/kb-whisper", got.Endpoint) + assert.Nil(t, got.Reconciliation, "reconciliation does not apply, so none is recorded") +} + +func TestEnsemble_AnAutoLanguageUploadFallsBackWhenItsEngineFailsAndStillSkipsTheBaseEngine(t *testing.T) { + r := newEnsemble("base text", "alt text", nil, errors.New("engine down"), true, &verdictAdj{}) + + got, err := r.e.Transcribe(autoCtx(), "/a.wav") + + require.NoError(t, err) + assert.Equal(t, []string{"iguana/kb-whisper", "berget/kb-whisper"}, *r.calls) + assert.Equal(t, "berget/kb-whisper", got.Endpoint) +} + +func TestEnsemble_AnAutoLanguageUploadWithNoFallbackFailsWithoutTouchingTheBaseEngine(t *testing.T) { + r := newEnsemble("base text", "alt text", nil, errors.New("engine down"), false, &verdictAdj{}) + + _, err := r.e.Transcribe(autoCtx(), "/a.wav") + + require.Error(t, err) + assert.Equal(t, []string{"iguana/kb-whisper"}, *r.calls) +} + +func TestEnsemble_ACancelledAutoLanguageJobDoesNotReachTheFallback(t *testing.T) { + r := newEnsemble("base text", "alt text", nil, errors.New("engine down"), true, &verdictAdj{}) + ctx, cancel := context.WithCancel(autoCtx()) + cancel() + + _, err := r.e.Transcribe(ctx, "/a.wav") + + require.Error(t, err) + assert.NotContains(t, *r.calls, "berget/kb-whisper", "a stopped job must not ship its audio to a cloud endpoint") +} diff --git a/internal/adapters/transcribe/gateway.go b/internal/adapters/transcribe/gateway.go index 106798a..a73aa22 100644 --- a/internal/adapters/transcribe/gateway.go +++ b/internal/adapters/transcribe/gateway.go @@ -71,9 +71,20 @@ func (g *Gateway) Transcribe(ctx context.Context, path string) (domain.Transcrip return domain.Transcription{}, fmt.Errorf("transcribe: stat audio: %w", err) } + // The language is one request field and a closed set: sv is sent, auto sends nothing so + // the engine detects it, and anything else never leaves this process (tapir#61). + fields := [][2]string{{"model", g.model}, {"response_format", "verbose_json"}} + switch lang := domain.LanguageFrom(ctx); lang { + case domain.LangSwedish: + fields = append(fields, [2]string{"language", string(lang)}) + case domain.LangAuto: + default: + return domain.Transcription{}, fmt.Errorf("transcribe: language %q is not one tapir sends", lang) + } + var prefix bytes.Buffer mw := multipart.NewWriter(&prefix) - for _, kv := range [][2]string{{"model", g.model}, {"response_format", "verbose_json"}, {"language", "sv"}} { + for _, kv := range fields { if err := mw.WriteField(kv[0], kv[1]); err != nil { return domain.Transcription{}, fmt.Errorf("transcribe: build request: %w", err) } diff --git a/internal/adapters/transcribe/gateway_test.go b/internal/adapters/transcribe/gateway_test.go index 97c655e..53ff943 100644 --- a/internal/adapters/transcribe/gateway_test.go +++ b/internal/adapters/transcribe/gateway_test.go @@ -255,3 +255,48 @@ func TestGateway_AnErrorBodyIsBoundedToTwoHundredBytesInTheStatusError(t *testin require.ErrorAs(t, err, &se) assert.LessOrEqual(t, len(se.Snippet), 200, "a gateway's whole body must not become part of an error") } + +// The language travels on the context (tapir#61); the request field is the only place it +// takes effect, so these observe the request the gateway fake received. + +func TestGateway_SendsLanguageSvWhenTheContextCarriesNoneAndForSwedish(t *testing.T) { + for name, ctx := range map[string]context.Context{ + "no language in the context": context.Background(), + "swedish": domain.WithLanguage(context.Background(), domain.LangSwedish), + } { + t.Run(name, func(t *testing.T) { + var got seen + srv := serve(t, 200, verboseJSON, &got) + g := transcribe.NewGateway(srv.URL+"/v1", "k", "iguana/kb-whisper", srv.Client()) + + _, err := g.Transcribe(ctx, wavFile(t, "RIFF-synthetic-pcm")) + + require.NoError(t, err) + assert.Equal(t, "sv", got.fields["language"]) + }) + } +} + +func TestGateway_OmitsTheLanguageFieldForAnAutoContextSoTheEngineDetectsIt(t *testing.T) { + var got seen + srv := serve(t, 200, verboseJSON, &got) + g := transcribe.NewGateway(srv.URL+"/v1", "k", "iguana/kb-whisper", srv.Client()) + + _, err := g.Transcribe(domain.WithLanguage(context.Background(), domain.LangAuto), wavFile(t, "RIFF-synthetic-pcm")) + + require.NoError(t, err) + _, present := got.fields["language"] + assert.False(t, present, "an auto upload must not be forced to Swedish") + assert.Equal(t, "iguana/kb-whisper", got.fields["model"], "the rest of the request is unchanged") +} + +func TestGateway_ALanguageOutsideTheClosedSetIsRefusedBeforeAnythingIsSent(t *testing.T) { + var got seen + srv := serve(t, 200, verboseJSON, &got) + g := transcribe.NewGateway(srv.URL+"/v1", "k", "iguana/kb-whisper", srv.Client()) + + _, err := g.Transcribe(domain.WithLanguage(context.Background(), domain.Language("en")), wavFile(t, "RIFF-synthetic-pcm")) + + require.Error(t, err) + assert.Empty(t, got.method, "no request may reach the engine with a language the closed set does not hold") +} diff --git a/internal/adapters/transcribe/timeout_test.go b/internal/adapters/transcribe/timeout_test.go index fed4489..7d03167 100644 --- a/internal/adapters/transcribe/timeout_test.go +++ b/internal/adapters/transcribe/timeout_test.go @@ -102,3 +102,45 @@ func TestEnsemble_AHungEngineIsASoftFailureAndTheOtherEnginesTextIsUsed(t *testi assert.Contains(t, got.Reconciliation.Reason, "timeout") assert.False(t, errors.Is(err, context.DeadlineExceeded)) } + +// WithTimeout sits in front of both local engines in production (cmd/tapir/uploads.go). It +// must hand the engine the job's context, not a fresh one: the language and the +// cancellation both travel on it (tapir#61). + +func TestWithTimeout_TheEngineStillReceivesTheLanguageOfTheJob(t *testing.T) { + for name, tt := range map[string]struct { + ctx context.Context + wantField bool + }{ + "auto leaves the field out": {domain.WithLanguage(context.Background(), domain.LangAuto), false}, + "swedish sends sv": {domain.WithLanguage(context.Background(), domain.LangSwedish), true}, + } { + t.Run(name, func(t *testing.T) { + var got seen + srv := serve(t, 200, verboseJSON, &got) + b := transcribe.WithTimeout(transcribe.NewGateway(srv.URL+"/v1", "k", "iguana/kb-whisper", srv.Client()), time.Minute) + + _, err := b.Transcribe(tt.ctx, wavFile(t, "RIFF-synthetic-pcm")) + + require.NoError(t, err) + lang, present := got.fields["language"] + assert.Equal(t, tt.wantField, present) + if tt.wantField { + assert.Equal(t, "sv", lang) + } + }) + } +} + +func TestWithTimeout_ACancelledJobIsNotRestartedByTheWrapper(t *testing.T) { + var got seen + srv := serve(t, 200, verboseJSON, &got) + b := transcribe.WithTimeout(transcribe.NewGateway(srv.URL+"/v1", "k", "iguana/kb-whisper", srv.Client()), time.Minute) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + _, err := b.Transcribe(ctx, wavFile(t, "RIFF-synthetic-pcm")) + + require.Error(t, err) + assert.Empty(t, got.method, "a cancelled job must not send its audio anywhere through the wrapper") +} diff --git a/internal/domain/domain.go b/internal/domain/domain.go index 548b51d..3dfb828 100644 --- a/internal/domain/domain.go +++ b/internal/domain/domain.go @@ -243,9 +243,12 @@ const ( // never stored in, or merged with, the shared public-caption Transcript (ADR-021). // Content and Segments are empty until Status is UploadDone. type UploadedTranscript struct { - ID string - UserID string - Title string + ID string + UserID string + Title string + // Language is what the upload was (or will be) transcribed as: sv, or auto for + // "Other" (tapir#61). + Language Language Status UploadStatus Content string Segments []Segment diff --git a/internal/domain/language.go b/internal/domain/language.go new file mode 100644 index 0000000..def4f07 --- /dev/null +++ b/internal/domain/language.go @@ -0,0 +1,47 @@ +package domain + +import "context" + +// Language is the spoken language an upload is transcribed as. It is chosen by the user on +// the upload form and reaches the engines as one request field, so it is a closed set, +// never free text (tapir#61). +type Language string + +const ( + // LangSwedish sends language=sv: the Swedish engines' own language, and what every + // upload was transcribed as before the choice existed. + LangSwedish Language = "sv" + // LangAuto sends no language, so the engine detects it. It is what "Other" on the form + // means: the Swedish-only engines are not used for it. + LangAuto Language = "auto" +) + +// ParseLanguage reads a form value. An empty value is Swedish (a client that predates the +// field); anything that is not exactly a known value is refused rather than normalised. +func ParseLanguage(s string) (Language, bool) { + switch Language(s) { + case "", LangSwedish: + return LangSwedish, true + case LangAuto: + return LangAuto, true + } + return "", false +} + +type languageKey struct{} + +// WithLanguage returns a context that carries the language of the job it serves. The +// Transcriber interface does not change: engines read it with LanguageFrom, the same way +// they read the progress Reporter (ADR-035). +func WithLanguage(ctx context.Context, l Language) context.Context { + return context.WithValue(ctx, languageKey{}, l) +} + +// LanguageFrom returns the language carried by ctx, or Swedish when it carries none, so a +// caller that sets nothing behaves as it always did. +func LanguageFrom(ctx context.Context) Language { + if l, ok := ctx.Value(languageKey{}).(Language); ok && l != "" { + return l + } + return LangSwedish +} diff --git a/internal/domain/language_test.go b/internal/domain/language_test.go new file mode 100644 index 0000000..ba5db04 --- /dev/null +++ b/internal/domain/language_test.go @@ -0,0 +1,41 @@ +package domain_test + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + + "git.d-ma.be/mathias/tapir/internal/domain" +) + +func TestParseLanguage_KnownValuesAndEmptyAreAcceptedEverythingElseIsNot(t *testing.T) { + tests := []struct { + in string + want domain.Language + wantOK bool + }{ + {"sv", domain.LangSwedish, true}, + {"auto", domain.LangAuto, true}, + {"", domain.LangSwedish, true}, // a client that sends no field keeps today's behaviour + {"SV", "", false}, // exact match: the value reaches a request field + {" sv", "", false}, + {"sv\n", "", false}, + {"en", "", false}, // a specific other language is not offered yet + {"sv,en", "", false}, + {"../sv", "", false}, + } + for _, tt := range tests { + t.Run(tt.in, func(t *testing.T) { + got, ok := domain.ParseLanguage(tt.in) + assert.Equal(t, tt.wantOK, ok) + assert.Equal(t, tt.want, got) + }) + } +} + +func TestLanguageFrom_IsSwedishWhenTheContextCarriesNoneAndOtherwiseWhatWasSet(t *testing.T) { + assert.Equal(t, domain.LangSwedish, domain.LanguageFrom(context.Background())) + ctx := domain.WithLanguage(context.Background(), domain.LangAuto) + assert.Equal(t, domain.LangAuto, domain.LanguageFrom(ctx)) +} diff --git a/internal/uploads/intake.go b/internal/uploads/intake.go index 4de7f31..a75b8b3 100644 --- a/internal/uploads/intake.go +++ b/internal/uploads/intake.go @@ -25,7 +25,7 @@ type Prober interface { // Store is the slice of the private transcript store the intake needs. type Store interface { - CreateUpload(ctx context.Context, userID, title string, hadVideo bool) (string, error) + CreateUpload(ctx context.Context, userID, title string, lang domain.Language, hadVideo bool) (string, error) FailUpload(ctx context.Context, userID, id string, kind domain.UploadErrorKind) error } @@ -35,6 +35,9 @@ type Job struct { UserID string UploadID string Path string + // Language is what the user chose for this recording; empty means Swedish (a job + // queued before the choice existed). + Language domain.Language // Duration is what the file CLAIMED when probed. The worker compares it with what // was really converted; zero means unknown (ADR-035 review). Duration time.Duration @@ -135,7 +138,7 @@ const busyMessage = "Too many recordings are waiting to be transcribed. Please t // validation yields a *RejectError and creates NO row — rejection is synchronous // and leaves nothing behind. On success the temp file is handed to the job and the // worker deletes it; on every failure path Accept deletes it itself. -func (in *Intake) Accept(ctx context.Context, userID, title string, src io.Reader) (string, error) { +func (in *Intake) Accept(ctx context.Context, userID, title string, lang domain.Language, src io.Reader) (string, error) { // Refuse before spooling when the queue has no room, if the submitter can say so: a full // queue otherwise cost up to the upload cap on disk, an ffprobe run and a failed row for // every refused attempt. @@ -190,11 +193,11 @@ func (in *Intake) Accept(ctx context.Context, userID, title string, src io.Reade return "", err } - id, err := in.store.CreateUpload(ctx, userID, title, probe.HasVideo) + id, err := in.store.CreateUpload(ctx, userID, title, lang, probe.HasVideo) if err != nil { return "", fmt.Errorf("uploads: create: %w", err) } - if err := in.sub.Submit(Job{UserID: userID, UploadID: id, Path: path, Duration: probe.Duration}); err != nil { + if err := in.sub.Submit(Job{UserID: userID, UploadID: id, Path: path, Language: lang, Duration: probe.Duration}); err != nil { if errors.Is(err, ErrQueueFull) { // The user's to retry, not a server fault: refuse it as such, and do not leave // the row queued -- that is the stuck spinner. diff --git a/internal/uploads/intake_busy_test.go b/internal/uploads/intake_busy_test.go index 81cc563..bc2c18c 100644 --- a/internal/uploads/intake_busy_test.go +++ b/internal/uploads/intake_busy_test.go @@ -23,7 +23,7 @@ func TestAccept_AFullQueueIsABusyRejectionNotAServerErrorAndLeavesNothingBehind( r := newRig(t) r.sub.err = uploads.ErrQueueFull - _, err := r.in.Accept(context.Background(), "user-1", "t", strings.NewReader("data")) + _, err := r.in.Accept(context.Background(), "user-1", "t", domain.LangSwedish, strings.NewReader("data")) var rej *uploads.RejectError require.True(t, errors.As(err, &rej), "a full queue must be a rejection the web layer can answer with 429: %v", err) @@ -48,7 +48,7 @@ func TestAccept_ANoRoomQueueIsRefusedBeforeAnythingIsSpooledProbedOrStored(t *te full := &roomSubmitter{room: false} in := uploads.NewIntake(r.st, r.pr, full, uploads.Limits{MaxBytes: 1024, MaxDuration: 90 * time.Minute}, r.tmp) - _, err := in.Accept(context.Background(), "user-1", "t", strings.NewReader("data")) + _, err := in.Accept(context.Background(), "user-1", "t", domain.LangSwedish, strings.NewReader("data")) var rej *uploads.RejectError require.True(t, errors.As(err, &rej)) @@ -62,7 +62,7 @@ func TestAccept_ANoRoomQueueIsRefusedBeforeAnythingIsSpooledProbedOrStored(t *te func TestAccept_ASubmitterThatCannotSayStillWorksAsBefore(t *testing.T) { r := newRig(t) - _, err := r.in.Accept(context.Background(), "user-1", "t", strings.NewReader("data")) + _, err := r.in.Accept(context.Background(), "user-1", "t", domain.LangSwedish, strings.NewReader("data")) require.NoError(t, err, "a Submitter without HasRoom is accepted: the check at Submit remains") } @@ -81,7 +81,7 @@ func (b *brokenBody) Read(p []byte) (int, error) { func TestAccept_AnErrorReadingTheClientsBodyIsMarkedAsTheClientsAndAWrappedSubmitErrorIsNot(t *testing.T) { r := newRig(t) - _, err := r.in.Accept(context.Background(), "user-1", "t", &brokenBody{n: 1}) + _, err := r.in.Accept(context.Background(), "user-1", "t", domain.LangSwedish, &brokenBody{n: 1}) require.Error(t, err) assert.ErrorIs(t, err, uploads.ErrClientRead, "a dropped client body is the sender's to retry") @@ -90,7 +90,7 @@ func TestAccept_AnErrorReadingTheClientsBodyIsMarkedAsTheClientsAndAWrappedSubmi // A fault that is ours must NOT carry the marker, even if it wraps the same sentinel. r2 := newRig(t) r2.sub.err = fmt.Errorf("db: %w", io.ErrUnexpectedEOF) - _, err = r2.in.Accept(context.Background(), "user-1", "t", strings.NewReader("data")) + _, err = r2.in.Accept(context.Background(), "user-1", "t", domain.LangSwedish, strings.NewReader("data")) require.Error(t, err) assert.NotErrorIs(t, err, uploads.ErrClientRead) } diff --git a/internal/uploads/intake_concurrency_test.go b/internal/uploads/intake_concurrency_test.go index c245903..41aa88c 100644 --- a/internal/uploads/intake_concurrency_test.go +++ b/internal/uploads/intake_concurrency_test.go @@ -15,6 +15,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "git.d-ma.be/mathias/tapir/internal/domain" "git.d-ma.be/mathias/tapir/internal/uploads" ) @@ -31,7 +32,7 @@ func startSlowReceive(t *testing.T, r *rig, in *uploads.Intake, user string, wan pr, pw := io.Pipe() s := &slowReceive{w: pw, done: make(chan error, 1)} go func() { - _, err := in.Accept(context.Background(), user, "t", pr) + _, err := in.Accept(context.Background(), user, "t", domain.LangSwedish, pr) s.done <- err }() require.Eventually(t, func() bool { return len(r.tmpFiles(t)) == wantFiles }, 2*time.Second, 5*time.Millisecond, @@ -80,7 +81,7 @@ func TestAccept_APerUserLimitOnConcurrentReceivesRefusesTheNextBodyBeforeSpoolin startSlowReceive(t, r, in, "user-1", 1) startSlowReceive(t, r, in, "user-1", 2) - _, err := in.Accept(context.Background(), "user-1", "t", strings.NewReader("data")) + _, err := in.Accept(context.Background(), "user-1", "t", domain.LangSwedish, strings.NewReader("data")) requireBusy(t, err) assert.Len(t, r.tmpFiles(t), 2, "the refused body was never spooled") @@ -98,7 +99,7 @@ func TestAccept_AnOverallLimitOnConcurrentReceivesRefusesTheNextUser(t *testing. startSlowReceive(t, r, in, u, i+1) } - _, err := in.Accept(context.Background(), "d", "t", strings.NewReader("data")) + _, err := in.Accept(context.Background(), "d", "t", domain.LangSwedish, strings.NewReader("data")) requireBusy(t, err) assert.Len(t, r.tmpFiles(t), 3) @@ -112,7 +113,7 @@ func TestAccept_AFinishedReceiveFreesItsSlot(t *testing.T) { first := startSlowReceive(t, r, in, "user-1", 1) require.NoError(t, first.finish(t)) - _, err := in.Accept(context.Background(), "user-1", "t", strings.NewReader("data")) + _, err := in.Accept(context.Background(), "user-1", "t", domain.LangSwedish, strings.NewReader("data")) require.NoError(t, err, "the slot is released when Accept returns") } @@ -122,10 +123,10 @@ func TestAccept_AFailedReceiveFreesItsSlot(t *testing.T) { in := uploads.NewIntake(r.st, r.pr, r.sub, uploads.Limits{MaxBytes: 1024, MaxDuration: time.Hour}, r.tmp) in.MaxReceivesPerUser = 1 in.MaxReceives = 5 - _, err := in.Accept(context.Background(), "user-1", "t", &brokenBody{n: 1}) + _, err := in.Accept(context.Background(), "user-1", "t", domain.LangSwedish, &brokenBody{n: 1}) require.ErrorIs(t, err, uploads.ErrClientRead) - _, err = in.Accept(context.Background(), "user-1", "t", strings.NewReader("data")) + _, err = in.Accept(context.Background(), "user-1", "t", domain.LangSwedish, strings.NewReader("data")) require.NoError(t, err, "a dropped client must not leak a slot") } diff --git a/internal/uploads/intake_defaults_test.go b/internal/uploads/intake_defaults_test.go index c51d2c9..68f8d19 100644 --- a/internal/uploads/intake_defaults_test.go +++ b/internal/uploads/intake_defaults_test.go @@ -12,6 +12,7 @@ import ( "testing" "time" + "git.d-ma.be/mathias/tapir/internal/domain" "git.d-ma.be/mathias/tapir/internal/uploads" ) @@ -21,11 +22,11 @@ func TestNewIntake_DefaultsAllowTwoReceivesPerUserAndFourOverall(t *testing.T) { startSlowReceive(t, r, in, "a", 1) startSlowReceive(t, r, in, "a", 2) - _, err := in.Accept(context.Background(), "a", "t", strings.NewReader("x")) + _, err := in.Accept(context.Background(), "a", "t", domain.LangSwedish, strings.NewReader("x")) requireBusy(t, err) // a third body from one user startSlowReceive(t, r, in, "b", 3) startSlowReceive(t, r, in, "b", 4) - _, err = in.Accept(context.Background(), "c", "t", strings.NewReader("x")) + _, err = in.Accept(context.Background(), "c", "t", domain.LangSwedish, strings.NewReader("x")) requireBusy(t, err) // a fifth body overall } diff --git a/internal/uploads/intake_test.go b/internal/uploads/intake_test.go index 4fc8dbd..020afaf 100644 --- a/internal/uploads/intake_test.go +++ b/internal/uploads/intake_test.go @@ -21,13 +21,15 @@ import ( type fakeStore struct { created []string // titles + langs []domain.Language hadVideo []bool failed map[string]domain.UploadErrorKind nextID string } -func (f *fakeStore) CreateUpload(_ context.Context, _, title string, hadVideo bool) (string, error) { +func (f *fakeStore) CreateUpload(_ context.Context, _, title string, lang domain.Language, hadVideo bool) (string, error) { f.created = append(f.created, title) + f.langs = append(f.langs, lang) f.hadVideo = append(f.hadVideo, hadVideo) return f.nextID, nil } @@ -98,7 +100,7 @@ func TestAccept_ValidAudioIsStoredProbedQueuedAndSubmitted(t *testing.T) { r := newRig(t) audio := bytes.Repeat([]byte("a"), 500) - id, err := r.in.Accept(context.Background(), "user-1", "Teams export", bytes.NewReader(audio)) + id, err := r.in.Accept(context.Background(), "user-1", "Teams export", domain.LangSwedish, bytes.NewReader(audio)) require.NoError(t, err) assert.Equal(t, "up-1", id) @@ -106,7 +108,7 @@ func TestAccept_ValidAudioIsStoredProbedQueuedAndSubmitted(t *testing.T) { assert.Equal(t, audio, r.pr.seen, "the prober must see the complete uploaded bytes, on disk") require.Len(t, r.sub.jobs, 1) job := r.sub.jobs[0] - assert.Equal(t, uploads.Job{UserID: "user-1", UploadID: "up-1", Path: job.Path, Duration: 10 * time.Minute}, job, + assert.Equal(t, uploads.Job{UserID: "user-1", UploadID: "up-1", Path: job.Path, Duration: 10 * time.Minute, Language: domain.LangSwedish}, job, "the job carries what the file CLAIMED, so the worker can compare it with what was converted") assert.Equal(t, filepath.Dir(job.Path), r.tmp, "the temp file lives in the configured directory") got, err := os.ReadFile(job.Path) @@ -117,7 +119,7 @@ func TestAccept_ValidAudioIsStoredProbedQueuedAndSubmitted(t *testing.T) { func TestAccept_OversizeBodyIsRejectedAndLeavesNothingBehind(t *testing.T) { r := newRig(t) - _, err := r.in.Accept(context.Background(), "user-1", "big", strings.NewReader(strings.Repeat("a", 1025))) + _, err := r.in.Accept(context.Background(), "user-1", "big", domain.LangSwedish, strings.NewReader(strings.Repeat("a", 1025))) var rej *uploads.RejectError require.True(t, errors.As(err, &rej)) @@ -132,7 +134,7 @@ func TestAccept_OversizeBodyIsRejectedAndLeavesNothingBehind(t *testing.T) { func TestAccept_BodyExactlyAtTheLimitIsAccepted(t *testing.T) { r := newRig(t) - _, err := r.in.Accept(context.Background(), "user-1", "edge", strings.NewReader(strings.Repeat("a", 1024))) + _, err := r.in.Accept(context.Background(), "user-1", "edge", domain.LangSwedish, strings.NewReader(strings.Repeat("a", 1024))) require.NoError(t, err) } @@ -152,7 +154,7 @@ func TestAccept_ProbeVerdictRejectsWithoutRowAndRemovesTheFile(t *testing.T) { r := newRig(t) r.pr.probe = tt.probe - _, err := r.in.Accept(context.Background(), "user-1", "t", strings.NewReader("data")) + _, err := r.in.Accept(context.Background(), "user-1", "t", domain.LangSwedish, strings.NewReader("data")) var rej *uploads.RejectError require.True(t, errors.As(err, &rej)) @@ -169,7 +171,7 @@ func TestAccept_UnreadableFileIsARejectionButAProberFaultIsAnError(t *testing.T) r := newRig(t) r.pr.err = uploads.ErrUnreadable - _, err := r.in.Accept(context.Background(), "user-1", "t", strings.NewReader("not media")) + _, err := r.in.Accept(context.Background(), "user-1", "t", domain.LangSwedish, strings.NewReader("not media")) var rej *uploads.RejectError require.True(t, errors.As(err, &rej), "the user's file is the problem: tell them") @@ -180,7 +182,7 @@ func TestAccept_UnreadableFileIsARejectionButAProberFaultIsAnError(t *testing.T) r := newRig(t) r.pr.err = errors.New("exec: ffprobe: executable file not found") - _, err := r.in.Accept(context.Background(), "user-1", "t", strings.NewReader("data")) + _, err := r.in.Accept(context.Background(), "user-1", "t", domain.LangSwedish, strings.NewReader("data")) require.Error(t, err) var rej *uploads.RejectError @@ -194,7 +196,7 @@ func TestAccept_ClientDisconnectMidUploadLeavesNothingBehind(t *testing.T) { r := newRig(t) body := io.MultiReader(strings.NewReader("partial"), iotest.ErrReader(errors.New("client went away"))) - _, err := r.in.Accept(context.Background(), "user-1", "t", body) + _, err := r.in.Accept(context.Background(), "user-1", "t", domain.LangSwedish, body) require.Error(t, err) assert.Empty(t, r.pr.paths) @@ -206,7 +208,7 @@ func TestAccept_SubmitFailureFailsTheRowHonestlyAndRemovesTheFile(t *testing.T) r := newRig(t) r.sub.err = errors.New("worker stopped") - _, err := r.in.Accept(context.Background(), "user-1", "t", strings.NewReader("data")) + _, err := r.in.Accept(context.Background(), "user-1", "t", domain.LangSwedish, strings.NewReader("data")) require.Error(t, err) assert.Equal(t, domain.UploadErrInterrupted, r.st.failed["up-1"], @@ -228,7 +230,7 @@ func TestAccept_AFileWithAVideoTrackIsAcceptedAndRecordsThatItHadOne(t *testing. r := newRig(t) r.pr.probe = tt.probe - id, err := r.in.Accept(context.Background(), "user-1", "t", strings.NewReader("data")) + id, err := r.in.Accept(context.Background(), "user-1", "t", domain.LangSwedish, strings.NewReader("data")) require.NoError(t, err) assert.Equal(t, "up-1", id) @@ -238,3 +240,14 @@ func TestAccept_AFileWithAVideoTrackIsAcceptedAndRecordsThatItHadOne(t *testing. }) } } + +func TestAccept_TheChosenLanguageIsStoredAndHandedToTheJob(t *testing.T) { + r := newRig(t) + + _, err := r.in.Accept(context.Background(), "user-1", "t", domain.LangAuto, strings.NewReader("data")) + + require.NoError(t, err) + assert.Equal(t, []domain.Language{domain.LangAuto}, r.st.langs, "the row records what the user chose") + require.Len(t, r.sub.jobs, 1) + assert.Equal(t, domain.LangAuto, r.sub.jobs[0].Language, "the worker is told, so the engines are") +} diff --git a/internal/uploads/worker.go b/internal/uploads/worker.go index a29fecf..917f5fb 100644 --- a/internal/uploads/worker.go +++ b/internal/uploads/worker.go @@ -33,6 +33,13 @@ type Transcriber interface { Transcribe(ctx context.Context, wavPath string) (domain.Transcription, error) } +// LanguagePlanner is a Planner whose phases depend on the language of the job: a recording +// that is not Swedish skips the second engine and the reconcile, and the page should not list +// them as coming (tapir#61). The worker prefers it over Planner. +type LanguagePlanner interface { + PlanFor(lang domain.Language) []progress.Planned +} + // Planner is an optional extra on a Transcriber: it declares its own phases (for // example one per engine) so the page can show them. A Transcriber that does not // implement it gets a single "transcribing" phase from the worker. Whatever a Planner @@ -251,8 +258,14 @@ func (w *Worker) Process(ctx context.Context, job Job) { } plan := []progress.Planned{{Key: "received", Label: "Received"}, {Key: "audio", Label: "Preparing the audio"}} + lang := job.Language + if lang == "" { + lang = domain.LangSwedish // queued before the choice existed + } planner, planned := w.tr.(Planner) - if planned { + if lp, ok := w.tr.(LanguagePlanner); ok { + plan = append(plan, lp.PlanFor(lang)...) + } else if planned { plan = append(plan, planner.Plan()...) } else { plan = append(plan, progress.Planned{Key: "transcribe", Label: "Transcribing"}) @@ -313,7 +326,7 @@ func (w *Worker) Process(ctx context.Context, job Job) { if !planned { tr.Begin("transcribe", 0) } - t, err := w.tr.Transcribe(progress.With(jobCtx, tr), wav) + t, err := w.tr.Transcribe(domain.WithLanguage(progress.With(jobCtx, tr), lang), wav) if err != nil { log.Error("upload: transcribe", "kind", errkind.Of(err), "elapsed", time.Since(start).Round(time.Second)) abort(domain.UploadErrTranscriptionFailed) diff --git a/internal/uploads/worker_progress_test.go b/internal/uploads/worker_progress_test.go index 58faf8c..7168bb7 100644 --- a/internal/uploads/worker_progress_test.go +++ b/internal/uploads/worker_progress_test.go @@ -213,3 +213,47 @@ func TestWorkerProgress_AStoreFailureLogsItsSQLStateNeverTheValueItQuotes(t *tes assert.Contains(t, out, "upload: store transcript") assert.Contains(t, out, "kind=sql_22P02") } + +// langPlannedTr plans per language, like the Ensemble: an auto job has no base engine. +type langPlannedTr struct{ *fakeTranscriber } + +func (langPlannedTr) Plan() []progress.Planned { return plannedTr{}.Plan() } +func (langPlannedTr) PlanFor(l domain.Language) []progress.Planned { + if l == domain.LangAuto { + return []progress.Planned{{Key: "engine_b", Label: "Transcribing with B"}} + } + return plannedTr{}.Plan() +} + +func (p langPlannedTr) Transcribe(ctx context.Context, wav string) (domain.Transcription, error) { + r := progress.From(ctx) + if domain.LanguageFrom(ctx) != domain.LangAuto { + r.Begin("engine_a", 0) + r.Done("engine_a") + } + r.Begin("engine_b", 0) + r.Done("engine_b") + return p.fakeTranscriber.Transcribe(ctx, wav) +} + +func TestWorkerProgress_AJobGetsThePlanOfItsLanguage(t *testing.T) { + for name, tt := range map[string]struct { + lang domain.Language + want []string + }{ + "auto": {domain.LangAuto, []string{"received:done", "audio:done", "engine_b:done", "save:done"}}, + "swedish": {domain.LangSwedish, []string{"received:done", "audio:done", "engine_a:done", "engine_b:done", "save:done"}}, + "no language (queued before the choice existed)": {"", []string{"received:done", "audio:done", "engine_a:done", "engine_b:done", "save:done"}}, + } { + t.Run(name, func(t *testing.T) { + r := newWRig(t, time.Minute) + w := newProgressWorker(t, r, langPlannedTr{r.tr}) + job := r.spool(t, "a") + job.Language = tt.lang + + w.Process(context.Background(), job) + + assert.Equal(t, tt.want, keysAndStates(r.st.lastSteps())) + }) + } +} diff --git a/internal/uploads/worker_test.go b/internal/uploads/worker_test.go index 728e404..66b722e 100644 --- a/internal/uploads/worker_test.go +++ b/internal/uploads/worker_test.go @@ -126,6 +126,7 @@ type fakeTranscriber struct { result domain.Transcription err error paths []string + langs []domain.Language // what each call's context carried // block, when non-nil, holds every call until it is closed or ctx ends. block chan struct{} running atomic.Int32 @@ -134,6 +135,7 @@ type fakeTranscriber struct { func (f *fakeTranscriber) Transcribe(ctx context.Context, wav string) (domain.Transcription, error) { f.paths = append(f.paths, wav) + f.langs = append(f.langs, domain.LanguageFrom(ctx)) n := f.running.Add(1) defer f.running.Add(-1) for { @@ -337,3 +339,24 @@ func (dbErr) Error() string { return `ERROR: invalid input syntax for type json: "zebrakaka secret words"` } func (dbErr) SQLState() string { return "22P02" } + +func TestWorker_TheTranscriberSeesTheLanguageOfTheJobOnItsContext(t *testing.T) { + for name, tt := range map[string]struct { + job domain.Language + want domain.Language + }{ + "auto": {domain.LangAuto, domain.LangAuto}, + "swedish": {domain.LangSwedish, domain.LangSwedish}, + "a job with no language": {"", domain.LangSwedish}, // queued before the field existed + } { + t.Run(name, func(t *testing.T) { + r := newWRig(t, time.Minute) + job := r.spool(t, "a") + job.Language = tt.job + + r.w.Process(context.Background(), job) + + assert.Equal(t, []domain.Language{tt.want}, r.tr.langs) + }) + } +} diff --git a/internal/web/uploads.go b/internal/web/uploads.go index 921a42f..0560088 100644 --- a/internal/web/uploads.go +++ b/internal/web/uploads.go @@ -28,7 +28,7 @@ import ( // *uploads.Intake satisfies it. It is the whole of what the handlers know about // validation, spooling and queueing. type Uploader interface { - Accept(ctx context.Context, userID, title string, src io.Reader) (id string, err error) + Accept(ctx context.Context, userID, title string, lang domain.Language, src io.Reader) (id string, err error) } const ( @@ -36,6 +36,8 @@ const ( // and the title field; the cap itself is enforced on the file part by Intake. uploadOverhead = 1 << 20 maxTitleRunes = 200 + // maxLanguageBytes bounds what is read of the language field; a valid value is at most four bytes, so a longer one fails the parse. + maxLanguageBytes = 16 // uploadPollEvery is how often an unfinished upload re-asks for its status. uploadPollEvery = "2s" // serviceSilentAfter is how long a job may go without any signal (heartbeat or @@ -89,6 +91,7 @@ func (a *App) handleUploadPost(w http.ResponseWriter, r *http.Request) { return } title := "" + lang := domain.LangSwedish for { part, err := mr.NextPart() if errors.Is(err, io.EOF) { @@ -121,11 +124,29 @@ func (a *App) handleUploadPost(w http.ResponseWriter, r *http.Request) { return } title = cleanTitle(string(b)) + case "language": + // A closed set read before the file: anything else is refused before the file + // is spooled, so a crafted value never reaches an engine (tapir#61). + b, err := io.ReadAll(io.LimitReader(part, maxLanguageBytes+1)) + if err != nil { + if a.abandoned(w, r) { + return + } + a.renderUploadsPage(w, r, userID, http.StatusBadRequest, "That upload could not be read. Please try again.") + return + } + l, ok := domain.ParseLanguage(string(b)) + if !ok { + a.logger().Info("upload rejected", "reason", "language") + a.renderUploadsPage(w, r, userID, http.StatusBadRequest, "Choose a language from the list.") + return + } + lang = l case "file": if title == "" { title = titleFromFilename(part.FileName()) } - id, err := a.Uploads.Accept(r.Context(), userID, title, part) + id, err := a.Uploads.Accept(r.Context(), userID, title, lang, part) if err != nil { a.uploadFailure(w, r, userID, err) return @@ -300,6 +321,14 @@ func cloudProvider(endpoint string) (string, bool) { return "", false } +// languageLabel is how the page names the language a recording was transcribed as. +func languageLabel(l domain.Language) string { + if l == domain.LangAuto { + return "Other (detected by the engine)" + } + return "Swedish" +} + // hasNote reports whether any phase carries a note. A note is the only warning for a // recording that is shorter than its file claimed, so a record that has one is shown open. func hasNote(steps []domain.ProgressStep) bool { diff --git a/internal/web/uploads.templ b/internal/web/uploads.templ index bc20c1f..60f9e4b 100644 --- a/internal/web/uploads.templ +++ b/internal/web/uploads.templ @@ -24,6 +24,14 @@ templ UploadsPage(v uploadsView) { }

+

+ +

@@ -49,6 +57,7 @@ templ UploadPage(u domain.UploadedTranscript, chatEnabled bool, now time.Time) { @Layout("Tapir — " + u.Title) {

{ u.Title }

+

Language: { languageLabel(u.Language) }

@uploadStatus(u, chatEnabled, now)

All recordings

diff --git a/internal/web/uploads_disconnect_handler_test.go b/internal/web/uploads_disconnect_handler_test.go index a5adb69..00f47fb 100644 --- a/internal/web/uploads_disconnect_handler_test.go +++ b/internal/web/uploads_disconnect_handler_test.go @@ -14,6 +14,8 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + + "git.d-ma.be/mathias/tapir/internal/domain" ) // A client that hangs up in the middle of an upload is routine, not a server fault. The @@ -85,7 +87,7 @@ func TestUploadPost_AClientThatHangsUpAnywhereInTheBodyIsLoggedAtInfoAndNeverAsA // context is cancelled, then Accept fails the way a read from a dead connection does. type goneMidAccept struct{ cancel context.CancelFunc } -func (g goneMidAccept) Accept(context.Context, string, string, io.Reader) (string, error) { +func (g goneMidAccept) Accept(context.Context, string, string, domain.Language, io.Reader) (string, error) { g.cancel() return "", fmt.Errorf("receive: %w", io.ErrClosedPipe) } diff --git a/internal/web/uploads_handler_test.go b/internal/web/uploads_handler_test.go index b130ff1..806d5cc 100644 --- a/internal/web/uploads_handler_test.go +++ b/internal/web/uploads_handler_test.go @@ -16,6 +16,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "git.d-ma.be/mathias/tapir/internal/domain" "git.d-ma.be/mathias/tapir/internal/uploads" "git.d-ma.be/mathias/tapir/internal/web" ) @@ -132,7 +133,7 @@ func TestUploadPost_AcceptedAudioQueuesARowAndRedirectsToIt(t *testing.T) { assert.Equal(t, "Sprint review", title) assert.Equal(t, "/uploads/"+id, rec.Header().Get("Location")) require.Len(t, sub.jobs, 1) - assert.Equal(t, uploads.Job{UserID: userID, UploadID: id, Path: sub.jobs[0].Path, Duration: audioOK.Duration}, sub.jobs[0]) + assert.Equal(t, uploads.Job{UserID: userID, UploadID: id, Path: sub.jobs[0].Path, Language: domain.LangSwedish, Duration: audioOK.Duration}, sub.jobs[0]) spooled, err := os.ReadFile(sub.jobs[0].Path) require.NoError(t, err) assert.Equal(t, "synthetic audio bytes", string(spooled)) diff --git a/internal/web/uploads_language_handler_test.go b/internal/web/uploads_language_handler_test.go new file mode 100644 index 0000000..73ea735 --- /dev/null +++ b/internal/web/uploads_language_handler_test.go @@ -0,0 +1,151 @@ +package web_test + +// The language of a recording is chosen on the upload form (tapir#61). These drive the real +// Router, Intake and Postgres store; the submitter is the process boundary and records the +// job the worker would receive. + +import ( + "bytes" + "context" + "mime/multipart" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "git.d-ma.be/mathias/tapir/internal/domain" +) + +// langReq builds an upload whose parts come in the order given: "title", "language" and +// "file" (the file part always carries the same synthetic bytes). +func langReq(t *testing.T, order []string, language string) *http.Request { + t.Helper() + var buf bytes.Buffer + mw := multipart.NewWriter(&buf) + for _, part := range order { + switch part { + case "title": + require.NoError(t, mw.WriteField("title", "m")) + case "language": + require.NoError(t, mw.WriteField("language", language)) + case "file": + fw, err := mw.CreateFormFile("file", "m.wav") + require.NoError(t, err) + _, err = fw.Write([]byte("RIFF....WAVE")) + require.NoError(t, err) + } + } + require.NoError(t, mw.Close()) + req := httptest.NewRequest(http.MethodPost, "/uploads", &buf) + req.Header.Set("Content-Type", mw.FormDataContentType()) + return req +} + +func rowLanguage(t *testing.T, location string) domain.Language { + t.Helper() + id := strings.TrimPrefix(location, "/uploads/") + got, err := newStore(t).GetUpload(context.Background(), userID, id) + require.NoError(t, err) + return got.Language +} + +func TestUploadPost_TheLanguageFieldReachesTheJobAndTheRow(t *testing.T) { + for _, tt := range []struct { + name string + language string + want domain.Language + }{ + {"other", "auto", domain.LangAuto}, + {"swedish", "sv", domain.LangSwedish}, + } { + t.Run(tt.name, func(t *testing.T) { + app, sub := newUploadApp(t, audioOK, 200<<20) + + rec := do(t, app, langReq(t, []string{"title", "language", "file"}, tt.language)) + + require.Equal(t, http.StatusSeeOther, rec.Code) + require.Len(t, sub.jobs, 1) + assert.Equal(t, tt.want, sub.jobs[0].Language) + assert.Equal(t, tt.want, rowLanguage(t, rec.Header().Get("Location"))) + }) + } +} + +func TestUploadPost_NoLanguageFieldIsSwedishAsBefore(t *testing.T) { + app, sub := newUploadApp(t, audioOK, 200<<20) + + rec := do(t, app, langReq(t, []string{"title", "file"}, "")) + + require.Equal(t, http.StatusSeeOther, rec.Code) + require.Len(t, sub.jobs, 1) + assert.Equal(t, domain.LangSwedish, sub.jobs[0].Language) + assert.Equal(t, domain.LangSwedish, rowLanguage(t, rec.Header().Get("Location"))) +} + +// A part after the file is never read (the file part ends the stream), so a client that +// orders its fields wrongly gets Swedish, the old behaviour, not a guess. +func TestUploadPost_ALanguageFieldAfterTheFileIsNotReadAndTheUploadIsSwedish(t *testing.T) { + app, sub := newUploadApp(t, audioOK, 200<<20) + + rec := do(t, app, langReq(t, []string{"title", "file", "language"}, "auto")) + + require.Equal(t, http.StatusSeeOther, rec.Code) + require.Len(t, sub.jobs, 1) + assert.Equal(t, domain.LangSwedish, sub.jobs[0].Language) +} + +func TestUploadPost_ALanguageOutsideTheClosedSetIsRefusedBeforeAnythingIsSpooledOrStored(t *testing.T) { + for _, bad := range []string{"en", "SV", " sv", "sv,auto", "../auto", strings.Repeat("a", 5000)} { + t.Run(bad[:min(len(bad), 12)], func(t *testing.T) { + app, sub := newUploadApp(t, audioOK, 200<<20) + + rec := do(t, app, langReq(t, []string{"title", "language", "file"}, bad)) + + assert.Equal(t, http.StatusBadRequest, rec.Code) + assert.Empty(t, sub.jobs, "a refused post must not enqueue transcription") + rows, err := newStore(t).ListUploads(context.Background(), userID) + require.NoError(t, err) + assert.Empty(t, rows, "and must leave no row") + assert.Contains(t, body(t, rec), "Choose a language from the list.") + }) + } +} + +func TestUploadPage_TheFormOffersSwedishPreselectedAndOtherBeforeTheFilePart(t *testing.T) { + app, _ := newUploadApp(t, audioOK, 200<<20) + + b := body(t, do(t, app, get("/uploads"))) + + sel := strings.Index(b, `name="language"`) + file := strings.Index(b, `name="file"`) + require.True(t, sel >= 0, "the form has a language field") + assert.True(t, sel < file, "the field must come before the file part: the stream ends at the file") + assert.Regexp(t, `