lisone/ds2api
1
0
Fork 0
mirror of https://github.com/CJackHwang/ds2api.git synced 2026-05-12 12:17:47 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: add unified response history session management across Claude, Gemini, and OpenAI API backends

Browse Source 
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
CJACK
2026-05-03 17:24:38 +08:00
parent 5e55cf36d8
commit c099a6f7bf
28 changed files with 776 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
internal/chathistory/store.go
View File

@@ -43,6 +43,7 @@ type Entry struct {
Status string `json:"status"`
CallerID string `json:"caller_id,omitempty"`
AccountID string `json:"account_id,omitempty"`
Surface string `json:"surface,omitempty"`
Model string `json:"model,omitempty"`
Stream bool `json:"stream"`
UserInput string `json:"user_input,omitempty"`
@@ -72,6 +73,7 @@ type SummaryEntry struct {
Status string `json:"status"`
CallerID string `json:"caller_id,omitempty"`
AccountID string `json:"account_id,omitempty"`
Surface string `json:"surface,omitempty"`
Model string `json:"model,omitempty"`
Stream bool `json:"stream"`
UserInput string `json:"user_input,omitempty"`
@@ -92,6 +94,7 @@ type File struct {
type StartParams struct {
CallerID string
AccountID string
Surface string
Model string
Stream bool
UserInput string
@@ -271,6 +274,7 @@ func (s *Store) Start(params StartParams) (Entry, error) {
Status: "streaming",
CallerID: strings.TrimSpace(params.CallerID),
AccountID: strings.TrimSpace(params.AccountID),
Surface: strings.TrimSpace(params.Surface),
Model: strings.TrimSpace(params.Model),
Stream: params.Stream,
UserInput: strings.TrimSpace(params.UserInput),
@@ -546,10 +550,13 @@ func (s *Store) rebuildIndexLocked() {
summaries = append(summaries, summaryFromEntry(item))
}
sort.Slice(summaries, func(i, j int) bool {
if summaries[i].UpdatedAt == summaries[j].UpdatedAt {
return summaries[i].CreatedAt > summaries[j].CreatedAt
if summaries[i].CreatedAt == summaries[j].CreatedAt {
if summaries[i].Revision == summaries[j].Revision {
return summaries[i].UpdatedAt > summaries[j].UpdatedAt
}
return summaries[i].Revision > summaries[j].Revision
}
return summaries[i].UpdatedAt > summaries[j].UpdatedAt
return summaries[i].CreatedAt > summaries[j].CreatedAt
})
if s.state.Limit < DisabledLimit || !isAllowedLimit(s.state.Limit) {
s.state.Limit = DefaultLimit
@@ -593,6 +600,7 @@ func summaryFromEntry(item Entry) SummaryEntry {
Status: item.Status,
CallerID: item.CallerID,
AccountID: item.AccountID,
Surface: item.Surface,
Model: item.Model,
Stream: item.Stream,
UserInput: item.UserInput,

31
internal/chathistory/store_test.go
View File

@@ -8,6 +8,7 @@ import (
"strings"
"sync"
"testing"
"time"
"unicode/utf8"
)
@@ -494,6 +495,36 @@ func TestStoreWritesOnlyChangedDetailFiles(t *testing.T) {
}
}
func TestStoreOrdersByCreationTimeNotStreamingUpdates(t *testing.T) {
path := filepath.Join(t.TempDir(), "chat_history.json")
store := New(path)
first, err := store.Start(StartParams{UserInput: "first"})
if err != nil {
t.Fatalf("start first failed: %v", err)
}
time.Sleep(time.Millisecond)
second, err := store.Start(StartParams{UserInput: "second"})
if err != nil {
t.Fatalf("start second failed: %v", err)
}
time.Sleep(time.Millisecond)
if _, err := store.Update(first.ID, UpdateParams{Status: "streaming", Content: "still running"}); err != nil {
t.Fatalf("update first failed: %v", err)
}
snapshot, err := store.Snapshot()
if err != nil {
t.Fatalf("snapshot failed: %v", err)
}
if len(snapshot.Items) != 2 {
t.Fatalf("expected two items, got %#v", snapshot.Items)
}
if snapshot.Items[0].ID != second.ID || snapshot.Items[1].ID != first.ID {
t.Fatalf("expected creation-time order to stay stable, got %#v", snapshot.Items)
}
}
func TestUpdatePreservesContentWhenNewContentIsEmpty(t *testing.T) {
path := filepath.Join(t.TempDir(), "chat_history.json")
store := New(path)

79
internal/httpapi/claude/current_input_file_test.go
View File

@@ -5,15 +5,25 @@ import (
"io"
"net/http"
"net/http/httptest"
"path/filepath"
"strings"
"testing"
"ds2api/internal/auth"
"ds2api/internal/chathistory"
dsclient "ds2api/internal/deepseek/client"
)
type claudeCurrentInputAuth struct{}
type claudeHistoryConfig struct {
aliases map[string]string
}
func (m claudeHistoryConfig) ModelAliases() map[string]string { return m.aliases }
func (claudeHistoryConfig) CurrentInputFileEnabled() bool { return false }
func (claudeHistoryConfig) CurrentInputFileMinChars() int { return 0 }
func (claudeCurrentInputAuth) Determine(*http.Request) (*auth.RequestAuth, error) {
return &auth.RequestAuth{
DeepSeekToken: "direct-token",
@@ -22,6 +32,50 @@ func (claudeCurrentInputAuth) Determine(*http.Request) (*auth.RequestAuth, error
}, nil
}
func TestClaudeDirectRecordsResponseHistory(t *testing.T) {
ds := &claudeCurrentInputDS{}
historyStore := chathistory.New(filepath.Join(t.TempDir(), "history.json"))
h := &Handler{
Store: claudeHistoryConfig{aliases: map[string]string{"claude-sonnet-4-6": "deepseek-v4-flash"}},
Auth: claudeCurrentInputAuth{},
DS: ds,
ChatHistory: historyStore,
}
reqBody := `{"model":"claude-sonnet-4-6","messages":[{"role":"user","content":"hello from claude"}],"max_tokens":1024}`
req := httptest.NewRequest(http.MethodPost, "/v1/messages", strings.NewReader(reqBody))
req.Header.Set("Content-Type", "application/json")
rec := httptest.NewRecorder()
h.Messages(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("expected 200, got %d body=%s", rec.Code, rec.Body.String())
}
snapshot, err := historyStore.Snapshot()
if err != nil {
t.Fatalf("snapshot history: %v", err)
}
if len(snapshot.Items) != 1 {
t.Fatalf("expected one history item, got %d", len(snapshot.Items))
}
item, err := historyStore.Get(snapshot.Items[0].ID)
if err != nil {
t.Fatalf("get history item: %v", err)
}
if item.Surface != "claude.messages" {
t.Fatalf("unexpected surface: %q", item.Surface)
}
if item.Model != "claude-sonnet-4-6" {
t.Fatalf("unexpected model: %q", item.Model)
}
if item.UserInput != "hello from claude" {
t.Fatalf("unexpected user input: %q", item.UserInput)
}
if item.Content != "ok" {
t.Fatalf("expected raw upstream content, got %q", item.Content)
}
}
func (claudeCurrentInputAuth) Release(*auth.RequestAuth) {}
type claudeCurrentInputDS struct {
@@ -53,10 +107,12 @@ func (d *claudeCurrentInputDS) CallCompletion(_ context.Context, _ *auth.Request
func TestClaudeDirectAppliesCurrentInputFile(t *testing.T) {
ds := &claudeCurrentInputDS{}
historyStore := chathistory.New(filepath.Join(t.TempDir(), "history.json"))
h := &Handler{
Store: mockClaudeConfig{aliases: map[string]string{"claude-sonnet-4-6": "deepseek-v4-flash"}},
Auth: claudeCurrentInputAuth{},
DS: ds,
Store: mockClaudeConfig{aliases: map[string]string{"claude-sonnet-4-6": "deepseek-v4-flash"}},
Auth: claudeCurrentInputAuth{},
DS: ds,
ChatHistory: historyStore,
}
reqBody := `{"model":"claude-sonnet-4-6","messages":[{"role":"user","content":"hello from claude"}],"max_tokens":1024}`
req := httptest.NewRequest(http.MethodPost, "/v1/messages", strings.NewReader(reqBody))
@@ -82,4 +138,21 @@ func TestClaudeDirectAppliesCurrentInputFile(t *testing.T) {
if !strings.Contains(prompt, "Continue from the latest state in the attached DS2API_HISTORY.txt context.") {
t.Fatalf("expected continuation prompt, got %q", prompt)
}
snapshot, err := historyStore.Snapshot()
if err != nil {
t.Fatalf("snapshot history: %v", err)
}
if len(snapshot.Items) != 1 {
t.Fatalf("expected one history item, got %d", len(snapshot.Items))
}
full, err := historyStore.Get(snapshot.Items[0].ID)
if err != nil {
t.Fatalf("get history item: %v", err)
}
if full.HistoryText != string(ds.uploads[0].Data) {
t.Fatalf("expected uploaded current input file to be persisted in history text")
}
if len(full.Messages) != 1 || !strings.Contains(full.Messages[0].Content, "Continue from the latest state in the attached DS2API_HISTORY.txt context.") {
t.Fatalf("expected persisted message to match upstream continuation prompt, got %#v", full.Messages)
}
}

58
internal/httpapi/claude/handler_messages.go
View File

@@ -2,6 +2,7 @@ package claude
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
@@ -15,8 +16,10 @@ import (
"ds2api/internal/completionruntime"
"ds2api/internal/config"
claudefmt "ds2api/internal/format/claude"
"ds2api/internal/httpapi/openai/history"
"ds2api/internal/httpapi/requestbody"
"ds2api/internal/promptcompat"
"ds2api/internal/responsehistory"
streamengine "ds2api/internal/stream"
"ds2api/internal/translatorcliproxy"
"ds2api/internal/util"
@@ -79,38 +82,71 @@ func (h *Handler) handleClaudeDirect(w http.ResponseWriter, r *http.Request) boo
return true
}
defer h.Auth.Release(a)
if norm.Standard.Stream {
h.handleClaudeDirectStream(w, r, a, norm.Standard)
stdReq, err := h.applyCurrentInputFile(r.Context(), a, norm.Standard)
if err != nil {
status, message := mapCurrentInputFileError(err)
writeClaudeError(w, status, message)
return true
}
result, outErr := completionruntime.ExecuteNonStreamWithRetry(r.Context(), h.DS, a, norm.Standard, completionruntime.Options{
historySession := responsehistory.Start(responsehistory.StartParams{
Store: h.ChatHistory,
Request: r,
Auth: a,
Surface: "claude.messages",
Standard: stdReq,
})
if stdReq.Stream {
h.handleClaudeDirectStream(w, r, a, stdReq, historySession)
return true
}
result, outErr := completionruntime.ExecuteNonStreamWithRetry(r.Context(), h.DS, a, stdReq, completionruntime.Options{
StripReferenceMarkers: stripReferenceMarkersEnabled(),
RetryEnabled: true,
CurrentInputFile: h.Store,
})
if outErr != nil {
if historySession != nil {
historySession.ErrorTurn(outErr.Status, outErr.Message, outErr.Code, result.Turn)
}
writeClaudeError(w, outErr.Status, outErr.Message)
return true
}
if historySession != nil {
historySession.SuccessTurn(http.StatusOK, result.Turn, responsehistory.GenericUsage(result.Turn))
}
writeJSON(w, http.StatusOK, claudefmt.BuildMessageResponseFromTurn(
fmt.Sprintf("msg_%d", time.Now().UnixNano()),
norm.Standard.ResponseModel,
stdReq.ResponseModel,
result.Turn,
exposeThinking,
))
return true
}
func (h *Handler) handleClaudeDirectStream(w http.ResponseWriter, r *http.Request, a *auth.RequestAuth, stdReq promptcompat.StandardRequest) {
func (h *Handler) applyCurrentInputFile(ctx context.Context, a *auth.RequestAuth, stdReq promptcompat.StandardRequest) (promptcompat.StandardRequest, error) {
if h == nil {
return stdReq, nil
}
return (history.Service{Store: h.Store, DS: h.DS}).ApplyCurrentInputFile(ctx, a, stdReq)
}
func mapCurrentInputFileError(err error) (int, string) {
return history.MapError(err)
}
func (h *Handler) handleClaudeDirectStream(w http.ResponseWriter, r *http.Request, a *auth.RequestAuth, stdReq promptcompat.StandardRequest, historySession *responsehistory.Session) {
start, outErr := completionruntime.StartCompletion(r.Context(), h.DS, a, stdReq, completionruntime.Options{
CurrentInputFile: h.Store,
})
if outErr != nil {
if historySession != nil {
historySession.Error(outErr.Status, outErr.Message, outErr.Code, "", "")
}
writeClaudeError(w, outErr.Status, outErr.Message)
return
}
streamReq := start.Request
h.handleClaudeStreamRealtime(w, r, start.Response, streamReq.ResponseModel, streamReq.Messages, streamReq.Thinking, streamReq.Search, streamReq.ToolNames, streamReq.ToolsRaw)
h.handleClaudeStreamRealtime(w, r, start.Response, streamReq.ResponseModel, streamReq.Messages, streamReq.Thinking, streamReq.Search, streamReq.ToolNames, streamReq.ToolsRaw, historySession)
}
func (h *Handler) proxyViaOpenAI(w http.ResponseWriter, r *http.Request, store ConfigReader) bool {
@@ -264,10 +300,17 @@ func stripClaudeThinkingBlocks(raw []byte) []byte {
return out
}
func (h *Handler) handleClaudeStreamRealtime(w http.ResponseWriter, r *http.Request, resp *http.Response, model string, messages []any, thinkingEnabled, searchEnabled bool, toolNames []string, toolsRaw any) {
func (h *Handler) handleClaudeStreamRealtime(w http.ResponseWriter, r *http.Request, resp *http.Response, model string, messages []any, thinkingEnabled, searchEnabled bool, toolNames []string, toolsRaw any, historySessions ...*responsehistory.Session) {
var historySession *responsehistory.Session
if len(historySessions) > 0 {
historySession = historySessions[0]
}
defer func() { _ = resp.Body.Close() }()
if resp.StatusCode != http.StatusOK {
body, _ := io.ReadAll(resp.Body)
if historySession != nil {
historySession.Error(resp.StatusCode, strings.TrimSpace(string(body)), "error", "", "")
}
writeClaudeError(w, http.StatusInternalServerError, string(body))
return
}
@@ -294,6 +337,7 @@ func (h *Handler) handleClaudeStreamRealtime(w http.ResponseWriter, r *http.Requ
toolNames,
toolsRaw,
buildClaudePromptTokenText(messages, thinkingEnabled),
historySession,
)
streamRuntime.sendMessageStart()

10
internal/httpapi/claude/handler_routes.go
View File

@@ -6,6 +6,7 @@ import (
"github.com/go-chi/chi/v5"
"ds2api/internal/chathistory"
"ds2api/internal/config"
dsprotocol "ds2api/internal/deepseek/protocol"
"ds2api/internal/textclean"
@@ -16,10 +17,11 @@ import (
var writeJSON = util.WriteJSON
type Handler struct {
Store ConfigReader
Auth AuthResolver
DS DeepSeekCaller
OpenAI OpenAIChatRunner
Store ConfigReader
Auth AuthResolver
DS DeepSeekCaller
OpenAI OpenAIChatRunner
ChatHistory *chathistory.Store
}
func stripReferenceMarkersEnabled() bool {

10
internal/httpapi/claude/stream_runtime_core.go
View File

@@ -6,6 +6,7 @@ import (
"strings"
"time"
"ds2api/internal/responsehistory"
"ds2api/internal/sse"
streamengine "ds2api/internal/stream"
"ds2api/internal/toolcall"
@@ -46,6 +47,7 @@ type claudeStreamRuntime struct {
textEmitted bool
ended bool
upstreamErr string
history *responsehistory.Session
}
func newClaudeStreamRuntime(
@@ -60,6 +62,7 @@ func newClaudeStreamRuntime(
toolNames []string,
toolsRaw any,
promptTokenText string,
history *responsehistory.Session,
) *claudeStreamRuntime {
return &claudeStreamRuntime{
w: w,
@@ -74,6 +77,7 @@ func newClaudeStreamRuntime(
toolNames: toolNames,
toolsRaw: toolsRaw,
promptTokenText: promptTokenText,
history: history,
messageID: fmt.Sprintf("msg_%d", time.Now().UnixNano()),
thinkingBlockIndex: -1,
textBlockIndex: -1,
@@ -232,5 +236,11 @@ func (s *claudeStreamRuntime) onParsed(parsed sse.LineResult) streamengine.Parse
}
}
if s.history != nil {
s.history.Progress(
responsehistory.ThinkingForArchive(s.rawThinking.String(), s.toolDetectionThinking.String(), s.thinking.String()),
responsehistory.TextForArchive(s.rawText.String(), s.text.String()),
)
}
return streamengine.ParsedDecision{ContentSeen: contentSeen}
}

16
internal/httpapi/claude/stream_runtime_finalize.go
View File

@@ -2,6 +2,7 @@ package claude
import (
"ds2api/internal/assistantturn"
"ds2api/internal/responsehistory"
"ds2api/internal/sse"
"ds2api/internal/toolcall"
"ds2api/internal/toolstream"
@@ -175,6 +176,15 @@ func (s *claudeStreamRuntime) finalize(stopReason string) {
if outcome.HasToolCalls {
stopReason = "tool_use"
}
if s.history != nil {
s.history.Success(
200,
responsehistory.ThinkingForArchive(turn.RawThinking, turn.DetectionThinking, turn.Thinking),
responsehistory.TextForArchive(turn.RawText, turn.Text),
stopReason,
responsehistory.GenericUsage(turn),
)
}
s.send("message_delta", map[string]any{
"type": "message_delta",
@@ -191,10 +201,16 @@ func (s *claudeStreamRuntime) finalize(stopReason string) {
func (s *claudeStreamRuntime) onFinalize(reason streamengine.StopReason, scannerErr error) {
if string(reason) == "upstream_error" {
if s.history != nil {
s.history.Error(500, s.upstreamErr, "upstream_error", responsehistory.ThinkingForArchive(s.rawThinking.String(), s.toolDetectionThinking.String(), s.thinking.String()), responsehistory.TextForArchive(s.rawText.String(), s.text.String()))
}
s.sendError(s.upstreamErr)
return
}
if scannerErr != nil {
if s.history != nil {
s.history.Error(500, scannerErr.Error(), "error", responsehistory.ThinkingForArchive(s.rawThinking.String(), s.toolDetectionThinking.String(), s.thinking.String()), responsehistory.TextForArchive(s.rawText.String(), s.text.String()))
}
s.sendError(scannerErr.Error())
return
}

42
internal/httpapi/gemini/handler_generate.go
View File

@@ -2,6 +2,7 @@ package gemini
import (
"bytes"
"context"
"encoding/json"
"errors"
"io"
@@ -14,8 +15,10 @@ import (
"ds2api/internal/assistantturn"
"ds2api/internal/auth"
"ds2api/internal/completionruntime"
"ds2api/internal/httpapi/openai/history"
"ds2api/internal/httpapi/requestbody"
"ds2api/internal/promptcompat"
"ds2api/internal/responsehistory"
"ds2api/internal/sse"
"ds2api/internal/toolcall"
"ds2api/internal/translatorcliproxy"
@@ -76,8 +79,21 @@ func (h *Handler) handleGeminiDirect(w http.ResponseWriter, r *http.Request, str
return true
}
defer h.Auth.Release(a)
stdReq, err = h.applyCurrentInputFile(r.Context(), a, stdReq)
if err != nil {
status, message := mapCurrentInputFileError(err)
writeGeminiError(w, status, message)
return true
}
historySession := responsehistory.Start(responsehistory.StartParams{
Store: h.ChatHistory,
Request: r,
Auth: a,
Surface: "gemini.generate_content",
Standard: stdReq,
})
if stream {
h.handleGeminiDirectStream(w, r, a, stdReq)
h.handleGeminiDirectStream(w, r, a, stdReq, historySession)
return true
}
result, outErr := completionruntime.ExecuteNonStreamWithRetry(r.Context(), h.DS, a, stdReq, completionruntime.Options{
@@ -86,23 +102,43 @@ func (h *Handler) handleGeminiDirect(w http.ResponseWriter, r *http.Request, str
CurrentInputFile: h.Store,
})
if outErr != nil {
if historySession != nil {
historySession.ErrorTurn(outErr.Status, outErr.Message, outErr.Code, result.Turn)
}
writeGeminiError(w, outErr.Status, outErr.Message)
return true
}
if historySession != nil {
historySession.SuccessTurn(http.StatusOK, result.Turn, responsehistory.GenericUsage(result.Turn))
}
writeJSON(w, http.StatusOK, buildGeminiGenerateContentResponseFromTurn(result.Turn))
return true
}
func (h *Handler) handleGeminiDirectStream(w http.ResponseWriter, r *http.Request, a *auth.RequestAuth, stdReq promptcompat.StandardRequest) {
func (h *Handler) applyCurrentInputFile(ctx context.Context, a *auth.RequestAuth, stdReq promptcompat.StandardRequest) (promptcompat.StandardRequest, error) {
if h == nil {
return stdReq, nil
}
return (history.Service{Store: h.Store, DS: h.DS}).ApplyCurrentInputFile(ctx, a, stdReq)
}
func mapCurrentInputFileError(err error) (int, string) {
return history.MapError(err)
}
func (h *Handler) handleGeminiDirectStream(w http.ResponseWriter, r *http.Request, a *auth.RequestAuth, stdReq promptcompat.StandardRequest, historySession *responsehistory.Session) {
start, outErr := completionruntime.StartCompletion(r.Context(), h.DS, a, stdReq, completionruntime.Options{
CurrentInputFile: h.Store,
})
if outErr != nil {
if historySession != nil {
historySession.Error(outErr.Status, outErr.Message, outErr.Code, "", "")
}
writeGeminiError(w, outErr.Status, outErr.Message)
return
}
streamReq := start.Request
h.handleStreamGenerateContent(w, r, start.Response, streamReq.ResponseModel, streamReq.PromptTokenText, streamReq.Thinking, streamReq.Search, streamReq.ToolNames, streamReq.ToolsRaw)
h.handleStreamGenerateContent(w, r, start.Response, streamReq.ResponseModel, streamReq.PromptTokenText, streamReq.Thinking, streamReq.Search, streamReq.ToolNames, streamReq.ToolsRaw, historySession)
}
func (h *Handler) proxyViaOpenAI(w http.ResponseWriter, r *http.Request, stream bool) bool {

10
internal/httpapi/gemini/handler_routes.go
View File

@@ -5,6 +5,7 @@ import (
"github.com/go-chi/chi/v5"
"ds2api/internal/chathistory"
"ds2api/internal/textclean"
"ds2api/internal/util"
)
@@ -12,10 +13,11 @@ import (
var writeJSON = util.WriteJSON
type Handler struct {
Store ConfigReader
Auth AuthResolver
DS DeepSeekCaller
OpenAI OpenAIChatRunner
Store ConfigReader
Auth AuthResolver
DS DeepSeekCaller
OpenAI OpenAIChatRunner
ChatHistory *chathistory.Store
}
//nolint:unused // used by native Gemini stream/non-stream runtime helpers.

31
internal/httpapi/gemini/handler_stream_runtime.go
View File

@@ -9,15 +9,23 @@ import (
"ds2api/internal/assistantturn"
dsprotocol "ds2api/internal/deepseek/protocol"
"ds2api/internal/responsehistory"
"ds2api/internal/sse"
streamengine "ds2api/internal/stream"
)
//nolint:unused // retained for native Gemini stream handling path.
func (h *Handler) handleStreamGenerateContent(w http.ResponseWriter, r *http.Request, resp *http.Response, model, finalPrompt string, thinkingEnabled, searchEnabled bool, toolNames []string, toolsRaw any) {
func (h *Handler) handleStreamGenerateContent(w http.ResponseWriter, r *http.Request, resp *http.Response, model, finalPrompt string, thinkingEnabled, searchEnabled bool, toolNames []string, toolsRaw any, historySessions ...*responsehistory.Session) {
var historySession *responsehistory.Session
if len(historySessions) > 0 {
historySession = historySessions[0]
}
defer func() { _ = resp.Body.Close() }()
if resp.StatusCode != http.StatusOK {
body, _ := io.ReadAll(resp.Body)
if historySession != nil {
historySession.Error(resp.StatusCode, strings.TrimSpace(string(body)), "error", "", "")
}
writeGeminiError(w, resp.StatusCode, strings.TrimSpace(string(body)))
return
}
@@ -29,7 +37,7 @@ func (h *Handler) handleStreamGenerateContent(w http.ResponseWriter, r *http.Req
rc := http.NewResponseController(w)
_, canFlush := w.(http.Flusher)
runtime := newGeminiStreamRuntime(w, rc, canFlush, model, finalPrompt, thinkingEnabled, searchEnabled, stripReferenceMarkersEnabled(), toolNames, toolsRaw)
runtime := newGeminiStreamRuntime(w, rc, canFlush, model, finalPrompt, thinkingEnabled, searchEnabled, stripReferenceMarkersEnabled(), toolNames, toolsRaw, historySession)
initialType := "text"
if thinkingEnabled {
@@ -70,6 +78,7 @@ type geminiStreamRuntime struct {
accumulator *assistantturn.Accumulator
contentFilter bool
responseMessageID int
history *responsehistory.Session
}
//nolint:unused // retained for native Gemini stream handling path.
@@ -84,6 +93,7 @@ func newGeminiStreamRuntime(
stripReferenceMarkers bool,
toolNames []string,
toolsRaw any,
history *responsehistory.Session,
) *geminiStreamRuntime {
return &geminiStreamRuntime{
w: w,
@@ -97,6 +107,7 @@ func newGeminiStreamRuntime(
stripReferenceMarkers: stripReferenceMarkers,
toolNames: toolNames,
toolsRaw: toolsRaw,
history: history,
accumulator: assistantturn.NewAccumulator(assistantturn.AccumulatorOptions{
ThinkingEnabled: thinkingEnabled,
SearchEnabled: searchEnabled,
@@ -170,6 +181,13 @@ func (s *geminiStreamRuntime) onParsed(parsed sse.LineResult) streamengine.Parse
"modelVersion": s.model,
})
}
if s.history != nil {
rawText, text, rawThinking, thinking, detectionThinking := s.accumulator.Snapshot()
s.history.Progress(
responsehistory.ThinkingForArchive(rawThinking, detectionThinking, thinking),
responsehistory.TextForArchive(rawText, text),
)
}
return streamengine.ParsedDecision{ContentSeen: accumulated.ContentSeen}
}
@@ -193,6 +211,15 @@ func (s *geminiStreamRuntime) finalize() {
ToolsRaw: s.toolsRaw,
})
outcome := assistantturn.FinalizeTurn(turn, assistantturn.FinalizeOptions{})
if s.history != nil {
s.history.Success(
http.StatusOK,
responsehistory.ThinkingForArchive(turn.RawThinking, turn.DetectionThinking, turn.Thinking),
responsehistory.TextForArchive(turn.RawText, turn.Text),
assistantturn.FinishReason(turn),
responsehistory.GenericUsage(turn),
)
}
if s.bufferContent {
parts := buildGeminiPartsFromTurn(turn)

33
internal/httpapi/gemini/handler_test.go
View File

@@ -7,12 +7,14 @@ import (
"io"
"net/http"
"net/http/httptest"
"path/filepath"
"strings"
"testing"
"github.com/go-chi/chi/v5"
"ds2api/internal/auth"
"ds2api/internal/chathistory"
dsclient "ds2api/internal/deepseek/client"
)
@@ -138,10 +140,12 @@ func TestGeminiDirectAppliesCurrentInputFile(t *testing.T) {
ds := &testGeminiDS{
resp: makeGeminiUpstreamResponse(`data: {"p":"response/content","v":"ok"}`),
}
historyStore := chathistory.New(filepath.Join(t.TempDir(), "history.json"))
h := &Handler{
Store: testGeminiConfig{},
Auth: testGeminiAuth{},
DS: ds,
Store: testGeminiConfig{},
Auth: testGeminiAuth{},
DS: ds,
ChatHistory: historyStore,
}
reqBody := `{"contents":[{"role":"user","parts":[{"text":"hello from gemini"}]}]}`
req := httptest.NewRequest(http.MethodPost, "/v1beta/models/gemini-2.5-pro:generateContent", strings.NewReader(reqBody))
@@ -172,6 +176,29 @@ func TestGeminiDirectAppliesCurrentInputFile(t *testing.T) {
if !strings.Contains(prompt, "Continue from the latest state in the attached DS2API_HISTORY.txt context.") {
t.Fatalf("expected continuation prompt, got %q", prompt)
}
snapshot, err := historyStore.Snapshot()
if err != nil {
t.Fatalf("snapshot history: %v", err)
}
if len(snapshot.Items) != 1 {
t.Fatalf("expected one history item, got %d", len(snapshot.Items))
}
full, err := historyStore.Get(snapshot.Items[0].ID)
if err != nil {
t.Fatalf("get history item: %v", err)
}
if full.Surface != "gemini.generate_content" {
t.Fatalf("unexpected surface: %q", full.Surface)
}
if full.Content != "ok" {
t.Fatalf("expected raw upstream content, got %q", full.Content)
}
if full.HistoryText != string(ds.uploadCalls[0].Data) {
t.Fatalf("expected uploaded current input file to be persisted in history text")
}
if len(full.Messages) != 1 || !strings.Contains(full.Messages[0].Content, "Continue from the latest state in the attached DS2API_HISTORY.txt context.") {
t.Fatalf("expected persisted message to match upstream continuation prompt, got %#v", full.Messages)
}
}
func TestGeminiRoutesRegistered(t *testing.T) {

7
internal/httpapi/openai/chat/chat_history.go
View File

@@ -14,9 +14,6 @@ import (
"ds2api/internal/promptcompat"
)
const adminWebUISourceHeader = "X-Ds2-Source"
const adminWebUISourceValue = "admin-webui-api-tester"
type chatHistorySession struct {
store *chathistory.Store
entryID string
@@ -40,6 +37,7 @@ func startChatHistory(store *chathistory.Store, r *http.Request, a *auth.Request
entry, err := store.Start(chathistory.StartParams{
CallerID: strings.TrimSpace(a.CallerID),
AccountID: strings.TrimSpace(a.AccountID),
Surface: "openai.chat_completions",
Model: strings.TrimSpace(stdReq.ResponseModel),
Stream: stdReq.Stream,
UserInput: extractSingleUserInput(stdReq.Messages),
@@ -50,6 +48,7 @@ func startChatHistory(store *chathistory.Store, r *http.Request, a *auth.Request
startParams := chathistory.StartParams{
CallerID: strings.TrimSpace(a.CallerID),
AccountID: strings.TrimSpace(a.AccountID),
Surface: "openai.chat_completions",
Model: strings.TrimSpace(stdReq.ResponseModel),
Stream: stdReq.Stream,
UserInput: extractSingleUserInput(stdReq.Messages),
@@ -82,7 +81,7 @@ func shouldCaptureChatHistory(r *http.Request) bool {
if isVercelStreamPrepareRequest(r) || isVercelStreamReleaseRequest(r) {
return false
}
return strings.TrimSpace(r.Header.Get(adminWebUISourceHeader)) != adminWebUISourceValue
return true
}
func extractSingleUserInput(messages []any) string {

8
internal/httpapi/openai/chat/chat_history_test.go
View File

@@ -294,7 +294,7 @@ func TestHandleStreamContextCancelledMarksHistoryStopped(t *testing.T) {
}
}
func TestChatCompletionsSkipsAdminWebUISource(t *testing.T) {
func TestChatCompletionsRecordsAdminWebUISource(t *testing.T) {
historyStore := newTestChatHistoryStore(t)
h := &Handler{
Store: mockOpenAIConfig{},
@@ -307,7 +307,7 @@ func TestChatCompletionsSkipsAdminWebUISource(t *testing.T) {
req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(reqBody))
req.Header.Set("Authorization", "Bearer direct-token")
req.Header.Set("Content-Type", "application/json")
req.Header.Set(adminWebUISourceHeader, adminWebUISourceValue)
req.Header.Set("X-Ds2-Source", "admin-webui-api-tester")
rec := httptest.NewRecorder()
h.ChatCompletions(rec, req)
@@ -318,8 +318,8 @@ func TestChatCompletionsSkipsAdminWebUISource(t *testing.T) {
if err != nil {
t.Fatalf("snapshot failed: %v", err)
}
if len(snapshot.Items) != 0 {
t.Fatalf("expected admin webui source to be skipped, got %#v", snapshot.Items)
if len(snapshot.Items) != 1 {
t.Fatalf("expected admin webui source to be recorded, got %#v", snapshot.Items)
}
}

12
internal/httpapi/openai/responses/empty_retry_runtime.go
View File

@@ -10,11 +10,12 @@ import (
"ds2api/internal/config"
dsprotocol "ds2api/internal/deepseek/protocol"
"ds2api/internal/promptcompat"
"ds2api/internal/responsehistory"
streamengine "ds2api/internal/stream"
)
func (h *Handler) handleResponsesStreamWithRetry(w http.ResponseWriter, r *http.Request, a *auth.RequestAuth, resp *http.Response, payload map[string]any, pow, owner, responseID, model, finalPrompt string, refFileTokens int, thinkingEnabled, searchEnabled bool, toolNames []string, toolsRaw any, toolChoice promptcompat.ToolChoicePolicy, traceID string) {
streamRuntime, initialType, ok := h.prepareResponsesStreamRuntime(w, resp, owner, responseID, model, finalPrompt, refFileTokens, thinkingEnabled, searchEnabled, toolNames, toolsRaw, toolChoice, traceID)
func (h *Handler) handleResponsesStreamWithRetry(w http.ResponseWriter, r *http.Request, a *auth.RequestAuth, resp *http.Response, payload map[string]any, pow, owner, responseID, model, finalPrompt string, refFileTokens int, thinkingEnabled, searchEnabled bool, toolNames []string, toolsRaw any, toolChoice promptcompat.ToolChoicePolicy, traceID string, historySession *responsehistory.Session) {
streamRuntime, initialType, ok := h.prepareResponsesStreamRuntime(w, resp, owner, responseID, model, finalPrompt, refFileTokens, thinkingEnabled, searchEnabled, toolNames, toolsRaw, toolChoice, traceID, historySession)
if !ok {
return
}
@@ -55,10 +56,13 @@ func (h *Handler) handleResponsesStreamWithRetry(w http.ResponseWriter, r *http.
}
}
func (h *Handler) prepareResponsesStreamRuntime(w http.ResponseWriter, resp *http.Response, owner, responseID, model, finalPrompt string, refFileTokens int, thinkingEnabled, searchEnabled bool, toolNames []string, toolsRaw any, toolChoice promptcompat.ToolChoicePolicy, traceID string) (*responsesStreamRuntime, string, bool) {
func (h *Handler) prepareResponsesStreamRuntime(w http.ResponseWriter, resp *http.Response, owner, responseID, model, finalPrompt string, refFileTokens int, thinkingEnabled, searchEnabled bool, toolNames []string, toolsRaw any, toolChoice promptcompat.ToolChoicePolicy, traceID string, historySession *responsehistory.Session) (*responsesStreamRuntime, string, bool) {
if resp.StatusCode != http.StatusOK {
defer func() { _ = resp.Body.Close() }()
body, _ := io.ReadAll(resp.Body)
if historySession != nil {
historySession.Error(resp.StatusCode, strings.TrimSpace(string(body)), "error", "", "")
}
writeOpenAIError(w, resp.StatusCode, strings.TrimSpace(string(body)))
return nil, "", false
}
@@ -78,7 +82,7 @@ func (h *Handler) prepareResponsesStreamRuntime(w http.ResponseWriter, resp *htt
h.toolcallFeatureMatchEnabled() && h.toolcallEarlyEmitHighConfidence(),
toolChoice, traceID, func(obj map[string]any) {
h.getResponseStore().put(owner, responseID, obj)
},
}, historySession,
)
streamRuntime.refFileTokens = refFileTokens
streamRuntime.sendCreated()

1
internal/httpapi/openai/responses/empty_retry_runtime_test.go
View File

@@ -47,6 +47,7 @@ func TestConsumeResponsesStreamAttemptMarksContextCancelledState(t *testing.T) {
promptcompat.DefaultToolChoicePolicy(),
"",
nil,
nil,
)
resp := makeResponsesOpenAISSEHTTPResponse(
`data: {"p":"response/content","v":"hello"}`,

20
internal/httpapi/openai/responses/responses_handler.go
View File

@@ -18,6 +18,7 @@ import (
dsprotocol "ds2api/internal/deepseek/protocol"
openaifmt "ds2api/internal/format/openai"
"ds2api/internal/promptcompat"
"ds2api/internal/responsehistory"
"ds2api/internal/sse"
streamengine "ds2api/internal/stream"
)
@@ -95,6 +96,13 @@ func (h *Handler) Responses(w http.ResponseWriter, r *http.Request) {
}
responseID := "resp_" + strings.ReplaceAll(uuid.NewString(), "-", "")
historySession := responsehistory.Start(responsehistory.StartParams{
Store: h.ChatHistory,
Request: r,
Auth: a,
Surface: "openai.responses",
Standard: stdReq,
})
if !stdReq.Stream {
result, outErr := completionruntime.ExecuteNonStreamWithRetry(r.Context(), h.DS, a, stdReq, completionruntime.Options{
StripReferenceMarkers: stripReferenceMarkersEnabled(),
@@ -102,9 +110,15 @@ func (h *Handler) Responses(w http.ResponseWriter, r *http.Request) {
CurrentInputFile: h.Store,
})
if outErr != nil {
if historySession != nil {
historySession.ErrorTurn(outErr.Status, outErr.Message, outErr.Code, result.Turn)
}
writeOpenAIErrorWithCode(w, outErr.Status, outErr.Message, outErr.Code)
return
}
if historySession != nil {
historySession.SuccessTurn(http.StatusOK, result.Turn, assistantturn.OpenAIResponsesUsage(result.Turn))
}
responseObj := openaifmt.BuildResponseObjectWithToolCalls(responseID, stdReq.ResponseModel, result.Turn.Prompt, result.Turn.Thinking, result.Turn.Text, result.Turn.ToolCalls, stdReq.ToolsRaw)
responseObj["usage"] = assistantturn.OpenAIResponsesUsage(result.Turn)
h.getResponseStore().put(owner, responseID, responseObj)
@@ -116,13 +130,16 @@ func (h *Handler) Responses(w http.ResponseWriter, r *http.Request) {
CurrentInputFile: h.Store,
})
if outErr != nil {
if historySession != nil {
historySession.Error(outErr.Status, outErr.Message, outErr.Code, "", "")
}
writeOpenAIErrorWithCode(w, outErr.Status, outErr.Message, outErr.Code)
return
}
streamReq := start.Request
refFileTokens := streamReq.RefFileTokens
h.handleResponsesStreamWithRetry(w, r, a, start.Response, start.Payload, start.Pow, owner, responseID, streamReq.ResponseModel, streamReq.PromptTokenText, refFileTokens, streamReq.Thinking, streamReq.Search, streamReq.ToolNames, streamReq.ToolsRaw, streamReq.ToolChoice, traceID)
h.handleResponsesStreamWithRetry(w, r, a, start.Response, start.Payload, start.Pow, owner, responseID, streamReq.ResponseModel, streamReq.PromptTokenText, refFileTokens, streamReq.Thinking, streamReq.Search, streamReq.ToolNames, streamReq.ToolsRaw, streamReq.ToolChoice, traceID, historySession)
}
func (h *Handler) handleResponsesNonStream(w http.ResponseWriter, resp *http.Response, owner, responseID, model, finalPrompt string, refFileTokens int, thinkingEnabled, searchEnabled bool, toolNames []string, toolsRaw any, toolChoice promptcompat.ToolChoicePolicy, traceID string) {
@@ -198,6 +215,7 @@ func (h *Handler) handleResponsesStream(w http.ResponseWriter, r *http.Request,
func(obj map[string]any) {
h.getResponseStore().put(owner, responseID, obj)
},
nil,
)
streamRuntime.refFileTokens = refFileTokens
streamRuntime.sendCreated()

100
internal/httpapi/openai/responses/responses_history_test.go Normal file
View File

@@ -0,0 +1,100 @@
package responses
import (
"context"
"io"
"net/http"
"net/http/httptest"
"path/filepath"
"strings"
"testing"
"github.com/go-chi/chi/v5"
"ds2api/internal/auth"
"ds2api/internal/chathistory"
dsclient "ds2api/internal/deepseek/client"
)
type responsesHistoryDS struct {
payload map[string]any
}
func (d *responsesHistoryDS) CreateSession(context.Context, *auth.RequestAuth, int) (string, error) {
return "session-id", nil
}
func (d *responsesHistoryDS) GetPow(context.Context, *auth.RequestAuth, int) (string, error) {
return "pow", nil
}
func (d *responsesHistoryDS) UploadFile(context.Context, *auth.RequestAuth, dsclient.UploadFileRequest, int) (*dsclient.UploadFileResult, error) {
return &dsclient.UploadFileResult{ID: "file-id"}, nil
}
func (d *responsesHistoryDS) CallCompletion(_ context.Context, _ *auth.RequestAuth, payload map[string]any, _ string, _ int) (*http.Response, error) {
d.payload = payload
return &http.Response{
StatusCode: http.StatusOK,
Header: make(http.Header),
Body: io.NopCloser(strings.NewReader("data: {\"p\":\"response/content\",\"v\":\"ok\"}\n")),
}, nil
}
func (d *responsesHistoryDS) DeleteSessionForToken(context.Context, string, string) (*dsclient.DeleteSessionResult, error) {
return &dsclient.DeleteSessionResult{Success: true}, nil
}
func (d *responsesHistoryDS) DeleteAllSessionsForToken(context.Context, string) error {
return nil
}
func TestResponsesRecordsResponseHistory(t *testing.T) {
store, resolver := newDirectTokenResolver(t)
historyStore := chathistory.New(filepath.Join(t.TempDir(), "history.json"))
ds := &responsesHistoryDS{}
h := &Handler{
Store: store,
Auth: resolver,
DS: ds,
ChatHistory: historyStore,
}
r := chi.NewRouter()
RegisterRoutes(r, h)
req := httptest.NewRequest(http.MethodPost, "/v1/responses", strings.NewReader(`{"model":"deepseek-v4-flash","input":"hello responses"}`))
req.Header.Set("Authorization", "Bearer direct-token")
req.Header.Set("Content-Type", "application/json")
rec := httptest.NewRecorder()
r.ServeHTTP(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("expected 200, got %d body=%s", rec.Code, rec.Body.String())
}
if ds.payload == nil {
t.Fatalf("expected upstream payload to be sent")
}
snapshot, err := historyStore.Snapshot()
if err != nil {
t.Fatalf("snapshot history: %v", err)
}
if len(snapshot.Items) != 1 {
t.Fatalf("expected one history item, got %d", len(snapshot.Items))
}
item, err := historyStore.Get(snapshot.Items[0].ID)
if err != nil {
t.Fatalf("get history item: %v", err)
}
if item.Surface != "openai.responses" {
t.Fatalf("unexpected surface: %q", item.Surface)
}
if !strings.Contains(item.UserInput, "Continue from the latest state in the attached DS2API_HISTORY.txt context.") {
t.Fatalf("unexpected user input: %q", item.UserInput)
}
if !strings.Contains(item.HistoryText, "hello responses") {
t.Fatalf("expected original input in persisted history text, got %q", item.HistoryText)
}
if item.Content != "ok" {
t.Fatalf("expected raw upstream content, got %q", item.Content)
}
}

22
internal/httpapi/openai/responses/responses_stream_runtime_core.go
View File

@@ -10,6 +10,7 @@ import (
openaifmt "ds2api/internal/format/openai"
"ds2api/internal/httpapi/openai/shared"
"ds2api/internal/promptcompat"
"ds2api/internal/responsehistory"
"ds2api/internal/sse"
streamengine "ds2api/internal/stream"
"ds2api/internal/toolstream"
@@ -61,6 +62,7 @@ type responsesStreamRuntime struct {
finalErrorCode string
persistResponse func(obj map[string]any)
history *responsehistory.Session
}
func newResponsesStreamRuntime(
@@ -80,6 +82,7 @@ func newResponsesStreamRuntime(
toolChoice promptcompat.ToolChoicePolicy,
traceID string,
persistResponse func(obj map[string]any),
history *responsehistory.Session,
) *responsesStreamRuntime {
return &responsesStreamRuntime{
w: w,
@@ -106,6 +109,7 @@ func newResponsesStreamRuntime(
toolChoice: toolChoice,
traceID: traceID,
persistResponse: persistResponse,
history: history,
accumulator: shared.StreamAccumulator{
ThinkingEnabled: thinkingEnabled,
SearchEnabled: searchEnabled,
@@ -138,6 +142,9 @@ func (s *responsesStreamRuntime) failResponse(status int, message, code string)
if s.persistResponse != nil {
s.persistResponse(failedResp)
}
if s.history != nil {
s.history.Error(status, message, code, responsehistory.ThinkingForArchive(s.accumulator.RawThinking.String(), s.accumulator.ToolDetectionThinking.String(), s.accumulator.Thinking.String()), responsehistory.TextForArchive(s.accumulator.RawText.String(), s.accumulator.Text.String()))
}
s.sendEvent("response.failed", openaifmt.BuildResponsesFailedPayload(s.responseID, s.model, status, message, code))
s.sendDone()
}
@@ -214,6 +221,15 @@ func (s *responsesStreamRuntime) finalize(finishReason string, deferEmptyOutput
if s.persistResponse != nil {
s.persistResponse(obj)
}
if s.history != nil {
s.history.Success(
http.StatusOK,
responsehistory.ThinkingForArchive(turn.RawThinking, turn.DetectionThinking, turn.Thinking),
responsehistory.TextForArchive(turn.RawText, turn.Text),
outcome.FinishReason,
assistantturn.OpenAIResponsesUsage(turn),
)
}
s.sendEvent("response.completed", openaifmt.BuildResponsesCompletedPayload(obj))
s.sendDone()
return true
@@ -272,5 +288,11 @@ func (s *responsesStreamRuntime) onParsed(parsed sse.LineResult) streamengine.Pa
}
batch.flush()
if s.history != nil {
s.history.Progress(
responsehistory.ThinkingForArchive(s.accumulator.RawThinking.String(), s.accumulator.ToolDetectionThinking.String(), s.accumulator.Thinking.String()),
responsehistory.TextForArchive(s.accumulator.RawText.String(), s.accumulator.Text.String()),
)
}
return streamengine.ParsedDecision{ContentSeen: accumulated.ContentSeen}
}

289
internal/responsehistory/session.go Normal file
View File

@@ -0,0 +1,289 @@
package responsehistory
import (
"errors"
"net/http"
"strings"
"time"
"ds2api/internal/assistantturn"
"ds2api/internal/auth"
"ds2api/internal/chathistory"
"ds2api/internal/config"
"ds2api/internal/prompt"
"ds2api/internal/promptcompat"
)
type Session struct {
store *chathistory.Store
entryID string
startedAt time.Time
lastPersist time.Time
startParams chathistory.StartParams
disabled bool
}
type StartParams struct {
Store *chathistory.Store
Request *http.Request
Auth *auth.RequestAuth
Surface string
Standard promptcompat.StandardRequest
}
func Start(params StartParams) *Session {
if params.Store == nil || params.Request == nil || params.Auth == nil {
return nil
}
if !params.Store.Enabled() || !shouldCapture(params.Request) {
return nil
}
startParams := chathistory.StartParams{
CallerID: strings.TrimSpace(params.Auth.CallerID),
AccountID: strings.TrimSpace(params.Auth.AccountID),
Surface: strings.TrimSpace(params.Surface),
Model: strings.TrimSpace(params.Standard.ResponseModel),
Stream: params.Standard.Stream,
UserInput: ExtractSingleUserInput(params.Standard.Messages),
Messages: ExtractAllMessages(params.Standard.Messages),
HistoryText: params.Standard.HistoryText,
FinalPrompt: params.Standard.FinalPrompt,
}
entry, err := params.Store.Start(startParams)
session := &Session{
store: params.Store,
entryID: entry.ID,
startedAt: time.Now(),
lastPersist: time.Now(),
startParams: startParams,
}
if err != nil {
if entry.ID == "" {
config.Logger.Warn("[response_history] start failed", "surface", startParams.Surface, "error", err)
return nil
}
config.Logger.Warn("[response_history] start persisted in memory after write failure", "surface", startParams.Surface, "error", err)
}
return session
}
func shouldCapture(r *http.Request) bool {
if r == nil || r.URL == nil {
return false
}
if strings.TrimSpace(r.URL.Query().Get("__stream_prepare")) == "1" {
return false
}
if strings.TrimSpace(r.URL.Query().Get("__stream_release")) == "1" {
return false
}
return true
}
func ExtractSingleUserInput(messages []any) string {
for i := len(messages) - 1; i >= 0; i-- {
msg, ok := messages[i].(map[string]any)
if !ok {
continue
}
role := strings.ToLower(strings.TrimSpace(asString(msg["role"])))
if role != "user" {
continue
}
if normalized := strings.TrimSpace(prompt.NormalizeContent(msg["content"])); normalized != "" {
return normalized
}
}
return ""
}
func ExtractAllMessages(messages []any) []chathistory.Message {
out := make([]chathistory.Message, 0, len(messages))
for _, raw := range messages {
msg, ok := raw.(map[string]any)
if !ok {
continue
}
role := strings.ToLower(strings.TrimSpace(asString(msg["role"])))
content := strings.TrimSpace(prompt.NormalizeContent(msg["content"]))
if role == "" || content == "" {
continue
}
out = append(out, chathistory.Message{
Role: role,
Content: content,
})
}
return out
}
func (s *Session) Progress(thinking, content string) {
if s == nil || s.store == nil || s.disabled {
return
}
now := time.Now()
if now.Sub(s.lastPersist) < 250*time.Millisecond {
return
}
s.lastPersist = now
s.persistUpdate(chathistory.UpdateParams{
Status: "streaming",
ReasoningContent: thinking,
Content: content,
StatusCode: http.StatusOK,
ElapsedMs: time.Since(s.startedAt).Milliseconds(),
})
}
func (s *Session) Success(statusCode int, thinking, content, finishReason string, usage map[string]any) {
if s == nil || s.store == nil || s.disabled {
return
}
s.persistUpdate(chathistory.UpdateParams{
Status: "success",
ReasoningContent: thinking,
Content: content,
StatusCode: statusCode,
ElapsedMs: time.Since(s.startedAt).Milliseconds(),
FinishReason: finishReason,
Usage: usage,
Completed: true,
})
}
func (s *Session) Error(statusCode int, message, finishReason, thinking, content string) {
if s == nil || s.store == nil || s.disabled {
return
}
s.persistUpdate(chathistory.UpdateParams{
Status: "error",
ReasoningContent: thinking,
Content: content,
Error: message,
StatusCode: statusCode,
ElapsedMs: time.Since(s.startedAt).Milliseconds(),
FinishReason: finishReason,
Completed: true,
})
}
func (s *Session) SuccessTurn(statusCode int, turn assistantturn.Turn, usage map[string]any) {
outcome := assistantturn.FinalizeTurn(turn, assistantturn.FinalizeOptions{})
s.Success(
statusCode,
ThinkingForArchive(turn.RawThinking, turn.DetectionThinking, turn.Thinking),
TextForArchive(turn.RawText, turn.Text),
outcome.FinishReason,
usage,
)
}
func (s *Session) ErrorTurn(statusCode int, message, finishReason string, turn assistantturn.Turn) {
s.Error(
statusCode,
message,
finishReason,
ThinkingForArchive(turn.RawThinking, turn.DetectionThinking, turn.Thinking),
TextForArchive(turn.RawText, turn.Text),
)
}
func TextForArchive(raw, visible string) string {
if strings.TrimSpace(raw) != "" {
return raw
}
return visible
}
func ThinkingForArchive(raw, detection, visible string) string {
if strings.TrimSpace(raw) != "" {
return raw
}
if strings.TrimSpace(detection) != "" {
return detection
}
return visible
}
func GenericUsage(turn assistantturn.Turn) map[string]any {
return map[string]any{
"input_tokens": turn.Usage.InputTokens,
"output_tokens": turn.Usage.OutputTokens,
"reasoning_tokens": turn.Usage.ReasoningTokens,
"total_tokens": turn.Usage.TotalTokens,
}
}
func (s *Session) retryMissingEntry() bool {
if s == nil || s.store == nil || s.disabled {
return false
}
entry, err := s.store.Start(s.startParams)
if errors.Is(err, chathistory.ErrDisabled) {
s.disabled = true
return false
}
if entry.ID == "" {
if err != nil {
config.Logger.Warn("[response_history] recreate missing entry failed", "surface", s.startParams.Surface, "error", err)
}
return false
}
s.entryID = entry.ID
if err != nil {
config.Logger.Warn("[response_history] recreate missing entry persisted in memory after write failure", "surface", s.startParams.Surface, "error", err)
}
return true
}
func (s *Session) persistUpdate(params chathistory.UpdateParams) {
if s == nil || s.store == nil || s.disabled {
return
}
if _, err := s.store.Update(s.entryID, params); err != nil {
s.handlePersistError(params, err)
}
}
func (s *Session) handlePersistError(params chathistory.UpdateParams, err error) {
if err == nil || s == nil {
return
}
if errors.Is(err, chathistory.ErrDisabled) {
s.disabled = true
return
}
if isMissingError(err) {
if s.retryMissingEntry() {
if _, retryErr := s.store.Update(s.entryID, params); retryErr != nil {
if errors.Is(retryErr, chathistory.ErrDisabled) || isMissingError(retryErr) {
s.disabled = true
return
}
config.Logger.Warn("[response_history] retry after missing entry failed", "surface", s.startParams.Surface, "error", retryErr)
}
return
}
s.disabled = true
return
}
config.Logger.Warn("[response_history] update failed", "surface", s.startParams.Surface, "error", err)
}
func isMissingError(err error) bool {
if err == nil {
return false
}
return strings.Contains(strings.ToLower(err.Error()), "not found")
}
func asString(v any) string {
switch x := v.(type) {
case string:
return x
case nil:
return ""
default:
return strings.TrimSpace(prompt.NormalizeContent(x))
}
}

4
internal/server/router.go
View File

@@ -65,8 +65,8 @@ func NewApp() (*App, error) {
responsesHandler := &responses.Handler{Store: store, Auth: resolver, DS: dsClient, ChatHistory: chatHistoryStore}
filesHandler := &files.Handler{Store: store, Auth: resolver, DS: dsClient, ChatHistory: chatHistoryStore}
embeddingsHandler := &embeddings.Handler{Store: store, Auth: resolver, DS: dsClient, ChatHistory: chatHistoryStore}
claudeHandler := &claude.Handler{Store: store, Auth: resolver, DS: dsClient, OpenAI: chatHandler}
geminiHandler := &gemini.Handler{Store: store, Auth: resolver, DS: dsClient, OpenAI: chatHandler}
claudeHandler := &claude.Handler{Store: store, Auth: resolver, DS: dsClient, OpenAI: chatHandler, ChatHistory: chatHistoryStore}
geminiHandler := &gemini.Handler{Store: store, Auth: resolver, DS: dsClient, OpenAI: chatHandler, ChatHistory: chatHistoryStore}
adminHandler := &admin.Handler{Store: store, Pool: pool, DS: dsClient, OpenAI: chatHandler, ChatHistory: chatHistoryStore}
webuiHandler := webui.NewHandler()