feat: migrate Thin SDK to sidecar control session

README.md

@@ -2,8 +2,9 @@
 
 `pole-client-cpp` 是框架无关的 Pole C++ Thin SDK。它只负责：
 
-- 通过 Unix Domain Socket 与本 Pod Sidecar 建立内部 gRPC bootstrap 会话；
-- 接收 Sidecar 主动下发的一次性 HTTP、gRPC、Dubbo、Thrift listener 快照；
+- 通过 Unix Domain Socket 与本 Pod Sidecar 建立内部双向 `OpenControlSession`；
+- 首个服务端事件原子安装 HTTP、gRPC、Dubbo、Thrift listener 快照；
+- 注册或注销本地服务；SDK 在会话重连后重放仍处于 desired 状态的注册；
 - 为业务请求生成 `latticehub-target-namespace` 与 `latticehub-target-service` 元信息。
 
 业务请求不会经过 bootstrap gRPC；应用仍按原协议访问 SDK 返回的
@@ -31,6 +32,10 @@ target_link_libraries(your_target PRIVATE PoleClient::pole-client-cpp)
 `SidecarSessionOptions.socket_path` 或 `POLE_SIDECAR_SOCKET` 覆盖。SDK 不硬编码
 任何业务 listener 端口；会话断开后快照立即失效，重连并收到完整首帧后恢复。
 
+Specification 子模块固定在正式集成提交
+`2642bc29c0a512f4da84ec4eb862b1e1ceee9833`（`develop`，PR #25），该提交已包含
+`OpenControlSession` 契约及官方 C++ 生成方式。
+
 ## 使用
 
 ```cpp
@@ -40,6 +45,14 @@ target_link_libraries(your_target PRIVATE PoleClient::pole-client-cpp)
 auto session = pole::client::SidecarSession::Connect();
 const auto grpc_address = session->ListenerAddress(pole::client::Protocol::kGrpc);
 
+const auto registration_id = session->RegisterLocalService({
+    .service_namespace = "default",
+    .service = "orders",
+    .protocol = pole::client::Protocol::kGrpc,
+    .local_port = 50051,
+});
+session->UnregisterLocalService(registration_id);
+
 const pole::client::TargetService target("default", "orders");
 const auto metadata = target.EncodeMetadata();
 ```

context-kg/tasks/lessons.md

@@ -0,0 +1,4 @@
+# 经验记录
+
+- Protobuf 字段名 `namespace` 是 C++ 关键字，官方生成器会把访问器命名为 `namespace_()` / `set_namespace_()`；不能按其他语言习惯写成 `set_namespace()`。
+- C++ Release 构建会通过 `NDEBUG` 把 `assert(expression)` 连同 expression 本身删除；集成测试不能把 `Read`、`Write` 等有副作用操作只放在 assert 参数里，至少要确保测试翻译单元启用断言或改用始终求值的检查函数。

context-kg/tasks/todo.md

@@ -0,0 +1,13 @@
+# Thin SDK OpenControlSession migration
+
+- [x] Verify the `codex/sidecar-service-session-v2` bootstrap contract.
+- [x] Replace legacy `OpenSession` with bidirectional `OpenControlSession`.
+- [x] Add local-service registration lifecycle and reconnect replay.
+- [x] Update documentation and focused tests.
+- [x] Run available validation and review the final diff.
+
+## Review
+
+- Contract source: Specification `develop` commit `2642bc29c0a512f4da84ec4eb862b1e1ceee9833`.
+- Passed: `git diff --check`; `protoc` descriptor validation for the formal bootstrap proto.
+- Blocked: CMake configuration requires Protobuf `5.26.1`; this environment offers only `35.1.0`.

include/pole/client/sidecar_session.h

@@ -4,7 +4,9 @@
 #include <condition_variable>
 #include <cstdint>
 #include <memory>
+#include <optional>
 #include <mutex>
+#include <map>
 #include <stdexcept>
 #include <string>
 #include <thread>
@@ -26,6 +28,26 @@ enum class Protocol : std::uint8_t {
   kThrift,
 };
 
+struct LocalServiceRegistration {
+  std::string service_namespace;
+  std::string service;
+  Protocol protocol;
+  std::uint16_t local_port;
+  std::string registration_id;
+};
+
+enum class LocalServiceState : std::uint8_t {
+  kUnspecified,
+  kRegistered,
+  kUnregistered,
+  kRejected,
+};
+
+struct LocalServiceStatus {
+  LocalServiceState state;
+  std::string message;
+};
+
 struct SidecarSessionOptions {
   std::string socket_path;
   std::string sdk_version = kSdkVersion;
@@ -55,23 +77,36 @@ class SidecarSession final {
   [[nodiscard]] const std::string& socket_path() const noexcept;
   [[nodiscard]] bool available() const;
   [[nodiscard]] std::string ListenerAddress(Protocol protocol) const;
+  std::string RegisterLocalService(LocalServiceRegistration registration);
+  bool UnregisterLocalService(const std::string& registration_id);
+  [[nodiscard]] std::optional<LocalServiceStatus> GetLocalServiceStatus(
+      const std::string& registration_id) const;
   void Close();
 
  private:
+  struct ActiveStream;
   struct ListenerSnapshot;
 
   explicit SidecarSession(SidecarSessionOptions options);
   void Run();
   void InstallSnapshot(std::shared_ptr<const ListenerSnapshot> snapshot);
   void InvalidateSnapshot();
   void SetInitializationError(std::string error);
+  void SendRegistration(const LocalServiceRegistration& registration);
+  void SendUnregistration(const std::string& registration_id);
+  void HandleLocalServiceStatus(LocalServiceStatus status, const std::string& registration_id);
 
   SidecarSessionOptions options_;
   mutable std::mutex mutex_;
   std::condition_variable condition_;
   std::shared_ptr<const ListenerSnapshot> listeners_;
+  std::map<std::string, LocalServiceRegistration> desired_registrations_;
+  std::map<std::string, LocalServiceStatus> local_service_statuses_;
   std::thread worker_;
+  std::mutex stream_mutex_;
+  ActiveStream* active_stream_ = nullptr;
   grpc::ClientContext* active_context_ = nullptr;
+  std::uint64_t next_registration_id_ = 1;
   bool initialized_ = false;
   bool closed_ = false;
   std::string initialization_error_;

src/sidecar_session.cc

@@ -3,6 +3,7 @@
 #include <algorithm>
 #include <array>
 #include <cstdlib>
+#include <map>
 #include <optional>
 #include <utility>
 
@@ -34,6 +35,36 @@ std::optional<Protocol> FromWireProtocol(BootstrapProtocol protocol) {
   }
 }
 
+BootstrapProtocol ToWireProtocol(Protocol protocol) {
+  switch (protocol) {
+    case Protocol::kHttp:
+      return BootstrapProtocol::PROTOCOL_HTTP;
+    case Protocol::kGrpc:
+      return BootstrapProtocol::PROTOCOL_GRPC;
+    case Protocol::kDubbo:
+      return BootstrapProtocol::PROTOCOL_DUBBO;
+    case Protocol::kThrift:
+      return BootstrapProtocol::PROTOCOL_THRIFT;
+  }
+  return BootstrapProtocol::PROTOCOL_UNSPECIFIED;
+}
+
+std::optional<LocalServiceState> FromWireLocalServiceState(
+    ::pole::sidecar::v1::LocalServiceState state) {
+  switch (state) {
+    case ::pole::sidecar::v1::LOCAL_SERVICE_STATE_REGISTERED:
+      return LocalServiceState::kRegistered;
+    case ::pole::sidecar::v1::LOCAL_SERVICE_STATE_UNREGISTERED:
+      return LocalServiceState::kUnregistered;
+    case ::pole::sidecar::v1::LOCAL_SERVICE_STATE_REJECTED:
+      return LocalServiceState::kRejected;
+    case ::pole::sidecar::v1::LOCAL_SERVICE_STATE_UNSPECIFIED:
+      return LocalServiceState::kUnspecified;
+    default:
+      return std::nullopt;
+  }
+}
+
 std::string ResolveSocketPath(const std::string& configured) {
   if (!configured.empty()) {
     return configured;
@@ -64,6 +95,11 @@ struct SidecarSession::ListenerSnapshot {
   std::array<std::uint16_t, 4> ports{};
 };
 
+struct SidecarSession::ActiveStream {
+  ::grpc::ClientReaderWriter<::pole::sidecar::v1::ClientEvent,
+                             ::pole::sidecar::v1::SidecarEvent>* writer;
+};
+
 SidecarUnavailableError::SidecarUnavailableError()
     : std::runtime_error("Pole Sidecar listener snapshot is unavailable") {}
 
@@ -105,6 +141,52 @@ std::string SidecarSession::ListenerAddress(Protocol protocol) const {
   return "127.0.0.1:" + std::to_string(listeners_->ports[ProtocolIndex(protocol)]);
 }
 
+std::string SidecarSession::RegisterLocalService(LocalServiceRegistration registration) {
+  if (registration.service_namespace.empty() || registration.service.empty() ||
+      registration.local_port == 0) {
+    throw std::invalid_argument("local service namespace, service, and local_port must be set");
+  }
+  {
+    std::lock_guard lock(mutex_);
+    if (closed_) {
+      throw std::runtime_error("Sidecar session is closed");
+    }
+    if (registration.registration_id.empty()) {
+      registration.registration_id = "cpp-" + std::to_string(next_registration_id_++);
+    }
+    const auto [_, inserted] = desired_registrations_.emplace(registration.registration_id, registration);
+    if (!inserted) {
+      throw std::invalid_argument("registration_id is already registered");
+    }
+  }
+  SendRegistration(registration);
+  return registration.registration_id;
+}
+
+bool SidecarSession::UnregisterLocalService(const std::string& registration_id) {
+  if (registration_id.empty()) {
+    throw std::invalid_argument("registration_id must not be empty");
+  }
+  {
+    std::lock_guard lock(mutex_);
+    if (desired_registrations_.erase(registration_id) == 0) {
+      return false;
+    }
+  }
+  SendUnregistration(registration_id);
+  return true;
+}
+
+std::optional<LocalServiceStatus> SidecarSession::GetLocalServiceStatus(
+    const std::string& registration_id) const {
+  std::lock_guard lock(mutex_);
+  const auto status = local_service_statuses_.find(registration_id);
+  if (status == local_service_statuses_.end()) {
+    return std::nullopt;
+  }
+  return status->second;
+}
+
 void SidecarSession::Close() {
   {
     std::lock_guard lock(mutex_);
@@ -113,6 +195,15 @@ void SidecarSession::Close() {
     }
     closed_ = true;
     listeners_.reset();
+  }
+  {
+    std::lock_guard stream_lock(stream_mutex_);
+    if (active_stream_ != nullptr) {
+      active_stream_->writer->WritesDone();
+    }
+  }
+  {
+    std::lock_guard lock(mutex_);
     if (active_context_ != nullptr) {
       active_context_->TryCancel();
     }
@@ -150,6 +241,37 @@ void SidecarSession::SetInitializationError(std::string error) {
   condition_.notify_all();
 }
 
+void SidecarSession::SendRegistration(const LocalServiceRegistration& registration) {
+  std::lock_guard stream_lock(stream_mutex_);
+  if (active_stream_ == nullptr) {
+    return;
+  }
+  ::pole::sidecar::v1::ClientEvent event;
+  auto* wire_registration = event.mutable_register_local_service();
+  wire_registration->set_registration_id(registration.registration_id);
+  wire_registration->set_namespace_(registration.service_namespace);
+  wire_registration->set_service(registration.service);
+  wire_registration->set_protocol(ToWireProtocol(registration.protocol));
+  wire_registration->set_local_port(registration.local_port);
+  active_stream_->writer->Write(event);
+}
+
+void SidecarSession::SendUnregistration(const std::string& registration_id) {
+  std::lock_guard stream_lock(stream_mutex_);
+  if (active_stream_ == nullptr) {
+    return;
+  }
+  ::pole::sidecar::v1::ClientEvent event;
+  event.mutable_unregister_local_service()->set_registration_id(registration_id);
+  active_stream_->writer->Write(event);
+}
+
+void SidecarSession::HandleLocalServiceStatus(LocalServiceStatus status,
+                                              const std::string& registration_id) {
+  std::lock_guard lock(mutex_);
+  local_service_statuses_[registration_id] = std::move(status);
+}
+
 void SidecarSession::Run() {
   auto retry_delay = options_.retry_initial_delay;
   while (true) {
@@ -172,18 +294,51 @@ void SidecarSession::Run() {
       active_context_ = &context;
     }
 
-    ::pole::sidecar::v1::ClientHello hello;
-    hello.set_sdk_language(kSdkLanguage);
-    hello.set_sdk_version(options_.sdk_version);
-    hello.add_supported_protocols(BootstrapProtocol::PROTOCOL_HTTP);
-    hello.add_supported_protocols(BootstrapProtocol::PROTOCOL_GRPC);
-    hello.add_supported_protocols(BootstrapProtocol::PROTOCOL_DUBBO);
-    hello.add_supported_protocols(BootstrapProtocol::PROTOCOL_THRIFT);
+    auto stream = stub->OpenControlSession(&context);
+    ::pole::sidecar::v1::ClientEvent hello_event;
+    auto* hello = hello_event.mutable_hello();
+    hello->set_sdk_language(kSdkLanguage);
+    hello->set_sdk_version(options_.sdk_version);
+    hello->add_supported_protocols(BootstrapProtocol::PROTOCOL_HTTP);
+    hello->add_supported_protocols(BootstrapProtocol::PROTOCOL_GRPC);
+    hello->add_supported_protocols(BootstrapProtocol::PROTOCOL_DUBBO);
+    hello->add_supported_protocols(BootstrapProtocol::PROTOCOL_THRIFT);
+    if (!stream->Write(hello_event)) {
+      stream->Finish();
+      {
+        std::lock_guard lock(mutex_);
+        active_context_ = nullptr;
+      }
+      SetInitializationError("failed to send ClientHello to Sidecar");
+      return;
+    }
+
+    ActiveStream active_stream{stream.get()};
+    {
+      std::lock_guard stream_lock(stream_mutex_);
+      active_stream_ = &active_stream;
+      std::map<std::string, LocalServiceRegistration> registrations;
+      {
+        std::lock_guard lock(mutex_);
+        registrations = desired_registrations_;
+      }
+      for (const auto& [_, registration] : registrations) {
+        ::pole::sidecar::v1::ClientEvent event;
+        auto* wire_registration = event.mutable_register_local_service();
+        wire_registration->set_registration_id(registration.registration_id);
+        wire_registration->set_namespace_(registration.service_namespace);
+        wire_registration->set_service(registration.service);
+        wire_registration->set_protocol(ToWireProtocol(registration.protocol));
+        wire_registration->set_local_port(registration.local_port);
+        if (!stream->Write(event)) {
+          break;
+        }
+      }
+    }
 
-    auto reader = stub->OpenSession(&context, hello);
     ::pole::sidecar::v1::SidecarEvent event;
     bool received_snapshot = false;
-    if (reader->Read(&event)) {
+    if (stream->Read(&event)) {
       if (!event.has_listener_snapshot()) {
         SetInitializationError("first Sidecar event must be listener_snapshot");
       } else {
@@ -215,12 +370,26 @@ void SidecarSession::Run() {
           InstallSnapshot(std::move(snapshot));
           received_snapshot = true;
           retry_delay = options_.retry_initial_delay;
-          while (reader->Read(&event)) {
+          while (stream->Read(&event)) {
+            if (event.has_local_service_status()) {
+              const auto state = FromWireLocalServiceState(event.local_service_status().state());
+              if (state.has_value()) {
+                HandleLocalServiceStatus(
+                    LocalServiceStatus{*state, event.local_service_status().message()},
+                    event.local_service_status().registration_id());
+              }
+            }
           }
         }
       }
     }
-    reader->Finish();
+    {
+      std::lock_guard stream_lock(stream_mutex_);
+      if (active_stream_ == &active_stream) {
+        active_stream_ = nullptr;
+      }
+    }
+    stream->Finish();
     {
       std::lock_guard lock(mutex_);
       active_context_ = nullptr;