diff --git a/README.md b/README.md index 87d88d4d..2f051614 100644 --- a/README.md +++ b/README.md @@ -195,6 +195,10 @@ docker run --rm -p 8080:8080 --env-file .env gomodel | --------------------------- | ------ | ------------------------------------------------------------------------------------------------------------ | | `/v1/chat/completions` | POST | Chat completions (streaming supported) | | `/v1/responses` | POST | OpenAI Responses API | +| `/v1/conversations` | POST | Create a conversation (gateway-managed) | +| `/v1/conversations/{id}` | GET | Retrieve a conversation | +| `/v1/conversations/{id}` | POST | Replace conversation metadata in full | +| `/v1/conversations/{id}` | DELETE | Delete a conversation | | `/v1/embeddings` | POST | Text embeddings | | `/v1/models` | GET | List available models | | `/v1/files` | POST | Upload a file (OpenAI-compatible multipart) | diff --git a/docs/advanced/conversations-api.mdx b/docs/advanced/conversations-api.mdx new file mode 100644 index 00000000..64ddd703 --- /dev/null +++ b/docs/advanced/conversations-api.mdx @@ -0,0 +1,103 @@ +--- +title: "Conversations API" +description: "Create, retrieve, update, and delete OpenAI-compatible conversations through GoModel." +icon: "messages-square" +tag: "Beta" +--- + +## Overview + +GoModel exposes OpenAI-compatible Conversations API endpoints under +`/v1/conversations`. + +Conversations are a **gateway-managed resource**. The OpenAI Conversations API +is provider-specific, so GoModel generates the conversation ID and stores the +conversation itself. This keeps `/v1/conversations` available and consistent +regardless of which provider routes your model traffic, and requires no +provider configuration. + +## Supported endpoints + +| Endpoint | Behavior | +| --- | --- | +| `POST /v1/conversations` | Creates a conversation. Accepts optional `items` and `metadata`. | +| `GET /v1/conversations/{id}` | Returns a stored conversation. | +| `POST /v1/conversations/{id}` | Replaces the conversation `metadata` in full. | +| `DELETE /v1/conversations/{id}` | Deletes a stored conversation. | + +## Conversation object + +```json +{ + "id": "conv_a1b2c3d4e5f6...", + "object": "conversation", + "created_at": 1741900000, + "metadata": {} +} +``` + +`metadata` is always present and serializes as `{}` when empty. + +## Limits + +GoModel enforces the OpenAI Conversations limits so the public contract stays +compatible: + +- `items` — at most 20 entries on create. +- `metadata` — at most 16 key-value pairs; keys up to 64 characters; values up + to 512 characters. + +The `items` array is accepted and stored with the conversation. It is not yet +exposed through a conversation items listing endpoint. + +## Storage and retention + +Conversations are held in an in-memory store. They survive across requests but +not process restarts. Retention is bounded: conversations expire after 30 days +and the store keeps at most 10,000 conversations, evicting the oldest first. + +## Errors + +GoModel returns OpenAI-compatible errors: + +- `400 invalid_request_error` — invalid body, missing `metadata` on update, or a + limit exceeded (the `param` field names the offending field). +- `404 not_found_error` — the conversation ID does not exist. + +## Examples + +Create a conversation: + +```bash +curl http://localhost:8080/v1/conversations \ + -H "Authorization: Bearer $GOMODEL_MASTER_KEY" \ + -H "Content-Type: application/json" \ + -d '{ + "metadata": { "topic": "support" } + }' +``` + +Retrieve it: + +```bash +curl http://localhost:8080/v1/conversations/conv_abc123 \ + -H "Authorization: Bearer $GOMODEL_MASTER_KEY" +``` + +Update its metadata (replaces all metadata): + +```bash +curl http://localhost:8080/v1/conversations/conv_abc123 \ + -H "Authorization: Bearer $GOMODEL_MASTER_KEY" \ + -H "Content-Type: application/json" \ + -d '{ + "metadata": { "topic": "billing" } + }' +``` + +Delete it: + +```bash +curl -X DELETE http://localhost:8080/v1/conversations/conv_abc123 \ + -H "Authorization: Bearer $GOMODEL_MASTER_KEY" +``` diff --git a/docs/docs.json b/docs/docs.json index 08c4c35b..a32d09b6 100644 --- a/docs/docs.json +++ b/docs/docs.json @@ -61,6 +61,7 @@ "advanced/config-yaml", "advanced/resilience", "advanced/responses-api", + "advanced/conversations-api", "advanced/anthropic-messages-api", "advanced/guardrails", "advanced/workflows", diff --git a/internal/conversationstore/store.go b/internal/conversationstore/store.go new file mode 100644 index 00000000..d5764091 --- /dev/null +++ b/internal/conversationstore/store.go @@ -0,0 +1,84 @@ +// Package conversationstore provides persistence for the OpenAI-compatible +// Conversations lifecycle endpoints. +package conversationstore + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + "time" + + "gomodel/internal/core" +) + +// ErrNotFound indicates a requested conversation was not found. +var ErrNotFound = errors.New("conversation not found") + +// StoredConversation keeps the public conversation snapshot separate from +// gateway-only metadata (initial items, owning user path, request id). +type StoredConversation struct { + Conversation *core.Conversation `json:"conversation"` + Items []json.RawMessage `json:"items,omitempty"` + UserPath string `json:"user_path,omitempty"` + RequestID string `json:"request_id,omitempty"` + StoredAt time.Time `json:"stored_at,omitempty"` + ExpiresAt time.Time `json:"expires_at,omitempty"` +} + +// Store defines persistence operations for the Conversations lifecycle API. +type Store interface { + Create(ctx context.Context, conversation *StoredConversation) error + Get(ctx context.Context, id string) (*StoredConversation, error) + Update(ctx context.Context, conversation *StoredConversation) error + Delete(ctx context.Context, id string) error + Close() error +} + +func cloneConversation(src *StoredConversation) (*StoredConversation, error) { + if src == nil { + return nil, fmt.Errorf("conversation is nil") + } + normalized := normalizeStoredConversation(src) + b, err := json.Marshal(normalized) + if err != nil { + return nil, fmt.Errorf("marshal conversation: %w", err) + } + var dst StoredConversation + if err := json.Unmarshal(b, &dst); err != nil { + return nil, fmt.Errorf("unmarshal conversation: %w", err) + } + return &dst, nil +} + +func normalizeStoredConversation(src *StoredConversation) *StoredConversation { + if src == nil { + return nil + } + + normalized := *src + normalized.UserPath = strings.TrimSpace(normalized.UserPath) + normalized.RequestID = strings.TrimSpace(normalized.RequestID) + + if src.Conversation != nil { + conversationCopy := *src.Conversation + if conversationCopy.Metadata != nil { + metadataCopy := make(map[string]string, len(conversationCopy.Metadata)) + for key, value := range conversationCopy.Metadata { + metadataCopy[key] = value + } + conversationCopy.Metadata = metadataCopy + } + normalized.Conversation = &conversationCopy + } + + if len(src.Items) > 0 { + normalized.Items = make([]json.RawMessage, 0, len(src.Items)) + for _, item := range src.Items { + normalized.Items = append(normalized.Items, core.CloneRawJSON(item)) + } + } + + return &normalized +} diff --git a/internal/conversationstore/store_memory.go b/internal/conversationstore/store_memory.go new file mode 100644 index 00000000..e54c4160 --- /dev/null +++ b/internal/conversationstore/store_memory.go @@ -0,0 +1,252 @@ +package conversationstore + +import ( + "context" + "fmt" + "sort" + "sync" + "time" +) + +const ( + // DefaultMemoryStoreTTL bounds in-memory conversation retention by age. + // It mirrors the OpenAI Conversations retention window (~30 days). + DefaultMemoryStoreTTL = 30 * 24 * time.Hour + // DefaultMemoryStoreMaxEntries bounds in-memory conversation retention by count. + DefaultMemoryStoreMaxEntries = 10000 + // DefaultMemoryStoreCleanupInterval limits full expired-entry sweeps. + DefaultMemoryStoreCleanupInterval = time.Minute +) + +// MemoryStore keeps conversation snapshots in process memory. +// Data survives across requests but not process restarts. +type MemoryStore struct { + mu sync.RWMutex + items map[string]*StoredConversation + ttl time.Duration + maxEntries int + lastCleanup time.Time + cleanupInterval time.Duration +} + +// MemoryStoreOption configures bounded in-memory conversation retention. +type MemoryStoreOption func(*MemoryStore) + +// WithTTL expires stored conversations after ttl. Non-positive values disable TTL. +func WithTTL(ttl time.Duration) MemoryStoreOption { + return func(s *MemoryStore) { + s.ttl = ttl + } +} + +// WithMaxEntries caps stored conversations with FIFO eviction. Non-positive values disable the cap. +func WithMaxEntries(maxEntries int) MemoryStoreOption { + return func(s *MemoryStore) { + s.maxEntries = maxEntries + } +} + +// WithUnboundedRetention disables default in-memory retention bounds. +func WithUnboundedRetention() MemoryStoreOption { + return func(s *MemoryStore) { + s.ttl = 0 + s.maxEntries = 0 + } +} + +// NewMemoryStore creates an empty in-memory conversation store. +// By default retention is bounded; pass WithUnboundedRetention to opt out. +func NewMemoryStore(options ...MemoryStoreOption) *MemoryStore { + store := &MemoryStore{ + items: make(map[string]*StoredConversation), + ttl: DefaultMemoryStoreTTL, + maxEntries: DefaultMemoryStoreMaxEntries, + cleanupInterval: DefaultMemoryStoreCleanupInterval, + } + for _, option := range options { + if option != nil { + option(store) + } + } + return store +} + +// Create stores a new conversation snapshot. +func (s *MemoryStore) Create(_ context.Context, conversation *StoredConversation) error { + if conversation == nil || conversation.Conversation == nil || conversation.Conversation.ID == "" { + return fmt.Errorf("conversation id is required") + } + + c, err := cloneConversation(conversation) + if err != nil { + return err + } + + now := time.Now().UTC() + prepareStoredConversationForMemory(c, now, s.ttl) + + s.mu.Lock() + defer s.mu.Unlock() + s.cleanupExpiredLocked(now) + if conversationExpired(c, now) { + return nil + } + if existing, exists := s.items[c.Conversation.ID]; exists { + if !conversationExpired(existing, now) { + return fmt.Errorf("conversation already exists: %s", c.Conversation.ID) + } + delete(s.items, c.Conversation.ID) + } + s.items[c.Conversation.ID] = c + s.enforceMaxEntriesLocked() + return nil +} + +// Get retrieves one conversation snapshot by id. +func (s *MemoryStore) Get(_ context.Context, id string) (*StoredConversation, error) { + now := time.Now().UTC() + s.mu.Lock() + s.cleanupExpiredLocked(now) + conversation, ok := s.items[id] + if !ok { + s.mu.Unlock() + return nil, ErrNotFound + } + if conversationExpired(conversation, now) { + delete(s.items, id) + s.mu.Unlock() + return nil, ErrNotFound + } + s.mu.Unlock() + return cloneConversation(conversation) +} + +// Update replaces an existing conversation snapshot. +func (s *MemoryStore) Update(_ context.Context, conversation *StoredConversation) error { + if conversation == nil || conversation.Conversation == nil || conversation.Conversation.ID == "" { + return fmt.Errorf("conversation id is required") + } + c, err := cloneConversation(conversation) + if err != nil { + return err + } + + now := time.Now().UTC() + s.mu.Lock() + defer s.mu.Unlock() + s.cleanupExpiredLocked(now) + existing, exists := s.items[c.Conversation.ID] + if !exists { + return ErrNotFound + } + if conversationExpired(existing, now) { + delete(s.items, c.Conversation.ID) + return ErrNotFound + } + if c.StoredAt.IsZero() { + c.StoredAt = existing.StoredAt + } + if c.ExpiresAt.IsZero() { + c.ExpiresAt = existing.ExpiresAt + } + prepareStoredConversationForMemory(c, now, s.ttl) + if conversationExpired(c, now) { + delete(s.items, c.Conversation.ID) + return ErrNotFound + } + s.items[c.Conversation.ID] = c + s.enforceMaxEntriesLocked() + return nil +} + +// Delete removes one conversation snapshot by id. +func (s *MemoryStore) Delete(_ context.Context, id string) error { + s.mu.Lock() + defer s.mu.Unlock() + now := time.Now().UTC() + s.cleanupExpiredLocked(now) + conversation, exists := s.items[id] + if !exists { + return ErrNotFound + } + // Expired entries report as not found, matching Get and Update, even when + // the throttled cleanup sweep has not removed them yet. + if conversationExpired(conversation, now) { + delete(s.items, id) + return ErrNotFound + } + delete(s.items, id) + return nil +} + +// Close releases resources (no-op for memory store). +func (s *MemoryStore) Close() error { + return nil +} + +func prepareStoredConversationForMemory(conversation *StoredConversation, now time.Time, ttl time.Duration) { + if conversation.StoredAt.IsZero() { + conversation.StoredAt = now + } + if ttl > 0 && conversation.ExpiresAt.IsZero() { + conversation.ExpiresAt = conversation.StoredAt.Add(ttl) + } +} + +func (s *MemoryStore) cleanupExpiredLocked(now time.Time) { + if s.ttl <= 0 { + return + } + if s.cleanupInterval > 0 && !s.lastCleanup.IsZero() && now.Sub(s.lastCleanup) < s.cleanupInterval { + return + } + s.lastCleanup = now + for id, conversation := range s.items { + if conversationExpired(conversation, now) { + delete(s.items, id) + } + } +} + +func (s *MemoryStore) enforceMaxEntriesLocked() { + if s.maxEntries <= 0 { + return + } + overLimit := len(s.items) - s.maxEntries + if overLimit <= 0 { + return + } + + entries := make([]memoryStoreEntry, 0, len(s.items)) + for id, conversation := range s.items { + entries = append(entries, memoryStoreEntry{ + id: id, + storedAt: conversationStoredAt(conversation), + }) + } + sort.Slice(entries, func(i, j int) bool { + if entries[i].storedAt.Equal(entries[j].storedAt) { + return entries[i].id < entries[j].id + } + return entries[i].storedAt.Before(entries[j].storedAt) + }) + for i := 0; i < overLimit && i < len(entries); i++ { + delete(s.items, entries[i].id) + } +} + +type memoryStoreEntry struct { + id string + storedAt time.Time +} + +func conversationExpired(conversation *StoredConversation, now time.Time) bool { + return conversation != nil && !conversation.ExpiresAt.IsZero() && !conversation.ExpiresAt.After(now) +} + +func conversationStoredAt(conversation *StoredConversation) time.Time { + if conversation == nil || conversation.StoredAt.IsZero() { + return time.Time{} + } + return conversation.StoredAt +} diff --git a/internal/conversationstore/store_memory_test.go b/internal/conversationstore/store_memory_test.go new file mode 100644 index 00000000..8cc96081 --- /dev/null +++ b/internal/conversationstore/store_memory_test.go @@ -0,0 +1,166 @@ +package conversationstore + +import ( + "context" + "errors" + "testing" + "time" + + "gomodel/internal/core" +) + +func storedConversation(id string, storedAt time.Time) *StoredConversation { + return &StoredConversation{ + Conversation: &core.Conversation{ + ID: id, + Object: core.ConversationObject, + Metadata: map[string]string{}, + }, + StoredAt: storedAt, + } +} + +func TestMemoryStoreCreateGetUpdateDelete(t *testing.T) { + ctx := context.Background() + store := NewMemoryStore() + + if err := store.Create(ctx, storedConversation("conv_1", time.Time{})); err != nil { + t.Fatalf("Create() error = %v", err) + } + + got, err := store.Get(ctx, "conv_1") + if err != nil { + t.Fatalf("Get() error = %v", err) + } + if got.Conversation.ID != "conv_1" { + t.Fatalf("id = %q, want conv_1", got.Conversation.ID) + } + + got.Conversation.Metadata = map[string]string{"k": "v"} + if err := store.Update(ctx, got); err != nil { + t.Fatalf("Update() error = %v", err) + } + updated, err := store.Get(ctx, "conv_1") + if err != nil { + t.Fatalf("Get() after update error = %v", err) + } + if updated.Conversation.Metadata["k"] != "v" { + t.Fatalf("metadata[k] = %q, want v", updated.Conversation.Metadata["k"]) + } + + if err := store.Delete(ctx, "conv_1"); err != nil { + t.Fatalf("Delete() error = %v", err) + } + if _, err := store.Get(ctx, "conv_1"); !errors.Is(err, ErrNotFound) { + t.Fatalf("Get() after delete error = %v, want ErrNotFound", err) + } +} + +func TestMemoryStoreCreateRejectsDuplicate(t *testing.T) { + ctx := context.Background() + store := NewMemoryStore() + + if err := store.Create(ctx, storedConversation("conv_dup", time.Time{})); err != nil { + t.Fatalf("Create() error = %v", err) + } + if err := store.Create(ctx, storedConversation("conv_dup", time.Time{})); err == nil { + t.Fatal("Create() duplicate error = nil, want error") + } +} + +func TestMemoryStoreUpdateMissingReturnsNotFound(t *testing.T) { + ctx := context.Background() + store := NewMemoryStore() + + if err := store.Update(ctx, storedConversation("conv_missing", time.Time{})); !errors.Is(err, ErrNotFound) { + t.Fatalf("Update() error = %v, want ErrNotFound", err) + } +} + +func TestMemoryStoreDeleteMissingReturnsNotFound(t *testing.T) { + if err := NewMemoryStore().Delete(context.Background(), "conv_missing"); !errors.Is(err, ErrNotFound) { + t.Fatalf("Delete() error = %v, want ErrNotFound", err) + } +} + +func TestMemoryStoreDeleteExpiredReturnsNotFound(t *testing.T) { + ctx := context.Background() + store := NewMemoryStore(WithTTL(time.Second)) + + if err := store.Create(ctx, storedConversation("conv_expired", time.Now().UTC().Add(-2*time.Second))); err != nil { + t.Fatalf("Create() error = %v", err) + } + if err := store.Delete(ctx, "conv_expired"); !errors.Is(err, ErrNotFound) { + t.Fatalf("Delete() error = %v, want ErrNotFound", err) + } +} + +func TestMemoryStoreExpiresConversations(t *testing.T) { + ctx := context.Background() + store := NewMemoryStore(WithTTL(time.Second)) + + if err := store.Create(ctx, storedConversation("conv_old", time.Now().UTC().Add(-2*time.Second))); err != nil { + t.Fatalf("Create() error = %v", err) + } + if _, err := store.Get(ctx, "conv_old"); !errors.Is(err, ErrNotFound) { + t.Fatalf("Get() error = %v, want ErrNotFound", err) + } +} + +func TestMemoryStoreMaxEntriesEvictsOldest(t *testing.T) { + ctx := context.Background() + store := NewMemoryStore(WithTTL(0), WithMaxEntries(2)) + now := time.Now().UTC() + + for _, conversation := range []*StoredConversation{ + storedConversation("conv_1", now.Add(-3*time.Second)), + storedConversation("conv_2", now.Add(-2*time.Second)), + storedConversation("conv_3", now.Add(-1*time.Second)), + } { + if err := store.Create(ctx, conversation); err != nil { + t.Fatalf("Create(%s) error = %v", conversation.Conversation.ID, err) + } + } + + if _, err := store.Get(ctx, "conv_1"); !errors.Is(err, ErrNotFound) { + t.Fatalf("Get(conv_1) error = %v, want ErrNotFound", err) + } + for _, id := range []string{"conv_2", "conv_3"} { + if _, err := store.Get(ctx, id); err != nil { + t.Fatalf("Get(%s) error = %v", id, err) + } + } +} + +func TestMemoryStoreDefaultRetentionIsBounded(t *testing.T) { + store := NewMemoryStore() + + if store.ttl != DefaultMemoryStoreTTL { + t.Fatalf("ttl = %s, want %s", store.ttl, DefaultMemoryStoreTTL) + } + if store.maxEntries != DefaultMemoryStoreMaxEntries { + t.Fatalf("maxEntries = %d, want %d", store.maxEntries, DefaultMemoryStoreMaxEntries) + } +} + +func TestMemoryStoreGetReturnsIsolatedCopy(t *testing.T) { + ctx := context.Background() + store := NewMemoryStore() + if err := store.Create(ctx, storedConversation("conv_iso", time.Time{})); err != nil { + t.Fatalf("Create() error = %v", err) + } + + first, err := store.Get(ctx, "conv_iso") + if err != nil { + t.Fatalf("Get() error = %v", err) + } + first.Conversation.Metadata["mutated"] = "true" + + second, err := store.Get(ctx, "conv_iso") + if err != nil { + t.Fatalf("Get() error = %v", err) + } + if _, mutated := second.Conversation.Metadata["mutated"]; mutated { + t.Fatal("stored conversation mutated through returned copy") + } +} diff --git a/internal/core/conversations.go b/internal/core/conversations.go new file mode 100644 index 00000000..c2c13c11 --- /dev/null +++ b/internal/core/conversations.go @@ -0,0 +1,110 @@ +package core + +import ( + "bytes" + "encoding/json" + "fmt" + "unicode/utf8" +) + +// ConversationObject is the value of the "object" field on a conversation. +const ConversationObject = "conversation" + +// ConversationDeletedObject is the value of the "object" field returned by +// DELETE /v1/conversations/{id}. +const ConversationDeletedObject = "conversation.deleted" + +// Conversation limits mirror the OpenAI Conversations API so the gateway keeps +// an OpenAI-compatible public contract. +const ( + // MaxConversationInitialItems caps the items array accepted by + // POST /v1/conversations. + MaxConversationInitialItems = 20 + + maxConversationMetadataPairs = 16 + maxConversationMetadataKeyLength = 64 + maxConversationMetadataValueLength = 512 +) + +// Conversation is the OpenAI-compatible conversation resource returned by the +// /v1/conversations endpoints. +type Conversation struct { + ID string `json:"id"` + Object string `json:"object"` // "conversation" + CreatedAt int64 `json:"created_at"` + Metadata map[string]string `json:"metadata"` +} + +// ConversationDeleteResponse is returned by DELETE /v1/conversations/{id}. +type ConversationDeleteResponse struct { + ID string `json:"id"` + Object string `json:"object"` // "conversation.deleted" + Deleted bool `json:"deleted"` +} + +// ConversationCreateRequest is the accepted body for POST /v1/conversations. +// Items are stored as opaque JSON so the gateway accepts any item shape the +// client sends without constraining future item-list support. +type ConversationCreateRequest struct { + Items []json.RawMessage `json:"items,omitempty"` + Metadata map[string]string `json:"metadata,omitempty"` +} + +// ConversationUpdateRequest is the accepted body for POST /v1/conversations/{id}. +// Metadata is a pointer so the handler can tell an absent field apart from an +// explicit empty object: OpenAI requires metadata on update. +type ConversationUpdateRequest struct { + Metadata *map[string]string `json:"metadata"` +} + +// DecodeConversationCreateRequest parses a conversation create body. An empty +// body is treated as an empty request (a conversation with no items/metadata). +func DecodeConversationCreateRequest(data []byte) (*ConversationCreateRequest, error) { + req := &ConversationCreateRequest{} + if len(bytes.TrimSpace(data)) == 0 { + return req, nil + } + if err := json.Unmarshal(data, req); err != nil { + return nil, err + } + return req, nil +} + +// DecodeConversationUpdateRequest parses a conversation update body. An empty +// body decodes to a request with no metadata, which the handler rejects. +func DecodeConversationUpdateRequest(data []byte) (*ConversationUpdateRequest, error) { + req := &ConversationUpdateRequest{} + if len(bytes.TrimSpace(data)) == 0 { + return req, nil + } + if err := json.Unmarshal(data, req); err != nil { + return nil, err + } + return req, nil +} + +// ValidateConversationMetadata enforces the OpenAI metadata limits (at most 16 +// pairs, keys up to 64 characters, values up to 512 characters). It returns nil +// when the metadata is acceptable. +func ValidateConversationMetadata(metadata map[string]string) *GatewayError { + if len(metadata) > maxConversationMetadataPairs { + return NewInvalidRequestError( + fmt.Sprintf("metadata supports at most %d key-value pairs", maxConversationMetadataPairs), nil, + ).WithParam("metadata") + } + for key, value := range metadata { + // Limits are character (rune) counts, not byte lengths, so multi-byte + // keys and values are not rejected prematurely. + if utf8.RuneCountInString(key) > maxConversationMetadataKeyLength { + return NewInvalidRequestError( + fmt.Sprintf("metadata keys support at most %d characters", maxConversationMetadataKeyLength), nil, + ).WithParam("metadata") + } + if utf8.RuneCountInString(value) > maxConversationMetadataValueLength { + return NewInvalidRequestError( + fmt.Sprintf("metadata values support at most %d characters", maxConversationMetadataValueLength), nil, + ).WithParam("metadata") + } + } + return nil +} diff --git a/internal/core/conversations_test.go b/internal/core/conversations_test.go new file mode 100644 index 00000000..5a59e680 --- /dev/null +++ b/internal/core/conversations_test.go @@ -0,0 +1,117 @@ +package core + +import ( + "strings" + "testing" +) + +func TestDecodeConversationCreateRequest(t *testing.T) { + t.Run("empty body", func(t *testing.T) { + req, err := DecodeConversationCreateRequest(nil) + if err != nil { + t.Fatalf("DecodeConversationCreateRequest() error = %v", err) + } + if req == nil || len(req.Items) != 0 || len(req.Metadata) != 0 { + t.Fatalf("got %+v, want empty request", req) + } + }) + + t.Run("items and metadata", func(t *testing.T) { + req, err := DecodeConversationCreateRequest([]byte(`{"items":[{"type":"message"}],"metadata":{"k":"v"}}`)) + if err != nil { + t.Fatalf("DecodeConversationCreateRequest() error = %v", err) + } + if len(req.Items) != 1 { + t.Fatalf("items = %d, want 1", len(req.Items)) + } + if req.Metadata["k"] != "v" { + t.Fatalf("metadata[k] = %q, want v", req.Metadata["k"]) + } + }) + + t.Run("invalid json", func(t *testing.T) { + if _, err := DecodeConversationCreateRequest([]byte(`{`)); err == nil { + t.Fatal("DecodeConversationCreateRequest() error = nil, want error") + } + }) +} + +func TestDecodeConversationUpdateRequest(t *testing.T) { + t.Run("absent metadata", func(t *testing.T) { + req, err := DecodeConversationUpdateRequest([]byte(`{}`)) + if err != nil { + t.Fatalf("DecodeConversationUpdateRequest() error = %v", err) + } + if req.Metadata != nil { + t.Fatalf("metadata = %v, want nil", req.Metadata) + } + }) + + t.Run("explicit empty metadata", func(t *testing.T) { + req, err := DecodeConversationUpdateRequest([]byte(`{"metadata":{}}`)) + if err != nil { + t.Fatalf("DecodeConversationUpdateRequest() error = %v", err) + } + if req.Metadata == nil { + t.Fatal("metadata = nil, want non-nil empty map") + } + if len(*req.Metadata) != 0 { + t.Fatalf("metadata = %v, want empty", *req.Metadata) + } + }) +} + +func TestValidateConversationMetadata(t *testing.T) { + t.Run("nil is valid", func(t *testing.T) { + if err := ValidateConversationMetadata(nil); err != nil { + t.Fatalf("ValidateConversationMetadata(nil) = %v, want nil", err) + } + }) + + t.Run("too many pairs", func(t *testing.T) { + metadata := make(map[string]string, maxConversationMetadataPairs+1) + for i := 0; i <= maxConversationMetadataPairs; i++ { + metadata[string(rune('a'+i))] = "v" + } + err := ValidateConversationMetadata(metadata) + if err == nil { + t.Fatal("ValidateConversationMetadata() = nil, want error") + } + if err.Param == nil || *err.Param != "metadata" { + t.Fatalf("param = %v, want metadata", err.Param) + } + }) + + t.Run("key too long", func(t *testing.T) { + err := ValidateConversationMetadata(map[string]string{ + strings.Repeat("k", maxConversationMetadataKeyLength+1): "v", + }) + if err == nil { + t.Fatal("ValidateConversationMetadata() = nil, want error") + } + }) + + t.Run("value too long", func(t *testing.T) { + err := ValidateConversationMetadata(map[string]string{ + "k": strings.Repeat("v", maxConversationMetadataValueLength+1), + }) + if err == nil { + t.Fatal("ValidateConversationMetadata() = nil, want error") + } + }) + + t.Run("multi-byte runes counted as characters not bytes", func(t *testing.T) { + // "é" is one rune but two UTF-8 bytes: a key/value at the rune limit + // stays valid even though its byte length exceeds the limit. + key := strings.Repeat("é", maxConversationMetadataKeyLength) + value := strings.Repeat("é", maxConversationMetadataValueLength) + if err := ValidateConversationMetadata(map[string]string{key: value}); err != nil { + t.Fatalf("ValidateConversationMetadata() = %v, want nil", err) + } + + tooLongKey := strings.Repeat("é", maxConversationMetadataKeyLength+1) + if err := ValidateConversationMetadata(map[string]string{tooLongKey: "v"}); err == nil { + t.Fatal("ValidateConversationMetadata() = nil, want error for over-limit rune count") + } + }) +} diff --git a/internal/core/endpoints.go b/internal/core/endpoints.go index 8b9b0cf5..a62d396e 100644 --- a/internal/core/endpoints.go +++ b/internal/core/endpoints.go @@ -21,6 +21,7 @@ type Operation string const ( OperationChatCompletions Operation = "chat_completions" OperationResponses Operation = "responses" + OperationConversations Operation = "conversations" OperationEmbeddings Operation = "embeddings" OperationBatches Operation = "batches" OperationFiles Operation = "files" @@ -68,6 +69,17 @@ func describeEndpointPath(path string) EndpointDescriptor { Dialect: "openai_compat", Operation: OperationResponses, } + case path == "/v1/conversations" || strings.HasPrefix(path, "/v1/conversations/"): + // Conversations are a gateway-managed resource store (no provider + // call), but they are an ingress-managed /v1 endpoint and are + // classified as a model interaction path so they appear in request + // and audit logs alongside the other /v1 endpoints. + return EndpointDescriptor{ + ModelInteraction: true, + IngressManaged: true, + Dialect: "openai_compat", + Operation: OperationConversations, + } case path == "/v1/embeddings": return EndpointDescriptor{ ModelInteraction: true, @@ -123,6 +135,12 @@ func bodyModeForEndpoint(method, path string, operation Operation) BodyMode { return BodyModeJSON } return BodyModeNone + case OperationConversations: + // POST creates (/v1/conversations) or updates (/v1/conversations/{id}). + if method == http.MethodPost { + return BodyModeJSON + } + return BodyModeNone case OperationBatches: switch method { case http.MethodPost: diff --git a/internal/core/endpoints_test.go b/internal/core/endpoints_test.go index e27a4cc8..a05c5cb1 100644 --- a/internal/core/endpoints_test.go +++ b/internal/core/endpoints_test.go @@ -18,6 +18,8 @@ func TestDescribeEndpointPath(t *testing.T) { {path: "/v1/chat/completions/", managed: true, dialect: "openai_compat", operation: OperationChatCompletions, bodyMode: BodyModeJSON, interaction: true}, {path: "/v1/responses/resp_1", managed: true, dialect: "openai_compat", operation: OperationResponses, bodyMode: BodyModeNone, interaction: true}, {path: "/v1/responses/resp_1/input_items", managed: true, dialect: "openai_compat", operation: OperationResponses, bodyMode: BodyModeNone, interaction: true}, + {path: "/v1/conversations", managed: true, dialect: "openai_compat", operation: OperationConversations, bodyMode: BodyModeNone, interaction: true}, + {path: "/v1/conversations/conv_1", managed: true, dialect: "openai_compat", operation: OperationConversations, bodyMode: BodyModeNone, interaction: true}, {path: "/v1/batches", managed: true, dialect: "openai_compat", operation: OperationBatches, bodyMode: BodyModeNone, interaction: true}, {path: "/v1/embeddings/", managed: true, dialect: "openai_compat", operation: OperationEmbeddings, bodyMode: BodyModeJSON, interaction: true}, {path: "/v1/files/file_1", managed: true, dialect: "openai_compat", operation: OperationFiles, bodyMode: BodyModeNone, interaction: true}, @@ -63,6 +65,10 @@ func TestDescribeEndpoint_UsesMethodForBodyMode(t *testing.T) { {method: http.MethodDelete, path: "/v1/responses/resp_1", bodyMode: BodyModeNone}, {method: http.MethodPost, path: "/v1/responses/input_tokens", bodyMode: BodyModeJSON}, {method: http.MethodPost, path: "/v1/responses/compact", bodyMode: BodyModeJSON}, + {method: http.MethodPost, path: "/v1/conversations", bodyMode: BodyModeJSON}, + {method: http.MethodPost, path: "/v1/conversations/conv_1", bodyMode: BodyModeJSON}, + {method: http.MethodGet, path: "/v1/conversations/conv_1", bodyMode: BodyModeNone}, + {method: http.MethodDelete, path: "/v1/conversations/conv_1", bodyMode: BodyModeNone}, {method: http.MethodPost, path: "/v1/files", bodyMode: BodyModeMultipart}, {method: http.MethodPost, path: "/v1/files/", bodyMode: BodyModeMultipart}, {method: http.MethodGet, path: "/v1/files/file_1", bodyMode: BodyModeNone}, diff --git a/internal/server/conversation_handlers.go b/internal/server/conversation_handlers.go new file mode 100644 index 00000000..fe6fc223 --- /dev/null +++ b/internal/server/conversation_handlers.go @@ -0,0 +1,77 @@ +package server + +import ( + "github.com/labstack/echo/v5" +) + +// CreateConversation handles POST /v1/conversations. +// +// Conversations are a gateway-managed resource: GoModel generates the +// conversation id and stores the conversation locally, so the endpoint behaves +// identically regardless of which provider serves model traffic. +// +// @Summary Create a conversation +// @Tags conversations +// @Accept json +// @Produce json +// @Security BearerAuth +// @Param request body core.ConversationCreateRequest false "Conversation create request" +// @Success 200 {object} core.Conversation +// @Failure 400 {object} core.OpenAIErrorEnvelope +// @Failure 401 {object} core.OpenAIErrorEnvelope +// @Failure 500 {object} core.OpenAIErrorEnvelope +// @Router /v1/conversations [post] +func (h *Handler) CreateConversation(c *echo.Context) error { + return h.conversations().CreateConversation(c) +} + +// GetConversation handles GET /v1/conversations/{id}. +// +// @Summary Get a conversation +// @Tags conversations +// @Produce json +// @Security BearerAuth +// @Param id path string true "Conversation ID" +// @Success 200 {object} core.Conversation +// @Failure 400 {object} core.OpenAIErrorEnvelope +// @Failure 401 {object} core.OpenAIErrorEnvelope +// @Failure 404 {object} core.OpenAIErrorEnvelope +// @Router /v1/conversations/{id} [get] +func (h *Handler) GetConversation(c *echo.Context) error { + return h.conversations().GetConversation(c) +} + +// UpdateConversation handles POST /v1/conversations/{id}. +// +// @Summary Update a conversation +// @Description Replaces the conversation metadata in full. +// @Tags conversations +// @Accept json +// @Produce json +// @Security BearerAuth +// @Param id path string true "Conversation ID" +// @Param request body core.ConversationUpdateRequest true "Conversation update request" +// @Success 200 {object} core.Conversation +// @Failure 400 {object} core.OpenAIErrorEnvelope +// @Failure 401 {object} core.OpenAIErrorEnvelope +// @Failure 404 {object} core.OpenAIErrorEnvelope +// @Router /v1/conversations/{id} [post] +func (h *Handler) UpdateConversation(c *echo.Context) error { + return h.conversations().UpdateConversation(c) +} + +// DeleteConversation handles DELETE /v1/conversations/{id}. +// +// @Summary Delete a conversation +// @Tags conversations +// @Produce json +// @Security BearerAuth +// @Param id path string true "Conversation ID" +// @Success 200 {object} core.ConversationDeleteResponse +// @Failure 400 {object} core.OpenAIErrorEnvelope +// @Failure 401 {object} core.OpenAIErrorEnvelope +// @Failure 404 {object} core.OpenAIErrorEnvelope +// @Router /v1/conversations/{id} [delete] +func (h *Handler) DeleteConversation(c *echo.Context) error { + return h.conversations().DeleteConversation(c) +} diff --git a/internal/server/conversation_handlers_test.go b/internal/server/conversation_handlers_test.go new file mode 100644 index 00000000..fdc5cf92 --- /dev/null +++ b/internal/server/conversation_handlers_test.go @@ -0,0 +1,262 @@ +package server + +import ( + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "gomodel/internal/core" +) + +func createConversation(t *testing.T, srv *Server, body string) core.Conversation { + t.Helper() + req := httptest.NewRequest(http.MethodPost, "/v1/conversations", strings.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + rec := httptest.NewRecorder() + srv.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("create status = %d, want 200 (%s)", rec.Code, rec.Body.String()) + } + var conversation core.Conversation + if err := json.Unmarshal(rec.Body.Bytes(), &conversation); err != nil { + t.Fatalf("decode conversation: %v", err) + } + return conversation +} + +func TestConversationCreateReturnsOpenAICompatibleObject(t *testing.T) { + srv := New(&mockProvider{}, nil) + + conversation := createConversation(t, srv, `{"metadata":{"topic":"demo"}}`) + + if !strings.HasPrefix(conversation.ID, "conv_") { + t.Fatalf("id = %q, want conv_ prefix", conversation.ID) + } + if conversation.Object != "conversation" { + t.Fatalf("object = %q, want conversation", conversation.Object) + } + if conversation.CreatedAt <= 0 { + t.Fatalf("created_at = %d, want positive", conversation.CreatedAt) + } + if conversation.Metadata["topic"] != "demo" { + t.Fatalf("metadata[topic] = %q, want demo", conversation.Metadata["topic"]) + } +} + +func TestConversationCreateEmptyBodyYieldsEmptyMetadataObject(t *testing.T) { + srv := New(&mockProvider{}, nil) + + req := httptest.NewRequest(http.MethodPost, "/v1/conversations", nil) + rec := httptest.NewRecorder() + srv.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("create status = %d, want 200 (%s)", rec.Code, rec.Body.String()) + } + var conversation core.Conversation + if err := json.Unmarshal(rec.Body.Bytes(), &conversation); err != nil { + t.Fatalf("decode conversation: %v", err) + } + // metadata must be an empty object rather than null, matching OpenAI. + if conversation.Metadata == nil || len(conversation.Metadata) != 0 { + t.Fatalf("metadata = %#v, want empty object", conversation.Metadata) + } +} + +func TestConversationCreateAcceptsItems(t *testing.T) { + srv := New(&mockProvider{}, nil) + + conversation := createConversation(t, srv, + `{"items":[{"type":"message","role":"user","content":"hello"}]}`) + if conversation.ID == "" { + t.Fatal("conversation id is empty") + } +} + +func TestConversationGetRoundTrip(t *testing.T) { + srv := New(&mockProvider{}, nil) + created := createConversation(t, srv, `{"metadata":{"k":"v"}}`) + + req := httptest.NewRequest(http.MethodGet, "/v1/conversations/"+created.ID, nil) + rec := httptest.NewRecorder() + srv.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("get status = %d, want 200 (%s)", rec.Code, rec.Body.String()) + } + var got core.Conversation + if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { + t.Fatalf("decode conversation: %v", err) + } + if got.ID != created.ID || got.Metadata["k"] != "v" { + t.Fatalf("get conversation = %+v, want id %s metadata k=v", got, created.ID) + } +} + +func TestConversationUpdateReplacesMetadata(t *testing.T) { + srv := New(&mockProvider{}, nil) + created := createConversation(t, srv, `{"metadata":{"old":"value","keep":"gone"}}`) + + req := httptest.NewRequest(http.MethodPost, "/v1/conversations/"+created.ID, + strings.NewReader(`{"metadata":{"new":"value"}}`)) + req.Header.Set("Content-Type", "application/json") + rec := httptest.NewRecorder() + srv.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("update status = %d, want 200 (%s)", rec.Code, rec.Body.String()) + } + var updated core.Conversation + if err := json.Unmarshal(rec.Body.Bytes(), &updated); err != nil { + t.Fatalf("decode conversation: %v", err) + } + if updated.Metadata["new"] != "value" { + t.Fatalf("metadata[new] = %q, want value", updated.Metadata["new"]) + } + if _, ok := updated.Metadata["old"]; ok { + t.Fatal("metadata still carries replaced key 'old'") + } +} + +func TestConversationDeleteRemovesConversation(t *testing.T) { + srv := New(&mockProvider{}, nil) + created := createConversation(t, srv, `{}`) + + delReq := httptest.NewRequest(http.MethodDelete, "/v1/conversations/"+created.ID, nil) + delRec := httptest.NewRecorder() + srv.ServeHTTP(delRec, delReq) + if delRec.Code != http.StatusOK { + t.Fatalf("delete status = %d, want 200 (%s)", delRec.Code, delRec.Body.String()) + } + var deleted core.ConversationDeleteResponse + if err := json.Unmarshal(delRec.Body.Bytes(), &deleted); err != nil { + t.Fatalf("decode delete response: %v", err) + } + if deleted.ID != created.ID || deleted.Object != "conversation.deleted" || !deleted.Deleted { + t.Fatalf("delete response = %+v, want deleted %s", deleted, created.ID) + } + + getReq := httptest.NewRequest(http.MethodGet, "/v1/conversations/"+created.ID, nil) + getRec := httptest.NewRecorder() + srv.ServeHTTP(getRec, getReq) + if getRec.Code != http.StatusNotFound { + t.Fatalf("get after delete status = %d, want 404 (%s)", getRec.Code, getRec.Body.String()) + } +} + +// TestConversationEndpointErrors covers the validation and not-found error +// paths. Each case is independent of stored state: update metadata validation +// runs before the conversation is loaded, so a missing id still exercises it. +func TestConversationEndpointErrors(t *testing.T) { + bigItems := make([]string, core.MaxConversationInitialItems+1) + for i := range bigItems { + bigItems[i] = `{"type":"message","role":"user","content":"x"}` + } + bigMetadata := make([]string, 17) + for i := range bigMetadata { + bigMetadata[i] = fmt.Sprintf(`"key%d":"value"`, i) + } + + tests := []struct { + name string + method string + path string + body string + wantStatus int + wantErrorType core.ErrorType + wantErrorParam string + }{ + { + name: "get missing conversation", + method: http.MethodGet, + path: "/v1/conversations/conv_missing", + wantStatus: http.StatusNotFound, + wantErrorType: core.ErrorTypeNotFound, + }, + { + name: "update missing conversation", + method: http.MethodPost, + path: "/v1/conversations/conv_missing", + body: `{"metadata":{}}`, + wantStatus: http.StatusNotFound, + wantErrorType: core.ErrorTypeNotFound, + }, + { + name: "delete missing conversation", + method: http.MethodDelete, + path: "/v1/conversations/conv_missing", + wantStatus: http.StatusNotFound, + wantErrorType: core.ErrorTypeNotFound, + }, + { + name: "update without metadata", + method: http.MethodPost, + path: "/v1/conversations/conv_missing", + body: `{}`, + wantStatus: http.StatusBadRequest, + wantErrorType: core.ErrorTypeInvalidRequest, + wantErrorParam: "metadata", + }, + { + name: "create with too many items", + method: http.MethodPost, + path: "/v1/conversations", + body: fmt.Sprintf(`{"items":[%s]}`, strings.Join(bigItems, ",")), + wantStatus: http.StatusBadRequest, + wantErrorType: core.ErrorTypeInvalidRequest, + wantErrorParam: "items", + }, + { + name: "create with too much metadata", + method: http.MethodPost, + path: "/v1/conversations", + body: fmt.Sprintf(`{"metadata":{%s}}`, strings.Join(bigMetadata, ",")), + wantStatus: http.StatusBadRequest, + wantErrorType: core.ErrorTypeInvalidRequest, + wantErrorParam: "metadata", + }, + { + name: "create with invalid json", + method: http.MethodPost, + path: "/v1/conversations", + body: `{`, + wantStatus: http.StatusBadRequest, + wantErrorType: core.ErrorTypeInvalidRequest, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + srv := New(&mockProvider{}, nil) + + var body io.Reader + if tt.body != "" { + body = strings.NewReader(tt.body) + } + req := httptest.NewRequest(tt.method, tt.path, body) + if tt.body != "" { + req.Header.Set("Content-Type", "application/json") + } + rec := httptest.NewRecorder() + srv.ServeHTTP(rec, req) + + if rec.Code != tt.wantStatus { + t.Fatalf("status = %d, want %d (%s)", rec.Code, tt.wantStatus, rec.Body.String()) + } + + var envelope core.OpenAIErrorEnvelope + if err := json.Unmarshal(rec.Body.Bytes(), &envelope); err != nil { + t.Fatalf("decode error envelope: %v", err) + } + if tt.wantErrorType != "" && envelope.Error.Type != tt.wantErrorType { + t.Fatalf("error type = %q, want %q", envelope.Error.Type, tt.wantErrorType) + } + if tt.wantErrorParam != "" { + if envelope.Error.Param == nil || *envelope.Error.Param != tt.wantErrorParam { + t.Fatalf("error param = %v, want %q", envelope.Error.Param, tt.wantErrorParam) + } + } + }) + } +} diff --git a/internal/server/handlers.go b/internal/server/handlers.go index 5044c2ed..f97a05e0 100644 --- a/internal/server/handlers.go +++ b/internal/server/handlers.go @@ -9,6 +9,7 @@ import ( "gomodel/internal/auditlog" batchstore "gomodel/internal/batch" + "gomodel/internal/conversationstore" "gomodel/internal/core" "gomodel/internal/filestore" "gomodel/internal/responsecache" @@ -35,6 +36,7 @@ type Handler struct { fileStore filestore.Store responseStore responsestore.Store responseStoreMu sync.RWMutex + conversationStore conversationstore.Store normalizePassthroughV1Prefix bool enabledPassthroughProviders map[string]struct{} responseCache *responsecache.ResponseCacheMiddleware @@ -99,6 +101,10 @@ func newHandlerWithAuthorizer( responsestore.WithTTL(responsestore.DefaultMemoryStoreTTL), responsestore.WithMaxEntries(responsestore.DefaultMemoryStoreMaxEntries), ), + conversationStore: conversationstore.NewMemoryStore( + conversationstore.WithTTL(conversationstore.DefaultMemoryStoreTTL), + conversationstore.WithMaxEntries(conversationstore.DefaultMemoryStoreMaxEntries), + ), normalizePassthroughV1Prefix: true, enabledPassthroughProviders: normalizeEnabledPassthroughProviders(defaultEnabledPassthroughProviders), } @@ -136,6 +142,16 @@ func (h *Handler) SetResponseStore(store responsestore.Store) { } } +// SetConversationStore replaces the conversation store used by the +// Conversations lifecycle endpoints. +// nil is ignored to keep an always-available fallback memory store. +func (h *Handler) SetConversationStore(store conversationstore.Store) { + if store == nil { + return + } + h.conversationStore = store +} + func (h *Handler) translatedInference() *translatedInferenceService { h.translatedSvcOnce.Do(func() { s := &translatedInferenceService{ @@ -196,6 +212,10 @@ func (h *Handler) nativeResponses() *nativeResponseService { } } +func (h *Handler) conversations() *conversationService { + return &conversationService{conversationStore: h.conversationStore} +} + func (h *Handler) currentResponseStore() responsestore.Store { h.responseStoreMu.RLock() defer h.responseStoreMu.RUnlock() diff --git a/internal/server/http.go b/internal/server/http.go index 703439b2..7ba1a63c 100644 --- a/internal/server/http.go +++ b/internal/server/http.go @@ -21,6 +21,7 @@ import ( "gomodel/internal/admin/dashboard" "gomodel/internal/auditlog" batchstore "gomodel/internal/batch" + "gomodel/internal/conversationstore" "gomodel/internal/core" "gomodel/internal/filestore" "gomodel/internal/responsecache" @@ -34,6 +35,7 @@ type Server struct { handler *Handler responseCacheMiddleware *responsecache.ResponseCacheMiddleware responseStore responsestore.Store + conversationStore conversationstore.Store } const ( @@ -67,6 +69,7 @@ type Config struct { BatchStore batchstore.Store // Optional: Batch lifecycle persistence store FileStore filestore.Store // Optional: File provider mapping persistence store ResponseStore responsestore.Store // Optional: Responses lifecycle persistence store + ConversationStore conversationstore.Store // Optional: Conversations lifecycle persistence store LogOnlyModelInteractions bool // Only log AI model endpoints (default: true) DisablePassthroughRoutes bool // Disable /p/{provider}/{endpoint} route registration EnabledPassthroughProviders []string // Provider types enabled on /p/{provider}/... passthrough routes @@ -147,6 +150,9 @@ func New(provider core.RoutableProvider, cfg *Config) *Server { if cfg != nil && cfg.ResponseStore != nil { handler.SetResponseStore(cfg.ResponseStore) } + if cfg != nil && cfg.ConversationStore != nil { + handler.SetConversationStore(cfg.ConversationStore) + } // Build list of paths that skip authentication authSkipPaths := []string{"/health"} @@ -310,6 +316,10 @@ func New(provider core.RoutableProvider, cfg *Config) *Server { e.GET("/v1/responses/:id", handler.GetResponse) e.DELETE("/v1/responses/:id", handler.DeleteResponse) e.POST("/v1/responses", handler.Responses) + e.POST("/v1/conversations", handler.CreateConversation) + e.GET("/v1/conversations/:id", handler.GetConversation) + e.POST("/v1/conversations/:id", handler.UpdateConversation) + e.DELETE("/v1/conversations/:id", handler.DeleteConversation) e.POST("/v1/embeddings", handler.Embeddings) e.POST("/v1/files", handler.CreateFile) e.GET("/v1/files", handler.ListFiles) @@ -352,6 +362,7 @@ func New(provider core.RoutableProvider, cfg *Config) *Server { handler: handler, responseCacheMiddleware: rcm, responseStore: handler.currentResponseStore(), + conversationStore: handler.conversationStore, } } @@ -395,7 +406,8 @@ func (s *Server) StartWithListener(ctx context.Context, listener net.Listener) e // Shutdown releases server resources. The HTTP server itself is stopped by // cancelling the context passed to Start; this method drains any in-flight -// response cache writes, closes the cache store, and closes the response store. +// response cache writes, closes the cache store, and closes the response and +// conversation stores. func (s *Server) Shutdown(_ context.Context) error { var firstErr error if s.responseCacheMiddleware != nil { @@ -412,6 +424,15 @@ func (s *Server) Shutdown(_ context.Context) error { } } } + if s.conversationStore != nil { + if err := s.conversationStore.Close(); err != nil { + if firstErr == nil { + firstErr = err + } else { + slog.Warn("conversation store close failed during shutdown", "error", err) + } + } + } return firstErr } diff --git a/internal/server/native_conversation_service.go b/internal/server/native_conversation_service.go new file mode 100644 index 00000000..d6814109 --- /dev/null +++ b/internal/server/native_conversation_service.go @@ -0,0 +1,203 @@ +package server + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "strings" + "time" + + "github.com/google/uuid" + "github.com/labstack/echo/v5" + + "gomodel/internal/auditlog" + "gomodel/internal/conversationstore" + "gomodel/internal/core" +) + +// conversationService owns the gateway-managed Conversations lifecycle endpoints. +// Conversations are stored locally rather than proxied: the resource is +// OpenAI-specific, so a gateway-owned store keeps /v1/conversations available +// uniformly regardless of which provider routes model traffic. +type conversationService struct { + conversationStore conversationstore.Store +} + +// CreateConversation handles POST /v1/conversations. +func (s *conversationService) CreateConversation(c *echo.Context) error { + ctx, requestID := requestContextWithRequestID(c.Request()) + auditConversationEntry(c) + + body, err := requestBodyBytes(c) + if err != nil { + return handleError(c, core.NewInvalidRequestError("invalid request body: "+err.Error(), err)) + } + req, err := core.DecodeConversationCreateRequest(body) + if err != nil { + return handleError(c, core.NewInvalidRequestError("invalid request body: "+err.Error(), err)) + } + if len(req.Items) > core.MaxConversationInitialItems { + return handleError(c, core.NewInvalidRequestError( + fmt.Sprintf("items supports at most %d entries", core.MaxConversationInitialItems), nil, + ).WithParam("items")) + } + if verr := core.ValidateConversationMetadata(req.Metadata); verr != nil { + return handleError(c, verr) + } + + now := time.Now().UTC() + conversation := &core.Conversation{ + ID: generatedConversationID(), + Object: core.ConversationObject, + CreatedAt: now.Unix(), + Metadata: normalizedConversationMetadata(req.Metadata), + } + stored := &conversationstore.StoredConversation{ + Conversation: conversation, + Items: cloneRawConversationItems(req.Items), + UserPath: core.UserPathFromContext(ctx), + RequestID: requestID, + StoredAt: now, + } + if err := s.conversationStore.Create(ctx, stored); err != nil { + return handleError(c, core.NewProviderError("conversation_store", http.StatusInternalServerError, "failed to persist conversation", err)) + } + return c.JSON(http.StatusOK, conversation) +} + +// GetConversation handles GET /v1/conversations/{id}. +func (s *conversationService) GetConversation(c *echo.Context) error { + ctx, _ := requestContextWithRequestID(c.Request()) + auditConversationEntry(c) + + id, err := conversationIDFromRequest(c) + if err != nil { + return handleError(c, err) + } + stored, err := s.loadStoredConversation(ctx, id) + if err != nil { + return handleError(c, err) + } + return c.JSON(http.StatusOK, stored.Conversation) +} + +// UpdateConversation handles POST /v1/conversations/{id}. The metadata in the +// request replaces the conversation's metadata in full, matching OpenAI. +func (s *conversationService) UpdateConversation(c *echo.Context) error { + ctx, _ := requestContextWithRequestID(c.Request()) + auditConversationEntry(c) + + id, err := conversationIDFromRequest(c) + if err != nil { + return handleError(c, err) + } + body, err := requestBodyBytes(c) + if err != nil { + return handleError(c, core.NewInvalidRequestError("invalid request body: "+err.Error(), err)) + } + req, err := core.DecodeConversationUpdateRequest(body) + if err != nil { + return handleError(c, core.NewInvalidRequestError("invalid request body: "+err.Error(), err)) + } + if req.Metadata == nil { + return handleError(c, core.NewInvalidRequestError("metadata is required", nil).WithParam("metadata")) + } + if verr := core.ValidateConversationMetadata(*req.Metadata); verr != nil { + return handleError(c, verr) + } + + stored, err := s.loadStoredConversation(ctx, id) + if err != nil { + return handleError(c, err) + } + stored.Conversation.Metadata = normalizedConversationMetadata(*req.Metadata) + if err := s.conversationStore.Update(ctx, stored); err != nil { + if errors.Is(err, conversationstore.ErrNotFound) { + return handleError(c, conversationNotFound(id)) + } + return handleError(c, core.NewProviderError("conversation_store", http.StatusInternalServerError, "failed to update conversation", err)) + } + return c.JSON(http.StatusOK, stored.Conversation) +} + +// DeleteConversation handles DELETE /v1/conversations/{id}. +func (s *conversationService) DeleteConversation(c *echo.Context) error { + ctx, _ := requestContextWithRequestID(c.Request()) + auditConversationEntry(c) + + id, err := conversationIDFromRequest(c) + if err != nil { + return handleError(c, err) + } + if err := s.conversationStore.Delete(ctx, id); err != nil { + if errors.Is(err, conversationstore.ErrNotFound) { + return handleError(c, conversationNotFound(id)) + } + return handleError(c, core.NewProviderError("conversation_store", http.StatusInternalServerError, "failed to delete conversation", err)) + } + return c.JSON(http.StatusOK, &core.ConversationDeleteResponse{ + ID: id, + Object: core.ConversationDeletedObject, + Deleted: true, + }) +} + +func (s *conversationService) loadStoredConversation(ctx context.Context, id string) (*conversationstore.StoredConversation, error) { + if s.conversationStore == nil { + return nil, conversationNotFound(id) + } + stored, err := s.conversationStore.Get(ctx, id) + if err != nil { + if errors.Is(err, conversationstore.ErrNotFound) { + return nil, conversationNotFound(id) + } + return nil, core.NewProviderError("conversation_store", http.StatusInternalServerError, "failed to load conversation", err) + } + if stored == nil || stored.Conversation == nil { + return nil, core.NewProviderError("conversation_store", http.StatusInternalServerError, "stored conversation payload missing", nil) + } + return stored, nil +} + +func conversationIDFromRequest(c *echo.Context) (string, error) { + id := strings.TrimSpace(c.Param("id")) + if id == "" { + return "", core.NewInvalidRequestError("conversation id is required", nil) + } + return id, nil +} + +func conversationNotFound(id string) *core.GatewayError { + return core.NewNotFoundError("conversation not found: " + id) +} + +func generatedConversationID() string { + return "conv_" + strings.ReplaceAll(uuid.NewString(), "-", "") +} + +// normalizedConversationMetadata always returns a non-nil map so the serialized +// conversation carries an empty object rather than null, matching OpenAI. +func normalizedConversationMetadata(metadata map[string]string) map[string]string { + normalized := make(map[string]string, len(metadata)) + for key, value := range metadata { + normalized[key] = value + } + return normalized +} + +func cloneRawConversationItems(items []json.RawMessage) []json.RawMessage { + if len(items) == 0 { + return nil + } + cloned := make([]json.RawMessage, 0, len(items)) + for _, item := range items { + cloned = append(cloned, core.CloneRawJSON(item)) + } + return cloned +} + +func auditConversationEntry(c *echo.Context) { + auditlog.EnrichEntry(c, "conversation", "") +}