返回 DeepSeek-Reasonix
compact_commit_emit_test.go
根目录 / internal / agent / compact_commit_emit_test.go
1 package agent
2
3 import (
4 "context"
5 "errors"
6 "os"
7 "path/filepath"
8 "reflect"
9 "strings"
10 "sync"
11 "testing"
12 "time"
13
14 "reasonix/internal/event"
15 "reasonix/internal/fileutil"
16 "reasonix/internal/provider"
17 "reasonix/internal/tool"
18 )
19
20 // reentrantSnapshotSink re-enters ContextMaintenanceSnapshot on every emit,
21 // which takes compactionMu. commitSummaryProjection must unlock before Emit.
22 type reentrantSnapshotSink struct {
23 agent *Agent
24 mu sync.Mutex
25 n int
26 }
27
28 type modelContextRecorderStub struct {
29 result SessionModelContextCommitResult
30 err error
31 commit SessionModelContextCommit
32 }
33
34 type modelContextRecorderStep struct {
35 result SessionModelContextCommitResult
36 err error
37 }
38
39 type sequencedModelContextRecorder struct {
40 steps []modelContextRecorderStep
41 commits []SessionModelContextCommit
42 }
43
44 func (*modelContextRecorderStub) CheckpointSession(context.Context, SessionCheckpointBoundary) error {
45 return nil
46 }
47
48 func (r *modelContextRecorderStub) RecordSessionModelContext(_ context.Context, commit SessionModelContextCommit) (SessionModelContextCommitResult, error) {
49 r.commit = commit
50 return r.result, r.err
51 }
52
53 func (*sequencedModelContextRecorder) CheckpointSession(context.Context, SessionCheckpointBoundary) error {
54 return nil
55 }
56
57 func (r *sequencedModelContextRecorder) RecordSessionModelContext(_ context.Context, commit SessionModelContextCommit) (SessionModelContextCommitResult, error) {
58 r.commits = append(r.commits, cloneSessionModelContextCommit(commit))
59 if len(r.steps) == 0 {
60 return SessionModelContextCommitResult{}, errors.New("unexpected model context commit")
61 }
62 step := r.steps[0]
63 r.steps = r.steps[1:]
64 return step.result, step.err
65 }
66
67 func (s *reentrantSnapshotSink) Emit(e event.Event) {
68 if e.Kind != event.ContextMaintenanceEvent {
69 return
70 }
71 s.mu.Lock()
72 s.n++
73 s.mu.Unlock()
74 if s.agent != nil {
75 _ = s.agent.ContextMaintenanceSnapshot()
76 }
77 }
78
79 func TestCommitSummaryEmitsOutsideCompactionLock(t *testing.T) {
80 prov := &fakeProvider{reply: "digest for reentrant emit"}
81 sess := &Session{Messages: []provider.Message{
82 {Role: provider.RoleSystem, Content: "sys"},
83 {Role: provider.RoleUser, Content: "task"},
84 {Role: provider.RoleAssistant, Content: strings.Repeat("work line\n", 800)},
85 {Role: provider.RoleUser, Content: "continue"},
86 {Role: provider.RoleAssistant, Content: strings.Repeat("more work\n", 800)},
87 {Role: provider.RoleUser, Content: "tail"},
88 {Role: provider.RoleAssistant, Content: "ok"},
89 }}
90 path := filepath.Join(t.TempDir(), "session.jsonl")
91 sink := &reentrantSnapshotSink{}
92 a := New(prov, tool.NewRegistry(), sess, Options{
93 ContextWindow: 20_000, CompactRatio: 0.5, RecentKeep: 2,
94 SessionPath: path, WorkspaceID: "ws", ModelRef: "p/m",
95 }, sink)
96 sink.agent = a
97 if err := a.CompactNow(context.Background(), ""); err != nil {
98 t.Fatalf("CompactNow: %v", err)
99 }
100 sink.mu.Lock()
101 n := sink.n
102 sink.mu.Unlock()
103 if n == 0 {
104 t.Fatal("expected context_maintenance emit after checkpoint install")
105 }
106 if got := a.currentProjectionVersion(); got != 1 {
107 t.Fatalf("projection version = %d, want 1", got)
108 }
109 }
110
111 func TestCommitSummaryRetainsAcceptedProjectionOnDurabilityFailure(t *testing.T) {
112 recorder := &modelContextRecorderStub{
113 result: SessionModelContextCommitResult{Accepted: true},
114 err: errors.New("injected flush failure"),
115 }
116 a := New(&fakeProvider{reply: "durability failure digest"}, tool.NewRegistry(), &Session{Messages: []provider.Message{
117 {Role: provider.RoleSystem, Content: "sys"},
118 {Role: provider.RoleUser, Content: "task"},
119 {Role: provider.RoleAssistant, Content: strings.Repeat("work line\n", 800)},
120 {Role: provider.RoleUser, Content: "continue"},
121 {Role: provider.RoleAssistant, Content: strings.Repeat("more work\n", 800)},
122 {Role: provider.RoleUser, Content: "tail"},
123 {Role: provider.RoleAssistant, Content: "ok"},
124 }}, Options{ContextWindow: 20_000, CompactRatio: 0.5, RecentKeep: 2, SessionCheckpointer: recorder}, event.Discard)
125
126 if err := a.CompactNow(context.Background(), ""); err == nil || !strings.Contains(err.Error(), "injected flush failure") {
127 t.Fatalf("CompactNow error = %v, want durability failure", err)
128 }
129 if got := a.currentProjectionVersion(); got != 1 {
130 t.Fatalf("accepted projection version = %d, want 1", got)
131 }
132 if len(recorder.commit.Messages) == 0 || recorder.commit.OperationID == "" {
133 t.Fatalf("recorder received incomplete commit: %+v", recorder.commit)
134 }
135 if a.sess.checkpointState != "pending" {
136 t.Fatalf("checkpoint state = %q, want pending", a.sess.checkpointState)
137 }
138 }
139
140 func TestPendingProjectionBlocksModelUntilExactCommitIsDurable(t *testing.T) {
141 recorder := &sequencedModelContextRecorder{steps: []modelContextRecorderStep{
142 {result: SessionModelContextCommitResult{Accepted: true}, err: errors.New("initial flush failure")},
143 {result: SessionModelContextCommitResult{Accepted: true}, err: errors.New("retry flush failure")},
144 {result: SessionModelContextCommitResult{Accepted: true, Durable: true}},
145 }}
146 prov := &scriptedProvider{name: "model", turns: [][]provider.Chunk{
147 {{Type: provider.ChunkText, Text: "durable retry digest"}, {Type: provider.ChunkDone}},
148 {{Type: provider.ChunkText, Text: "continued"}, {Type: provider.ChunkDone}},
149 }}
150 a := New(prov, tool.NewRegistry(), &Session{Messages: []provider.Message{
151 {Role: provider.RoleSystem, Content: "sys"},
152 {Role: provider.RoleUser, Content: "task"},
153 {Role: provider.RoleAssistant, Content: strings.Repeat("work line\n", 800)},
154 {Role: provider.RoleUser, Content: "continue"},
155 {Role: provider.RoleAssistant, Content: strings.Repeat("more work\n", 800)},
156 {Role: provider.RoleUser, Content: "tail"},
157 {Role: provider.RoleAssistant, Content: "ok"},
158 }}, Options{ContextWindow: 20_000, CompactRatio: 0.5, RecentKeep: 2, SessionCheckpointer: recorder}, event.Discard)
159
160 if err := a.CompactNow(t.Context(), ""); err == nil || !strings.Contains(err.Error(), "initial flush failure") {
161 t.Fatalf("CompactNow error = %v, want initial durability failure", err)
162 }
163 if prov.call != 1 {
164 t.Fatalf("provider calls after summary = %d, want 1", prov.call)
165 }
166 if err := a.Run(t.Context(), "must stay blocked"); err == nil || !strings.Contains(err.Error(), "retry flush failure") {
167 t.Fatalf("blocked Run error = %v, want retry durability failure", err)
168 }
169 if prov.call != 1 {
170 t.Fatalf("provider dispatched with pending durability: calls=%d, want 1", prov.call)
171 }
172 if err := a.Run(t.Context(), "continue after durable"); err != nil {
173 t.Fatalf("Run after durable retry: %v", err)
174 }
175 if prov.call != 2 {
176 t.Fatalf("provider calls after recovery = %d, want 2", prov.call)
177 }
178 if len(recorder.commits) != 3 {
179 t.Fatalf("model context commits = %d, want 3", len(recorder.commits))
180 }
181 for i := 1; i < len(recorder.commits); i++ {
182 if !reflect.DeepEqual(recorder.commits[0], recorder.commits[i]) {
183 t.Fatalf("retry %d changed accepted commit\nfirst=%+v\nretry=%+v", i, recorder.commits[0], recorder.commits[i])
184 }
185 }
186 if a.sess.pendingModelContextCommit != nil || a.sess.checkpointState != "applied" {
187 t.Fatalf("pending checkpoint not cleared: pending=%+v state=%q", a.sess.pendingModelContextCommit, a.sess.checkpointState)
188 }
189 }
190
191 func TestCommitSummaryRollsBackRejectedProjection(t *testing.T) {
192 recorder := &modelContextRecorderStub{err: errors.New("injected prepare failure")}
193 a := New(&fakeProvider{reply: "rejected projection digest"}, tool.NewRegistry(), &Session{Messages: []provider.Message{
194 {Role: provider.RoleSystem, Content: "sys"},
195 {Role: provider.RoleUser, Content: "task"},
196 {Role: provider.RoleAssistant, Content: strings.Repeat("work line\n", 800)},
197 {Role: provider.RoleUser, Content: "continue"},
198 {Role: provider.RoleAssistant, Content: strings.Repeat("more work\n", 800)},
199 {Role: provider.RoleUser, Content: "tail"},
200 {Role: provider.RoleAssistant, Content: "ok"},
201 }}, Options{ContextWindow: 20_000, CompactRatio: 0.5, RecentKeep: 2, SessionCheckpointer: recorder}, event.Discard)
202
203 if err := a.CompactNow(context.Background(), ""); err == nil || !strings.Contains(err.Error(), "injected prepare failure") {
204 t.Fatalf("CompactNow error = %v, want prepare failure", err)
205 }
206 if got := a.currentProjectionVersion(); got != 0 {
207 t.Fatalf("rejected projection version = %d, want 0", got)
208 }
209 }
210
211 // TestCommitSurvivesPostPublishDirSyncFailure locks the publish contract:
212 // after rename the checkpoint is committed. A parent-dir fsync failure must
213 // not roll back in-memory generation/projection (memory/disk fork).
214 func TestCommitSurvivesPostPublishDirSyncFailure(t *testing.T) {
215 restore := fileutil.SetSyncParentDirForTest(func(string) error {
216 return errors.New("injected parent dir fsync failure")
217 })
218 t.Cleanup(restore)
219
220 prov := &fakeProvider{reply: "digest after dir-sync fault"}
221 sess := &Session{Messages: []provider.Message{
222 {Role: provider.RoleSystem, Content: "sys"},
223 {Role: provider.RoleUser, Content: "task"},
224 {Role: provider.RoleAssistant, Content: strings.Repeat("work line\n", 800)},
225 {Role: provider.RoleUser, Content: "continue"},
226 {Role: provider.RoleAssistant, Content: strings.Repeat("more work\n", 800)},
227 {Role: provider.RoleUser, Content: "tail"},
228 {Role: provider.RoleAssistant, Content: "ok"},
229 }}
230 path := filepath.Join(t.TempDir(), "session.jsonl")
231 a := New(prov, tool.NewRegistry(), sess, Options{
232 ContextWindow: 20_000, CompactRatio: 0.5, RecentKeep: 2,
233 SessionPath: path, WorkspaceID: "ws", ModelRef: "p/m",
234 }, event.Discard)
235 if err := a.CompactNow(context.Background(), ""); err != nil {
236 t.Fatalf("CompactNow with post-publish dir sync fault: %v", err)
237 }
238 memVer := a.currentProjectionVersion()
239 if memVer != 1 {
240 t.Fatalf("memory projection version = %d, want 1", memVer)
241 }
242 disk, ok, err := LoadCompactionState(path)
243 if err != nil || !ok {
244 t.Fatalf("load disk checkpoint: ok=%v err=%v", ok, err)
245 }
246 if disk.Projection.ProjectionVersion != memVer {
247 t.Fatalf("disk/memory fork: disk=%d mem=%d", disk.Projection.ProjectionVersion, memVer)
248 }
249 if disk.Generation != a.sess.compactionState.Generation {
250 t.Fatalf("generation fork: disk=%d mem=%d", disk.Generation, a.sess.compactionState.Generation)
251 }
252 }
253
254 // TestBlockedReceiptSurvivesPostPublishDirSyncFailure ensures a failed summary
255 // still installs the generation-scoped receipt in memory when only parent-dir
256 // fsync fails after rename — otherwise the next Prepare pays for another summary.
257 func TestBlockedReceiptSurvivesPostPublishDirSyncFailure(t *testing.T) {
258 restore := fileutil.SetSyncParentDirForTest(func(string) error {
259 return errors.New("injected parent dir fsync failure")
260 })
261 t.Cleanup(restore)
262
263 const window = 10_000
264 messages := []provider.Message{
265 {Role: provider.RoleSystem, Content: "system"},
266 {Role: provider.RoleUser, Content: "task"},
267 {Role: provider.RoleAssistant, Content: strings.Repeat("old work ", 500)},
268 {Role: provider.RoleUser, Content: "current"},
269 {Role: provider.RoleAssistant, Content: "tail"},
270 }
271 path := filepath.Join(t.TempDir(), "session.jsonl")
272 prov := &failingSummaryProvider{}
273 a := New(prov, tool.NewRegistry(), &Session{Messages: append([]provider.Message(nil), messages...)}, Options{
274 ContextWindow: window, CompactRatio: 0.85, RecentKeep: 2,
275 WorkspaceID: "workspace", ModelRef: "model",
276 }, event.Discard)
277 a.BindSessionPath(path, true)
278
279 policy := ContextPreparePolicy{Trigger: CompactionTriggerPressure, ObservedInputTokens: 8600}
280 if _, err := a.contextManager().Prepare(context.Background(), policy); err != nil {
281 t.Fatalf("above-ratio failure should not reject: %v", err)
282 }
283 if prov.calls != 1 {
284 t.Fatalf("summary calls = %d, want 1", prov.calls)
285 }
286 if a.sess.compactionState.LastReceipt == nil {
287 t.Fatal("memory lost blocked/failed receipt after post-publish dir-sync fault")
288 }
289 if status := a.sess.compactionState.LastReceipt.Status; status != "blocked" && status != "failed" {
290 t.Fatalf("receipt status = %q", status)
291 }
292 disk, ok, err := LoadCompactionState(path)
293 if err != nil || !ok || disk.LastReceipt == nil {
294 t.Fatalf("disk receipt missing: ok=%v err=%v", ok, err)
295 }
296 if disk.Generation != a.sess.compactionState.Generation {
297 t.Fatalf("blocked generation fork: disk=%d mem=%d", disk.Generation, a.sess.compactionState.Generation)
298 }
299 if _, err := a.contextManager().Prepare(context.Background(), policy); err != nil {
300 t.Fatal(err)
301 }
302 if prov.calls != 1 {
303 t.Fatalf("same generation re-summarized after dir-sync fault: calls=%d", prov.calls)
304 }
305 }
306
307 func TestLoadProjectionSidecarDoesNotRewriteExactKey(t *testing.T) {
308 path := filepath.Join(t.TempDir(), "session.jsonl")
309 msgs := []provider.Message{
310 {Role: provider.RoleSystem, Content: "sys"},
311 {Role: provider.RoleUser, Content: "u"},
312 }
313 hash := coveredPrefixHash(msgs, len(msgs))
314 key := promptCacheKey("ws", BranchID(path), "p/m")
315 st := CompactionState{
316 SchemaVersion: compactionStateSchemaCurrent,
317 TranscriptVersion: 0,
318 PromptCacheKey: key,
319 Projection: ContextProjection{
320 Messages: msgs, CoveredCount: len(msgs), CoveredPrefixHash: hash,
321 ProjectionVersion: 3, TranscriptVersion: 0,
322 },
323 UpdatedAt: time.Now().UTC(),
324 }
325 if err := SaveCompactionState(path, st); err != nil {
326 t.Fatal(err)
327 }
328 before, err := os.ReadFile(ContextStatePath(path))
329 if err != nil {
330 t.Fatal(err)
331 }
332 a := New(nil, tool.NewRegistry(), &Session{Messages: append([]provider.Message(nil), msgs...)}, Options{
333 SessionPath: path, WorkspaceID: "ws", ModelRef: "p/m",
334 }, event.Discard)
335 a.LoadProjectionSidecar(path)
336 if a.currentProjectionVersion() != 3 {
337 t.Fatalf("version = %d, want 3", a.currentProjectionVersion())
338 }
339 after, err := os.ReadFile(ContextStatePath(path))
340 if err != nil {
341 t.Fatal(err)
342 }
343 if string(before) != string(after) {
344 t.Fatalf("exact-key restore rewrote sidecar (%d -> %d bytes)", len(before), len(after))
345 }
346 }
347
348 func TestSaveCompactionStateStripsLegacyWriterFields(t *testing.T) {
349 path := filepath.Join(t.TempDir(), "session.jsonl")
350 st := CompactionState{
351 SchemaVersion: compactionStateSchemaCurrent,
352 TranscriptVersion: 1,
353 PromptCacheKey: "k",
354 LastTrigger: CompactionTriggerPressure,
355 LastMode: CompactionModeSummarized,
356 LastSourceTokens: 1000,
357 LastResultTokens: 200,
358 BlockedInputHash: "legacy-blocked",
359 BlockedReason: "legacy",
360 LastReceipt: &ContextMaintenanceReceipt{
361 Status: "applied", Action: "summary", ProjectionVersion: 1,
362 InputHash: "in", OutputHash: "out",
363 },
364 }
365 if err := SaveCompactionState(path, st); err != nil {
366 t.Fatal(err)
367 }
368 raw, err := os.ReadFile(ContextStatePath(path))
369 if err != nil {
370 t.Fatal(err)
371 }
372 for _, banned := range []string{
373 `"last_trigger"`, `"last_mode"`, `"last_source_tokens"`,
374 `"last_result_tokens"`, `"blocked_input_hash"`, `"blocked_reason"`,
375 } {
376 if strings.Contains(string(raw), banned) {
377 t.Fatalf("new writer re-emitted %s:\n%s", banned, raw)
378 }
379 }
380 got, ok, err := LoadCompactionState(path)
381 if err != nil || !ok {
382 t.Fatalf("load: ok=%v err=%v", ok, err)
383 }
384 if got.LastMode != "" || got.LastTrigger != "" || got.BlockedInputHash != "" {
385 t.Fatalf("legacy mirrors present after save: %+v", got)
386 }
387 if got.LastReceipt == nil || got.LastReceipt.Status != "applied" {
388 t.Fatalf("receipt lost: %+v", got.LastReceipt)
389 }
390 }
391
392 func TestLoadProjectionSidecarNormalizesNativeKeyOnce(t *testing.T) {
393 path := filepath.Join(t.TempDir(), "session.jsonl")
394 msgs := []provider.Message{
395 {Role: provider.RoleSystem, Content: "sys"},
396 {Role: provider.RoleUser, Content: "u"},
397 }
398 hash := coveredPrefixHash(msgs, len(msgs))
399 key := promptCacheKey("ws", BranchID(path), "p/m")
400 st := CompactionState{
401 SchemaVersion: compactionStateSchemaCurrent,
402 TranscriptVersion: 0,
403 PromptCacheKey: key + "|context-editing-native-anthropic",
404 Projection: ContextProjection{
405 Messages: msgs, CoveredCount: len(msgs), CoveredPrefixHash: hash,
406 ProjectionVersion: 2, TranscriptVersion: 0,
407 },
408 UpdatedAt: time.Now().UTC(),
409 }
410 if err := SaveCompactionState(path, st); err != nil {
411 t.Fatal(err)
412 }
413 a := New(nil, tool.NewRegistry(), &Session{Messages: append([]provider.Message(nil), msgs...)}, Options{
414 SessionPath: path, WorkspaceID: "ws", ModelRef: "p/m",
415 }, event.Discard)
416 a.LoadProjectionSidecar(path)
417 if a.currentProjectionVersion() != 2 {
418 t.Fatalf("version = %d, want 2", a.currentProjectionVersion())
419 }
420 loaded, ok, err := LoadCompactionState(path)
421 if err != nil || !ok {
422 t.Fatalf("reload: ok=%v err=%v", ok, err)
423 }
424 if loaded.PromptCacheKey != key {
425 t.Fatalf("PromptCacheKey = %q, want normalized %q", loaded.PromptCacheKey, key)
426 }
427 before, err := os.ReadFile(ContextStatePath(path))
428 if err != nil {
429 t.Fatal(err)
430 }
431 a2 := New(nil, tool.NewRegistry(), &Session{Messages: append([]provider.Message(nil), msgs...)}, Options{
432 SessionPath: path, WorkspaceID: "ws", ModelRef: "p/m",
433 }, event.Discard)
434 a2.LoadProjectionSidecar(path)
435 after, err := os.ReadFile(ContextStatePath(path))
436 if err != nil {
437 t.Fatal(err)
438 }
439 if string(before) != string(after) {
440 t.Fatal("second restore rewrote already-normalized sidecar")
441 }
442 }
443
443 lines GO