| 1 | package serve |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "log/slog" |
| 7 | "net/http" |
| 8 | "strings" |
| 9 | "time" |
| 10 | |
| 11 | "reasonix/internal/agent" |
| 12 | "reasonix/internal/control" |
| 13 | "reasonix/internal/event" |
| 14 | "reasonix/internal/session" |
| 15 | ) |
| 16 | |
| 17 | // reclaim is the remote side's way back: it asks the local writer to yield |
| 18 | // the session, waits for the lease to come free, then re-owns the session. |
| 19 | // The local side demotes passively — it sees reclaimRequested on its next |
| 20 | // frame push or heartbeat — so exactly one side speaks at any moment. |
| 21 | func (s *Server) reclaim(w http.ResponseWriter, r *http.Request) { |
| 22 | var body handoffRequest |
| 23 | if err := decodeTakeoverJSON(w, r, &body); err != nil || strings.TrimSpace(body.SessionPath) == "" { |
| 24 | if err == nil { |
| 25 | http.Error(w, "missing sessionPath", http.StatusBadRequest) |
| 26 | } |
| 27 | return |
| 28 | } |
| 29 | if isSessionIDRoute(body.SessionPath) { |
| 30 | s.reclaimIdentity(w, r, body) |
| 31 | return |
| 32 | } |
| 33 | mode := parseHandoffMode(body.Mode) |
| 34 | timeout := handoffTimeout(body.TimeoutMs) |
| 35 | realPath, err := s.resolveSessionPath(body.SessionPath) |
| 36 | if err != nil { |
| 37 | http.Error(w, err.Error(), http.StatusBadRequest) |
| 38 | return |
| 39 | } |
| 40 | canonical := agent.CanonicalSessionPath(realPath) |
| 41 | |
| 42 | s.mirrorMu.Lock() |
| 43 | m, ok := s.mirrored[canonical] |
| 44 | if !ok { |
| 45 | s.mirrorMu.Unlock() |
| 46 | if s.serveHoldsSession(realPath) { |
| 47 | w.WriteHeader(http.StatusNoContent) |
| 48 | return |
| 49 | } |
| 50 | // A local holder that never adopted has no mirror forwarder to signal, |
| 51 | // so the reclaim can only wait for the lease to free. Cap that wait |
| 52 | // short: the caller needs feedback, not a two-minute hang. |
| 53 | if leaseHeldByForeignRuntime(realPath) { |
| 54 | slog.Info("serve: reclaim on un-mirrored foreign-held session (adopter absent)", |
| 55 | "session", canonical) |
| 56 | deadline := time.Now().Add(10 * time.Second) |
| 57 | for leaseHeldByForeignRuntime(realPath) { |
| 58 | if time.Now().After(deadline) { |
| 59 | http.Error(w, "session is held by a local Reasonix window that never registered a mirror; close that window or retry after it exits", http.StatusConflict) |
| 60 | return |
| 61 | } |
| 62 | time.Sleep(handoffPollInterval) |
| 63 | } |
| 64 | s.bindMu.Lock() |
| 65 | defer s.bindMu.Unlock() |
| 66 | s.resumeSession(w, r, realPath) |
| 67 | return |
| 68 | } |
| 69 | http.Error(w, "session is not held by any known runtime", http.StatusConflict) |
| 70 | return |
| 71 | } |
| 72 | m.reclaimRequested = true |
| 73 | m.reclaimMode = mode |
| 74 | m.phase = mirrorPhaseReclaimRequested |
| 75 | s.mirrored[canonical] = m |
| 76 | s.mirrorMu.Unlock() |
| 77 | s.bc.Emit(event.Event{ |
| 78 | Kind: event.Notice, |
| 79 | Code: event.NoticeCodeSessionReclaimRequested, |
| 80 | Text: "The remote side asked to take this session back.", |
| 81 | SessionPath: canonical, |
| 82 | }) |
| 83 | slog.Info("serve: reclaim requested", "session", canonical, "mode", string(mode)) |
| 84 | |
| 85 | deadline := time.Now().Add(timeout) |
| 86 | for leaseHeldByForeignRuntime(realPath) { |
| 87 | if time.Now().After(deadline) { |
| 88 | http.Error(w, "local writer did not yield the session; retry", http.StatusConflict) |
| 89 | return |
| 90 | } |
| 91 | time.Sleep(handoffPollInterval) |
| 92 | } |
| 93 | |
| 94 | s.bindMu.Lock() |
| 95 | defer s.bindMu.Unlock() |
| 96 | current, ok := s.mirroredEntry(realPath) |
| 97 | if !ok || current.mirrorID != m.mirrorID { |
| 98 | if s.serveHoldsSession(realPath) { |
| 99 | w.WriteHeader(http.StatusNoContent) |
| 100 | return |
| 101 | } |
| 102 | http.Error(w, "mirror generation changed during reclaim", http.StatusConflict) |
| 103 | return |
| 104 | } |
| 105 | s.reclaimMirroredLocked(w, realPath, current) |
| 106 | } |
| 107 | |
| 108 | func (s *Server) serveHoldsSession(realPath string) bool { |
| 109 | cur := s.ctl() |
| 110 | if cur != nil && agent.CanonicalSessionPath(cur.SessionPath()) == agent.CanonicalSessionPath(realPath) { |
| 111 | return true |
| 112 | } |
| 113 | return s.detachedBusy(realPath) |
| 114 | } |
| 115 | |
| 116 | // serveHoldsIdentity reports whether this serve process runs ref anywhere: on |
| 117 | // the foreground or as a detached background session. The writer-lock probe |
| 118 | // cannot tell the two apart from a foreign holder, so every identity ownership |
| 119 | // answer must consult this first. |
| 120 | func (s *Server) serveHoldsIdentity(ref session.SessionRef) bool { |
| 121 | if concrete, ok := s.ctl().(*control.Controller); ok { |
| 122 | if current, bound := concrete.SessionRef(); bound && current == ref { |
| 123 | return true |
| 124 | } |
| 125 | } |
| 126 | return s.detachedIdentityHolder(ref) != nil |
| 127 | } |
| 128 | |
| 129 | // reclaimIdentity is the remote side's way back for a final-format identity: |
| 130 | // ask the mirroring writer to yield, watch the writer lock go free, then |
| 131 | // re-own the session by attaching the foreground through OpenSession. |
| 132 | func (s *Server) reclaimIdentity(w http.ResponseWriter, r *http.Request, body handoffRequest) { |
| 133 | route := strings.TrimSpace(body.SessionPath) |
| 134 | mode := parseHandoffMode(body.Mode) |
| 135 | timeout := handoffTimeout(body.TimeoutMs) |
| 136 | ref, dir, err := s.resolveSessionIdentity(route) |
| 137 | if err != nil { |
| 138 | http.Error(w, err.Error(), http.StatusBadRequest) |
| 139 | return |
| 140 | } |
| 141 | |
| 142 | s.mirrorMu.Lock() |
| 143 | m, ok := s.mirrored[mirrorKey(route)] |
| 144 | if !ok { |
| 145 | s.mirrorMu.Unlock() |
| 146 | if s.serveHoldsIdentity(ref) { |
| 147 | w.WriteHeader(http.StatusNoContent) |
| 148 | return |
| 149 | } |
| 150 | // A final-format session held by a local writer that never adopted has |
| 151 | // no mirror forwarder to signal. Wait briefly for the writer lock, then |
| 152 | // re-own directly if it went free. |
| 153 | if session.ProbeWriterHeld(dir) { |
| 154 | slog.Info("serve: reclaim on un-mirrored foreign-held identity (adopter absent)", "session", route) |
| 155 | deadline := time.Now().Add(10 * time.Second) |
| 156 | for session.ProbeWriterHeld(dir) { |
| 157 | if time.Now().After(deadline) { |
| 158 | http.Error(w, "session is held by a local Reasonix window that never registered a mirror; close that window or retry after it exits", http.StatusConflict) |
| 159 | return |
| 160 | } |
| 161 | time.Sleep(handoffPollInterval) |
| 162 | } |
| 163 | s.bindMu.Lock() |
| 164 | defer s.bindMu.Unlock() |
| 165 | s.reclaimIdentityLocked(w, r.Context(), route, ref, mirroredSession{}) |
| 166 | return |
| 167 | } |
| 168 | http.Error(w, "session is not held by any known runtime", http.StatusConflict) |
| 169 | return |
| 170 | } |
| 171 | m.reclaimRequested = true |
| 172 | m.reclaimMode = mode |
| 173 | m.phase = mirrorPhaseReclaimRequested |
| 174 | s.mirrored[mirrorKey(route)] = m |
| 175 | s.mirrorMu.Unlock() |
| 176 | s.bc.Emit(event.Event{ |
| 177 | Kind: event.Notice, |
| 178 | Code: event.NoticeCodeSessionReclaimRequested, |
| 179 | Text: "The remote side asked to take this session back.", |
| 180 | SessionPath: route, |
| 181 | }) |
| 182 | slog.Info("serve: reclaim requested", "session", route, "mode", string(mode)) |
| 183 | |
| 184 | deadline := time.Now().Add(timeout) |
| 185 | for session.ProbeWriterHeld(dir) { |
| 186 | if time.Now().After(deadline) { |
| 187 | http.Error(w, "local writer did not yield the session; retry", http.StatusConflict) |
| 188 | return |
| 189 | } |
| 190 | time.Sleep(handoffPollInterval) |
| 191 | } |
| 192 | |
| 193 | s.bindMu.Lock() |
| 194 | defer s.bindMu.Unlock() |
| 195 | current, ok := s.mirroredEntry(route) |
| 196 | if !ok || current.mirrorID != m.mirrorID { |
| 197 | if s.serveHoldsIdentity(ref) { |
| 198 | w.WriteHeader(http.StatusNoContent) |
| 199 | return |
| 200 | } |
| 201 | http.Error(w, "mirror generation changed during reclaim", http.StatusConflict) |
| 202 | return |
| 203 | } |
| 204 | s.reclaimIdentityLocked(w, r.Context(), route, ref, current) |
| 205 | } |
| 206 | |
| 207 | // reclaimIdentityLocked re-owns a final-format identity. OpenSession both |
| 208 | // acquires the writer lease and republishes the foreground; only then does the |
| 209 | // mirror entry clear. Callers hold bindMu. An empty mirror ID marks an |
| 210 | // un-mirrored foreign holder that has since released. |
| 211 | func (s *Server) reclaimIdentityLocked(w http.ResponseWriter, ctx context.Context, route string, ref session.SessionRef, mirror mirroredSession) { |
| 212 | if mirror.mirrorID != "" { |
| 213 | s.touchMirrored(route, mirror.mirrorID, mirrorPhaseRecovering) |
| 214 | } |
| 215 | concrete, ok := s.ctl().(*control.Controller) |
| 216 | if !ok || !concrete.UsesExclusiveSession() { |
| 217 | http.Error(w, "session runtime unavailable", http.StatusInternalServerError) |
| 218 | return |
| 219 | } |
| 220 | if cur, bound := concrete.SessionRef(); !bound || cur != ref { |
| 221 | if err := concrete.Snapshot(); err != nil { |
| 222 | http.Error(w, "snapshot current session: "+err.Error(), http.StatusInternalServerError) |
| 223 | return |
| 224 | } |
| 225 | } |
| 226 | if _, err := concrete.OpenSession(ctx, ref); err != nil { |
| 227 | if errors.Is(err, session.ErrWriterOwned) { |
| 228 | http.Error(w, "local writer still holds the session; retry", http.StatusConflict) |
| 229 | } else { |
| 230 | http.Error(w, "open session: "+err.Error(), http.StatusConflict) |
| 231 | } |
| 232 | return |
| 233 | } |
| 234 | // The re-owned identity is the foreground again: refresh the frame tag so |
| 235 | // live turns carry the reclaimed session's id (the pre-reclaim tag points |
| 236 | // elsewhere and the desktop pump would drop the frames). |
| 237 | s.setControllerPath(concrete, "") |
| 238 | if mirror.mirrorID != "" { |
| 239 | if _, ok := s.clearMirrored(route, mirror.mirrorID); !ok { |
| 240 | http.Error(w, "mirror generation changed", http.StatusConflict) |
| 241 | return |
| 242 | } |
| 243 | } |
| 244 | w.Header().Set(sessionIDHeader, ref.SessionID) |
| 245 | s.announceSessionChanged("", false) |
| 246 | s.broadcastReclaimed(route) |
| 247 | w.WriteHeader(http.StatusNoContent) |
| 248 | s.replayPendingPromptsBroadcast() |
| 249 | } |
| 250 | |
| 251 | // reclaimMirroredLocked acquires the returning writer's reservation, reloads |
| 252 | // and binds the controller, and only then clears the matching mirror epoch. |
| 253 | // Callers hold bindMu. |
| 254 | func (s *Server) reclaimMirroredLocked(w http.ResponseWriter, realPath string, mirror mirroredSession) { |
| 255 | current, ok := s.mirroredEntry(realPath) |
| 256 | if !ok || current.mirrorID != mirror.mirrorID { |
| 257 | http.Error(w, "mirror generation changed", http.StatusConflict) |
| 258 | return |
| 259 | } |
| 260 | s.touchMirrored(realPath, mirror.mirrorID, mirrorPhaseRecovering) |
| 261 | cur := s.ctl() |
| 262 | if cur == nil || s.leases == nil { |
| 263 | http.Error(w, "session runtime unavailable", http.StatusInternalServerError) |
| 264 | return |
| 265 | } |
| 266 | canonical := agent.CanonicalSessionPath(realPath) |
| 267 | if agent.CanonicalSessionPath(cur.SessionPath()) != canonical && !s.foregroundMirroredLocked() { |
| 268 | if err := cur.Snapshot(); err != nil { |
| 269 | http.Error(w, "snapshot current session: "+err.Error(), http.StatusInternalServerError) |
| 270 | return |
| 271 | } |
| 272 | } |
| 273 | previous, err := s.acquireReturningLease(realPath, mirror) |
| 274 | if err != nil { |
| 275 | if errors.Is(err, agent.ErrSessionLeaseHeld) { |
| 276 | http.Error(w, sessionInUseError(err), http.StatusConflict) |
| 277 | } else { |
| 278 | http.Error(w, "session lease: "+err.Error(), http.StatusInternalServerError) |
| 279 | } |
| 280 | return |
| 281 | } |
| 282 | committed := false |
| 283 | defer func() { |
| 284 | if committed { |
| 285 | if previous != nil { |
| 286 | previous.RetireDetached() |
| 287 | } |
| 288 | return |
| 289 | } |
| 290 | s.rollbackReclaimLease(cur, previous) |
| 291 | }() |
| 292 | loaded, err := agent.LoadSession(realPath) |
| 293 | if err != nil { |
| 294 | http.Error(w, "load session: "+err.Error(), http.StatusBadRequest) |
| 295 | return |
| 296 | } |
| 297 | if !s.commitLoadedResume(w, cur, loaded, realPath) { |
| 298 | return |
| 299 | } |
| 300 | if _, ok := s.clearMirrored(realPath, mirror.mirrorID); !ok { |
| 301 | http.Error(w, "mirror generation changed", http.StatusConflict) |
| 302 | return |
| 303 | } |
| 304 | committed = true |
| 305 | s.bc.ResetSessionPath(realPath) |
| 306 | s.announceSessionChanged(realPath, false) |
| 307 | s.broadcastReclaimed(realPath) |
| 308 | w.WriteHeader(http.StatusNoContent) |
| 309 | s.replayPendingPromptsBroadcast() |
| 310 | } |
| 311 | |
| 312 | // rollbackReclaimLease restores the controller and keeper that were detached |
| 313 | // while a mirrored target was acquired. commitLoadedResume can reject after |
| 314 | // Resume (for example when a test hook rotates the current controller), so the |
| 315 | // source transcript is reloaded and re-authorized before the failed target |
| 316 | // lease is retired. |
| 317 | func (s *Server) rollbackReclaimLease(cur control.SessionAPI, previous *control.SessionLeaseKeeper) { |
| 318 | failed := s.leases.Split() |
| 319 | if previous == nil { |
| 320 | if failed != nil { |
| 321 | failed.Release() |
| 322 | } |
| 323 | return |
| 324 | } |
| 325 | previousPath := previous.HeldPath() |
| 326 | loaded, err := agent.LoadSession(previousPath) |
| 327 | if err == nil { |
| 328 | err = previous.BindSessionAuthority(loaded) |
| 329 | } |
| 330 | if err == nil { |
| 331 | cur.Resume(loaded, previousPath) |
| 332 | } else { |
| 333 | slog.Error("serve: restore source after failed reclaim", "err", err) |
| 334 | } |
| 335 | s.leases.Adopt(previous) |
| 336 | if ctrl, ok := cur.(*control.Controller); ok && err == nil { |
| 337 | if bindErr := s.leases.BindControllerAuthority(ctrl); bindErr != nil { |
| 338 | slog.Error("serve: restore source authority after failed reclaim", "err", bindErr) |
| 339 | } |
| 340 | } |
| 341 | if failed != nil { |
| 342 | // The same controller may already be restored through s.leases. Retire |
| 343 | // only the failed target lease without clearing that shared authority. |
| 344 | failed.RetireDetached() |
| 345 | } |
| 346 | } |
| 347 | |
| 348 | func (s *Server) acquireReturningLease(realPath string, mirror mirroredSession) (*control.SessionLeaseKeeper, error) { |
| 349 | info, err := agent.LoadSessionLeaseInfo(realPath) |
| 350 | if err == nil && info != nil && info.HandoffTo == agent.SessionWriterID() && |
| 351 | info.HandoffID == mirror.returnHandoffID && info.WriterID == mirror.targetWriterID { |
| 352 | return s.leases.RebindDetachingWithHandoff(realPath, mirror.targetWriterID, mirror.returnHandoffID) |
| 353 | } |
| 354 | return s.leases.RebindDetaching(realPath) |
| 355 | } |
| 356 | |
| 357 | func (s *Server) broadcastReclaimed(realPath string) { |
| 358 | s.bc.Emit(event.Event{ |
| 359 | Kind: event.Notice, |
| 360 | Code: event.NoticeCodeSessionReclaimed, |
| 361 | Text: "This session is driven remotely again.", |
| 362 | SessionPath: mirrorKey(realPath), |
| 363 | }) |
| 364 | } |
| 365 | |
| 366 | // maybeAutoReclaimMirrored recovers a mirror whose writer vanished without |
| 367 | // calling /mirror-end (killed window, laptop died). The OS releases the lease |
| 368 | // with the process; once the entry is stale and the lease is free, hand the |
| 369 | // session back to the remote side; a non-nil result closes after the attempt. |
| 370 | func (s *Server) maybeAutoReclaimMirrored(path string) <-chan struct{} { |
| 371 | m, ok := s.mirroredEntry(path) |
| 372 | if !ok { |
| 373 | return nil |
| 374 | } |
| 375 | if time.Since(m.lastContact) < mirrorStaleAfter { |
| 376 | return nil |
| 377 | } |
| 378 | if leaseHeldByForeignRuntime(path) { |
| 379 | // The writer is alive but quiet (or another runtime took the file). |
| 380 | // Push the staleness window so a chatty-but-healthy writer never |
| 381 | // gets reclaimed under itself. |
| 382 | s.touchMirrored(path, m.mirrorID, "") |
| 383 | return nil |
| 384 | } |
| 385 | if m.reclaimRequested { |
| 386 | // The vanished writer's OS lock is gone, so finish the request. Skipping |
| 387 | // it leaves the mirror stuck in read-only spectator mode. |
| 388 | slog.Info("serve: completing outstanding reclaim for vanished writer", |
| 389 | "session", agent.CanonicalSessionPath(path)) |
| 390 | } |
| 391 | done := make(chan struct{}) |
| 392 | go func() { |
| 393 | defer close(done) |
| 394 | s.bindMu.Lock() |
| 395 | defer s.bindMu.Unlock() |
| 396 | current, ok := s.mirroredEntry(path) |
| 397 | if !ok || current.mirrorID != m.mirrorID { |
| 398 | return |
| 399 | } |
| 400 | recorder := &statusRecorder{header: http.Header{}} |
| 401 | if isSessionIDRoute(path) { |
| 402 | if ref, _, err := s.resolveSessionIdentity(path); err == nil { |
| 403 | s.reclaimIdentityLocked(recorder, context.Background(), path, ref, current) |
| 404 | } |
| 405 | } else { |
| 406 | s.reclaimMirroredLocked(recorder, path, current) |
| 407 | } |
| 408 | if recorder.status >= http.StatusBadRequest { |
| 409 | slog.Warn("serve: auto-reclaim of stale mirror failed", "session", path, "status", recorder.status) |
| 410 | return |
| 411 | } |
| 412 | slog.Info("serve: stale mirror auto-reclaimed", "session", path) |
| 413 | }() |
| 414 | return done |
| 415 | } |
| 416 |