feat(daemon): session registry, PTY reader, debounced render scheduler, attach/resize

commit: 7e2f16e754dfd8e0a72322db77dd91de9db4daa6
author: brice · 3mo ago · 2026-07-02 11:59 UTC
parent: 844c846
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +1307 -18

M internal/usecase/daemon/daemon.go +753 -18

@@ -1,29 +1,764 @@
1		-	// Package daemon holds vev's server-side session multiplexer use case.
	1	+	// Package daemon holds vev's server-side session multiplexer use case: the
	2	+	// accept loop, the ephemeral/named session registry, the per-window PTY reader
	3	+	// and VT screen, and the per-client debounced render scheduler.
2	4		//
3		-	// The daemon core — accepting client connections, spawning PTY-backed
4		-	// sessions, and fanning I/O between them — is implemented in a later
5		-	// milestone (Task 11). This file provides only the Serve entry point that
6		-	// the app layer wires the hidden --daemon path to, so the client, CLI, and
7		-	// auto-spawn machinery can be built and tested against a stable seam now.
	5	+	// Concurrency model (one window per session for the MVP; multi-window is M3):
	6	+	//
	7	+	// - Serve runs the accept loop. Each accepted connection is handled by its
	8	+	// own goroutine (handleConn): it reads the first frame and routes it to a
	9	+	// session create/attach, a list, or a kill.
	10	+	// - Per session there are exactly two long-lived goroutines: the PTY reader
	11	+	// (drains child output into the VT screen and pokes a cap-1 dirty channel)
	12	+	// and the render scheduler (debounces dirties and paints the attached
	13	+	// client). Both are tied to the session context and unwind when the
	14	+	// session is killed (pty.Close unblocks the reader; ctx cancel stops the
	15	+	// scheduler).
	16	+	// - The daemon exits (Serve returns) when the last session is removed, or
	17	+	// when the parent context is cancelled (graceful shutdown notifies any
	18	+	// attached clients with ReasonServerShutdown).
	19	+	//
	20	+	// Locking: a session's screen and per-client renderer shadow are both guarded
	21	+	// by window.mu; the attached-client pointer by session.mu; the registry by
	22	+	// Daemon.mu. When more than one is held the order is always
	23	+	// Daemon.mu > session.mu, and (for the transport) attachedClient.sendMu >
	24	+	// window.mu — the PTY reader only ever takes window.mu, so it never blocks on
	25	+	// a slow client.
8	26		package daemon
9	27
10	28		import (
	29	+	"context"
11	30		"errors"
	31	+	"fmt"
	32	+	"log/slog"
	33	+	"os"
	34	+	"sort"
	35	+	"strconv"
	36	+	"strings"
	37	+	"sync"
	38	+	"time"
12	39
	40	+	"github.com/bnema/vev/internal/domain"
13	41		"github.com/bnema/vev/internal/ports"
	42	+	"github.com/bnema/vev/pkg/renderer"
	43	+	"github.com/bnema/vev/pkg/vt"
14	44		)
15	45
16		-	// ErrNotImplemented is returned by Serve until the daemon core lands in
17		-	// Task 11. It is deliberately a distinct sentinel so the app layer (and
18		-	// tests) can recognise the not-yet-implemented state.
19		-	var ErrNotImplemented = errors.New("vev: daemon core not implemented yet (Task 11)")
	46	+	// debounceInterval is the fixed window opened on the first dirty signal: all
	47	+	// further dirties arriving within it are absorbed and coalesced into a single
	48	+	// render. It bounds render frequency to ~100Hz per session.
	49	+	const debounceInterval = 10 * time.Millisecond
20	50
21		-	// Serve runs the daemon's accept loop over ln, owning ln for its lifetime.
22		-	//
23		-	// TODO(Task 11): implement the real accept/session/dispatch loop. For now
24		-	// it accepts the listener and immediately reports ErrNotImplemented without
25		-	// closing it (the caller owns ln and closes it).
26		-	func Serve(ln ports.Listener) error {
27		-	_ = ln
28		-	return ErrNotImplemented
	51	+	// ptyReadBufSize is the PTY reader's read buffer.
	52	+	const ptyReadBufSize = 32 * 1024
	53	+
	54	+	// defaultSize is used when a client's Hello carries no valid dimensions.
	55	+	var defaultSize = domain.Size{Cols: 80, Rows: 24}
	56	+
	57	+	// Daemon is vev's server-side multiplexer. Construct it with New and drive it
	58	+	// with Serve.
	59	+	type Daemon struct {
	60	+	mu sync.Mutex
	61	+	sessions map[domain.SessionID]*session
	62	+	nextID uint64
	63	+
	64	+	ptys ports.PTYFactory
	65	+	clock ports.Clock
	66	+	log *slog.Logger
	67	+	baseEnv []string
	68	+	shell string
	69	+	shellArgs []string
	70	+
	71	+	serveCtx context.Context
	72	+	serveCancel context.CancelFunc
	73	+
	74	+	// hardCtx force-closes connection transports on shutdown, but only after
	75	+	// shutdownAll has delivered graceful Detached notices — keeping it separate
	76	+	// from serveCtx so a parent-context cancel never races the notice.
	77	+	hardCtx context.Context
	78	+	hardCancel context.CancelFunc
	79	+
	80	+	done chan struct{}
	81	+	doneOnce sync.Once
	82	+
	83	+	sessWg sync.WaitGroup // PTY reader + scheduler goroutines
	84	+	connWg sync.WaitGroup // per-connection handler goroutines
	85	+	}
	86	+
	87	+	// session is a single multiplexed session. For the MVP it owns exactly one
	88	+	// window and at most one attached client.
	89	+	type session struct {
	90	+	id domain.SessionID
	91	+	name string
	92	+	ephemeral bool
	93	+	win *window
	94	+
	95	+	ctx context.Context
	96	+	cancel context.CancelFunc
	97	+
	98	+	mu sync.Mutex // guards client
	99	+	client *attachedClient
	100	+	}
	101	+
	102	+	// window is one PTY-backed screen. dirty is a cap-1 channel so bursty output
	103	+	// applies back-pressure by collapsing (a full buffer means "a render is
	104	+	// already pending"), never by blocking the reader.
	105	+	type window struct {
	106	+	pty ports.PTY
	107	+	mu sync.Mutex // guards screen and every attached renderer's shadow
	108	+	screen *vt.Screen
	109	+	dirty chan struct{}
	110	+	size domain.Size
	111	+	}
	112	+
	113	+	// attachedClient is a client currently attached to a session's window. rend is
	114	+	// its private renderer shadow (so each client gets output minimised against
	115	+	// what it has actually seen). sendMu serialises the two senders — the render
	116	+	// scheduler and the connection handler — so the transport's single-writer
	117	+	// contract holds and Draw→Send stays atomic (correct output ordering).
	118	+	type attachedClient struct {
	119	+	tr ports.Transport
	120	+	rend *renderer.Renderer
	121	+	size domain.Size
	122	+	sendMu sync.Mutex
	123	+	}
	124	+
	125	+	// send serialises a frame onto the client's transport.
	126	+	func (ac *attachedClient) send(f ports.Frame) error {
	127	+	ac.sendMu.Lock()
	128	+	defer ac.sendMu.Unlock()
	129	+	return ac.tr.Send(f)
	130	+	}
	131	+
	132	+	// Option customises a Daemon at construction.
	133	+	type Option func(*Daemon)
	134	+
	135	+	// WithShell overrides the command (and its args) each session spawns. The
	136	+	// default is $SHELL (or /bin/sh) with no arguments; tests use this to run a
	137	+	// deterministic program.
	138	+	func WithShell(cmd string, args []string) Option {
	139	+	return func(d *Daemon) {
	140	+	d.shell = cmd
	141	+	d.shellArgs = args
	142	+	}
	143	+	}
	144	+
	145	+	// New constructs a Daemon. ptys spawns PTY-backed children, clock drives the
	146	+	// render debounce, and log receives diagnostics (defaults to slog.Default).
	147	+	func New(ptys ports.PTYFactory, clock ports.Clock, log slog.Logger, opts ...Option) Daemon {
	148	+	if log == nil {
	149	+	log = slog.Default()
	150	+	}
	151	+	shell := os.Getenv("SHELL")
	152	+	if shell == "" {
	153	+	shell = "/bin/sh"
	154	+	}
	155	+	d := &Daemon{
	156	+	sessions: make(map[domain.SessionID]*session),
	157	+	ptys: ptys,
	158	+	clock: clock,
	159	+	log: log,
	160	+	baseEnv: os.Environ(),
	161	+	shell: shell,
	162	+	done: make(chan struct{}),
	163	+	}
	164	+	for _, o := range opts {
	165	+	o(d)
	166	+	}
	167	+	return d
	168	+	}
	169	+
	170	+	// Serve runs the accept loop over l, owning it for the loop's lifetime. It
	171	+	// returns when the last session is removed or ctx is cancelled; on the latter
	172	+	// path attached clients are detached with ReasonServerShutdown.
	173	+	func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error {
	174	+	d.serveCtx, d.serveCancel = context.WithCancel(ctx)
	175	+	defer d.serveCancel()
	176	+	d.hardCtx, d.hardCancel = context.WithCancel(context.Background())
	177	+	defer d.hardCancel()
	178	+
	179	+	// Break the accept loop when either the parent context is cancelled or the
	180	+	// registry drains to empty: both close the listener, which fails Accept.
	181	+	go func() {
	182	+	select {
	183	+	case <-d.serveCtx.Done():
	184	+	case <-d.done:
	185	+	}
	186	+	_ = l.Close()
	187	+	}()
	188	+
	189	+	for {
	190	+	tr, err := l.Accept()
	191	+	if err != nil {
	192	+	break
	193	+	}
	194	+	d.connWg.Add(1)
	195	+	go func() {
	196	+	defer d.connWg.Done()
	197	+	d.handleConn(tr)
	198	+	}()
	199	+	}
	200	+
	201	+	// Tear down: first notify any attached clients and close their PTYs and
	202	+	// transports (killSession), THEN force-close whatever connections are still
	203	+	// parked (control conns, in-flight handshakes), then cancel the serve
	204	+	// context and wait for every goroutine.
	205	+	d.shutdownAll(ports.ReasonServerShutdown)
	206	+	d.hardCancel()
	207	+	d.serveCancel()
	208	+	d.sessWg.Wait()
	209	+	d.connWg.Wait()
	210	+	return nil
	211	+	}
	212	+
	213	+	// shutdownAll kills every live session, snapshotting under the registry lock so
	214	+	// killSession (which relocks) never runs while the lock is held.
	215	+	func (d *Daemon) shutdownAll(reason uint8) {
	216	+	d.mu.Lock()
	217	+	snapshot := make([]*session, 0, len(d.sessions))
	218	+	for _, s := range d.sessions {
	219	+	snapshot = append(snapshot, s)
	220	+	}
	221	+	d.mu.Unlock()
	222	+	for _, s := range snapshot {
	223	+	d.killSession(s, reason)
	224	+	}
	225	+	}
	226	+
	227	+	// handleConn reads the first frame off a fresh connection and routes it. A
	228	+	// context watcher closes the transport on serve-context cancel so a handler
	229	+	// parked in Recv (mid-handshake or between input frames) unwinds on shutdown.
	230	+	func (d *Daemon) handleConn(tr ports.Transport) {
	231	+	stop := context.AfterFunc(d.hardCtx, func() { _ = tr.Close() })
	232	+	defer stop()
	233	+
	234	+	first, err := tr.Recv()
	235	+	if err != nil {
	236	+	_ = tr.Close()
	237	+	return
	238	+	}
	239	+	switch first.Type {
	240	+	case ports.MsgList:
	241	+	d.handleList(tr)
	242	+	case ports.MsgKill:
	243	+	d.handleKill(tr, first)
	244	+	case ports.MsgHello:
	245	+	d.handleHello(tr, first)
	246	+	default:
	247	+	_ = tr.Send(frameError(ports.ErrInternal, "expected hello"))
	248	+	_ = tr.Close()
	249	+	}
	250	+	}
	251	+
	252	+	// handleList replies with the current session listing and closes the (one-shot
	253	+	// control) connection.
	254	+	func (d *Daemon) handleList(tr ports.Transport) {
	255	+	defer tr.Close()
	256	+
	257	+	d.mu.Lock()
	258	+	infos := make([]ports.SessionInfo, 0, len(d.sessions))
	259	+	for _, s := range d.sessions {
	260	+	s.mu.Lock()
	261	+	attached := s.client != nil
	262	+	s.mu.Unlock()
	263	+	infos = append(infos, ports.SessionInfo{
	264	+	SessionID: string(s.id),
	265	+	Name: s.name,
	266	+	Ephemeral: s.ephemeral,
	267	+	Windows: 1,
	268	+	Attached: attached,
	269	+	})
	270	+	}
	271	+	d.mu.Unlock()
	272	+
	273	+	sort.Slice(infos, func(i, j int) bool { return infos[i].Name < infos[j].Name })
	274	+	_ = tr.Send(frameSessions(infos))
	275	+	}
	276	+
	277	+	// handleKill terminates the named session (if any) and closes the control
	278	+	// connection; the resulting EOF is the client's success signal.
	279	+	func (d *Daemon) handleKill(tr ports.Transport, f ports.Frame) {
	280	+	defer tr.Close()
	281	+
	282	+	k, err := ports.UnmarshalKill(f.Payload)
	283	+	if err != nil {
	284	+	_ = tr.Send(frameError(ports.ErrInternal, "malformed kill request"))
	285	+	return
	286	+	}
	287	+
	288	+	d.mu.Lock()
	289	+	target := d.findByNameLocked(k.Name)
	290	+	d.mu.Unlock()
	291	+
	292	+	if target == nil {
	293	+	_ = tr.Send(frameError(ports.ErrNoSuchSession, "no such session: "+k.Name))
	294	+	return
	295	+	}
	296	+	d.killSession(target, ports.ReasonSessionKilled)
	297	+	}
	298	+
	299	+	// handleHello runs the attach handshake: version check, intent routing,
	300	+	// Welcome, guaranteed first paint, then the per-connection input loop.
	301	+	func (d *Daemon) handleHello(tr ports.Transport, f ports.Frame) {
	302	+	h, err := ports.UnmarshalHello(f.Payload)
	303	+	if err != nil {
	304	+	_ = tr.Send(frameError(ports.ErrInternal, "malformed hello"))
	305	+	_ = tr.Close()
	306	+	return
	307	+	}
	308	+	if h.Version != ports.ProtocolVersion {
	309	+	_ = tr.Send(frameError(ports.ErrVersionMismatch, "protocol version mismatch"))
	310	+	_ = tr.Close()
	311	+	return
	312	+	}
	313	+
	314	+	sess, ac, rerr := d.route(h, tr)
	315	+	if rerr != nil {
	316	+	var pe *protoErr
	317	+	if errors.As(rerr, &pe) {
	318	+	_ = tr.Send(frameError(pe.code, pe.text))
	319	+	} else {
	320	+	_ = tr.Send(frameError(ports.ErrInternal, rerr.Error()))
	321	+	}
	322	+	_ = tr.Close()
	323	+	return
	324	+	}
	325	+
	326	+	if err := ac.send(frameWelcome(sess)); err != nil {
	327	+	d.clientGone(sess, ac, false)
	328	+	return
	329	+	}
	330	+	d.firstPaint(sess, ac, h.Size)
	331	+	d.runConnLoop(sess, ac)
	332	+	_ = tr.Close()
	333	+	}
	334	+
	335	+	// protoErr is a session-level rejection carrying a wire ErrorMsg code.
	336	+	type protoErr struct {
	337	+	code uint16
	338	+	text string
	339	+	}
	340	+
	341	+	func (e *protoErr) Error() string { return e.text }
	342	+
	343	+	// route resolves a Hello to a session and a freshly attached client, creating
	344	+	// the session for ephemeral/new intents.
	345	+	func (d Daemon) route(h ports.Hello, tr ports.Transport) (session, *attachedClient, error) {
	346	+	sz := h.Size
	347	+	if !sz.Valid() {
	348	+	sz = defaultSize
	349	+	}
	350	+
	351	+	d.mu.Lock()
	352	+	switch h.Intent {
	353	+	case ports.IntentEphemeral:
	354	+	name := d.allocEphemeralNameLocked()
	355	+	sess, err := d.createSessionLocked(name, true, sz)
	356	+	d.mu.Unlock()
	357	+	if err != nil {
	358	+	return nil, nil, err
	359	+	}
	360	+	return sess, d.attachClient(sess, tr, sz), nil
	361	+
	362	+	case ports.IntentNew:
	363	+	if h.Name == "" {
	364	+	d.mu.Unlock()
	365	+	return nil, nil, &protoErr{ports.ErrInternal, "empty session name"}
	366	+	}
	367	+	if d.nameTakenLocked(h.Name) {
	368	+	d.mu.Unlock()
	369	+	return nil, nil, &protoErr{ports.ErrNameTaken, "session name already in use: " + h.Name}
	370	+	}
	371	+	sess, err := d.createSessionLocked(h.Name, false, sz)
	372	+	d.mu.Unlock()
	373	+	if err != nil {
	374	+	return nil, nil, err
	375	+	}
	376	+	return sess, d.attachClient(sess, tr, sz), nil
	377	+
	378	+	case ports.IntentAttach:
	379	+	sess := d.findByNameLocked(h.Name)
	380	+	d.mu.Unlock()
	381	+	if sess == nil {
	382	+	return nil, nil, &protoErr{ports.ErrNoSuchSession, "no such session: " + h.Name}
	383	+	}
	384	+	return sess, d.attachClient(sess, tr, sz), nil
	385	+
	386	+	default:
	387	+	d.mu.Unlock()
	388	+	return nil, nil, &protoErr{ports.ErrInternal, "unknown intent"}
	389	+	}
	390	+	}
	391	+
	392	+	// createSessionLocked spawns a PTY-backed session and starts its reader and
	393	+	// scheduler goroutines. Caller holds d.mu.
	394	+	func (d Daemon) createSessionLocked(name string, ephemeral bool, sz domain.Size) (session, error) {
	395	+	pty, err := d.ptys.Open(d.shell, d.shellArgs, d.childEnv(name), sz)
	396	+	if err != nil {
	397	+	return nil, fmt.Errorf("daemon: spawning session %q: %w", name, err)
	398	+	}
	399	+
	400	+	id := domain.SessionID(fmt.Sprintf("sess-%d", d.nextID))
	401	+	d.nextID++
	402	+
	403	+	win := &window{
	404	+	pty: pty,
	405	+	screen: vt.NewScreen(sz.Cols, sz.Rows),
	406	+	dirty: make(chan struct{}, 1),
	407	+	size: sz,
	408	+	}
	409	+	sctx, cancel := context.WithCancel(d.serveCtx)
	410	+	sess := &session{
	411	+	id: id,
	412	+	name: name,
	413	+	ephemeral: ephemeral,
	414	+	win: win,
	415	+	ctx: sctx,
	416	+	cancel: cancel,
	417	+	}
	418	+	d.sessions[id] = sess
	419	+
	420	+	d.sessWg.Add(2)
	421	+	go d.ptyReader(sess)
	422	+	go d.scheduler(sess)
	423	+	return sess, nil
	424	+	}
	425	+
	426	+	// attachClient makes ac the session's current client, displacing any prior one
	427	+	// (which is notified with ReasonDetach and disconnected — its own conn handler
	428	+	// then unwinds without killing the session, since it is no longer current).
	429	+	func (d Daemon) attachClient(sess session, tr ports.Transport, sz domain.Size) *attachedClient {
	430	+	ac := &attachedClient{
	431	+	tr: tr,
	432	+	rend: renderer.New(renderer.Capabilities{}),
	433	+	size: sz,
	434	+	}
	435	+	sess.mu.Lock()
	436	+	old := sess.client
	437	+	sess.client = ac
	438	+	sess.mu.Unlock()
	439	+
	440	+	if old != nil {
	441	+	_ = old.send(frameDetached(ports.ReasonDetach))
	442	+	_ = old.tr.Close()
	443	+	}
	444	+	return ac
	445	+	}
	446	+
	447	+	// firstPaint guarantees the freshly attached client sees the full screen: if
	448	+	// the window size differs from the client's it resizes first (which paints a
	449	+	// full redraw), otherwise it forces a full paint against the fresh renderer.
	450	+	func (d Daemon) firstPaint(sess session, ac *attachedClient, clientSize domain.Size) {
	451	+	win := sess.win
	452	+	win.mu.Lock()
	453	+	wsz := win.size
	454	+	win.mu.Unlock()
	455	+
	456	+	if clientSize.Valid() && wsz != clientSize {
	457	+	d.resize(sess, ac, clientSize)
	458	+	return
	459	+	}
	460	+	d.paint(sess, ac, true)
	461	+	}
	462	+
	463	+	// runConnLoop is the per-connection input router: it pumps client messages
	464	+	// until detach, EOF, or a transport error.
	465	+	func (d Daemon) runConnLoop(sess session, ac *attachedClient) {
	466	+	win := sess.win
	467	+	for {
	468	+	f, err := ac.tr.Recv()
	469	+	if err != nil {
	470	+	d.clientGone(sess, ac, false)
	471	+	return
	472	+	}
	473	+	switch f.Type {
	474	+	case ports.MsgInput:
	475	+	if in, derr := ports.UnmarshalInput(f.Payload); derr == nil {
	476	+	_, _ = win.pty.Write(in.Data)
	477	+	}
	478	+	case ports.MsgResize:
	479	+	if rz, derr := ports.UnmarshalResize(f.Payload); derr == nil {
	480	+	d.resize(sess, ac, rz.Size)
	481	+	}
	482	+	case ports.MsgDetach:
	483	+	d.clientGone(sess, ac, true)
	484	+	return
	485	+	case ports.MsgPing:
	486	+	_ = ac.send(framePong())
	487	+	default:
	488	+	// Unknown/out-of-band client messages are ignored so a newer
	489	+	// client can add message types without breaking an older daemon.
	490	+	}
	491	+	}
	492	+	}
	493	+
	494	+	// clientGone detaches ac if it is still the session's current client. An
	495	+	// ephemeral session dies with its client; a named one survives headless. When
	496	+	// explicit (a MsgDetach), the client is acked with ReasonDetach.
	497	+	func (d Daemon) clientGone(sess session, ac *attachedClient, explicit bool) {
	498	+	if !sess.detachIfCurrent(ac) {
	499	+	return // already displaced by a newer client; nothing to do
	500	+	}
	501	+	if explicit {
	502	+	_ = ac.send(frameDetached(ports.ReasonDetach))
	503	+	}
	504	+	_ = ac.tr.Close()
	505	+	if sess.ephemeral {
	506	+	d.killSession(sess, ports.ReasonSessionKilled)
	507	+	}
	508	+	}
	509	+
	510	+	// detachIfCurrent clears the client iff ac is the current one, reporting
	511	+	// whether it acted.
	512	+	func (s session) detachIfCurrent(ac attachedClient) bool {
	513	+	s.mu.Lock()
	514	+	defer s.mu.Unlock()
	515	+	if s.client == ac {
	516	+	s.client = nil
	517	+	return true
	518	+	}
	519	+	return false
	520	+	}
	521	+
	522	+	// ptyReader drains child output into the VT screen and pokes the dirty channel
	523	+	// (non-blocking: a full channel already means a render is pending). On any read
	524	+	// error (EOF when the child exits) it kills the session.
	525	+	func (d Daemon) ptyReader(sess session) {
	526	+	defer d.sessWg.Done()
	527	+	win := sess.win
	528	+	buf := make([]byte, ptyReadBufSize)
	529	+	for {
	530	+	n, err := win.pty.Read(buf)
	531	+	if n > 0 {
	532	+	win.mu.Lock()
	533	+	win.screen.Write(buf[:n])
	534	+	win.mu.Unlock()
	535	+	select {
	536	+	case win.dirty <- struct{}{}:
	537	+	default:
	538	+	}
	539	+	}
	540	+	if err != nil {
	541	+	d.killSession(sess, ports.ReasonSessionKilled)
	542	+	return
	543	+	}
	544	+	}
	545	+	}
	546	+
	547	+	// scheduler debounces dirty signals: the first opens a fixed window during
	548	+	// which further dirties are absorbed, then it renders exactly once.
	549	+	func (d Daemon) scheduler(sess session) {
	550	+	defer d.sessWg.Done()
	551	+	win := sess.win
	552	+	for {
	553	+	select {
	554	+	case <-sess.ctx.Done():
	555	+	return
	556	+	case <-win.dirty:
	557	+	}
	558	+
	559	+	timer := d.clock.NewTimer(debounceInterval)
	560	+	absorb:
	561	+	for {
	562	+	select {
	563	+	case <-sess.ctx.Done():
	564	+	timer.Stop()
	565	+	return
	566	+	case <-win.dirty:
	567	+	// Coalesced into the pending render.
	568	+	case <-timer.C():
	569	+	break absorb
	570	+	}
	571	+	}
	572	+	d.render(sess)
	573	+	}
	574	+	}
	575	+
	576	+	// render paints the current client, or (when detached) just clears accumulated
	577	+	// damage so it never grows unbounded while headless.
	578	+	func (d Daemon) render(sess session) {
	579	+	win := sess.win
	580	+	sess.mu.Lock()
	581	+	ac := sess.client
	582	+	sess.mu.Unlock()
	583	+
	584	+	if ac == nil {
	585	+	win.mu.Lock()
	586	+	win.screen.ClearDamage()
	587	+	win.mu.Unlock()
	588	+	return
	589	+	}
	590	+	d.paint(sess, ac, false)
	591	+	}
	592	+
	593	+	// paint draws the screen for ac and sends the resulting bytes. The renderer
	594	+	// shadow is read/written under win.mu; sendMu makes Draw→Send atomic per client
	595	+	// so the scheduler and a concurrent resize never interleave their output.
	596	+	func (d Daemon) paint(sess session, ac *attachedClient, reset bool) {
	597	+	win := sess.win
	598	+
	599	+	ac.sendMu.Lock()
	600	+	win.mu.Lock()
	601	+	if reset {
	602	+	ac.rend.Reset()
	603	+	}
	604	+	data, err := ac.rend.Draw(win.screen.Frame, win.screen.Damage())
	605	+	win.screen.ClearDamage()
	606	+	win.mu.Unlock()
	607	+
	608	+	var serr error
	609	+	if err == nil && len(data) > 0 {
	610	+	serr = ac.tr.Send(frameOutput(data))
	611	+	}
	612	+	ac.sendMu.Unlock()
	613	+
	614	+	if err != nil {
	615	+	d.log.Error("render draw failed", "err", err, "session", sess.name)
	616	+	return
	617	+	}
	618	+	if serr != nil {
	619	+	d.detachOnSendError(sess, ac)
	620	+	}
	621	+	}
	622	+
	623	+	// resize applies a client size change in the plan's strict order: pty.Resize,
	624	+	// then screen.Resize under the window lock, then a renderer reset + full
	625	+	// redraw, then send.
	626	+	func (d Daemon) resize(sess session, ac *attachedClient, sz domain.Size) {
	627	+	if !sz.Valid() {
	628	+	return
	629	+	}
	630	+	win := sess.win
	631	+
	632	+	if err := win.pty.Resize(sz); err != nil {
	633	+	d.log.Warn("pty resize failed", "err", err, "session", sess.name)
	634	+	}
	635	+
	636	+	ac.sendMu.Lock()
	637	+	win.mu.Lock()
	638	+	win.screen.Resize(sz.Cols, sz.Rows)
	639	+	win.size = sz
	640	+	ac.rend.Reset()
	641	+	data, err := ac.rend.Draw(win.screen.Frame, win.screen.Damage())
	642	+	win.screen.ClearDamage()
	643	+	win.mu.Unlock()
	644	+
	645	+	var serr error
	646	+	if err == nil && len(data) > 0 {
	647	+	serr = ac.tr.Send(frameOutput(data))
	648	+	}
	649	+	ac.sendMu.Unlock()
	650	+
	651	+	if err != nil {
	652	+	d.log.Error("resize draw failed", "err", err, "session", sess.name)
	653	+	return
	654	+	}
	655	+	if serr != nil {
	656	+	d.detachOnSendError(sess, ac)
	657	+	}
	658	+	}
	659	+
	660	+	// detachOnSendError drops a client whose transport failed but keeps the session
	661	+	// running headless (per plan: output errors never kill the session).
	662	+	func (d Daemon) detachOnSendError(sess session, ac *attachedClient) {
	663	+	if sess.detachIfCurrent(ac) {
	664	+	_ = ac.tr.Close()
	665	+	d.log.Warn("detached client after send error", "session", sess.name)
	666	+	}
	667	+	}
	668	+
	669	+	// killSession removes a session and tears down its resources. It is
	670	+	// idempotent: only the caller that wins the registry delete acts. When the
	671	+	// registry empties it signals daemon shutdown.
	672	+	func (d Daemon) killSession(sess session, reason uint8) {
	673	+	d.mu.Lock()
	674	+	if _, ok := d.sessions[sess.id]; !ok {
	675	+	d.mu.Unlock()
	676	+	return
	677	+	}
	678	+	delete(d.sessions, sess.id)
	679	+	empty := len(d.sessions) == 0
	680	+	d.mu.Unlock()
	681	+
	682	+	sess.mu.Lock()
	683	+	ac := sess.client
	684	+	sess.client = nil
	685	+	sess.mu.Unlock()
	686	+	if ac != nil {
	687	+	_ = ac.send(frameDetached(reason))
	688	+	_ = ac.tr.Close()
	689	+	}
	690	+
	691	+	sess.cancel()
	692	+	_ = sess.win.pty.Close()
	693	+
	694	+	if empty {
	695	+	d.doneOnce.Do(func() { close(d.done) })
	696	+	}
	697	+	}
	698	+
	699	+	// allocEphemeralNameLocked returns the lowest free decimal name. Caller holds
	700	+	// d.mu.
	701	+	func (d *Daemon) allocEphemeralNameLocked() string {
	702	+	used := make(map[string]struct{}, len(d.sessions))
	703	+	for _, s := range d.sessions {
	704	+	used[s.name] = struct{}{}
	705	+	}
	706	+	for i := 0; ; i++ {
	707	+	n := strconv.Itoa(i)
	708	+	if _, taken := used[n]; !taken {
	709	+	return n
	710	+	}
	711	+	}
	712	+	}
	713	+
	714	+	func (d Daemon) findByNameLocked(name string) session {
	715	+	for _, s := range d.sessions {
	716	+	if s.name == name {
	717	+	return s
	718	+	}
	719	+	}
	720	+	return nil
	721	+	}
	722	+
	723	+	func (d *Daemon) nameTakenLocked(name string) bool { return d.findByNameLocked(name) != nil }
	724	+
	725	+	// childEnv builds the session child's environment: the daemon's own, with TERM
	726	+	// and VEV forced to well-known values.
	727	+	func (d *Daemon) childEnv(name string) []string {
	728	+	out := make([]string, 0, len(d.baseEnv)+2)
	729	+	for _, e := range d.baseEnv {
	730	+	if strings.HasPrefix(e, "TERM=") \|\| strings.HasPrefix(e, "VEV=") {
	731	+	continue
	732	+	}
	733	+	out = append(out, e)
	734	+	}
	735	+	return append(out, "TERM=xterm-256color", "VEV="+name)
	736	+	}
	737	+
	738	+	func frameWelcome(s *session) ports.Frame {
	739	+	return ports.Frame{Type: ports.MsgWelcome, Payload: ports.MarshalWelcome(ports.Welcome{
	740	+	SessionID: string(s.id),
	741	+	SessionName: s.name,
	742	+	Ephemeral: s.ephemeral,
	743	+	})}
	744	+	}
	745	+
	746	+	func frameError(code uint16, text string) ports.Frame {
	747	+	return ports.Frame{Type: ports.MsgError, Payload: ports.MarshalErrorMsg(ports.ErrorMsg{Code: code, Text: text})}
	748	+	}
	749	+
	750	+	func frameOutput(b []byte) ports.Frame {
	751	+	return ports.Frame{Type: ports.MsgOutput, Payload: ports.MarshalOutput(ports.Output{Data: b})}
	752	+	}
	753	+
	754	+	func frameDetached(reason uint8) ports.Frame {
	755	+	return ports.Frame{Type: ports.MsgDetached, Payload: ports.MarshalDetached(ports.Detached{Reason: reason})}
	756	+	}
	757	+
	758	+	func framePong() ports.Frame {
	759	+	return ports.Frame{Type: ports.MsgPong, Payload: ports.MarshalPong(ports.Pong{})}
	760	+	}
	761	+
	762	+	func frameSessions(infos []ports.SessionInfo) ports.Frame {
	763	+	return ports.Frame{Type: ports.MsgSessions, Payload: ports.MarshalSessions(ports.Sessions{Sessions: infos})}
29	764		}

@@ -1,29 +1,764 @@
1	-	// Package daemon holds vev's server-side session multiplexer use case.	1	+	// Package daemon holds vev's server-side session multiplexer use case: the
			2	+	// accept loop, the ephemeral/named session registry, the per-window PTY reader
			3	+	// and VT screen, and the per-client debounced render scheduler.
2		//	4		//
3	-	// The daemon core — accepting client connections, spawning PTY-backed	5	+	// Concurrency model (one window per session for the MVP; multi-window is M3):
4	-	// sessions, and fanning I/O between them — is implemented in a later	6	+	//
5	-	// milestone (Task 11). This file provides only the Serve entry point that	7	+	// - Serve runs the accept loop. Each accepted connection is handled by its
6	-	// the app layer wires the hidden --daemon path to, so the client, CLI, and	8	+	// own goroutine (handleConn): it reads the first frame and routes it to a
7	-	// auto-spawn machinery can be built and tested against a stable seam now.	9	+	// session create/attach, a list, or a kill.
			10	+	// - Per session there are exactly two long-lived goroutines: the PTY reader
			11	+	// (drains child output into the VT screen and pokes a cap-1 dirty channel)
			12	+	// and the render scheduler (debounces dirties and paints the attached
			13	+	// client). Both are tied to the session context and unwind when the
			14	+	// session is killed (pty.Close unblocks the reader; ctx cancel stops the
			15	+	// scheduler).
			16	+	// - The daemon exits (Serve returns) when the last session is removed, or
			17	+	// when the parent context is cancelled (graceful shutdown notifies any
			18	+	// attached clients with ReasonServerShutdown).
			19	+	//
			20	+	// Locking: a session's screen and per-client renderer shadow are both guarded
			21	+	// by window.mu; the attached-client pointer by session.mu; the registry by
			22	+	// Daemon.mu. When more than one is held the order is always
			23	+	// Daemon.mu > session.mu, and (for the transport) attachedClient.sendMu >
			24	+	// window.mu — the PTY reader only ever takes window.mu, so it never blocks on
			25	+	// a slow client.
8		package daemon	26		package daemon
9			27
10		import (	28		import (
			29	+	"context"
11		"errors"	30		"errors"
			31	+	"fmt"
			32	+	"log/slog"
			33	+	"os"
			34	+	"sort"
			35	+	"strconv"
			36	+	"strings"
			37	+	"sync"
			38	+	"time"
12			39
			40	+	"github.com/bnema/vev/internal/domain"
13		"github.com/bnema/vev/internal/ports"	41		"github.com/bnema/vev/internal/ports"
			42	+	"github.com/bnema/vev/pkg/renderer"
			43	+	"github.com/bnema/vev/pkg/vt"
14		)	44		)
15			45
16	-	// ErrNotImplemented is returned by Serve until the daemon core lands in	46	+	// debounceInterval is the fixed window opened on the first dirty signal: all
17	-	// Task 11. It is deliberately a distinct sentinel so the app layer (and	47	+	// further dirties arriving within it are absorbed and coalesced into a single
18	-	// tests) can recognise the not-yet-implemented state.	48	+	// render. It bounds render frequency to ~100Hz per session.
19	-	var ErrNotImplemented = errors.New("vev: daemon core not implemented yet (Task 11)")	49	+	const debounceInterval = 10 * time.Millisecond
20			50
21	-	// Serve runs the daemon's accept loop over ln, owning ln for its lifetime.	51	+	// ptyReadBufSize is the PTY reader's read buffer.
22	-	//	52	+	const ptyReadBufSize = 32 * 1024
23	-	// TODO(Task 11): implement the real accept/session/dispatch loop. For now	53	+
24	-	// it accepts the listener and immediately reports ErrNotImplemented without	54	+	// defaultSize is used when a client's Hello carries no valid dimensions.
25	-	// closing it (the caller owns ln and closes it).	55	+	var defaultSize = domain.Size{Cols: 80, Rows: 24}
26	-	func Serve(ln ports.Listener) error {	56	+
27	-	_ = ln	57	+	// Daemon is vev's server-side multiplexer. Construct it with New and drive it
28	-	return ErrNotImplemented	58	+	// with Serve.
			59	+	type Daemon struct {
			60	+	mu sync.Mutex
			61	+	sessions map[domain.SessionID]*session
			62	+	nextID uint64
			63	+
			64	+	ptys ports.PTYFactory
			65	+	clock ports.Clock
			66	+	log *slog.Logger
			67	+	baseEnv []string
			68	+	shell string
			69	+	shellArgs []string
			70	+
			71	+	serveCtx context.Context
			72	+	serveCancel context.CancelFunc
			73	+
			74	+	// hardCtx force-closes connection transports on shutdown, but only after
			75	+	// shutdownAll has delivered graceful Detached notices — keeping it separate
			76	+	// from serveCtx so a parent-context cancel never races the notice.
			77	+	hardCtx context.Context
			78	+	hardCancel context.CancelFunc
			79	+
			80	+	done chan struct{}
			81	+	doneOnce sync.Once
			82	+
			83	+	sessWg sync.WaitGroup // PTY reader + scheduler goroutines
			84	+	connWg sync.WaitGroup // per-connection handler goroutines
			85	+	}
			86	+
			87	+	// session is a single multiplexed session. For the MVP it owns exactly one
			88	+	// window and at most one attached client.
			89	+	type session struct {
			90	+	id domain.SessionID
			91	+	name string
			92	+	ephemeral bool
			93	+	win *window
			94	+
			95	+	ctx context.Context
			96	+	cancel context.CancelFunc
			97	+
			98	+	mu sync.Mutex // guards client
			99	+	client *attachedClient
			100	+	}
			101	+
			102	+	// window is one PTY-backed screen. dirty is a cap-1 channel so bursty output
			103	+	// applies back-pressure by collapsing (a full buffer means "a render is
			104	+	// already pending"), never by blocking the reader.
			105	+	type window struct {
			106	+	pty ports.PTY
			107	+	mu sync.Mutex // guards screen and every attached renderer's shadow
			108	+	screen *vt.Screen
			109	+	dirty chan struct{}
			110	+	size domain.Size
			111	+	}
			112	+
			113	+	// attachedClient is a client currently attached to a session's window. rend is
			114	+	// its private renderer shadow (so each client gets output minimised against
			115	+	// what it has actually seen). sendMu serialises the two senders — the render
			116	+	// scheduler and the connection handler — so the transport's single-writer
			117	+	// contract holds and Draw→Send stays atomic (correct output ordering).
			118	+	type attachedClient struct {
			119	+	tr ports.Transport
			120	+	rend *renderer.Renderer
			121	+	size domain.Size
			122	+	sendMu sync.Mutex
			123	+	}
			124	+
			125	+	// send serialises a frame onto the client's transport.
			126	+	func (ac *attachedClient) send(f ports.Frame) error {
			127	+	ac.sendMu.Lock()
			128	+	defer ac.sendMu.Unlock()
			129	+	return ac.tr.Send(f)
			130	+	}
			131	+
			132	+	// Option customises a Daemon at construction.
			133	+	type Option func(*Daemon)
			134	+
			135	+	// WithShell overrides the command (and its args) each session spawns. The
			136	+	// default is $SHELL (or /bin/sh) with no arguments; tests use this to run a
			137	+	// deterministic program.
			138	+	func WithShell(cmd string, args []string) Option {
			139	+	return func(d *Daemon) {
			140	+	d.shell = cmd
			141	+	d.shellArgs = args
			142	+	}
			143	+	}
			144	+
			145	+	// New constructs a Daemon. ptys spawns PTY-backed children, clock drives the
			146	+	// render debounce, and log receives diagnostics (defaults to slog.Default).
			147	+	func New(ptys ports.PTYFactory, clock ports.Clock, log slog.Logger, opts ...Option) Daemon {
			148	+	if log == nil {
			149	+	log = slog.Default()
			150	+	}
			151	+	shell := os.Getenv("SHELL")
			152	+	if shell == "" {
			153	+	shell = "/bin/sh"
			154	+	}
			155	+	d := &Daemon{
			156	+	sessions: make(map[domain.SessionID]*session),
			157	+	ptys: ptys,
			158	+	clock: clock,
			159	+	log: log,
			160	+	baseEnv: os.Environ(),
			161	+	shell: shell,
			162	+	done: make(chan struct{}),
			163	+	}
			164	+	for _, o := range opts {
			165	+	o(d)
			166	+	}
			167	+	return d
			168	+	}
			169	+
			170	+	// Serve runs the accept loop over l, owning it for the loop's lifetime. It
			171	+	// returns when the last session is removed or ctx is cancelled; on the latter
			172	+	// path attached clients are detached with ReasonServerShutdown.
			173	+	func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error {
			174	+	d.serveCtx, d.serveCancel = context.WithCancel(ctx)
			175	+	defer d.serveCancel()
			176	+	d.hardCtx, d.hardCancel = context.WithCancel(context.Background())
			177	+	defer d.hardCancel()
			178	+
			179	+	// Break the accept loop when either the parent context is cancelled or the
			180	+	// registry drains to empty: both close the listener, which fails Accept.
			181	+	go func() {
			182	+	select {
			183	+	case <-d.serveCtx.Done():
			184	+	case <-d.done:
			185	+	}
			186	+	_ = l.Close()
			187	+	}()
			188	+
			189	+	for {
			190	+	tr, err := l.Accept()
			191	+	if err != nil {
			192	+	break
			193	+	}
			194	+	d.connWg.Add(1)
			195	+	go func() {
			196	+	defer d.connWg.Done()
			197	+	d.handleConn(tr)
			198	+	}()
			199	+	}
			200	+
			201	+	// Tear down: first notify any attached clients and close their PTYs and
			202	+	// transports (killSession), THEN force-close whatever connections are still
			203	+	// parked (control conns, in-flight handshakes), then cancel the serve
			204	+	// context and wait for every goroutine.
			205	+	d.shutdownAll(ports.ReasonServerShutdown)
			206	+	d.hardCancel()
			207	+	d.serveCancel()
			208	+	d.sessWg.Wait()
			209	+	d.connWg.Wait()
			210	+	return nil
			211	+	}
			212	+
			213	+	// shutdownAll kills every live session, snapshotting under the registry lock so
			214	+	// killSession (which relocks) never runs while the lock is held.
			215	+	func (d *Daemon) shutdownAll(reason uint8) {
			216	+	d.mu.Lock()
			217	+	snapshot := make([]*session, 0, len(d.sessions))
			218	+	for _, s := range d.sessions {
			219	+	snapshot = append(snapshot, s)
			220	+	}
			221	+	d.mu.Unlock()
			222	+	for _, s := range snapshot {
			223	+	d.killSession(s, reason)
			224	+	}
			225	+	}
			226	+
			227	+	// handleConn reads the first frame off a fresh connection and routes it. A
			228	+	// context watcher closes the transport on serve-context cancel so a handler
			229	+	// parked in Recv (mid-handshake or between input frames) unwinds on shutdown.
			230	+	func (d *Daemon) handleConn(tr ports.Transport) {
			231	+	stop := context.AfterFunc(d.hardCtx, func() { _ = tr.Close() })
			232	+	defer stop()
			233	+
			234	+	first, err := tr.Recv()
			235	+	if err != nil {
			236	+	_ = tr.Close()
			237	+	return
			238	+	}
			239	+	switch first.Type {
			240	+	case ports.MsgList:
			241	+	d.handleList(tr)
			242	+	case ports.MsgKill:
			243	+	d.handleKill(tr, first)
			244	+	case ports.MsgHello:
			245	+	d.handleHello(tr, first)
			246	+	default:
			247	+	_ = tr.Send(frameError(ports.ErrInternal, "expected hello"))
			248	+	_ = tr.Close()
			249	+	}
			250	+	}
			251	+
			252	+	// handleList replies with the current session listing and closes the (one-shot
			253	+	// control) connection.
			254	+	func (d *Daemon) handleList(tr ports.Transport) {
			255	+	defer tr.Close()
			256	+
			257	+	d.mu.Lock()
			258	+	infos := make([]ports.SessionInfo, 0, len(d.sessions))
			259	+	for _, s := range d.sessions {
			260	+	s.mu.Lock()
			261	+	attached := s.client != nil
			262	+	s.mu.Unlock()
			263	+	infos = append(infos, ports.SessionInfo{
			264	+	SessionID: string(s.id),
			265	+	Name: s.name,
			266	+	Ephemeral: s.ephemeral,
			267	+	Windows: 1,
			268	+	Attached: attached,
			269	+	})
			270	+	}
			271	+	d.mu.Unlock()
			272	+
			273	+	sort.Slice(infos, func(i, j int) bool { return infos[i].Name < infos[j].Name })
			274	+	_ = tr.Send(frameSessions(infos))
			275	+	}
			276	+
			277	+	// handleKill terminates the named session (if any) and closes the control
			278	+	// connection; the resulting EOF is the client's success signal.
			279	+	func (d *Daemon) handleKill(tr ports.Transport, f ports.Frame) {
			280	+	defer tr.Close()
			281	+
			282	+	k, err := ports.UnmarshalKill(f.Payload)
			283	+	if err != nil {
			284	+	_ = tr.Send(frameError(ports.ErrInternal, "malformed kill request"))
			285	+	return
			286	+	}
			287	+
			288	+	d.mu.Lock()
			289	+	target := d.findByNameLocked(k.Name)
			290	+	d.mu.Unlock()
			291	+
			292	+	if target == nil {
			293	+	_ = tr.Send(frameError(ports.ErrNoSuchSession, "no such session: "+k.Name))
			294	+	return
			295	+	}
			296	+	d.killSession(target, ports.ReasonSessionKilled)
			297	+	}
			298	+
			299	+	// handleHello runs the attach handshake: version check, intent routing,
			300	+	// Welcome, guaranteed first paint, then the per-connection input loop.
			301	+	func (d *Daemon) handleHello(tr ports.Transport, f ports.Frame) {
			302	+	h, err := ports.UnmarshalHello(f.Payload)
			303	+	if err != nil {
			304	+	_ = tr.Send(frameError(ports.ErrInternal, "malformed hello"))
			305	+	_ = tr.Close()
			306	+	return
			307	+	}
			308	+	if h.Version != ports.ProtocolVersion {
			309	+	_ = tr.Send(frameError(ports.ErrVersionMismatch, "protocol version mismatch"))
			310	+	_ = tr.Close()
			311	+	return
			312	+	}
			313	+
			314	+	sess, ac, rerr := d.route(h, tr)
			315	+	if rerr != nil {
			316	+	var pe *protoErr
			317	+	if errors.As(rerr, &pe) {
			318	+	_ = tr.Send(frameError(pe.code, pe.text))
			319	+	} else {
			320	+	_ = tr.Send(frameError(ports.ErrInternal, rerr.Error()))
			321	+	}
			322	+	_ = tr.Close()
			323	+	return
			324	+	}
			325	+
			326	+	if err := ac.send(frameWelcome(sess)); err != nil {
			327	+	d.clientGone(sess, ac, false)
			328	+	return
			329	+	}
			330	+	d.firstPaint(sess, ac, h.Size)
			331	+	d.runConnLoop(sess, ac)
			332	+	_ = tr.Close()
			333	+	}
			334	+
			335	+	// protoErr is a session-level rejection carrying a wire ErrorMsg code.
			336	+	type protoErr struct {
			337	+	code uint16
			338	+	text string
			339	+	}
			340	+
			341	+	func (e *protoErr) Error() string { return e.text }
			342	+
			343	+	// route resolves a Hello to a session and a freshly attached client, creating
			344	+	// the session for ephemeral/new intents.
			345	+	func (d Daemon) route(h ports.Hello, tr ports.Transport) (session, *attachedClient, error) {
			346	+	sz := h.Size
			347	+	if !sz.Valid() {
			348	+	sz = defaultSize
			349	+	}
			350	+
			351	+	d.mu.Lock()
			352	+	switch h.Intent {
			353	+	case ports.IntentEphemeral:
			354	+	name := d.allocEphemeralNameLocked()
			355	+	sess, err := d.createSessionLocked(name, true, sz)
			356	+	d.mu.Unlock()
			357	+	if err != nil {
			358	+	return nil, nil, err
			359	+	}
			360	+	return sess, d.attachClient(sess, tr, sz), nil
			361	+
			362	+	case ports.IntentNew:
			363	+	if h.Name == "" {
			364	+	d.mu.Unlock()
			365	+	return nil, nil, &protoErr{ports.ErrInternal, "empty session name"}
			366	+	}
			367	+	if d.nameTakenLocked(h.Name) {
			368	+	d.mu.Unlock()
			369	+	return nil, nil, &protoErr{ports.ErrNameTaken, "session name already in use: " + h.Name}
			370	+	}
			371	+	sess, err := d.createSessionLocked(h.Name, false, sz)
			372	+	d.mu.Unlock()
			373	+	if err != nil {
			374	+	return nil, nil, err
			375	+	}
			376	+	return sess, d.attachClient(sess, tr, sz), nil
			377	+
			378	+	case ports.IntentAttach:
			379	+	sess := d.findByNameLocked(h.Name)
			380	+	d.mu.Unlock()
			381	+	if sess == nil {
			382	+	return nil, nil, &protoErr{ports.ErrNoSuchSession, "no such session: " + h.Name}
			383	+	}
			384	+	return sess, d.attachClient(sess, tr, sz), nil
			385	+
			386	+	default:
			387	+	d.mu.Unlock()
			388	+	return nil, nil, &protoErr{ports.ErrInternal, "unknown intent"}
			389	+	}
			390	+	}
			391	+
			392	+	// createSessionLocked spawns a PTY-backed session and starts its reader and
			393	+	// scheduler goroutines. Caller holds d.mu.
			394	+	func (d Daemon) createSessionLocked(name string, ephemeral bool, sz domain.Size) (session, error) {
			395	+	pty, err := d.ptys.Open(d.shell, d.shellArgs, d.childEnv(name), sz)
			396	+	if err != nil {
			397	+	return nil, fmt.Errorf("daemon: spawning session %q: %w", name, err)
			398	+	}
			399	+
			400	+	id := domain.SessionID(fmt.Sprintf("sess-%d", d.nextID))
			401	+	d.nextID++
			402	+
			403	+	win := &window{
			404	+	pty: pty,
			405	+	screen: vt.NewScreen(sz.Cols, sz.Rows),
			406	+	dirty: make(chan struct{}, 1),
			407	+	size: sz,
			408	+	}
			409	+	sctx, cancel := context.WithCancel(d.serveCtx)
			410	+	sess := &session{
			411	+	id: id,
			412	+	name: name,
			413	+	ephemeral: ephemeral,
			414	+	win: win,
			415	+	ctx: sctx,
			416	+	cancel: cancel,
			417	+	}
			418	+	d.sessions[id] = sess
			419	+
			420	+	d.sessWg.Add(2)
			421	+	go d.ptyReader(sess)
			422	+	go d.scheduler(sess)
			423	+	return sess, nil
			424	+	}
			425	+
			426	+	// attachClient makes ac the session's current client, displacing any prior one
			427	+	// (which is notified with ReasonDetach and disconnected — its own conn handler
			428	+	// then unwinds without killing the session, since it is no longer current).
			429	+	func (d Daemon) attachClient(sess session, tr ports.Transport, sz domain.Size) *attachedClient {
			430	+	ac := &attachedClient{
			431	+	tr: tr,
			432	+	rend: renderer.New(renderer.Capabilities{}),
			433	+	size: sz,
			434	+	}
			435	+	sess.mu.Lock()
			436	+	old := sess.client
			437	+	sess.client = ac
			438	+	sess.mu.Unlock()
			439	+
			440	+	if old != nil {
			441	+	_ = old.send(frameDetached(ports.ReasonDetach))
			442	+	_ = old.tr.Close()
			443	+	}
			444	+	return ac
			445	+	}
			446	+
			447	+	// firstPaint guarantees the freshly attached client sees the full screen: if
			448	+	// the window size differs from the client's it resizes first (which paints a
			449	+	// full redraw), otherwise it forces a full paint against the fresh renderer.
			450	+	func (d Daemon) firstPaint(sess session, ac *attachedClient, clientSize domain.Size) {
			451	+	win := sess.win
			452	+	win.mu.Lock()
			453	+	wsz := win.size
			454	+	win.mu.Unlock()
			455	+
			456	+	if clientSize.Valid() && wsz != clientSize {
			457	+	d.resize(sess, ac, clientSize)
			458	+	return
			459	+	}
			460	+	d.paint(sess, ac, true)
			461	+	}
			462	+
			463	+	// runConnLoop is the per-connection input router: it pumps client messages
			464	+	// until detach, EOF, or a transport error.
			465	+	func (d Daemon) runConnLoop(sess session, ac *attachedClient) {
			466	+	win := sess.win
			467	+	for {
			468	+	f, err := ac.tr.Recv()
			469	+	if err != nil {
			470	+	d.clientGone(sess, ac, false)
			471	+	return
			472	+	}
			473	+	switch f.Type {
			474	+	case ports.MsgInput:
			475	+	if in, derr := ports.UnmarshalInput(f.Payload); derr == nil {
			476	+	_, _ = win.pty.Write(in.Data)
			477	+	}
			478	+	case ports.MsgResize:
			479	+	if rz, derr := ports.UnmarshalResize(f.Payload); derr == nil {
			480	+	d.resize(sess, ac, rz.Size)
			481	+	}
			482	+	case ports.MsgDetach:
			483	+	d.clientGone(sess, ac, true)
			484	+	return
			485	+	case ports.MsgPing:
			486	+	_ = ac.send(framePong())
			487	+	default:
			488	+	// Unknown/out-of-band client messages are ignored so a newer
			489	+	// client can add message types without breaking an older daemon.
			490	+	}
			491	+	}
			492	+	}
			493	+
			494	+	// clientGone detaches ac if it is still the session's current client. An
			495	+	// ephemeral session dies with its client; a named one survives headless. When
			496	+	// explicit (a MsgDetach), the client is acked with ReasonDetach.
			497	+	func (d Daemon) clientGone(sess session, ac *attachedClient, explicit bool) {
			498	+	if !sess.detachIfCurrent(ac) {
			499	+	return // already displaced by a newer client; nothing to do
			500	+	}
			501	+	if explicit {
			502	+	_ = ac.send(frameDetached(ports.ReasonDetach))
			503	+	}
			504	+	_ = ac.tr.Close()
			505	+	if sess.ephemeral {
			506	+	d.killSession(sess, ports.ReasonSessionKilled)
			507	+	}
			508	+	}
			509	+
			510	+	// detachIfCurrent clears the client iff ac is the current one, reporting
			511	+	// whether it acted.
			512	+	func (s session) detachIfCurrent(ac attachedClient) bool {
			513	+	s.mu.Lock()
			514	+	defer s.mu.Unlock()
			515	+	if s.client == ac {
			516	+	s.client = nil
			517	+	return true
			518	+	}
			519	+	return false
			520	+	}
			521	+
			522	+	// ptyReader drains child output into the VT screen and pokes the dirty channel
			523	+	// (non-blocking: a full channel already means a render is pending). On any read
			524	+	// error (EOF when the child exits) it kills the session.
			525	+	func (d Daemon) ptyReader(sess session) {
			526	+	defer d.sessWg.Done()
			527	+	win := sess.win
			528	+	buf := make([]byte, ptyReadBufSize)
			529	+	for {
			530	+	n, err := win.pty.Read(buf)
			531	+	if n > 0 {
			532	+	win.mu.Lock()
			533	+	win.screen.Write(buf[:n])
			534	+	win.mu.Unlock()
			535	+	select {
			536	+	case win.dirty <- struct{}{}:
			537	+	default:
			538	+	}
			539	+	}
			540	+	if err != nil {
			541	+	d.killSession(sess, ports.ReasonSessionKilled)
			542	+	return
			543	+	}
			544	+	}
			545	+	}
			546	+
			547	+	// scheduler debounces dirty signals: the first opens a fixed window during
			548	+	// which further dirties are absorbed, then it renders exactly once.
			549	+	func (d Daemon) scheduler(sess session) {
			550	+	defer d.sessWg.Done()
			551	+	win := sess.win
			552	+	for {
			553	+	select {
			554	+	case <-sess.ctx.Done():
			555	+	return
			556	+	case <-win.dirty:
			557	+	}
			558	+
			559	+	timer := d.clock.NewTimer(debounceInterval)
			560	+	absorb:
			561	+	for {
			562	+	select {
			563	+	case <-sess.ctx.Done():
			564	+	timer.Stop()
			565	+	return
			566	+	case <-win.dirty:
			567	+	// Coalesced into the pending render.
			568	+	case <-timer.C():
			569	+	break absorb
			570	+	}
			571	+	}
			572	+	d.render(sess)
			573	+	}
			574	+	}
			575	+
			576	+	// render paints the current client, or (when detached) just clears accumulated
			577	+	// damage so it never grows unbounded while headless.
			578	+	func (d Daemon) render(sess session) {
			579	+	win := sess.win
			580	+	sess.mu.Lock()
			581	+	ac := sess.client
			582	+	sess.mu.Unlock()
			583	+
			584	+	if ac == nil {
			585	+	win.mu.Lock()
			586	+	win.screen.ClearDamage()
			587	+	win.mu.Unlock()
			588	+	return
			589	+	}
			590	+	d.paint(sess, ac, false)
			591	+	}
			592	+
			593	+	// paint draws the screen for ac and sends the resulting bytes. The renderer
			594	+	// shadow is read/written under win.mu; sendMu makes Draw→Send atomic per client
			595	+	// so the scheduler and a concurrent resize never interleave their output.
			596	+	func (d Daemon) paint(sess session, ac *attachedClient, reset bool) {
			597	+	win := sess.win
			598	+
			599	+	ac.sendMu.Lock()
			600	+	win.mu.Lock()
			601	+	if reset {
			602	+	ac.rend.Reset()
			603	+	}
			604	+	data, err := ac.rend.Draw(win.screen.Frame, win.screen.Damage())
			605	+	win.screen.ClearDamage()
			606	+	win.mu.Unlock()
			607	+
			608	+	var serr error
			609	+	if err == nil && len(data) > 0 {
			610	+	serr = ac.tr.Send(frameOutput(data))
			611	+	}
			612	+	ac.sendMu.Unlock()
			613	+
			614	+	if err != nil {
			615	+	d.log.Error("render draw failed", "err", err, "session", sess.name)
			616	+	return
			617	+	}
			618	+	if serr != nil {
			619	+	d.detachOnSendError(sess, ac)
			620	+	}
			621	+	}
			622	+
			623	+	// resize applies a client size change in the plan's strict order: pty.Resize,
			624	+	// then screen.Resize under the window lock, then a renderer reset + full
			625	+	// redraw, then send.
			626	+	func (d Daemon) resize(sess session, ac *attachedClient, sz domain.Size) {
			627	+	if !sz.Valid() {
			628	+	return
			629	+	}
			630	+	win := sess.win
			631	+
			632	+	if err := win.pty.Resize(sz); err != nil {
			633	+	d.log.Warn("pty resize failed", "err", err, "session", sess.name)
			634	+	}
			635	+
			636	+	ac.sendMu.Lock()
			637	+	win.mu.Lock()
			638	+	win.screen.Resize(sz.Cols, sz.Rows)
			639	+	win.size = sz
			640	+	ac.rend.Reset()
			641	+	data, err := ac.rend.Draw(win.screen.Frame, win.screen.Damage())
			642	+	win.screen.ClearDamage()
			643	+	win.mu.Unlock()
			644	+
			645	+	var serr error
			646	+	if err == nil && len(data) > 0 {
			647	+	serr = ac.tr.Send(frameOutput(data))
			648	+	}
			649	+	ac.sendMu.Unlock()
			650	+
			651	+	if err != nil {
			652	+	d.log.Error("resize draw failed", "err", err, "session", sess.name)
			653	+	return
			654	+	}
			655	+	if serr != nil {
			656	+	d.detachOnSendError(sess, ac)
			657	+	}
			658	+	}
			659	+
			660	+	// detachOnSendError drops a client whose transport failed but keeps the session
			661	+	// running headless (per plan: output errors never kill the session).
			662	+	func (d Daemon) detachOnSendError(sess session, ac *attachedClient) {
			663	+	if sess.detachIfCurrent(ac) {
			664	+	_ = ac.tr.Close()
			665	+	d.log.Warn("detached client after send error", "session", sess.name)
			666	+	}
			667	+	}
			668	+
			669	+	// killSession removes a session and tears down its resources. It is
			670	+	// idempotent: only the caller that wins the registry delete acts. When the
			671	+	// registry empties it signals daemon shutdown.
			672	+	func (d Daemon) killSession(sess session, reason uint8) {
			673	+	d.mu.Lock()
			674	+	if _, ok := d.sessions[sess.id]; !ok {
			675	+	d.mu.Unlock()
			676	+	return
			677	+	}
			678	+	delete(d.sessions, sess.id)
			679	+	empty := len(d.sessions) == 0
			680	+	d.mu.Unlock()
			681	+
			682	+	sess.mu.Lock()
			683	+	ac := sess.client
			684	+	sess.client = nil
			685	+	sess.mu.Unlock()
			686	+	if ac != nil {
			687	+	_ = ac.send(frameDetached(reason))
			688	+	_ = ac.tr.Close()
			689	+	}
			690	+
			691	+	sess.cancel()
			692	+	_ = sess.win.pty.Close()
			693	+
			694	+	if empty {
			695	+	d.doneOnce.Do(func() { close(d.done) })
			696	+	}
			697	+	}
			698	+
			699	+	// allocEphemeralNameLocked returns the lowest free decimal name. Caller holds
			700	+	// d.mu.
			701	+	func (d *Daemon) allocEphemeralNameLocked() string {
			702	+	used := make(map[string]struct{}, len(d.sessions))
			703	+	for _, s := range d.sessions {
			704	+	used[s.name] = struct{}{}
			705	+	}
			706	+	for i := 0; ; i++ {
			707	+	n := strconv.Itoa(i)
			708	+	if _, taken := used[n]; !taken {
			709	+	return n
			710	+	}
			711	+	}
			712	+	}
			713	+
			714	+	func (d Daemon) findByNameLocked(name string) session {
			715	+	for _, s := range d.sessions {
			716	+	if s.name == name {
			717	+	return s
			718	+	}
			719	+	}
			720	+	return nil
			721	+	}
			722	+
			723	+	func (d *Daemon) nameTakenLocked(name string) bool { return d.findByNameLocked(name) != nil }
			724	+
			725	+	// childEnv builds the session child's environment: the daemon's own, with TERM
			726	+	// and VEV forced to well-known values.
			727	+	func (d *Daemon) childEnv(name string) []string {
			728	+	out := make([]string, 0, len(d.baseEnv)+2)
			729	+	for _, e := range d.baseEnv {
			730	+	if strings.HasPrefix(e, "TERM=") \|\| strings.HasPrefix(e, "VEV=") {
			731	+	continue
			732	+	}
			733	+	out = append(out, e)
			734	+	}
			735	+	return append(out, "TERM=xterm-256color", "VEV="+name)
			736	+	}
			737	+
			738	+	func frameWelcome(s *session) ports.Frame {
			739	+	return ports.Frame{Type: ports.MsgWelcome, Payload: ports.MarshalWelcome(ports.Welcome{
			740	+	SessionID: string(s.id),
			741	+	SessionName: s.name,
			742	+	Ephemeral: s.ephemeral,
			743	+	})}
			744	+	}
			745	+
			746	+	func frameError(code uint16, text string) ports.Frame {
			747	+	return ports.Frame{Type: ports.MsgError, Payload: ports.MarshalErrorMsg(ports.ErrorMsg{Code: code, Text: text})}
			748	+	}
			749	+
			750	+	func frameOutput(b []byte) ports.Frame {
			751	+	return ports.Frame{Type: ports.MsgOutput, Payload: ports.MarshalOutput(ports.Output{Data: b})}
			752	+	}
			753	+
			754	+	func frameDetached(reason uint8) ports.Frame {
			755	+	return ports.Frame{Type: ports.MsgDetached, Payload: ports.MarshalDetached(ports.Detached{Reason: reason})}
			756	+	}
			757	+
			758	+	func framePong() ports.Frame {
			759	+	return ports.Frame{Type: ports.MsgPong, Payload: ports.MarshalPong(ports.Pong{})}
			760	+	}
			761	+
			762	+	func frameSessions(infos []ports.SessionInfo) ports.Frame {
			763	+	return ports.Frame{Type: ports.MsgSessions, Payload: ports.MarshalSessions(ports.Sessions{Sessions: infos})}
29		}	764		}

A internal/usecase/daemon/daemon_test.go +554 -0

@@ -0,0 +1,554 @@
	1	+	package daemon
	2	+
	3	+	import (
	4	+	"context"
	5	+	"io"
	6	+	"log/slog"
	7	+	"sync"
	8	+	"sync/atomic"
	9	+	"testing"
	10	+	"time"
	11	+
	12	+	"github.com/stretchr/testify/mock"
	13	+	"github.com/stretchr/testify/require"
	14	+
	15	+	"github.com/bnema/vev/internal/domain"
	16	+	"github.com/bnema/vev/internal/ports"
	17	+	portsmocks "github.com/bnema/vev/internal/ports/mocks"
	18	+	"github.com/bnema/vev/pkg/renderer"
	19	+	"github.com/bnema/vev/pkg/vt"
	20	+	)
	21	+
	22	+	// --- test doubles -----------------------------------------------------------
	23	+
	24	+	// stubClock returns timers whose channel never fires, so a scheduler under it
	25	+	// blocks in its debounce loop until the session context is cancelled. Used by
	26	+	// every test that is not specifically exercising the debounce.
	27	+	type stubClock struct{}
	28	+
	29	+	func (stubClock) Now() time.Time { return time.Time{} }
	30	+	func (stubClock) NewTimer(time.Duration) ports.Timer { return stubTimer{} }
	31	+
	32	+	type stubTimer struct{}
	33	+
	34	+	func (stubTimer) C() <-chan time.Time { return nil }
	35	+	func (stubTimer) Reset(time.Duration) bool { return false }
	36	+	func (stubTimer) Stop() bool { return true }
	37	+
	38	+	func newTestDaemon(t testing.T, ptys ports.PTYFactory, clk ports.Clock) Daemon {
	39	+	t.Helper()
	40	+	d := New(ptys, clk, slog.New(slog.NewTextHandler(io.Discard, nil)))
	41	+	d.serveCtx, d.serveCancel = context.WithCancel(context.Background())
	42	+	d.hardCtx, d.hardCancel = context.WithCancel(context.Background())
	43	+	t.Cleanup(d.serveCancel)
	44	+	t.Cleanup(d.hardCancel)
	45	+	return d
	46	+	}
	47	+
	48	+	// newBlockingPTY returns a MockPTY whose Read blocks until release is called,
	49	+	// then reports io.EOF (the "child exited" signal).
	50	+	func newBlockingPTY(t testing.T) (portsmocks.MockPTY, func()) {
	51	+	t.Helper()
	52	+	p := portsmocks.NewMockPTY(t)
	53	+	release := make(chan struct{})
	54	+	var once sync.Once
	55	+	unblock := func() { once.Do(func() { close(release) }) }
	56	+	p.EXPECT().Read(mock.Anything).RunAndReturn(func(b []byte) (int, error) {
	57	+	<-release
	58	+	return 0, io.EOF
	59	+	}).Maybe()
	60	+	p.EXPECT().Write(mock.Anything).RunAndReturn(func(b []byte) (int, error) { return len(b), nil }).Maybe()
	61	+	p.EXPECT().Resize(mock.Anything).Return(nil).Maybe()
	62	+	// Close unblocks a parked Read, exactly as the real PTY does (its Close
	63	+	// closes the master fd, failing the in-flight read).
	64	+	p.EXPECT().Close().RunAndReturn(func() error { unblock(); return nil }).Maybe()
	65	+	p.EXPECT().Pid().Return(4242).Maybe()
	66	+	return p, unblock
	67	+	}
	68	+
	69	+	func newFactory(t testing.T, p ports.PTY) portsmocks.MockPTYFactory {
	70	+	t.Helper()
	71	+	f := portsmocks.NewMockPTYFactory(t)
	72	+	f.EXPECT().Open(mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(p, nil).Maybe()
	73	+	return f
	74	+	}
	75	+
	76	+	// newConn scripts a MockTransport: Recv yields first then more (in order),
	77	+	// then blocks until the connection is released or Closed (returning io.EOF).
	78	+	// Every Send is captured on the returned channel.
	79	+	func newConn(t testing.T, first ports.Frame, more ...ports.Frame) (portsmocks.MockTransport, chan ports.Frame, func()) {
	80	+	t.Helper()
	81	+	tr := portsmocks.NewMockTransport(t)
	82	+	sends := make(chan ports.Frame, 64)
	83	+	recvCh := make(chan ports.Frame, 1+len(more))
	84	+	recvCh <- first
	85	+	for _, f := range more {
	86	+	recvCh <- f
	87	+	}
	88	+	done := make(chan struct{})
	89	+	var once sync.Once
	90	+	closeDone := func() { once.Do(func() { close(done) }) }
	91	+
	92	+	tr.EXPECT().Recv().RunAndReturn(func() (ports.Frame, error) {
	93	+	select {
	94	+	case f := <-recvCh:
	95	+	return f, nil
	96	+	case <-done:
	97	+	return ports.Frame{}, io.EOF
	98	+	}
	99	+	}).Maybe()
	100	+	tr.EXPECT().Send(mock.Anything).RunAndReturn(func(f ports.Frame) error {
	101	+	sends <- f
	102	+	return nil
	103	+	}).Maybe()
	104	+	tr.EXPECT().Close().RunAndReturn(func() error { closeDone(); return nil }).Maybe()
	105	+	return tr, sends, closeDone
	106	+	}
	107	+
	108	+	func mustHello(intent uint8, name string, sz domain.Size) ports.Frame {
	109	+	return ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(ports.Hello{
	110	+	Version: ports.ProtocolVersion, Intent: intent, Name: name, Size: sz, TermEnv: "xterm-256color",
	111	+	})}
	112	+	}
	113	+
	114	+	// awaitFrame waits for the next frame of type typ on ch, failing on timeout.
	115	+	func awaitFrame(t *testing.T, ch chan ports.Frame, typ ports.MsgType) ports.Frame {
	116	+	t.Helper()
	117	+	deadline := time.After(2 * time.Second)
	118	+	for {
	119	+	select {
	120	+	case f := <-ch:
	121	+	if f.Type == typ {
	122	+	return f
	123	+	}
	124	+	case <-deadline:
	125	+	t.Fatalf("timed out waiting for frame type %d", typ)
	126	+	}
	127	+	}
	128	+	}
	129	+
	130	+	func firstSession(d Daemon) session {
	131	+	d.mu.Lock()
	132	+	defer d.mu.Unlock()
	133	+	for _, s := range d.sessions {
	134	+	return s
	135	+	}
	136	+	return nil
	137	+	}
	138	+
	139	+	func sessionCount(d *Daemon) int {
	140	+	d.mu.Lock()
	141	+	defer d.mu.Unlock()
	142	+	return len(d.sessions)
	143	+	}
	144	+
	145	+	// --- handshake --------------------------------------------------------------
	146	+
	147	+	func TestHandshakeEphemeralHappy(t *testing.T) {
	148	+	p, releasePTY := newBlockingPTY(t)
	149	+	d := newTestDaemon(t, newFactory(t, p), stubClock{})
	150	+	tr, sends, releaseConn := newConn(t, mustHello(ports.IntentEphemeral, "", domain.Size{Cols: 80, Rows: 24}))
	151	+
	152	+	var hg sync.WaitGroup
	153	+	hg.Add(1)
	154	+	go func() { defer hg.Done(); d.handleConn(tr) }()
	155	+
	156	+	w := awaitFrame(t, sends, ports.MsgWelcome)
	157	+	welcome, err := ports.UnmarshalWelcome(w.Payload)
	158	+	require.NoError(t, err)
	159	+	require.Equal(t, "0", welcome.SessionName)
	160	+	require.True(t, welcome.Ephemeral)
	161	+	awaitFrame(t, sends, ports.MsgOutput) // guaranteed first paint
	162	+	require.Equal(t, 1, sessionCount(d))
	163	+
	164	+	releaseConn()
	165	+	releasePTY()
	166	+	hg.Wait()
	167	+	d.sessWg.Wait()
	168	+	}
	169	+
	170	+	func TestHandshakeNewHappy(t *testing.T) {
	171	+	p, releasePTY := newBlockingPTY(t)
	172	+	d := newTestDaemon(t, newFactory(t, p), stubClock{})
	173	+	tr, sends, releaseConn := newConn(t, mustHello(ports.IntentNew, "work", domain.Size{Cols: 80, Rows: 24}))
	174	+
	175	+	var hg sync.WaitGroup
	176	+	hg.Add(1)
	177	+	go func() { defer hg.Done(); d.handleConn(tr) }()
	178	+
	179	+	w := awaitFrame(t, sends, ports.MsgWelcome)
	180	+	welcome, err := ports.UnmarshalWelcome(w.Payload)
	181	+	require.NoError(t, err)
	182	+	require.Equal(t, "work", welcome.SessionName)
	183	+	require.False(t, welcome.Ephemeral)
	184	+
	185	+	sess := firstSession(d)
	186	+	require.NotNil(t, sess)
	187	+
	188	+	releaseConn()
	189	+	d.killSession(sess, ports.ReasonServerShutdown)
	190	+	releasePTY()
	191	+	hg.Wait()
	192	+	d.sessWg.Wait()
	193	+	}
	194	+
	195	+	func TestHandshakeVersionMismatch(t *testing.T) {
	196	+	// No Open expectation: the factory must never be asked to spawn a PTY.
	197	+	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
	198	+	bad := ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(ports.Hello{
	199	+	Version: ports.ProtocolVersion + 99, Intent: ports.IntentEphemeral, Size: domain.Size{Cols: 80, Rows: 24},
	200	+	})}
	201	+	tr, sends, _ := newConn(t, bad)
	202	+
	203	+	d.handleConn(tr) // returns after the rejection; no session, no goroutines
	204	+
	205	+	e := awaitFrame(t, sends, ports.MsgError)
	206	+	em, err := ports.UnmarshalErrorMsg(e.Payload)
	207	+	require.NoError(t, err)
	208	+	require.Equal(t, ports.ErrVersionMismatch, em.Code)
	209	+	require.Equal(t, 0, sessionCount(d))
	210	+	}
	211	+
	212	+	func TestHandshakeNameTaken(t *testing.T) {
	213	+	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
	214	+	// Pre-populate a bare session (no goroutines) to collide with.
	215	+	d.sessions[domain.SessionID("x")] = &session{id: "x", name: "work"}
	216	+
	217	+	tr, sends, _ := newConn(t, mustHello(ports.IntentNew, "work", domain.Size{Cols: 80, Rows: 24}))
	218	+	d.handleConn(tr)
	219	+
	220	+	e := awaitFrame(t, sends, ports.MsgError)
	221	+	em, _ := ports.UnmarshalErrorMsg(e.Payload)
	222	+	require.Equal(t, ports.ErrNameTaken, em.Code)
	223	+	}
	224	+
	225	+	func TestHandshakeNoSuchSession(t *testing.T) {
	226	+	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
	227	+	tr, sends, _ := newConn(t, mustHello(ports.IntentAttach, "ghost", domain.Size{Cols: 80, Rows: 24}))
	228	+	d.handleConn(tr)
	229	+
	230	+	e := awaitFrame(t, sends, ports.MsgError)
	231	+	em, _ := ports.UnmarshalErrorMsg(e.Payload)
	232	+	require.Equal(t, ports.ErrNoSuchSession, em.Code)
	233	+	}
	234	+
	235	+	// --- ephemeral numbering ----------------------------------------------------
	236	+
	237	+	func TestEphemeralNumberingReuse(t *testing.T) {
	238	+	d := New(nil, stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)))
	239	+	// Emulate two ephemeral sessions "0" and "1".
	240	+	d.sessions["a"] = &session{id: "a", name: "0"}
	241	+	d.sessions["b"] = &session{id: "b", name: "1"}
	242	+	require.Equal(t, "2", d.allocEphemeralNameLocked())
	243	+
	244	+	// Kill "0": the freed number is reused before "2".
	245	+	delete(d.sessions, "a")
	246	+	require.Equal(t, "0", d.allocEphemeralNameLocked())
	247	+	}
	248	+
	249	+	// --- attach replace ---------------------------------------------------------
	250	+
	251	+	func TestAttachReplaceDetachesOld(t *testing.T) {
	252	+	p, releasePTY := newBlockingPTY(t)
	253	+	d := newTestDaemon(t, newFactory(t, p), stubClock{})
	254	+
	255	+	// Client A creates and attaches to ephemeral "0".
	256	+	trA, sendsA, releaseA := newConn(t, mustHello(ports.IntentEphemeral, "", domain.Size{Cols: 80, Rows: 24}))
	257	+	var hg sync.WaitGroup
	258	+	hg.Add(1)
	259	+	go func() { defer hg.Done(); d.handleConn(trA) }()
	260	+	awaitFrame(t, sendsA, ports.MsgWelcome)
	261	+	awaitFrame(t, sendsA, ports.MsgOutput)
	262	+	sess := firstSession(d)
	263	+	require.NotNil(t, sess)
	264	+
	265	+	// Client B attaches to the same session, displacing A.
	266	+	trB, sendsB, releaseB := newConn(t, mustHello(ports.IntentAttach, "0", domain.Size{Cols: 80, Rows: 24}))
	267	+	hg.Add(1)
	268	+	go func() { defer hg.Done(); d.handleConn(trB) }()
	269	+	awaitFrame(t, sendsB, ports.MsgWelcome)
	270	+
	271	+	// A is notified it was detached.
	272	+	dA := awaitFrame(t, sendsA, ports.MsgDetached)
	273	+	det, _ := ports.UnmarshalDetached(dA.Payload)
	274	+	require.Equal(t, ports.ReasonDetach, det.Reason)
	275	+
	276	+	// B is now the current client.
	277	+	sess.mu.Lock()
	278	+	require.NotNil(t, sess.client)
	279	+	sess.mu.Unlock()
	280	+
	281	+	releaseA()
	282	+	releaseB()
	283	+	d.killSession(sess, ports.ReasonServerShutdown)
	284	+	releasePTY()
	285	+	hg.Wait()
	286	+	d.sessWg.Wait()
	287	+	}
	288	+
	289	+	// --- detach semantics -------------------------------------------------------
	290	+
	291	+	func TestDetachKillsEphemeral(t *testing.T) {
	292	+	p, releasePTY := newBlockingPTY(t)
	293	+	d := newTestDaemon(t, newFactory(t, p), stubClock{})
	294	+	tr, sends, _ := newConn(t,
	295	+	mustHello(ports.IntentEphemeral, "", domain.Size{Cols: 80, Rows: 24}),
	296	+	ports.Frame{Type: ports.MsgDetach, Payload: ports.MarshalDetach(ports.Detach{})},
	297	+	)
	298	+
	299	+	var hg sync.WaitGroup
	300	+	hg.Add(1)
	301	+	go func() { defer hg.Done(); d.handleConn(tr) }()
	302	+	awaitFrame(t, sends, ports.MsgWelcome)
	303	+
	304	+	require.Eventually(t, func() bool { return sessionCount(d) == 0 }, 2time.Second, 5time.Millisecond,
	305	+	"ephemeral session must die on client detach")
	306	+
	307	+	releasePTY()
	308	+	hg.Wait()
	309	+	d.sessWg.Wait()
	310	+	}
	311	+
	312	+	func TestDetachKeepsNamed(t *testing.T) {
	313	+	p, releasePTY := newBlockingPTY(t)
	314	+	d := newTestDaemon(t, newFactory(t, p), stubClock{})
	315	+	tr, sends, _ := newConn(t,
	316	+	mustHello(ports.IntentNew, "keep", domain.Size{Cols: 80, Rows: 24}),
	317	+	ports.Frame{Type: ports.MsgDetach, Payload: ports.MarshalDetach(ports.Detach{})},
	318	+	)
	319	+
	320	+	var hg sync.WaitGroup
	321	+	hg.Add(1)
	322	+	go func() { defer hg.Done(); d.handleConn(tr) }()
	323	+	awaitFrame(t, sends, ports.MsgWelcome)
	324	+	awaitFrame(t, sends, ports.MsgDetached) // ack for explicit detach
	325	+	hg.Wait() // handler returns after detach
	326	+
	327	+	require.Equal(t, 1, sessionCount(d), "named session survives detach")
	328	+	sess := firstSession(d)
	329	+	sess.mu.Lock()
	330	+	require.Nil(t, sess.client, "named session is headless after detach")
	331	+	sess.mu.Unlock()
	332	+
	333	+	d.killSession(sess, ports.ReasonServerShutdown)
	334	+	releasePTY()
	335	+	d.sessWg.Wait()
	336	+	}
	337	+
	338	+	// --- render scheduler debounce ----------------------------------------------
	339	+
	340	+	func TestSchedulerDebounceCoalesces(t *testing.T) {
	341	+	mc := portsmocks.NewMockClock(t)
	342	+	mt := portsmocks.NewMockTimer(t)
	343	+	timerCh := make(chan time.Time, 1)
	344	+	newTimerCalled := make(chan struct{}, 4)
	345	+
	346	+	mc.EXPECT().NewTimer(debounceInterval).RunAndReturn(func(time.Duration) ports.Timer {
	347	+	select {
	348	+	case newTimerCalled <- struct{}{}:
	349	+	default:
	350	+	}
	351	+	return mt
	352	+	}).Maybe()
	353	+	mt.EXPECT().C().Return(timerCh).Maybe()
	354	+	mt.EXPECT().Stop().Return(true).Maybe()
	355	+
	356	+	var outputs atomic.Int32
	357	+	gotOutput := make(chan struct{}, 1)
	358	+	tr := portsmocks.NewMockTransport(t)
	359	+	tr.EXPECT().Send(mock.Anything).RunAndReturn(func(f ports.Frame) error {
	360	+	if f.Type == ports.MsgOutput {
	361	+	outputs.Add(1)
	362	+	select {
	363	+	case gotOutput <- struct{}{}:
	364	+	default:
	365	+	}
	366	+	}
	367	+	return nil
	368	+	}).Maybe()
	369	+
	370	+	d := New(nil, mc, slog.New(slog.NewTextHandler(io.Discard, nil)))
	371	+	sctx, cancel := context.WithCancel(context.Background())
	372	+	win := &window{screen: vt.NewScreen(20, 5), dirty: make(chan struct{}, 1), size: domain.Size{Cols: 20, Rows: 5}}
	373	+	ac := &attachedClient{tr: tr, rend: renderer.New(renderer.Capabilities{})}
	374	+	sess := &session{id: "s", name: "s", win: win, ctx: sctx, cancel: cancel, client: ac}
	375	+
	376	+	win.mu.Lock()
	377	+	win.screen.Write([]byte("hi"))
	378	+	win.mu.Unlock()
	379	+
	380	+	d.sessWg.Add(1)
	381	+	go d.scheduler(sess)
	382	+
	383	+	// First dirty opens the debounce window.
	384	+	win.dirty <- struct{}{}
	385	+	<-newTimerCalled
	386	+
	387	+	// A burst of further dirties inside the window must be absorbed.
	388	+	for i := 0; i < 5; i++ {
	389	+	select {
	390	+	case win.dirty <- struct{}{}:
	391	+	default:
	392	+	}
	393	+	}
	394	+
	395	+	// Fire the timer once: exactly one render.
	396	+	timerCh <- time.Now()
	397	+	<-gotOutput
	398	+
	399	+	cancel()
	400	+	d.sessWg.Wait()
	401	+	require.Equal(t, int32(1), outputs.Load(), "N dirties in one window must render exactly once")
	402	+	}
	403	+
	404	+	// --- resize ordering --------------------------------------------------------
	405	+
	406	+	func TestResizeOrdersPTYBeforeScreen(t *testing.T) {
	407	+	newSize := domain.Size{Cols: 100, Rows: 30}
	408	+
	409	+	p := portsmocks.NewMockPTY(t)
	410	+	var screenWidthAtResize int
	411	+	win := &window{screen: vt.NewScreen(80, 24), dirty: make(chan struct{}, 1), size: domain.Size{Cols: 80, Rows: 24}}
	412	+	p.EXPECT().Resize(newSize).RunAndReturn(func(sz domain.Size) error {
	413	+	// The screen must not yet be resized when the PTY is: proves order.
	414	+	screenWidthAtResize = win.screen.Frame.Width
	415	+	return nil
	416	+	}).Once()
	417	+	win.pty = p
	418	+
	419	+	var gotOutput atomic.Bool
	420	+	tr := portsmocks.NewMockTransport(t)
	421	+	tr.EXPECT().Send(mock.Anything).RunAndReturn(func(f ports.Frame) error {
	422	+	if f.Type == ports.MsgOutput {
	423	+	gotOutput.Store(true)
	424	+	}
	425	+	return nil
	426	+	}).Maybe()
	427	+
	428	+	d := New(nil, stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)))
	429	+	ac := &attachedClient{tr: tr, rend: renderer.New(renderer.Capabilities{})}
	430	+	sess := &session{id: "s", name: "s", win: win, client: ac}
	431	+
	432	+	d.resize(sess, ac, newSize)
	433	+
	434	+	require.Equal(t, 80, screenWidthAtResize, "pty.Resize must run before screen.Resize")
	435	+	require.Equal(t, 100, win.screen.Frame.Width, "screen resized after pty")
	436	+	require.True(t, gotOutput.Load(), "resize forces a full redraw output")
	437	+	}
	438	+
	439	+	// --- reader EOF -> registry-empty shutdown ----------------------------------
	440	+
	441	+	func TestReaderEOFRemovesSessionAndSignalsShutdown(t *testing.T) {
	442	+	p := portsmocks.NewMockPTY(t)
	443	+	// Read returns EOF immediately (child already gone).
	444	+	p.EXPECT().Read(mock.Anything).Return(0, io.EOF).Maybe()
	445	+	p.EXPECT().Close().Return(nil).Maybe()
	446	+	p.EXPECT().Pid().Return(1).Maybe()
	447	+
	448	+	d := newTestDaemon(t, newFactory(t, p), stubClock{})
	449	+	tr, sends, _ := newConn(t, mustHello(ports.IntentEphemeral, "", domain.Size{Cols: 80, Rows: 24}))
	450	+
	451	+	var hg sync.WaitGroup
	452	+	hg.Add(1)
	453	+	go func() { defer hg.Done(); d.handleConn(tr) }()
	454	+
	455	+	// The client is detached with ReasonSessionKilled when the child exits.
	456	+	det := awaitFrame(t, sends, ports.MsgDetached)
	457	+	dm, _ := ports.UnmarshalDetached(det.Payload)
	458	+	require.Equal(t, ports.ReasonSessionKilled, dm.Reason)
	459	+
	460	+	select {
	461	+	case <-d.done:
	462	+	case <-time.After(2 * time.Second):
	463	+	t.Fatal("registry-empty shutdown was not signalled")
	464	+	}
	465	+	require.Equal(t, 0, sessionCount(d))
	466	+
	467	+	hg.Wait()
	468	+	d.sessWg.Wait()
	469	+	}
	470	+
	471	+	// --- full Serve lifecycle ---------------------------------------------------
	472	+
	473	+	func TestServeReturnsWhenLastSessionExits(t *testing.T) {
	474	+	p, releasePTY := newBlockingPTY(t)
	475	+	tr, sends, _ := newConn(t, mustHello(ports.IntentEphemeral, "", domain.Size{Cols: 80, Rows: 24}))
	476	+
	477	+	l := portsmocks.NewMockListener(t)
	478	+	connCh := make(chan ports.Transport, 1)
	479	+	connCh <- tr
	480	+	closed := make(chan struct{})
	481	+	var once sync.Once
	482	+	l.EXPECT().Accept().RunAndReturn(func() (ports.Transport, error) {
	483	+	select {
	484	+	case c := <-connCh:
	485	+	return c, nil
	486	+	case <-closed:
	487	+	return nil, io.EOF
	488	+	}
	489	+	}).Maybe()
	490	+	l.EXPECT().Close().RunAndReturn(func() error { once.Do(func() { close(closed) }); return nil }).Maybe()
	491	+	l.EXPECT().Addr().Return("mock").Maybe()
	492	+
	493	+	d := New(newFactory(t, p), stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)))
	494	+
	495	+	served := make(chan error, 1)
	496	+	go func() { served <- d.Serve(context.Background(), l) }()
	497	+
	498	+	awaitFrame(t, sends, ports.MsgWelcome)
	499	+	awaitFrame(t, sends, ports.MsgOutput)
	500	+
	501	+	// Child exits -> session removed -> registry empties -> Serve returns.
	502	+	releasePTY()
	503	+
	504	+	select {
	505	+	case err := <-served:
	506	+	require.NoError(t, err)
	507	+	case <-time.After(3 * time.Second):
	508	+	t.Fatal("Serve did not return after last session exited")
	509	+	}
	510	+	}
	511	+
	512	+	func TestServeGracefulShutdownOnContextCancel(t *testing.T) {
	513	+	p, releasePTY := newBlockingPTY(t)
	514	+	defer releasePTY()
	515	+	tr, sends, _ := newConn(t, mustHello(ports.IntentNew, "long", domain.Size{Cols: 80, Rows: 24}))
	516	+
	517	+	l := portsmocks.NewMockListener(t)
	518	+	connCh := make(chan ports.Transport, 1)
	519	+	connCh <- tr
	520	+	closed := make(chan struct{})
	521	+	var once sync.Once
	522	+	l.EXPECT().Accept().RunAndReturn(func() (ports.Transport, error) {
	523	+	select {
	524	+	case c := <-connCh:
	525	+	return c, nil
	526	+	case <-closed:
	527	+	return nil, io.EOF
	528	+	}
	529	+	}).Maybe()
	530	+	l.EXPECT().Close().RunAndReturn(func() error { once.Do(func() { close(closed) }); return nil }).Maybe()
	531	+	l.EXPECT().Addr().Return("mock").Maybe()
	532	+
	533	+	d := New(newFactory(t, p), stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)))
	534	+	ctx, cancel := context.WithCancel(context.Background())
	535	+
	536	+	served := make(chan error, 1)
	537	+	go func() { served <- d.Serve(ctx, l) }()
	538	+
	539	+	awaitFrame(t, sends, ports.MsgWelcome)
	540	+	awaitFrame(t, sends, ports.MsgOutput)
	541	+
	542	+	cancel() // graceful shutdown
	543	+
	544	+	det := awaitFrame(t, sends, ports.MsgDetached)
	545	+	dm, _ := ports.UnmarshalDetached(det.Payload)
	546	+	require.Equal(t, ports.ReasonServerShutdown, dm.Reason)
	547	+
	548	+	select {
	549	+	case err := <-served:
	550	+	require.NoError(t, err)
	551	+	case <-time.After(3 * time.Second):
	552	+	t.Fatal("Serve did not return after context cancel")
	553	+	}
	554	+	}

@@ -0,0 +1,554 @@
	1	+	package daemon
	2	+
	3	+	import (
	4	+	"context"
	5	+	"io"
	6	+	"log/slog"
	7	+	"sync"
	8	+	"sync/atomic"
	9	+	"testing"
	10	+	"time"
	11	+
	12	+	"github.com/stretchr/testify/mock"
	13	+	"github.com/stretchr/testify/require"
	14	+
	15	+	"github.com/bnema/vev/internal/domain"
	16	+	"github.com/bnema/vev/internal/ports"
	17	+	portsmocks "github.com/bnema/vev/internal/ports/mocks"
	18	+	"github.com/bnema/vev/pkg/renderer"
	19	+	"github.com/bnema/vev/pkg/vt"
	20	+	)
	21	+
	22	+	// --- test doubles -----------------------------------------------------------
	23	+
	24	+	// stubClock returns timers whose channel never fires, so a scheduler under it
	25	+	// blocks in its debounce loop until the session context is cancelled. Used by
	26	+	// every test that is not specifically exercising the debounce.
	27	+	type stubClock struct{}
	28	+
	29	+	func (stubClock) Now() time.Time { return time.Time{} }
	30	+	func (stubClock) NewTimer(time.Duration) ports.Timer { return stubTimer{} }
	31	+
	32	+	type stubTimer struct{}
	33	+
	34	+	func (stubTimer) C() <-chan time.Time { return nil }
	35	+	func (stubTimer) Reset(time.Duration) bool { return false }
	36	+	func (stubTimer) Stop() bool { return true }
	37	+
	38	+	func newTestDaemon(t testing.T, ptys ports.PTYFactory, clk ports.Clock) Daemon {
	39	+	t.Helper()
	40	+	d := New(ptys, clk, slog.New(slog.NewTextHandler(io.Discard, nil)))
	41	+	d.serveCtx, d.serveCancel = context.WithCancel(context.Background())
	42	+	d.hardCtx, d.hardCancel = context.WithCancel(context.Background())
	43	+	t.Cleanup(d.serveCancel)
	44	+	t.Cleanup(d.hardCancel)
	45	+	return d
	46	+	}
	47	+
	48	+	// newBlockingPTY returns a MockPTY whose Read blocks until release is called,
	49	+	// then reports io.EOF (the "child exited" signal).
	50	+	func newBlockingPTY(t testing.T) (portsmocks.MockPTY, func()) {
	51	+	t.Helper()
	52	+	p := portsmocks.NewMockPTY(t)
	53	+	release := make(chan struct{})
	54	+	var once sync.Once
	55	+	unblock := func() { once.Do(func() { close(release) }) }
	56	+	p.EXPECT().Read(mock.Anything).RunAndReturn(func(b []byte) (int, error) {
	57	+	<-release
	58	+	return 0, io.EOF
	59	+	}).Maybe()
	60	+	p.EXPECT().Write(mock.Anything).RunAndReturn(func(b []byte) (int, error) { return len(b), nil }).Maybe()
	61	+	p.EXPECT().Resize(mock.Anything).Return(nil).Maybe()
	62	+	// Close unblocks a parked Read, exactly as the real PTY does (its Close
	63	+	// closes the master fd, failing the in-flight read).
	64	+	p.EXPECT().Close().RunAndReturn(func() error { unblock(); return nil }).Maybe()
	65	+	p.EXPECT().Pid().Return(4242).Maybe()
	66	+	return p, unblock
	67	+	}
	68	+
	69	+	func newFactory(t testing.T, p ports.PTY) portsmocks.MockPTYFactory {
	70	+	t.Helper()
	71	+	f := portsmocks.NewMockPTYFactory(t)
	72	+	f.EXPECT().Open(mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(p, nil).Maybe()
	73	+	return f
	74	+	}
	75	+
	76	+	// newConn scripts a MockTransport: Recv yields first then more (in order),
	77	+	// then blocks until the connection is released or Closed (returning io.EOF).
	78	+	// Every Send is captured on the returned channel.
	79	+	func newConn(t testing.T, first ports.Frame, more ...ports.Frame) (portsmocks.MockTransport, chan ports.Frame, func()) {
	80	+	t.Helper()
	81	+	tr := portsmocks.NewMockTransport(t)
	82	+	sends := make(chan ports.Frame, 64)
	83	+	recvCh := make(chan ports.Frame, 1+len(more))
	84	+	recvCh <- first
	85	+	for _, f := range more {
	86	+	recvCh <- f
	87	+	}
	88	+	done := make(chan struct{})
	89	+	var once sync.Once
	90	+	closeDone := func() { once.Do(func() { close(done) }) }
	91	+
	92	+	tr.EXPECT().Recv().RunAndReturn(func() (ports.Frame, error) {
	93	+	select {
	94	+	case f := <-recvCh:
	95	+	return f, nil
	96	+	case <-done:
	97	+	return ports.Frame{}, io.EOF
	98	+	}
	99	+	}).Maybe()
	100	+	tr.EXPECT().Send(mock.Anything).RunAndReturn(func(f ports.Frame) error {
	101	+	sends <- f
	102	+	return nil
	103	+	}).Maybe()
	104	+	tr.EXPECT().Close().RunAndReturn(func() error { closeDone(); return nil }).Maybe()
	105	+	return tr, sends, closeDone
	106	+	}
	107	+
	108	+	func mustHello(intent uint8, name string, sz domain.Size) ports.Frame {
	109	+	return ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(ports.Hello{
	110	+	Version: ports.ProtocolVersion, Intent: intent, Name: name, Size: sz, TermEnv: "xterm-256color",
	111	+	})}
	112	+	}
	113	+
	114	+	// awaitFrame waits for the next frame of type typ on ch, failing on timeout.
	115	+	func awaitFrame(t *testing.T, ch chan ports.Frame, typ ports.MsgType) ports.Frame {
	116	+	t.Helper()
	117	+	deadline := time.After(2 * time.Second)
	118	+	for {
	119	+	select {
	120	+	case f := <-ch:
	121	+	if f.Type == typ {
	122	+	return f
	123	+	}
	124	+	case <-deadline:
	125	+	t.Fatalf("timed out waiting for frame type %d", typ)
	126	+	}
	127	+	}
	128	+	}
	129	+
	130	+	func firstSession(d Daemon) session {
	131	+	d.mu.Lock()
	132	+	defer d.mu.Unlock()
	133	+	for _, s := range d.sessions {
	134	+	return s
	135	+	}
	136	+	return nil
	137	+	}
	138	+
	139	+	func sessionCount(d *Daemon) int {
	140	+	d.mu.Lock()
	141	+	defer d.mu.Unlock()
	142	+	return len(d.sessions)
	143	+	}
	144	+
	145	+	// --- handshake --------------------------------------------------------------
	146	+
	147	+	func TestHandshakeEphemeralHappy(t *testing.T) {
	148	+	p, releasePTY := newBlockingPTY(t)
	149	+	d := newTestDaemon(t, newFactory(t, p), stubClock{})
	150	+	tr, sends, releaseConn := newConn(t, mustHello(ports.IntentEphemeral, "", domain.Size{Cols: 80, Rows: 24}))
	151	+
	152	+	var hg sync.WaitGroup
	153	+	hg.Add(1)
	154	+	go func() { defer hg.Done(); d.handleConn(tr) }()
	155	+
	156	+	w := awaitFrame(t, sends, ports.MsgWelcome)
	157	+	welcome, err := ports.UnmarshalWelcome(w.Payload)
	158	+	require.NoError(t, err)
	159	+	require.Equal(t, "0", welcome.SessionName)
	160	+	require.True(t, welcome.Ephemeral)
	161	+	awaitFrame(t, sends, ports.MsgOutput) // guaranteed first paint
	162	+	require.Equal(t, 1, sessionCount(d))
	163	+
	164	+	releaseConn()
	165	+	releasePTY()
	166	+	hg.Wait()
	167	+	d.sessWg.Wait()
	168	+	}
	169	+
	170	+	func TestHandshakeNewHappy(t *testing.T) {
	171	+	p, releasePTY := newBlockingPTY(t)
	172	+	d := newTestDaemon(t, newFactory(t, p), stubClock{})
	173	+	tr, sends, releaseConn := newConn(t, mustHello(ports.IntentNew, "work", domain.Size{Cols: 80, Rows: 24}))
	174	+
	175	+	var hg sync.WaitGroup
	176	+	hg.Add(1)
	177	+	go func() { defer hg.Done(); d.handleConn(tr) }()
	178	+
	179	+	w := awaitFrame(t, sends, ports.MsgWelcome)
	180	+	welcome, err := ports.UnmarshalWelcome(w.Payload)
	181	+	require.NoError(t, err)
	182	+	require.Equal(t, "work", welcome.SessionName)
	183	+	require.False(t, welcome.Ephemeral)
	184	+
	185	+	sess := firstSession(d)
	186	+	require.NotNil(t, sess)
	187	+
	188	+	releaseConn()
	189	+	d.killSession(sess, ports.ReasonServerShutdown)
	190	+	releasePTY()
	191	+	hg.Wait()
	192	+	d.sessWg.Wait()
	193	+	}
	194	+
	195	+	func TestHandshakeVersionMismatch(t *testing.T) {
	196	+	// No Open expectation: the factory must never be asked to spawn a PTY.
	197	+	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
	198	+	bad := ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(ports.Hello{
	199	+	Version: ports.ProtocolVersion + 99, Intent: ports.IntentEphemeral, Size: domain.Size{Cols: 80, Rows: 24},
	200	+	})}
	201	+	tr, sends, _ := newConn(t, bad)
	202	+
	203	+	d.handleConn(tr) // returns after the rejection; no session, no goroutines
	204	+
	205	+	e := awaitFrame(t, sends, ports.MsgError)
	206	+	em, err := ports.UnmarshalErrorMsg(e.Payload)
	207	+	require.NoError(t, err)
	208	+	require.Equal(t, ports.ErrVersionMismatch, em.Code)
	209	+	require.Equal(t, 0, sessionCount(d))
	210	+	}
	211	+
	212	+	func TestHandshakeNameTaken(t *testing.T) {
	213	+	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
	214	+	// Pre-populate a bare session (no goroutines) to collide with.
	215	+	d.sessions[domain.SessionID("x")] = &session{id: "x", name: "work"}
	216	+
	217	+	tr, sends, _ := newConn(t, mustHello(ports.IntentNew, "work", domain.Size{Cols: 80, Rows: 24}))
	218	+	d.handleConn(tr)
	219	+
	220	+	e := awaitFrame(t, sends, ports.MsgError)
	221	+	em, _ := ports.UnmarshalErrorMsg(e.Payload)
	222	+	require.Equal(t, ports.ErrNameTaken, em.Code)
	223	+	}
	224	+
	225	+	func TestHandshakeNoSuchSession(t *testing.T) {
	226	+	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
	227	+	tr, sends, _ := newConn(t, mustHello(ports.IntentAttach, "ghost", domain.Size{Cols: 80, Rows: 24}))
	228	+	d.handleConn(tr)
	229	+
	230	+	e := awaitFrame(t, sends, ports.MsgError)
	231	+	em, _ := ports.UnmarshalErrorMsg(e.Payload)
	232	+	require.Equal(t, ports.ErrNoSuchSession, em.Code)
	233	+	}
	234	+
	235	+	// --- ephemeral numbering ----------------------------------------------------
	236	+
	237	+	func TestEphemeralNumberingReuse(t *testing.T) {
	238	+	d := New(nil, stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)))
	239	+	// Emulate two ephemeral sessions "0" and "1".
	240	+	d.sessions["a"] = &session{id: "a", name: "0"}
	241	+	d.sessions["b"] = &session{id: "b", name: "1"}
	242	+	require.Equal(t, "2", d.allocEphemeralNameLocked())
	243	+
	244	+	// Kill "0": the freed number is reused before "2".
	245	+	delete(d.sessions, "a")
	246	+	require.Equal(t, "0", d.allocEphemeralNameLocked())
	247	+	}
	248	+
	249	+	// --- attach replace ---------------------------------------------------------
	250	+
	251	+	func TestAttachReplaceDetachesOld(t *testing.T) {
	252	+	p, releasePTY := newBlockingPTY(t)
	253	+	d := newTestDaemon(t, newFactory(t, p), stubClock{})
	254	+
	255	+	// Client A creates and attaches to ephemeral "0".
	256	+	trA, sendsA, releaseA := newConn(t, mustHello(ports.IntentEphemeral, "", domain.Size{Cols: 80, Rows: 24}))
	257	+	var hg sync.WaitGroup
	258	+	hg.Add(1)
	259	+	go func() { defer hg.Done(); d.handleConn(trA) }()
	260	+	awaitFrame(t, sendsA, ports.MsgWelcome)
	261	+	awaitFrame(t, sendsA, ports.MsgOutput)
	262	+	sess := firstSession(d)
	263	+	require.NotNil(t, sess)
	264	+
	265	+	// Client B attaches to the same session, displacing A.
	266	+	trB, sendsB, releaseB := newConn(t, mustHello(ports.IntentAttach, "0", domain.Size{Cols: 80, Rows: 24}))
	267	+	hg.Add(1)
	268	+	go func() { defer hg.Done(); d.handleConn(trB) }()
	269	+	awaitFrame(t, sendsB, ports.MsgWelcome)
	270	+
	271	+	// A is notified it was detached.
	272	+	dA := awaitFrame(t, sendsA, ports.MsgDetached)
	273	+	det, _ := ports.UnmarshalDetached(dA.Payload)
	274	+	require.Equal(t, ports.ReasonDetach, det.Reason)
	275	+
	276	+	// B is now the current client.
	277	+	sess.mu.Lock()
	278	+	require.NotNil(t, sess.client)
	279	+	sess.mu.Unlock()
	280	+
	281	+	releaseA()
	282	+	releaseB()
	283	+	d.killSession(sess, ports.ReasonServerShutdown)
	284	+	releasePTY()
	285	+	hg.Wait()
	286	+	d.sessWg.Wait()
	287	+	}
	288	+
	289	+	// --- detach semantics -------------------------------------------------------
	290	+
	291	+	func TestDetachKillsEphemeral(t *testing.T) {
	292	+	p, releasePTY := newBlockingPTY(t)
	293	+	d := newTestDaemon(t, newFactory(t, p), stubClock{})
	294	+	tr, sends, _ := newConn(t,
	295	+	mustHello(ports.IntentEphemeral, "", domain.Size{Cols: 80, Rows: 24}),
	296	+	ports.Frame{Type: ports.MsgDetach, Payload: ports.MarshalDetach(ports.Detach{})},
	297	+	)
	298	+
	299	+	var hg sync.WaitGroup
	300	+	hg.Add(1)
	301	+	go func() { defer hg.Done(); d.handleConn(tr) }()
	302	+	awaitFrame(t, sends, ports.MsgWelcome)
	303	+
	304	+	require.Eventually(t, func() bool { return sessionCount(d) == 0 }, 2time.Second, 5time.Millisecond,
	305	+	"ephemeral session must die on client detach")
	306	+
	307	+	releasePTY()
	308	+	hg.Wait()
	309	+	d.sessWg.Wait()
	310	+	}
	311	+
	312	+	func TestDetachKeepsNamed(t *testing.T) {
	313	+	p, releasePTY := newBlockingPTY(t)
	314	+	d := newTestDaemon(t, newFactory(t, p), stubClock{})
	315	+	tr, sends, _ := newConn(t,
	316	+	mustHello(ports.IntentNew, "keep", domain.Size{Cols: 80, Rows: 24}),
	317	+	ports.Frame{Type: ports.MsgDetach, Payload: ports.MarshalDetach(ports.Detach{})},
	318	+	)
	319	+
	320	+	var hg sync.WaitGroup
	321	+	hg.Add(1)
	322	+	go func() { defer hg.Done(); d.handleConn(tr) }()
	323	+	awaitFrame(t, sends, ports.MsgWelcome)
	324	+	awaitFrame(t, sends, ports.MsgDetached) // ack for explicit detach
	325	+	hg.Wait() // handler returns after detach
	326	+
	327	+	require.Equal(t, 1, sessionCount(d), "named session survives detach")
	328	+	sess := firstSession(d)
	329	+	sess.mu.Lock()
	330	+	require.Nil(t, sess.client, "named session is headless after detach")
	331	+	sess.mu.Unlock()
	332	+
	333	+	d.killSession(sess, ports.ReasonServerShutdown)
	334	+	releasePTY()
	335	+	d.sessWg.Wait()
	336	+	}
	337	+
	338	+	// --- render scheduler debounce ----------------------------------------------
	339	+
	340	+	func TestSchedulerDebounceCoalesces(t *testing.T) {
	341	+	mc := portsmocks.NewMockClock(t)
	342	+	mt := portsmocks.NewMockTimer(t)
	343	+	timerCh := make(chan time.Time, 1)
	344	+	newTimerCalled := make(chan struct{}, 4)
	345	+
	346	+	mc.EXPECT().NewTimer(debounceInterval).RunAndReturn(func(time.Duration) ports.Timer {
	347	+	select {
	348	+	case newTimerCalled <- struct{}{}:
	349	+	default:
	350	+	}
	351	+	return mt
	352	+	}).Maybe()
	353	+	mt.EXPECT().C().Return(timerCh).Maybe()
	354	+	mt.EXPECT().Stop().Return(true).Maybe()
	355	+
	356	+	var outputs atomic.Int32
	357	+	gotOutput := make(chan struct{}, 1)
	358	+	tr := portsmocks.NewMockTransport(t)
	359	+	tr.EXPECT().Send(mock.Anything).RunAndReturn(func(f ports.Frame) error {
	360	+	if f.Type == ports.MsgOutput {
	361	+	outputs.Add(1)
	362	+	select {
	363	+	case gotOutput <- struct{}{}:
	364	+	default:
	365	+	}
	366	+	}
	367	+	return nil
	368	+	}).Maybe()
	369	+
	370	+	d := New(nil, mc, slog.New(slog.NewTextHandler(io.Discard, nil)))
	371	+	sctx, cancel := context.WithCancel(context.Background())
	372	+	win := &window{screen: vt.NewScreen(20, 5), dirty: make(chan struct{}, 1), size: domain.Size{Cols: 20, Rows: 5}}
	373	+	ac := &attachedClient{tr: tr, rend: renderer.New(renderer.Capabilities{})}
	374	+	sess := &session{id: "s", name: "s", win: win, ctx: sctx, cancel: cancel, client: ac}
	375	+
	376	+	win.mu.Lock()
	377	+	win.screen.Write([]byte("hi"))
	378	+	win.mu.Unlock()
	379	+
	380	+	d.sessWg.Add(1)
	381	+	go d.scheduler(sess)
	382	+
	383	+	// First dirty opens the debounce window.
	384	+	win.dirty <- struct{}{}
	385	+	<-newTimerCalled
	386	+
	387	+	// A burst of further dirties inside the window must be absorbed.
	388	+	for i := 0; i < 5; i++ {
	389	+	select {
	390	+	case win.dirty <- struct{}{}:
	391	+	default:
	392	+	}
	393	+	}
	394	+
	395	+	// Fire the timer once: exactly one render.
	396	+	timerCh <- time.Now()
	397	+	<-gotOutput
	398	+
	399	+	cancel()
	400	+	d.sessWg.Wait()
	401	+	require.Equal(t, int32(1), outputs.Load(), "N dirties in one window must render exactly once")
	402	+	}
	403	+
	404	+	// --- resize ordering --------------------------------------------------------
	405	+
	406	+	func TestResizeOrdersPTYBeforeScreen(t *testing.T) {
	407	+	newSize := domain.Size{Cols: 100, Rows: 30}
	408	+
	409	+	p := portsmocks.NewMockPTY(t)
	410	+	var screenWidthAtResize int
	411	+	win := &window{screen: vt.NewScreen(80, 24), dirty: make(chan struct{}, 1), size: domain.Size{Cols: 80, Rows: 24}}
	412	+	p.EXPECT().Resize(newSize).RunAndReturn(func(sz domain.Size) error {
	413	+	// The screen must not yet be resized when the PTY is: proves order.
	414	+	screenWidthAtResize = win.screen.Frame.Width
	415	+	return nil
	416	+	}).Once()
	417	+	win.pty = p
	418	+
	419	+	var gotOutput atomic.Bool
	420	+	tr := portsmocks.NewMockTransport(t)
	421	+	tr.EXPECT().Send(mock.Anything).RunAndReturn(func(f ports.Frame) error {
	422	+	if f.Type == ports.MsgOutput {
	423	+	gotOutput.Store(true)
	424	+	}
	425	+	return nil
	426	+	}).Maybe()
	427	+
	428	+	d := New(nil, stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)))
	429	+	ac := &attachedClient{tr: tr, rend: renderer.New(renderer.Capabilities{})}
	430	+	sess := &session{id: "s", name: "s", win: win, client: ac}
	431	+
	432	+	d.resize(sess, ac, newSize)
	433	+
	434	+	require.Equal(t, 80, screenWidthAtResize, "pty.Resize must run before screen.Resize")
	435	+	require.Equal(t, 100, win.screen.Frame.Width, "screen resized after pty")
	436	+	require.True(t, gotOutput.Load(), "resize forces a full redraw output")
	437	+	}
	438	+
	439	+	// --- reader EOF -> registry-empty shutdown ----------------------------------
	440	+
	441	+	func TestReaderEOFRemovesSessionAndSignalsShutdown(t *testing.T) {
	442	+	p := portsmocks.NewMockPTY(t)
	443	+	// Read returns EOF immediately (child already gone).
	444	+	p.EXPECT().Read(mock.Anything).Return(0, io.EOF).Maybe()
	445	+	p.EXPECT().Close().Return(nil).Maybe()
	446	+	p.EXPECT().Pid().Return(1).Maybe()
	447	+
	448	+	d := newTestDaemon(t, newFactory(t, p), stubClock{})
	449	+	tr, sends, _ := newConn(t, mustHello(ports.IntentEphemeral, "", domain.Size{Cols: 80, Rows: 24}))
	450	+
	451	+	var hg sync.WaitGroup
	452	+	hg.Add(1)
	453	+	go func() { defer hg.Done(); d.handleConn(tr) }()
	454	+
	455	+	// The client is detached with ReasonSessionKilled when the child exits.
	456	+	det := awaitFrame(t, sends, ports.MsgDetached)
	457	+	dm, _ := ports.UnmarshalDetached(det.Payload)
	458	+	require.Equal(t, ports.ReasonSessionKilled, dm.Reason)
	459	+
	460	+	select {
	461	+	case <-d.done:
	462	+	case <-time.After(2 * time.Second):
	463	+	t.Fatal("registry-empty shutdown was not signalled")
	464	+	}
	465	+	require.Equal(t, 0, sessionCount(d))
	466	+
	467	+	hg.Wait()
	468	+	d.sessWg.Wait()
	469	+	}
	470	+
	471	+	// --- full Serve lifecycle ---------------------------------------------------
	472	+
	473	+	func TestServeReturnsWhenLastSessionExits(t *testing.T) {
	474	+	p, releasePTY := newBlockingPTY(t)
	475	+	tr, sends, _ := newConn(t, mustHello(ports.IntentEphemeral, "", domain.Size{Cols: 80, Rows: 24}))
	476	+
	477	+	l := portsmocks.NewMockListener(t)
	478	+	connCh := make(chan ports.Transport, 1)
	479	+	connCh <- tr
	480	+	closed := make(chan struct{})
	481	+	var once sync.Once
	482	+	l.EXPECT().Accept().RunAndReturn(func() (ports.Transport, error) {
	483	+	select {
	484	+	case c := <-connCh:
	485	+	return c, nil
	486	+	case <-closed:
	487	+	return nil, io.EOF
	488	+	}
	489	+	}).Maybe()
	490	+	l.EXPECT().Close().RunAndReturn(func() error { once.Do(func() { close(closed) }); return nil }).Maybe()
	491	+	l.EXPECT().Addr().Return("mock").Maybe()
	492	+
	493	+	d := New(newFactory(t, p), stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)))
	494	+
	495	+	served := make(chan error, 1)
	496	+	go func() { served <- d.Serve(context.Background(), l) }()
	497	+
	498	+	awaitFrame(t, sends, ports.MsgWelcome)
	499	+	awaitFrame(t, sends, ports.MsgOutput)
	500	+
	501	+	// Child exits -> session removed -> registry empties -> Serve returns.
	502	+	releasePTY()
	503	+
	504	+	select {
	505	+	case err := <-served:
	506	+	require.NoError(t, err)
	507	+	case <-time.After(3 * time.Second):
	508	+	t.Fatal("Serve did not return after last session exited")
	509	+	}
	510	+	}
	511	+
	512	+	func TestServeGracefulShutdownOnContextCancel(t *testing.T) {
	513	+	p, releasePTY := newBlockingPTY(t)
	514	+	defer releasePTY()
	515	+	tr, sends, _ := newConn(t, mustHello(ports.IntentNew, "long", domain.Size{Cols: 80, Rows: 24}))
	516	+
	517	+	l := portsmocks.NewMockListener(t)
	518	+	connCh := make(chan ports.Transport, 1)
	519	+	connCh <- tr
	520	+	closed := make(chan struct{})
	521	+	var once sync.Once
	522	+	l.EXPECT().Accept().RunAndReturn(func() (ports.Transport, error) {
	523	+	select {
	524	+	case c := <-connCh:
	525	+	return c, nil
	526	+	case <-closed:
	527	+	return nil, io.EOF
	528	+	}
	529	+	}).Maybe()
	530	+	l.EXPECT().Close().RunAndReturn(func() error { once.Do(func() { close(closed) }); return nil }).Maybe()
	531	+	l.EXPECT().Addr().Return("mock").Maybe()
	532	+
	533	+	d := New(newFactory(t, p), stubClock{}, slog.New(slog.NewTextHandler(io.Discard, nil)))
	534	+	ctx, cancel := context.WithCancel(context.Background())
	535	+
	536	+	served := make(chan error, 1)
	537	+	go func() { served <- d.Serve(ctx, l) }()
	538	+
	539	+	awaitFrame(t, sends, ports.MsgWelcome)
	540	+	awaitFrame(t, sends, ports.MsgOutput)
	541	+
	542	+	cancel() // graceful shutdown
	543	+
	544	+	det := awaitFrame(t, sends, ports.MsgDetached)
	545	+	dm, _ := ports.UnmarshalDetached(det.Payload)
	546	+	require.Equal(t, ports.ReasonServerShutdown, dm.Reason)
	547	+
	548	+	select {
	549	+	case err := <-served:
	550	+	require.NoError(t, err)
	551	+	case <-time.After(3 * time.Second):
	552	+	t.Fatal("Serve did not return after context cancel")
	553	+	}
	554	+	}