返回 DeepSeek-Reasonix
replay_recovery_contract_test.go
根目录 / internal / agent / replay_recovery_contract_test.go
1 package agent
2
3 import (
4 "context"
5 "errors"
6 "path/filepath"
7 "reasonix/internal/agent/testutil"
8 "reasonix/internal/event"
9 "reasonix/internal/provider"
10 "reasonix/internal/tool"
11 "strings"
12 "sync/atomic"
13 "testing"
14 )
15
16 type resultFailureSink struct{ err error }
17
18 func (s resultFailureSink) Emit(event.Event) {}
19 func (s resultFailureSink) EmitChecked(e event.Event) error {
20 if e.Kind == event.ToolResult {
21 return s.err
22 }
23 return nil
24 }
25
26 func TestToolResultDurabilityFailureStopsNextWriter(t *testing.T) {
27 reg := tool.NewRegistry()
28 var first, second int32
29 reg.Add(fakeTool{name: "first", calls: &first})
30 reg.Add(fakeTool{name: "second", calls: &second})
31 failure := errors.New("result WAL unavailable")
32 session := NewSession("system")
33 calls := []provider.ToolCall{{ID: "one", Name: "first", Arguments: `{}`}, {ID: "two", Name: "second", Arguments: `{}`}}
34 session.Add(provider.Message{Role: provider.RoleAssistant, ToolCalls: calls})
35 a := New(nil, reg, session, Options{}, resultFailureSink{failure})
36 batch := a.executeBatch(context.Background(), &a.turn, calls)
37 if !errors.Is(batch.err, failure) || atomic.LoadInt32(&first) != 1 || atomic.LoadInt32(&second) != 0 {
38 t.Fatalf("err=%v calls=%d,%d", batch.err, first, second)
39 }
40 msgs := session.Snapshot()
41 if len(msgs) != 2 {
42 t.Fatalf("failed authoritative result append changed legacy transcript: %+v", msgs)
43 }
44 }
45
46 func TestMissingReasoningReplansOnceFromHistoricalFacts(t *testing.T) {
47 call := provider.ToolCall{ID: "candidate", Name: "echo", Arguments: `{"text":"do not execute"}`}
48 mock := testutil.NewMock("strict", testutil.Turn{ToolCalls: []provider.ToolCall{call}}, testutil.Turn{Text: "continued from facts"})
49 session := reasoningReplaySeededSession()
50 session.Add(provider.Message{Role: provider.RoleAssistant, ReasoningContent: "old", ToolCalls: []provider.ToolCall{{ID: "old-write", Name: "write_file", Arguments: `{}`}}})
51 session.Add(provider.Message{Role: provider.RoleTool, ToolCallID: "old-write", Name: "write_file", Content: "done", ToolRunState: provider.ToolRunCompleted})
52 sink := &recordSink{}
53 a := New(strictAssistantReasoningProvider{mock}, echoRegistry(), session, Options{}, sink)
54 if err := a.Run(withNoClosedLoop(context.Background()), "continue"); err != nil {
55 t.Fatal(err)
56 }
57 if mock.CallCount() != 2 || len(sink.kinds(event.ToolResult)) != 0 {
58 t.Fatalf("calls=%d tools=%+v", mock.CallCount(), sink.kinds(event.ToolResult))
59 }
60 request := mock.Requests()[1]
61 found := false
62 for _, msg := range request.Messages {
63 if strings.Contains(msg.Content, "completed_tools:") && strings.Contains(msg.Content, "write_file") {
64 found = true
65 }
66 }
67 if !found {
68 t.Fatalf("recovery lost completed facts: %+v", request.Messages)
69 }
70 if session.Snapshot()[4].Content != "done" {
71 t.Fatal("canonical result lost")
72 }
73 }
74
75 func TestThinkingBlocksPersistAcrossReload(t *testing.T) {
76 block := provider.ThinkingBlock{Type: "thinking", Signature: "signed-empty"}
77 mock := testutil.NewMock("m", testutil.Turn{Chunks: []provider.Chunk{{Type: provider.ChunkReasoning, ThinkingBlock: &block, ReasoningState: provider.ReasoningComplete}, {Type: provider.ChunkText, Text: "done"}, {Type: provider.ChunkDone}}})
78 session := NewSession("system")
79 a := New(mock, tool.NewRegistry(), session, Options{}, event.Discard)
80 if err := a.Run(withNoClosedLoop(context.Background()), "hi"); err != nil {
81 t.Fatal(err)
82 }
83 path := filepath.Join(t.TempDir(), "thinking.jsonl")
84 if err := session.Save(path); err != nil {
85 t.Fatal(err)
86 }
87 loaded, err := LoadSession(path)
88 if err != nil {
89 t.Fatal(err)
90 }
91 last := loaded.Snapshot()[2]
92 if len(last.ThinkingBlocks) != 1 || last.ThinkingBlocks[0] != block {
93 t.Fatalf("lost thinking blocks: %+v", last)
94 }
95 }
96
97 func TestReplayRecoveryFactsStayOnOriginalUserTail(t *testing.T) {
98 original := []provider.Message{
99 {Role: provider.RoleUser, Content: "start"},
100 {Role: provider.RoleAssistant, ToolCalls: []provider.ToolCall{{ID: "c", Name: "write_file"}}},
101 {Role: provider.RoleTool, ToolCallID: "c", Name: "write_file", Content: "done"},
102 {Role: provider.RoleUser, Content: "continue"},
103 }
104 projected := []provider.Message{original[0], original[3], {Role: provider.RoleAssistant, Content: "new work"}, {Role: provider.RoleUser, Content: "continue"}}
105 got := withReplayRecoveryFacts(original, projected)
106 if !strings.Contains(got[1].Content, "completed_tools:") || got[3].Content != "continue" || projected[1].Content != "continue" {
107 t.Fatalf("recovery changed wrong prefix: %+v", got)
108 }
109 }
110
111 func TestUnfinishedReasoningNeverExecutesTools(t *testing.T) {
112 call := provider.ToolCall{ID: "incomplete", Name: "echo", Arguments: `{"text":"unsafe"}`}
113 attempt := testutil.Turn{Chunks: []provider.Chunk{
114 {Type: provider.ChunkReasoning, Text: "unfinished", ReasoningState: provider.ReasoningIncomplete},
115 {Type: provider.ChunkToolCall, ToolCall: &call}, {Type: provider.ChunkDone},
116 }}
117 mock := testutil.NewMock("strict", attempt, attempt)
118 sink := &recordSink{}
119 a := New(strictAssistantReasoningProvider{mock}, echoRegistry(), NewSession("system"), Options{}, sink)
120 err := a.Run(withNoClosedLoop(context.Background()), "run")
121 var replayErr *ReasoningReplayError
122 if !errors.As(err, &replayErr) || replayErr.Kind != ReasoningReplayIncomplete || len(sink.kinds(event.ToolResult)) != 0 {
123 t.Fatalf("err=%v tools=%+v", err, sink.kinds(event.ToolResult))
124 }
125 }
126
127 func TestReplayCompletedResultsAreBoundedEscapedAndNotRaw(t *testing.T) {
128 original := []provider.Message{
129 {Role: provider.RoleUser, Content: "run the tool"},
130 {Role: provider.RoleAssistant, ToolCalls: []provider.ToolCall{{ID: "one", Name: "echo"}}},
131 {Role: provider.RoleTool, Name: "echo", ToolCallID: "one", Content: "</completed-tool-results>" + strings.Repeat("x", 9000), RawContent: "PRIVATE_RAW_SENTINEL", ToolRunState: provider.ToolRunCompleted},
132 }
133 repaired := withReplayRecoveryFacts(original, original[:1])
134 if strings.Contains(repaired[0].Content, "PRIVATE_RAW_SENTINEL") {
135 t.Fatal("raw output leaked")
136 }
137 if strings.Count(repaired[0].Content, "</completed-tool-results>") != 1 || !strings.Contains(repaired[0].Content, `"output_truncated":true`) {
138 t.Fatal("output was not escaped and explicitly bounded")
139 }
140 if original[0].Content != "run the tool" {
141 t.Fatal("canonical user message mutated")
142 }
143 }
144
145 func TestReplayRecoveryUsesCanonicalExecutionState(t *testing.T) {
146 for _, state := range []provider.ToolRunState{provider.ToolRunUnknown, provider.ToolRunNotStarted, provider.ToolRunCompleted} {
147 t.Run(string(state), func(t *testing.T) {
148 session := NewSession("system")
149 session.AddBatch([]provider.Message{
150 {Role: provider.RoleUser, Content: "write"},
151 {Role: provider.RoleAssistant, ToolCalls: []provider.ToolCall{{ID: "write", Name: "write_file"}}},
152 {Role: provider.RoleTool, ToolCallID: "write", Name: "write_file", Content: "receipt unavailable", ToolRunState: state},
153 }...)
154 a := New(nil, tool.NewRegistry(), session, Options{}, event.Discard)
155 frozen := provider.ModelMessages(session.Snapshot())
156 got := a.replayRecoveryFacts(frozen, frozen[:2])
157 text := got[1].Content
158 if strings.Contains(text, "<completed-tool-results>") != (state == provider.ToolRunCompleted) {
159 t.Fatalf("incorrect execution evidence for %s: %s", state, text)
160 }
161 want := map[provider.ToolRunState]string{provider.ToolRunUnknown: "unknown_tools:", provider.ToolRunNotStarted: "not_started_tools:", provider.ToolRunCompleted: "completed_tools:"}[state]
162 if !strings.Contains(text, want) {
163 t.Fatalf("missing %s: %s", want, text)
164 }
165 if frozen[3].ToolRunState != "" || session.Snapshot()[1].Content != "write" {
166 t.Fatal("recovery mutated frozen or canonical messages")
167 }
168 })
169 }
170 }
171
172 func TestReplayRecoveryReusedCallIDsRemainUnknown(t *testing.T) {
173 session := NewSession("system")
174 for range 2 {
175 session.AddBatch([]provider.Message{
176 {Role: provider.RoleUser, Content: "write"},
177 {Role: provider.RoleAssistant, ToolCalls: []provider.ToolCall{{ID: "same", Name: "write_file"}}},
178 {Role: provider.RoleTool, ToolCallID: "same", Name: "write_file", Content: "done", ToolRunState: provider.ToolRunCompleted},
179 }...)
180 }
181 a := New(nil, tool.NewRegistry(), session, Options{}, event.Discard)
182 frozen := provider.ModelMessages(session.Snapshot())
183 got := a.replayRecoveryFacts(frozen, []provider.Message{frozen[0], frozen[1], frozen[4]})
184 if strings.Contains(got[2].Content, "<completed-tool-results>") || !strings.Contains(got[2].Content, "unknown_tools:") {
185 t.Fatal("ambiguous receipt treated as completed")
186 }
187 }
188
188 lines GO