transcriptor-ai/cmd/transcriptor-ai/main.go

// Command transcriptor-ai orchestrates local, real-time meeting transcription:
// it drives audiotee for audio capture, qwen-asr for speech recognition, and
// serves the resulting transcript live over SSE. See CLAUDE.md for the full
// architecture and the reasoning behind it.
//
// This is a capture+VAD+ASR+SSE test harness: it runs audiotee, feeds
// detected speech segments to qwen_asr (one subprocess call per segment, one
// worker per track), and publishes the results both to stdout and to an SSE
// endpoint (`curl -N http://<http-addr>/events` is the terminal client for
// now — see CLAUDE.md on why SSE, not WebSocket). Post-processing (fuzzy
// glossary correction, LLM reread) isn't wired in yet.
package main

import (
	"context"
	"flag"
	"fmt"
	"net/http"
	"os"
	"os/signal"
	"path/filepath"
	"sync"
	"syscall"
	"time"

	"github.com/stephanetailland/transcriptor-ai/internal/asr"
	"github.com/stephanetailland/transcriptor-ai/internal/capture"
	"github.com/stephanetailland/transcriptor-ai/internal/transcript"
	"github.com/stephanetailland/transcriptor-ai/internal/vad"
)

func main() {
	if err := run(); err != nil {
		fmt.Fprintln(os.Stderr, "transcriptor-ai:", err)
		os.Exit(1)
	}
}

func run() error {
	audioteeBinary := flag.String("audiotee-binary", "", "path to the audiotee binary (optional; defaults to PATH then ~/bin/audiotee — see internal/capture)")
	asrBinary := flag.String("asr-binary", "", "path to the qwen_asr binary (required)")
	asrModelDir := flag.String("asr-model-dir", "", "path to the qwen_asr model directory (required)")
	prompt := flag.String("prompt", "", "glossary biasing prompt passed to qwen_asr")
	httpAddr := flag.String("http-addr", ":8420", "address to serve the live transcript SSE endpoint on")
	transcriptFile := flag.String("transcript-file", "", "path to append the live transcript to as plain text (optional; not written if empty)")
	captureSystem := flag.Bool("capture-system", true, "transcribe the system audio track (audiotee's system tap always runs regardless of this flag — disabling it only stops transcribing that track, see internal/capture.Config.CaptureMic doc)")
	captureMic := flag.Bool("capture-mic", true, "capture and transcribe the microphone track")
	flag.Parse()

	if *asrBinary == "" || *asrModelDir == "" {
		return fmt.Errorf("both -asr-binary and -asr-model-dir are required")
	}
	if !*captureSystem && !*captureMic {
		return fmt.Errorf("at least one of -capture-system or -capture-mic must be enabled")
	}

	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
	defer stop()

	systemPath := filepath.Join(os.TempDir(), "transcriptor-ai-system.pcm")
	micPath := filepath.Join(os.TempDir(), "transcriptor-ai-mic.pcm")

	systemOut, err := os.Create(systemPath)
	if err != nil {
		return fmt.Errorf("create system output file: %w", err)
	}
	defer systemOut.Close()

	micOut, err := os.Create(micPath)
	if err != nil {
		return fmt.Errorf("create mic output file: %w", err)
	}
	defer micOut.Close()

	var fileWriter *transcript.FileWriter
	if *transcriptFile != "" {
		fileWriter, err = transcript.NewFileWriter(*transcriptFile)
		if err != nil {
			return err
		}
		defer fileWriter.Close()
		fmt.Fprintln(os.Stderr, "transcript file ->", *transcriptFile)
	}

	broadcaster := transcript.NewBroadcaster()
	httpServer := &http.Server{Addr: *httpAddr, Handler: sseMux(broadcaster)}
	httpErrs := make(chan error, 1)
	go func() {
		if err := httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed {
			httpErrs <- err
		}
	}()

	const sampleRate = 16000
	c := capture.New(capture.Config{SampleRate: sampleRate, AudioteePath: *audioteeBinary, CaptureMic: *captureMic})
	if err := c.Start(ctx); err != nil {
		return fmt.Errorf("start capture: %w", err)
	}

	fmt.Fprintln(os.Stderr, "capturing... Ctrl+C to stop")
	fmt.Fprintln(os.Stderr, "system track  ->", systemPath)
	fmt.Fprintln(os.Stderr, "mic track     ->", micPath)
	fmt.Fprintln(os.Stderr, "live transcript (SSE) -> curl -N http://"+httpAddrForDisplay(*httpAddr)+"/events")

	// Tee raw chunks to disk (for listening back later) while also feeding
	// them to VAD. Buffered so the tee doesn't block segmentation.
	tee := make(chan capture.Chunk, 64)
	go func() {
		defer close(tee)
		for chunk := range c.Chunks() {
			switch chunk.Track {
			case capture.TrackSystem:
				systemOut.Write(chunk.Data)
			case capture.TrackMic:
				micOut.Write(chunk.Data)
			}
			tee <- chunk
		}
	}()

	detector := vad.New(vad.DefaultConfig(sampleRate))
	segments := detector.Run(tee)

	// Two independent transcribers (2 tracks, 2 ASR engines — see
	// CLAUDE.md), each processing its own track's segments sequentially. A
	// fresh qwen_asr subprocess per segment is fast enough (~0.15-0.2x
	// realtime measured) that this doesn't need to be more concurrent than
	// that.
	transcriber := asr.New(asr.Config{BinaryPath: *asrBinary, ModelDir: *asrModelDir, Prompt: *prompt})
	// transcribeTrack uses its own background context, not the
	// shutdown-signal ctx above: segments still queued when Ctrl+C arrives
	// (including the ones flushed by c.Stop() below) must still get
	// transcribed during the graceful drain, not fail with "context
	// canceled" because the shutdown signal already canceled ctx.
	asrCtx := context.Background()
	var systemSegs, micSegs chan vad.Segment
	var wg sync.WaitGroup
	if *captureSystem {
		systemSegs = make(chan vad.Segment, 8)
		wg.Add(1)
		go func() { defer wg.Done(); transcribeTrack(asrCtx, transcriber, systemSegs, broadcaster, fileWriter) }()
	}
	if *captureMic {
		micSegs = make(chan vad.Segment, 8)
		wg.Add(1)
		go func() { defer wg.Done(); transcribeTrack(asrCtx, transcriber, micSegs, broadcaster, fileWriter) }()
	}

loop:
	for {
		select {
		case seg, ok := <-segments:
			if !ok {
				break loop
			}
			logSegment(seg)
			dispatch(seg, systemSegs, micSegs)
		case err := <-c.Errs():
			fmt.Fprintln(os.Stderr, "capture error:", err)
		case err := <-httpErrs:
			fmt.Fprintln(os.Stderr, "http server error:", err)
		case <-ctx.Done():
			break loop
		}
	}

	// A graceful shutdown can take a few seconds (draining queued segments
	// through qwen_asr, up to ~2s each on the 1.7B model) with no visible
	// progress otherwise, which reads as "hung" — say so, and let a second
	// Ctrl+C skip the wait and exit immediately instead of forcing the user
	// to keep hitting it.
	fmt.Fprintln(os.Stderr, "shutting down, waiting for in-flight transcriptions to finish (Ctrl+C again to force quit)...")
	forceQuit := make(chan os.Signal, 1)
	signal.Notify(forceQuit, syscall.SIGINT, syscall.SIGTERM)
	go func() {
		<-forceQuit
		fmt.Fprintln(os.Stderr, "\nforcing immediate exit")
		os.Exit(1)
	}()

	c.Stop()
	// Drain any segments flushed by Stop() closing the chunk stream.
	for seg := range segments {
		logSegment(seg)
		dispatch(seg, systemSegs, micSegs)
	}

	// Closing a nil channel panics — only close the ones actually created
	// (i.e. for tracks that were enabled).
	if systemSegs != nil {
		close(systemSegs)
	}
	if micSegs != nil {
		close(micSegs)
	}
	wg.Wait() // let in-flight/queued transcriptions finish before exiting, and
	// before their results are published — otherwise a connected SSE client
	// would miss the tail end of the transcript.

	shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	_ = httpServer.Shutdown(shutdownCtx)

	return nil
}

// dispatch routes seg to the channel for its track. Sending to a nil
// channel (a disabled track — see -capture-system/-capture-mic) blocks
// forever in Go, so it must be checked, not just sent to unconditionally;
// a disabled track's segments are silently dropped (VAD still runs on
// every track's chunks regardless — negligible cost — but a disabled
// track never reaches ASR).
func dispatch(seg vad.Segment, systemSegs, micSegs chan<- vad.Segment) {
	switch seg.Track {
	case capture.TrackSystem:
		if systemSegs != nil {
			systemSegs <- seg
		}
	case capture.TrackMic:
		if micSegs != nil {
			micSegs <- seg
		}
	}
}

func sseMux(b *transcript.Broadcaster) http.Handler {
	mux := http.NewServeMux()
	mux.Handle("/events", b)
	return mux
}

// httpAddrForDisplay turns a ":8420"-style listen address into something
// pasteable ("localhost:8420") for the printed curl hint.
func httpAddrForDisplay(addr string) string {
	if len(addr) > 0 && addr[0] == ':' {
		return "localhost" + addr
	}
	return addr
}

func transcribeTrack(ctx context.Context, t *asr.Transcriber, segs <-chan vad.Segment, b *transcript.Broadcaster, fw *transcript.FileWriter) {
	for seg := range segs {
		text, err := t.Transcribe(ctx, seg.Data)
		if err != nil {
			fmt.Fprintf(os.Stderr, "[%s] asr error: %v\n", seg.Track, err)
			continue
		}
		if text == "" {
			continue
		}
		msg := transcript.Message{Track: seg.Track, Text: text, IsFinal: seg.IsFinal, Timestamp: seg.End}
		b.Publish(msg)
		if fw != nil {
			// Loud, not fatal: losing one line to a transient write error
			// shouldn't take down a live transcription session — but stay
			// visible, since silent data loss would defeat the point of
			// this file existing at all.
			if err := fw.Write(msg); err != nil {
				fmt.Fprintf(os.Stderr, "[%s] transcript file write error: %v\n", seg.Track, err)
			}
		}

		finality := "final"
		if !seg.IsFinal {
			finality = "partial"
		}
		fmt.Printf("[%s/%s] %s\n", seg.Track, finality, text)
	}
}

func logSegment(seg vad.Segment) {
	kind := "final"
	if !seg.IsFinal {
		kind = "force-cut"
	}
	fmt.Fprintf(os.Stderr, "[%s] %-6s dur=%-6s bytes=%d (%s)\n",
		seg.Start.Format("15:04:05.000"), seg.Track, seg.End.Sub(seg.Start).Round(10*time.Millisecond), len(seg.Data), kind)
}