diff --git a/include/neug/main/connection.h b/include/neug/main/connection.h
index 6d4a94d3c..4f3df319d 100644
--- a/include/neug/main/connection.h
+++ b/include/neug/main/connection.h
@@ -17,8 +17,10 @@
 #include <glog/logging.h>
 
 #include <atomic>
+#include <functional>
 #include <memory>
 #include <string>
+#include <utility>
 #include <vector>
 
 #include "neug/compiler/planner/graph_planner.h"
@@ -83,10 +85,14 @@ class NeugDB;
  */
 class Connection {
  public:
+  using CloseCallback = std::function<void()>;
+
   Connection(GraphSnapshotStore& snapshot_store,
-             std::shared_ptr<QueryProcessor> query_processor)
+             std::shared_ptr<QueryProcessor> query_processor,
+             CloseCallback close_callback = {})
       : snapshot_store_(snapshot_store),
         query_processor_(query_processor),
+        close_callback_(std::move(close_callback)),
         is_closed_(false) {}
   ~Connection() { Close(); }
 
@@ -204,6 +210,7 @@ class Connection {
   GraphSnapshotStore& snapshot_store_;
 
   std::shared_ptr<QueryProcessor> query_processor_;
+  CloseCallback close_callback_;
 
   std::atomic<bool> is_closed_{false};
 };
diff --git a/include/neug/main/connection_manager.h b/include/neug/main/connection_manager.h
index b10fa92bd..b5de66559 100644
--- a/include/neug/main/connection_manager.h
+++ b/include/neug/main/connection_manager.h
@@ -15,6 +15,7 @@
 #pragma once
 
 #include <stddef.h>
+#include <functional>
 #include <memory>
 #include <mutex>
 #include <vector>
@@ -39,7 +40,8 @@ class ConnectionManager {
         config_(config) {}
   ~ConnectionManager() { Close(); }
 
-  std::shared_ptr<Connection> CreateConnection();
+  std::shared_ptr<Connection> CreateConnection(
+      std::function<void()> close_callback = {});
 
   /**
    * @brief Close all connections managed by the connection manager.
diff --git a/include/neug/main/neug_db.h b/include/neug/main/neug_db.h
index bdbd6344e..333b47417 100644
--- a/include/neug/main/neug_db.h
+++ b/include/neug/main/neug_db.h
@@ -250,6 +250,8 @@ class NeugDB {
    *
    * @note In READ_ONLY mode, multiple connections can be created.
    * @note In READ_WRITE mode, only one write connection is allowed.
+   * @note Embedded connections cannot be opened while a NeugDBService is
+   * active.
    * @note Connections share the planner instance for efficiency.
    *
    * @throws std::runtime_error if database is not open or closed
@@ -273,8 +275,8 @@ class NeugDB {
 
   /**
    * @brief Remove all connection from the database.
-   * @note This method is used to remove all connection when tp svc created, to
-   * remove the handle from the database.
+   * @note This method is used during database shutdown to close and remove
+   * managed connection handles.
    */
   void CloseAllConnection();
 
@@ -319,14 +321,22 @@ class NeugDB {
    * resets last_ts_ to 0.
    */
   void createCheckpoint(bool reopen = true);
+  void ValidateCanStartTPService() const;
+  void registerTPService();
+  void unregisterTPService() noexcept;
+  void registerAPConnection();
+  void unregisterAPConnection() noexcept;
 
   friend class NeugDBSession;
+  friend class PyDatabase;
   friend class neug::NeugDBService;
 
   timestamp_t last_compaction_ts_;
   timestamp_t last_ts_;
   // Configuration and settings
   std::atomic<bool> closed_;
+  // 0 means idle, -1 means TP service active, >0 counts AP connections.
+  std::atomic<int32_t> mode_state_;
   bool is_pure_memory_;
   int max_thread_num_;
   NeugDBConfig config_;
diff --git a/include/neug/server/neug_db_service.h b/include/neug/server/neug_db_service.h
index 8c1f7dcd5..3dc07c6b2 100644
--- a/include/neug/server/neug_db_service.h
+++ b/include/neug/server/neug_db_service.h
@@ -97,11 +97,19 @@ class NeugDBService {
    * @param db Reference to the NeuG database that will handle queries
    *
    * @note The database should be opened and ready before creating the service
+   * @throws std::runtime_error if embedded connections are active
    */
   NeugDBService(neug::NeugDB& db, const ServiceConfig& config = ServiceConfig())
       : db_(db), db_config_(db_.config()), compact_thread_running_(false) {
-    db_.CloseAllConnection();
-    init(config);
+    db_.registerTPService();
+    tp_service_registered_ = true;
+    try {
+      init(config);
+    } catch (...) {
+      db_.unregisterTPService();
+      tp_service_registered_ = false;
+      throw;
+    }
   }
 
   /**
@@ -251,6 +259,7 @@ class NeugDBService {
 
   std::thread compact_thread_;
   bool compact_thread_running_ = false;
+  bool tp_service_registered_ = false;
 
   std::atomic<bool> running_{false};
   std::mutex mtx_;
diff --git a/src/main/connection.cc b/src/main/connection.cc
index f5ff54ddb..f2e18704a 100644
--- a/src/main/connection.cc
+++ b/src/main/connection.cc
@@ -15,6 +15,8 @@
 
 #include "neug/main/connection.h"
 
+#include <exception>
+
 #include "neug/main/neug_db.h"
 #include "neug/main/query_request.h"
 #include "neug/utils/pb_utils.h"
@@ -34,10 +36,26 @@ std::string Connection::GetSchema() const {
 }
 
 void Connection::Close() {
-  if (is_closed_.load(std::memory_order_relaxed)) {
+  if (is_closed_.exchange(true, std::memory_order_acq_rel)) {
     LOG(WARNING) << "Connection is already closed.";
     return;
   }
+  struct CloseCallbackGuard {
+    CloseCallback& callback;
+    ~CloseCallbackGuard() noexcept {
+      if (!callback) {
+        return;
+      }
+      try {
+        callback();
+      } catch (const std::exception& e) {
+        LOG(ERROR) << "Connection close callback failed: " << e.what();
+      } catch (...) {
+        LOG(ERROR) << "Connection close callback failed.";
+      }
+    }
+  } close_callback_guard{close_callback_};
+
   LOG(INFO) << "Closing connection.";
 
   // Clean up all temporary schemas created during this session.
@@ -69,8 +87,6 @@ void Connection::Close() {
   if (!temp_edges.empty() || !temp_vertices.empty()) {
     query_processor_->clear_cache();
   }
-
-  is_closed_.store(true);
 }
 
 result<QueryResult> Connection::Query(const std::string& query_string,
diff --git a/src/main/connection_manager.cc b/src/main/connection_manager.cc
index 233c1f2ef..9fbc02cba 100644
--- a/src/main/connection_manager.cc
+++ b/src/main/connection_manager.cc
@@ -16,6 +16,7 @@
 #include "neug/main/connection_manager.h"
 
 #include <glog/logging.h>
+#include <algorithm>
 #include <ostream>
 #include "neug/config.h"
 #include "neug/main/connection.h"
@@ -35,20 +36,30 @@ void ConnectionManager::ConnectionManager::Close() {
   read_only_connections_.clear();
 }
 
-std::shared_ptr<Connection> ConnectionManager::CreateConnection() {
+std::shared_ptr<Connection> ConnectionManager::CreateConnection(
+    std::function<void()> close_callback) {
   std::lock_guard<std::mutex> lock(connection_mutex_);
   if (config_.mode == DBMode::READ_ONLY) {
-    auto conn = std::make_shared<Connection>(snapshot_store_, query_processor_);
+    read_only_connections_.erase(
+        std::remove_if(read_only_connections_.begin(),
+                       read_only_connections_.end(),
+                       [](const auto& conn) { return conn->IsClosed(); }),
+        read_only_connections_.end());
+    auto conn = std::make_shared<Connection>(snapshot_store_, query_processor_,
+                                             close_callback);
     read_only_connections_.push_back(conn);
     return conn;
   } else if (config_.mode == DBMode::READ_WRITE) {
+    if (read_write_connection_ && read_write_connection_->IsClosed()) {
+      read_write_connection_.reset();
+    }
     if (read_write_connection_) {
       LOG(ERROR) << "There is already a read-write connection constructed.";
       THROW_TX_STATE_CONFLICT(
           "There is already a read-write connection constructed.");
     }
-    read_write_connection_ =
-        std::make_shared<Connection>(snapshot_store_, query_processor_);
+    read_write_connection_ = std::make_shared<Connection>(
+        snapshot_store_, query_processor_, close_callback);
     return read_write_connection_;
   } else {
     THROW_RUNTIME_ERROR("Invalid mode.");
@@ -61,6 +72,7 @@ void ConnectionManager::RemoveConnection(std::shared_ptr<Connection> conn) {
     for (auto it = read_only_connections_.begin();
          it != read_only_connections_.end(); ++it) {
       if (*it == conn) {
+        conn->Close();
         read_only_connections_.erase(it);
         VLOG(10) << "Removed a read-only connection.";
         return;
@@ -69,6 +81,7 @@ void ConnectionManager::RemoveConnection(std::shared_ptr<Connection> conn) {
     LOG(ERROR) << "Connection not found in read-only connections.";
   } else if (config_.mode == DBMode::READ_WRITE) {
     if (read_write_connection_ == conn) {
+      conn->Close();
       read_write_connection_.reset();
       VLOG(10) << "Removed the read-write connection.";
       return;
diff --git a/src/main/neug_db.cc b/src/main/neug_db.cc
index 720fd2c07..67d6ec3d2 100644
--- a/src/main/neug_db.cc
+++ b/src/main/neug_db.cc
@@ -46,6 +46,18 @@
 
 namespace neug {
 
+namespace {
+
+void ThrowTPServiceModeConflict(int32_t state) {
+  if (state > 0) {
+    THROW_RUNTIME_ERROR(
+        "Cannot start TP service while embedded connections are active.");
+  }
+  THROW_RUNTIME_ERROR("Cannot start TP service while TP service is active.");
+}
+
+}  // namespace
+
 inline std::string allocator_prefix(const std::string& allocator_dir,
                                     int thread_id) {
   return (std::filesystem::path(allocator_dir) /
@@ -79,6 +91,7 @@ NeugDB::NeugDB()
     : last_compaction_ts_(0),
       last_ts_(0),
       closed_(true),
+      mode_state_(0),
       is_pure_memory_(false),
       max_thread_num_(1) {}
 
@@ -178,7 +191,17 @@ void NeugDB::Close() {
 }
 
 std::shared_ptr<Connection> NeugDB::Connect() {
-  return connection_manager_->CreateConnection();
+  if (IsClosed() || !connection_manager_) {
+    THROW_RUNTIME_ERROR("NeugDB instance is not open.");
+  }
+  registerAPConnection();
+  try {
+    return connection_manager_->CreateConnection(
+        [this]() { unregisterAPConnection(); });
+  } catch (...) {
+    unregisterAPConnection();
+    throw;
+  }
 }
 
 void NeugDB::RemoveConnection(std::shared_ptr<Connection> conn) {
@@ -187,6 +210,61 @@ void NeugDB::RemoveConnection(std::shared_ptr<Connection> conn) {
 
 void NeugDB::CloseAllConnection() { connection_manager_->Close(); }
 
+void NeugDB::ValidateCanStartTPService() const {
+  const auto state = mode_state_.load(std::memory_order_acquire);
+  if (state != 0) {
+    ThrowTPServiceModeConflict(state);
+  }
+}
+
+void NeugDB::registerTPService() {
+  int32_t expected = 0;
+  if (mode_state_.compare_exchange_strong(expected, -1,
+                                          std::memory_order_acq_rel,
+                                          std::memory_order_acquire)) {
+    return;
+  }
+  ThrowTPServiceModeConflict(expected);
+}
+
+void NeugDB::unregisterTPService() noexcept {
+  int32_t expected = -1;
+  if (!mode_state_.compare_exchange_strong(expected, 0,
+                                           std::memory_order_acq_rel,
+                                           std::memory_order_acquire)) {
+    LOG(ERROR) << "Unexpected NeugDB mode state when unregistering TP service: "
+               << expected;
+  }
+}
+
+void NeugDB::registerAPConnection() {
+  auto state = mode_state_.load(std::memory_order_acquire);
+  while (true) {
+    if (state < 0) {
+      THROW_RUNTIME_ERROR(
+          "Cannot open embedded connection while TP service is active.");
+    }
+    if (mode_state_.compare_exchange_weak(state, state + 1,
+                                          std::memory_order_acq_rel,
+                                          std::memory_order_acquire)) {
+      return;
+    }
+  }
+}
+
+void NeugDB::unregisterAPConnection() noexcept {
+  auto state = mode_state_.load(std::memory_order_acquire);
+  while (state > 0) {
+    if (mode_state_.compare_exchange_weak(state, state - 1,
+                                          std::memory_order_acq_rel,
+                                          std::memory_order_acquire)) {
+      return;
+    }
+  }
+  LOG(ERROR) << "Unexpected NeugDB mode state when unregistering AP connection: "
+             << state;
+}
+
 void NeugDB::preprocessConfig() {
   if (config_.max_thread_num < 0) {
     THROW_INVALID_ARGUMENT_EXCEPTION(
diff --git a/src/server/neug_db_service.cc b/src/server/neug_db_service.cc
index b847e0e95..1b7346a0c 100644
--- a/src/server/neug_db_service.cc
+++ b/src/server/neug_db_service.cc
@@ -65,6 +65,10 @@ NeugDBService::~NeugDBService() {
     hdl_mgr_->Stop();
     hdl_mgr_.reset();
   }
+  if (tp_service_registered_) {
+    db_.unregisterTPService();
+    tp_service_registered_ = false;
+  }
 }
 
 const ServiceConfig& NeugDBService::GetServiceConfig() const {
diff --git a/tests/transaction/test_compact_transaction.cc b/tests/transaction/test_compact_transaction.cc
index 078abfa4a..93b48192e 100644
--- a/tests/transaction/test_compact_transaction.cc
+++ b/tests/transaction/test_compact_transaction.cc
@@ -17,6 +17,7 @@
 #include "neug/server/neug_db_service.h"
 #include "neug/storages/graph/graph_interface.h"
 #include "neug/transaction/compact_transaction.h"
+#include "neug/transaction/update_transaction.h"
 #include "neug/transaction/version_manager.h"
 
 #include <atomic>
@@ -106,6 +107,19 @@ class CompactTransactionTest : public ::testing::Test {
     });
     return edge_count;
   }
+
+  void delete_person(neug::NeugDBService& service, int64_t id) {
+    auto sess = service.AcquireSession();
+    auto txn = sess->GetUpdateTransaction();
+    neug::StorageTPUpdateInterface interface(txn);
+    const auto person_label = txn.schema().get_vertex_label_id("person");
+    neug::vid_t vertex_id = 0;
+    ASSERT_TRUE(txn.GetVertexIndex(person_label,
+                                   neug::execution::Value::INT64(id),
+                                   vertex_id));
+    ASSERT_TRUE(interface.DeleteVertex(person_label, vertex_id));
+    ASSERT_TRUE(txn.Commit());
+  }
 };
 
 // Commit, Abort, and destructor (auto-abort) should all preserve data.
@@ -163,12 +177,7 @@ TEST_F(CompactTransactionTest, DeleteThenCompactPurgesData) {
   db.Open(config);
   auto svc = std::make_shared<neug::NeugDBService>(db);
 
-  // Delete person id=2 via Cypher
-  {
-    auto conn = db.Connect();
-    EXPECT_TRUE(conn->Query("MATCH (v:person) WHERE v.id = 2 DELETE v;"));
-    conn->Close();
-  }
+  delete_person(*svc, 2);
 
   // Verify deletion visible before compact
   {
@@ -220,12 +229,7 @@ TEST_F(CompactTransactionTest, CompactAndReopenPersistsData) {
     db.Open(config);
     auto svc = std::make_shared<neug::NeugDBService>(db);
 
-    // Delete person id=1 via Cypher
-    {
-      auto conn = db.Connect();
-      EXPECT_TRUE(conn->Query("MATCH (v:person) WHERE v.id = 1 DELETE v;"));
-      conn->Close();
-    }
+    delete_person(*svc, 1);
 
     // Compact explicitly before close
     {
diff --git a/tests/transaction/test_wal_replay.cc b/tests/transaction/test_wal_replay.cc
index 8352e3ea2..9fb6eb0dc 100644
--- a/tests/transaction/test_wal_replay.cc
+++ b/tests/transaction/test_wal_replay.cc
@@ -76,6 +76,7 @@ void create_person_schema(neug::NeugDB& db) {
   assert_query_ok(
       *conn,
       "CREATE NODE TABLE person(id INT64, name STRING, PRIMARY KEY(id));");
+  conn->Close();
 }
 
 bool replayed_graph_matches(neug::NeugDB& db) {
diff --git a/tests/unittest/test_db_svc.cc b/tests/unittest/test_db_svc.cc
index a64d6bb23..07771f77d 100644
--- a/tests/unittest/test_db_svc.cc
+++ b/tests/unittest/test_db_svc.cc
@@ -46,6 +46,7 @@ class NeugDBServiceTest : public ::testing::Test {
     // Load modern graph
     auto conn = db_->Connect();
     load_modern_graph(conn);
+    conn->Close();
 
     // Configure service
     config_.query_port = 19999;  // Use non-standard port to avoid conflicts
@@ -154,6 +155,35 @@ TEST_F(NeugDBServiceTest, ServiceThreadNumCannotExceedDatabaseMaxThreadNum) {
                neug::exception::InvalidArgumentException);
 }
 
+TEST_F(NeugDBServiceTest, ServiceCreationRejectsActiveEmbeddedConnection) {
+  auto conn = db_->Connect();
+
+  EXPECT_THROW(neug::NeugDBService service(*db_, config_),
+               neug::exception::RuntimeError);
+
+  conn->Close();
+}
+
+TEST_F(NeugDBServiceTest, EmbeddedConnectionRejectedWhileServiceActive) {
+  {
+    neug::NeugDBService service(*db_, config_);
+
+    EXPECT_THROW({ auto conn = db_->Connect(); },
+                 neug::exception::RuntimeError);
+  }
+
+  auto conn = db_->Connect();
+  ASSERT_NE(conn, nullptr);
+  conn->Close();
+}
+
+TEST_F(NeugDBServiceTest, ClosedEmbeddedConnectionDoesNotBlockService) {
+  auto conn = db_->Connect();
+  conn->Close();
+
+  EXPECT_NO_THROW(neug::NeugDBService service(*db_, config_));
+}
+
 TEST_F(NeugDBServiceTest, ConcurrentSessionOperations) {
   neug::NeugDBService service(*db_, config_);
   const int num_threads = 4;
diff --git a/tools/python_bind/neug/database.py b/tools/python_bind/neug/database.py
index bec15ccf1..d8cb68146 100644
--- a/tools/python_bind/neug/database.py
+++ b/tools/python_bind/neug/database.py
@@ -258,8 +258,8 @@ def serve(
         """
         Start the database server for handling remote connections(TP mode).
         This method is used to start the database server for handling remote connections.
-        When db.serve() is called, the database will switch to the TP mode, and all the connections to the local database
-        will be closed. After that, no new connections to the local database will be allowed.
+        When db.serve() is called, the database will switch to TP mode. All local database connections must already be
+        closed. After that, no new local database connections will be allowed.
         It will start a server that listens on a specific port, and clients can connect to the server to interact with the
         database. User could use Session to connect to the server. For detail usage, please refer to the
         documentation of Session.
@@ -323,8 +323,8 @@ def serve(
                 raise RuntimeError(
                     "Cannot start the server while there are open async connections to the local database."
                 )
-        # We should not clear the connections here, because the connection maybe held by the user.
-        # Instead, we will close all connections when the server is stopped.
+        # Keep closed connection handles in the wrapper because they may still be
+        # held by the user. Active handles were rejected above.
         if self._serving:
             logger.warning("Database server is already running.")
             return
diff --git a/tools/python_bind/src/py_database.cc b/tools/python_bind/src/py_database.cc
index f4b35075c..2ca1230a2 100644
--- a/tools/python_bind/src/py_database.cc
+++ b/tools/python_bind/src/py_database.cc
@@ -79,6 +79,7 @@ void PyDatabase::initialize(pybind11::handle& m) {
 }
 
 PyConnection PyDatabase::connect() {
+  std::lock_guard<std::recursive_mutex> lock(mtx_);
   if (!database) {
     THROW_RUNTIME_ERROR("Database is not initialized.");
   }
@@ -88,6 +89,8 @@ PyConnection PyDatabase::connect() {
 std::string PyDatabase::serve(int port, const std::string& host,
                               int32_t thread_num, bool blocking) {
 #ifdef BUILD_HTTP_SERVER
+  std::lock_guard<std::recursive_mutex> lock(mtx_);
+
   if (!database) {
     THROW_RUNTIME_ERROR("Database is not initialized.");
   }
@@ -115,8 +118,7 @@ std::string PyDatabase::serve(int port, const std::string& host,
    * doing this, we make sure all changes made during AP mode is persisted.
    */
 
-  std::lock_guard<std::recursive_mutex> lock(mtx_);
-
+  database->ValidateCanStartTPService();
   database->Close();
   database->Open(database->config());
   neug::ServiceConfig config;
diff --git a/tools/python_bind/tests/test_tp_service.py b/tools/python_bind/tests/test_tp_service.py
index c70be9407..53ff43bc1 100644
--- a/tools/python_bind/tests/test_tp_service.py
+++ b/tools/python_bind/tests/test_tp_service.py
@@ -106,6 +106,32 @@ def test_start_service_on_pure_memory_db():
     db.close()
 
 
+def test_serve_rejects_active_embedded_connection(tmp_path):
+    db_dir = str(tmp_path / "test_serve_rejects_active_embedded_connection")
+    shutil.rmtree(db_dir, ignore_errors=True)
+    db = Database(db_dir, "w")
+    conn = db.connect()
+    try:
+        with pytest.raises(Exception, match="open connections|embedded connections"):
+            db.serve(19002, "127.0.0.1", False)
+    finally:
+        conn.close()
+        db.close()
+
+
+def test_connect_rejected_while_service_active(tmp_path):
+    db_dir = str(tmp_path / "test_connect_rejected_while_service_active")
+    shutil.rmtree(db_dir, ignore_errors=True)
+    db = Database(db_dir, "w")
+    try:
+        db.serve(19003, "127.0.0.1", False)
+        with pytest.raises(Exception, match="server is running|TP service"):
+            db.connect()
+    finally:
+        db.stop_serving()
+        db.close()
+
+
 def test_start_serving_and_dump(tmp_path):
     db_dir = str(tmp_path / "test_start_serving_and_dump")
     shutil.rmtree(db_dir, ignore_errors=True)