| 1 | package control |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "path/filepath" |
| 9 | "strings" |
| 10 | "testing" |
| 11 | |
| 12 | "reasonix/internal/agent" |
| 13 | "reasonix/internal/agent/testutil" |
| 14 | "reasonix/internal/event" |
| 15 | "reasonix/internal/provider" |
| 16 | "reasonix/internal/session" |
| 17 | "reasonix/internal/tool" |
| 18 | "reasonix/internal/transcript" |
| 19 | ) |
| 20 | |
| 21 | func TestTranscriptFollowAssignsUniqueIdentityToOutsideTurnNotices(t *testing.T) { |
| 22 | for _, afterTurn := range []bool{false, true} { |
| 23 | name := "before-turn" |
| 24 | if afterTurn { |
| 25 | name = "after-turn" |
| 26 | } |
| 27 | t.Run(name, func(t *testing.T) { |
| 28 | c, _, runtime := newTranscriptBoundaryController(t, testutil.Turn{Text: "answer"}, event.Discard) |
| 29 | if afterTurn { |
| 30 | if err := c.RunTurn(t.Context(), "question"); err != nil { |
| 31 | t.Fatal(err) |
| 32 | } |
| 33 | } |
| 34 | bodies := []string{strings.Repeat("a", 70_000), strings.Repeat("b", 70_000)} |
| 35 | for _, body := range bodies { |
| 36 | c.notice(body) |
| 37 | } |
| 38 | response, err := c.TranscriptFollow(t.Context(), transcript.FollowRequest{}) |
| 39 | if err != nil { |
| 40 | t.Fatal(err) |
| 41 | } |
| 42 | t.Cleanup(func() { |
| 43 | _, _ = c.TranscriptFollow(context.Background(), transcript.FollowRequest{Subscription: response.Subscription, Close: true}) |
| 44 | }) |
| 45 | var notices []transcript.Record |
| 46 | for _, record := range response.Snapshot.Records { |
| 47 | if record.Message.Role == "notice" && len(record.Message.Content) > 0 { |
| 48 | notices = append(notices, record) |
| 49 | } |
| 50 | } |
| 51 | if len(notices) != 2 || notices[0].ID == notices[1].ID { |
| 52 | t.Fatalf("outside-turn notice identities = %+v", notices) |
| 53 | } |
| 54 | for index, record := range notices { |
| 55 | if record.ID == "" || record.Message.RecordID != record.ID || len(record.Refs) != 1 || record.Refs[0].RecordID != record.ID { |
| 56 | t.Fatalf("notice %d identity/ref mismatch: %+v", index, record) |
| 57 | } |
| 58 | chunk, err := runtime.Transcript().Content(transcript.ContentRequest{ContentRef: record.Refs[0]}) |
| 59 | if err != nil || !strings.HasPrefix(bodies[index], chunk.Data) || chunk.Data == "" { |
| 60 | t.Fatalf("notice %d content alias: chunk=%q err=%v", index, chunk.Data, err) |
| 61 | } |
| 62 | } |
| 63 | }) |
| 64 | } |
| 65 | } |
| 66 | |
| 67 | func TestTranscriptFollowMakesAcceptedBatchBeyondDisplayBudgetPageable(t *testing.T) { |
| 68 | c, _, runtime := newTranscriptBoundaryController(t, testutil.Turn{}, event.Discard) |
| 69 | var events []session.Event |
| 70 | for i := range 160 { |
| 71 | body, _ := json.Marshal(map[string]any{"message": provider.Message{ID: fmt.Sprintf("accepted-%03d", i), Role: provider.RoleAssistant, Content: "accepted body"}}) |
| 72 | events = append(events, session.Event{Kind: "message/complete", Payload: body}) |
| 73 | } |
| 74 | commit, err := runtime.Session().Append(t.Context(), session.Batch{OperationID: "large-batch", Events: events}) |
| 75 | if err != nil { |
| 76 | t.Fatal(err) |
| 77 | } |
| 78 | // Do not flush: Follow must bridge the accepted/durable boundary itself. |
| 79 | response, err := c.TranscriptFollow(t.Context(), transcript.FollowRequest{}) |
| 80 | if err != nil { |
| 81 | t.Fatal(err) |
| 82 | } |
| 83 | defer c.TranscriptFollow(context.Background(), transcript.FollowRequest{Subscription: response.Subscription, Close: true}) |
| 84 | if response.Snapshot == nil || response.History == nil || response.History.SnapshotSequence != commit.LastSequence() || response.Snapshot.CoveredThroughSeq != commit.LastSequence() || runtime.Transcript().Boundary().DurableSeq < commit.LastSequence() || !response.History.HasOlder || len(response.History.Messages) != 32 { |
| 85 | t.Fatalf("accepted history outside the display tail is unreachable: %+v", response) |
| 86 | } |
| 87 | } |
| 88 | |
| 89 | func TestTranscriptCancellationKeepsPartialAnswerAndReasoningRecoverable(t *testing.T) { |
| 90 | ctx, cancel := context.WithCancel(t.Context()) |
| 91 | defer cancel() |
| 92 | var streamedID string |
| 93 | sink := event.FuncSink(func(e event.Event) { |
| 94 | if e.Kind == event.Text { |
| 95 | streamedID = e.MessageID |
| 96 | cancel() |
| 97 | } |
| 98 | }) |
| 99 | c, service, runtime := newTranscriptBoundaryController(t, testutil.Turn{Reasoning: "partial reasoning", Text: "partial answer"}, sink) |
| 100 | if err := c.RunTurn(ctx, "question"); !errors.Is(err, context.Canceled) { |
| 101 | t.Fatalf("cancelled turn error=%v", err) |
| 102 | } |
| 103 | if streamedID == "" { |
| 104 | t.Fatal("fixture did not stream before cancellation") |
| 105 | } |
| 106 | cut, err := c.TranscriptSnapshot(transcript.PageRequest{}) |
| 107 | if err != nil { |
| 108 | t.Fatal(err) |
| 109 | } |
| 110 | if cut.Runtime.Status != event.TurnInterrupted || len(cut.ActiveAttempts) != 0 { |
| 111 | t.Fatalf("cancelled view remains active: status=%q attempts=%d", cut.Runtime.Status, len(cut.ActiveAttempts)) |
| 112 | } |
| 113 | visible := 0 |
| 114 | for _, record := range cut.Records { |
| 115 | if record.Message.Content == "partial answer" && record.Message.Reasoning == "partial reasoning" { |
| 116 | visible++ |
| 117 | } |
| 118 | } |
| 119 | if visible != 1 { |
| 120 | t.Errorf("cancellation lost or duplicated visible prefix: matchingRows=%d records=%+v", visible, cut.Records) |
| 121 | } |
| 122 | history, err := service.Query().History(t.Context(), runtime.Ref()) |
| 123 | if err != nil { |
| 124 | t.Fatal(err) |
| 125 | } |
| 126 | recoverable := false |
| 127 | for _, message := range history { |
| 128 | recoverable = recoverable || message.LocalOnly && message.Content == "partial answer" && message.ReasoningContent == "partial reasoning" |
| 129 | } |
| 130 | if !recoverable { |
| 131 | t.Fatal("cancelled partial output is absent from persistent recovery history") |
| 132 | } |
| 133 | state := runtime.StateSnapshot().Session |
| 134 | if state.DurableSequence != state.EventSequence { |
| 135 | t.Fatalf("interrupted terminal published before persistence: durable=%d accepted=%d", state.DurableSequence, state.EventSequence) |
| 136 | } |
| 137 | } |
| 138 | |
| 139 | func TestTranscriptFollowInitialHistoryIsReadyAndPinnedToDurableCut(t *testing.T) { |
| 140 | c, _, runtime := newTranscriptBoundaryController(t, testutil.Turn{Reasoning: "saved reasoning", Text: "saved final answer"}, event.Discard) |
| 141 | if err := c.RunTurn(t.Context(), "question"); err != nil { |
| 142 | t.Fatal(err) |
| 143 | } |
| 144 | response, err := c.TranscriptFollow(t.Context(), transcript.FollowRequest{}) |
| 145 | if err != nil { |
| 146 | t.Fatal(err) |
| 147 | } |
| 148 | t.Cleanup(func() { |
| 149 | _, _ = c.TranscriptFollow(context.Background(), transcript.FollowRequest{Subscription: response.Subscription, Close: true}) |
| 150 | }) |
| 151 | if response.Snapshot == nil || response.History == nil || response.History.Status != "ready" { |
| 152 | t.Fatalf("initial follow exposed an unready history page: %+v", response) |
| 153 | } |
| 154 | cut := response.Snapshot |
| 155 | if cut.DurableSeq == 0 || cut.DurableSeq != runtime.StateSnapshot().Session.DurableSequence || response.History.SnapshotSequence != cut.DurableSeq { |
| 156 | t.Fatalf("history and view were sampled at different durable cuts: view=%d history=%d durable=%d", cut.DurableSeq, response.History.SnapshotSequence, runtime.StateSnapshot().Session.DurableSequence) |
| 157 | } |
| 158 | var finalID string |
| 159 | for _, row := range cut.Records { |
| 160 | if row.Message.Content == "saved final answer" { |
| 161 | finalID = row.Message.MessageID |
| 162 | } |
| 163 | } |
| 164 | if finalID == "" { |
| 165 | t.Fatal("initial follow omitted the saved final answer") |
| 166 | } |
| 167 | found := false |
| 168 | for _, message := range response.History.Messages { |
| 169 | if message.EventSequence > cut.DurableSeq { |
| 170 | t.Fatal("history page includes a message beyond its declared cut") |
| 171 | } |
| 172 | found = found || message.MessageID == finalID |
| 173 | } |
| 174 | if !found { |
| 175 | t.Fatal("initial historical page omitted the completed view's final message") |
| 176 | } |
| 177 | } |
| 178 | |
| 179 | // Exercise the same exclusive runtime used by Desktop and Serve. The legacy |
| 180 | // controller-only ledger has a different sequence space and must not hide |
| 181 | // regressions in this path. Only provider output is scripted; persistence, |
| 182 | // the agent, runtime and display projection are real. |
| 183 | func newTranscriptBoundaryController(t *testing.T, turn testutil.Turn, sink event.Sink) (*Controller, *session.Service, *session.Runtime) { |
| 184 | t.Helper() |
| 185 | service, err := session.NewService("desktop", session.NewFilesystemPersistence(filepath.Join(t.TempDir(), "sessions"))) |
| 186 | if err != nil { |
| 187 | t.Fatal(err) |
| 188 | } |
| 189 | runtime, err := service.Create(t.Context(), session.CreateOptions{SessionID: "transcript-boundary"}) |
| 190 | if err != nil { |
| 191 | t.Fatal(err) |
| 192 | } |
| 193 | executor := agent.New(testutil.NewMock("test", turn), tool.NewRegistry(), agent.NewSession("system"), agent.Options{}, event.Discard) |
| 194 | c := newOwnedTestController(t, Options{ |
| 195 | Runner: executor, Executor: executor, Sink: sink, |
| 196 | SessionService: service, SessionRuntime: runtime, ExclusiveSession: true, |
| 197 | }) |
| 198 | return c, service, runtime |
| 199 | } |
| 200 | |
| 201 | func transcriptBoundaryMessage(snapshot transcript.Snapshot, id string) (transcript.Message, bool) { |
| 202 | for _, records := range [][]transcript.Record{snapshot.Records, snapshot.ActiveRecords} { |
| 203 | for _, record := range records { |
| 204 | if record.Message.MessageID == id { |
| 205 | return record.Message, true |
| 206 | } |
| 207 | } |
| 208 | } |
| 209 | return transcript.Message{}, false |
| 210 | } |
| 211 | |
| 212 | func TestTranscriptRuntimeSnapshotRetainsStreamingPrefixAndState(t *testing.T) { |
| 213 | var c *Controller |
| 214 | var runtime *session.Runtime |
| 215 | observed := false |
| 216 | sink := event.FuncSink(func(e event.Event) { |
| 217 | if e.Kind != event.Text { |
| 218 | return |
| 219 | } |
| 220 | observed = true |
| 221 | snapshot, err := c.TranscriptSnapshot(transcript.PageRequest{}) |
| 222 | if err != nil { |
| 223 | t.Errorf("snapshot during output: %v", err) |
| 224 | return |
| 225 | } |
| 226 | if phase := runtime.StateSnapshot().Phase; phase != session.RuntimeRunning { |
| 227 | t.Errorf("fixture is not streaming: phase=%q", phase) |
| 228 | } |
| 229 | if snapshot.Runtime.Status != event.TurnInProgress { |
| 230 | t.Errorf("streaming snapshot loses running state: status=%q", snapshot.Runtime.Status) |
| 231 | } |
| 232 | message, found := transcriptBoundaryMessage(snapshot, e.MessageID) |
| 233 | if !found || message.Content != "visible answer" || message.Reasoning != "visible thinking" { |
| 234 | t.Errorf("streaming snapshot loses assistant prefix: found=%v content=%q reasoning=%q", found, message.Content, message.Reasoning) |
| 235 | } |
| 236 | matched := false |
| 237 | for _, attempt := range snapshot.ActiveAttempts { |
| 238 | if attempt.MessageID == e.MessageID && attempt.ID != "" { |
| 239 | matched = true |
| 240 | } |
| 241 | } |
| 242 | if !matched { |
| 243 | t.Error("streaming snapshot omits the active assistant attempt") |
| 244 | } |
| 245 | }) |
| 246 | c, _, runtime = newTranscriptBoundaryController(t, testutil.Turn{Reasoning: "visible thinking", Text: "visible answer"}, sink) |
| 247 | if err := c.RunTurn(t.Context(), "question"); err != nil { |
| 248 | t.Fatal(err) |
| 249 | } |
| 250 | if !observed { |
| 251 | t.Fatal("provider output was never published") |
| 252 | } |
| 253 | } |
| 254 | |
| 255 | func TestTranscriptRuntimeTransientRevisionDoesNotAdvanceBusinessCoverage(t *testing.T) { |
| 256 | var c *Controller |
| 257 | var runtime *session.Runtime |
| 258 | var cuts []transcript.Snapshot |
| 259 | sink := event.FuncSink(func(e event.Event) { |
| 260 | if e.Kind != event.Text { |
| 261 | return |
| 262 | } |
| 263 | snapshot, err := c.TranscriptSnapshot(transcript.PageRequest{}) |
| 264 | if err != nil { |
| 265 | t.Errorf("snapshot during output: %v", err) |
| 266 | return |
| 267 | } |
| 268 | if want := runtime.StateSnapshot().Session.EventSequence; snapshot.CoveredThroughSeq != want { |
| 269 | t.Errorf("snapshot mixes business and transient sequences: coverage=%d business=%d", snapshot.CoveredThroughSeq, want) |
| 270 | } |
| 271 | cuts = append(cuts, snapshot) |
| 272 | }) |
| 273 | c, _, runtime = newTranscriptBoundaryController(t, testutil.Turn{Chunks: []provider.Chunk{ |
| 274 | {Type: provider.ChunkText, Text: "first "}, |
| 275 | {Type: provider.ChunkText, Text: "second"}, |
| 276 | {Type: provider.ChunkDone}, |
| 277 | }}, sink) |
| 278 | if err := c.RunTurn(t.Context(), "question"); err != nil { |
| 279 | t.Fatal(err) |
| 280 | } |
| 281 | if len(cuts) != 2 { |
| 282 | t.Fatalf("expected two streaming cuts, got %d", len(cuts)) |
| 283 | } |
| 284 | if cuts[0].CoveredThroughSeq != cuts[1].CoveredThroughSeq { |
| 285 | t.Errorf("text-only chunks advance business coverage: before=%d after=%d", cuts[0].CoveredThroughSeq, cuts[1].CoveredThroughSeq) |
| 286 | } |
| 287 | if cuts[1].ProjectionRevision <= cuts[0].ProjectionRevision { |
| 288 | t.Errorf("new streaming prefix reused the previous view revision: before=%d after=%d", cuts[0].ProjectionRevision, cuts[1].ProjectionRevision) |
| 289 | } |
| 290 | } |
| 291 | |
| 292 | func TestTranscriptRuntimePublishesCommittedAnswerBeforeCompletion(t *testing.T) { |
| 293 | var c *Controller |
| 294 | var service *session.Service |
| 295 | var runtime *session.Runtime |
| 296 | var finalID string |
| 297 | completed := false |
| 298 | sink := event.FuncSink(func(e event.Event) { |
| 299 | switch e.Kind { |
| 300 | case event.Message: |
| 301 | if e.Text != "final answer" { |
| 302 | return |
| 303 | } |
| 304 | finalID = e.MessageID |
| 305 | messages, err := service.Query().History(t.Context(), runtime.Ref()) |
| 306 | if err != nil { |
| 307 | t.Errorf("read committed answer: %v", err) |
| 308 | return |
| 309 | } |
| 310 | found := false |
| 311 | for _, message := range messages { |
| 312 | if message.ID == finalID && message.Content == "final answer" { |
| 313 | found = true |
| 314 | } |
| 315 | } |
| 316 | if !found { |
| 317 | t.Error("final message was published before its business commit") |
| 318 | } |
| 319 | case event.TurnDone: |
| 320 | completed = true |
| 321 | state := runtime.StateSnapshot().Session |
| 322 | if state.DurableSequence != state.EventSequence { |
| 323 | t.Errorf("completion precedes persistence barrier: durable=%d committed=%d", state.DurableSequence, state.EventSequence) |
| 324 | } |
| 325 | snapshot, err := c.TranscriptSnapshot(transcript.PageRequest{}) |
| 326 | if err != nil { |
| 327 | t.Errorf("completion snapshot: %v", err) |
| 328 | return |
| 329 | } |
| 330 | message, found := transcriptBoundaryMessage(snapshot, finalID) |
| 331 | if !found || message.Content != "final answer" || message.Reasoning != "final reasoning" { |
| 332 | t.Errorf("completion snapshot loses committed answer: found=%v content=%q reasoning=%q", found, message.Content, message.Reasoning) |
| 333 | } |
| 334 | if snapshot.Runtime.Status != event.TurnCompleted || len(snapshot.ActiveAttempts) != 0 { |
| 335 | t.Errorf("completion snapshot is not settled: status=%q activeAttempts=%d", snapshot.Runtime.Status, len(snapshot.ActiveAttempts)) |
| 336 | } |
| 337 | } |
| 338 | }) |
| 339 | c, service, runtime = newTranscriptBoundaryController(t, testutil.Turn{Reasoning: "final reasoning", Text: "final answer"}, sink) |
| 340 | if err := c.RunTurn(t.Context(), "question"); err != nil { |
| 341 | t.Fatal(err) |
| 342 | } |
| 343 | if finalID == "" || !completed { |
| 344 | t.Fatalf("missing final publications: message=%q completed=%v", finalID, completed) |
| 345 | } |
| 346 | // A repeated read must not hit a cache that advertises the final business |
| 347 | // sequence but still contains only the history preceding the final answer. |
| 348 | for range 2 { |
| 349 | snapshot, err := c.TranscriptSnapshot(transcript.PageRequest{}) |
| 350 | if err != nil { |
| 351 | t.Fatal(err) |
| 352 | } |
| 353 | message, found := transcriptBoundaryMessage(snapshot, finalID) |
| 354 | if !found || message.Content != "final answer" { |
| 355 | t.Fatal("settled history reread lost the final answer") |
| 356 | } |
| 357 | } |
| 358 | } |
| 359 |