| 1 | package serve |
| 2 | |
| 3 | import ( |
| 4 | "bufio" |
| 5 | "bytes" |
| 6 | "context" |
| 7 | "encoding/json" |
| 8 | "fmt" |
| 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/boot" |
| 20 | "reasonix/internal/config" |
| 21 | "reasonix/internal/control" |
| 22 | "reasonix/internal/eventwire" |
| 23 | "reasonix/internal/provider" |
| 24 | ) |
| 25 | |
| 26 | // withForeignWriterLease models the local writer as a separate process. The |
| 27 | // real probe answers false for leases held by the calling process, so tests |
| 28 | // substitute one backed by the writer lease they hold in-process. |
| 29 | func withForeignWriterLease(t *testing.T, session string, held *atomic.Bool) { |
| 30 | t.Helper() |
| 31 | prev := leaseHeldByForeignRuntime |
| 32 | canonical := agent.CanonicalSessionPath(session) |
| 33 | leaseHeldByForeignRuntime = func(path string) bool { |
| 34 | return held.Load() && agent.CanonicalSessionPath(path) == canonical |
| 35 | } |
| 36 | t.Cleanup(func() { leaseHeldByForeignRuntime = prev }) |
| 37 | } |
| 38 | |
| 39 | // extendSessionOnDisk appends a message as the local writer would: load the |
| 40 | // transcript (establishing its CAS baseline), add the turn, save. |
| 41 | func extendSessionOnDisk(t *testing.T, path, content string) { |
| 42 | t.Helper() |
| 43 | loaded, err := agent.LoadSession(path) |
| 44 | if err != nil { |
| 45 | t.Fatalf("writer load: %v", err) |
| 46 | } |
| 47 | loaded.Add(provider.Message{Role: provider.RoleUser, Content: content}) |
| 48 | if err := loaded.Save(path); err != nil { |
| 49 | t.Fatalf("writer save: %v", err) |
| 50 | } |
| 51 | } |
| 52 | |
| 53 | // runningForeverController keeps RuntimeStatus busy so drain-mode handoff has |
| 54 | // something to wait on. |
| 55 | type runningForeverController struct { |
| 56 | *control.Controller |
| 57 | } |
| 58 | |
| 59 | func (c *runningForeverController) RuntimeStatus() control.RuntimeStatus { |
| 60 | return control.RuntimeStatus{Running: true} |
| 61 | } |
| 62 | |
| 63 | type snapshotFailController struct { |
| 64 | *control.Controller |
| 65 | } |
| 66 | |
| 67 | func (c *snapshotFailController) Snapshot() error { return fmt.Errorf("snapshot failed") } |
| 68 | |
| 69 | type ownershipFixture struct { |
| 70 | server *Server |
| 71 | srv *httptest.Server |
| 72 | leases *control.SessionLeaseKeeper |
| 73 | active string |
| 74 | dir string |
| 75 | grant mirrorGrant |
| 76 | } |
| 77 | |
| 78 | func newOwnershipFixture(t *testing.T) *ownershipFixture { |
| 79 | t.Helper() |
| 80 | dir := t.TempDir() |
| 81 | active := filepath.Join(dir, "active.jsonl") |
| 82 | saveServeTestSession(t, active) |
| 83 | |
| 84 | bc := NewBroadcaster() |
| 85 | exec := agent.New(nil, nil, agent.NewSession("sys"), agent.Options{}, bc) |
| 86 | ctrl := control.New(control.Options{Executor: exec, Sink: bc, SessionDir: dir, SessionPath: active}) |
| 87 | server := newLifecycleTestServer(t, ctrl, bc, config.ServeConfig{}) |
| 88 | leases := control.NewSessionLeaseKeeper() |
| 89 | if err := leases.Rebind(active); err != nil { |
| 90 | t.Fatalf("seed lease on active: %v", err) |
| 91 | } |
| 92 | server.SetSessionLeases(leases) |
| 93 | t.Cleanup(func() { |
| 94 | leases.Release() |
| 95 | ctrl.Close() |
| 96 | }) |
| 97 | fixture := &ownershipFixture{server: server, leases: leases, active: active, dir: dir} |
| 98 | fixture.srv = httptest.NewServer(server.Handler()) |
| 99 | t.Cleanup(fixture.srv.Close) |
| 100 | return fixture |
| 101 | } |
| 102 | |
| 103 | func (f *ownershipFixture) post(t *testing.T, path string, body any) (int, string) { |
| 104 | t.Helper() |
| 105 | payload, err := json.Marshal(body) |
| 106 | if err != nil { |
| 107 | t.Fatal(err) |
| 108 | } |
| 109 | resp, err := http.Post(f.srv.URL+path, "application/json", strings.NewReader(string(payload))) |
| 110 | if err != nil { |
| 111 | t.Fatal(err) |
| 112 | } |
| 113 | defer resp.Body.Close() |
| 114 | var buf bytes.Buffer |
| 115 | _, _ = buf.ReadFrom(resp.Body) |
| 116 | return resp.StatusCode, strings.TrimSpace(buf.String()) |
| 117 | } |
| 118 | |
| 119 | func (f *ownershipFixture) get(t *testing.T, path string) (int, string) { |
| 120 | t.Helper() |
| 121 | resp, err := http.Get(f.srv.URL + path) |
| 122 | if err != nil { |
| 123 | t.Fatal(err) |
| 124 | } |
| 125 | defer resp.Body.Close() |
| 126 | var buf bytes.Buffer |
| 127 | _, _ = buf.ReadFrom(resp.Body) |
| 128 | return resp.StatusCode, strings.TrimSpace(buf.String()) |
| 129 | } |
| 130 | |
| 131 | func (f *ownershipFixture) ownershipView(t *testing.T, session string) ownershipView { |
| 132 | t.Helper() |
| 133 | status, body := f.get(t, "/ownership?session="+filepath.ToSlash(session)) |
| 134 | if status != http.StatusOK { |
| 135 | t.Fatalf("GET /ownership status = %d (body %q)", status, body) |
| 136 | } |
| 137 | var view ownershipView |
| 138 | if err := json.Unmarshal([]byte(body), &view); err != nil { |
| 139 | t.Fatalf("decode ownership view: %v (body %q)", err, body) |
| 140 | } |
| 141 | return view |
| 142 | } |
| 143 | |
| 144 | // handoffForce performs the takeover a confirmed local window would issue. |
| 145 | func (f *ownershipFixture) handoffForce(t *testing.T, mode string) (int, string) { |
| 146 | t.Helper() |
| 147 | status, body := f.post(t, "/handoff", map[string]any{ |
| 148 | "sessionPath": f.active, |
| 149 | "targetWriterId": agent.SessionWriterID(), |
| 150 | "force": true, |
| 151 | "mode": mode, |
| 152 | "timeoutMs": 3000, |
| 153 | }) |
| 154 | if status == http.StatusOK { |
| 155 | if err := json.Unmarshal([]byte(body), &f.grant); err != nil { |
| 156 | t.Fatalf("decode handoff grant: %v", err) |
| 157 | } |
| 158 | } |
| 159 | return status, body |
| 160 | } |
| 161 | |
| 162 | func (f *ownershipFixture) acquireHandedOff(t *testing.T) *agent.SessionLease { |
| 163 | t.Helper() |
| 164 | lease, err := agent.TryAcquireSessionLeaseWithHandoff(f.active, f.grant.SourceWriterID, f.grant.HandoffID) |
| 165 | if err != nil { |
| 166 | t.Fatalf("acquire handed-off lease: %v", err) |
| 167 | } |
| 168 | return lease |
| 169 | } |
| 170 | |
| 171 | // TestOwnershipReportsHolderStates covers the free / serve / other triangle a |
| 172 | // takeover prompt is built from. |
| 173 | func TestOwnershipReportsHolderStates(t *testing.T) { |
| 174 | f := newOwnershipFixture(t) |
| 175 | |
| 176 | if view := f.ownershipView(t, f.active); view.Holder != "serve" { |
| 177 | t.Fatalf("foreground holder = %q, want serve", view.Holder) |
| 178 | } |
| 179 | |
| 180 | other := filepath.Join(f.dir, "other.jsonl") |
| 181 | saveServeTestSession(t, other) |
| 182 | if view := f.ownershipView(t, other); view.Holder != "free" { |
| 183 | t.Fatalf("untouched session holder = %q, want free", view.Holder) |
| 184 | } |
| 185 | |
| 186 | // A writer in another process holds the lease; the in-process test lease |
| 187 | // would read as "self", so model it through the probe seam. |
| 188 | var held atomic.Bool |
| 189 | held.Store(true) |
| 190 | withForeignWriterLease(t, other, &held) |
| 191 | if view := f.ownershipView(t, other); view.Holder != "other" { |
| 192 | t.Fatalf("foreign-held session holder = %q, want other", view.Holder) |
| 193 | } |
| 194 | held.Store(false) |
| 195 | } |
| 196 | |
| 197 | // TestHandoffReleasesLeaseAndGatesMutations walks the core takeover: after a |
| 198 | // forced handoff the local side can acquire the lease, every foreground |
| 199 | // mutation is refused with the takeover wording, /history follows the file |
| 200 | // (the writer's turns, not Serve's frozen memory), and /status flags |
| 201 | // takenOver for the remote surface. |
| 202 | func TestHandoffReleasesLeaseAndGatesMutations(t *testing.T) { |
| 203 | f := newOwnershipFixture(t) |
| 204 | |
| 205 | status, body := f.handoffForce(t, "wait") |
| 206 | if status != http.StatusOK { |
| 207 | t.Fatalf("handoff status = %d, want 200 (body %q)", status, body) |
| 208 | } |
| 209 | |
| 210 | writerLease := f.acquireHandedOff(t) |
| 211 | defer writerLease.Release() |
| 212 | |
| 213 | if view := f.ownershipView(t, f.active); view.Holder != "external" || !view.Mirrored || !view.TakenOver { |
| 214 | t.Fatalf("post-handoff ownership = %+v, want external mirror", view) |
| 215 | } |
| 216 | |
| 217 | status, body = f.post(t, "/submit", map[string]string{"input": "hello"}) |
| 218 | if status != http.StatusConflict || !strings.Contains(body, "taken over by a local Reasonix") { |
| 219 | t.Fatalf("mirrored submit = %d %q, want 409 takeover refusal", status, body) |
| 220 | } |
| 221 | |
| 222 | // The writer extends the transcript; Serve must serve the file's content. |
| 223 | extendSessionOnDisk(t, f.active, "writer turn") |
| 224 | status, body = f.get(t, "/history") |
| 225 | if status != http.StatusOK || !strings.Contains(body, "writer turn") { |
| 226 | t.Fatalf("mirrored history = %d %q, want the writer's turn from disk", status, body) |
| 227 | } |
| 228 | |
| 229 | status, body = f.get(t, "/status?runtime=1") |
| 230 | if status != http.StatusOK || !strings.Contains(body, `"takenOver":true`) { |
| 231 | t.Fatalf("mirrored status = %d %q, want takenOver", status, body) |
| 232 | } |
| 233 | |
| 234 | status, body = f.get(t, "/sessions") |
| 235 | if status != http.StatusOK || !strings.Contains(body, `"takenOver":true`) { |
| 236 | t.Fatalf("sessions list = %d %q, want takenOver row", status, body) |
| 237 | } |
| 238 | } |
| 239 | |
| 240 | func TestHandoffSnapshotFailureKeepsServeLeaseAndMirrorUnpublished(t *testing.T) { |
| 241 | dir := t.TempDir() |
| 242 | active := filepath.Join(dir, "active.jsonl") |
| 243 | saveServeTestSession(t, active) |
| 244 | bc := NewBroadcaster() |
| 245 | base := control.New(control.Options{Sink: bc, SessionDir: dir, SessionPath: active}) |
| 246 | ctrl := &snapshotFailController{Controller: base} |
| 247 | server := New(ctrl, bc, config.ServeConfig{}) |
| 248 | leases := control.NewSessionLeaseKeeper() |
| 249 | defer leases.Release() |
| 250 | defer base.Close() |
| 251 | if err := leases.Rebind(active); err != nil { |
| 252 | t.Fatal(err) |
| 253 | } |
| 254 | server.SetSessionLeases(leases) |
| 255 | srv := httptest.NewServer(server.Handler()) |
| 256 | defer srv.Close() |
| 257 | payload, _ := json.Marshal(map[string]any{ |
| 258 | "sessionPath": active, "targetWriterId": "target", "force": true, "mode": "wait", |
| 259 | }) |
| 260 | resp, err := http.Post(srv.URL+"/handoff", "application/json", bytes.NewReader(payload)) |
| 261 | if err != nil { |
| 262 | t.Fatal(err) |
| 263 | } |
| 264 | body := readAllOrFatal(t, resp) |
| 265 | if resp.StatusCode != http.StatusInternalServerError || !strings.Contains(body, "snapshot") { |
| 266 | t.Fatalf("handoff snapshot failure = %d %q, want 500", resp.StatusCode, body) |
| 267 | } |
| 268 | if got := leases.HeldPath(); got != agent.CanonicalSessionPath(active) { |
| 269 | t.Fatalf("held path = %q, want active", got) |
| 270 | } |
| 271 | if _, ok := server.mirroredEntry(active); ok { |
| 272 | t.Fatal("snapshot failure published a mirror") |
| 273 | } |
| 274 | } |
| 275 | |
| 276 | // TestHandoffRefusedWhileAttachedWithoutForce proves an unconfirmed takeover |
| 277 | // cannot yank the session while a remote client is watching. |
| 278 | func TestHandoffRefusedWhileAttachedWithoutForce(t *testing.T) { |
| 279 | f := newOwnershipFixture(t) |
| 280 | events := subscribeServeEvents(t, f.srv.URL+"/events?all=1") |
| 281 | defer events.close() |
| 282 | |
| 283 | status, body := f.post(t, "/handoff", map[string]any{"sessionPath": f.active, "targetWriterId": agent.SessionWriterID()}) |
| 284 | if status != http.StatusConflict || !strings.Contains(body, "force") { |
| 285 | t.Fatalf("unforced handoff with subscriber = %d %q, want 409 force guidance", status, body) |
| 286 | } |
| 287 | if view := f.ownershipView(t, f.active); view.Holder != "serve" { |
| 288 | t.Fatalf("holder after refused handoff = %q, want serve", view.Holder) |
| 289 | } |
| 290 | } |
| 291 | |
| 292 | // TestExternalFramesReachSubscriber proves the mirror: after takeover the |
| 293 | // writer's frames land on the remote SSE stream tagged to the mirrored |
| 294 | // session and marked current, and heartbeats surface reclaim requests. |
| 295 | func TestExternalFramesReachSubscriber(t *testing.T) { |
| 296 | f := newOwnershipFixture(t) |
| 297 | events := subscribeServeEvents(t, f.srv.URL+"/events?all=1") |
| 298 | defer events.close() |
| 299 | |
| 300 | if status, body := f.handoffForce(t, "wait"); status != http.StatusOK { |
| 301 | t.Fatalf("handoff status = %d (body %q)", status, body) |
| 302 | } |
| 303 | // Drain the takeover notice, then push a writer frame. Runtime state is a |
| 304 | // first-class frame and may be published while handoff changes ownership, so |
| 305 | // do not make the lifecycle assertion depend on incidental queue order. |
| 306 | var notice eventwire.Event |
| 307 | deadline := time.Now().Add(3 * time.Second) |
| 308 | for notice.Code != "session_taken_over" { |
| 309 | if err := events.next(¬ice, time.Until(deadline)); err != nil { |
| 310 | t.Fatalf("expected taken_over notice, got %+v (%v)", notice, err) |
| 311 | } |
| 312 | } |
| 313 | |
| 314 | status, body := f.post(t, "/external/frames", map[string]any{ |
| 315 | "sessionPath": f.active, |
| 316 | "mirrorId": f.grant.MirrorID, |
| 317 | "frames": []map[string]any{{"kind": "text", "text": "writer says hi"}}, |
| 318 | }) |
| 319 | if status != http.StatusOK { |
| 320 | t.Fatalf("external frames status = %d (body %q)", status, body) |
| 321 | } |
| 322 | |
| 323 | var frame eventwire.Event |
| 324 | if err := events.next(&frame, 3*time.Second); err != nil { |
| 325 | t.Fatalf("subscriber did not receive mirrored frame: %v", err) |
| 326 | } |
| 327 | canonical := agent.CanonicalSessionPath(f.active) |
| 328 | if frame.Kind != "text" || frame.Text != "writer says hi" || frame.SessionPath != canonical || !frame.SessionCurrent { |
| 329 | t.Fatalf("mirrored frame = %+v, want current text frame on %q", frame, canonical) |
| 330 | } |
| 331 | |
| 332 | // A heartbeat (empty frames) reports no pending reclaim. |
| 333 | var resp externalFramesResponse |
| 334 | status, body = f.post(t, "/external/frames", map[string]any{"sessionPath": f.active, "mirrorId": f.grant.MirrorID, "frames": []map[string]any{}}) |
| 335 | if status != http.StatusOK || json.Unmarshal([]byte(body), &resp) != nil || resp.ReclaimRequested { |
| 336 | t.Fatalf("heartbeat = %d %q, want reclaimRequested=false", status, body) |
| 337 | } |
| 338 | } |
| 339 | |
| 340 | // TestReclaimRestoresRemoteOwnership covers the reverse transition: the |
| 341 | // remote side reclaims, the local writer sees the request on its heartbeat, |
| 342 | // yields the lease, and Serve re-owns the session with the writer's turns |
| 343 | // reloaded from disk. |
| 344 | func TestReclaimRestoresRemoteOwnership(t *testing.T) { |
| 345 | f := newOwnershipFixture(t) |
| 346 | if status, body := f.handoffForce(t, "wait"); status != http.StatusOK { |
| 347 | t.Fatalf("handoff status = %d (body %q)", status, body) |
| 348 | } |
| 349 | |
| 350 | writerLease := f.acquireHandedOff(t) |
| 351 | var writerHeld atomic.Bool |
| 352 | writerHeld.Store(true) |
| 353 | withForeignWriterLease(t, f.active, &writerHeld) |
| 354 | extendSessionOnDisk(t, f.active, "writer turn") |
| 355 | |
| 356 | type reclaimResult struct { |
| 357 | status int |
| 358 | body string |
| 359 | } |
| 360 | done := make(chan reclaimResult, 1) |
| 361 | go func() { |
| 362 | status, body := f.post(t, "/reclaim", map[string]any{ |
| 363 | "sessionPath": f.active, |
| 364 | "mode": "wait", |
| 365 | "timeoutMs": 5000, |
| 366 | }) |
| 367 | done <- reclaimResult{status, body} |
| 368 | }() |
| 369 | |
| 370 | deadline := time.Now().Add(3 * time.Second) |
| 371 | for time.Now().Before(deadline) { |
| 372 | if f.ownershipView(t, f.active).ReclaimRequested { |
| 373 | break |
| 374 | } |
| 375 | time.Sleep(50 * time.Millisecond) |
| 376 | } |
| 377 | if view := f.ownershipView(t, f.active); !view.ReclaimRequested { |
| 378 | t.Fatal("reclaim request never became visible to the writer") |
| 379 | } |
| 380 | var heartbeat externalFramesResponse |
| 381 | status, body := f.post(t, "/external/frames", map[string]any{"sessionPath": f.active, "mirrorId": f.grant.MirrorID, "frames": []map[string]any{}}) |
| 382 | if status != http.StatusOK || json.Unmarshal([]byte(body), &heartbeat) != nil || !heartbeat.ReclaimRequested { |
| 383 | t.Fatalf("writer heartbeat = %d %q, want reclaimRequested=true", status, body) |
| 384 | } |
| 385 | |
| 386 | writerHeld.Store(false) |
| 387 | writerLease.Release() |
| 388 | select { |
| 389 | case res := <-done: |
| 390 | if res.status != http.StatusNoContent { |
| 391 | t.Fatalf("reclaim status = %d (body %q)", res.status, res.body) |
| 392 | } |
| 393 | case <-time.After(5 * time.Second): |
| 394 | t.Fatal("reclaim did not complete after the writer yielded") |
| 395 | } |
| 396 | |
| 397 | if got := f.leases.HeldPath(); got != agent.CanonicalSessionPath(f.active) { |
| 398 | t.Fatalf("post-reclaim lease = %q, want the reclaimed session", got) |
| 399 | } |
| 400 | if view := f.ownershipView(t, f.active); view.Holder != "serve" || view.Mirrored { |
| 401 | t.Fatalf("post-reclaim ownership = %+v, want serve without mirror", view) |
| 402 | } |
| 403 | status, body = f.get(t, "/history") |
| 404 | if status != http.StatusOK || !strings.Contains(body, "writer turn") { |
| 405 | t.Fatalf("post-reclaim history = %d %q, want the writer's turn reloaded", status, body) |
| 406 | } |
| 407 | } |
| 408 | |
| 409 | // TestMirrorEndRequiresReleasedLease proves the writer's farewell only ends |
| 410 | // the mirror once its lease is actually gone, then hands speaking rights |
| 411 | // straight back to the remote side. |
| 412 | func TestMirrorEndRequiresReleasedLease(t *testing.T) { |
| 413 | f := newOwnershipFixture(t) |
| 414 | if status, body := f.handoffForce(t, "wait"); status != http.StatusOK { |
| 415 | t.Fatalf("handoff status = %d (body %q)", status, body) |
| 416 | } |
| 417 | |
| 418 | writerLease := f.acquireHandedOff(t) |
| 419 | var writerHeld atomic.Bool |
| 420 | writerHeld.Store(true) |
| 421 | withForeignWriterLease(t, f.active, &writerHeld) |
| 422 | if status, body := f.post(t, "/mirror-end", map[string]string{"sessionPath": f.active, "mirrorId": f.grant.MirrorID}); status != http.StatusConflict { |
| 423 | t.Fatalf("mirror-end with live writer = %d %q, want 409", status, body) |
| 424 | } |
| 425 | if view := f.ownershipView(t, f.active); !view.Mirrored { |
| 426 | t.Fatal("mirror was cleared under a live writer") |
| 427 | } |
| 428 | |
| 429 | writerHeld.Store(false) |
| 430 | writerLease.Release() |
| 431 | if status, body := f.post(t, "/mirror-end", map[string]string{"sessionPath": f.active, "mirrorId": f.grant.MirrorID}); status != http.StatusNoContent { |
| 432 | t.Fatalf("mirror-end after release = %d %q, want 204", status, body) |
| 433 | } |
| 434 | if got := f.leases.HeldPath(); got != agent.CanonicalSessionPath(f.active) { |
| 435 | t.Fatalf("post mirror-end lease = %q, want the session re-owned", got) |
| 436 | } |
| 437 | if view := f.ownershipView(t, f.active); view.Holder != "serve" || view.Mirrored { |
| 438 | t.Fatalf("post mirror-end ownership = %+v, want serve without mirror", view) |
| 439 | } |
| 440 | } |
| 441 | |
| 442 | func TestMirrorEndLoadFailureKeepsMirrorRetryable(t *testing.T) { |
| 443 | f := newOwnershipFixture(t) |
| 444 | if status, body := f.handoffForce(t, "wait"); status != http.StatusOK { |
| 445 | t.Fatalf("handoff = %d %q", status, body) |
| 446 | } |
| 447 | writerLease := f.acquireHandedOff(t) |
| 448 | if err := writerLease.ReleaseForHandoff(f.grant.SourceWriterID, f.grant.ReturnHandoffID); err != nil { |
| 449 | t.Fatal(err) |
| 450 | } |
| 451 | if err := os.WriteFile(agent.SessionEventLogPath(f.active), []byte(`{"schema_version":999,"type":"replace"}`+"\n"), 0o600); err != nil { |
| 452 | t.Fatal(err) |
| 453 | } |
| 454 | status, body := f.post(t, "/mirror-end", map[string]string{"sessionPath": f.active, "mirrorId": f.grant.MirrorID}) |
| 455 | if status < http.StatusBadRequest { |
| 456 | t.Fatalf("mirror-end with unloadable session = %d %q, want failure", status, body) |
| 457 | } |
| 458 | if _, ok := f.server.mirroredEntry(f.active); !ok { |
| 459 | t.Fatal("load failure cleared mirror generation") |
| 460 | } |
| 461 | } |
| 462 | |
| 463 | func TestMirrorEndCommitFailureRestoresPreviousServeSession(t *testing.T) { |
| 464 | f := newOwnershipFixture(t) |
| 465 | if status, body := f.handoffForce(t, "wait"); status != http.StatusOK { |
| 466 | t.Fatalf("handoff = %d %q", status, body) |
| 467 | } |
| 468 | writerLease := f.acquireHandedOff(t) |
| 469 | previousPath := filepath.Join(f.dir, "previous.jsonl") |
| 470 | saveServeTestSession(t, previousPath) |
| 471 | previousLoaded, err := agent.LoadSession(previousPath) |
| 472 | if err != nil { |
| 473 | t.Fatal(err) |
| 474 | } |
| 475 | previousCtrl := f.server.ctl() |
| 476 | if err := f.leases.Rebind(previousPath); err != nil { |
| 477 | t.Fatal(err) |
| 478 | } |
| 479 | if err := f.leases.BindSessionAuthority(previousLoaded); err != nil { |
| 480 | t.Fatal(err) |
| 481 | } |
| 482 | previousCtrl.Resume(previousLoaded, previousPath) |
| 483 | if concrete, ok := previousCtrl.(*control.Controller); ok { |
| 484 | if err := f.leases.BindControllerAuthority(concrete); err != nil { |
| 485 | t.Fatal(err) |
| 486 | } |
| 487 | f.server.setControllerPath(concrete, previousPath) |
| 488 | } |
| 489 | previousPath = f.leases.HeldPath() |
| 490 | if err := writerLease.ReleaseForHandoff(f.grant.SourceWriterID, f.grant.ReturnHandoffID); err != nil { |
| 491 | t.Fatal(err) |
| 492 | } |
| 493 | |
| 494 | replacement := control.New(control.Options{SessionDir: f.dir, SessionPath: filepath.Join(f.dir, "replacement.jsonl")}) |
| 495 | defer replacement.Close() |
| 496 | resumeBindHookForTest = func() { |
| 497 | f.server.mu.Lock() |
| 498 | f.server.ctrl = replacement |
| 499 | f.server.mu.Unlock() |
| 500 | } |
| 501 | t.Cleanup(func() { resumeBindHookForTest = nil }) |
| 502 | status, _ := f.post(t, "/mirror-end", map[string]string{"sessionPath": f.active, "mirrorId": f.grant.MirrorID}) |
| 503 | resumeBindHookForTest = nil |
| 504 | f.server.mu.Lock() |
| 505 | f.server.ctrl = previousCtrl |
| 506 | f.server.mu.Unlock() |
| 507 | if status != http.StatusConflict { |
| 508 | t.Fatalf("commit-raced mirror-end status = %d, want 409", status) |
| 509 | } |
| 510 | if got := f.leases.HeldPath(); got != previousPath { |
| 511 | t.Fatalf("restored lease = %q, want %q", got, previousPath) |
| 512 | } |
| 513 | if got := agent.CanonicalSessionPath(previousCtrl.SessionPath()); got != previousPath { |
| 514 | t.Fatalf("restored controller path = %q, want %q", got, previousPath) |
| 515 | } |
| 516 | if err := previousCtrl.Snapshot(); err != nil { |
| 517 | t.Fatalf("restored controller lost authority: %v", err) |
| 518 | } |
| 519 | if !f.server.sessionMirrored(f.active) { |
| 520 | t.Fatal("commit failure cleared mirror generation") |
| 521 | } |
| 522 | if status, body := f.post(t, "/mirror-end", map[string]string{"sessionPath": f.active, "mirrorId": f.grant.MirrorID}); status != http.StatusNoContent { |
| 523 | t.Fatalf("retry mirror-end = %d %q, want 204", status, body) |
| 524 | } |
| 525 | } |
| 526 | |
| 527 | // TestHandoffWaitsOnRunningForeground proves drain mode refuses (rather than |
| 528 | // interrupting) while the foreground turn is still running, and that the |
| 529 | // refusal names the interrupt escape hatch. |
| 530 | func TestHandoffWaitsOnRunningForeground(t *testing.T) { |
| 531 | dir := t.TempDir() |
| 532 | active := filepath.Join(dir, "active.jsonl") |
| 533 | saveServeTestSession(t, active) |
| 534 | |
| 535 | bc := NewBroadcaster() |
| 536 | ctrl := &runningForeverController{Controller: control.New(control.Options{Sink: bc, SessionDir: dir, SessionPath: active})} |
| 537 | server := newLifecycleTestServer(t, ctrl, bc, config.ServeConfig{}) |
| 538 | leases := control.NewSessionLeaseKeeper() |
| 539 | defer leases.Release() |
| 540 | if err := leases.Rebind(active); err != nil { |
| 541 | t.Fatal(err) |
| 542 | } |
| 543 | server.SetSessionLeases(leases) |
| 544 | srv := httptest.NewServer(server.Handler()) |
| 545 | defer srv.Close() |
| 546 | |
| 547 | payload, _ := json.Marshal(map[string]any{"sessionPath": active, "targetWriterId": agent.SessionWriterID(), "force": true, "mode": "wait", "timeoutMs": 400}) |
| 548 | resp, err := http.Post(srv.URL+"/handoff", "application/json", strings.NewReader(string(payload))) |
| 549 | if err != nil { |
| 550 | t.Fatal(err) |
| 551 | } |
| 552 | body := readAllOrFatal(t, resp) |
| 553 | if resp.StatusCode != http.StatusConflict || !strings.Contains(body, "mode=interrupt") { |
| 554 | t.Fatalf("busy wait handoff = %d %q, want 409 with interrupt hint", resp.StatusCode, body) |
| 555 | } |
| 556 | if view := ownershipViewFromServer(t, srv, active); view.Holder != "serve" { |
| 557 | t.Fatalf("holder after refused busy handoff = %q, want serve", view.Holder) |
| 558 | } |
| 559 | } |
| 560 | |
| 561 | // TestNewOnMirroredForegroundRotates proves /new still works for the remote |
| 562 | // user after a takeover: the mirrored controller cannot rotate in place, so |
| 563 | // Serve publishes a replacement and keeps the mirrored transcript untouched. |
| 564 | func TestNewOnMirroredForegroundRotates(t *testing.T) { |
| 565 | f := newOwnershipFixture(t) |
| 566 | // busyDetach needs a tagged foreground controller and an injectable |
| 567 | // replacement builder (production would boot real providers). |
| 568 | foreground, ok := f.server.ctl().(*control.Controller) |
| 569 | if !ok { |
| 570 | t.Fatal("foreground controller is not a *control.Controller") |
| 571 | } |
| 572 | f.server.RegisterSessionTag(foreground, NewSessionTagSink(f.server.bc)) |
| 573 | f.server.buildControllerWithOptions = func(_ context.Context, _ string, opts boot.Options) (*control.Controller, error) { |
| 574 | return control.New(control.Options{Sink: opts.Sink, SessionDir: opts.SessionDir}), nil |
| 575 | } |
| 576 | |
| 577 | if status, body := f.handoffForce(t, "wait"); status != http.StatusOK { |
| 578 | t.Fatalf("handoff status = %d (body %q)", status, body) |
| 579 | } |
| 580 | status, body := f.post(t, "/new", map[string]any{}) |
| 581 | if status != http.StatusNoContent { |
| 582 | t.Fatalf("/new on mirrored foreground = %d %q, want 204", status, body) |
| 583 | } |
| 584 | |
| 585 | canonical := agent.CanonicalSessionPath(f.active) |
| 586 | if got := agent.CanonicalSessionPath(f.server.ctl().SessionPath()); got == canonical { |
| 587 | t.Fatalf("/new kept the foreground on the mirrored session %q", got) |
| 588 | } |
| 589 | if !f.server.sessionMirrored(f.active) { |
| 590 | t.Fatal("/new cleared the mirror; the local writer was cut off") |
| 591 | } |
| 592 | if held := f.leases.HeldPath(); held == canonical || held == "" { |
| 593 | t.Fatalf("post-rotation lease = %q, want the replacement session", held) |
| 594 | } |
| 595 | } |
| 596 | |
| 597 | func readAllOrFatal(t *testing.T, resp *http.Response) string { |
| 598 | t.Helper() |
| 599 | defer resp.Body.Close() |
| 600 | var buf bytes.Buffer |
| 601 | _, _ = buf.ReadFrom(resp.Body) |
| 602 | return strings.TrimSpace(buf.String()) |
| 603 | } |
| 604 | |
| 605 | func ownershipViewFromServer(t *testing.T, srv *httptest.Server, session string) ownershipView { |
| 606 | t.Helper() |
| 607 | resp, err := http.Get(srv.URL + "/ownership?session=" + filepath.ToSlash(session)) |
| 608 | if err != nil { |
| 609 | t.Fatal(err) |
| 610 | } |
| 611 | body := readAllOrFatal(t, resp) |
| 612 | var view ownershipView |
| 613 | if err := json.Unmarshal([]byte(body), &view); err != nil { |
| 614 | t.Fatalf("decode ownership view: %v (body %q)", err, body) |
| 615 | } |
| 616 | return view |
| 617 | } |
| 618 | |
| 619 | // serveEventStream reads an SSE endpoint frame by frame. |
| 620 | type serveEventStream struct { |
| 621 | lines chan string |
| 622 | stop chan struct{} |
| 623 | } |
| 624 | |
| 625 | func subscribeServeEvents(t *testing.T, url string) *serveEventStream { |
| 626 | t.Helper() |
| 627 | req, err := http.NewRequest(http.MethodGet, url, nil) |
| 628 | if err != nil { |
| 629 | t.Fatal(err) |
| 630 | } |
| 631 | req.Header.Set("Accept", "text/event-stream") |
| 632 | resp, err := http.DefaultClient.Do(req) |
| 633 | if err != nil { |
| 634 | t.Fatal(err) |
| 635 | } |
| 636 | if resp.StatusCode != http.StatusOK { |
| 637 | resp.Body.Close() |
| 638 | t.Fatalf("subscribe %s status = %d", url, resp.StatusCode) |
| 639 | } |
| 640 | stream := &serveEventStream{lines: make(chan string, 64), stop: make(chan struct{})} |
| 641 | go func() { |
| 642 | defer close(stream.lines) |
| 643 | defer resp.Body.Close() |
| 644 | scanner := bufio.NewScanner(resp.Body) |
| 645 | for scanner.Scan() { |
| 646 | select { |
| 647 | case <-stream.stop: |
| 648 | return |
| 649 | default: |
| 650 | } |
| 651 | line := scanner.Text() |
| 652 | if data, ok := strings.CutPrefix(line, "data: "); ok { |
| 653 | select { |
| 654 | case stream.lines <- data: |
| 655 | case <-stream.stop: |
| 656 | return |
| 657 | } |
| 658 | } |
| 659 | } |
| 660 | }() |
| 661 | t.Cleanup(func() { |
| 662 | select { |
| 663 | case <-stream.stop: |
| 664 | default: |
| 665 | close(stream.stop) |
| 666 | } |
| 667 | }) |
| 668 | return stream |
| 669 | } |
| 670 | |
| 671 | func (s *serveEventStream) next(out any, timeout time.Duration) error { |
| 672 | select { |
| 673 | case line := <-s.lines: |
| 674 | return json.Unmarshal([]byte(line), out) |
| 675 | case <-time.After(timeout): |
| 676 | return fmt.Errorf("timed out waiting for an SSE frame") |
| 677 | } |
| 678 | } |
| 679 | |
| 680 | func (s *serveEventStream) close() { |
| 681 | select { |
| 682 | case <-s.stop: |
| 683 | default: |
| 684 | close(s.stop) |
| 685 | } |
| 686 | } |
| 687 | |
| 688 | // TestAdoptRegistersDirectlyOpenedSession proves a local runtime that opened a |
| 689 | // session without any handoff can announce it to Serve, making it watchable |
| 690 | // and reclaimable by the remote side. |
| 691 | func TestAdoptRegistersDirectlyOpenedSession(t *testing.T) { |
| 692 | f := newOwnershipFixture(t) |
| 693 | other := filepath.Join(f.dir, "other.jsonl") |
| 694 | saveServeTestSession(t, other) |
| 695 | |
| 696 | // The local runtime owns it; Serve does not. Adopt succeeds. |
| 697 | writerLease, err := agent.TryAcquireSessionLease(other) |
| 698 | if err != nil { |
| 699 | t.Fatal(err) |
| 700 | } |
| 701 | defer writerLease.Release() |
| 702 | status, body := f.post(t, "/adopt", map[string]string{"sessionPath": other, "writerId": agent.SessionWriterID()}) |
| 703 | if status != http.StatusOK || !strings.Contains(body, "adopted") { |
| 704 | t.Fatalf("adopt status = %d %q, want 200 adopted", status, body) |
| 705 | } |
| 706 | if view := f.ownershipView(t, other); view.Holder != "external" || !view.Mirrored { |
| 707 | t.Fatalf("post-adopt ownership = %+v, want external mirror", view) |
| 708 | } |
| 709 | // Idempotent second adopt. |
| 710 | status, _ = f.post(t, "/adopt", map[string]string{"sessionPath": other, "writerId": agent.SessionWriterID()}) |
| 711 | if status != http.StatusOK { |
| 712 | t.Fatalf("second adopt status = %d, want 200", status) |
| 713 | } |
| 714 | } |
| 715 | |
| 716 | // TestAdoptRefusedForServeHeldSession proves sessions Serve itself holds go |
| 717 | // through /handoff, not /adopt. |
| 718 | func TestAdoptRefusedForServeHeldSession(t *testing.T) { |
| 719 | f := newOwnershipFixture(t) |
| 720 | status, body := f.post(t, "/adopt", map[string]string{"sessionPath": f.active, "writerId": agent.SessionWriterID()}) |
| 721 | if status != http.StatusConflict || !strings.Contains(body, "handoff") { |
| 722 | t.Fatalf("adopt of serve-held session = %d %q, want 409 handoff hint", status, body) |
| 723 | } |
| 724 | } |
| 725 | |
| 726 | func TestAdoptRequiresLiveClaimedWriter(t *testing.T) { |
| 727 | f := newOwnershipFixture(t) |
| 728 | other := filepath.Join(f.dir, "other.jsonl") |
| 729 | saveServeTestSession(t, other) |
| 730 | writerLease, err := agent.TryAcquireSessionLease(other) |
| 731 | if err != nil { |
| 732 | t.Fatal(err) |
| 733 | } |
| 734 | defer writerLease.Release() |
| 735 | |
| 736 | status, body := f.post(t, "/adopt", map[string]string{"sessionPath": other, "writerId": "not-the-owner"}) |
| 737 | if status != http.StatusConflict || !strings.Contains(body, "claimed writer") { |
| 738 | t.Fatalf("adopt with false owner = %d %q, want 409", status, body) |
| 739 | } |
| 740 | if view := f.ownershipView(t, other); view.Mirrored { |
| 741 | t.Fatalf("false adoption published mirror: %+v", view) |
| 742 | } |
| 743 | } |
| 744 | |
| 745 | func TestMirrorGenerationFencesFramesAndEnd(t *testing.T) { |
| 746 | f := newOwnershipFixture(t) |
| 747 | if status, body := f.handoffForce(t, "wait"); status != http.StatusOK { |
| 748 | t.Fatalf("handoff = %d %q", status, body) |
| 749 | } |
| 750 | writerLease := f.acquireHandedOff(t) |
| 751 | defer writerLease.Release() |
| 752 | |
| 753 | status, _ := f.post(t, "/external/frames", map[string]any{ |
| 754 | "sessionPath": f.active, "mirrorId": "old-generation", "frames": []eventwire.Event{}, |
| 755 | }) |
| 756 | if status != http.StatusConflict { |
| 757 | t.Fatalf("stale frames status = %d, want 409", status) |
| 758 | } |
| 759 | status, _ = f.post(t, "/mirror-end", map[string]string{"sessionPath": f.active, "mirrorId": "old-generation"}) |
| 760 | if status != http.StatusConflict { |
| 761 | t.Fatalf("stale mirror-end status = %d, want 409", status) |
| 762 | } |
| 763 | if view := f.ownershipView(t, f.active); !view.Mirrored || view.Holder != "external" { |
| 764 | t.Fatalf("stale generation changed ownership: %+v", view) |
| 765 | } |
| 766 | } |
| 767 | |
| 768 | func TestExternalFramesRequestLimits(t *testing.T) { |
| 769 | f := newOwnershipFixture(t) |
| 770 | if status, body := f.handoffForce(t, "wait"); status != http.StatusOK { |
| 771 | t.Fatalf("handoff = %d %q", status, body) |
| 772 | } |
| 773 | writerLease := f.acquireHandedOff(t) |
| 774 | defer writerLease.Release() |
| 775 | |
| 776 | frames := make([]eventwire.Event, externalFramesMaxCount+1) |
| 777 | status, _ := f.post(t, "/external/frames", map[string]any{ |
| 778 | "sessionPath": f.active, "mirrorId": f.grant.MirrorID, "frames": frames, |
| 779 | }) |
| 780 | if status != http.StatusRequestEntityTooLarge { |
| 781 | t.Fatalf("too many frames status = %d, want 413", status) |
| 782 | } |
| 783 | |
| 784 | oversized := `{"sessionPath":` + fmt.Sprintf("%q", f.active) + `,"mirrorId":` + fmt.Sprintf("%q", f.grant.MirrorID) + `,"padding":"` + strings.Repeat("x", externalFramesMaxBody) + `"}` |
| 785 | resp, err := http.Post(f.srv.URL+"/external/frames", "application/json", strings.NewReader(oversized)) |
| 786 | if err != nil { |
| 787 | t.Fatal(err) |
| 788 | } |
| 789 | defer resp.Body.Close() |
| 790 | if resp.StatusCode != http.StatusRequestEntityTooLarge { |
| 791 | t.Fatalf("oversized body status = %d, want 413", resp.StatusCode) |
| 792 | } |
| 793 | } |
| 794 | |
| 795 | // TestMirroredStatusAndHistoryBySession proves the read-only endpoints answer |
| 796 | // a spectator selecting a mirrored session with the file-backed view. |
| 797 | func TestMirroredStatusAndHistoryBySession(t *testing.T) { |
| 798 | f := newOwnershipFixture(t) |
| 799 | other := filepath.Join(f.dir, "other.jsonl") |
| 800 | saveServeTestSession(t, other) |
| 801 | writerLease, err := agent.TryAcquireSessionLease(other) |
| 802 | if err != nil { |
| 803 | t.Fatal(err) |
| 804 | } |
| 805 | defer writerLease.Release() |
| 806 | if status, body := f.post(t, "/adopt", map[string]string{"sessionPath": other, "writerId": agent.SessionWriterID()}); status != http.StatusOK { |
| 807 | t.Fatalf("adopt failed: %d %q", status, body) |
| 808 | } |
| 809 | extendSessionOnDisk(t, other, "writer turn") |
| 810 | |
| 811 | status, body := f.get(t, "/history?session="+filepath.ToSlash(other)) |
| 812 | if status != http.StatusOK || !strings.Contains(body, "writer turn") { |
| 813 | t.Fatalf("spectator history = %d %q, want the writer's turn", status, body) |
| 814 | } |
| 815 | status, body = f.get(t, "/status?runtime=1&session="+filepath.ToSlash(other)) |
| 816 | if status != http.StatusOK || !strings.Contains(body, `"takenOver":true`) { |
| 817 | t.Fatalf("spectator status = %d %q, want takenOver", status, body) |
| 818 | } |
| 819 | } |
| 820 | |
| 821 | // TestResumeSpectatorMountOnMirroredSession proves /resume on a session a |
| 822 | // local runtime owns returns 200 (read-only spectator mount) without taking |
| 823 | // ownership, so any client version can attach and render the mirrored view. |
| 824 | func TestResumeSpectatorMountOnMirroredSession(t *testing.T) { |
| 825 | f := newOwnershipFixture(t) |
| 826 | other := filepath.Join(f.dir, "other.jsonl") |
| 827 | saveServeTestSession(t, other) |
| 828 | writerLease, err := agent.TryAcquireSessionLease(other) |
| 829 | if err != nil { |
| 830 | t.Fatal(err) |
| 831 | } |
| 832 | defer writerLease.Release() |
| 833 | if status, body := f.post(t, "/adopt", map[string]string{"sessionPath": other, "writerId": agent.SessionWriterID()}); status != http.StatusOK { |
| 834 | t.Fatalf("adopt failed: %d %q", status, body) |
| 835 | } |
| 836 | status, body := f.post(t, "/resume", map[string]string{"path": other}) |
| 837 | if status != http.StatusNoContent { |
| 838 | t.Fatalf("spectator resume = %d %q, want 204", status, body) |
| 839 | } |
| 840 | // Ownership must stay external: no lease transfer happened. |
| 841 | if view := f.ownershipView(t, other); view.Holder != "external" || !view.Mirrored { |
| 842 | t.Fatalf("post-resume ownership = %+v, want external mirror", view) |
| 843 | } |
| 844 | // The foreground controller must not have moved onto the spectator target. |
| 845 | if got := f.server.ctl().SessionPath(); got == other { |
| 846 | t.Fatalf("spectator resume switched the foreground onto %q", got) |
| 847 | } |
| 848 | // The spectator's writes are refused by the expected-path fence: it is |
| 849 | // pinned to a session the foreground controller does not own. |
| 850 | payload, _ := json.Marshal(map[string]string{"input": "hello"}) |
| 851 | req, _ := http.NewRequest(http.MethodPost, f.srv.URL+"/submit", strings.NewReader(string(payload))) |
| 852 | req.Header.Set("Content-Type", "application/json") |
| 853 | req.Header.Set("X-Reasonix-Expected-Session-Path", agent.CanonicalSessionPath(other)) |
| 854 | resp, err := http.DefaultClient.Do(req) |
| 855 | if err != nil { |
| 856 | t.Fatal(err) |
| 857 | } |
| 858 | body, _ = readAll(resp) |
| 859 | if resp.StatusCode != http.StatusConflict || !strings.Contains(body, "taken over by a local Reasonix") { |
| 860 | t.Fatalf("spectator submit = %d %q, want 409 takeover refusal", resp.StatusCode, body) |
| 861 | } |
| 862 | } |
| 863 | |
| 864 | // TestSpectatorSwitchCommandsPassTheFence proves a read-only spectator pinned |
| 865 | // to a local-owned session can still run foreground-switch commands (/new) — |
| 866 | // that is how the remote side leaves its pin and regains the ability to act. |
| 867 | func TestSpectatorSwitchCommandsPassTheFence(t *testing.T) { |
| 868 | f := newOwnershipFixture(t) |
| 869 | other := filepath.Join(f.dir, "other.jsonl") |
| 870 | saveServeTestSession(t, other) |
| 871 | writerLease, err := agent.TryAcquireSessionLease(other) |
| 872 | if err != nil { |
| 873 | t.Fatal(err) |
| 874 | } |
| 875 | defer writerLease.Release() |
| 876 | if status, body := f.post(t, "/adopt", map[string]string{"sessionPath": other, "writerId": agent.SessionWriterID()}); status != http.StatusOK { |
| 877 | t.Fatalf("adopt failed: %d %q", status, body) |
| 878 | } |
| 879 | // Spectator mounts on the mirrored session. |
| 880 | if status, body := f.post(t, "/resume", map[string]string{"path": other}); status != http.StatusNoContent { |
| 881 | t.Fatalf("spectator resume = %d %q", status, body) |
| 882 | } |
| 883 | // /new as the spectator: expected path is the mirrored pin, foreground is |
| 884 | // elsewhere — the switch fence must let it through and rotate the |
| 885 | // foreground to a fresh session. |
| 886 | payload, _ := json.Marshal(map[string]any{}) |
| 887 | req, _ := http.NewRequest(http.MethodPost, f.srv.URL+"/new", strings.NewReader(string(payload))) |
| 888 | req.Header.Set("Content-Type", "application/json") |
| 889 | req.Header.Set("X-Reasonix-Expected-Session-Path", agent.CanonicalSessionPath(other)) |
| 890 | resp, err := http.DefaultClient.Do(req) |
| 891 | if err != nil { |
| 892 | t.Fatal(err) |
| 893 | } |
| 894 | body, _ := readAll(resp) |
| 895 | if resp.StatusCode != http.StatusNoContent { |
| 896 | t.Fatalf("spectator /new = %d %q, want 204", resp.StatusCode, body) |
| 897 | } |
| 898 | if got := f.server.ctl().SessionPath(); got == other { |
| 899 | t.Fatalf("/new did not rotate the foreground off the mirrored pin") |
| 900 | } |
| 901 | // The mirrored session stays mirrored (local owner untouched). |
| 902 | if view := f.ownershipView(t, other); !view.Mirrored || view.Holder != "external" { |
| 903 | t.Fatalf("post-/new ownership = %+v, want external mirror", view) |
| 904 | } |
| 905 | } |
| 906 | |
| 907 | // TestAutoReclaimCompletesOutstandingReclaim proves a writer that vanished |
| 908 | // after a reclaim was requested cannot leave the session mirrored forever: |
| 909 | // once the entry goes stale and the lease is free, the outstanding reclaim |
| 910 | // completes on the recovery sweep instead of being skipped by the flag. |
| 911 | func TestAutoReclaimCompletesOutstandingReclaim(t *testing.T) { |
| 912 | f := newOwnershipFixture(t) |
| 913 | other := filepath.Join(f.dir, "vanished.jsonl") |
| 914 | saveServeTestSession(t, other) |
| 915 | |
| 916 | writerLease, err := agent.TryAcquireSessionLease(other) |
| 917 | if err != nil { |
| 918 | t.Fatal(err) |
| 919 | } |
| 920 | held := &atomic.Bool{} |
| 921 | held.Store(true) |
| 922 | withForeignWriterLease(t, other, held) |
| 923 | |
| 924 | if status, body := f.post(t, "/adopt", map[string]any{"sessionPath": other, "writerId": agent.SessionWriterID()}); status != http.StatusOK { |
| 925 | t.Fatalf("adopt = %d %q", status, body) |
| 926 | } |
| 927 | // The writer ignores the reclaim: the wait times out, the flag stays set. |
| 928 | status, body := f.post(t, "/reclaim", map[string]any{"sessionPath": other, "timeoutMs": 200}) |
| 929 | if status != http.StatusConflict { |
| 930 | t.Fatalf("reclaim against silent writer = %d %q, want 409", status, body) |
| 931 | } |
| 932 | if view := f.ownershipView(t, other); !view.Mirrored || !view.ReclaimRequested { |
| 933 | t.Fatalf("post-timeout ownership = %+v, want mirrored with reclaim requested", view) |
| 934 | } |
| 935 | |
| 936 | canonical := agent.CanonicalSessionPath(other) |
| 937 | backdate := func() { |
| 938 | f.server.mirrorMu.Lock() |
| 939 | defer f.server.mirrorMu.Unlock() |
| 940 | if m, ok := f.server.mirrored[canonical]; ok { |
| 941 | m.lastContact = time.Now().Add(-2 * mirrorStaleAfter) |
| 942 | f.server.mirrored[canonical] = m |
| 943 | } |
| 944 | } |
| 945 | |
| 946 | // Stale but still leased: the recovery sweep must stand down. |
| 947 | backdate() |
| 948 | f.server.maybeAutoReclaimMirrored(other) |
| 949 | if view := f.ownershipView(t, other); !view.Mirrored { |
| 950 | t.Fatal("auto-reclaim cleared a mirror whose lease is still held") |
| 951 | } |
| 952 | |
| 953 | // The writer dies silently: lease gone, no mirror-end, no heartbeats. The |
| 954 | // outstanding reclaim completes and the remote side can own the session. |
| 955 | held.Store(false) |
| 956 | writerLease.Release() |
| 957 | backdate() |
| 958 | done := f.server.maybeAutoReclaimMirrored(other) |
| 959 | if done == nil { |
| 960 | t.Fatal("stale mirror recovery did not start") |
| 961 | } |
| 962 | <-done |
| 963 | if view := f.ownershipView(t, other); view.Mirrored { |
| 964 | t.Fatalf("stale mirror with outstanding reclaim was never cleared: %+v", view) |
| 965 | } |
| 966 | } |
| 967 |