返回 DeepSeek-Reasonix
bootstrap.go
根目录 / internal / remote / bootstrap / bootstrap.go
1 // Package bootstrap starts and manages a detached `reasonix serve` process on
2 // a remote host over an established SSH connection. It detects the remote
3 // OS/arch, locates or installs reasonix, launches serve bound to a random
4 // loopback port with a file-based token (never in argv), and records the
5 // result under the remote ~/.reasonix/remote so a later reconnect can reuse
6 // it. V1 targets Linux and macOS remotes.
7 package bootstrap
8
9 import (
10 "context"
11 "crypto/rand"
12 "encoding/hex"
13 "errors"
14 "fmt"
15 "io"
16 "net"
17 "strconv"
18 "strings"
19 "time"
20
21 "reasonix/internal/remote"
22 "reasonix/internal/remote/sftpfs"
23 )
24
25 // Conn is the subset of *remote.Client bootstrap needs. *remote.Client
26 // satisfies it directly; tests inject a fake. bootstrap depends on remote
27 // (never the reverse), so using remote.ExecResult here introduces no cycle.
28 type Conn interface {
29 Exec(ctx context.Context, cmd string) (remote.ExecResult, error)
30 SFTP() (*sftpfs.FS, error)
31 }
32
33 // Install strategies.
34 const (
35 InstallAuto = "auto"
36 InstallNPM = "npm"
37 InstallUpload = "upload"
38 InstallNever = "never"
39 )
40
41 // MinServeVersion is retained for display/informational use only. Usability is
42 // decided by probing `serve --help` for the --port-file flag (see locate), not
43 // by a version number: --port-file/--token-file ship in this change, so no
44 // released version satisfies a numeric gate, and the release number this change
45 // lands in is unknown at authoring time.
46 const MinServeVersion = "flag:port-file"
47
48 // Options configures EnsureServe.
49 type Options struct {
50 Workspace string // remote workspace path (may start with ~)
51 Install string // auto|npm|upload|never
52 LocalBinary string // path to the running reasonix binary, for same-platform upload
53 LocalGOOS string // GOOS of LocalBinary
54 LocalGOARCH string // GOARCH of LocalBinary
55 ProductVersion string // exact local release used for a cross-platform official download
56 FetchBinary func(context.Context, string, string, string) ([]byte, error) // local verified release fetcher
57 MinVersion string // minimum acceptable remote version
58 Progress func(step, detail string) // optional progress callback
59 Clock func() time.Time // nil => time.Now
60 // CredentialProxy installs a tunnel-backed provider and a scoped virtual
61 // token on the remote. The real provider key never leaves the desktop.
62 CredentialProxy *CredentialProxyOptions
63 // BrowserBroker is consulted right before a fresh launch, only when the
64 // located binary advertises ServeBrowserBrokerMarker; it opens the reverse
65 // forward and returns what the serve's environment must carry.
66 BrowserBroker func(context.Context) (*BrowserBrokerOptions, error)
67 }
68
69 func (o Options) progress(step, detail string) {
70 if o.Progress != nil {
71 o.Progress(step, detail)
72 }
73 }
74
75 func (o Options) clock() func() time.Time {
76 if o.Clock != nil {
77 return o.Clock
78 }
79 return time.Now
80 }
81
82 // Result is the outcome of EnsureServe.
83 type Result struct {
84 State ServeState
85 Token string // the pre-shared auth token (read from or written to TokenFile)
86 Reused bool // true when an already-running serve was reused
87 // CredentialConfigChanged is true when the credential-proxy heal rewrote
88 // the remote config while REUSING a serve — that serve's in-memory
89 // providers still reflect the previous config and must be reloaded.
90 CredentialConfigChanged bool
91 }
92
93 // EnsureServe returns a running serve for (host, workspace), starting one if
94 // needed. It is also the reconnect path: an existing live process is reused.
95 func EnsureServe(ctx context.Context, conn Conn, opts Options) (Result, error) {
96 fs, err := conn.SFTP()
97 if err != nil {
98 return Result{}, err
99 }
100 home, err := fs.RealPath(ctx, "~")
101 if err != nil {
102 return Result{}, fmt.Errorf("bootstrap: resolve remote home: %w", err)
103 }
104 workspace, err := resolveWorkspace(ctx, fs, opts.Workspace, home)
105 if err != nil {
106 return Result{}, err
107 }
108 paths := pathsFor(home, workspace)
109
110 requireLaunchArgs, credentialChanged, err := prepareCredentialProxy(ctx, conn, fs, opts, home, workspace, paths)
111 if err != nil {
112 return Result{}, err
113 }
114 // 1. Reuse a live process if the recorded pid is still running and exposes
115 // every Serve contract required by this desktop.
116 if st, tok, ok := tryReuse(ctx, conn, fs, paths, workspace, requireLaunchArgs...); ok {
117 opts.progress("reuse", st.Addr)
118 return Result{State: st, Token: tok, Reused: true, CredentialConfigChanged: credentialChanged}, nil
119 }
120
121 // 2. Detect remote platform.
122 opts.progress("detect", "")
123 unameRes, err := conn.Exec(ctx, "uname -sm")
124 if err != nil {
125 return Result{}, fmt.Errorf("bootstrap: uname: %w", err)
126 }
127 goos, goarch, err := ParseUname(string(unameRes.Stdout))
128 if err != nil {
129 return Result{}, err
130 }
131
132 // 3. Locate or install a usable reasonix.
133 bin, version, err := ensureBinary(ctx, conn, fs, opts, home, goos, goarch, paths)
134 if err != nil {
135 return Result{}, err
136 }
137
138 // 4. Serialize only the short launch/publish section across every client.
139 // Another caller may have completed while this one was locating/installing,
140 // so re-check state after acquiring the remote lock.
141 opts.progress("waiting_lock", "")
142 lock, err := acquireServeLock(ctx, fs, paths, opts.clock())
143 if err != nil {
144 return Result{}, err
145 }
146 defer lock.release()
147 if st, tok, ok := tryReuse(ctx, conn, fs, paths, workspace, requireLaunchArgs...); ok {
148 opts.progress("reuse", st.Addr)
149 return Result{State: st, Token: tok, Reused: true, CredentialConfigChanged: credentialChanged}, nil
150 }
151
152 // 5. Stage the replacement token, retire incompatible Serve, then publish.
153 freshToken, err := generateToken()
154 if err != nil {
155 return Result{}, err
156 }
157 stagedTokenFile, err := stageServeToken(ctx, fs, paths, freshToken)
158 if err != nil {
159 return Result{}, err
160 }
161 defer cleanupStagedServeToken(fs, stagedTokenFile)
162 // Retire an incompatible Serve only after its replacement is ready to launch
163 // inside the lock, so preparation failures do not interrupt existing work.
164 if err := retireIncompatibleServe(ctx, conn, fs, paths, workspace, requireLaunchArgs); err != nil {
165 return Result{}, err
166 }
167 if err := fs.Rename(ctx, stagedTokenFile, paths.TokenFile); err != nil {
168 return Result{}, fmt.Errorf("bootstrap: publish token: %w", err)
169 }
170 opts.progress("launch", "")
171 launchRes, err := conn.Exec(ctx, LaunchCommand(bin, workspace, paths, opts.CredentialProxy, resolveBrowserBroker(ctx, conn, bin, opts)))
172 if err != nil {
173 cleanupFailedLaunch(conn, fs, paths, 0)
174 return Result{}, fmt.Errorf("bootstrap: launch: %w", err)
175 }
176 pid, _ := strconv.Atoi(strings.TrimSpace(string(launchRes.Stdout)))
177
178 // 6. Poll the newly-created port file for the real bound address. The launch
179 // command removes stale port/pid files before forking.
180 opts.progress("health_check", "")
181 addr, err := pollPortFile(ctx, fs, paths.PortFile, opts.clock())
182 if err != nil {
183 cleanupFailedLaunch(conn, fs, paths, pid)
184 return Result{}, err
185 }
186 if filePID, perr := readPIDFile(ctx, fs, paths.PidFile); perr == nil {
187 pid = filePID // --pid-file is authoritative when available.
188 }
189 if pid <= 0 || !pidIsServe(ctx, conn, pid, paths) {
190 cleanupFailedLaunch(conn, fs, paths, pid)
191 return Result{}, errors.New("bootstrap: launched process did not become the expected reasonix serve")
192 }
193
194 st := ServeState{
195 PID: pid,
196 Addr: addr,
197 Workspace: workspace,
198 Version: version,
199 ServeCaps: ServeCapsToken,
200 TokenFile: paths.TokenFile,
201 LogFile: paths.LogFile,
202 StartedAt: nowUnix(opts.clock()),
203 }
204 data, err := MarshalState(st)
205 if err != nil {
206 cleanupFailedLaunch(conn, fs, paths, pid)
207 return Result{}, err
208 }
209 if err := fs.WriteFileAtomic(ctx, paths.StateJSON, data, 0o600); err != nil {
210 cleanupFailedLaunch(conn, fs, paths, pid)
211 return Result{}, fmt.Errorf("bootstrap: write state: %w", err)
212 }
213 opts.progress("ready", addr)
214 return Result{State: st, Token: freshToken}, nil
215 }
216
217 func stageServeToken(ctx context.Context, fs *sftpfs.FS, paths StatePaths, token string) (string, error) {
218 if err := fs.MkdirAll(ctx, paths.Dir); err != nil {
219 return "", err
220 }
221 staged := paths.TokenFile + ".next"
222 if err := fs.WriteFileAtomic(ctx, staged, []byte(token+"\n"), 0o600); err != nil {
223 cleanupStagedServeToken(fs, staged)
224 return "", fmt.Errorf("bootstrap: stage token: %w", err)
225 }
226 return staged, nil
227 }
228
229 func cleanupStagedServeToken(fs *sftpfs.FS, staged string) {
230 ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
231 defer cancel()
232 _ = fs.Remove(ctx, staged, false)
233 }
234
235 func prepareCredentialProxy(ctx context.Context, conn Conn, fs *sftpfs.FS, opts Options, home, workspace string, paths StatePaths) ([]string, bool, error) {
236 if opts.CredentialProxy == nil {
237 return nil, false, nil
238 }
239 opts.progress("credential_proxy", "")
240 changed, err := ensureCredentialProvider(ctx, fs, home, opts.CredentialProxy)
241 if err != nil {
242 return nil, false, err
243 }
244 required := []string{"--model " + opts.CredentialProxy.Provider}
245 return required, changed, nil
246 }
247
248 // Status reads the recorded state and reports whether the process is alive.
249 func Status(ctx context.Context, conn Conn, workspace string) (ServeState, bool, error) {
250 fs, err := conn.SFTP()
251 if err != nil {
252 return ServeState{}, false, err
253 }
254 home, err := fs.RealPath(ctx, "~")
255 if err != nil {
256 return ServeState{}, false, err
257 }
258 ws, err := resolveWorkspace(ctx, fs, workspace, home)
259 if err != nil {
260 return ServeState{}, false, err
261 }
262 paths := pathsFor(home, ws)
263 st, err := readState(ctx, fs, paths.StateJSON)
264 if err != nil {
265 return ServeState{}, false, nil // no state => not running
266 }
267 alive := st.Workspace == ws && validServeAddr(st.Addr) && pidIsServe(ctx, conn, st.PID, paths)
268 return st, alive, nil
269 }
270
271 // Stop terminates the recorded process and removes its state files.
272 func Stop(ctx context.Context, conn Conn, workspace string) error {
273 fs, err := conn.SFTP()
274 if err != nil {
275 return err
276 }
277 home, err := fs.RealPath(ctx, "~")
278 if err != nil {
279 return err
280 }
281 ws, err := resolveWorkspace(ctx, fs, workspace, home)
282 if err != nil {
283 return err
284 }
285 paths := pathsFor(home, ws)
286 st, err := readState(ctx, fs, paths.StateJSON)
287 if err != nil {
288 return nil // nothing recorded
289 }
290 // Only signal the pid if it is still OUR serve: a recycled PID now owned by
291 // an unrelated process must never be TERM/KILLed.
292 if st.PID > 0 {
293 if _, err := conn.Exec(ctx, StopCommand(st.PID, paths)); err != nil {
294 return fmt.Errorf("bootstrap: stop pid %d: %w", st.PID, err)
295 }
296 }
297 _ = fs.Remove(ctx, paths.StateJSON, false)
298 _ = fs.Remove(ctx, paths.TokenFile, false)
299 _ = fs.Remove(ctx, paths.PortFile, false)
300 _ = fs.Remove(ctx, paths.PidFile, false)
301 return nil
302 }
303
304 // Logs writes up to n tail lines of the serve log to w.
305 func Logs(ctx context.Context, conn Conn, workspace string, n int, w io.Writer) error {
306 fs, err := conn.SFTP()
307 if err != nil {
308 return err
309 }
310 home, err := fs.RealPath(ctx, "~")
311 if err != nil {
312 return err
313 }
314 ws, err := resolveWorkspace(ctx, fs, workspace, home)
315 if err != nil {
316 return err
317 }
318 paths := pathsFor(home, ws)
319 res, err := conn.Exec(ctx, LogsCommand(paths.LogFile, n))
320 if err != nil {
321 return err
322 }
323 _, err = w.Write(res.Stdout)
324 return err
325 }
326
327 func tryReuse(ctx context.Context, conn Conn, fs *sftpfs.FS, paths StatePaths, workspace string, requireArgs ...string) (ServeState, string, bool) {
328 st, err := readState(ctx, fs, paths.StateJSON)
329 if err != nil || st.PID <= 0 || st.Addr == "" {
330 return ServeState{}, "", false
331 }
332 if workspace != "" && st.Workspace != workspace {
333 return ServeState{}, "", false
334 }
335 if !validServeAddr(st.Addr) || !pidIsServe(ctx, conn, st.PID, paths, requireArgs...) {
336 return ServeState{}, "", false
337 }
338 if st.ServeCaps != ServeCapsToken && !supportsRequiredServeCapabilities(ctx, conn, st.PID) {
339 return ServeState{}, "", false
340 }
341 // The state record is informational; the workspace-derived path is the
342 // authority, so a tampered record cannot make us read an arbitrary file.
343 tok, err := readToken(ctx, fs, paths.TokenFile)
344 if err != nil {
345 return ServeState{}, "", false
346 }
347 return st, tok, true
348 }
349
350 // stopMismatchedServe TERMs a live serve whose command line lacks the
351 // required launch args: reuse would route model calls under the wrong
352 // credential setup, and a plain relaunch would orphan the process.
353 func stopMismatchedServe(ctx context.Context, conn Conn, fs *sftpfs.FS, paths StatePaths, workspace string, requireArgs []string) error {
354 if len(requireArgs) == 0 {
355 return nil
356 }
357 st, err := readState(ctx, fs, paths.StateJSON)
358 if err != nil || st.PID <= 0 || !validServeAddr(st.Addr) || st.Workspace != workspace {
359 return nil
360 }
361 if pidIsServe(ctx, conn, st.PID, paths) && !pidIsServe(ctx, conn, st.PID, paths, requireArgs...) {
362 if _, err := conn.Exec(ctx, StopCommand(st.PID, paths)); err != nil {
363 return fmt.Errorf("bootstrap: stop mismatched serve: %w", err)
364 }
365 }
366 return nil
367 }
368
369 func retireIncompatibleServe(ctx context.Context, conn Conn, fs *sftpfs.FS, paths StatePaths, workspace string, requireArgs []string) error {
370 if err := stopMismatchedServe(ctx, conn, fs, paths, workspace, requireArgs); err != nil {
371 return err
372 }
373 return stopOutdatedServe(ctx, conn, fs, paths, workspace)
374 }
375
376 // stopOutdatedServe retires a live process whose binary lacks the wire and
377 // healing contracts required by the desktop. Leaving it alive would retain the
378 // workspace lease and race the replacement process.
379 func stopOutdatedServe(ctx context.Context, conn Conn, fs *sftpfs.FS, paths StatePaths, workspace string) error {
380 st, err := readState(ctx, fs, paths.StateJSON)
381 if err != nil || st.PID <= 0 || !validServeAddr(st.Addr) || st.Workspace != workspace {
382 return nil
383 }
384 if !pidIsServe(ctx, conn, st.PID, paths) {
385 return nil
386 }
387 if st.ServeCaps == ServeCapsToken || supportsRequiredServeCapabilities(ctx, conn, st.PID) {
388 return nil
389 }
390 if _, stopErr := conn.Exec(ctx, StopCommand(st.PID, paths)); stopErr != nil {
391 return fmt.Errorf("bootstrap: stop outdated serve: %w", stopErr)
392 }
393 return nil
394 }
395
396 func supportsRequiredServeCapabilities(ctx context.Context, conn Conn, pid int) bool {
397 res, err := conn.Exec(ctx, SupportsRequiredServeCapabilitiesCommand(pid))
398 return err == nil && strings.TrimSpace(string(res.Stdout)) == "yes"
399 }
400
401 // pidIsServe reports whether pid is running AND is a reasonix serve process,
402 // so PID reuse cannot make an unrelated process look like a live serve.
403 func pidIsServe(ctx context.Context, conn Conn, pid int, paths StatePaths, requireArgs ...string) bool {
404 if pid <= 0 {
405 return false
406 }
407 res, err := conn.Exec(ctx, ServeAliveCommand(pid, paths, requireArgs...))
408 if err != nil {
409 return false
410 }
411 return strings.TrimSpace(string(res.Stdout)) == "1"
412 }
413
414 func readState(ctx context.Context, fs *sftpfs.FS, path string) (ServeState, error) {
415 data, _, _, err := fs.ReadFile(ctx, path, 1<<20)
416 if err != nil {
417 return ServeState{}, err
418 }
419 return UnmarshalState(data)
420 }
421
422 func readToken(ctx context.Context, fs *sftpfs.FS, path string) (string, error) {
423 data, _, _, err := fs.ReadFile(ctx, path, 64<<10)
424 if err != nil {
425 return "", err
426 }
427 tok := strings.TrimSpace(string(data))
428 if tok == "" {
429 return "", errors.New("bootstrap: empty token file")
430 }
431 return tok, nil
432 }
433
434 func pollPortFile(ctx context.Context, fs *sftpfs.FS, portFile string, clock func() time.Time) (string, error) {
435 deadline := clock().Add(20 * time.Second)
436 for {
437 data, _, _, err := fs.ReadFile(ctx, portFile, 128)
438 if err == nil {
439 addr := strings.TrimSpace(string(data))
440 if validServeAddr(addr) {
441 return addr, nil
442 }
443 }
444 if clock().After(deadline) {
445 return "", errors.New("bootstrap: timed out waiting for serve to report its port")
446 }
447 select {
448 case <-ctx.Done():
449 return "", ctx.Err()
450 case <-time.After(250 * time.Millisecond):
451 }
452 }
453 }
454
455 func validServeAddr(addr string) bool {
456 host, portText, err := net.SplitHostPort(strings.TrimSpace(addr))
457 if err != nil || host != "127.0.0.1" {
458 return false
459 }
460 port, err := strconv.Atoi(portText)
461 return err == nil && port > 0 && port <= 65535
462 }
463
464 func readPIDFile(ctx context.Context, fs *sftpfs.FS, pidFile string) (int, error) {
465 data, _, _, err := fs.ReadFile(ctx, pidFile, 64)
466 if err != nil {
467 return 0, err
468 }
469 pid, err := strconv.Atoi(strings.TrimSpace(string(data)))
470 if err != nil || pid <= 0 {
471 return 0, errors.New("bootstrap: invalid serve pid file")
472 }
473 return pid, nil
474 }
475
476 func cleanupFailedLaunch(conn Conn, fs *sftpfs.FS, paths StatePaths, pid int) {
477 ctx, cancel := context.WithTimeout(context.Background(), 7*time.Second)
478 defer cancel()
479 if pid <= 0 {
480 pid, _ = readPIDFile(ctx, fs, paths.PidFile)
481 }
482 if pid > 0 {
483 _, _ = conn.Exec(ctx, StopCommand(pid, paths))
484 }
485 _ = fs.Remove(ctx, paths.StateJSON, false)
486 _ = fs.Remove(ctx, paths.TokenFile, false)
487 _ = fs.Remove(ctx, paths.PortFile, false)
488 _ = fs.Remove(ctx, paths.PidFile, false)
489 }
490
491 func resolveWorkspace(ctx context.Context, fs *sftpfs.FS, workspace, home string) (string, error) {
492 workspace = strings.TrimSpace(workspace)
493 if workspace == "" {
494 return home, nil
495 }
496 if workspace == "~" {
497 return home, nil
498 }
499 if after, ok0 := strings.CutPrefix(workspace, "~/"); ok0 {
500 return strings.TrimRight(home, "/") + "/" + after, nil
501 }
502 if strings.HasPrefix(workspace, "/") {
503 return workspace, nil
504 }
505 // Relative to home.
506 return strings.TrimRight(home, "/") + "/" + workspace, nil
507 }
508
509 func generateToken() (string, error) {
510 var b [32]byte
511 if _, err := rand.Read(b[:]); err != nil {
512 return "", err
513 }
514 return hex.EncodeToString(b[:]), nil
515 }
516
516 lines GO