From d09aff68740b2ce38cdfb54cf01aaec1d3903cc9 Mon Sep 17 00:00:00 2001 From: m3ta-chiron Date: Sun, 18 Oct 2026 17:10:36 +0200 Subject: [PATCH] feat(talkd): Daemon mit HTTP-API auf 127.0.0.1, Queue-Betrieb und Shutdown MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - HTTP-Server nur auf 127.0.0.1 (Port aus config server.port/TALK_PORT), klarer Fehler wenn der Port belegt ist - Ansagen laufen über die unveränderte Provider-Kette (pro Item gebaut, Pin/Voice je Anfrage); /status meldet den aktiven Provider - SIGINT/SIGTERM: erst HTTP dicht, dann Queue schließen (laufende Ansage wird abgebrochen, Worker-Exit wird abgewartet) - Mute ohne Config-Dir: dokumentierte Warnung statt Absturz --- cmd/talkd/main.go | 249 ++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 242 insertions(+), 7 deletions(-) diff --git a/cmd/talkd/main.go b/cmd/talkd/main.go index a4b709b..1115875 100644 --- a/cmd/talkd/main.go +++ b/cmd/talkd/main.go @@ -1,13 +1,248 @@ -// talkd ist der Daemon für talk 2.0 (Queue, Mute-State, HTTP-API, MCP). +// talkd ist der Daemon für talk 2.0: HTTP-API auf 127.0.0.1 (Default-Port +// 7800, server.port/TALK_PORT), serialisierte Ansage-Queue mit Prioritäten +// und now-Sprung, persistenter Mute-State (Datei mute im Config-Dir). // -// Dieser Stand (Task 01) ist ein Platzhalter: talkd startet, loggt eine Zeile -// und beendet sich mit Exit-Code 0. Die eigentliche Daemon-Implementierung -// (HTTP-Server auf 127.0.0.1, serialisierte Ansage-Queue, persistentes Mute) -// entsteht in Task 04. +// Task 04: POST /speak, GET /status, POST /cancel, POST /mute, POST /unmute. +// Rate-Limit/Ruhezeiten folgen in Task 05, CORS/MCP in Task 06 — die +// Rate-Limit-Struktur existiert bereits (internal/queue/rate.go), ist hier +// aber bewusst noch nicht angebunden (Task-04-Default: unbegrenzt). package main -import "log" +import ( + "context" + "fmt" + "log" + "net" + "net/http" + "os" + "os/signal" + "path/filepath" + "strings" + "sync" + "syscall" + "time" + + "code.m3ta.dev/m3tam3re/talk/internal/api" + "code.m3ta.dev/m3tam3re/talk/internal/audio" + "code.m3ta.dev/m3tam3re/talk/internal/config" + "code.m3ta.dev/m3tam3re/talk/internal/langdetect" + "code.m3ta.dev/m3tam3re/talk/internal/provider" + "code.m3ta.dev/m3tam3re/talk/internal/queue" +) + +// Build-Informationen. Werden beim Release per -ldflags injiziert (gleicher +// Vertrag wie cmd/talk, siehe .goreleaser.yaml in Task 11). Default: dev. +var ( + version = "dev" + commit = "" +) func main() { - log.Println("talkd: Platzhalter — Daemon-Implementierung folgt in Task 04") + log.SetFlags(log.LstdFlags) + log.SetPrefix("talkd: ") + if err := run(); err != nil { + log.Fatal(err) + } +} + +// run startet den Daemon und kehrt erst beim Herunterfahren (SIGINT/ +// SIGTERM) zurück. Abbrüche der laufenden Ansage beim Shutdown sind +// beabsichtigt: talkd soll auf Signale prompt reagieren. +func run() error { + getenv := os.Getenv + + cfg, err := config.Load(config.Options{Getenv: getenv}) + if err != nil { + return err + } + + // Audio-Sink (audio.sink; "null" verwirft tonlos — CI/Server). + // Ketten-, Queue- und Sink-Protokoll laufen über denselben Logger. + lw := lineLogger{} + sink, err := audio.NewSink(cfg.Audio.Sink, lw) + if err != nil { + return err + } + + // Basiskette nur für die /status-Auflösung „aktiver Provider" (erste + // verfügbare Stufe; Available ist billig — Key-/Endpunkt-Binary-Checks + // ohne Netz-Call). Das eigentliche Abspielen baut pro Item eine eigene + // Kette (Pin/Voice je Anfrage) — die Kette aus internal/provider wird + // unverändert wiederverwendet. + baseChain, err := provider.NewChain(chainConfigFor(cfg, "", "", getenv, nil)) + if err != nil { + return err + } + + state := &providerState{} + speak := func(ctx context.Context, item queue.Item) error { + state.set(resolveProviderName(baseChain, item)) + lang := item.Lang + if lang == "" { + lang = langdetect.Detect(item.Text) + } + chain, err := provider.NewChain(chainConfigFor(cfg, item.Provider, item.Voice, getenv, sink)) + if err != nil { + return err + } + return chain.Speak(ctx, item.Text, lang) + } + + q := queue.New(speak, lw) + + // Persistenter Mute-State (Datei mute im Config-Dir). Ohne Config-Dir + // läuft der Daemon ohne Mute weiter (dokumentierte Warnung) statt zu + // scheitern — alles andere an talk darf ja auch ohne Config-Datei. + var mutePath string + if dir, err := config.Dir(getenv); err == nil { + mutePath = filepath.Join(dir, queue.MuteFileName) + } else { + log.Printf("warnung: Config-Dir nicht bestimmbar — Mute-State deaktiviert (%v)", err) + } + mute := queue.MuteStore{Path: mutePath} + + handler := api.NewHandler(api.HandlerConfig{ + Queue: q, + Mute: mute, + Version: versionString(), + KnownProviders: provider.Names(), + Provider: func() string { + if cur, ok := q.Current(); ok { + return resolveProviderName(baseChain, cur) + } + if last := state.get(); last != "" { + return last + } + return resolveProviderName(baseChain, queue.Item{}) + }, + }) + + addr := fmt.Sprintf("127.0.0.1:%d", cfg.Server.Port) + ln, err := net.Listen("tcp", addr) + if err != nil { + return fmt.Errorf("Port belegen (%s): %w — läuft bereits ein talkd?", addr, err) + } + srv := &http.Server{Handler: handler, ReadHeaderTimeout: 5 * time.Second} + errCh := make(chan error, 1) + go func() { errCh <- srv.Serve(ln) }() + log.Printf("listening on http://%s (version %s, audio sink %s, mute=%v)", + addr, versionString(), sink.Name(), mute.Muted()) + + sig := make(chan os.Signal, 1) + signal.Notify(sig, os.Interrupt, syscall.SIGTERM) + + select { + case err := <-errCh: + q.Close() + return fmt.Errorf("http server: %w", err) + case s := <-sig: + log.Printf("Signal %v empfangen — fahre herunter (laufende Ansage wird abgebrochen)", s) + } + + // Reihenfolge: erst HTTP dicht (keine neuen Anfragen), dann die Queue + // schließen (laufende Ansage abbrechen, Worker beenden und darauf + // warten). So läuft nie eine Ansage nach dem Daemon-Ende weiter. + shutdownCtx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + if err := srv.Shutdown(shutdownCtx); err != nil { + log.Printf("http shutdown: %v", err) + } + q.Close() + return nil +} + +// versionString rendert die Daemon-Version für /status. +func versionString() string { + if commit == "" { + return version + } + return fmt.Sprintf("%s (%s)", version, commit) +} + +// chainConfigFor überführt Config + Item-Pins in die ChainConfig — dieselbe +// Zusammensetzung wie im CLI-Pfad (cmd/talk/say.go), nur dass der Daemon +// sie pro Ansage baut (Pin/Voice sind Anfrage-Properties). player ist nur +// für die Abspiel-Kette gesetzt (die Basiskette braucht ihn nicht). +func chainConfigFor(cfg config.Config, pin, voice string, getenv func(string) string, player provider.Player) provider.ChainConfig { + return provider.ChainConfig{ + Pin: pickProvider(pin, cfg.Provider), + Voice: voice, + Voices: cfg.Voice, + Endpoints: endpointsOf(cfg), + Model: cfg.Providers["openai"].Model, + Getenv: getenv, + Player: player, + Log: lineLogger{}, + } +} + +// pickProvider löst den Provider-Pin auf: Anfrage-/Flag-Pin schlägt +// `provider = "…"` aus der config.toml; leer = komplette Default-Kette. +func pickProvider(pin, cfgValue string) string { + if pin != "" { + return pin + } + return cfgValue +} + +// endpointsOf überführt die Provider-Einstellungen der Config in die +// Endpunkt-Map der Kette. +func endpointsOf(cfg config.Config) map[string]string { + endpoints := make(map[string]string, len(cfg.Providers)) + for name, settings := range cfg.Providers { + if settings.Endpoint != "" { + endpoints[name] = settings.Endpoint + } + } + return endpoints +} + +// resolveProviderName löst den „aktiven Provider" für /status auf: Pin des +// Items, sonst erste verfügbare Stufe der Basiskette. Available() ist +// bewusst billig (Key-/Binary-Prüfung ohne Netz-Call), deshalb darf das +// /status on demand aufrufen. +func resolveProviderName(chain *provider.Chain, item queue.Item) string { + if item.Provider != "" { + return item.Provider + } + for _, p := range chain.Tiers() { + if p.Available(context.Background()) { + return p.Name() + } + } + return "keine" +} + +// providerState merkt sich den zuletzt aktiven Provider (für /status im +// Leerlauf). Geschrieben wird ausschließlich vom Queue-Worker, gelesen von +// HTTP-Handlern — daher der eigene kleine Mutex. +type providerState struct { + mu sync.Mutex + name string +} + +func (s *providerState) set(name string) { + s.mu.Lock() + s.name = name + s.mu.Unlock() +} + +func (s *providerState) get() string { + s.mu.Lock() + defer s.mu.Unlock() + return s.name +} + +// lineLogger reicht Zeilen an den Standard-Logger weiter (Zeitstempel + +// talkd-Präfix), statt sie roh zu schreiben: Provider-Kette, Queue und +// null-Sink schreiben zeilenweise, so entsteht ein einheitliches Log. +type lineLogger struct{} + +func (lineLogger) Write(p []byte) (int, error) { + for _, line := range strings.Split(strings.TrimRight(string(p), "\n"), "\n") { + if line != "" { + log.Print(line) + } + } + return len(p), nil }