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 paneProcessCtx, paneProcessCancel := context.WithCancel(context.Background())
592 d := &Daemon{
593 sessions: make(map[domain.SessionID]*session),
594 stopped: make(map[string]stoppedSession),
595 creating: make(map[string]struct{}),
596 parked: make(map[uint64]*parkedAttachment),
597 parking: make(map[uint64]*parkingAttachment),
598 paneProcessCtx: paneProcessCtx,
599 paneProcessCancel: paneProcessCancel,
600 ptys: ptys,
601 clock: clock,
602 log: log,
603 baseEnv: os.Environ(),
604 shell: defaultShellCommand,
605 dirOrHome: dirOrHome,
606 done: make(chan struct{}),
607 restoreDone: make(chan struct{}),
608 animWake: make(chan struct{}, 1),
609 snapshotJobs: make(chan *snapshotCapture, snapshotQueueCapacity),
610 snapshotAdmitted: make(map[*snapshotCapture]struct{}),
611 snapshotWake: make(chan struct{}, 1),
612 notices: newNoticeCenter(),
613 remoteCatalog: newRemoteCatalogState(),
614 resumeParkGrace: defaultResumeParkGrace,
615 barScripts: &barScriptState{
616 cfg: barConfigFromDomain(domain.Defaults().Bar),
617 outputs: make(map[domain.SessionID]barScriptOutputs),
618 lastRefresh: make(map[domain.SessionID]time.Time),
619 lastContext: make(map[domain.SessionID]barScriptContext),
620 running: make(map[domain.SessionID]bool),
621 reload: make(chan struct{}, 1),
622 },
623 }
624 defaults := domain.Defaults()
625 defaultFloating := defaults.Floating
626 d.floatingConfig.Store(&defaultFloating)
627 defaultCopy := defaults.Copy
628 d.copyConfig.Store(&defaultCopy)
629 defaultPalette := defaults.Palette
630 d.paletteConfig.Store(&defaultPalette)
631 for _, o := range opts {
632 o(d)
633 }
634 if d.dirOrHome == nil {
635 d.dirOrHome = dirOrHome
636 }
637 if d.bindings.Load() == nil {
638 d.bindings.Store(keys.DefaultBindings())
639 }
640 if d.codeOverrides.Load() == nil {
641 empty := map[string]string{}
642 d.codeOverrides.Store(&empty)
643 }
644 if d.restoreProcessAllowlist.Load() == nil {
645 allow := buildRestoreProcessAllowlist(domain.DefaultSnapshotRestoreProcesses())
646 d.restoreProcessAllowlist.Store(&allow)
647 }
648 records := d.catalogueRecords
649 var maxSeq uint64
650 var maxCreatedAt int64
651 hasCreatedAt := false
652 for _, r := range records {
653 if d.catalogueRecordsProvided {
654 state, done := initialSessionState(r)
655 d.stopped[r.Name] = stoppedSessionFromRecord(r, state, done)
656 } else {
657 d.stopped[r.Name] = stoppedSessionFromRecord(r, ports.SessionDown, nil)
658 }
659 if !hasCreatedAt || r.CreatedAt > maxCreatedAt {
660 maxCreatedAt = r.CreatedAt
661 hasCreatedAt = true
662 }
663 if r.LastUsedSeq > maxSeq {
664 maxSeq = r.LastUsedSeq
665 }
666 }
667 if hasCreatedAt {
668 d.lastAllocatedCreatedAt = maxCreatedAt
669 }
670 d.mruSeq.Store(maxSeq)
671 d.loadRemoteCatalogCache()
672 return d
673}
674
675// Serve runs the accept loop over l, owning it for the loop's lifetime. It
676// returns when the last session is removed or ctx is cancelled; on the latter
677// path attached clients are detached with ReasonServerShutdown.
678func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error {
679 d.serveCtx, d.serveCancel = context.WithCancel(ctx)
680 defer d.serveCancel()
681 d.hardCtx, d.hardCancel = context.WithCancel(context.Background())
682 defer d.hardCancel()
683
684 d.sessWg.Go(func() {
685 d.attentionAnimator(d.serveCtx)
686 })
687 d.sessWg.Go(func() {
688 d.barScriptPoller(d.serveCtx)
689 })
690 if d.persistEnabled {
691 d.sessWg.Go(func() {
692 d.cwdSampler(d.serveCtx)
693 })
694 }
695 if d.snapsEnabled {
696 d.startSnapshotEncodeWorker()
697 d.sessWg.Go(func() {
698 d.snapshotRepositorySaver(d.serveCtx)
699 })
700 d.startSnapshotRestoration()
701 } else {
702 d.closeRestoreDone()
703 }
704 // Break the accept loop when either the parent context is cancelled or the
705 // registry drains to empty: both close the listener, which fails Accept.
706 go func() {
707 select {
708 case <-d.serveCtx.Done():
709 case <-d.done:
710 }
711 _ = l.Close()
712 }()
713
714 for {
715 tr, err := l.Accept()
716 if err != nil {
717 if d.serveCtx.Err() != nil {
718 d.log.Info("accept loop exiting", "err", err, "reason", "context canceled")
719 } else {
720 d.log.Warn("accept loop exiting", "err", err)
721 }
722 break
723 }
724 d.connWg.Go(func() {
725 d.handleConn(tr)
726 })
727 }
728
729 // Tear down. shutdownAll marks the daemon closing (route now rejects any
730 // racing Hello) and kills every session: PTYs are closed and contexts
731 // cancelled unconditionally, with the Detached notices sent asynchronously
732 // under a deadline. Wait for those notices (bounded — a wedged client's
733 // transport is force-closed by the deadline) before hard-closing the
734 // remaining parked connections, so graceful notices are never raced by the
735 // force-close. Finally drain the conn handlers, run one defensive sweep in
736 // case any ordering ever leaves a session behind, and join the session
737 // goroutines (readers unblock via pty.Close, coordinators via ctx cancel).
738 // One deadline owns every terminal checkpoint wait and the final worker
739 // join. A repository is allowed to ignore cancellation, so spending a fresh
740 // interval at either stage would make shutdown exceed its documented bound.
741 var snapshotDeadline *snapshotShutdownDeadline
742 if d.snapsEnabled {
743 snapshotDeadline = newSnapshotShutdownDeadline(d.clock)
744 defer snapshotDeadline.stop()
745 }
746 d.shutdownAllWithSnapshotDeadline(ports.ReasonServerShutdown, snapshotDeadline)
747 d.waitNotifies()
748 d.hardCancel()
749 d.serveCancel()
750 // Wait for handlers before flushing snapshots: a handler that already
751 // entered killSession may still submit its terminal capture after the
752 // registry snapshot has been removed.
753 d.connWg.Wait()
754 d.attachmentCleanupWg.Wait()
755 // Forced terminal checkpoints run before shutdown stops producers or the
756 // worker. StopDurableWriters reports every session whose final capture still
757 // owns work when the shared checkpoint budget expires. Persist one notice per
758 // affected session, then retain ownership until every writer actually exits.
759 stopCtx, stopCancel := snapshotStopContext(snapshotDeadline)
760 timedOutSessions := d.StopDurableWriters(stopCtx)
761 stopCancel()
762 for _, name := range timedOutSessions {
763 d.persistShutdownSnapshotFailure(name, context.DeadlineExceeded)
764 }
765 d.WaitDurableWriters()
766 d.shutdownAllWithSnapshotDeadline(ports.ReasonServerShutdown, snapshotDeadline)
767 d.waitSessionWorkersWithSnapshotDeadline(snapshotDeadline)
768 d.waitNotifies()
769 if err := d.flushCatalogue(); err != nil {
770 d.log.Warn("flushing session catalogue at shutdown failed", "err", err)
771 }
772 if d.catalogue != nil {
773 if err := d.catalogue.Close(); err != nil {
774 d.log.Warn("closing session catalogue failed", "err", err)
775 }
776 }
777 return nil
778}
779
780// shutdownAll marks the daemon closing and kills every live session. Setting
781// closing under the same lock as the snapshot guarantees no session can be
782// inserted after the snapshot: route rejects once closing is set, and both run
783// under d.mu. killSession (which relocks) runs after the lock is released.
784func (d *Daemon) shutdownAll(reason uint8) (checkpointIncomplete bool) {
785 return d.shutdownAllWithSnapshotDeadline(reason, nil)
786}
787
788func (d *Daemon) shutdownAllWithSnapshotDeadline(reason uint8, deadline *snapshotShutdownDeadline) (checkpointIncomplete bool) {
789 d.cancelRemotePickerRefresh()
790 d.closeMoveLifecycles()
791 d.mu.Lock()
792 d.closing = true
793 d.purgeAllParkingLocked()
794 parkedRetirements := d.purgeAllParkedLocked()
795 snapshot := d.sessionsSnapshotLocked()
796 empty := len(snapshot) == 0
797 d.mu.Unlock()
798 d.finishParkedAttachmentRetirements(parkedRetirements)
799 d.log.Info("graceful shutdown begin", "reason", reason, "sessions", len(snapshot))
800 if empty {
801 d.doneOnce.Do(func() { close(d.done) })
802 return false
803 }
804 for _, s := range snapshot {
805 // Cancellation and PTY closure must not wait behind a teardown owner that
806 // is blocked in snapshot publication or purge work.
807 s.mu.Lock()
808 name := s.name
809 ephemeral := s.ephemeral
810 s.mu.Unlock()
811 s.stopInMemoryLifecycle()
812 if err := d.killSessionWithSnapshotDeadline(s, reason, false, deadline, nil); err != nil {
813 checkpointIncomplete = true
814 d.log.Error("closing session with unpersisted terminal state", "err", err)
815 }
816 if !ephemeral && deadline != nil {
817 select {
818 case <-deadline.Done():
819 checkpointIncomplete = true
820 d.persistShutdownSnapshotFailure(name, context.DeadlineExceeded)
821 default:
822 }
823 }
824 }
825 // Signal Serve even when a concurrent role transition made this initial
826 // pass abort; Serve owns the bounded defensive teardown passes.
827 d.doneOnce.Do(func() { close(d.done) })
828 return checkpointIncomplete
829}
830
831// waitSessionWorkersWithSnapshotDeadline joins in-memory workers when possible,
832// but never extends Serve beyond the shared repository shutdown budget. By this
833// point connection handlers are joined and every session has been cancelled, so
834// no new session worker can be registered while the waiter is active.
835func (d *Daemon) waitSessionWorkersWithSnapshotDeadline(deadline *snapshotShutdownDeadline) {
836 done := make(chan struct{})
837 go func() {
838 d.sessWg.Wait()
839 close(done)
840 }()
841 if deadline == nil {
842 <-done
843 return
844 }
845 select {
846 case <-done:
847 case <-deadline.Done():
848 }
849}
850
851// snapshotShutdownDeadline is a single-use shutdown budget shared by forced
852// checkpoints and the final snapshot worker join. Its done channel is never
853// closed by callers, so a detached worker can safely retain session state until
854// an uncooperative repository call returns.
855type snapshotShutdownDeadline struct {
856 done chan struct{}
857 stopCh chan struct{}
858 finished chan struct{}
859 stopOnce sync.Once
860}
861
862func newSnapshotShutdownDeadline(clock ports.Clock) *snapshotShutdownDeadline {
863 deadline := &snapshotShutdownDeadline{done: make(chan struct{}), stopCh: make(chan struct{}), finished: make(chan struct{})}
864 timer := clock.NewTimer(snapshotFinalFlushTimeout)
865 go func() {
866 defer close(deadline.finished)
867 select {
868 case <-timer.C():
869 close(deadline.done)
870 case <-deadline.stopCh:
871 timer.Stop()
872 }
873 }()
874 return deadline
875}
876
877func (d *snapshotShutdownDeadline) stop() {
878 if d != nil {
879 d.stopOnce.Do(func() {
880 close(d.stopCh)
881 <-d.finished
882 })
883 }
884}
885
886func (d *snapshotShutdownDeadline) Done() <-chan struct{} {
887 if d == nil {
888 return nil
889 }
890 return d.done
891}
892
893func snapshotStopContext(deadline *snapshotShutdownDeadline) (context.Context, context.CancelFunc) {
894 ctx, cancel := context.WithCancel(context.Background())
895 if deadline == nil {
896 return ctx, cancel
897 }
898 select {
899 case <-deadline.Done():
900 cancel()
901 return ctx, cancel
902 default:
903 }
904 stop := make(chan struct{})
905 go func() {
906 select {
907 case <-deadline.Done():
908 cancel()
909 case <-stop:
910 }
911 }()
912 var stopOnce sync.Once
913 return ctx, func() {
914 stopOnce.Do(func() { close(stop) })
915 cancel()
916 }
917}
918
919// registerSessionLocked publishes an exact attachment-session identity. Caller
920// holds d.mu.
921func (d *Daemon) registerSessionLocked(entry *session) bool {
922 if entry == nil {
923 return false
924 }
925 core := entry.core()
926 if core == nil || core.id == "" || d.sessions[core.id] != nil {
927 return false
928 }
929 d.sessions[core.id] = entry
930 return true
931}
932
933// unregisterSessionLocked removes only the exact registered identity. Caller
934// holds d.mu.
935func (d *Daemon) unregisterSessionLocked(entry *session) bool {
936 if entry == nil {
937 return false
938 }
939 core := entry.core()
940 if core == nil || d.sessions[core.id] != entry {
941 return false
942 }
943 delete(d.sessions, core.id)
944 return true
945}
946
947func (d *Daemon) sessionsSnapshotLocked() []*session {
948 return sessionsSnapshot(d.sessions)
949}
950
951// handleConn reads the first frame off a fresh connection and routes it. A
952// context watcher closes the transport on serve-context cancel or handshake
953// timeout so blocked transport operations unwind.
954func (d *Daemon) handleConn(tr ports.Transport) {
955 handshakeCtx, timedOut, finishHandshake := d.newHandshakeContext(d.hardCtx)
956 stopTransport := watchHandshakeTransport(handshakeCtx, tr)
957 defer finishHandshake()
958 defer stopTransport()
959
960 var first ports.Frame
961 err := boundedHandshakeOperation(handshakeCtx, tr, func() error {
962 var recvErr error
963 first, recvErr = tr.Recv()
964 return recvErr
965 })
966 if err != nil {
967 err = handshakeContextError(d.hardCtx, timedOut, err)
968 d.log.Warn("connection closed before hello", "err", err)
969 _ = tr.Close()
970 return
971 }
972 switch first.Type {
973 case ports.MsgList:
974 stopTransport()
975 finishHandshake()
976 d.handleList(tr)
977 case ports.MsgCommand:
978 stopTransport()
979 finishHandshake()
980 if err := d.handleCommand(tr, first); err != nil {
981 d.log.Warn("command handler failed", "err", err)
982 }
983 case ports.MsgRemotePreviewRequest:
984 stopTransport()
985 finishHandshake()
986 if err := d.handleRemotePreview(tr, first); err != nil {
987 d.log.Warn("remote preview handler failed", "err", err)
988 }
989 case ports.MsgKill:
990 stopTransport()
991 finishHandshake()
992 d.handleKill(tr, first)
993 case ports.MsgHello:
994 d.handleHelloWithContext(handshakeCtx, timedOut, stopTransport, finishHandshake, tr, first)
995 default:
996 d.log.Warn("hello rejected", "err", "expected hello", "type", first.Type)
997 _ = boundedHandshakeOperation(handshakeCtx, tr, func() error {
998 return tr.Send(frameError(ports.ErrInternal, "expected hello"))
999 })
1000 stopTransport()
1001 finishHandshake()
1002 _ = tr.Close()
1003 }
1004}
1005
1006// handleList replies with the current session listing and closes the (one-shot
1007// control) connection.
1008func (d *Daemon) handleList(tr ports.Transport) {
1009 defer func() { _ = tr.Close() }()
1010
1011 d.mu.Lock()
1012 infos := make([]ports.SessionInfo, 0, len(d.sessions)+len(d.stopped))
1013 liveNames := make(map[string]struct{}, len(d.sessions))
1014 for _, s := range d.sessions {
1015 if s == nil {
1016 continue
1017 }
1018 s.mu.Lock()
1019 info := ports.SessionInfo{
1020 SessionID: string(s.id),
1021 Name: s.name,
1022 State: ports.SessionUp,
1023 Ephemeral: s.ephemeral,
1024 Tabs: uint16(len(s.tabs)),
1025 Attached: len(s.attachments) != 0,
1026 }
1027 liveNames[s.name] = struct{}{}
1028 s.mu.Unlock()
1029 infos = append(infos, info)
1030 }
1031 for name, stopped := range d.stopped {
1032 if _, live := liveNames[name]; live {
1033 continue
1034 }
1035 state := ports.SessionDown
1036 if stopped.purging || stopped.state == ports.SessionBroken {
1037 // Purge is the dominant externally visible state: restoration must
1038 // never make a deletion-reserved record appear attachable.
1039 state = ports.SessionBroken
1040 }
1041 infos = append(infos, ports.SessionInfo{Name: name, State: state})
1042 }
1043 d.mu.Unlock()
1044
1045 sort.Slice(infos, func(i, j int) bool { return infos[i].Name < infos[j].Name })
1046 _ = d.boundedControlSend(tr, frameSessions(infos))
1047}
1048
1049// handleKill terminates the requested live session or stopped named session,
1050// or all sessions, and closes the control connection; the resulting EOF is the
1051// client's success signal.
1052func (d *Daemon) handleKill(tr ports.Transport, f ports.Frame) {
1053 defer func() { _ = tr.Close() }()
1054
1055 k, err := ports.UnmarshalKill(f.Payload)
1056 if err != nil {
1057 _ = d.boundedControlSend(tr, frameError(ports.ErrInternal, "malformed kill request"))
1058 return
1059 }
1060 if k.All {
1061 d.shutdownAll(ports.ReasonServerShutdown)
1062 return
1063 }
1064
1065 d.mu.Lock()
1066 target := d.findByNameLocked(k.Name)
1067 if target == nil {
1068 if _, ok := d.stopped[k.Name]; ok {
1069 d.mu.Unlock()
1070 // Stopped sessions use the same catalogue-first, incarnation-second
1071 // deletion order as live and offline purges.
1072 if err := d.retryStoppedPurge(k.Name); err != nil {
1073 d.log.Warn("deleting stopped session failed", "err", err, "session", k.Name)
1074 _ = d.boundedControlSend(tr, frameError(ports.ErrInternal, "deleting stopped session failed"))
1075 }
1076 return
1077 }
1078 }
1079 d.mu.Unlock()
1080
1081 if target == nil {
1082 _ = d.boundedControlSend(tr, frameError(ports.ErrNoSuchSession, "no such session: "+k.Name))
1083 return
1084 }
1085 if err := d.killSession(target, ports.ReasonSessionKilled, true); err != nil {
1086 _ = d.boundedControlSend(tr, frameError(ports.ErrInternal, "deleting persisted session failed"))
1087 }
1088}
1089
1090// handleHello runs the attach handshake for direct package callers. Accepted
1091// connections use handleHelloWithContext so the deadline also covers the first
1092// frame read in handleConn.
1093func (d *Daemon) handleHello(tr ports.Transport, f ports.Frame) {
1094 handshakeCtx, timedOut, finishHandshake := d.newHandshakeContext(context.Background())
1095 stopTransport := watchHandshakeTransport(handshakeCtx, tr)
1096 defer finishHandshake()
1097 defer stopTransport()
1098 d.handleHelloWithContext(handshakeCtx, timedOut, stopTransport, finishHandshake, tr, f)
1099}
1100
1101// handleHelloWithContext runs the attach handshake: version check, intent
1102// routing, Welcome, guaranteed first paint, then the per-connection input loop.
1103func (d *Daemon) handleHelloWithContext(handshakeCtx context.Context, timedOut <-chan struct{}, stopHandshakeTransport, finishHandshake func(), tr ports.Transport, f ports.Frame) {
1104 if err := handshakeContextError(context.Background(), timedOut, handshakeCtx.Err()); err != nil {
1105 _ = tr.Close()
1106 return
1107 }
1108 sendHandshake := func(frame ports.Frame) error {
1109 err := boundedHandshakeOperation(handshakeCtx, tr, func() error { return tr.Send(frame) })
1110 if err != nil {
1111 return handshakeContextError(context.Background(), timedOut, err)
1112 }
1113 return nil
1114 }
1115 h, err := ports.UnmarshalHello(f.Payload)
1116 if err != nil {
1117 if version, ok := ports.PeekHelloVersion(f.Payload); ok && version != ports.ProtocolVersion {
1118 d.log.Warn("hello rejected", "err", "protocol version mismatch", "version", version, "expected", ports.ProtocolVersion)
1119 _ = sendHandshake(frameError(ports.ErrVersionMismatch, "protocol version mismatch"))
1120 } else {
1121 d.log.Warn("hello rejected", "err", err)
1122 _ = sendHandshake(frameError(ports.ErrInternal, "malformed hello"))
1123 }
1124 _ = tr.Close()
1125 return
1126 }
1127 if h.Version != ports.ProtocolVersion {
1128 d.log.Warn("hello rejected", "err", "protocol version mismatch", "version", h.Version, "expected", ports.ProtocolVersion, "intent", h.Intent, "session", h.Name)
1129 _ = sendHandshake(frameError(ports.ErrVersionMismatch, "protocol version mismatch"))
1130 _ = tr.Close()
1131 return
1132 }
1133
1134 sess, ac, rerr := d.routeWithContext(handshakeCtx, h, tr)
1135 if rerr != nil {
1136 d.log.Warn("hello rejected", "err", rerr, "intent", h.Intent, "session", h.Name)
1137 if pe, ok := errors.AsType[*protoErr](rerr); ok {
1138 _ = sendHandshake(frameError(pe.code, pe.text))
1139 } else {
1140 _ = sendHandshake(frameError(ports.ErrInternal, rerr.Error()))
1141 }
1142 _ = tr.Close()
1143 return
1144 }
1145
1146 welcomeSent := false
1147 //nolint:contextcheck // handshakeCtx is intentionally not propagated: cancellation teardown must complete its geometry and ownership cleanup.
1148 failAttachment := func() {
1149 d.failHandshakeAttachment(sess, ac, tr, welcomeSent)
1150 }
1151 if err := handshakeCtx.Err(); err != nil {
1152 failAttachment()
1153 return
1154 }
1155 expected := ac.transportSnapshot()
1156 welcomeToken := sess.attachmentToken(ac, tr)
1157 welcomeTicket, admitted := ac.beginAttachmentEffect(welcomeToken)
1158 if expected.transport != tr || !admitted || welcomeToken.ac == nil {
1159 if admitted {
1160 welcomeTicket.End()
1161 }
1162 failAttachment()
1163 return
1164 }
1165 if err := handshakeCtx.Err(); err != nil {
1166 welcomeTicket.End()
1167 failAttachment()
1168 return
1169 }
1170 welcomeDone, welcomeErr := boundedHandshakeOperationTracked(handshakeCtx, tr, func() error {
1171 frame, err := frameWelcome(sess, d.resumeTokenSnapshot(ac))
1172 if err != nil {
1173 return err
1174 }
1175 return ac.sendExpectedTransportForAttachment(expected, frame, welcomeTicket)
1176 })
1177 if welcomeErr != nil {
1178 <-welcomeDone
1179 welcomeTicket.End()
1180 failAttachment()
1181 return
1182 }
1183 welcomeSent = true
1184 // Release Welcome's effect before discovering post-handshake authority so a
1185 // replacement blocked behind the send can publish its generation and lease.
1186 welcomeTicket.End()
1187 postWelcomeToken, postWelcomeTicket, admitted := ac.beginCurrentAttachmentEffectContext(handshakeCtx, sess, tr)
1188 if !admitted {
1189 failAttachment()
1190 return
1191 }
1192 postWelcomeLease := postWelcomeToken.lease
1193 if postWelcomeToken.ac == nil {
1194 postWelcomeTicket.End()
1195 failAttachment()
1196 return
1197 }
1198 postWelcomeRC := sess.renderCoordinator()
1199 if postWelcomeLease != nil && (postWelcomeRC == nil || !postWelcomeRC.markAttachmentReady(postWelcomeLease)) {
1200 postWelcomeTicket.End()
1201 // The attachment was detached while Welcome was in flight; never let
1202 // this stale handshake emit an Output frame.
1203 failAttachment()
1204 return
1205 }
1206 paintToken := sess.attachmentToken(ac, tr)
1207 painted := make(chan bool, 1)
1208 paintDone, paintErr := boundedHandshakeOperationTracked(handshakeCtx, tr, func() error {
1209 //nolint:contextcheck // firstPaintForTransition is capability-bounded; the surrounding handshake operation owns cancellation.
1210 painted <- d.firstPaintForTransition(paintToken)
1211 return nil
1212 })
1213 if paintErr != nil {
1214 <-paintDone
1215 postWelcomeTicket.End()
1216 failAttachment()
1217 return
1218 }
1219 if !<-painted || handshakeCtx.Err() != nil {
1220 postWelcomeTicket.End()
1221 failAttachment()
1222 return
1223 }
1224 if !d.commitResumeClaim(ac) {
1225 postWelcomeTicket.End()
1226 failAttachment()
1227 return
1228 }
1229 if ac.startupOverlay == ports.StartupOverlaySessionPicker {
1230 d.enterPicker(sess, ac)
1231 }
1232 postWelcomeTicket.End()
1233 stopHandshakeTransport()
1234 finishHandshake()
1235 d.runConnLoop(ac)
1236 _ = tr.Close()
1237}
1238
1239// protoErr is a session-level rejection carrying a wire ErrorMsg code.
1240type protoErr struct {
1241 code uint16
1242 text string
1243}
1244
1245func (e *protoErr) Error() string { return e.text }
1246
1247// finishAttach completes an attachment prepared while d.mu is held. Every
1248// caller must hold d.mu on entry; finishAttach transfers ownership by
1249// unlocking d.mu before returning, including every error path. It publishes
1250// terminal and role state before releasing d.mu, then defers coordinator
1251// cleanup so obsolete workers never delay the new handshake.
1252func (d *Daemon) finishAttach(sess *session, tr ports.Transport, sz domain.Size, term terminalEnv, h ports.Hello) (*attachedClient, error) {
1253 initialTabIndex := -1
1254 if h.RemoteTarget != nil {
1255 var ok bool
1256 initialTabIndex, ok = remoteTargetTabIndexLocked(sess, *h.RemoteTarget)
1257 if !ok {
1258 if tr != nil {
1259 _ = tr.Close()
1260 }
1261 d.mu.Unlock()
1262 return nil, &protoErr{ports.ErrNoSuchTarget, "remote tab no longer exists"}
1263 }
1264 }
1265 if h.PreferredTabID != "" {
1266 // Route memory is a best-effort attachment cursor, not attach authority.
1267 // It overrides point-in-time picker tab metadata when still present and
1268 // otherwise falls back to the session's normal first-tab repair.
1269 initialTabIndex = preferredTabIndex(sess, h.PreferredTabID)
1270 }
1271 // Session state is the sole source for future PTY children. Update it before
1272 // publishing the attachment; existing PTYs keep their original environment.
1273 // A picker handoff deliberately leaves the daemon-owned environment and CWD
1274 // untouched, even though Hello retains those fields for direct CLI clients.
1275 sess.mu.Lock()
1276 if h.EnvironmentPolicy != ports.EnvironmentPolicyDaemonOwned {
1277 sess.env = copyEnvironment(h.Env)
1278 }
1279 sess.terminal = term
1280 sess.mu.Unlock()
1281 opts := attachClientOptions{
1282 clientID: h.ClientID,
1283 resumeCapable: true,
1284 maxOutputInFlight: normalizeOutputWindow(h.MaxOutputInFlight),
1285 navigationCapabilities: h.NavigationCapabilities,
1286 startupOverlay: h.StartupOverlay,
1287 remoteOrigin: h.RemoteOrigin,
1288 }
1289 if opts.remoteOrigin == "" && h.RemoteTarget != nil {
1290 opts.remoteOrigin = h.RemoteTarget.DisplayOrigin
1291 }
1292 ac := d.prepareAttachedClientLocked(tr, sz, opts)
1293 d.mu.Unlock()
1294 result, err := d.transitionAttachment(attachmentTransitionRequest{
1295 target: sess,
1296 next: ac,
1297 expectedTransport: ac.transportSnapshot(),
1298 activateTargetTab: initialTabIndex >= 0,
1299 targetTabIndex: initialTabIndex,
1300 ready: false,
1301 })
1302 if err != nil {
1303 if tr != nil {
1304 _ = tr.Close()
1305 }
1306 return nil, err
1307 }
1308 d.finishAttachedClient(sess, ac, opts)
1309 d.deferAttachmentTransitionCleanups(result)
1310 return ac, nil
1311}
1312
1313func preferredTabIndex(sess *session, preferred domain.TabStableID) int {
1314 if sess == nil || preferred == "" {
1315 return -1
1316 }
1317 sess.mu.Lock()
1318 defer sess.mu.Unlock()
1319 for i, candidate := range sess.tabs {
1320 if candidate != nil && domain.TabStableID(candidate.stableID) == preferred {
1321 return i
1322 }
1323 }
1324 return -1
1325}
1326
1327func (d *Daemon) waitForTargetRestore(ctx context.Context, name string) error {
1328 d.mu.Lock()
1329 var (
1330 done chan struct{}
1331 stopped stoppedSession
1332 ok bool
1333 )
1334 if sess := d.findByNameLocked(name); sess != nil {
1335 sess.mu.Lock()
1336 done = sess.restoreDone
1337 sess.mu.Unlock()
1338 } else {
1339 stopped, ok = d.stopped[name]
1340 if !ok || stopped.purging {
1341 d.mu.Unlock()
1342 return nil
1343 }
1344 done = stopped.restoreDone
1345 }
1346 d.mu.Unlock()
1347
1348 if done != nil {
1349 select {
1350 case <-done:
1351 case <-ctx.Done():
1352 return ctx.Err()
1353 }
1354 }
1355
1356 d.mu.Lock()
1357 defer d.mu.Unlock()
1358 if d.findByNameLocked(name) != nil {
1359 return nil
1360 }
1361 stopped, ok = d.stopped[name]
1362 if !ok || stopped.record.Name == "" {
1363 return nil
1364 }
1365 if stopped.state == ports.SessionBroken {
1366 return &protoErr{ports.ErrInternal, "session durable state is broken: " + name}
1367 }
1368 if stopped.record.Committed == nil {
1369 return nil
1370 }
1371 return &protoErr{ports.ErrInternal, "session was not restored into this daemon: " + name}
1372}
1373
1374// route resolves a Hello to a session and a freshly attached client, creating
1375// the session for ephemeral/new intents. Direct package callers retain the
1376// daemon lifetime context; inbound handshakes use routeWithContext.
1377func (d *Daemon) validateExactSessionTargetLocked(target ports.ExactSessionTarget) error {
1378 if sess := d.findByNameLocked(target.SessionName); sess != nil {
1379 if sess.incarnation != target.LifecycleID {
1380 return &protoErr{ports.ErrNoSuchSession, "session lifecycle has changed: " + target.SessionName}
1381 }
1382 return nil
1383 }
1384 stopped, ok := d.stopped[target.SessionName]
1385 if !ok || stopped.purging || stopped.incarnation != target.LifecycleID {
1386 return &protoErr{ports.ErrNoSuchSession, "no such session lifecycle: " + target.SessionName}
1387 }
1388 return nil
1389}
1390
1391func (d *Daemon) validateExactSessionTarget(ctx context.Context, target ports.ExactSessionTarget) error {
1392 if err := target.Validate(); err != nil {
1393 return &protoErr{ports.ErrNoSuchSession, "invalid exact session target"}
1394 }
1395 if err := d.waitForTargetRestore(ctx, target.SessionName); err != nil {
1396 return err
1397 }
1398
1399 d.mu.Lock()
1400 defer d.mu.Unlock()
1401 return d.validateExactSessionTargetLocked(target)
1402}
1403
1404func (d *Daemon) route(h ports.Hello, tr ports.Transport) (*session, *attachedClient, error) {
1405 ctx := d.serveCtx
1406 if ctx == nil {
1407 ctx = context.Background()
1408 }
1409 return d.routeWithContext(ctx, h, tr)
1410}
1411
1412func (d *Daemon) routeWithContext(ctx context.Context, h ports.Hello, tr ports.Transport) (*session, *attachedClient, error) {
1413 if ctx == nil {
1414 ctx = context.Background()
1415 }
1416 if err := ctx.Err(); err != nil {
1417 return nil, nil, err
1418 }
1419 sz := h.Size
1420 if !sz.Valid() {
1421 return nil, nil, &protoErr{ports.ErrInternal, "invalid terminal size"}
1422 }
1423 term := terminalEnv{TrueColor: h.TrueColor}
1424 if h.Intent == ports.IntentResume && h.ResumeToken == 0 {
1425 return nil, nil, &protoErr{ports.ErrNoSuchSession, "resume token is required"}
1426 }
1427 if h.ExactTarget != nil && h.ResumeToken == 0 && h.Name != h.ExactTarget.SessionName {
1428 return nil, nil, &protoErr{ports.ErrNoSuchSession, "exact session target name mismatch"}
1429 }
1430 if h.RemoteTarget != nil && h.ResumeToken == 0 {
1431 return d.routeRemoteTargetWithContext(ctx, h, tr)
1432 }
1433 if h.ExactTarget != nil && h.ResumeToken == 0 {
1434 if err := d.validateExactSessionTarget(ctx, *h.ExactTarget); err != nil {
1435 return nil, nil, err
1436 }
1437 }
1438
1439 // A non-zero token is an authoritative resume credential. If it is unknown,
1440 // expired, or raced with lifecycle teardown, fail closed instead of routing
1441 // the Hello as an ordinary attach that could create or replace ownership.
1442 // The one legitimate pre-park race is a still-active same-client credential
1443 // for the requested session; hand that into the park/resume lifecycle.
1444 if h.ResumeToken != 0 {
1445 d.mu.Lock()
1446 parkedAtStart := d.parked[h.ResumeToken]
1447 d.mu.Unlock()
1448 if parkedAtStart == nil {
1449 if sess, ac, ok, err := d.resumeLiveAttachment(h, tr, sz); err != nil {
1450 if errors.Is(err, errResumeTokenLifecycleRace) {
1451 // Live recovery parked then lost a competing resumeParked
1452 // race; keep the fail-closed wire response instead of
1453 // leaking the internal sentinel to handleHello as ErrInternal.
1454 return nil, nil, &protoErr{ports.ErrNoSuchSession, "resume token is no longer valid"}
1455 }
1456 return nil, nil, err
1457 } else if ok {
1458 return sess, ac, nil
1459 }
1460 return nil, nil, &protoErr{ports.ErrNoSuchSession, "resume token is no longer valid"}
1461 }
1462 if sess, ac, ok, err := d.resumeParked(h, tr, sz); err == nil {
1463 if ok {
1464 return sess, ac, nil
1465 }
1466 return nil, nil, &protoErr{ports.ErrNoSuchSession, "resume token is no longer valid"}
1467 } else if errors.Is(err, errResumeTokenLifecycleRace) {
1468 // The parked entry was replaced while this handshake waited for
1469 // its send lock. Fail closed on the original credential rather
1470 // than falling through to ordinary attach/create routing.
1471 return nil, nil, &protoErr{ports.ErrNoSuchSession, "resume token is no longer valid"}
1472 } else {
1473 return nil, nil, err
1474 }
1475 }
1476 if h.Intent == ports.IntentResume || h.Intent == ports.IntentAttach {
1477 if err := d.waitForTargetRestore(ctx, h.Name); err != nil {
1478 return nil, nil, err
1479 }
1480 }
1481 if h.EnvironmentPolicy == ports.EnvironmentPolicyDaemonOwned && h.RemoteTarget == nil &&
1482 (h.Intent == ports.IntentNew || h.Intent == ports.IntentEphemeral) {
1483 return nil, nil, &protoErr{ports.ErrNoSuchTarget, "daemon-owned environment requires an exact remote target"}
1484 }
1485 d.mu.Lock()
1486 // Shutdown/create interlock: once shutdown has begun (last session removed,
1487 // or shutdownAll started) no new session may be created and no attach may
1488 // proceed — the teardown snapshot has already been (or is being) taken, so
1489 // anything inserted now would leak its PTY and hang Serve.
1490 if d.closing {
1491 d.mu.Unlock()
1492 return nil, nil, &protoErr{ports.ErrServerShutdown, "daemon is shutting down"}
1493 }
1494 // The first exact-target check happens before restore I/O. Recheck while
1495 // holding d.mu so a same-name lifecycle replacement cannot slip between
1496 // validation and attachment creation.
1497 if h.ExactTarget != nil && h.ResumeToken == 0 {
1498 if err := d.validateExactSessionTargetLocked(*h.ExactTarget); err != nil {
1499 d.mu.Unlock()
1500 return nil, nil, err
1501 }
1502 }
1503 switch h.Intent {
1504 case ports.IntentEphemeral:
1505 name := d.allocEphemeralNameLocked()
1506 sess, err := d.createSessionLockedWithMode(name, true, h.Cwd, sz, term, h.Env)
1507 if err != nil {
1508 d.mu.Unlock()
1509 return nil, nil, err
1510 }
1511 ac, err := d.finishRouteAttach(sess, tr, sz, term, h, true, true)
1512 return sess, ac, err
1513
1514 case ports.IntentNew:
1515 if h.Name == "" {
1516 d.mu.Unlock()
1517 return nil, nil, &protoErr{ports.ErrInvalidSessionName, "empty session name"}
1518 }
1519 if err := domain.ValidateSessionName(h.Name); err != nil {
1520 d.mu.Unlock()
1521 return nil, nil, &protoErr{ports.ErrInvalidSessionName, err.Error()}
1522 }
1523 if d.nameLiveOrStoppedLocked(h.Name) {
1524 d.mu.Unlock()
1525 return nil, nil, &protoErr{ports.ErrNameTaken, "session name already in use: " + h.Name}
1526 }
1527 sess, err := d.createSessionLockedWithMode(h.Name, false, h.Cwd, sz, term, h.Env)
1528 if err != nil {
1529 d.mu.Unlock()
1530 return nil, nil, err
1531 }
1532 ac, err := d.finishRouteAttach(sess, tr, sz, term, h, true, true)
1533 return sess, ac, err
1534
1535 case ports.IntentAttach:
1536 sess := d.findByNameLocked(h.Name)
1537 created := false
1538 if sess == nil {
1539 stopped, ok := d.stopped[h.Name]
1540 if !ok || stopped.purging {
1541 d.mu.Unlock()
1542 return nil, nil, &protoErr{ports.ErrNoSuchSession, "no such session: " + h.Name}
1543 }
1544 cwd := d.dirOrHome(stopped.cwd)
1545 env := h.Env
1546 if h.EnvironmentPolicy == ports.EnvironmentPolicyDaemonOwned {
1547 env = copyEnvironment(d.baseEnv)
1548 }
1549 var err error
1550 sess, err = d.createSessionLockedWithMode(h.Name, false, cwd, sz, term, env, stopped.tabNames)
1551 if err != nil {
1552 d.mu.Unlock()
1553 return nil, nil, err
1554 }
1555 created = true
1556 }
1557 ac, err := d.finishRouteAttach(sess, tr, sz, term, h, created, false)
1558 return sess, ac, err
1559
1560 default:
1561 d.mu.Unlock()
1562 return nil, nil, &protoErr{ports.ErrInternal, "unknown intent"}
1563 }
1564}