diff --git a/RSYNC_COMPAT.md b/RSYNC_COMPAT.md
index 5c5770b..c817767 100644
--- a/RSYNC_COMPAT.md
+++ b/RSYNC_COMPAT.md
@@ -259,12 +259,53 @@ why plain `--append` works on the normal atomic path, not only with `--inplace`.
 | `--super` | Receiver attempts super-user activities | ❌ Not Implemented | |
 | `--fake-super` | Store/recover privileged attrs via xattrs | ❌ Not Implemented | |
 | `--open-noatime` | Avoid changing access time when opening files | ❌ Not Implemented | |
-| `--numeric-ids` | Do not map uid/gid by name | ❌ Not Implemented | |
-| `--usermap=STRING` | Map usernames | ❌ Not Implemented | |
-| `--groupmap=STRING` | Map group names | ❌ Not Implemented | |
-| `--chown=USER:GROUP` | Map owner and group | ❌ Not Implemented | |
+| `--numeric-ids` | Do not map uid/gid by name | ✅ Implemented | Ownership is applied through FastSync's opt-in identity path (see the Phase-4 identity notes below). `--numeric-ids` is a mapping-policy modifier: when applying ownership it uses the transmitted numeric uid/gid directly, skipping the name lookup. Without an ownership-affecting option it is inert (FastSync only applies ownership when the user opts in). It does not need `-M` to be parsed, but ownership is only applied when metadata (hence the source uid/gid) is actually transmitted (see the notes) |
+| `--usermap=STRING` | Map usernames | ✅ Implemented | Opt-in ownership application. rsync subset implemented: comma-separated `FROM:TO` rules evaluated in order, first match wins; `FROM`/`TO` are group/user names (resolved on the SOURCE machine at parse time), `*` (FROM matches any id / TO = the receiving process's current euid), and an `@N` or bare `N` numeric id. Rules are carried over the wire as resolved numeric id pairs; the receiver applies a matching rule (else falls back to `--chown`, `--numeric-ids`, then a best-effort name lookup) via an fd-relative `fchown`. Malformed/unresolvable specs are rejected with a clear error, never a silent no-op. Implies metadata preservation so the source uid/gid travel. Only effective when the receiver can actually change ownership (root or membership); otherwise it warns and continues |
+| `--groupmap=STRING` | Map group names | ✅ Implemented | Same rsync subset and semantics as `--usermap` but for the group (gid) side and the group databases. See the Phase-4 identity notes |
+| `--chown=USER:GROUP` | Map owner and group | ✅ Implemented | Opt-in ownership override applied receiver-side. Forms: `USER:GROUP`, `USER` (owner only), `:GROUP` (group only); a `*` for USER/GROUP means the current/root user or group as appropriate; an `@N`/bare `N` numeric id is accepted. A `:` inside a name may be escaped as `\:`. Equivalent to a trailing `*:*` usermap+groupmap rule (so an explicit `--usermap`/`--groupmap` match wins). Malformed or unresolvable specs are clear parse errors. Implies metadata preservation. Only effective when the receiver has permission to chown; otherwise it warns and continues (rsync parity) |
 | `--copy-as=USER[:GROUP]` | Perform the copy as another user/group | ❌ Not Implemented | |
 
+**Phase-4 identity notes:** `--numeric-ids`, `--usermap`, `--groupmap`, and
+`--chown` are real. They introduce a **controlled, opt-in, privilege-gated**
+ownership-application path on the receiver: plain `-M`/`--preserve` still does
+NOT apply client-supplied ownership (FastSync's deliberate conservative
+default, byte-for-byte backward compatible); ownership is only attempted once a
+client explicitly requests an ownership-affecting option. Application goes
+through an fd-relative `fchown()` in the receiver's metadata-restore path (after
+the file is fully written, before timestamps are set), so it is confined and
+symlink-safe — never a path-based `chown`. When the receiver lacks permission
+(typically non-root, e.g. the CI `nobody` user) `EPERM`/`EACCES` is logged as a
+warning and the transfer CONTINUES with exit status success, matching rsync.
+A no-op default means existing transfers are unaffected.
+
+Resolution of the destination uid/gid on the receiver: a matching
+`--usermap`/`--groupmap` rule wins; else the matching `--chown` side; else, with
+`--numeric-ids`, the transmitted numeric id is used raw (no name lookup); else a
+best-effort name lookup on the receiver's own account databases (skipped when
+the transmitted id has no name present there). `--chown` enforces the receiver
+side and is validated at parse time (malformed specs are clear errors, never a
+silent no-op).
+
+Wire/version: the config frame gained `numeric_ids`, `chown_uid_set`,
+`chown_uid`, `chown_gid_set`, `chown_gid`, and the `usermap`/`groupmap` tables
+(count-delimited lists of resolved int32 FROM/TO id pairs), so
+`PROTOCOL_VERSION` was bumped **2.10.0 → 2.11.0** (peers must match). All new
+fields cross `config_send`/`config_receive` with full symmetry and are validated
+on receive (bounded map sizes below `MAX_IDENTITY_MAP`, ids `>=` the `-1`
+sentinels).
+
+Documented divergences from rsync: because FastSync transmits only numeric
+uid/gid (not names) on the wire, name-based values (`--usermap`/`--groupmap`
+names, `--chown` names) are resolved to numbers at CLI parse time against the
+**client (sender) machine's** account databases; this reproduces rsync's
+semantics on a shared-account source/destination and is documented for a
+genuinely different destination. The interesting named-value subset is
+supported (`*` FROM wildcard, `*` TO = current user, `@N`/bare-`N` numerics); a
+lone-`@` "use the FROM value unchanged" rsync form is not implemented. Also
+unlike rsync, plain `-M` never applies ownership and `--usermap`/`--groupmap`/
+`--chown` each imply metadata preservation so the source uid/gid actually travel
+(the flags only take effect where ownership is being preserved/applied).
+
 ## 9. Symlink Handling
 
 | Flag | Rsync Description | FastSync Status | Notes |
diff --git a/src/client/client_cli.c b/src/client/client_cli.c
index f258675..6652511 100644
--- a/src/client/client_cli.c
+++ b/src/client/client_cli.c
@@ -6,6 +6,7 @@
 #include "delta.h"
 #include "file_list.h"
 #include "filter.h"
+#include "identity.h"
 #include "log.h"
 #include "protocol.h"
 #include "transport_tcp.h"
@@ -530,6 +531,7 @@ static const OptionEntry OPTION_TABLE[] = {
     {"--from0", "-0", OPT_FLAG, offsetof(Config, from0)},
     {"--cvs-exclude", "-C", OPT_FLAG, offsetof(Config, cvs_exclude)},
     {"-F", NULL, OPT_FLAG, offsetof(Config, per_dir_filter)},
+    {"--numeric-ids", NULL, OPT_FLAG, offsetof(Config, numeric_ids)},
 };
 
 /* Only boolean options with no required argument are safe to negate. */
@@ -1108,6 +1110,42 @@ int parse_args(Config* config, int argc, char* argv[], int* positional_args,
       }
       if (set_basis_dest_option(config, BASIS_DEST_LINK, argv[++i], "--link-dest") != 0)
         return -1;
+    } else if (strncmp(argv[i], "--usermap=", 10) == 0) {
+      if (identity_parse_map(config, argv[i] + 10, false) != 0)
+        return -1;
+      config->use_metadata = true;
+    } else if (opt_is(argv[i], "--usermap", NULL)) {
+      if (i + 1 >= argc) {
+        log_message(LOG_LEVEL_ERROR, "missing argument for %s", argv[i]);
+        return -1;
+      }
+      if (identity_parse_map(config, argv[++i], false) != 0)
+        return -1;
+      config->use_metadata = true;
+    } else if (strncmp(argv[i], "--groupmap=", 11) == 0) {
+      if (identity_parse_map(config, argv[i] + 11, true) != 0)
+        return -1;
+      config->use_metadata = true;
+    } else if (opt_is(argv[i], "--groupmap", NULL)) {
+      if (i + 1 >= argc) {
+        log_message(LOG_LEVEL_ERROR, "missing argument for %s", argv[i]);
+        return -1;
+      }
+      if (identity_parse_map(config, argv[++i], true) != 0)
+        return -1;
+      config->use_metadata = true;
+    } else if (strncmp(argv[i], "--chown=", 8) == 0) {
+      if (identity_parse_chown(config, argv[i] + 8) != 0)
+        return -1;
+      config->use_metadata = true;
+    } else if (opt_is(argv[i], "--chown", NULL)) {
+      if (i + 1 >= argc) {
+        log_message(LOG_LEVEL_ERROR, "missing argument for %s", argv[i]);
+        return -1;
+      }
+      if (identity_parse_chown(config, argv[++i]) != 0)
+        return -1;
+      config->use_metadata = true;
     } else if (argv[i][0] == '-') {
       char* escaped = output_escape(argv[i], false);
       fprintf(stderr, "Unknown option: %s\n", escaped ? escaped : "<allocation failed>");
diff --git a/src/client/usage.c b/src/client/usage.c
index 6ad9f15..f9eae40 100644
--- a/src/client/usage.c
+++ b/src/client/usage.c
@@ -129,6 +129,19 @@ void print_usage(void) {
   printf("  -M, --preserve      Preserve file metadata\n");
   printf("  -E, --executability Preserve executable permission bits\n");
   printf("  --chmod <changes>   Modify transferred permissions (rsync syntax)\n");
+  printf("  --numeric-ids       Do not map uid/gid by name: use the source numeric\n");
+  printf("                      ids directly when applying ownership\n");
+  printf("  --usermap=MAP       Map usernames when applying ownership: comma-separated\n");
+  printf("                      FROM:TO rules, first match wins.  FROM/TO are names\n");
+  printf("                      (resolved on the source machine), * (match any /\n");
+  printf("                      current user), or @N numeric ids.  e.g. *:nobody\n");
+  printf("  --groupmap=MAP      Map group names when applying ownership (same syntax)\n");
+  printf("  --chown=USER:GROUP  Override the ownership of transferred files.  Forms:\n");
+  printf("                      USER:GROUP, USER (owner only), :GROUP (group only); a\n");
+  printf("                      value of * means the current/root user as appropriate.\n");
+  printf("                      Names resolve on the source machine; @N for numerics.\n");
+  printf("                      Note: -M is already FastSync's preserve flag; these use\n");
+  printf("                      long forms only.\n");
   printf("  --chunk-size <n>    Chunk size in bytes (default: %d)\n", DEFAULT_CHUNK_SIZE);
   printf("  --source-dir <path> Source directory\n");
   printf("  --dest-dir <path>   Destination directory\n");
diff --git a/src/server/server.c b/src/server/server.c
index b042ffb..7ca8afc 100644
--- a/src/server/server.c
+++ b/src/server/server.c
@@ -1,6 +1,7 @@
 #include "config.h"
 #include "delay_updates.h"
 #include "file.h"
+#include "identity.h"
 #include "log.h"
 #include "multiprocessing.h"
 #include "protocol.h"
@@ -210,6 +211,10 @@ void handler(int file_descriptor) {
       return;
     }
   }
+  /* Preserve the negotiated identity policy for the fd-relative ownership
+     apply path.  Each connection is its own forked process, so this
+     per-process snapshot never races another connection. */
+  identity_set_active(config);
   if (config->use_multithreading) {
     Queue* q = queue_create(100, file_destroy);
     if (q == NULL) {
@@ -225,6 +230,7 @@ void handler(int file_descriptor) {
       config_delete(config);
       close(file_descriptor);
       protocol_session_unbind();
+      identity_clear_active();
       return;
     }
     protocol_session_set_max_alloc(&context->session, config->max_alloc);
@@ -253,6 +259,7 @@ void handler(int file_descriptor) {
         thrd_join(writer, NULL);
       pipeline_context_receiver_destroy(context);
       protocol_session_unbind();
+      identity_clear_active();
       return;
     }
     int receiver_result;
@@ -303,6 +310,7 @@ void handler(int file_descriptor) {
     config_delete(config);
   }
   protocol_session_unbind();
+  identity_clear_active();
   close(file_descriptor);
 }
 
diff --git a/src/shared/config.c b/src/shared/config.c
index b4e51b1..16f3b34 100644
--- a/src/shared/config.c
+++ b/src/shared/config.c
@@ -3,6 +3,7 @@
 #include "delay_updates.h"
 #include "delta.h"
 #include "file_list.h"
+#include "identity.h"
 #include "log.h"
 #include "protocol.h"
 #include "utils.h"
@@ -136,6 +137,15 @@ static void config_set_defaults(Config* config) {
   config->skip_compress_suffixes = NULL;
   config->skip_compress_count = 0;
   config->skip_compress_set = false;
+  config->numeric_ids = false;
+  config->chown_uid_set = false;
+  config->chown_uid = 0;
+  config->chown_gid_set = false;
+  config->chown_gid = 0;
+  config->usermap = NULL;
+  config->usermap_count = 0;
+  config->groupmap = NULL;
+  config->groupmap_count = 0;
   config->delay_context = NULL;
 }
 
@@ -178,6 +188,7 @@ static bool validate_received_config(const Config* config) {
          valid_wire_bool(config->partial) && valid_wire_bool(config->delete_before) &&
          valid_wire_bool(config->checksum) && valid_wire_bool(config->eight_bit_output) &&
          checksum_algo_valid(config->checksum_algo) && config_has_valid_delete_timing(config) &&
+         identity_wire_valid(config) &&
          !(config->skip_compress_set && config->use_chunk_serialization) &&
          /* --append / --append-verify tail resume needs the per-file check,
             which chunk serialization -s disables: reject on the receiver too
@@ -379,6 +390,12 @@ void config_delete(Config* config) {
       free(config->skip_compress_suffixes[i]);
     free(config->skip_compress_suffixes);
   }
+  free(config->usermap);
+  config->usermap = NULL;
+  config->usermap_count = 0;
+  free(config->groupmap);
+  config->groupmap = NULL;
+  config->groupmap_count = 0;
   if (config->filters) {
     array_list_delete(config->filters);
   }
@@ -673,6 +690,60 @@ static bool receive_checksum_options(int fd, Config* c) {
   return receive_n_data(fd, &c->checksum_seed, sizeof(c->checksum_seed));
 }
 
+/* --numeric-ids / --usermap / --groupmap / --chown (identity mapping).  The
+ * receiver needs these to apply the ownership the client requested, so they
+ * cross the config frame.  Trailing fields; protocol 2.11.0. */
+static bool send_identity_map(int fd, const IdentityMap* map, int count) {
+  if (!send_int(fd, count))
+    return false;
+  for (int i = 0; i < count; i++) {
+    if (!send_int(fd, map[i].from) || !send_int(fd, map[i].to))
+      return false;
+  }
+  return true;
+}
+
+static bool send_identity_options(int fd, const Config* c) {
+  return send_int(fd, c->numeric_ids) && send_int(fd, c->chown_uid_set) &&
+         send_int(fd, c->chown_uid) && send_int(fd, c->chown_gid_set) &&
+         send_int(fd, c->chown_gid) && send_identity_map(fd, c->usermap, c->usermap_count) &&
+         send_identity_map(fd, c->groupmap, c->groupmap_count);
+}
+
+static bool receive_identity_map(int fd, int* pcount, IdentityMap** pmap) {
+  int count;
+  if (!receive_int(fd, &count) || count < 0 || count > MAX_IDENTITY_MAP)
+    return false;
+  if (count > 0) {
+    IdentityMap* map = calloc((size_t)count, sizeof(IdentityMap));
+    if (!map)
+      return false;
+    for (int i = 0; i < count; i++) {
+      if (!receive_int(fd, &map[i].from) || !receive_int(fd, &map[i].to)) {
+        free(map);
+        return false;
+      }
+    }
+    *pmap = map;
+  }
+  *pcount = count;
+  return true;
+}
+
+static bool receive_identity_options(int fd, Config* c) {
+  int numeric_ids;
+  if (!receive_int(fd, &numeric_ids) || !valid_wire_bool(numeric_ids))
+    return false;
+  c->numeric_ids = numeric_ids != 0;
+  if (!receive_wire_bool(fd, &c->chown_uid_set) || !receive_int(fd, &c->chown_uid) ||
+      !receive_wire_bool(fd, &c->chown_gid_set) || !receive_int(fd, &c->chown_gid))
+    return false;
+  if (c->chown_uid < IDENTITY_MATCH_ANY || c->chown_gid < IDENTITY_MATCH_ANY)
+    return false;
+  return receive_identity_map(fd, &c->usermap_count, &c->usermap) &&
+         receive_identity_map(fd, &c->groupmap_count, &c->groupmap);
+}
+
 bool config_send(int file_descriptor, const Config* config) {
   protocol_session_set_max_alloc(NULL, config->max_alloc);
   if (!send_core_fields(file_descriptor, config) || !send_delta_fields(file_descriptor, config) ||
@@ -680,7 +751,8 @@ bool config_send(int file_descriptor, const Config* config) {
       !send_selection_options(file_descriptor, config) ||
       !send_resume_options(file_descriptor, config) ||
       !send_basis_options(file_descriptor, config) || !send_fuzzy_option(file_descriptor, config) ||
-      !send_checksum_options(file_descriptor, config))
+      !send_checksum_options(file_descriptor, config) ||
+      !send_identity_options(file_descriptor, config))
     return false;
   Status status;
   if (!receive_status(file_descriptor, &status))
@@ -715,7 +787,8 @@ Config* config_receive(int file_descriptor) {
       !receive_resume_options(file_descriptor, config) ||
       !receive_basis_options(file_descriptor, config) ||
       !receive_fuzzy_option(file_descriptor, config) ||
-      !receive_checksum_options(file_descriptor, config))
+      !receive_checksum_options(file_descriptor, config) ||
+      !receive_identity_options(file_descriptor, config))
     goto error;
   if (config->compress_choice[0] != '\0' && strcmp(config->compress_choice, "zstd") != 0 &&
       strcmp(config->compress_choice, "none") != 0) {
diff --git a/src/shared/config.h b/src/shared/config.h
index 840f45f..7542e1c 100644
--- a/src/shared/config.h
+++ b/src/shared/config.h
@@ -29,6 +29,17 @@ typedef struct BasisDest {
   char* path; /* relative to the destination root (receiver-confined) */
 } BasisDest;
 
+/* One resolved FROM:TO identity-mapping rule (--usermap / --groupmap).  Both
+ * fields are numeric ids.  IDENTITY_MATCH_ANY (-1) in `from` is rsync's '*'
+ * wildcard (matches any transmitted id); IDENTITY_CURRENT (-1) in `to` makes
+ * the receiver resolve the receiving process's own current euid/egid at apply
+ * time.  Names are resolved to numbers at parse time on the client (see
+ * identity.h for the exact subset). */
+typedef struct {
+  int32_t from;
+  int32_t to;
+} IdentityMap;
+
 typedef struct Config {
   char* version;
   char* send_directory;
@@ -252,16 +263,44 @@ typedef struct Config {
   int skip_compress_count;
   bool skip_compress_set;
 
+  // Issue #131: Identity mapping.  These configure whether and how the receiver
+  // applies ownership when it is actually preserved/applied.  ALL of them cross
+  // the wire (protocol 2.11.0) so the receiver resolves and applies ownership
+  // with the exact policy the client requested.  Plain -M/--preserve still does
+  // NOT apply ownership (FastSync's deliberate conservative default); it is
+  // only attempted when at least one of these is set (see identity.h).
+  /* --numeric-ids: no name lookup, use the transmitted numeric ids raw. */
+  bool numeric_ids;
+  /* --chown USER (owner) override; IDENTITY_CURRENT = the receiver's euid. */
+  bool chown_uid_set;
+  int32_t chown_uid;
+  /* --chown :GROUP (group) override; IDENTITY_CURRENT = the receiver's egid. */
+  bool chown_gid_set;
+  int32_t chown_gid;
+  /* --usermap / --groupmap entries, in order (first match wins). */
+  IdentityMap* usermap;
+  int usermap_count;
+  IdentityMap* groupmap;
+  int groupmap_count;
+
   // Receiver-side runtime staging registry for --delay-updates.  Never sent
   // over the wire and never set on the sender side.
   DelayUpdatesContext* delay_context;
 } Config;
 
-#define PROTOCOL_VERSION "2.10.0"
+#define PROTOCOL_VERSION "2.11.0"
 #define DEFAULT_CHUNK_SIZE (10 * 1024 * 1024)
 /* Upper bound on total basis-dir entries (rsync caps --link-dest at 20). */
 #define MAX_BASIS_DIRS 64
 
+/* Identity-mapping sentinels and bounds (see identity.h for semantics).
+ * IDENTITY_MATCH_ANY is a usermap/groupmap FROM '*' (matches any id);
+ * IDENTITY_CURRENT is a chown / map TO '*' (resolve to the receiver's current
+ * euid/egid at apply time). */
+#define IDENTITY_MATCH_ANY (-1)
+#define IDENTITY_CURRENT (-1)
+#define MAX_IDENTITY_MAP 128
+
 Config* config_create(void);
 void config_delete(Config* config);
 bool config_send(int file_descriptor, const Config* config);
diff --git a/src/shared/identity.c b/src/shared/identity.c
new file mode 100644
index 0000000..db412a4
--- /dev/null
+++ b/src/shared/identity.c
@@ -0,0 +1,457 @@
+#include "identity.h"
+#include "log.h"
+#include "utils.h"
+#include <errno.h>
+#include <grp.h>
+#include <limits.h>
+#include <pwd.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/stat.h>
+#include <unistd.h>
+
+/* The active identity snapshot lives in a per-process global.  The TCP server
+ * forks one child process per connection, so a connection never shares this
+ * with another; within a connection the multithreaded receiver reads it without
+ * mutation.  This is what lets the fd-relative metadata path consult the
+ * negotiated policy without threading a Config through every write helper. */
+typedef struct {
+  bool numeric_ids;
+  bool chown_uid_set;
+  int32_t chown_uid;
+  bool chown_gid_set;
+  int32_t chown_gid;
+  IdentityMap* usermap;
+  int usermap_count;
+  IdentityMap* groupmap;
+  int groupmap_count;
+  bool set;
+} IdentityActive;
+
+static IdentityActive g_identity;
+
+static void identity_active_reset(void) {
+  free(g_identity.usermap);
+  free(g_identity.groupmap);
+  g_identity.usermap = NULL;
+  g_identity.groupmap = NULL;
+  g_identity.usermap_count = 0;
+  g_identity.groupmap_count = 0;
+  g_identity.numeric_ids = false;
+  g_identity.chown_uid_set = false;
+  g_identity.chown_uid = 0;
+  g_identity.chown_gid_set = false;
+  g_identity.chown_gid = 0;
+  g_identity.set = false;
+}
+
+void identity_clear_active(void) {
+  identity_active_reset();
+}
+
+void identity_set_active(const Config* config) {
+  identity_active_reset();
+  if (!config)
+    return;
+  g_identity.numeric_ids = config->numeric_ids;
+  g_identity.chown_uid_set = config->chown_uid_set;
+  g_identity.chown_uid = config->chown_uid;
+  g_identity.chown_gid_set = config->chown_gid_set;
+  g_identity.chown_gid = config->chown_gid;
+  if (config->usermap_count > 0) {
+    g_identity.usermap = calloc((size_t)config->usermap_count, sizeof(IdentityMap));
+    if (g_identity.usermap) {
+      memcpy(g_identity.usermap, config->usermap,
+             (size_t)config->usermap_count * sizeof(IdentityMap));
+      g_identity.usermap_count = config->usermap_count;
+    }
+  }
+  if (config->groupmap_count > 0) {
+    g_identity.groupmap = calloc((size_t)config->groupmap_count, sizeof(IdentityMap));
+    if (g_identity.groupmap) {
+      memcpy(g_identity.groupmap, config->groupmap,
+             (size_t)config->groupmap_count * sizeof(IdentityMap));
+      g_identity.groupmap_count = config->groupmap_count;
+    }
+  }
+  g_identity.set = true;
+  /* A root receiver would honor any client-supplied ownership request (a
+     --usermap/--groupmap/--chown, or raw ids under --numeric-ids).  Surface
+     that prominently; a privileged daemon applying arbitrary client ownership
+     is a deliberate, opt-in choice the operator should be aware of. */
+  if (geteuid() == 0)
+    log_message(LOG_LEVEL_WARNING,
+                "identity mapping active and running as root: client-supplied "
+                "ownership (usermap/groupmap/chown/numeric-ids) will be honored; "
+                "run the daemon as an unprivileged user unless intended");
+}
+
+bool identity_active_enabled(void) {
+  /* --numeric-ids alone is a policy modifier (how ids are resolved WHERE
+     ownership is otherwise preserved), not itself an ownership-application
+     trigger, so it is deliberately excluded from this set: standalone it stays
+     inert, matching its siblings only when combined with -M/--preserve. */
+  return g_identity.set &&
+         (g_identity.chown_uid_set || g_identity.chown_gid_set ||
+          g_identity.usermap_count > 0 || g_identity.groupmap_count > 0);
+}
+
+bool identity_wire_valid(const Config* config) {
+  if (!config)
+    return false;
+  if (config->usermap_count < 0 || config->usermap_count > MAX_IDENTITY_MAP ||
+      config->groupmap_count < 0 || config->groupmap_count > MAX_IDENTITY_MAP)
+    return false;
+  if (config->chown_uid_set && config->chown_uid < IDENTITY_MATCH_ANY)
+    return false;
+  if (config->chown_gid_set && config->chown_gid < IDENTITY_MATCH_ANY)
+    return false;
+  for (int i = 0; i < config->usermap_count; i++) {
+    if (config->usermap[i].from < IDENTITY_MATCH_ANY || config->usermap[i].to < IDENTITY_CURRENT)
+      return false;
+  }
+  for (int i = 0; i < config->groupmap_count; i++) {
+    if (config->groupmap[i].from < IDENTITY_MATCH_ANY || config->groupmap[i].to < IDENTITY_CURRENT)
+      return false;
+  }
+  return true;
+}
+
+/* ---- CLI-time name/number resolution ---- */
+
+/* Parse a single FROM/TO token into an int32 id.  Returns 0 on success, -1 on a
+ * malformed or unresolvable token.  When is_group, name lookups use the group
+ * database; otherwise the user database.  A `*` token returns IDENTITY_MATCH_ANY
+ * / IDENTITY_CURRENT (the same -1 value, disambiguated by the caller's
+ * position).  An `@`-prefixed or bare-decimal token is a numeric id. */
+static int identity_resolve_token(const char* token, bool is_group, int32_t* out) {
+  if (!token || *token == '\0')
+    return -1;
+  if (strcmp(token, "*") == 0) {
+    *out = IDENTITY_MATCH_ANY;
+    return 0;
+  }
+  const char* num = (token[0] == '@') ? token + 1 : token;
+  if (*num != '\0') {
+    bool all_digits = true;
+    for (const char* p = num; *p; p++)
+      if (*p < '0' || *p > '9')
+        all_digits = false;
+    if (all_digits) {
+      char* endptr = NULL;
+      errno = 0;
+      long val = strtol(num, &endptr, 10);
+      if (errno == 0 && endptr && *endptr == '\0' && val >= 0 && val <= INT32_MAX) {
+        *out = (int32_t)val;
+        return 0;
+      }
+      return -1;
+    }
+  }
+  /* A name (or a name-like numeric that failed strict numeric parse). */
+  if (is_group) {
+    struct group* gr = getgrnam(token);
+    if (!gr)
+      return -1;
+    *out = (int32_t)gr->gr_gid;
+    return 0;
+  }
+  struct passwd* pw = getpwnam(token);
+  if (!pw)
+    return -1;
+  *out = (int32_t)pw->pw_uid;
+  return 0;
+}
+
+static int identity_append_rule(IdentityMap** map, int* count, int32_t from, int32_t to) {
+  if (*count >= MAX_IDENTITY_MAP)
+    return -1;
+  IdentityMap* grown = realloc(*map, (size_t)(*count + 1) * sizeof(IdentityMap));
+  if (!grown)
+    return -1;
+  *map = grown;
+  (*map)[*count].from = from;
+  (*map)[*count].to = to;
+  (*count)++;
+  return 0;
+}
+
+int identity_parse_map(Config* config, const char* value, bool is_group) {
+  if (!config || !value || *value == '\0') {
+    log_message(LOG_LEVEL_ERROR, "%smap requires a value", is_group ? "--group" : "--user");
+    return -1;
+  }
+  char* list = str_dup(value);
+  if (!list)
+    return -1;
+  const char* optname = is_group ? "--groupmap" : "--usermap";
+  char* saveptr = NULL;
+  for (char* rule = strtok_r(list, ",", &saveptr); rule; rule = strtok_r(NULL, ",", &saveptr)) {
+    char* colon = strchr(rule, ':');
+    if (!colon || colon == rule) {
+      free(list);
+      log_message(LOG_LEVEL_ERROR, "%s rules must be FROM:TO (got '%s')", optname, rule);
+      return -1;
+    }
+    *colon = '\0';
+    char* from_token = rule;
+    char* to_token = colon + 1;
+    if (*to_token == '\0') {
+      free(list);
+      log_message(LOG_LEVEL_ERROR, "%s rule 'FROM:' is missing the TO value (got '%s')", optname,
+                  value);
+      return -1;
+    }
+    int32_t from_id, to_id;
+    if (identity_resolve_token(from_token, is_group, &from_id) != 0 ||
+        identity_resolve_token(to_token, is_group, &to_id) != 0) {
+      free(list);
+      log_message(LOG_LEVEL_ERROR,
+                  "%s could not resolve '%s' (name must exist on the source; use "
+                  "@N for a numeric id)",
+                  optname, value);
+      return -1;
+    }
+    if (identity_append_rule(is_group ? &config->groupmap : &config->usermap,
+                             is_group ? &config->groupmap_count : &config->usermap_count, from_id,
+                             to_id) != 0) {
+      free(list);
+      log_message(LOG_LEVEL_ERROR, "%s has too many rules (max %d)", optname, MAX_IDENTITY_MAP);
+      return -1;
+    }
+  }
+  free(list);
+  return 0;
+}
+
+/* Split --chown=USER:GROUP on the first UNESCAPED colon, honoring backslash
+ * escapes (a `\:` is a literal colon inside a name; a lone backslash before any
+ * other character is kept verbatim).  Both sides are returned as malloc'd
+ * strings (the absent side is NULL). */
+static int identity_split_chown(const char* value, char** puser, char** pgroup) {
+  size_t len = strlen(value);
+  char* user = malloc(len + 1);
+  char* group = malloc(len + 1);
+  if (!user || !group) {
+    free(user);
+    free(group);
+    return -1;
+  }
+  const char* p = value;
+  size_t ui = 0;
+  bool split_seen = false;
+  size_t gi = 0;
+  while (*p) {
+    if (*p == '\\' && p[1] == ':') {
+      /* an escaped colon: a literal ':' in the current side's name */
+      if (split_seen)
+        group[gi++] = ':';
+      else
+        user[ui++] = ':';
+      p += 2;
+      continue;
+    }
+    if (*p == ':') {
+      split_seen = true;
+      p++;
+      continue;
+    }
+    if (split_seen)
+      group[gi++] = *p;
+    else
+      user[ui++] = *p;
+    p++;
+  }
+  user[ui] = '\0';
+  group[gi] = '\0';
+  char* u = str_dup(user);
+  char* g = str_dup(group);
+  free(user);
+  free(group);
+  if (!u || !g) {
+    free(u);
+    free(g);
+    return -1;
+  }
+  *puser = u;
+  *pgroup = g;
+  return 0;
+}
+
+int identity_parse_chown(Config* config, const char* value) {
+  if (!config || !value || *value == '\0') {
+    log_message(LOG_LEVEL_ERROR, "--chown requires a value (USER:GROUP, USER, or :GROUP)");
+    return -1;
+  }
+  /* Reject more than one UNESCAPED colon (a name or group may not contain an
+   * unescaped ':' in the spec).  The scan is escape-aware: a `\:` is a literal
+   * colon inside a name, not a field separator. */
+  int colons = 0;
+  bool saw_colon = false;
+  const char* p = value;
+  while (*p) {
+    if (*p == '\\' && p[1] == ':') {
+      p += 2;
+      continue;
+    }
+    if (*p == ':') {
+      colons++;
+      saw_colon = true;
+    }
+    p++;
+  }
+  if (colons > 1) {
+    log_message(LOG_LEVEL_ERROR, "--chown must have at most one ':' (got '%s')", value);
+    return -1;
+  }
+
+  char *user = NULL, *group = NULL;
+  if (identity_split_chown(value, &user, &group) != 0) {
+    log_message(LOG_LEVEL_ERROR, "memory allocation failed for --chown");
+    return -1;
+  }
+  int ret = 0;
+  if (!saw_colon) {
+    /* --chown=USER: owner only. */
+    if (*user == '\0') {
+      log_message(LOG_LEVEL_ERROR, "--chown requires a user or group (got '%s')", value);
+      ret = -1;
+    } else if (identity_resolve_token(user, false, &config->chown_uid) != 0) {
+      log_message(LOG_LEVEL_ERROR,
+                  "--chown could not resolve user '%s' (use a name that exists "
+                  "on the source, '*', or @N)",
+                  value);
+      ret = -1;
+    } else {
+      config->chown_uid_set = true;
+    }
+  } else {
+    /* --chown=USER:GROUP, --chown=:GROUP, --chown=USER: */
+    if (*user != '\0') {
+      if (identity_resolve_token(user, false, &config->chown_uid) != 0) {
+        log_message(LOG_LEVEL_ERROR, "--chown could not resolve user '%s'", value);
+        ret = -1;
+        goto done;
+      }
+      config->chown_uid_set = true;
+    }
+    if (*group != '\0') {
+      if (identity_resolve_token(group, true, &config->chown_gid) != 0) {
+        log_message(LOG_LEVEL_ERROR, "--chown could not resolve group '%s'", value);
+        ret = -1;
+        goto done;
+      }
+      config->chown_gid_set = true;
+    }
+    if (!*user && !*group) {
+      log_message(LOG_LEVEL_ERROR, "--chown must set a user, a group, or both (got '%s')", value);
+      ret = -1;
+    }
+  }
+done:
+  free(user);
+  free(group);
+  return ret;
+}
+
+/* ---- Receiver-side ownership application ---- */
+
+static bool identity_map_lookup(const IdentityMap* map, int count, int32_t source_id,
+                                int32_t* out_to) {
+  for (int i = 0; i < count; i++) {
+    if (map[i].from == IDENTITY_MATCH_ANY || map[i].from == source_id) {
+      *out_to = map[i].to;
+      return true;
+    }
+  }
+  return false;
+}
+
+void identity_apply_ownership(int fd, int32_t source_uid, int32_t source_gid) {
+  /* Ownership application is OFF unless the client requested an identity flag.
+   * This is the controlled gate: a default (or plain -M) transfer never changes
+   * ownership, byte-for-byte preserving FastSync's existing behavior. */
+  if (!identity_active_enabled() || fd < 0)
+    return;
+  struct stat st;
+  if (fstat(fd, &st) != 0)
+    return;
+
+  bool set_uid = false;
+  bool set_gid = false;
+  uid_t uid = 0;
+  gid_t gid = 0;
+
+  int32_t target;
+  if (identity_map_lookup(g_identity.usermap, g_identity.usermap_count, source_uid, &target)) {
+    uid = target == IDENTITY_CURRENT ? geteuid() : (uid_t)target;
+    set_uid = true;
+  } else if (g_identity.chown_uid_set) {
+    uid = g_identity.chown_uid == IDENTITY_CURRENT ? geteuid() : (uid_t)g_identity.chown_uid;
+    set_uid = true;
+  } else if (g_identity.numeric_ids) {
+    uid = (uid_t)source_uid;
+    set_uid = true;
+  } else {
+    /* Best-effort name mapping against the receiver's own database: if the
+     * transmitted (numeric) id resolves to a name present on this machine,
+     * re-resolve it.  On a shared-account host this is the identity operation;
+     * when the id has no name here, the user side is left alone. */
+    struct passwd* pw = getpwuid((uid_t)source_uid);
+    if (pw) {
+      const struct passwd* mapped = getpwnam(pw->pw_name);
+      if (mapped) {
+        uid = mapped->pw_uid;
+        set_uid = true;
+      }
+    }
+  }
+
+  if (identity_map_lookup(g_identity.groupmap, g_identity.groupmap_count, source_gid, &target)) {
+    gid = target == IDENTITY_CURRENT ? getegid() : (gid_t)target;
+    set_gid = true;
+  } else if (g_identity.chown_gid_set) {
+    gid = g_identity.chown_gid == IDENTITY_CURRENT ? getegid() : (gid_t)g_identity.chown_gid;
+    set_gid = true;
+  } else if (g_identity.numeric_ids) {
+    gid = (gid_t)source_gid;
+    set_gid = true;
+  } else {
+    struct group* gr = getgrgid((gid_t)source_gid);
+    if (gr) {
+      const struct group* mapped = getgrnam(gr->gr_name);
+      if (mapped) {
+        gid = mapped->gr_gid;
+        set_gid = true;
+      }
+    }
+  }
+
+  if (!set_uid && !set_gid)
+    return;
+  /* An unset side keeps the file's current id so the other side can change. */
+  if (!set_uid)
+    uid = st.st_uid;
+  if (!set_gid)
+    gid = st.st_gid;
+
+  /* Only call fchown when the target differs (avoid needless syscalls and any
+   * chance of clearing setuid/setgid on an already-correct file). */
+  if (st.st_uid == uid && st.st_gid == gid)
+    return;
+
+  if (fchown(fd, uid, gid) != 0) {
+    /* EPERM/EACCES are expected when the receiver is not privileged (e.g. the
+     * CI `nobody` user): warn and continue, never abort the transfer.  Any
+     * other error (EIO/EROFS/ENOSPC/...) is a real failure and must not be
+     * silently downgraded to a warning. */
+    if (errno == EPERM || errno == EACCES)
+      log_message(LOG_LEVEL_WARNING,
+                  "could not apply ownership (uid=%ld gid=%ld): %s; leaving as-is", (long)uid,
+                  (long)gid, strerror(errno));
+    else
+      log_message(LOG_LEVEL_ERROR, "failed to apply ownership (uid=%ld gid=%ld): %s", (long)uid,
+                  (long)gid, strerror(errno));
+  }
+}
\ No newline at end of file
diff --git a/src/shared/identity.h b/src/shared/identity.h
new file mode 100644
index 0000000..9143770
--- /dev/null
+++ b/src/shared/identity.h
@@ -0,0 +1,61 @@
+#ifndef IDENTITY_H
+#define IDENTITY_H
+
+#include "config.h"
+#include <stdbool.h>
+#include <stdint.h>
+#include <sys/types.h>
+
+/*
+ * Identity mapping: --numeric-ids / --usermap / --groupmap / --chown.
+ *
+ * FastSync transmits uid/gid numerically (int32 on the wire) and, by design,
+ * NEVER applies client-supplied ownership unless a user explicitly opts in with
+ * an identity flag below.  This module is the controlled, opt-in,
+ * privilege-gated path for applying ownership on the receiver: the wire config
+ * is snapshotted once per connection via identity_set_active() and applied
+ * through an fd-relative fchown() in the receiver's metadata-restore path.
+ *
+ * Because only numeric ids cross the wire, name-based values are resolved to
+ * numbers at CLI parse time using the CLIENT (sender) machine's databases.  On
+ * a shared-account source/destination this reproduces rsync's semantics; a
+ * genuinely different destination database is a documented divergence (see
+ * RSYNC_COMPAT.md).
+ */
+
+/* Parse one --usermap= / --groupmap= value (comma-separated FROM:TO rules,
+ * first match wins) into config->usermap / config->groupmap.  is_group selects
+ * the group tables and name databases.  Returns 0 on success, -1 on a
+ * malformed spec or an unresolvable name (never a silent no-op). */
+int identity_parse_map(Config* config, const char* value, bool is_group);
+
+/* Parse --chown=USER:GROUP.  Supports USER:GROUP, USER (owner only), :GROUP
+ * (group only), '*' (current/root as appropriate) and numeric ids.  Returns 0
+ * on success, -1 on a malformed spec / unresolvable name. */
+int identity_parse_chown(Config* config, const char* value);
+
+/* Receiver-side snapshot of the negotiated identity config.  The server calls
+ * identity_set_active() once per connection (before any file write) using the
+ * config received over the wire; the snapshot is a deep copy so the caller may
+ * free its Config immediately.  identity_clear_active() releases it. */
+void identity_set_active(const Config* config);
+void identity_clear_active(void);
+
+/* True when any ownership-affecting identity option is present in the active
+ * snapshot.  Ownership stays OFF ("do not apply") for every transfer that
+ * requests none of them, preserving FastSync's existing behavior. */
+bool identity_active_enabled(void);
+
+/* Apply the negotiated ownership to an already-written file descriptor.
+ * source_uid/source_gid are the transmitted numeric ids.  Resolution order:
+ * a matching usermap/groupmap rule, then --chown, then --numeric-ids (raw),
+ * then a best-effort name lookup on the receiver's own databases (skipped when
+ * the transmitted id has no name on this system).  Only calls fchown() when the
+ * result differs from the current value; EPERM/EACCES are logged and ignored,
+ * never fatal (rsync parity: the transfer must not abort). */
+void identity_apply_ownership(int fd, int32_t source_uid, int32_t source_gid);
+
+/* Receiver-side wire validation of the resolved identity fields. */
+bool identity_wire_valid(const Config* config);
+
+#endif
\ No newline at end of file
diff --git a/src/shared/metadata.c b/src/shared/metadata.c
index 80d50cb..94ef772 100644
--- a/src/shared/metadata.c
+++ b/src/shared/metadata.c
@@ -1,5 +1,6 @@
 #include "metadata.h"
 #include "file.h"
+#include "identity.h"
 #include "log.h"
 #include "protocol.h"
 #include "utils.h"
@@ -249,7 +250,15 @@ bool file_restore_metadata_fd(int fd, const FileMetadata* metadata, bool preserv
   mode_t safe_mode = metadata_mode(metadata, current.st_mode, preserve_executability);
   if (fchmod(fd, safe_mode) != 0)
     ok = false;
-  /* Client uid/gid values are deliberately not authoritative. */
+  /* Client uid/gid values are deliberately not authoritative UNLESS the client
+     explicitly opted in with an identity flag (--numeric-ids / --usermap /
+     --groupmap / --chown).  identity_apply_ownership is the controlled,
+     privilege-gated path: it consults the negotiated policy, resolves the
+     target ids, and applies them via an fd-relative fchown() that is confined
+     to the just-written file (EPERM/EACCES are logged, never fatal).  With no
+     identity flag set it is a no-op, so a default or plain -M transfer keeps
+     FastSync's existing behavior of never applying client ownership. */
+  identity_apply_ownership(fd, (int32_t)metadata->uid, (int32_t)metadata->gid);
   struct timespec times[2] = {{.tv_sec = 0, .tv_nsec = UTIME_OMIT},
                               {.tv_sec = metadata->mtime_sec, .tv_nsec = metadata->mtime_nsec}};
   if (futimens(fd, times) != 0)
diff --git a/tests/integration/test_features.py b/tests/integration/test_features.py
index 4d8aedb..eec27ff 100644
--- a/tests/integration/test_features.py
+++ b/tests/integration/test_features.py
@@ -3695,3 +3695,89 @@ class TestFuzzy:
         assert proxy.client_to_server > len(new_bytes) // 2, \
             "--no-fuzzy should leave the default whole-file behavior intact"
 
+
+
+class TestIdentityMapping:
+    """Ownership-application flags (--numeric-ids / --usermap / --groupmap /
+    --chown).  In CI the receiver usually runs unprivileged, so ownership apply
+    is expected to fail from lack of privilege: the transfer must STILL succeed
+    and exit 0 (the receiver warns and continues, rsync parity).  The only
+    assertion that requires the ownership to actually change is gated on
+    os.geteuid() == 0 so it is skipped (not failed) as a non-root user."""
+
+    def test_numeric_ids_transfer_succeeds_unprivileged(self, shared_server):
+        source = os.path.join(TEST_DATA_DIR, "identity_num_source")
+        dest = os.path.join(TEST_DATA_DIR, "identity_num_dest")
+        clean_dir(source)
+        clean_dir(dest)
+        with open(os.path.join(source, "f.txt"), "wb") as f:
+            f.write(b"hello identity")
+        result, _ = run_client(source, dest,
+                               flags=["-M", "--numeric-ids"],
+                               port=shared_server.port)
+        assert result.returncode == 0, \
+            f"exit {result.returncode}: {(result.stderr or '')[:200]}"
+        received = get_dest_received_dir(dest, source)
+        with open(os.path.join(received, "f.txt"), "rb") as f:
+            assert f.read() == b"hello identity"
+
+    def test_usermap_and_groupmap_and_chown_succeed_unprivileged(self, shared_server):
+        source = os.path.join(TEST_DATA_DIR, "identity_map_source")
+        dest = os.path.join(TEST_DATA_DIR, "identity_map_dest")
+        clean_dir(source)
+        clean_dir(dest)
+        with open(os.path.join(source, "f.txt"), "wb") as f:
+            f.write(b"mapped")
+        result, _ = run_client(
+            source, dest,
+            flags=["-M", "--usermap=@1000:@1001", "--groupmap=@100:@101", "--chown=@2000:@2001"],
+            port=shared_server.port)
+        assert result.returncode == 0, \
+            f"exit {result.returncode}: {(result.stderr or '')[:200]}"
+        received = get_dest_received_dir(dest, source)
+        with open(os.path.join(received, "f.txt"), "rb") as f:
+            assert f.read() == b"mapped"
+
+    @pytest.mark.skipif(os.geteuid() != 0, reason="only root can change ownership")
+    def test_numeric_ids_applies_ownership_as_root(self, shared_server):
+        source = os.path.join(TEST_DATA_DIR, "identity_root_source")
+        dest = os.path.join(TEST_DATA_DIR, "identity_root_dest")
+        clean_dir(source)
+        clean_dir(dest)
+        src_file = os.path.join(source, "f.txt")
+        with open(src_file, "wb") as f:
+            f.write(b"owner")
+        os.chown(src_file, 12345, 12346)
+        result, _ = run_client(source, dest,
+                               flags=["-M", "--numeric-ids"],
+                               port=shared_server.port)
+        assert result.returncode == 0, \
+            f"exit {result.returncode}: {(result.stderr or '')[:200]}"
+        received = get_dest_received_dir(dest, source)
+        dst_file = os.path.join(received, "f.txt")
+        assert os.path.exists(dst_file)
+        st = os.stat(dst_file)
+        assert st.st_uid == 12345 and st.st_gid == 12346, \
+            f"owner not applied: uid={st.st_uid} gid={st.st_gid}"
+
+    @pytest.mark.skipif(os.geteuid() != 0, reason="only root can change ownership")
+    def test_chown_overrides_ownership_as_root(self, shared_server):
+        source = os.path.join(TEST_DATA_DIR, "identity_chown_root_source")
+        dest = os.path.join(TEST_DATA_DIR, "identity_chown_root_dest")
+        clean_dir(source)
+        clean_dir(dest)
+        src_file = os.path.join(source, "f.txt")
+        with open(src_file, "wb") as f:
+            f.write(b"root chown")
+        os.chown(src_file, 1, 1)
+        result, _ = run_client(source, dest,
+                               flags=["-M", "--chown=@12345:@54321"],
+                               port=shared_server.port)
+        assert result.returncode == 0, \
+            f"exit {result.returncode}: {(result.stderr or '')[:200]}"
+        received = get_dest_received_dir(dest, source)
+        dst_file = os.path.join(received, "f.txt")
+        assert os.path.exists(dst_file)
+        st = os.stat(dst_file)
+        assert st.st_uid == 12345 and st.st_gid == 54321, \
+            f"--chown not applied: uid={st.st_uid} gid={st.st_gid}"
diff --git a/tests/test_client_cli.c b/tests/test_client_cli.c
index ae3d00a..56928e0 100644
--- a/tests/test_client_cli.c
+++ b/tests/test_client_cli.c
@@ -7,6 +7,8 @@
 #include "log.h"
 #include "test_utils.h"
 #include "utils.h"
+#include <pwd.h>
+#include <stdint.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
@@ -2083,8 +2085,169 @@ static void test_validate_config_append_verify_rejects_whole_file() {
   EXPECT_FALSE(validate_config(cfg));
   config_delete(cfg);
 }
+/* --numeric-ids is a plain boolean flag. */
+static void test_parse_args_numeric_ids() {
+  Config* cfg = config_create();
+  char* argv[] = {"fastsync", "--numeric-ids", "/src", "/dst"};
+  int positional_args[2];
+  int positional_count = 0;
+  EXPECT_EQ_INT(parse_args(cfg, 4, argv, positional_args, &positional_count), 0);
+  EXPECT_TRUE(cfg->numeric_ids);
+  config_delete(cfg);
+}
+
+/* --usermap / --groupmap resolve an rsync subset into numeric FROM:TO pairs and
+ * imply metadata preservation (so the source uid/gid travel on the wire). */
+static void test_parse_args_usermap() {
+  Config* cfg = config_create();
+  char* argv[] = {"fastsync", "--usermap=@1000:@1001", "/src", "/dst"};
+  int positional_args[2];
+  int positional_count = 0;
+  EXPECT_EQ_INT(parse_args(cfg, 4, argv, positional_args, &positional_count), 0);
+  EXPECT_TRUE(cfg->use_metadata);
+  EXPECT_EQ_INT(cfg->usermap_count, 1);
+  EXPECT_EQ_INT(cfg->usermap[0].from, 1000);
+  EXPECT_EQ_INT(cfg->usermap[0].to, 1001);
+  config_delete(cfg);
+
+  /* Space form, multiple rules, comma-separated. */
+  cfg = config_create();
+  positional_count = 0;
+  char* argv2[] = {"fastsync", "--usermap", "@1:@2,@3:@4", "/src", "/dst"};
+  EXPECT_EQ_INT(parse_args(cfg, 5, argv2, positional_args, &positional_count), 0);
+  EXPECT_EQ_INT(cfg->usermap_count, 2);
+  EXPECT_EQ_INT(cfg->usermap[0].from, 1);
+  EXPECT_EQ_INT(cfg->usermap[0].to, 2);
+  EXPECT_EQ_INT(cfg->usermap[1].from, 3);
+  EXPECT_EQ_INT(cfg->usermap[1].to, 4);
+  config_delete(cfg);
+
+  /* '*' FROM means match any id; '*' TO means current user. */
+  cfg = config_create();
+  positional_count = 0;
+  char* argv3[] = {"fastsync", "--usermap=*:@2000", "/src", "/dst"};
+  EXPECT_EQ_INT(parse_args(cfg, 4, argv3, positional_args, &positional_count), 0);
+  EXPECT_EQ_INT(cfg->usermap[0].from, IDENTITY_MATCH_ANY);
+  EXPECT_EQ_INT(cfg->usermap[0].to, 2000);
+  config_delete(cfg);
+}
+
+static void test_parse_args_groupmap() {
+  Config* cfg = config_create();
+  char* argv[] = {"fastsync", "--groupmap=@100:@101", "/src", "/dst"};
+  int positional_args[2];
+  int positional_count = 0;
+  EXPECT_EQ_INT(parse_args(cfg, 4, argv, positional_args, &positional_count), 0);
+  EXPECT_TRUE(cfg->use_metadata);
+  EXPECT_EQ_INT(cfg->groupmap_count, 1);
+  EXPECT_EQ_INT(cfg->groupmap[0].from, 100);
+  EXPECT_EQ_INT(cfg->groupmap[0].to, 101);
+  config_delete(cfg);
+}
+
+/* A name in a map can be resolved to a number via the local user database. */
+static void test_parse_args_usermap_name_resolution() {
+  struct passwd* self = getpwuid(geteuid());
+  if (!self)
+    return; /* cannot construct a resolvable name deterministically */
+  char map_value[128];
+  snprintf(map_value, sizeof(map_value), "%s:@0", self->pw_name);
+  Config* cfg = config_create();
+  int positional_args[2];
+  int positional_count = 0;
+  char* argv[] = {"fastsync", (char*)"--usermap", map_value, "/src", "/dst"};
+  EXPECT_EQ_INT(parse_args(cfg, 5, argv, positional_args, &positional_count), 0);
+  EXPECT_EQ_INT(cfg->usermap_count, 1);
+  EXPECT_EQ_INT(cfg->usermap[0].from, (int32_t)self->pw_uid);
+  config_delete(cfg);
+}
+
+/* --chown parses USER:GROUP / USER / :GROUP, numeric ids, and '*'. */
+static void test_parse_args_chown() {
+  Config* cfg = config_create();
+  char* argv[] = {"fastsync", "--chown=@1000:@1001", "/src", "/dst"};
+  int positional_args[2];
+  int positional_count = 0;
+  EXPECT_EQ_INT(parse_args(cfg, 4, argv, positional_args, &positional_count), 0);
+  EXPECT_TRUE(cfg->use_metadata);
+  EXPECT_TRUE(cfg->chown_uid_set);
+  EXPECT_EQ_INT(cfg->chown_uid, 1000);
+  EXPECT_TRUE(cfg->chown_gid_set);
+  EXPECT_EQ_INT(cfg->chown_gid, 1001);
+  config_delete(cfg);
+
+  /* --chown=:GROUP sets only the group. */
+  cfg = config_create();
+  positional_count = 0;
+  char* argv2[] = {"fastsync", "--chown=:@1001", "/src", "/dst"};
+  EXPECT_EQ_INT(parse_args(cfg, 4, argv2, positional_args, &positional_count), 0);
+  EXPECT_FALSE(cfg->chown_uid_set);
+  EXPECT_TRUE(cfg->chown_gid_set);
+  EXPECT_EQ_INT(cfg->chown_gid, 1001);
+  config_delete(cfg);
+
+  /* --chown=USER sets only the owner. */
+  cfg = config_create();
+  positional_count = 0;
+  char* argv3[] = {"fastsync", "--chown=@1000", "/src", "/dst"};
+  EXPECT_EQ_INT(parse_args(cfg, 4, argv3, positional_args, &positional_count), 0);
+  EXPECT_TRUE(cfg->chown_uid_set);
+  EXPECT_EQ_INT(cfg->chown_uid, 1000);
+  EXPECT_FALSE(cfg->chown_gid_set);
+  config_delete(cfg);
+
+  /* '*' means current user/group. */
+  cfg = config_create();
+  positional_count = 0;
+  char* argv4[] = {"fastsync", "--chown=*:*", "/src", "/dst"};
+  EXPECT_EQ_INT(parse_args(cfg, 4, argv4, positional_args, &positional_count), 0);
+  EXPECT_TRUE(cfg->chown_uid_set);
+  EXPECT_EQ_INT(cfg->chown_uid, IDENTITY_CURRENT);
+  EXPECT_TRUE(cfg->chown_gid_set);
+  EXPECT_EQ_INT(cfg->chown_gid, IDENTITY_CURRENT);
+  config_delete(cfg);
+}
+
+/* Malformed identity specs are rejected, never silently ignored. */
+static void test_parse_args_rejects_malformed_identity() {
+  struct {
+    const char* opt;
+    const char* val;
+  } bad[] = {
+      {"--usermap", "@1000"},
+      {"--usermap", ":1000"},
+      {"--usermap", "definitely_not_a_real_user_zzz:@1"},
+      {"--groupmap", "@1"},
+      {"--groupmap", "no_such_group_qqq:x"},
+      {"--chown", "a:b:c"},
+      {"--chown", "no_such_user_zzz:"},
+  };
+  for (size_t i = 0; i < sizeof(bad) / sizeof(bad[0]); i++) {
+    Config* cfg = config_create();
+    char* argv[] = {"fastsync", (char*)bad[i].opt, (char*)bad[i].val, "/src", "/dst"};
+    int positional_args[2];
+    int positional_count = 0;
+    EXPECT_EQ_INT(parse_args(cfg, 5, argv, positional_args, &positional_count), -1);
+    config_delete(cfg);
+  }
+
+  /* An option with a missing value fails at the CLI layer. */
+  Config* cfg = config_create();
+  char* argv[] = {"fastsync", "--chown"};
+  int positional_args[2];
+  int positional_count = 0;
+  EXPECT_EQ_INT(parse_args(cfg, 2, argv, positional_args, &positional_count), -1);
+  config_delete(cfg);
+}
+
 void test_client_cli() {
   test_validate_config_required_paths();
+  test_parse_args_numeric_ids();
+  test_parse_args_usermap();
+  test_parse_args_groupmap();
+  test_parse_args_usermap_name_resolution();
+  test_parse_args_chown();
+  test_parse_args_rejects_malformed_identity();
   test_parse_args_append();
   test_parse_args_append_verify();
   test_parse_args_append_both();
diff --git a/tests/test_config.c b/tests/test_config.c
index 2dc6c2b..255420e 100644
--- a/tests/test_config.c
+++ b/tests/test_config.c
@@ -908,6 +908,102 @@ static void test_config_receive_rejects_invalid_checksum_algo() {
   EXPECT_FALSE(roundtrip_config_ok(c));
   config_delete(c);
 }
+/* The identity-mapping fields (--numeric-ids / --usermap / --groupmap /
+   --chown) cross the config wire unchanged: the receiver needs them to apply
+   ownership with the same policy the client requested. */
+static void test_config_identity_wire_roundtrip() {
+  if (is_running_under_valgrind())
+    return;
+  Config* send_cfg = config_create();
+  EXPECT_NOT_NULL(send_cfg);
+  send_cfg->send_directory = str_dup("/send/src");
+  send_cfg->receive_root_directory = str_dup("/send/dst");
+  send_cfg->numeric_ids = true;
+  send_cfg->chown_uid_set = true;
+  send_cfg->chown_uid = 1001;
+  send_cfg->chown_gid_set = true;
+  send_cfg->chown_gid = IDENTITY_CURRENT;
+  send_cfg->usermap_count = 2;
+  send_cfg->usermap = calloc(2, sizeof(IdentityMap));
+  send_cfg->usermap[0].from = IDENTITY_MATCH_ANY;
+  send_cfg->usermap[0].to = 65534;
+  send_cfg->usermap[1].from = 1000;
+  send_cfg->usermap[1].to = 1000;
+  send_cfg->groupmap_count = 1;
+  send_cfg->groupmap = calloc(1, sizeof(IdentityMap));
+  send_cfg->groupmap[0].from = 0;
+  send_cfg->groupmap[0].to = IDENTITY_CURRENT;
+
+  int p[2];
+  EXPECT_EQ_INT(socketpair(AF_UNIX, SOCK_STREAM, 0, p), 0);
+  io_set_fds(p[0], p[1]);
+  io_set_bwlimit(0);
+  pid_t pid = fork();
+  if (pid == 0) {
+    close(p[1]);
+    io_set_fds(p[0], p[0]);
+    Config* recv = config_receive(p[0]);
+    bool ok = recv != NULL;
+    if (ok) {
+      ok = recv->numeric_ids && recv->chown_uid_set && recv->chown_uid == 1001 &&
+           recv->chown_gid_set && recv->chown_gid == IDENTITY_CURRENT && recv->usermap_count == 2 &&
+           recv->groupmap_count == 1 && recv->usermap[0].from == IDENTITY_MATCH_ANY &&
+           recv->usermap[0].to == 65534 && recv->usermap[1].from == 1000 &&
+           recv->usermap[1].to == 1000 && recv->groupmap[0].from == 0 &&
+           recv->groupmap[0].to == IDENTITY_CURRENT;
+    }
+    config_delete(recv);
+    close(p[0]);
+    close(p[1]);
+    _exit(ok ? 0 : 1);
+  } else {
+    close(p[0]);
+    io_set_fds(p[1], p[1]);
+    bool sent = config_send(p[1], send_cfg);
+    int status;
+    waitpid(pid, &status, 0);
+    close(p[1]);
+    config_delete(send_cfg);
+    EXPECT_TRUE(sent);
+    EXPECT_TRUE(WIFEXITED(status) && WEXITSTATUS(status) == 0);
+  }
+}
+
+/* The receiver must reject an out-of-range identity-map count or id on the
+   wire (defense against a malicious/oversized table). */
+static void test_config_receive_rejects_invalid_identity() {
+  if (is_running_under_valgrind())
+    return;
+  Config* c = config_create();
+  EXPECT_NOT_NULL(c);
+  c->send_directory = str_dup("/src");
+  c->receive_root_directory = str_dup("/dst");
+  c->usermap_count = 1;
+  c->usermap = calloc(1, sizeof(IdentityMap));
+  c->usermap[0].from = -2; /* below IDENTITY_MATCH_ANY */
+  c->usermap[0].to = 0;
+  EXPECT_FALSE(roundtrip_config_ok(c));
+  config_delete(c);
+
+  c = config_create();
+  EXPECT_NOT_NULL(c);
+  c->send_directory = str_dup("/src");
+  c->receive_root_directory = str_dup("/dst");
+  c->chown_uid_set = true;
+  c->chown_uid = -5;
+  EXPECT_FALSE(roundtrip_config_ok(c));
+  config_delete(c);
+
+  /* A well-formed identity config still round-trips through the shared helper. */
+  c = config_create();
+  EXPECT_NOT_NULL(c);
+  c->send_directory = str_dup("/src");
+  c->receive_root_directory = str_dup("/dst");
+  c->numeric_ids = true;
+  EXPECT_TRUE(roundtrip_config_ok(c));
+  config_delete(c);
+}
+
 void test_config() {
   test_config_lifecycle();
   test_config_ssh_dest();
@@ -932,6 +1028,8 @@ void test_config() {
     test_config_basis_normalization();
     test_config_checksum_options_wire_roundtrip();
     test_config_receive_rejects_invalid_checksum_algo();
+    test_config_identity_wire_roundtrip();
+    test_config_receive_rejects_invalid_identity();
   }
   test_config_delete_timing_early_helper();
   test_config_is_remote_dest();