feat(p6-batch): residual-batch write/read driver + codec

Implements the client-only residual-batch feature end-to-end:
- src/shared/batch.{c,h}: self-contained single-file batch codec using the
  existing chunk_serialize/chunk_deserialize codec (byte-identical by
  construction).  Magic+format-version header (metadata mode is persisted into
  the header so a batch is self-describing across machines), length-prefixed
  chunk records, bounded reads that reject malformed/truncated/oversized
  records cleanly.
- src/client/client_send.c: write_batch_from_source (deterministic separate
  scan pass, loads every chunk's file images, emits header+records) and
  apply_batch_to_dest (local apply to a destination root via
  file_save_to_disk_full).  No wire change, no server involved.
- src/client/client_validation.c: --write-batch XOR --only-write-batch;
  --read-batch exclusive with both; --read-batch needs only a DEST,
  --only-write-batch only a SOURCE.
- src/client/client_cli.c: main() drives the three batch modes without
  connecting/transferring for read/only-write; --write-batch runs the live
  transfer (single-threaded so the config survives) then emits the batch.
- tests/test_batch.{c,h} (unit: byte-identical roundtrip with and without
  metadata; bad-magic/truncated/oversized rejection) + tests/integration/
  test_batch.py (only-write no-server, read-batch no-source roundtrip,
  --write-batch with a live transfer, conflict rejections).
- clang-format: realign PART-1 config.h comment block.

No PROTOCOL_VERSION bump, no config-frame field, no server flag.

src/shared/batch.c

@@ -0,0 +1,160 @@
+#include "batch.h"
+#include "data.h"
+#include "file.h"
+#include "file_receive.h"
+#include "log.h"
+#include <errno.h>
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h>
+
+/* Serialization metadata mode for the batch stream, captured from the config at
+ * batch_write_header time.  The header persists it into the file so a batch is
+ * self-describing: batch_read_apply re-reads it from the file (not from the
+ * reading config), so a batch written with -M is applied identically by an
+ * invoking process regardless of its own -M setting.  The batch driver is a
+ * single sequential scan pass within one thread, so this module-level flag is
+ * safe. */
+static bool batch_metadata_mode = false;
+
+static bool write_all_bytes(int fd, const void* data, size_t size) {
+  const unsigned char* p = (const unsigned char*)data;
+  size_t done = 0;
+  while (done < size) {
+    ssize_t n = write(fd, p + done, size - done);
+    if (n < 0 && errno == EINTR)
+      continue;
+    if (n <= 0)
+      return false;
+    done += (size_t)n;
+  }
+  return true;
+}
+
+bool batch_write_header(int fd, const Config* config) {
+  if (fd < 0)
+    return false;
+  batch_metadata_mode = config != NULL && config->use_metadata;
+  if (!write_all_bytes(fd, BATCH_MAGIC, BATCH_MAGIC_LEN))
+    return false;
+  unsigned char version = BATCH_FORMAT_VERSION;
+  if (!write_all_bytes(fd, &version, 1))
+    return false;
+  unsigned char mode = batch_metadata_mode ? 1 : 0;
+  return write_all_bytes(fd, &mode, 1);
+}
+
+bool batch_write_chunk(int fd, Chunk* chunk) {
+  if (fd < 0 || chunk == NULL)
+    return false;
+  Data* serialized = chunk_serialize(chunk, batch_metadata_mode);
+  if (serialized == NULL)
+    return false;
+  bool ok = false;
+  unsigned long long length = (unsigned long long)serialized->size;
+  if (length > BATCH_MAX_RECORD) {
+    log_message(LOG_LEVEL_ERROR, "batch: record size %llu exceeds the %llu-byte cap", length,
+                (unsigned long long)BATCH_MAX_RECORD);
+  } else if (write_all_bytes(fd, &length, sizeof(length)) &&
+             (length == 0 || write_all_bytes(fd, serialized->data, (size_t)length))) {
+    ok = true;
+  }
+  data_destroy(serialized);
+  return ok;
+}
+
+/* Read exactly `size` bytes.  Returns true on success.  On reaching EOF, sets
+ * *clean_eof only when no bytes had been read yet (a clean boundary) and returns
+ * that value, so a truncated record (EOF mid-read) yields false. */
+static bool read_exact(int fd, void* data, size_t size, bool* clean_eof) {
+  unsigned char* p = (unsigned char*)data;
+  size_t done = 0;
+  while (done < size) {
+    ssize_t n = read(fd, p + done, size - done);
+    if (n < 0 && errno == EINTR)
+      continue;
+    if (n == 0) {
+      if (clean_eof)
+        *clean_eof = done == 0;
+      return done == 0;
+    }
+    if (n < 0)
+      return false;
+    done += (size_t)n;
+  }
+  if (clean_eof)
+    *clean_eof = false;
+  return true;
+}
+
+int batch_read_apply(int fd, const Config* config, const char* dest_root) {
+  if (fd < 0 || dest_root == NULL || dest_root[0] == '\0')
+    return -1;
+
+  char magic[BATCH_MAGIC_LEN];
+  bool eof = false;
+  if (!read_exact(fd, magic, BATCH_MAGIC_LEN, &eof) || eof ||
+      memcmp(magic, BATCH_MAGIC, BATCH_MAGIC_LEN) != 0) {
+    log_message(LOG_LEVEL_ERROR, "batch: malformed header (bad magic)");
+    return -1;
+  }
+  unsigned char version;
+  if (!read_exact(fd, &version, 1, &eof) || eof || version != BATCH_FORMAT_VERSION) {
+    log_message(LOG_LEVEL_ERROR, "batch: malformed header (bad or missing format version)");
+    return -1;
+  }
+  unsigned char mode;
+  if (!read_exact(fd, &mode, 1, &eof) || eof || (mode != 0 && mode != 1)) {
+    log_message(LOG_LEVEL_ERROR, "batch: malformed header (bad metadata flag)");
+    return -1;
+  }
+  bool use_metadata = mode == 1;
+
+  while (1) {
+    unsigned long long length;
+    if (!read_exact(fd, &length, sizeof(length), &eof)) {
+      log_message(LOG_LEVEL_ERROR, "batch: truncated length prefix");
+      return -1;
+    }
+    if (eof)
+      break; /* clean end of stream */
+    if (length == 0 || length > BATCH_MAX_RECORD) {
+      log_message(LOG_LEVEL_ERROR, "batch: rejected record length %llu (valid range 1..%llu)",
+                  length, (unsigned long long)BATCH_MAX_RECORD);
+      return -1;
+    }
+    char* record = (char*)malloc((size_t)length);
+    if (record == NULL) {
+      log_message(LOG_LEVEL_ERROR, "batch: could not allocate a %llu-byte record", length);
+      return -1;
+    }
+    if (!read_exact(fd, record, (size_t)length, &eof)) {
+      log_message(LOG_LEVEL_ERROR, "batch: truncated chunk record");
+      free(record);
+      return -1;
+    }
+    Data* data = data_create(record, (size_t)length);
+    if (data == NULL)
+      return -1; /* data_create frees `record` on failure */
+    Chunk* chunk = chunk_deserialize(data, use_metadata);
+    data_destroy(data);
+    if (chunk == NULL) {
+      log_message(LOG_LEVEL_ERROR, "batch: rejected malformed chunk record");
+      return -1;
+    }
+    for (int i = 0; i < chunk->element_count; i++) {
+      File* file = chunk->items[i];
+      chunk->items[i] = NULL;
+      if (file == NULL)
+        continue;
+      FileSaveResult result = file_save_to_disk_full(dest_root, file, config);
+      file_destroy(file);
+      if (result == FILE_SAVE_ERROR) {
+        chunk_destroy(chunk);
+        return -1;
+      }
+    }
+    chunk_destroy(chunk);
+  }
+  return 0;
+}