| 1 | package agent |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "os" |
| 8 | "path/filepath" |
| 9 | "strings" |
| 10 | "sync" |
| 11 | "testing" |
| 12 | "time" |
| 13 | |
| 14 | "reasonix/internal/fileutil" |
| 15 | "reasonix/internal/provider" |
| 16 | "reasonix/internal/store" |
| 17 | ) |
| 18 | |
| 19 | func TestDAGConcurrentAppendersKeepBothHeads(t *testing.T) { |
| 20 | path := dagTestSession(t) |
| 21 | _, base := dagLinearLog(t, path) |
| 22 | dagAppend(t, path, sessionDAGEntry{Type: sessionDAGTypeFork, Head: SessionMainHead, NewHead: "B", From: "U2", Kind: HeadKindConcurrent, Writer: "w-b", At: base.Add(time.Minute)}) |
| 23 | const perWriter = 25 |
| 24 | var wg sync.WaitGroup |
| 25 | run := func(head, writer, parent string) { |
| 26 | defer wg.Done() |
| 27 | for i := range perWriter { |
| 28 | id := head + "-" + string(rune('a'+i%26)) + string(rune('a'+i/26)) |
| 29 | e := dagMessageEntry(t, head, parent, "", dagMsg(provider.RoleUser, head, id), base.Add(time.Duration(i)*time.Millisecond)) |
| 30 | e.Writer = writer |
| 31 | unlock, err := lockSessionFile(path) |
| 32 | if err != nil { |
| 33 | t.Errorf("lock: %v", err) |
| 34 | return |
| 35 | } |
| 36 | _, err = appendSessionDAGEntries(path, []sessionDAGEntry{e}, false) |
| 37 | unlock() |
| 38 | if err != nil { |
| 39 | t.Errorf("append: %v", err) |
| 40 | return |
| 41 | } |
| 42 | parent = id |
| 43 | } |
| 44 | } |
| 45 | wg.Add(2) |
| 46 | go run(SessionMainHead, "w-a", "U2") |
| 47 | go run("B", "w-b", "U2") |
| 48 | wg.Wait() |
| 49 | st := dagReplay(t, path) |
| 50 | if st.damaged { |
| 51 | t.Fatal("interleaved appends damaged the log") |
| 52 | } |
| 53 | if got := len(dagChain(st, SessionMainHead)); got != 4+perWriter { |
| 54 | t.Fatalf("main chain length %d", got) |
| 55 | } |
| 56 | if got := len(dagChain(st, "B")); got != 4+perWriter { |
| 57 | t.Fatalf("B chain length %d", got) |
| 58 | } |
| 59 | if st.writers["w-a"] == nil || st.writers["w-b"] == nil { |
| 60 | t.Fatalf("writers = %v", st.writers) |
| 61 | } |
| 62 | entries, _ := os.ReadDir(filepath.Dir(path)) |
| 63 | for _, entry := range entries { |
| 64 | if store.IsSessionTranscriptName(entry.Name()) { |
| 65 | t.Fatalf("concurrent writers created a transcript copy: %s", entry.Name()) |
| 66 | } |
| 67 | } |
| 68 | } |
| 69 | |
| 70 | func TestDAGIncrementalReplayFromKnownTail(t *testing.T) { |
| 71 | path := dagTestSession(t) |
| 72 | _, base := dagLinearLog(t, path) |
| 73 | st := dagReplay(t, path) |
| 74 | tail := st.lastGoodEnd |
| 75 | dagAppend(t, path, dagMessageEntry(t, SessionMainHead, "U2", "", dagMsg(provider.RoleAssistant, "a2", "A2"), base.Add(time.Hour))) |
| 76 | if err := st.replayFrom(context.Background(), tail, defaultSessionReplayLimits); err != nil { |
| 77 | t.Fatal(err) |
| 78 | } |
| 79 | if st.heads[SessionMainHead].leaf != "A2" || st.records != 6 || st.lastGoodEnd <= tail { |
| 80 | t.Fatalf("leaf %q records %d tail %d>%d", st.heads[SessionMainHead].leaf, st.records, st.lastGoodEnd, tail) |
| 81 | } |
| 82 | } |
| 83 | |
| 84 | func TestDAGUpgradeFromSchemaOneKeepsIDsAndTranscript(t *testing.T) { |
| 85 | useSchemaOneLog(t) |
| 86 | path := dagTestSession(t) |
| 87 | v1 := &Session{Messages: []provider.Message{ |
| 88 | {Role: provider.RoleSystem, Content: "sys"}, |
| 89 | {Role: provider.RoleUser, Content: "q1", CreatedAt: 1}, |
| 90 | {Role: provider.RoleAssistant, Content: "a1"}, |
| 91 | }} |
| 92 | if err := v1.Save(path); err != nil { |
| 93 | t.Fatalf("v1 save: %v", err) |
| 94 | } |
| 95 | checkpointBefore, _ := os.ReadFile(path) |
| 96 | loaded, err := LoadSession(path) |
| 97 | if err != nil { |
| 98 | t.Fatal(err) |
| 99 | } |
| 100 | if _, ok := loaded.Head(); ok { |
| 101 | t.Fatal("schema-1 session must not report a head") |
| 102 | } |
| 103 | now := time.Date(2026, 9, 8, 12, 0, 0, 0, time.UTC) |
| 104 | inFlight := &InFlightTurnMeta{ID: "turn-7", StartMessageIndex: 3, PreserveUser: true, StartedAt: now.Add(-time.Minute)} |
| 105 | if err := upgradeSessionLogToDAG(path, loaded.Messages, nil, inFlight, now); err != nil { |
| 106 | t.Fatalf("upgrade: %v", err) |
| 107 | } |
| 108 | probe, err := probeSessionEventLog(path) |
| 109 | if err != nil || !probe.dag { |
| 110 | t.Fatalf("probe after upgrade = %+v err=%v", probe, err) |
| 111 | } |
| 112 | st := dagReplay(t, path) |
| 113 | if st.generation != 1 || st.upgradedFrom != sessionEventSchemaVersion { |
| 114 | t.Fatalf("header generation=%d upgradedFrom=%d", st.generation, st.upgradedFrom) |
| 115 | } |
| 116 | if turn := st.heads[SessionMainHead].openTurn; turn == nil || turn.turn != "turn-7" || turn.leaf != loaded.Messages[2].ID || !turn.preserveUser { |
| 117 | t.Fatalf("open turn = %+v", turn) |
| 118 | } |
| 119 | again, err := LoadSession(path) |
| 120 | if err != nil { |
| 121 | t.Fatal(err) |
| 122 | } |
| 123 | if len(again.Messages) != len(loaded.Messages) { |
| 124 | t.Fatalf("len %d vs %d", len(again.Messages), len(loaded.Messages)) |
| 125 | } |
| 126 | for i := range loaded.Messages { |
| 127 | if again.Messages[i].ID != loaded.Messages[i].ID || again.Messages[i].Content != loaded.Messages[i].Content { |
| 128 | t.Fatalf("message %d changed across upgrade: %+v vs %+v", i, again.Messages[i], loaded.Messages[i]) |
| 129 | } |
| 130 | } |
| 131 | if ref, ok := again.Head(); !ok || ref.HeadID != SessionMainHead || ref.LeafID != loaded.Messages[2].ID { |
| 132 | t.Fatalf("head after upgrade = %+v ok=%v", ref, ok) |
| 133 | } |
| 134 | if b, _ := os.ReadFile(path); string(b) != string(checkpointBefore) { |
| 135 | t.Fatal("upgrade must not touch the .jsonl checkpoint") |
| 136 | } |
| 137 | if err := writeSessionDAGIndex(context.Background(), path, st); err != nil { |
| 138 | t.Fatal(err) |
| 139 | } |
| 140 | idx, err := ReadSessionHeadIndex(path) |
| 141 | if err != nil || idx == nil || !idx.Current(path) || idx.SelectedHead != SessionMainHead || idx.MessageCount != 3 || len(idx.Heads) != 1 { |
| 142 | t.Fatalf("index = %+v err=%v", idx, err) |
| 143 | } |
| 144 | if _, err := readSessionEventIndex(path); err == nil { |
| 145 | t.Fatal("schema-1 index reader must reject the schema-2 index") |
| 146 | } |
| 147 | dagAppend(t, path, dagMessageEntry(t, SessionMainHead, loaded.Messages[2].ID, "", dagMsg(provider.RoleUser, "q2", "N1"), now)) |
| 148 | if idx.Current(path) { |
| 149 | t.Fatal("index must go stale once the log grows") |
| 150 | } |
| 151 | } |
| 152 | |
| 153 | func TestDAGRotationDropsUnreachableAndAppliesRedactions(t *testing.T) { |
| 154 | path := dagTestSession(t) |
| 155 | _, base := dagLinearLog(t, path) |
| 156 | replacement, err := encodeSessionDAGMessage(dagMsg(provider.RoleAssistant, "[gone]", "")) |
| 157 | if err != nil { |
| 158 | t.Fatal(err) |
| 159 | } |
| 160 | patchedU2, err := encodeSessionDAGMessage(provider.Message{Role: provider.RoleUser, Content: "q2", Edited: true}) |
| 161 | if err != nil { |
| 162 | t.Fatal(err) |
| 163 | } |
| 164 | dagAppend(t, path, |
| 165 | sessionDAGEntry{Type: sessionDAGTypeFork, Head: SessionMainHead, NewHead: "F", From: "U1", Kind: HeadKindFork, Name: "side", At: base.Add(time.Minute)}, |
| 166 | dagMessageEntry(t, "F", "U1", "", dagMsg(provider.RoleAssistant, "side-answer-secret", "F1"), base.Add(2*time.Minute)), |
| 167 | sessionDAGEntry{Type: sessionDAGTypeRetire, Head: "F", At: base.Add(3 * time.Minute)}, |
| 168 | sessionDAGEntry{Type: sessionDAGTypeRedact, Head: SessionMainHead, Targets: map[string]json.RawMessage{"A1": replacement}, At: base.Add(4 * time.Minute)}, |
| 169 | sessionDAGEntry{Type: sessionDAGTypePatch, Head: SessionMainHead, Target: "U2", Msgs: patchedU2, At: base.Add(5 * time.Minute)}, |
| 170 | sessionDAGEntry{Type: sessionDAGTypeRename, Head: SessionMainHead, Name: "primary", At: base.Add(6 * time.Minute)}, |
| 171 | sessionDAGEntry{Type: sessionDAGTypeCompaction, Head: SessionMainHead, CoveredLeaf: "A1", CoveredCount: 3, PrefixHash: "h", At: base.Add(7 * time.Minute)}, |
| 172 | sessionDAGEntry{Type: sessionDAGTypeSelect, Head: SessionMainHead, At: base.Add(8 * time.Minute)}, |
| 173 | ) |
| 174 | st := dagReplay(t, path) |
| 175 | if err := rotateSessionDAG(path, st, base.Add(time.Hour)); err != nil { |
| 176 | t.Fatalf("rotate: %v", err) |
| 177 | } |
| 178 | raw, _ := os.ReadFile(store.SessionEventLog(path)) |
| 179 | if strings.Contains(string(raw), "side-answer-secret") || strings.Contains(string(raw), `"a1"`) { |
| 180 | t.Fatalf("rotated log still carries dropped or redacted bytes:\n%s", raw) |
| 181 | } |
| 182 | if !strings.Contains(string(raw), `"dropped":["F1"]`) || !strings.Contains(string(raw), `"tombstones":["A1"]`) { |
| 183 | t.Fatalf("checkpoint manifest missing:\n%s", raw) |
| 184 | } |
| 185 | after := dagReplay(t, path) |
| 186 | if after.generation != 2 || after.damaged || len(after.nodes) != 4 || len(after.heads) != 1 { |
| 187 | t.Fatalf("after rotation generation=%d damaged=%v nodes=%d heads=%d", after.generation, after.damaged, len(after.nodes), len(after.heads)) |
| 188 | } |
| 189 | msgs, _ := after.materialize(SessionMainHead) |
| 190 | if got := dagContents(msgs); strings.Join(got, ",") != "sys,q1,[gone],q2" { |
| 191 | t.Fatalf("rotated chain %v", got) |
| 192 | } |
| 193 | if !msgs[3].Edited || msgs[2].ID != "A1" || msgs[3].ID != "U2" { |
| 194 | t.Fatalf("patch fold or ids lost: %+v", msgs) |
| 195 | } |
| 196 | main := after.heads[SessionMainHead] |
| 197 | if main.name != "primary" || main.compaction == nil || main.compaction.coveredLeaf != "A1" || after.selected != SessionMainHead { |
| 198 | t.Fatalf("main head metadata lost: name=%q compaction=%+v selected=%q", main.name, main.compaction, after.selected) |
| 199 | } |
| 200 | if len(after.patches) != 0 || len(after.redactions) != 0 { |
| 201 | t.Fatal("rotation must fold overlays physically") |
| 202 | } |
| 203 | } |
| 204 | |
| 205 | func TestDAGSingleWriterProof(t *testing.T) { |
| 206 | path := dagTestSession(t) |
| 207 | _, base := dagLinearLog(t, path) |
| 208 | st := dagReplay(t, path) |
| 209 | now := time.Now().UTC() |
| 210 | var denied *SessionRotationDeniedError |
| 211 | if err := sessionDAGSingleWriterProof(path, st, now); !errors.As(err, &denied) || !strings.Contains(denied.Reason, "lease") { |
| 212 | t.Fatalf("without lease err = %v", err) |
| 213 | } |
| 214 | lease, err := TryAcquireSessionLease(path) |
| 215 | if err != nil { |
| 216 | t.Fatalf("lease: %v", err) |
| 217 | } |
| 218 | defer lease.Release() |
| 219 | if err := sessionDAGSingleWriterProof(path, st, now); err != nil { |
| 220 | t.Fatalf("with lease err = %v", err) |
| 221 | } |
| 222 | recent := dagMessageEntry(t, SessionMainHead, "U2", "", dagMsg(provider.RoleAssistant, "other", "X1"), now.Add(-10*time.Second)) |
| 223 | recent.Writer = "other-writer" |
| 224 | dagAppend(t, path, recent) |
| 225 | st = dagReplay(t, path) |
| 226 | if err := sessionDAGSingleWriterProof(path, st, now); !errors.As(err, &denied) || !strings.Contains(denied.Reason, "other-writer") { |
| 227 | t.Fatalf("recent foreign writer err = %v", err) |
| 228 | } |
| 229 | if err := sessionDAGSingleWriterProof(path, st, now.Add(sessionDAGWriterQuietPeriod+time.Second)); err != nil { |
| 230 | t.Fatalf("quiet foreign writer err = %v", err) |
| 231 | } |
| 232 | _ = base |
| 233 | } |
| 234 | |
| 235 | func TestDAGLogOversizedUsesLiveChains(t *testing.T) { |
| 236 | path := dagTestSession(t) |
| 237 | dagLinearLog(t, path) |
| 238 | st := dagReplay(t, path) |
| 239 | if sessionDAGLogOversized(st) { |
| 240 | t.Fatal("small log must not be oversized") |
| 241 | } |
| 242 | st.size = sessionEventLogCompactFloor*2 + 1 |
| 243 | if !sessionDAGLogOversized(st) { |
| 244 | t.Fatal("log far beyond its live chains must be oversized") |
| 245 | } |
| 246 | } |
| 247 | |
| 248 | func TestDAGCrashPointsLeaveLogUntouched(t *testing.T) { |
| 249 | path := dagTestSession(t) |
| 250 | _, base := dagLinearLog(t, path) |
| 251 | logPath := store.SessionEventLog(path) |
| 252 | before, _ := os.ReadFile(logPath) |
| 253 | for _, op := range []string{"dag-append", "dag-rotate"} { |
| 254 | fileutil.CrashPoint = func(got, _ string) { |
| 255 | if got == op { |
| 256 | panic("crash:" + op) |
| 257 | } |
| 258 | } |
| 259 | func() { |
| 260 | defer func() { |
| 261 | if r := recover(); r == nil { |
| 262 | t.Fatalf("%s: crash point did not fire", op) |
| 263 | } |
| 264 | }() |
| 265 | switch op { |
| 266 | case "dag-append": |
| 267 | dagAppend(t, path, dagMessageEntry(t, SessionMainHead, "U2", "", dagMsg(provider.RoleUser, "lost", "L1"), base)) |
| 268 | case "dag-rotate": |
| 269 | _ = rotateSessionDAG(path, dagReplay(t, path), base) |
| 270 | } |
| 271 | }() |
| 272 | fileutil.CrashPoint = nil |
| 273 | after, _ := os.ReadFile(logPath) |
| 274 | if string(after) != string(before) { |
| 275 | t.Fatalf("%s: log changed despite crash", op) |
| 276 | } |
| 277 | if leftovers, _ := filepath.Glob(filepath.Join(filepath.Dir(path), "*.tmp")); len(leftovers) != 0 { |
| 278 | t.Fatalf("%s: temp files left behind: %v", op, leftovers) |
| 279 | } |
| 280 | } |
| 281 | } |
| 282 |