| 1 | package serve |
| 2 | |
| 3 | import ( |
| 4 | "bufio" |
| 5 | "context" |
| 6 | "encoding/json" |
| 7 | "errors" |
| 8 | "io" |
| 9 | "net/http" |
| 10 | "net/http/httptest" |
| 11 | "os" |
| 12 | "path/filepath" |
| 13 | "strings" |
| 14 | "sync/atomic" |
| 15 | "testing" |
| 16 | "time" |
| 17 | |
| 18 | "reasonix/internal/agent" |
| 19 | "reasonix/internal/billing" |
| 20 | "reasonix/internal/boot" |
| 21 | "reasonix/internal/config" |
| 22 | "reasonix/internal/control" |
| 23 | "reasonix/internal/event" |
| 24 | "reasonix/internal/eventwire" |
| 25 | ) |
| 26 | |
| 27 | type replayStubController struct{ *control.Controller } |
| 28 | |
| 29 | func (replayStubController) SessionPath() string { return "/sessions/a.jsonl" } |
| 30 | func (replayStubController) ReplayPendingPromptsWith(factory func() event.Sink) { |
| 31 | factory().Emit(event.Event{Kind: event.ApprovalRequest}) |
| 32 | } |
| 33 | |
| 34 | func sessionPathFromFrame(t *testing.T, frame []byte) string { |
| 35 | t.Helper() |
| 36 | var wired eventwire.Event |
| 37 | if err := json.Unmarshal(frame, &wired); err != nil { |
| 38 | t.Fatalf("decode event frame: %v\nframe=%s", err, frame) |
| 39 | } |
| 40 | return wired.SessionPath |
| 41 | } |
| 42 | |
| 43 | type replayRaceController struct { |
| 44 | control.SessionAPI |
| 45 | path string |
| 46 | onPath func() |
| 47 | replayed chan string |
| 48 | } |
| 49 | |
| 50 | func (c *replayRaceController) SessionPath() string { |
| 51 | if c.onPath != nil { |
| 52 | c.onPath() |
| 53 | } |
| 54 | return c.path |
| 55 | } |
| 56 | |
| 57 | func (c *replayRaceController) ReplayPendingPromptsWith(factory func() event.Sink) { |
| 58 | c.replayed <- c.path |
| 59 | factory().Emit(event.Event{Kind: event.ApprovalRequest}) |
| 60 | } |
| 61 | |
| 62 | type closeProbeController struct { |
| 63 | *control.Controller |
| 64 | closed atomic.Bool |
| 65 | } |
| 66 | |
| 67 | func (c *closeProbeController) Close() { |
| 68 | c.closed.Store(true) |
| 69 | c.Controller.Close() |
| 70 | } |
| 71 | |
| 72 | type retiringCloseProbeController struct { |
| 73 | *control.Controller |
| 74 | closeStarted chan struct{} |
| 75 | closeRelease chan struct{} |
| 76 | } |
| 77 | |
| 78 | func (c *retiringCloseProbeController) Close() { |
| 79 | close(c.closeStarted) |
| 80 | <-c.closeRelease |
| 81 | c.Controller.Close() |
| 82 | } |
| 83 | |
| 84 | func TestDetachedSessionRemainsBusyUntilCloseFinishes(t *testing.T) { |
| 85 | path := filepath.Join(t.TempDir(), "retiring.jsonl") |
| 86 | ctrl := &retiringCloseProbeController{ |
| 87 | Controller: control.New(control.Options{SessionPath: path}), |
| 88 | closeStarted: make(chan struct{}), |
| 89 | closeRelease: make(chan struct{}), |
| 90 | } |
| 91 | t.Cleanup(func() { |
| 92 | select { |
| 93 | case <-ctrl.closeRelease: |
| 94 | default: |
| 95 | close(ctrl.closeRelease) |
| 96 | } |
| 97 | }) |
| 98 | server := New(control.New(control.Options{}), NewBroadcaster(), config.ServeConfig{}) |
| 99 | tag := NewSessionTagSink(server.bc) |
| 100 | tag.SetPath(path) |
| 101 | detached, err := server.registerDetached(ctrl, nil, tag) |
| 102 | if err != nil { |
| 103 | t.Fatal(err) |
| 104 | } |
| 105 | select { |
| 106 | case <-ctrl.closeStarted: |
| 107 | case <-time.After(2 * time.Second): |
| 108 | t.Fatal("detached controller did not start retiring") |
| 109 | } |
| 110 | if !server.detachedBusy(path) { |
| 111 | t.Fatal("retiring session disappeared before controller Close finished") |
| 112 | } |
| 113 | if got := server.takeDetached(path); got != nil { |
| 114 | t.Fatal("retiring controller remained reattachable") |
| 115 | } |
| 116 | close(ctrl.closeRelease) |
| 117 | select { |
| 118 | case <-detached.done: |
| 119 | case <-time.After(2 * time.Second): |
| 120 | t.Fatal("detached retirement did not finish") |
| 121 | } |
| 122 | if server.detachedBusy(path) { |
| 123 | t.Fatal("retired session remained busy after controller Close finished") |
| 124 | } |
| 125 | } |
| 126 | |
| 127 | func TestServerCloseClosesPublishedForegroundReplacement(t *testing.T) { |
| 128 | bc := NewBroadcaster() |
| 129 | first := &closeProbeController{Controller: control.New(control.Options{Sink: bc})} |
| 130 | replacement := &closeProbeController{Controller: control.New(control.Options{Sink: bc})} |
| 131 | server := New(first, bc, config.ServeConfig{}) |
| 132 | if !server.publishControllerSwap(first, replacement, replacement.SessionPath()) { |
| 133 | t.Fatal("replacement publication failed") |
| 134 | } |
| 135 | server.Close() |
| 136 | if !replacement.closed.Load() { |
| 137 | t.Fatal("server shutdown left the published foreground controller open") |
| 138 | } |
| 139 | } |
| 140 | |
| 141 | func TestStaleRecoveryCannotOverwritePublishedForegroundRoute(t *testing.T) { |
| 142 | dir := t.TempDir() |
| 143 | oldPath := filepath.Join(dir, "old.jsonl") |
| 144 | newPath := filepath.Join(dir, "new.jsonl") |
| 145 | recoveryPath := filepath.Join(dir, "old-recovery.jsonl") |
| 146 | bc := NewBroadcaster() |
| 147 | old := control.New(control.Options{SessionPath: oldPath}) |
| 148 | next := control.New(control.Options{SessionPath: newPath}) |
| 149 | server := New(old, bc, config.ServeConfig{}) |
| 150 | t.Cleanup(old.Close) |
| 151 | t.Cleanup(server.Close) |
| 152 | tag := NewSessionTagSink(bc) |
| 153 | server.RegisterSessionTag(old, tag) |
| 154 | recoverOld := server.sessionRecoveryHandler(old, nil) |
| 155 | if !server.publishControllerSwap(old, next, newPath) { |
| 156 | t.Fatal("foreground replacement publication failed") |
| 157 | } |
| 158 | if err := recoverOld(control.SessionRecoveryInfo{RecoveryPath: recoveryPath}); err != nil { |
| 159 | t.Fatal(err) |
| 160 | } |
| 161 | if got, want := bc.CurrentSession(), agent.CanonicalSessionPath(newPath); got != want { |
| 162 | t.Fatalf("stale recovery changed foreground route to %q, want %q", got, want) |
| 163 | } |
| 164 | } |
| 165 | |
| 166 | func TestCapturedRecoveryCallbackFollowsDetachedKeeper(t *testing.T) { |
| 167 | dir := t.TempDir() |
| 168 | oldPath := filepath.Join(dir, "old.jsonl") |
| 169 | targetPath := filepath.Join(dir, "target.jsonl") |
| 170 | recoveryPath := filepath.Join(dir, "old-recovery.jsonl") |
| 171 | old := control.New(control.Options{SessionPath: oldPath}) |
| 172 | server := newLifecycleTestServer(t, old, NewBroadcaster(), config.ServeConfig{}) |
| 173 | leases := control.NewSessionLeaseKeeper() |
| 174 | defer leases.Release() |
| 175 | if err := leases.Rebind(oldPath); err != nil { |
| 176 | t.Fatal(err) |
| 177 | } |
| 178 | if err := server.SetSessionLeases(leases); err != nil { |
| 179 | t.Fatal(err) |
| 180 | } |
| 181 | captured := server.sessionRecoveryHandler(old, leases) |
| 182 | detached, err := leases.RebindDetaching(targetPath) |
| 183 | if err != nil { |
| 184 | t.Fatal(err) |
| 185 | } |
| 186 | defer detached.Release() |
| 187 | replacement := control.New(control.Options{SessionPath: targetPath}) |
| 188 | defer replacement.Close() |
| 189 | if err := leases.BindControllerAuthority(replacement); err != nil { |
| 190 | t.Fatal(err) |
| 191 | } |
| 192 | if err := captured(control.SessionRecoveryInfo{RecoveryPath: recoveryPath}); err != nil { |
| 193 | t.Fatal(err) |
| 194 | } |
| 195 | if got, want := detached.HeldPath(), agent.CanonicalSessionPath(recoveryPath); got != want { |
| 196 | t.Fatalf("captured recovery moved detached keeper to %q, want %q", got, want) |
| 197 | } |
| 198 | if got, want := leases.HeldPath(), agent.CanonicalSessionPath(targetPath); got != want { |
| 199 | t.Fatalf("captured recovery corrupted foreground keeper: got %q, want %q", got, want) |
| 200 | } |
| 201 | } |
| 202 | |
| 203 | func TestBusyNewRejectsUntaggedLegacyController(t *testing.T) { |
| 204 | dir := t.TempDir() |
| 205 | path := filepath.Join(dir, "legacy.jsonl") |
| 206 | bc := NewBroadcaster() |
| 207 | ctrl := control.New(control.Options{Runner: blockingRunner{}, Sink: bc, SessionDir: dir, SessionPath: path}) |
| 208 | server := New(ctrl, bc, config.ServeConfig{}) |
| 209 | httpServer := httptest.NewServer(server.Handler()) |
| 210 | defer httpServer.Close() |
| 211 | defer ctrl.Close() |
| 212 | ctrl.Submit("keep running") |
| 213 | waitRunning(t, ctrl) |
| 214 | resp, err := http.Post(httpServer.URL+"/new", "application/json", nil) |
| 215 | if err != nil { |
| 216 | t.Fatal(err) |
| 217 | } |
| 218 | defer resp.Body.Close() |
| 219 | if resp.StatusCode != http.StatusConflict { |
| 220 | body, _ := io.ReadAll(resp.Body) |
| 221 | t.Fatalf("untagged busy /new = %d, want 409: %s", resp.StatusCode, body) |
| 222 | } |
| 223 | if server.ctl() != control.SessionAPI(ctrl) || !ctrl.Running() { |
| 224 | t.Fatal("legacy controller was detached without a session tag") |
| 225 | } |
| 226 | ctrl.Cancel() |
| 227 | waitNotRunning(t, ctrl) |
| 228 | } |
| 229 | |
| 230 | func TestReplayPendingPromptsBroadcastTagsFrames(t *testing.T) { |
| 231 | bc := NewBroadcaster() |
| 232 | ctrl := control.New(control.Options{Sink: bc}) |
| 233 | server := New(replayStubController{ctrl}, bc, config.ServeConfig{}) |
| 234 | all, stop := bc.SubscribeAll() |
| 235 | defer stop() |
| 236 | server.replayPendingPromptsBroadcast() |
| 237 | select { |
| 238 | case frame := <-all: |
| 239 | if got, want := sessionPathFromFrame(t, frame), agent.CanonicalSessionPath("/sessions/a.jsonl"); got != want { |
| 240 | t.Fatalf("replayed frame session path = %q, want %q: %s", got, want, frame) |
| 241 | } |
| 242 | default: |
| 243 | t.Fatal("pending prompt was not replayed") |
| 244 | } |
| 245 | } |
| 246 | |
| 247 | func TestEventsReplayUsesControllerCapturedWithPath(t *testing.T) { |
| 248 | bc := NewBroadcaster() |
| 249 | baseA := control.New(control.Options{Sink: bc}) |
| 250 | baseB := control.New(control.Options{Sink: bc}) |
| 251 | replayed := make(chan string, 2) |
| 252 | a := &replayRaceController{SessionAPI: baseA, path: "/sessions/a.jsonl", replayed: replayed} |
| 253 | b := &replayRaceController{SessionAPI: baseB, path: "/sessions/b.jsonl", replayed: replayed} |
| 254 | server := New(a, bc, config.ServeConfig{}) |
| 255 | promotionStarted := make(chan struct{}) |
| 256 | promotionDone := make(chan struct{}) |
| 257 | a.onPath = func() { |
| 258 | a.onPath = nil |
| 259 | go func() { |
| 260 | close(promotionStarted) |
| 261 | server.bindMu.Lock() |
| 262 | if !server.publishControllerSwap(a, b, b.path) { |
| 263 | t.Error("controller promotion failed") |
| 264 | } |
| 265 | bc.Emit(event.Event{Kind: event.AskRequest, SessionPath: b.path}) |
| 266 | server.bindMu.Unlock() |
| 267 | close(promotionDone) |
| 268 | }() |
| 269 | <-promotionStarted |
| 270 | } |
| 271 | |
| 272 | httpServer := httptest.NewServer(server.Handler()) |
| 273 | defer httpServer.Close() |
| 274 | req, err := http.NewRequestWithContext(t.Context(), http.MethodGet, httpServer.URL+"/events", nil) |
| 275 | if err != nil { |
| 276 | t.Fatal(err) |
| 277 | } |
| 278 | resp, err := http.DefaultClient.Do(req) |
| 279 | if err != nil { |
| 280 | t.Fatal(err) |
| 281 | } |
| 282 | defer resp.Body.Close() |
| 283 | scanner := bufio.NewScanner(resp.Body) |
| 284 | nextData := func() string { |
| 285 | t.Helper() |
| 286 | for scanner.Scan() { |
| 287 | if line := scanner.Text(); strings.HasPrefix(line, "data: ") { |
| 288 | return strings.TrimPrefix(line, "data: ") |
| 289 | } |
| 290 | } |
| 291 | t.Fatalf("event stream ended before the next frame: %v", scanner.Err()) |
| 292 | return "" |
| 293 | } |
| 294 | if first := nextData(); sessionPathFromFrame(t, []byte(first)) != agent.CanonicalSessionPath(a.path) { |
| 295 | t.Fatalf("first frame = %s, want controller A replay", first) |
| 296 | } |
| 297 | select { |
| 298 | case <-promotionDone: |
| 299 | case <-time.After(2 * time.Second): |
| 300 | t.Fatal("controller promotion remained blocked after subscription") |
| 301 | } |
| 302 | if second := nextData(); sessionPathFromFrame(t, []byte(second)) != agent.CanonicalSessionPath(b.path) { |
| 303 | t.Fatalf("second frame = %s, want promoted controller B prompt", second) |
| 304 | } |
| 305 | } |
| 306 | |
| 307 | func TestSlashNewRefreshesControllerTagAndForegroundRoute(t *testing.T) { |
| 308 | dir := t.TempDir() |
| 309 | path := filepath.Join(dir, "current.jsonl") |
| 310 | saveServeTestSession(t, path) |
| 311 | bc := NewBroadcaster() |
| 312 | tag := NewSessionTagSink(bc) |
| 313 | tag.SetPath(path) |
| 314 | loaded, err := agent.LoadSession(path) |
| 315 | if err != nil { |
| 316 | t.Fatal(err) |
| 317 | } |
| 318 | exec := agent.New(nil, nil, loaded, agent.Options{}, tag) |
| 319 | ctrl := control.New(control.Options{Executor: exec, Sink: tag, SessionDir: dir, SessionPath: path, Label: "test"}) |
| 320 | server := newLifecycleTestServer(t, ctrl, bc, config.ServeConfig{}) |
| 321 | server.RegisterSessionTag(ctrl, tag) |
| 322 | leases := control.NewSessionLeaseKeeper() |
| 323 | defer leases.Release() |
| 324 | if err := leases.Rebind(path); err != nil { |
| 325 | t.Fatal(err) |
| 326 | } |
| 327 | if err := server.SetSessionLeases(leases); err != nil { |
| 328 | t.Fatal(err) |
| 329 | } |
| 330 | all, stop := bc.SubscribeAll() |
| 331 | defer stop() |
| 332 | httpServer := httptest.NewServer(server.Handler()) |
| 333 | defer httpServer.Close() |
| 334 | resp, err := http.Post(httpServer.URL+"/submit", "application/json", strings.NewReader(`{"input":"/new"}`)) |
| 335 | if err != nil { |
| 336 | t.Fatal(err) |
| 337 | } |
| 338 | resp.Body.Close() |
| 339 | if resp.StatusCode != http.StatusNoContent { |
| 340 | t.Fatalf("slash /new submit status = %d, want 204", resp.StatusCode) |
| 341 | } |
| 342 | select { |
| 343 | case frame := <-all: |
| 344 | deadline := time.After(2 * time.Second) |
| 345 | for !strings.Contains(string(frame), `"kind":"notice"`) || !strings.Contains(string(frame), `"text":"new session"`) { |
| 346 | select { |
| 347 | case frame = <-all: |
| 348 | case <-deadline: |
| 349 | t.Fatal("slash /new did not publish its completion notice") |
| 350 | } |
| 351 | } |
| 352 | newPath := agent.CanonicalSessionPath(ctrl.SessionPath()) |
| 353 | if newPath == agent.CanonicalSessionPath(path) || tag.Path() != newPath || bc.CurrentSession() != newPath { |
| 354 | t.Fatalf("slash /new routing = controller %q tag %q broadcaster %q frame=%s", newPath, tag.Path(), bc.CurrentSession(), frame) |
| 355 | } |
| 356 | if got := sessionPathFromFrame(t, frame); got != newPath { |
| 357 | t.Fatalf("slash /new notice session path = %q, want %q: %s", got, newPath, frame) |
| 358 | } |
| 359 | case <-time.After(2 * time.Second): |
| 360 | t.Fatal("slash /new emitted no event") |
| 361 | } |
| 362 | } |
| 363 | |
| 364 | type backgroundJobOnlyController struct{ *control.Controller } |
| 365 | |
| 366 | func (c *backgroundJobOnlyController) RuntimeStatus() control.RuntimeStatus { |
| 367 | return control.RuntimeStatus{BackgroundJobs: 1, Cancellable: true} |
| 368 | } |
| 369 | |
| 370 | type detachedProviderHealProbe struct { |
| 371 | *control.Controller |
| 372 | closed atomic.Bool |
| 373 | } |
| 374 | |
| 375 | func (c *detachedProviderHealProbe) RuntimeStatus() control.RuntimeStatus { |
| 376 | return control.RuntimeStatus{BackgroundJobs: 1, Cancellable: true} |
| 377 | } |
| 378 | |
| 379 | func (c *detachedProviderHealProbe) Close() { |
| 380 | c.closed.Store(true) |
| 381 | c.Controller.Close() |
| 382 | } |
| 383 | |
| 384 | func TestProviderHealSynchronouslyRetiresDetachedControllers(t *testing.T) { |
| 385 | path := filepath.Join(t.TempDir(), "background.jsonl") |
| 386 | ctrl := &detachedProviderHealProbe{Controller: control.New(control.Options{SessionPath: path})} |
| 387 | server := New(control.New(control.Options{}), NewBroadcaster(), config.ServeConfig{}) |
| 388 | tag := NewSessionTagSink(server.bc) |
| 389 | tag.SetPath(path) |
| 390 | if _, err := server.registerDetached(ctrl, nil, tag); err != nil { |
| 391 | t.Fatal(err) |
| 392 | } |
| 393 | server.retireDetachedForProviderHeal() |
| 394 | if !ctrl.closed.Load() { |
| 395 | t.Fatal("provider heal returned before the detached controller closed") |
| 396 | } |
| 397 | if server.detachedBusy(path) { |
| 398 | t.Fatal("provider heal left the detached controller reattachable") |
| 399 | } |
| 400 | } |
| 401 | |
| 402 | func TestBuildTaggedInheritsWorkspacePlacement(t *testing.T) { |
| 403 | root := t.TempDir() |
| 404 | sessionDir := filepath.Join(root, "sessions") |
| 405 | if err := os.MkdirAll(sessionDir, 0o755); err != nil { |
| 406 | t.Fatal(err) |
| 407 | } |
| 408 | bc := NewBroadcaster() |
| 409 | ctrl := control.New(control.Options{Sink: bc, SessionDir: sessionDir, WorkspaceRoot: root}) |
| 410 | server := New(ctrl, bc, config.ServeConfig{}) |
| 411 | server.SetControllerBuildOptions(boot.Options{MaxSteps: 7, MaxStepsKey: "--max-steps", AgentPreset: "delivery"}) |
| 412 | var got boot.Options |
| 413 | server.buildControllerWithOptions = func(_ context.Context, _ string, opts boot.Options) (*control.Controller, error) { |
| 414 | got = opts |
| 415 | return control.New(control.Options{Sink: opts.Sink, SessionDir: opts.SessionDir, WorkspaceRoot: opts.WorkspaceRoot}), nil |
| 416 | } |
| 417 | built, _, err := server.buildTagged(context.Background(), "provider/model", false) |
| 418 | if err != nil { |
| 419 | t.Fatal(err) |
| 420 | } |
| 421 | defer built.Close() |
| 422 | if got.SessionDir != sessionDir || got.WorkspaceRoot != root { |
| 423 | t.Fatalf("build placement = (%q, %q), want (%q, %q)", got.SessionDir, got.WorkspaceRoot, sessionDir, root) |
| 424 | } |
| 425 | if got.MaxSteps != 7 || got.MaxStepsKey != "--max-steps" || got.AgentPreset != "delivery" { |
| 426 | t.Fatalf("process build options = max:%d key:%q preset:%q, want 7/--max-steps/delivery", got.MaxSteps, got.MaxStepsKey, got.AgentPreset) |
| 427 | } |
| 428 | } |
| 429 | |
| 430 | func TestSessionTagSinkBuffersUntilReplacementCommit(t *testing.T) { |
| 431 | bc := NewBroadcaster() |
| 432 | all, stop := bc.SubscribeAll() |
| 433 | defer stop() |
| 434 | tag := NewSessionTagSink(bc) |
| 435 | path := filepath.Join(t.TempDir(), "replacement.jsonl") |
| 436 | |
| 437 | tag.Emit(event.Event{Kind: event.Notice, Text: "booted"}) |
| 438 | tag.PrimePath(path) |
| 439 | if len(all) != 0 { |
| 440 | t.Fatal("replacement boot event leaked before publication committed") |
| 441 | } |
| 442 | tag.Activate() |
| 443 | if len(all) != 1 { |
| 444 | t.Fatalf("activation flushed %d frames, want 1", len(all)) |
| 445 | } |
| 446 | var frame eventwire.Event |
| 447 | if err := json.Unmarshal(<-all, &frame); err != nil { |
| 448 | t.Fatal(err) |
| 449 | } |
| 450 | if frame.Kind != "notice" || frame.SessionPath != agent.CanonicalSessionPath(path) { |
| 451 | t.Fatalf("activated boot frame = %+v", frame) |
| 452 | } |
| 453 | } |
| 454 | |
| 455 | func TestBuildTaggedFailureDiscardsBufferedBootEvents(t *testing.T) { |
| 456 | bc := NewBroadcaster() |
| 457 | all, stop := bc.SubscribeAll() |
| 458 | defer stop() |
| 459 | server := New(control.New(control.Options{}), bc, config.ServeConfig{}) |
| 460 | server.buildControllerWithOptions = func(_ context.Context, _ string, opts boot.Options) (*control.Controller, error) { |
| 461 | opts.Sink.Emit(event.Event{Kind: event.Notice, Text: "booting replacement"}) |
| 462 | return nil, errors.New("build failed") |
| 463 | } |
| 464 | if _, _, err := server.buildTagged(context.Background(), "provider/model", false); err == nil { |
| 465 | t.Fatal("buildTagged unexpectedly succeeded") |
| 466 | } |
| 467 | if len(all) != 0 { |
| 468 | t.Fatal("failed replacement leaked a buffered boot event") |
| 469 | } |
| 470 | } |
| 471 | |
| 472 | type balanceProbeController struct { |
| 473 | *control.Controller |
| 474 | calls atomic.Int32 |
| 475 | } |
| 476 | |
| 477 | func (c *balanceProbeController) Balance(context.Context) (*billing.Balance, error) { |
| 478 | c.calls.Add(1) |
| 479 | return &billing.Balance{Available: true}, nil |
| 480 | } |
| 481 | |
| 482 | func (c *balanceProbeController) RuntimeStatus() control.RuntimeStatus { |
| 483 | return control.RuntimeStatus{Running: true, PendingPrompt: true, Cancellable: true} |
| 484 | } |
| 485 | |
| 486 | func (c *balanceProbeController) RuntimeStateSnapshot() event.RuntimeStateSnapshot { |
| 487 | state := c.Controller.RuntimeStateSnapshot() |
| 488 | state.Phase, state.Running, state.PendingPrompt, state.Cancellable = "executing", true, true, true |
| 489 | return state |
| 490 | } |
| 491 | |
| 492 | func TestStatusRuntimeQuerySkipsBalance(t *testing.T) { |
| 493 | bc := NewBroadcaster() |
| 494 | ctrl := &balanceProbeController{Controller: control.New(control.Options{Sink: bc})} |
| 495 | srv := httptest.NewServer(New(ctrl, bc, config.ServeConfig{}).Handler()) |
| 496 | defer srv.Close() |
| 497 | |
| 498 | full, err := http.Get(srv.URL + "/status") |
| 499 | if err != nil { |
| 500 | t.Fatal(err) |
| 501 | } |
| 502 | _, _ = io.Copy(io.Discard, full.Body) |
| 503 | full.Body.Close() |
| 504 | before := ctrl.calls.Load() |
| 505 | if before == 0 { |
| 506 | t.Fatal("full status did not fetch balance") |
| 507 | } |
| 508 | lite, err := http.Get(srv.URL + "/status?runtime=1") |
| 509 | if err != nil { |
| 510 | t.Fatal(err) |
| 511 | } |
| 512 | body, _ := io.ReadAll(lite.Body) |
| 513 | lite.Body.Close() |
| 514 | if ctrl.calls.Load() != before { |
| 515 | t.Fatal("runtime status fetched balance") |
| 516 | } |
| 517 | for _, want := range []string{`"running":true`, `"pendingPrompt":true`, `"cancellable":true`} { |
| 518 | if !strings.Contains(string(body), want) { |
| 519 | t.Fatalf("runtime status missing %s: %s", want, body) |
| 520 | } |
| 521 | } |
| 522 | } |
| 523 | |
| 524 | func TestBusyResumeDetachesAndReattachesRunningController(t *testing.T) { |
| 525 | dir := t.TempDir() |
| 526 | aPath := filepath.Join(dir, "a.jsonl") |
| 527 | bPath := filepath.Join(dir, "b.jsonl") |
| 528 | saveServeTestSession(t, aPath) |
| 529 | saveServeTestSession(t, bPath) |
| 530 | |
| 531 | bc := NewBroadcaster() |
| 532 | tag := NewSessionTagSink(bc) |
| 533 | tag.SetPath(aPath) |
| 534 | ctrlA := control.New(control.Options{Runner: blockingRunner{}, Sink: tag, SessionDir: dir, SessionPath: aPath, Label: "test"}) |
| 535 | server := newLifecycleTestServer(t, ctrlA, bc, config.ServeConfig{}) |
| 536 | server.RegisterSessionTag(ctrlA, tag) |
| 537 | leases := control.NewSessionLeaseKeeper() |
| 538 | defer leases.Release() |
| 539 | if err := leases.Rebind(aPath); err != nil { |
| 540 | t.Fatal(err) |
| 541 | } |
| 542 | if err := server.SetSessionLeases(leases); err != nil { |
| 543 | t.Fatal(err) |
| 544 | } |
| 545 | server.buildControllerWithOptions = func(_ context.Context, _ string, opts boot.Options) (*control.Controller, error) { |
| 546 | return control.New(control.Options{Runner: blockingRunner{}, Sink: opts.Sink, SessionDir: opts.SessionDir, WorkspaceRoot: opts.WorkspaceRoot, Label: "test"}), nil |
| 547 | } |
| 548 | srv := httptest.NewServer(server.Handler()) |
| 549 | defer srv.Close() |
| 550 | defer server.CloseBackground() |
| 551 | |
| 552 | ctrlA.Submit("keep running") |
| 553 | waitRunning(t, ctrlA) |
| 554 | postResume := func(path string) { |
| 555 | payload, _ := json.Marshal(map[string]string{"path": path}) |
| 556 | resp, err := http.Post(srv.URL+"/resume", "application/json", strings.NewReader(string(payload))) |
| 557 | if err != nil { |
| 558 | t.Fatal(err) |
| 559 | } |
| 560 | defer resp.Body.Close() |
| 561 | if resp.StatusCode != http.StatusNoContent { |
| 562 | body, _ := io.ReadAll(resp.Body) |
| 563 | t.Fatalf("resume %q = %d: %s", path, resp.StatusCode, body) |
| 564 | } |
| 565 | } |
| 566 | |
| 567 | postResume(bPath) |
| 568 | wantB, err := filepath.EvalSymlinks(bPath) |
| 569 | if err != nil { |
| 570 | t.Fatal(err) |
| 571 | } |
| 572 | if got := filepath.Clean(server.ctl().SessionPath()); got != filepath.Clean(wantB) { |
| 573 | t.Fatalf("foreground session = %q, want b", got) |
| 574 | } |
| 575 | if !ctrlA.Running() { |
| 576 | t.Fatal("switched-away session stopped instead of running in background") |
| 577 | } |
| 578 | postResume(aPath) |
| 579 | if server.ctl() != control.SessionAPI(ctrlA) { |
| 580 | t.Fatal("reattach did not restore the original controller") |
| 581 | } |
| 582 | if !ctrlA.Running() { |
| 583 | t.Fatal("running turn was lost during reattach") |
| 584 | } |
| 585 | ctrlA.Cancel() |
| 586 | waitNotRunning(t, ctrlA) |
| 587 | } |
| 588 |