| 1 | //go:build live |
| 2 | |
| 3 | package agent |
| 4 | |
| 5 | import ( |
| 6 | "context" |
| 7 | "fmt" |
| 8 | "os" |
| 9 | "path/filepath" |
| 10 | "sync/atomic" |
| 11 | "testing" |
| 12 | "time" |
| 13 | |
| 14 | "reasonix/internal/event" |
| 15 | "reasonix/internal/provider" |
| 16 | "reasonix/internal/tool" |
| 17 | "reasonix/internal/tool/builtin" |
| 18 | ) |
| 19 | |
| 20 | // A real model prepares the built-in write. We checkpoint its actual intent, |
| 21 | // interrupt after the disk effect, and reopen only the pre-result checkpoint. |
| 22 | // The only model tool is a writer confined to an empty disposable directory. |
| 23 | func TestLiveOfficialWriteAfterEffectResume(t *testing.T) { |
| 24 | key := os.Getenv("DEEPSEEK_API_KEY") |
| 25 | if key == "" { |
| 26 | t.Skip("DEEPSEEK_API_KEY not set") |
| 27 | } |
| 28 | for _, protocol := range []string{"chat", "responses", "anthropic"} { |
| 29 | t.Run(protocol, func(t *testing.T) { |
| 30 | p := officialMatrixProvider(t, key, "deepseek-v4-flash", protocol, "high", "") |
| 31 | runLiveWriteAfterEffectResume(t, p, protocol) |
| 32 | }) |
| 33 | } |
| 34 | } |
| 35 | |
| 36 | func runLiveWriteAfterEffectResume(t *testing.T, p provider.Provider, label string) { |
| 37 | t.Helper() |
| 38 | root := t.TempDir() |
| 39 | target := filepath.Join(root, "marker.txt") |
| 40 | state := filepath.Join(t.TempDir(), "session.jsonl") |
| 41 | lease, err := TryAcquireSessionLease(state) |
| 42 | if err != nil { |
| 43 | t.Fatal(err) |
| 44 | } |
| 45 | defer lease.Release() |
| 46 | ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second) |
| 47 | defer cancel() |
| 48 | var writes atomic.Int32 |
| 49 | reg := tool.NewRegistry() |
| 50 | for _, w := range builtin.ConfineWriters([]string{root}, builtin.SessionDataGuard{}, builtin.ManagedConfigPaths{}) { |
| 51 | if w.Name() == "write_file" { |
| 52 | reg.Add(builtin.BindFileWriteReceipt(w, func(string, bool, []byte) { writes.Add(1); cancel() })) |
| 53 | } |
| 54 | } |
| 55 | sess := NewSession("Follow the user's precise file request. Never repeat a file write whose expected contents are already satisfied.") |
| 56 | sink := &liveWriteCheckpointSink{session: sess, path: state} |
| 57 | a := New(p, reg, sess, Options{MaxSteps: 4, MaxOutputTokens: 2048, MissingReasoningWarnStateDir: t.TempDir()}, sink) |
| 58 | a.SetSessionPath(state) |
| 59 | if err := a.Run(ctx, fmt.Sprintf("Use write_file once to write exactly the text live-write-marker (no newline) to %s. Then report completion.", target)); err == nil { |
| 60 | t.Fatal("expected interrupted run") |
| 61 | } |
| 62 | data, err := os.ReadFile(target) |
| 63 | if err != nil || string(data) != "live-write-marker" { |
| 64 | t.Fatalf("actual file mismatch, err=%v bytes=%d", err, len(data)) |
| 65 | } |
| 66 | if writes.Load() != 1 || sink.checkpoints.Load() != 1 { |
| 67 | t.Fatalf("writes=%d checkpoints=%d", writes.Load(), sink.checkpoints.Load()) |
| 68 | } |
| 69 | reopened, err := LoadSession(state) |
| 70 | if err != nil { |
| 71 | t.Fatal(err) |
| 72 | } |
| 73 | intents, results, unknown := 0, 0, 0 |
| 74 | for _, m := range reopened.Snapshot() { |
| 75 | for _, c := range m.ToolCalls { |
| 76 | intents += len(c.WriteIntents) |
| 77 | } |
| 78 | if m.Role == provider.RoleTool { |
| 79 | if provider.ToolResultRunState(m) == provider.ToolRunUnknown { |
| 80 | unknown++ |
| 81 | } else { |
| 82 | results++ |
| 83 | } |
| 84 | } |
| 85 | } |
| 86 | if intents != 1 || results != 0 || unknown != 1 { |
| 87 | t.Fatalf("pre-result checkpoint intents=%d results=%d unknown=%d", intents, results, unknown) |
| 88 | } |
| 89 | resume := New(p, reg, reopened, Options{MaxSteps: 4, MaxOutputTokens: 2048, MissingReasoningWarnStateDir: t.TempDir()}, event.Discard) |
| 90 | resume.SetSessionPath(state) |
| 91 | next, done := context.WithTimeout(context.Background(), 90*time.Second) |
| 92 | defer done() |
| 93 | history := snapshotLiveWriteHistory(t, resume) |
| 94 | if err := resume.Run(next, "Continue from the interrupted operation. The host has read the target file and verified that its contents are exactly live-write-marker. This establishes current file state, not a historical execution outcome. Do not rewrite satisfied content; just report whether the target is satisfied."); err != nil { |
| 95 | t.Fatal(err) |
| 96 | } |
| 97 | data, err = os.ReadFile(target) |
| 98 | if err != nil || string(data) != "live-write-marker" || writes.Load() != 1 { |
| 99 | t.Fatalf("resume changed file or repeated write: writes=%d err=%v", writes.Load(), err) |
| 100 | } |
| 101 | if snapshotLiveWriteHistory(t, resume) != history { |
| 102 | t.Fatal("continuation rewrote historical execution facts") |
| 103 | } |
| 104 | t.Logf("protocol=%s intent_checkpoints=%d disk_writes=%d original_results=%d reopened_messages=%d", label, sink.checkpoints.Load(), writes.Load(), results, len(reopened.Snapshot())) |
| 105 | } |
| 106 | |
| 107 | type liveWriteCheckpointSink struct { |
| 108 | session *Session |
| 109 | path string |
| 110 | checkpoints atomic.Int32 |
| 111 | } |
| 112 | |
| 113 | func (s *liveWriteCheckpointSink) Emit(event.Event) {} |
| 114 | func (s *liveWriteCheckpointSink) EmitChecked(e event.Event) error { |
| 115 | if e.WriteIntent { |
| 116 | if err := s.session.Save(s.path); err != nil { |
| 117 | return err |
| 118 | } |
| 119 | s.checkpoints.Add(1) |
| 120 | } |
| 121 | return nil |
| 122 | } |
| 123 |