1// Package daemon holds vev's server-side session multiplexer use case.
2package daemon
3
4import (
5 "context"
6 "crypto/rand"
7 "errors"
8 "fmt"
9 "os"
10 "slices"
11 "strconv"
12 "strings"
13 "sync"
14 "sync/atomic"
15 "time"
16
17 "github.com/bnema/vev/internal/domain"
18 "github.com/bnema/vev/internal/ports"
19 "github.com/bnema/vev/internal/usecase/layout"
20)
21
22var (
23 errSessionNameRequired = errors.New("name required")
24 errSessionNameInUse = errors.New("name already in use")
25)
26
27type terminalEnv struct {
28 TrueColor bool
29}
30
31type session struct {
32 sessionCore
33
34 ctx context.Context
35 cancel context.CancelFunc
36
37 // dispatchMu serializes user-initiated mutating command dispatch (palette
38 // and vev-cmd control) for this session, so a one-shot command cannot
39 // resolve focus against one state and mutate a later one. Lock order:
40 // dispatchMu strictly before d.mu, sess.mu, or any tab mu.
41 dispatchMu sync.Mutex
42 // teardownMu guards teardown ownership and is the outermost per-session
43 // lifecycle lock: never acquire it while holding daemon, session, snapshot,
44 // tab, or pane locks. The owner performs teardown without holding teardownMu;
45 // duplicate daemon-shutdown callers can therefore bound their ownership wait
46 // with Serve's shared snapshot deadline.
47 teardownMu sync.Mutex
48 teardownActive bool
49 teardownDone chan struct{}
50 moveReservations uint
51 teardownWaiters uint
52 teardownChanged chan struct{}
53 lifecycleStopOnce sync.Once
54 // sessionCore.mu guards tabs, attachment membership, restoreDone, clipFiles,
55 // and clipboard queue state.
56 // restoreDone remains attached to a restored session after it is published in
57 // the live registry, so racing attaches cannot bypass restoration completion.
58 restoreDone chan struct{}
59 themeMu sync.Mutex
60 tabs []*tab
61 clipboardQueue []clipboardForward
62 clipboardWorkerRunning bool
63 cwd string
64 terminal terminalEnv
65 // env is the authoritative immutable environment snapshot for future PTY children.
66 // It is guarded by mu and is always copied on ingress and egress.
67 env []string
68 snapDirty atomic.Bool
69 snapEligible atomic.Bool
70 // snapshotMu serializes mutation revisions and repository publication
71 // generations with worker completion. It is intentionally independent from
72 // mu: persistence never holds session state locks while encoding or writing.
73 snapshotMu sync.Mutex
74 snapshotGeneration uint64 // newest mutation revision
75 snapshotPublishedGeneration uint64 // newest repository generation
76 snapshotPublishedCheckpoint *domain.CheckpointRef
77 snapshotPublishedMutationRevision uint64
78 snapshotNextEligibleAt time.Time
79 // The coordinator state below is guarded by snapshotMu. A capture can be
80 // queued globally or in flight, never more than one of each per session.
81 // Quarantine cancels the session publication context before destructive
82 // repository operations and publicationDone joins a started publication.
83 snapshotPending bool
84 snapshotPendingCaptures uint
85 // snapshotForcedGeneration is the newest mutation revision a forced
86 // checkpoint must publish. It survives an older routine capture so worker
87 // completion can enqueue exactly one forced successor.
88 snapshotForcedGeneration uint64
89 snapshotQueuedCapture *snapshotCapture
90 snapshotInFlightCapture *snapshotCapture
91 snapshotPublicationDone chan struct{}
92 snapshotPublicationContext context.Context
93 snapshotPublicationCancel context.CancelFunc
94 snapshotQuarantined bool
95 // snapshotChunkCache retains encoded sealed history only for this named
96 // session. snapshotMu owns it so cache touches never contend with pane
97 // state or mutate bytes retained by queued/in-flight publications.
98 snapshotChunkCache *snapshotChunkCache
99 snapshotChanged chan struct{}
100 snapshotWake chan struct{}
101 // syncGen makes synchronized-output watchdog generations unique across all
102 // panes in this session.
103 syncGen atomic.Uint64
104 // layoutApplyMu serializes whole-session resize prepare/apply/admit/publish
105 // transactions. It is deliberately not an architecture lock and may span
106 // external PTY.Resize calls.
107 layoutApplyMu sync.Mutex
108 // clipFiles records clipboard-image-transfer temp file paths (see
109 // clipboard.go) written for this session, removed best-effort in
110 // killSession.
111 clipFiles []string
112
113 // floatingLaunchMu is ownership synchronization for external floating
114 // launches. It is separate from mu so PTYFactory.Open never holds a
115 // session, tab, or daemon architecture lock.
116 floatingLaunchMu sync.Mutex
117 floatingLaunchStopping bool
118 floatingLaunches map[*floatingLaunch]struct{}
119}
120
121func (s *session) nameSnapshot() string {
122 if s == nil {
123 return ""
124 }
125 s.mu.Lock()
126 defer s.mu.Unlock()
127 return s.name
128}
129
130// tab is a pane layout container; pane owns PTY/screen/scrollback/render scheduling state.
131type tab struct {
132 mu sync.Mutex // guards tree, panes, floating, nextPaneID, size, and pane map membership
133
134 // layoutGeneration invalidates prepared geometry whenever live tab layout
135 // state changes. layoutApplyMu serializes the lock-free PTY apply boundary.
136 layoutGeneration uint64
137 layoutApplyMu sync.Mutex
138 // layoutRetryMu owns the one bounded delayed retry worker for accepted
139 // tiled-layout degradation. Its context is derived from ctx, so tab/session
140 // teardown cancels a waiting retry before it can touch PTY state.
141 layoutRetryMu sync.Mutex
142 layoutRetryCancel context.CancelFunc
143 layoutRetryRunning bool
144
145 stableID string
146 name string
147 tree *layout.Tree
148 panes map[layout.PaneID]*pane
149 nextPaneID int
150 size domain.Size
151 ctx context.Context
152 cancel context.CancelFunc
153 // floating is independent from the normal layout tree and pane map.
154 floating floatingSlot
155
156 // attention fields are guarded by the owning session.mu.
157 attention bool
158 attentionAt time.Time
159 attentionVisiblePaint bool
160}
161
162// attachedClient is a client currently attached to a session's tab. rend is
163// its private renderer shadow (so each client gets output minimised against
164// what it has actually seen). sendMu serialises the two senders — the render
165// coordinator and the connection handler — so the transport's single-writer
166
167func (d *Daemon) touchMRU(entry *session) {
168 if d == nil || entry == nil || entry.core() == nil {
169 return
170 }
171 core := entry.core()
172 seq := d.mruSeq.Add(1)
173 for {
174 old := core.mruAt.Load()
175 if old >= seq {
176 return
177 }
178 if core.mruAt.CompareAndSwap(old, seq) {
179 break
180 }
181 }
182 entry.mu.Lock()
183 if !entry.ephemeral {
184 d.markCatalogueDirty(entry.persistRecordLocked(max(d.nowUnixNano(), entry.createdAt, int64(1))).MetadataUpdate())
185 }
186 entry.mu.Unlock()
187}
188
189func (d *Daemon) createSessionLocked(name string, ephemeral bool, cwd string, sz domain.Size, term terminalEnv, env []string, restoredTabNames ...[]string) (*session, error) {
190 return d.createSessionLockedWithMode(name, ephemeral, cwd, sz, term, env, restoredTabNames...)
191}
192
193func (d *Daemon) createSessionLockedWithMode(name string, ephemeral bool, cwd string, sz domain.Size, term terminalEnv, env []string, restoredTabNames ...[]string) (*session, error) {
194 env = copyEnvironment(env)
195 if _, reserved := d.creating[name]; reserved {
196 return nil, errSessionNameInUse
197 }
198 d.creating[name] = struct{}{}
199 defer delete(d.creating, name)
200 stopped, resuming := d.stopped[name]
201 var authoritative domain.CatalogueRecord
202 var authoritativeExists bool
203 if !ephemeral && resuming && d.persistEnabled {
204 d.mu.Unlock()
205 var err error
206 authoritative, authoritativeExists, err = d.catalogueRecord(name)
207 d.mu.Lock()
208 if err != nil {
209 return nil, domain.UserErr(domain.NoticeSessionSpawn, "couldn't read session catalogue", err)
210 }
211 if d.closing {
212 return nil, &protoErr{ports.ErrServerShutdown, "daemon is shutting down"}
213 }
214 stopped, resuming = d.stopped[name]
215 }
216 var createdAt int64
217 var incarnation domain.IncarnationID
218 if !ephemeral {
219 if resuming {
220 // A stopped session is the same lifecycle when resumed; it must retain
221 // the persisted identity rather than receive a fresh timestamp.
222 createdAt = stopped.createdAt
223 incarnation = stopped.incarnation
224 if authoritativeExists {
225 createdAt = authoritative.CreatedAt
226 incarnation = authoritative.IncarnationID
227 }
228 if createdAt > d.lastAllocatedCreatedAt {
229 d.lastAllocatedCreatedAt = createdAt
230 }
231 } else {
232 var err error
233 createdAt, err = d.allocateLifecycleCreatedAtLocked()
234 if err != nil {
235 return nil, domain.UserErr(domain.NoticeSessionSpawn, "couldn't create session", err)
236 }
237 }
238 }
239 if incarnation == (domain.IncarnationID{}) {
240 var err error
241 incarnation, err = domain.NewSessionLifecycleID(rand.Reader)
242 if err != nil {
243 return nil, domain.UserErr(domain.NoticeSessionSpawn, "couldn't create session", fmt.Errorf("generate lifecycle identity: %w", err))
244 }
245 }
246
247 if d.ptys == nil {
248 return nil, domain.UserErr(domain.NoticeSessionSpawn, "couldn't create session", errors.New("daemon: PTY factory is unavailable"))
249 }
250 tbSize := contentSize(sz)
251 var names []string
252 var tabRecords []domain.CatalogueTabRecord
253 if resuming && len(stopped.tabRecords) != 0 {
254 tabRecords = append([]domain.CatalogueTabRecord(nil), stopped.tabRecords...)
255 names = make([]string, len(tabRecords))
256 for i, record := range tabRecords {
257 names[i] = record.Name
258 }
259 } else if len(restoredTabNames) > 0 {
260 names = append([]string(nil), restoredTabNames[0]...)
261 }
262 tabs := make([]*tab, 0, max(1, len(names)))
263 for i := range max(1, len(names)) {
264 tabStableID, paneStableID, err := d.newTabPaneStableIDs()
265 if i < len(tabRecords) && tabRecords[i].StableID != "" {
266 tabStableID = string(tabRecords[i].StableID)
267 }
268 if err != nil {
269 closeTabs(tabs)
270 return nil, domain.UserErr(domain.NoticeSessionSpawn, "couldn't create session", err)
271 }
272 command, args := d.ptyCommand(env)
273 lifetime := d.newPaneProcessLifetime(d.serveCtx)
274 pty, err := d.ptys.Open(lifetime.ctx, command, args, childEnvFrom(env, name, tabStableID, paneStableID, term), cwd, tbSize)
275 if err != nil {
276 lifetime.abort()
277 if pty != nil {
278 _ = pty.Close()
279 }
280 closeTabs(tabs)
281 d.log.Warn("pty spawn failed", "err", err, "session", name, "kind", "session")
282 return nil, domain.UserErr(domain.NoticeSessionSpawn, "couldn't create session: shell failed to start", err)
283 }
284 tb := newTabWithStableID(tabStableID, paneStableID, pty, tbSize)
285 if !lifetime.publish(tb.focusedPane()) {
286 lifetime.abort()
287 _ = pty.Close()
288 closeTabs(tabs)
289 return nil, domain.UserErr(domain.NoticeSessionSpawn, "couldn't create session: shell failed to start", lifetime.ctx.Err())
290 }
291 if i < len(names) {
292 tb.name = names[i]
293 }
294 tabs = append(tabs, tb)
295 }
296
297 id := domain.SessionID(fmt.Sprintf("sess-%d", d.nextID))
298 d.nextID++
299
300 sctx, cancel := context.WithCancel(d.serveCtx)
301 for _, tb := range tabs {
302 tb.ctx, tb.cancel = context.WithCancel(sctx)
303 }
304 lastUsedSeq := uint64(0)
305 if !ephemeral && resuming {
306 lastUsedSeq = stopped.lastUsedSeq
307 }
308 if lastUsedSeq == 0 {
309 lastUsedSeq = d.mruSeq.Add(1)
310 }
311 sess := &session{
312 sessionCore: sessionCore{
313 id: id,
314 name: name,
315 ephemeral: ephemeral,
316 createdAt: createdAt,
317 incarnation: incarnation,
318 },
319 ctx: sctx,
320 cancel: cancel,
321 tabs: tabs,
322 cwd: cwd,
323 terminal: term,
324 env: env,
325 snapshotWake: d.snapshotWake,
326 }
327 if !ephemeral && name != "" {
328 sess.snapshotChunkCache = newSnapshotChunkCache(snapshotChunkCacheLimit)
329 }
330 sess.mruAt.Store(lastUsedSeq)
331 sess.snapEligible.Store(!ephemeral && name != "")
332 // These tabs are still private construction state. Publish every tiled
333 // owner before the session enters d.sessions or any PTY reader starts.
334 for _, tb := range tabs {
335 publishTiledPaneOwners(sess, tb)
336 }
337 var rollbackDurable func() error
338 if !ephemeral && d.persistEnabled {
339 record := domain.CatalogueRecord{Name: name, IncarnationID: incarnation, Cwd: cwd, CreatedAt: createdAt, UpdatedAt: createdAt, LastUsedSeq: lastUsedSeq, TabNames: names, TabRecords: tabRecordsFromTabs(tabs)}
340 if d.recovery == nil {
341 closeTabs(tabs)
342 cancel()
343 return nil, domain.UserErr(domain.NoticeSessionSpawn, "couldn't create session", errors.New("durable session authority is not configured"))
344 }
345 var err error
346 if resuming && authoritativeExists {
347 err = d.catalogue.UpdateMetadata(record.MetadataUpdate())
348 if err == nil {
349 rollbackDurable = func() error { return d.catalogue.Replace(name, authoritative) }
350 err = d.catalogue.Sync()
351 }
352 } else {
353 record, err = d.recovery.Create(d.serveCtx, record)
354 if err == nil {
355 rollbackDurable = func() error {
356 current, ok, recordErr := d.catalogue.Record(name)
357 if recordErr != nil {
358 return recordErr
359 }
360 if !ok {
361 return nil
362 }
363 if current.IncarnationID != record.IncarnationID {
364 return errors.New("daemon: created session authority changed before rollback")
365 }
366 return d.catalogue.Delete(name)
367 }
368 }
369 }
370 if err != nil {
371 if rollbackDurable != nil {
372 err = errors.Join(err, rollbackDurable())
373 }
374 closeTabs(tabs)
375 cancel()
376 return nil, domain.UserErr(domain.NoticeSessionSpawn, "couldn't create session", err)
377 }
378 sess.incarnation = record.IncarnationID
379 if resuming && authoritativeExists && authoritative.Committed != nil && authoritative.IncarnationID == sess.incarnation {
380 checkpoint := *authoritative.Committed
381 sess.snapshotPublishedGeneration = checkpoint.Generation
382 sess.snapshotPublishedCheckpoint = &checkpoint
383 }
384 }
385 if !d.registerSessionLocked(sess) {
386 err := errors.New("daemon: session registry rejected local session")
387 if rollbackDurable != nil {
388 err = errors.Join(err, rollbackDurable())
389 }
390 closeTabs(tabs)
391 cancel()
392 return nil, domain.UserErr(domain.NoticeSessionSpawn, "couldn't create session", err)
393 }
394 if !ephemeral {
395 // Keep stopped lifecycle authority intact until runtime publication and
396 // its durable side effects have both succeeded.
397 delete(d.stopped, name)
398 }
399 d.log.Info("session created", "session", name, "id", id, "ephemeral", ephemeral)
400 for i, tb := range tabs {
401 d.log.Info("tab created", "session", name, "tab", i)
402 d.startTabGoroutines(sess, tb)
403 }
404 return sess, nil
405}
406
407func closeTabs(tabs []*tab) {
408 for _, tb := range tabs {
409 tb.closeAllPanes()
410 }
411}
412
413func (d *Daemon) createSessionAndSwitch(from *session, ac *attachedClient, name string) error {
414 if name == "" {
415 return errSessionNameRequired
416 }
417 if err := domain.ValidateSessionName(name); err != nil {
418 return err
419 }
420 sz := ac.sizeSnapshot()
421 d.mu.Lock()
422 if d.closing {
423 d.mu.Unlock()
424 return errors.New("daemon is shutting down")
425 }
426 if d.nameLiveOrStoppedLocked(name) {
427 d.mu.Unlock()
428 return errSessionNameInUse
429 }
430 from.mu.Lock()
431 cwd := from.cwd
432 term := from.terminal
433 env := copyEnvironment(from.env)
434 _, attached := from.attachments[ac]
435 from.mu.Unlock()
436 if !attached {
437 d.mu.Unlock()
438 return errors.New("client detached")
439 }
440
441 newSess, err := d.createSessionLockedWithMode(name, false, cwd, sz, term, env)
442 d.mu.Unlock()
443 if err != nil {
444 return err
445 }
446 // The newly created target is registered before the attachment transition, but
447 // remains headless. Freeze/drain must occur with no daemon/routing/session
448 // lock held; the transition revalidates both lifecycles before publication.
449 transition, err := d.transitionAttachment(attachmentTransitionRequest{
450 source: from,
451 target: newSess,
452 next: ac,
453
454 expectedTransport: ac.transportSnapshot(),
455 ready: true,
456 })
457 if err != nil {
458 _ = d.killSession(newSess, ports.ReasonSessionKilled, true)
459 return err
460 }
461
462 d.touchMRU(newSess)
463 d.log.Info("client attached", "session", newSess.name, "resume", ac.resumeCapable)
464 d.deferAttachmentTransitionCleanups(transition)
465 d.firstPaintForTransition(transition.published)
466 return nil
467}
468
469// createSessionAndSwitchForAttachment creates no target until the initiating role
470// ticket has ended and the exact source capability is frozen and revalidated.
471func (d *Daemon) createSessionAndSwitchForAttachment(token attachmentConnectionToken, name string) error {
472 if name == "" {
473 return errSessionNameRequired
474 }
475 if err := domain.ValidateSessionName(name); err != nil {
476 return err
477 }
478 if token.sess == nil || token.ac == nil || token.effect == nil {
479 return errAttachmentTransition
480 }
481 token.effect.bindActionEnd(d, "create-session")
482 token.effect.End()
483
484 var created *session
485 transition, err := d.transitionAttachment(attachmentTransitionRequest{
486 source: token.sess, next: token.ac,
487
488 expectedTransport: token.transport, sourceToken: &token, action: "create-session",
489 copySourceEnvironment: true, ready: true,
490 createTargetLocked: func() (*session, error) {
491 source := token.sess
492 if source == nil {
493 return nil, errAttachmentTransition
494 }
495 if d.closing {
496 return nil, errors.New("daemon is shutting down")
497 }
498 if d.nameLiveOrStoppedLocked(name) {
499 return nil, errSessionNameInUse
500 }
501 source.mu.Lock()
502 cwd, term, env := source.cwd, source.terminal, copyEnvironment(source.env)
503 source.mu.Unlock()
504 var createErr error
505 created, createErr = d.createSessionLockedWithMode(name, false, cwd, token.ac.sizeSnapshot(), term, env)
506 return created, createErr
507 },
508 })
509 if err != nil {
510 if created != nil {
511 _ = d.killSession(created, ports.ReasonSessionKilled, true)
512 }
513 return err
514 }
515
516 d.touchMRU(created)
517 d.log.Info("client attached", "session", created.name, "resume", token.ac.resumeCapable)
518 d.deferAttachmentTransitionCleanups(transition)
519 d.firstPaintForTransition(transition.published)
520 return nil
521}
522
523func (d *Daemon) createTab(sess *session, sz domain.Size) error {
524 return d.createTabForAttachment(sess, nil, sz)
525}
526
527func (d *Daemon) createTabForAttachment(sess *session, ac *attachedClient, _ domain.Size) error {
528 sess.mu.Lock()
529 name := sess.name
530 cwd := sess.cwd
531 term := sess.terminal
532 env := copyEnvironment(sess.env)
533 attachments := sess.snapshotAttachmentsLocked()
534 sess.mu.Unlock()
535 tbSize := contentSize(sess.fullViewportSize())
536 tabStableID, paneStableID, err := d.newTabPaneStableIDs()
537 if err != nil {
538 return err
539 }
540 command, args := d.ptyCommand(env)
541 lifetime := d.newPaneProcessLifetime(sess.ctx)
542 pty, err := d.ptys.Open(lifetime.ctx, command, args, childEnvFrom(env, name, tabStableID, paneStableID, term), cwd, tbSize)
543 if err != nil {
544 lifetime.abort()
545 if pty != nil {
546 _ = pty.Close()
547 }
548 d.log.Warn("pty spawn failed", "err", err, "session", name, "kind", "tab")
549 return domain.UserErr(domain.NoticeTabSpawn, "couldn't open tab: shell failed to start", err)
550 }
551 tb := newTabWithStableID(tabStableID, paneStableID, pty, tbSize)
552 themeClient := ac
553 if themeClient == nil && len(attachments) != 0 {
554 themeClient = attachments[0]
555 }
556 if themeClient != nil {
557 t := d.effectiveTheme(themeClient.getClientTheme())
558 tb.mu.Lock()
559 p := tb.focusedPane()
560 tb.mu.Unlock()
561 if p != nil {
562 p.mu.Lock()
563 applyPaneThemeLocked(p, t, false)
564 p.mu.Unlock()
565 }
566 }
567 d.mu.Lock()
568 if d.closing || d.sessions[sess.id] != sess || sess.ctx.Err() != nil {
569 d.mu.Unlock()
570 lifetime.abort()
571 _ = pty.Close()
572 return errors.New("daemon: session closed")
573 }
574 tb.ctx, tb.cancel = context.WithCancel(sess.ctx)
575 sess.mu.Lock()
576 p := tb.focusedPane()
577 if !lifetime.publish(p) {
578 sess.mu.Unlock()
579 d.mu.Unlock()
580 lifetime.abort()
581 _ = pty.Close()
582 return errors.New("daemon: session closed")
583 }
584 // Publish ownership before appending the tab to the visible session.
585 publishPaneOwner(p, sess, tb, 0)
586 sess.tabs = append(sess.tabs, tb)
587 tabIndex := len(sess.tabs) - 1
588 if !sess.ephemeral {
589 d.markCatalogueDirty(sess.persistRecordLocked(max(d.nowUnixNano(), sess.createdAt, int64(1))).MetadataUpdate())
590 }
591 if ac != nil {
592 sess.activateAttachmentViewLocked(ac, tabIndex)
593 }
594 sess.mu.Unlock()
595 d.mu.Unlock()
596 d.log.Info("tab created", "session", name, "tab", tabIndex)
597 d.startTabGoroutines(sess, tb)
598 d.activateTab(sess, tb)
599 markSnapshotDirty(sess)
600 return nil
601}
602
603func newTab(pty ports.PTY, sz domain.Size) *tab {
604 return newTabWithStableID(fallbackStableID("t"), fallbackStableID("p"), pty, sz)
605}
606
607func newTabWithStableID(tabStableID, paneStableID string, pty ports.PTY, sz domain.Size) *tab {
608 id := layout.PaneID("pane-1")
609 p := newPaneWithStableID(id, paneStableID, pty, sz)
610 return &tab{
611 stableID: tabStableID,
612 tree: layout.NewTree(id),
613 panes: map[layout.PaneID]*pane{id: p},
614 nextPaneID: 2,
615 size: sz,
616 }
617}
618
619func (tb *tab) bumpLayoutGenerationLocked() {
620 tb.layoutGeneration++
621}
622
623func (tb *tab) focusedPane() *pane {
624 if tb == nil || tb.tree == nil || tb.panes == nil {
625 return nil
626 }
627 return tb.panes[tb.tree.Focus]
628}
629
630func (tb *tab) panesSnapshot() []*pane {
631 if tb == nil {
632 return nil
633 }
634 out := make([]*pane, 0, len(tb.panes))
635 for _, p := range tb.panes {
636 out = append(out, p)
637 }
638 return out
639}
640
641func (tb *tab) closeAllPanes() {
642 tb.mu.Lock()
643 panes := tb.panesSnapshot()
644 // Invalidate before cancellation so a concurrently unblocked floating
645 // reader cannot reap a subsequently reused slot.
646 floating := tb.takeFloatingLocked()
647 tb.mu.Unlock()
648 if floating != nil {
649 panes = append(panes, floating)
650 }
651 for _, p := range panes {
652 closePaneProcess(p)
653 }
654}
655
656// tabChromeRows is the vertical chrome (tab bar + status bar) between a
657// client viewport and a tab's content area.
658const tabChromeRows = 2
659
660func tabSize(clientSize domain.Size) domain.Size {
661 return contentSize(clientSize)
662}
663
664// contentSize maps the negotiated viewport to PTY geometry.
665func contentSize(clientSize domain.Size) domain.Size {
666 if !clientSize.Valid() {
667 clientSize = defaultSize
668 }
669 rows := max(clientSize.Rows-tabChromeRows, 1)
670 return domain.Size{Cols: clientSize.Cols, Rows: rows}
671}
672
673// fullViewportSize derives a full client-equivalent viewport from the active
674// tab's retained content size, for actions that need a viewport while no
675// client is attached. It falls back to defaultSize when no valid size exists.
676func (s *session) fullViewportSize() domain.Size {
677 s.mu.Lock()
678 var tb *tab
679 if len(s.tabs) != 0 {
680 tb = s.tabs[0]
681 }
682 s.mu.Unlock()
683
684 var content domain.Size
685 if tb != nil {
686 tb.mu.Lock()
687 content = tb.size
688 tb.mu.Unlock()
689 }
690 if !content.Valid() {
691 return defaultSize
692 }
693 return domain.Size{Cols: content.Cols, Rows: content.Rows + tabChromeRows}
694}
695
696func (d *Daemon) startTabGoroutines(sess *session, tb *tab) {
697 tb.mu.Lock()
698 panes := tb.panesSnapshot()
699 tb.mu.Unlock()
700 for _, p := range panes {
701 d.startPaneGoroutines(sess, tb, p)
702 }
703}
704
705func (d *Daemon) startPaneGoroutines(sess *session, tb *tab, p *pane) {
706 if p == nil {
707 return
708 }
709 owner := p.ownerSnapshot()
710 if owner == nil {
711 // Teardown may win after publication but before reader startup. In that
712 // case the process is already closed and no ownerless reader may start.
713 return
714 }
715 if owner.session != sess || owner.tab != tb {
716 return
717 }
718 sess.mu.Lock()
719 name := sess.name
720 sess.mu.Unlock()
721 d.log.Info("pane created", "session", name, "pane", p.id)
722 // Scheduler ownership was removed; this launch creates exactly one reader.
723 d.sessWg.Add(1)
724 go d.readPanePTY(p)
725}
726
727func (s *session) detachIfCurrent(ac *attachedClient) bool {
728 s.mu.Lock()
729 defer s.mu.Unlock()
730 if !attachmentRegisteredLocked(s, ac) || ac.currentAttachmentSession() != s {
731 return false
732 }
733 s.unregisterAttachmentLocked(ac)
734 ac.setSession(nil)
735 return true
736}
737
738// detachIfCurrent publishes terminal attachment invalidation through the attachment
739// gate. Freeze/drain happens before routing or session locks are acquired.
740func (d *Daemon) detachIfCurrent(sess *session, ac *attachedClient) bool {
741 return d.detachIfCurrentTransport(sess, ac, transportSnapshot{})
742}
743
744// detachIfCurrentTransport is detachIfCurrent with an optional exact-link
745// fence. When expected.transport is non-nil, a rebound incarnation of the same
746// attachment object cannot be detached by a stale connection loop.
747func (d *Daemon) detachIfCurrentTransport(sess *session, ac *attachedClient, expected transportSnapshot) bool {
748 if sess == nil || ac == nil {
749 return false
750 }
751 frozen := freezeAttachmentEffectGates(ac)
752 defer frozen.unfreeze()
753 if d.afterDetachAttachmentEffectsFrozen != nil {
754 d.afterDetachAttachmentEffectsFrozen()
755 }
756 d.notices.routingMu.Lock()
757 sess.mu.Lock()
758 current := attachmentRegisteredLocked(sess, ac) && ac.currentAttachmentSession() == sess
759 if current && expected.transport != nil && !ac.transportSnapshotCurrent(expected) {
760 current = false
761 }
762 if current {
763 sess.unregisterAttachmentLocked(ac)
764 ac.connectionGeneration.Add(1)
765 ac.setSession(nil)
766 ac.invalidateFrozenAttachmentCapability()
767 }
768 sess.mu.Unlock()
769 d.notices.routingMu.Unlock()
770 return current
771}
772
773// detachIfAttachmentCurrent invalidates only the exact attachment generation and transport
774// whose admitted send failed. A queued stale error cannot detach a later
775// attachment publication or a rebound transport on the same attachment object.
776func (d *Daemon) detachIfAttachmentCurrent(token attachmentConnectionToken) bool {
777 return d.detachIfAttachmentCurrentUntil(token, nil)
778}
779
780func (d *Daemon) detachIfAttachmentCurrentUntil(token attachmentConnectionToken, done func() <-chan struct{}) bool {
781 if token.sess == nil || token.ac == nil || token.transport.transport == nil {
782 return false
783 }
784 // Attachments without the session's optional primary render lease still
785 // own an independent connection lifecycle and detach through the same exact
786 // transport/generation fence.
787 if token.lease == nil {
788 return d.detachIfCurrentTransport(token.sess, token.ac, token.transport)
789 }
790 frozen := freezeAttachmentEffectGatesWith(attachmentEffectFreezeOptions{done: done}, token.ac)
791 defer frozen.unfreeze()
792 if !frozen.acquired || !frozen.drained {
793 return false
794 }
795 if d.afterDetachAttachmentEffectsFrozen != nil {
796 d.afterDetachAttachmentEffectsFrozen()
797 }
798 d.mu.Lock()
799 core := token.sess.core()
800 if core == nil || d.sessions[core.id] != token.sess {
801 d.mu.Unlock()
802 return false
803 }
804 d.notices.routingMu.Lock()
805 core.mu.Lock()
806 coordinator := core.coordinator.Load()
807 if coordinator != nil {
808 coordinator.mu.Lock()
809 }
810 req := attachmentTransitionRequest{
811 source: token.sess, next: token.ac, expectedTransport: token.transport,
812 }
813 current := coordinator != nil && transitionSourceTokenCurrentLocked(token, token.sess, coordinator, req)
814 if current {
815 unregisterAttachmentSessionLocked(token.sess, token.ac)
816 token.ac.connectionGeneration.Add(1)
817 token.ac.setSession(nil)
818 token.ac.invalidateFrozenAttachmentCapability()
819 }
820 if coordinator != nil {
821 coordinator.mu.Unlock()
822 }
823 core.mu.Unlock()
824 d.notices.routingMu.Unlock()
825 d.mu.Unlock()
826 return current
827}
828
829func (d *Daemon) renameSession(sess *session, name string) error {
830 if name == "" {
831 return errSessionNameRequired
832 }
833 if err := domain.ValidateSessionName(name); err != nil {
834 return err
835 }
836
837 // Hold daemon then session state across the durable identity write. The
838 // incarnation-keyed snapshot namespace is never relocated.
839 d.mu.Lock()
840 if taken := d.findByNameLocked(name); taken != nil && taken != sess {
841 d.mu.Unlock()
842 return errSessionNameInUse
843 }
844 sess.mu.Lock()
845 if stopped, ok := d.stopped[name]; ok && stopped.name != sess.name {
846 sess.mu.Unlock()
847 d.mu.Unlock()
848 return errSessionNameInUse
849 }
850 oldName := sess.name
851 wasEphemeral := sess.ephemeral
852 createdAt := sess.createdAt
853 priorIncarnation := sess.incarnation
854 if wasEphemeral {
855 var err error
856 createdAt, err = d.allocateLifecycleCreatedAtLocked()
857 if err != nil {
858 sess.mu.Unlock()
859 d.mu.Unlock()
860 return err
861 }
862 if sess.incarnation == (domain.IncarnationID{}) {
863 sess.incarnation, err = domain.NewSessionLifecycleID(rand.Reader)
864 if err != nil {
865 sess.mu.Unlock()
866 d.mu.Unlock()
867 return fmt.Errorf("generate durable identity: %w", err)
868 }
869 }
870 }
871 lastUsedSeq := sess.mruAt.Load()
872 record := sess.persistRecordLocked(max(d.nowUnixNano(), sess.createdAt, int64(1)))
873 record.Name = name
874 record.CreatedAt = createdAt
875 record.LastUsedSeq = lastUsedSeq
876 rollback := func(err error) error {
877 sess.incarnation = priorIncarnation
878 sess.mu.Unlock()
879 d.mu.Unlock()
880 return err
881 }
882 // Durable named renames are one atomic catalogue batch. The incarnation-keyed
883 // snapshot namespace is intentionally untouched.
884 if (wasEphemeral || oldName != name) && d.persistEnabled {
885 if d.recovery == nil {
886 return rollback(errors.New("durable session authority is not configured"))
887 }
888 {
889 var committed domain.CatalogueRecord
890 var err error
891 if wasEphemeral {
892 committed, err = d.recovery.Create(d.serveCtx, record)
893 } else {
894 committed, err = d.recovery.Rename(d.serveCtx, oldName, name)
895 }
896 if err != nil {
897 return rollback(err)
898 }
899 record = committed
900 }
901 }
902
903 delete(d.stopped, oldName)
904 delete(d.stopped, name)
905 sess.name = name
906 sess.createdAt = createdAt
907 sess.incarnation = record.IncarnationID
908 sess.ephemeral = false
909 sess.snapEligible.Store(name != "")
910 sess.mu.Unlock()
911 d.mu.Unlock()
912 if wasEphemeral {
913 sess.snapshotMu.Lock()
914 sess.snapshotChunkCache = newSnapshotChunkCache(snapshotChunkCacheLimit)
915 sess.snapshotMu.Unlock()
916 }
917 markSnapshotDirty(sess)
918 return nil
919}
920
921func (d *Daemon) renameTab(sess *session, tb *tab, name string) error {
922 if tb == nil {
923 return errors.New("tab required")
924 }
925 d.mu.Lock()
926 if d.sessions[sess.id] != sess {
927 d.mu.Unlock()
928 return errors.New("daemon: session closed")
929 }
930 sess.mu.Lock()
931 if !slices.Contains(sess.tabs, tb) {
932 sess.mu.Unlock()
933 d.mu.Unlock()
934 return errors.New("tab not found")
935 }
936 tb.name = name
937 if !sess.ephemeral {
938 d.markCatalogueDirty(sess.persistRecordLocked(max(d.nowUnixNano(), sess.createdAt, int64(1))).MetadataUpdate())
939 }
940 sess.mu.Unlock()
941 d.mu.Unlock()
942 return nil
943}
944
945func tabRecordsFromTabs(tabs []*tab) []domain.CatalogueTabRecord {
946 if len(tabs) == 0 {
947 return nil
948 }
949 records := make([]domain.CatalogueTabRecord, len(tabs))
950 for i, tb := range tabs {
951 if tb == nil {
952 continue
953 }
954 records[i] = domain.CatalogueTabRecord{StableID: domain.TabStableID(tb.stableID), Name: tb.name}
955 }
956 return records
957}
958
959func (s *session) persistRecordLocked(updatedAt int64) domain.CatalogueRecord {
960 createdAt := s.createdAt
961 tabNames := make([]string, len(s.tabs))
962 lastCustom := -1
963 for i, tb := range s.tabs {
964 tabNames[i] = tb.name
965 if tb.name != "" {
966 lastCustom = i
967 }
968 }
969 if lastCustom == -1 {
970 tabNames = nil
971 } else {
972 tabNames = tabNames[:lastCustom+1]
973 }
974 return domain.CatalogueRecord{Name: s.name, IncarnationID: s.incarnation, Cwd: s.cwd, CreatedAt: createdAt, UpdatedAt: updatedAt, LastUsedSeq: s.mruAt.Load(), TabNames: tabNames, TabRecords: tabRecordsFromTabs(s.tabs)}
975}
976
977// closeTab is the public close boundary for callers that are not already
978// inside session.runMutation. Mutation dispatchers use closeTabLocked.
979func (d *Daemon) closeTab(sess *session, tb *tab, repaint bool) error {
980 if sess == nil {
981 return layout.ErrNotFound
982 }
983 sess.dispatchMu.Lock()
984 defer sess.dispatchMu.Unlock()
985 return d.closeTabLocked(sess, tb, repaint)
986}
987
988// closeTabLocked requires sess.dispatchMu before entering daemon or session
989// locks. Existing runMutation callers use it to avoid nested dispatch locking.
990func (d *Daemon) closeTabLocked(sess *session, tb *tab, repaint bool) error {
991 return d.closeTabLockedWithEffect(sess, tb, repaint, nil)
992}
993
994func (d *Daemon) closeTabLockedWithEffect(sess *session, tb *tab, repaint bool, effect *attachmentEffectTicket) error {
995 if sess == nil || tb == nil {
996 return layout.ErrNotFound
997 }
998 d.mu.Lock()
999 if d.sessions[sess.id] != sess {
1000 d.mu.Unlock()
1001 return errors.New("daemon: session closed")
1002 }
1003 sess.mu.Lock()
1004 idx := -1
1005 for i, w := range sess.tabs {
1006 if w == tb {
1007 idx = i
1008 break
1009 }
1010 }
1011 if idx == -1 {
1012 sess.mu.Unlock()
1013 d.mu.Unlock()
1014 return errors.New("tab not found")
1015 }
1016 if len(sess.tabs) == 1 {
1017 name := sess.name
1018 sess.mu.Unlock()
1019 d.mu.Unlock()
1020 var err error
1021 if effect != nil {
1022 effect.bindActionEnd(d, "close-tab")
1023 err = d.killSessionForAttachment(sess, ports.ReasonSessionKilled, false, effect.connectionToken(), "close-tab")
1024 } else {
1025 err = d.killSession(sess, ports.ReasonSessionKilled, false)
1026 }
1027 if err != nil {
1028 d.log.Warn("closing last tab failed", "session", name, "err", err)
1029 d.reportError(sess, domain.UserErr(domain.NoticeSnapshotSaturated,
1030 "couldn't close tab: session state not yet saved; try again", err))
1031 return err
1032 }
1033 d.log.Info("tab closed", "session", name, "last", true)
1034 return nil
1035 }
1036 ringing := tb.attention
1037 sess.prepareAttachmentViewsForRemovedTabLocked(tb, idx)
1038 sess.tabs = append(sess.tabs[:idx], sess.tabs[idx+1:]...)
1039 attachments := sess.snapshotAttachmentsLocked()
1040 name := sess.name
1041 if !sess.ephemeral {
1042 d.markCatalogueDirty(sess.persistRecordLocked(max(d.nowUnixNano(), sess.createdAt, int64(1))).MetadataUpdate())
1043 }
1044 sess.invalidateViewsLocked()
1045 sess.mu.Unlock()
1046 d.mu.Unlock()
1047 d.log.Info("tab closed", "session", name)
1048 markSnapshotDirty(sess)
1049
1050 d.clearDestroyedTabPreview(tb)
1051 tb.mu.Lock()
1052 panes := tb.panesSnapshot()
1053 tb.mu.Unlock()
1054 for _, ac := range attachments {
1055 ac.pruneCaptureFrames(panes...)
1056 }
1057 if tb.cancel != nil {
1058 tb.cancel()
1059 }
1060 for _, ac := range attachments {
1061 d.teardownFloating(tb, ac)
1062 }
1063 if rc := sess.renderCoordinator(); rc != nil {
1064 for _, p := range panes {
1065 rc.noteSyncPaneRemoved(p)
1066 }
1067 }
1068 tb.closeAllPanes()
1069 if repaint {
1070 for _, ac := range attachments {
1071 d.invalidateRender(sess, ac, true, "session.go")
1072 }
1073 }
1074 if ringing {
1075 d.repaintAllAttachedClients()
1076 }
1077 return nil
1078}
1079
1080// ptyReader drains child output into the VT screen and pokes the dirty channel
1081// (non-blocking: a full channel already means a render is pending). On any read
1082
1083// beginSnapshotPurge commits a logical named-session kill before its live
1084// identity or persisted metadata is removed. It never runs under daemon or
1085// session locks.
1086func (d *Daemon) beginSnapshotPurge(_ string, _ domain.IncarnationID) error {
1087 if d.persistEnabled && d.recovery == nil {
1088 return errors.New("durable session authority is not configured")
1089 }
1090 // The lifecycle coordinator performs the complete ordered protocol after
1091 // snapshot writers have joined.
1092 return nil
1093}
1094
1095// finishSnapshotPurge removes catalogue metadata first, then deletes the
1096// incarnation directory. Startup garbage collection removes the directory if
1097// the second step is interrupted.
1098func (d *Daemon) finishSnapshotPurge(ctx context.Context, name string, _ domain.IncarnationID) error {
1099 if !d.persistEnabled {
1100 return nil
1101 }
1102 if d.recovery == nil {
1103 return errors.New("durable session authority is not configured")
1104 }
1105 if ctx == nil {
1106 ctx = d.serveCtx
1107 }
1108 purgeCtx, cancel := context.WithTimeout(ctx, snapshotFinalFlushTimeout)
1109 defer cancel()
1110 return d.recovery.Delete(purgeCtx, name)
1111}
1112
1113// retryStoppedPurge completes a previously closed session's durable purge.
1114// It keeps the stopped record hidden until all sources and metadata cross their
1115// deletion boundaries.
1116func (d *Daemon) retryStoppedPurge(name string) error {
1117 return d.retryStoppedPurgeContext(d.serveCtx, name)
1118}
1119
1120func (d *Daemon) retryStoppedPurgeContext(ctx context.Context, name string) error {
1121 d.mu.Lock()
1122 stopped, ok := d.stopped[name]
1123 if !ok {
1124 d.mu.Unlock()
1125 return nil
1126 }
1127 stopped.purging = true
1128 d.stopped[name] = stopped
1129
1130 if err := d.beginSnapshotPurge(name, stopped.incarnation); err != nil {
1131 d.mu.Unlock()
1132 return err
1133 }
1134 if err := d.finishSnapshotPurge(ctx, name, stopped.incarnation); err != nil {
1135 d.mu.Unlock()
1136 return err
1137 }
1138 if current, ok := d.stopped[name]; ok && current.purging {
1139 delete(d.stopped, name)
1140 }
1141 d.mu.Unlock()
1142 return nil
1143}
1144
1145func (d *Daemon) killSession(sess *session, reason uint8, purge bool) error {
1146 return d.killSessionWithSnapshotDeadline(sess, reason, purge, nil, nil)
1147}
1148
1149type sessionKillAdmission struct {
1150 token attachmentConnectionToken
1151 action string
1152}
1153
1154type sessionKillParticipants struct {
1155 attachments map[*attachedClient]struct{}
1156 transports map[*attachedClient]transportSnapshot
1157 effectGates []*attachedClient
1158 interrupts []attachmentTransportInterrupt
1159 source *session
1160 sourceToken *attachmentConnectionToken
1161 target *session
1162}
1163
1164func (d *Daemon) killSessionForAttachment(sess *session, reason uint8, purge bool, token attachmentConnectionToken, action string) error {
1165 return d.killSessionWithSnapshotDeadline(sess, reason, purge, nil, &sessionKillAdmission{token: token, action: action})
1166}
1167
1168// snapshotSessionKillParticipants captures attachment membership under the canonical
1169// architecture locks, but releases every lock before any attachment gate is frozen or
1170// drained. The initiator is part of the same deduplicated global gate set even
1171// when it belongs to a different source session.
1172func (d *Daemon) snapshotSessionKillParticipants(target *session, admission *sessionKillAdmission) (sessionKillParticipants, bool) {
1173 snapshot := sessionKillParticipants{target: target}
1174 if admission != nil {
1175 token := admission.token
1176 if token.sess == nil || token.ac == nil || token.transport.transport == nil {
1177 return sessionKillParticipants{}, false
1178 }
1179 snapshot.source = token.sess
1180 if snapshot.source == nil {
1181 return sessionKillParticipants{}, false
1182 }
1183 snapshot.sourceToken = &admission.token
1184 }
1185
1186 d.mu.Lock()
1187 if d.sessions[target.id] != target || snapshot.source != nil && d.sessions[snapshot.source.id] != snapshot.source {
1188 d.mu.Unlock()
1189 return sessionKillParticipants{}, false
1190 }
1191 d.notices.routingMu.Lock()
1192 source := snapshot.source
1193 if source == nil {
1194 source = target
1195 }
1196 unlockSessions := lockAttachmentSessions(source, target)
1197 snapshot.attachments = make(map[*attachedClient]struct{}, len(target.attachments))
1198 snapshot.transports = make(map[*attachedClient]transportSnapshot, len(target.attachments))
1199 for ac := range target.attachments {
1200 snapshot.attachments[ac] = struct{}{}
1201 snapshot.transports[ac] = ac.transportSnapshot()
1202 }
1203
1204 unlockSessions()
1205 d.notices.routingMu.Unlock()
1206 d.mu.Unlock()
1207
1208 seen := make(map[*attachedClient]struct{}, 1+len(snapshot.attachments))
1209 add := func(ac *attachedClient) {
1210 if ac == nil {
1211 return
1212 }
1213 if _, duplicate := seen[ac]; duplicate {
1214 return
1215 }
1216 seen[ac] = struct{}{}
1217 snapshot.effectGates = append(snapshot.effectGates, ac)
1218 }
1219 if admission != nil {
1220 add(admission.token.ac)
1221 }
1222 for ac := range snapshot.attachments {
1223 add(ac)
1224 }
1225 for ac, transport := range snapshot.transports {
1226 if transport.transport != nil {
1227 snapshot.interrupts = append(snapshot.interrupts, attachmentTransportInterrupt{ac: ac, transport: transport})
1228 }
1229 }
1230 return snapshot, true
1231}
1232
1233// sessionKillParticipantsCurrent repeats lifecycle, source-token, and exact
1234// target-membership validation while every discovered attachment gate is frozen.
1235func (d *Daemon) sessionKillParticipantsCurrent(snapshot sessionKillParticipants) bool {
1236 d.mu.Lock()
1237 defer d.mu.Unlock()
1238 if d.sessions[snapshot.target.id] != snapshot.target ||
1239 snapshot.source != nil && d.sessions[snapshot.source.id] != snapshot.source {
1240 return false
1241 }
1242 d.notices.routingMu.Lock()
1243 defer d.notices.routingMu.Unlock()
1244 source := snapshot.source
1245 if source == nil {
1246 source = snapshot.target
1247 }
1248 sourceCoordinator := source.renderCoordinator()
1249 targetCoordinator := snapshot.target.renderCoordinator()
1250 unlockSessions := lockAttachmentSessions(source, snapshot.target)
1251 defer unlockSessions()
1252 unlockCoordinators := lockAttachmentCoordinators(source, sourceCoordinator, snapshot.target, targetCoordinator)
1253 defer unlockCoordinators()
1254 return sessionKillParticipantsCurrentLocked(snapshot, sourceCoordinator)
1255}
1256
1257// sessionKillParticipantsCurrentLocked requires the ordered source/target
1258// session and coordinator locks.
1259func sessionKillParticipantsCurrentLocked(snapshot sessionKillParticipants, sourceCoordinator *renderCoordinator) bool {
1260 target := snapshot.target
1261 if len(target.attachments) != len(snapshot.attachments) {
1262 return false
1263 }
1264 for ac := range snapshot.attachments {
1265 if _, current := target.attachments[ac]; !current {
1266 return false
1267 }
1268 }
1269
1270 for ac, transport := range snapshot.transports {
1271 if !ac.transportSnapshotCurrent(transport) {
1272 return false
1273 }
1274 }
1275 if snapshot.sourceToken == nil {
1276 return true
1277 }
1278 token := *snapshot.sourceToken
1279 req := attachmentTransitionRequest{
1280 source: token.sess, next: token.ac, expectedTransport: token.transport,
1281 }
1282 return transitionSourceTokenCurrentLocked(token, token.sess, sourceCoordinator, req)
1283}
1284
1285// stopInMemoryLifecycle cancels every session producer and closes every PTY
1286// independently of teardown ownership. Repository work may remain blocked, but
1287// repeated shutdown passes cannot leave pane readers or launches running.
1288func (s *session) stopInMemoryLifecycle() {
1289 if s == nil {
1290 return
1291 }
1292 s.lifecycleStopOnce.Do(func() {
1293 s.stopFloatingLaunches()
1294 if s.cancel != nil {
1295 s.cancel()
1296 }
1297 s.mu.Lock()
1298 tabs := append([]*tab(nil), s.tabs...)
1299 s.mu.Unlock()
1300 for _, tb := range tabs {
1301 tb.closeAllPanes()
1302 }
1303 })
1304}
1305
1306func (s *session) teardownChangeLocked() chan struct{} {
1307 if s.teardownChanged == nil {
1308 s.teardownChanged = make(chan struct{})
1309 }
1310 return s.teardownChanged
1311}
1312
1313func (s *session) signalTeardownChangedLocked() {
1314 close(s.teardownChangeLocked())
1315 s.teardownChanged = make(chan struct{})
1316}
1317
1318// beginTeardown waits for ownership without holding teardownMu across teardown.
1319// A false result means Serve's shared shutdown budget expired while another
1320// lifecycle path owned the session. Ordinary callers wait and retry ownership,
1321// preserving failed-teardown and stopped-purge retry semantics.
1322func (s *session) beginTeardown(deadline *snapshotShutdownDeadline) bool {
1323 for {
1324 if deadline != nil {
1325 select {
1326 case <-deadline.Done():
1327 return false
1328 default:
1329 }
1330 }
1331
1332 s.teardownMu.Lock()
1333 if !s.teardownActive && s.moveReservations == 0 {
1334 if deadline != nil {
1335 select {
1336 case <-deadline.Done():
1337 s.teardownMu.Unlock()
1338 return false
1339 default:
1340 }
1341 }
1342 s.teardownActive = true
1343 s.teardownDone = make(chan struct{})
1344 s.signalTeardownChangedLocked()
1345 s.teardownMu.Unlock()
1346 return true
1347 }
1348 s.teardownWaiters++
1349 s.signalTeardownChangedLocked()
1350 changed := s.teardownChangeLocked()
1351 s.teardownMu.Unlock()
1352
1353 timedOut := false
1354 if deadline == nil {
1355 <-changed
1356 } else {
1357 select {
1358 case <-changed:
1359 case <-deadline.Done():
1360 timedOut = true
1361 }
1362 }
1363 s.teardownMu.Lock()
1364 s.teardownWaiters--
1365 s.signalTeardownChangedLocked()
1366 s.teardownMu.Unlock()
1367 if timedOut {
1368 return false
1369 }
1370 }
1371}
1372
1373func (s *session) finishTeardown() {
1374 s.teardownMu.Lock()
1375 s.teardownActive = false
1376 close(s.teardownDone)
1377 s.teardownDone = nil
1378 s.signalTeardownChangedLocked()
1379 s.teardownMu.Unlock()
1380}
1381
1382// killSessionWithSnapshotDeadline shares Serve's shutdown budget with its
1383// terminal checkpoint and coordinator join. A timed-out repository call keeps
1384// only immutable capture state; it never observes closed worker channels or
1385// session-owned cache that teardown has released.
1386func (d *Daemon) killSessionWithSnapshotDeadline(sess *session, reason uint8, purge bool, deadline *snapshotShutdownDeadline, admission *sessionKillAdmission) error {
1387 if sess == nil {
1388 return nil
1389 }
1390 // Discover every attachment whose membership can be invalidated before freezing
1391 // any one of them. The common gate helper deduplicates and freezes the whole
1392 // immutable identity set in the same global order used by transitions.
1393 participants, ok := d.snapshotSessionKillParticipants(sess, admission)
1394 if !ok {
1395 return nil
1396 }
1397 if admission != nil {
1398 d.endActionAttachmentEffect(admission.token.effect, admission.action)
1399 }
1400 action := ""
1401 if admission != nil {
1402 action = admission.action
1403 }
1404 if d.afterAttachmentEffectParticipantsSnapshotted != nil {
1405 d.afterAttachmentEffectParticipantsSnapshotted(action, participants.effectGates)
1406 }
1407 var drainDone func() <-chan struct{}
1408 var drainDeadline *attachmentEffectDrainDeadline
1409 if deadline != nil {
1410 drainDone = deadline.Done
1411 } else {
1412 drainDeadline = newAttachmentEffectDrainDeadline(d.clock)
1413 drainDone = drainDeadline.Done
1414 defer drainDeadline.stop()
1415 }
1416 frozen := freezeAttachmentEffectGatesWith(attachmentEffectFreezeOptions{interrupts: participants.interrupts, done: drainDone, afterFrozen: func(ac *attachedClient) {
1417 if d.afterAttachmentEffectGateFrozen != nil {
1418 d.afterAttachmentEffectGateFrozen(action, ac)
1419 }
1420 }}, participants.effectGates...)
1421 defer func() { frozen.unfreeze() }()
1422 if !frozen.acquired {
1423 // The acquisition helper has already rolled back exactly the gates it
1424 // owned. Do not enter architecture locks or publish terminal ownership
1425 // changes unless the complete canonical participant set is frozen.
1426 return nil
1427 }
1428 if !d.sessionKillParticipantsCurrent(participants) {
1429 return nil
1430 }
1431
1432 // Pane EOF, explicit close, and daemon shutdown can all converge on the same
1433 // session. Only one owner may create the terminal mutation and checkpoint;
1434 // later callers observe the completed registry transition instead of
1435 // manufacturing a newer generation that the owner will quarantine. A
1436 // shutdown duplicate abandons this wait when the shared budget expires.
1437 if !sess.beginTeardown(deadline) {
1438 return nil
1439 }
1440 defer sess.finishTeardown()
1441
1442 d.mu.Lock()
1443 current := d.sessions[sess.id]
1444 d.mu.Unlock()
1445 if current != sess {
1446 if purge {
1447 sess.mu.Lock()
1448 name := sess.name
1449 sess.mu.Unlock()
1450 return d.retryStoppedPurge(name)
1451 }
1452 return nil
1453 }
1454
1455 ringing := sess.anyAttention()
1456 sess.mu.Lock()
1457 isEphemeral := sess.ephemeral
1458 name := sess.name
1459 incarnation := sess.incarnation
1460 sess.mu.Unlock()
1461 // Join snapshot publication before changing live identity or deleting the
1462 // incarnation directory, so an older publication cannot recreate it.
1463 if purge && !isEphemeral {
1464 if err := d.beginSnapshotPurge(name, incarnation); err != nil {
1465 return err
1466 }
1467 if d.snapshotRepository != nil {
1468 <-quarantineSnapshotCoordinator(sess)
1469 }
1470 }
1471 if !purge && !isEphemeral && d.persistEnabled {
1472 d.refreshSessionCwd(sess)
1473 }
1474 var terminalSnapshotErr error
1475 retainSnapshotRetry := false
1476 if d.snapsEnabled && !isEphemeral && !purge {
1477 markSnapshotDirty(sess)
1478 sess.snapshotMu.Lock()
1479 terminalGeneration := sess.snapshotGeneration
1480 sess.snapshotMu.Unlock()
1481 if !d.scheduleFinalSnapshot(sess) || !d.waitForSnapshotGenerationWithDeadline(sess, terminalGeneration, deadline) {
1482 sess.mu.Lock()
1483 name := sess.name
1484 sess.mu.Unlock()
1485 terminalSnapshotErr = fmt.Errorf("retain final snapshot for session %q: snapshot worker unavailable, saturated, or timed out", name)
1486 if reason != ports.ReasonServerShutdown {
1487 return terminalSnapshotErr
1488 }
1489 // Preserve any already-queued routine capture until the blocked
1490 // worker can safely discard it. This retains the dirty generation
1491 // and forced retry intent without adding another queue entry.
1492 retainSnapshotRetry = true
1493 d.persistShutdownSnapshotFailure(name, terminalSnapshotErr)
1494 }
1495 }
1496 if !isEphemeral {
1497 // Joining first preserves byte ownership for a publication that is still
1498 // encoding while this session is being torn down. Once Serve's shared
1499 // budget expires, the worker is cancelled by its one final join instead;
1500 // keep the cache reachable because an uncooperative writer may still be
1501 // encoding immutable state.
1502 var quarantineDone <-chan struct{}
1503 if retainSnapshotRetry {
1504 quarantineDone = quarantineSnapshotCoordinatorRetainingQueuedCapture(sess)
1505 } else {
1506 quarantineDone = quarantineSnapshotCoordinator(sess)
1507 }
1508 joined := true
1509 if deadline != nil {
1510 select {
1511 case <-quarantineDone:
1512 case <-deadline.Done():
1513 joined = false
1514 }
1515 } else {
1516 <-quarantineDone
1517 }
1518 if joined {
1519 sess.snapshotMu.Lock()
1520 sess.snapshotChunkCache = nil
1521 sess.snapshotMu.Unlock()
1522 }
1523 }
1524 // Revalidate the exact source capability, target lifecycle, and attachment
1525 // memberships under canonical architecture locks immediately before
1526 // publication. A changed preflight aborts without mutating either lifecycle.
1527 d.mu.Lock()
1528 if d.sessions[sess.id] != sess || participants.source != nil && d.sessions[participants.source.id] != participants.source {
1529 d.mu.Unlock()
1530 return nil
1531 }
1532 d.notices.routingMu.Lock()
1533 source := participants.source
1534 if source == nil {
1535 source = sess
1536 }
1537 sourceCoordinator := source.renderCoordinator()
1538 targetCoordinator := sess.renderCoordinator()
1539 unlockSessions := lockAttachmentSessions(source, sess)
1540 unlockCoordinators := lockAttachmentCoordinators(source, sourceCoordinator, sess, targetCoordinator)
1541 if !sessionKillParticipantsCurrentLocked(participants, sourceCoordinator) {
1542 unlockCoordinators()
1543 unlockSessions()
1544 d.notices.routingMu.Unlock()
1545 d.mu.Unlock()
1546 return nil
1547 }
1548
1549 // d.mu -> routingMu -> ordered sessions -> ordered coordinators are held
1550 // here. Registry removal and every attachment invalidation are one publication
1551 // while the complete immutable gate set remains frozen.
1552 if !d.unregisterSessionLocked(sess) {
1553 unlockCoordinators()
1554 unlockSessions()
1555 d.notices.routingMu.Unlock()
1556 d.mu.Unlock()
1557 return errAttachmentTransition
1558 }
1559 d.clearBarScriptsForSession(sess.id)
1560 d.purgeParkingForSessionLocked(sess)
1561 parkedRetirements := d.purgeParkedForSessionLocked(sess)
1562 attachments := make([]detachedAttachmentSnapshot, 0, len(participants.effectGates))
1563 capture := func(attached *attachedClient) {
1564 if attached == nil {
1565 return
1566 }
1567 attachments = append(attachments, detachedAttachmentSnapshot{ac: attached, transport: participants.transports[attached]})
1568 attached.connectionGeneration.Add(1)
1569 attached.setSession(nil)
1570 attached.invalidateFrozenAttachmentCapability()
1571 }
1572 for attached := range sess.attachments {
1573 capture(attached)
1574 }
1575 clear(sess.attachments)
1576 stoppedName := sess.name
1577 stoppedCwd := sess.cwd
1578 createdAt := sess.createdAt
1579 stoppedRecord := sess.persistRecordLocked(max(d.nowUnixNano(), sess.createdAt, int64(1)))
1580 tabNames := stoppedRecord.TabNames
1581 tabRecords := stoppedRecord.TabRecords
1582 ephemeral := sess.ephemeral
1583 unlockCoordinators()
1584 unlockSessions()
1585 d.notices.routingMu.Unlock()
1586 if !ephemeral {
1587 stopped := stoppedSession{name: stoppedName, cwd: stoppedCwd, createdAt: createdAt, incarnation: incarnation, lastUsedSeq: sess.mruAt.Load(), tabNames: tabNames, tabRecords: tabRecords, record: stoppedRecord, state: ports.SessionDown, purging: purge}
1588 d.stopped[stoppedName] = stopped
1589 }
1590 empty := len(d.sessions) == 0
1591 if empty {
1592 // Shutdown is now inevitable and irreversible (doneOnce below): stop
1593 // route from inserting new sessions from this instant, while we still
1594 // hold the same lock route checks under.
1595 d.closing = true
1596 }
1597 d.mu.Unlock()
1598 d.finishParkedAttachmentRetirements(parkedRetirements)
1599 frozen.unfreeze()
1600 frozen = frozenAttachmentEffectGates{}
1601 d.log.Info("session closed", "session", stoppedName, "id", sess.id, "ephemeral", ephemeral, "purge", purge, "reason", reason)
1602
1603 var purgeErr error
1604 if rc := sess.renderCoordinator(); rc != nil {
1605 // Terminal teardown has two phases: this session owner first prevents any
1606 // new worker registration and detaches all tokens, then stops and waits
1607 // outside coordinator callbacks and session locks.
1608 rc.beginSessionTeardown().finish()
1609 rc.waitForTimerWorkers()
1610 }
1611 for _, attachment := range attachments {
1612 d.unregisterPreview(attachment.ac)
1613 attachment.ac.clearCaptureFrames()
1614 }
1615
1616 // Prevent queued launches from entering Open, cancel the parent context, and
1617 // close PTYs. Serve may already have performed this phase while repository
1618 // work owned by this teardown was blocked.
1619 sess.stopInMemoryLifecycle()
1620 sess.mu.Lock()
1621 tabs := append([]*tab(nil), sess.tabs...)
1622 clipFiles := sess.clipFiles
1623 sess.clipFiles = nil
1624 sess.mu.Unlock()
1625 for _, tb := range tabs {
1626 d.clearDestroyedTabPreview(tb)
1627 for _, attachment := range attachments {
1628 d.teardownFloating(tb, attachment.ac)
1629 }
1630 }
1631 for _, path := range clipFiles {
1632 if err := os.Remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
1633 d.log.Warn("removing clipboard temp file failed", "err", err, "path", path)
1634 }
1635 }
1636 // The coordinator and all panes are now stopped, so no producer can publish
1637 // another generation after this destructive source deletion. Keep the hidden
1638 // stopped record when deletion fails so a repeated live kill can retry.
1639 if !ephemeral && purge {
1640 d.mu.Lock()
1641 sess.mu.Lock()
1642 err := d.finishSnapshotPurge(d.serveCtx, stoppedName, incarnation)
1643 sess.mu.Unlock()
1644 if err != nil {
1645 purgeErr = errors.Join(purgeErr, err)
1646 d.log.Warn("finishing session snapshot purge failed", "err", err, "session", stoppedName)
1647 } else if stopped, ok := d.stopped[stoppedName]; ok && stopped.purging {
1648 delete(d.stopped, stoppedName)
1649 }
1650 d.mu.Unlock()
1651 }
1652 if empty {
1653 d.doneOnce.Do(func() { close(d.done) })
1654 } else if ringing {
1655 d.repaintAllAttachedClients()
1656 }
1657
1658 for _, attachment := range attachments {
1659 d.notifyDetachedSnapshotAsync(attachment, reason)
1660 }
1661 return errors.Join(purgeErr, terminalSnapshotErr)
1662}
1663
1664// allocEphemeralNameLocked returns the lowest free decimal name. Caller holds
1665// d.mu.
1666func (d *Daemon) allocEphemeralNameLocked() string {
1667 used := make(map[string]struct{}, len(d.sessions)+len(d.stopped))
1668 for _, s := range d.sessions {
1669 if s != nil {
1670 used[s.name] = struct{}{}
1671 }
1672 }
1673 for name := range d.stopped {
1674 used[name] = struct{}{}
1675 }
1676 for i := 0; ; i++ {
1677 n := strconv.Itoa(i)
1678 if _, taken := used[n]; !taken {
1679 return n
1680 }
1681 }
1682}
1683
1684func (d *Daemon) findByNameLocked(name string) *session {
1685 for _, s := range d.sessions {
1686 if s != nil && s.name == name {
1687 return s
1688 }
1689 }
1690 return nil
1691}
1692
1693func (d *Daemon) nameLiveOrStoppedLocked(name string) bool {
1694 if d.findByNameLocked(name) != nil {
1695 return true
1696 }
1697 if _, ok := d.stopped[name]; ok {
1698 return true
1699 }
1700 _, reserved := d.creating[name]
1701 return reserved
1702}
1703
1704func (d *Daemon) cwdSampler(ctx context.Context) {
1705 t := d.clock.NewTimer(5 * time.Second)
1706 defer t.Stop()
1707 for {
1708 select {
1709 case <-ctx.Done():
1710 return
1711 case <-t.C():
1712 }
1713 if d.procCwd != nil {
1714 d.refreshNamedSessionCwds()
1715 }
1716 if err := d.flushCatalogue(); err != nil {
1717 d.log.Warn("flushing session catalogue failed", "err", err)
1718 }
1719 t.Reset(5 * time.Second)
1720 }
1721}
1722
1723func (d *Daemon) refreshNamedSessionCwds() {
1724 d.mu.Lock()
1725 sessions := d.sessionsSnapshotLocked()
1726 d.mu.Unlock()
1727 for _, sess := range sessions {
1728 sess.mu.Lock()
1729 ephemeral := sess.ephemeral
1730 sess.mu.Unlock()
1731 if !ephemeral {
1732 d.refreshSessionCwd(sess)
1733 }
1734 }
1735}
1736
1737func (d *Daemon) refreshSessionCwd(sess *session) {
1738 if d.procCwd == nil {
1739 return
1740 }
1741 tb := sess.firstTab()
1742 if tb == nil {
1743 return
1744 }
1745 tb.mu.Lock()
1746 p := tb.focusedPane()
1747 tb.mu.Unlock()
1748 if p == nil {
1749 return
1750 }
1751 cwd, err := d.procCwd(p.pty.Pid())
1752 if err != nil || cwd == "" {
1753 return
1754 }
1755 d.mu.Lock()
1756 if d.sessions[sess.id] != sess {
1757 d.mu.Unlock()
1758 return
1759 }
1760 sess.mu.Lock()
1761 if sess.cwd == cwd {
1762 sess.mu.Unlock()
1763 d.mu.Unlock()
1764 return
1765 }
1766 sess.cwd = cwd
1767 if !sess.ephemeral {
1768 d.markCatalogueDirty(sess.persistRecordLocked(max(d.nowUnixNano(), sess.createdAt, int64(1))).MetadataUpdate())
1769 }
1770 sess.mu.Unlock()
1771 d.mu.Unlock()
1772 markSnapshotDirty(sess)
1773}
1774
1775// childEnv retains the daemon-environment helper for daemon-local legacy callers.
1776// Interactive PTY launch paths use childEnvFrom with their session snapshot.
1777func (d *Daemon) childEnv(name, tabStableID, paneStableID string, term terminalEnv) []string {
1778 return childEnvFrom(d.baseEnv, name, tabStableID, paneStableID, term)
1779}
1780
1781func copyEnvironment(env []string) []string {
1782 return append([]string(nil), env...)
1783}
1784
1785func environmentEntry(entry string) (name, value string, ok bool) {
1786 name, value, ok = strings.Cut(entry, "=")
1787 return name, value, ok
1788}
1789
1790func terminalEnvFromEnvironment(env []string) terminalEnv {
1791 var term, colorTerm string
1792 for _, entry := range env {
1793 name, value, ok := environmentEntry(entry)
1794 if !ok {
1795 continue
1796 }
1797 switch name {
1798 case "TERM":
1799 term = value
1800 case "COLORTERM":
1801 colorTerm = value
1802 }
1803 }
1804 return terminalEnv{TrueColor: ports.DetectTrueColor(term, colorTerm)}
1805}
1806
1807func (d *Daemon) ptyCommand(env []string) (string, []string) {
1808 if d.shellOverride {
1809 return d.shell, append([]string(nil), d.shellArgs...)
1810 }
1811 return shellFromEnvironment(env), nil
1812}
1813
1814func shellFromEnvironment(env []string) string {
1815 shell := ""
1816 for _, entry := range env {
1817 if name, value, ok := environmentEntry(entry); ok && name == "SHELL" {
1818 shell = value
1819 }
1820 }
1821 if shell == "" {
1822 return "/bin/sh"
1823 }
1824 return shell
1825}
1826
1827// childEnvFrom preserves the supplied environment byte-for-byte and in order,
1828// except for exact reserved variable names which vev owns.
1829func childEnvFrom(env []string, name, tabStableID, paneStableID string, term terminalEnv) []string {
1830 out := make([]string, 0, len(env)+4)
1831 for _, entry := range env {
1832 key, _, ok := environmentEntry(entry)
1833 if ok && (key == "TERM" || key == "COLORTERM" || key == "TERM_PROGRAM" || key == "VEV") {
1834 continue
1835 }
1836 out = append(out, entry)
1837 }
1838 if term.TrueColor {
1839 out = append(out, "TERM=xterm-direct", "COLORTERM=truecolor")
1840 } else {
1841 out = append(out, "TERM=xterm-256color")
1842 }
1843 return append(out,
1844 "TERM_PROGRAM=vev",
1845 "VEV=session="+escapeVEVComponent(name)+",tab="+tabStableID+",pane="+paneStableID,
1846 )
1847}