fix(tui): harden gateway runtime adapter handshake and event delivery

Generated with [codeagent](https://github.com/qbox/codeagent)
Co-authored-by: pionxe <148670367+pionxe@users.noreply.github.com>

Author: xgopilot

internal/cli/gateway_runtime_bridge.go

@@ -414,15 +414,7 @@ func isRuntimeNotFoundError(err error) bool {
 	if err == nil {
 		return false
 	}
-	if errors.Is(err, os.ErrNotExist) {
-		return true
-	}
-	normalized := strings.ToLower(strings.TrimSpace(err.Error()))
-	return strings.Contains(normalized, "not found") ||
-		strings.Contains(normalized, "file does not exist") ||
-		strings.Contains(normalized, "does not exist") ||
-		strings.Contains(normalized, "no such file") ||
-		strings.Contains(normalized, "no such file or directory")
+	return errors.Is(err, agentsession.ErrSessionNotFound) || errors.Is(err, os.ErrNotExist)
 }
 
 var _ gateway.RuntimePort = (*gatewayRuntimePortBridge)(nil)

internal/cli/gateway_runtime_bridge_test.go

@@ -351,7 +351,7 @@ func TestGatewayRuntimePortBridgeLoadSessionNotFoundBranches(t *testing.T) {
 	t.Parallel()
 
 	base := &runtimeStub{
-		loadErr: errors.New("session not found"),
+		loadErr: agentsession.ErrSessionNotFound,
 	}
 	bridgeWithoutCreator, err := newGatewayRuntimePortBridge(context.Background(), &runtimeWithoutCreator{base: base})
 	if err != nil {
@@ -367,7 +367,7 @@ func TestGatewayRuntimePortBridgeLoadSessionNotFoundBranches(t *testing.T) {
 	}
 
 	stub := &runtimeStub{
-		loadErr:   errors.New("file does not exist"),
+		loadErr:   os.ErrNotExist,
 		createErr: errors.New("create failed"),
 	}
 	bridgeWithCreator, err := newGatewayRuntimePortBridge(context.Background(), stub)
@@ -390,6 +390,12 @@ func TestIsRuntimeNotFoundErrorIncludesOSErrNotExist(t *testing.T) {
 	if !isRuntimeNotFoundError(os.ErrNotExist) {
 		t.Fatalf("os.ErrNotExist should be treated as runtime not found")
 	}
+	if !isRuntimeNotFoundError(agentsession.ErrSessionNotFound) {
+		t.Fatalf("ErrSessionNotFound should be treated as runtime not found")
+	}
+	if isRuntimeNotFoundError(errors.New("session not found")) {
+		t.Fatalf("plain string not-found error should not be treated as runtime not found")
+	}
 }
 
 func TestGatewayRuntimePortBridgeRuntimeMethodErrors(t *testing.T) {
@@ -431,7 +437,7 @@ func TestGatewayRuntimePortBridgeRuntimeMethodErrors(t *testing.T) {
 func TestGatewayRuntimePortBridgeLoadSessionUpsertWhenMissing(t *testing.T) {
 	now := time.Now()
 	stub := &runtimeStub{
-		loadErr: errors.New("file does not exist"),
+		loadErr: agentsession.ErrSessionNotFound,
 		createSession: agentsession.Session{
 			ID:        "session-new",
 			Title:     "New Session",
@@ -464,7 +470,7 @@ func TestGatewayRuntimePortBridgeLoadSessionUpsertWhenMissing(t *testing.T) {
 	}
 }
 
-func TestGatewayRuntimePortBridgeLoadSessionUpsertWhenMissingNoSuchFile(t *testing.T) {
+func TestGatewayRuntimePortBridgeLoadSessionNoUpsertOnPlainStringNotFoundError(t *testing.T) {
 	now := time.Now()
 	stub := &runtimeStub{
 		loadErr: errors.New("open sessions/session-new.json: no such file"),
@@ -482,18 +488,15 @@ func TestGatewayRuntimePortBridgeLoadSessionUpsertWhenMissingNoSuchFile(t *testi
 	}
 	defer bridge.Close()
 
-	session, err := bridge.LoadSession(context.Background(), gateway.LoadSessionInput{
+	_, err = bridge.LoadSession(context.Background(), gateway.LoadSessionInput{
 		SubjectID: testBridgeSubjectID,
 		SessionID: " session-new ",
 	})
-	if err != nil {
-		t.Fatalf("load_session upsert no such file: %v", err)
-	}
-	if stub.createID != "session-new" {
-		t.Fatalf("create id = %q, want %q", stub.createID, "session-new")
+	if err == nil || err.Error() != "open sessions/session-new.json: no such file" {
+		t.Fatalf("expected original string error passthrough, got %v", err)
 	}
-	if session.ID != "session-new" {
-		t.Fatalf("upsert session id = %q, want %q", session.ID, "session-new")
+	if stub.createID != "" {
+		t.Fatalf("create should not be called for plain string error, got createID=%q", stub.createID)
 	}
 }
 

internal/tui/services/gateway_rpc_client.go

@@ -5,7 +5,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"log"
 	"net"
 	"strings"
 	"sync"
@@ -413,14 +412,12 @@ func (c *GatewayRPCClient) startNotificationDispatcher() {
 	})
 }
 
-// enqueueNotification 以无阻塞方式投递通知；队列满时丢弃并记录告警，避免反压读循环。
+// enqueueNotification 以阻塞方式投递通知，确保 gateway.event 不会因队列满被静默丢弃。
 func (c *GatewayRPCClient) enqueueNotification(notification gatewayRPCNotification) {
 	select {
 	case <-c.closed:
 		return
 	case c.notificationQueue <- notification:
-	default:
-		log.Printf("gateway rpc client: drop notification method=%q reason=queue_full", notification.Method)
 	}
 }
 

internal/tui/services/gateway_rpc_client_additional_test.go

@@ -9,6 +9,7 @@ import (
 	"net"
 	"os"
 	"path/filepath"
+	"strconv"
 	"strings"
 	"sync"
 	"testing"
@@ -456,6 +457,59 @@ func TestGatewayRPCClientNotificationDispatcherStopsOnQueueClose(t *testing.T) {
 	client.notificationWG.Wait()
 }
 
+func TestGatewayRPCClientEnqueueNotificationDoesNotDropUnderQueuePressure(t *testing.T) {
+	t.Parallel()
+
+	const total = 256
+	client := &GatewayRPCClient{
+		closed:            make(chan struct{}),
+		pending:           make(map[string]chan gatewayRPCResponse),
+		notifications:     make(chan gatewayRPCNotification, 1),
+		notificationQueue: make(chan gatewayRPCNotification, 1),
+	}
+	client.startNotificationDispatcher()
+	t.Cleanup(func() { _ = client.Close() })
+
+	receivedCh := make(chan struct{}, total)
+	go func() {
+		for range client.Notifications() {
+			receivedCh <- struct{}{}
+		}
+	}()
+
+	var enqueueWG sync.WaitGroup
+	for i := 0; i < total; i++ {
+		enqueueWG.Add(1)
+		go func(index int) {
+			defer enqueueWG.Done()
+			client.enqueueNotification(gatewayRPCNotification{
+				Method: protocol.MethodGatewayEvent,
+				Params: json.RawMessage(`{"index":` + strconv.Itoa(index) + `}`),
+			})
+		}(i)
+	}
+
+	waitDone := make(chan struct{})
+	go func() {
+		enqueueWG.Wait()
+		close(waitDone)
+	}()
+
+	select {
+	case <-waitDone:
+	case <-time.After(5 * time.Second):
+		t.Fatalf("enqueue notifications timed out under queue pressure")
+	}
+
+	for i := 0; i < total; i++ {
+		select {
+		case <-receivedCh:
+		case <-time.After(5 * time.Second):
+			t.Fatalf("expected %d notifications, got %d", total, i)
+		}
+	}
+}
+
 func TestGatewayRPCClientWriteRequestFailure(t *testing.T) {
 	t.Parallel()
 

internal/tui/services/gateway_rpc_client_test.go

@@ -219,7 +219,7 @@ func TestGatewayRPCClientCallWithEmptyMethodReturnsError(t *testing.T) {
 	}
 }
 
-func TestGatewayRPCClientReadLoopDoesNotBlockOnNotifications(t *testing.T) {
+func TestGatewayRPCClientReadLoopSustainsBackpressureWhenNotificationsAreConsumed(t *testing.T) {
 	tokenFile, _ := createTestAuthTokenFile(t)
 
 	client, err := NewGatewayRPCClient(GatewayRPCClientOptions{
@@ -257,6 +257,11 @@ func TestGatewayRPCClientReadLoopDoesNotBlockOnNotifications(t *testing.T) {
 	}
 	t.Cleanup(func() { _ = client.Close() })
 
+	go func() {
+		for range client.Notifications() {
+		}
+	}()
+
 	callErr := client.CallWithOptions(
 		context.Background(),
 		protocol.MethodGatewayPing,

internal/tui/services/remote_runtime_adapter.go

@@ -71,19 +71,20 @@ func NewRemoteRuntimeAdapter(options RemoteRuntimeAdapterOptions) (*RemoteRuntim
 	if timeout <= 0 {
 		timeout = defaultRemoteRuntimeTimeout
 	}
+	retryCount := normalizeRemoteRuntimeRetryCount(options.RetryCount)
 
 	rpcClient, err := newGatewayRPCClientFactory(GatewayRPCClientOptions{
 		ListenAddress:  strings.TrimSpace(options.ListenAddress),
 		TokenFile:      strings.TrimSpace(options.TokenFile),
 		RequestTimeout: timeout,
-		RetryCount:     options.RetryCount,
+		RetryCount:     retryCount,
 	})
 	if err != nil {
 		return nil, err
 	}
 
 	streamClient := newGatewayStreamClientFactory(rpcClient.Notifications())
-	adapter := newRemoteRuntimeAdapterWithClients(rpcClient, streamClient, timeout, options.RetryCount)
+	adapter := newRemoteRuntimeAdapterWithClients(rpcClient, streamClient, timeout, retryCount)
 
 	ctx, cancel := context.WithTimeout(context.Background(), timeout)
 	defer cancel()
@@ -100,6 +101,7 @@ func newRemoteRuntimeAdapterWithClients(
 	timeout time.Duration,
 	retryCount int,
 ) *RemoteRuntimeAdapter {
+	retryCount = normalizeRemoteRuntimeRetryCount(retryCount)
 	adapter := &RemoteRuntimeAdapter{
 		rpcClient:    rpcClient,
 		streamClient: streamClient,
@@ -113,7 +115,7 @@ func newRemoteRuntimeAdapterWithClients(
 	return adapter
 }
 
-// Submit 将用户输入提交到网关：先 authenticate，再 loadSession 预热，随后 bindStream，最后 run。
+// Submit 将用户输入提交到网关：先 authenticate，再 bindStream，随后 loadSession，最后 run。
 func (r *RemoteRuntimeAdapter) Submit(ctx context.Context, input agentruntime.PrepareInput) error {
 	sessionID := strings.TrimSpace(input.SessionID)
 	if sessionID == "" {
@@ -127,10 +129,10 @@ func (r *RemoteRuntimeAdapter) Submit(ctx context.Context, input agentruntime.Pr
 	if err := r.authenticate(ctx); err != nil {
 		return err
 	}
-	if err := r.preloadSession(ctx, sessionID); err != nil {
+	if err := r.bindStream(ctx, sessionID, runID); err != nil {
 		return err
 	}
-	if err := r.bindStream(ctx, sessionID, runID); err != nil {
+	if err := r.preloadSession(ctx, sessionID); err != nil {
 		return err
 	}
 
@@ -495,11 +497,23 @@ func (r *RemoteRuntimeAdapter) setActiveRun(runID string, sessionID string) {
 func (r *RemoteRuntimeAdapter) clearActiveRun(runID string) {
 	r.activeMu.Lock()
 	defer r.activeMu.Unlock()
-	if strings.TrimSpace(runID) == "" || strings.EqualFold(strings.TrimSpace(runID), strings.TrimSpace(r.activeRunID)) {
+	normalizedRunID := strings.TrimSpace(runID)
+	if normalizedRunID == "" {
+		return
+	}
+	if strings.EqualFold(normalizedRunID, strings.TrimSpace(r.activeRunID)) {
 		r.activeRunID = ""
 	}
 }
 
+// normalizeRemoteRuntimeRetryCount 统一归一化重试次数，避免零值关闭默认重试兜底。
+func normalizeRemoteRuntimeRetryCount(retryCount int) int {
+	if retryCount <= 0 {
+		return defaultGatewayRPCRetryCount
+	}
+	return retryCount
+}
+
 func (r *RemoteRuntimeAdapter) activeRun() (string, string) {
 	r.activeMu.Lock()
 	defer r.activeMu.Unlock()

internal/tui/services/remote_runtime_adapter_additional_test.go

@@ -72,13 +72,17 @@ func TestNewRemoteRuntimeAdapterBranches(t *testing.T) {
 	adapter, err := NewRemoteRuntimeAdapter(RemoteRuntimeAdapterOptions{
 		ListenAddress:  "ok",
 		RequestTimeout: -1,
+		RetryCount:     0,
 	})
 	if err != nil {
 		t.Fatalf("NewRemoteRuntimeAdapter() error = %v", err)
 	}
 	if adapter.timeout != defaultRemoteRuntimeTimeout {
 		t.Fatalf("timeout = %v, want %v", adapter.timeout, defaultRemoteRuntimeTimeout)
 	}
+	if adapter.retryCount != defaultGatewayRPCRetryCount {
+		t.Fatalf("retryCount = %d, want %d", adapter.retryCount, defaultGatewayRPCRetryCount)
+	}
 	_ = adapter.Close()
 }
 
@@ -304,6 +308,62 @@ func TestRemoteRuntimeAdapterEventObservationAndActiveRunState(t *testing.T) {
 	if runID != "" {
 		t.Fatalf("expected cleared run id, got %q", runID)
 	}
+
+	adapter.setActiveRun("run-c", "session-c")
+	adapter.observeEvent(agentruntime.RuntimeEvent{Type: agentruntime.EventError})
+	runID, sessionID = adapter.activeRun()
+	if runID != "run-c" || sessionID != "session-c" {
+		t.Fatalf("event error without run id should not clear active run, got run=%q session=%q", runID, sessionID)
+	}
+}
+
+func TestNewRemoteRuntimeAdapterWithClientsNormalizesRetryCount(t *testing.T) {
+	t.Parallel()
+
+	adapter := newRemoteRuntimeAdapterWithClients(
+		&stubRemoteRPCClient{notifications: make(chan gatewayRPCNotification)},
+		&stubRemoteStreamClient{events: make(chan agentruntime.RuntimeEvent)},
+		time.Second,
+		0,
+	)
+	t.Cleanup(func() { _ = adapter.Close() })
+
+	if adapter.retryCount != defaultGatewayRPCRetryCount {
+		t.Fatalf("retryCount = %d, want %d", adapter.retryCount, defaultGatewayRPCRetryCount)
+	}
+}
+
+func TestRemoteRuntimeAdapterUsesDefaultRetryWhenOptionsZero(t *testing.T) {
+	t.Parallel()
+
+	rpcClient := &stubRemoteRPCClient{
+		frames: map[string]gateway.MessageFrame{
+			protocol.MethodGatewayListSessions: {
+				Type:    gateway.FrameTypeAck,
+				Action:  gateway.FrameActionListSessions,
+				Payload: map[string]any{"sessions": []gateway.SessionSummary{}},
+			},
+		},
+		notifications: make(chan gatewayRPCNotification),
+	}
+	adapter := newRemoteRuntimeAdapterWithClients(
+		rpcClient,
+		&stubRemoteStreamClient{events: make(chan agentruntime.RuntimeEvent)},
+		time.Second,
+		0,
+	)
+	t.Cleanup(func() { _ = adapter.Close() })
+
+	if _, err := adapter.ListSessions(context.Background()); err != nil {
+		t.Fatalf("ListSessions() error = %v", err)
+	}
+	options, ok := rpcClient.snapshotOptions()[protocol.MethodGatewayListSessions]
+	if !ok {
+		t.Fatalf("expected listSessions call options to be captured")
+	}
+	if options.Retries != defaultGatewayRPCRetryCount {
+		t.Fatalf("listSessions retries = %d, want %d", options.Retries, defaultGatewayRPCRetryCount)
+	}
 }
 
 func TestRemoteRuntimeAdapterLoadSessionAndCancelErrorPaths(t *testing.T) {
@@ -347,9 +407,13 @@ func TestRemoteRuntimeAdapterSubmitAndCompactErrorPaths(t *testing.T) {
 	if err := adapter.Submit(context.Background(), agentruntime.PrepareInput{}); err == nil || !strings.Contains(err.Error(), "bind failed") {
 		t.Fatalf("expected bind failed submit error, got %v", err)
 	}
-	params := rpcClient.snapshotParams()[protocol.MethodGatewayLoadSession].(protocol.LoadSessionParams)
-	if strings.TrimSpace(params.SessionID) == "" {
-		t.Fatalf("Submit() should generate default session id")
+	methods := rpcClient.snapshotMethods()
+	if len(methods) != 1 || methods[0] != protocol.MethodGatewayBindStream {
+		t.Fatalf("Submit() should fail after bindStream and before loadSession, methods=%#v", methods)
+	}
+	bindParams, ok := rpcClient.snapshotParams()[protocol.MethodGatewayBindStream].(protocol.BindStreamParams)
+	if !ok || strings.TrimSpace(bindParams.SessionID) == "" {
+		t.Fatalf("Submit() should generate default session id for bindStream, params=%#v", rpcClient.snapshotParams()[protocol.MethodGatewayBindStream])
 	}
 
 	rpcClient.authErr = errors.New("auth failed")