diff --git a/doc/third-party/ipi.md b/doc/third-party/ipi.md
index 33a90c56c8..ee2765873a 100644
--- a/doc/third-party/ipi.md
+++ b/doc/third-party/ipi.md
@@ -41,3 +41,17 @@ The option **`port`** should be the same as that in input.xml:
 The option **`graph_file`** provides the file name of the frozen model. The model can have either double or single float precision interface.
 
 The `dp_ipi` gets the atom names from an [XYZ file](https://en.wikipedia.org/wiki/XYZ_file_format) provided by **`coord_file`** (meanwhile ignores all coordinates in it) and translates the names to atom types by rules provided by **`atom_type`**.
+
+The client accepts the standard i-PI `INIT` message (replica ID, payload length,
+and payload). Its model and atom types remain configured by the JSON input;
+initialization payloads are read in chunks with no additional size cap beyond
+the signed 32-bit protocol length limit. Negative payload lengths are rejected
+with an error.
+
+Socket reads assemble each requested protocol field completely, including when
+its bytes arrive in separate packets. Interrupted reads are retried. If the
+server closes the connection before a field is complete, the client rejects the
+incomplete read and exits with an error. Bytes received before EOF have already
+been read from the socket, and verbose mode may have printed earlier complete
+INIT chunks. To request a successful client shutdown, send the standard `EXIT`
+message rather than closing the socket without it.
diff --git a/source/ipi/driver.cc b/source/ipi/driver.cc
index 8e863398d3..c1bdd492fe 100644
--- a/source/ipi/driver.cc
+++ b/source/ipi/driver.cc
@@ -1,4 +1,5 @@
 // SPDX-License-Identifier: LGPL-3.0-or-later
+#include <algorithm>
 #include <cstdint>
 #include <exception>
 #include <fstream>
@@ -162,11 +163,33 @@ int main(int argc, char* argv[]) {
       }
     } else if (header_str == "INIT") {
       assert(4 == sizeof(int32_t));
+      int32_t replica_id;
+      readbuffer_(&socket, (char*)(&replica_id), sizeof(int32_t));
       readbuffer_(&socket, (char*)(&cbuf), sizeof(int32_t));
-      readbuffer_(&socket, initbuffer, cbuf);
+      if (cbuf < 0) {
+        std::cerr << "dp_ipi: INIT payload length must be nonnegative."
+                  << std::endl;
+        return 1;
+      }
+      if (b_verb) {
+        std::cout << "Init sys from wrapper, using ";
+      }
+      // This driver is configured by its input file. Consume the optional
+      // initialization string without imposing a fixed payload-size limit.
+      while (cbuf > 0) {
+        const int chunk =
+            std::min(cbuf, static_cast<int32_t>(sizeof(initbuffer)));
+        readbuffer_(&socket, initbuffer, chunk);
+        if (b_verb) {
+          std::cout.write(initbuffer, chunk);
+        }
+        cbuf -= chunk;
+      }
       if (b_verb) {
-        std::cout << "Init sys from wrapper, using " << initbuffer << std::endl;
+        std::cout << std::endl;
       }
+      isinit = true;
+      hasdata = false;
     } else if (header_str == "POSDATA") {
       assert(8 == sizeof(double));
 
@@ -179,6 +202,12 @@ int main(int argc, char* argv[]) {
 
       // get number of atoms
       readbuffer_(&socket, (char*)(&cbuf), sizeof(int32_t));
+      if (cbuf <= 0 || static_cast<size_t>(cbuf) != dtype.size()) {
+        std::cerr << "dp_ipi: POSDATA atom count " << cbuf
+                  << " does not match configured atom count " << dtype.size()
+                  << "." << std::endl;
+        return 1;
+      }
       if (natoms < 0) {
         natoms = cbuf;
         if (b_verb) {
@@ -225,6 +254,8 @@ int main(int argc, char* argv[]) {
       writebuffer_(&socket, (char*)(&cbuf), sizeof(int32_t));
       writebuffer_(&socket, msg_nothing, 7);
       hasdata = false;
+    } else if (header_str == "EXIT") {
+      break;
     } else {
       std::cerr << "unexpected header " << std::endl;
       return 1;
diff --git a/source/ipi/include/sockets.h b/source/ipi/include/sockets.h
index 150b7c1a69..24e17a4a2d 100644
--- a/source/ipi/include/sockets.h
+++ b/source/ipi/include/sockets.h
@@ -39,7 +39,9 @@ void open_socket_(int* psockfd, int* inet, int* port, const char* host);
 */
 void writebuffer_(int* psockfd, char* data, int len);
 
-/* Reads from a socket.
+/* Reads exactly len bytes from a socket. Interrupted reads are retried.
+   Exits on premature EOF, another read error, or a negative length.
+   A zero-length request succeeds without reading.
    Args:
    psockfd: The id of the socket that will be read from.
    data: The storage array for data read from the socket.
diff --git a/source/ipi/src/sockets.c b/source/ipi/src/sockets.c
index 6bfd881730..bcd28322a5 100644
--- a/source/ipi/src/sockets.c
+++ b/source/ipi/src/sockets.c
@@ -110,6 +110,34 @@ int deepmd_write_all(int sockfd,
   return 0;
 }
 
+int deepmd_read_all(int sockfd,
+                    char* data,
+                    size_t len,
+                    deepmd_socket_read_fn read_fn) {
+  size_t received = 0;
+  if (read_fn == NULL || (data == NULL && len != 0)) {
+    errno = EINVAL;
+    return -1;
+  }
+  while (received < len) {
+    ssize_t count = read_fn(sockfd, data + received, len - received);
+    if (count > 0) {
+      if ((size_t)count > len - received) {
+        errno = EIO;
+        return -1;
+      }
+      received += (size_t)count;
+    } else if (count == 0) {
+      // A partial frame must never be consumed as a complete protocol field.
+      errno = ECONNRESET;
+      return -1;
+    } else if (errno != EINTR) {
+      return -1;
+    }
+  }
+  return 0;
+}
+
 void error(const char* msg)
 // Prints an error message and then exits.
 {
@@ -213,17 +241,11 @@ void readbuffer_(int* psockfd, char* data, int len)
 */
 
 {
-  int n, nr;
-  int sockfd = *psockfd;
-
-  n = nr = read(sockfd, data, len);
-
-  while (nr > 0 && n < len) {
-    nr = read(sockfd, &data[n], len - n);
-    n += nr;
+  if (len < 0) {
+    errno = EINVAL;
+    error("Error reading from socket: invalid buffer length");
   }
-
-  if (n == 0) {
+  if (deepmd_read_all(*psockfd, data, (size_t)len, read) < 0) {
     error("Error reading from socket: server has quit or connection broke");
   }
 }
diff --git a/source/ipi/src/sockets_internal.h b/source/ipi/src/sockets_internal.h
index c4c66e8831..6f9a567443 100644
--- a/source/ipi/src/sockets_internal.h
+++ b/source/ipi/src/sockets_internal.h
@@ -19,3 +19,10 @@ int deepmd_write_all(int sockfd,
                      const char* data,
                      size_t len,
                      deepmd_socket_write_fn write_fn);
+
+// Receive exactly len bytes; EOF before completion is an error.
+typedef ssize_t (*deepmd_socket_read_fn)(int, void*, size_t);
+int deepmd_read_all(int sockfd,
+                    char* data,
+                    size_t len,
+                    deepmd_socket_read_fn read_fn);
diff --git a/source/ipi/tests/test_driver_exit.py b/source/ipi/tests/test_driver_exit.py
new file mode 100644
index 0000000000..fd423c823a
--- /dev/null
+++ b/source/ipi/tests/test_driver_exit.py
@@ -0,0 +1,73 @@
+# SPDX-License-Identifier: LGPL-3.0-or-later
+import json
+import os
+import socket
+import subprocess
+import tempfile
+import unittest
+from pathlib import (
+    Path,
+)
+
+
+@unittest.skipIf(
+    os.environ.get("ENABLE_PYTORCH", "1") != "1",
+    reason="Skip test because PyTorch support is not enabled.",
+)
+class TestDPIPIExit(unittest.TestCase):
+    def test_exit_header_stops_driver_cleanly(self) -> None:
+        tests_path = Path(__file__).parent.parent.parent / "tests"
+        model_file = tests_path / "infer" / "deeppot_sea.pth"
+
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            tmp_path = Path(tmp_dir)
+            coord_file = tmp_path / "coord.xyz"
+            config_file = tmp_path / "config.json"
+            socket_name = f"dpipi_exit_{os.getpid()}_{tmp_path.name}"
+            socket_path = Path("/tmp") / f"ipi_{socket_name}"
+            coord_file.write_text(
+                "1\nEXIT protocol regression test\nO 0.0 0.0 0.0\n",
+                encoding="utf-8",
+            )
+            config_file.write_text(
+                json.dumps(
+                    {
+                        "verbose": False,
+                        "use_unix": True,
+                        "port": 31415,
+                        "host": socket_name,
+                        "graph_file": str(model_file),
+                        "coord_file": str(coord_file),
+                        "atom_type": {"O": 0},
+                    }
+                ),
+                encoding="utf-8",
+            )
+
+            server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+            process = None
+            try:
+                server.bind(str(socket_path))
+                server.listen(1)
+                server.settimeout(30)
+                process = subprocess.Popen(
+                    ["dp_ipi", str(config_file)],
+                    stdout=subprocess.PIPE,
+                    stderr=subprocess.PIPE,
+                    text=True,
+                )
+                connection, _ = server.accept()
+                with connection:
+                    connection.sendall(b"EXIT        ")
+                stdout, stderr = process.communicate(timeout=60)
+                self.assertEqual(
+                    process.returncode,
+                    0,
+                    msg=f"stdout:\n{stdout}\nstderr:\n{stderr}",
+                )
+            finally:
+                if process is not None and process.poll() is None:
+                    process.kill()
+                    process.wait()
+                server.close()
+                socket_path.unlink(missing_ok=True)
diff --git a/source/ipi/tests/test_driver_initialization.py b/source/ipi/tests/test_driver_initialization.py
new file mode 100644
index 0000000000..8c0637323e
--- /dev/null
+++ b/source/ipi/tests/test_driver_initialization.py
@@ -0,0 +1,112 @@
+# SPDX-License-Identifier: LGPL-3.0-or-later
+"""Validate framing of the standard i-PI initialization message."""
+
+import json
+import os
+import socket
+import struct
+import subprocess
+import tempfile
+import unittest
+from pathlib import (
+    Path,
+)
+
+
+@unittest.skipIf(
+    os.environ.get("ENABLE_PYTORCH", "1") != "1",
+    reason="Skip test because PyTorch support is not enabled.",
+)
+class TestDPIPIInitialization(unittest.TestCase):
+    def exchange(
+        self, payload: bytes, length: int | None = None, *, verbose: bool = False
+    ) -> None:
+        """Check INIT framing, optionally advertising an invalid payload length.
+
+        A complete payload must preserve the following STATUS header. A short
+        payload is followed by EOF and must fail before its bytes are logged.
+        """
+        model = Path(__file__).resolve().parents[2] / "tests/infer/deeppot_sea.pth"
+        with tempfile.TemporaryDirectory() as temporary:
+            root = Path(temporary)
+            coordinate = root / "coord.xyz"
+            coordinate.write_text("1\nINIT protocol test\nO 0 0 0\n")
+            name = f"dpipi_init_{os.getpid()}_{root.name}"
+            address = Path("/tmp") / f"ipi_{name}"
+            config = root / "config.json"
+            config.write_text(
+                json.dumps(
+                    {
+                        "verbose": verbose,
+                        "use_unix": True,
+                        "port": 31415,
+                        "host": name,
+                        "graph_file": str(model),
+                        "coord_file": str(coordinate),
+                        "atom_type": {"O": 0},
+                    }
+                )
+            )
+            process = None
+            server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+            try:
+                server.bind(str(address))
+                server.listen(1)
+                server.settimeout(30)
+                process = subprocess.Popen(
+                    ["dp_ipi", str(config)],
+                    stdout=subprocess.PIPE,
+                    stderr=subprocess.PIPE,
+                    text=True,
+                )
+                connection, _ = server.accept()
+                with connection:
+                    connection.settimeout(10)
+                    size = len(payload) if length is None else length
+                    connection.sendall(b"INIT        " + struct.pack("=ii", 0, size))
+                    connection.sendall(payload)
+                    if size == len(payload):
+                        connection.sendall(b"STATUS      ")
+                        response = b""
+                        while len(response) < 12:
+                            part = connection.recv(12 - len(response))
+                            self.assertTrue(part, "driver closed before STATUS reply")
+                            response += part
+                        self.assertEqual(response, b"READY       ")
+                        connection.sendall(b"EXIT        ")
+                    elif size > len(payload):
+                        connection.shutdown(socket.SHUT_WR)
+                stdout, stderr = process.communicate(timeout=30)
+                self.assertEqual(
+                    process.returncode,
+                    0 if size == len(payload) else (1 if size < 0 else 255),
+                    msg=f"stdout:\n{stdout}\nstderr:\n{stderr}",
+                )
+                if size > len(payload):
+                    self.assertIn("Error reading from socket", stderr)
+                    self.assertNotIn(payload.decode(), stdout)
+                if size < 0:
+                    self.assertIn(
+                        "dp_ipi: INIT payload length must be nonnegative.", stderr
+                    )
+            finally:
+                if process is not None and process.poll() is None:
+                    process.kill()
+                    process.wait()
+                server.close()
+                address.unlink(missing_ok=True)
+
+    def test_empty_initialization_preserves_next_header(self) -> None:
+        self.exchange(b"")
+
+    def test_initialization_consumes_replica_id_and_payload(self) -> None:
+        self.exchange(b'{"temperature": 300}')
+
+    def test_initialization_accepts_payload_larger_than_read_buffer(self) -> None:
+        self.exchange(b"x" * 4097)
+
+    def test_negative_initialization_length_is_rejected(self) -> None:
+        self.exchange(b"", length=-1)
+
+    def test_truncated_verbose_initialization_is_rejected(self) -> None:
+        self.exchange(b"partial-init-marker", length=64, verbose=True)
diff --git a/source/ipi/tests/test_driver_input_validation.py b/source/ipi/tests/test_driver_input_validation.py
index 912626da4e..b8f2893558 100644
--- a/source/ipi/tests/test_driver_input_validation.py
+++ b/source/ipi/tests/test_driver_input_validation.py
@@ -50,10 +50,12 @@ def test_unknown_atom_name_is_rejected_before_model_loading(self) -> None:
             )
 
         self.assertEqual(result.returncode, 1, msg=result.stderr)
-        self.assertEqual(
-            result.stderr,
-            "dp_ipi: Unknown atom name 'Xx' in coordinate file: "
-            "no matching entry in atom_type.\n",
+        self.assertTrue(
+            result.stderr.endswith(
+                "dp_ipi: Unknown atom name 'Xx' in coordinate file: "
+                "no matching entry in atom_type.\n"
+            ),
+            msg=result.stderr,
         )
 
     def test_missing_atom_type_is_rejected_before_model_loading(self) -> None:
diff --git a/source/ipi/tests/test_driver_protocol_validation.py b/source/ipi/tests/test_driver_protocol_validation.py
new file mode 100644
index 0000000000..3cc26a2ab7
--- /dev/null
+++ b/source/ipi/tests/test_driver_protocol_validation.py
@@ -0,0 +1,93 @@
+# SPDX-License-Identifier: LGPL-3.0-or-later
+import json
+import os
+import socket
+import struct
+import subprocess
+import tempfile
+import unittest
+from pathlib import (
+    Path,
+)
+
+
+@unittest.skipIf(
+    os.environ.get("ENABLE_PYTORCH", "1") != "1",
+    reason="Skip test because PyTorch support is not enabled.",
+)
+class TestDPIPIProtocolValidation(unittest.TestCase):
+    def test_posdata_atom_count_must_match_configuration(self) -> None:
+        tests_path = Path(__file__).parent.parent.parent / "tests"
+        model_file = tests_path / "infer" / "deeppot_sea.pth"
+
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            tmp_path = Path(tmp_dir)
+            coord_file = tmp_path / "coord.xyz"
+            config_file = tmp_path / "config.json"
+            socket_name = f"dpipi_count_{os.getpid()}_{tmp_path.name}"
+            socket_path = Path("/tmp") / f"ipi_{socket_name}"
+            coord_file.write_text(
+                "1\nPOSDATA atom-count regression test\nO 0.0 0.0 0.0\n",
+                encoding="utf-8",
+            )
+            config_file.write_text(
+                json.dumps(
+                    {
+                        "verbose": False,
+                        "use_unix": True,
+                        "port": 31415,
+                        "host": socket_name,
+                        "graph_file": str(model_file),
+                        "coord_file": str(coord_file),
+                        "atom_type": {"O": 0},
+                    }
+                ),
+                encoding="utf-8",
+            )
+
+            server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+            process = None
+            try:
+                server.bind(str(socket_path))
+                server.listen(1)
+                server.settimeout(30)
+                process = subprocess.Popen(
+                    ["dp_ipi", str(config_file)],
+                    stdout=subprocess.PIPE,
+                    stderr=subprocess.PIPE,
+                    text=True,
+                )
+                try:
+                    connection, _ = server.accept()
+                except TimeoutError:
+                    if process.poll() is None:
+                        process.kill()
+                    stdout, stderr = process.communicate(timeout=5)
+                    self.fail(
+                        "dp_ipi did not connect within 30 seconds "
+                        f"(return code {process.returncode}).\n"
+                        f"stdout:\n{stdout}\nstderr:\n{stderr}"
+                    )
+                with connection:
+                    identity = (1.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 1.0)
+                    connection.sendall(b"POSDATA     ")
+                    connection.sendall(struct.pack("=9d", *identity))
+                    connection.sendall(struct.pack("=9d", *identity))
+                    connection.sendall(struct.pack("=i", 2))
+                stdout, stderr = process.communicate(timeout=60)
+                self.assertEqual(
+                    process.returncode,
+                    1,
+                    msg=f"stdout:\n{stdout}\nstderr:\n{stderr}",
+                )
+                self.assertIn(
+                    "dp_ipi: POSDATA atom count 2 does not match configured "
+                    "atom count 1.\n",
+                    stderr,
+                )
+            finally:
+                if process is not None and process.poll() is None:
+                    process.kill()
+                    process.wait()
+                server.close()
+                socket_path.unlink(missing_ok=True)
diff --git a/source/ipi/tests/test_sockets.c b/source/ipi/tests/test_sockets.c
index a649edf558..75638afd1b 100644
--- a/source/ipi/tests/test_sockets.c
+++ b/source/ipi/tests/test_sockets.c
@@ -63,6 +63,99 @@ static ssize_t overreporting_writer(int sockfd, const void* data, size_t len) {
   return (ssize_t)(len + 1);
 }
 
+static const char read_payload[] = "fragmented-read";
+static size_t read_offset;
+static int read_calls;
+static int read_failure;
+
+static ssize_t scripted_reader(int sockfd, void* data, size_t len) {
+  size_t count;
+  (void)sockfd;
+  ++read_calls;
+  // Interrupt both before any progress and between successful fragments.
+  if (read_calls == 1 || read_calls == 3) {
+    errno = EINTR;
+    return -1;
+  }
+  if (read_failure && read_offset == 3) {
+    if (read_failure == 1) {
+      return 0;
+    }
+    errno = EIO;
+    return -1;
+  }
+  count = len < 3 ? len : 3;
+  memcpy(data, read_payload + read_offset, count);
+  read_offset += count;
+  return (ssize_t)count;
+}
+
+static ssize_t overreporting_reader(int sockfd, void* data, size_t len) {
+  (void)sockfd;
+  (void)data;
+  return (ssize_t)(len + 1);
+}
+
+static int test_complete_reads(void) {
+  char data[sizeof(read_payload)];
+  int failure;
+  errno = 0;
+  CHECK(deepmd_read_all(123, data, 1, NULL) == -1);
+  CHECK(errno == EINVAL);
+  errno = 0;
+  CHECK(deepmd_read_all(123, NULL, 1, scripted_reader) == -1);
+  CHECK(errno == EINVAL);
+  read_calls = 0;
+  CHECK(deepmd_read_all(123, NULL, 0, scripted_reader) == 0);
+  CHECK(read_calls == 0);
+  for (failure = 0; failure < 3; ++failure) {
+    memset(data, '?', sizeof(data));
+    read_calls = 0;
+    read_offset = 0;
+    read_failure = failure;
+    errno = 0;
+    CHECK(deepmd_read_all(123, data, sizeof(data), scripted_reader) ==
+          (failure ? -1 : 0));
+    if (failure) {
+      CHECK(errno == (failure == 1 ? ECONNRESET : EIO));
+      CHECK(read_offset == 3);
+      CHECK(memcmp(data, read_payload, 3) == 0);
+      CHECK(data[3] == '?');
+    } else {
+      CHECK(memcmp(data, read_payload, sizeof(data)) == 0);
+      CHECK(read_calls > 3);
+    }
+  }
+  CHECK(deepmd_read_all(123, data, 1, overreporting_reader) == -1);
+  CHECK(errno == EIO);
+  return 0;
+}
+
+static int test_readbuffer_rejects_incomplete_frames(void) {
+  int length;
+  for (length = -1; length <= 8; length += 9) {
+    int channel[2], status;
+    pid_t child;
+    CHECK(socketpair(AF_UNIX, SOCK_STREAM, 0, channel) == 0);
+    CHECK(write(channel[0], "abc", 3) == 3);
+    CHECK(shutdown(channel[0], SHUT_WR) == 0);
+    child = fork();
+    CHECK(child >= 0);
+    if (child == 0) {
+      char data[8];
+      close(channel[0]);
+      readbuffer_(&channel[1], data, length);
+      _exit(0);
+    }
+    close(channel[0]);
+    close(channel[1]);
+    CHECK(waitpid(child, &status, 0) == child);
+    CHECK(WIFEXITED(status));
+    CHECK(WEXITSTATUS(status) == 255);
+  }
+  return 0;
+}
+
 static int test_unix_socket_path_bounds(void) {
   struct sockaddr_un address;
   const size_t prefix_length = strlen("/tmp/ipi_");
@@ -161,6 +254,10 @@ static int test_writebuffer_rejects_negative_length(void) {
 }
 
 int main(void) {
+  if (test_complete_reads() != 0 ||
+      test_readbuffer_rejects_incomplete_frames() != 0) {
+    return 1;
+  }
   if (test_unix_socket_path_bounds() != 0) {
     return 1;
   }