From 652d41ca8f7ec6e560707836aaa745e6ae6b8a8f Mon Sep 17 00:00:00 2001 From: dwrz Date: Fri, 30 Oct 2026 17:21:58 +0000 Subject: [PATCH] Refactor service --- internal/service/conversations.go | 187 ++++++++ internal/service/conversations_test.go | 200 +++++++++ internal/service/executions.go | 132 ++++++ internal/service/inference.go | 340 ++++++++++++++ internal/service/messages.go | 595 +++++++++++++++++++++++++ internal/service/service.go | 130 +++++- internal/service/service_test.go | 459 ++++++++++++++++--- 7 files changed, 1975 insertions(+), 68 deletions(-) create mode 100644 internal/service/conversations.go create mode 100644 internal/service/conversations_test.go create mode 100644 internal/service/executions.go create mode 100644 internal/service/inference.go create mode 100644 internal/service/messages.go diff --git a/internal/service/conversations.go b/internal/service/conversations.go new file mode 100644 index 0000000..7eb67f4 --- /dev/null +++ b/internal/service/conversations.go @@ -0,0 +1,187 @@ +package service + +import ( + "database/sql" + "encoding/json/v2" + "errors" + "log/slog" + "net/http" +) + +func (svc *Service) listConversations(w http.ResponseWriter, r *http.Request) { + var ( + ctx = r.Context() + log = ctx.Value(logKey).(*slog.Logger) + ) + + rows, err := svc.dbr.SelectConversationsUserId(ctx, "default") + if err != nil { + log.ErrorContext( + ctx, "failed to list conversations", + slog.Any("error", err), + ) + http.Error( + w, + http.StatusText(http.StatusInternalServerError), + http.StatusInternalServerError, + ) + return + } + + w.Header().Set("Content-Type", "application/json") + if err := json.MarshalWrite(w, rows); err != nil { + log.ErrorContext( + ctx, "failed to encode conversations response", + slog.Any("error", err), + ) + } +} + +func (svc *Service) createConversation(w http.ResponseWriter, r *http.Request) { + var ( + ctx = r.Context() + log = ctx.Value(logKey).(*slog.Logger) + req struct { + Title *string `json:"title"` + SystemMessage *string `json:"system_message"` + } + ) + + r.Body = http.MaxBytesReader(w, r.Body, 32<<20) + if err := json.UnmarshalRead(r.Body, &req); err != nil { + log.ErrorContext( + ctx, "failed to decode request", + slog.Any("error", err), + ) + http.Error(w, "invalid request", http.StatusBadRequest) + return + } + + system := svc.cfg.SystemMessage + if req.SystemMessage != nil { + system = *req.SystemMessage + } + saved, _, err := svc.newConversation( + ctx, "default", req.Title, system, + ) + if err != nil { + log.ErrorContext( + ctx, "failed to create conversation", + slog.Any("error", err), + ) + http.Error( + w, + http.StatusText(http.StatusInternalServerError), + http.StatusInternalServerError, + ) + return + } + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusCreated) + if err := json.MarshalWrite(w, saved); err != nil { + log.ErrorContext( + ctx, "failed to encode conversation response", + slog.Any("error", err), + ) + } +} + +func (svc *Service) getConversationHandler( + w http.ResponseWriter, + r *http.Request, +) { + var ( + ctx = r.Context() + log = ctx.Value(logKey).(*slog.Logger) + ) + + saved, err := svc.dbr.SelectConversationIdUserId( + ctx, r.PathValue("id"), "default", + ) + if errors.Is(err, sql.ErrNoRows) { + http.NotFound(w, r) + return + } + if err != nil { + log.ErrorContext( + ctx, "failed to load conversation", + slog.Any("error", err), + ) + http.Error( + w, + http.StatusText(http.StatusInternalServerError), + http.StatusInternalServerError, + ) + return + } + + w.Header().Set("Content-Type", "application/json") + if err := json.MarshalWrite(w, saved); err != nil { + log.ErrorContext( + ctx, "failed to encode conversation response", + slog.Any("error", err), + ) + } +} + +func (svc *Service) deleteConversation(w http.ResponseWriter, r *http.Request) { + var ( + ctx = r.Context() + log = ctx.Value(logKey).(*slog.Logger) + id = r.PathValue("id") + ) + + if _, err := svc.dbr.SelectConversationIdUserId( + ctx, id, "default", + ); err != nil { + if errors.Is(err, sql.ErrNoRows) { + http.NotFound(w, r) + return + } + log.ErrorContext( + ctx, "failed to load conversation", + slog.Any("error", err), + ) + http.Error( + w, + http.StatusText(http.StatusInternalServerError), + http.StatusInternalServerError, + ) + return + } + c, err := svc.getConversation(ctx, id) + if errors.Is(err, sql.ErrNoRows) { + http.NotFound(w, r) + return + } + if err != nil { + log.ErrorContext( + ctx, "failed to get conversation handle", + slog.Any("error", err), + ) + http.Error( + w, + http.StatusText(http.StatusInternalServerError), + http.StatusInternalServerError, + ) + return + } + if err := c.Delete(ctx); err != nil { + log.ErrorContext( + ctx, "failed to delete conversation", + slog.Any("error", err), + ) + http.Error( + w, + http.StatusText(http.StatusInternalServerError), + http.StatusInternalServerError, + ) + return + } + + svc.mu.Lock() + delete(svc.conversations, id) + svc.mu.Unlock() + w.WriteHeader(http.StatusNoContent) +} diff --git a/internal/service/conversations_test.go b/internal/service/conversations_test.go new file mode 100644 index 0000000..2d361a9 --- /dev/null +++ b/internal/service/conversations_test.go @@ -0,0 +1,200 @@ +package service + +import ( + "database/sql" + "encoding/json/v2" + "errors" + "log/slog" + "path/filepath" + "sync" + "testing" + + "code.chimeric.al/chimerical/odidere/internal/conversation" + "code.chimeric.al/chimerical/odidere/internal/db" + "code.chimeric.al/chimerical/odidere/internal/llm" + "code.chimeric.al/chimerical/odidere/internal/tool" +) + +// newTestConversationService returns a service with a fresh database. +func newTestConversationService(t *testing.T) *Service { + t.Helper() + var ( + ctx = t.Context() + log = slog.New(slog.DiscardHandler) + ) + database, err := db.Connect(ctx, db.Config{ + Path: filepath.Join(t.TempDir(), "conversations.db"), + }) + if err != nil { + t.Fatalf("Connect: %v", err) + } + t.Cleanup(func() { database.Close() }) + if err := database.ApplySchema(ctx, log); err != nil { + t.Fatalf("ApplySchema: %v", err) + } + tools, err := tool.NewRegistry(nil) + if err != nil { + t.Fatalf("NewRegistry: %v", err) + } + return &Service{ + db: database, + dbr: db.New(database.Reader), + dbw: db.New(database.Writer), + log: log, + tools: tools, + conversations: make(map[string]*conversation.Conversation), + } +} + +func TestNewConversation(t *testing.T) { + title := "Test conversation" + tests := []struct { + name string + title *string + system string + }{ + {name: "with system", title: &title, system: "Use short answers."}, + {name: "empty system"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + svc := newTestConversationService(t) + ctx := t.Context() + saved, c, err := svc.newConversation( + ctx, "default", tt.title, tt.system, + ) + if err != nil { + t.Fatalf("newConversation: %v", err) + } + if saved == nil || c == nil { + t.Fatal("newConversation returned a nil row or handle") + } + stored, err := svc.dbr.SelectConversationId(ctx, saved.Id) + if err != nil { + t.Fatalf("SelectConversationId: %v", err) + } + if stored.UserId != "default" { + t.Fatalf("UserId = %q, want default", stored.UserId) + } + if tt.title == nil { + if stored.Title != nil { + t.Fatalf("Title = %v, want nil", stored.Title) + } + } else if stored.Title == nil || *stored.Title != *tt.title { + t.Fatalf("Title = %v, want %q", stored.Title, *tt.title) + } + rows, err := svc.dbr.SelectMessagesConversationId(ctx, saved.Id) + if err != nil { + t.Fatalf("SelectMessagesConversationId: %v", err) + } + if len(rows) != 1 { + t.Fatalf("got %d messages, want 1", len(rows)) + } + if rows[0].Role != db.MessageRoleSystem || rows[0].Sequence != 0 { + t.Fatalf("unexpected system row: %+v", rows[0]) + } + var msg llm.Message + if err := json.Unmarshal(rows[0].Content, &msg); err != nil { + t.Fatalf("Unmarshal: %v", err) + } + if err := msg.Validate(); err != nil { + t.Fatalf("Validate: %v", err) + } + if msg.Role != llm.RoleSystem || msg.Text() != tt.system { + t.Fatalf("unexpected system message: %+v", msg) + } + if svc.conversations[saved.Id] != c { + t.Fatal("created handle was not registered") + } + got, err := svc.getConversation(ctx, saved.Id) + if err != nil { + t.Fatalf("getConversation: %v", err) + } + if got != c { + t.Fatal("getConversation returned a different handle") + } + }) + } +} + +func TestGetConversation(t *testing.T) { + svc := newTestConversationService(t) + ctx := t.Context() + saved, err := svc.dbw.InsertConversation( + ctx, db.InsertConversationParams{UserId: "default"}, + ) + if err != nil { + t.Fatalf("InsertConversation: %v", err) + } + + // Concurrent retrievals must share one handle for the saved row. + var ( + wg sync.WaitGroup + handles [8]*conversation.Conversation + errs [8]error + ) + for i := range handles { + wg.Go(func() { + handles[i], errs[i] = svc.getConversation(ctx, saved.Id) + }) + } + wg.Wait() + for i, c := range handles { + if errs[i] != nil { + t.Fatalf("getConversation: %v", errs[i]) + } + if c == nil || c != handles[0] || c != svc.conversations[saved.Id] { + t.Fatal("getConversation did not reuse the registered handle") + } + } + + // A cached handle must not hide deletion of its saved row. + if err := svc.dbw.DeleteConversationId(ctx, saved.Id); err != nil { + t.Fatalf("DeleteConversationId: %v", err) + } + for _, id := range []string{saved.Id, "missing"} { + c, err := svc.getConversation(ctx, id) + if !errors.Is(err, sql.ErrNoRows) || c != nil { + t.Fatalf("getConversation(%q) = %v, %v; want nil, sql.ErrNoRows", + id, c, err) + } + } + if _, ok := svc.conversations["missing"]; ok { + t.Fatal("getConversation registered a missing conversation") + } +} + +func TestNewConversationRollback(t *testing.T) { + for _, name := range []string{"missing owner", "missing logger", "missing tools"} { + t.Run(name, func(t *testing.T) { + svc := newTestConversationService(t) + ctx := t.Context() + userId := "default" + switch name { + case "missing owner": + userId = "missing" + case "missing logger": + svc.log = nil + case "missing tools": + svc.tools = nil + } + saved, c, err := svc.newConversation(ctx, userId, nil, "System") + if err == nil || saved != nil || c != nil { + t.Fatalf("newConversation = %v, %v, %v; want nil, nil, error", + saved, c, err) + } + var conversations, messages int + if err := svc.db.Reader.QueryRowContext(ctx, ` +SELECT + (SELECT count(*) FROM conversations), + (SELECT count(*) FROM messages) +`).Scan(&conversations, &messages); err != nil { + t.Fatalf("count rows: %v", err) + } + if conversations != 0 || messages != 0 || len(svc.conversations) != 0 { + t.Fatalf("got %d conversations, %d messages, %d handles; want none", + conversations, messages, len(svc.conversations)) + } + }) + } +} diff --git a/internal/service/executions.go b/internal/service/executions.go new file mode 100644 index 0000000..4c60269 --- /dev/null +++ b/internal/service/executions.go @@ -0,0 +1,132 @@ +package service + +import ( + "database/sql" + "encoding/json/v2" + "errors" + "log/slog" + "net/http" + + "code.chimeric.al/chimerical/odidere/internal/db" +) + +func (svc *Service) listExecutions(w http.ResponseWriter, r *http.Request) { + var ( + ctx = r.Context() + log = ctx.Value(logKey).(*slog.Logger) + ) + + rows, err := svc.dbr.SelectPendingExecutions(ctx) + if err != nil { + log.ErrorContext( + ctx, "failed to list pending executions", + slog.Any("error", err), + ) + http.Error( + w, + http.StatusText(http.StatusInternalServerError), + http.StatusInternalServerError, + ) + return + } + + w.Header().Set("Content-Type", "application/json") + if err := json.MarshalWrite(w, rows); err != nil { + log.ErrorContext( + ctx, "failed to encode executions response", + slog.Any("error", err), + ) + } +} + +func (svc *Service) updateExecution(w http.ResponseWriter, r *http.Request) { + var ( + ctx = r.Context() + log = ctx.Value(logKey).(*slog.Logger) + req struct { + Status db.ExecutionStatus `json:"status"` + } + ) + + r.Body = http.MaxBytesReader(w, r.Body, 32<<20) + if err := json.UnmarshalRead(r.Body, &req); err != nil { + log.ErrorContext( + ctx, "failed to decode request", + slog.Any("error", err), + ) + http.Error(w, "invalid request", http.StatusBadRequest) + return + } + if req.Status != db.ExecutionStatusApproved && + req.Status != db.ExecutionStatusDenied { + http.Error(w, "invalid status", http.StatusBadRequest) + return + } + + tx, err := svc.db.Writer.BeginTx(ctx, nil) + if err != nil { + log.ErrorContext( + ctx, "failed to begin execution transaction", + slog.Any("error", err), + ) + http.Error( + w, + http.StatusText(http.StatusInternalServerError), + http.StatusInternalServerError, + ) + return + } + defer tx.Rollback() + + qtx := svc.dbw.WithTx(tx) + execution, err := qtx.SelectExecutionId(ctx, r.PathValue("id")) + if errors.Is(err, sql.ErrNoRows) { + http.NotFound(w, r) + return + } + if err != nil { + log.ErrorContext( + ctx, "failed to load execution", + slog.Any("error", err), + ) + http.Error( + w, + http.StatusText(http.StatusInternalServerError), + http.StatusInternalServerError, + ) + return + } + if execution.Status != db.ExecutionStatusPending { + http.Error(w, "execution is not pending", http.StatusConflict) + return + } + if err := qtx.UpdateExecutionStatus(ctx, db.UpdateExecutionStatusParams{ + Id: execution.Id, + Status: req.Status, + }); err != nil { + log.ErrorContext( + ctx, "failed to update execution status", + slog.Any("error", err), + ) + http.Error( + w, + http.StatusText(http.StatusInternalServerError), + http.StatusInternalServerError, + ) + return + } + if err := tx.Commit(); err != nil { + log.ErrorContext( + ctx, "failed to commit execution transaction", + slog.Any("error", err), + ) + http.Error( + w, + http.StatusText(http.StatusInternalServerError), + http.StatusInternalServerError, + ) + return + } + + w.WriteHeader(http.StatusNoContent) +} diff --git a/internal/service/inference.go b/internal/service/inference.go new file mode 100644 index 0000000..3af8aa9 --- /dev/null +++ b/internal/service/inference.go @@ -0,0 +1,340 @@ +package service + +import ( + "database/sql" + "encoding/json/v2" + "errors" + "fmt" + "log/slog" + "net/http" + "strings" + "time" + + "code.chimeric.al/chimerical/odidere/internal/conversation" + "code.chimeric.al/chimerical/odidere/internal/llm" +) + +func (svc *Service) inference(w http.ResponseWriter, r *http.Request) { + var ( + ctx = r.Context() + log = ctx.Value(logKey).(*slog.Logger) + id = r.PathValue("id") + req struct { + Message *llm.Message `json:"message"` + Provider string `json:"provider"` + Model string `json:"model"` + MaxIterations int `json:"max_iterations"` + ThinkingBudgetTokens *int `json:"thinking_budget_tokens"` + Timeout time.Duration `json:"timeout"` + Stream *bool `json:"stream"` + } + ) + + flusher, ok := w.(http.Flusher) + if !ok { + http.Error( + w, + "streaming not supported", + http.StatusInternalServerError, + ) + return + } + + r.Body = http.MaxBytesReader(w, r.Body, 32<<20) + if err := json.UnmarshalRead(r.Body, &req); err != nil { + log.ErrorContext( + ctx, "failed to decode request", + slog.Any("error", err), + ) + http.Error(w, "invalid request", http.StatusBadRequest) + return + } + + provider := req.Provider + if provider == "" { + provider = svc.cfg.DefaultModel.Provider + } + client := svc.llms[provider] + if client == 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 + } + if allowed, ok := svc.allowedModels[provider]; ok { + base, _, _ := strings.Cut(model, ":") + if _, ok := allowed[base]; !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 + } + } + + if _, err := svc.dbr.SelectConversationIdUserId( + ctx, id, "default", + ); err != nil { + if errors.Is(err, sql.ErrNoRows) { + http.NotFound(w, r) + return + } + log.ErrorContext( + ctx, "failed to load conversation", + slog.Any("error", err), + ) + http.Error( + w, + http.StatusText(http.StatusInternalServerError), + http.StatusInternalServerError, + ) + return + } + c, err := svc.getConversation(ctx, id) + if errors.Is(err, sql.ErrNoRows) { + http.NotFound(w, r) + return + } + if err != nil { + log.ErrorContext( + ctx, "failed to get conversation handle", + slog.Any("error", err), + ) + http.Error( + w, + http.StatusText(http.StatusInternalServerError), + http.StatusInternalServerError, + ) + return + } + + stream := true + if req.Stream != nil { + stream = *req.Stream + } + inf, err := c.Reserve(svc.ctx, client, conversation.InferenceRequest{ + Message: req.Message, + Model: model, + MaxIterations: req.MaxIterations, + ThinkingBudgetTokens: req.ThinkingBudgetTokens, + Timeout: req.Timeout, + Stream: stream, + }) + if errors.Is(err, conversation.ErrBusy) { + http.Error( + w, + http.StatusText(http.StatusConflict), + http.StatusConflict, + ) + return + } + if err != nil { + log.ErrorContext( + ctx, "failed to reserve inference", + slog.Any("error", err), + ) + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + + // Register the background run before returning so Service.Run waits + // for it after HTTP shutdown completes. + svc.wg.Go(func() { + if err := inf.Run(); err != nil { + log.ErrorContext( + ctx, "background inference failed", + slog.Any("error", err), + ) + } + }) + + w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("Connection", "keep-alive") + w.Header().Set("Content-Type", "text/event-stream") + for e := range inf.Events(ctx) { + data, err := json.Marshal(e) + 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", e.Type(), data) + flusher.Flush() + } +} + +func (svc *Service) stop(w http.ResponseWriter, r *http.Request) { + var ( + ctx = r.Context() + log = ctx.Value(logKey).(*slog.Logger) + id = r.PathValue("id") + ) + + if _, err := svc.dbr.SelectConversationIdUserId( + ctx, id, "default", + ); err != nil { + if errors.Is(err, sql.ErrNoRows) { + http.NotFound(w, r) + return + } + log.ErrorContext( + ctx, "failed to load conversation", + slog.Any("error", err), + ) + http.Error( + w, + http.StatusText(http.StatusInternalServerError), + http.StatusInternalServerError, + ) + return + } + c, err := svc.getConversation(ctx, id) + if errors.Is(err, sql.ErrNoRows) { + http.NotFound(w, r) + return + } + if err != nil { + log.ErrorContext( + ctx, "failed to get conversation handle", + slog.Any("error", err), + ) + http.Error( + w, + http.StatusText(http.StatusInternalServerError), + http.StatusInternalServerError, + ) + return + } + + c.Stop() + w.WriteHeader(http.StatusNoContent) +} + +func (svc *Service) subscribe(w http.ResponseWriter, r *http.Request) { + var ( + ctx = r.Context() + log = ctx.Value(logKey).(*slog.Logger) + id = r.PathValue("id") + req struct { + Manifest []struct { + Id string `json:"id"` + UpdatedAt time.Time `json:"updated_at"` + } `json:"manifest"` + } + ) + + flusher, ok := w.(http.Flusher) + if !ok { + http.Error( + w, + "streaming not supported", + http.StatusInternalServerError, + ) + return + } + + r.Body = http.MaxBytesReader(w, r.Body, 32<<20) + if err := json.UnmarshalRead(r.Body, &req); err != nil { + log.ErrorContext( + ctx, "failed to decode request", + slog.Any("error", err), + ) + http.Error(w, "invalid request", http.StatusBadRequest) + return + } + + if _, err := svc.dbr.SelectConversationIdUserId( + ctx, id, "default", + ); err != nil { + if errors.Is(err, sql.ErrNoRows) { + http.NotFound(w, r) + return + } + log.ErrorContext( + ctx, "failed to load conversation", + slog.Any("error", err), + ) + http.Error( + w, + http.StatusText(http.StatusInternalServerError), + http.StatusInternalServerError, + ) + return + } + c, err := svc.getConversation(ctx, id) + if errors.Is(err, sql.ErrNoRows) { + http.NotFound(w, r) + return + } + if err != nil { + log.ErrorContext( + ctx, "failed to get conversation handle", + slog.Any("error", err), + ) + http.Error( + w, + http.StatusText(http.StatusInternalServerError), + http.StatusInternalServerError, + ) + return + } + + manifest := make([]conversation.MessageManifest, len(req.Manifest)) + for i, m := range req.Manifest { + manifest[i] = conversation.MessageManifest{ + Id: m.Id, + UpdatedAt: m.UpdatedAt, + } + } + events, err := c.Subscribe(ctx, manifest) + if err != nil { + log.ErrorContext( + ctx, "failed to subscribe to conversation", + slog.Any("error", err), + ) + http.Error( + w, + http.StatusText(http.StatusInternalServerError), + http.StatusInternalServerError, + ) + return + } + + w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("Connection", "keep-alive") + w.Header().Set("Content-Type", "text/event-stream") + for e := range events { + data, err := json.Marshal(e) + 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", e.Type(), data) + flusher.Flush() + } +} diff --git a/internal/service/messages.go b/internal/service/messages.go new file mode 100644 index 0000000..84ecec6 --- /dev/null +++ b/internal/service/messages.go @@ -0,0 +1,595 @@ +package service + +import ( + "database/sql" + "encoding/json/v2" + "errors" + "log/slog" + "net/http" + "time" + + "code.chimeric.al/chimerical/odidere/internal/conversation" + "code.chimeric.al/chimerical/odidere/internal/db" + "code.chimeric.al/chimerical/odidere/internal/llm" +) + +func (svc *Service) insertMessage(w http.ResponseWriter, r *http.Request) { + var ( + ctx = r.Context() + log = ctx.Value(logKey).(*slog.Logger) + id = r.PathValue("id") + req struct { + Message *llm.Message `json:"message"` + MessageId string `json:"message_id"` + UpdatedAt time.Time `json:"updated_at"` + } + ) + + if _, err := svc.dbr.SelectConversationIdUserId( + ctx, id, "default", + ); err != nil { + if errors.Is(err, sql.ErrNoRows) { + http.NotFound(w, r) + return + } + log.ErrorContext( + ctx, "failed to load conversation", + slog.Any("error", err), + ) + http.Error( + w, + http.StatusText(http.StatusInternalServerError), + http.StatusInternalServerError, + ) + return + } + c, err := svc.getConversation(ctx, id) + if errors.Is(err, sql.ErrNoRows) { + http.NotFound(w, r) + return + } + if err != nil { + log.ErrorContext( + ctx, "failed to get conversation handle", + slog.Any("error", err), + ) + http.Error( + w, + http.StatusText(http.StatusInternalServerError), + http.StatusInternalServerError, + ) + return + } + + r.Body = http.MaxBytesReader(w, r.Body, 32<<20) + if err := json.UnmarshalRead(r.Body, &req); err != nil { + log.ErrorContext( + ctx, "failed to decode request", + slog.Any("error", err), + ) + http.Error(w, "invalid request", http.StatusBadRequest) + return + } + if req.Message == nil { + http.Error(w, "invalid request", http.StatusBadRequest) + return + } + + var saved *db.Message + if req.Message.Role == llm.RoleTool { + if req.MessageId == "" || req.UpdatedAt.IsZero() { + http.Error(w, "invalid request", http.StatusBadRequest) + return + } + saved, err = c.AddToolResult( + ctx, + req.MessageId, + req.Message.ToolCallID, + req.UpdatedAt, + req.Message.Text(), + ) + } else { + saved, err = c.InsertMessage(ctx, *req.Message) + } + if err != nil { + switch { + case errors.Is(err, conversation.ErrBusy), + errors.Is(err, conversation.ErrRevisionConflict): + http.Error( + w, + http.StatusText(http.StatusConflict), + http.StatusConflict, + ) + case errors.Is(err, sql.ErrNoRows): + http.NotFound(w, r) + default: + log.ErrorContext( + ctx, "failed to insert message", + slog.Any("error", err), + ) + http.Error(w, err.Error(), http.StatusBadRequest) + } + return + } + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusCreated) + if err := json.MarshalWrite(w, saved); err != nil { + log.ErrorContext( + ctx, "failed to encode message response", + slog.Any("error", err), + ) + } +} + +func (svc *Service) updateMessage(w http.ResponseWriter, r *http.Request) { + var ( + ctx = r.Context() + log = ctx.Value(logKey).(*slog.Logger) + id = r.PathValue("id") + mid = r.PathValue("mid") + req struct { + Message *llm.Message `json:"message"` + UpdatedAt time.Time `json:"updated_at"` + } + ) + + if _, err := svc.dbr.SelectConversationIdUserId( + ctx, id, "default", + ); err != nil { + if errors.Is(err, sql.ErrNoRows) { + http.NotFound(w, r) + return + } + log.ErrorContext( + ctx, "failed to load conversation", + slog.Any("error", err), + ) + http.Error( + w, + http.StatusText(http.StatusInternalServerError), + http.StatusInternalServerError, + ) + return + } + c, err := svc.getConversation(ctx, id) + if errors.Is(err, sql.ErrNoRows) { + http.NotFound(w, r) + return + } + if err != nil { + log.ErrorContext( + ctx, "failed to get conversation handle", + slog.Any("error", err), + ) + http.Error( + w, + http.StatusText(http.StatusInternalServerError), + http.StatusInternalServerError, + ) + return + } + + r.Body = http.MaxBytesReader(w, r.Body, 32<<20) + if err := json.UnmarshalRead(r.Body, &req); err != nil { + log.ErrorContext( + ctx, "failed to decode request", + slog.Any("error", err), + ) + http.Error(w, "invalid request", http.StatusBadRequest) + return + } + if req.Message == nil || req.UpdatedAt.IsZero() { + http.Error(w, "invalid request", http.StatusBadRequest) + return + } + + var saved *db.Message + if req.Message.Role == llm.RoleTool { + // mid identifies the parent assistant message, while + // UpdatedAt is the tool result message's updated_at. + saved, err = c.UpdateToolResult( + ctx, + mid, + req.Message.ToolCallID, + req.UpdatedAt, + req.Message.Text(), + ) + } else { + saved, err = c.UpdateMessage( + ctx, mid, req.UpdatedAt, *req.Message, + ) + } + if err != nil { + switch { + case errors.Is(err, conversation.ErrBusy), + errors.Is(err, conversation.ErrRevisionConflict): + http.Error( + w, + http.StatusText(http.StatusConflict), + http.StatusConflict, + ) + case errors.Is(err, sql.ErrNoRows): + http.NotFound(w, r) + default: + log.ErrorContext( + ctx, "failed to update message", + slog.Any("error", err), + ) + http.Error(w, err.Error(), http.StatusBadRequest) + } + return + } + + w.Header().Set("Content-Type", "application/json") + if err := json.MarshalWrite(w, saved); err != nil { + log.ErrorContext( + ctx, "failed to encode message response", + slog.Any("error", err), + ) + } +} + +func (svc *Service) deleteMessage(w http.ResponseWriter, r *http.Request) { + var ( + ctx = r.Context() + log = ctx.Value(logKey).(*slog.Logger) + id = r.PathValue("id") + mid = r.PathValue("mid") + req struct { + UpdatedAt time.Time `json:"updated_at"` + } + ) + + if _, err := svc.dbr.SelectConversationIdUserId( + ctx, id, "default", + ); err != nil { + if errors.Is(err, sql.ErrNoRows) { + http.NotFound(w, r) + return + } + log.ErrorContext( + ctx, "failed to load conversation", + slog.Any("error", err), + ) + http.Error( + w, + http.StatusText(http.StatusInternalServerError), + http.StatusInternalServerError, + ) + return + } + c, err := svc.getConversation(ctx, id) + if errors.Is(err, sql.ErrNoRows) { + http.NotFound(w, r) + return + } + if err != nil { + log.ErrorContext( + ctx, "failed to get conversation handle", + slog.Any("error", err), + ) + http.Error( + w, + http.StatusText(http.StatusInternalServerError), + http.StatusInternalServerError, + ) + return + } + + r.Body = http.MaxBytesReader(w, r.Body, 32<<20) + if err := json.UnmarshalRead(r.Body, &req); err != nil { + log.ErrorContext( + ctx, "failed to decode request", + slog.Any("error", err), + ) + http.Error(w, "invalid request", http.StatusBadRequest) + return + } + if req.UpdatedAt.IsZero() { + http.Error(w, "invalid request", http.StatusBadRequest) + return + } + + if err := c.DeleteMessage(ctx, mid, req.UpdatedAt); err != nil { + switch { + case errors.Is(err, conversation.ErrBusy), + errors.Is(err, conversation.ErrRevisionConflict): + http.Error( + w, + http.StatusText(http.StatusConflict), + http.StatusConflict, + ) + case errors.Is(err, sql.ErrNoRows): + http.NotFound(w, r) + default: + log.ErrorContext( + ctx, "failed to delete message", + slog.Any("error", err), + ) + http.Error( + w, + http.StatusText(http.StatusInternalServerError), + http.StatusInternalServerError, + ) + } + return + } + + w.WriteHeader(http.StatusNoContent) +} + +func (svc *Service) addToolCall(w http.ResponseWriter, r *http.Request) { + var ( + ctx = r.Context() + log = ctx.Value(logKey).(*slog.Logger) + id = r.PathValue("id") + mid = r.PathValue("mid") + req struct { + Call *llm.ToolCall `json:"call"` + UpdatedAt time.Time `json:"updated_at"` + } + ) + + if _, err := svc.dbr.SelectConversationIdUserId( + ctx, id, "default", + ); err != nil { + if errors.Is(err, sql.ErrNoRows) { + http.NotFound(w, r) + return + } + log.ErrorContext( + ctx, "failed to load conversation", + slog.Any("error", err), + ) + http.Error( + w, + http.StatusText(http.StatusInternalServerError), + http.StatusInternalServerError, + ) + return + } + c, err := svc.getConversation(ctx, id) + if errors.Is(err, sql.ErrNoRows) { + http.NotFound(w, r) + return + } + if err != nil { + log.ErrorContext( + ctx, "failed to get conversation handle", + slog.Any("error", err), + ) + http.Error( + w, + http.StatusText(http.StatusInternalServerError), + http.StatusInternalServerError, + ) + return + } + + r.Body = http.MaxBytesReader(w, r.Body, 32<<20) + if err := json.UnmarshalRead(r.Body, &req); err != nil { + log.ErrorContext( + ctx, "failed to decode request", + slog.Any("error", err), + ) + http.Error(w, "invalid request", http.StatusBadRequest) + return + } + if req.Call == nil || req.UpdatedAt.IsZero() { + http.Error(w, "invalid request", http.StatusBadRequest) + return + } + + saved, err := c.AddToolCall(ctx, mid, req.UpdatedAt, *req.Call) + if err != nil { + switch { + case errors.Is(err, conversation.ErrBusy), + errors.Is(err, conversation.ErrRevisionConflict): + http.Error( + w, + http.StatusText(http.StatusConflict), + http.StatusConflict, + ) + case errors.Is(err, sql.ErrNoRows): + http.NotFound(w, r) + default: + log.ErrorContext( + ctx, "failed to add tool call", + slog.Any("error", err), + ) + http.Error(w, err.Error(), http.StatusBadRequest) + } + return + } + + w.Header().Set("Content-Type", "application/json") + if err := json.MarshalWrite(w, saved); err != nil { + log.ErrorContext( + ctx, "failed to encode message response", + slog.Any("error", err), + ) + } +} + +func (svc *Service) updateToolCall(w http.ResponseWriter, r *http.Request) { + var ( + ctx = r.Context() + log = ctx.Value(logKey).(*slog.Logger) + id = r.PathValue("id") + mid = r.PathValue("mid") + cid = r.PathValue("cid") + req struct { + Call *llm.ToolCall `json:"call"` + UpdatedAt time.Time `json:"updated_at"` + } + ) + + if _, err := svc.dbr.SelectConversationIdUserId( + ctx, id, "default", + ); err != nil { + if errors.Is(err, sql.ErrNoRows) { + http.NotFound(w, r) + return + } + log.ErrorContext( + ctx, "failed to load conversation", + slog.Any("error", err), + ) + http.Error( + w, + http.StatusText(http.StatusInternalServerError), + http.StatusInternalServerError, + ) + return + } + c, err := svc.getConversation(ctx, id) + if errors.Is(err, sql.ErrNoRows) { + http.NotFound(w, r) + return + } + if err != nil { + log.ErrorContext( + ctx, "failed to get conversation handle", + slog.Any("error", err), + ) + http.Error( + w, + http.StatusText(http.StatusInternalServerError), + http.StatusInternalServerError, + ) + return + } + + r.Body = http.MaxBytesReader(w, r.Body, 32<<20) + if err := json.UnmarshalRead(r.Body, &req); err != nil { + log.ErrorContext( + ctx, "failed to decode request", + slog.Any("error", err), + ) + http.Error(w, "invalid request", http.StatusBadRequest) + return + } + if req.Call == nil || req.UpdatedAt.IsZero() || req.Call.ID != cid { + http.Error(w, "invalid request", http.StatusBadRequest) + return + } + + saved, err := c.UpdateToolCall( + ctx, mid, cid, req.UpdatedAt, *req.Call, + ) + if err != nil { + switch { + case errors.Is(err, conversation.ErrBusy), + errors.Is(err, conversation.ErrRevisionConflict): + http.Error( + w, + http.StatusText(http.StatusConflict), + http.StatusConflict, + ) + case errors.Is(err, sql.ErrNoRows): + http.NotFound(w, r) + default: + log.ErrorContext( + ctx, "failed to update tool call", + slog.Any("error", err), + ) + http.Error(w, err.Error(), http.StatusBadRequest) + } + return + } + + w.Header().Set("Content-Type", "application/json") + if err := json.MarshalWrite(w, saved); err != nil { + log.ErrorContext( + ctx, "failed to encode message response", + slog.Any("error", err), + ) + } +} + +func (svc *Service) deleteToolCall(w http.ResponseWriter, r *http.Request) { + var ( + ctx = r.Context() + log = ctx.Value(logKey).(*slog.Logger) + id = r.PathValue("id") + mid = r.PathValue("mid") + cid = r.PathValue("cid") + req struct { + UpdatedAt time.Time `json:"updated_at"` + } + ) + + if _, err := svc.dbr.SelectConversationIdUserId( + ctx, id, "default", + ); err != nil { + if errors.Is(err, sql.ErrNoRows) { + http.NotFound(w, r) + return + } + log.ErrorContext( + ctx, "failed to load conversation", + slog.Any("error", err), + ) + http.Error( + w, + http.StatusText(http.StatusInternalServerError), + http.StatusInternalServerError, + ) + return + } + c, err := svc.getConversation(ctx, id) + if errors.Is(err, sql.ErrNoRows) { + http.NotFound(w, r) + return + } + if err != nil { + log.ErrorContext( + ctx, "failed to get conversation handle", + slog.Any("error", err), + ) + http.Error( + w, + http.StatusText(http.StatusInternalServerError), + http.StatusInternalServerError, + ) + return + } + + r.Body = http.MaxBytesReader(w, r.Body, 32<<20) + if err := json.UnmarshalRead(r.Body, &req); err != nil { + log.ErrorContext( + ctx, "failed to decode request", + slog.Any("error", err), + ) + http.Error(w, "invalid request", http.StatusBadRequest) + return + } + if req.UpdatedAt.IsZero() { + http.Error(w, "invalid request", http.StatusBadRequest) + return + } + + if err := c.DeleteToolCall(ctx, mid, cid, req.UpdatedAt); err != nil { + switch { + case errors.Is(err, conversation.ErrBusy), + errors.Is(err, conversation.ErrRevisionConflict): + http.Error( + w, + http.StatusText(http.StatusConflict), + http.StatusConflict, + ) + case errors.Is(err, sql.ErrNoRows): + http.NotFound(w, r) + default: + log.ErrorContext( + ctx, "failed to delete tool call", + slog.Any("error", err), + ) + http.Error(w, err.Error(), http.StatusBadRequest) + } + return + } + + w.WriteHeader(http.StatusNoContent) +} diff --git a/internal/service/service.go b/internal/service/service.go index 57e0587..c3faba7 100644 --- a/internal/service/service.go +++ b/internal/service/service.go @@ -65,6 +65,7 @@ type Service struct { ctx context.Context cancel context.CancelFunc + wg sync.WaitGroup mu sync.Mutex conversations map[string]*conversation.Conversation } @@ -331,6 +332,18 @@ func (svc *Service) ServeHTTP(w http.ResponseWriter, r *http.Request) { // 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", @@ -463,13 +476,21 @@ func (svc *Service) Run(ctx context.Context) error { if err := svc.server.Shutdown(shutdownCtx); err != nil { 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. + // 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 <-cronCtx.Done(): + case <-done: case <-shutdownCtx.Done(): shutdownErr = fmt.Errorf( - "wait for jobs: %w", shutdownCtx.Err(), + "wait for background work: %w", + shutdownCtx.Err(), ) } } @@ -487,6 +508,107 @@ func (svc *Service) Run(ctx context.Context) error { 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 diff --git a/internal/service/service_test.go b/internal/service/service_test.go index 8f86492..063b59f 100644 --- a/internal/service/service_test.go +++ b/internal/service/service_test.go @@ -1,15 +1,81 @@ package service import ( - "encoding/json" + "context" + "encoding/json/v2" + "fmt" + "log/slog" "net/http" "net/http/httptest" + "path/filepath" "strings" "testing" + "time" + "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/llm" ) +func newTestService(t *testing.T, handler http.HandlerFunc) *Service { + t.Helper() + if handler == nil { + handler = func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + fmt.Fprint(w, `{ + "choices":[{ + "message":{"role":"assistant","content":"ok"}, + "finish_reason":"stop" + }] + }`) + } + } + srv := httptest.NewServer(handler) + t.Cleanup(srv.Close) + + var ( + ctx = t.Context() + log = slog.New(slog.DiscardHandler) + ) + database, err := db.Connect(ctx, db.Config{ + Path: filepath.Join(t.TempDir(), "service.db"), + }) + if err != nil { + t.Fatalf("Connect: %v", err) + } + t.Cleanup(func() { database.Close() }) + if err := database.ApplySchema(ctx, log); err != nil { + t.Fatalf("ApplySchema: %v", err) + } + + cfg := &config.Config{ + Address: ":0", + Database: config.Database{ + Path: filepath.Join(t.TempDir(), "service.db"), + }, + DefaultModel: config.ModelConfig{ + Provider: "test", + Model: "test-model", + }, + Providers: []config.ProviderConfig{{ + Name: "test", + URL: srv.URL, + Models: []string{"test-model"}, + }}, + STT: config.STT{URL: "http://localhost:8178"}, + TTS: config.TTS{URL: "http://localhost:8880", Voice: "af"}, + SystemMessage: "Default system.", + } + cfg.ApplyDefaults() + + svc, err := New(cfg, log, database) + if err != nil { + t.Fatalf("New: %v", err) + } + return svc +} + func TestStatusHandler(t *testing.T) { svc := &Service{} @@ -27,83 +93,348 @@ func TestStatusHandler(t *testing.T) { } } -func TestResponseIncludesStatsInJSON(t *testing.T) { - resp := Response{ - Messages: []llm.Message{ - {Role: llm.RoleAssistant, ContentParts: []llm.ContentPart{{ - Type: llm.ContentTypeText, - Text: "hello", - }}}, - }, - Provider: "test-provider", - Model: "test-model", - Usage: &llm.Usage{ - PromptTokens: 10, - CompletionTokens: 5, - TotalTokens: 15, - PromptTokensDetails: &llm.TokenDetails{ - CachedTokens: 3, - }, - CompletionTokensDetails: &llm.TokenDetails{ - ReasoningTokens: 2, - }, - }, - Timings: &llm.Timings{ - PromptMS: 50.5, - PromptPerSecond: 198.0, - PredictedMS: 100.0, - PredictedPerSecond: 50.0, - }, +func TestConversationHandlers(t *testing.T) { + svc := newTestService(t, nil) + + w := httptest.NewRecorder() + svc.ServeHTTP(w, httptest.NewRequest( + http.MethodPost, "/conversations", + strings.NewReader(`{"title":"Chat"}`), + )) + if w.Code != http.StatusCreated { + t.Fatalf("create status = %d, want 201: %s", w.Code, w.Body.String()) + } + var created db.Conversation + if err := json.Unmarshal(w.Body.Bytes(), &created); err != nil { + t.Fatalf("Unmarshal: %v", err) + } + if created.Title == nil || *created.Title != "Chat" { + t.Fatalf("Title = %v, want Chat", created.Title) } - // Verify JSON encoding includes stats fields - var buf strings.Builder - enc := json.NewEncoder(&buf) - if err := enc.Encode(resp); err != nil { - t.Fatalf("encode Response: %v", err) + w = httptest.NewRecorder() + svc.ServeHTTP(w, httptest.NewRequest( + http.MethodGet, "/conversations", nil, + )) + if w.Code != http.StatusOK { + t.Fatalf("list status = %d, want 200", w.Code) + } + var listed []db.Conversation + if err := json.Unmarshal(w.Body.Bytes(), &listed); err != nil { + t.Fatalf("Unmarshal list: %v", err) + } + if len(listed) != 1 || listed[0].Id != created.Id { + t.Fatalf("listed = %+v, want [%s]", listed, created.Id) } - jsonStr := buf.String() - if !strings.Contains(jsonStr, `"prompt_tokens":10`) { - t.Errorf("JSON missing prompt_tokens: %s", jsonStr) + w = httptest.NewRecorder() + svc.ServeHTTP(w, httptest.NewRequest( + http.MethodGet, "/conversations/"+created.Id, nil, + )) + if w.Code != http.StatusOK { + t.Fatalf("get status = %d, want 200", w.Code) } - if !strings.Contains(jsonStr, `"completion_tokens":5`) { - t.Errorf("JSON missing completion_tokens: %s", jsonStr) + + w = httptest.NewRecorder() + svc.ServeHTTP(w, httptest.NewRequest( + http.MethodDelete, "/conversations/"+created.Id, nil, + )) + if w.Code != http.StatusNoContent { + t.Fatalf("delete status = %d, want 204", w.Code) } - if !strings.Contains(jsonStr, `"total_tokens":15`) { - t.Errorf("JSON missing total_tokens: %s", jsonStr) + if _, ok := svc.conversations[created.Id]; ok { + t.Fatal("deleted conversation handle remained in map") } - if !strings.Contains(jsonStr, `"prompt_ms":50.5`) { - t.Errorf("JSON missing prompt_ms: %s", jsonStr) - } - if !strings.Contains(jsonStr, `"predicted_ms":100`) { - t.Errorf("JSON missing predicted_ms: %s", jsonStr) + + w = httptest.NewRecorder() + svc.ServeHTTP(w, httptest.NewRequest( + http.MethodGet, "/conversations/"+created.Id, nil, + )) + if w.Code != http.StatusNotFound { + t.Fatalf("get deleted status = %d, want 404", w.Code) } } -func TestStreamMessageIncludesUsageInJSON(t *testing.T) { - msg := StreamMessage{ - Delta: "hello", - Provider: "test-provider", - Model: "test-model", - Usage: &llm.Usage{ - PromptTokens: 10, - CompletionTokens: 5, - TotalTokens: 15, - }, +func TestMessageAndToolCallHandlers(t *testing.T) { + svc := newTestService(t, nil) + saved, _, err := svc.newConversation( + t.Context(), "default", nil, "System", + ) + if err != nil { + t.Fatalf("newConversation: %v", err) + } + base := "/conversations/" + saved.Id + "/messages" + + // Insert and update a user message. + w := httptest.NewRecorder() + svc.ServeHTTP(w, httptest.NewRequest( + http.MethodPost, base, + strings.NewReader(`{"message":{"role":"user","content":"hi"}}`), + )) + if w.Code != http.StatusCreated { + t.Fatalf("insert user = %d: %s", w.Code, w.Body.String()) + } + var user db.Message + if err := json.Unmarshal(w.Body.Bytes(), &user); err != nil { + t.Fatalf("Unmarshal: %v", err) } - var buf strings.Builder - enc := json.NewEncoder(&buf) - if err := enc.Encode(msg); err != nil { - t.Fatalf("encode StreamMessage: %v", err) + w = httptest.NewRecorder() + svc.ServeHTTP(w, httptest.NewRequest( + http.MethodPut, base+"/"+user.Id, + strings.NewReader(fmt.Sprintf( + `{"message":{"role":"user","content":"hello"},"updated_at":%q}`, + user.UpdatedAt.Format(time.RFC3339Nano), + )), + )) + if w.Code != http.StatusOK { + t.Fatalf("update user = %d: %s", w.Code, w.Body.String()) + } + if err := json.Unmarshal(w.Body.Bytes(), &user); err != nil { + t.Fatalf("Unmarshal: %v", err) } - jsonStr := buf.String() - if !strings.Contains(jsonStr, `"prompt_tokens":10`) { - t.Errorf("JSON missing prompt_tokens: %s", jsonStr) + // Insert an assistant message, edit its tool call, and edit its result. + w = httptest.NewRecorder() + svc.ServeHTTP(w, httptest.NewRequest( + http.MethodPost, base, + strings.NewReader(`{"message":{"role":"assistant","content":"run"}}`), + )) + if w.Code != http.StatusCreated { + t.Fatalf("insert assistant = %d: %s", w.Code, w.Body.String()) } - if !strings.Contains(jsonStr, `"total_tokens":15`) { - t.Errorf("JSON missing total_tokens: %s", jsonStr) + var asst db.Message + if err := json.Unmarshal(w.Body.Bytes(), &asst); err != nil { + t.Fatalf("Unmarshal: %v", err) + } + + asstPath := base + "/" + asst.Id + w = httptest.NewRecorder() + svc.ServeHTTP(w, httptest.NewRequest( + http.MethodPost, asstPath+"/calls", + strings.NewReader(fmt.Sprintf( + `{"call":{"id":"c1","type":"function","function":{"name":"fn","arguments":"{}"}},"updated_at":%q}`, + asst.UpdatedAt.Format(time.RFC3339Nano), + )), + )) + if w.Code != http.StatusOK { + t.Fatalf("add call = %d: %s", w.Code, w.Body.String()) + } + if err := json.Unmarshal(w.Body.Bytes(), &asst); err != nil { + t.Fatalf("Unmarshal: %v", err) + } + + w = httptest.NewRecorder() + svc.ServeHTTP(w, httptest.NewRequest( + http.MethodPut, asstPath+"/calls/c1", + strings.NewReader(fmt.Sprintf( + `{"call":{"id":"c1","type":"function","function":{"name":"fn2","arguments":"{}"}},"updated_at":%q}`, + asst.UpdatedAt.Format(time.RFC3339Nano), + )), + )) + if w.Code != http.StatusOK { + t.Fatalf("update call = %d: %s", w.Code, w.Body.String()) + } + if err := json.Unmarshal(w.Body.Bytes(), &asst); err != nil { + t.Fatalf("Unmarshal: %v", err) + } + + w = httptest.NewRecorder() + svc.ServeHTTP(w, httptest.NewRequest( + http.MethodPost, base, + strings.NewReader(fmt.Sprintf( + `{"message":{"role":"tool","tool_call_id":"c1","content":"out"},"message_id":%q,"updated_at":%q}`, + asst.Id, asst.UpdatedAt.Format(time.RFC3339Nano), + )), + )) + if w.Code != http.StatusCreated { + t.Fatalf("add tool result = %d: %s", w.Code, w.Body.String()) + } + var result db.Message + if err := json.Unmarshal(w.Body.Bytes(), &result); err != nil { + t.Fatalf("Unmarshal: %v", err) + } + + w = httptest.NewRecorder() + svc.ServeHTTP(w, httptest.NewRequest( + http.MethodPut, asstPath, + strings.NewReader(fmt.Sprintf( + `{"message":{"role":"tool","tool_call_id":"c1","content":"new"},"updated_at":%q}`, + result.UpdatedAt.Format(time.RFC3339Nano), + )), + )) + if w.Code != http.StatusOK { + t.Fatalf("update tool result = %d: %s", w.Code, w.Body.String()) + } + + w = httptest.NewRecorder() + svc.ServeHTTP(w, httptest.NewRequest( + http.MethodDelete, asstPath+"/calls/c1", + strings.NewReader(fmt.Sprintf( + `{"updated_at":%q}`, + asst.UpdatedAt.Format(time.RFC3339Nano), + )), + )) + if w.Code != http.StatusNoContent { + t.Fatalf("delete call = %d: %s", w.Code, w.Body.String()) + } + + w = httptest.NewRecorder() + svc.ServeHTTP(w, httptest.NewRequest( + http.MethodDelete, base+"/"+user.Id, + strings.NewReader(fmt.Sprintf( + `{"updated_at":%q}`, + user.UpdatedAt.Format(time.RFC3339Nano), + )), + )) + if w.Code != http.StatusNoContent { + t.Fatalf("delete user = %d: %s", w.Code, w.Body.String()) + } +} + +func TestExecutionHandlers(t *testing.T) { + svc := newTestService(t, nil) + _, c, err := svc.newConversation( + t.Context(), "default", nil, "System", + ) + if err != nil { + t.Fatalf("newConversation: %v", err) + } + asst, err := c.InsertMessage(t.Context(), llm.Message{ + Role: llm.RoleAssistant, + ToolCalls: []llm.ToolCall{{ + ID: "c1", + Type: llm.ToolTypeFunction, + Function: llm.Function{Name: "fn", Arguments: "{}"}, + }}, + }) + if err != nil { + t.Fatalf("InsertMessage: %v", err) + } + exec, err := svc.dbw.InsertExecution( + t.Context(), db.InsertExecutionParams{ + MessageId: asst.Id, + ToolCallId: "c1", + Status: db.ExecutionStatusPending, + }, + ) + if err != nil { + t.Fatalf("InsertExecution: %v", err) + } + + w := httptest.NewRecorder() + svc.ServeHTTP(w, httptest.NewRequest( + http.MethodGet, "/executions", nil, + )) + if w.Code != http.StatusOK { + t.Fatalf("list status = %d", w.Code) + } + var pending []db.Execution + if err := json.Unmarshal(w.Body.Bytes(), &pending); err != nil { + t.Fatalf("Unmarshal: %v", err) + } + if len(pending) != 1 || pending[0].Id != exec.Id { + t.Fatalf("pending = %+v, want [%s]", pending, exec.Id) + } + + w = httptest.NewRecorder() + svc.ServeHTTP(w, httptest.NewRequest( + http.MethodPut, "/executions/"+exec.Id, + strings.NewReader(`{"status":"approved"}`), + )) + if w.Code != http.StatusNoContent { + t.Fatalf("approve status = %d: %s", w.Code, w.Body.String()) + } + + w = httptest.NewRecorder() + svc.ServeHTTP(w, httptest.NewRequest( + http.MethodPut, "/executions/"+exec.Id, + strings.NewReader(`{"status":"denied"}`), + )) + if w.Code != http.StatusConflict { + t.Fatalf("second decision status = %d, want 409", w.Code) + } +} + +func TestInferenceSubscribeAndStopHandlers(t *testing.T) { + started := make(chan struct{}) + release := make(chan struct{}) + svc := newTestService(t, func(w http.ResponseWriter, r *http.Request) { + close(started) + select { + case <-release: + case <-r.Context().Done(): + return + } + w.Header().Set("Content-Type", "application/json") + fmt.Fprint(w, `{ + "choices":[{ + "message":{"role":"assistant","content":"reply"}, + "finish_reason":"stop" + }] + }`) + }) + saved, _, err := svc.newConversation( + t.Context(), "default", nil, "System", + ) + if err != nil { + t.Fatalf("newConversation: %v", err) + } + + // Disconnect the inference client while the background run continues. + reqCtx, cancelReq := context.WithCancel(t.Context()) + doneReq := make(chan *httptest.ResponseRecorder, 1) + go func() { + w := httptest.NewRecorder() + req := httptest.NewRequestWithContext( + reqCtx, + http.MethodPost, + "/conversations/"+saved.Id+"/inference", + strings.NewReader(`{"message":{"role":"user","content":"hi"},"stream":false}`), + ) + svc.ServeHTTP(w, req) + doneReq <- w + }() + + <-started + cancelReq() + <-doneReq + + // A second inference while reserved returns 409 Conflict. + w := httptest.NewRecorder() + svc.ServeHTTP(w, httptest.NewRequest( + http.MethodPost, + "/conversations/"+saved.Id+"/inference", + strings.NewReader(`{"stream":false}`), + )) + if w.Code != http.StatusConflict { + t.Fatalf("concurrent inference = %d, want 409", w.Code) + } + + close(release) + svc.wg.Wait() + + w = httptest.NewRecorder() + svc.ServeHTTP(w, httptest.NewRequest( + http.MethodPost, + "/conversations/"+saved.Id+"/subscribe", + strings.NewReader(`{}`), + )) + if w.Code != http.StatusOK { + t.Fatalf("subscribe status = %d: %s", w.Code, w.Body.String()) + } + body := w.Body.String() + if strings.Count(body, "event: "+string(conversation.EventMessageSaved)) != 3 { + t.Fatalf("subscribe events = %s", body) + } + + w = httptest.NewRecorder() + svc.ServeHTTP(w, httptest.NewRequest( + http.MethodPost, + "/conversations/"+saved.Id+"/stop", + nil, + )) + if w.Code != http.StatusNoContent { + t.Fatalf("stop status = %d, want 204", w.Code) } }