返回 DeepSeek-Reasonix
session_takeover_serve.go
根目录 / internal / cli / session_takeover_serve.go
1 package cli
2
3 // Discovery side of CLI takeover: find the resident serve processes recorded
4 // under <Reasonix home>/remote, ask one of them for the session through POST
5 // /handoff, and prompt before taking a held session over.
6
7 import (
8 "bytes"
9 "context"
10 "encoding/json"
11 "errors"
12 "fmt"
13 "io"
14 "net/http"
15 "net/http/cookiejar"
16 "os"
17 "path/filepath"
18 "strings"
19 "time"
20
21 "reasonix/internal/agent"
22 "reasonix/internal/config"
23 "reasonix/internal/control"
24 "reasonix/internal/remote/bootstrap"
25 "reasonix/internal/store"
26 )
27
28 // cliTakeoverTimeout bounds the drain window of a wait-mode takeover.
29 const cliTakeoverTimeout = 2 * time.Minute
30
31 type cliServeRecord struct {
32 pid int
33 base string
34 token string
35 }
36
37 type cliTakeoverGrant struct {
38 SessionPath string `json:"sessionPath"`
39 MirrorID string `json:"mirrorId"`
40 HandoffID string `json:"handoffId,omitempty"`
41 ReturnHandoffID string `json:"returnHandoffId"`
42 SourceWriterID string `json:"sourceWriterId"`
43 TargetWriterID string `json:"targetWriterId"`
44 }
45
46 // cliServeProcessAlive is the PID probe used to prune serve state files
47 // whose process is gone. Variable so tests can model dead and live records.
48 var cliServeProcessAlive = webInstanceProcessAlive
49
50 // discoverCLIServes enumerates resident serve processes recorded under
51 // <Reasonix home>/remote. This machine is the SSH target in the takeover
52 // scenario, so the bootstrap's SFTP-written state files are local files here.
53 // Records whose PID no longer exists are skipped: a restarted desktop
54 // respawns the serve on a new port, and dialing the stale address only
55 // produces connection-refused noise that can shadow a live record's result.
56 func discoverCLIServes() []cliServeRecord {
57 dir := config.RemoteStateDir()
58 if dir == "" {
59 return nil
60 }
61 entries, err := os.ReadDir(dir)
62 if err != nil {
63 return nil
64 }
65 var out []cliServeRecord
66 for _, entry := range entries {
67 name := entry.Name()
68 if entry.IsDir() || !strings.HasPrefix(name, "serve-") || !strings.HasSuffix(name, ".json") {
69 continue
70 }
71 data, err := os.ReadFile(filepath.Join(dir, name))
72 if err != nil {
73 continue
74 }
75 state, err := bootstrap.UnmarshalState(data)
76 if err != nil || state.PID <= 0 || !cliServeProcessAlive(state.PID) {
77 continue
78 }
79 slug := strings.TrimSuffix(strings.TrimPrefix(name, "serve-"), ".json")
80 addr := state.Addr
81 if port, err := os.ReadFile(filepath.Join(dir, store.RemoteServePortName(slug))); err == nil {
82 if trimmed := strings.TrimSpace(string(port)); trimmed != "" {
83 addr = trimmed
84 }
85 }
86 if addr == "" {
87 continue
88 }
89 token := ""
90 if data, err := os.ReadFile(filepath.Join(dir, store.RemoteServeTokenName(slug))); err == nil {
91 token = strings.TrimSpace(string(data))
92 }
93 if token == "" {
94 continue
95 }
96 out = append(out, cliServeRecord{pid: state.PID, base: "http://" + addr, token: token})
97 }
98 return out
99 }
100
101 var discoverCLIServesForTakeover = discoverCLIServes
102
103 // cliServeForPID finds the resident serve holding the lease by matching the
104 // holder PID the lease error reported.
105 func cliServeForPID(pid int) *cliServeRecord {
106 records := discoverCLIServes()
107 for i := range records {
108 if records[i].pid == pid {
109 return &records[i]
110 }
111 }
112 return nil
113 }
114
115 func cliServeClient(ctx context.Context, record cliServeRecord) (*http.Client, error) {
116 jar, err := cookiejar.New(nil)
117 if err != nil {
118 return nil, err
119 }
120 client := &http.Client{Jar: jar}
121 auth, _ := json.Marshal(map[string]string{"token": record.token})
122 authReq, err := http.NewRequestWithContext(ctx, http.MethodPost, record.base+"/auth/token", bytes.NewReader(auth))
123 if err != nil {
124 return nil, err
125 }
126 authReq.Header.Set("Content-Type", "application/json")
127 authResp, err := client.Do(authReq)
128 if err != nil {
129 return nil, &cliServeUnreachableError{err: err}
130 }
131 _, _ = io.Copy(io.Discard, authResp.Body)
132 authResp.Body.Close()
133 if authResp.StatusCode != http.StatusNoContent {
134 return nil, fmt.Errorf("serve auth: status %d", authResp.StatusCode)
135 }
136 return client, nil
137 }
138
139 // cliTakeoverHeldSession requests a target-writer reservation and consumes it
140 // through leases. The previous keeper binding is retained if either step
141 // fails; callers commit their controller only after this returns a binding.
142 func cliTakeoverHeldSession(sessionPath string, leaseErr error, leases *control.SessionLeaseKeeper, manager *cliTakeoverManager) (*cliTakeoverBinding, error) {
143 if manager != nil && manager.Reclaiming() {
144 return nil, fmt.Errorf("the remote side is reclaiming the current session")
145 }
146 pid := 0
147 var leaseError *agent.SessionLeaseError
148 if errors.As(leaseErr, &leaseError) && leaseError != nil && leaseError.Info != nil {
149 pid = leaseError.Info.PID
150 }
151 if pid <= 0 {
152 return nil, fmt.Errorf("%w; no local serve identity to take over from", agent.ErrSessionLeaseHeld)
153 }
154 record := cliServeForPID(pid)
155 if record == nil {
156 // The holder PID does not match any discovered serve (stale state
157 // file, serve restart). The holder is on this machine, so try every
158 // local serve: the one holding the session will accept the handoff.
159 records := discoverCLIServes()
160 if len(records) == 0 {
161 return nil, fmt.Errorf("%w; holder pid %d is not a resident serve on this machine and no local serve is running", agent.ErrSessionLeaseHeld, pid)
162 }
163 var lastErr error
164 for i := range records {
165 binding, err := cliTakeoverFromServe(sessionPath, &records[i], leases, manager)
166 if err == nil {
167 return binding, nil
168 }
169 lastErr = err
170 }
171 return nil, lastErr
172 }
173 return cliTakeoverFromServe(sessionPath, record, leases, manager)
174 }
175
176 // postCLITakeoverHandoff exchanges one serve's /handoff for a grant. Both the
177 // legacy path-lease flow and the final-format identity flow validate the same
178 // wire contract; only the target key and post-grant reservation differ.
179 func postCLITakeoverHandoff(ctx context.Context, client *http.Client, base, sessionPath, errPrefix string) (cliTakeoverGrant, error) {
180 body, _ := json.Marshal(map[string]any{
181 "sessionPath": sessionPath, "targetWriterId": agent.SessionWriterID(),
182 "force": true, "mode": "wait", "timeoutMs": cliTakeoverTimeout.Milliseconds(),
183 })
184 req, err := http.NewRequestWithContext(ctx, http.MethodPost, base+"/handoff", bytes.NewReader(body))
185 if err == nil {
186 req.Header.Set("Content-Type", "application/json")
187 }
188 if err != nil {
189 return cliTakeoverGrant{}, err
190 }
191 resp, err := client.Do(req)
192 if err != nil {
193 return cliTakeoverGrant{}, fmt.Errorf("%s: %w", errPrefix, &cliServeUnreachableError{err: err})
194 }
195 defer resp.Body.Close()
196 respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<16))
197 if resp.StatusCode != http.StatusOK {
198 return cliTakeoverGrant{}, fmt.Errorf("%s: %s", errPrefix, strings.TrimSpace(string(respBody)))
199 }
200 var grant cliTakeoverGrant
201 if json.Unmarshal(respBody, &grant) != nil || grant.MirrorID == "" || grant.HandoffID == "" ||
202 grant.ReturnHandoffID == "" || grant.SourceWriterID == "" || grant.TargetWriterID != agent.SessionWriterID() {
203 return cliTakeoverGrant{}, fmt.Errorf("%s: invalid handoff grant", errPrefix)
204 }
205 return grant, nil
206 }
207
208 // cliTakeoverFromServe executes the handoff against one specific serve.
209 func cliTakeoverFromServe(sessionPath string, record *cliServeRecord, leases *control.SessionLeaseKeeper, manager *cliTakeoverManager) (*cliTakeoverBinding, error) {
210 pid := record.pid
211 ctx, cancel := context.WithTimeout(context.Background(), cliTakeoverTimeout+15*time.Second)
212 defer cancel()
213 client, err := cliServeClient(ctx, *record)
214 if err != nil {
215 return nil, fmt.Errorf("takeover from local serve (pid %d): %w", pid, err)
216 }
217 grant, err := postCLITakeoverHandoff(ctx, client, record.base, sessionPath, fmt.Sprintf("takeover from local serve (pid %d)", pid))
218 if err != nil {
219 return nil, err
220 }
221 binding := &cliTakeoverBinding{path: sessionPath, record: *record, client: client, grant: grant}
222 if manager != nil {
223 current, _, _, _ := manager.snapshot()
224 if current != nil && !manager.Returned() && agent.CanonicalSessionPath(current.path) != agent.CanonicalSessionPath(sessionPath) {
225 binding.priorMirror = current
226 }
227 }
228 previous, err := leases.RebindDetachingWithHandoff(sessionPath, grant.SourceWriterID, grant.HandoffID)
229 if err != nil {
230 cliEndFailedHandoff(binding)
231 return nil, err
232 }
233 binding.previous = previous
234 return binding, nil
235 }
236
237 // cliSessionTakeoverCandidate reports whether leaseErr describes a session
238 // /takeover can actually take: the lease info must identify a holder, and at
239 // least one resident serve must exist on this machine to hand the session
240 // over. The holder does not have to be a discovered serve — a serve's state
241 // file PID drifts across restarts and desktop reconnects, and the takeover
242 // execution falls back to trying every local serve — but with no serve at all
243 // the holder is another CLI or an unrelated runtime that has no handoff
244 // endpoint, and offering /takeover would only promise a command that must
245 // fail. The refusal then names the holder and the close hint instead.
246 func cliSessionTakeoverCandidate(leaseErr error) bool {
247 var leaseError *agent.SessionLeaseError
248 if !errors.As(leaseErr, &leaseError) || leaseError == nil || leaseError.Info == nil {
249 return false
250 }
251 return len(discoverCLIServesForTakeover()) > 0
252 }
253
254 // promptSessionTakeover asks on the terminal (pre-TUI startup) whether to take
255 // the held session over. Non-interactive sessions answer no.
256 func promptSessionTakeover(leaseErr error) bool {
257 if !isInteractive() {
258 return false
259 }
260 fmt.Fprintf(os.Stderr, "%s\n", sessionLeaseResumeRefusal(leaseErr))
261 fmt.Fprint(os.Stderr, "take over the session from this machine's resident serve? [y/N] ")
262 answer, err := readCLITakeoverAnswer()
263 if err != nil {
264 return false
265 }
266 answer = strings.ToLower(strings.TrimSpace(answer))
267 return answer == "y" || answer == "yes"
268 }
269
270 func readCLITakeoverAnswer() (string, error) {
271 buf := make([]byte, 64)
272 n, err := os.Stdin.Read(buf)
273 if n > 0 {
274 return string(buf[:n]), nil
275 }
276 return "", err
277 }
278
278 lines GO