Merge pull request #127 from bnema/fix/final-snapshot-shutdown

fix(daemon): serialize final session snapshots

commit: 841e10ac87a666daae85b2c14782f2591cc51134
author: bnema · 3mo ago · 2026-07-24 17:02 UTC
committer: GitHub · 3mo ago · 2026-07-24 17:02 UTC
merge of: cd0e8bd + a8af417 · diff against the first parent
signature: GPG UnverifiedB5690EEEBB952194 The key is unknown

3 files · +252 -34

M internal/usecase/daemon/daemon.go +25 -2


@@ -518,8 +518,8 @@ func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error {
518	518		// worker. The same deadline bounds the worker's final drain, including a
519	519		// repository call that ignores cancellation.
520	520		d.stopSnapshotEncodeWorkerWithDeadline(snapshotDeadline)
521		-	d.shutdownAll(ports.ReasonServerShutdown)
522		-	d.sessWg.Wait()
	521	+	d.shutdownAllWithSnapshotDeadline(ports.ReasonServerShutdown, snapshotDeadline)
	522	+	d.waitSessionWorkersWithSnapshotDeadline(snapshotDeadline)
523	523		d.waitNotifies()
524	524		if err := d.persist.Close(); err != nil {
525	525		d.log.Warn("closing session persister failed", "err", err)

@@ -550,6 +550,9 @@ func (d Daemon) shutdownAllWithSnapshotDeadline(reason uint8, deadline snapsho
550	550		return false
551	551		}
552	552		for _, s := range snapshot {
	553	+	// Cancellation and PTY closure must not wait behind a teardown owner that
	554	+	// is blocked in snapshot publication or purge work.
	555	+	s.stopInMemoryLifecycle()
553	556		if err := d.killSessionWithSnapshotDeadline(s, reason, false, deadline); err != nil {
554	557		checkpointIncomplete = true
555	558		d.log.Error("closing session with unpersisted terminal state", "err", err)

@@ -558,6 +561,26 @@ func (d Daemon) shutdownAllWithSnapshotDeadline(reason uint8, deadline snapsho
558	561		return checkpointIncomplete
559	562		}
560	563
	564	+	// waitSessionWorkersWithSnapshotDeadline joins in-memory workers when possible,
	565	+	// but never extends Serve beyond the shared repository shutdown budget. By this
	566	+	// point connection handlers are joined and every session has been cancelled, so
	567	+	// no new session worker can be registered while the waiter is active.
	568	+	func (d Daemon) waitSessionWorkersWithSnapshotDeadline(deadline snapshotShutdownDeadline) {
	569	+	done := make(chan struct{})
	570	+	go func() {
	571	+	d.sessWg.Wait()
	572	+	close(done)
	573	+	}()
	574	+	if deadline == nil {
	575	+	<-done
	576	+	return
	577	+	}
	578	+	select {
	579	+	case <-done:
	580	+	case <-deadline.Done():
	581	+	}
	582	+	}
	583	+
561	584		// snapshotShutdownDeadline is a single-use shutdown budget shared by forced
562	585		// checkpoints and the final snapshot worker join. Its done channel is never
563	586		// closed by callers, so a detached worker can safely retain session state until


@@ -518,8 +518,8 @@ func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error {
518		// worker. The same deadline bounds the worker's final drain, including a	518		// worker. The same deadline bounds the worker's final drain, including a
519		// repository call that ignores cancellation.	519		// repository call that ignores cancellation.
520		d.stopSnapshotEncodeWorkerWithDeadline(snapshotDeadline)	520		d.stopSnapshotEncodeWorkerWithDeadline(snapshotDeadline)
521	-	d.shutdownAll(ports.ReasonServerShutdown)	521	+	d.shutdownAllWithSnapshotDeadline(ports.ReasonServerShutdown, snapshotDeadline)
522	-	d.sessWg.Wait()	522	+	d.waitSessionWorkersWithSnapshotDeadline(snapshotDeadline)
523		d.waitNotifies()	523		d.waitNotifies()
524		if err := d.persist.Close(); err != nil {	524		if err := d.persist.Close(); err != nil {
525		d.log.Warn("closing session persister failed", "err", err)	525		d.log.Warn("closing session persister failed", "err", err)

@@ -550,6 +550,9 @@ func (d Daemon) shutdownAllWithSnapshotDeadline(reason uint8, deadline snapsho
550		return false	550		return false
551		}	551		}
552		for _, s := range snapshot {	552		for _, s := range snapshot {
			553	+	// Cancellation and PTY closure must not wait behind a teardown owner that
			554	+	// is blocked in snapshot publication or purge work.
			555	+	s.stopInMemoryLifecycle()
553		if err := d.killSessionWithSnapshotDeadline(s, reason, false, deadline); err != nil {	556		if err := d.killSessionWithSnapshotDeadline(s, reason, false, deadline); err != nil {
554		checkpointIncomplete = true	557		checkpointIncomplete = true
555		d.log.Error("closing session with unpersisted terminal state", "err", err)	558		d.log.Error("closing session with unpersisted terminal state", "err", err)

@@ -558,6 +561,26 @@ func (d Daemon) shutdownAllWithSnapshotDeadline(reason uint8, deadline snapsho
558		return checkpointIncomplete	561		return checkpointIncomplete
559		}	562		}
560			563
			564	+	// waitSessionWorkersWithSnapshotDeadline joins in-memory workers when possible,
			565	+	// but never extends Serve beyond the shared repository shutdown budget. By this
			566	+	// point connection handlers are joined and every session has been cancelled, so
			567	+	// no new session worker can be registered while the waiter is active.
			568	+	func (d Daemon) waitSessionWorkersWithSnapshotDeadline(deadline snapshotShutdownDeadline) {
			569	+	done := make(chan struct{})
			570	+	go func() {
			571	+	d.sessWg.Wait()
			572	+	close(done)
			573	+	}()
			574	+	if deadline == nil {
			575	+	<-done
			576	+	return
			577	+	}
			578	+	select {
			579	+	case <-done:
			580	+	case <-deadline.Done():
			581	+	}
			582	+	}
			583	+
561		// snapshotShutdownDeadline is a single-use shutdown budget shared by forced	584		// snapshotShutdownDeadline is a single-use shutdown budget shared by forced
562		// checkpoints and the final snapshot worker join. Its done channel is never	585		// checkpoints and the final snapshot worker join. Its done channel is never
563		// closed by callers, so a detached worker can safely retain session state until	586		// closed by callers, so a detached worker can safely retain session state until

M internal/usecase/daemon/session.go +115 -12


@@ -41,10 +41,17 @@ type session struct {
41	41		// resolve focus against one state and mutate a later one. Lock order:
42	42		// dispatchMu strictly before d.mu, sess.mu, or any tab mu.
43	43		dispatchMu sync.Mutex
44		-	// teardownMu is the outermost per-session lifecycle lock. It serializes
45		-	// competing pane-exit, control, and daemon-shutdown teardown paths before
46		-	// they acquire daemon, session, snapshot, or pane locks.
	44	+	// teardownMu guards teardown ownership and is the outermost per-session
	45	+	// lifecycle lock: never acquire it while holding daemon, session, snapshot,
	46	+	// tab, or pane locks. The owner performs teardown without holding teardownMu;
	47	+	// duplicate daemon-shutdown callers can therefore bound their ownership wait
	48	+	// with Serve's shared snapshot deadline.
47	49		teardownMu sync.Mutex
	50	+	teardownActive bool
	51	+	teardownDone chan struct{}
	52	+	teardownWaiters uint
	53	+	teardownChanged chan struct{}
	54	+	lifecycleStopOnce sync.Once
48	55		mu sync.Mutex // guards tabs, active, client, clipFiles, and clipboard queue state
49	56		themeMu sync.Mutex
50	57		tabs []*tab

@@ -907,6 +914,101 @@ func (d Daemon) killSession(sess session, reason uint8, purge bool) error {
907	914		return d.killSessionWithSnapshotDeadline(sess, reason, purge, nil)
908	915		}
909	916
	917	+	// stopInMemoryLifecycle cancels every session producer and closes every PTY
	918	+	// independently of teardown ownership. Repository work may remain blocked, but
	919	+	// repeated shutdown passes cannot leave pane readers or launches running.
	920	+	func (s *session) stopInMemoryLifecycle() {
	921	+	if s == nil {
	922	+	return
	923	+	}
	924	+	s.lifecycleStopOnce.Do(func() {
	925	+	s.stopFloatingLaunches()
	926	+	s.cancel()
	927	+	s.mu.Lock()
	928	+	tabs := append([]*tab(nil), s.tabs...)
	929	+	s.mu.Unlock()
	930	+	for _, tb := range tabs {
	931	+	tb.closeAllPanes()
	932	+	}
	933	+	})
	934	+	}
	935	+
	936	+	func (s *session) teardownChangeLocked() chan struct{} {
	937	+	if s.teardownChanged == nil {
	938	+	s.teardownChanged = make(chan struct{})
	939	+	}
	940	+	return s.teardownChanged
	941	+	}
	942	+
	943	+	func (s *session) signalTeardownChangedLocked() {
	944	+	close(s.teardownChangeLocked())
	945	+	s.teardownChanged = make(chan struct{})
	946	+	}
	947	+
	948	+	// beginTeardown waits for ownership without holding teardownMu across teardown.
	949	+	// A false result means Serve's shared shutdown budget expired while another
	950	+	// lifecycle path owned the session. Ordinary callers wait and retry ownership,
	951	+	// preserving failed-teardown and stopped-purge retry semantics.
	952	+	func (s session) beginTeardown(deadline snapshotShutdownDeadline) bool {
	953	+	for {
	954	+	if deadline != nil {
	955	+	select {
	956	+	case <-deadline.Done():
	957	+	return false
	958	+	default:
	959	+	}
	960	+	}
	961	+
	962	+	s.teardownMu.Lock()
	963	+	if !s.teardownActive {
	964	+	if deadline != nil {
	965	+	select {
	966	+	case <-deadline.Done():
	967	+	s.teardownMu.Unlock()
	968	+	return false
	969	+	default:
	970	+	}
	971	+	}
	972	+	s.teardownActive = true
	973	+	s.teardownDone = make(chan struct{})
	974	+	s.signalTeardownChangedLocked()
	975	+	s.teardownMu.Unlock()
	976	+	return true
	977	+	}
	978	+	done := s.teardownDone
	979	+	s.teardownWaiters++
	980	+	s.signalTeardownChangedLocked()
	981	+	s.teardownMu.Unlock()
	982	+
	983	+	timedOut := false
	984	+	if deadline == nil {
	985	+	<-done
	986	+	} else {
	987	+	select {
	988	+	case <-done:
	989	+	case <-deadline.Done():
	990	+	timedOut = true
	991	+	}
	992	+	}
	993	+	s.teardownMu.Lock()
	994	+	s.teardownWaiters--
	995	+	s.signalTeardownChangedLocked()
	996	+	s.teardownMu.Unlock()
	997	+	if timedOut {
	998	+	return false
	999	+	}
	1000	+	}
	1001	+	}
	1002	+
	1003	+	func (s *session) finishTeardown() {
	1004	+	s.teardownMu.Lock()
	1005	+	s.teardownActive = false
	1006	+	close(s.teardownDone)
	1007	+	s.teardownDone = nil
	1008	+	s.signalTeardownChangedLocked()
	1009	+	s.teardownMu.Unlock()
	1010	+	}
	1011	+
910	1012		// killSessionWithSnapshotDeadline shares Serve's shutdown budget with its
911	1013		// terminal checkpoint and coordinator join. A timed-out repository call keeps
912	1014		// only immutable capture state; it never observes closed worker channels or

@@ -918,9 +1020,12 @@ func (d Daemon) killSessionWithSnapshotDeadline(sess session, reason uint8, pu
918	1020		// Pane EOF, explicit close, and daemon shutdown can all converge on the same
919	1021		// session. Only one owner may create the terminal mutation and checkpoint;
920	1022		// later callers observe the completed registry transition instead of
921		-	// manufacturing a newer generation that the owner will quarantine.
922		-	sess.teardownMu.Lock()
923		-	defer sess.teardownMu.Unlock()
	1023	+	// manufacturing a newer generation that the owner will quarantine. A
	1024	+	// shutdown duplicate abandons this wait when the shared budget expires.
	1025	+	if !sess.beginTeardown(deadline) {
	1026	+	return nil
	1027	+	}
	1028	+	defer sess.finishTeardown()
924	1029
925	1030		d.mu.Lock()
926	1031		current := d.sessions[sess.id]

@@ -1058,11 +1163,10 @@ func (d Daemon) killSessionWithSnapshotDeadline(sess session, reason uint8, pu
1058	1163		ac.clearCaptureFrames()
1059	1164		}
1060	1165
1061		-	// Prevent queued launches from entering Open, then cancel the parent
1062		-	// context. The launch worker owns any late PTY result and closes it rather
1063		-	// than publishing it into the torn-down tab.
1064		-	sess.stopFloatingLaunches()
1065		-	sess.cancel()
	1166	+	// Prevent queued launches from entering Open, cancel the parent context, and
	1167	+	// close PTYs. Serve may already have performed this phase while repository
	1168	+	// work owned by this teardown was blocked.
	1169	+	sess.stopInMemoryLifecycle()
1066	1170		sess.mu.Lock()
1067	1171		tabs := append([]*tab(nil), sess.tabs...)
1068	1172		clipFiles := sess.clipFiles

@@ -1071,7 +1175,6 @@ func (d Daemon) killSessionWithSnapshotDeadline(sess session, reason uint8, pu
1071	1175		for _, tb := range tabs {
1072	1176		d.clearDestroyedTabPreview(tb)
1073	1177		d.teardownFloating(tb, ac)
1074		-	tb.closeAllPanes()
1075	1178		}
1076	1179		for _, path := range clipFiles {
1077	1180		if err := os.Remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {


@@ -41,10 +41,17 @@ type session struct {
41		// resolve focus against one state and mutate a later one. Lock order:	41		// resolve focus against one state and mutate a later one. Lock order:
42		// dispatchMu strictly before d.mu, sess.mu, or any tab mu.	42		// dispatchMu strictly before d.mu, sess.mu, or any tab mu.
43		dispatchMu sync.Mutex	43		dispatchMu sync.Mutex
44	-	// teardownMu is the outermost per-session lifecycle lock. It serializes	44	+	// teardownMu guards teardown ownership and is the outermost per-session
45	-	// competing pane-exit, control, and daemon-shutdown teardown paths before	45	+	// lifecycle lock: never acquire it while holding daemon, session, snapshot,
46	-	// they acquire daemon, session, snapshot, or pane locks.	46	+	// tab, or pane locks. The owner performs teardown without holding teardownMu;
			47	+	// duplicate daemon-shutdown callers can therefore bound their ownership wait
			48	+	// with Serve's shared snapshot deadline.
47		teardownMu sync.Mutex	49		teardownMu sync.Mutex
			50	+	teardownActive bool
			51	+	teardownDone chan struct{}
			52	+	teardownWaiters uint
			53	+	teardownChanged chan struct{}
			54	+	lifecycleStopOnce sync.Once
48		mu sync.Mutex // guards tabs, active, client, clipFiles, and clipboard queue state	55		mu sync.Mutex // guards tabs, active, client, clipFiles, and clipboard queue state
49		themeMu sync.Mutex	56		themeMu sync.Mutex
50		tabs []*tab	57		tabs []*tab

@@ -907,6 +914,101 @@ func (d Daemon) killSession(sess session, reason uint8, purge bool) error {
907		return d.killSessionWithSnapshotDeadline(sess, reason, purge, nil)	914		return d.killSessionWithSnapshotDeadline(sess, reason, purge, nil)
908		}	915		}
909			916
			917	+	// stopInMemoryLifecycle cancels every session producer and closes every PTY
			918	+	// independently of teardown ownership. Repository work may remain blocked, but
			919	+	// repeated shutdown passes cannot leave pane readers or launches running.
			920	+	func (s *session) stopInMemoryLifecycle() {
			921	+	if s == nil {
			922	+	return
			923	+	}
			924	+	s.lifecycleStopOnce.Do(func() {
			925	+	s.stopFloatingLaunches()
			926	+	s.cancel()
			927	+	s.mu.Lock()
			928	+	tabs := append([]*tab(nil), s.tabs...)
			929	+	s.mu.Unlock()
			930	+	for _, tb := range tabs {
			931	+	tb.closeAllPanes()
			932	+	}
			933	+	})
			934	+	}
			935	+
			936	+	func (s *session) teardownChangeLocked() chan struct{} {
			937	+	if s.teardownChanged == nil {
			938	+	s.teardownChanged = make(chan struct{})
			939	+	}
			940	+	return s.teardownChanged
			941	+	}
			942	+
			943	+	func (s *session) signalTeardownChangedLocked() {
			944	+	close(s.teardownChangeLocked())
			945	+	s.teardownChanged = make(chan struct{})
			946	+	}
			947	+
			948	+	// beginTeardown waits for ownership without holding teardownMu across teardown.
			949	+	// A false result means Serve's shared shutdown budget expired while another
			950	+	// lifecycle path owned the session. Ordinary callers wait and retry ownership,
			951	+	// preserving failed-teardown and stopped-purge retry semantics.
			952	+	func (s session) beginTeardown(deadline snapshotShutdownDeadline) bool {
			953	+	for {
			954	+	if deadline != nil {
			955	+	select {
			956	+	case <-deadline.Done():
			957	+	return false
			958	+	default:
			959	+	}
			960	+	}
			961	+
			962	+	s.teardownMu.Lock()
			963	+	if !s.teardownActive {
			964	+	if deadline != nil {
			965	+	select {
			966	+	case <-deadline.Done():
			967	+	s.teardownMu.Unlock()
			968	+	return false
			969	+	default:
			970	+	}
			971	+	}
			972	+	s.teardownActive = true
			973	+	s.teardownDone = make(chan struct{})
			974	+	s.signalTeardownChangedLocked()
			975	+	s.teardownMu.Unlock()
			976	+	return true
			977	+	}
			978	+	done := s.teardownDone
			979	+	s.teardownWaiters++
			980	+	s.signalTeardownChangedLocked()
			981	+	s.teardownMu.Unlock()
			982	+
			983	+	timedOut := false
			984	+	if deadline == nil {
			985	+	<-done
			986	+	} else {
			987	+	select {
			988	+	case <-done:
			989	+	case <-deadline.Done():
			990	+	timedOut = true
			991	+	}
			992	+	}
			993	+	s.teardownMu.Lock()
			994	+	s.teardownWaiters--
			995	+	s.signalTeardownChangedLocked()
			996	+	s.teardownMu.Unlock()
			997	+	if timedOut {
			998	+	return false
			999	+	}
			1000	+	}
			1001	+	}
			1002	+
			1003	+	func (s *session) finishTeardown() {
			1004	+	s.teardownMu.Lock()
			1005	+	s.teardownActive = false
			1006	+	close(s.teardownDone)
			1007	+	s.teardownDone = nil
			1008	+	s.signalTeardownChangedLocked()
			1009	+	s.teardownMu.Unlock()
			1010	+	}
			1011	+
910		// killSessionWithSnapshotDeadline shares Serve's shutdown budget with its	1012		// killSessionWithSnapshotDeadline shares Serve's shutdown budget with its
911		// terminal checkpoint and coordinator join. A timed-out repository call keeps	1013		// terminal checkpoint and coordinator join. A timed-out repository call keeps
912		// only immutable capture state; it never observes closed worker channels or	1014		// only immutable capture state; it never observes closed worker channels or

@@ -918,9 +1020,12 @@ func (d Daemon) killSessionWithSnapshotDeadline(sess session, reason uint8, pu
918		// Pane EOF, explicit close, and daemon shutdown can all converge on the same	1020		// Pane EOF, explicit close, and daemon shutdown can all converge on the same
919		// session. Only one owner may create the terminal mutation and checkpoint;	1021		// session. Only one owner may create the terminal mutation and checkpoint;
920		// later callers observe the completed registry transition instead of	1022		// later callers observe the completed registry transition instead of
921	-	// manufacturing a newer generation that the owner will quarantine.	1023	+	// manufacturing a newer generation that the owner will quarantine. A
922	-	sess.teardownMu.Lock()	1024	+	// shutdown duplicate abandons this wait when the shared budget expires.
923	-	defer sess.teardownMu.Unlock()	1025	+	if !sess.beginTeardown(deadline) {
			1026	+	return nil
			1027	+	}
			1028	+	defer sess.finishTeardown()
924			1029
925		d.mu.Lock()	1030		d.mu.Lock()
926		current := d.sessions[sess.id]	1031		current := d.sessions[sess.id]

@@ -1058,11 +1163,10 @@ func (d Daemon) killSessionWithSnapshotDeadline(sess session, reason uint8, pu
1058		ac.clearCaptureFrames()	1163		ac.clearCaptureFrames()
1059		}	1164		}
1060			1165
1061	-	// Prevent queued launches from entering Open, then cancel the parent	1166	+	// Prevent queued launches from entering Open, cancel the parent context, and
1062	-	// context. The launch worker owns any late PTY result and closes it rather	1167	+	// close PTYs. Serve may already have performed this phase while repository
1063	-	// than publishing it into the torn-down tab.	1168	+	// work owned by this teardown was blocked.
1064	-	sess.stopFloatingLaunches()	1169	+	sess.stopInMemoryLifecycle()
1065	-	sess.cancel()
1066		sess.mu.Lock()	1170		sess.mu.Lock()
1067		tabs := append([]*tab(nil), sess.tabs...)	1171		tabs := append([]*tab(nil), sess.tabs...)
1068		clipFiles := sess.clipFiles	1172		clipFiles := sess.clipFiles

@@ -1071,7 +1175,6 @@ func (d Daemon) killSessionWithSnapshotDeadline(sess session, reason uint8, pu
1071		for _, tb := range tabs {	1175		for _, tb := range tabs {
1072		d.clearDestroyedTabPreview(tb)	1176		d.clearDestroyedTabPreview(tb)
1073		d.teardownFloating(tb, ac)	1177		d.teardownFloating(tb, ac)
1074	-	tb.closeAllPanes()
1075		}	1178		}
1076		for _, path := range clipFiles {	1179		for _, path := range clipFiles {
1077		if err := os.Remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {	1180		if err := os.Remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {

M internal/usecase/daemon/snapshot_scheduler_test.go +112 -20


@@ -4,7 +4,6 @@ import (
4	4		"context"
5	5		"errors"
6	6		"io"
7		-	"runtime"
8	7		"sync"
9	8		"sync/atomic"
10	9		"testing"

@@ -443,6 +442,95 @@ func TestForcedSnapshotShutdownTimeoutRetainsRetryableStateAndNotice(t *testing.
443	442		}
444	443		}
445	444
	445	+	func TestServeShutdownDeadlineBoundsPaneExitTeardownOwner(t *testing.T) {
	446	+	pty, releasePTY := newBlockingPTY(t)
	447	+	clock := newServeShutdownClock()
	448	+	d := newTestDaemon(t, newFactory(t, pty), clock)
	449	+	repository := portsmocks.NewMockSnapshotRepository(t)
	450	+	WithSnapshotRepository(repository, nil)(d)
	451	+	repository.EXPECT().List(mock.Anything).Return(nil, nil).Maybe()
	452	+
	453	+	publicationStarted := make(chan struct{})
	454	+	releasePublication := make(chan struct{})
	455	+	var releasePublicationOnce sync.Once
	456	+	releaseSnapshot := func() { releasePublicationOnce.Do(func() { close(releasePublication) }) }
	457	+	repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
	458	+	require.Equal(t, uint64(1), publication.Generation)
	459	+	close(publicationStarted)
	460	+	<-releasePublication
	461	+	return nil
	462	+	}).Once()
	463	+
	464	+	tr, sends, _ := newConn(t, mustHello(ports.IntentNew, "work", domain.Size{Cols: 80, Rows: 24}))
	465	+	listener := serveSnapshotListener(t, tr)
	466	+	ctx, cancel := context.WithCancel(context.Background())
	467	+	defer cancel()
	468	+	served := make(chan error, 1)
	469	+	go func() { served <- d.Serve(ctx, listener) }()
	470	+
	471	+	awaitFrame(t, sends, ports.MsgWelcome)
	472	+	sess := firstSession(d)
	473	+	require.NotNil(t, sess)
	474	+
	475	+	// Releasing the real pane reader drives EOF through ptyReader, reapPane, and
	476	+	// closePane. That ordinary lifecycle path must own teardown before Serve's
	477	+	// context cancellation enters its competing shutdownAll path.
	478	+	releasePTY()
	479	+	select {
	480	+	case <-publicationStarted:
	481	+	case <-time.After(testWaitTimeout):
	482	+	t.Fatal("pane EOF teardown did not publish its terminal snapshot")
	483	+	}
	484	+	// The pane-exit owner has an independent final-snapshot timer. Keep it
	485	+	// unfired until cleanup so Serve must use, and remain bounded by, its shared
	486	+	// deadline.
	487	+	ownerDeadlineTimer := clock.nextFinalTimer(t)
	488	+
	489	+	// Cleanup is the only place that releases the competing owner. It also fires
	490	+	// that owner's independent budget and joins the real pane reader, proving no
	491	+	// teardown goroutine is leaked without making Serve depend on repository I/O.
	492	+	t.Cleanup(func() {
	493	+	releaseSnapshot()
	494	+	ownerDeadlineTimer.fire()
	495	+	d.sessWg.Wait()
	496	+	})
	497	+
	498	+	cancel()
	499	+	shutdownDeadlineTimer := clock.nextFinalTimer(t)
	500	+	shutdownDeadlineTimer.fire()
	501	+
	502	+	select {
	503	+	case err := <-served:
	504	+	require.NoError(t, err)
	505	+	case <-time.After(testWaitTimeout):
	506	+	t.Fatal("Serve did not return after its shared deadline while the pane-exit owner remained blocked")
	507	+	}
	508	+
	509	+	sess.snapshotMu.Lock()
	510	+	generation := sess.snapshotGeneration
	511	+	sess.snapshotMu.Unlock()
	512	+	require.Equal(t, uint64(1), generation, "Serve shutdown must not create a competing terminal generation")
	513	+	require.Equal(t, int64(snapshotFinalFlushTimeout), clock.finalBudget.Load(), "Serve must spend only its shared snapshot deadline")
	514	+	}
	515	+
	516	+	func awaitTeardownWaiters(t testing.T, sess session, want uint) {
	517	+	t.Helper()
	518	+	for {
	519	+	sess.teardownMu.Lock()
	520	+	waiters := sess.teardownWaiters
	521	+	changed := sess.teardownChangeLocked()
	522	+	sess.teardownMu.Unlock()
	523	+	if waiters == want {
	524	+	return
	525	+	}
	526	+	select {
	527	+	case <-changed:
	528	+	case <-time.After(testWaitTimeout):
	529	+	t.Fatalf("teardown ownership waiters = %d, want %d", waiters, want)
	530	+	}
	531	+	}
	532	+	}
	533	+
446	534		func TestConcurrentSessionTeardownPublishesOneFinalSnapshot(t *testing.T) {
447	535		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
448	536		repository := portsmocks.NewMockSnapshotRepository(t)

@@ -466,34 +554,28 @@ func TestConcurrentSessionTeardownPublishesOneFinalSnapshot(t *testing.T) {
466	554		d.sessions[sess.id] = sess
467	555
468	556		first := make(chan error, 1)
469		-	go func() { first <- d.killSession(sess, ports.ReasonServerShutdown, false) }()
	557	+	go func() { first <- d.killSession(sess, ports.ReasonSessionKilled, false) }()
470	558		select {
471	559		case <-publicationStarted:
472	560		case <-time.After(testWaitTimeout):
473		-	t.Fatal("first terminal snapshot did not reach repository")
	561	+	t.Fatal("ordinary teardown did not publish its terminal snapshot")
474	562		}
475	563
	564	+	sess.teardownMu.Lock()
	565	+	require.True(t, sess.teardownActive)
	566	+	require.NotNil(t, sess.teardownDone)
	567	+	sess.teardownMu.Unlock()
	568	+
476	569		const duplicateCallers = 8
477		-	started := make(chan struct{}, duplicateCallers)
478	570		duplicates := make(chan error, duplicateCallers)
479	571		for range duplicateCallers {
480	572		go func() {
481		-	started <- struct{}{}
482		-	duplicates <- d.killSession(sess, ports.ReasonServerShutdown, false)
	573	+	duplicates <- d.killSession(sess, ports.ReasonSessionKilled, false)
483	574		}()
484	575		}
485		-	for range duplicateCallers {
486		-	<-started
487		-	}
488		-	for range 1000 {
489		-	runtime.Gosched()
490		-	sess.snapshotMu.Lock()
491		-	generation := sess.snapshotGeneration
492		-	sess.snapshotMu.Unlock()
493		-	if generation > 1 {
494		-	break
495		-	}
496		-	}
	576	+	// Synchronize on the ownership state itself: every duplicate is waiting
	577	+	// behind the distinct ordinary teardown before its publication is released.
	578	+	awaitTeardownWaiters(t, sess, duplicateCallers)
497	579
498	580		sess.snapshotMu.Lock()
499	581		generation := sess.snapshotGeneration

@@ -501,9 +583,19 @@ func TestConcurrentSessionTeardownPublishesOneFinalSnapshot(t *testing.T) {
501	583		require.Equal(t, uint64(1), generation, "duplicate teardown callers must not create synthetic terminal generations")
502	584
503	585		releaseOnce.Do(func() { close(releasePublication) })
504		-	require.NoError(t, <-first)
	586	+	select {
	587	+	case err := <-first:
	588	+	require.NoError(t, err)
	589	+	case <-time.After(testWaitTimeout):
	590	+	t.Fatal("ordinary teardown did not finish")
	591	+	}
505	592		for range duplicateCallers {
506		-	require.NoError(t, <-duplicates)
	593	+	select {
	594	+	case err := <-duplicates:
	595	+	require.NoError(t, err)
	596	+	case <-time.After(testWaitTimeout):
	597	+	t.Fatal("duplicate teardown did not observe the completed registry transition")
	598	+	}
507	599		}
508	600		sess.snapshotMu.Lock()
509	601		generation = sess.snapshotGeneration


@@ -4,7 +4,6 @@ import (
4		"context"	4		"context"
5		"errors"	5		"errors"
6		"io"	6		"io"
7	-	"runtime"
8		"sync"	7		"sync"
9		"sync/atomic"	8		"sync/atomic"
10		"testing"	9		"testing"

@@ -443,6 +442,95 @@ func TestForcedSnapshotShutdownTimeoutRetainsRetryableStateAndNotice(t *testing.
443		}	442		}
444		}	443		}
445			444
			445	+	func TestServeShutdownDeadlineBoundsPaneExitTeardownOwner(t *testing.T) {
			446	+	pty, releasePTY := newBlockingPTY(t)
			447	+	clock := newServeShutdownClock()
			448	+	d := newTestDaemon(t, newFactory(t, pty), clock)
			449	+	repository := portsmocks.NewMockSnapshotRepository(t)
			450	+	WithSnapshotRepository(repository, nil)(d)
			451	+	repository.EXPECT().List(mock.Anything).Return(nil, nil).Maybe()
			452	+
			453	+	publicationStarted := make(chan struct{})
			454	+	releasePublication := make(chan struct{})
			455	+	var releasePublicationOnce sync.Once
			456	+	releaseSnapshot := func() { releasePublicationOnce.Do(func() { close(releasePublication) }) }
			457	+	repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
			458	+	require.Equal(t, uint64(1), publication.Generation)
			459	+	close(publicationStarted)
			460	+	<-releasePublication
			461	+	return nil
			462	+	}).Once()
			463	+
			464	+	tr, sends, _ := newConn(t, mustHello(ports.IntentNew, "work", domain.Size{Cols: 80, Rows: 24}))
			465	+	listener := serveSnapshotListener(t, tr)
			466	+	ctx, cancel := context.WithCancel(context.Background())
			467	+	defer cancel()
			468	+	served := make(chan error, 1)
			469	+	go func() { served <- d.Serve(ctx, listener) }()
			470	+
			471	+	awaitFrame(t, sends, ports.MsgWelcome)
			472	+	sess := firstSession(d)
			473	+	require.NotNil(t, sess)
			474	+
			475	+	// Releasing the real pane reader drives EOF through ptyReader, reapPane, and
			476	+	// closePane. That ordinary lifecycle path must own teardown before Serve's
			477	+	// context cancellation enters its competing shutdownAll path.
			478	+	releasePTY()
			479	+	select {
			480	+	case <-publicationStarted:
			481	+	case <-time.After(testWaitTimeout):
			482	+	t.Fatal("pane EOF teardown did not publish its terminal snapshot")
			483	+	}
			484	+	// The pane-exit owner has an independent final-snapshot timer. Keep it
			485	+	// unfired until cleanup so Serve must use, and remain bounded by, its shared
			486	+	// deadline.
			487	+	ownerDeadlineTimer := clock.nextFinalTimer(t)
			488	+
			489	+	// Cleanup is the only place that releases the competing owner. It also fires
			490	+	// that owner's independent budget and joins the real pane reader, proving no
			491	+	// teardown goroutine is leaked without making Serve depend on repository I/O.
			492	+	t.Cleanup(func() {
			493	+	releaseSnapshot()
			494	+	ownerDeadlineTimer.fire()
			495	+	d.sessWg.Wait()
			496	+	})
			497	+
			498	+	cancel()
			499	+	shutdownDeadlineTimer := clock.nextFinalTimer(t)
			500	+	shutdownDeadlineTimer.fire()
			501	+
			502	+	select {
			503	+	case err := <-served:
			504	+	require.NoError(t, err)
			505	+	case <-time.After(testWaitTimeout):
			506	+	t.Fatal("Serve did not return after its shared deadline while the pane-exit owner remained blocked")
			507	+	}
			508	+
			509	+	sess.snapshotMu.Lock()
			510	+	generation := sess.snapshotGeneration
			511	+	sess.snapshotMu.Unlock()
			512	+	require.Equal(t, uint64(1), generation, "Serve shutdown must not create a competing terminal generation")
			513	+	require.Equal(t, int64(snapshotFinalFlushTimeout), clock.finalBudget.Load(), "Serve must spend only its shared snapshot deadline")
			514	+	}
			515	+
			516	+	func awaitTeardownWaiters(t testing.T, sess session, want uint) {
			517	+	t.Helper()
			518	+	for {
			519	+	sess.teardownMu.Lock()
			520	+	waiters := sess.teardownWaiters
			521	+	changed := sess.teardownChangeLocked()
			522	+	sess.teardownMu.Unlock()
			523	+	if waiters == want {
			524	+	return
			525	+	}
			526	+	select {
			527	+	case <-changed:
			528	+	case <-time.After(testWaitTimeout):
			529	+	t.Fatalf("teardown ownership waiters = %d, want %d", waiters, want)
			530	+	}
			531	+	}
			532	+	}
			533	+
446		func TestConcurrentSessionTeardownPublishesOneFinalSnapshot(t *testing.T) {	534		func TestConcurrentSessionTeardownPublishesOneFinalSnapshot(t *testing.T) {
447		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})	535		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
448		repository := portsmocks.NewMockSnapshotRepository(t)	536		repository := portsmocks.NewMockSnapshotRepository(t)

@@ -466,34 +554,28 @@ func TestConcurrentSessionTeardownPublishesOneFinalSnapshot(t *testing.T) {
466		d.sessions[sess.id] = sess	554		d.sessions[sess.id] = sess
467			555
468		first := make(chan error, 1)	556		first := make(chan error, 1)
469	-	go func() { first <- d.killSession(sess, ports.ReasonServerShutdown, false) }()	557	+	go func() { first <- d.killSession(sess, ports.ReasonSessionKilled, false) }()
470		select {	558		select {
471		case <-publicationStarted:	559		case <-publicationStarted:
472		case <-time.After(testWaitTimeout):	560		case <-time.After(testWaitTimeout):
473	-	t.Fatal("first terminal snapshot did not reach repository")	561	+	t.Fatal("ordinary teardown did not publish its terminal snapshot")
474		}	562		}
475			563
			564	+	sess.teardownMu.Lock()
			565	+	require.True(t, sess.teardownActive)
			566	+	require.NotNil(t, sess.teardownDone)
			567	+	sess.teardownMu.Unlock()
			568	+
476		const duplicateCallers = 8	569		const duplicateCallers = 8
477	-	started := make(chan struct{}, duplicateCallers)
478		duplicates := make(chan error, duplicateCallers)	570		duplicates := make(chan error, duplicateCallers)
479		for range duplicateCallers {	571		for range duplicateCallers {
480		go func() {	572		go func() {
481	-	started <- struct{}{}	573	+	duplicates <- d.killSession(sess, ports.ReasonSessionKilled, false)
482	-	duplicates <- d.killSession(sess, ports.ReasonServerShutdown, false)
483		}()	574		}()
484		}	575		}
485	-	for range duplicateCallers {	576	+	// Synchronize on the ownership state itself: every duplicate is waiting
486	-	<-started	577	+	// behind the distinct ordinary teardown before its publication is released.
487	-	}	578	+	awaitTeardownWaiters(t, sess, duplicateCallers)
488	-	for range 1000 {
489	-	runtime.Gosched()
490	-	sess.snapshotMu.Lock()
491	-	generation := sess.snapshotGeneration
492	-	sess.snapshotMu.Unlock()
493	-	if generation > 1 {
494	-	break
495	-	}
496	-	}
497			579
498		sess.snapshotMu.Lock()	580		sess.snapshotMu.Lock()
499		generation := sess.snapshotGeneration	581		generation := sess.snapshotGeneration

@@ -501,9 +583,19 @@ func TestConcurrentSessionTeardownPublishesOneFinalSnapshot(t *testing.T) {
501		require.Equal(t, uint64(1), generation, "duplicate teardown callers must not create synthetic terminal generations")	583		require.Equal(t, uint64(1), generation, "duplicate teardown callers must not create synthetic terminal generations")
502			584
503		releaseOnce.Do(func() { close(releasePublication) })	585		releaseOnce.Do(func() { close(releasePublication) })
504	-	require.NoError(t, <-first)	586	+	select {
			587	+	case err := <-first:
			588	+	require.NoError(t, err)
			589	+	case <-time.After(testWaitTimeout):
			590	+	t.Fatal("ordinary teardown did not finish")
			591	+	}
505		for range duplicateCallers {	592		for range duplicateCallers {
506	-	require.NoError(t, <-duplicates)	593	+	select {
			594	+	case err := <-duplicates:
			595	+	require.NoError(t, err)
			596	+	case <-time.After(testWaitTimeout):
			597	+	t.Fatal("duplicate teardown did not observe the completed registry transition")
			598	+	}
507		}	599		}
508		sess.snapshotMu.Lock()	600		sess.snapshotMu.Lock()
509		generation = sess.snapshotGeneration	601		generation = sess.snapshotGeneration