| 1 | package serve |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "net/http" |
| 7 | "net/http/httptest" |
| 8 | "os" |
| 9 | "path/filepath" |
| 10 | "strings" |
| 11 | "testing" |
| 12 | "time" |
| 13 | |
| 14 | "reasonix/internal/agent" |
| 15 | "reasonix/internal/config" |
| 16 | "reasonix/internal/control" |
| 17 | "reasonix/internal/event" |
| 18 | "reasonix/internal/provider" |
| 19 | "reasonix/internal/session" |
| 20 | ) |
| 21 | |
| 22 | // retireExclusiveForeground releases every writer deterministically (the |
| 23 | // foreground plus any runtime a handoff rotated away from) so Windows temp-dir |
| 24 | // cleanup does not race the idle-retirement TTL. |
| 25 | func retireExclusiveForeground(t *testing.T, ctrl *control.Controller, service *session.Service) { |
| 26 | t.Helper() |
| 27 | ctrl.Close() |
| 28 | if service != nil { |
| 29 | _ = service.CloseAll(context.Background()) |
| 30 | } |
| 31 | } |
| 32 | |
| 33 | // newIdentityLifecycleServe wraps the lifecycle test server with the frame |
| 34 | // tag an exclusive controller's production host would register, so identity |
| 35 | // transitions can assert how live frames get stamped. |
| 36 | func newIdentityLifecycleServe(t *testing.T, ctrl *control.Controller, ref session.SessionRef) *Server { |
| 37 | t.Helper() |
| 38 | bc := NewBroadcaster() |
| 39 | lifecycle := newLifecycleTestServer(t, ctrl, bc, config.ServeConfig{}) |
| 40 | tag := newSessionTagSink(bc) |
| 41 | tag.SetIdentity("", ref.SessionID) |
| 42 | lifecycle.RegisterSessionTag(ctrl, tag) |
| 43 | return lifecycle |
| 44 | } |
| 45 | |
| 46 | // openIdentityWriter opens the final-format session's writer through a bare |
| 47 | // persistence handle, standing in for the local runtime that takes over. The |
| 48 | // returned session keeps the writer lock until Close. |
| 49 | func openIdentityWriter(t *testing.T, root string, ref session.SessionRef) *session.Session { |
| 50 | t.Helper() |
| 51 | handle, err := session.NewFilesystemPersistence(root).Open(ref.SessionID, session.ReadWrite) |
| 52 | if err != nil { |
| 53 | t.Fatalf("open identity writer: %v", err) |
| 54 | } |
| 55 | return handle |
| 56 | } |
| 57 | |
| 58 | func identityRoot(t *testing.T, service *session.Service, ref session.SessionRef) string { |
| 59 | t.Helper() |
| 60 | dir, err := service.SessionDir(t.Context(), ref) |
| 61 | if err != nil { |
| 62 | t.Fatalf("resolve identity dir: %v", err) |
| 63 | } |
| 64 | return filepath.Dir(dir) |
| 65 | } |
| 66 | |
| 67 | func serveBody(t *testing.T, method, url, body string) (*http.Response, string) { |
| 68 | t.Helper() |
| 69 | req, err := http.NewRequest(method, url, strings.NewReader(body)) |
| 70 | if err != nil { |
| 71 | t.Fatal(err) |
| 72 | } |
| 73 | req.Header.Set("Content-Type", "application/json") |
| 74 | resp, err := http.DefaultClient.Do(req) |
| 75 | if err != nil { |
| 76 | t.Fatal(err) |
| 77 | } |
| 78 | defer resp.Body.Close() |
| 79 | raw := make([]byte, 0, 4<<10) |
| 80 | buf := make([]byte, 4<<10) |
| 81 | for { |
| 82 | n, readErr := resp.Body.Read(buf) |
| 83 | raw = append(raw, buf[:n]...) |
| 84 | if readErr != nil { |
| 85 | break |
| 86 | } |
| 87 | } |
| 88 | return resp, string(raw) |
| 89 | } |
| 90 | |
| 91 | // TestIdentityHandoffReleasesWriterAndOwnershipTracks proves the release half: |
| 92 | // /handoff on a final-format identity rotates the foreground, drops the writer |
| 93 | // lock before the grant is answered, and /ownership reports the external |
| 94 | // holder while the taker keeps the lock. |
| 95 | func TestIdentityHandoffReleasesWriterAndOwnershipTracks(t *testing.T) { |
| 96 | _, ctrl, service, current := newExclusiveSessionServe(t) |
| 97 | root := identityRoot(t, service, current) |
| 98 | lifecycle := newIdentityLifecycleServe(t, ctrl, current) |
| 99 | ts := httptest.NewServer(lifecycle.Handler()) |
| 100 | defer ts.Close() |
| 101 | route := "session-id:" + current.SessionID |
| 102 | |
| 103 | resp, raw := serveBody(t, http.MethodGet, ts.URL+"/ownership?session="+route, "") |
| 104 | if resp.StatusCode != http.StatusOK { |
| 105 | t.Fatalf("ownership status = %d body %s", resp.StatusCode, raw) |
| 106 | } |
| 107 | var view ownershipView |
| 108 | if err := json.Unmarshal([]byte(raw), &view); err != nil { |
| 109 | t.Fatal(err) |
| 110 | } |
| 111 | if view.Holder != "serve" || view.Running { |
| 112 | t.Fatalf("before handoff view = %+v, want serve holder idle", view) |
| 113 | } |
| 114 | defer retireExclusiveForeground(t, ctrl, service) |
| 115 | |
| 116 | resp, raw = serveBody(t, http.MethodPost, ts.URL+"/handoff", `{"sessionPath":"`+route+`","targetWriterId":"taker-writer","force":true,"mode":"wait","timeoutMs":2000}`) |
| 117 | if resp.StatusCode != http.StatusOK { |
| 118 | t.Fatalf("handoff status = %d body %s", resp.StatusCode, raw) |
| 119 | } |
| 120 | var grant mirrorGrant |
| 121 | if err := json.Unmarshal([]byte(raw), &grant); err != nil { |
| 122 | t.Fatal(err) |
| 123 | } |
| 124 | if grant.MirrorID == "" || grant.ReturnHandoffID == "" || grant.SourceWriterID == "" || |
| 125 | grant.TargetWriterID != "taker-writer" || grant.SessionPath != route { |
| 126 | t.Fatalf("handoff grant = %+v", grant) |
| 127 | } |
| 128 | if ref, bound := ctrl.SessionRef(); bound { |
| 129 | t.Fatalf("foreground still bound to %q after handoff; release must not allocate a replacement identity", ref.SessionID) |
| 130 | } |
| 131 | // The frame tag must follow the released foreground: a tag still pointing |
| 132 | // at the handed-off identity misroutes every subsequent live frame. |
| 133 | if tag := lifecycle.tagFor(ctrl); tag == nil { |
| 134 | t.Fatal("frame tag missing after handoff release") |
| 135 | } else if tag.path != "" || tag.sessionID != "" { |
| 136 | t.Fatalf("frame tag after handoff = %+v, want no route until the next identity is allocated", tag) |
| 137 | } |
| 138 | if session.ProbeWriterHeld(filepath.Join(root, current.SessionID)) { |
| 139 | t.Fatal("writer lock still held after handoff grant") |
| 140 | } |
| 141 | |
| 142 | // The taker acquires the released writer; ownership flips to external. |
| 143 | writer := openIdentityWriter(t, root, current) |
| 144 | defer writer.Close(t.Context()) |
| 145 | resp, raw = serveBody(t, http.MethodGet, ts.URL+"/ownership?session="+route, "") |
| 146 | if err := json.Unmarshal([]byte(raw), &view); err != nil { |
| 147 | t.Fatal(err) |
| 148 | } |
| 149 | if resp.StatusCode != http.StatusOK || view.Holder != "external" || !view.TakenOver { |
| 150 | t.Fatalf("after takeover view = %+v (status %d)", view, resp.StatusCode) |
| 151 | } |
| 152 | } |
| 153 | |
| 154 | // TestIdentityResumeMountsSpectatorWhenWriterHeld proves the attach contract: |
| 155 | // /resume for an identity another runtime writes answers 204 with the |
| 156 | // taken-over header instead of a hard failure, and /history serves the cold |
| 157 | // event log so the spectator can render. |
| 158 | func TestIdentityResumeMountsSpectatorWhenWriterHeld(t *testing.T) { |
| 159 | _, ctrl, service, current := newExclusiveSessionServe(t) |
| 160 | root := identityRoot(t, service, current) |
| 161 | ts := httptest.NewServer(newLifecycleTestServer(t, ctrl, NewBroadcaster(), config.ServeConfig{}).Handler()) |
| 162 | defer ts.Close() |
| 163 | route := "session-id:" + current.SessionID |
| 164 | |
| 165 | // Detach the foreground from the identity first so the resume path cannot |
| 166 | // short-circuit onto the already-bound current session. |
| 167 | if _, err := ctrl.BindFreshSession(t.Context(), "spectator-fresh"); err != nil { |
| 168 | t.Fatal(err) |
| 169 | } |
| 170 | if err := service.Close(t.Context(), current); err != nil { |
| 171 | t.Fatalf("close rotated-out runtime: %v", err) |
| 172 | } |
| 173 | writer := openIdentityWriter(t, root, current) |
| 174 | defer writer.Close(t.Context()) |
| 175 | |
| 176 | resp, raw := serveBody(t, http.MethodPost, ts.URL+"/resume", `{"sessionId":"`+current.SessionID+`"}`) |
| 177 | if resp.StatusCode != http.StatusNoContent { |
| 178 | t.Fatalf("spectator resume status = %d body %s", resp.StatusCode, raw) |
| 179 | } |
| 180 | if resp.Header.Get(sessionTakenOverHeader) == "" { |
| 181 | t.Fatal("spectator resume omitted the taken-over header") |
| 182 | } |
| 183 | if resp.Header.Get(sessionIDHeader) != current.SessionID { |
| 184 | t.Fatalf("spectator resume session header = %q", resp.Header.Get(sessionIDHeader)) |
| 185 | } |
| 186 | |
| 187 | resp, raw = serveBody(t, http.MethodGet, ts.URL+"/history?session="+route, "") |
| 188 | if resp.StatusCode != http.StatusOK { |
| 189 | t.Fatalf("spectator history status = %d body %s", resp.StatusCode, raw) |
| 190 | } |
| 191 | |
| 192 | resp, raw = serveBody(t, http.MethodGet, ts.URL+"/status?session="+route, "") |
| 193 | if resp.StatusCode != http.StatusOK { |
| 194 | t.Fatalf("spectator status code = %d body %s", resp.StatusCode, raw) |
| 195 | } |
| 196 | var status map[string]any |
| 197 | if err := json.Unmarshal([]byte(raw), &status); err != nil { |
| 198 | t.Fatal(err) |
| 199 | } |
| 200 | if taken, _ := status["takenOver"].(bool); !taken { |
| 201 | t.Fatalf("spectator status missing takenOver: %v", status) |
| 202 | } |
| 203 | retireExclusiveForeground(t, ctrl, service) |
| 204 | } |
| 205 | |
| 206 | // TestIdentityAdoptRegistersWriterAfterServeRestart pins the canonical |
| 207 | // re-registration path used when a CLI survives the resident serve restarting. |
| 208 | func TestIdentityAdoptRegistersWriterAfterServeRestart(t *testing.T) { |
| 209 | _, ctrl, service, current := newExclusiveSessionServe(t) |
| 210 | root := identityRoot(t, service, current) |
| 211 | lifecycle := newLifecycleTestServer(t, ctrl, NewBroadcaster(), config.ServeConfig{}) |
| 212 | ts := httptest.NewServer(lifecycle.Handler()) |
| 213 | defer ts.Close() |
| 214 | defer retireExclusiveForeground(t, ctrl, service) |
| 215 | route := "session-id:" + current.SessionID |
| 216 | |
| 217 | if _, err := ctrl.BindFreshSession(t.Context(), "adopt-fresh"); err != nil { |
| 218 | t.Fatal(err) |
| 219 | } |
| 220 | if err := service.Close(t.Context(), current); err != nil { |
| 221 | t.Fatalf("close rotated-out runtime: %v", err) |
| 222 | } |
| 223 | writer := openIdentityWriter(t, root, current) |
| 224 | defer writer.Close(t.Context()) |
| 225 | |
| 226 | resp, raw := serveBody(t, http.MethodPost, ts.URL+"/adopt", `{"sessionPath":"`+route+`","writerId":"surviving-cli"}`) |
| 227 | if resp.StatusCode != http.StatusOK { |
| 228 | t.Fatalf("identity adopt status = %d body %s", resp.StatusCode, raw) |
| 229 | } |
| 230 | var grant mirrorGrant |
| 231 | if err := json.Unmarshal([]byte(raw), &grant); err != nil { |
| 232 | t.Fatal(err) |
| 233 | } |
| 234 | if grant.SessionPath != route || grant.MirrorID == "" || grant.TargetWriterID != "surviving-cli" { |
| 235 | t.Fatalf("identity adopt grant = %+v", grant) |
| 236 | } |
| 237 | if mirrored, ok := lifecycle.mirroredEntry(route); !ok || mirrored.targetWriterID != "surviving-cli" { |
| 238 | t.Fatalf("identity mirror after adopt = %+v, present=%v", mirrored, ok) |
| 239 | } |
| 240 | } |
| 241 | |
| 242 | // TestIdentityReclaimReattachesForeground proves the return half: once the |
| 243 | // local writer drops the lock, /reclaim re-owns the identity for the serve and |
| 244 | // clears the mirror bookkeeping. |
| 245 | func TestIdentityReclaimReattachesForeground(t *testing.T) { |
| 246 | _, ctrl, service, current := newExclusiveSessionServe(t) |
| 247 | root := identityRoot(t, service, current) |
| 248 | lifecycle := newIdentityLifecycleServe(t, ctrl, current) |
| 249 | ts := httptest.NewServer(lifecycle.Handler()) |
| 250 | defer ts.Close() |
| 251 | route := "session-id:" + current.SessionID |
| 252 | |
| 253 | resp, raw := serveBody(t, http.MethodPost, ts.URL+"/handoff", `{"sessionPath":"`+route+`","targetWriterId":"taker-writer","force":true,"mode":"wait","timeoutMs":2000}`) |
| 254 | if resp.StatusCode != http.StatusOK { |
| 255 | t.Fatalf("handoff status = %d body %s", resp.StatusCode, raw) |
| 256 | } |
| 257 | writer := openIdentityWriter(t, root, current) |
| 258 | if err := writer.Close(t.Context()); err != nil { |
| 259 | t.Fatalf("taker release: %v", err) |
| 260 | } |
| 261 | |
| 262 | resp, raw = serveBody(t, http.MethodPost, ts.URL+"/reclaim", `{"sessionPath":"`+route+`","mode":"wait","timeoutMs":5000}`) |
| 263 | if resp.StatusCode != http.StatusNoContent { |
| 264 | t.Fatalf("reclaim status = %d body %s", resp.StatusCode, raw) |
| 265 | } |
| 266 | // After the reclaim the identity-selected status must answer ownership |
| 267 | // explicitly false: clients apply present fields only, so an omitted |
| 268 | // takenOver would pin the spectator banner forever. |
| 269 | resp, raw = serveBody(t, http.MethodGet, ts.URL+"/status?session="+route, "") |
| 270 | if resp.StatusCode != http.StatusOK { |
| 271 | t.Fatalf("post-reclaim status code = %d body %s", resp.StatusCode, raw) |
| 272 | } |
| 273 | var reclaimed map[string]any |
| 274 | if err := json.Unmarshal([]byte(raw), &reclaimed); err != nil { |
| 275 | t.Fatal(err) |
| 276 | } |
| 277 | if taken, _ := reclaimed["takenOver"].(bool); taken { |
| 278 | t.Fatalf("post-reclaim status still reports takenOver: %v", reclaimed) |
| 279 | } |
| 280 | if ref, bound := ctrl.SessionRef(); !bound || ref != current { |
| 281 | t.Fatalf("foreground ref after reclaim = %+v (bound %v), want %+v", ref, bound, current) |
| 282 | } |
| 283 | // The frame tag must follow the re-owned identity: a stale tag stamps live |
| 284 | // frames with another session and identity-routed subscribers drop them. |
| 285 | if tag := lifecycle.tagFor(ctrl); tag == nil || tag.path != "" || tag.sessionID != current.SessionID { |
| 286 | t.Fatalf("frame tag after reclaim = %+v, want identity %q", tag, current.SessionID) |
| 287 | } |
| 288 | if _, still := lifecycle.mirroredEntry(route); still { |
| 289 | t.Fatal("mirror entry survived reclaim") |
| 290 | } |
| 291 | defer retireExclusiveForeground(t, ctrl, service) |
| 292 | if !session.ProbeWriterHeld(filepath.Join(root, current.SessionID)) { |
| 293 | t.Fatal("serve did not re-acquire the writer after reclaim") |
| 294 | } |
| 295 | } |
| 296 | |
| 297 | // TestIdentityHandoffRefusesForeignHolder pins the guard: a handoff for an |
| 298 | // identity this serve does not run is refused instead of granting a session |
| 299 | // the serve cannot release. |
| 300 | func TestIdentityHandoffRefusesForeignHolder(t *testing.T) { |
| 301 | _, ctrl, service, _ := newExclusiveSessionServe(t) |
| 302 | ts := httptest.NewServer(newLifecycleTestServer(t, ctrl, NewBroadcaster(), config.ServeConfig{}).Handler()) |
| 303 | defer ts.Close() |
| 304 | resp, raw := serveBody(t, http.MethodPost, ts.URL+"/handoff", `{"sessionPath":"session-id:does-not-exist","targetWriterId":"taker","force":true}`) |
| 305 | if resp.StatusCode != http.StatusBadRequest { |
| 306 | t.Fatalf("unknown identity handoff status = %d body %s", resp.StatusCode, raw) |
| 307 | } |
| 308 | retireExclusiveForeground(t, ctrl, service) |
| 309 | } |
| 310 | |
| 311 | // shortMirrorEndWait shrinks the farewell's release wait so the tests pin the |
| 312 | // protocol (which probe re-owns, how many probes the bound allows) rather than |
| 313 | // wall-clock time. |
| 314 | func shortMirrorEndWait(t *testing.T, polls int) { |
| 315 | t.Helper() |
| 316 | wait, poll := mirrorEndReleaseWait, mirrorEndReleasePoll |
| 317 | mirrorEndReleasePoll = 2 * time.Millisecond |
| 318 | mirrorEndReleaseWait = time.Duration(polls) * mirrorEndReleasePoll |
| 319 | t.Cleanup(func() { |
| 320 | mirrorEndReleaseWait, mirrorEndReleasePoll = wait, poll |
| 321 | mirrorEndProbeHookForTest = nil |
| 322 | }) |
| 323 | } |
| 324 | |
| 325 | // handoffIdentityForTest hands the fixture's identity to "taker-writer" and |
| 326 | // returns the grant. |
| 327 | func handoffIdentityForTest(t *testing.T, url, route string) mirrorGrant { |
| 328 | t.Helper() |
| 329 | resp, raw := serveBody(t, http.MethodPost, url+"/handoff", `{"sessionPath":"`+route+`","targetWriterId":"taker-writer","force":true,"mode":"wait","timeoutMs":2000}`) |
| 330 | if resp.StatusCode != http.StatusOK { |
| 331 | t.Fatalf("handoff status = %d body %s", resp.StatusCode, raw) |
| 332 | } |
| 333 | var grant mirrorGrant |
| 334 | if err := json.Unmarshal([]byte(raw), &grant); err != nil { |
| 335 | t.Fatal(err) |
| 336 | } |
| 337 | return grant |
| 338 | } |
| 339 | |
| 340 | // TestIdentityMirrorEndAcceptsLiveWriter pins the farewell contract: the |
| 341 | // writer's return transaction sends mirror-end before process exit, so the |
| 342 | // writer lock may still be held past the release wait and the serve must accept |
| 343 | // (204) instead of 409ing a call its own protocol ordering requires. The |
| 344 | // outstanding return then belongs to the stale auto-reclaim, and the wait is |
| 345 | // bounded by the configured number of probes. |
| 346 | func TestIdentityMirrorEndAcceptsLiveWriter(t *testing.T) { |
| 347 | const polls = 3 |
| 348 | shortMirrorEndWait(t, polls) |
| 349 | _, ctrl, service, current := newExclusiveSessionServe(t) |
| 350 | root := identityRoot(t, service, current) |
| 351 | lifecycle := newIdentityLifecycleServe(t, ctrl, current) |
| 352 | ts := httptest.NewServer(lifecycle.Handler()) |
| 353 | defer ts.Close() |
| 354 | route := "session-id:" + current.SessionID |
| 355 | grant := handoffIdentityForTest(t, ts.URL, route) |
| 356 | writer := openIdentityWriter(t, root, current) |
| 357 | defer writer.Close(t.Context()) |
| 358 | probes := 0 |
| 359 | mirrorEndProbeHookForTest = func(int) { probes++ } |
| 360 | |
| 361 | resp, raw := serveBody(t, http.MethodPost, ts.URL+"/mirror-end", `{"sessionPath":"`+route+`","mirrorId":"`+grant.MirrorID+`"}`) |
| 362 | if resp.StatusCode != http.StatusNoContent { |
| 363 | t.Fatalf("mirror-end with live writer status = %d body %s", resp.StatusCode, raw) |
| 364 | } |
| 365 | if ref, bound := ctrl.SessionRef(); bound && ref == current { |
| 366 | t.Fatal("mirror-end re-owned the identity under a live writer") |
| 367 | } |
| 368 | if _, mirrored := lifecycle.mirroredEntry(route); !mirrored { |
| 369 | t.Fatal("mirror entry dropped although the writer never released; the auto-reclaim has nothing to finish") |
| 370 | } |
| 371 | if probes < 1 || probes > polls+2 { |
| 372 | t.Fatalf("farewell probed %d times, want between 1 and %d (bounded wait)", probes, polls+2) |
| 373 | } |
| 374 | retireExclusiveForeground(t, ctrl, service) |
| 375 | } |
| 376 | |
| 377 | // The desktop releases its runtime right after sending mirror-end; the |
| 378 | // farewell must re-own the identity on the first probe that sees the lock |
| 379 | // free instead of answering 204 and leaving the remote side read-only until |
| 380 | // the 30 s stale auto-reclaim. |
| 381 | func TestIdentityMirrorEndReclaimsOnceWriterReleases(t *testing.T) { |
| 382 | shortMirrorEndWait(t, 50) |
| 383 | _, ctrl, service, current := newExclusiveSessionServe(t) |
| 384 | root := identityRoot(t, service, current) |
| 385 | lifecycle := newIdentityLifecycleServe(t, ctrl, current) |
| 386 | ts := httptest.NewServer(lifecycle.Handler()) |
| 387 | defer ts.Close() |
| 388 | route := "session-id:" + current.SessionID |
| 389 | grant := handoffIdentityForTest(t, ts.URL, route) |
| 390 | writer := openIdentityWriter(t, root, current) |
| 391 | heldProbes := 0 |
| 392 | mirrorEndProbeHookForTest = func(attempt int) { |
| 393 | heldProbes++ |
| 394 | if attempt == 0 { |
| 395 | // The writer's teardown lands after the farewell was sent. |
| 396 | if err := writer.Close(t.Context()); err != nil { |
| 397 | t.Errorf("release taker writer: %v", err) |
| 398 | } |
| 399 | } |
| 400 | } |
| 401 | |
| 402 | resp, raw := serveBody(t, http.MethodPost, ts.URL+"/mirror-end", `{"sessionPath":"`+route+`","mirrorId":"`+grant.MirrorID+`"}`) |
| 403 | if resp.StatusCode != http.StatusNoContent { |
| 404 | t.Fatalf("mirror-end status = %d body %s", resp.StatusCode, raw) |
| 405 | } |
| 406 | if heldProbes != 1 { |
| 407 | t.Fatalf("farewell saw the lock held on %d probes, want exactly 1: the release must be picked up by the very next probe", heldProbes) |
| 408 | } |
| 409 | if ref, bound := ctrl.SessionRef(); !bound || ref != current { |
| 410 | t.Fatalf("foreground after farewell = %+v (bound %v), want the reclaimed identity %+v", ref, bound, current) |
| 411 | } |
| 412 | if _, mirrored := lifecycle.mirroredEntry(route); mirrored { |
| 413 | t.Fatal("mirror entry survived the farewell reclaim") |
| 414 | } |
| 415 | defer retireExclusiveForeground(t, ctrl, service) |
| 416 | if !session.ProbeWriterHeld(filepath.Join(root, current.SessionID)) { |
| 417 | t.Fatal("serve did not re-acquire the writer after the farewell") |
| 418 | } |
| 419 | } |
| 420 | |
| 421 | // An identity history read runs outside bindMu. A rotation or handoff landing |
| 422 | // between the read and the response must be reported as a changed runtime, as |
| 423 | // transcriptBoundRead already does, instead of answering the new route with |
| 424 | // the outgoing controller's transcript. |
| 425 | func TestHistoryIdentityRouteDetectsRuntimeChangeDuringRead(t *testing.T) { |
| 426 | _, ctrl, service, current := newExclusiveSessionServe(t) |
| 427 | lifecycle := newIdentityLifecycleServe(t, ctrl, current) |
| 428 | ts := httptest.NewServer(lifecycle.Handler()) |
| 429 | defer ts.Close() |
| 430 | defer retireExclusiveForeground(t, ctrl, service) |
| 431 | route := "session-id:" + current.SessionID |
| 432 | |
| 433 | historyIdentityReadHookForTest = func() { |
| 434 | if _, err := ctrl.BindFreshSession(context.Background(), "rotated-mid-read"); err != nil { |
| 435 | t.Errorf("rotate during read: %v", err) |
| 436 | } |
| 437 | } |
| 438 | t.Cleanup(func() { historyIdentityReadHookForTest = nil }) |
| 439 | resp, raw := serveBody(t, http.MethodGet, ts.URL+"/history?session="+route, "") |
| 440 | if resp.StatusCode != http.StatusConflict || !strings.Contains(raw, "transcript runtime changed during read") { |
| 441 | t.Fatalf("history across a mid-read rotation = %d %q, want 409 runtime changed", resp.StatusCode, raw) |
| 442 | } |
| 443 | historyIdentityReadHookForTest = nil |
| 444 | if err := service.Close(t.Context(), current); err != nil { |
| 445 | t.Fatalf("close rotated-out runtime: %v", err) |
| 446 | } |
| 447 | // With the rotation settled the route is served cold from the event log. |
| 448 | resp, raw = serveBody(t, http.MethodGet, ts.URL+"/history?session="+route, "") |
| 449 | if resp.StatusCode != http.StatusOK { |
| 450 | t.Fatalf("history after rotation = %d body %s", resp.StatusCode, raw) |
| 451 | } |
| 452 | } |
| 453 | |
| 454 | // TestSessionsFoldsEngineMirrorOfLegacyTranscript pins the listing fold: the |
| 455 | // engine mirrors an in-flight legacy transcript into a final-format event log |
| 456 | // keyed by the legacy branch id; the /sessions view must keep one row for that |
| 457 | // conversation instead of a transcript row plus its mirror. |
| 458 | func TestSessionsFoldsEngineMirrorOfLegacyTranscript(t *testing.T) { |
| 459 | legacyDir := t.TempDir() |
| 460 | legacy := filepath.Join(legacyDir, "mirror-me.jsonl") |
| 461 | if err := os.WriteFile(legacy, []byte(`{"role":"user","content":"hi"}`+"\n"), 0o600); err != nil { |
| 462 | t.Fatal(err) |
| 463 | } |
| 464 | v4Root := filepath.Join(t.TempDir(), "sessions-v4") |
| 465 | persistence := session.NewFilesystemPersistence(v4Root) |
| 466 | mirror, err := persistence.Create(session.CreateOptions{SessionID: "mirror-me"}) |
| 467 | if err != nil { |
| 468 | t.Fatal(err) |
| 469 | } |
| 470 | if err := mirror.Close(t.Context()); err != nil { |
| 471 | t.Fatal(err) |
| 472 | } |
| 473 | service, err := session.NewService("serve-test", persistence) |
| 474 | if err != nil { |
| 475 | t.Fatal(err) |
| 476 | } |
| 477 | exec := agent.New(nil, nil, agent.NewSession("system"), agent.Options{}, event.Discard) |
| 478 | ctrl := control.New(control.Options{Executor: exec, SessionDir: legacyDir, SessionService: service, ExclusiveSession: true}) |
| 479 | if _, err := ctrl.BindFreshSession(t.Context(), "foreground"); err != nil { |
| 480 | t.Fatal(err) |
| 481 | } |
| 482 | t.Cleanup(ctrl.Close) |
| 483 | srv := newLifecycleTestServer(t, ctrl, NewBroadcaster(), config.ServeConfig{}) |
| 484 | recorder := httptest.NewRecorder() |
| 485 | srv.sessions(recorder, httptest.NewRequest(http.MethodGet, "/sessions", nil)) |
| 486 | var rows []sessionListEntry |
| 487 | if err := json.Unmarshal(recorder.Body.Bytes(), &rows); err != nil { |
| 488 | t.Fatal(err) |
| 489 | } |
| 490 | legacyKey := agent.CanonicalSessionPath(legacy) |
| 491 | sawLegacy, sawMirror := false, false |
| 492 | for _, row := range rows { |
| 493 | if agent.CanonicalSessionPath(row.Path) == legacyKey { |
| 494 | sawLegacy = true |
| 495 | } |
| 496 | if row.SessionID == "mirror-me" { |
| 497 | sawMirror = true |
| 498 | } |
| 499 | } |
| 500 | if !sawLegacy { |
| 501 | t.Fatalf("legacy row missing from listing: %+v", rows) |
| 502 | } |
| 503 | if sawMirror { |
| 504 | t.Fatalf("engine mirror listed beside its transcript: %+v", rows) |
| 505 | } |
| 506 | retireExclusiveForeground(t, ctrl, service) |
| 507 | } |
| 508 | |
| 509 | // TestIdentityStatusAnswersFreeWriterWithRouteMatch pins the serve-restart |
| 510 | // recovery: a spectator identity whose writer exited and whose mirror entry |
| 511 | // was lost to the restart must get an explicit route-matching status with |
| 512 | // takenOver=false — the foreground snapshot names a different session and a |
| 513 | // pinned tab would discard it, leaving the banner stuck until re-attach. |
| 514 | func TestIdentityStatusAnswersFreeWriterWithRouteMatch(t *testing.T) { |
| 515 | _, ctrl, service, current := newExclusiveSessionServe(t) |
| 516 | lifecycle := newIdentityLifecycleServe(t, ctrl, current) |
| 517 | ts := httptest.NewServer(lifecycle.Handler()) |
| 518 | defer ts.Close() |
| 519 | route := "session-id:" + current.SessionID |
| 520 | |
| 521 | // Move the foreground off the identity and free its writer, mimicking a |
| 522 | // post-restart world where nothing holds the session. |
| 523 | if _, err := ctrl.BindFreshSession(t.Context(), "elsewhere"); err != nil { |
| 524 | t.Fatal(err) |
| 525 | } |
| 526 | if err := service.Close(t.Context(), current); err != nil { |
| 527 | t.Fatalf("release identity writer: %v", err) |
| 528 | } |
| 529 | |
| 530 | resp, raw := serveBody(t, http.MethodGet, ts.URL+"/status?session="+route, "") |
| 531 | if resp.StatusCode != http.StatusOK { |
| 532 | t.Fatalf("status code = %d body %s", resp.StatusCode, raw) |
| 533 | } |
| 534 | var status map[string]any |
| 535 | if err := json.Unmarshal([]byte(raw), &status); err != nil { |
| 536 | t.Fatal(err) |
| 537 | } |
| 538 | if sid, _ := status["sessionId"].(string); sid != current.SessionID { |
| 539 | t.Fatalf("status sessionId = %v, want the queried identity", status["sessionId"]) |
| 540 | } |
| 541 | if taken, _ := status["takenOver"].(bool); taken { |
| 542 | t.Fatalf("free-writer identity status still reports takenOver: %v", status) |
| 543 | } |
| 544 | retireExclusiveForeground(t, ctrl, service) |
| 545 | } |
| 546 | |
| 547 | // countSessions returns the /sessions row count so identity lifecycle tests |
| 548 | // can pin what a transition persists. |
| 549 | func countSessions(t *testing.T, url string) int { |
| 550 | t.Helper() |
| 551 | resp, raw := serveBody(t, http.MethodGet, url+"/sessions", "") |
| 552 | if resp.StatusCode != http.StatusOK { |
| 553 | t.Fatalf("sessions status = %d body %s", resp.StatusCode, raw) |
| 554 | } |
| 555 | var rows []sessionListEntry |
| 556 | if err := json.Unmarshal([]byte(raw), &rows); err != nil { |
| 557 | t.Fatal(err) |
| 558 | } |
| 559 | return len(rows) |
| 560 | } |
| 561 | |
| 562 | // A handoff releases authority; it is not a conversation. The legacy keeper |
| 563 | // only unbinds, and the identity path must match: no replacement identity is |
| 564 | // created until the user actually starts one, so handoff/reclaim cycles do not |
| 565 | // litter /sessions with empty rows. /new on the released foreground still |
| 566 | // allocates on demand. |
| 567 | func TestIdentityHandoffDoesNotPersistReplacementSession(t *testing.T) { |
| 568 | _, ctrl, service, current := newExclusiveSessionServe(t) |
| 569 | lifecycle := newIdentityLifecycleServe(t, ctrl, current) |
| 570 | ts := httptest.NewServer(lifecycle.Handler()) |
| 571 | defer ts.Close() |
| 572 | defer retireExclusiveForeground(t, ctrl, service) |
| 573 | route := "session-id:" + current.SessionID |
| 574 | |
| 575 | before := countSessions(t, ts.URL) |
| 576 | resp, raw := serveBody(t, http.MethodPost, ts.URL+"/handoff", `{"sessionPath":"`+route+`","targetWriterId":"taker-writer","force":true,"mode":"wait","timeoutMs":2000}`) |
| 577 | if resp.StatusCode != http.StatusOK { |
| 578 | t.Fatalf("handoff status = %d body %s", resp.StatusCode, raw) |
| 579 | } |
| 580 | if after := countSessions(t, ts.URL); after != before { |
| 581 | t.Fatalf("/sessions rows after handoff = %d, want %d (handoff persisted a replacement session)", after, before) |
| 582 | } |
| 583 | if _, bound := ctrl.SessionRef(); bound { |
| 584 | t.Fatal("foreground is bound after handoff; nothing should be allocated until the next turn") |
| 585 | } |
| 586 | for _, msg := range ctrl.History() { |
| 587 | if msg.Role != provider.RoleSystem { |
| 588 | t.Fatalf("released foreground still carries the handed-off conversation: %+v", msg) |
| 589 | } |
| 590 | } |
| 591 | // The released foreground stays usable: /new allocates exactly one fresh |
| 592 | // identity on demand. |
| 593 | resp, raw = serveBody(t, http.MethodPost, ts.URL+"/new", "") |
| 594 | if resp.StatusCode != http.StatusNoContent { |
| 595 | t.Fatalf("/new after handoff status = %d body %s", resp.StatusCode, raw) |
| 596 | } |
| 597 | fresh, bound := ctrl.SessionRef() |
| 598 | if !bound || fresh == current { |
| 599 | t.Fatalf("/new after handoff bound %+v (bound %v), want a fresh identity", fresh, bound) |
| 600 | } |
| 601 | if got := countSessions(t, ts.URL); got != before+1 { |
| 602 | t.Fatalf("/sessions rows after /new = %d, want %d", got, before+1) |
| 603 | } |
| 604 | } |
| 605 | |
| 606 | // A turn admitted between the unlocked quiet probe and the locked release must |
| 607 | // be refused with the legacy path's busy-again error, not raced: otherwise the |
| 608 | // foreground is unbound mid-turn, the close fails as busy, and the caller gets |
| 609 | // a 500 with an orphaned running runtime. |
| 610 | func TestIdentityHandoffRefusesTurnAdmittedAfterQuietProbe(t *testing.T) { |
| 611 | _, ctrl, service, current := newExclusiveSessionServeWithOptions(t, func(opts *control.Options) { |
| 612 | opts.Runner = blockingRunner{} |
| 613 | }) |
| 614 | root := identityRoot(t, service, current) |
| 615 | lifecycle := newIdentityLifecycleServe(t, ctrl, current) |
| 616 | ts := httptest.NewServer(lifecycle.Handler()) |
| 617 | defer ts.Close() |
| 618 | defer retireExclusiveForeground(t, ctrl, service) |
| 619 | route := "session-id:" + current.SessionID |
| 620 | |
| 621 | handoffIdentityBeforeLockHookForTest = func() { |
| 622 | // POST /chat was admitted right after the probe saw an idle foreground. |
| 623 | ctrl.Submit("keep running") |
| 624 | waitRunning(t, ctrl) |
| 625 | } |
| 626 | t.Cleanup(func() { handoffIdentityBeforeLockHookForTest = nil }) |
| 627 | defer func() { |
| 628 | ctrl.Cancel() |
| 629 | waitNotRunning(t, ctrl) |
| 630 | }() |
| 631 | |
| 632 | resp, raw := serveBody(t, http.MethodPost, ts.URL+"/handoff", `{"sessionPath":"`+route+`","targetWriterId":"taker-writer","force":true,"mode":"wait","timeoutMs":2000}`) |
| 633 | if resp.StatusCode != http.StatusConflict || !strings.Contains(raw, errHandoffBusyAgain.Error()) { |
| 634 | t.Fatalf("handoff against a freshly admitted turn = %d %q, want 409 busy-again", resp.StatusCode, raw) |
| 635 | } |
| 636 | if ref, bound := ctrl.SessionRef(); !bound || ref != current { |
| 637 | t.Fatalf("foreground binding after refused handoff = %+v (bound %v), want %+v", ref, bound, current) |
| 638 | } |
| 639 | if _, mirrored := lifecycle.mirroredEntry(route); mirrored { |
| 640 | t.Fatal("refused handoff registered a mirror entry") |
| 641 | } |
| 642 | if !session.ProbeWriterHeld(filepath.Join(root, current.SessionID)) { |
| 643 | t.Fatal("refused handoff dropped the writer lock") |
| 644 | } |
| 645 | if !ctrl.Running() { |
| 646 | t.Fatal("refused handoff interrupted the admitted turn") |
| 647 | } |
| 648 | } |
| 649 | |
| 650 | // The controller can report idle while the runtime is still finalizing the |
| 651 | // turn's terminal commit; closing such a runtime is refused as busy. The |
| 652 | // handoff must report busy-again from the locked re-check and leave the |
| 653 | // binding intact, then succeed once the runtime settles. |
| 654 | func TestIdentityHandoffRefusesFinalizingRuntimeAndRecovers(t *testing.T) { |
| 655 | _, ctrl, service, current := newExclusiveSessionServe(t) |
| 656 | lifecycle := newIdentityLifecycleServe(t, ctrl, current) |
| 657 | ts := httptest.NewServer(lifecycle.Handler()) |
| 658 | defer ts.Close() |
| 659 | defer retireExclusiveForeground(t, ctrl, service) |
| 660 | route := "session-id:" + current.SessionID |
| 661 | runtime, ok := service.Runtime(current) |
| 662 | if !ok { |
| 663 | t.Fatal("current runtime is not published") |
| 664 | } |
| 665 | generation := ctrl.ExecutionGeneration() |
| 666 | handoffIdentityBeforeLockHookForTest = func() { |
| 667 | runtime.NoteExecution(generation, session.RuntimeFinalizing, "terminal_commit") |
| 668 | } |
| 669 | t.Cleanup(func() { handoffIdentityBeforeLockHookForTest = nil }) |
| 670 | |
| 671 | body := `{"sessionPath":"` + route + `","targetWriterId":"taker-writer","force":true,"mode":"wait","timeoutMs":2000}` |
| 672 | resp, raw := serveBody(t, http.MethodPost, ts.URL+"/handoff", body) |
| 673 | if resp.StatusCode != http.StatusConflict || !strings.Contains(raw, errHandoffBusyAgain.Error()) { |
| 674 | t.Fatalf("handoff against a finalizing runtime = %d %q, want 409 busy-again", resp.StatusCode, raw) |
| 675 | } |
| 676 | if ref, bound := ctrl.SessionRef(); !bound || ref != current { |
| 677 | t.Fatalf("foreground binding after refused handoff = %+v (bound %v), want %+v", ref, bound, current) |
| 678 | } |
| 679 | if _, mirrored := lifecycle.mirroredEntry(route); mirrored { |
| 680 | t.Fatal("refused handoff registered a mirror entry") |
| 681 | } |
| 682 | |
| 683 | handoffIdentityBeforeLockHookForTest = nil |
| 684 | runtime.NoteExecution(generation, session.RuntimeIdle, "") |
| 685 | resp, raw = serveBody(t, http.MethodPost, ts.URL+"/handoff", body) |
| 686 | if resp.StatusCode != http.StatusOK { |
| 687 | t.Fatalf("handoff after the runtime settled = %d %q, want 200", resp.StatusCode, raw) |
| 688 | } |
| 689 | if _, bound := ctrl.SessionRef(); bound { |
| 690 | t.Fatal("foreground still bound after the successful handoff") |
| 691 | } |
| 692 | } |
| 693 | |
| 694 | // registerDetachedIdentityHolderForTest parks ctrl in the background registry |
| 695 | // the way a detached legacy controller ends up there after upgrading to an |
| 696 | // identity mid-turn: keyed by its former transcript path while SessionRef |
| 697 | // reports the identity. The close-on-idle watcher is omitted because these |
| 698 | // tests assert ownership predicates, not idle retirement; done is pre-closed so |
| 699 | // takeDetached hands the entry over immediately. |
| 700 | func (s *Server) registerDetachedIdentityHolderForTest(key string, ctrl *control.Controller, tag *sessionTagSink) *detachedSession { |
| 701 | done := make(chan struct{}) |
| 702 | close(done) |
| 703 | d := &detachedSession{ |
| 704 | path: agent.CanonicalSessionPath(key), ctrl: ctrl, tag: tag, |
| 705 | force: make(chan struct{}), reattach: make(chan struct{}), done: done, |
| 706 | } |
| 707 | s.detachedMu.Lock() |
| 708 | s.detached[d.path] = d |
| 709 | s.detachedMu.Unlock() |
| 710 | return d |
| 711 | } |
| 712 | |
| 713 | // A detached background session that runs an identity is still this serve's |
| 714 | // writer. /ownership must say so instead of "other", /adopt must refuse the |
| 715 | // claim instead of registering a mirror over our own writer, /status must not |
| 716 | // pin the tab read-only, and /handoff must release the detached holder the way |
| 717 | // the legacy detached handoff does. |
| 718 | func TestIdentityOwnershipCoversDetachedHolder(t *testing.T) { |
| 719 | _, ctrl, service, current := newExclusiveSessionServe(t) |
| 720 | root := identityRoot(t, service, current) |
| 721 | lifecycle := newIdentityLifecycleServe(t, ctrl, current) |
| 722 | ts := httptest.NewServer(lifecycle.Handler()) |
| 723 | defer ts.Close() |
| 724 | defer retireExclusiveForeground(t, ctrl, service) |
| 725 | |
| 726 | background, err := service.Create(t.Context(), session.CreateOptions{SessionID: "background"}) |
| 727 | if err != nil { |
| 728 | t.Fatal(err) |
| 729 | } |
| 730 | if err := service.Close(t.Context(), background.Ref()); err != nil { |
| 731 | t.Fatal(err) |
| 732 | } |
| 733 | exec := agent.New(nil, nil, agent.NewSession("system"), agent.Options{}, event.Discard) |
| 734 | bg := control.New(control.Options{Runner: blockingRunner{}, Executor: exec, SessionDir: ctrl.SessionDir(), SessionService: service, ExclusiveSession: true}) |
| 735 | if _, err := bg.OpenSession(t.Context(), background.Ref()); err != nil { |
| 736 | t.Fatal(err) |
| 737 | } |
| 738 | tag := newSessionTagSink(lifecycle.bc) |
| 739 | tag.SetIdentity("", background.Ref().SessionID) |
| 740 | lifecycle.RegisterSessionTag(bg, tag) |
| 741 | bg.Submit("keep running") |
| 742 | waitRunning(t, bg) |
| 743 | lifecycle.registerDetachedIdentityHolderForTest(filepath.Join(ctrl.SessionDir(), "upgraded.jsonl"), bg, tag) |
| 744 | closed := false |
| 745 | defer func() { |
| 746 | if !closed { |
| 747 | bg.Cancel() |
| 748 | waitNotRunning(t, bg) |
| 749 | bg.Close() |
| 750 | } |
| 751 | }() |
| 752 | route := "session-id:" + background.Ref().SessionID |
| 753 | |
| 754 | resp, raw := serveBody(t, http.MethodGet, ts.URL+"/ownership?session="+route, "") |
| 755 | var view ownershipView |
| 756 | if err := json.Unmarshal([]byte(raw), &view); err != nil { |
| 757 | t.Fatal(err) |
| 758 | } |
| 759 | if resp.StatusCode != http.StatusOK || view.Holder != "serve" || !view.Running { |
| 760 | t.Fatalf("ownership of a detached identity = %+v (status %d), want serve holder running", view, resp.StatusCode) |
| 761 | } |
| 762 | resp, raw = serveBody(t, http.MethodPost, ts.URL+"/adopt", `{"sessionPath":"`+route+`","writerId":"impostor"}`) |
| 763 | if resp.StatusCode != http.StatusConflict { |
| 764 | t.Fatalf("adopt of our own detached identity = %d body %s, want 409", resp.StatusCode, raw) |
| 765 | } |
| 766 | if _, mirrored := lifecycle.mirroredEntry(route); mirrored { |
| 767 | t.Fatal("adopt registered a mirror over this serve's own detached writer") |
| 768 | } |
| 769 | resp, raw = serveBody(t, http.MethodGet, ts.URL+"/status?session="+route, "") |
| 770 | var status map[string]any |
| 771 | if err := json.Unmarshal([]byte(raw), &status); err != nil { |
| 772 | t.Fatal(err) |
| 773 | } |
| 774 | if taken, _ := status["takenOver"].(bool); resp.StatusCode != http.StatusOK || taken { |
| 775 | t.Fatalf("status of a detached identity reports takenOver: %v (status %d)", status, resp.StatusCode) |
| 776 | } |
| 777 | |
| 778 | // The detached holder is handed off like a legacy detached session: the |
| 779 | // turn is interrupted, the writer lock drops, the controller is retired. |
| 780 | resp, raw = serveBody(t, http.MethodPost, ts.URL+"/handoff", `{"sessionPath":"`+route+`","targetWriterId":"taker-writer","force":true,"mode":"interrupt","timeoutMs":5000}`) |
| 781 | if resp.StatusCode != http.StatusOK { |
| 782 | t.Fatalf("handoff of a detached identity = %d body %s", resp.StatusCode, raw) |
| 783 | } |
| 784 | closed = true |
| 785 | if lifecycle.detachedBusy(filepath.Join(ctrl.SessionDir(), "upgraded.jsonl")) { |
| 786 | t.Fatal("handed-off detached holder is still registered") |
| 787 | } |
| 788 | if session.ProbeWriterHeld(filepath.Join(root, background.Ref().SessionID)) { |
| 789 | t.Fatal("writer lock still held after handing off the detached identity") |
| 790 | } |
| 791 | if _, mirrored := lifecycle.mirroredEntry(route); !mirrored { |
| 792 | t.Fatal("handoff of the detached identity did not register a mirror entry") |
| 793 | } |
| 794 | if ref, bound := ctrl.SessionRef(); !bound || ref != current { |
| 795 | t.Fatalf("foreground changed while handing off a detached identity: %+v (bound %v)", ref, bound) |
| 796 | } |
| 797 | } |
| 798 |