返回 DeepSeek-Reasonix
session_runtime.go
根目录 / desktop / session_runtime.go
1 package main
2
3 import (
4 "context"
5 "crypto/rand"
6 "encoding/hex"
7 "errors"
8 "fmt"
9 "os"
10 "strings"
11 "time"
12
13 "reasonix/internal/agent"
14 )
15
16 type SessionRuntimePhase string
17
18 const (
19 sessionRuntimeStarting SessionRuntimePhase = "starting"
20 sessionRuntimeReady SessionRuntimePhase = "ready"
21 sessionRuntimeLeaseBlocked SessionRuntimePhase = "lease_blocked"
22 sessionRuntimeFailed SessionRuntimePhase = "failed"
23 sessionRuntimeClosing SessionRuntimePhase = "closing"
24 )
25
26 type SessionRuntimeIssue struct {
27 Code string `json:"code"`
28 Message string `json:"message"`
29 Retryable bool `json:"retryable"`
30 HolderPID int `json:"holderPid,omitempty"`
31 HolderHost string `json:"holderHost,omitempty"`
32 AcquiredAt string `json:"acquiredAt,omitempty"`
33 }
34
35 type SessionRuntimeView struct {
36 Phase SessionRuntimePhase `json:"phase"`
37 Epoch string `json:"epoch"`
38 Issue *SessionRuntimeIssue `json:"issue,omitempty"`
39 }
40
41 // desktopSessionRuntime is the process-local ownership record for one writable
42 // session. WorkspaceTab still carries compatibility projections of controller
43 // and lease fields while the desktop code migrates, but this registry is the
44 // authority that prevents a second local build from competing for the same
45 // session path.
46 //
47 // All fields are guarded by App.mu.
48 type desktopSessionRuntime struct {
49 ID, Key, Epoch string
50 Phase SessionRuntimePhase
51 Issue *SessionRuntimeIssue
52 Owner *WorkspaceTab
53 suppressStartupRestore bool
54 readyCh chan struct{}
55 }
56
57 func newSessionRuntimeID(prefix string) string {
58 var b [12]byte
59 if _, err := rand.Read(b[:]); err == nil {
60 return prefix + "_" + hex.EncodeToString(b[:])
61 }
62 return prefix + "_" + time.Now().UTC().Format("20060102150405.000000000")
63 }
64
65 func cloneSessionRuntimeIssue(issue *SessionRuntimeIssue) *SessionRuntimeIssue {
66 if issue == nil {
67 return nil
68 }
69 copy := *issue
70 return &copy
71 }
72
73 func sessionRuntimeIssueForError(err error) *SessionRuntimeIssue {
74 if err == nil {
75 return nil
76 }
77 issue := &SessionRuntimeIssue{
78 Code: "startup_failed",
79 Message: userFacingSessionLeaseError("", err).Error(),
80 Retryable: false,
81 }
82 if !errors.Is(err, agent.ErrSessionLeaseHeld) {
83 return issue
84 }
85 issue.Code = "session_lease_held"
86 issue.Retryable = true
87 var leaseErr *agent.SessionLeaseError
88 if errors.As(err, &leaseErr) && leaseErr != nil && leaseErr.Info != nil {
89 issue.HolderPID = leaseErr.Info.PID
90 issue.HolderHost = strings.TrimSpace(leaseErr.Info.Hostname)
91 if !leaseErr.Info.AcquiredAt.IsZero() {
92 issue.AcquiredAt = leaseErr.Info.AcquiredAt.UTC().Format(time.RFC3339)
93 }
94 }
95 return issue
96 }
97
98 func (a *App) newSessionRuntimeLocked(tab *WorkspaceTab, key string) *desktopSessionRuntime {
99 if existing := a.runtimeBySessionKey[key]; key != "" && existing != nil && existing.Owner != tab {
100 // A second tab may begin restoring the same persisted path before it
101 // reaches claimSessionRuntime. Do not overwrite the first starting
102 // placeholder; the later build will wait for and attach to it.
103 key = ""
104 }
105 rt := &desktopSessionRuntime{
106 ID: newSessionRuntimeID("runtime"),
107 Key: key,
108 Epoch: newSessionRuntimeID("epoch"),
109 Phase: sessionRuntimeStarting,
110 Owner: tab,
111 readyCh: make(chan struct{}),
112 }
113 if tab != nil && tab.Ctrl != nil && tab.Ready {
114 rt.Phase = sessionRuntimeReady
115 closeRuntimeReadyChannelLocked(rt)
116 }
117 if a.runtimeByID == nil {
118 a.runtimeByID = map[string]*desktopSessionRuntime{}
119 }
120 if a.runtimeBySessionKey == nil {
121 a.runtimeBySessionKey = map[string]*desktopSessionRuntime{}
122 }
123 a.runtimeByID[rt.ID] = rt
124 if key != "" {
125 a.runtimeBySessionKey[key] = rt
126 }
127 if tab != nil {
128 tab.runtimeID = rt.ID
129 if tab.sink != nil {
130 tab.sink.setRuntimeEpoch(rt.Epoch)
131 }
132 }
133 return rt
134 }
135
136 func (a *App) runtimeForTabLocked(tab *WorkspaceTab) *desktopSessionRuntime {
137 if tab == nil || strings.TrimSpace(tab.runtimeID) == "" {
138 return nil
139 }
140 rt := a.runtimeByID[tab.runtimeID]
141 if rt == nil || rt.Owner != tab {
142 return nil
143 }
144 return rt
145 }
146
147 // runtimeEpochForTabLocked returns the generation already transferred to tab.
148 // Callers use it when reattaching an existing runtime: the generation must be
149 // announced before that runtime replays any pending prompt.
150 func (a *App) runtimeEpochForTabLocked(tab *WorkspaceTab) string {
151 if rt := a.runtimeForTabLocked(tab); rt != nil {
152 return rt.Epoch
153 }
154 if tab != nil && tab.sink != nil {
155 return tab.sink.runtimeEpochSnapshot()
156 }
157 return ""
158 }
159
160 func (a *App) runtimeOwnerLiveLocked(rt *desktopSessionRuntime) bool {
161 if rt == nil || rt.Owner == nil {
162 return false
163 }
164 if a.tabs[rt.Owner.ID] == rt.Owner {
165 return true
166 }
167 for _, detached := range a.detachedSessions {
168 if detached == rt.Owner {
169 return true
170 }
171 }
172 return false
173 }
174
175 func (a *App) removeSessionRuntimeMappingsLocked(rt *desktopSessionRuntime) {
176 if rt == nil {
177 return
178 }
179 for key, candidate := range a.runtimeBySessionKey {
180 if candidate == rt {
181 delete(a.runtimeBySessionKey, key)
182 }
183 }
184 delete(a.runtimeByID, rt.ID)
185 }
186
187 func closeRuntimeReadyChannelLocked(rt *desktopSessionRuntime) {
188 if rt == nil || rt.readyCh == nil {
189 return
190 }
191 close(rt.readyCh)
192 rt.readyCh = nil
193 }
194
195 func (a *App) setSessionRuntimePhaseLocked(tab *WorkspaceTab, phase SessionRuntimePhase, err error) {
196 if tab == nil {
197 return
198 }
199 rt := a.runtimeForTabLocked(tab)
200 if rt == nil {
201 key := sessionRuntimeKey(tab.SessionPath)
202 if existing := a.runtimeBySessionKey[key]; key != "" && existing != nil && existing.Owner != tab {
203 return
204 }
205 rt = a.newSessionRuntimeLocked(tab, key)
206 }
207 rt.Phase = phase
208 rt.Issue = sessionRuntimeIssueForError(err)
209 rt.suppressStartupRestore = false
210 if phase == sessionRuntimeStarting {
211 rt.Issue = nil
212 if rt.readyCh == nil {
213 rt.readyCh = make(chan struct{})
214 }
215 } else {
216 closeRuntimeReadyChannelLocked(rt)
217 }
218 if tab.sink != nil {
219 tab.sink.setRuntimeEpoch(rt.Epoch)
220 }
221 }
222
223 func (a *App) advanceSessionRuntimeEpochLocked(tab *WorkspaceTab) string {
224 if tab == nil {
225 return ""
226 }
227 rt := a.runtimeForTabLocked(tab)
228 if rt == nil {
229 rt = a.newSessionRuntimeLocked(tab, sessionRuntimeKey(tab.SessionPath))
230 }
231 rt.Epoch = newSessionRuntimeID("epoch")
232 // A final-format session already has the process-generation identity that
233 // fences late events and prompt answers. Reuse that exact epoch in the
234 // Desktop registry instead of inventing a second, UI-only generation.
235 if _, runtime, exclusive := exclusiveSessionBinding(tab.Ctrl); exclusive && runtime != nil {
236 rt.Epoch = runtime.StateSnapshot().Epoch
237 }
238 rt.Phase = sessionRuntimeReady
239 rt.Issue = nil
240 rt.suppressStartupRestore = false
241 closeRuntimeReadyChannelLocked(rt)
242 if tab.sink != nil {
243 tab.sink.setRuntimeEpoch(rt.Epoch)
244 }
245 if binder, ok := tab.Ctrl.(interface{ BindTranscriptRuntimeEpoch(string) }); ok {
246 binder.BindTranscriptRuntimeEpoch(rt.Epoch)
247 }
248 return rt.Epoch
249 }
250
251 func (a *App) sessionRuntimeViewLocked(tab *WorkspaceTab) SessionRuntimeView {
252 if tab == nil {
253 return SessionRuntimeView{Phase: sessionRuntimeStarting}
254 }
255 if rt := a.runtimeForTabLocked(tab); rt != nil {
256 return SessionRuntimeView{
257 Phase: rt.Phase,
258 Epoch: rt.Epoch,
259 Issue: cloneSessionRuntimeIssue(rt.Issue),
260 }
261 }
262 view := SessionRuntimeView{Phase: sessionRuntimeStarting}
263 switch {
264 case tab.Ctrl != nil && tab.Ready:
265 view.Phase = sessionRuntimeReady
266 case tab.StartupErrLeaseHeld:
267 view.Phase = sessionRuntimeLeaseBlocked
268 view.Issue = sessionRuntimeIssueForError(&agent.SessionLeaseError{})
269 case strings.TrimSpace(tab.StartupErr) != "":
270 view.Phase = sessionRuntimeFailed
271 view.Issue = &SessionRuntimeIssue{Code: "startup_failed", Message: tab.StartupErr}
272 }
273 return view
274 }
275
276 func (a *App) bindSessionRuntimeKeyLocked(tab *WorkspaceTab, path string) bool {
277 if tab == nil {
278 return false
279 }
280 key := sessionRuntimeKey(path)
281 if key == "" {
282 return true
283 }
284 if existing := a.runtimeBySessionKey[key]; existing != nil && existing.Owner != tab {
285 return false
286 }
287 rt := a.runtimeForTabLocked(tab)
288 if rt == nil {
289 a.newSessionRuntimeLocked(tab, key)
290 return true
291 }
292 if rt.Key != key {
293 for alias, candidate := range a.runtimeBySessionKey {
294 if candidate == rt {
295 delete(a.runtimeBySessionKey, alias)
296 }
297 }
298 a.unregisterDetachedRuntimeLocked(tab)
299 }
300 rt.Key = key
301 a.runtimeBySessionKey[key] = rt
302 return true
303 }
304
305 type sessionRuntimePathTransition struct {
306 runtime *desktopSessionRuntime
307 owner *WorkspaceTab
308 oldKey string
309 targetKey string
310 expectedEpoch string
311 }
312
313 func (a *App) reserveSessionRuntimePath(tab *WorkspaceTab, path string) (sessionRuntimePathTransition, error) {
314 targetKey := sessionRuntimeKey(path)
315 if tab == nil || targetKey == "" {
316 return sessionRuntimePathTransition{}, nil
317 }
318 a.mu.Lock()
319 defer a.mu.Unlock()
320 if existing := a.runtimeBySessionKey[targetKey]; existing != nil && existing.Owner != tab {
321 return sessionRuntimePathTransition{}, fmt.Errorf("%w: local runtime already owns session", agent.ErrSessionLeaseHeld)
322 }
323 rt := a.runtimeForTabLocked(tab)
324 if rt == nil {
325 // A path transition must retain the source identity until commit. Using
326 // targetKey here would make a failed first rebind forget the still-live
327 // source controller and its lease.
328 rt = a.newSessionRuntimeLocked(tab, sessionRuntimeKey(tab.currentSessionIdentity()))
329 }
330 transition := sessionRuntimePathTransition{
331 runtime: rt,
332 owner: tab,
333 oldKey: rt.Key,
334 targetKey: targetKey,
335 expectedEpoch: rt.Epoch,
336 }
337 // Keep the old key mapped until the lease rebind succeeds. The target alias
338 // prevents another local startup from claiming it during the off-lock file
339 // operation.
340 a.runtimeBySessionKey[targetKey] = rt
341 return transition, nil
342 }
343
344 func (a *App) commitSessionRuntimePath(transition sessionRuntimePathTransition) {
345 if transition.runtime == nil || transition.targetKey == "" {
346 return
347 }
348 a.mu.Lock()
349 defer a.mu.Unlock()
350 rt := transition.runtime
351 if transition.oldKey != "" && transition.oldKey != transition.targetKey && a.runtimeBySessionKey[transition.oldKey] == rt {
352 delete(a.runtimeBySessionKey, transition.oldKey)
353 }
354 rt.Key = transition.targetKey
355 a.runtimeBySessionKey[transition.targetKey] = rt
356 }
357
358 // commitSessionRuntimePathLocked commits a previously reserved path only when
359 // the same runtime generation still owns both aliases. It lets controller
360 // swaps make the registry update part of their single App.mu commit.
361 func (a *App) commitSessionRuntimePathLocked(transition sessionRuntimePathTransition) bool {
362 if transition.runtime == nil || transition.targetKey == "" {
363 return false
364 }
365 if !a.sessionRuntimePathTransitionValidLocked(transition) {
366 return false
367 }
368 rt := transition.runtime
369 if transition.oldKey != "" && transition.oldKey != transition.targetKey && a.runtimeBySessionKey[transition.oldKey] == rt {
370 delete(a.runtimeBySessionKey, transition.oldKey)
371 }
372 rt.Key = transition.targetKey
373 a.runtimeBySessionKey[transition.targetKey] = rt
374 return true
375 }
376
377 func (a *App) sessionRuntimePathTransitionValidLocked(transition sessionRuntimePathTransition) bool {
378 if transition.runtime == nil || transition.targetKey == "" {
379 return false
380 }
381 rt := transition.runtime
382 if a.runtimeByID[rt.ID] != rt ||
383 rt.Owner != transition.owner ||
384 rt.Epoch != transition.expectedEpoch ||
385 (rt.Key != transition.oldKey && rt.Key != transition.targetKey) ||
386 a.runtimeBySessionKey[transition.targetKey] != rt {
387 return false
388 }
389 return true
390 }
391
392 func (a *App) rollbackSessionRuntimePath(transition sessionRuntimePathTransition) {
393 if transition.runtime == nil || transition.targetKey == "" || transition.targetKey == transition.oldKey {
394 return
395 }
396 a.mu.Lock()
397 if a.runtimeBySessionKey[transition.targetKey] == transition.runtime {
398 delete(a.runtimeBySessionKey, transition.targetKey)
399 }
400 a.mu.Unlock()
401 }
402
403 // claimSessionRuntime reserves path for tab, or waits for/attaches the existing
404 // local runtime. The caller owns its not-yet-published candidate controller; a
405 // true return means that candidate must be closed because tab now uses the
406 // already registered runtime.
407 func (a *App) claimSessionRuntime(tab *WorkspaceTab, path string, ctx context.Context) bool {
408 key := sessionRuntimeKey(path)
409 if tab == nil || key == "" {
410 return false
411 }
412 for {
413 a.mu.Lock()
414 if tab.removed || a.tabs[tab.ID] != tab {
415 a.mu.Unlock()
416 return false
417 }
418 rt := a.runtimeBySessionKey[key]
419 if rt != nil && !a.runtimeOwnerLiveLocked(rt) {
420 a.removeSessionRuntimeMappingsLocked(rt)
421 rt = nil
422 }
423 switch {
424 case rt == nil:
425 // Detached runtimes created before the admission registry was
426 // populated are a compatibility edge. Attach them before claiming
427 // the key so applyRuntimeTab can publish one authoritative runtime
428 // instead of leaving behind an unused placeholder.
429 if matched := a.liveRuntimeTabMatchingLocked(tab, path); matched != nil && matched.Ctrl != nil {
430 identity := runtimeAttachIdentity(matched, path)
431 a.mu.Unlock()
432 return a.attachExistingSessionRuntime(tab, identity, a.ctx)
433 }
434 a.bindSessionRuntimeKeyLocked(tab, path)
435 a.mu.Unlock()
436 return false
437 case rt.Owner == tab:
438 a.mu.Unlock()
439 // The target may own the starting placeholder while a legacy
440 // visible/detached runtime for the same session predates the
441 // registry. Let attachExistingSessionRuntime adopt that usable
442 // controller; otherwise this remains the owner build.
443 return a.attachExistingSessionRuntime(tab, path, a.ctx)
444 case rt.Phase == sessionRuntimeStarting && rt.readyCh != nil:
445 wait := rt.readyCh
446 a.mu.Unlock()
447 select {
448 case <-wait:
449 continue
450 case <-ctx.Done():
451 return false
452 case <-time.After(250 * time.Millisecond):
453 // Re-check owner liveness even when a superseded build exited
454 // before publishing a terminal phase.
455 continue
456 }
457 default:
458 a.mu.Unlock()
459 return a.attachExistingSessionRuntime(tab, path, a.ctx)
460 }
461 }
462 }
463
464 func (a *App) releaseSessionRuntimeLocked(tab *WorkspaceTab) {
465 rt := a.runtimeForTabLocked(tab)
466 if rt == nil {
467 if tab != nil {
468 tab.runtimeID = ""
469 }
470 return
471 }
472 rt.Phase = sessionRuntimeClosing
473 closeRuntimeReadyChannelLocked(rt)
474 a.removeSessionRuntimeMappingsLocked(rt)
475 if tab != nil {
476 tab.runtimeID = ""
477 }
478 }
479
480 func sameCurrentProcessLease(err error) bool {
481 var leaseErr *agent.SessionLeaseError
482 if !errors.As(err, &leaseErr) || leaseErr == nil || leaseErr.Info == nil {
483 return false
484 }
485 if leaseErr.Info.PID != os.Getpid() || leaseErr.Info.WriterID != agent.SessionWriterID() {
486 return false
487 }
488 host, _ := os.Hostname()
489 return strings.TrimSpace(leaseErr.Info.Hostname) == strings.TrimSpace(host)
490 }
491
492 // sessionParentLive reports whether a desktop tab or detached runtime in this
493 // process currently owns, or is still building, the requested session. It is
494 // intentionally checked before stale-subagent cleanup probes the durable lease:
495 // a starting tab has published SessionPath but may not have bound that lease yet.
496 func (a *App) sessionParentLive(sessionPath string) bool {
497 return a.sessionParentLiveForBuild(sessionPath, nil)
498 }
499
500 // subagentParentProbeForBuild excludes an initial build's own unbound tab: that
501 // build can safely repair its crash leftovers before it binds the session lease.
502 // Other live tabs remain protected from the sweep.
503 func (a *App) subagentParentProbeForBuild(building *WorkspaceTab) func(string) bool {
504 return func(sessionPath string) bool {
505 return a.sessionParentLiveForBuild(sessionPath, building)
506 }
507 }
508
509 func (a *App) sessionParentLiveForBuild(sessionPath string, building *WorkspaceTab) bool {
510 key := sessionRuntimeKey(sessionPath)
511 if a == nil || key == "" {
512 return false
513 }
514
515 a.mu.RLock()
516 defer a.mu.RUnlock()
517 if rt := a.runtimeBySessionKey[key]; rt != nil && a.runtimeOwnerLiveLocked(rt) &&
518 !(rt.Owner == building && building.Ctrl == nil) {
519 return true
520 }
521 liveTab := func(tab *WorkspaceTab) bool {
522 if tab == nil {
523 return false
524 }
525 if tab == building && tab.Ctrl == nil {
526 return false
527 }
528 if sessionRuntimeKey(tab.SessionPath) == key {
529 return true
530 }
531 return tab.Ctrl != nil && sessionRuntimeKey(tab.Ctrl.SessionPath()) == key
532 }
533 for _, tab := range a.tabs {
534 if liveTab(tab) {
535 return true
536 }
537 }
538 for _, tab := range a.detachedSessions {
539 if liveTab(tab) {
540 return true
541 }
542 }
543 return false
544 }
545
545 lines GO