From 546a9315d517bcc60c493f52f59bf90a4432a43c Mon Sep 17 00:00:00 2001 From: OpenClaw Dev Date: Sun, 25 Oct 2026 06:51:20 +0100 Subject: [PATCH] fix(llm): strip inline reasoning from Dahl/MiniMax streams; dedupe identical parallel tool calls; hermetic config tests --- internal/agent/agent.go | 13 +++- internal/agent/agent_test.go | 26 +++++++ internal/config/presets_test.go | 11 +++ internal/llm/openai.go | 15 +++- internal/llm/think.go | 129 ++++++++++++++++++++++++++++++++ internal/llm/think_test.go | 75 +++++++++++++++++++ 6 files changed, 264 insertions(+), 5 deletions(-) create mode 100644 internal/llm/think.go create mode 100644 internal/llm/think_test.go diff --git a/internal/agent/agent.go b/internal/agent/agent.go index e696a33..93c60da 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -178,11 +178,20 @@ func (a *Agent) Run(parent context.Context, sessionID, userText string, emit fun if len(resp.ToolCalls) == 0 { return resp.Content, nil } + // Some models (seen with GLM/DeepSeek on Dahl) emit the same call several times in + // parallel; identical name+arguments run once and share the result. + seen := map[string]string{} for _, tc := range resp.ToolCalls { emit(Event{Type: "tool_call", Tool: tc.Name, Args: tc.Arguments, CallID: tc.ID}) - out := "error: cancelled by user" - if !cancelled() { + key := tc.Name + "\x00" + tc.Arguments + out, dup := seen[key] + switch { + case dup: + case cancelled(): + out = "error: cancelled by user" + default: out = a.Tools.Call(ctx, tc.Name, tc.Arguments) + seen[key] = out } emit(Event{Type: "tool_result", Tool: tc.Name, Text: clip(out, 2000), CallID: tc.ID}) // Every tool call gets a result so the history stays valid for the provider. diff --git a/internal/agent/agent_test.go b/internal/agent/agent_test.go index b56e8ab..5645632 100644 --- a/internal/agent/agent_test.go +++ b/internal/agent/agent_test.go @@ -206,3 +206,29 @@ func TestSummarizeNowManual(t *testing.T) { t.Fatalf("%v %v", sm, err) } } + +func TestDuplicateToolCallsRunOnce(t *testing.T) { + st := newStore(t) + ss, _ := st.CreateSession(context.Background(), "") + n := 0 + now := func() time.Time { n++; return time.Date(2026, 10, 25, 6, 0, 0, 0, time.UTC) } + dup := []llm.ToolCall{{ID: "a", Name: "time_now", Arguments: `{"timezone":"UTC"}`}, {ID: "b", Name: "time_now", Arguments: `{"timezone":"UTC"}`}, {ID: "c", Name: "time_now", Arguments: `{}`}} + p := &scripted{resps: []llm.Response{{ToolCalls: dup}, {Content: "ok"}}} + a := &Agent{LLM: p, Store: st, Tools: tools.NewRegistry(tools.TimeNow{Now: now}), MaxIterations: 3} + if _, err := a.Run(context.Background(), ss.ID, "zeit?", nil); err != nil { + t.Fatal(err) + } + if n != 2 { + t.Fatalf("tool executed %d times, want 2", n) + } + msgs, _ := st.Messages(context.Background(), ss.ID, 0) + tools := 0 + for _, m := range msgs { + if m.Role == llm.RoleTool { + tools++ + } + } + if tools != 3 { + t.Fatalf("every call id needs a result, got %d", tools) + } +} diff --git a/internal/config/presets_test.go b/internal/config/presets_test.go index d2e957c..8bc4074 100644 --- a/internal/config/presets_test.go +++ b/internal/config/presets_test.go @@ -7,6 +7,7 @@ import ( ) func load(t *testing.T, yaml string) (Config, error) { + t.Helper() t.Helper() p := filepath.Join(t.TempDir(), "c.yaml") os.WriteFile(p, []byte("auth_token: aaaaaaaaaaaaaaaaaaaa\n"+yaml), 0o600) @@ -74,3 +75,13 @@ func TestAllPresetsComplete(t *testing.T) { } } } + +// TestMain keeps tests hermetic when the developer shell has BYTECLAW_* or provider keys set. +func TestMain(m *testing.M) { + for _, k := range []string{"BYTECLAW_LISTEN", "BYTECLAW_DATA_DIR", "BYTECLAW_WORKSPACE", "BYTECLAW_LLM_PROVIDER", + "BYTECLAW_LLM_BASE_URL", "BYTECLAW_LLM_API_KEY", "BYTECLAW_LLM_MODEL", "BYTECLAW_ALLOWED_ORIGINS", "BYTECLAW_AUTH_TOKEN", + "OPENAI_API_KEY", "OPENROUTER_API_KEY", "DAHL_API_KEY", "GONKAROUTER_API_KEY"} { + os.Unsetenv(k) + } + os.Exit(m.Run()) +} diff --git a/internal/llm/openai.go b/internal/llm/openai.go index 6c91ebf..5cdbef4 100644 --- a/internal/llm/openai.go +++ b/internal/llm/openai.go @@ -131,6 +131,7 @@ type partialCall struct { func parseStream(r io.Reader, onDelta DeltaFunc) (Response, error) { var out Response var text strings.Builder + var tf thinkFilter calls := map[int]*partialCall{} sc := bufio.NewScanner(r) sc.Buffer(make([]byte, 64*1024), 4<<20) @@ -152,9 +153,11 @@ func parseStream(r io.Reader, onDelta DeltaFunc) (Response, error) { } for _, c := range ch.Choices { if c.Delta.Content != "" { - text.WriteString(c.Delta.Content) - if onDelta != nil { - onDelta(c.Delta.Content) + if v := tf.push(c.Delta.Content); v != "" { + text.WriteString(v) + if onDelta != nil { + onDelta(v) + } } } for i, tc := range c.Delta.ToolCalls { @@ -183,6 +186,12 @@ func parseStream(r io.Reader, onDelta DeltaFunc) (Response, error) { if err := sc.Err(); err != nil { return out, fmt.Errorf("read stream: %w", err) } + if v := tf.finish(); v != "" { + text.WriteString(v) + if onDelta != nil { + onDelta(v) + } + } out.Content = text.String() idxs := make([]int, 0, len(calls)) for i := range calls { diff --git a/internal/llm/think.go b/internal/llm/think.go new file mode 100644 index 0000000..7a6ef71 --- /dev/null +++ b/internal/llm/think.go @@ -0,0 +1,129 @@ +package llm + +import "strings" + +// thinkFilter removes inline reasoning blocks ("…") that some +// OpenAI-compatible backends (e.g. MiniMax-M2.x on vLLM, seen on Dahl) put into +// the normal content stream instead of a separate reasoning field. Tags may be +// split across chunks. Text inside a block is never streamed to the user. +type thinkFilter struct { + pending string // undecided tail that may be the start of a tag + inThink bool + think strings.Builder // reasoning text of the current (unclosed) block + out strings.Builder // visible content + sawOpen bool +} + +const ( + openTag = "" + closeTag = "" + gap = "\n\n\n" +) + +// push consumes a chunk and returns the text that may be shown now. +func (f *thinkFilter) push(s string) string { + buf := f.pending + s + f.pending = "" + var vis strings.Builder + for buf != "" { + tag := openTag + if f.inThink { + tag = closeTag + } + i, n := strings.Index(buf, tag), len(tag) + // Dahl/vLLM sometimes swallows "" and leaves only a "\n\n\n" gap; + // treat that gap as the end of the block so the answer still streams live. + if f.inThink { + if j := strings.Index(buf, gap); j >= 0 && (i < 0 || j < i) { + i, n = j, len(gap) + } + } + if i >= 0 { + f.emit(&vis, buf[:i]) + buf = buf[i+n:] + if f.inThink { + f.think.Reset() + } else { + f.sawOpen = true + } + f.inThink = !f.inThink + continue + } + // a stray closing tag outside a block is dropped + if !f.inThink { + if i := strings.Index(buf, closeTag); i >= 0 { + f.emit(&vis, buf[:i]) + buf = buf[i+len(closeTag):] + continue + } + } + keep := partialSuffix(buf, tag) + if f.inThink { + if k := partialSuffix(buf, gap); k > keep { + keep = k + } + } else { + if k := partialSuffix(buf, closeTag); k > keep { + keep = k + } + } + f.emit(&vis, buf[:len(buf)-keep]) + f.pending = buf[len(buf)-keep:] + buf = "" + } + return vis.String() +} + +func (f *thinkFilter) emit(vis *strings.Builder, s string) { + if s == "" { + return + } + if f.inThink { + f.think.WriteString(s) + return + } + // drop whitespace directly after a closed block at the start of the answer + if f.out.Len() == 0 { + s = strings.TrimLeft(s, " \n\r\t") + if s == "" { + return + } + } + f.out.WriteString(s) + vis.WriteString(s) +} + +// finish flushes the remainder. If a block was never closed (observed when the +// backend swallows ""), the answer is recovered from the text after the +// first blank-line gap so the user does not end up with nothing. +func (f *thinkFilter) finish() string { + var vis strings.Builder + if f.pending != "" { + p := f.pending + f.pending = "" + f.emit(&vis, p) + } + if f.inThink { + t := f.think.String() + f.inThink = false + f.think.Reset() + ans := "" + if i := strings.Index(t, "\n\n\n"); i >= 0 { + ans = t[i:] + } else if f.out.Len() == 0 { + ans = t // better a visible reasoning than an empty reply + } + f.emit(&vis, strings.TrimSpace(ans)) + } + return vis.String() +} + +// partialSuffix returns the length of the longest suffix of s that is a proper prefix of tag. +func partialSuffix(s, tag string) int { + for n := len(tag) - 1; n > 0; n-- { + if len(s) >= n && strings.HasSuffix(s, tag[:n]) { + return n + } + } + return 0 +} diff --git a/internal/llm/think_test.go b/internal/llm/think_test.go new file mode 100644 index 0000000..d51a699 --- /dev/null +++ b/internal/llm/think_test.go @@ -0,0 +1,75 @@ +package llm + +import ( + "strings" + "testing" +) + +func run(chunks ...string) (string, string) { + var f thinkFilter + var streamed strings.Builder + for _, c := range chunks { + streamed.WriteString(f.push(c)) + } + streamed.WriteString(f.finish()) + return streamed.String(), f.out.String() +} + +func TestThinkFilter(t *testing.T) { + cases := []struct { + name string + chunks []string + want string + }{ + {"plain", []string{"Hallo ", "Welt"}, "Hallo Welt"}, + {"block", []string{"reasoning\n\n\nHallo!"}, "Hallo!"}, + {"split tags", []string{"abc", "\n\nHal", "lo"}, "Hallo"}, + {"char by char", strings.Split("xJa", ""), "Ja"}, + {"stray close", []string{"Hallo!\n\nHallo!"}, "Hallo!\n\nHallo!"}, + {"unclosed with gap (Dahl)", []string{"The user asks.\n\n\nHallo! Wie geht's?"}, "Hallo! Wie geht's?"}, + {"gap streams live", []string{"a\n", "\n", "\nAntw", "ort"}, "Antwort"}, + {"gap outside block kept", []string{"A\n\n\nB"}, "A\n\n\nB"}, + {"unclosed no gap", []string{"nur Gedanken"}, "nur Gedanken"}, + {"lt not a tag", []string{"a c <", "3"}, "a c <3"}, + {"trailing partial", []string{"x The user"}}]} + +data: {"choices":[{"delta":{"content":" greets.\n\n\nHallo!"}}]} + +data: {"choices":[{"delta":{},"finish_reason":"stop"}]} + +data: [DONE] +` + var deltas []string + resp, err := parseStream(strings.NewReader(sse), func(s string) { deltas = append(deltas, s) }) + if err != nil || resp.Content != "Hallo!" || strings.Join(deltas, "") != "Hallo!" { + t.Fatalf("%q %v %v", resp.Content, deltas, err) + } +} + +// The answer after a swallowed must be streamed incrementally, not only at the end. +func TestThinkGapStreamsIncrementally(t *testing.T) { + var f thinkFilter + if v := f.push("Überlegung\n\n\nHal"); v != "Hal" { + t.Fatalf("got %q", v) + } + if v := f.push("lo"); v != "lo" { + t.Fatalf("got %q", v) + } +}