@@ -555,6 +555,52 @@ def _read_tree(repo: Any, args: List[str], kwargs: Dict[str, Any]) -> bytes:
     return b""
 
 
+def _checked_signature(signature: Any) -> Any:
+    # Git strips "crud" at the edges and angle brackets/newlines within identities.
+    crud = bytes(range(33)) + b".,:;<>\"'\\"
+    for value in (signature.name, signature.email):
+        if not value or value != value.strip(crud) or any(char in value for char in b"<>\r\n\0"):
+            raise _Unsupported("identity normalization")
+    return signature
+
+
+def _signature(env: Dict[str, Any], role: str) -> Any:
+    prefix = "GIT_" + role + "_"
+    name, email, date = (env.get(prefix + field) for field in ("NAME", "EMAIL", "DATE"))
+    match = re.fullmatch(r"(-?\d+) ([+-])(\d\d)(\d\d)", date or "")
+    if not name or not email or match is None:
+        raise _Unsupported("identity resolution")
+    seconds, sign, hours, minutes = match.groups()
+    offset = (int(hours) * 60 + int(minutes)) * 60 * (-1 if sign == "-" else 1)
+    return _checked_signature(gix.Signature(name, email, int(seconds), offset))
+
+
+def _commit_tree(repo: Any, args: List[str], kwargs: Dict[str, Any]) -> bytes:
+    if len(args) < 2 or args[0] != "--no-gpg-sign" or args[2::2] != ["-p"] * len(args[2::2]):
+        raise _Unsupported("commit options")
+    if len(args) % 2 or any(setting.lower() != "i18n.commitencoding=utf-8" for setting in kwargs.get("_config", ())):
+        raise _Unsupported("commit encoding (GIX-6)")
+    data = _input(kwargs)
+    try:
+        message = data.decode("utf-8")
+    except UnicodeDecodeError:
+        raise _Unsupported("commit message bytes (GIX-6)") from None
+    parents = args[3::2]
+    if len(set(parents)) != len(parents):
+        raise _Unsupported("duplicate parents")
+    author = _signature(kwargs.get("env", {}), "AUTHOR")
+    committer = _signature(kwargs.get("env", {}), "COMMITTER")
+    tree = _oid(repo, args[1])
+    # Native commit creation accepts literal IDs, whereas commit-tree verifies kinds.
+    if repo.find_header(tree).kind() != "tree" or any(
+        repo.find_header(_oid(repo, p)).kind() != "commit" for p in parents
+    ):
+        raise _Unsupported("commit object kinds")
+    commit = _write("commit_tree", lambda: repo.new_commit_as(committer, author, message, tree, parents))
+    kwargs["istream"].seek(len(data), 1)
+    return str(commit.id).encode("ascii") + b"\n"
+
+
 _HANDLERS: Dict[str, Callable[[Any, List[str], Dict[str, Any]], bytes]] = {
     "rev_parse": _rev_parse,
     "ls_tree": _ls_tree,
@@ -566,6 +612,7 @@ def _read_tree(repo: Any, args: List[str], kwargs: Dict[str, Any]) -> bytes:
     "hash_object": _hash_object,
     "mktree": _mktree,
     "read_tree": _read_tree,
+    "commit_tree": _commit_tree,
     "config": _config,
     "worktree": _worktree,
 }
 
@@ -7,7 +7,7 @@
 
 import pytest
 
-from git import Git, Repo, _backend
+from git import Actor, Commit, Git, Repo, _backend
 from git.exc import GitCommandError, InvalidGitRepositoryError
 from gitdb import IStream
 
@@ -55,6 +55,45 @@ def test_native_worktree_inventory_includes_bare_main(repo, tmp_path):
                     assert current.git._call_process_safe("worktree", "list", "--porcelain", "-z") == expected
 
 
+def test_native_objects_trees_commits_and_index_reads(repo):
+    actor = Actor("Example", "example@example.invalid")
+    paths = ["file", "dir/with space", "dir/unicode-é", "--option"]
+    for name in paths:
+        path = Path(repo.working_dir, name)
+        path.parent.mkdir(exist_ok=True)
+        path.write_bytes(name.encode())
+    index = repo.index
+    # All setup has finished; these operations must succeed without a subprocess.
+    with patch.object(Git, "execute", side_effect=AssertionError("unexpected CLI call")):
+        index.add(paths)
+        tree = index.write_tree()
+        assert {blob.path for blob in tree.traverse() if blob.type == "blob"} == set(paths)
+        assert tree["file"].data_stream.read() == b"file"
+        assert set(repo.index.entries) == {(name, 0) for name in paths}
+        stream = IStream("blob", 4, BytesIO(b"data"))
+        assert repo.odb.store(stream) is stream
+        assert repo.odb.stream(stream.binsha).read(2) == b"da"
+    commit = index.commit("message without a trailing newline", author=actor, committer=actor, skip_hooks=True)
+    with patch.object(Git, "execute", side_effect=AssertionError("unexpected CLI call")):
+        assert repo.commit(commit.hexsha).message == "message without a trailing newline"
+        assert repo.merge_base(commit, commit) == [commit]
+        assert repo.is_ancestor(commit, commit)
+        assert commit.replace(message="changed").message == "changed"
+
+
+def test_native_tree_and_commit_bytes_match_cli(repo):
+    actor = Actor("Example", "example@example.invalid")
+    Path(repo.working_dir, "file").write_bytes(b"payload\n")
+    index = repo.index
+    index.add(["file"])
+    native_tree = index.write_tree()
+    commit = index.commit("message", author=actor, committer=actor, skip_hooks=True)
+    native = commit.replace(message="a changed message")
+    with patch.object(_backend, "gix", None):
+        assert index.write_tree().hexsha == native_tree.hexsha
+        assert commit.replace(message="a changed message").hexsha == native.hexsha
+
+
 def test_unknown_command_and_storage_environment_use_cli(repo):
     before = _backend.statistics()
     assert repo.git._call_process_safe("status", "--porcelain") == ""
@@ -100,6 +139,21 @@ def test_tree_writes_validate_child_object_kinds(repo, mode, kind):
             repo.git._call_process_safe("mktree", "-z", "--missing", istream=stream)
 
 
+def test_commit_identity_cleanup_matches_git(repo):
+    actor = Actor(" Name. ", " email@example.invalid ")
+    tree = repo.index.write_tree()
+    kwargs = {
+        "author": actor,
+        "committer": actor,
+        "author_date": "1100000000 +0000",
+        "commit_date": "1100000000 +0000",
+        "parent_commits": [],
+    }
+    commit = Commit.create_from_tree(repo, tree, "message", **kwargs)
+    with patch.object(_backend, "gix", None):
+        assert commit == Commit.create_from_tree(repo, tree, "message", **kwargs)
+
+
 def test_partial_native_stream_does_not_corrupt_next_read(repo):
     one = repo.odb.store(IStream("blob", 6, BytesIO(b"abcdef"))).binsha
     two = repo.odb.store(IStream("blob", 3, BytesIO(b"xyz"))).binsha