| 1 | package main |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "io" |
| 7 | "log/slog" |
| 8 | "net/http" |
| 9 | "strings" |
| 10 | "time" |
| 11 | |
| 12 | "reasonix/internal/agent" |
| 13 | "reasonix/internal/config" |
| 14 | ) |
| 15 | |
| 16 | // takeoverAfterAdoptGrantHookForTest pauses a direct local-session adoption |
| 17 | // after Serve rotates the mirror generation but before Desktop publishes it. |
| 18 | // Production leaves it nil. |
| 19 | var takeoverAfterAdoptGrantHookForTest func() |
| 20 | |
| 21 | var discoverLocalTakeoverServesForAdopt = discoverLocalTakeoverServes |
| 22 | |
| 23 | func (a *App) adoptSessionFromLocalServe(tabID, sessionPath string) { |
| 24 | if a.adoptSessionFromLocalServeOnce(tabID, sessionPath) { |
| 25 | return |
| 26 | } |
| 27 | // The announce can race a restart, token rotation, or transient discovery. |
| 28 | // Retry once after backoff while the tab still shows the session; otherwise |
| 29 | // the remote side can see the foreign lease but cannot reclaim it cleanly. |
| 30 | time.AfterFunc(serveProbeBackoffWindow, func() { |
| 31 | if tab := a.tabByID(tabID); tab != nil && !tab.ReadOnly && |
| 32 | strings.TrimSpace(tab.currentSessionPath()) == strings.TrimSpace(sessionPath) { |
| 33 | a.adoptSessionFromLocalServeOnce(tabID, sessionPath) |
| 34 | } |
| 35 | }) |
| 36 | } |
| 37 | |
| 38 | type takeoverAdoptFence struct { |
| 39 | tab *WorkspaceTab |
| 40 | sink *tabEventSink |
| 41 | epoch string |
| 42 | revision uint64 |
| 43 | } |
| 44 | |
| 45 | func (a *App) beginTakeoverAdopt(tabID, sessionPath, key string) (takeoverAdoptFence, bool) { |
| 46 | a.mu.RLock() |
| 47 | tab := a.tabByIDLocked(tabID) |
| 48 | fence := takeoverAdoptFence{tab: tab} |
| 49 | valid := tab != nil && !tab.ReadOnly && sessionRuntimeKey(tab.currentSessionPath()) == key && tab.sink != nil |
| 50 | if valid { |
| 51 | fence.sink = tab.sink |
| 52 | fence.epoch = a.runtimeEpochForTabLocked(tab) |
| 53 | } |
| 54 | a.mu.RUnlock() |
| 55 | if !valid { |
| 56 | return takeoverAdoptFence{}, false |
| 57 | } |
| 58 | a.takeoverMu.Lock() |
| 59 | defer a.takeoverMu.Unlock() |
| 60 | if a.takeoverMirrors[key] != nil { |
| 61 | return takeoverAdoptFence{}, false |
| 62 | } |
| 63 | if a.takeoverAdoptRevisions == nil { |
| 64 | a.takeoverAdoptRevisions = map[string]uint64{} |
| 65 | } |
| 66 | fence.revision = a.takeoverAdoptRevisions[key] + 1 |
| 67 | a.takeoverAdoptRevisions[key] = fence.revision |
| 68 | return fence, true |
| 69 | } |
| 70 | |
| 71 | func newTakeoverMirror(app *App, key, tabID, sessionPath string, sink *tabEventSink, record takeoverServeRecord, client *http.Client, grant takeoverGrant) *takeoverMirror { |
| 72 | return &takeoverMirror{ |
| 73 | app: app, key: key, tabID: tabID, sessionPath: sessionPath, sink: sink, |
| 74 | record: record, client: client, grant: grant, bindingRevision: 1, |
| 75 | stop: make(chan struct{}), done: make(chan struct{}), wake: make(chan struct{}, 1), |
| 76 | } |
| 77 | } |
| 78 | |
| 79 | // commitTakeoverAdopt publishes an adoption only while the initiating tab, |
| 80 | // runtime epoch, session path, sink, and adoption revision are all current. |
| 81 | // runtimeRebuildMu closes the final validation-to-attach gap against session |
| 82 | // switches; an older overlapping /adopt response loses to the latest revision. |
| 83 | func (a *App) commitTakeoverAdopt(fence takeoverAdoptFence, key, tabID, sessionPath string, record takeoverServeRecord, client *http.Client, grant takeoverGrant) bool { |
| 84 | a.runtimeRebuildMu.Lock() |
| 85 | defer a.runtimeRebuildMu.Unlock() |
| 86 | a.mu.RLock() |
| 87 | tab := a.tabByIDLocked(tabID) |
| 88 | valid := tab != nil && tab == fence.tab && !tab.ReadOnly && tab.sink == fence.sink && |
| 89 | a.runtimeEpochForTabLocked(tab) == fence.epoch && sessionRuntimeKey(tab.currentSessionPath()) == key |
| 90 | a.mu.RUnlock() |
| 91 | if !valid { |
| 92 | return false |
| 93 | } |
| 94 | m := newTakeoverMirror(a, key, tabID, sessionPath, fence.sink, record, client, grant) |
| 95 | a.takeoverMu.Lock() |
| 96 | currentRevision := a.takeoverAdoptRevisions[key] |
| 97 | if currentRevision != fence.revision || a.takeoverMirrors[key] != nil { |
| 98 | a.takeoverMu.Unlock() |
| 99 | return false |
| 100 | } |
| 101 | if a.takeoverMirrors == nil { |
| 102 | a.takeoverMirrors = map[string]*takeoverMirror{} |
| 103 | } |
| 104 | a.takeoverMirrors[key] = m |
| 105 | delete(a.takeoverAdoptRevisions, key) |
| 106 | a.takeoverMu.Unlock() |
| 107 | fence.sink.setTakeoverMirror(m) |
| 108 | go m.run(client, record) |
| 109 | return true |
| 110 | } |
| 111 | |
| 112 | // adoptSessionFromLocalServeOnce announces a directly-opened local session to |
| 113 | // a resident serve. It reports false when no serve could be told, so the |
| 114 | // caller can retry. |
| 115 | func (a *App) adoptSessionFromLocalServeOnce(tabID, sessionPath string) bool { |
| 116 | key := sessionRuntimeKey(sessionPath) |
| 117 | if key == "" { |
| 118 | return true |
| 119 | } |
| 120 | fence, attempt := a.beginTakeoverAdopt(tabID, sessionPath, key) |
| 121 | if !attempt { |
| 122 | return true |
| 123 | } |
| 124 | defer func() { |
| 125 | a.takeoverMu.Lock() |
| 126 | if a.takeoverAdoptRevisions[key] == fence.revision && a.takeoverMirrors[key] == nil { |
| 127 | delete(a.takeoverAdoptRevisions, key) |
| 128 | } |
| 129 | a.takeoverMu.Unlock() |
| 130 | }() |
| 131 | ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) |
| 132 | defer cancel() |
| 133 | for _, serve := range discoverLocalTakeoverServesForAdopt() { |
| 134 | workspaceDir := config.ProjectSessionDir(serve.state.Workspace) |
| 135 | if workspaceDir == "" || !pathWithinDir(sessionPath, workspaceDir) { |
| 136 | continue |
| 137 | } |
| 138 | client, err := takeoverClient(ctx, serve) |
| 139 | if err != nil { |
| 140 | continue |
| 141 | } |
| 142 | view, err := takeoverOwnership(ctx, client, serve.base, sessionPath) |
| 143 | if err != nil { |
| 144 | continue |
| 145 | } |
| 146 | if view.Holder == "serve" || view.Holder == "external" { |
| 147 | return true |
| 148 | } |
| 149 | body, err := json.Marshal(map[string]string{"sessionPath": sessionPath, "writerId": agent.SessionWriterID()}) |
| 150 | if err != nil { |
| 151 | continue |
| 152 | } |
| 153 | resp, err := serveDo(ctx, client, http.MethodPost, serveURL(serve.base, "/adopt"), body) |
| 154 | if err != nil { |
| 155 | continue |
| 156 | } |
| 157 | respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<16)) |
| 158 | resp.Body.Close() |
| 159 | if resp.StatusCode != http.StatusOK { |
| 160 | continue |
| 161 | } |
| 162 | var grant takeoverGrant |
| 163 | if json.Unmarshal(respBody, &grant) != nil || grant.MirrorID == "" || grant.ReturnHandoffID == "" || grant.SourceWriterID == "" || |
| 164 | grant.TargetWriterID != agent.SessionWriterID() || sessionRuntimeKey(grant.SessionPath) != key { |
| 165 | continue |
| 166 | } |
| 167 | if hook := takeoverAfterAdoptGrantHookForTest; hook != nil { |
| 168 | hook() |
| 169 | } |
| 170 | if !a.commitTakeoverAdopt(fence, key, tabID, sessionPath, serve, client, grant) { |
| 171 | a.endFailedTakeover(serve, client, grant) |
| 172 | return true |
| 173 | } |
| 174 | slog.Info("desktop: local session adopted by serve for remote spectating", |
| 175 | "tab", tabID, "session", sessionPath, "serve", serve.base) |
| 176 | return true |
| 177 | } |
| 178 | return false |
| 179 | } |
| 180 |