| 1 | package agent |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "os" |
| 9 | "path/filepath" |
| 10 | "reasonix/internal/fileops" |
| 11 | "reflect" |
| 12 | "strings" |
| 13 | "testing" |
| 14 | |
| 15 | "reasonix/internal/provider" |
| 16 | "reasonix/internal/store" |
| 17 | ) |
| 18 | |
| 19 | func TestDisplayPagerDistinguishesAncientEventsFromDamagedCheckpoint(t *testing.T) { |
| 20 | for _, tc := range []struct { |
| 21 | name, body string |
| 22 | unsupported bool |
| 23 | }{ |
| 24 | {"kind", "{\"kind\":\"user.message\",\"text\":\"old\"}\n", true}, |
| 25 | {"type", "{\"type\":\"model.final\",\"content\":\"old\"}\n", true}, |
| 26 | {"missing-role", "{\"content\":\"broken\"}\n", false}, |
| 27 | {"mixed", "{\"role\":\"user\",\"content\":\"valid\"}\n{\"kind\":\"model.final\",\"content\":\"foreign\"}\n", false}, |
| 28 | } { |
| 29 | t.Run(tc.name, func(t *testing.T) { |
| 30 | path := filepath.Join(t.TempDir(), "source.jsonl") |
| 31 | if err := os.WriteFile(path, []byte(tc.body), 0600); err != nil { |
| 32 | t.Fatal(err) |
| 33 | } |
| 34 | pager, err := OpenDisplayPager(t.Context(), path, filepath.Join(t.TempDir(), "cache.sqlite")) |
| 35 | if pager != nil { |
| 36 | pager.Close() |
| 37 | t.Fatal("invalid checkpoint published a pager") |
| 38 | } |
| 39 | if err == nil || errors.Is(err, ErrDisplayFormatUnsupported) != tc.unsupported { |
| 40 | t.Fatalf("unsupported=%v: %v", tc.unsupported, err) |
| 41 | } |
| 42 | }) |
| 43 | } |
| 44 | } |
| 45 | |
| 46 | func TestDisplayPagerRejectsSameSizeRewriteWithRestoredMtime(t *testing.T) { |
| 47 | path := filepath.Join(t.TempDir(), "source.jsonl") |
| 48 | before := []byte("{\"role\":\"user\",\"content\":\"one\"}\n") |
| 49 | if err := os.WriteFile(path, before, 0600); err != nil { |
| 50 | t.Fatal(err) |
| 51 | } |
| 52 | info, err := os.Stat(path) |
| 53 | if err != nil { |
| 54 | t.Fatal(err) |
| 55 | } |
| 56 | cache := filepath.Join(t.TempDir(), "index.sqlite") |
| 57 | p, err := OpenDisplayPager(t.Context(), path, cache) |
| 58 | if err != nil { |
| 59 | t.Fatal(err) |
| 60 | } |
| 61 | defer p.Close() |
| 62 | if err := os.WriteFile(path, []byte("{\"role\":\"user\",\"content\":\"two\"}\n"), 0600); err != nil { |
| 63 | t.Fatal(err) |
| 64 | } |
| 65 | if err := os.Chtimes(path, info.ModTime(), info.ModTime()); err != nil { |
| 66 | t.Fatal(err) |
| 67 | } |
| 68 | current, err := os.Stat(path) |
| 69 | if err != nil { |
| 70 | t.Fatal(err) |
| 71 | } |
| 72 | _, version := fileops.DiskSnapshot(path, current) |
| 73 | if version == p.sourceVersion { |
| 74 | t.Skip("filesystem does not expose a change-time version") |
| 75 | } |
| 76 | if err := p.Validate(); !errors.Is(err, ErrDisplaySourceChanged) { |
| 77 | t.Fatalf("same-size source replacement accepted: %v", err) |
| 78 | } |
| 79 | oldDigest := p.Header.ContentDigest |
| 80 | p.Close() |
| 81 | next, err := OpenDisplayPager(t.Context(), path, cache) |
| 82 | if err != nil { |
| 83 | t.Fatal(err) |
| 84 | } |
| 85 | defer next.Close() |
| 86 | if next.Header.ContentDigest == oldDigest { |
| 87 | t.Fatal("persistent cache trusted size and restored mtime alone") |
| 88 | } |
| 89 | } |
| 90 | |
| 91 | func TestDisplayPagerRebuildsUntrustedSidecarWithoutModifyingSource(t *testing.T) { |
| 92 | path := filepath.Join(t.TempDir(), "source.jsonl") |
| 93 | body := []byte("{\"role\":\"user\",\"content\":\"original\"}\n") |
| 94 | if err := os.WriteFile(path, body, 0600); err != nil { |
| 95 | t.Fatal(err) |
| 96 | } |
| 97 | sidecar := store.SessionDisplayIndex(path) |
| 98 | if err := os.WriteFile(sidecar, []byte("{broken"), 0600); err != nil { |
| 99 | t.Fatal(err) |
| 100 | } |
| 101 | p, err := OpenDisplayPager(t.Context(), path, filepath.Join(t.TempDir(), "index.sqlite")) |
| 102 | if err != nil { |
| 103 | t.Fatal(err) |
| 104 | } |
| 105 | defer p.Close() |
| 106 | if p.Header.MessageCount != 1 { |
| 107 | t.Fatalf("rebuilt count=%d", p.Header.MessageCount) |
| 108 | } |
| 109 | after, _ := os.ReadFile(path) |
| 110 | idx, _ := os.ReadFile(sidecar) |
| 111 | if string(after) != string(body) || string(idx) != "{broken" { |
| 112 | t.Fatal("display preparation repaired authoritative storage or its sidecar") |
| 113 | } |
| 114 | } |
| 115 | |
| 116 | func TestDisplayPagerCheckpointIsBoundedAndReadOnly(t *testing.T) { |
| 117 | dir := t.TempDir() |
| 118 | path := filepath.Join(dir, "old.jsonl") |
| 119 | var original []byte |
| 120 | for _, m := range displayIndexTestMessages() { |
| 121 | body, err := json.Marshal(m) |
| 122 | if err != nil { |
| 123 | t.Fatal(err) |
| 124 | } |
| 125 | original = append(original, append(body, '\n')...) |
| 126 | } |
| 127 | if err := os.WriteFile(path, original, 0600); err != nil { |
| 128 | t.Fatal(err) |
| 129 | } |
| 130 | cache := filepath.Join(t.TempDir(), "display.sqlite") |
| 131 | p, err := OpenDisplayPager(t.Context(), path, cache) |
| 132 | if err != nil { |
| 133 | t.Fatal(err) |
| 134 | } |
| 135 | if len(p.Header.Entries) != 0 || p.Header.MessageCount != len(displayIndexTestMessages()) || p.Header.AuthoredTurns != 2 { |
| 136 | t.Fatalf("unbounded or incorrect header: %+v", p.Header) |
| 137 | } |
| 138 | entries, err := p.Entries(1, 3) |
| 139 | if err != nil || len(entries) != 2 || entries[0].Role != provider.RoleUser { |
| 140 | t.Fatalf("page: %+v %v", entries, err) |
| 141 | } |
| 142 | if _, err := p.Entries(0, 501); err == nil { |
| 143 | t.Fatal("unbounded page accepted") |
| 144 | } |
| 145 | turns, err := p.TurnEntries(2, 1) |
| 146 | if err != nil || len(turns) != 1 || !turns[0].StartsTurn || turns[0].AuthoredTurn != 2 { |
| 147 | t.Fatalf("authored-turn page: %+v %v", turns, err) |
| 148 | } |
| 149 | var queryID, parentID, unused int |
| 150 | var plan string |
| 151 | if err := p.DB.QueryRowContext(t.Context(), `EXPLAIN QUERY PLAN SELECT entry FROM entries WHERE turn>=2 AND json_extract(CAST(entry AS TEXT),'$.starts_turn')=1 ORDER BY turn,position LIMIT 1`).Scan(&queryID, &parentID, &unused, &plan); err != nil || !strings.Contains(plan, "SEARCH entries USING INDEX entries_authored_turn") { |
| 152 | t.Fatalf("outline prefix scan: %s %v", plan, err) |
| 153 | } |
| 154 | if _, err := p.TurnEntries(1, 1001); err == nil { |
| 155 | t.Fatal("unbounded outline accepted") |
| 156 | } |
| 157 | canceled, cancel := context.WithCancel(t.Context()) |
| 158 | cancel() |
| 159 | if _, err := p.WithContext(canceled).TurnEntries(1, 1); !errors.Is(err, context.Canceled) { |
| 160 | t.Fatalf("canceled outline read: %v", err) |
| 161 | } |
| 162 | digest := p.Header.ContentDigest |
| 163 | if err := p.Close(); err != nil { |
| 164 | t.Fatal(err) |
| 165 | } |
| 166 | p, err = OpenDisplayPager(t.Context(), path, cache) |
| 167 | if err != nil { |
| 168 | t.Fatal(err) |
| 169 | } |
| 170 | defer p.Close() |
| 171 | if p.Header.ContentDigest != digest { |
| 172 | t.Fatal("reopening changed snapshot identity") |
| 173 | } |
| 174 | got, _ := os.ReadFile(path) |
| 175 | if !reflect.DeepEqual(got, original) { |
| 176 | t.Fatal("display preparation changed authoritative content") |
| 177 | } |
| 178 | if _, err := os.Stat(store.SessionDisplayIndex(path)); !os.IsNotExist(err) { |
| 179 | t.Fatal("read created a session sidecar") |
| 180 | } |
| 181 | replacement := filepath.Join(dir, "replacement") |
| 182 | if err := os.WriteFile(replacement, original, 0600); err != nil { |
| 183 | t.Fatal(err) |
| 184 | } |
| 185 | if err := os.Rename(replacement, path); err != nil { |
| 186 | t.Fatal(err) |
| 187 | } |
| 188 | if err := p.Validate(); err == nil { |
| 189 | t.Fatal("source replacement retained stale offsets") |
| 190 | } |
| 191 | } |
| 192 | |
| 193 | func TestDisplayPagerCancellationDoesNotPublishPartialGeneration(t *testing.T) { |
| 194 | path := filepath.Join(t.TempDir(), "old.jsonl") |
| 195 | if err := os.WriteFile(path, []byte("{\"role\":\"user\",\"content\":\"one\"}\n"), 0600); err != nil { |
| 196 | t.Fatal(err) |
| 197 | } |
| 198 | cache := filepath.Join(t.TempDir(), "display.sqlite") |
| 199 | ctx, cancel := context.WithCancel(t.Context()) |
| 200 | cancel() |
| 201 | if p, err := OpenDisplayPager(ctx, path, cache); err == nil { |
| 202 | p.Close() |
| 203 | t.Fatal("cancelled build succeeded") |
| 204 | } |
| 205 | p, err := OpenDisplayPager(t.Context(), path, cache) |
| 206 | if err != nil { |
| 207 | t.Fatal(err) |
| 208 | } |
| 209 | defer p.Close() |
| 210 | if p.Header.MessageCount != 1 { |
| 211 | t.Fatal("cancelled generation was published") |
| 212 | } |
| 213 | } |
| 214 | |
| 215 | func TestDisplayPagerRejectsNewEventAuthority(t *testing.T) { |
| 216 | for _, existingEmpty := range []bool{false, true} { |
| 217 | t.Run(fmt.Sprint(existingEmpty), func(t *testing.T) { |
| 218 | path := filepath.Join(t.TempDir(), "checkpoint.jsonl") |
| 219 | if err := os.WriteFile(path, []byte("{\"role\":\"user\",\"content\":\"one\"}\n"), 0600); err != nil { |
| 220 | t.Fatal(err) |
| 221 | } |
| 222 | log := store.SessionEventLog(path) |
| 223 | if existingEmpty { |
| 224 | if err := os.WriteFile(log, nil, 0600); err != nil { |
| 225 | t.Fatal(err) |
| 226 | } |
| 227 | } |
| 228 | p, err := OpenDisplayPager(t.Context(), path, filepath.Join(t.TempDir(), "index.sqlite")) |
| 229 | if err != nil { |
| 230 | t.Fatal(err) |
| 231 | } |
| 232 | defer p.Close() |
| 233 | // No checkpoint byte or timestamp changes. Even an incomplete new |
| 234 | // event log prevents the old view from claiming complete authority. |
| 235 | if err := os.WriteFile(log, []byte("{\"schema_version\":2}\n"), 0600); err != nil { |
| 236 | t.Fatal(err) |
| 237 | } |
| 238 | if err := p.Validate(); !errors.Is(err, ErrDisplaySourceChanged) { |
| 239 | t.Fatalf("new event authority accepted: %v", err) |
| 240 | } |
| 241 | }) |
| 242 | } |
| 243 | } |
| 244 |