| 1 | package serve |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "log/slog" |
| 8 | "net/http" |
| 9 | "strings" |
| 10 | "time" |
| 11 | |
| 12 | "reasonix/internal/agent" |
| 13 | "reasonix/internal/control" |
| 14 | "reasonix/internal/event" |
| 15 | "reasonix/internal/session" |
| 16 | ) |
| 17 | |
| 18 | type handoffRequest struct { |
| 19 | SessionPath string `json:"sessionPath"` |
| 20 | TargetWriterID string `json:"targetWriterId"` |
| 21 | Force bool `json:"force"` |
| 22 | Mode string `json:"mode"` |
| 23 | TimeoutMs int `json:"timeoutMs"` |
| 24 | } |
| 25 | |
| 26 | // handoff releases a session Serve holds so a local runtime on this machine |
| 27 | // can take it over. With force unset it refuses while a remote client is |
| 28 | // attached — the caller is expected to have confirmed the takeover with its |
| 29 | // user via GET /ownership. wait drains a running turn; interrupt cancels it. |
| 30 | // Final-format identities may arrive as a "session-id:<id>" route or the |
| 31 | // explicit sessionId field; both take the identity release path, which swaps |
| 32 | // the writer lease instead of a path lease. |
| 33 | func (s *Server) handoff(w http.ResponseWriter, r *http.Request) { |
| 34 | var body handoffRequest |
| 35 | if err := decodeTakeoverJSON(w, r, &body); err != nil || strings.TrimSpace(body.SessionPath) == "" || strings.TrimSpace(body.TargetWriterID) == "" { |
| 36 | if err == nil { |
| 37 | http.Error(w, "missing sessionPath or targetWriterId", http.StatusBadRequest) |
| 38 | } |
| 39 | return |
| 40 | } |
| 41 | if isSessionIDRoute(body.SessionPath) { |
| 42 | s.handoffIdentity(w, r, strings.TrimSpace(body.SessionPath), strings.TrimSpace(body.TargetWriterID), body) |
| 43 | return |
| 44 | } |
| 45 | mode := parseHandoffMode(body.Mode) |
| 46 | realPath, err := s.resolveSessionPath(body.SessionPath) |
| 47 | if err != nil { |
| 48 | http.Error(w, err.Error(), http.StatusBadRequest) |
| 49 | return |
| 50 | } |
| 51 | if existing, ok := s.mirroredEntry(realPath); ok { |
| 52 | if existing.targetWriterID != strings.TrimSpace(body.TargetWriterID) { |
| 53 | http.Error(w, "session is already handed off to another writer", http.StatusConflict) |
| 54 | return |
| 55 | } |
| 56 | writeJSON(w, existing.grant("already_handed_off")) |
| 57 | return |
| 58 | } |
| 59 | if !body.Force && s.bc.Subscribers() > 0 { |
| 60 | http.Error(w, "session is attached to a remote client; retry with force after confirming the takeover", http.StatusConflict) |
| 61 | return |
| 62 | } |
| 63 | timeout := handoffTimeout(body.TimeoutMs) |
| 64 | |
| 65 | // Drain or cancel outside bindMu: waiting inside would freeze every other |
| 66 | // command for up to the whole timeout. |
| 67 | if err := s.quietSessionForHandoff(realPath, mode, timeout); err != nil { |
| 68 | http.Error(w, err.Error(), http.StatusConflict) |
| 69 | return |
| 70 | } |
| 71 | |
| 72 | s.bindMu.Lock() |
| 73 | m, err := s.handoffLocked(realPath, strings.TrimSpace(body.TargetWriterID)) |
| 74 | s.bindMu.Unlock() |
| 75 | if err != nil { |
| 76 | http.Error(w, err.Error(), statusForHandoffError(err)) |
| 77 | return |
| 78 | } |
| 79 | writeJSON(w, m.grant("handed_off")) |
| 80 | } |
| 81 | |
| 82 | // handoffIdentity releases a final-format identity the serve's foreground |
| 83 | // currently writes. The single-writer credential is the session directory's |
| 84 | // writer lock, so the release is: quiesce the foreground turn, flush, drop the |
| 85 | // foreground's binding without allocating a replacement identity, and |
| 86 | // synchronously close the handed-off runtime — its writer lock drops before the |
| 87 | // grant is answered. |
| 88 | func (s *Server) handoffIdentity(w http.ResponseWriter, r *http.Request, route, targetWriterID string, body handoffRequest) { |
| 89 | ref, _, err := s.resolveSessionIdentity(route) |
| 90 | if err != nil { |
| 91 | http.Error(w, err.Error(), http.StatusBadRequest) |
| 92 | return |
| 93 | } |
| 94 | if existing, ok := s.mirroredEntry(route); ok { |
| 95 | if existing.targetWriterID != targetWriterID { |
| 96 | http.Error(w, "session is already handed off to another writer", http.StatusConflict) |
| 97 | return |
| 98 | } |
| 99 | writeJSON(w, existing.grant("already_handed_off")) |
| 100 | return |
| 101 | } |
| 102 | if !body.Force && s.bc.Subscribers() > 0 { |
| 103 | http.Error(w, "session is attached to a remote client; retry with force after confirming the takeover", http.StatusConflict) |
| 104 | return |
| 105 | } |
| 106 | mode := parseHandoffMode(body.Mode) |
| 107 | timeout := handoffTimeout(body.TimeoutMs) |
| 108 | // Drain or cancel outside bindMu, mirroring the legacy path. |
| 109 | if err := s.quietIdentityForHandoff(ref, mode, timeout); err != nil { |
| 110 | http.Error(w, err.Error(), statusForHandoffError(err)) |
| 111 | return |
| 112 | } |
| 113 | if handoffIdentityBeforeLockHookForTest != nil { |
| 114 | handoffIdentityBeforeLockHookForTest() |
| 115 | } |
| 116 | |
| 117 | s.bindMu.Lock() |
| 118 | m, err := s.handoffIdentityLocked(r.Context(), route, ref, targetWriterID) |
| 119 | s.bindMu.Unlock() |
| 120 | if err != nil { |
| 121 | http.Error(w, err.Error(), statusForHandoffError(err)) |
| 122 | return |
| 123 | } |
| 124 | writeJSON(w, m.grant("handed_off")) |
| 125 | } |
| 126 | |
| 127 | // quietHandoffTarget answers the drain/cancel poll for one handoff target: |
| 128 | // held reports whether this serve currently runs the target, busy whether a |
| 129 | // turn is still active on it. Callers cancel toward idle in interrupt mode. |
| 130 | type quietHandoffTarget func() (ctrl control.SessionAPI, held, busy bool) |
| 131 | |
| 132 | // quietHandoffLoop waits for (or cancels toward) an idle handoff target before |
| 133 | // the release transaction runs. It re-checks under bindMu afterwards: turn |
| 134 | // admission holds bindMu, so once the caller holds it and the target is idle, |
| 135 | // no new turn can start on it. |
| 136 | func quietHandoffLoop(target quietHandoffTarget, mode handoffMode, timeout time.Duration) error { |
| 137 | deadline := time.Now().Add(timeout) |
| 138 | for { |
| 139 | ctrl, held, busy := target() |
| 140 | if !held { |
| 141 | return errSessionNotHeld |
| 142 | } |
| 143 | if busy && mode == handoffModeInterrupt && ctrl != nil { |
| 144 | ctrl.Cancel() |
| 145 | } |
| 146 | if !busy { |
| 147 | return nil |
| 148 | } |
| 149 | if time.Now().After(deadline) { |
| 150 | if mode == handoffModeInterrupt { |
| 151 | return fmt.Errorf("session did not stop within %s; retry", timeout) |
| 152 | } |
| 153 | return fmt.Errorf("session is still running after %s; retry with mode=interrupt to cancel it", timeout) |
| 154 | } |
| 155 | time.Sleep(handoffPollInterval) |
| 156 | } |
| 157 | } |
| 158 | |
| 159 | // quietIdentityForHandoff waits for (or cancels toward) an idle holder of the |
| 160 | // identity — the foreground or a detached session — before the release |
| 161 | // transaction runs. |
| 162 | func (s *Server) quietIdentityForHandoff(ref session.SessionRef, mode handoffMode, timeout time.Duration) error { |
| 163 | return quietHandoffLoop(func() (control.SessionAPI, bool, bool) { |
| 164 | if cur, ok := s.ctl().(*control.Controller); ok { |
| 165 | if current, bound := cur.SessionRef(); bound && current == ref { |
| 166 | return cur, true, controllerHasActiveRuntimeWork(cur) |
| 167 | } |
| 168 | } |
| 169 | if d := s.detachedIdentityHolder(ref); d != nil { |
| 170 | return d.ctrl, true, controllerHasActiveRuntimeWork(d.ctrl) |
| 171 | } |
| 172 | return nil, false, false |
| 173 | }, mode, timeout) |
| 174 | } |
| 175 | |
| 176 | // handoffIdentityBeforeLockHookForTest runs between the unlocked quiet probe |
| 177 | // and the locked release so tests can admit a turn in that window. |
| 178 | var handoffIdentityBeforeLockHookForTest func() |
| 179 | |
| 180 | // identityHolderLocked resolves which controller of this serve runs ref: the |
| 181 | // foreground, else the detached session bound to it. Callers hold bindMu. |
| 182 | func (s *Server) identityHolderLocked(ref session.SessionRef) (*control.Controller, *detachedSession) { |
| 183 | if cur, ok := s.ctl().(*control.Controller); ok && cur.UsesExclusiveSession() { |
| 184 | if current, bound := cur.SessionRef(); bound && current == ref { |
| 185 | return cur, nil |
| 186 | } |
| 187 | } |
| 188 | if d := s.detachedIdentityHolder(ref); d != nil { |
| 189 | if concrete, ok := d.ctrl.(*control.Controller); ok { |
| 190 | return concrete, d |
| 191 | } |
| 192 | } |
| 193 | return nil, nil |
| 194 | } |
| 195 | |
| 196 | // handoffIdentityLocked performs the identity release. Callers hold bindMu and |
| 197 | // have already quieted the holder; the busy state is re-checked here because |
| 198 | // turn admission also holds bindMu, so an idle holder observed under the lock |
| 199 | // cannot start a turn before the release completes. |
| 200 | func (s *Server) handoffIdentityLocked(ctx context.Context, route string, ref session.SessionRef, targetWriterID string) (mirroredSession, error) { |
| 201 | holder, detached := s.identityHolderLocked(ref) |
| 202 | if holder == nil { |
| 203 | return mirroredSession{}, errSessionNotHeld |
| 204 | } |
| 205 | if controllerHasActiveRuntimeWork(holder) { |
| 206 | return mirroredSession{}, errHandoffBusyAgain |
| 207 | } |
| 208 | service := holder.SessionService() |
| 209 | if service == nil { |
| 210 | return mirroredSession{}, errors.New("handoff: session service unavailable") |
| 211 | } |
| 212 | // The runtime can still be finalizing a turn the controller already reports |
| 213 | // as done; Close would refuse it as busy, so treat it as busy up front. |
| 214 | if live, ok := service.Runtime(ref); ok && live.Snapshot().Phase.Busy() { |
| 215 | return mirroredSession{}, errHandoffBusyAgain |
| 216 | } |
| 217 | m, err := newMirroredSession(route, agent.SessionWriterID(), targetWriterID, mirrorPhasePending) |
| 218 | if err != nil { |
| 219 | return mirroredSession{}, fmt.Errorf("handoff: create generation: %w", err) |
| 220 | } |
| 221 | var taken *detachedSession |
| 222 | if detached != nil { |
| 223 | // Transfer ownership from the close-on-idle watcher before releasing, |
| 224 | // exactly as the legacy detached handoff does; a retiring entry is |
| 225 | // already closing and must not be handed off. |
| 226 | taken = s.takeDetached(detached.path) |
| 227 | if taken == nil { |
| 228 | return mirroredSession{}, errHandoffBusyAgain |
| 229 | } |
| 230 | if controllerHasActiveRuntimeWork(holder) { |
| 231 | _, _ = s.registerDetached(taken.ctrl, taken.keeper, taken.tag) |
| 232 | return mirroredSession{}, errHandoffBusyAgain |
| 233 | } |
| 234 | } |
| 235 | restore := func() { |
| 236 | s.reattachAfterFailedHandoff(ctx, holder, ref) |
| 237 | if taken != nil { |
| 238 | _, _ = s.registerDetached(taken.ctrl, taken.keeper, taken.tag) |
| 239 | } |
| 240 | } |
| 241 | // Release authority the way the legacy lease keeper does: flush and unbind, |
| 242 | // allocating nothing. The holder is left never-bound, so the next turn or |
| 243 | // /new allocates lazily and no empty canonical row is left in /sessions. |
| 244 | if err := holder.ReleaseSessionForHandoff(); err != nil { |
| 245 | restore() |
| 246 | return mirroredSession{}, fmt.Errorf("handoff: release session binding: %w", err) |
| 247 | } |
| 248 | // Deterministic writer release: Close drops the writer lock now instead of |
| 249 | // waiting out the idle-retirement TTL, so the taker's open cannot race a |
| 250 | // lingering lease. A refused close re-attaches the holder to the live runtime. |
| 251 | if err := service.Close(ctx, ref); err != nil { |
| 252 | restore() |
| 253 | if errors.Is(err, session.ErrRuntimeBusy) { |
| 254 | return mirroredSession{}, errHandoffBusyAgain |
| 255 | } |
| 256 | return mirroredSession{}, fmt.Errorf("handoff: release session writer: %w", err) |
| 257 | } |
| 258 | if taken != nil { |
| 259 | if taken.keeper != nil { |
| 260 | taken.keeper.Release() |
| 261 | } |
| 262 | holder.Close() |
| 263 | s.forgetSessionTag(holder) |
| 264 | } else { |
| 265 | // Re-point the frame tag: a stale tag stamps live frames with the |
| 266 | // handed-off identity and the desktop pump drops them as background. |
| 267 | s.setControllerPath(holder, "") |
| 268 | } |
| 269 | s.markMirrored(m) |
| 270 | slog.Info("serve: final-format session handed off to local runtime", "session", route) |
| 271 | s.bc.Emit(event.Event{ |
| 272 | Kind: event.Notice, |
| 273 | Level: event.LevelWarn, |
| 274 | Code: event.NoticeCodeSessionTakenOver, |
| 275 | Text: "This session was taken over by a local Reasonix window and is read-only here.", |
| 276 | Detail: "A Reasonix window on this machine took over the conversation. It keeps streaming here; use \"take back\" to reclaim it.", |
| 277 | SessionPath: route, |
| 278 | }) |
| 279 | return m, nil |
| 280 | } |
| 281 | |
| 282 | // reattachAfterFailedHandoff restores the foreground's binding to an identity |
| 283 | // whose release did not complete. The runtime is still the service's live |
| 284 | // instance (a refused close never retires it), so OpenSession re-binds and |
| 285 | // re-projects it; only when even that fails is the controller left unbound, |
| 286 | // which the next turn resolves by allocating lazily. |
| 287 | func (s *Server) reattachAfterFailedHandoff(ctx context.Context, concrete *control.Controller, ref session.SessionRef) { |
| 288 | if cur, bound := concrete.SessionRef(); bound && cur == ref { |
| 289 | return |
| 290 | } |
| 291 | if _, err := concrete.OpenSession(ctx, ref); err != nil { |
| 292 | slog.Error("serve: re-attach identity after failed handoff", "session", ref.SessionID, "err", err) |
| 293 | } |
| 294 | } |
| 295 | |
| 296 | func parseHandoffMode(raw string) handoffMode { |
| 297 | if handoffMode(raw) == handoffModeInterrupt { |
| 298 | return handoffModeInterrupt |
| 299 | } |
| 300 | return handoffModeWait |
| 301 | } |
| 302 | |
| 303 | func handoffTimeout(ms int) time.Duration { |
| 304 | if ms <= 0 { |
| 305 | return handoffDefaultTimeout |
| 306 | } |
| 307 | return time.Duration(ms) * time.Millisecond |
| 308 | } |
| 309 | |
| 310 | func statusForHandoffError(err error) int { |
| 311 | switch { |
| 312 | case errors.Is(err, errSessionNotHeld), errors.Is(err, errHandoffBusyAgain): |
| 313 | return http.StatusConflict |
| 314 | default: |
| 315 | return http.StatusInternalServerError |
| 316 | } |
| 317 | } |
| 318 | |
| 319 | var ( |
| 320 | errSessionNotHeld = errors.New("session is not held by this serve process") |
| 321 | errHandoffBusyAgain = errors.New("session became busy again during handoff; retry") |
| 322 | ) |
| 323 | |
| 324 | // quietSessionForHandoff waits for (or cancels toward) an idle session before |
| 325 | // the binding transaction runs, covering both the foreground and a detached |
| 326 | // holder of the path. |
| 327 | func (s *Server) quietSessionForHandoff(realPath string, mode handoffMode, timeout time.Duration) error { |
| 328 | return quietHandoffLoop(func() (control.SessionAPI, bool, bool) { |
| 329 | cur := s.ctl() |
| 330 | if cur != nil && agent.CanonicalSessionPath(cur.SessionPath()) == agent.CanonicalSessionPath(realPath) { |
| 331 | return cur, true, controllerHasActiveRuntimeWork(cur) |
| 332 | } |
| 333 | if !s.detachedBusy(realPath) { |
| 334 | return nil, false, false |
| 335 | } |
| 336 | s.detachedMu.Lock() |
| 337 | d := s.detached[agent.CanonicalSessionPath(realPath)] |
| 338 | ctrl := control.SessionAPI(nil) |
| 339 | if d != nil { |
| 340 | ctrl = d.ctrl |
| 341 | } |
| 342 | s.detachedMu.Unlock() |
| 343 | return ctrl, ctrl != nil, ctrl != nil && controllerHasActiveRuntimeWork(ctrl) |
| 344 | }, mode, timeout) |
| 345 | } |
| 346 | |
| 347 | // handoffLocked performs the release transaction. Callers hold bindMu and |
| 348 | // have already quieted the session. |
| 349 | func (s *Server) handoffLocked(realPath, targetWriterID string) (mirroredSession, error) { |
| 350 | cur := s.ctl() |
| 351 | canonical := agent.CanonicalSessionPath(realPath) |
| 352 | info, err := agent.LoadSessionLeaseInfo(realPath) |
| 353 | if err != nil || info == nil || strings.TrimSpace(info.WriterID) == "" { |
| 354 | return mirroredSession{}, fmt.Errorf("handoff: current lease identity unavailable") |
| 355 | } |
| 356 | m, err := newMirroredSession(canonical, info.WriterID, targetWriterID, mirrorPhasePending) |
| 357 | if err != nil { |
| 358 | return mirroredSession{}, fmt.Errorf("handoff: create generation: %w", err) |
| 359 | } |
| 360 | switch { |
| 361 | case cur != nil && agent.CanonicalSessionPath(cur.SessionPath()) == canonical: |
| 362 | if controllerHasActiveRuntimeWork(cur) { |
| 363 | return mirroredSession{}, errHandoffBusyAgain |
| 364 | } |
| 365 | // Flush the transcript while this process still owns the file, then |
| 366 | // hand the lease over. Rebind("") also unbinds write authority, so a |
| 367 | // later save fails closed instead of racing the new writer. |
| 368 | if err := cur.Snapshot(); err != nil { |
| 369 | return mirroredSession{}, fmt.Errorf("handoff: snapshot session: %w", err) |
| 370 | } |
| 371 | if s.leases == nil { |
| 372 | return mirroredSession{}, fmt.Errorf("handoff: lease keeper unavailable") |
| 373 | } |
| 374 | if err := s.leases.ReleaseForHandoff(targetWriterID, m.handoffID); err != nil { |
| 375 | return mirroredSession{}, fmt.Errorf("handoff: release session lease: %w", err) |
| 376 | } |
| 377 | case s.detachedBusy(realPath): |
| 378 | detached := s.takeDetached(realPath) |
| 379 | if detached == nil { |
| 380 | return mirroredSession{}, errHandoffBusyAgain |
| 381 | } |
| 382 | if controllerHasActiveRuntimeWork(detached.ctrl) { |
| 383 | _, _ = s.registerDetached(detached.ctrl, detached.keeper, detached.tag) |
| 384 | return mirroredSession{}, errHandoffBusyAgain |
| 385 | } |
| 386 | if err := detached.ctrl.Snapshot(); err != nil { |
| 387 | _, _ = s.registerDetached(detached.ctrl, detached.keeper, detached.tag) |
| 388 | return mirroredSession{}, fmt.Errorf("handoff: snapshot detached session: %w", err) |
| 389 | } |
| 390 | if detached.keeper == nil { |
| 391 | _, _ = s.registerDetached(detached.ctrl, detached.keeper, detached.tag) |
| 392 | return mirroredSession{}, fmt.Errorf("handoff: detached lease keeper unavailable") |
| 393 | } |
| 394 | if err := detached.keeper.ReleaseForHandoff(targetWriterID, m.handoffID); err != nil { |
| 395 | _, _ = s.registerDetached(detached.ctrl, detached.keeper, detached.tag) |
| 396 | return mirroredSession{}, fmt.Errorf("handoff: release detached session lease: %w", err) |
| 397 | } |
| 398 | detached.ctrl.Close() |
| 399 | if concrete, ok := detached.ctrl.(*control.Controller); ok { |
| 400 | s.forgetSessionTag(concrete) |
| 401 | } |
| 402 | default: |
| 403 | return mirroredSession{}, errSessionNotHeld |
| 404 | } |
| 405 | s.markMirrored(m) |
| 406 | slog.Info("serve: session handed off to local runtime", "session", canonical) |
| 407 | s.bc.Emit(event.Event{ |
| 408 | Kind: event.Notice, |
| 409 | Level: event.LevelWarn, |
| 410 | Code: event.NoticeCodeSessionTakenOver, |
| 411 | Text: "This session was taken over by a local Reasonix window and is read-only here.", |
| 412 | Detail: "A Reasonix window on this machine took over the conversation. It keeps streaming here; use \"take back\" to reclaim it.", |
| 413 | SessionPath: canonical, |
| 414 | }) |
| 415 | return m, nil |
| 416 | } |
| 417 |