// Package service orchestrates the odidere voice assistant server. // It coordinates the HTTP server, LLM clients, and handles // graceful shutdown. package service import ( "context" "embed" "encoding/json/v2" "fmt" "html/template" "io/fs" "log/slog" "net/http" "os" "os/signal" "runtime/debug" "strings" "sync" "syscall" "time" "uuid" "code.chimeric.al/chimerical/odidere/internal/config" "code.chimeric.al/chimerical/odidere/internal/conversation" "code.chimeric.al/chimerical/odidere/internal/db" "code.chimeric.al/chimerical/odidere/internal/job" "code.chimeric.al/chimerical/odidere/internal/llm" "code.chimeric.al/chimerical/odidere/internal/service/templates" "code.chimeric.al/chimerical/odidere/internal/tool" "github.com/robfig/cron/v3" ) // Context keys for request-scoped values. type key string const ( logKey key = "log" idKey key = "id" ipKey key = "ip" modelsTimeout = 10 * time.Second ) //go:embed all:static/* var static embed.FS // Service is the main application coordinator. // It owns the HTTP server and all processing clients. type Service struct { cfg *config.Config cron *cron.Cron db *db.SQLite dbr *db.Queries dbw *db.Queries jobs []*job.Job llms map[string]*llm.Client log *slog.Logger allowedModels map[string]map[string]struct{} mux *http.ServeMux server *http.Server tmpl *template.Template tools *tool.Registry ctx context.Context cancel context.CancelFunc wg sync.WaitGroup mu sync.Mutex conversations map[string]*conversation.Conversation } // New creates a Service from the provided configuration. // It initializes all clients and the HTTP server. func New( cfg *config.Config, log *slog.Logger, sdb *db.SQLite, ) (*Service, error) { if cfg == nil { return nil, fmt.Errorf("config is nil") } if err := cfg.Validate(); err != nil { return nil, fmt.Errorf("invalid config: %w", err) } if log == nil { return nil, fmt.Errorf("logger is nil") } if sdb == nil || sdb.Writer == nil || sdb.Reader == nil { return nil, fmt.Errorf("database is nil") } ctx, cancel := context.WithCancel(context.Background()) var svc = &Service{ cfg: cfg, db: sdb, dbr: db.New(sdb.Reader), dbw: db.New(sdb.Writer), log: log, mux: http.NewServeMux(), ctx: ctx, cancel: cancel, conversations: make(map[string]*conversation.Conversation), } // Setup tool registry. registry, err := tool.NewRegistry(cfg.Tools) if err != nil { return nil, fmt.Errorf("load tools: %v", err) } svc.tools = registry // Create LLM clients for each provider. svc.llms = make(map[string]*llm.Client, len(cfg.Providers)) for _, pc := range cfg.Providers { client, err := llm.NewClient( llm.Config{ Key: pc.Key, Provider: pc.Name, Timeout: pc.Timeout, URL: pc.URL, }, log, ) if err != nil { return nil, fmt.Errorf( "create LLM client for provider %q: %w", pc.Name, err, ) } svc.llms[pc.Name] = client log.Info( "LLM provider registered", slog.String("provider", pc.Name), slog.String("url", pc.URL), ) } if svc.llms[cfg.DefaultModel.Provider] == nil { return nil, fmt.Errorf("unknown default provider") } // Build model allowed list map: provider name -> set of base model IDs. svc.allowedModels = make( map[string]map[string]struct{}, len(cfg.Providers), ) for _, pc := range cfg.Providers { if len(pc.Models) == 0 { continue } allowed := make(map[string]struct{}, len(pc.Models)) for _, m := range pc.Models { base, _, _ := strings.Cut(m, ":") allowed[base] = struct{}{} } svc.allowedModels[pc.Name] = allowed } // Convert job configs to jobs. jobs := make([]*job.Job, 0, len(cfg.Jobs)) for _, jc := range cfg.Jobs { // Resolve the provider and model for this job. provider, modelID := jc.Model.Provider, jc.Model.Model if provider == "" { provider = cfg.DefaultModel.Provider } if modelID == "" { modelID = cfg.DefaultModel.Model } llmc := svc.llms[provider] if llmc == nil { return nil, fmt.Errorf( "create job %q: no LLM client for provider %q", jc.Name, provider, ) } jc.Model = config.ModelConfig{ Provider: provider, Model: modelID, } if jc.SystemMessage == "" { jc.SystemMessage = cfg.SystemMessage } j, err := job.New(jc, llmc, log) if err != nil { return nil, fmt.Errorf( "create job %q: %w", jc.Name, err, ) } jobs = append(jobs, j) } svc.jobs = jobs // Parse templates. tmpl, err := templates.Parse() if err != nil { return nil, fmt.Errorf("parse templates: %v", err) } svc.tmpl = tmpl // Setup static file server. staticFS, err := fs.Sub(static, "static") if err != nil { return nil, fmt.Errorf("setup static fs: %v", err) } // Register routes. svc.mux.HandleFunc("GET /", svc.home) svc.mux.HandleFunc("GET /status", svc.status) svc.mux.Handle( "GET /static/", http.StripPrefix( "/static/", http.FileServer(http.FS(staticFS)), ), ) svc.mux.HandleFunc("GET /models", svc.models) svc.mux.HandleFunc("GET /conversations", svc.listConversations) svc.mux.HandleFunc("POST /conversations", svc.createConversation) svc.mux.HandleFunc("GET /conversations/{id}", svc.getConversationHandler) svc.mux.HandleFunc("DELETE /conversations/{id}", svc.deleteConversation) svc.mux.HandleFunc( "POST /conversations/{id}/inference", svc.inference, ) svc.mux.HandleFunc("POST /conversations/{id}/stop", svc.stop) svc.mux.HandleFunc( "POST /conversations/{id}/subscribe", svc.subscribe, ) svc.mux.HandleFunc( "POST /conversations/{id}/messages", svc.insertMessage, ) svc.mux.HandleFunc( "PUT /conversations/{id}/messages/{mid}", svc.updateMessage, ) svc.mux.HandleFunc( "DELETE /conversations/{id}/messages/{mid}", svc.deleteMessage, ) svc.mux.HandleFunc( "POST /conversations/{id}/messages/{mid}/calls", svc.addToolCall, ) svc.mux.HandleFunc( "PUT /conversations/{id}/messages/{mid}/calls/{cid}", svc.updateToolCall, ) svc.mux.HandleFunc( "DELETE /conversations/{id}/messages/{mid}/calls/{cid}", svc.deleteToolCall, ) svc.mux.HandleFunc("GET /executions", svc.listExecutions) svc.mux.HandleFunc("PUT /executions/{id}", svc.updateExecution) svc.server = &http.Server{ Addr: cfg.Address, Handler: svc, } return svc, nil } // ServeHTTP implements http.Handler. It logs requests, assigns a UUID, // sets context values, handles panics, and delegates to the mux. func (svc *Service) ServeHTTP(w http.ResponseWriter, r *http.Request) { var ( start = time.Now() id = uuid.NewV4().String() ip = func() string { if ip := r.Header.Get("X-Forwarded-For"); ip != "" { before, _, found := strings.Cut(ip, ",") if found { return strings.TrimSpace(before) } return ip } return r.RemoteAddr }() log = svc.log.With(slog.Group( "request", slog.String("id", id), slog.String("ip", ip), slog.String("method", r.Method), slog.String("path", r.URL.Path), )) ) // Log completion time. defer func() { log.InfoContext( r.Context(), "completed", slog.Duration("duration", time.Since(start)), ) }() // Panic recovery. defer func() { if err := recover(); err != nil { log.ErrorContext( r.Context(), "panic recovered", slog.Any("error", err), slog.String("stack", string(debug.Stack())), ) http.Error( w, http.StatusText( http.StatusInternalServerError, ), http.StatusInternalServerError, ) } }() // Enrich context with request-scoped values. ctx := r.Context() ctx = context.WithValue(ctx, logKey, log) ctx = context.WithValue(ctx, idKey, id) ctx = context.WithValue(ctx, ipKey, ip) r = r.WithContext(ctx) log.InfoContext(ctx, "handling") // Always set request ID header. w.Header().Set("X-Request-ID", id) // Pass the request on to the multiplexer. svc.mux.ServeHTTP(w, r) } // Run starts the service and blocks until shutdown. // Shutdown is triggered by SIGINT or SIGTERM. // // The return value tells the caller when it is safe to close the // database: // // - nil: all HTTP handlers returned, and all inference calls and // scheduled jobs finished. The caller can close the database. // - non-nil: shutdown did not finish within the configured // ShutdownTimeout, or the HTTP server failed. Handlers, // inference calls, or jobs can still use the database. Do not // close the database. Exit the process instead. The database // runs in WAL mode, so committed data stays safe when the // process exits without a close. func (svc *Service) Run(ctx context.Context) error { svc.log.Info( "starting odidere", slog.Group( "llm", slog.String( "default_provider", svc.cfg.DefaultModel.Provider, ), slog.String( "default_model", svc.cfg.DefaultModel.Model, ), slog.Int("providers", len(svc.cfg.Providers)), ), slog.Group( "server", slog.String("address", svc.cfg.Address), ), slog.Group( "tools", slog.Int("count", len(svc.tools.List())), slog.Any( "names", strings.Join(svc.tools.List(), ","), ), ), slog.Group( "tts_url", slog.String("url", svc.cfg.TTS.URL), slog.String("default_voice", svc.cfg.TTS.Voice), ), slog.Group( "stt_url", slog.String("url", svc.cfg.STT.URL), ), slog.Group( "jobs", slog.Int("count", svc.JobCount()), ), slog.Any("shutdown_timeout", svc.cfg.ShutdownTimeout), ) // Setup signal handling for graceful shutdown. ctx, cancel := signal.NotifyContext( ctx, os.Interrupt, syscall.SIGTERM, ) defer cancel() svc.ctx, svc.cancel = context.WithCancel(ctx) defer svc.cancel() // Register jobs with cron scheduler. svc.cron = cron.New() for _, j := range svc.jobs { entry, err := svc.cron.AddFunc(j.Schedule(), func() { title := j.Name() system := j.SystemMessage() _, c, err := svc.newConversation( svc.ctx, j.UserId(), &title, system, ) if err != nil { svc.log.Error("failed to create job conversation", slog.String("name", j.Name()), slog.Any("error", err), ) return } if err := j.Run(svc.ctx, c); err != nil { svc.log.Error("job failed", slog.String("name", j.Name()), slog.Any("error", err), ) } }) if err != nil { svc.log.Error("failed to schedule job", slog.String("name", j.Name()), slog.Any("error", err), ) continue } svc.log.Info("job scheduled", slog.String("name", j.Name()), slog.String("schedule", j.Schedule()), slog.Int("entry_id", int(entry)), ) } svc.cron.Start() // Start HTTP server in background. var errs = make(chan error, 1) go func() { svc.log.Info( "HTTP server listening", slog.String("address", svc.cfg.Address), ) if err := svc.server.ListenAndServe(); err != nil && err != http.ErrServerClosed { errs <- err } close(errs) }() // Wait for shutdown signal or server error. var runErr error select { case <-ctx.Done(): svc.log.Info("shutdown signal received") case err := <-errs: if err != nil { runErr = fmt.Errorf("server error: %w", err) } } // Use one deadline for HTTP shutdown and job completion. // The shutdown context must remain active after svc.ctx is canceled. shutdownCtx, shutdownCancel := context.WithTimeout( context.Background(), svc.cfg.ShutdownTimeout, ) defer shutdownCancel() svc.cancel() svc.log.Info("stopping job scheduler") cronCtx := svc.cron.Stop() var shutdownErr error svc.log.Info("shutting down HTTP server") if err := svc.server.Shutdown(shutdownCtx); err != nil { shutdownErr = fmt.Errorf("HTTP shutdown: %w", err) } else { // Jobs and HTTP inference can save tool results after // cancellation. Wait until both return before the caller // closes the database. done := make(chan struct{}) go func() { <-cronCtx.Done() svc.wg.Wait() close(done) }() select { case <-done: case <-shutdownCtx.Done(): shutdownErr = fmt.Errorf( "wait for background work: %w", shutdownCtx.Err(), ) } } // On failure, main exits without closing the database because // handlers or jobs may still use it. if shutdownErr != nil { if runErr != nil { return fmt.Errorf("%w; %w", runErr, shutdownErr) } return shutdownErr } svc.log.Info("terminating") return runErr } // newConversation saves a conversation and its initial system message. // It registers the handle only after the transaction commits. func (svc *Service) newConversation( ctx context.Context, userId string, title *string, system string, ) (*db.Conversation, *conversation.Conversation, error) { msg := llm.Message{ Role: llm.RoleSystem, ContentParts: []llm.ContentPart{{ Type: llm.ContentTypeText, Text: system, }}, } if err := msg.Validate(); err != nil { return nil, nil, fmt.Errorf("invalid system message: %w", err) } content, err := json.Marshal(msg) if err != nil { return nil, nil, fmt.Errorf("marshal system message: %w", err) } tx, err := svc.db.Writer.BeginTx(ctx, nil) if err != nil { return nil, nil, fmt.Errorf("begin tx: %w", err) } defer tx.Rollback() qtx := svc.dbw.WithTx(tx) saved, err := qtx.InsertConversation( ctx, db.InsertConversationParams{ Title: title, UserId: userId, }, ) if err != nil { return nil, nil, fmt.Errorf("insert conversation: %w", err) } if _, err := qtx.InsertMessage(ctx, db.InsertMessageParams{ Content: content, ConversationId: saved.Id, Role: db.MessageRoleSystem, Sequence: 0, }); err != nil { return nil, nil, fmt.Errorf("insert system message: %w", err) } c, err := conversation.New( conversation.Config{Id: saved.Id}, svc.db, svc.log, svc.tools, ) if err != nil { return nil, nil, fmt.Errorf("create conversation handle: %w", err) } if err := tx.Commit(); err != nil { return nil, nil, fmt.Errorf("commit conversation tx: %w", err) } svc.mu.Lock() // A retrieval can register a handle after the commit. if existing := svc.conversations[saved.Id]; existing != nil { c = existing } else { svc.conversations[saved.Id] = c } svc.mu.Unlock() return saved, c, nil } // getConversation checks saved state before it returns a cached handle. func (svc *Service) getConversation( ctx context.Context, id string, ) (*conversation.Conversation, error) { if _, err := svc.dbr.SelectConversationId(ctx, id); err != nil { return nil, fmt.Errorf("select conversation: %w", err) } svc.mu.Lock() defer svc.mu.Unlock() if c := svc.conversations[id]; c != nil { return c, nil } c, err := conversation.New( conversation.Config{Id: id}, svc.db, svc.log, svc.tools, ) if err != nil { return nil, fmt.Errorf("create conversation handle: %w", err) } svc.conversations[id] = c return c, nil } // ListJobs returns all registered jobs. func (svc *Service) ListJobs() []*job.Job { return svc.jobs } // JobCount returns the number of registered jobs. func (svc *Service) JobCount() int { return len(svc.jobs) } func (svc *Service) home(w http.ResponseWriter, r *http.Request) { var ( ctx = r.Context() log = ctx.Value(logKey).(*slog.Logger) ) if r.URL.Path != "/" { http.NotFound(w, r) return } w.Header().Set("Content-Type", "text/html; charset=utf-8") if err := svc.tmpl.ExecuteTemplate( w, "index.gohtml", struct { TTSURL string TTSDefaultVoice string STTURL string }{ TTSURL: svc.cfg.TTS.URL, TTSDefaultVoice: svc.cfg.TTS.Voice, STTURL: svc.cfg.STT.URL, }, ); err != nil { log.ErrorContext( ctx, "template error", slog.Any("error", err), ) http.Error( w, http.StatusText(http.StatusInternalServerError), http.StatusInternalServerError, ) } } // status returns server status. func (svc *Service) status(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) } // ModelStatus represents the availability status of a model. type ModelStatus string const ( // ModelStatusAvailable indicates the model is available for use. ModelStatusAvailable ModelStatus = "available" // ModelStatusNotFound indicates the model is configured but not found // in the provider's model list. ModelStatusNotFound ModelStatus = "not_found" // ModelStatusProviderError indicates the provider could not be reached // to verify the model. ModelStatusProviderError ModelStatus = "provider_error" ) // Model represents a model in the /models response. type Model struct { Model string `json:"model"` Status ModelStatus `json:"status"` ContextSize *int `json:"context_size,omitempty"` } // Models is the response format for the /models endpoint. type Models struct { Providers map[string][]Model `json:"providers"` DefaultProvider string `json:"default_provider"` DefaultModel string `json:"default_model"` } // models returns available LLM models from all configured providers. func (svc *Service) models(w http.ResponseWriter, r *http.Request) { var ( ctx = r.Context() log = ctx.Value(logKey).(*slog.Logger) mu sync.Mutex wg sync.WaitGroup providers = make(map[string][]Model, len(svc.llms)) ) cfg := make(map[string]config.ProviderConfig, len(svc.cfg.Providers)) for _, pc := range svc.cfg.Providers { cfg[pc.Name] = pc } for name, client := range svc.llms { wg.Add(1) go func(name string, c *llm.Client, pc config.ProviderConfig) { defer wg.Done() var result = []Model{} fetchCtx, cancel := context.WithTimeout( ctx, modelsTimeout, ) defer cancel() models, err := c.ListModels(fetchCtx) if err != nil { log.WarnContext( ctx, "failed to list models for provider", slog.String("provider", name), slog.Any("error", err), ) for _, m := range pc.Models { result = append(result, Model{ Model: m, Status: ModelStatusProviderError, }) } mu.Lock() providers[name] = result mu.Unlock() return } available := make(map[string]struct{}, len(models)) contextSizes := make(map[string]int, len(models)) for _, m := range models { baseID, _, _ := strings.Cut(m.ID, ":") available[baseID] = struct{}{} cs := m.ContextLength if cs == 0 && m.Meta != nil { cs = m.Meta.NCtx } if cs > 0 { contextSizes[baseID] = cs } } var toCheck []string if len(pc.Models) > 0 { toCheck = pc.Models } else { toCheck = make([]string, 0, len(models)) for _, m := range models { toCheck = append(toCheck, m.ID) } } for _, m := range toCheck { base, _, _ := strings.Cut(m, ":") mi := Model{Model: m} if _, ok := available[base]; ok { mi.Status = ModelStatusAvailable if cs, ok := contextSizes[base]; ok { mi.ContextSize = &cs } } else { mi.Status = ModelStatusNotFound } result = append(result, mi) } mu.Lock() providers[name] = result mu.Unlock() }(name, client, cfg[name]) } wg.Wait() w.Header().Set("Content-Type", "application/json") if err := json.MarshalWrite(w, Models{ Providers: providers, DefaultProvider: svc.cfg.DefaultModel.Provider, DefaultModel: svc.cfg.DefaultModel.Model, }); err != nil { log.ErrorContext(ctx, "failed to encode models response", slog.Any("error", err)) } }