refactor(sse): redesign with session model and type-safe events

Introduce a Session-based API that encapsulates topic management
inside the package, removing the exported TopicKey and context.Value
pattern from consumers. EventType is now a typed enum (Stdout, Stderr,
Done) replacing loose string constants and hardcoded "done". Server
is no longer an http.Handler — sessions handle the upgrade via
Session.ServeHTTP. Publish returns errors instead of silently
swallowing them; the management layer logs failures via a publish()
helper. OnSession now writes 403 Forbidden on rejection instead of
returning a misleading 200. SSE Shutdown is wired into the management
server lifecycle after Serve returns. The logrus dependency moves from
the sse package to the management package where logging actually
happens.

Author: ihexon

.gitignore

@@ -9,3 +9,4 @@ CLAUDE.md
 task.md
 ANALYSIS.md
 test_20runs.sh
+.qoder

pkg/service/management/server.go

@@ -13,8 +13,9 @@ import (
 	ssev2 "linuxvm/pkg/sse"
 	"net/http"
 	"sync"
+	"time"
 
-	"github.com/google/uuid"
+	"github.com/sirupsen/logrus"
 )
 
 type Server struct {
@@ -51,19 +52,24 @@ func NewServer(machine Machine) (*Server, error) {
 	return &Server{
 		machine: machine,
 		srv:     httpv2.NewUnixSockHTTPServer("management-api", config.Endpoints.ManagementAPI),
-		sse:     ssev2.NewSSEServer(),
+		sse:     ssev2.NewServer(),
 	}, nil
 }
 
 func (s *Server) Start(ctx context.Context) error {
-	// new management api
 	s.srv.Mux.HandleFunc("/v2/healthz", s.handleHealth)
 	s.srv.Mux.HandleFunc("/v2/vmconfig", s.handleVMConfig)
 	s.srv.Mux.HandleFunc("/v2/attach", s.handleAttach)
 	s.srv.Mux.HandleFunc("/v2/exec", s.handleExec)
 	s.srv.Mux.HandleFunc("/v2/stop", s.handleRequestVMStop)
 
-	return s.srv.Serve(ctx)
+	err := s.srv.Serve(ctx)
+
+	shutdownCtx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+	_ = s.sse.Shutdown(shutdownCtx)
+
+	return err
 }
 
 type Info struct {
@@ -122,18 +128,18 @@ func (s *Server) handleExec(w http.ResponseWriter, r *http.Request) {
 		http.Error(w, "invalid json", http.StatusBadRequest)
 		return
 	}
-	topic := "sess-" + uuid.NewString()
-	ctx, cancel := context.WithCancel(context.WithValue(r.Context(), ssev2.TopicKey, topic)) //nolint:staticcheck
+	sess := s.sse.BeginSession()
+	ctx, cancel := context.WithCancel(r.Context())
 	defer cancel()
-	go s.executeCommand(ctx, cancel, topic, req)
-	s.sse.ServeHTTP(w, r.WithContext(ctx))
+	go s.executeCommand(ctx, cancel, sess, req)
+	sess.ServeHTTP(w, r.WithContext(ctx))
 }
 
-func (s *Server) executeCommand(ctx context.Context, cancel context.CancelFunc, topic string, req execRequest) {
+func (s *Server) executeCommand(ctx context.Context, cancel context.CancelFunc, sess *ssev2.Session, req execRequest) {
 	defer cancel()
 	proc, err := sshsvc.GuestExec(ctx, s.machine.SSHTarget(), req.Bin, req.Args...)
 	if err != nil {
-		s.sse.Publish(topic, ssev2.TypeErr, "guest exec failed: "+err.Error())
+		publish(sess, ssev2.Stderr, "guest exec failed: "+err.Error())
 		return
 	}
 	var wg sync.WaitGroup
@@ -143,21 +149,27 @@ func (s *Server) executeCommand(ctx context.Context, cancel context.CancelFunc,
 		sc := bufio.NewScanner(proc.StdoutPipeReader)
 		sc.Buffer(make([]byte, 64*1024), 1<<20)
 		for sc.Scan() {
-			s.sse.Publish(topic, ssev2.TypeOut, sc.Text())
+			publish(sess, ssev2.Stdout, sc.Text())
 		}
 	}()
 	go func() {
 		defer wg.Done()
 		sc := bufio.NewScanner(proc.StderrPipeReader)
 		sc.Buffer(make([]byte, 64*1024), 1<<20)
 		for sc.Scan() {
-			s.sse.Publish(topic, ssev2.TypeErr, sc.Text())
+			publish(sess, ssev2.Stderr, sc.Text())
 		}
 	}()
 	wg.Wait()
 	if err := <-proc.ErrChan; err != nil {
-		s.sse.Publish(topic, ssev2.TypeErr, "wait: "+err.Error())
+		publish(sess, ssev2.Stderr, "wait: "+err.Error())
 		return
 	}
-	s.sse.Publish(topic, "done", "done")
+	publish(sess, ssev2.Done, "done")
 }
+
+func publish(sess *ssev2.Session, typ ssev2.EventType, data string) {
+	if err := sess.Publish(typ, data); err != nil {
+		logrus.Warnf("sse: publish failed on session %s: %v", sess.Topic(), err)
+	}
+}

pkg/sse/sse_server.go

@@ -3,32 +3,37 @@
 package sse
 
 import (
+	"context"
 	"net/http"
 
-	"github.com/sirupsen/logrus"
-	"github.com/tmaxmax/go-sse"
+	"github.com/google/uuid"
+	gosse "github.com/tmaxmax/go-sse"
 )
 
-// SSE message types
-const (
-	TypeOut = "out"
-	TypeErr = "error"
+// EventType identifies the kind of SSE event sent to clients.
+type EventType string
 
-	TopicKey = "sseTopicKey"
+const (
+	Stdout EventType = "out"
+	Stderr EventType = "error"
+	Done   EventType = "done"
 )
 
-// Server wraps the SSE server with helper methods.
+// Server creates and manages SSE sessions. It owns the underlying
+// pub-sub provider and is responsible for shutting it down on exit.
 type Server struct {
-	server *sse.Server
+	inner *gosse.Server
 }
 
-func NewSSEServer() *Server {
+// NewServer creates an SSE server. Sessions are created via BeginSession;
+// the server itself is not an http.Handler.
+func NewServer() *Server {
 	return &Server{
-		server: &sse.Server{
+		inner: &gosse.Server{
 			OnSession: func(w http.ResponseWriter, r *http.Request) ([]string, bool) {
-				topic, ok := r.Context().Value(TopicKey).(string)
+				topic, ok := r.Context().Value(internalTopicKey).(string)
 				if !ok || topic == "" {
-					logrus.Warn("sse: empty topic in session")
+					w.WriteHeader(http.StatusForbidden)
 					return nil, false
 				}
 				return []string{topic}, true
@@ -37,16 +42,45 @@ func NewSSEServer() *Server {
 	}
 }
 
-func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
-	s.server.ServeHTTP(w, r)
+// Shutdown closes the SSE provider, stopping all active subscriptions.
+func (s *Server) Shutdown(ctx context.Context) error {
+	return s.inner.Shutdown(ctx)
 }
 
-func (s *Server) Publish(topic, msgType, data string) {
-	msg := &sse.Message{}
-	msg.AppendData(data)
-	msg.Type = sse.Type(msgType)
+// Session represents a single SSE client connection tied to a unique topic.
+// Each session publishes events on its own topic, isolating streams between clients.
+type Session struct {
+	server *Server
+	topic  string
+}
 
-	if err := s.server.Publish(msg, topic); err != nil {
-		logrus.Warnf("sse: failed to publish message: %v", err)
+// BeginSession creates a new SSE session with a unique topic.
+func (s *Server) BeginSession() *Session {
+	return &Session{
+		server: s,
+		topic:  "sess-" + uuid.NewString(),
 	}
 }
+
+// Topic returns the session's unique identifier, useful for logging.
+func (sess *Session) Topic() string {
+	return sess.topic
+}
+
+// ServeHTTP upgrades the HTTP request into an SSE connection for this session.
+func (sess *Session) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	ctx := context.WithValue(r.Context(), internalTopicKey, sess.topic) //nolint:staticcheck
+	sess.server.inner.ServeHTTP(w, r.WithContext(ctx))
+}
+
+// Publish sends an event of the given type with the provided data to this session's clients.
+func (sess *Session) Publish(typ EventType, data string) error {
+	msg := &gosse.Message{}
+	msg.AppendData(data)
+	msg.Type = gosse.Type(string(typ))
+	return sess.server.inner.Publish(msg, sess.topic)
+}
+
+type contextKeyType struct{}
+
+var internalTopicKey = contextKeyType{}