Merge pull request #2657 from lc6464/fix-deepseek-v4-thinking-history

fix(reasoning): persist canonical history for DeepSeek and web chat
This commit is contained in:
美電球 2026-04-25 15:08:48 +08:00 committed by GitHub
commit 41f4d95597
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
17 changed files with 1172 additions and 73 deletions

View file

@ -1720,6 +1720,38 @@ func (m *messageToolProvider) GetDefaultModel() string {
return "message-tool-model" return "message-tool-model"
} }
type reasoningVisibleToolProvider struct {
filePath string
calls int
}
func (m *reasoningVisibleToolProvider) Chat(
ctx context.Context,
messages []providers.Message,
tools []providers.ToolDefinition,
model string,
opts map[string]any,
) (*providers.LLMResponse, error) {
m.calls++
if m.calls == 1 {
return &providers.LLMResponse{
Content: "I'll inspect that file now.",
ReasoningContent: "Read the file before answering.",
ToolCalls: []providers.ToolCall{{
ID: "call_read_file",
Type: "function",
Name: "read_file",
Arguments: map[string]any{"path": m.filePath},
}},
}, nil
}
return &providers.LLMResponse{Content: "DONE"}, nil
}
func (m *reasoningVisibleToolProvider) GetDefaultModel() string {
return "reasoning-visible-tool-model"
}
type artifactThenSendProvider struct { type artifactThenSendProvider struct {
calls int calls int
} }
@ -1866,6 +1898,28 @@ func TestToolFeedbackExplanationFromResponse_UsesCurrentContentFirst(t *testing.
} }
} }
func TestSideQuestionResponseContent_FallsBackWhenContentIsWhitespace(t *testing.T) {
response := &providers.LLMResponse{
Content: " \n\t ",
ReasoningContent: "reasoning fallback",
}
if got := sideQuestionResponseContent(response); got != "reasoning fallback" {
t.Fatalf("sideQuestionResponseContent() = %q, want %q", got, "reasoning fallback")
}
}
func TestResponseReasoningContent_FallsBackWhenReasoningIsWhitespace(t *testing.T) {
response := &providers.LLMResponse{
Reasoning: " \n\t ",
ReasoningContent: "structured reasoning fallback",
}
if got := responseReasoningContent(response); got != "structured reasoning fallback" {
t.Fatalf("responseReasoningContent() = %q, want %q", got, "structured reasoning fallback")
}
}
func TestToolFeedbackExplanationFromResponse_UsesExplicitToolCallExtraContent(t *testing.T) { func TestToolFeedbackExplanationFromResponse_UsesExplicitToolCallExtraContent(t *testing.T) {
response := &providers.LLMResponse{ response := &providers.LLMResponse{
ToolCalls: []providers.ToolCall{{ ToolCalls: []providers.ToolCall{{
@ -3974,6 +4028,182 @@ func TestProcessMessage_PublishesToolFeedbackWhenEnabled(t *testing.T) {
} }
} }
func TestProcessMessage_PersistsReasoningContentInSessionHistory(t *testing.T) {
tmpDir := t.TempDir()
cfg := &config.Config{
Agents: config.AgentsConfig{
Defaults: config.AgentDefaults{
Workspace: tmpDir,
ModelName: "test-model",
MaxTokens: 4096,
MaxToolIterations: 10,
},
},
}
msgBus := bus.NewMessageBus()
provider := &reasoningContentProvider{
response: "final answer",
reasoningContent: "thinking trace",
}
al := NewAgentLoop(cfg, msgBus, provider)
response, err := al.processMessage(context.Background(), bus.InboundMessage{
Channel: "pico",
SenderID: "user1",
ChatID: "pico:test-session",
Content: "hello",
})
if err != nil {
t.Fatalf("processMessage() error = %v", err)
}
if response != "final answer" {
t.Fatalf("processMessage() response = %q, want %q", response, "final answer")
}
store := al.GetRegistry().GetDefaultAgent().Sessions
sessionKeys := store.ListSessions()
if len(sessionKeys) != 1 {
t.Fatalf("session keys = %v, want exactly 1 active session", sessionKeys)
}
history := store.GetHistory(sessionKeys[0])
if len(history) < 2 {
t.Fatalf("session history len = %d, want at least 2", len(history))
}
last := history[len(history)-1]
if last.Role != "assistant" {
t.Fatalf("last message role = %q, want assistant", last.Role)
}
if last.Content != "final answer" {
t.Fatalf("last message content = %q, want %q", last.Content, "final answer")
}
if last.ReasoningContent != "thinking trace" {
t.Fatalf("last message reasoning_content = %q, want %q", last.ReasoningContent, "thinking trace")
}
}
func TestProcessMessage_PersistsReasoningToolResponseAsSingleAssistantRecord(t *testing.T) {
tmpDir := t.TempDir()
inspectPath := filepath.Join(tmpDir, "inspect.txt")
if err := os.WriteFile(inspectPath, []byte("inspect me"), 0o644); err != nil {
t.Fatalf("WriteFile(inspectPath) error = %v", err)
}
cfg := config.DefaultConfig()
cfg.Agents.Defaults.Workspace = tmpDir
cfg.Agents.Defaults.ModelName = "test-model"
cfg.Agents.Defaults.MaxTokens = 4096
cfg.Agents.Defaults.MaxToolIterations = 10
msgBus := bus.NewMessageBus()
provider := &reasoningVisibleToolProvider{filePath: inspectPath}
al := NewAgentLoop(cfg, msgBus, provider)
response, err := al.processMessage(context.Background(), bus.InboundMessage{
Channel: "telegram",
SenderID: "user1",
ChatID: "chat1",
Content: "hello",
})
if err != nil {
t.Fatalf("processMessage() error = %v", err)
}
if response != "DONE" {
t.Fatalf("processMessage() response = %q, want %q", response, "DONE")
}
store := al.GetRegistry().GetDefaultAgent().Sessions
sessionKeys := store.ListSessions()
if len(sessionKeys) != 1 {
t.Fatalf("session keys = %v, want exactly 1 active session", sessionKeys)
}
history := store.GetHistory(sessionKeys[0])
if len(history) < 3 {
t.Fatalf("session history len = %d, want at least 3", len(history))
}
var assistantWithToolCall *providers.Message
for i := range history {
msg := history[i]
if msg.Role == "assistant" && len(msg.ToolCalls) > 0 {
assistantWithToolCall = &msg
break
}
}
if assistantWithToolCall == nil {
t.Fatal("expected assistant history record with tool_calls")
}
if assistantWithToolCall.Content != "I'll inspect that file now." {
t.Fatalf("assistant content = %q, want %q", assistantWithToolCall.Content, "I'll inspect that file now.")
}
if assistantWithToolCall.ReasoningContent != "Read the file before answering." {
t.Fatalf("assistant reasoning_content = %q, want preserved", assistantWithToolCall.ReasoningContent)
}
if len(assistantWithToolCall.ToolCalls) != 1 {
t.Fatalf("assistant tool calls = %+v, want single read_file tool", assistantWithToolCall.ToolCalls)
}
if got := providers.NormalizeToolCall(assistantWithToolCall.ToolCalls[0]).Name; got != "read_file" {
t.Fatalf("assistant tool calls = %+v, want single read_file tool", assistantWithToolCall.ToolCalls)
}
sessionDir := filepath.Join(tmpDir, "sessions")
entries, err := os.ReadDir(sessionDir)
if err != nil {
t.Fatalf("ReadDir(%q) error = %v", sessionDir, err)
}
var jsonlPath string
for _, entry := range entries {
if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".jsonl") {
continue
}
jsonlPath = filepath.Join(sessionDir, entry.Name())
break
}
if jsonlPath == "" {
t.Fatal("expected session jsonl file to be created")
}
data, err := os.ReadFile(jsonlPath)
if err != nil {
t.Fatalf("ReadFile(%q) error = %v", jsonlPath, err)
}
lines := strings.Split(strings.TrimSpace(string(data)), "\n")
if len(lines) < 3 {
t.Fatalf("jsonl lines = %d, want at least 3", len(lines))
}
matchingRecords := 0
for _, line := range lines {
var msg providers.Message
if err := json.Unmarshal([]byte(line), &msg); err != nil {
t.Fatalf("Unmarshal(jsonl line) error = %v", err)
}
if msg.Role != "assistant" {
continue
}
if msg.Content == "I'll inspect that file now." || msg.ReasoningContent == "Read the file before answering." {
matchingRecords++
toolName := ""
if len(msg.ToolCalls) == 1 {
toolName = providers.NormalizeToolCall(msg.ToolCalls[0]).Name
}
if msg.Content != "I'll inspect that file now." ||
msg.ReasoningContent != "Read the file before answering." ||
len(msg.ToolCalls) != 1 ||
toolName != "read_file" {
t.Fatalf("assistant jsonl record = %+v, want content+reasoning+tool_calls in one line", msg)
}
}
}
if matchingRecords != 1 {
t.Fatalf("matching assistant jsonl records = %d, want exactly 1 canonical assistant record", matchingRecords)
}
}
func TestProcessMessage_DoesNotLeakReasoningContentInToolFeedback(t *testing.T) { func TestProcessMessage_DoesNotLeakReasoningContentInToolFeedback(t *testing.T) {
tmpDir := t.TempDir() tmpDir := t.TempDir()
heartbeatFile := filepath.Join(tmpDir, "tool-feedback-reasoning.txt") heartbeatFile := filepath.Join(tmpDir, "tool-feedback-reasoning.txt")

View file

@ -6,6 +6,7 @@ import (
"context" "context"
"encoding/json" "encoding/json"
"fmt" "fmt"
"maps"
"path/filepath" "path/filepath"
"strings" "strings"
"time" "time"
@ -478,17 +479,28 @@ func sideQuestionResponseContent(response *providers.LLMResponse) string {
if response == nil { if response == nil {
return "" return ""
} }
if response.Content != "" { if strings.TrimSpace(response.Content) != "" {
return response.Content return response.Content
} }
return response.ReasoningContent return responseReasoningContent(response)
}
func responseReasoningContent(response *providers.LLMResponse) string {
if response == nil {
return ""
}
if strings.TrimSpace(response.Reasoning) != "" {
return response.Reasoning
}
if strings.TrimSpace(response.ReasoningContent) != "" {
return response.ReasoningContent
}
return ""
} }
func shallowCloneLLMOptions(opts map[string]any) map[string]any { func shallowCloneLLMOptions(opts map[string]any) map[string]any {
clone := make(map[string]any, len(opts)) clone := make(map[string]any, len(opts))
for k, v := range opts { maps.Copy(clone, opts)
clone[k] = v
}
return clone return clone
} }

View file

@ -40,8 +40,12 @@ func (p *Pipeline) Finalize(
ts.setPhase(TurnPhaseFinalizing) ts.setPhase(TurnPhaseFinalizing)
ts.setFinalContent(finalContent) ts.setFinalContent(finalContent)
if !ts.opts.NoHistory { if !ts.opts.NoHistory {
finalMsg := providers.Message{Role: "assistant", Content: finalContent} finalMsg := providers.Message{
ts.agent.Sessions.AddMessage(ts.sessionKey, finalMsg.Role, finalMsg.Content) Role: "assistant",
Content: finalContent,
ReasoningContent: responseReasoningContent(exec.response),
}
ts.agent.Sessions.AddFullMessage(ts.sessionKey, finalMsg)
ts.recordPersistedMessage(finalMsg) ts.recordPersistedMessage(finalMsg)
ts.ingestMessage(turnCtx, al, finalMsg) ts.ingestMessage(turnCtx, al, finalMsg)
if err := ts.agent.Sessions.Save(ts.sessionKey); err != nil { if err := ts.agent.Sessions.Save(ts.sessionKey); err != nil {

View file

@ -382,10 +382,7 @@ func (p *Pipeline) CallLLM(
} }
} }
reasoningContent := exec.response.Reasoning reasoningContent := responseReasoningContent(exec.response)
if reasoningContent == "" {
reasoningContent = exec.response.ReasoningContent
}
if ts.channel == "pico" { if ts.channel == "pico" {
go al.publishPicoReasoning(turnCtx, reasoningContent, ts.chatID) go al.publishPicoReasoning(turnCtx, reasoningContent, ts.chatID)
} else { } else {
@ -494,7 +491,7 @@ func (p *Pipeline) CallLLM(
assistantMsg := providers.Message{ assistantMsg := providers.Message{
Role: "assistant", Role: "assistant",
Content: exec.response.Content, Content: exec.response.Content,
ReasoningContent: exec.response.ReasoningContent, ReasoningContent: reasoningContent,
} }
for _, tc := range exec.normalizedToolCalls { for _, tc := range exec.normalizedToolCalls {
argumentsJSON, _ := json.Marshal(tc.Arguments) argumentsJSON, _ := json.Marshal(tc.Arguments)

View file

@ -10,6 +10,7 @@ import (
"github.com/sipeed/picoclaw/pkg/logger" "github.com/sipeed/picoclaw/pkg/logger"
"github.com/sipeed/picoclaw/pkg/providers" "github.com/sipeed/picoclaw/pkg/providers"
"github.com/sipeed/picoclaw/pkg/providers/messageutil"
"github.com/sipeed/picoclaw/pkg/tools" "github.com/sipeed/picoclaw/pkg/tools"
) )
@ -623,6 +624,10 @@ func (e *ephemeralSessionStore) AddMessage(_, role, content string) {
} }
func (e *ephemeralSessionStore) AddFullMessage(_ string, msg providers.Message) { func (e *ephemeralSessionStore) AddFullMessage(_ string, msg providers.Message) {
if messageutil.IsTransientAssistantThoughtMessage(msg) {
return
}
e.mu.Lock() e.mu.Lock()
defer e.mu.Unlock() defer e.mu.Unlock()
e.history = append(e.history, msg) e.history = append(e.history, msg)
@ -652,6 +657,7 @@ func (e *ephemeralSessionStore) SetSummary(_, summary string) {
func (e *ephemeralSessionStore) SetHistory(_ string, history []providers.Message) { func (e *ephemeralSessionStore) SetHistory(_ string, history []providers.Message) {
e.mu.Lock() e.mu.Lock()
defer e.mu.Unlock() defer e.mu.Unlock()
history = messageutil.FilterInvalidHistoryMessages(history)
e.history = make([]providers.Message, len(history)) e.history = make([]providers.Message, len(history))
copy(e.history, history) copy(e.history, history)
e.truncateLocked() e.truncateLocked()

View file

@ -10,12 +10,14 @@ import (
"log" "log"
"os" "os"
"path/filepath" "path/filepath"
"sort"
"strings" "strings"
"sync" "sync"
"time" "time"
"github.com/sipeed/picoclaw/pkg/fileutil" "github.com/sipeed/picoclaw/pkg/fileutil"
"github.com/sipeed/picoclaw/pkg/providers" "github.com/sipeed/picoclaw/pkg/providers"
"github.com/sipeed/picoclaw/pkg/providers/messageutil"
) )
const ( const (
@ -405,12 +407,9 @@ func (s *JSONLStore) promoteAliasHistoryLocked(
} }
func (s *JSONLStore) sessionHasVisibleContentLocked(sessionKey string, meta SessionMeta) (bool, error) { func (s *JSONLStore) sessionHasVisibleContentLocked(sessionKey string, meta SessionMeta) (bool, error) {
if meta.Count-meta.Skip > 0 || strings.TrimSpace(meta.Summary) != "" { if strings.TrimSpace(meta.Summary) != "" {
return true, nil return true, nil
} }
if meta.Count != 0 || meta.Skip != 0 {
return false, nil
}
history, err := readMessages(s.jsonlPath(sessionKey), meta.Skip) history, err := readMessages(s.jsonlPath(sessionKey), meta.Skip)
if err != nil { if err != nil {
return false, err return false, err
@ -482,6 +481,9 @@ func readMessages(path string, skip int) ([]providers.Message, error) {
lineNum, filepath.Base(path), err) lineNum, filepath.Base(path), err)
continue continue
} }
if messageutil.IsTransientAssistantThoughtMessage(msg) {
continue
}
msgs = append(msgs, msg) msgs = append(msgs, msg)
} }
if scanner.Err() != nil { if scanner.Err() != nil {
@ -494,28 +496,44 @@ func readMessages(path string, skip int) ([]providers.Message, error) {
return msgs, nil return msgs, nil
} }
// countLines counts the total number of non-empty lines in a .jsonl file. // scanRetainedMessageLines returns the total number of non-empty raw JSONL
// Used by TruncateHistory to reconcile a stale meta.Count without // lines plus the raw line numbers that survive readMessages filtering.
// the overhead of unmarshaling every message. // TruncateHistory uses this to compute keepLast against retained messages
func countLines(path string) (int, error) { // while preserving the raw-line skip offset stored in metadata.
func scanRetainedMessageLines(path string) (int, []int, error) {
f, err := os.Open(path) f, err := os.Open(path)
if os.IsNotExist(err) { if os.IsNotExist(err) {
return 0, nil return 0, []int{}, nil
} }
if err != nil { if err != nil {
return 0, fmt.Errorf("memory: open jsonl: %w", err) return 0, nil, fmt.Errorf("memory: open jsonl: %w", err)
} }
defer f.Close() defer f.Close()
n := 0 rawCount := 0
retained := make([]int, 0)
scanner := bufio.NewScanner(f) scanner := bufio.NewScanner(f)
scanner.Buffer(make([]byte, 0, 64*1024), maxLineSize) scanner.Buffer(make([]byte, 0, 64*1024), maxLineSize)
for scanner.Scan() { for scanner.Scan() {
if len(scanner.Bytes()) > 0 { line := scanner.Bytes()
n++ if len(line) == 0 {
continue
} }
rawCount++
var msg providers.Message
if err := json.Unmarshal(line, &msg); err != nil {
continue
}
if messageutil.IsTransientAssistantThoughtMessage(msg) {
continue
}
retained = append(retained, rawCount)
} }
return n, scanner.Err() if err := scanner.Err(); err != nil {
return 0, nil, err
}
return rawCount, retained, nil
} }
func (s *JSONLStore) AddMessage( func (s *JSONLStore) AddMessage(
@ -535,6 +553,10 @@ func (s *JSONLStore) AddFullMessage(
// addMsg is the shared implementation for AddMessage and AddFullMessage. // addMsg is the shared implementation for AddMessage and AddFullMessage.
func (s *JSONLStore) addMsg(sessionKey string, msg providers.Message) error { func (s *JSONLStore) addMsg(sessionKey string, msg providers.Message) error {
if messageutil.IsTransientAssistantThoughtMessage(msg) {
return nil
}
l := s.sessionLock(sessionKey) l := s.sessionLock(sessionKey)
l.Lock() l.Lock()
defer l.Unlock() defer l.Unlock()
@ -655,24 +677,26 @@ func (s *JSONLStore) TruncateHistory(
return err return err
} }
// Always reconcile meta.Count with the actual line count on disk. rawCount, retainedRawLines, scanErr := scanRetainedMessageLines(s.jsonlPath(sessionKey))
// A crash between the JSONL append and the meta update in addMsg if scanErr != nil {
// leaves meta.Count stale (e.g. file has 101 lines but meta says return scanErr
// 100). Counting lines is cheap — no unmarshal, just a scan — and
// TruncateHistory is not a hot path, so always re-count.
n, countErr := countLines(s.jsonlPath(sessionKey))
if countErr != nil {
return countErr
} }
meta.Count = n meta.Count = rawCount
if meta.Skip > meta.Count {
if keepLast <= 0 {
meta.Skip = meta.Count meta.Skip = meta.Count
} else { }
effective := meta.Count - meta.Skip
if keepLast < effective { activeStart := sort.Search(len(retainedRawLines), func(i int) bool {
meta.Skip = meta.Count - keepLast return retainedRawLines[i] > meta.Skip
} })
activeRetainedCount := len(retainedRawLines) - activeStart
switch {
case keepLast <= 0 || activeRetainedCount == 0:
meta.Skip = meta.Count
case keepLast < activeRetainedCount:
activeRawLines := retainedRawLines[activeStart:]
meta.Skip = activeRawLines[activeRetainedCount-keepLast-1]
} }
meta.UpdatedAt = time.Now() meta.UpdatedAt = time.Now()
@ -684,6 +708,8 @@ func (s *JSONLStore) SetHistory(
sessionKey string, sessionKey string,
history []providers.Message, history []providers.Message,
) error { ) error {
history = messageutil.FilterInvalidHistoryMessages(history)
l := s.sessionLock(sessionKey) l := s.sessionLock(sessionKey)
l.Lock() l.Lock()
defer l.Unlock() defer l.Unlock()
@ -762,6 +788,8 @@ func (s *JSONLStore) Compact(
func (s *JSONLStore) rewriteJSONL( func (s *JSONLStore) rewriteJSONL(
sessionKey string, msgs []providers.Message, sessionKey string, msgs []providers.Message,
) error { ) error {
msgs = messageutil.FilterInvalidHistoryMessages(msgs)
var buf bytes.Buffer var buf bytes.Buffer
for i, msg := range msgs { for i, msg := range msgs {
line, err := json.Marshal(msg) line, err := json.Marshal(msg)

View file

@ -6,8 +6,10 @@ import (
"os" "os"
"path/filepath" "path/filepath"
"reflect" "reflect"
"strings"
"sync" "sync"
"testing" "testing"
"time"
"github.com/sipeed/picoclaw/pkg/providers" "github.com/sipeed/picoclaw/pkg/providers"
) )
@ -155,6 +157,27 @@ func TestAddFullMessage_ToolCallID(t *testing.T) {
} }
} }
func TestAddFullMessage_DropsTransientAssistantThought(t *testing.T) {
store := newTestStore(t)
ctx := context.Background()
err := store.AddFullMessage(ctx, "transient-thought", providers.Message{
Role: "assistant",
ReasoningContent: "internal chain of thought",
})
if err != nil {
t.Fatalf("AddFullMessage: %v", err)
}
history, err := store.GetHistory(ctx, "transient-thought")
if err != nil {
t.Fatalf("GetHistory: %v", err)
}
if len(history) != 0 {
t.Fatalf("expected transient thought to be discarded, got %d messages", len(history))
}
}
func TestGetHistory_EmptySession(t *testing.T) { func TestGetHistory_EmptySession(t *testing.T) {
store := newTestStore(t) store := newTestStore(t)
ctx := context.Background() ctx := context.Background()
@ -243,6 +266,46 @@ func TestSetSummary_GetSummary(t *testing.T) {
} }
} }
func TestSetHistory_DropsTransientAssistantThought(t *testing.T) {
store := newTestStore(t)
ctx := context.Background()
newHistory := []providers.Message{
{Role: "user", Content: "hello"},
{Role: "assistant", ReasoningContent: "internal chain of thought"},
{Role: "assistant", Content: "visible answer", ReasoningContent: "visible thought"},
}
err := store.SetHistory(ctx, "replace", newHistory)
if err != nil {
t.Fatalf("SetHistory: %v", err)
}
history, err := store.GetHistory(ctx, "replace")
if err != nil {
t.Fatalf("GetHistory: %v", err)
}
if len(history) != 2 {
t.Fatalf("expected transient thought to be removed, got %d messages", len(history))
}
if history[0].Role != "user" || history[0].Content != "hello" {
t.Fatalf("history[0] = %+v, want user/hello", history[0])
}
if history[1].Role != "assistant" || history[1].Content != "visible answer" ||
history[1].ReasoningContent != "visible thought" {
t.Fatalf("history[1] = %+v, want assistant visible answer with reasoning", history[1])
}
data, err := os.ReadFile(store.jsonlPath("replace"))
if err != nil {
t.Fatalf("ReadFile(jsonl): %v", err)
}
lines := strings.Split(strings.TrimSpace(string(data)), "\n")
if len(lines) != 2 {
t.Fatalf("jsonl line count = %d, want 2", len(lines))
}
}
func TestSessionMetaScopeAndAliasesPersist(t *testing.T) { func TestSessionMetaScopeAndAliasesPersist(t *testing.T) {
store := newTestStore(t) store := newTestStore(t)
ctx := context.Background() ctx := context.Background()
@ -733,6 +796,56 @@ func TestTruncateHistory_StaleMetaCount(t *testing.T) {
} }
} }
func TestTruncateHistory_IgnoresTransientThoughtForKeepLast(t *testing.T) {
store := newTestStore(t)
ctx := context.Background()
sessionKey := "transient-keep-last"
now := time.Now()
rawJSONL := strings.Join([]string{
`{"role":"user","content":"a"}`,
`{"role":"assistant","content":"b"}`,
`{"role":"assistant","content":"","reasoning_content":"dangling thought"}`,
`{"role":"user","content":"c"}`,
`{"role":"assistant","content":"d"}`,
}, "\n") + "\n"
if err := os.WriteFile(store.jsonlPath(sessionKey), []byte(rawJSONL), 0o644); err != nil {
t.Fatalf("WriteFile(jsonl): %v", err)
}
if err := store.writeMeta(sessionKey, SessionMeta{
Key: sessionKey,
Count: 5,
Skip: 0,
CreatedAt: now,
UpdatedAt: now,
}); err != nil {
t.Fatalf("writeMeta: %v", err)
}
if err := store.TruncateHistory(ctx, sessionKey, 2); err != nil {
t.Fatalf("TruncateHistory: %v", err)
}
history, err := store.GetHistory(ctx, sessionKey)
if err != nil {
t.Fatalf("GetHistory: %v", err)
}
if len(history) != 2 {
t.Fatalf("expected 2 retained messages, got %d", len(history))
}
if history[0].Content != "c" || history[1].Content != "d" {
t.Fatalf("kept history = %+v, want c,d", history)
}
meta, err := store.readMeta(sessionKey)
if err != nil {
t.Fatalf("readMeta: %v", err)
}
if meta.Skip != 2 {
t.Fatalf("meta.Skip = %d, want 2 raw lines skipped", meta.Skip)
}
}
func TestCrashRecovery_PartialLine(t *testing.T) { func TestCrashRecovery_PartialLine(t *testing.T) {
store := newTestStore(t) store := newTestStore(t)
ctx := context.Background() ctx := context.Background()

View file

@ -178,7 +178,7 @@ func CreateProviderFromConfig(cfg *config.ModelConfig) (LLMProvider, string, err
if apiBase == "" { if apiBase == "" {
apiBase = getDefaultAPIBase(protocol) apiBase = getDefaultAPIBase(protocol)
} }
return NewHTTPProviderWithMaxTokensFieldAndRequestTimeout( provider := NewHTTPProviderWithMaxTokensFieldAndRequestTimeout(
cfg.APIKey(), cfg.APIKey(),
apiBase, apiBase,
cfg.Proxy, cfg.Proxy,
@ -187,7 +187,9 @@ func CreateProviderFromConfig(cfg *config.ModelConfig) (LLMProvider, string, err
cfg.RequestTimeout, cfg.RequestTimeout,
cfg.ExtraBody, cfg.ExtraBody,
cfg.CustomHeaders, cfg.CustomHeaders,
), modelID, nil )
provider.SetProviderName(protocol)
return provider, modelID, nil
case "azure", "azure-openai": case "azure", "azure-openai":
// Azure OpenAI uses deployment-based URLs, api-key header auth, // Azure OpenAI uses deployment-based URLs, api-key header auth,
@ -257,7 +259,7 @@ func CreateProviderFromConfig(cfg *config.ModelConfig) (LLMProvider, string, err
if apiBase == "" { if apiBase == "" {
apiBase = getDefaultAPIBase(protocol) apiBase = getDefaultAPIBase(protocol)
} }
return NewHTTPProviderWithMaxTokensFieldAndRequestTimeout( provider := NewHTTPProviderWithMaxTokensFieldAndRequestTimeout(
cfg.APIKey(), cfg.APIKey(),
apiBase, apiBase,
cfg.Proxy, cfg.Proxy,
@ -266,7 +268,9 @@ func CreateProviderFromConfig(cfg *config.ModelConfig) (LLMProvider, string, err
cfg.RequestTimeout, cfg.RequestTimeout,
cfg.ExtraBody, cfg.ExtraBody,
cfg.CustomHeaders, cfg.CustomHeaders,
), modelID, nil )
provider.SetProviderName(protocol)
return provider, modelID, nil
case "gemini": case "gemini":
if cfg.APIKey() == "" && cfg.APIBase == "" { if cfg.APIKey() == "" && cfg.APIBase == "" {
@ -302,7 +306,7 @@ func CreateProviderFromConfig(cfg *config.ModelConfig) (LLMProvider, string, err
if _, ok := extraBody["reasoning_split"]; !ok { if _, ok := extraBody["reasoning_split"]; !ok {
extraBody["reasoning_split"] = true extraBody["reasoning_split"] = true
} }
return NewHTTPProviderWithMaxTokensFieldAndRequestTimeout( provider := NewHTTPProviderWithMaxTokensFieldAndRequestTimeout(
cfg.APIKey(), cfg.APIKey(),
apiBase, apiBase,
cfg.Proxy, cfg.Proxy,
@ -311,7 +315,9 @@ func CreateProviderFromConfig(cfg *config.ModelConfig) (LLMProvider, string, err
cfg.RequestTimeout, cfg.RequestTimeout,
extraBody, extraBody,
cfg.CustomHeaders, cfg.CustomHeaders,
), modelID, nil )
provider.SetProviderName(protocol)
return provider, modelID, nil
case "anthropic": case "anthropic":
if cfg.AuthMethod == "oauth" || cfg.AuthMethod == "token" { if cfg.AuthMethod == "oauth" || cfg.AuthMethod == "token" {
@ -330,7 +336,7 @@ func CreateProviderFromConfig(cfg *config.ModelConfig) (LLMProvider, string, err
if cfg.APIKey() == "" { if cfg.APIKey() == "" {
return nil, "", fmt.Errorf("api_key is required for anthropic protocol (model: %s)", cfg.Model) return nil, "", fmt.Errorf("api_key is required for anthropic protocol (model: %s)", cfg.Model)
} }
return NewHTTPProviderWithMaxTokensFieldAndRequestTimeout( provider := NewHTTPProviderWithMaxTokensFieldAndRequestTimeout(
cfg.APIKey(), cfg.APIKey(),
apiBase, apiBase,
cfg.Proxy, cfg.Proxy,
@ -339,7 +345,9 @@ func CreateProviderFromConfig(cfg *config.ModelConfig) (LLMProvider, string, err
cfg.RequestTimeout, cfg.RequestTimeout,
cfg.ExtraBody, cfg.ExtraBody,
cfg.CustomHeaders, cfg.CustomHeaders,
), modelID, nil )
provider.SetProviderName(protocol)
return provider, modelID, nil
case "anthropic-messages": case "anthropic-messages":
// Anthropic Messages API with native format (HTTP-based, no SDK) // Anthropic Messages API with native format (HTTP-based, no SDK)

View file

@ -77,3 +77,10 @@ func (p *HTTPProvider) GetDefaultModel() string {
func (p *HTTPProvider) SupportsNativeSearch() bool { func (p *HTTPProvider) SupportsNativeSearch() bool {
return p.delegate.SupportsNativeSearch() return p.delegate.SupportsNativeSearch()
} }
func (p *HTTPProvider) SetProviderName(providerName string) {
if p == nil || p.delegate == nil {
return
}
p.delegate.SetProviderName(providerName)
}

View file

@ -0,0 +1,38 @@
package messageutil
import (
"strings"
"github.com/sipeed/picoclaw/pkg/providers/protocoltypes"
)
// IsTransientAssistantThoughtMessage reports whether msg is an invalid
// reasoning-only assistant history record. These "hanging" thought messages
// are not a canonical persisted format and should be discarded instead of
// replayed or reconstructed.
func IsTransientAssistantThoughtMessage(msg protocoltypes.Message) bool {
return msg.Role == "assistant" &&
strings.TrimSpace(msg.Content) == "" &&
strings.TrimSpace(msg.ReasoningContent) != "" &&
len(msg.ToolCalls) == 0 &&
len(msg.Media) == 0 &&
len(msg.Attachments) == 0 &&
strings.TrimSpace(msg.ToolCallID) == ""
}
// FilterInvalidHistoryMessages removes invalid persisted history records such
// as transient assistant thought-only messages.
func FilterInvalidHistoryMessages(history []protocoltypes.Message) []protocoltypes.Message {
if len(history) == 0 {
return []protocoltypes.Message{}
}
filtered := make([]protocoltypes.Message, 0, len(history))
for _, msg := range history {
if IsTransientAssistantThoughtMessage(msg) {
continue
}
filtered = append(filtered, msg)
}
return filtered
}

View file

@ -15,6 +15,7 @@ import (
"time" "time"
"github.com/sipeed/picoclaw/pkg/providers/common" "github.com/sipeed/picoclaw/pkg/providers/common"
"github.com/sipeed/picoclaw/pkg/providers/messageutil"
"github.com/sipeed/picoclaw/pkg/providers/protocoltypes" "github.com/sipeed/picoclaw/pkg/providers/protocoltypes"
) )
@ -34,6 +35,7 @@ type (
type Provider struct { type Provider struct {
apiKey string apiKey string
apiBase string apiBase string
providerName string
maxTokensField string // Field name for max tokens (e.g., "max_completion_tokens" for o1/glm models) maxTokensField string // Field name for max tokens (e.g., "max_completion_tokens" for o1/glm models)
httpClient *http.Client httpClient *http.Client
extraBody map[string]any // Additional fields to inject into request body extraBody map[string]any // Additional fields to inject into request body
@ -95,6 +97,12 @@ func WithCustomHeaders(customHeaders map[string]string) Option {
} }
} }
func WithProviderName(providerName string) Option {
return func(p *Provider) {
p.providerName = strings.ToLower(strings.TrimSpace(providerName))
}
}
func NewProvider(apiKey, apiBase, proxy string, opts ...Option) *Provider { func NewProvider(apiKey, apiBase, proxy string, opts ...Option) *Provider {
p := &Provider{ p := &Provider{
apiKey: apiKey, apiKey: apiKey,
@ -136,7 +144,7 @@ func (p *Provider) buildRequestBody(
requestBody := map[string]any{ requestBody := map[string]any{
"model": model, "model": model,
"messages": common.SerializeMessages(messages), "messages": common.SerializeMessages(p.prepareMessagesForRequest(messages)),
} }
// When fallback uses a different provider (e.g. DeepSeek), that provider must not inject web_search_preview. // When fallback uses a different provider (e.g. DeepSeek), that provider must not inject web_search_preview.
@ -196,6 +204,111 @@ func (p *Provider) applyCustomHeaders(req *http.Request) {
} }
} }
func (p *Provider) SetProviderName(providerName string) {
p.providerName = strings.ToLower(strings.TrimSpace(providerName))
}
func (p *Provider) prepareMessagesForRequest(messages []Message) []Message {
if len(messages) == 0 {
return nil
}
if p.isDeepSeekReasoningProvider() {
return filterDeepSeekReasoningMessages(messages)
}
return stripReasoningMessages(messages)
}
func (p *Provider) isDeepSeekReasoningProvider() bool {
return p.providerName == "deepseek" || isDeepSeekHost(p.apiBase)
}
func isDeepSeekHost(apiBase string) bool {
parsed, err := url.Parse(strings.TrimSpace(apiBase))
if err != nil {
return false
}
host := strings.ToLower(strings.TrimSpace(parsed.Hostname()))
return host == "deepseek.com" || strings.HasSuffix(host, ".deepseek.com")
}
func filterDeepSeekReasoningMessages(messages []Message) []Message {
out := make([]Message, 0, len(messages))
start := 0
flush := func(end int) {
if end <= start {
return
}
out = append(out, filterDeepSeekReasoningTurn(messages[start:end])...)
start = end
}
for i := 1; i < len(messages); i++ {
if messages[i].Role == "user" {
flush(i)
}
}
flush(len(messages))
return out
}
func filterDeepSeekReasoningTurn(messages []Message) []Message {
hasToolInteraction := false
for _, msg := range messages {
if msg.Role == "tool" || (msg.Role == "assistant" && len(msg.ToolCalls) > 0) {
hasToolInteraction = true
break
}
}
out := make([]Message, 0, len(messages))
for _, msg := range messages {
if messageutil.IsTransientAssistantThoughtMessage(msg) {
continue
}
cloned := msg
if cloned.Role == "assistant" && strings.TrimSpace(cloned.ReasoningContent) != "" && !hasToolInteraction {
cloned.ReasoningContent = ""
}
if assistantMessageEmpty(cloned) {
continue
}
out = append(out, cloned)
}
return out
}
func stripReasoningMessages(messages []Message) []Message {
out := make([]Message, 0, len(messages))
for _, msg := range messages {
if messageutil.IsTransientAssistantThoughtMessage(msg) {
continue
}
cloned := msg
cloned.ReasoningContent = ""
if assistantMessageEmpty(cloned) {
continue
}
out = append(out, cloned)
}
return out
}
func assistantMessageEmpty(msg Message) bool {
return msg.Role == "assistant" &&
strings.TrimSpace(msg.Content) == "" &&
strings.TrimSpace(msg.ReasoningContent) == "" &&
len(msg.ToolCalls) == 0 &&
len(msg.Media) == 0 &&
len(msg.Attachments) == 0 &&
strings.TrimSpace(msg.ToolCallID) == ""
}
func (p *Provider) Chat( func (p *Provider) Chat(
ctx context.Context, ctx context.Context,
messages []Message, messages []Message,

View file

@ -202,7 +202,7 @@ func TestProviderChat_ParsesReasoningContent(t *testing.T) {
} }
} }
func TestProviderChat_PreservesReasoningContentInHistory(t *testing.T) { func TestProviderChat_StripsReasoningContentForNonDeepSeekHistory(t *testing.T) {
var requestBody map[string]any var requestBody map[string]any
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@ -225,8 +225,6 @@ func TestProviderChat_PreservesReasoningContentInHistory(t *testing.T) {
p := NewProvider("key", server.URL, "") p := NewProvider("key", server.URL, "")
// Simulate a multi-turn conversation where the assistant's previous
// reply included reasoning_content (e.g. from kimi-k2.5).
messages := []Message{ messages := []Message{
{Role: "user", Content: "What is 1+1?"}, {Role: "user", Content: "What is 1+1?"},
{Role: "assistant", Content: "2", ReasoningContent: "Let me think... 1+1=2"}, {Role: "assistant", Content: "2", ReasoningContent: "Let me think... 1+1=2"},
@ -238,7 +236,6 @@ func TestProviderChat_PreservesReasoningContentInHistory(t *testing.T) {
t.Fatalf("Chat() error = %v", err) t.Fatalf("Chat() error = %v", err)
} }
// Verify reasoning_content is preserved in the serialized request.
reqMessages, ok := requestBody["messages"].([]any) reqMessages, ok := requestBody["messages"].([]any)
if !ok { if !ok {
t.Fatalf("messages is not []any: %T", requestBody["messages"]) t.Fatalf("messages is not []any: %T", requestBody["messages"])
@ -247,11 +244,288 @@ func TestProviderChat_PreservesReasoningContentInHistory(t *testing.T) {
if !ok { if !ok {
t.Fatalf("assistant message is not map[string]any: %T", reqMessages[1]) t.Fatalf("assistant message is not map[string]any: %T", reqMessages[1])
} }
if assistantMsg["reasoning_content"] != "Let me think... 1+1=2" { if _, exists := assistantMsg["reasoning_content"]; exists {
t.Errorf("reasoning_content not preserved in request, got %v", assistantMsg["reasoning_content"]) t.Fatalf(
"reasoning_content should be stripped for non-DeepSeek providers, got %v",
assistantMsg["reasoning_content"],
)
} }
} }
func TestProviderChat_DeepSeekOmitsReasoningContentForNonToolTurnHistory(t *testing.T) {
var requestBody map[string]any
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if err := json.NewDecoder(r.Body).Decode(&requestBody); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
resp := map[string]any{
"choices": []map[string]any{
{
"message": map[string]any{"content": "ok"},
"finish_reason": "stop",
},
},
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(resp)
}))
defer server.Close()
p := NewProvider("key", server.URL, "")
p.apiBase = "https://api.deepseek.com/v1"
p.httpClient = &http.Client{
Transport: roundTripperFunc(func(r *http.Request) (*http.Response, error) {
r.URL, _ = url.Parse(server.URL + r.URL.Path)
return http.DefaultTransport.RoundTrip(r)
}),
}
messages := []Message{
{Role: "user", Content: "What is 1+1?"},
{Role: "assistant", Content: "2", ReasoningContent: "Let me think... 1+1=2"},
{Role: "user", Content: "What about 2+2?"},
}
_, err := p.Chat(t.Context(), messages, nil, "deepseek-v4-flash", nil)
if err != nil {
t.Fatalf("Chat() error = %v", err)
}
reqMessages, ok := requestBody["messages"].([]any)
if !ok {
t.Fatalf("messages is not []any: %T", requestBody["messages"])
}
assistantMsg, ok := reqMessages[1].(map[string]any)
if !ok {
t.Fatalf("assistant message is not map[string]any: %T", reqMessages[1])
}
if _, exists := assistantMsg["reasoning_content"]; exists {
t.Fatalf(
"reasoning_content should be omitted for DeepSeek non-tool turns, got %v",
assistantMsg["reasoning_content"],
)
}
}
func TestProviderChat_DeepSeekPreservesReasoningContentForToolTurnHistory(t *testing.T) {
var requestBody map[string]any
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if err := json.NewDecoder(r.Body).Decode(&requestBody); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
resp := map[string]any{
"choices": []map[string]any{
{
"message": map[string]any{"content": "ok"},
"finish_reason": "stop",
},
},
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(resp)
}))
defer server.Close()
p := NewProvider("key", server.URL, "")
p.SetProviderName("deepseek")
messages := []Message{
{Role: "user", Content: "How's the weather tomorrow?"},
{
Role: "assistant",
Content: "Let me check the date first.",
ReasoningContent: "I need tomorrow's date before checking the weather.",
ToolCalls: []ToolCall{{
ID: "call_1",
Type: "function",
Function: &FunctionCall{
Name: "get_date",
Arguments: "{}",
},
}},
},
{Role: "tool", ToolCallID: "call_1", Content: "2026-04-24"},
{
Role: "assistant",
Content: "Tomorrow is 2026-04-25.",
ReasoningContent: "Now I can share the final answer.",
},
{Role: "user", Content: "What about Guangzhou?"},
}
_, err := p.Chat(t.Context(), messages, nil, "deepseek-v4-flash", nil)
if err != nil {
t.Fatalf("Chat() error = %v", err)
}
reqMessages, ok := requestBody["messages"].([]any)
if !ok {
t.Fatalf("messages is not []any: %T", requestBody["messages"])
}
if len(reqMessages) != len(messages) {
t.Fatalf("len(messages) = %d, want %d", len(reqMessages), len(messages))
}
firstAssistant, ok := reqMessages[1].(map[string]any)
if !ok {
t.Fatalf("first assistant message is not map[string]any: %T", reqMessages[1])
}
if firstAssistant["reasoning_content"] != "I need tomorrow's date before checking the weather." {
t.Fatalf("first assistant reasoning_content = %v, want preserved", firstAssistant["reasoning_content"])
}
finalAssistant, ok := reqMessages[3].(map[string]any)
if !ok {
t.Fatalf("final assistant message is not map[string]any: %T", reqMessages[3])
}
if finalAssistant["reasoning_content"] != "Now I can share the final answer." {
t.Fatalf("final assistant reasoning_content = %v, want preserved", finalAssistant["reasoning_content"])
}
}
func TestProviderChat_HistoryCanonicalizationMatrix(t *testing.T) {
baseMessages := []Message{
{Role: "user", Content: "turn1"},
{Role: "assistant", Content: "plain visible", ReasoningContent: "plain thought"},
{Role: "user", Content: "turn2"},
{
Role: "assistant",
Content: "",
ReasoningContent: "tool thought",
ToolCalls: []ToolCall{{
ID: "call_read_file",
Type: "function",
Function: &FunctionCall{
Name: "read_file",
Arguments: `{"path":"README.md"}`,
},
}},
},
{Role: "tool", ToolCallID: "call_read_file", Content: "file content"},
{Role: "user", Content: "turn3"},
{
Role: "assistant",
Content: "tool visible only",
ToolCalls: []ToolCall{{
ID: "call_list_dir",
Type: "function",
Function: &FunctionCall{
Name: "list_dir",
Arguments: `{"path":"."}`,
},
}},
},
{Role: "tool", ToolCallID: "call_list_dir", Content: "dir listing"},
{Role: "user", Content: "turn4"},
{
Role: "assistant",
Content: "tool visible and thought",
ReasoningContent: "tool mixed thought",
ToolCalls: []ToolCall{{
ID: "call_exec",
Type: "function",
Function: &FunctionCall{
Name: "exec",
Arguments: `{"command":"pwd"}`,
},
}},
},
{Role: "tool", ToolCallID: "call_exec", Content: "pwd output"},
{Role: "user", Content: "current turn"},
}
captureRequestMessages := func(t *testing.T, providerName string) []map[string]any {
t.Helper()
var requestBody map[string]any
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if err := json.NewDecoder(r.Body).Decode(&requestBody); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
resp := map[string]any{
"choices": []map[string]any{
{
"message": map[string]any{"content": "ok"},
"finish_reason": "stop",
},
},
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(resp)
}))
defer server.Close()
p := NewProvider("key", server.URL, "")
if providerName != "" {
p.SetProviderName(providerName)
}
_, err := p.Chat(t.Context(), baseMessages, nil, "test-model", nil)
if err != nil {
t.Fatalf("Chat() error = %v", err)
}
rawMessages, ok := requestBody["messages"].([]any)
if !ok {
t.Fatalf("messages is not []any: %T", requestBody["messages"])
}
out := make([]map[string]any, 0, len(rawMessages))
for i, raw := range rawMessages {
msg, ok := raw.(map[string]any)
if !ok {
t.Fatalf("messages[%d] is %T, want map[string]any", i, raw)
}
out = append(out, msg)
}
return out
}
t.Run("deepseek", func(t *testing.T) {
msgs := captureRequestMessages(t, "deepseek")
if len(msgs) != len(baseMessages) {
t.Fatalf("len(messages) = %d, want %d", len(msgs), len(baseMessages))
}
if _, ok := msgs[1]["reasoning_content"]; ok {
t.Fatalf(
"turn1 reasoning_content should be stripped for DeepSeek non-tool turn, got %v",
msgs[1]["reasoning_content"],
)
}
if msgs[3]["reasoning_content"] != "tool thought" {
t.Fatalf("turn2 reasoning_content = %v, want preserved", msgs[3]["reasoning_content"])
}
if _, ok := msgs[6]["reasoning_content"]; ok {
t.Fatalf("turn3 reasoning_content should be absent, got %v", msgs[6]["reasoning_content"])
}
if msgs[9]["reasoning_content"] != "tool mixed thought" {
t.Fatalf("turn4 reasoning_content = %v, want preserved", msgs[9]["reasoning_content"])
}
if msgs[9]["content"] != "tool visible and thought" {
t.Fatalf("turn4 content = %v, want preserved", msgs[9]["content"])
}
})
t.Run("non-deepseek", func(t *testing.T) {
msgs := captureRequestMessages(t, "")
for i, msg := range msgs {
if _, ok := msg["reasoning_content"]; ok {
t.Fatalf(
"messages[%d] reasoning_content should be stripped for non-DeepSeek providers, got %v",
i,
msg["reasoning_content"],
)
}
}
})
}
func TestProviderChat_HTTPError(t *testing.T) { func TestProviderChat_HTTPError(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
http.Error(w, "bad request", http.StatusBadRequest) http.Error(w, "bad request", http.StatusBadRequest)

View file

@ -9,6 +9,7 @@ import (
"time" "time"
"github.com/sipeed/picoclaw/pkg/providers" "github.com/sipeed/picoclaw/pkg/providers"
"github.com/sipeed/picoclaw/pkg/providers/messageutil"
) )
type Session struct { type Session struct {
@ -69,6 +70,10 @@ func (sm *SessionManager) AddMessage(sessionKey, role, content string) {
// AddFullMessage adds a complete message with tool calls and tool call ID to the session. // AddFullMessage adds a complete message with tool calls and tool call ID to the session.
// This is used to save the full conversation flow including tool calls and tool results. // This is used to save the full conversation flow including tool calls and tool results.
func (sm *SessionManager) AddFullMessage(sessionKey string, msg providers.Message) { func (sm *SessionManager) AddFullMessage(sessionKey string, msg providers.Message) {
if messageutil.IsTransientAssistantThoughtMessage(msg) {
return
}
sm.mu.Lock() sm.mu.Lock()
defer sm.mu.Unlock() defer sm.mu.Unlock()
@ -196,8 +201,7 @@ func (sm *SessionManager) Save(key string) error {
Updated: stored.Updated, Updated: stored.Updated,
} }
if len(stored.Messages) > 0 { if len(stored.Messages) > 0 {
snapshot.Messages = make([]providers.Message, len(stored.Messages)) snapshot.Messages = messageutil.FilterInvalidHistoryMessages(stored.Messages)
copy(snapshot.Messages, stored.Messages)
} else { } else {
snapshot.Messages = []providers.Message{} snapshot.Messages = []providers.Message{}
} }
@ -270,6 +274,7 @@ func (sm *SessionManager) loadSessions() error {
if err := json.Unmarshal(data, &session); err != nil { if err := json.Unmarshal(data, &session); err != nil {
continue continue
} }
session.Messages = messageutil.FilterInvalidHistoryMessages(session.Messages)
sm.sessions[session.Key] = &session sm.sessions[session.Key] = &session
} }
@ -290,6 +295,7 @@ func (sm *SessionManager) SetHistory(key string, history []providers.Message) {
session, ok := sm.sessions[key] session, ok := sm.sessions[key]
if ok { if ok {
history = messageutil.FilterInvalidHistoryMessages(history)
// Create a deep copy to strictly isolate internal state // Create a deep copy to strictly isolate internal state
// from the caller's slice. // from the caller's slice.
msgs := make([]providers.Message, len(history)) msgs := make([]providers.Message, len(history))

View file

@ -16,6 +16,7 @@ import (
"github.com/sipeed/picoclaw/pkg/config" "github.com/sipeed/picoclaw/pkg/config"
"github.com/sipeed/picoclaw/pkg/memory" "github.com/sipeed/picoclaw/pkg/memory"
"github.com/sipeed/picoclaw/pkg/providers" "github.com/sipeed/picoclaw/pkg/providers"
"github.com/sipeed/picoclaw/pkg/providers/messageutil"
"github.com/sipeed/picoclaw/pkg/session" "github.com/sipeed/picoclaw/pkg/session"
"github.com/sipeed/picoclaw/pkg/utils" "github.com/sipeed/picoclaw/pkg/utils"
) )
@ -49,6 +50,7 @@ type sessionListItem struct {
type sessionChatMessage struct { type sessionChatMessage struct {
Role string `json:"role"` Role string `json:"role"`
Content string `json:"content"` Content string `json:"content"`
Kind string `json:"kind,omitempty"`
Media []string `json:"media,omitempty"` Media []string `json:"media,omitempty"`
Attachments []sessionChatAttachment `json:"attachments,omitempty"` Attachments []sessionChatAttachment `json:"attachments,omitempty"`
} }
@ -154,6 +156,9 @@ func (h *Handler) readSessionMessages(path string, skip int) ([]providers.Messag
if err := json.Unmarshal(line, &msg); err != nil { if err := json.Unmarshal(line, &msg); err != nil {
continue continue
} }
if messageutil.IsTransientAssistantThoughtMessage(msg) {
continue
}
msgs = append(msgs, msg) msgs = append(msgs, msg)
} }
if err := scanner.Err(); err != nil { if err := scanner.Err(); err != nil {
@ -474,6 +479,18 @@ func sessionChatMessagePreview(msg sessionChatMessage) string {
} }
func visibleSessionMessages(messages []providers.Message, toolFeedbackMaxArgsLength int) []sessionChatMessage { func visibleSessionMessages(messages []providers.Message, toolFeedbackMaxArgsLength int) []sessionChatMessage {
return sessionTranscriptMessages(messages, toolFeedbackMaxArgsLength, false)
}
func detailSessionMessages(messages []providers.Message, toolFeedbackMaxArgsLength int) []sessionChatMessage {
return sessionTranscriptMessages(messages, toolFeedbackMaxArgsLength, true)
}
func sessionTranscriptMessages(
messages []providers.Message,
toolFeedbackMaxArgsLength int,
includeThoughts bool,
) []sessionChatMessage {
transcript := make([]sessionChatMessage, 0, len(messages)) transcript := make([]sessionChatMessage, 0, len(messages))
for _, msg := range messages { for _, msg := range messages {
@ -495,11 +512,14 @@ func visibleSessionMessages(messages []providers.Message, toolFeedbackMaxArgsLen
} }
case "assistant": case "assistant":
// Reasoning-only assistant messages are transient display artifacts and if messageutil.IsTransientAssistantThoughtMessage(msg) {
// should not be restored from session history.
if assistantMessageTransientThought(msg) {
continue continue
} }
if includeThoughts {
if thoughtMsg, ok := assistantThoughtMessage(msg); ok {
transcript = append(transcript, thoughtMsg)
}
}
toolSummaryMessages := visibleAssistantToolSummaryMessages(msg.ToolCalls, toolFeedbackMaxArgsLength) toolSummaryMessages := visibleAssistantToolSummaryMessages(msg.ToolCalls, toolFeedbackMaxArgsLength)
if len(toolSummaryMessages) > 0 { if len(toolSummaryMessages) > 0 {
@ -681,18 +701,25 @@ func sessionAttachmentType(attachment providers.Attachment) string {
} }
} }
func assistantMessageTransientThought(msg providers.Message) bool {
return strings.TrimSpace(msg.Content) == "" &&
strings.TrimSpace(msg.ReasoningContent) != "" &&
len(msg.ToolCalls) == 0 &&
len(msg.Media) == 0 &&
len(msg.Attachments) == 0
}
func assistantMessageInternalOnly(msg providers.Message) bool { func assistantMessageInternalOnly(msg providers.Message) bool {
return strings.TrimSpace(msg.Content) == handledToolResponseSummaryText return strings.TrimSpace(msg.Content) == handledToolResponseSummaryText
} }
func assistantThoughtMessage(msg providers.Message) (sessionChatMessage, bool) {
reasoning := strings.TrimSpace(msg.ReasoningContent)
if reasoning == "" {
return sessionChatMessage{}, false
}
if reasoning == strings.TrimSpace(msg.Content) {
return sessionChatMessage{}, false
}
return sessionChatMessage{
Role: "assistant",
Content: reasoning,
Kind: "thought",
}, true
}
func visibleAssistantToolSummaryMessages( func visibleAssistantToolSummaryMessages(
toolCalls []providers.ToolCall, toolCalls []providers.ToolCall,
toolFeedbackMaxArgsLength int, toolFeedbackMaxArgsLength int,
@ -986,7 +1013,7 @@ func (h *Handler) handleGetSession(w http.ResponseWriter, r *http.Request) {
} }
} }
messages := visibleSessionMessages(sess.Messages, toolFeedbackMaxArgsLength) messages := detailSessionMessages(sess.Messages, toolFeedbackMaxArgsLength)
w.Header().Set("Content-Type", "application/json") w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]any{ json.NewEncoder(w).Encode(map[string]any{

View file

@ -8,6 +8,7 @@ import (
"path/filepath" "path/filepath"
"strings" "strings"
"testing" "testing"
"time"
"github.com/sipeed/picoclaw/pkg/config" "github.com/sipeed/picoclaw/pkg/config"
"github.com/sipeed/picoclaw/pkg/memory" "github.com/sipeed/picoclaw/pkg/memory"
@ -101,6 +102,64 @@ func TestHandleListSessions_JSONLStorage(t *testing.T) {
} }
} }
func TestHandleListSessions_TransientThoughtDoesNotInflateMessageCount(t *testing.T) {
configPath, cleanup := setupOAuthTestEnv(t)
defer cleanup()
dir := sessionsTestDir(t, configPath)
sessionKey := legacyPicoSessionPrefix + "history-jsonl-transient"
base := filepath.Join(dir, sanitizeSessionKey(sessionKey))
now := time.Now().UTC()
rawJSONL := strings.Join([]string{
`{"role":"user","content":"keep me"}`,
`{"role":"assistant","content":"","reasoning_content":"dangling thought"}`,
`{"role":"assistant","content":"and me"}`,
}, "\n") + "\n"
if err := os.WriteFile(base+".jsonl", []byte(rawJSONL), 0o644); err != nil {
t.Fatalf("WriteFile(jsonl) error = %v", err)
}
metaData, err := json.Marshal(memory.SessionMeta{
Key: sessionKey,
Count: 3,
Skip: 0,
CreatedAt: now,
UpdatedAt: now,
})
if err != nil {
t.Fatalf("Marshal(meta) error = %v", err)
}
if err := os.WriteFile(base+".meta.json", metaData, 0o644); err != nil {
t.Fatalf("WriteFile(meta) error = %v", err)
}
h := NewHandler(configPath)
mux := http.NewServeMux()
h.RegisterRoutes(mux)
rec := httptest.NewRecorder()
req := httptest.NewRequest(http.MethodGet, "/api/sessions", nil)
mux.ServeHTTP(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("status = %d, want %d, body=%s", rec.Code, http.StatusOK, rec.Body.String())
}
var items []sessionListItem
if err := json.Unmarshal(rec.Body.Bytes(), &items); err != nil {
t.Fatalf("Unmarshal() error = %v", err)
}
if len(items) != 1 {
t.Fatalf("len(items) = %d, want 1", len(items))
}
if items[0].ID != "history-jsonl-transient" {
t.Fatalf("items[0].ID = %q, want %q", items[0].ID, "history-jsonl-transient")
}
if items[0].MessageCount != 2 {
t.Fatalf("items[0].MessageCount = %d, want 2 after dropping transient thought", items[0].MessageCount)
}
}
func TestHandleListSessions_TitleUsesFirstUserMessage(t *testing.T) { func TestHandleListSessions_TitleUsesFirstUserMessage(t *testing.T) {
configPath, cleanup := setupOAuthTestEnv(t) configPath, cleanup := setupOAuthTestEnv(t)
defer cleanup() defer cleanup()
@ -423,7 +482,7 @@ func TestHandleSessions_JSONLScopeDiscovery(t *testing.T) {
} }
} }
func TestHandleGetSession_OmitsTransientThoughtMessages(t *testing.T) { func TestHandleGetSession_SkipsTransientThoughtMessages(t *testing.T) {
configPath, cleanup := setupOAuthTestEnv(t) configPath, cleanup := setupOAuthTestEnv(t)
defer cleanup() defer cleanup()
@ -460,6 +519,7 @@ func TestHandleGetSession_OmitsTransientThoughtMessages(t *testing.T) {
Messages []struct { Messages []struct {
Role string `json:"role"` Role string `json:"role"`
Content string `json:"content"` Content string `json:"content"`
Kind string `json:"kind"`
} `json:"messages"` } `json:"messages"`
} }
if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil { if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil {
@ -476,6 +536,180 @@ func TestHandleGetSession_OmitsTransientThoughtMessages(t *testing.T) {
} }
} }
func TestHandleGetSession_ReconstructsThoughtFromAssistantReasoningContent(t *testing.T) {
configPath, cleanup := setupOAuthTestEnv(t)
defer cleanup()
dir := sessionsTestDir(t, configPath)
store, err := memory.NewJSONLStore(dir)
if err != nil {
t.Fatalf("NewJSONLStore() error = %v", err)
}
sessionKey := picoSessionPrefix + "detail-reasoning-content"
for _, msg := range []providers.Message{
{Role: "user", Content: "hello"},
{Role: "assistant", Content: "final visible answer", ReasoningContent: "internal chain of thought"},
} {
if err := store.AddFullMessage(nil, sessionKey, msg); err != nil {
t.Fatalf("AddFullMessage() error = %v", err)
}
}
h := NewHandler(configPath)
mux := http.NewServeMux()
h.RegisterRoutes(mux)
rec := httptest.NewRecorder()
req := httptest.NewRequest(http.MethodGet, "/api/sessions/detail-reasoning-content", nil)
mux.ServeHTTP(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("status = %d, want %d, body=%s", rec.Code, http.StatusOK, rec.Body.String())
}
var resp struct {
Messages []struct {
Role string `json:"role"`
Content string `json:"content"`
Kind string `json:"kind"`
} `json:"messages"`
}
if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil {
t.Fatalf("Unmarshal() error = %v", err)
}
if len(resp.Messages) != 3 {
t.Fatalf("len(resp.Messages) = %d, want 3", len(resp.Messages))
}
if resp.Messages[1].Role != "assistant" ||
resp.Messages[1].Content != "internal chain of thought" ||
resp.Messages[1].Kind != "thought" {
t.Fatalf("thought message = %#v, want assistant thought/internal chain of thought", resp.Messages[1])
}
if resp.Messages[2].Role != "assistant" || resp.Messages[2].Content != "final visible answer" {
t.Fatalf("final message = %#v, want assistant/final visible answer", resp.Messages[2])
}
}
func TestHandleGetSession_ReconstructsRefreshMatrixForThoughtAndToolSummary(t *testing.T) {
configPath, cleanup := setupOAuthTestEnv(t)
defer cleanup()
dir := sessionsTestDir(t, configPath)
store, err := memory.NewJSONLStore(dir)
if err != nil {
t.Fatalf("NewJSONLStore() error = %v", err)
}
sessionKey := picoSessionPrefix + "detail-refresh-matrix"
for _, msg := range []providers.Message{
{Role: "user", Content: "turn1"},
{Role: "assistant", Content: "plain visible", ReasoningContent: "plain thought"},
{Role: "user", Content: "turn2"},
{
Role: "assistant",
ReasoningContent: "tool thought",
ToolCalls: []providers.ToolCall{{
ID: "call_read_file",
Type: "function",
Function: &providers.FunctionCall{
Name: "read_file",
Arguments: `{"path":"README.md"}`,
},
}},
},
{Role: "tool", ToolCallID: "call_read_file", Content: "file result"},
{Role: "user", Content: "turn3"},
{
Role: "assistant",
Content: "tool visible only",
ToolCalls: []providers.ToolCall{{
ID: "call_list_dir",
Type: "function",
Function: &providers.FunctionCall{
Name: "list_dir",
Arguments: `{"path":"."}`,
},
}},
},
{Role: "tool", ToolCallID: "call_list_dir", Content: "dir result"},
{Role: "user", Content: "turn4"},
{
Role: "assistant",
Content: "tool visible and thought",
ReasoningContent: "tool mixed thought",
ToolCalls: []providers.ToolCall{{
ID: "call_exec",
Type: "function",
Function: &providers.FunctionCall{
Name: "exec",
Arguments: `{"command":"pwd"}`,
},
}},
},
{Role: "tool", ToolCallID: "call_exec", Content: "pwd result"},
} {
if err := store.AddFullMessage(nil, sessionKey, msg); err != nil {
t.Fatalf("AddFullMessage() error = %v", err)
}
}
h := NewHandler(configPath)
mux := http.NewServeMux()
h.RegisterRoutes(mux)
rec := httptest.NewRecorder()
req := httptest.NewRequest(http.MethodGet, "/api/sessions/detail-refresh-matrix", nil)
mux.ServeHTTP(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("status = %d, want %d, body=%s", rec.Code, http.StatusOK, rec.Body.String())
}
var resp struct {
Messages []struct {
Role string `json:"role"`
Content string `json:"content"`
Kind string `json:"kind"`
} `json:"messages"`
}
if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil {
t.Fatalf("Unmarshal() error = %v", err)
}
if len(resp.Messages) != 13 {
t.Fatalf("len(resp.Messages) = %d, want 13", len(resp.Messages))
}
assertMessage := func(index int, role, kind, content string) {
t.Helper()
msg := resp.Messages[index]
if msg.Role != role || msg.Kind != kind || msg.Content != content {
t.Fatalf("messages[%d] = %#v, want role=%q kind=%q content=%q", index, msg, role, kind, content)
}
}
assertMessage(0, "user", "", "turn1")
assertMessage(1, "assistant", "thought", "plain thought")
assertMessage(2, "assistant", "", "plain visible")
assertMessage(3, "user", "", "turn2")
assertMessage(4, "assistant", "thought", "tool thought")
if !strings.Contains(resp.Messages[5].Content, "`read_file`") {
t.Fatalf("messages[5] = %#v, want read_file tool summary", resp.Messages[5])
}
assertMessage(6, "user", "", "turn3")
if !strings.Contains(resp.Messages[7].Content, "`list_dir`") {
t.Fatalf("messages[7] = %#v, want list_dir tool summary", resp.Messages[7])
}
assertMessage(8, "assistant", "", "tool visible only")
assertMessage(9, "user", "", "turn4")
assertMessage(10, "assistant", "thought", "tool mixed thought")
if !strings.Contains(resp.Messages[11].Content, "`exec`") {
t.Fatalf("messages[11] = %#v, want exec tool summary", resp.Messages[11])
}
assertMessage(12, "assistant", "", "tool visible and thought")
}
func TestHandleGetSession_ReconstructsVisibleMessageToolOutputWithoutDuplicateSummary(t *testing.T) { func TestHandleGetSession_ReconstructsVisibleMessageToolOutputWithoutDuplicateSummary(t *testing.T) {
configPath, cleanup := setupOAuthTestEnv(t) configPath, cleanup := setupOAuthTestEnv(t)
defer cleanup() defer cleanup()

View file

@ -14,6 +14,7 @@ export interface SessionDetail {
messages: { messages: {
role: "user" | "assistant" role: "user" | "assistant"
content: string content: string
kind?: "normal" | "thought"
media?: string[] media?: string[]
attachments?: { attachments?: {
type?: "image" | "audio" | "video" | "file" type?: "image" | "audio" | "video" | "file"

View file

@ -45,7 +45,8 @@ export async function loadSessionMessages(
id: `hist-${index}-${Date.now()}`, id: `hist-${index}-${Date.now()}`,
role: message.role, role: message.role,
content: message.content, content: message.content,
kind: message.role === "assistant" ? "normal" : undefined, kind:
message.role === "assistant" ? (message.kind ?? "normal") : undefined,
attachments: toChatAttachments({ attachments: toChatAttachments({
media: message.media, media: message.media,
attachments: message.attachments, attachments: message.attachments,