diff --git a/internal/service/service.go b/internal/service/service.go index 9c2a735..57e0587 100644 --- a/internal/service/service.go +++ b/internal/service/service.go @@ -6,7 +6,7 @@ package service import ( "context" "embed" - "encoding/json" + "encoding/json/v2" "fmt" "html/template" "io/fs" @@ -19,13 +19,16 @@ import ( "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/google/uuid" "github.com/robfig/cron/v3" ) @@ -48,6 +51,9 @@ var static embed.FS 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 @@ -55,20 +61,49 @@ type Service struct { mux *http.ServeMux server *http.Server tmpl *template.Template - tools *llm.Registry + tools *tool.Registry + + ctx context.Context + cancel context.CancelFunc + 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) (*Service, error) { +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, - log: log, - mux: http.NewServeMux(), + 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 := llm.NewRegistry(cfg.Tools) + registry, err := tool.NewRegistry(cfg.Tools) if err != nil { return nil, fmt.Errorf("load tools: %v", err) } @@ -79,12 +114,11 @@ func New(cfg *config.Config, log *slog.Logger) (*Service, error) { for _, pc := range cfg.Providers { client, err := llm.NewClient( llm.Config{ - Key: pc.Key, - SystemMessage: cfg.SystemMessage, - Timeout: pc.Timeout, - URL: pc.URL, + Key: pc.Key, + Provider: pc.Name, + Timeout: pc.Timeout, + URL: pc.URL, }, - registry, log, ) if err != nil { @@ -100,6 +134,9 @@ func New(cfg *config.Config, log *slog.Logger) (*Service, error) { 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( @@ -121,19 +158,12 @@ func New(cfg *config.Config, log *slog.Logger) (*Service, error) { jobs := make([]*job.Job, 0, len(cfg.Jobs)) for _, jc := range cfg.Jobs { // Resolve the provider and model for this job. - var provider, modelID string - if jc.Model.Provider != "" { - if jc.Model.Model == "" { - return nil, fmt.Errorf( - "job %q: model.provider set but model.model is empty", - jc.Name, - ) - } - provider = jc.Model.Provider - modelID = jc.Model.Model - } else { - provider = svc.cfg.DefaultModel.Provider - modelID = svc.cfg.DefaultModel.Model + 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 { @@ -143,7 +173,14 @@ func New(cfg *config.Config, log *slog.Logger) (*Service, error) { provider, ) } - j, err := job.New(jc, log, llmc, modelID) + 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, @@ -175,9 +212,46 @@ func New(cfg *config.Config, log *slog.Logger) (*Service, error) { "/static/", http.FileServer(http.FS(staticFS)), ), ) - svc.mux.HandleFunc("POST /v1/chat/voice", svc.chat) - svc.mux.HandleFunc("POST /v1/chat/voice/stream", svc.chatStream) - svc.mux.HandleFunc("GET /v1/models", svc.models) + 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, @@ -192,7 +266,7 @@ func New(cfg *config.Config, log *slog.Logger) (*Service, error) { func (svc *Service) ServeHTTP(w http.ResponseWriter, r *http.Request) { var ( start = time.Now() - id = uuid.NewString() + id = uuid.NewV4().String() ip = func() string { if ip := r.Header.Get("X-Forwarded-For"); ip != "" { before, _, found := strings.Cut(ip, ",") @@ -296,7 +370,7 @@ func (svc *Service) Run(ctx context.Context) error { "jobs", slog.Int("count", svc.JobCount()), ), - slog.Any("shutdown_timeout", svc.cfg.GetShutdownTimeout()), + slog.Any("shutdown_timeout", svc.cfg.ShutdownTimeout), ) // Setup signal handling for graceful shutdown. @@ -305,26 +379,29 @@ func (svc *Service) Run(ctx context.Context) error { 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() { - result := j.Run(ctx) - if result.Error != nil { + 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", result.Error), - slog.Duration( - "duration", result.Duration, - ), - ) - } else { - svc.log.Info("job completed", - slog.String("name", j.Name()), - slog.Duration( - "duration", result.Duration, - ), + slog.Any("error", err), ) } }) @@ -358,36 +435,56 @@ func (svc *Service) Run(ctx context.Context) error { }() // 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 { - return fmt.Errorf("server error: %w", err) + runErr = fmt.Errorf("server error: %w", err) } } - // Graceful shutdown with timeout. + // 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.GetShutdownTimeout(), + svc.cfg.ShutdownTimeout, ) defer shutdownCancel() - // Stop cron scheduler — in-flight jobs continue to completion. - svc.log.Info("stopping job scheduler") - svc.cron.Stop() + 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 { - svc.log.Warn( - "shutdown timeout reached", - slog.Any("error", err), - ) + shutdownErr = fmt.Errorf("HTTP shutdown: %w", err) + } else { + // Jobs can save tool results after cancellation. Wait until + // their callbacks return before the caller closes the database. + select { + case <-cronCtx.Done(): + case <-shutdownCtx.Done(): + shutdownErr = fmt.Errorf( + "wait for jobs: %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 nil + return runErr } // ListJobs returns all registered jobs. @@ -438,338 +535,6 @@ func (svc *Service) status(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) } -// Request is the incoming request format for the chat endpoints. -type Request struct { - // Messages is the conversation history. - Messages []llm.Message `json:"messages"` - // Provider is the LLM provider name. If empty, the default provider is used. - Provider string `json:"provider,omitempty"` - // Model is the LLM model ID. If empty, the default model is used. - Model string `json:"model,omitempty"` - // SystemMessage overrides the configured system message for this request. - SystemMessage string `json:"system_message,omitempty"` - // MaxIterations caps the number of agent loop turns (0 = unlimited). - MaxIterations int `json:"max_iterations,omitempty"` - // ThinkingBudgetTokens is the llama.cpp reasoning budget (-1 = unlimited, 0 = off). - ThinkingBudgetTokens *int `json:"thinking_budget_tokens,omitempty"` -} - -// Response is the response format for chat and voice endpoints. -type Response struct { - // Messages is the full list of messages generated during the completions request, - // including tool calls and tool results. - Messages []llm.Message `json:"messages,omitempty"` - // Provider is the LLM provider used for the response. - Provider string `json:"used_provider,omitempty"` - // Model is the LLM model used for the response. - Model string `json:"used_model,omitempty"` - // Usage is token usage from the API response. - Usage *llm.Usage `json:"usage,omitempty"` - // Timings is llama.cpp timing info from the API response. - Timings *llm.Timings `json:"timings,omitempty"` -} - -// chat processes text chat requests. -func (svc *Service) chat(w http.ResponseWriter, r *http.Request) { - var ( - ctx = r.Context() - log = ctx.Value(logKey).(*slog.Logger) - ) - - // Parse request. - r.Body = http.MaxBytesReader(w, r.Body, 32<<20) - var req Request - if err := json.NewDecoder(r.Body).Decode(&req); err != nil { - log.ErrorContext( - ctx, - "failed to decode request", - slog.Any("error", err), - ) - http.Error(w, "invalid request", http.StatusBadRequest) - return - } - - // Validate messages. - if len(req.Messages) == 0 { - http.Error(w, "messages required", http.StatusBadRequest) - return - } - log.DebugContext( - ctx, - "messages", - slog.Any("data", req.Messages), - ) - - // Resolve provider and model. - provider := req.Provider - if provider == "" { - provider = svc.cfg.DefaultModel.Provider - } - llmc := svc.llms[provider] - if llmc == nil { - log.ErrorContext( - ctx, - "unknown provider", - slog.String("provider", provider), - ) - http.Error( - w, - fmt.Sprintf("unknown provider: %q", provider), - http.StatusBadRequest, - ) - return - } - - model := req.Model - if model == "" { - model = svc.cfg.DefaultModel.Model - } - // Validate model against provider allowed list, if configured. - if allowed, ok := svc.allowedModels[provider]; ok { - m, _, _ := strings.Cut(model, ":") - if _, ok := allowed[m]; !ok { - log.ErrorContext( - ctx, - "model not in allowed list", - slog.String("provider", provider), - slog.String("model", model), - ) - http.Error( - w, - fmt.Sprintf( - "model %q not allowed for provider %q", - model, provider, - ), - http.StatusBadRequest, - ) - return - } - } - - res, err := llmc.Completions( - ctx, llm.CompletionsParams{ - Model: model, - SystemMessage: req.SystemMessage, - MaxIterations: req.MaxIterations, - Messages: req.Messages, - ThinkingBudgetTokens: req.ThinkingBudgetTokens, - }, - ) - if err != nil { - log.ErrorContext( - ctx, - "LLM request failed", - slog.Any("error", err), - ) - http.Error(w, "LLM error", http.StatusInternalServerError) - return - } - if len(res.Messages) == 0 { - http.Error( - w, - "no response from LLM", - http.StatusInternalServerError, - ) - return - } - final := res.Messages[len(res.Messages)-1] - log.DebugContext( - ctx, - "LLM response", - slog.String("text", final.Text()), - slog.String("provider", provider), - slog.String("model", model), - ) - - w.Header().Set("Content-Type", "application/json") - if err := json.NewEncoder(w).Encode(Response{ - Messages: res.Messages, - Provider: provider, - Model: model, - Usage: res.Usage, - Timings: res.Timings, - }); err != nil { - log.ErrorContext( - ctx, - "failed to json encode response", - slog.Any("error", err), - ) - } -} - -// StreamMessage is the SSE event payload for the streaming chat endpoint. -type StreamMessage struct { - // Error is an error message, if any. - Error string `json:"error,omitempty"` - // Delta is a text fragment from the assistant response. - Delta string `json:"delta,omitempty"` - // ReasoningDelta is a reasoning text fragment from the assistant response. - ReasoningDelta string `json:"reasoning_delta,omitempty"` - // Message is the chat completion message. - Message *llm.Message `json:"message,omitempty"` - // Provider is the LLM provider used for the response. - Provider string `json:"provider,omitempty"` - // Model is the LLM model used for the response. - Model string `json:"model,omitempty"` - // Usage is token usage from the API response. - Usage *llm.Usage `json:"usage,omitempty"` - // Timings is llama.cpp timing info from the API response. - Timings *llm.Timings `json:"timings,omitempty"` -} - -// chatStream processes chat requests with streaming SSE output. -func (svc *Service) chatStream(w http.ResponseWriter, r *http.Request) { - var ( - ctx = r.Context() - log = ctx.Value(logKey).(*slog.Logger) - ) - - // Check that the response writer supports flushing. - flusher, ok := w.(http.Flusher) - if !ok { - http.Error( - w, - "streaming not supported", - http.StatusInternalServerError, - ) - return - } - - // Parse request. - r.Body = http.MaxBytesReader(w, r.Body, 32<<20) - var req Request - if err := json.NewDecoder(r.Body).Decode(&req); err != nil { - log.ErrorContext( - ctx, - "failed to decode request", - slog.Any("error", err), - ) - http.Error(w, "invalid request", http.StatusBadRequest) - return - } - - // Validate messages. - if len(req.Messages) == 0 { - http.Error(w, "messages required", http.StatusBadRequest) - return - } - - // Resolve provider and model. - provider := req.Provider - if provider == "" { - provider = svc.cfg.DefaultModel.Provider - } - model := req.Model - if model == "" { - model = svc.cfg.DefaultModel.Model - } - llmc := svc.llms[provider] - if llmc == nil { - log.ErrorContext( - ctx, - "unknown provider", - slog.String("provider", provider), - ) - http.Error( - w, - fmt.Sprintf("unknown provider: %q", provider), - http.StatusBadRequest, - ) - return - } - - // Validate model against provider allowed list, if configured. - if allowed, ok := svc.allowedModels[provider]; ok { - m, _, _ := strings.Cut(model, ":") - if _, ok := allowed[m]; !ok { - log.ErrorContext( - ctx, - "model not in allowed list", - slog.String("provider", provider), - slog.String("model", model), - ) - http.Error( - w, - fmt.Sprintf( - "model %q not allowed for provider %q", - model, provider, - ), - http.StatusBadRequest, - ) - return - } - } - - // Set SSE headers. - w.Header().Set("Cache-Control", "no-cache") - w.Header().Set("Connection", "keep-alive") - w.Header().Set("Content-Type", "text/event-stream") - - // Helper to send an SSE event. - send := func(event string, msg StreamMessage) { - data, err := json.Marshal(msg) - if err != nil { - log.ErrorContext(ctx, "failed to marshal SSE event", - slog.Any("error", err), - ) - return - } - fmt.Fprintf(w, "event: %s\ndata: %s\n\n", event, data) - flusher.Flush() - } - - // Start streaming LLM completions request. - var ( - events = make(chan llm.StreamEvent) - errs = make(chan error, 1) - ) - go func() { - errs <- llmc.CompletionsStream( - ctx, llm.CompletionsParams{ - Model: model, - SystemMessage: req.SystemMessage, - MaxIterations: req.MaxIterations, - Messages: req.Messages, - ThinkingBudgetTokens: req.ThinkingBudgetTokens, - }, events, - ) - close(errs) - }() - - // Consume events and send as SSE. - for evt := range events { - msg := StreamMessage{ - Delta: evt.Delta, - ReasoningDelta: evt.ReasoningDelta, - Provider: provider, - Model: model, - } - if evt.Message.Role != "" { - msg.Message = &evt.Message - } - if evt.Usage != nil { - msg.Usage = evt.Usage - } - if evt.Timings != nil { - msg.Timings = evt.Timings - } - send(evt.Type, msg) - } - if err := <-errs; err != nil { - log.ErrorContext( - ctx, - "LLM stream failed", - slog.Any("error", err), - ) - msg := llm.Message{Role: llm.RoleAssistant} - send("message", StreamMessage{ - Message: &msg, - Error: fmt.Sprintf("LLM error: %v", err), - }) - return - } -} - // ModelStatus represents the availability status of a model. type ModelStatus string @@ -784,14 +549,14 @@ const ( ModelStatusProviderError ModelStatus = "provider_error" ) -// Model represents a model in the /v1/models response. +// 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 /v1/models endpoint. +// Models is the response format for the /models endpoint. type Models struct { Providers map[string][]Model `json:"providers"` DefaultProvider string `json:"default_provider"` @@ -891,7 +656,7 @@ func (svc *Service) models(w http.ResponseWriter, r *http.Request) { wg.Wait() w.Header().Set("Content-Type", "application/json") - if err := json.NewEncoder(w).Encode(Models{ + if err := json.MarshalWrite(w, Models{ Providers: providers, DefaultProvider: svc.cfg.DefaultModel.Provider, DefaultModel: svc.cfg.DefaultModel.Model,