diff --git a/CHANGELOG.md b/CHANGELOG.md
index e07e7f3..836c7c5 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,15 @@
 # Changelog
 
+## [0.5.0] - 2026-10-06
+
+### Added
+- Optional `overwrite` file patterns on `upload`, `sync` and `plan_upload` copy selected directory files even when sizes match.
+  Exclusions take precedence. Downloads and deletion settings retain their existing behavior.
+- Optional `upload_exclude` on `sync` overrides upload exclusions without changing the initial download.
+  `None` inherits `exclude`; an empty list clears upload exclusions.
+  With `delete_remote=True`, remote objects matched by `exclude` but not by `upload_exclude` are deleted unless a local file exists at that path.
+  Use `delete_remote=False` to preserve these objects. For a later upload of recording markers, close the first mirror and use a new mirror with `upload(..., interval=None, delete=False)`.
+
 ## [0.4.0] - 2026-09-24
 
 ### Added
diff --git a/README.md b/README.md
index 8e74782..7dad71c 100644
--- a/README.md
+++ b/README.md
@@ -87,7 +87,7 @@ Registers a path for download. Ensures local copy matches S3 immediately.
 
 **Returns**: `pathlib.Path` to the local directory/file.
 
-### `pos3.upload(remote, local=None, interval=300, delete=True, sync_on_error=False, exclude=None)`
+### `pos3.upload(remote, local=None, interval=300, delete=True, sync_on_error=False, exclude=None, profile=None, overwrite=None)`
 
 Registers a local path for upload. Uploads on exit and optionally in background.
 - `remote`: Destination S3 URL.
@@ -98,11 +98,26 @@ Registers a local path for upload. Uploads on exit and optionally in background.
 
 **Returns**: `pathlib.Path` to the local directory/file.
 
-### `pos3.sync(remote, local=None, interval=300, delete_local=True, delete_remote=True, sync_on_error=False, exclude=None)`
+Pass `overwrite=[".recording-attempts.sqlite", "run_metadata_*.yaml"]` to copy matching files even when the remote size matches. This selection applies to file paths relative to a directory source, or the local basename of a single-file source. Patterns use `PurePosixPath.match`, which matches from the right. A directory name does not select its contents. For example, a local `journal` file matches `"*"` or `"journal"`, even when uploaded to a different remote key. Excluded files remain excluded. The default compares file sizes. This option does not change deletion settings. `plan_upload(..., overwrite=...)` reports the same selection without transfers.
+
+### `pos3.sync(remote, local=None, interval=300, delete_local=True, delete_remote=True, sync_on_error=False, exclude=None, profile=None, overwrite=None, upload_exclude=None)`
 
 Bi-directional helper. Performs `download()` then registers `upload()`. Useful for jobs that work on existing files, like when you resume training from a checkpoint.
 - `delete_local`: Cleanup local files during download.
 - `delete_remote`: Cleanup remote files during upload. carefully consider setting to `False` when resuming jobs to avoid deleting history.
+- `overwrite`: Select local files for forced upload using directory-relative paths or a single-file basename. The initial download still compares sizes. Existing same-size local files stay local and can replace remote bytes on upload. Use a fresh local root when the remote contents must be authoritative.
+- `upload_exclude`: Override exclusions for uploads only. `None` uses `exclude` for both directions. An empty list clears upload exclusions. The initial download always uses `exclude`. With `delete_remote=True`, remote objects matched by `exclude` but not by `upload_exclude` are deleted unless a local file exists at that path. Use `delete_remote=False` to preserve remote objects when narrowing or clearing upload exclusions.
+
+Use `upload_exclude=[".unfinished"]` with `delete_remote=False` to download existing recording markers while withholding new open markers from periodic uploads. After recording and the first mirror context have stopped, a one-shot `upload` in a new `mirror()` context can include markers that remain. Set `delete=False` on this upload to preserve remote objects absent locally.
+
+```python
+with pos3.mirror():
+    local = pos3.sync(remote, local, delete_remote=False, upload_exclude=[".unfinished"])
+    # Run and stop recording before leaving this context.
+
+with pos3.mirror():
+    pos3.upload(remote, local, interval=None, delete=False)
+```
 
 **Returns**: `pathlib.Path` to the local directory/file.
 
diff --git a/pos3/__init__.py b/pos3/__init__.py
index 842e6d6..00f2d41 100644
--- a/pos3/__init__.py
+++ b/pos3/__init__.py
@@ -193,7 +193,13 @@ def _filter_fileinfo(fileinfo_iter: Iterator[FileInfo], exclude: list[str] | Non
             yield info
 
 
-def _compute_sync_diff(source: Iterator[FileInfo], target: Iterator[FileInfo]) -> tuple[list[FileInfo], list[FileInfo]]:
+def _compute_sync_diff(
+    source: Iterator[FileInfo],
+    target: Iterator[FileInfo],
+    overwrite: list[str] | None = None,
+    *,
+    source_name: str = ".",
+) -> tuple[list[FileInfo], list[FileInfo]]:
     source_map: dict[str, FileInfo] = {info.relative_path: info for info in source}
     target_map: dict[str, FileInfo] = {info.relative_path: info for info in target}
 
@@ -207,7 +213,10 @@ def _compute_sync_diff(source: Iterator[FileInfo], target: Iterator[FileInfo]) -
         elif source_info.is_dir != target_info.is_dir:
             to_delete.append(target_info)
             to_copy.append(source_info)
-        elif not source_info.is_dir and source_info.size != target_info.size:
+        elif not source_info.is_dir and (
+            source_info.size != target_info.size
+            or any(PurePosixPath(relative_path or source_name).match(pattern) for pattern in overwrite or [])
+        ):
             to_copy.append(source_info)
 
     for relative_path, target_info in target_map.items():
@@ -263,6 +272,7 @@ class _UploadRegistration:
     sync_on_error: bool
     exclude: list[str] | None
     profile: Profile | None = None
+    overwrite: list[str] | None = None
     last_sync: float = 0.0
     # The user's original URL, preserving trailing-slash intent. _sync_uploads
     # parses it raw so `s3://bucket/data/` skips head_object('data') in
@@ -281,6 +291,7 @@ def __eq__(self, other):
             and self.sync_on_error == other.sync_on_error
             and self.exclude == other.exclude
             and self.profile == other.profile
+            and self.overwrite == other.overwrite
         )
 
 
@@ -440,6 +451,7 @@ def upload(
         sync_on_error,
         exclude: list[str] | None = None,
         profile: str | Profile | None = None,
+        overwrite: list[str] | None = None,
     ) -> Path:
         """
         Register (and perform if needed) an upload from a local directory or file to a remote S3 bucket path.
@@ -485,6 +497,7 @@ def upload(
             profile=effective_profile,
             last_sync=0,
             raw_remote=remote,
+            overwrite=None if overwrite is None else list(overwrite),
         )
 
         with self._lock:
@@ -557,6 +570,7 @@ def plan_upload(
         local: str | Path | None = None,
         exclude: list[str] | None = None,
         profile: str | Profile | None = None,
+        overwrite: list[str] | None = None,
     ) -> TransferPlan:
         """Compute the :class:`TransferPlan` ``upload()`` would execute.
 
@@ -589,6 +603,8 @@ def plan_upload(
         to_copy, to_delete = _compute_sync_diff(
             _filter_fileinfo(_scan_local(source), exclude),
             _filter_fileinfo(self._scan_s3(scan_bucket, scan_prefix, effective_profile), exclude),
+            overwrite=overwrite,
+            source_name=source.name,
         )
         copies: list[tuple[str, str]] = []
         for info in to_copy:
@@ -615,6 +631,8 @@ def sync(
         sync_on_error: bool,
         exclude: list[str] | None = None,
         profile: str | Profile | None = None,
+        overwrite: list[str] | None = None,
+        upload_exclude: list[str] | None = None,
     ) -> Path:
         # Let download() and upload() handle profile resolution and normalization
         local_path = self.download(remote, local, delete_local, exclude, profile)
@@ -625,7 +643,16 @@ def sync(
         effective_profile = self._effective_profile(profile, remote)
         # Unregister the download to allow upload registration for the same remote
         self._downloads.pop((normalized, effective_profile), None)
-        return self.upload(remote, local_path, interval, delete_remote, sync_on_error, exclude, profile)
+        return self.upload(
+            remote,
+            local_path,
+            interval,
+            delete_remote,
+            sync_on_error,
+            exclude if upload_exclude is None else upload_exclude,
+            profile,
+            overwrite,
+        )
 
     def ls(self, prefix: str, recursive: bool = False, profile: str | Profile | None = None) -> list[str]:
         """Lists objects under the given prefix, working for both local directories and S3 prefixes."""
@@ -757,7 +784,7 @@ def _final_sync(self, had_error: bool = False) -> None:
                 self._sync_uploads(uploads)
 
     def _sync_uploads(self, registrations: Iterable[_UploadRegistration]) -> None:
-        tasks: list[tuple[str, Path, bool, list[str] | None, Profile | None]] = []
+        tasks: list[tuple[str, Path, bool, list[str] | None, Profile | None, list[str] | None]] = []
         for registration in registrations:
             if registration.local_path.exists():
                 tasks.append(
@@ -771,6 +798,7 @@ def _sync_uploads(self, registrations: Iterable[_UploadRegistration]) -> None:
                         registration.delete,
                         registration.exclude,
                         registration.profile,
+                        registration.overwrite,
                     )
                 )
 
@@ -781,13 +809,15 @@ def _sync_uploads(self, registrations: Iterable[_UploadRegistration]) -> None:
         to_remove: list[tuple[str, str, Profile | None]] = []
         total_bytes = 0
 
-        for remote, local_path, delete, exclude, profile in tasks:
+        for remote, local_path, delete, exclude, profile, overwrite in tasks:
             logger.debug("Syncing upload: %s from %s (delete=%s)", remote, local_path, delete)
             scan_bucket, scan_prefix = _parse_s3_url(remote)
             bucket, out_prefix = _parse_s3_url(_normalize_s3_url(remote))
             to_copy, to_delete = _compute_sync_diff(
                 _filter_fileinfo(_scan_local(local_path), exclude),
                 _filter_fileinfo(self._scan_s3(scan_bucket, scan_prefix, profile), exclude),
+                overwrite=overwrite,
+                source_name=local_path.name,
             )
 
             for info in to_copy:
@@ -1151,6 +1181,7 @@ def upload(
     sync_on_error: bool = False,
     exclude: list[str] | None = None,
     profile: str | Profile | None = None,
+    overwrite: list[str] | None = None,
 ) -> Path:
     """
     Register a local path for upload. Uploads on exit and optionally in background.
@@ -1162,12 +1193,13 @@ def upload(
         delete: If True (default), deletes S3 files NOT present locally.
         sync_on_error: If True, syncs even if the context exits with an exception.
         profile: S3 profile name or Profile config for custom endpoints.
+        overwrite: Relative-path or single-file basename patterns copied even when sizes match.
 
     Returns:
         Path to the local directory/file.
     """
     mirror_obj = _require_active_mirror()
-    return mirror_obj.upload(remote, local, interval, delete, sync_on_error, exclude, profile)
+    return mirror_obj.upload(remote, local, interval, delete, sync_on_error, exclude, profile, overwrite)
 
 
 def sync(
@@ -1179,6 +1211,8 @@ def sync(
     sync_on_error: bool = False,
     exclude: list[str] | None = None,
     profile: str | Profile | None = None,
+    overwrite: list[str] | None = None,
+    upload_exclude: list[str] | None = None,
 ) -> Path:
     """
     Bi-directional helper. Performs download() then registers upload().
@@ -1186,13 +1220,17 @@ def sync(
     Args:
         delete_local: Cleanup local files during download.
         delete_remote: Cleanup remote files during upload.
+        overwrite: Relative-path or single-file basename patterns copied even when sizes match.
+        upload_exclude: Upload-only exclusions. None inherits exclude; an empty list clears upload exclusions.
         profile: S3 profile name or Profile config for custom endpoints.
 
     Returns:
         Path to the local directory/file.
     """
     mirror_obj = _require_active_mirror()
-    return mirror_obj.sync(remote, local, interval, delete_local, delete_remote, sync_on_error, exclude, profile)
+    return mirror_obj.sync(
+        remote, local, interval, delete_local, delete_remote, sync_on_error, exclude, profile, overwrite, upload_exclude
+    )
 
 
 def ls(prefix: str, recursive: bool = False, profile: str | Profile | None = None) -> list[str]:
@@ -1232,6 +1270,7 @@ def plan_upload(
     local: str | Path | None = None,
     exclude: list[str] | None = None,
     profile: str | Profile | None = None,
+    overwrite: list[str] | None = None,
 ) -> TransferPlan:
     """Return the :class:`TransferPlan` ``upload()`` would execute.
 
@@ -1242,7 +1281,7 @@ def plan_upload(
     skips registrations with a missing local_path).
     """
     mirror_obj = _require_active_mirror()
-    return mirror_obj.plan_upload(remote, local, exclude, profile)
+    return mirror_obj.plan_upload(remote, local, exclude, profile, overwrite)
 
 
 __all__ = [
diff --git a/pyproject.toml b/pyproject.toml
index abfd7fe..87da497 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "pos3"
-version = "0.4.0"
+version = "0.5.0"
 description = "S3 Simple Sync - Make using S3 as simple as using local files"
 readme = "README.md"
 requires-python = ">=3.11"
diff --git a/tests/test_overwrite.py b/tests/test_overwrite.py
new file mode 100644
index 0000000..1aaef3d
--- /dev/null
+++ b/tests/test_overwrite.py
@@ -0,0 +1,274 @@
+from pathlib import Path
+
+import pytest
+from botocore.exceptions import ClientError
+
+import pos3
+
+
+@pytest.fixture
+def storage(monkeypatch):
+    class Storage:
+        def __init__(self):
+            self.objects = {}
+            self.uploads = []
+            self.deletes = []
+
+        def head_object(self, *, Bucket, Key):
+            if (Bucket, Key) not in self.objects:
+                raise ClientError({"Error": {"Code": "404"}}, "HeadObject")
+            return {"ContentLength": len(self.objects[Bucket, Key])}
+
+        def get_paginator(self, operation):
+            assert operation == "list_objects_v2"
+            return self
+
+        def paginate(self, *, Bucket, Prefix):
+            yield {
+                "Contents": [
+                    {"Key": key, "Size": len(body)}
+                    for (bucket, key), body in self.objects.items()
+                    if bucket == Bucket and key.startswith(Prefix)
+                ]
+            }
+
+        def upload_file(self, filename, bucket, key, *, Callback):
+            body = Path(filename).read_bytes()
+            self.objects[bucket, key] = body
+            self.uploads.append(key)
+            Callback(len(body))
+
+        def put_object(self, *, Bucket, Key, Body):
+            self.objects[Bucket, Key] = Body
+
+        def download_file(self, bucket, key, filename, *, Callback):
+            body = self.objects[bucket, key]
+            Path(filename).write_bytes(body)
+            Callback(len(body))
+
+        def delete_object(self, *, Bucket, Key):
+            self.deletes.append(Key)
+            del self.objects[Bucket, Key]
+
+    result = Storage()
+    monkeypatch.setattr(pos3._Mirror, "_get_client", lambda self, profile=None: result)
+    return result
+
+
+@pytest.mark.parametrize("method", ["upload", "sync"])
+@pytest.mark.parametrize("patterns", [None, [], [".recording-attempts.sqlite", "run_metadata_*.yaml"]])
+def test_selected_same_size_files_are_overwritten_without_deleting_remote_objects(tmp_path, storage, method, patterns):
+    names = [".recording-attempts.sqlite", "run_metadata_1.yaml", "episode.parquet"]
+    for name in names:
+        (tmp_path / name).write_bytes(b"new")
+        storage.objects["bucket", "run/" + name] = b"old"
+    storage.objects["bucket", "run/remote-only"] = b"keep"
+    with pos3.mirror(show_progress=False):
+        plan = pos3.plan_upload("s3://bucket/run/", tmp_path, overwrite=patterns)
+        kwargs = {"delete": False} if method == "upload" else {"delete_remote": False}
+        getattr(pos3, method)("s3://bucket/run/", tmp_path, interval=None, overwrite=patterns, **kwargs)
+        expected = names[:2] if patterns else []
+        assert {remote for _, remote in plan.to_copy} == {"s3://bucket/run/" + name for name in expected}
+        assert storage.uploads == []
+    assert set(storage.uploads) == {"run/" + name for name in expected}
+    assert storage.deletes == []
+    assert storage.objects["bucket", "run/remote-only"] == b"keep"
+    for name in names:
+        assert storage.objects["bucket", "run/" + name] == (b"new" if name in expected else b"old")
+    with pos3.mirror(show_progress=False):
+        readback = pos3.download("s3://bucket/run/", local=tmp_path / "readback")
+    for name in names:
+        assert (readback / name).read_bytes() == (b"new" if name in expected else b"old")
+
+
+def test_overwrite_selection_survives_registration_replay_and_respects_exclusion(tmp_path, storage):
+    (tmp_path / "journal").write_bytes(b"new")
+    storage.objects["bucket", "run/journal"] = b"old"
+    with pos3.mirror(show_progress=False):
+        pos3.upload(
+            "s3://bucket/run/", tmp_path, interval=None, delete=False, overwrite=["journal"], exclude=["journal"]
+        )
+        pos3.upload(
+            "s3://bucket/run/", tmp_path, interval=None, delete=False, overwrite=["journal"], exclude=["journal"]
+        )
+        with pytest.raises(ValueError, match="different parameters"):
+            pos3.upload("s3://bucket/run/", tmp_path, interval=None, delete=False, overwrite=[], exclude=["journal"])
+        assert pos3.plan_upload("s3://bucket/run/", tmp_path, overwrite=["journal"], exclude=["journal"]).to_copy == []
+    assert storage.uploads == []
+    assert storage.objects["bucket", "run/journal"] == b"old"
+
+
+def test_sync_overwrite_keeps_size_only_download_and_uploads_selected_local_bytes(tmp_path, storage):
+    local = tmp_path / "journal"
+    local.write_bytes(b"local")
+    storage.objects["bucket", "run/journal"] = b"other"
+    with pos3.mirror(show_progress=False):
+        pos3.sync("s3://bucket/run/", tmp_path, interval=None, delete_remote=False, overwrite=["journal"])
+        assert local.read_bytes() == b"local"
+        assert storage.objects["bucket", "run/journal"] == b"other"
+    assert storage.objects["bucket", "run/journal"] == b"local"
+
+
+def test_directory_name_does_not_select_its_contents(tmp_path, storage):
+    directory = tmp_path / "state"
+    directory.mkdir()
+    (directory / "journal").write_bytes(b"new")
+    storage.objects["bucket", "run/state/journal"] = b"old"
+    storage.objects["bucket", "run/state/"] = b""
+    with pos3.mirror(show_progress=False):
+        assert pos3.plan_upload("s3://bucket/run/", tmp_path, overwrite=["state"]).to_copy == []
+        pos3.upload("s3://bucket/run/", tmp_path, interval=None, delete=False, overwrite=["state"])
+    assert storage.objects["bucket", "run/state/journal"] == b"old"
+
+
+@pytest.mark.parametrize("method", ["upload", "sync"])
+@pytest.mark.parametrize("patterns", [["journal"], ["state/journal"]])
+def test_right_relative_patterns_select_and_copy_nested_same_size_files(tmp_path, storage, method, patterns):
+    directory = tmp_path / "state"
+    directory.mkdir()
+    for name in ["journal", "other"]:
+        (directory / name).write_bytes(b"new")
+        storage.objects["bucket", "run/state/" + name] = b"old"
+    storage.objects["bucket", "run/state/remote-only"] = b"keep"
+    with pos3.mirror(show_progress=False):
+        plan = pos3.plan_upload("s3://bucket/run/", tmp_path, overwrite=patterns)
+        assert plan.to_copy == [(str(directory / "journal"), "s3://bucket/run/state/journal")]
+        kwargs = {"delete": False} if method == "upload" else {"delete_remote": False}
+        getattr(pos3, method)("s3://bucket/run/", tmp_path, interval=None, overwrite=patterns, **kwargs)
+        assert (directory / "journal").read_bytes() == b"new"
+        assert storage.objects["bucket", "run/state/journal"] == b"old"
+    assert storage.uploads == ["run/state/journal"]
+    assert storage.objects["bucket", "run/state/journal"] == b"new"
+    assert storage.objects["bucket", "run/state/other"] == b"old"
+    assert storage.objects["bucket", "run/state/remote-only"] == b"keep"
+    assert storage.deletes == []
+
+
+@pytest.mark.parametrize("method", ["upload", "sync"])
+@pytest.mark.parametrize(
+    "patterns,selected",
+    [
+        (None, False),
+        ([], False),
+        (["*"], True),
+        (["journal"], True),
+        (["remote-journal"], False),
+        (["state/journal"], False),
+    ],
+)
+def test_single_file_overwrite_matches_source_basename_and_retains_remote_objects(
+    tmp_path, storage, method, patterns, selected
+):
+    local = tmp_path / "journal"
+    local.write_bytes(b"new")
+    remote = "s3://bucket/run/remote-journal"
+    storage.objects["bucket", "run/remote-journal"] = b"old"
+    storage.objects["bucket", "run/remote-only"] = b"keep"
+    with pos3.mirror(show_progress=False):
+        plan = pos3.plan_upload(remote, local, overwrite=patterns)
+        assert plan.to_delete == []
+        assert storage.uploads == []
+        assert storage.deletes == []
+        kwargs = {"delete": False} if method == "upload" else {"delete_remote": False}
+        getattr(pos3, method)(remote, local, interval=None, overwrite=patterns, **kwargs)
+        assert local.read_bytes() == b"new"
+        assert storage.objects["bucket", "run/remote-journal"] == b"old"
+    assert storage.objects["bucket", "run/remote-journal"] == (b"new" if selected else b"old")
+    assert plan.to_copy == ([(str(local), remote)] if selected else [])
+    assert storage.uploads == (["run/remote-journal"] if selected else [])
+    assert storage.deletes == []
+    assert storage.objects["bucket", "run/remote-only"] == b"keep"
+    with pos3.mirror(show_progress=False):
+        readback = pos3.download(remote, local=tmp_path / "readback")
+    assert readback.read_bytes() == (b"new" if selected else b"old")
+
+
+@pytest.mark.parametrize(
+    "upload_exclude,expected", [(None, {"upload"}), ([], {"general", "upload"}), (["upload"], {"general"})]
+)
+def test_sync_upload_exclusions_keep_the_download_exclusions(tmp_path, storage, upload_exclude, expected):
+    storage.objects["bucket", "run/general"] = b"excluded download"
+    storage.objects["bucket", "run/baseline"] = b"retained"
+    with pos3.mirror(show_progress=False):
+        pos3.sync(
+            "s3://bucket/run/",
+            tmp_path,
+            interval=None,
+            delete_remote=False,
+            exclude=["general"],
+            upload_exclude=upload_exclude,
+        )
+        assert not (tmp_path / "general").exists()
+        assert (tmp_path / "baseline").read_bytes() == b"retained"
+        (tmp_path / "general").write_bytes(b"new")
+        (tmp_path / "upload").write_bytes(b"new")
+    assert set(storage.uploads) == {"run/" + name for name in expected}
+    assert storage.deletes == []
+
+
+def test_sync_can_download_baseline_markers_without_uploading_new_open_markers(tmp_path, storage):
+    storage.objects["bucket", "run/baseline/.unfinished"] = b"unfinished"
+    storage.objects["bucket", "run/foreign"] = b"retained"
+    with pos3.mirror(show_progress=False):
+        pos3.sync("s3://bucket/run/", tmp_path, interval=None, delete_remote=False, upload_exclude=[".unfinished"])
+        assert (tmp_path / "baseline/.unfinished").read_bytes() == b"unfinished"
+        (tmp_path / "new").mkdir()
+        (tmp_path / "new/.unfinished").write_bytes(b"unfinished")
+        (tmp_path / "new/data").write_bytes(b"recording")
+    assert ("bucket", "run/new/.unfinished") not in storage.objects
+    assert storage.objects["bucket", "run/new/data"] == b"recording"
+    assert storage.objects["bucket", "run/baseline/.unfinished"] == b"unfinished"
+    assert storage.objects["bucket", "run/foreign"] == b"retained"
+    assert storage.deletes == []
+
+    storage.objects["bucket", "run/late-foreign"] = b"arrived after phase one"
+    assert not (tmp_path / "late-foreign").exists()
+    with pos3.mirror(show_progress=False):
+        pos3.upload("s3://bucket/run/", tmp_path, interval=None, delete=False)
+    assert storage.objects["bucket", "run/new/.unfinished"] == b"unfinished"
+    assert storage.objects["bucket", "run/baseline/.unfinished"] == b"unfinished"
+    assert storage.objects["bucket", "run/foreign"] == b"retained"
+    assert storage.objects["bucket", "run/late-foreign"] == b"arrived after phase one"
+    assert storage.deletes == []
+
+
+@pytest.mark.parametrize("delete_remote", [False, True])
+@pytest.mark.parametrize("upload_exclude", [None, []])
+def test_sync_narrowed_upload_exclusions_obey_remote_deletion(tmp_path, storage, delete_remote, upload_exclude):
+    storage.objects["bucket", "run/general"] = b"excluded download"
+    storage.objects["bucket", "run/baseline"] = b"retained"
+    with pos3.mirror(show_progress=False):
+        pos3.sync(
+            "s3://bucket/run/",
+            tmp_path,
+            interval=None,
+            delete_remote=delete_remote,
+            exclude=["general"],
+            upload_exclude=upload_exclude,
+        )
+        assert not (tmp_path / "general").exists()
+        assert (tmp_path / "baseline").read_bytes() == b"retained"
+    if delete_remote and upload_exclude == []:
+        assert ("bucket", "run/general") not in storage.objects
+        assert storage.deletes == ["run/general"]
+    else:
+        assert storage.objects["bucket", "run/general"] == b"excluded download"
+        assert storage.deletes == []
+    assert storage.objects["bucket", "run/baseline"] == b"retained"
+    assert storage.uploads == []
+
+
+def test_registered_overwrite_selection_survives_caller_list_mutation(tmp_path, storage):
+    for name in ["journal", "other"]:
+        (tmp_path / name).write_bytes(b"new")
+        storage.objects["bucket", "run/" + name] = b"old"
+    patterns = ["journal"]
+    with pos3.mirror(show_progress=False):
+        pos3.upload("s3://bucket/run/", tmp_path, interval=None, delete=False, overwrite=patterns)
+        patterns[:] = ["other"]
+        with pytest.raises(ValueError, match="different parameters"):
+            pos3.upload("s3://bucket/run/", tmp_path, interval=None, delete=False, overwrite=["other"])
+    assert storage.uploads == ["run/journal"]
+    assert storage.objects["bucket", "run/journal"] == b"new"
+    assert storage.objects["bucket", "run/other"] == b"old"
+    assert storage.deletes == []
diff --git a/uv.lock b/uv.lock
index 2b40744..6f4692f 100644
--- a/uv.lock
+++ b/uv.lock
@@ -223,7 +223,7 @@ wheels = [
 
 [[package]]
 name = "pos3"
-version = "0.4.0"
+version = "0.5.0"
 source = { editable = "." }
 dependencies = [
     { name = "boto3" },