| 1 | package serve |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "encoding/json" |
| 7 | "fmt" |
| 8 | "io" |
| 9 | "net/http" |
| 10 | "net/url" |
| 11 | "strconv" |
| 12 | "strings" |
| 13 | "time" |
| 14 | |
| 15 | "reasonix/internal/config" |
| 16 | "reasonix/internal/control" |
| 17 | ) |
| 18 | |
| 19 | var modelSettingsSourceClient = &http.Client{Transport: &http.Transport{Proxy: nil}, Timeout: 20 * time.Second} |
| 20 | |
| 21 | func requestModelSettingsSource(ctx context.Context, settings *config.ModelRuntimeSettings, request config.ModelSettingsSourceRequest) (config.ModelSettingsSourceResponse, error) { |
| 22 | var response config.ModelSettingsSourceResponse |
| 23 | endpoint, err := url.Parse(settings.ProxyURL) |
| 24 | if err != nil || endpoint.Scheme != "http" || (endpoint.Hostname() != "127.0.0.1" && endpoint.Hostname() != "::1") || endpoint.User != nil { |
| 25 | return response, fmt.Errorf("model settings source requires the local credential tunnel") |
| 26 | } |
| 27 | endpoint.Path, endpoint.RawQuery, endpoint.Fragment = "/model-settings-source", "", "" |
| 28 | body, err := json.Marshal(request) |
| 29 | if err != nil { |
| 30 | return response, err |
| 31 | } |
| 32 | req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint.String(), bytes.NewReader(body)) |
| 33 | if err != nil { |
| 34 | return response, err |
| 35 | } |
| 36 | req.Header.Set("Authorization", "Bearer "+settings.SourceToken) |
| 37 | req.Header.Set("Content-Type", "application/json") |
| 38 | res, err := modelSettingsSourceClient.Do(req) |
| 39 | if err != nil { |
| 40 | return response, fmt.Errorf("Desktop model settings source is unavailable") |
| 41 | } |
| 42 | defer res.Body.Close() |
| 43 | if res.StatusCode != http.StatusOK { |
| 44 | return response, fmt.Errorf("Desktop cannot apply saved model settings (status %d); check its available models and connection", res.StatusCode) |
| 45 | } |
| 46 | if err := json.NewDecoder(io.LimitReader(res.Body, 4<<20)).Decode(&response); err != nil || response.Version != 1 || response.Revision == "" { |
| 47 | return response, fmt.Errorf("invalid Desktop model settings acknowledgement") |
| 48 | } |
| 49 | return response, nil |
| 50 | } |
| 51 | |
| 52 | func sourceModelRef(settings *config.ModelRuntimeSettings, remoteRef string) string { |
| 53 | for source, target := range settings.References { |
| 54 | if target == remoteRef && strings.Contains(source, "/") { |
| 55 | return source |
| 56 | } |
| 57 | } |
| 58 | return remoteRef |
| 59 | } |
| 60 | |
| 61 | // New HTTP turns and autonomous FIFO dispatch share this boundary. A source |
| 62 | // failure leaves the existing controller and queued work intact. Approvals, |
| 63 | // steers, and children remain on their already accepted runtime. |
| 64 | func (s *Server) refreshRunModelSettingsLocked(ctx context.Context) error { |
| 65 | return s.refreshModelSettingsOwnerLocked(ctx, modelSettingsRuntimeOwner{ |
| 66 | current: s.ctl, settings: &s.managedModels, offerID: &s.modelSettingsOfferID, apply: s.switchModelLocked, |
| 67 | }) |
| 68 | } |
| 69 | |
| 70 | type modelSettingsRuntimeOwner struct { |
| 71 | current func() control.SessionAPI |
| 72 | settings **config.ModelRuntimeSettings |
| 73 | offerID *string |
| 74 | apply func(context.Context, string) error |
| 75 | } |
| 76 | |
| 77 | func (s *Server) refreshModelSettingsOwnerLocked(ctx context.Context, owner modelSettingsRuntimeOwner) error { |
| 78 | for { |
| 79 | if err := ctx.Err(); err != nil { |
| 80 | return err |
| 81 | } |
| 82 | current := owner.current() |
| 83 | settings := (*owner.settings) |
| 84 | var offered *config.ModelRuntimeSettings |
| 85 | ref := current.ModelRef() |
| 86 | var sourceRequest config.ModelSettingsSourceRequest |
| 87 | if settings != nil && settings.SourceToken != "" { |
| 88 | offerID, err := config.NewModelSettingsOfferID() |
| 89 | if err != nil { |
| 90 | return err |
| 91 | } |
| 92 | endpoint, err := url.Parse(settings.ProxyURL) |
| 93 | if err != nil { |
| 94 | return fmt.Errorf("invalid model settings tunnel") |
| 95 | } |
| 96 | port, _ := strconv.Atoi(endpoint.Port()) |
| 97 | status := s.modelSettingsStatusLocked() |
| 98 | sourceRequest = config.ModelSettingsSourceRequest{Mode: "prepare", OfferID: offerID, PreviousOfferID: (*owner.offerID), Model: sourceModelRef(settings, ref), AppliedRevision: settings.Revision, RemotePort: port, OwnedRevisions: status.OwnedRevisions, UnversionedOwners: status.UnversionedOwners} |
| 99 | (*owner.offerID) = offerID |
| 100 | sourceRequest.ModelSettingsOwnership = status.ModelSettingsOwnership |
| 101 | response, err := requestModelSettingsSource(ctx, settings, sourceRequest) |
| 102 | if err != nil { |
| 103 | return err |
| 104 | } |
| 105 | if response.Settings != nil { |
| 106 | if err := validateModelSettingsSourceOffer(offerID, response); err != nil { |
| 107 | return err |
| 108 | } |
| 109 | offered, ref = response.Settings, response.Ref |
| 110 | } else if response.Revision != settings.Revision { |
| 111 | return fmt.Errorf("changed model settings require a complete resolver") |
| 112 | } |
| 113 | } |
| 114 | needsApply := offered != nil && (offered.Revision != settings.Revision || ref != current.ModelRef()) |
| 115 | if snapshot, ok := current.(interface { |
| 116 | ModelSettingsState() (string, string, error) |
| 117 | }); ok { |
| 118 | applied, desired, err := snapshot.ModelSettingsState() |
| 119 | if err != nil { |
| 120 | return err |
| 121 | } |
| 122 | needsApply = needsApply || applied != desired |
| 123 | } |
| 124 | var applyErr error |
| 125 | if needsApply { |
| 126 | if offered != nil { |
| 127 | (*owner.settings) = offered |
| 128 | } |
| 129 | applyErr = owner.apply(ctx, ref) |
| 130 | if applyErr != nil { |
| 131 | (*owner.settings) = settings |
| 132 | } |
| 133 | } |
| 134 | if settings != nil && settings.SourceToken != "" { |
| 135 | ackSettings := settings |
| 136 | if offered != nil { |
| 137 | ackSettings = offered |
| 138 | } |
| 139 | status := s.modelSettingsStatusLocked() |
| 140 | sourceRequest.Mode = "finish" |
| 141 | sourceRequest.ModelSettingsOwnership = status.ModelSettingsOwnership |
| 142 | sourceRequest.PreviousOfferID = "" |
| 143 | sourceRequest.OwnedRevisions, sourceRequest.UnversionedOwners = status.OwnedRevisions, status.UnversionedOwners |
| 144 | ack, ackErr := requestModelSettingsSource(ctx, ackSettings, sourceRequest) |
| 145 | if ackErr == nil { |
| 146 | (*owner.offerID) = "" |
| 147 | } |
| 148 | if applyErr != nil { |
| 149 | return fmt.Errorf("saved model settings could not be applied: %w", applyErr) |
| 150 | } |
| 151 | if ackErr != nil { |
| 152 | return ackErr |
| 153 | } |
| 154 | if ack.Revision != (*owner.settings).Revision { |
| 155 | continue // a save overtook the candidate; no new run was admitted |
| 156 | } |
| 157 | } |
| 158 | if applyErr != nil { |
| 159 | return applyErr |
| 160 | } |
| 161 | if changed, err := runtimeModelSettingsChanged(owner.current()); err != nil { |
| 162 | return err |
| 163 | } else if changed { |
| 164 | continue |
| 165 | } |
| 166 | return nil |
| 167 | } |
| 168 | } |
| 169 | |
| 170 | func validateModelSettingsSourceOffer(id string, response config.ModelSettingsSourceResponse) error { |
| 171 | if response.Settings.OfferID != id || response.Settings.Revision != response.Revision || response.Ref == "" || response.Settings.SourceToken == "" { |
| 172 | return fmt.Errorf("model settings offer does not match its request") |
| 173 | } |
| 174 | return nil |
| 175 | } |
| 176 | |
| 177 | func runtimeModelSettingsChanged(ctrl control.SessionAPI) (bool, error) { |
| 178 | snapshot, ok := ctrl.(interface { |
| 179 | ModelSettingsState() (string, string, error) |
| 180 | }) |
| 181 | if !ok { |
| 182 | return false, nil |
| 183 | } |
| 184 | applied, desired, err := snapshot.ModelSettingsState() |
| 185 | return applied != desired, err |
| 186 | } |
| 187 | |
| 188 | func (s *Server) beforeInboxDispatch(ctrl *control.Controller) (func(), error) { |
| 189 | if s.ctl() != ctrl { |
| 190 | return s.beforeDetachedInboxDispatch(ctrl) |
| 191 | } |
| 192 | s.bindMu.Lock() |
| 193 | if s.ctl() != ctrl { |
| 194 | s.bindMu.Unlock() |
| 195 | return nil, control.ErrInboxRuntimeUnpublished |
| 196 | } |
| 197 | if ctrl.Running() { |
| 198 | s.bindMu.Unlock() |
| 199 | return nil, control.ErrTurnRunning |
| 200 | } |
| 201 | ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) |
| 202 | err := s.refreshRunModelSettingsLocked(ctx) |
| 203 | cancel() |
| 204 | if err != nil { |
| 205 | s.bindMu.Unlock() |
| 206 | return nil, err |
| 207 | } |
| 208 | if current := s.ctl(); current != ctrl { |
| 209 | s.bindMu.Unlock() |
| 210 | if replacement, ok := current.(*control.Controller); ok { |
| 211 | replacement.NotifyInboxRuntimeReady() |
| 212 | } |
| 213 | return nil, control.ErrInboxRuntimeUnpublished |
| 214 | } |
| 215 | return s.bindMu.Unlock, nil |
| 216 | } |
| 217 |