fix(gateway): stop TunnelManager deadlocking on its own reservation (#2597)

Co-authored-by: Kandev Agent <agent@kandev.dev>
Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

apps/backend/internal/gateway/websocket/port_tunnel.go

@@ -2,6 +2,7 @@ package websocket
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"net"
 	"net/http"
@@ -24,7 +25,8 @@ type TunnelInfo struct {
 	TunnelPort int `json:"tunnel_port"`
 }
 
-// tunnelEntry tracks a running tunnel HTTP server.
+// tunnelEntry tracks a running tunnel HTTP server. Entries reach m.tunnels
+// only once they are complete, so every reader can dereference them.
 type tunnelEntry struct {
 	port       int
 	tunnelPort int
@@ -33,6 +35,25 @@ type tunnelEntry struct {
 	cancel     context.CancelFunc
 }
 
+// pendingTunnel is a StartTunnel that is still resolving and binding. It is
+// kept out of m.tunnels so that a concurrent caller waits for the in-flight
+// bind and shares its outcome instead of observing a half-built entry, and so
+// that StopTunnel, ListTunnels, InvalidateSession and Shutdown never see one.
+// tunnelPort and err are written before done is closed and read only after.
+type pendingTunnel struct {
+	done       chan struct{}
+	tunnelPort int
+	err        error
+}
+
+// errStartCanceled reports a start that was abandoned mid-bind because
+// InvalidateSession or Shutdown dropped its reservation.
+var errStartCanceled = errors.New("tunnel start canceled by session invalidation or shutdown")
+
+// errManagerClosed reports a start refused because Shutdown already ran.
+// Shutdown is terminal: nothing may bind a listener it will never close.
+var errManagerClosed = errors.New("tunnel manager is shut down")
+
 // TunnelManager manages per-session port tunnels that bind dedicated host
 // ports and reverse-proxy traffic to agentctl's port-proxy endpoint.
 // Unlike the path-based PortProxyHandler, tunnels serve apps at the root
@@ -42,7 +63,9 @@ type TunnelManager struct {
 	logger       *logger.Logger
 
 	mu      sync.Mutex
-	tunnels map[string]*tunnelEntry // key: "sessionId:port"
+	closed  bool                      // set by Shutdown; refuses every later start
+	tunnels map[string]*tunnelEntry   // key: "sessionId:port"; complete entries only
+	pending map[string]*pendingTunnel // key: "sessionId:port"; starts still in flight
 }
 
 // NewTunnelManager creates a new TunnelManager.
@@ -51,6 +74,7 @@ func NewTunnelManager(lifecycleMgr *lifecycle.Manager, log *logger.Logger) *Tunn
 		lifecycleMgr: lifecycleMgr,
 		logger:       log.WithFields(zap.String("component", "port-tunnel")),
 		tunnels:      make(map[string]*tunnelEntry),
+		pending:      make(map[string]*pendingTunnel),
 	}
 }
 
@@ -60,20 +84,16 @@ func NewTunnelManager(lifecycleMgr *lifecycle.Manager, log *logger.Logger) *Tunn
 func (m *TunnelManager) StartTunnel(sessionID string, port int, tunnelPort int) (int, error) {
 	cacheKey := sessionID + ":" + strconv.Itoa(port)
 
-	// Check for existing tunnel under lock to prevent duplicate creation.
-	m.mu.Lock()
-	if entry, ok := m.tunnels[cacheKey]; ok {
-		tp := entry.tunnelPort
-		m.mu.Unlock()
-		return tp, nil
+	started, existingPort, err := m.claimStart(cacheKey)
+	if started == nil {
+		// Another caller already owns this session:port, either as a running
+		// tunnel or as an in-flight start we just waited for.
+		return existingPort, err
 	}
-	// Reserve the key with a nil entry to prevent concurrent creation.
-	m.tunnels[cacheKey] = nil
-	m.mu.Unlock()
 
-	target, authToken, ln, err := m.resolveAndBind(cacheKey, sessionID, tunnelPort)
+	target, authToken, ln, err := m.resolveAndBind(sessionID, tunnelPort)
 	if err != nil {
-		return 0, err
+		return m.finishStart(cacheKey, started, 0, err)
 	}
 	actualPort := ln.Addr().(*net.TCPAddr).Port
 
@@ -89,9 +109,14 @@ func (m *TunnelManager) StartTunnel(sessionID string, port int, tunnelPort int)
 		cancel:     cancel,
 	}
 
-	m.mu.Lock()
-	m.tunnels[cacheKey] = entry
-	m.mu.Unlock()
+	if !m.publish(cacheKey, started, entry) {
+		cancel()
+		_ = ln.Close()
+		m.logger.Info("tunnel start canceled while binding",
+			zap.String("session_id", sessionID),
+			zap.Int("port", port))
+		return m.finishStart(cacheKey, started, 0, errStartCanceled)
+	}
 
 	m.serveTunnel(ctx, srv, ln, cacheKey, sessionID, port, actualPort)
 
@@ -100,43 +125,94 @@ func (m *TunnelManager) StartTunnel(sessionID string, port int, tunnelPort int)
 		zap.Int("port", port),
 		zap.Int("tunnel_port", actualPort))
 
-	return actualPort, nil
+	return m.finishStart(cacheKey, started, actualPort, nil)
+}
+
+// claimStart reserves cacheKey for the calling StartTunnel. It returns a
+// non-nil pendingTunnel when the caller owns the start; otherwise it returns
+// the result of the tunnel that already exists or was being built
+// concurrently, having waited for the latter to settle.
+func (m *TunnelManager) claimStart(cacheKey string) (*pendingTunnel, int, error) {
+	m.mu.Lock()
+	closed := m.closed
+	entry, running := m.tunnels[cacheKey]
+	inFlight, starting := m.pending[cacheKey]
+	var started *pendingTunnel
+	if !closed && !running && !starting {
+		started = &pendingTunnel{done: make(chan struct{})}
+		m.pending[cacheKey] = started
+	}
+	m.mu.Unlock()
+
+	switch {
+	case closed:
+		return nil, 0, errManagerClosed
+	case running:
+		return nil, entry.tunnelPort, nil
+	case starting:
+		<-inFlight.done // Never while holding m.mu: the owner needs it to finish.
+		return nil, inFlight.tunnelPort, inFlight.err
+	default:
+		return started, 0, nil
+	}
+}
+
+// publish installs a finished tunnel under cacheKey, before the reservation is
+// released, so a caller arriving in between finds the tunnel rather than
+// starting a second one. It reports false when the reservation is no longer
+// registered, which means InvalidateSession or Shutdown tore this session down
+// while the bind was in flight: the caller must then abandon its listener
+// instead of installing a tunnel nobody will ever close.
+func (m *TunnelManager) publish(cacheKey string, started *pendingTunnel, entry *tunnelEntry) bool {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	if m.pending[cacheKey] != started {
+		return false
+	}
+	m.tunnels[cacheKey] = entry
+	return true
+}
+
+// finishStart releases the reservation taken by claimStart and hands its
+// outcome to every caller waiting on the same session:port. The identity check
+// matters: once a teardown has canceled this reservation another start may own
+// the key, and that newcomer's reservation must survive.
+func (m *TunnelManager) finishStart(cacheKey string, started *pendingTunnel, tunnelPort int, err error) (int, error) {
+	started.tunnelPort = tunnelPort
+	started.err = err
+
+	m.mu.Lock()
+	if m.pending[cacheKey] == started {
+		delete(m.pending, cacheKey)
+	}
+	m.mu.Unlock()
+
+	close(started.done)
+	return tunnelPort, err
 }
 
 // resolveAndBind resolves the agentctl target URL for the session and binds a
-// local TCP listener for the tunnel. On failure it cleans up the placeholder
-// entry reserved by StartTunnel.
-func (m *TunnelManager) resolveAndBind(cacheKey, sessionID string, tunnelPort int) (*url.URL, string, net.Listener, error) {
-	cleanup := func() {
-		m.mu.Lock()
-		if m.tunnels[cacheKey] == nil {
-			delete(m.tunnels, cacheKey)
-		}
-		m.mu.Unlock()
-	}
-
+// local TCP listener for the tunnel.
+func (m *TunnelManager) resolveAndBind(sessionID string, tunnelPort int) (*url.URL, string, net.Listener, error) {
 	execution, ok := m.lifecycleMgr.GetExecutionBySessionID(sessionID)
 	if !ok {
-		cleanup()
 		return nil, "", nil, fmt.Errorf("session not found or no active execution")
 	}
 
 	agentctlClient := execution.GetAgentCtlClient()
 	if agentctlClient == nil {
-		cleanup()
 		return nil, "", nil, fmt.Errorf("agentctl client not available")
 	}
 
 	target, err := url.Parse(agentctlClient.BaseURL())
 	if err != nil {
-		cleanup()
 		return nil, "", nil, fmt.Errorf("failed to parse agentctl URL: %w", err)
 	}
 	authToken := agentctlClient.AuthToken()
 
 	ln, err := net.Listen("tcp", fmt.Sprintf(":%d", tunnelPort))
 	if err != nil {
-		cleanup()
 		if netutil.IsAddrInUse(err) {
 			return nil, "", nil, fmt.Errorf("port %d is already in use, choose a different port", tunnelPort)
 		}
@@ -219,6 +295,7 @@ func (m *TunnelManager) InvalidateSession(sessionID string) {
 			delete(m.tunnels, key)
 		}
 	}
+	canceled := m.cancelPendingStarts(prefix)
 	m.mu.Unlock()
 
 	for _, entry := range toStop {
@@ -226,10 +303,11 @@ func (m *TunnelManager) InvalidateSession(sessionID string) {
 		_ = entry.ln.Close()
 	}
 
-	if len(toStop) > 0 {
+	if len(toStop) > 0 || canceled > 0 {
 		m.logger.Info("invalidated session tunnels",
 			zap.String("session_id", sessionID),
-			zap.Int("count", len(toStop)))
+			zap.Int("count", len(toStop)),
+			zap.Int("canceled_starts", canceled))
 	}
 }
 
@@ -241,6 +319,8 @@ func (m *TunnelManager) Shutdown() {
 		entries = append(entries, entry)
 	}
 	m.tunnels = make(map[string]*tunnelEntry)
+	canceled := m.cancelPendingStarts("")
+	m.closed = true
 	m.mu.Unlock()
 
 	for _, entry := range entries {
@@ -248,11 +328,29 @@ func (m *TunnelManager) Shutdown() {
 		_ = entry.ln.Close()
 	}
 
-	if len(entries) > 0 {
-		m.logger.Info("shutdown all tunnels", zap.Int("count", len(entries)))
+	if len(entries) > 0 || canceled > 0 {
+		m.logger.Info("shutdown all tunnels",
+			zap.Int("count", len(entries)),
+			zap.Int("canceled_starts", canceled))
 	}
 }
 
+// cancelPendingStarts drops every reservation whose key has the given prefix
+// ("" for all of them) and reports how many were dropped. A start whose
+// reservation is gone abandons its listener in publish rather than installing a
+// tunnel into a session, or a manager, that has already been torn down.
+// The caller must hold m.mu.
+func (m *TunnelManager) cancelPendingStarts(prefix string) int {
+	canceled := 0
+	for key := range m.pending {
+		if strings.HasPrefix(key, prefix) {
+			delete(m.pending, key)
+			canceled++
+		}
+	}
+	return canceled
+}
+
 func (m *TunnelManager) removeTunnel(cacheKey string) {
 	m.mu.Lock()
 	defer m.mu.Unlock()

apps/backend/internal/gateway/websocket/port_tunnel_concurrency_test.go

@@ -0,0 +1,223 @@
+package websocket
+
+import (
+	"net/http"
+	"testing"
+	"time"
+
+	"github.com/kandev/kandev/internal/agent/runtime/lifecycle"
+	"github.com/kandev/kandev/internal/common/logger"
+)
+
+// startOutcome is one concurrent StartTunnel result. A panic in a racing
+// goroutine is recovered and reported here so it fails the test that caused it
+// instead of taking the whole test binary down with it.
+type startOutcome struct {
+	port      int
+	err       error
+	recovered any
+}
+
+// startTunnelRacing runs StartTunnel in its own goroutine, releasing gate first
+// so every caller enters the reservation window together.
+func startTunnelRacing(
+	t *testing.T,
+	manager *TunnelManager,
+	gate <-chan struct{},
+	sessionID string,
+	port int,
+) <-chan startOutcome {
+	t.Helper()
+	out := make(chan startOutcome, 1)
+	go func() {
+		outcome := startOutcome{}
+		defer func() {
+			outcome.recovered = recover()
+			out <- outcome
+		}()
+		<-gate
+		outcome.port, outcome.err = manager.StartTunnel(sessionID, port, 0)
+	}()
+	return out
+}
+
+// awaitStart collects one racing StartTunnel result under a bound. A leaked
+// m.mu blocks the caller forever, so an unbounded receive here would hang the
+// suite instead of reporting the defect.
+func awaitStart(t *testing.T, what string, out <-chan startOutcome) startOutcome {
+	t.Helper()
+	timer := time.NewTimer(wsTestTimeout)
+	defer timer.Stop()
+	select {
+	case outcome := <-out:
+		if outcome.recovered != nil {
+			t.Errorf("%s panicked: %v", what, outcome.recovered)
+		}
+		return outcome
+	case <-timer.C:
+		t.Fatalf("%s did not return within %s: TunnelManager.mu is held by a goroutine that never released it", what, wsTestTimeout)
+		return startOutcome{}
+	}
+}
+
+// mustNotBlock fails if fn has not returned within wsTestTimeout. Every
+// TunnelManager method takes m.mu, so a lock leaked by a panicking StartTunnel
+// shows up here as a hang rather than as an error.
+func mustNotBlock(t *testing.T, what string, fn func()) {
+	t.Helper()
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		fn()
+	}()
+	timer := time.NewTimer(wsTestTimeout)
+	defer timer.Stop()
+	select {
+	case <-done:
+	case <-timer.C:
+		t.Fatalf("%s blocked for %s: TunnelManager.mu was never released", what, wsTestTimeout)
+	}
+}
+
+// newTunnelRaceBackend returns a lifecycle manager whose named sessions all
+// resolve to one fake agentctl, so tunnel starts against them succeed.
+func newTunnelRaceBackend(t *testing.T, log *logger.Logger, sessionIDs ...string) *lifecycle.Manager {
+	t.Helper()
+	upstream := newFakeAgentctl(t, func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusOK)
+	})
+	lifecycleMgr := newProxyTestManager(t, log)
+	for _, sessionID := range sessionIDs {
+		addExecutionForURL(t, lifecycleMgr, sessionID, upstream.server.URL, "", log)
+	}
+	return lifecycleMgr
+}
+
+// newBoundedTunnelManager returns a manager whose cleanup cannot hang the
+// suite: Shutdown takes m.mu, so a leaked lock would block teardown forever.
+func newBoundedTunnelManager(t *testing.T, lifecycleMgr *lifecycle.Manager, log *logger.Logger) *TunnelManager {
+	t.Helper()
+	manager := NewTunnelManager(lifecycleMgr, log)
+	t.Cleanup(func() { mustNotBlock(t, "Shutdown", manager.Shutdown) })
+	return manager
+}
+
+func newRacingTunnelManager(t *testing.T, sessionIDs ...string) *TunnelManager {
+	t.Helper()
+	log, _ := observedTerminalLogger(t)
+	return newBoundedTunnelManager(t, newTunnelRaceBackend(t, log, sessionIDs...), log)
+}
+
+// Two clicks on "expose port" arrive as two ports.tunnel.start actions, and
+// client.go dispatches each in its own goroutine. Both callers must end up
+// sharing a single tunnel instead of observing the reservation the first one
+// left behind mid-bind.
+func TestStartTunnelIsSafeUnderConcurrentDuplicateStarts(t *testing.T) {
+	manager := newRacingTunnelManager(t, "sess-race")
+
+	gate := make(chan struct{})
+	first := startTunnelRacing(t, manager, gate, "sess-race", 3000)
+	second := startTunnelRacing(t, manager, gate, "sess-race", 3000)
+	close(gate)
+
+	firstOutcome := awaitStart(t, "first StartTunnel", first)
+	secondOutcome := awaitStart(t, "second StartTunnel", second)
+
+	for _, outcome := range []struct {
+		what string
+		got  startOutcome
+	}{{"first", firstOutcome}, {"second", secondOutcome}} {
+		if outcome.got.err != nil {
+			t.Fatalf("%s StartTunnel() error = %v, want nil", outcome.what, outcome.got.err)
+		}
+		if outcome.got.port <= 0 {
+			t.Fatalf("%s StartTunnel() port = %d, want an OS-assigned port", outcome.what, outcome.got.port)
+		}
+	}
+	if firstOutcome.port != secondOutcome.port {
+		t.Fatalf("concurrent starts returned different ports (%d and %d), want one shared tunnel",
+			firstOutcome.port, secondOutcome.port)
+	}
+
+	listed := tunnelList(t, manager, "sess-race")
+	if len(listed) != 1 {
+		t.Fatalf("tunnels = %v, want exactly 1 (a duplicate start must not bind a second listener)", listed)
+	}
+	if listed[0].TunnelPort != firstOutcome.port {
+		t.Fatalf("listed tunnel port = %d, want the started %d", listed[0].TunnelPort, firstOutcome.port)
+	}
+	if status, _ := getThroughTunnel(t, newTunnelClient(t), firstOutcome.port, "/"); status != http.StatusOK {
+		t.Fatalf("shared tunnel status = %d, want %d", status, http.StatusOK)
+	}
+}
+
+// The serious half of the defect: StartTunnel dereferenced the reservation
+// while holding m.mu behind a non-deferred Unlock, so a panic there stranded
+// the lock and every later start, stop, list and invalidate blocked for the
+// process lifetime.
+func TestTunnelManagerStaysUsableAfterConcurrentDuplicateStarts(t *testing.T) {
+	manager := newRacingTunnelManager(t, "sess-usable")
+
+	gate := make(chan struct{})
+	first := startTunnelRacing(t, manager, gate, "sess-usable", 3000)
+	second := startTunnelRacing(t, manager, gate, "sess-usable", 3000)
+	close(gate)
+	tunnelPort := awaitStart(t, "first StartTunnel", first).port
+	awaitStart(t, "second StartTunnel", second)
+
+	mustNotBlock(t, "ListTunnels after a duplicate start", func() {
+		_ = manager.ListTunnels("sess-usable")
+	})
+	mustNotBlock(t, "StartTunnel for another port after a duplicate start", func() {
+		if _, err := manager.StartTunnel("sess-usable", 4000, 0); err != nil {
+			t.Errorf("StartTunnel(4000) error = %v, want nil", err)
+		}
+	})
+	mustNotBlock(t, "InvalidateSession after a duplicate start", func() {
+		manager.InvalidateSession("sess-nobody")
+	})
+	mustNotBlock(t, "StopTunnel after a duplicate start", func() {
+		if err := manager.StopTunnel("sess-usable", 3000); err != nil {
+			t.Errorf("StopTunnel() error = %v, want nil", err)
+		}
+	})
+	assertPortClosed(t, tunnelPort)
+
+	mustNotBlock(t, "Shutdown after a duplicate start", manager.Shutdown)
+}
+
+// Every reservation must be released even when many callers pile onto the same
+// key and the start fails, or the session:port pair stays permanently unusable.
+func TestConcurrentFailingStartsReleaseTheReservation(t *testing.T) {
+	manager := newRacingTunnelManager(t)
+
+	const callers = 8
+	gate := make(chan struct{})
+	outs := make([]<-chan startOutcome, 0, callers)
+	for range callers {
+		outs = append(outs, startTunnelRacing(t, manager, gate, "sess-missing", 3000))
+	}
+	close(gate)
+
+	for i, out := range outs {
+		outcome := awaitStart(t, "failing StartTunnel", out)
+		if outcome.err == nil {
+			t.Fatalf("caller %d: StartTunnel() error = nil, want a failure", i)
+		}
+		if outcome.port != 0 {
+			t.Fatalf("caller %d: StartTunnel() port = %d, want 0 on failure", i, outcome.port)
+		}
+	}
+
+	// The key is usable again: nothing was left reserved by the failed round.
+	log, _ := observedTerminalLogger(t)
+	upstream := newFakeAgentctl(t, func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusOK)
+	})
+	addExecutionForURL(t, manager.lifecycleMgr, "sess-missing", upstream.server.URL, "", log)
+	mustNotBlock(t, "StartTunnel after concurrent failures", func() {
+		if _, err := manager.StartTunnel("sess-missing", 3000, 0); err != nil {
+			t.Errorf("StartTunnel after the failed round: %v", err)
+		}
+	})
+}

apps/backend/internal/gateway/websocket/port_tunnel_reservation_test.go

@@ -0,0 +1,142 @@
+package websocket
+
+import (
+	"errors"
+	"testing"
+	"time"
+)
+
+// reserveInFlightStart seeds the reservation StartTunnel takes while it
+// resolves and binds, and returns the settle function the owning start would
+// call. Binding is a real syscall of unbounded duration, so this window is wide
+// enough in production for every other method to run inside it.
+func reserveInFlightStart(t *testing.T, manager *TunnelManager, cacheKey string) (*pendingTunnel, func(port int, err error)) {
+	t.Helper()
+	started := &pendingTunnel{done: make(chan struct{})}
+	manager.mu.Lock()
+	manager.pending[cacheKey] = started
+	manager.mu.Unlock()
+
+	var settled bool
+	settle := func(port int, err error) {
+		if settled {
+			return
+		}
+		settled = true
+		_, _ = manager.finishStart(cacheKey, started, port, err)
+	}
+	// Registered immediately: a t.Fatal below must not strand callers that are
+	// already blocked on this reservation.
+	t.Cleanup(func() { settle(0, errors.New("test cleanup")) })
+	return started, settle
+}
+
+// A start that has not finished binding is invisible to every other method:
+// none of them can observe a half-built tunnel, and none of them blocks on the
+// in-flight bind.
+func TestInFlightStartIsInvisibleToTheOtherReaders(t *testing.T) {
+	manager := newRacingTunnelManager(t)
+	_, settle := reserveInFlightStart(t, manager, "sess-inflight:3000")
+
+	mustNotBlock(t, "ListTunnels during an in-flight start", func() {
+		if listed := tunnelList(t, manager, "sess-inflight"); len(listed) != 0 {
+			t.Errorf("ListTunnels() = %v during an in-flight start, want none", listed)
+		}
+	})
+	mustNotBlock(t, "StopTunnel during an in-flight start", func() {
+		if err := manager.StopTunnel("sess-inflight", 3000); err == nil {
+			t.Error("StopTunnel() error = nil during an in-flight start, want a not-found error")
+		}
+	})
+	mustNotBlock(t, "InvalidateSession for another session during an in-flight start", func() {
+		manager.InvalidateSession("sess-unrelated")
+	})
+	// A start for a different session:port is not blocked by this reservation.
+	mustNotBlock(t, "StartTunnel for another key during an in-flight start", func() {
+		if _, err := manager.StartTunnel("sess-other", 3000, 0); err == nil {
+			t.Error("StartTunnel(sess-other) error = nil, want the missing-session failure")
+		}
+	})
+	// Last, because Shutdown is terminal.
+	mustNotBlock(t, "Shutdown during an in-flight start", manager.Shutdown)
+
+	settle(41234, nil)
+}
+
+// A caller that loses the reservation race waits for the in-flight bind and
+// shares its result, rather than binding a second listener or reading a
+// half-built entry.
+func TestDuplicateStartWaitsForTheInFlightResult(t *testing.T) {
+	tests := []struct {
+		name      string
+		port      int
+		err       error
+		wantPort  int
+		wantError string
+	}{
+		{name: "in-flight start succeeds", port: 41234, wantPort: 41234},
+		{name: "in-flight start fails", err: errors.New("bind exploded"), wantError: "bind exploded"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			manager := newRacingTunnelManager(t)
+			_, settle := reserveInFlightStart(t, manager, "sess-wait:3000")
+
+			gate := make(chan struct{})
+			close(gate)
+			waiter := startTunnelRacing(t, manager, gate, "sess-wait", 3000)
+
+			// The waiter must be parked without holding m.mu, so an unrelated
+			// reader still runs while it waits.
+			mustNotBlock(t, "ListTunnels while a duplicate start waits", func() {
+				_ = manager.ListTunnels("sess-wait")
+			})
+			// ListTunnels above is the m.mu liveness proof. This select is a
+			// secondary sanity-check that the waiter has not returned early;
+			// 50ms is ample for an in-process channel wait to park.
+			select {
+			case outcome := <-waiter:
+				t.Fatalf("duplicate StartTunnel returned %+v before the in-flight start settled", outcome)
+			case <-time.After(50 * time.Millisecond):
+			}
+
+			settle(tt.port, tt.err)
+
+			outcome := awaitStart(t, "duplicate StartTunnel", waiter)
+			if outcome.port != tt.wantPort {
+				t.Fatalf("duplicate StartTunnel() port = %d, want the in-flight %d", outcome.port, tt.wantPort)
+			}
+			if tt.wantError == "" {
+				if outcome.err != nil {
+					t.Fatalf("duplicate StartTunnel() error = %v, want nil", outcome.err)
+				}
+				return
+			}
+			if outcome.err == nil || outcome.err.Error() != tt.wantError {
+				t.Fatalf("duplicate StartTunnel() error = %v, want %q", outcome.err, tt.wantError)
+			}
+		})
+	}
+}
+
+// The reservation is released even when the start fails, so the same
+// session:port can be started again.
+func TestFinishStartReleasesTheReservation(t *testing.T) {
+	manager := newRacingTunnelManager(t)
+	_, settle := reserveInFlightStart(t, manager, "sess-release:3000")
+	settle(0, errors.New("bind exploded"))
+
+	manager.mu.Lock()
+	_, stillReserved := manager.pending["sess-release:3000"]
+	manager.mu.Unlock()
+	if stillReserved {
+		t.Fatal("finishStart left the reservation in place")
+	}
+
+	// A later start owns the key again instead of waiting on a dead reservation.
+	mustNotBlock(t, "StartTunnel after a released reservation", func() {
+		if _, err := manager.StartTunnel("sess-release", 3000, 0); err == nil {
+			t.Error("StartTunnel() error = nil, want the missing-session failure")
+		}
+	})
+}

apps/backend/internal/gateway/websocket/port_tunnel_serve_test.go

@@ -180,8 +180,12 @@ func TestStartTunnelReleasesReservationWhenResolutionFails(t *testing.T) {
 				t.Fatalf("StartTunnel() port = %d, want 0 on failure", port)
 			}
 			manager.mu.Lock()
-			_, stillReserved := manager.tunnels[tt.sessionID+":3000"]
+			_, stillRunning := manager.tunnels[tt.sessionID+":3000"]
+			_, stillReserved := manager.pending[tt.sessionID+":3000"]
 			manager.mu.Unlock()
+			if stillRunning {
+				t.Fatal("a failed start left an entry in the tunnel map")
+			}
 			if stillReserved {
 				t.Fatal("the reserved cache key was not released after the failure")
 			}

apps/backend/internal/gateway/websocket/port_tunnel_teardown_test.go

@@ -0,0 +1,181 @@
+package websocket
+
+import (
+	"errors"
+	"net/http"
+	"testing"
+)
+
+// startTunnelRacingOnPort is startTunnelRacing for a caller that needs to know
+// the tunnel port up front, so the port can be checked whichever way the race
+// went.
+func startTunnelRacingOnPort(
+	t *testing.T,
+	manager *TunnelManager,
+	gate <-chan struct{},
+	sessionID string,
+	port, tunnelPort int,
+) <-chan startOutcome {
+	t.Helper()
+	out := make(chan startOutcome, 1)
+	go func() {
+		outcome := startOutcome{}
+		defer func() {
+			outcome.recovered = recover()
+			out <- outcome
+		}()
+		<-gate
+		outcome.port, outcome.err = manager.StartTunnel(sessionID, port, tunnelPort)
+	}()
+	return out
+}
+
+// A start still binding when its session or the whole manager is torn down must
+// be canceled. Its reservation is the cancellation token: once the teardown has
+// dropped it, publish refuses to install the tunnel, so no listener survives a
+// teardown that could not see it.
+func TestTeardownCancelsInFlightStarts(t *testing.T) {
+	tests := []struct {
+		name     string
+		teardown func(*TunnelManager)
+	}{
+		{name: "Shutdown", teardown: func(m *TunnelManager) { m.Shutdown() }},
+		{name: "InvalidateSession", teardown: func(m *TunnelManager) { m.InvalidateSession("sess-teardown") }},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			manager := newRacingTunnelManager(t)
+			started, settle := reserveInFlightStart(t, manager, "sess-teardown:3000")
+
+			tt.teardown(manager)
+
+			manager.mu.Lock()
+			_, stillReserved := manager.pending["sess-teardown:3000"]
+			manager.mu.Unlock()
+			if stillReserved {
+				t.Fatalf("%s left the in-flight reservation in place", tt.name)
+			}
+
+			// The owning start now finishes its bind and must abandon it.
+			if manager.publish("sess-teardown:3000", started, &tunnelEntry{port: 3000, tunnelPort: 41000}) {
+				t.Fatalf("publish installed a tunnel after %s canceled its reservation", tt.name)
+			}
+			manager.mu.Lock()
+			installed := len(manager.tunnels)
+			manager.mu.Unlock()
+			if installed != 0 {
+				t.Fatalf("tunnels after a canceled start = %d, want 0", installed)
+			}
+
+			settle(0, errStartCanceled)
+		})
+	}
+}
+
+// A canceled reservation must not take a newer one down with it: after a
+// teardown, the next start owns the key, and the canceled start's finishStart
+// must leave that newcomer's reservation alone.
+func TestFinishStartKeepsANewerReservation(t *testing.T) {
+	manager := newRacingTunnelManager(t)
+	canceled, settleCanceled := reserveInFlightStart(t, manager, "sess-newer:3000")
+	manager.InvalidateSession("sess-newer")
+
+	newer := &pendingTunnel{done: make(chan struct{})}
+	manager.mu.Lock()
+	manager.pending["sess-newer:3000"] = newer
+	manager.mu.Unlock()
+	t.Cleanup(func() { _, _ = manager.finishStart("sess-newer:3000", newer, 0, errors.New("test cleanup")) })
+
+	settleCanceled(0, errStartCanceled)
+
+	manager.mu.Lock()
+	owner := manager.pending["sess-newer:3000"]
+	manager.mu.Unlock()
+	if owner != newer {
+		t.Fatalf("the canceled start's finishStart dropped the newer reservation (owner = %p, want %p)", owner, newer)
+	}
+	if manager.publish("sess-newer:3000", canceled, &tunnelEntry{port: 3000}) {
+		t.Fatal("publish installed a tunnel for the canceled start while a newer one owned the key")
+	}
+}
+
+// End-to-end property, whatever the interleaving: a StartTunnel racing Shutdown
+// either publishes a tunnel that Shutdown then closes, or is canceled and
+// closes its own listener. Nothing may still be listening afterwards, and the
+// manager must be left empty.
+func TestStartTunnelRacingShutdownLeavesNoListener(t *testing.T) {
+	log, _ := observedTerminalLogger(t)
+	backend := newTunnelRaceBackend(t, log, "sess-shutdown-race")
+
+	const attempts = 20
+	published, canceled, refused, unbound := 0, 0, 0, 0
+	for range attempts {
+		manager := newBoundedTunnelManager(t, backend, log)
+		wanted := reserveLoopbackPort(t)
+
+		gate := make(chan struct{})
+		start := startTunnelRacingOnPort(t, manager, gate, "sess-shutdown-race", 3000, wanted)
+		shutdownDone := make(chan struct{})
+		go func() {
+			defer close(shutdownDone)
+			<-gate
+			manager.Shutdown()
+		}()
+		close(gate)
+
+		outcome := awaitStart(t, "StartTunnel racing Shutdown", start)
+		mustNotBlock(t, "Shutdown racing StartTunnel", func() { <-shutdownDone })
+
+		switch {
+		case outcome.err == nil:
+			published++
+			assertPortClosed(t, wanted) // Shutdown drained the published entry.
+		case errors.Is(outcome.err, errStartCanceled):
+			canceled++
+			assertPortClosed(t, wanted) // The canceled start closed its own listener.
+		case errors.Is(outcome.err, errManagerClosed):
+			refused++
+			assertPortClosed(t, wanted) // Refused before binding anything.
+		default:
+			// The reserved port was taken between reserve and bind; another
+			// process owns it, so its state says nothing about this manager.
+			unbound++
+		}
+
+		manager.mu.Lock()
+		leftTunnels, leftPending := len(manager.tunnels), len(manager.pending)
+		manager.mu.Unlock()
+		if leftTunnels != 0 || leftPending != 0 {
+			t.Fatalf("after the race: tunnels = %d, pending = %d, want 0 and 0", leftTunnels, leftPending)
+		}
+	}
+	t.Logf("outcomes over %d attempts: published=%d canceled=%d refused=%d port-unavailable=%d",
+		attempts, published, canceled, refused, unbound)
+}
+
+// A tunnel published before a teardown is still closed by it; this pins the
+// ordering the property test above relies on.
+func TestShutdownClosesATunnelPublishedBeforeIt(t *testing.T) {
+	manager := newRacingTunnelManager(t, "sess-ordered")
+
+	tunnelPort, err := manager.StartTunnel("sess-ordered", 3000, 0)
+	if err != nil {
+		t.Fatalf("StartTunnel: %v", err)
+	}
+	if status, _ := getThroughTunnel(t, newTunnelClient(t), tunnelPort, "/"); status != http.StatusOK {
+		t.Fatalf("status before shutdown = %d, want %d", status, http.StatusOK)
+	}
+
+	manager.Shutdown()
+	assertPortClosed(t, tunnelPort)
+
+	// Shutdown is terminal. A start afterwards must be refused rather than bind
+	// a listener that nothing is left to close.
+	again, err := manager.StartTunnel("sess-ordered", 3000, 0)
+	if !errors.Is(err, errManagerClosed) {
+		t.Fatalf("StartTunnel after Shutdown error = %v, want %v", err, errManagerClosed)
+	}
+	if again != 0 {
+		t.Fatalf("StartTunnel after Shutdown port = %d, want 0", again)
+	}
+}

apps/backend/internal/gateway/websocket/port_tunnel_test.go

@@ -46,7 +46,7 @@ func TestResolveAndBindReportsOccupiedPort(t *testing.T) {
 	port := occupied.Addr().(*net.TCPAddr).Port
 
 	mgr := NewTunnelManager(lifecycleMgr, log)
-	_, _, _, err = mgr.resolveAndBind("session:1", "session", port)
+	_, _, _, err = mgr.resolveAndBind("session", port)
 	if err == nil {
 		t.Fatal("resolveAndBind() = nil, want occupied-port error")
 	}