1// Package daemon holds vev's server-side session multiplexer use case.
2package daemon
3
4import (
5 "context"
6 "errors"
7 "log/slog"
8 "math"
9 "os"
10 "reflect"
11 "sort"
12 "sync"
13 "sync/atomic"
14 "time"
15
16 "github.com/bnema/vev/internal/domain"
17 "github.com/bnema/vev/internal/ports"
18 "github.com/bnema/vev/internal/usecase/keys"
19 recoveryusecase "github.com/bnema/vev/internal/usecase/recovery"
20)
21
22// Scheduler debounce bounds. Idle updates use the minimum for low latency;
23// sustained floods step toward the maximum to reduce frame/syscall pressure.
24const (
25 minDebounceInterval = 2 * time.Millisecond
26 maxSyncUpdateDuration = 500 * time.Millisecond
27)
28
29const debounceInterval = minDebounceInterval
30
31// ptyReadBufSize is the PTY reader's read buffer.
32const ptyReadBufSize = 32 * 1024
33
34// New and restored panes are bounded by both history rows and cells. The
35// cell budget keeps a very wide terminal from retaining disproportionately
36// more scrollback than the normal 160-column case.
37const (
38 defaultScrollbackRows = 10_000
39 defaultScrollbackCells = 12_000 * 160
40)
41
42// detachNotifyTimeout bounds the best-effort Detached notification on the
43// detach/kill/shutdown paths: if a wedged client (full kernel send buffer)
44// blocks the write for this long, its transport is force-closed, which fails
45// the in-flight send. Teardown is never gated on a client draining its socket.
46const detachNotifyTimeout = time.Second
47
48// maxUnackedOutputStates caps how many output states may be in flight (sent but
49// not yet acked by the client) before paint defers rather than composing
50// another diff. It bounds the daemon's paint rate to the client's ack rate, so
51// heavy output degrades to lower fps on a slow link instead of overflowing the
52// transport. It must stay well under the datagram carriage's 32-frame client
53// window so its reliable queue never fills from painting alone.
54const maxUnackedOutputStates = 8
55
56// normalizeOutputWindow bounds the untrusted Hello value. Zero deliberately
57// means the legacy/default window, so malformed or absent values remain safe.
58func normalizeOutputWindow(window uint8) uint8 {
59 if window == 0 || window > maxUnackedOutputStates {
60 return maxUnackedOutputStates
61 }
62 return window
63}
64
65const defaultResumeParkGrace = 15 * time.Minute
66
67// defaultSize is retained for headless layout helpers that have no client
68// window; Hello routing rejects invalid dimensions instead of using it.
69var defaultSize = domain.Size{Cols: 80, Rows: 24}
70
71type Daemon struct {
72 mu sync.Mutex
73 sessions map[domain.SessionID]*session
74 stopped map[string]stoppedSession
75 // creating reserves names while durable creation I/O runs without mu.
76 creating map[string]struct{}
77 nextID uint64
78 // lastAllocatedCreatedAt is the named-session lifecycle timestamp high-water
79 // mark. It is guarded by mu and prevents a wall-clock regression from
80 // reusing a lifecycle identity.
81 lastAllocatedCreatedAt int64
82 mruSeq atomic.Uint64
83 // pickerSort is the picker ordering mode for this daemon's lifetime
84 // (pickerSortMode); not persisted across restarts.
85 pickerSort atomic.Uint32
86 // closing marks that shutdown has irreversibly begun. It is set under mu,
87 // atomically with the event that makes shutdown inevitable (the registry
88 // emptying in killSession, or shutdownAll starting), and checked by route
89 // under the same mutex — so a Hello racing shutdown can never insert a new
90 // session that nobody would tear down.
91 closing bool
92
93 // moveLifecycleMu is the daemon-level admission gate for transferable
94 // ownership changes. It is acquired before any session teardownMu, and is
95 // never held while a move, teardown, or external operation runs.
96 moveLifecycleMu sync.Mutex
97 moveLifecycleClosing bool
98 moveLifecycleActive uint
99 moveLifecycleChanged chan struct{}
100 // paneProcessCtx roots transferable pane processes outside any one session.
101 // Shutdown cancels it only after the global move gate drains.
102 paneProcessCtx context.Context
103 paneProcessCancel context.CancelFunc
104
105 // notifies holds one completion channel per in-flight async Detached
106 // notification (guarded by mu, pruned on insert). Channels rather than a
107 // WaitGroup: notifications are spawned from arbitrary goroutines while
108 // Serve may be waiting, and WaitGroup forbids Add-from-zero concurrent
109 // with Wait.
110 notifies []chan struct{}
111 parked map[uint64]*parkedAttachment
112 // parking tracks resume-capable attachments from before detach clears the
113 // live seat until parkAttachment publishes the token into parked.
114 // IntentResume waits on the matching entry instead of treating the live
115 // credential as unknown across that gap.
116 parking map[uint64]*parkingAttachment
117
118 attnMu sync.Mutex
119 animFrame int
120 animWake chan struct{}
121
122 paletteRecentMu sync.Mutex
123 paletteRecent []string
124 // beforeRecentSessionHandoff is a deterministic test seam for the narrow
125 // interval between JRS validation and its committed hand-off.
126 beforeRecentSessionHandoff func()
127 // beforeCopyModeRevalidate is a deterministic test seam between staging a
128 // copy-mode candidate and revalidating its pane membership.
129 beforeCopyModeRevalidate func()
130 // beforeCopyMouseMap is a deterministic seam after an immutable copy-input
131 // snapshot is captured and before its mapped position is applied.
132 beforeCopyMouseMap func()
133 // beforeSessionResizePublication runs after final external PTY validation
134 // and before coordinator epoch admission. It is a deterministic regression
135 // seam for stale resize publication.
136 beforeSessionResizePublication func()
137 // beforeResizeOwnerPostEffect pauses after a resize's optimistic owner check
138 // and immediately before a post-commit effect is published. Tests use it to
139 // move a pane through the ordered resize fences at the former TOCTOU window.
140 beforeResizeOwnerPostEffect func(resizeOwnerPostEffect)
141 // afterResizeCommitSendLocked is a deterministic lock-order seam. It runs
142 // after publishResizeCommit acquires attachment sendMu and before it acquires
143 // owner fences. Tests must not perform external work from this callback.
144 afterResizeCommitSendLocked func()
145 // afterConnectionSessionSnapshot is a deterministic test seam after the
146 // connection loop reads its current session and before it captures that role.
147 afterConnectionSessionSnapshot func(*session)
148 // afterAttachmentFrameDispatch is a deterministic test seam after the
149 // connection loop snapshots a token and before a decoded frame takes effect.
150 afterAttachmentFrameDispatch func(attachmentConnectionToken)
151 // afterAttachmentEffectParticipantsSnapshotted observes immutable role-gate
152 // participants after architecture preflight locks are released and before
153 // the globally ordered freeze begins.
154 afterAttachmentEffectParticipantsSnapshotted func(string, []*attachedClient)
155 // afterAttachmentEffectGateFrozen observes each participant after it is frozen in
156 // immutable identity order and before drain/publication continues.
157 afterAttachmentEffectGateFrozen func(string, *attachedClient)
158 // afterAttachmentEffectsFrozen observes the lock-free boundary after all affected
159 // attachment gates are frozen and drained, before architecture publication.
160 afterAttachmentEffectsFrozen func()
161 // afterAttachmentTransitionCoordinatorsLocked is a deterministic lock-order
162 // seam used by transition validation tests.
163 afterAttachmentTransitionCoordinatorsLocked func()
164 // beforeMoveDispatch, afterMoveLifecycleReserved, and
165 // afterMovePaneSourceSnapshot are test-only seams. They run before dispatch
166 // admission, after lifecycle reservation, and after the pre-fence source
167 // attachment snapshot respectively. beforeMovePaneCommit runs inside the
168 // non-failing publication section. None is set in production, and none may
169 // perform external work while locks are held.
170 beforeMoveDispatch func()
171 afterMoveLifecycleReserved func()
172 afterMoveLifecycleGateBeforeTeardownLocks func()
173 afterMovePaneSourceSnapshot func()
174 afterMoveTabSourceSnapshot func()
175 beforeMovePaneCommit func()
176 beforeMoveTabCommit func()
177 // afterDetachAttachmentEffectsFrozen observes terminal detach after it wins the
178 // attachment gate but before it checks session ownership.
179 afterDetachAttachmentEffectsFrozen func()
180 // beforeClientGoneDetach pauses clientGone after the stale-transport
181 // precheck and before exact transport/incarnation detach validation.
182 beforeClientGoneDetach func()
183 // beforeMarkParkingInFlight pauses resumeLiveAttachment after the live
184 // credential matches and before markParkingInFlight, so tests can win
185 // explicit detach or session removal before late marker publication.
186 beforeMarkParkingInFlight func()
187 // afterClientGoneDetach pauses finishClientGone after detach won and before
188 // parkAttachment. The parking-in-flight marker must already be published.
189 afterClientGoneDetach func()
190 // afterResumeLiveDetach pauses resumeLiveAttachment after detach won and
191 // before parkAttachment. The parking-in-flight marker must already be
192 // published so a concurrent same-token resume can wait the gap.
193 afterResumeLiveDetach func()
194 // afterParkingWaitArmed observes waitParkingInFlight after it has resolved a
195 // matching in-flight entry and before it blocks on done.
196 afterParkingWaitArmed func()
197 // beforeResumeParkedSendMu pauses resumeParked after the initial parked
198 // lookup and before the attachment send lock, so tests can consume or
199 // replace the credential while the handshake waits.
200 beforeResumeParkedSendMu func()
201 // afterAttachmentEffectAdmitted is a deterministic test seam after a frame/paint
202 // reserves its exact capability and before its first observable mutation.
203 afterAttachmentEffectAdmitted func(attachmentConnectionToken)
204 // beforeFirstPaintSendWait is a test-only seam immediately before a
205 // transition first paint waits for the attachment send lock.
206 beforeFirstPaintSendWait func(attachmentConnectionToken)
207 // afterDelayedKeyEffectAttempt observes whether a timer callback acquired a
208 // fresh exact capability before producing PTY, action, or overlay effects.
209 afterDelayedKeyEffectAttempt func(bool)
210 // afterActionAttachmentEffectEnded observes action-specific admission release. It
211 // is a deterministic seam for proving no role-bound mutation follows release.
212 afterActionAttachmentEffectEnded func(string)
213 // beforeAttachmentSendErrorCleanup pauses asynchronous render-failure retirement
214 // after the render ticket ends and before exact lifecycle validation.
215 beforeAttachmentSendErrorCleanup func(attachmentConnectionToken)
216 afterAttachmentSendErrorCleanup func()
217 ptys ports.PTYFactory
218 clock ports.Clock
219 log *slog.Logger
220 runtimeObserver ports.RuntimeObserver
221 baseEnv []string
222 shell string
223 shellArgs []string
224 shellOverride bool
225 persistEnabled bool
226 catalogue ports.Catalogue
227 catalogueRecords []domain.CatalogueRecord
228 catalogueRecordsProvided bool
229 // snapshotRepository is the sole checkpoint storage contract.
230 snapshotRepository ports.SnapshotRepository
231 recovery *recoveryusecase.Coordinator
232 maintenanceWorkerCancel context.CancelFunc
233 maintenanceWorkerDone chan struct{}
234 // restoreWorkerDone is the restoration goroutine's ownership signal. Startup
235 // restoration reconciles durable checkpoints, so it is a durable writer and
236 // is guarded by snapshotWorkerMu with the other two.
237 restoreWorkerDone chan struct{}
238 snapsEnabled bool
239 noticeStore ports.NoticeStore
240 snapshotJobs chan *snapshotCapture
241 // snapshotAdmitted contains every capture accepted by either worker queue,
242 // including captures buffered in snapshotJobs. Guarded by snapshotWorkerMu.
243 snapshotAdmitted map[*snapshotCapture]struct{}
244 // snapshotWake wakes the repository scheduler when a session becomes dirty
245 // or an attempt completes. It is never closed and producers only send
246 // non-blockingly.
247 snapshotWake chan struct{}
248 snapshotWorkerMu sync.Mutex
249 snapshotWorkerID uint64
250 snapshotWorkerCtx context.Context
251 snapshotWorkerCancel context.CancelFunc
252 snapshotWorkerDone chan struct{}
253 snapshotWorkerFlush chan struct{}
254 snapshotWorkerFinalWake chan struct{}
255 // snapshotFinalJobs coalesces terminal captures by session when the bounded
256 // regular queue is full. It retains at most snapshotFinalQueueCapacity named
257 // sessions, each with only its newest terminal state while the worker blocks.
258 snapshotFinalJobs map[*session]*snapshotCapture
259 snapshotFinalOrder []*session
260 snapshotWorkerClosing bool
261 snapshotWorkerInFlight *snapshotCapture
262 // snapshotNoticeMu guards the active global persistence failure signature.
263 // It is separate from snapshotWorkerMu so notice routing cannot block a
264 // producer or a repository worker.
265 snapshotNoticeMu sync.Mutex
266 snapshotActiveFailureSignature string
267 shutdownNoticeMu sync.Mutex
268 shutdownNoticedSessions map[string]struct{}
269 restoreDone chan struct{}
270 restoreOnce sync.Once
271 procCwd func(int) (string, error)
272 procComm func(int) (string, error)
273 procArgv func(int) ([]string, error)
274 procGroupArgv func(int, int) ([]string, error)
275 dirOrHome func(string) string
276 bindings atomic.Pointer[keys.Bindings]
277 codeOverrides atomic.Pointer[map[string]string]
278 restoreProcessAllowlist atomic.Pointer[map[string]struct{}]
279 floatingConfig atomic.Pointer[domain.FloatingConfig]
280 copyConfig atomic.Pointer[domain.CopyConfig]
281 paletteConfig atomic.Pointer[domain.PaletteConfig]
282 navConfig atomic.Pointer[domain.NavConfig]
283 tabsConfig atomic.Pointer[domain.TabsConfig]
284 themeConfig atomic.Pointer[themeConfigSnapshot]
285 barScripts *barScriptState
286 notices *noticeCenter
287 resumeParkGrace time.Duration
288 // remoteCatalog owns cache-derived discovery state independently of the
289 // live attachment registry. Its cache reads and writes never hold d.mu.
290 remoteCatalog remoteCatalogState
291 remotePreview remotePreviewState
292 remoteHostStore ports.RemoteHostStore
293 remoteCatalogClient ports.RemoteCatalogClient
294 remoteCatalogCache ports.RemoteCatalogCache
295 remotePreviewClient ports.RemotePreviewClient
296 remoteDialerFactory ports.RemoteDialerFactory
297 remoteTransportMode ports.RemoteTransportMode
298 // tempDir overrides os.TempDir() for clipboard-image-transfer writes
299 // (see clipboard.go); empty means use os.TempDir().
300 tempDir string
301
302 serveCtx context.Context
303 serveCancel context.CancelFunc
304
305 // hardCtx force-closes connection transports on shutdown, but only after
306 // shutdownAll has delivered graceful Detached notices — keeping it separate
307 // from serveCtx so a parent-context cancel never races the notice.
308 hardCtx context.Context
309 hardCancel context.CancelFunc
310
311 done chan struct{}
312 doneOnce sync.Once
313
314 // sessWg owns attention animation, bar-script polling, CWD sampling,
315 // snapshot save/restore, and floating-session launch goroutines.
316 sessWg sync.WaitGroup
317 connWg sync.WaitGroup // per-connection handler goroutines
318 // attachmentCleanupWg owns replacement transport closes and retired render
319 // worker joins. Only connection handlers add work; Serve waits for those
320 // handlers before joining this group, so no Add races its terminal Wait.
321 attachmentCleanupWg sync.WaitGroup
322}
323
324type parkedAttachment struct {
325 sess *session
326 ac *attachedClient
327 pickerGeneration uint64
328 timer ports.Timer
329 claimed bool
330 done chan struct{}
331 doneOnce sync.Once
332}
333
334// parkingAttachment is the observable detach→park lifecycle for one resume
335// credential. done closes when the token is published into parked or parking
336// is abandoned.
337type parkingAttachment struct {
338 sess *session
339 ac *attachedClient
340 done chan struct{}
341 doneOnce sync.Once
342}
343
344func (p *parkingAttachment) closeDone() {
345 if p == nil || p.done == nil {
346 return
347 }
348 p.doneOnce.Do(func() { close(p.done) })
349}
350
351// session is a single multiplexed session. It owns one or more full-screen
352
353type stoppedSession struct {
354 name string
355 cwd string
356 createdAt int64
357 incarnation domain.IncarnationID
358 lastUsedSeq uint64
359 tabNames []string
360 tabRecords []domain.CatalogueTabRecord
361 purging bool
362 record domain.CatalogueRecord
363 state ports.SessionState
364 restoreDone chan struct{}
365}
366
367type Option func(*Daemon)
368
369// WithRuntimeObserver accepts only a composition-root serialized observer.
370// The application owns its lifecycle; the daemon never creates or closes a
371// second reporting worker around it.
372func WithRuntimeObserver(observer ports.SerializedRuntimeObserver) Option {
373 return func(d *Daemon) { d.runtimeObserver = observer }
374}
375
376// WithRemoteDiscovery installs the remote discovery ports used by the daemon.
377// The composition root validates mode before constructing the daemon.
378func WithRemoteDiscovery(store ports.RemoteHostStore, catalog ports.RemoteCatalogClient, cache ports.RemoteCatalogCache, dialer ports.RemoteDialerFactory, mode ports.RemoteTransportMode) Option {
379 return func(d *Daemon) {
380 d.remoteHostStore = store
381 d.remoteCatalogClient = catalog
382 d.remoteCatalogCache = cache
383 d.remoteDialerFactory = dialer
384 d.remoteTransportMode = mode
385 }
386}
387
388// WithRemotePreview installs the optional non-attaching remote viewport client.
389func WithRemotePreview(client ports.RemotePreviewClient) Option {
390 return func(d *Daemon) { d.remotePreviewClient = client }
391}
392
393// WithShell overrides the command (and its args) each session spawns. The
394// default is $SHELL (or /bin/sh) with no arguments; tests use this to run a
395// deterministic program.
396func WithShell(cmd string, args []string) Option {
397 return func(d *Daemon) {
398 d.shell = cmd
399 d.shellArgs = args
400 d.shellOverride = true
401 }
402}
403
404// WithCatalogue installs the singular catalogue and the strictly opened
405// records that define the daemon's expected-session registry at startup.
406func WithCatalogue(catalogue ports.Catalogue, records []domain.CatalogueRecord) Option {
407 return func(d *Daemon) {
408 d.catalogue = catalogue
409 d.persistEnabled = catalogue != nil
410 d.catalogueRecords = append([]domain.CatalogueRecord(nil), records...)
411 d.catalogueRecordsProvided = true
412 }
413}
414
415func (d *Daemon) catalogueRecord(name string) (domain.CatalogueRecord, bool, error) {
416 if d == nil || d.catalogue == nil {
417 return domain.CatalogueRecord{}, false, nil
418 }
419 return d.catalogue.Record(name)
420}
421
422// markCatalogueDirty buffers metadata. The periodic catalogue timer, the next
423// identity write, or shutdown provides the durability barrier.
424func (d *Daemon) markCatalogueDirty(update domain.CatalogueMetadataUpdate) {
425 if d == nil || d.catalogue == nil {
426 return
427 }
428 if err := d.catalogue.UpdateMetadata(update); err != nil {
429 d.log.Warn("buffering session metadata failed", "err", err, "session", update.Name)
430 }
431}
432
433func (d *Daemon) flushCatalogue() error {
434 if d == nil || d.catalogue == nil {
435 return nil
436 }
437 return d.catalogue.Sync()
438}
439
440// WithRecoveryCoordinator installs the durable recovery coordinator.
441func WithRecoveryCoordinator(coordinator *recoveryusecase.Coordinator) Option {
442 return func(d *Daemon) {
443 d.recovery = coordinator
444 }
445}
446
447// WithSnapshotRepository enables content-addressed incremental snapshots.
448func WithSnapshotRepository(repository ports.SnapshotRepository) Option {
449 return func(d *Daemon) {
450 if isNilSnapshotRepository(repository) {
451 return
452 }
453 d.snapshotRepository = repository
454 d.snapsEnabled = true
455 }
456}
457
458func isNilSnapshotRepository(repository ports.SnapshotRepository) bool {
459 if repository == nil {
460 return true
461 }
462 value := reflect.ValueOf(repository)
463 switch value.Kind() {
464 case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, reflect.Pointer, reflect.Slice:
465 return value.IsNil()
466 default:
467 return false
468 }
469}
470
471// WithNoticeStore enables persisting undeliverable notices across daemon
472// restarts. A nil store keeps the daemon in no-op notice-persistence mode.
473func WithNoticeStore(store ports.NoticeStore) Option {
474 return func(d *Daemon) { d.noticeStore = store }
475}
476
477// WithDurableMaintenance retains the application wiring for the one
478// pre-publication GC pass. Standalone users without an explicitly supplied
479// coordinator receive the same canonical coordinator path.
480func WithDurableMaintenance(catalogue ports.Catalogue, repository ports.SnapshotRepository) Option {
481 return func(d *Daemon) {
482 if d.recovery == nil {
483 d.recovery = recoveryusecase.NewCoordinator(catalogue, repository, nil)
484 }
485 }
486}
487
488// WithCwdReader overrides the process cwd reader used for persistence tests.
489func WithCwdReader(fn func(int) (string, error)) Option {
490 return func(d *Daemon) {
491 if fn != nil {
492 d.procCwd = fn
493 }
494 }
495}
496
497// WithProcessInspector installs the platform process-inspection implementation.
498func WithProcessInspector(ins ports.ProcessInspector) Option {
499 return func(d *Daemon) {
500 if ins == nil {
501 return
502 }
503 d.procCwd = ins.Cwd
504 d.procComm = ins.Comm
505 d.procArgv = ins.Argv
506 d.procGroupArgv = ins.GroupArgv
507 }
508}
509
510// WithDirOrHome installs path fallback behavior from the application layer.
511func WithDirOrHome(fn func(string) string) Option {
512 return func(d *Daemon) {
513 if fn != nil {
514 d.dirOrHome = fn
515 }
516 }
517}
518
519// WithTempDir overrides the directory clipboard-image-transfer writes temp
520// files into (production default: os.TempDir()); tests use this with
521// t.TempDir() so writes are isolated and auto-cleaned.
522func WithTempDir(dir string) Option {
523 return func(d *Daemon) {
524 d.tempDir = dir
525 }
526}
527
528// WithBarScriptCommandRunner installs the shell command runner used by bar scripts.
529func WithBarScriptCommandRunner(runner ports.ShellCommandRunner) Option {
530 return func(d *Daemon) {
531 if runner != nil {
532 d.barScripts.runner = barScriptRunner{runner: runner}
533 }
534 }
535}
536
537// WithResumeParkGrace overrides how long detached resume-capable clients stay
538// parked for reconnection. Non-positive durations keep the default.
539func WithResumeParkGrace(grace time.Duration) Option {
540 return func(d *Daemon) {
541 if grace > 0 {
542 d.resumeParkGrace = grace
543 }
544 }
545}
546
547// WithConfig applies the initial user configuration.
548func WithConfig(cfg domain.Config) Option {
549 return func(d *Daemon) {
550 d.ApplyConfig(cfg)
551 }
552}
553
554// allocateLifecycleCreatedAtLocked returns a unique lifecycle timestamp. Caller
555// holds d.mu. The persisted high-water mark is loaded by New before any named
556// session can be resumed or created.
557func (d *Daemon) allocateLifecycleCreatedAtLocked() (int64, error) {
558 now := d.nowUnixNano()
559 if now <= d.lastAllocatedCreatedAt {
560 if d.lastAllocatedCreatedAt == math.MaxInt64 {
561 return 0, errors.New("daemon: lifecycle identities exhausted")
562 }
563 now = d.lastAllocatedCreatedAt + 1
564 }
565 d.lastAllocatedCreatedAt = now
566 return now, nil
567}
568
569func (d *Daemon) nowUnixNano() int64 {
570 return d.clock.Now().UnixNano()
571}
572
573type systemClock struct{}
574type systemTimer struct{ *time.Timer }
575
576func (systemClock) Now() time.Time { return time.Now() }
577func (systemClock) NewTimer(delay time.Duration) ports.Timer {
578 return systemTimer{Timer: time.NewTimer(delay)}
579}
580func (t systemTimer) C() <-chan time.Time { return t.Timer.C }
581
582// New constructs a Daemon. ptys spawns PTY-backed children, clock drives the
583// render debounce, and log receives diagnostics (defaults to slog.Default).
584func New(ptys ports.PTYFactory, clock ports.Clock, log *slog.Logger, opts ...Option) *Daemon {
585 if log == nil {
586 log = slog.Default()
587 }
588 if clock == nil {
589 clock = systemClock{}
590 }
591 shell := os.Getenv("SHELL")
592 if shell == "" {
593 shell = "/bin/sh"
594 }
595 paneProcessCtx, paneProcessCancel := context.WithCancel(context.Background())
596 d := &Daemon{
597 sessions: make(map[domain.SessionID]*session),
598 stopped: make(map[string]stoppedSession),
599 creating: make(map[string]struct{}),
600 parked: make(map[uint64]*parkedAttachment),
601 parking: make(map[uint64]*parkingAttachment),
602 paneProcessCtx: paneProcessCtx,
603 paneProcessCancel: paneProcessCancel,
604 ptys: ptys,
605 clock: clock,
606 log: log,
607 baseEnv: os.Environ(),
608 shell: shell,
609 dirOrHome: dirOrHome,
610 done: make(chan struct{}),
611 restoreDone: make(chan struct{}),
612 animWake: make(chan struct{}, 1),
613 snapshotJobs: make(chan *snapshotCapture, snapshotQueueCapacity),
614 snapshotAdmitted: make(map[*snapshotCapture]struct{}),
615 snapshotWake: make(chan struct{}, 1),
616 notices: newNoticeCenter(),
617 remoteCatalog: newRemoteCatalogState(),
618 resumeParkGrace: defaultResumeParkGrace,
619 barScripts: &barScriptState{
620 cfg: barConfigFromDomain(domain.Defaults().Bar),
621 outputs: make(map[domain.SessionID]barScriptOutputs),
622 lastRefresh: make(map[domain.SessionID]time.Time),
623 lastContext: make(map[domain.SessionID]barScriptContext),
624 running: make(map[domain.SessionID]bool),
625 reload: make(chan struct{}, 1),
626 },
627 }
628 defaults := domain.Defaults()
629 defaultFloating := defaults.Floating
630 d.floatingConfig.Store(&defaultFloating)
631 defaultCopy := defaults.Copy
632 d.copyConfig.Store(&defaultCopy)
633 defaultPalette := defaults.Palette
634 d.paletteConfig.Store(&defaultPalette)
635 for _, o := range opts {
636 o(d)
637 }
638 if d.dirOrHome == nil {
639 d.dirOrHome = dirOrHome
640 }
641 if d.bindings.Load() == nil {
642 d.bindings.Store(keys.DefaultBindings())
643 }
644 if d.codeOverrides.Load() == nil {
645 empty := map[string]string{}
646 d.codeOverrides.Store(&empty)
647 }
648 if d.restoreProcessAllowlist.Load() == nil {
649 allow := buildRestoreProcessAllowlist(domain.DefaultSnapshotRestoreProcesses())
650 d.restoreProcessAllowlist.Store(&allow)
651 }
652 records := d.catalogueRecords
653 var maxSeq uint64
654 var maxCreatedAt int64
655 hasCreatedAt := false
656 for _, r := range records {
657 if d.catalogueRecordsProvided {
658 state, done := initialSessionState(r)
659 d.stopped[r.Name] = stoppedSessionFromRecord(r, state, done)
660 } else {
661 d.stopped[r.Name] = stoppedSessionFromRecord(r, ports.SessionDown, nil)
662 }
663 if !hasCreatedAt || r.CreatedAt > maxCreatedAt {
664 maxCreatedAt = r.CreatedAt
665 hasCreatedAt = true
666 }
667 if r.LastUsedSeq > maxSeq {
668 maxSeq = r.LastUsedSeq
669 }
670 }
671 if hasCreatedAt {
672 d.lastAllocatedCreatedAt = maxCreatedAt
673 }
674 d.mruSeq.Store(maxSeq)
675 d.loadRemoteCatalogCache()
676 return d
677}
678
679// Serve runs the accept loop over l, owning it for the loop's lifetime. It
680// returns when the last session is removed or ctx is cancelled; on the latter
681// path attached clients are detached with ReasonServerShutdown.
682func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error {
683 d.serveCtx, d.serveCancel = context.WithCancel(ctx)
684 defer d.serveCancel()
685 d.hardCtx, d.hardCancel = context.WithCancel(context.Background())
686 defer d.hardCancel()
687
688 d.sessWg.Go(func() {
689 d.attentionAnimator(d.serveCtx)
690 })
691 d.sessWg.Go(func() {
692 d.barScriptPoller(d.serveCtx)
693 })
694 if d.persistEnabled {
695 d.sessWg.Go(func() {
696 d.cwdSampler(d.serveCtx)
697 })
698 }
699 if d.snapsEnabled {
700 d.startSnapshotEncodeWorker()
701 d.sessWg.Go(func() {
702 d.snapshotRepositorySaver(d.serveCtx)
703 })
704 d.startSnapshotRestoration()
705 } else {
706 d.closeRestoreDone()
707 }
708 // Break the accept loop when either the parent context is cancelled or the
709 // registry drains to empty: both close the listener, which fails Accept.
710 go func() {
711 select {
712 case <-d.serveCtx.Done():
713 case <-d.done:
714 }
715 _ = l.Close()
716 }()
717
718 for {
719 tr, err := l.Accept()
720 if err != nil {
721 if d.serveCtx.Err() != nil {
722 d.log.Info("accept loop exiting", "err", err, "reason", "context canceled")
723 } else {
724 d.log.Warn("accept loop exiting", "err", err)
725 }
726 break
727 }
728 d.connWg.Go(func() {
729 d.handleConn(tr)
730 })
731 }
732
733 // Tear down. shutdownAll marks the daemon closing (route now rejects any
734 // racing Hello) and kills every session: PTYs are closed and contexts
735 // cancelled unconditionally, with the Detached notices sent asynchronously
736 // under a deadline. Wait for those notices (bounded — a wedged client's
737 // transport is force-closed by the deadline) before hard-closing the
738 // remaining parked connections, so graceful notices are never raced by the
739 // force-close. Finally drain the conn handlers, run one defensive sweep in
740 // case any ordering ever leaves a session behind, and join the session
741 // goroutines (readers unblock via pty.Close, coordinators via ctx cancel).
742 // One deadline owns every terminal checkpoint wait and the final worker
743 // join. A repository is allowed to ignore cancellation, so spending a fresh
744 // interval at either stage would make shutdown exceed its documented bound.
745 var snapshotDeadline *snapshotShutdownDeadline
746 if d.snapsEnabled {
747 snapshotDeadline = newSnapshotShutdownDeadline(d.clock)
748 defer snapshotDeadline.stop()
749 }
750 d.shutdownAllWithSnapshotDeadline(ports.ReasonServerShutdown, snapshotDeadline)
751 d.waitNotifies()
752 d.hardCancel()
753 d.serveCancel()
754 // Wait for handlers before flushing snapshots: a handler that already
755 // entered killSession may still submit its terminal capture after the
756 // registry snapshot has been removed.
757 d.connWg.Wait()
758 d.attachmentCleanupWg.Wait()
759 // Forced terminal checkpoints run before shutdown stops producers or the
760 // worker. StopDurableWriters reports every session whose final capture still
761 // owns work when the shared checkpoint budget expires. Persist one notice per
762 // affected session, then retain ownership until every writer actually exits.
763 stopCtx, stopCancel := snapshotStopContext(snapshotDeadline)
764 timedOutSessions := d.StopDurableWriters(stopCtx)
765 stopCancel()
766 for _, name := range timedOutSessions {
767 d.persistShutdownSnapshotFailure(name, context.DeadlineExceeded)
768 }
769 d.WaitDurableWriters()
770 d.shutdownAllWithSnapshotDeadline(ports.ReasonServerShutdown, snapshotDeadline)
771 d.waitSessionWorkersWithSnapshotDeadline(snapshotDeadline)
772 d.waitNotifies()
773 if err := d.flushCatalogue(); err != nil {
774 d.log.Warn("flushing session catalogue at shutdown failed", "err", err)
775 }
776 if d.catalogue != nil {
777 if err := d.catalogue.Close(); err != nil {
778 d.log.Warn("closing session catalogue failed", "err", err)
779 }
780 }
781 return nil
782}
783
784// shutdownAll marks the daemon closing and kills every live session. Setting
785// closing under the same lock as the snapshot guarantees no session can be
786// inserted after the snapshot: route rejects once closing is set, and both run
787// under d.mu. killSession (which relocks) runs after the lock is released.
788func (d *Daemon) shutdownAll(reason uint8) (checkpointIncomplete bool) {
789 return d.shutdownAllWithSnapshotDeadline(reason, nil)
790}
791
792func (d *Daemon) shutdownAllWithSnapshotDeadline(reason uint8, deadline *snapshotShutdownDeadline) (checkpointIncomplete bool) {
793 d.cancelRemotePickerRefresh()
794 d.closeMoveLifecycles()
795 d.mu.Lock()
796 d.closing = true
797 d.purgeAllParkingLocked()
798 parkedRetirements := d.purgeAllParkedLocked()
799 snapshot := d.sessionsSnapshotLocked()
800 empty := len(snapshot) == 0
801 d.mu.Unlock()
802 d.finishParkedAttachmentRetirements(parkedRetirements)
803 d.log.Info("graceful shutdown begin", "reason", reason, "sessions", len(snapshot))
804 if empty {
805 d.doneOnce.Do(func() { close(d.done) })
806 return false
807 }
808 for _, s := range snapshot {
809 // Cancellation and PTY closure must not wait behind a teardown owner that
810 // is blocked in snapshot publication or purge work.
811 s.mu.Lock()
812 name := s.name
813 ephemeral := s.ephemeral
814 s.mu.Unlock()
815 s.stopInMemoryLifecycle()
816 if err := d.killSessionWithSnapshotDeadline(s, reason, false, deadline, nil); err != nil {
817 checkpointIncomplete = true
818 d.log.Error("closing session with unpersisted terminal state", "err", err)
819 }
820 if !ephemeral && deadline != nil {
821 select {
822 case <-deadline.Done():
823 checkpointIncomplete = true
824 d.persistShutdownSnapshotFailure(name, context.DeadlineExceeded)
825 default:
826 }
827 }
828 }
829 // Signal Serve even when a concurrent role transition made this initial
830 // pass abort; Serve owns the bounded defensive teardown passes.
831 d.doneOnce.Do(func() { close(d.done) })
832 return checkpointIncomplete
833}
834
835// waitSessionWorkersWithSnapshotDeadline joins in-memory workers when possible,
836// but never extends Serve beyond the shared repository shutdown budget. By this
837// point connection handlers are joined and every session has been cancelled, so
838// no new session worker can be registered while the waiter is active.
839func (d *Daemon) waitSessionWorkersWithSnapshotDeadline(deadline *snapshotShutdownDeadline) {
840 done := make(chan struct{})
841 go func() {
842 d.sessWg.Wait()
843 close(done)
844 }()
845 if deadline == nil {
846 <-done
847 return
848 }
849 select {
850 case <-done:
851 case <-deadline.Done():
852 }
853}
854
855// snapshotShutdownDeadline is a single-use shutdown budget shared by forced
856// checkpoints and the final snapshot worker join. Its done channel is never
857// closed by callers, so a detached worker can safely retain session state until
858// an uncooperative repository call returns.
859type snapshotShutdownDeadline struct {
860 done chan struct{}
861 stopCh chan struct{}
862 finished chan struct{}
863 stopOnce sync.Once
864}
865
866func newSnapshotShutdownDeadline(clock ports.Clock) *snapshotShutdownDeadline {
867 deadline := &snapshotShutdownDeadline{done: make(chan struct{}), stopCh: make(chan struct{}), finished: make(chan struct{})}
868 timer := clock.NewTimer(snapshotFinalFlushTimeout)
869 go func() {
870 defer close(deadline.finished)
871 select {
872 case <-timer.C():
873 close(deadline.done)
874 case <-deadline.stopCh:
875 timer.Stop()
876 }
877 }()
878 return deadline
879}
880
881func (d *snapshotShutdownDeadline) stop() {
882 if d != nil {
883 d.stopOnce.Do(func() {
884 close(d.stopCh)
885 <-d.finished
886 })
887 }
888}
889
890func (d *snapshotShutdownDeadline) Done() <-chan struct{} {
891 if d == nil {
892 return nil
893 }
894 return d.done
895}
896
897func snapshotStopContext(deadline *snapshotShutdownDeadline) (context.Context, context.CancelFunc) {
898 ctx, cancel := context.WithCancel(context.Background())
899 if deadline == nil {
900 return ctx, cancel
901 }
902 select {
903 case <-deadline.Done():
904 cancel()
905 return ctx, cancel
906 default:
907 }
908 stop := make(chan struct{})
909 go func() {
910 select {
911 case <-deadline.Done():
912 cancel()
913 case <-stop:
914 }
915 }()
916 var stopOnce sync.Once
917 return ctx, func() {
918 stopOnce.Do(func() { close(stop) })
919 cancel()
920 }
921}
922
923// registerSessionLocked publishes an exact attachment-session identity. Caller
924// holds d.mu.
925func (d *Daemon) registerSessionLocked(entry *session) bool {
926 if entry == nil {
927 return false
928 }
929 core := entry.core()
930 if core == nil || core.id == "" || d.sessions[core.id] != nil {
931 return false
932 }
933 d.sessions[core.id] = entry
934 return true
935}
936
937// unregisterSessionLocked removes only the exact registered identity. Caller
938// holds d.mu.
939func (d *Daemon) unregisterSessionLocked(entry *session) bool {
940 if entry == nil {
941 return false
942 }
943 core := entry.core()
944 if core == nil || d.sessions[core.id] != entry {
945 return false
946 }
947 delete(d.sessions, core.id)
948 return true
949}
950
951func (d *Daemon) sessionsSnapshotLocked() []*session {
952 return sessionsSnapshot(d.sessions)
953}
954
955// handleConn reads the first frame off a fresh connection and routes it. A
956// context watcher closes the transport on serve-context cancel or handshake
957// timeout so blocked transport operations unwind.
958func (d *Daemon) handleConn(tr ports.Transport) {
959 handshakeCtx, timedOut, finishHandshake := d.newHandshakeContext(d.hardCtx)
960 stopTransport := watchHandshakeTransport(handshakeCtx, tr)
961 defer finishHandshake()
962 defer stopTransport()
963
964 var first ports.Frame
965 err := boundedHandshakeOperation(handshakeCtx, tr, func() error {
966 var recvErr error
967 first, recvErr = tr.Recv()
968 return recvErr
969 })
970 if err != nil {
971 err = handshakeContextError(d.hardCtx, timedOut, err)
972 d.log.Warn("connection closed before hello", "err", err)
973 _ = tr.Close()
974 return
975 }
976 switch first.Type {
977 case ports.MsgList:
978 stopTransport()
979 finishHandshake()
980 d.handleList(tr)
981 case ports.MsgCommand:
982 stopTransport()
983 finishHandshake()
984 if err := d.handleCommand(tr, first); err != nil {
985 d.log.Warn("command handler failed", "err", err)
986 }
987 case ports.MsgRemotePreviewRequest:
988 stopTransport()
989 finishHandshake()
990 if err := d.handleRemotePreview(tr, first); err != nil {
991 d.log.Warn("remote preview handler failed", "err", err)
992 }
993 case ports.MsgKill:
994 stopTransport()
995 finishHandshake()
996 d.handleKill(tr, first)
997 case ports.MsgHello:
998 d.handleHelloWithContext(handshakeCtx, timedOut, stopTransport, finishHandshake, tr, first)
999 default:
1000 d.log.Warn("hello rejected", "err", "expected hello", "type", first.Type)
1001 _ = boundedHandshakeOperation(handshakeCtx, tr, func() error {
1002 return tr.Send(frameError(ports.ErrInternal, "expected hello"))
1003 })
1004 stopTransport()
1005 finishHandshake()
1006 _ = tr.Close()
1007 }
1008}
1009
1010// handleList replies with the current session listing and closes the (one-shot
1011// control) connection.
1012func (d *Daemon) handleList(tr ports.Transport) {
1013 defer func() { _ = tr.Close() }()
1014
1015 d.mu.Lock()
1016 infos := make([]ports.SessionInfo, 0, len(d.sessions)+len(d.stopped))
1017 liveNames := make(map[string]struct{}, len(d.sessions))
1018 for _, s := range d.sessions {
1019 if s == nil {
1020 continue
1021 }
1022 s.mu.Lock()
1023 info := ports.SessionInfo{
1024 SessionID: string(s.id),
1025 Name: s.name,
1026 State: ports.SessionUp,
1027 Ephemeral: s.ephemeral,
1028 Tabs: uint16(len(s.tabs)),
1029 Attached: len(s.attachments) != 0,
1030 }
1031 liveNames[s.name] = struct{}{}
1032 s.mu.Unlock()
1033 infos = append(infos, info)
1034 }
1035 for name, stopped := range d.stopped {
1036 if _, live := liveNames[name]; live {
1037 continue
1038 }
1039 state := ports.SessionDown
1040 if stopped.purging || stopped.state == ports.SessionBroken {
1041 // Purge is the dominant externally visible state: restoration must
1042 // never make a deletion-reserved record appear attachable.
1043 state = ports.SessionBroken
1044 }
1045 infos = append(infos, ports.SessionInfo{Name: name, State: state})
1046 }
1047 d.mu.Unlock()
1048
1049 sort.Slice(infos, func(i, j int) bool { return infos[i].Name < infos[j].Name })
1050 _ = d.boundedControlSend(tr, frameSessions(infos))
1051}
1052
1053// handleKill terminates the requested live session or stopped named session,
1054// or all sessions, and closes the control connection; the resulting EOF is the
1055// client's success signal.
1056func (d *Daemon) handleKill(tr ports.Transport, f ports.Frame) {
1057 defer func() { _ = tr.Close() }()
1058
1059 k, err := ports.UnmarshalKill(f.Payload)
1060 if err != nil {
1061 _ = d.boundedControlSend(tr, frameError(ports.ErrInternal, "malformed kill request"))
1062 return
1063 }
1064 if k.All {
1065 d.shutdownAll(ports.ReasonServerShutdown)
1066 return
1067 }
1068
1069 d.mu.Lock()
1070 target := d.findByNameLocked(k.Name)
1071 if target == nil {
1072 if _, ok := d.stopped[k.Name]; ok {
1073 d.mu.Unlock()
1074 // Stopped sessions use the same catalogue-first, incarnation-second
1075 // deletion order as live and offline purges.
1076 if err := d.retryStoppedPurge(k.Name); err != nil {
1077 d.log.Warn("deleting stopped session failed", "err", err, "session", k.Name)
1078 _ = d.boundedControlSend(tr, frameError(ports.ErrInternal, "deleting stopped session failed"))
1079 }
1080 return
1081 }
1082 }
1083 d.mu.Unlock()
1084
1085 if target == nil {
1086 _ = d.boundedControlSend(tr, frameError(ports.ErrNoSuchSession, "no such session: "+k.Name))
1087 return
1088 }
1089 if err := d.killSession(target, ports.ReasonSessionKilled, true); err != nil {
1090 _ = d.boundedControlSend(tr, frameError(ports.ErrInternal, "deleting persisted session failed"))
1091 }
1092}
1093
1094// handleHello runs the attach handshake for direct package callers. Accepted
1095// connections use handleHelloWithContext so the deadline also covers the first
1096// frame read in handleConn.
1097func (d *Daemon) handleHello(tr ports.Transport, f ports.Frame) {
1098 handshakeCtx, timedOut, finishHandshake := d.newHandshakeContext(context.Background())
1099 stopTransport := watchHandshakeTransport(handshakeCtx, tr)
1100 defer finishHandshake()
1101 defer stopTransport()
1102 d.handleHelloWithContext(handshakeCtx, timedOut, stopTransport, finishHandshake, tr, f)
1103}
1104
1105// handleHelloWithContext runs the attach handshake: version check, intent
1106// routing, Welcome, guaranteed first paint, then the per-connection input loop.
1107func (d *Daemon) handleHelloWithContext(handshakeCtx context.Context, timedOut <-chan struct{}, stopHandshakeTransport, finishHandshake func(), tr ports.Transport, f ports.Frame) {
1108 if err := handshakeContextError(context.Background(), timedOut, handshakeCtx.Err()); err != nil {
1109 _ = tr.Close()
1110 return
1111 }
1112 sendHandshake := func(frame ports.Frame) error {
1113 err := boundedHandshakeOperation(handshakeCtx, tr, func() error { return tr.Send(frame) })
1114 if err != nil {
1115 return handshakeContextError(context.Background(), timedOut, err)
1116 }
1117 return nil
1118 }
1119 h, err := ports.UnmarshalHello(f.Payload)
1120 if err != nil {
1121 if version, ok := ports.PeekHelloVersion(f.Payload); ok && version != ports.ProtocolVersion {
1122 d.log.Warn("hello rejected", "err", "protocol version mismatch", "version", version, "expected", ports.ProtocolVersion)
1123 _ = sendHandshake(frameError(ports.ErrVersionMismatch, "protocol version mismatch"))
1124 } else {
1125 d.log.Warn("hello rejected", "err", err)
1126 _ = sendHandshake(frameError(ports.ErrInternal, "malformed hello"))
1127 }
1128 _ = tr.Close()
1129 return
1130 }
1131 if h.Version != ports.ProtocolVersion {
1132 d.log.Warn("hello rejected", "err", "protocol version mismatch", "version", h.Version, "expected", ports.ProtocolVersion, "intent", h.Intent, "session", h.Name)
1133 _ = sendHandshake(frameError(ports.ErrVersionMismatch, "protocol version mismatch"))
1134 _ = tr.Close()
1135 return
1136 }
1137
1138 sess, ac, rerr := d.routeWithContext(handshakeCtx, h, tr)
1139 if rerr != nil {
1140 d.log.Warn("hello rejected", "err", rerr, "intent", h.Intent, "session", h.Name)
1141 if pe, ok := errors.AsType[*protoErr](rerr); ok {
1142 _ = sendHandshake(frameError(pe.code, pe.text))
1143 } else {
1144 _ = sendHandshake(frameError(ports.ErrInternal, rerr.Error()))
1145 }
1146 _ = tr.Close()
1147 return
1148 }
1149
1150 welcomeSent := false
1151 //nolint:contextcheck // handshakeCtx is intentionally not propagated: cancellation teardown must complete its geometry and ownership cleanup.
1152 failAttachment := func() {
1153 d.failHandshakeAttachment(sess, ac, tr, welcomeSent)
1154 }
1155 if err := handshakeCtx.Err(); err != nil {
1156 failAttachment()
1157 return
1158 }
1159 expected := ac.transportSnapshot()
1160 welcomeToken := sess.attachmentToken(ac, tr)
1161 welcomeTicket, admitted := ac.beginAttachmentEffect(welcomeToken)
1162 if expected.transport != tr || !admitted || welcomeToken.ac == nil {
1163 if admitted {
1164 welcomeTicket.End()
1165 }
1166 failAttachment()
1167 return
1168 }
1169 if err := handshakeCtx.Err(); err != nil {
1170 welcomeTicket.End()
1171 failAttachment()
1172 return
1173 }
1174 welcomeDone, welcomeErr := boundedHandshakeOperationTracked(handshakeCtx, tr, func() error {
1175 frame, err := frameWelcome(sess, d.resumeTokenSnapshot(ac))
1176 if err != nil {
1177 return err
1178 }
1179 return ac.sendExpectedTransportForAttachment(expected, frame, welcomeTicket)
1180 })
1181 if welcomeErr != nil {
1182 <-welcomeDone
1183 welcomeTicket.End()
1184 failAttachment()
1185 return
1186 }
1187 welcomeSent = true
1188 // Release Welcome's effect before discovering post-handshake authority so a
1189 // replacement blocked behind the send can publish its generation and lease.
1190 welcomeTicket.End()
1191 postWelcomeToken, postWelcomeTicket, admitted := ac.beginCurrentAttachmentEffectContext(handshakeCtx, sess, tr)
1192 if !admitted {
1193 failAttachment()
1194 return
1195 }
1196 postWelcomeLease := postWelcomeToken.lease
1197 if postWelcomeToken.ac == nil {
1198 postWelcomeTicket.End()
1199 failAttachment()
1200 return
1201 }
1202 postWelcomeRC := sess.renderCoordinator()
1203 if postWelcomeLease != nil && (postWelcomeRC == nil || !postWelcomeRC.markAttachmentReady(postWelcomeLease)) {
1204 postWelcomeTicket.End()
1205 // The attachment was detached while Welcome was in flight; never let
1206 // this stale handshake emit an Output frame.
1207 failAttachment()
1208 return
1209 }
1210 paintToken := sess.attachmentToken(ac, tr)
1211 painted := make(chan bool, 1)
1212 paintDone, paintErr := boundedHandshakeOperationTracked(handshakeCtx, tr, func() error {
1213 //nolint:contextcheck // firstPaintForTransition is capability-bounded; the surrounding handshake operation owns cancellation.
1214 painted <- d.firstPaintForTransition(paintToken)
1215 return nil
1216 })
1217 if paintErr != nil {
1218 <-paintDone
1219 postWelcomeTicket.End()
1220 failAttachment()
1221 return
1222 }
1223 if !<-painted || handshakeCtx.Err() != nil {
1224 postWelcomeTicket.End()
1225 failAttachment()
1226 return
1227 }
1228 if !d.commitResumeClaim(ac) {
1229 postWelcomeTicket.End()
1230 failAttachment()
1231 return
1232 }
1233 if ac.startupOverlay == ports.StartupOverlaySessionPicker {
1234 d.enterPicker(sess, ac)
1235 }
1236 postWelcomeTicket.End()
1237 stopHandshakeTransport()
1238 finishHandshake()
1239 d.runConnLoop(ac)
1240 _ = tr.Close()
1241}
1242
1243// protoErr is a session-level rejection carrying a wire ErrorMsg code.
1244type protoErr struct {
1245 code uint16
1246 text string
1247}
1248
1249func (e *protoErr) Error() string { return e.text }
1250
1251// finishAttach completes an attachment prepared while d.mu is held. Every
1252// caller must hold d.mu on entry; finishAttach transfers ownership by
1253// unlocking d.mu before returning, including every error path. It publishes
1254// terminal and role state before releasing d.mu, then defers coordinator
1255// cleanup so obsolete workers never delay the new handshake.
1256func (d *Daemon) finishAttach(sess *session, tr ports.Transport, sz domain.Size, term terminalEnv, h ports.Hello) (*attachedClient, error) {
1257 initialTabIndex := -1
1258 if h.RemoteTarget != nil {
1259 var ok bool
1260 initialTabIndex, ok = remoteTargetTabIndexLocked(sess, *h.RemoteTarget)
1261 if !ok {
1262 if tr != nil {
1263 _ = tr.Close()
1264 }
1265 d.mu.Unlock()
1266 return nil, &protoErr{ports.ErrNoSuchTarget, "remote tab no longer exists"}
1267 }
1268 }
1269 if h.PreferredTabID != "" {
1270 // Route memory is a best-effort attachment cursor, not attach authority.
1271 // It overrides point-in-time picker tab metadata when still present and
1272 // otherwise falls back to the session's normal first-tab repair.
1273 initialTabIndex = preferredTabIndex(sess, h.PreferredTabID)
1274 }
1275 // Session state is the sole source for future PTY children. Update it before
1276 // publishing the attachment; existing PTYs keep their original environment.
1277 // A picker handoff deliberately leaves the daemon-owned environment and CWD
1278 // untouched, even though Hello retains those fields for direct CLI clients.
1279 sess.mu.Lock()
1280 if h.EnvironmentPolicy != ports.EnvironmentPolicyDaemonOwned {
1281 sess.env = copyEnvironment(h.Env)
1282 }
1283 sess.terminal = term
1284 sess.mu.Unlock()
1285 opts := attachClientOptions{
1286 clientID: h.ClientID,
1287 resumeCapable: true,
1288 maxOutputInFlight: normalizeOutputWindow(h.MaxOutputInFlight),
1289 navigationCapabilities: h.NavigationCapabilities,
1290 startupOverlay: h.StartupOverlay,
1291 remoteOrigin: h.RemoteOrigin,
1292 }
1293 if opts.remoteOrigin == "" && h.RemoteTarget != nil {
1294 opts.remoteOrigin = h.RemoteTarget.DisplayOrigin
1295 }
1296 ac := d.prepareAttachedClientLocked(tr, sz, opts)
1297 d.mu.Unlock()
1298 result, err := d.transitionAttachment(attachmentTransitionRequest{
1299 target: sess,
1300 next: ac,
1301 expectedTransport: ac.transportSnapshot(),
1302 activateTargetTab: initialTabIndex >= 0,
1303 targetTabIndex: initialTabIndex,
1304 ready: false,
1305 })
1306 if err != nil {
1307 if tr != nil {
1308 _ = tr.Close()
1309 }
1310 return nil, err
1311 }
1312 d.finishAttachedClient(sess, ac, opts)
1313 d.deferAttachmentTransitionCleanups(result)
1314 return ac, nil
1315}
1316
1317func preferredTabIndex(sess *session, preferred domain.TabStableID) int {
1318 if sess == nil || preferred == "" {
1319 return -1
1320 }
1321 sess.mu.Lock()
1322 defer sess.mu.Unlock()
1323 for i, candidate := range sess.tabs {
1324 if candidate != nil && domain.TabStableID(candidate.stableID) == preferred {
1325 return i
1326 }
1327 }
1328 return -1
1329}
1330
1331func (d *Daemon) waitForTargetRestore(ctx context.Context, name string) error {
1332 d.mu.Lock()
1333 var (
1334 done chan struct{}
1335 stopped stoppedSession
1336 ok bool
1337 )
1338 if sess := d.findByNameLocked(name); sess != nil {
1339 sess.mu.Lock()
1340 done = sess.restoreDone
1341 sess.mu.Unlock()
1342 } else {
1343 stopped, ok = d.stopped[name]
1344 if !ok || stopped.purging {
1345 d.mu.Unlock()
1346 return nil
1347 }
1348 done = stopped.restoreDone
1349 }
1350 d.mu.Unlock()
1351
1352 if done != nil {
1353 select {
1354 case <-done:
1355 case <-ctx.Done():
1356 return ctx.Err()
1357 }
1358 }
1359
1360 d.mu.Lock()
1361 defer d.mu.Unlock()
1362 if d.findByNameLocked(name) != nil {
1363 return nil
1364 }
1365 stopped, ok = d.stopped[name]
1366 if !ok || stopped.record.Name == "" {
1367 return nil
1368 }
1369 if stopped.state == ports.SessionBroken {
1370 return &protoErr{ports.ErrInternal, "session durable state is broken: " + name}
1371 }
1372 if stopped.record.Committed == nil {
1373 return nil
1374 }
1375 return &protoErr{ports.ErrInternal, "session was not restored into this daemon: " + name}
1376}
1377
1378// route resolves a Hello to a session and a freshly attached client, creating
1379// the session for ephemeral/new intents. Direct package callers retain the
1380// daemon lifetime context; inbound handshakes use routeWithContext.
1381func (d *Daemon) validateExactSessionTargetLocked(target ports.ExactSessionTarget) error {
1382 if sess := d.findByNameLocked(target.SessionName); sess != nil {
1383 if sess.incarnation != target.LifecycleID {
1384 return &protoErr{ports.ErrNoSuchSession, "session lifecycle has changed: " + target.SessionName}
1385 }
1386 return nil
1387 }
1388 stopped, ok := d.stopped[target.SessionName]
1389 if !ok || stopped.purging || stopped.incarnation != target.LifecycleID {
1390 return &protoErr{ports.ErrNoSuchSession, "no such session lifecycle: " + target.SessionName}
1391 }
1392 return nil
1393}
1394
1395func (d *Daemon) validateExactSessionTarget(ctx context.Context, target ports.ExactSessionTarget) error {
1396 if err := target.Validate(); err != nil {
1397 return &protoErr{ports.ErrNoSuchSession, "invalid exact session target"}
1398 }
1399 if err := d.waitForTargetRestore(ctx, target.SessionName); err != nil {
1400 return err
1401 }
1402
1403 d.mu.Lock()
1404 defer d.mu.Unlock()
1405 return d.validateExactSessionTargetLocked(target)
1406}
1407
1408func (d *Daemon) route(h ports.Hello, tr ports.Transport) (*session, *attachedClient, error) {
1409 ctx := d.serveCtx
1410 if ctx == nil {
1411 ctx = context.Background()
1412 }
1413 return d.routeWithContext(ctx, h, tr)
1414}
1415
1416func (d *Daemon) routeWithContext(ctx context.Context, h ports.Hello, tr ports.Transport) (*session, *attachedClient, error) {
1417 if ctx == nil {
1418 ctx = context.Background()
1419 }
1420 if err := ctx.Err(); err != nil {
1421 return nil, nil, err
1422 }
1423 sz := h.Size
1424 if !sz.Valid() {
1425 return nil, nil, &protoErr{ports.ErrInternal, "invalid terminal size"}
1426 }
1427 term := terminalEnv{TrueColor: h.TrueColor}
1428 if h.Intent == ports.IntentResume && h.ResumeToken == 0 {
1429 return nil, nil, &protoErr{ports.ErrNoSuchSession, "resume token is required"}
1430 }
1431 if h.ExactTarget != nil && h.ResumeToken == 0 && h.Name != h.ExactTarget.SessionName {
1432 return nil, nil, &protoErr{ports.ErrNoSuchSession, "exact session target name mismatch"}
1433 }
1434 if h.RemoteTarget != nil && h.ResumeToken == 0 {
1435 return d.routeRemoteTargetWithContext(ctx, h, tr)
1436 }
1437 if h.ExactTarget != nil && h.ResumeToken == 0 {
1438 if err := d.validateExactSessionTarget(ctx, *h.ExactTarget); err != nil {
1439 return nil, nil, err
1440 }
1441 }
1442
1443 // A non-zero token is an authoritative resume credential. If it is unknown,
1444 // expired, or raced with lifecycle teardown, fail closed instead of routing
1445 // the Hello as an ordinary attach that could create or replace ownership.
1446 // The one legitimate pre-park race is a still-active same-client credential
1447 // for the requested session; hand that into the park/resume lifecycle.
1448 if h.ResumeToken != 0 {
1449 d.mu.Lock()
1450 parkedAtStart := d.parked[h.ResumeToken]
1451 d.mu.Unlock()
1452 if parkedAtStart == nil {
1453 if sess, ac, ok, err := d.resumeLiveAttachment(h, tr, sz); err != nil {
1454 if errors.Is(err, errResumeTokenLifecycleRace) {
1455 // Live recovery parked then lost a competing resumeParked
1456 // race; keep the fail-closed wire response instead of
1457 // leaking the internal sentinel to handleHello as ErrInternal.
1458 return nil, nil, &protoErr{ports.ErrNoSuchSession, "resume token is no longer valid"}
1459 }
1460 return nil, nil, err
1461 } else if ok {
1462 return sess, ac, nil
1463 }
1464 return nil, nil, &protoErr{ports.ErrNoSuchSession, "resume token is no longer valid"}
1465 }
1466 if sess, ac, ok, err := d.resumeParked(h, tr, sz); err == nil {
1467 if ok {
1468 return sess, ac, nil
1469 }
1470 return nil, nil, &protoErr{ports.ErrNoSuchSession, "resume token is no longer valid"}
1471 } else if errors.Is(err, errResumeTokenLifecycleRace) {
1472 // The parked entry was replaced while this handshake waited for
1473 // its send lock. Fail closed on the original credential rather
1474 // than falling through to ordinary attach/create routing.
1475 return nil, nil, &protoErr{ports.ErrNoSuchSession, "resume token is no longer valid"}
1476 } else {
1477 return nil, nil, err
1478 }
1479 }
1480 if h.Intent == ports.IntentResume || h.Intent == ports.IntentAttach {
1481 if err := d.waitForTargetRestore(ctx, h.Name); err != nil {
1482 return nil, nil, err
1483 }
1484 }
1485 if h.EnvironmentPolicy == ports.EnvironmentPolicyDaemonOwned && h.RemoteTarget == nil &&
1486 (h.Intent == ports.IntentNew || h.Intent == ports.IntentEphemeral) {
1487 return nil, nil, &protoErr{ports.ErrNoSuchTarget, "daemon-owned environment requires an exact remote target"}
1488 }
1489 d.mu.Lock()
1490 // Shutdown/create interlock: once shutdown has begun (last session removed,
1491 // or shutdownAll started) no new session may be created and no attach may
1492 // proceed — the teardown snapshot has already been (or is being) taken, so
1493 // anything inserted now would leak its PTY and hang Serve.
1494 if d.closing {
1495 d.mu.Unlock()
1496 return nil, nil, &protoErr{ports.ErrServerShutdown, "daemon is shutting down"}
1497 }
1498 // The first exact-target check happens before restore I/O. Recheck while
1499 // holding d.mu so a same-name lifecycle replacement cannot slip between
1500 // validation and attachment creation.
1501 if h.ExactTarget != nil && h.ResumeToken == 0 {
1502 if err := d.validateExactSessionTargetLocked(*h.ExactTarget); err != nil {
1503 d.mu.Unlock()
1504 return nil, nil, err
1505 }
1506 }
1507 switch h.Intent {
1508 case ports.IntentEphemeral:
1509 name := d.allocEphemeralNameLocked()
1510 sess, err := d.createSessionLockedWithMode(name, true, h.Cwd, sz, term, h.Env)
1511 if err != nil {
1512 d.mu.Unlock()
1513 return nil, nil, err
1514 }
1515 ac, err := d.finishRouteAttach(sess, tr, sz, term, h, true, true)
1516 return sess, ac, err
1517
1518 case ports.IntentNew:
1519 if h.Name == "" {
1520 d.mu.Unlock()
1521 return nil, nil, &protoErr{ports.ErrInvalidSessionName, "empty session name"}
1522 }
1523 if err := domain.ValidateSessionName(h.Name); err != nil {
1524 d.mu.Unlock()
1525 return nil, nil, &protoErr{ports.ErrInvalidSessionName, err.Error()}
1526 }
1527 if d.nameLiveOrStoppedLocked(h.Name) {
1528 d.mu.Unlock()
1529 return nil, nil, &protoErr{ports.ErrNameTaken, "session name already in use: " + h.Name}
1530 }
1531 sess, err := d.createSessionLockedWithMode(h.Name, false, h.Cwd, sz, term, h.Env)
1532 if err != nil {
1533 d.mu.Unlock()
1534 return nil, nil, err
1535 }
1536 ac, err := d.finishRouteAttach(sess, tr, sz, term, h, true, true)
1537 return sess, ac, err
1538
1539 case ports.IntentAttach:
1540 sess := d.findByNameLocked(h.Name)
1541 created := false
1542 if sess == nil {
1543 stopped, ok := d.stopped[h.Name]
1544 if !ok || stopped.purging {
1545 d.mu.Unlock()
1546 return nil, nil, &protoErr{ports.ErrNoSuchSession, "no such session: " + h.Name}
1547 }
1548 cwd := d.dirOrHome(stopped.cwd)
1549 env := h.Env
1550 if h.EnvironmentPolicy == ports.EnvironmentPolicyDaemonOwned {
1551 env = copyEnvironment(d.baseEnv)
1552 }
1553 var err error
1554 sess, err = d.createSessionLockedWithMode(h.Name, false, cwd, sz, term, env, stopped.tabNames)
1555 if err != nil {
1556 d.mu.Unlock()
1557 return nil, nil, err
1558 }
1559 created = true
1560 }
1561 ac, err := d.finishRouteAttach(sess, tr, sz, term, h, created, false)
1562 return sess, ac, err
1563
1564 default:
1565 d.mu.Unlock()
1566 return nil, nil, &protoErr{ports.ErrInternal, "unknown intent"}
1567 }
1568}