Maven/internal/voice/server.go

// voice/server.go — the daemon-side TCP listener.
//
// Accept loop mirrors ipc/server.go shape: one Server, one goroutine per
// conn, recover per conn so a misbehaving client can't kill core. Two key
// differences from ipc:
//
//   - Each conn registers a Session in *Sessions before it reads the first
//     frame. The conn serves requests AND receives server-initiated Pushes
//     through the same conn (the voicesink calls Sessions.PushToMostRecent,
//     which finds the session by lastActive and writes a Push frame on
//     its conn). The conn's write side is therefore shared: serveConn's
//     Response writes vs. the voicesink's Push writes; both serialize via
//     the per-Session mutex.
//
//   - The handler is a voice.Handler, NOT a CoreAPI-style interface. The
//     single method, HandlePushToTalk(ctx, req, sessionID) → resp, does
//     the full reactive path (stt → router → action → replier → tts) and
//     returns the reply. The daemon provides a concrete impl wired to its
//     stt/tts/router/coreAPI; the voice package stays free of those imports
//     (it's just the wire surface).
package voice

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log"
	"net"
	"sync"
	"time"
)

// Handler — the daemon-side reactive path. The voice package defines the
// interface; the daemon wires a concrete handler that knows stt/tts/router.
// This keeps voice from importing every other package in the system.
//
// The handler is called on a per-conn goroutine; it must be safe for
// concurrent use by multiple callers (the daemon's impl routes through
// the wired singleton stt.Transcriber / tts.Synthesizer / router, all of
// which are concurrency-safe).
type Handler interface {
	HandlePushToTalk(ctx context.Context, req PushToTalkReq, sessionID uint64) (PushToTalkResp, error)
}

// Server — maven's client↔core network surface. Listens on a TCP address
// (inside the wg tunnel; bind to wg-egress only — the listener doesn't
// enforce that, the daemon's config picks the bind addr). Each conn is a
// session. Server holds *Sessions so the voicesink can ask
// PushToMostRecent.
type Server struct {
	addr     string
	handler  Handler
	sessions *Sessions

	ln   net.Listener
	wg   sync.WaitGroup
	done chan struct{}

	// Accepted conns, tracked so Close can shut them. Same defect as
	// ipc/server.go had: closing only the listener leaves every idle client
	// parked in readFrame, wg.Wait never returns, and the daemon dies to
	// SIGKILL without sealing the database.
	connMu sync.Mutex
	conns  map[net.Conn]struct{}
}

// closeGrace — how long Close waits for in-flight dispatches before dropping
// them. A push-to-talk turn can be mid-inference; abandoning one costs a reply,
// hanging costs every write since the last clean shutdown.
const closeGrace = 3 * time.Second

// NewServer builds a Server bound to addr (e.g. "127.0.0.1:9100" for a
// local-only smoke; production: a wg-tunnel address). handler is the
// reactive handler; sessions is shared with the voicesink (the daemon
// constructs one, passes to both Server and voicesink).
func NewServer(addr string, handler Handler, sessions *Sessions) *Server {
	return &Server{
		addr:     addr,
		handler:  handler,
		sessions: sessions,
		done:     make(chan struct{}),
	}
}

// Listen binds the TCP listener. Today's floor is plaintext — the auth
// cascade (mTLS, passkey) layers in at the same addr without changing the
// wire shape. Production picks a bind addr that's inside the wg tunnel
// (the wg layer IS the L0 floor); the listener doesn't enforce that, the
// config does.
func (s *Server) Listen() error {
	ln, err := net.Listen("tcp", s.addr)
	if err != nil {
		return fmt.Errorf("voice: listen %s: %w", s.addr, err)
	}
	s.ln = ln
	return nil
}

// Addr returns the bound TCP address (after Listen).
func (s *Server) Addr() string {
	if s.ln == nil {
		return s.addr
	}
	return s.ln.Addr().String()
}

// Serve accepts connections until the listener closes. Per-conn goroutine;
// per-conn recover so a misbehaving client can't crash core.
func (s *Server) Serve() error {
	if s.ln == nil {
		return fmt.Errorf("voice: serve before listen")
	}
	for {
		c, err := s.ln.Accept()
		if err != nil {
			select {
			case <-s.done:
				return nil
			default:
				return fmt.Errorf("voice: accept: %w", err)
			}
		}
		s.wg.Add(1)
		s.trackConn(c)
		go func(c net.Conn) {
			defer s.wg.Done()
			defer s.untrackConn(c)
			s.serveConn(c)
		}(c)
	}
}

// serveConn — one client's lifecycle. Registers a Session, reads requests
// in a loop, dispatches to Handler, writes Responses, removes the session
// on EOF / read error / ctx cancel.
func (s *Server) serveConn(c net.Conn) {
	defer c.Close()
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	// TODO(step-up): the auth handshake populates the surface from mTLS /
	// passkey enrollment. Today the floor sets SurfacePCClient (the
	// reference client's surface, capped at L3 per auth.MaxLayer). The
	// reference client doesn't carry passkey yet, so the floor is "you
	// got through the wg tunnel ⇒ you're on SurfacePCClient by name; the
	// passkey step-up will cap-before-L3 untrusted pop sessions later."
	sess := s.sessions.Add(c, SurfacePCClient)
	defer s.sessions.Remove(sess.ID)
	log.Printf("voice: client %d connected from %s", sess.ID, sess.RemoteAddr)

	for {
		var req Request
		if err := readFrame(c, &req); err != nil {
			if errors.Is(err, io.EOF) {
				// quiet disconnect; common during shutdown.
			} else {
				log.Printf("voice: client %d read: %v", sess.ID, err)
			}
			return
		}
		s.sessions.Touch(sess.ID, time.Now())

		result, err := s.safeDispatch(ctx, c.RemoteAddr(), sess.ID, req)
		resp := Response{ID: req.ID}
		if err != nil {
			resp.Error = rpcErr(err)
		} else {
			resp.Result = result
		}
		if err := writeFrame(c, &resp); err != nil {
			log.Printf("voice: client %d write: %v", sess.ID, err)
			return
		}
	}
}

func (s *Server) safeDispatch(ctx context.Context, addr net.Addr, sid uint64, req Request) (result json.RawMessage, err error) {
	defer func() {
		if r := recover(); r != nil {
			err = fmt.Errorf("voice: panic dispatching %s (sid %d): %v", req.Method, sid, r)
		}
	}()
	return s.dispatch(ctx, sid, req)
}

func (s *Server) dispatch(ctx context.Context, sid uint64, req Request) (json.RawMessage, error) {
	switch req.Method {
	case MethodPushToTalk:
		var p PushToTalkReq
		if err := unmarshalParams(req.Params, &p); err != nil {
			return nil, err
		}
		// Floor default: surface = SurfacePCClient (set here so a floor
		// client that didn't populate the wire field still gets the
		// reference surface). Production: handshake sets it; req.Surface
		// wins over the default.
		if p.Surface == "" {
			p.Surface = SurfacePCClient
		}
		resp, err := s.handler.HandlePushToTalk(ctx, p, sid)
		if err != nil {
			return nil, err
		}
		return marshalResult(resp), nil

	case MethodPong:
		// Pong updates lastActive (the Touch above already did it on the
		// read); no further action. Returns an empty success.
		return marshalResult(nil), nil

	default:
		return nil, fmt.Errorf("%w: %s", ErrUnknownMethod, req.Method)
	}
}

// Close stops accepting and waits for in-flight conns to drain.
func (s *Server) Close() error {
	select {
	case <-s.done:
		return nil
	default:
		close(s.done)
	}
	var err error
	if s.ln != nil {
		err = s.ln.Close()
	}
	// Close the accepted conns too, or a client that is merely idle keeps
	// serveConn blocked in readFrame forever.
	s.closeConns()
	if !waitTimeout(&s.wg, closeGrace) {
		log.Printf("voice: %d connection(s) still busy after %s, closing anyway", s.liveConns(), closeGrace)
	}
	return err
}

func (s *Server) trackConn(c net.Conn) {
	s.connMu.Lock()
	defer s.connMu.Unlock()
	if s.conns == nil {
		s.conns = make(map[net.Conn]struct{})
	}
	s.conns[c] = struct{}{}
}

func (s *Server) untrackConn(c net.Conn) {
	s.connMu.Lock()
	defer s.connMu.Unlock()
	delete(s.conns, c)
}

func (s *Server) liveConns() int {
	s.connMu.Lock()
	defer s.connMu.Unlock()
	return len(s.conns)
}

// closeConns unblocks every parked reader. serveConn's own defer closes the
// conn again; a second Close on a net.Conn is a harmless error.
func (s *Server) closeConns() {
	s.connMu.Lock()
	conns := make([]net.Conn, 0, len(s.conns))
	for c := range s.conns {
		conns = append(conns, c)
	}
	s.connMu.Unlock()
	for _, c := range conns {
		_ = c.Close()
	}
}

// waitTimeout waits on wg, but not forever. Reports whether it finished.
func waitTimeout(wg *sync.WaitGroup, d time.Duration) bool {
	done := make(chan struct{})
	go func() {
		wg.Wait()
		close(done)
	}()
	select {
	case <-done:
		return true
	case <-time.After(d):
		return false
	}
}

func unmarshalParams(raw json.RawMessage, v any) error {
	if len(raw) == 0 {
		raw = []byte("null")
	}
	if err := json.Unmarshal(raw, v); err != nil {
		return fmt.Errorf("%w: %v", ErrBadParams, err)
	}
	return nil
}

func marshalResult(v any) json.RawMessage {
	if v == nil {
		return json.RawMessage("null")
	}
	b, _ := json.Marshal(v)
	return b
}

// io.EOF — used by serveConn to detect a quiet client disconnect.