[codex] Fix named pipe server partial writes

include/SimpleNamedPipe/NamedPipeServer.hpp

@@ -162,6 +162,7 @@ namespace SimpleNamedPipe {
         void process_write_commands();
         void process_write_commands(size_t index);
         void post_next_write(size_t index);
+        bool post_next_read(size_t index, HANDLE completion_port, OVERLAPPED* ov);
         void handle_write_completion(size_t index, size_t bytes_transferred);
         void handle_close(size_t index, HANDLE completion_port);
         void cleanup_pending_operations(const std::error_code& reason);

include/SimpleNamedPipe/NamedPipeServer/NamedPipeServer.ipp

@@ -301,13 +301,30 @@ namespace SimpleNamedPipe {
                 }
 
                 if (err == ERROR_BROKEN_PIPE ||
-                    err == ERROR_NO_DATA) {
+                    err == ERROR_NO_DATA ||
+                    err == ERROR_PIPE_NOT_CONNECTED) {
                     notify_disconnected(index, std::error_code(err, std::system_category()));
                     DisconnectNamedPipe(m_pipes[index]);
                     reconnect_client(index, completion_port, &m_read_overlapped[index]);
                     continue;
                 }
 
+                if (err == ERROR_MORE_DATA &&
+                    index < MAX_CLIENTS &&
+                    ov == &m_read_overlapped[index]) {
+                    if (bytes_transferred > 0) {
+                        m_message_buffers[index].append(
+                            m_read_buffers[index].data(),
+                            bytes_transferred);
+                    }
+                    if (!m_is_connected[index].load(std::memory_order_acquire)) {
+                        continue;
+                    }
+
+                    post_next_read(index, completion_port, &m_read_overlapped[index]);
+                    continue;
+                }
+
                 notify_error(std::error_code(static_cast<int>(err), std::system_category()));
                 continue;
             }
@@ -329,39 +346,73 @@ namespace SimpleNamedPipe {
             } else
             // Handle read completion
             if (ov == &m_read_overlapped[index] && bytes_transferred > 0) {
-                err = GetLastError();
                 m_message_buffers[index].append(m_read_buffers[index].data(), bytes_transferred);
-                if (err != ERROR_MORE_DATA) {
-                    notify_message(index);
-                }
+                notify_message(index);
             }
 
             // Skip reading if still not connected
             if (!m_is_connected[index].load(std::memory_order_acquire)) continue;
 
-            DWORD dummy = 0;
-            OVERLAPPED* new_ov = &m_read_overlapped[index];
-            BOOL result = ReadFile(m_pipes[index], m_read_buffers[index].data(), static_cast<DWORD>(m_read_buffers[index].size()), &dummy, new_ov);
+            post_next_read(index, completion_port, &m_read_overlapped[index]);
+        }
 
-            err = GetLastError();
-            if (!result && err != ERROR_IO_PENDING) {
-                if (err == ERROR_BROKEN_PIPE ||
-                    err == ERROR_NO_DATA) {
-                    notify_disconnected(index, std::error_code(static_cast<int>(err), std::system_category()));
-                    DisconnectNamedPipe(m_pipes[index]);
-                    reconnect_client(index, completion_port, new_ov);
-                    continue;
-                } else
-                if (err == ERROR_OPERATION_ABORTED) {
-                    continue;
-                } else {
-                    notify_error(std::error_code(static_cast<int>(err), std::system_category()));
-                    continue;
+        notify_stop(config);
+    }
+
+    SIMPLE_NAMED_PIPE_INLINE bool NamedPipeServer::post_next_read(
+            size_t index,
+            HANDLE completion_port,
+            OVERLAPPED* ov) {
+        if (index >= MAX_CLIENTS) {
+            notify_error(make_error_code(NamedPipeErrc::ClientIndexOutOfRange));
+            return false;
+        }
+        if (!m_is_connected[index].load(std::memory_order_acquire)) {
+            return true;
+        }
+
+        for (;;) {
+            memset(ov, 0, sizeof(OVERLAPPED));
+
+            DWORD bytes_read = 0;
+            BOOL result = ReadFile(
+                m_pipes[index],
+                m_read_buffers[index].data(),
+                static_cast<DWORD>(m_read_buffers[index].size()),
+                &bytes_read,
+                ov);
+
+            DWORD err = GetLastError();
+            if (result || err == ERROR_IO_PENDING) {
+                return true;
+            }
+
+            if (err == ERROR_MORE_DATA) {
+                if (bytes_read > 0) {
+                    m_message_buffers[index].append(
+                        m_read_buffers[index].data(),
+                        bytes_read);
                 }
+                continue;
             }
-        }
 
-        notify_stop(config);
+            if (err == ERROR_BROKEN_PIPE ||
+                err == ERROR_NO_DATA ||
+                err == ERROR_PIPE_NOT_CONNECTED) {
+                notify_disconnected(
+                    index,
+                    std::error_code(static_cast<int>(err), std::system_category()));
+                DisconnectNamedPipe(m_pipes[index]);
+                reconnect_client(index, completion_port, ov);
+                return false;
+            }
+            if (err == ERROR_OPERATION_ABORTED) {
+                return false;
+            }
+
+            notify_error(std::error_code(static_cast<int>(err), std::system_category()));
+            return false;
+        }
     }
 
     // Process all accumulated write commands
@@ -417,21 +468,21 @@ namespace SimpleNamedPipe {
             return;
         }
 
-        size_t buffer_size = m_write_buffers[index].size();
         size_t msg_offset = cmd.offset;
         size_t remaining = (msg_offset < cmd.message.size())
             ? (cmd.message.size() - msg_offset)
             : 0;
-        size_t bytes_to_copy = (std::min)(buffer_size, remaining);
-        auto& buffer = m_write_buffers[index];
-        buffer.assign(cmd.message.begin() + msg_offset, cmd.message.begin() + msg_offset + bytes_to_copy);
-        cmd.offset += bytes_to_copy;
 
         OVERLAPPED* ov = &m_write_overlapped[index];
         memset(ov, 0, sizeof(OVERLAPPED));
 
         DWORD bytes_written = 0;
-        BOOL success = WriteFile(m_pipes[index], buffer.data(), static_cast<DWORD>(bytes_to_copy), &bytes_written, ov);
+        BOOL success = WriteFile(
+            m_pipes[index],
+            cmd.message.data() + msg_offset,
+            static_cast<DWORD>(remaining),
+            &bytes_written,
+            ov);
         DWORD err = GetLastError();
         if (!success && err != ERROR_IO_PENDING) {
             if (cmd.on_done) cmd.on_done(std::error_code(static_cast<int>(err), std::system_category()));
@@ -460,6 +511,7 @@ namespace SimpleNamedPipe {
             }
             if (err == ERROR_BROKEN_PIPE ||
                 err == ERROR_NO_DATA ||
+                err == ERROR_PIPE_NOT_CONNECTED ||
                 err == ERROR_OPERATION_ABORTED) {
                 DisconnectNamedPipe(m_pipes[index]);
                 continue;

tests/client_server_test.cpp

@@ -266,6 +266,71 @@ void basic_io_edges(TestRunner& tr) {
     tr.expect(harness.error_count() == 0, with_error("server reported error: ", harness.error()));
 }
 
+void large_messages_cross_transport_buffers(TestRunner& tr) {
+    ServerHarness harness("large_buffers", 64, 25);
+    harness.start();
+    if (!tr.expect(harness.wait_started(), "server did not start")) return;
+
+    auto client = make_client(harness.pipe_name, 37, 3000);
+    std::error_code ec;
+    if (!tr.expect(client->connect(&ec), with_error("client connect failed: ", ec))) return;
+    tr.expect(harness.wait_connected_at_least(1), "server did not observe client connect");
+
+    std::string payload;
+    payload.reserve(8192);
+    for (int i = 0; i < 8192; ++i) {
+        payload.push_back(static_cast<char>('A' + (i % 26)));
+    }
+
+    tr.expect(client->write(payload, &ec), with_error("large message write failed: ", ec));
+    std::string response;
+    tr.expect(client->read(response, 3000, &ec), with_error("large message read failed: ", ec));
+    const std::string expected = "Echo: " + payload;
+    if (!tr.expect(response == expected, "large message echo bytes did not match")) {
+        std::cerr << "expected bytes=" << expected.size()
+                  << " actual bytes=" << response.size() << "\n";
+    }
+    tr.expect(harness.wait_messages_at_least(1), "server did not receive large message");
+
+    client->close();
+    tr.expect(harness.wait_disconnected_at_least(1), "server did not observe client close");
+    harness.stop();
+    tr.expect(harness.error_count() == 0, with_error("server reported error: ", harness.error()));
+}
+
+void adjacent_messages_are_not_merged(TestRunner& tr) {
+    ServerHarness harness("adjacent_messages", 1024, 25);
+    harness.start();
+    if (!tr.expect(harness.wait_started(), "server did not start")) return;
+
+    auto client = make_client(harness.pipe_name, 1024, 3000);
+    std::error_code ec;
+    if (!tr.expect(client->connect(&ec), with_error("client connect failed: ", ec))) return;
+    tr.expect(harness.wait_connected_at_least(1), "server did not observe client connect");
+
+    const std::string first = "first-json-rpc-request";
+    const std::string second = "second-json-rpc-request";
+    tr.expect(client->write(first, &ec), with_error("first adjacent write failed: ", ec));
+    tr.expect(client->write(second, &ec), with_error("second adjacent write failed: ", ec));
+
+    std::string first_response;
+    std::string second_response;
+    tr.expect(client->read(first_response, 3000, &ec),
+              with_error("first adjacent read failed: ", ec));
+    tr.expect(client->read(second_response, 3000, &ec),
+              with_error("second adjacent read failed: ", ec));
+    tr.expect(first_response == "Echo: " + first,
+              "first adjacent response should echo only first message");
+    tr.expect(second_response == "Echo: " + second,
+              "second adjacent response should echo only second message");
+    tr.expect(harness.wait_messages_at_least(2), "server did not receive two adjacent messages");
+
+    client->close();
+    tr.expect(harness.wait_disconnected_at_least(1), "server did not observe client close");
+    harness.stop();
+    tr.expect(harness.error_count() == 0, with_error("server reported error: ", harness.error()));
+}
+
 void repeated_connect_and_open_contract(TestRunner& tr) {
     ServerHarness harness("repeat", 1024, 25);
     harness.start();
@@ -446,6 +511,8 @@ int main() {
     TestRunner runner;
 
     runner.run("basic_io_edges", basic_io_edges);
+    runner.run("large_messages_cross_transport_buffers", large_messages_cross_transport_buffers);
+    runner.run("adjacent_messages_are_not_merged", adjacent_messages_are_not_merged);
     runner.run("repeated_connect_and_open_contract", repeated_connect_and_open_contract);
     runner.run("server_disconnect_notifies_client", server_disconnect_notifies_client);
     runner.run("churn_clients", churn_clients);

tests/stubs/windows.h

@@ -63,6 +63,7 @@ using LPOVERLAPPED = OVERLAPPED*;
 #define ERROR_OPERATION_ABORTED 995
 #define ERROR_MORE_DATA 234
 #define ERROR_BROKEN_PIPE 109
+#define ERROR_PIPE_NOT_CONNECTED 233
 #define ERROR_SUCCESS 0
 #define ERROR_NO_DATA 232
 #define INFINITE 0xFFFFFFFF