| 1 | package session |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "encoding/json" |
| 7 | "errors" |
| 8 | "fmt" |
| 9 | "os" |
| 10 | "path/filepath" |
| 11 | "slices" |
| 12 | "strings" |
| 13 | "sync" |
| 14 | "testing" |
| 15 | |
| 16 | "reasonix/internal/event" |
| 17 | "reasonix/internal/provider" |
| 18 | ) |
| 19 | |
| 20 | // newSourceService builds the service and one live source session the same way |
| 21 | // service_test.go does. |
| 22 | func newSourceService(t *testing.T, sessionID string) (*Service, string, *Runtime) { |
| 23 | t.Helper() |
| 24 | root := filepath.Join(t.TempDir(), "sessions-v4") |
| 25 | service, err := NewService("local", NewFilesystemPersistence(root)) |
| 26 | if err != nil { |
| 27 | t.Fatal(err) |
| 28 | } |
| 29 | t.Cleanup(func() { _ = service.CloseAll(context.Background()) }) |
| 30 | runtime, err := service.Create(t.Context(), CreateOptions{SessionID: sessionID}) |
| 31 | if err != nil { |
| 32 | t.Fatal(err) |
| 33 | } |
| 34 | return service, root, runtime |
| 35 | } |
| 36 | |
| 37 | // appendCompletedTurn commits one whole turn as a single batch and returns that |
| 38 | // commit. The reply is an assistant message because only an assistant message |
| 39 | // becomes the turn's MessageID. |
| 40 | func appendCompletedTurn(t *testing.T, runtime *Runtime, turnID, messageID string) Commit { |
| 41 | t.Helper() |
| 42 | payload, err := json.Marshal(map[string]any{"message": provider.Message{ID: messageID, Role: provider.RoleAssistant, Content: messageID}}) |
| 43 | if err != nil { |
| 44 | t.Fatal(err) |
| 45 | } |
| 46 | commit, err := runtime.Session().Append(t.Context(), Batch{OperationID: turnID, TurnID: turnID, Events: []Event{ |
| 47 | {Kind: "turn/start"}, |
| 48 | {Kind: "message/complete", Payload: payload}, |
| 49 | {Kind: "turn/end", Payload: json.RawMessage(`{"status":"completed"}`)}, |
| 50 | }}) |
| 51 | if err != nil { |
| 52 | t.Fatal(err) |
| 53 | } |
| 54 | return commit |
| 55 | } |
| 56 | |
| 57 | // completedTurnTarget is the fork target of a turn committed by |
| 58 | // appendCompletedTurn. Its closing commit ends at turn/end, so the turn's end |
| 59 | // sequence is also its boundary sequence. |
| 60 | func completedTurnTarget(commit Commit, turnID, messageID string, number int) ForkTarget { |
| 61 | return ForkTarget{ |
| 62 | TurnID: turnID, TurnNumber: number, |
| 63 | StartSequence: commit.FirstSequence, EndSequence: commit.LastSequence(), |
| 64 | BoundarySequence: commit.LastSequence(), |
| 65 | Status: event.TurnCompleted, MessageID: messageID, Available: true, |
| 66 | } |
| 67 | } |
| 68 | |
| 69 | // turnSource is a source session with two completed turns, closed before a test |
| 70 | // reads it, so every read is a cold read with no runtime in this process. |
| 71 | type turnSource struct { |
| 72 | root string |
| 73 | service *Service |
| 74 | ref SessionRef |
| 75 | first Commit |
| 76 | second Commit |
| 77 | } |
| 78 | |
| 79 | func newClosedTurnSource(t *testing.T) turnSource { |
| 80 | t.Helper() |
| 81 | service, root, runtime := newSourceService(t, "source") |
| 82 | source := turnSource{root: root, service: service, ref: runtime.Ref()} |
| 83 | source.first = appendCompletedTurn(t, runtime, "turn-1", "message-1") |
| 84 | source.second = appendCompletedTurn(t, runtime, "turn-2", "message-2") |
| 85 | if err := service.Close(t.Context(), source.ref); err != nil { |
| 86 | t.Fatal(err) |
| 87 | } |
| 88 | return source |
| 89 | } |
| 90 | |
| 91 | // openTurnSource is a source session whose second turn has started without |
| 92 | // ending. Its runtime stays live so a test can fork while the source runs. |
| 93 | type openTurnSource struct { |
| 94 | root string |
| 95 | service *Service |
| 96 | ref SessionRef |
| 97 | first Commit |
| 98 | open Commit |
| 99 | } |
| 100 | |
| 101 | func newOpenTurnSource(t *testing.T) openTurnSource { |
| 102 | t.Helper() |
| 103 | service, root, runtime := newSourceService(t, "source") |
| 104 | source := openTurnSource{root: root, service: service, ref: runtime.Ref()} |
| 105 | source.first = appendCompletedTurn(t, runtime, "turn-1", "message-1") |
| 106 | open, err := runtime.Session().Append(t.Context(), Batch{OperationID: "turn-2-start", TurnID: "turn-2", Events: []Event{{Kind: "turn/start"}}}) |
| 107 | if err != nil { |
| 108 | t.Fatal(err) |
| 109 | } |
| 110 | source.open = open |
| 111 | return source |
| 112 | } |
| 113 | |
| 114 | // forkChildDirs names the published child sessions under a sessions root. The |
| 115 | // source session and the dot-prefixed derived caches are not children. |
| 116 | func forkChildDirs(t *testing.T, root, sourceID string) []string { |
| 117 | t.Helper() |
| 118 | entries, err := os.ReadDir(root) |
| 119 | if err != nil { |
| 120 | t.Fatal(err) |
| 121 | } |
| 122 | children := []string{} |
| 123 | for _, entry := range entries { |
| 124 | if entry.IsDir() && entry.Name() != sourceID && !strings.HasPrefix(entry.Name(), ".") { |
| 125 | children = append(children, entry.Name()) |
| 126 | } |
| 127 | } |
| 128 | slices.Sort(children) |
| 129 | return children |
| 130 | } |
| 131 | |
| 132 | func TestForkTargetsListCompletedTurnForClosedSession(t *testing.T) { |
| 133 | source := newClosedTurnSource(t) |
| 134 | set, err := source.service.ForkTargetSetFor(t.Context(), source.ref) |
| 135 | if err != nil { |
| 136 | t.Fatal(err) |
| 137 | } |
| 138 | if !set.Verifiable { |
| 139 | t.Fatal("closed source with turn records reported unverifiable history") |
| 140 | } |
| 141 | if len(set.Targets) != 2 { |
| 142 | t.Fatalf("targets = %+v", set.Targets) |
| 143 | } |
| 144 | if want := completedTurnTarget(source.first, "turn-1", "message-1", 1); set.Targets[0] != want { |
| 145 | t.Fatalf("first target = %+v, want %+v", set.Targets[0], want) |
| 146 | } |
| 147 | if want := completedTurnTarget(source.second, "turn-2", "message-2", 2); set.Targets[1] != want { |
| 148 | t.Fatalf("second target = %+v, want %+v", set.Targets[1], want) |
| 149 | } |
| 150 | snapshot, err := source.service.Query().Snapshot(t.Context(), source.ref) |
| 151 | if err != nil { |
| 152 | t.Fatal(err) |
| 153 | } |
| 154 | turns := snapshot.Projection.Turns |
| 155 | if len(turns) != 2 { |
| 156 | t.Fatalf("turns = %+v", turns) |
| 157 | } |
| 158 | if turns[0].BoundarySequence == 0 || turns[0].BoundarySequence != source.first.LastSequence() { |
| 159 | t.Fatalf("first boundary sequence = %d, want %d", turns[0].BoundarySequence, source.first.LastSequence()) |
| 160 | } |
| 161 | if turns[1].BoundarySequence != source.second.LastSequence() { |
| 162 | t.Fatalf("second boundary sequence = %d, want %d", turns[1].BoundarySequence, source.second.LastSequence()) |
| 163 | } |
| 164 | } |
| 165 | |
| 166 | func TestCreateForkFromColdSourceInheritsOnlyPrefixThroughTurn(t *testing.T) { |
| 167 | source := newClosedTurnSource(t) |
| 168 | result, err := source.service.CreateFork(t.Context(), ForkRequest{ |
| 169 | Source: source.ref, TurnID: "turn-1", BoundarySequence: source.first.LastSequence(), ChildID: "child-after-turn-1", OperationID: "fork-turn-1", |
| 170 | }) |
| 171 | if err != nil { |
| 172 | t.Fatal(err) |
| 173 | } |
| 174 | if want := (SessionRef{HostID: "local", SessionID: "child-after-turn-1"}); result.Child != want { |
| 175 | t.Fatalf("child ref = %+v, want %+v", result.Child, want) |
| 176 | } |
| 177 | if want := completedTurnTarget(source.first, "turn-1", "message-1", 1); result.Turn != want { |
| 178 | t.Fatalf("fork turn = %+v, want %+v", result.Turn, want) |
| 179 | } |
| 180 | child, err := Open(filepath.Join(source.root, "child-after-turn-1"), "child-after-turn-1") |
| 181 | if err != nil { |
| 182 | t.Fatal(err) |
| 183 | } |
| 184 | defer child.Close(context.Background()) |
| 185 | projection := child.Snapshot().Projection |
| 186 | if len(projection.Messages) != 1 || projection.Messages[0].ID != "message-1" { |
| 187 | t.Fatalf("child messages = %+v", projection.Messages) |
| 188 | } |
| 189 | if len(projection.Turns) != 1 || projection.Turns[0].TurnID != "turn-1" { |
| 190 | t.Fatalf("child turns = %+v", projection.Turns) |
| 191 | } |
| 192 | if projection.TurnID != "" { |
| 193 | t.Fatalf("child inherited open turn %q", projection.TurnID) |
| 194 | } |
| 195 | if got := child.Manifest().InheritedEvents; got != source.first.LastSequence() { |
| 196 | t.Fatalf("inherited events = %d, want %d", got, source.first.LastSequence()) |
| 197 | } |
| 198 | if got := child.Snapshot().EventSequence; got != source.first.LastSequence() { |
| 199 | t.Fatalf("child sequence = %d, want %d", got, source.first.LastSequence()) |
| 200 | } |
| 201 | } |
| 202 | |
| 203 | func TestForkTargetsIncludeOpenTrailingTurnWhileEarlierTurnStaysForkable(t *testing.T) { |
| 204 | source := newOpenTurnSource(t) |
| 205 | set, err := source.service.ForkTargetSetFor(t.Context(), source.ref) |
| 206 | if err != nil { |
| 207 | t.Fatal(err) |
| 208 | } |
| 209 | if !set.Verifiable || len(set.Targets) != 2 { |
| 210 | t.Fatalf("targets = %+v", set) |
| 211 | } |
| 212 | if want := completedTurnTarget(source.first, "turn-1", "message-1", 1); set.Targets[0] != want { |
| 213 | t.Fatalf("completed target = %+v, want %+v", set.Targets[0], want) |
| 214 | } |
| 215 | wantOpen := ForkTarget{ |
| 216 | TurnID: "turn-2", TurnNumber: 2, StartSequence: source.open.FirstSequence, |
| 217 | Status: event.TurnInProgress, Reason: ForkTurnOpen, |
| 218 | } |
| 219 | if set.Targets[1] != wantOpen { |
| 220 | t.Fatalf("open target = %+v, want %+v", set.Targets[1], wantOpen) |
| 221 | } |
| 222 | result, err := source.service.CreateFork(t.Context(), ForkRequest{ |
| 223 | Source: source.ref, TurnID: "turn-1", BoundarySequence: source.first.LastSequence(), ChildID: "child-after-turn-1", OperationID: "fork-turn-1", |
| 224 | }) |
| 225 | if err != nil { |
| 226 | t.Fatal(err) |
| 227 | } |
| 228 | child, err := Open(filepath.Join(source.root, result.Child.SessionID), result.Child.SessionID) |
| 229 | if err != nil { |
| 230 | t.Fatal(err) |
| 231 | } |
| 232 | defer child.Close(context.Background()) |
| 233 | projection := child.Snapshot().Projection |
| 234 | if len(projection.Messages) != 1 || len(projection.Turns) != 1 || projection.Turns[0].TurnID != "turn-1" { |
| 235 | t.Fatalf("child projection = %+v", projection) |
| 236 | } |
| 237 | after, err := source.service.ForkTargetSetFor(t.Context(), source.ref) |
| 238 | if err != nil { |
| 239 | t.Fatal(err) |
| 240 | } |
| 241 | if len(after.Targets) != 2 || after.Targets[1] != wantOpen { |
| 242 | t.Fatalf("source targets after fork = %+v", after.Targets) |
| 243 | } |
| 244 | runtime, ok := source.service.Runtime(source.ref) |
| 245 | if !ok { |
| 246 | t.Fatal("forking closed the live source runtime") |
| 247 | } |
| 248 | if turnID := runtime.Session().Snapshot().Projection.TurnID; turnID != "turn-2" { |
| 249 | t.Fatalf("source open turn = %q", turnID) |
| 250 | } |
| 251 | } |
| 252 | |
| 253 | func TestForkSequenceRefusesUnknownTurnAndCreateForkRefusesOpenTurn(t *testing.T) { |
| 254 | source := newOpenTurnSource(t) |
| 255 | if err := source.service.Close(t.Context(), source.ref); err != nil { |
| 256 | t.Fatal(err) |
| 257 | } |
| 258 | snapshot, err := source.service.Query().Snapshot(t.Context(), source.ref) |
| 259 | if err != nil { |
| 260 | t.Fatal(err) |
| 261 | } |
| 262 | projection := snapshot.Projection |
| 263 | if sequence, availability, err := ForkSequence(projection, "turn-1"); err != nil || sequence != source.first.LastSequence() || availability != ForkAvailable { |
| 264 | t.Fatalf("completed turn cut = %d, %s, %v", sequence, availability, err) |
| 265 | } |
| 266 | if sequence, availability, err := ForkSequence(projection, "turn-2"); err != nil || sequence != 0 || availability != ForkTurnOpen { |
| 267 | t.Fatalf("open turn cut = %d, %s, %v", sequence, availability, err) |
| 268 | } |
| 269 | sequence, availability, err := ForkSequence(projection, "no-such-turn") |
| 270 | if err != nil { |
| 271 | t.Fatal(err) |
| 272 | } |
| 273 | if sequence != 0 || availability != ForkHistoryUnverifiable { |
| 274 | t.Fatalf("unknown turn cut = %d, %s", sequence, availability) |
| 275 | } |
| 276 | if number, availability, err := ForkSequenceForNumber(projection, 1); err != nil || number != source.first.LastSequence() || availability != ForkAvailable { |
| 277 | t.Fatalf("turn number 1 cut = %d, %s, %v", number, availability, err) |
| 278 | } |
| 279 | if number, availability, err := ForkSequenceForNumber(projection, 3); err == nil || number != 0 || availability != ForkHistoryUnverifiable { |
| 280 | t.Fatalf("out-of-range turn number cut = %d, %s, %v", number, availability, err) |
| 281 | } |
| 282 | childID := "child-from-open-turn" |
| 283 | _, err = source.service.CreateFork(t.Context(), ForkRequest{Source: source.ref, TurnID: "turn-2", ChildID: childID, OperationID: "fork-open-turn"}) |
| 284 | var unavailable *ForkUnavailableError |
| 285 | if !errors.As(err, &unavailable) { |
| 286 | t.Fatalf("open-turn fork error = %v", err) |
| 287 | } |
| 288 | if unavailable.TurnID != "turn-2" || unavailable.Reason != ForkTurnOpen { |
| 289 | t.Fatalf("unavailable = %+v", unavailable) |
| 290 | } |
| 291 | if _, statErr := os.Stat(filepath.Join(source.root, childID)); !os.IsNotExist(statErr) { |
| 292 | t.Fatalf("refused fork left a child directory: %v", statErr) |
| 293 | } |
| 294 | } |
| 295 | |
| 296 | func TestCreateForkInheritsWholeAtomicCommitThatClosedTurn(t *testing.T) { |
| 297 | service, root, runtime := newSourceService(t, "source") |
| 298 | payload, err := json.Marshal(map[string]any{"message": provider.Message{ID: "message-1", Role: provider.RoleAssistant, Content: "one"}}) |
| 299 | if err != nil { |
| 300 | t.Fatal(err) |
| 301 | } |
| 302 | commit, err := runtime.Session().Append(t.Context(), Batch{OperationID: "turn-1", TurnID: "turn-1", Events: []Event{ |
| 303 | {Kind: "turn/start"}, |
| 304 | {Kind: "message/complete", Payload: payload}, |
| 305 | {Kind: "turn/end", Payload: json.RawMessage(`{"status":"completed"}`)}, |
| 306 | {Kind: "diagnostic", Payload: json.RawMessage(`{"note":"committed with the turn"}`)}, |
| 307 | }}) |
| 308 | if err != nil { |
| 309 | t.Fatal(err) |
| 310 | } |
| 311 | turns := runtime.Session().Snapshot().Projection.Turns |
| 312 | if len(turns) != 1 { |
| 313 | t.Fatalf("turns = %+v", turns) |
| 314 | } |
| 315 | if turns[0].EndSequence == commit.LastSequence() { |
| 316 | t.Fatal("turn/end is the last event of this batch, so it cannot show the cut covering the whole commit") |
| 317 | } |
| 318 | if turns[0].BoundarySequence != commit.LastSequence() { |
| 319 | t.Fatalf("boundary sequence = %d, want %d", turns[0].BoundarySequence, commit.LastSequence()) |
| 320 | } |
| 321 | result, err := service.CreateFork(t.Context(), ForkRequest{Source: runtime.Ref(), TurnID: "turn-1", BoundarySequence: commit.LastSequence(), ChildID: "child-atomic", OperationID: "fork-atomic"}) |
| 322 | if err != nil { |
| 323 | t.Fatal(err) |
| 324 | } |
| 325 | // turn/end is the third event of the batch and the diagnostic the fourth, so |
| 326 | // the cut must be the batch's last sequence, not the turn/end sequence. |
| 327 | if endEvent := commit.FirstSequence + 2; result.Turn.EndSequence != endEvent { |
| 328 | t.Fatalf("fork turn end = %d, want the turn/end event %d", result.Turn.EndSequence, endEvent) |
| 329 | } |
| 330 | if result.Turn.EndSequence != turns[0].EndSequence { |
| 331 | t.Fatalf("fork turn end = %d, want %d", result.Turn.EndSequence, turns[0].EndSequence) |
| 332 | } |
| 333 | inherited, err := Replay(filepath.Join(root, "child-atomic"), nil) |
| 334 | if err != nil { |
| 335 | t.Fatal(err) |
| 336 | } |
| 337 | if len(inherited) != 1 || inherited[0].LastSequence() != commit.LastSequence() || len(inherited[0].Events) != len(commit.Events) { |
| 338 | t.Fatalf("inherited commits = %+v", inherited) |
| 339 | } |
| 340 | if kind := inherited[0].Events[len(inherited[0].Events)-1].Kind; kind != "diagnostic" { |
| 341 | t.Fatalf("inherited last event = %q", kind) |
| 342 | } |
| 343 | child, err := Open(filepath.Join(root, "child-atomic"), "child-atomic") |
| 344 | if err != nil { |
| 345 | t.Fatal(err) |
| 346 | } |
| 347 | defer child.Close(context.Background()) |
| 348 | if got := child.Snapshot().EventSequence; got != commit.LastSequence() { |
| 349 | t.Fatalf("child sequence = %d, want %d", got, commit.LastSequence()) |
| 350 | } |
| 351 | childTurns := child.Snapshot().Projection.Turns |
| 352 | if len(childTurns) != 1 || childTurns[0].BoundarySequence != commit.LastSequence() { |
| 353 | t.Fatalf("child turns = %+v", childTurns) |
| 354 | } |
| 355 | } |
| 356 | |
| 357 | func TestCreateForkIsIdempotentPerOperationID(t *testing.T) { |
| 358 | source := newClosedTurnSource(t) |
| 359 | request := ForkRequest{Source: source.ref, TurnID: "turn-1", BoundarySequence: source.first.LastSequence(), OperationID: "fork-retry"} |
| 360 | first, err := source.service.CreateFork(t.Context(), request) |
| 361 | if err != nil { |
| 362 | t.Fatal(err) |
| 363 | } |
| 364 | second, err := source.service.CreateFork(t.Context(), request) |
| 365 | if err != nil { |
| 366 | t.Fatalf("retried fork: %v", err) |
| 367 | } |
| 368 | if first.Child != second.Child { |
| 369 | t.Fatalf("retry published %+v, want %+v", second.Child, first.Child) |
| 370 | } |
| 371 | if children := forkChildDirs(t, source.root, source.ref.SessionID); len(children) != 1 || children[0] != first.Child.SessionID { |
| 372 | t.Fatalf("child directories = %v", children) |
| 373 | } |
| 374 | other, err := source.service.CreateFork(t.Context(), ForkRequest{Source: source.ref, TurnID: "turn-1", BoundarySequence: source.first.LastSequence(), OperationID: "fork-other"}) |
| 375 | if err != nil { |
| 376 | t.Fatal(err) |
| 377 | } |
| 378 | if other.Child == first.Child { |
| 379 | t.Fatal("a different operation id reused the published child") |
| 380 | } |
| 381 | if children := forkChildDirs(t, source.root, source.ref.SessionID); len(children) != 2 { |
| 382 | t.Fatalf("child directories = %v", children) |
| 383 | } |
| 384 | } |
| 385 | |
| 386 | // TestCreateForkConcurrentSameOperationIDPublishesOneChild drives the race two |
| 387 | // callers of one operation id hit: neither holds a reservation, so both can pass |
| 388 | // the child-existence check before either publishes. Every caller must resolve |
| 389 | // as the same idempotent success rather than the losing rename reaching the |
| 390 | // surface as a hard failure, and the source must hold exactly one child. |
| 391 | func TestCreateForkConcurrentSameOperationIDPublishesOneChild(t *testing.T) { |
| 392 | source := newClosedTurnSource(t) |
| 393 | request := ForkRequest{Source: source.ref, TurnID: "turn-1", BoundarySequence: source.first.LastSequence(), OperationID: "fork-race"} |
| 394 | const callers = 4 |
| 395 | start := make(chan struct{}) |
| 396 | results := make([]ForkResult, callers) |
| 397 | errs := make([]error, callers) |
| 398 | var wait sync.WaitGroup |
| 399 | for caller := range callers { |
| 400 | wait.Go(func() { |
| 401 | <-start |
| 402 | results[caller], errs[caller] = source.service.CreateFork(t.Context(), request) |
| 403 | }) |
| 404 | } |
| 405 | close(start) |
| 406 | wait.Wait() |
| 407 | for caller := range callers { |
| 408 | if errs[caller] != nil { |
| 409 | t.Fatalf("caller %d: %v", caller, errs[caller]) |
| 410 | } |
| 411 | if results[caller].Child != results[0].Child { |
| 412 | t.Fatalf("caller %d published %+v, want %+v", caller, results[caller].Child, results[0].Child) |
| 413 | } |
| 414 | if want := completedTurnTarget(source.first, "turn-1", "message-1", 1); results[caller].Turn != want { |
| 415 | t.Fatalf("caller %d turn = %+v, want %+v", caller, results[caller].Turn, want) |
| 416 | } |
| 417 | } |
| 418 | children := forkChildDirs(t, source.root, source.ref.SessionID) |
| 419 | if len(children) != 1 || children[0] != results[0].Child.SessionID { |
| 420 | t.Fatalf("child directories = %v, want exactly [%s]", children, results[0].Child.SessionID) |
| 421 | } |
| 422 | } |
| 423 | |
| 424 | func TestCreateForkConcurrentDifferentOperationIDsPublishDistinctChildren(t *testing.T) { |
| 425 | source := newClosedTurnSource(t) |
| 426 | const callers = 4 |
| 427 | start := make(chan struct{}) |
| 428 | results := make([]ForkResult, callers) |
| 429 | errs := make([]error, callers) |
| 430 | var wait sync.WaitGroup |
| 431 | for caller := range callers { |
| 432 | wait.Go(func() { |
| 433 | <-start |
| 434 | results[caller], errs[caller] = source.service.CreateFork(t.Context(), ForkRequest{ |
| 435 | Source: source.ref, TurnID: "turn-1", BoundarySequence: source.first.LastSequence(), |
| 436 | OperationID: fmt.Sprintf("fork-distinct-%d", caller), |
| 437 | }) |
| 438 | }) |
| 439 | } |
| 440 | close(start) |
| 441 | wait.Wait() |
| 442 | children := map[string]bool{} |
| 443 | for caller := range callers { |
| 444 | if errs[caller] != nil { |
| 445 | t.Fatalf("caller %d: %v", caller, errs[caller]) |
| 446 | } |
| 447 | children[results[caller].Child.SessionID] = true |
| 448 | } |
| 449 | if len(children) != callers { |
| 450 | t.Fatalf("distinct operations published %d children: %+v", len(children), results) |
| 451 | } |
| 452 | } |
| 453 | |
| 454 | func TestCreateForkFromReadOnlySourceYieldsWritableChildAndLeavesSourceLog(t *testing.T) { |
| 455 | source := newClosedTurnSource(t) |
| 456 | logPath := filepath.Join(source.root, source.ref.SessionID, currentLogName) |
| 457 | before, err := os.ReadFile(logPath) |
| 458 | if err != nil { |
| 459 | t.Fatal(err) |
| 460 | } |
| 461 | if len(before) == 0 { |
| 462 | t.Fatal("source log is empty") |
| 463 | } |
| 464 | result, err := source.service.CreateFork(t.Context(), ForkRequest{Source: source.ref, TurnID: "turn-2", BoundarySequence: source.second.LastSequence(), ChildID: "child-from-read-only", OperationID: "fork-read-only"}) |
| 465 | if err != nil { |
| 466 | t.Fatal(err) |
| 467 | } |
| 468 | assertSourceLogUnchanged := func(when string) { |
| 469 | t.Helper() |
| 470 | after, readErr := os.ReadFile(logPath) |
| 471 | if readErr != nil { |
| 472 | t.Fatal(readErr) |
| 473 | } |
| 474 | if !bytes.Equal(before, after) { |
| 475 | t.Fatalf("source log bytes changed %s", when) |
| 476 | } |
| 477 | } |
| 478 | assertSourceLogUnchanged("while forking the read-only source") |
| 479 | child, err := Open(filepath.Join(source.root, result.Child.SessionID), result.Child.SessionID) |
| 480 | if err != nil { |
| 481 | t.Fatal(err) |
| 482 | } |
| 483 | defer child.Close(context.Background()) |
| 484 | if _, err := child.Append(t.Context(), Batch{OperationID: "child-turn", TurnID: "child-turn", Events: []Event{ |
| 485 | {Kind: "turn/start"}, {Kind: "turn/end", Payload: json.RawMessage(`{"status":"completed"}`)}, |
| 486 | }}); err != nil { |
| 487 | t.Fatalf("append to a child of a read-only source: %v", err) |
| 488 | } |
| 489 | if got := child.Snapshot().EventSequence; got != result.Turn.EndSequence+2 { |
| 490 | t.Fatalf("child sequence after append = %d, want %d", got, result.Turn.EndSequence+2) |
| 491 | } |
| 492 | assertSourceLogUnchanged("after writing the child") |
| 493 | } |
| 494 | |
| 495 | // TestCreateForkRefusesCutWhoseCommitLeavesAuthorityOpen covers the terminal but |
| 496 | // still unsafe boundary: the commit that closed the turn also opened an |
| 497 | // interaction nothing resolved. The cut covers that whole commit, so the child |
| 498 | // would inherit pending authority. The refusal must carry its own reason rather |
| 499 | // than being reported as an unverifiable boundary, and must publish nothing. |
| 500 | func TestCreateForkRefusesCutWhoseCommitLeavesAuthorityOpen(t *testing.T) { |
| 501 | service, root, runtime := newSourceService(t, "source") |
| 502 | message, err := json.Marshal(map[string]any{"message": provider.Message{ID: "message-1", Role: provider.RoleAssistant, Content: "one"}}) |
| 503 | if err != nil { |
| 504 | t.Fatal(err) |
| 505 | } |
| 506 | commit, err := runtime.Session().Append(t.Context(), Batch{OperationID: "turn-1", TurnID: "turn-1", Events: []Event{ |
| 507 | {Kind: "turn/start"}, |
| 508 | {Kind: "message/complete", Payload: message}, |
| 509 | {Kind: "interaction/created", Payload: json.RawMessage(`{"id":"interaction-1"}`)}, |
| 510 | {Kind: "turn/end", Payload: json.RawMessage(`{"status":"completed"}`)}, |
| 511 | }}) |
| 512 | if err != nil { |
| 513 | t.Fatal(err) |
| 514 | } |
| 515 | set, setErr := service.ForkTargetSetFor(t.Context(), runtime.Ref()) |
| 516 | if setErr != nil || len(set.Targets) != 1 || set.Targets[0].Available || set.Targets[0].Reason != ForkActiveAuthority { |
| 517 | t.Fatalf("unsafe target set = %+v, err=%v", set, setErr) |
| 518 | } |
| 519 | _, err = service.CreateFork(t.Context(), ForkRequest{Source: runtime.Ref(), TurnID: "turn-1", BoundarySequence: commit.LastSequence(), OperationID: "fork-open-authority"}) |
| 520 | var unavailable *ForkUnavailableError |
| 521 | if !errors.As(err, &unavailable) { |
| 522 | t.Fatalf("fork error = %v, want a *ForkUnavailableError", err) |
| 523 | } |
| 524 | if unavailable.Reason != ForkActiveAuthority { |
| 525 | t.Fatalf("reason = %q, want %q", unavailable.Reason, ForkActiveAuthority) |
| 526 | } |
| 527 | if children := forkChildDirs(t, root, "source"); len(children) != 0 { |
| 528 | t.Fatalf("refused fork published %v", children) |
| 529 | } |
| 530 | } |
| 531 | |
| 532 | func TestForkAvailabilityUsesTheCompleteClosingCommit(t *testing.T) { |
| 533 | service, _, runtime := newSourceService(t, "complete-commit") |
| 534 | message, _ := json.Marshal(map[string]any{"message": provider.Message{ID: "message-1", Role: provider.RoleAssistant, Content: "one"}}) |
| 535 | commit, err := runtime.Session().Append(t.Context(), Batch{OperationID: "turn-1", TurnID: "turn-1", Events: []Event{ |
| 536 | {Kind: "turn/start"}, |
| 537 | {Kind: "message/complete", Payload: message}, |
| 538 | {Kind: "interaction/created", Payload: json.RawMessage(`{"id":"interaction-1","state":"pending"}`)}, |
| 539 | {Kind: "tool/start", Payload: json.RawMessage(`{"id":"tool-1","name":"bash"}`)}, |
| 540 | {Kind: "turn/end", Payload: json.RawMessage(`{"status":"completed"}`)}, |
| 541 | // These records share the atomic commit with turn/end. Eligibility must be |
| 542 | // computed after both have cleared their authority. |
| 543 | {Kind: "interaction/resolved", Payload: json.RawMessage(`{"id":"interaction-1","state":"answered"}`)}, |
| 544 | {Kind: "tool/result", Payload: json.RawMessage(`{"id":"tool-1","name":"bash","output":"ok"}`)}, |
| 545 | }}) |
| 546 | if err != nil { |
| 547 | t.Fatal(err) |
| 548 | } |
| 549 | set, err := service.ForkTargetSetFor(t.Context(), runtime.Ref()) |
| 550 | if err != nil || len(set.Targets) != 1 || !set.Targets[0].Available || set.Targets[0].BoundarySequence != commit.LastSequence() { |
| 551 | t.Fatalf("complete closing commit = %+v, err=%v", set, err) |
| 552 | } |
| 553 | } |
| 554 | |
| 555 | func TestForkAvailabilityRejectsActiveToolAtClosingBoundary(t *testing.T) { |
| 556 | service, _, runtime := newSourceService(t, "active-tool") |
| 557 | _, err := runtime.Session().Append(t.Context(), Batch{OperationID: "turn-1", TurnID: "turn-1", Events: []Event{ |
| 558 | {Kind: "turn/start"}, |
| 559 | {Kind: "tool/start", Payload: json.RawMessage(`{"id":"tool-1","name":"bash"}`)}, |
| 560 | {Kind: "turn/end", Payload: json.RawMessage(`{"status":"completed"}`)}, |
| 561 | }}) |
| 562 | if err != nil { |
| 563 | t.Fatal(err) |
| 564 | } |
| 565 | set, err := service.ForkTargetSetFor(t.Context(), runtime.Ref()) |
| 566 | if err != nil || len(set.Targets) != 1 || set.Targets[0].Available || set.Targets[0].Reason != ForkActiveAuthority { |
| 567 | t.Fatalf("active-tool closing commit = %+v, err=%v", set, err) |
| 568 | } |
| 569 | } |
| 570 | |
| 571 | func TestForkTargetSetUnverifiableForMessageOnlyHistory(t *testing.T) { |
| 572 | service, _, runtime := newSourceService(t, "legacy") |
| 573 | payload, err := json.Marshal(map[string]any{ |
| 574 | "source": Source{Path: "/legacy/old.jsonl", Size: 128, SHA256: strings.Repeat("a", 64), Version: Codec}, |
| 575 | "messages": []provider.Message{{ID: "message-1", Role: provider.RoleUser, Content: "hello"}}, |
| 576 | }) |
| 577 | if err != nil { |
| 578 | t.Fatal(err) |
| 579 | } |
| 580 | if _, err := runtime.Session().Append(t.Context(), Batch{OperationID: "legacy-import", Events: []Event{{Kind: "legacy/import", Payload: payload}}}); err != nil { |
| 581 | t.Fatal(err) |
| 582 | } |
| 583 | set, err := service.ForkTargetSetFor(t.Context(), runtime.Ref()) |
| 584 | if err != nil { |
| 585 | t.Fatal(err) |
| 586 | } |
| 587 | if set.Verifiable || len(set.Targets) != 0 { |
| 588 | t.Fatalf("message-only history reported %+v", set) |
| 589 | } |
| 590 | snapshot := runtime.Session().Snapshot() |
| 591 | if len(snapshot.Projection.Messages) != 1 || snapshot.Projection.Messages[0].ID != "message-1" { |
| 592 | t.Fatalf("messages = %+v", snapshot.Projection.Messages) |
| 593 | } |
| 594 | if sequence, availability, err := ForkSequence(snapshot.Projection, "message-1"); err != nil || sequence != 0 || availability != ForkHistoryUnverifiable { |
| 595 | t.Fatalf("message identity resolved as a cut = %d, %s, %v", sequence, availability, err) |
| 596 | } |
| 597 | } |
| 598 |