Function runSession

internal/nodeagent/session.go:16–67

runSession 建立一次连接，发 hello 帧，进入收发循环。任何错误（包括正常 EOF） 都返回 error，由上层 Run 触发重连。

(ctx context.Context, cfg Config)

// runSession 建立一次连接，发 hello 帧，进入收发循环。任何错误（包括正常 EOF）
// 都返回 error，由上层 Run 触发重连。
func runSession(ctx context.Context, cfg Config) error {
	conn, err := cfg.Dialer(ctx)
	if err != nil {
		return fmt.Errorf("dial: %w", err)
	}
	defer conn.Close()

	// 整个 session 范围 ctx，stream 出错或 ctx 取消时统一收尾。
	sessionCtx, cancel := context.WithCancel(ctx)
	defer cancel()

	client := nodev1.NewNodeAgentClient(conn)
	stream, err := client.Session(sessionCtx)
	if err != nil {
		return fmt.Errorf("open session stream: %w", err)
	}

	s := newSession(sessionCtx, cancel, cfg, stream)

	// 1) hello 帧
	helloBody, err := cfg.HelloProvider()
	if err != nil {
		return fmt.Errorf("HelloProvider: %w", err)
	}
	if err := s.sendRaw(&nodev1.NodeMessage{Event: "hello", Body: helloBody}); err != nil {
		return fmt.Errorf("send hello: %w", err)
	}

	if cfg.OnConnected != nil {
		cfg.OnConnected(sessionCtx, s)
	}

	// Dispatcher 选配 SetSender(Sender)：用于流式 method 内部主动 push 帧。
	if setter, ok := cfg.Dispatcher.(interface{ SetSender(Sender) }); ok {
		setter.SetSender(s)
	}

	// 2) recv 循环（在当前 goroutine 跑），dispatch 派发到独立 goroutine。
	recvErr := s.recvLoop()
	s.shutdown()
	if recvErr != nil && !errors.Is(recvErr, io.EOF) && !errors.Is(recvErr, context.Canceled) {
		return recvErr
	}
	if ctx.Err() != nil {
		return ctx.Err()
	}
	if recvErr == nil {
		// 不应发生（recvLoop 只在 Recv 出错或 ctx 取消时返回 nil 是 EOF 情形）
		return errors.New("session ended")
	}
	return recvErr
}

// session 是单次 Session 的状态：
//   - inflight：reqID -> cancelFunc，用于响应 cancel_id；