tapir/cmd/tapir/uploads.go

package main

import (
	"context"
	"fmt"
	"log/slog"
	"net/http"
	"os"
	"path/filepath"
	"time"

	"git.d-ma.be/mathias/tapir/internal/adapters/audio"
	"git.d-ma.be/mathias/tapir/internal/adapters/llm"
	"git.d-ma.be/mathias/tapir/internal/adapters/transcribe"
	"git.d-ma.be/mathias/tapir/internal/config"
	"git.d-ma.be/mathias/tapir/internal/reconcile"
	"git.d-ma.be/mathias/tapir/internal/uploads"
	"git.d-ma.be/mathias/tapir/internal/web"
)

// uploadsStore is the store surface the upload wiring needs. *store.Store
// satisfies it.
type uploadsStore interface {
	uploads.Store
	uploads.WorkerStore
	FailInterruptedUploads(ctx context.Context) (int, error)
}

// buildUploads wires audio upload and private transcription (ADR-033). It returns
// nils when uploads are disabled — never a typed-nil Intake, which would mount the
// /uploads routes over nothing. When enabled it, in order: prepares a private spool
// directory, deletes any audio a crashed run left in it, fails the rows such a run
// left queued or transcribing (the audio is not retained, so they can never
// resume), and only then builds the intake and the worker. The caller runs the
// worker. The fallback endpoint is a cloud service; an empty
// TAPIR_TRANSCRIBE_FALLBACK_MODEL keeps private audio on the homelab.
func buildUploads(ctx context.Context, cfg config.Config, st uploadsStore, log *slog.Logger) (*uploads.Intake, *uploads.Worker, error) {
	if !cfg.UploadsEnabled {
		return nil, nil, nil
	}
	if cfg.GatewayURL == "" {
		return nil, nil, fmt.Errorf("uploads: TAPIR_UPLOADS_ENABLED is set but there is no TAPIR_GATEWAY_URL to transcribe through")
	}

	if err := os.MkdirAll(cfg.UploadTmpDir, 0o700); err != nil {
		return nil, nil, fmt.Errorf("uploads: spool dir: %w", err)
	}
	removeSpooledAudio(cfg.UploadTmpDir, log)

	n, err := st.FailInterruptedUploads(ctx)
	if err != nil {
		return nil, nil, fmt.Errorf("uploads: fail interrupted rows: %w", err)
	}
	if n > 0 {
		log.Warn("uploads: marked interrupted by a restart (audio is not retained; they must be re-uploaded)", "count", n)
	}

	worker := uploads.NewWorker(st, newConverter(cfg), buildTranscriber(cfg),
		cfg.UploadTmpDir, cfg.TranscribeTimeout, uploads.WithMaxDuration(cfg.UploadMaxDuration))
	intake := uploads.NewIntake(st, &audio.FFprobe{Bin: cfg.FFprobeBin}, worker,
		uploadLimits(cfg), cfg.UploadTmpDir)
	return intake, worker, nil
}

func uploadLimits(cfg config.Config) uploads.Limits {
	return uploads.Limits{MaxBytes: cfg.UploadMaxBytes, MaxDuration: cfg.UploadMaxDuration}
}

// removeSpooledAudio deletes the spool files this process family creates
// (upload-*, wav-*) and nothing else.
func removeSpooledAudio(dir string, log *slog.Logger) {
	for _, pat := range []string{"upload-*", "wav-*"} {
		matches, _ := filepath.Glob(filepath.Join(dir, pat))
		for _, m := range matches {
			if err := os.Remove(m); err != nil {
				log.Warn("uploads: remove leftover audio", "path", m, "err", err)
			}
		}
	}
}

// wireUploads connects a built Intake to the web App: the routes, the limits the
// form states, and the chat settings for an uploaded recording. A nil intake
// leaves the App untouched — assigning a nil *Intake to the Uploader interface
// would produce a non-nil interface and mount the routes over nothing.
func wireUploads(app *web.App, cfg config.Config, in *uploads.Intake) {
	if in == nil {
		return
	}
	app.Uploads = in
	app.UploadLimits = uploadLimits(cfg)
	app.UploadChatMaxChars = cfg.MaxUploadedTranscriptChars
	app.UploadChatModel = cfg.UploadChatModel
	app.UploadChatModels = uploadChatModels(cfg)
}

// newConverter is the ffmpeg adapter capped at the configured maximum duration, so a file
// whose header lies about its length cannot become hours of WAV (ADR-035 review).
func newConverter(cfg config.Config) *audio.FFmpeg {
	return &audio.FFmpeg{Bin: cfg.FFmpegBin, MaxSeconds: int(cfg.UploadMaxDuration.Seconds())}
}

// adjudicatorTimeout bounds one adjudication call. A call takes ~2 s on gemma4-26b and ~10 s
// on 31b; this is the wait before an unresponsive model is counted as an error.
const adjudicatorTimeout = 2 * time.Minute

// buildTranscriber builds what the worker transcribes with. By default that is the
// single-engine chain (the primary, then the optional cloud fallback). When
// TAPIR_RECONCILE_BASE_MODEL is set it is the two-engine ensemble (ADR-034): the base
// engine and the primary run on the same WAV and a LOCAL model reconciles them. Config
// has already refused any non-local model, so nothing here can send private audio or
// text to a cloud model for reconciliation; the cloud fallback is unchanged.
func buildTranscriber(cfg config.Config) uploads.Transcriber {
	// No overall client timeout: a 90 minute recording takes minutes. The deadline is
	// the worker's per-job context.
	client := &http.Client{}
	// Each LOCAL engine call gets its own deadline: a wedged engine does not answer, and
	// without one it used the whole job before the fallback could run.
	local := func(model string) transcribe.Backend {
		return transcribe.WithTimeout(transcribe.NewGateway(cfg.GatewayURL, cfg.GatewayKey, model, client), cfg.TranscribeEngineTimeout)
	}
	primary := local(cfg.TranscribeModel)
	var fallback transcribe.Backend
	if cfg.TranscribeFallbackModel != "" && cfg.TranscribeFallbackModel != cfg.TranscribeModel {
		fallback = transcribe.NewGateway(cfg.GatewayURL, cfg.GatewayKey, cfg.TranscribeFallbackModel, client)
	}
	if cfg.ReconcileBaseModel == "" {
		return transcribe.NewChain(primary, fallback)
	}

	base := local(cfg.ReconcileBaseModel)
	// Temperature 0 and a short reply: the adjudicator's accuracy was measured that way.
	chat := llm.New(cfg.GatewayURL, cfg.GatewayKey, cfg.ReconcileModel, adjudicatorTimeout,
		llm.WithTemperature(0), llm.WithMaxTokens(200))
	var adj reconcile.Adjudicator = reconcile.NewLLMAdjudicator(chat)
	if cfg.ReconcileCheckOrder {
		adj = reconcile.NewOrderChecked(adj)
	}
	return transcribe.NewEnsemble(base, primary, fallback, adj, reconcile.Options{
		BaseEngine: cfg.ReconcileBaseModel, AltEngine: cfg.TranscribeModel, Reconciler: cfg.ReconcileModel,
	})
}