From 30179e6e30018e72b2ec4d992df73de68ba19e78 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Thu, 21 May 2026 15:32:09 +0000
Subject: [PATCH 01/16] Add pos3 CLI: ls, download, upload (#10)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Introduces a console-script entry point so common pos3 operations are
one-liners from the shell instead of requiring a Python script.

- pos3 ls <prefix> [-r] [--profile NAME] — one full s3:// URL per line on stdout.
- pos3 download <url> [--local PATH] [--delete] [--exclude PATTERN]... [--profile NAME]
  — prints only the resulting local path to stdout (progress + logs go to
  stderr), so data_dir=$(pos3 download s3://bucket/dataset/) is safe.
- pos3 upload <url> [--local PATH] [--delete] [--exclude PATTERN]... [--profile NAME]
  — one-shot upload (no background loop, no interval). Source defaults to the
  cache path pos3 download would have produced; errors if the source is missing.

--delete defaults OFF in the CLI even though the Python API defaults to True:
CLI defaults are conservative for interactive shell use. --profile is honored
alongside the URL form s3://<profile>@bucket/...; URL wins on conflict, matching
existing Python precedence.

https://claude.ai/code/session_01QoQaZ2G6FsNpSZyStqTXas
---
 CHANGELOG.md      |  19 ++++
 README.md         |  33 ++++++
 pos3/cli.py       | 135 ++++++++++++++++++++++++
 pyproject.toml    |   3 +
 tests/test_cli.py | 262 ++++++++++++++++++++++++++++++++++++++++++++++
 5 files changed, 452 insertions(+)
 create mode 100644 pos3/cli.py
 create mode 100644 tests/test_cli.py

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 607c6d9..4356bad 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,24 @@
 # Changelog
 
+## [Unreleased]
+
+### Added
+- `pos3` console-script entry point with `ls`, `download`, `upload` subcommands
+  ([#10](https://github.com/Positronic-Robotics/pos3/issues/10)).
+  - `pos3 ls <prefix> [-r] [--profile NAME]` lists objects, one full `s3://` URL
+    per line on stdout.
+  - `pos3 download <url> [--local PATH] [--delete] [--exclude PATTERN]... [--profile NAME]`
+    prints only the resulting local path to stdout; progress and logs go to
+    stderr, so `data_dir=$(pos3 download s3://bucket/dataset/)` is safe.
+  - `pos3 upload <url> [--local PATH] [--delete] [--exclude PATTERN]... [--profile NAME]`
+    is one-shot (no background loop or interval). Source defaults to the cache
+    path `pos3 download` would have produced; errors if the source doesn't
+    exist.
+  - `--delete` defaults OFF for both `download` and `upload` (the Python API
+    defaults to `True`; the CLI is more conservative for interactive use).
+  - `--profile` is supported alongside the URL form `s3://<profile>@bucket/...`;
+    the URL form wins on conflict, matching the Python precedence.
+
 ## [0.3.0] - 2026-05-19
 
 ### Added
diff --git a/README.md b/README.md
index d207e67..dc3a661 100644
--- a/README.md
+++ b/README.md
@@ -113,6 +113,39 @@ Lists files/objects in a directory or S3 prefix.
 
 **Returns**: List of full S3 URLs or local paths.
 
+## CLI
+
+`pos3` ships a small command-line interface for the most common one-shot
+operations. After `uv pip install pos3` (or `pip install pos3`), `pos3` is on
+your `$PATH`:
+
+```bash
+# List objects (one full s3:// URL per line on stdout)
+pos3 ls s3://bucket/dataset/
+pos3 ls -r s3://bucket/dataset/
+
+# Download an S3 prefix or object into the cache (or a custom --local path).
+# Only the resulting local path is written to stdout — progress and logs go
+# to stderr — so it's safe to capture in a shell variable:
+data_dir=$(pos3 download s3://bucket/dataset/)
+
+# One-shot upload. Source defaults to the same cache path `pos3 download`
+# would have produced; --local overrides. Errors if the source doesn't exist.
+pos3 upload s3://bucket/results/ --local ./out/
+```
+
+All three subcommands accept `--profile NAME`. The URL form
+`s3://<profile>@bucket/...` takes precedence over `--profile` on conflict
+(matching the Python API).
+
+`--delete` defaults to **OFF** for both `download` and `upload`, even though
+the Python API defaults to `True`. CLI defaults are conservative for
+interactive shell use; pass `--delete` explicitly to mirror file removals.
+
+The CLI is one-shot only — no background sync, no `pos3 sync` subcommand.
+Use the Python `pos3.mirror()` context manager when you need an interval-based
+loop or bi-directional sync over a job's lifetime.
+
 ## Comparison with Libraries
 
 Why use `pos3` instead of other Python libraries?
diff --git a/pos3/cli.py b/pos3/cli.py
new file mode 100644
index 0000000..7caef05
--- /dev/null
+++ b/pos3/cli.py
@@ -0,0 +1,135 @@
+"""Command-line interface for pos3.
+
+Three subcommands — ``ls``, ``download``, ``upload`` — each running inside a
+short-lived ``pos3.mirror()`` context. ``upload`` is one-shot: no background
+interval, no sync loop.
+
+CLI defaults intentionally differ from the Python API: ``--delete`` defaults
+to OFF for both ``download`` and ``upload``, because the API's ``True``
+default is too destructive for interactive shell use. ``download`` writes
+only the resulting local path to stdout so the output is safe to capture in
+``$(pos3 download ...)``; progress bars and logs go to stderr.
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+from pathlib import Path
+
+from . import _is_s3_path, _require_active_mirror, download, ls, mirror, upload
+from .profiles import _resolve_profile, _url_profile
+
+
+def _build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(prog="pos3", description="pos3 command-line interface.")
+    subparsers = parser.add_subparsers(dest="command", required=True)
+
+    def add_profile(p: argparse.ArgumentParser) -> None:
+        p.add_argument(
+            "--profile",
+            metavar="NAME",
+            help="Named pos3 profile. Overridden by the URL form s3://<profile>@bucket/key.",
+        )
+
+    def add_transfer_args(p: argparse.ArgumentParser, local_help: str) -> None:
+        p.add_argument("--local", metavar="PATH", help=local_help)
+        p.add_argument(
+            "--delete",
+            action="store_true",
+            help="Delete files not present at the other end. Defaults to OFF in the CLI.",
+        )
+        p.add_argument(
+            "--exclude",
+            metavar="PATTERN",
+            action="append",
+            default=None,
+            help="Glob pattern to skip. May be passed multiple times.",
+        )
+        add_profile(p)
+
+    p_ls = subparsers.add_parser("ls", help="List objects under a prefix.")
+    p_ls.add_argument("prefix", help="S3 prefix (s3://bucket/key) or local path.")
+    p_ls.add_argument("-r", "--recursive", action="store_true", help="List subdirectories recursively.")
+    add_profile(p_ls)
+
+    p_dl = subparsers.add_parser("download", help="Download an S3 prefix or object to a local path.")
+    p_dl.add_argument("url", help="Source S3 URL (s3://bucket/key).")
+    add_transfer_args(p_dl, local_help="Destination path. Defaults to the cache path.")
+
+    p_up = subparsers.add_parser("upload", help="One-shot upload of a local path to S3.")
+    p_up.add_argument("url", help="Destination S3 URL (s3://bucket/key).")
+    add_transfer_args(p_up, local_help="Source path. Defaults to the cache path.")
+
+    return parser
+
+
+def _cmd_ls(args: argparse.Namespace) -> int:
+    with mirror(show_progress=False):
+        for item in ls(args.prefix, recursive=args.recursive, profile=args.profile):
+            print(item)
+    return 0
+
+
+def _cmd_download(args: argparse.Namespace) -> int:
+    with mirror(show_progress=True):
+        local_path = download(
+            args.url,
+            local=args.local,
+            delete=args.delete,
+            exclude=args.exclude,
+            profile=args.profile,
+        )
+    print(str(local_path))
+    return 0
+
+
+def _cmd_upload(args: argparse.Namespace) -> int:
+    with mirror(show_progress=True):
+        mirror_obj = _require_active_mirror()
+        if args.local:
+            source = Path(args.local).expanduser().resolve()
+        else:
+            # Resolve the same profile precedence (URL > --profile > context default)
+            # the upload() call will use, so the cache path we check matches the one
+            # upload() would target.
+            profile: str | None = args.profile
+            if _is_s3_path(args.url):
+                url_profile = _url_profile(args.url)
+                if url_profile is not None:
+                    profile = url_profile
+            effective_profile = _resolve_profile(profile) or mirror_obj.options.default_profile
+            source = mirror_obj.options.cache_path_for(args.url, effective_profile)
+        if not source.exists():
+            print(f"pos3 upload: source path does not exist: {source}", file=sys.stderr)
+            return 1
+        upload(
+            args.url,
+            local=source,
+            interval=None,
+            delete=args.delete,
+            exclude=args.exclude,
+            profile=args.profile,
+        )
+    return 0
+
+
+_COMMANDS = {
+    "ls": _cmd_ls,
+    "download": _cmd_download,
+    "upload": _cmd_upload,
+}
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = _build_parser()
+    args = parser.parse_args(argv)
+    try:
+        return _COMMANDS[args.command](args)
+    except ValueError as exc:
+        print(f"pos3 {args.command}: {exc}", file=sys.stderr)
+        return 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/pyproject.toml b/pyproject.toml
index 9421ebf..6699609 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -27,6 +27,9 @@ dependencies = [
 Homepage = "https://github.com/Positronic-Robotics/pos3"
 Repository = "https://github.com/Positronic-Robotics/pos3"
 
+[project.scripts]
+pos3 = "pos3.cli:main"
+
 [project.optional-dependencies]
 dev = [
     "pytest>=7.0",
diff --git a/tests/test_cli.py b/tests/test_cli.py
new file mode 100644
index 0000000..3d03f43
--- /dev/null
+++ b/tests/test_cli.py
@@ -0,0 +1,262 @@
+import tempfile
+from pathlib import Path
+from unittest.mock import Mock, patch
+
+import pytest
+from botocore.exceptions import ClientError
+
+from pos3.cli import main
+
+BOTO3_PATCH_TARGET = "pos3.profiles.boto3.client"
+
+
+def _make_404_error(*_args, **_kwargs):
+    raise ClientError({"Error": {"Code": "404"}}, "head_object")
+
+
+def _setup_s3_mock(mock_boto_client, paginate_return_value=None):
+    mock_s3 = Mock()
+    mock_boto_client.return_value = mock_s3
+    mock_s3.head_object.side_effect = _make_404_error
+    mock_paginator = Mock()
+    mock_s3.get_paginator.return_value = mock_paginator
+    mock_paginator.paginate.return_value = paginate_return_value or [{"Contents": []}]
+    return mock_s3
+
+
+class TestCliLs:
+    @patch(BOTO3_PATCH_TARGET)
+    def test_ls_prints_full_s3_urls(self, mock_boto_client, capsys):
+        paginate = [
+            {
+                "Contents": [
+                    {"Key": "data/file.txt", "Size": 5},
+                    {"Key": "data/sub/nested.txt", "Size": 10},
+                ]
+            }
+        ]
+        _setup_s3_mock(mock_boto_client, paginate)
+
+        rc = main(["ls", "s3://bucket/data"])
+
+        captured = capsys.readouterr()
+        assert rc == 0
+        lines = captured.out.strip().splitlines()
+        assert "s3://bucket/data/file.txt" in lines
+        # Non-recursive excludes nested items
+        assert "s3://bucket/data/sub/nested.txt" not in lines
+
+    @patch(BOTO3_PATCH_TARGET)
+    def test_ls_recursive(self, mock_boto_client, capsys):
+        paginate = [{"Contents": [{"Key": "data/sub/nested.txt", "Size": 10}]}]
+        _setup_s3_mock(mock_boto_client, paginate)
+
+        rc = main(["ls", "-r", "s3://bucket/data"])
+
+        captured = capsys.readouterr()
+        assert rc == 0
+        assert "s3://bucket/data/sub/nested.txt" in captured.out
+
+    def test_ls_local_path(self, capsys):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            base = Path(tmpdir)
+            (base / "file.txt").write_text("x")
+
+            rc = main(["ls", str(base)])
+
+        captured = capsys.readouterr()
+        assert rc == 0
+        assert str(base / "file.txt") in captured.out
+
+
+class TestCliDownload:
+    @patch(BOTO3_PATCH_TARGET)
+    def test_download_prints_only_local_path_to_stdout(self, mock_boto_client, capsys):
+        paginate = [{"Contents": [{"Key": "data/file.txt", "Size": 5}]}]
+        _setup_s3_mock(mock_boto_client, paginate)
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            local_dir = Path(tmpdir) / "dst"
+            rc = main(["download", "s3://bucket/data", "--local", str(local_dir)])
+
+        captured = capsys.readouterr()
+        assert rc == 0
+        # Exactly one line on stdout: the resulting local path.
+        lines = captured.out.strip().splitlines()
+        assert len(lines) == 1
+        assert lines[0] == str(local_dir.resolve())
+
+    @patch(BOTO3_PATCH_TARGET)
+    def test_download_default_does_not_delete(self, mock_boto_client):
+        """CLI default for --delete is OFF; orphan local files survive."""
+        paginate = [{"Contents": [{"Key": "data/file1.txt", "Size": 5}]}]
+        _setup_s3_mock(mock_boto_client, paginate)
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            local_dir = Path(tmpdir) / "data"
+            local_dir.mkdir()
+            (local_dir / "file1.txt").write_bytes(b"12345")
+            orphan = local_dir / "orphan.txt"
+            orphan.write_text("orphan")
+
+            rc = main(["download", "s3://bucket/data", "--local", str(local_dir)])
+
+            assert rc == 0
+            assert orphan.exists()
+
+    @patch(BOTO3_PATCH_TARGET)
+    def test_download_delete_flag_removes_orphans(self, mock_boto_client):
+        paginate = [{"Contents": [{"Key": "data/file1.txt", "Size": 5}]}]
+        _setup_s3_mock(mock_boto_client, paginate)
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            local_dir = Path(tmpdir) / "data"
+            local_dir.mkdir()
+            (local_dir / "file1.txt").write_bytes(b"12345")
+            orphan = local_dir / "orphan.txt"
+            orphan.write_text("orphan")
+
+            rc = main(["download", "s3://bucket/data", "--local", str(local_dir), "--delete"])
+
+            assert rc == 0
+            assert not orphan.exists()
+
+    @patch(BOTO3_PATCH_TARGET)
+    def test_download_exclude_multiple_patterns(self, mock_boto_client):
+        paginate = [
+            {
+                "Contents": [
+                    {"Key": "data/file.txt", "Size": 5},
+                    {"Key": "data/file.log", "Size": 10},
+                    {"Key": "data/file.tmp", "Size": 10},
+                ]
+            }
+        ]
+        mock_s3 = _setup_s3_mock(mock_boto_client, paginate)
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            local_dir = Path(tmpdir) / "dst"
+            rc = main(
+                [
+                    "download",
+                    "s3://bucket/data",
+                    "--local",
+                    str(local_dir),
+                    "--exclude",
+                    "*.log",
+                    "--exclude",
+                    "*.tmp",
+                ]
+            )
+
+        assert rc == 0
+        assert mock_s3.download_file.call_count == 1
+        downloaded_key = mock_s3.download_file.call_args_list[0][0][1]
+        assert "file.txt" in downloaded_key
+
+    @patch(BOTO3_PATCH_TARGET)
+    def test_download_unknown_profile_returns_error(self, mock_boto_client, capsys):
+        _setup_s3_mock(mock_boto_client)
+
+        rc = main(["download", "s3://bucket/data", "--profile", "no-such-profile"])
+
+        captured = capsys.readouterr()
+        assert rc == 1
+        assert "Unknown profile" in captured.err
+
+
+class TestCliUpload:
+    @patch(BOTO3_PATCH_TARGET)
+    def test_upload_errors_when_source_missing(self, mock_boto_client, capsys):
+        _setup_s3_mock(mock_boto_client)
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            missing = Path(tmpdir) / "does-not-exist"
+            rc = main(["upload", "s3://bucket/data", "--local", str(missing)])
+
+        captured = capsys.readouterr()
+        assert rc == 1
+        assert "does not exist" in captured.err
+        # Nothing should have been written to S3.
+        mock_boto_client.return_value.upload_file.assert_not_called()
+
+    @patch(BOTO3_PATCH_TARGET)
+    def test_upload_uploads_existing_local_source(self, mock_boto_client):
+        mock_s3 = _setup_s3_mock(mock_boto_client)
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            src = Path(tmpdir) / "src"
+            src.mkdir()
+            (src / "file.txt").write_text("content")
+
+            rc = main(["upload", "s3://bucket/data", "--local", str(src)])
+
+        assert rc == 0
+        assert mock_s3.upload_file.call_count >= 1
+
+    @patch(BOTO3_PATCH_TARGET)
+    def test_upload_default_source_is_cache_path(self, mock_boto_client):
+        """When --local is omitted, source defaults to the same cache path pos3 download
+        would have produced."""
+        mock_s3 = _setup_s3_mock(mock_boto_client)
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            cache_root = Path(tmpdir)
+            cache_path = cache_root / "_" / "bucket" / "data"
+            cache_path.mkdir(parents=True)
+            (cache_path / "file.txt").write_text("hi")
+
+            # Override the CLI's mirror() to anchor cache_root at our tmpdir, so
+            # the no-`--local` path resolves under it.
+            import pos3
+
+            real_mirror = pos3.mirror
+
+            def fixed_mirror(**_kwargs):
+                return real_mirror(cache_root=str(cache_root), show_progress=False)
+
+            with patch("pos3.cli.mirror", side_effect=fixed_mirror):
+                rc = main(["upload", "s3://bucket/data"])
+
+            assert rc == 0
+            assert mock_s3.upload_file.call_count >= 1
+
+    @patch(BOTO3_PATCH_TARGET)
+    def test_upload_default_does_not_delete(self, mock_boto_client):
+        """CLI default for --delete is OFF; remote orphans survive."""
+        # S3 has remote_only.txt; local has file.txt. Without --delete, remote should stay.
+        paginate = [{"Contents": [{"Key": "data/remote_only.txt", "Size": 5}]}]
+        mock_s3 = _setup_s3_mock(mock_boto_client, paginate)
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            src = Path(tmpdir) / "src"
+            src.mkdir()
+            (src / "file.txt").write_text("content")
+
+            rc = main(["upload", "s3://bucket/data", "--local", str(src)])
+
+        assert rc == 0
+        # No deletes should have been issued.
+        assert mock_s3.delete_object.call_count == 0
+
+    @patch(BOTO3_PATCH_TARGET)
+    def test_upload_delete_flag_removes_remote_orphans(self, mock_boto_client):
+        paginate = [{"Contents": [{"Key": "data/remote_only.txt", "Size": 5}]}]
+        mock_s3 = _setup_s3_mock(mock_boto_client, paginate)
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            src = Path(tmpdir) / "src"
+            src.mkdir()
+            (src / "file.txt").write_text("content")
+
+            rc = main(["upload", "s3://bucket/data", "--local", str(src), "--delete"])
+
+        assert rc == 0
+        assert mock_s3.delete_object.call_count >= 1
+
+
+class TestCliEntry:
+    def test_no_subcommand_exits_with_error(self, capsys):
+        with pytest.raises(SystemExit) as exc:
+            main([])
+        assert exc.value.code != 0

From 9da3a7b6befdaf9741e5de54767555263eb3b0e5 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Sat, 23 May 2026 16:02:59 +0000
Subject: [PATCH 02/16] Bump version to 0.3.1 for CLI release

v0.3.0 is already released; the pos3 CLI added in the previous commit
ships as 0.3.1. CHANGELOG section moves from [Unreleased] to [0.3.1].

https://claude.ai/code/session_01QoQaZ2G6FsNpSZyStqTXas
---
 CHANGELOG.md   | 2 +-
 pyproject.toml | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 4356bad..857005a 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,6 +1,6 @@
 # Changelog
 
-## [Unreleased]
+## [0.3.1] - 2026-05-21
 
 ### Added
 - `pos3` console-script entry point with `ls`, `download`, `upload` subcommands
diff --git a/pyproject.toml b/pyproject.toml
index 6699609..4d6627e 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "pos3"
-version = "0.3.0"
+version = "0.3.1"
 description = "S3 Simple Sync - Make using S3 as simple as using local files"
 readme = "README.md"
 requires-python = ">=3.11"

From db93dd27a06a4edce3918c683feb86c31781647d Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Sat, 23 May 2026 17:44:30 +0000
Subject: [PATCH 03/16] Add --dry-run/-n to pos3 download and upload

The flag is accepted only on the two transfer subcommands (ls rejects it).
Dry-run reuses the existing _compute_sync_diff so the plan reflects what a
real run would do; output is `aws s3 sync --dryrun`-style one-line-per-file
on stdout. No transfers, deletes, or directory creation happen. Synthesized
directory entries from _scan_s3 are skipped so the output is file-level.

https://claude.ai/code/session_01QoQaZ2G6FsNpSZyStqTXas
---
 CHANGELOG.md      |   3 ++
 README.md         |   8 +++
 pos3/cli.py       | 130 +++++++++++++++++++++++++++++++++++++++-------
 tests/test_cli.py | 108 ++++++++++++++++++++++++++++++++++++++
 4 files changed, 231 insertions(+), 18 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 857005a..9e6c7d4 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -18,6 +18,9 @@
     defaults to `True`; the CLI is more conservative for interactive use).
   - `--profile` is supported alongside the URL form `s3://<profile>@bucket/...`;
     the URL form wins on conflict, matching the Python precedence.
+  - `-n` / `--dry-run` on `download` and `upload` prints the planned per-file
+    actions to stdout (in `aws s3 sync --dryrun` style) and performs no
+    transfers, deletes, or directory creation.
 
 ## [0.3.0] - 2026-05-19
 
diff --git a/README.md b/README.md
index dc3a661..07a341b 100644
--- a/README.md
+++ b/README.md
@@ -132,6 +132,10 @@ data_dir=$(pos3 download s3://bucket/dataset/)
 # One-shot upload. Source defaults to the same cache path `pos3 download`
 # would have produced; --local overrides. Errors if the source doesn't exist.
 pos3 upload s3://bucket/results/ --local ./out/
+
+# Preview what download/upload would do, without touching anything.
+pos3 download -n s3://bucket/dataset/ --local ./data/ --delete
+pos3 upload -n s3://bucket/results/ --local ./out/
 ```
 
 All three subcommands accept `--profile NAME`. The URL form
@@ -142,6 +146,10 @@ All three subcommands accept `--profile NAME`. The URL form
 the Python API defaults to `True`. CLI defaults are conservative for
 interactive shell use; pass `--delete` explicitly to mirror file removals.
 
+`-n` / `--dry-run` is accepted on `download` and `upload` (not `ls`). It
+prints per-file plan lines to stdout in `aws s3 sync --dryrun` style and
+performs no transfers, deletes, or directory creation.
+
 The CLI is one-shot only — no background sync, no `pos3 sync` subcommand.
 Use the Python `pos3.mirror()` context manager when you need an interval-based
 loop or bi-directional sync over a job's lifetime.
diff --git a/pos3/cli.py b/pos3/cli.py
index 7caef05..3b63d40 100644
--- a/pos3/cli.py
+++ b/pos3/cli.py
@@ -9,6 +9,10 @@
 default is too destructive for interactive shell use. ``download`` writes
 only the resulting local path to stdout so the output is safe to capture in
 ``$(pos3 download ...)``; progress bars and logs go to stderr.
+
+``--dry-run`` / ``-n`` is accepted only on ``download`` and ``upload``: it
+prints the planned per-file actions to stdout in ``aws s3 sync --dryrun``
+style and performs no transfers, deletes, or directory creation.
 """
 
 from __future__ import annotations
@@ -17,7 +21,19 @@
 import sys
 from pathlib import Path
 
-from . import _is_s3_path, _require_active_mirror, download, ls, mirror, upload
+from . import (
+    _compute_sync_diff,
+    _filter_fileinfo,
+    _is_s3_path,
+    _make_s3_key,
+    _parse_s3_url,
+    _require_active_mirror,
+    _scan_local,
+    download,
+    ls,
+    mirror,
+    upload,
+)
 from .profiles import _resolve_profile, _url_profile
 
 
@@ -46,6 +62,12 @@ def add_transfer_args(p: argparse.ArgumentParser, local_help: str) -> None:
             default=None,
             help="Glob pattern to skip. May be passed multiple times.",
         )
+        p.add_argument(
+            "-n",
+            "--dry-run",
+            action="store_true",
+            help="Print the planned actions and exit without transferring or deleting anything.",
+        )
         add_profile(p)
 
     p_ls = subparsers.add_parser("ls", help="List objects under a prefix.")
@@ -72,6 +94,10 @@ def _cmd_ls(args: argparse.Namespace) -> int:
 
 
 def _cmd_download(args: argparse.Namespace) -> int:
+    if args.dry_run:
+        with mirror(show_progress=False):
+            _print_download_plan(args)
+        return 0
     with mirror(show_progress=True):
         local_path = download(
             args.url,
@@ -84,25 +110,39 @@ def _cmd_download(args: argparse.Namespace) -> int:
     return 0
 
 
+def _resolve_upload_source(args: argparse.Namespace) -> Path | None:
+    """Resolve the local source path, mirroring the precedence used by upload().
+
+    Prints an error and returns None if the source path does not exist.
+    """
+    mirror_obj = _require_active_mirror()
+    if args.local:
+        source = Path(args.local).expanduser().resolve()
+    else:
+        # Resolve the same profile precedence (URL > --profile > context default)
+        # the upload() call will use, so the cache path we check matches the one
+        # upload() would target.
+        profile: str | None = args.profile
+        if _is_s3_path(args.url):
+            url_profile = _url_profile(args.url)
+            if url_profile is not None:
+                profile = url_profile
+        effective_profile = _resolve_profile(profile) or mirror_obj.options.default_profile
+        source = mirror_obj.options.cache_path_for(args.url, effective_profile)
+    if not source.exists():
+        print(f"pos3 upload: source path does not exist: {source}", file=sys.stderr)
+        return None
+    return source
+
+
 def _cmd_upload(args: argparse.Namespace) -> int:
-    with mirror(show_progress=True):
-        mirror_obj = _require_active_mirror()
-        if args.local:
-            source = Path(args.local).expanduser().resolve()
-        else:
-            # Resolve the same profile precedence (URL > --profile > context default)
-            # the upload() call will use, so the cache path we check matches the one
-            # upload() would target.
-            profile: str | None = args.profile
-            if _is_s3_path(args.url):
-                url_profile = _url_profile(args.url)
-                if url_profile is not None:
-                    profile = url_profile
-            effective_profile = _resolve_profile(profile) or mirror_obj.options.default_profile
-            source = mirror_obj.options.cache_path_for(args.url, effective_profile)
-        if not source.exists():
-            print(f"pos3 upload: source path does not exist: {source}", file=sys.stderr)
+    with mirror(show_progress=not args.dry_run):
+        source = _resolve_upload_source(args)
+        if source is None:
             return 1
+        if args.dry_run:
+            _print_upload_plan(args, source)
+            return 0
         upload(
             args.url,
             local=source,
@@ -114,6 +154,60 @@ def _cmd_upload(args: argparse.Namespace) -> int:
     return 0
 
 
+def _print_download_plan(args: argparse.Namespace) -> None:
+    if not _is_s3_path(args.url):
+        return  # download() on a local path is a no-op pass-through.
+    mirror_obj = _require_active_mirror()
+    profile = mirror_obj._effective_profile(args.profile, args.url)
+    local_path = (
+        mirror_obj.options.cache_path_for(args.url, profile)
+        if args.local is None
+        else Path(args.local).expanduser().resolve()
+    )
+    bucket, prefix = _parse_s3_url(args.url)
+    to_copy, to_delete = _compute_sync_diff(
+        _filter_fileinfo(mirror_obj._scan_s3(bucket, prefix, profile), args.exclude),
+        _filter_fileinfo(_scan_local(local_path), args.exclude),
+    )
+    # Skip synthesized directory entries: only file-level actions matter for the user.
+    for info in to_copy:
+        if info.is_dir:
+            continue
+        s3_key = _make_s3_key(prefix, info)
+        dst = local_path / info.relative_path if info.relative_path else local_path
+        print(f"download: s3://{bucket}/{s3_key} to {dst}")
+    if args.delete:
+        for info in to_delete:
+            if info.is_dir:
+                continue
+            target = local_path / info.relative_path if info.relative_path else local_path
+            print(f"delete: {target}")
+
+
+def _print_upload_plan(args: argparse.Namespace, source: Path) -> None:
+    if not _is_s3_path(args.url):
+        return  # upload() on a local path is a no-op pass-through (it would mkdir).
+    mirror_obj = _require_active_mirror()
+    profile = mirror_obj._effective_profile(args.profile, args.url)
+    bucket, prefix = _parse_s3_url(args.url)
+    to_copy, to_delete = _compute_sync_diff(
+        _filter_fileinfo(_scan_local(source), args.exclude),
+        _filter_fileinfo(mirror_obj._scan_s3(bucket, prefix, profile), args.exclude),
+    )
+    for info in to_copy:
+        if info.is_dir:
+            continue
+        s3_key = _make_s3_key(prefix, info)
+        local = source / info.relative_path if info.relative_path else source
+        print(f"upload: {local} to s3://{bucket}/{s3_key}")
+    if args.delete:
+        for info in to_delete:
+            if info.is_dir:
+                continue
+            s3_key = _make_s3_key(prefix, info)
+            print(f"delete: s3://{bucket}/{s3_key}")
+
+
 _COMMANDS = {
     "ls": _cmd_ls,
     "download": _cmd_download,
diff --git a/tests/test_cli.py b/tests/test_cli.py
index 3d03f43..f1fe453 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -255,6 +255,114 @@ def test_upload_delete_flag_removes_remote_orphans(self, mock_boto_client):
         assert mock_s3.delete_object.call_count >= 1
 
 
+class TestCliDryRun:
+    @patch(BOTO3_PATCH_TARGET)
+    def test_download_dry_run_prints_plan_and_does_not_transfer(self, mock_boto_client, capsys):
+        paginate = [
+            {
+                "Contents": [
+                    {"Key": "data/file.txt", "Size": 5},
+                    {"Key": "data/sub/nested.txt", "Size": 7},
+                ]
+            }
+        ]
+        mock_s3 = _setup_s3_mock(mock_boto_client, paginate)
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            local_dir = Path(tmpdir) / "dst"
+            rc = main(["download", "-n", "s3://bucket/data", "--local", str(local_dir)])
+
+        captured = capsys.readouterr()
+        assert rc == 0
+        # No actual download was performed.
+        mock_s3.download_file.assert_not_called()
+        out_lines = captured.out.strip().splitlines()
+        # Two files planned, no extra trailing local-path line.
+        copy_lines = [line for line in out_lines if line.startswith("download:")]
+        assert len(copy_lines) == 2
+        assert any("s3://bucket/data/file.txt" in line for line in copy_lines)
+        assert any("s3://bucket/data/sub/nested.txt" in line for line in copy_lines)
+        assert all(" to " in line for line in copy_lines)
+        # No delete lines without --delete.
+        assert not any(line.startswith("delete:") for line in out_lines)
+
+    @patch(BOTO3_PATCH_TARGET)
+    def test_download_dry_run_with_delete_emits_delete_lines(self, mock_boto_client, capsys):
+        paginate = [{"Contents": [{"Key": "data/keep.txt", "Size": 5}]}]
+        mock_s3 = _setup_s3_mock(mock_boto_client, paginate)
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            local_dir = Path(tmpdir) / "data"
+            local_dir.mkdir()
+            (local_dir / "keep.txt").write_bytes(b"12345")
+            orphan = local_dir / "orphan.txt"
+            orphan.write_text("orphan")
+
+            rc = main(
+                ["download", "-n", "s3://bucket/data", "--local", str(local_dir), "--delete"]
+            )
+
+            assert rc == 0
+            # Dry-run must not touch the filesystem.
+            assert orphan.exists()
+
+        captured = capsys.readouterr()
+        delete_lines = [
+            line for line in captured.out.splitlines() if line.startswith("delete:")
+        ]
+        assert any(str(orphan) in line for line in delete_lines)
+        mock_s3.download_file.assert_not_called()
+
+    @patch(BOTO3_PATCH_TARGET)
+    def test_upload_dry_run_prints_plan_and_does_not_transfer(self, mock_boto_client, capsys):
+        mock_s3 = _setup_s3_mock(mock_boto_client)
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            src = Path(tmpdir) / "src"
+            src.mkdir()
+            (src / "file.txt").write_text("content")
+
+            rc = main(["upload", "-n", "s3://bucket/data", "--local", str(src)])
+
+        captured = capsys.readouterr()
+        assert rc == 0
+        mock_s3.upload_file.assert_not_called()
+        upload_lines = [
+            line for line in captured.out.splitlines() if line.startswith("upload:")
+        ]
+        assert len(upload_lines) == 1
+        assert "s3://bucket/data/file.txt" in upload_lines[0]
+        assert str(src / "file.txt") in upload_lines[0]
+
+    @patch(BOTO3_PATCH_TARGET)
+    def test_upload_dry_run_with_delete_emits_remote_delete_lines(self, mock_boto_client, capsys):
+        paginate = [{"Contents": [{"Key": "data/remote_only.txt", "Size": 5}]}]
+        mock_s3 = _setup_s3_mock(mock_boto_client, paginate)
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            src = Path(tmpdir) / "src"
+            src.mkdir()
+            (src / "file.txt").write_text("content")
+
+            rc = main(
+                ["upload", "-n", "s3://bucket/data", "--local", str(src), "--delete"]
+            )
+
+        captured = capsys.readouterr()
+        assert rc == 0
+        mock_s3.upload_file.assert_not_called()
+        mock_s3.delete_object.assert_not_called()
+        delete_lines = [
+            line for line in captured.out.splitlines() if line.startswith("delete:")
+        ]
+        assert any("s3://bucket/data/remote_only.txt" in line for line in delete_lines)
+
+    def test_dry_run_not_accepted_on_ls(self):
+        with pytest.raises(SystemExit) as exc:
+            main(["ls", "-n", "s3://bucket/data"])
+        assert exc.value.code != 0
+
+
 class TestCliEntry:
     def test_no_subcommand_exits_with_error(self, capsys):
         with pytest.raises(SystemExit) as exc:

From a60803fb317d0079fe75dc5c8323814809ee81dd Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Thu, 28 May 2026 14:21:10 +0000
Subject: [PATCH 04/16] CLI: reject non-s3:// urls for download and upload
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The Python API treats non-S3 inputs to download()/upload() as a local-path
passthrough — useful when calling code is polymorphic over local/remote
inputs, but in the CLI it meant `pos3 download bucket/path` (no `s3://`)
exited 0 and printed a local path without transferring anything. The CLI
help text on both commands already says "Source/Destination S3 URL", so a
typo could silently break shell pipelines.

Add an explicit s3:// guard in both _cmd_download and _cmd_upload. Tighten
the dry-run wording in the cli docstring, README, and CHANGELOG: dry-run
performs no transfers and no deletes, but the cache root is still mkdir'd
on mirror() entry the same way it is for any pos3 invocation.

`ls` is unchanged — it documents and supports both s3:// and local paths.

https://claude.ai/code/session_01QoQaZ2G6FsNpSZyStqTXas
---
 CHANGELOG.md      |  7 ++++++-
 README.md         |  7 ++++++-
 pos3/cli.py       | 30 ++++++++++++++++++------------
 tests/test_cli.py | 31 +++++++++++++++++++++++++++++++
 4 files changed, 61 insertions(+), 14 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 9e6c7d4..6417ff7 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -20,7 +20,12 @@
     the URL form wins on conflict, matching the Python precedence.
   - `-n` / `--dry-run` on `download` and `upload` prints the planned per-file
     actions to stdout (in `aws s3 sync --dryrun` style) and performs no
-    transfers, deletes, or directory creation.
+    transfers and no deletes. (The cache root is still initialized as it is
+    for any `pos3` invocation.)
+  - `download` and `upload` require an `s3://` URL. The Python API's
+    local-path passthrough still works in code; the CLI rejects non-S3
+    inputs with a clear error so a typo can't silently succeed. `ls` is
+    unchanged and still accepts both forms.
 
 ## [0.3.0] - 2026-05-19
 
diff --git a/README.md b/README.md
index 07a341b..e096ae8 100644
--- a/README.md
+++ b/README.md
@@ -148,7 +148,12 @@ interactive shell use; pass `--delete` explicitly to mirror file removals.
 
 `-n` / `--dry-run` is accepted on `download` and `upload` (not `ls`). It
 prints per-file plan lines to stdout in `aws s3 sync --dryrun` style and
-performs no transfers, deletes, or directory creation.
+performs no transfers and no deletes. (The cache root directory is
+initialized the same way it is for any `pos3` invocation.)
+
+`download` and `upload` require an `s3://` URL; non-S3 inputs are rejected
+with a non-zero exit. `ls` still accepts both `s3://` prefixes and local
+paths, matching the Python API.
 
 The CLI is one-shot only — no background sync, no `pos3 sync` subcommand.
 Use the Python `pos3.mirror()` context manager when you need an interval-based
diff --git a/pos3/cli.py b/pos3/cli.py
index 3b63d40..d8e6612 100644
--- a/pos3/cli.py
+++ b/pos3/cli.py
@@ -10,9 +10,10 @@
 only the resulting local path to stdout so the output is safe to capture in
 ``$(pos3 download ...)``; progress bars and logs go to stderr.
 
-``--dry-run`` / ``-n`` is accepted only on ``download`` and ``upload``: it
-prints the planned per-file actions to stdout in ``aws s3 sync --dryrun``
-style and performs no transfers, deletes, or directory creation.
+``--dry-run`` / ``-n`` (download and upload only) prints the planned
+per-file actions to stdout in ``aws s3 sync --dryrun`` style and performs
+no transfers and no deletes. (The cache root directory is initialized the
+same way it is for any pos3 invocation.)
 """
 
 from __future__ import annotations
@@ -94,6 +95,12 @@ def _cmd_ls(args: argparse.Namespace) -> int:
 
 
 def _cmd_download(args: argparse.Namespace) -> int:
+    if not _is_s3_path(args.url):
+        print(
+            f"pos3 download: url must be an s3:// URL, got: {args.url}",
+            file=sys.stderr,
+        )
+        return 1
     if args.dry_run:
         with mirror(show_progress=False):
             _print_download_plan(args)
@@ -122,11 +129,8 @@ def _resolve_upload_source(args: argparse.Namespace) -> Path | None:
         # Resolve the same profile precedence (URL > --profile > context default)
         # the upload() call will use, so the cache path we check matches the one
         # upload() would target.
-        profile: str | None = args.profile
-        if _is_s3_path(args.url):
-            url_profile = _url_profile(args.url)
-            if url_profile is not None:
-                profile = url_profile
+        url_profile = _url_profile(args.url)
+        profile = url_profile if url_profile is not None else args.profile
         effective_profile = _resolve_profile(profile) or mirror_obj.options.default_profile
         source = mirror_obj.options.cache_path_for(args.url, effective_profile)
     if not source.exists():
@@ -136,6 +140,12 @@ def _resolve_upload_source(args: argparse.Namespace) -> Path | None:
 
 
 def _cmd_upload(args: argparse.Namespace) -> int:
+    if not _is_s3_path(args.url):
+        print(
+            f"pos3 upload: url must be an s3:// URL, got: {args.url}",
+            file=sys.stderr,
+        )
+        return 1
     with mirror(show_progress=not args.dry_run):
         source = _resolve_upload_source(args)
         if source is None:
@@ -155,8 +165,6 @@ def _cmd_upload(args: argparse.Namespace) -> int:
 
 
 def _print_download_plan(args: argparse.Namespace) -> None:
-    if not _is_s3_path(args.url):
-        return  # download() on a local path is a no-op pass-through.
     mirror_obj = _require_active_mirror()
     profile = mirror_obj._effective_profile(args.profile, args.url)
     local_path = (
@@ -185,8 +193,6 @@ def _print_download_plan(args: argparse.Namespace) -> None:
 
 
 def _print_upload_plan(args: argparse.Namespace, source: Path) -> None:
-    if not _is_s3_path(args.url):
-        return  # upload() on a local path is a no-op pass-through (it would mkdir).
     mirror_obj = _require_active_mirror()
     profile = mirror_obj._effective_profile(args.profile, args.url)
     bucket, prefix = _parse_s3_url(args.url)
diff --git a/tests/test_cli.py b/tests/test_cli.py
index f1fe453..9eb871a 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -165,6 +165,37 @@ def test_download_unknown_profile_returns_error(self, mock_boto_client, capsys):
         assert "Unknown profile" in captured.err
 
 
+class TestCliUrlValidation:
+    @patch(BOTO3_PATCH_TARGET)
+    def test_download_rejects_non_s3_url(self, mock_boto_client, capsys):
+        mock_s3 = _setup_s3_mock(mock_boto_client)
+
+        rc = main(["download", "bucket/data", "--local", "/tmp/ignored"])
+
+        captured = capsys.readouterr()
+        assert rc == 1
+        assert "must be an s3:// URL" in captured.err
+        # Nothing should have happened on stdout, nothing on the wire.
+        assert captured.out == ""
+        mock_s3.download_file.assert_not_called()
+
+    @patch(BOTO3_PATCH_TARGET)
+    def test_upload_rejects_non_s3_url(self, mock_boto_client, capsys, tmp_path):
+        mock_s3 = _setup_s3_mock(mock_boto_client)
+        src = tmp_path / "src"
+        src.mkdir()
+        (src / "file.txt").write_text("x")
+
+        rc = main(["upload", str(tmp_path / "dst"), "--local", str(src)])
+
+        captured = capsys.readouterr()
+        assert rc == 1
+        assert "must be an s3:// URL" in captured.err
+        mock_s3.upload_file.assert_not_called()
+        # The bogus "destination" must not have been mkdir'd as a side effect.
+        assert not (tmp_path / "dst").exists()
+
+
 class TestCliUpload:
     @patch(BOTO3_PATCH_TARGET)
     def test_upload_errors_when_source_missing(self, mock_boto_client, capsys):

From e9da8ae8c9b7bfe18e363e02cb6a72ac2c50e733 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Thu, 28 May 2026 14:30:05 +0000
Subject: [PATCH 05/16] Propagate transfer failures via new pos3.TransferError

Per-worker failures in _process_futures were logged and swallowed, so a
download() / upload() with a failed S3 GET or PUT (403, network blip,
etc.) returned normally. Library callers got a path to a partial cache;
the new pos3 CLI exited 0 after printing that path, breaking the
`data_dir=$(pos3 download ...)` contract codex flagged.

Introduce pos3.TransferError(operation, failures). _process_futures now
collects per-worker exceptions and raises one TransferError once all
futures have been drained (we keep the "do as much as we can, then
report" semantics rather than fail-fast cancelling pending work). The
existing Mirror.download error handler already re-raises arbitrary
exceptions; for upload the failure surfaces from the mirror() context's
final sync. The CLI main() catches TransferError alongside ValueError
and exits 1 with the failure on stderr.

Tests: two new TestCliTransferFailures cases inject side_effect on
download_file / upload_file and assert rc == 1, no path on stdout for
download.

https://claude.ai/code/session_01QoQaZ2G6FsNpSZyStqTXas
---
 CHANGELOG.md      | 10 ++++++++++
 pos3/__init__.py  | 20 ++++++++++++++++++++
 pos3/cli.py       |  3 ++-
 tests/test_cli.py | 36 ++++++++++++++++++++++++++++++++++++
 4 files changed, 68 insertions(+), 1 deletion(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 6417ff7..106c0bc 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -27,6 +27,16 @@
     inputs with a clear error so a typo can't silently succeed. `ls` is
     unchanged and still accepts both forms.
 
+### Changed
+- Per-object transfer failures now raise `pos3.TransferError` instead of
+  being logged and swallowed. Previously, if any worker in a download or
+  upload batch failed, the error was sent to the logger and the call
+  returned normally — `data_dir = pos3.download(...)` could return a path
+  to a partial cache, and `pos3 download` exited 0 after a failed S3 GET.
+  Both now propagate. The new exception exposes `.operation` and
+  `.failures` (list of the underlying per-worker exceptions). The CLI
+  catches it and exits 1 with the failure on stderr.
+
 ## [0.3.0] - 2026-05-19
 
 ### Added
diff --git a/pos3/__init__.py b/pos3/__init__.py
index 6892179..e521735 100644
--- a/pos3/__init__.py
+++ b/pos3/__init__.py
@@ -80,12 +80,29 @@ def _make_s3_key(prefix: str, info: FileInfo) -> str:
     return key
 
 
+class TransferError(Exception):
+    """Raised when one or more S3 transfer or delete workers failed.
+
+    ``failures`` carries the per-worker exceptions; the message names the
+    operation (Download / Upload / Delete) and the failure count.
+    """
+
+    def __init__(self, operation: str, failures: list[BaseException]):
+        super().__init__(f"{operation}: {len(failures)} worker(s) failed: {failures[0]}")
+        self.operation = operation
+        self.failures = failures
+
+
 def _process_futures(futures, operation: str) -> None:
+    failures: list[BaseException] = []
     for future in futures:
         try:
             future.result()
         except Exception as exc:
             logger.error("%s failed: %s", operation, exc)
+            failures.append(exc)
+    if failures:
+        raise TransferError(operation, failures)
 
 
 @dataclass(frozen=True)
@@ -318,6 +335,7 @@ def download(
         Raises:
             FileNotFoundError: If remote is a local path that does not exist.
             ValueError: If download registration conflicts with an existing download or upload or parameters differ.
+            TransferError: If any object failed to download (was previously logged and swallowed).
         """
         effective_profile = self._effective_profile(profile, remote)
 
@@ -394,6 +412,8 @@ def upload(
 
         Raises:
             ValueError: If upload registration conflicts with an existing download or upload or parameters differ.
+            TransferError: Raised from the mirror context exit (or background sync) if any object failed to upload
+                or delete (was previously logged and swallowed).
         """
         effective_profile = self._effective_profile(profile, remote)
 
diff --git a/pos3/cli.py b/pos3/cli.py
index d8e6612..b457cb6 100644
--- a/pos3/cli.py
+++ b/pos3/cli.py
@@ -23,6 +23,7 @@
 from pathlib import Path
 
 from . import (
+    TransferError,
     _compute_sync_diff,
     _filter_fileinfo,
     _is_s3_path,
@@ -226,7 +227,7 @@ def main(argv: list[str] | None = None) -> int:
     args = parser.parse_args(argv)
     try:
         return _COMMANDS[args.command](args)
-    except ValueError as exc:
+    except (ValueError, TransferError) as exc:
         print(f"pos3 {args.command}: {exc}", file=sys.stderr)
         return 1
 
diff --git a/tests/test_cli.py b/tests/test_cli.py
index 9eb871a..1118279 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -286,6 +286,42 @@ def test_upload_delete_flag_removes_remote_orphans(self, mock_boto_client):
         assert mock_s3.delete_object.call_count >= 1
 
 
+class TestCliTransferFailures:
+    @patch(BOTO3_PATCH_TARGET)
+    def test_download_returns_nonzero_when_worker_fails(self, mock_boto_client, capsys, tmp_path):
+        paginate = [{"Contents": [{"Key": "data/file.txt", "Size": 5}]}]
+        mock_s3 = _setup_s3_mock(mock_boto_client, paginate)
+        mock_s3.download_file.side_effect = RuntimeError("boom")
+
+        local_dir = tmp_path / "dst"
+        rc = main(["download", "s3://bucket/data", "--local", str(local_dir)])
+
+        captured = capsys.readouterr()
+        assert rc == 1
+        # The success-path stdout (the local cache path) MUST NOT be printed
+        # on failure, since `data_dir=$(pos3 download …)` would treat the
+        # path as valid and downstream reads would silently use a partial cache.
+        assert captured.out == ""
+        assert "Download" in captured.err
+        assert "boom" in captured.err
+
+    @patch(BOTO3_PATCH_TARGET)
+    def test_upload_returns_nonzero_when_worker_fails(self, mock_boto_client, capsys, tmp_path):
+        mock_s3 = _setup_s3_mock(mock_boto_client)
+        mock_s3.upload_file.side_effect = RuntimeError("boom")
+
+        src = tmp_path / "src"
+        src.mkdir()
+        (src / "file.txt").write_text("content")
+
+        rc = main(["upload", "s3://bucket/data", "--local", str(src)])
+
+        captured = capsys.readouterr()
+        assert rc == 1
+        assert "Upload" in captured.err
+        assert "boom" in captured.err
+
+
 class TestCliDryRun:
     @patch(BOTO3_PATCH_TARGET)
     def test_download_dry_run_prints_plan_and_does_not_transfer(self, mock_boto_client, capsys):

From b99561e5edef61f838fa0fba766543feac371f24 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Thu, 28 May 2026 14:50:19 +0000
Subject: [PATCH 06/16] Keep background sync daemon alive across TransferError
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The previous commit made _process_futures raise, which was the right call
for one-shot download/upload (CLI must exit non-zero, library callers
must not silently get partial caches). But it also fired inside the
_background_worker daemon's _sync_uploads(due) call — a single transient
upload_file failure under `upload(..., interval=N)` would now kill the
thread and stop all future interval syncs for the rest of the mirror
context.

Wrap the _sync_uploads call inside the worker loop in try/except: log
and continue. Best-effort retry-next-tick is the right model for
periodic syncs. _final_sync (context exit) and one-shot download() still
propagate, so CLI behavior is unchanged.

Test: test_background_worker_survives_transfer_error registers an upload
with interval=1, makes the first upload_file call raise, and sleeps 2.5s.
call_count >= 2 confirms the daemon survived and retried.

https://claude.ai/code/session_01QoQaZ2G6FsNpSZyStqTXas
---
 CHANGELOG.md     |  6 +++++-
 pos3/__init__.py |  9 ++++++++-
 tests/test_s3.py | 29 +++++++++++++++++++++++++++++
 3 files changed, 42 insertions(+), 2 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 106c0bc..e83eaee 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -35,7 +35,11 @@
   to a partial cache, and `pos3 download` exited 0 after a failed S3 GET.
   Both now propagate. The new exception exposes `.operation` and
   `.failures` (list of the underlying per-worker exceptions). The CLI
-  catches it and exits 1 with the failure on stderr.
+  catches it and exits 1 with the failure on stderr. **Background
+  interval syncs** (`upload(..., interval=N)`) are best-effort: a
+  `TransferError` from one tick is logged and the daemon continues so
+  the next interval can retry. Only the final sync on context exit and
+  any one-shot call propagate.
 
 ## [0.3.0] - 2026-05-19
 
diff --git a/pos3/__init__.py b/pos3/__init__.py
index e521735..ca25868 100644
--- a/pos3/__init__.py
+++ b/pos3/__init__.py
@@ -546,7 +546,14 @@ def _background_worker(self) -> None:
                         registration.last_sync = now
                         due.append(registration)
 
-            self._sync_uploads(due)
+            try:
+                self._sync_uploads(due)
+            except Exception as exc:
+                # Background interval syncs are best-effort: a TransferError
+                # (or anything else) here must not kill the daemon thread —
+                # the next tick will retry. _final_sync still propagates on
+                # context exit so one-shot callers see definitive failures.
+                logger.error("Background sync iteration failed: %s", exc)
 
     def _final_sync(self, had_error: bool = False) -> None:
         with self._lock:
diff --git a/tests/test_s3.py b/tests/test_s3.py
index 98662a7..a0b17d6 100644
--- a/tests/test_s3.py
+++ b/tests/test_s3.py
@@ -243,6 +243,35 @@ def test_background_sync_uploads_repeatedly(self, mock_boto_client):
 
         assert mock_s3.upload_file.call_count >= 2
 
+    @patch(BOTO3_PATCH_TARGET)
+    def test_background_worker_survives_transfer_error(self, mock_boto_client):
+        """A TransferError from one interval-sync iteration must not kill the
+        daemon thread; subsequent ticks should keep retrying."""
+        mock_s3 = _setup_s3_mock(mock_boto_client)
+
+        call_count = {"n": 0}
+
+        def upload_side_effect(*_args, **_kwargs):
+            call_count["n"] += 1
+            if call_count["n"] == 1:
+                raise RuntimeError("transient")
+            # Subsequent calls succeed (no return value needed).
+
+        mock_s3.upload_file.side_effect = upload_side_effect
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            output = Path(tmpdir) / "output"
+            output.mkdir()
+            (output / "data.txt").write_text("content")
+
+            with s3.mirror(cache_root=tmpdir, show_progress=False):
+                s3.upload("s3://bucket/output", local=output, interval=1)
+                time.sleep(2.5)
+
+        # If the worker had died after the first failure, count would stay at 1.
+        # Surviving means at least one more attempt happened on a later tick.
+        assert call_count["n"] >= 2
+
     @patch(BOTO3_PATCH_TARGET)
     def test_upload_no_sync_on_error(self, mock_boto_client):
         """Test that uploads with sync_on_error=False don't sync when context exits with error."""

From 95f493adb1f95873d777c9a726b33de788745cc2 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Thu, 28 May 2026 14:57:28 +0000
Subject: [PATCH 07/16] Lift plan as public API; make Mirror constructor
 side-effect free
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Three connected cleanups motivated by the codex review cycle:

1. Add pos3.TransferPlan (frozen dataclass with to_copy / to_delete) and
   Mirror.plan_download / Mirror.plan_upload. These compute the set of
   (source, destination) copies and target deletes a real call would
   perform, without performing any of them. The CLI's _print_*_plan
   helpers shrink from ~25 lines reaching into 6 private helpers to ~12
   lines calling one public method. Library callers can now ask "would
   download() do anything?" the same way the CLI does.

2. Drop the eager cache_root.mkdir from _Mirror.__init__. The leaf
   directory is still mkdir'd on demand by _put_locally (it does
   target.parent.mkdir(parents=True)), so download() / upload() behavior
   is unchanged. Dry-run and plan_* paths are now genuinely side-effect
   free — closes codex's earlier P2 for real, not just in docs.

3. _final_sync(had_error=True) now catches TransferError from the
   cleanup sync, logs it, and lets the original app exception propagate.
   Without this, an experiment that raises AppError followed by a failed
   cleanup upload would see TransferError as the top-level cause —
   regression from the pre-TransferError logged-and-continued behavior.
   Codex's most recent P2 ("Preserve the original exception during
   sync_on_error cleanup").

Also documents _process_futures' asymmetric error model so the
background-worker try/except is no longer surprising.

Tests added: TestPlan (plan_download / plan_upload + non-S3 rejection),
TestMirrorConstructorIsSideEffectFree, and
TestFinalSyncPreservesOriginalException.

https://claude.ai/code/session_01QoQaZ2G6FsNpSZyStqTXas
---
 CHANGELOG.md     |  18 +++++--
 README.md        |   6 ++-
 pos3/__init__.py | 130 ++++++++++++++++++++++++++++++++++++++++++++-
 pos3/cli.py      |  61 ++++++---------------
 tests/test_s3.py | 135 +++++++++++++++++++++++++++++++++++++++++++++++
 5 files changed, 299 insertions(+), 51 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index e83eaee..b8b292b 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -20,12 +20,16 @@
     the URL form wins on conflict, matching the Python precedence.
   - `-n` / `--dry-run` on `download` and `upload` prints the planned per-file
     actions to stdout (in `aws s3 sync --dryrun` style) and performs no
-    transfers and no deletes. (The cache root is still initialized as it is
-    for any `pos3` invocation.)
+    transfers, no deletes, and no directory creation.
   - `download` and `upload` require an `s3://` URL. The Python API's
     local-path passthrough still works in code; the CLI rejects non-S3
     inputs with a clear error so a typo can't silently succeed. `ls` is
     unchanged and still accepts both forms.
+- `pos3.TransferPlan` dataclass plus `Mirror.plan_download(remote, ...)` and
+  `Mirror.plan_upload(remote, ...)` methods: compute the set of
+  `(source, destination)` copies and target deletes a real call would
+  perform, without performing any of them. The CLI's `-n` / `--dry-run`
+  is implemented on top of these.
 
 ### Changed
 - Per-object transfer failures now raise `pos3.TransferError` instead of
@@ -39,7 +43,15 @@
   interval syncs** (`upload(..., interval=N)`) are best-effort: a
   `TransferError` from one tick is logged and the daemon continues so
   the next interval can retry. Only the final sync on context exit and
-  any one-shot call propagate.
+  any one-shot call propagate. **Cleanup on error** — when the
+  `mirror()` body is unwinding with an exception, a `TransferError` from
+  the `sync_on_error=True` cleanup sync is logged but swallowed, so the
+  original application exception remains the visible cause.
+- `pos3.mirror()` no longer creates the cache root directory eagerly on
+  context entry. The leaf directory is still created on demand when a
+  file is actually downloaded, so the visible behavior of `download()` is
+  unchanged. Dry-run and `plan_*` paths are now genuinely side-effect
+  free on the local filesystem.
 
 ## [0.3.0] - 2026-05-19
 
diff --git a/README.md b/README.md
index e096ae8..a7deaa0 100644
--- a/README.md
+++ b/README.md
@@ -148,8 +148,10 @@ interactive shell use; pass `--delete` explicitly to mirror file removals.
 
 `-n` / `--dry-run` is accepted on `download` and `upload` (not `ls`). It
 prints per-file plan lines to stdout in `aws s3 sync --dryrun` style and
-performs no transfers and no deletes. (The cache root directory is
-initialized the same way it is for any `pos3` invocation.)
+performs no transfers, no deletes, and no local directory creation. The
+underlying `Mirror.plan_download` / `Mirror.plan_upload` methods are
+public — call them directly to inspect what `download()` / `upload()`
+*would* do.
 
 `download` and `upload` require an `s3://` URL; non-S3 inputs are rejected
 with a non-zero exit. `ls` still accepts both `s3://` prefixes and local
diff --git a/pos3/__init__.py b/pos3/__init__.py
index ca25868..5d7fd5b 100644
--- a/pos3/__init__.py
+++ b/pos3/__init__.py
@@ -93,7 +93,34 @@ def __init__(self, operation: str, failures: list[BaseException]):
         self.failures = failures
 
 
+@dataclass(frozen=True)
+class TransferPlan:
+    """The set of operations a download/upload *would* perform, if executed.
+
+    Returned by :meth:`_Mirror.plan_download` and :meth:`_Mirror.plan_upload`.
+
+    ``to_copy`` is a list of ``(source, destination)`` pairs as strings —
+    each pair is one ``s3://bucket/key`` URL and one local filesystem path,
+    direction depending on whether the plan is for download or upload.
+    ``to_delete`` is a list of destination URLs / paths that ``delete=True``
+    on the corresponding call would remove. Directory-only entries from
+    the underlying scan are filtered out so the plan only reflects file
+    operations.
+    """
+
+    to_copy: list[tuple[str, str]]
+    to_delete: list[str]
+
+
 def _process_futures(futures, operation: str) -> None:
+    """Drain ``futures`` and raise :class:`TransferError` if any worker failed.
+
+    Drains all futures before raising, so a 1-of-N failure does not cancel
+    the rest of the batch (preserves best-effort completion). Callers that
+    need to keep going across failures wrap this in ``try / except`` — the
+    only such caller today is :meth:`_Mirror._background_worker`, which
+    treats interval syncs as retry-next-tick.
+    """
     failures: list[BaseException] = []
     for future in futures:
         try:
@@ -258,8 +285,11 @@ def __eq__(self, other):
 class _Mirror:
     def __init__(self, options: _Options):
         self.options = options
+        # cache_root is resolved eagerly but NOT created — the per-file
+        # workers (_put_locally) mkdir the leaf directory on demand. Keeping
+        # the constructor side-effect free lets dry-run / planning paths
+        # construct a Mirror without mutating the filesystem.
         self.cache_root = Path(self.options.cache_root).expanduser().resolve()
-        self.cache_root.mkdir(parents=True, exist_ok=True)
 
         self._default_profile = options.default_profile
         self._clients: dict[Profile | None, Any] = {}
@@ -455,6 +485,89 @@ def upload(
 
         return local_path
 
+    def plan_download(
+        self,
+        remote: str,
+        local: str | Path | None = None,
+        exclude: list[str] | None = None,
+        profile: str | Profile | None = None,
+    ) -> TransferPlan:
+        """Compute the :class:`TransferPlan` ``download()`` would execute.
+
+        Reads from S3 and the local filesystem but performs no transfers,
+        no deletes, and no directory creation. Raises ``ValueError`` if
+        ``remote`` is not an ``s3://`` URL.
+        """
+        if not _is_s3_path(remote):
+            raise ValueError(f"plan_download requires an s3:// URL, got: {remote}")
+        effective_profile = self._effective_profile(profile, remote)
+        local_path = (
+            self.options.cache_path_for(remote, effective_profile)
+            if local is None
+            else Path(local).expanduser().resolve()
+        )
+        bucket, prefix = _parse_s3_url(remote)
+        to_copy, to_delete = _compute_sync_diff(
+            _filter_fileinfo(self._scan_s3(bucket, prefix, effective_profile), exclude),
+            _filter_fileinfo(_scan_local(local_path), exclude),
+        )
+        copies: list[tuple[str, str]] = []
+        for info in to_copy:
+            if info.is_dir:
+                continue
+            s3_key = _make_s3_key(prefix, info)
+            dst = local_path / info.relative_path if info.relative_path else local_path
+            copies.append((f"s3://{bucket}/{s3_key}", str(dst)))
+        deletes: list[str] = []
+        for info in to_delete:
+            if info.is_dir:
+                continue
+            target = local_path / info.relative_path if info.relative_path else local_path
+            deletes.append(str(target))
+        return TransferPlan(to_copy=copies, to_delete=deletes)
+
+    def plan_upload(
+        self,
+        remote: str,
+        local: str | Path | None = None,
+        exclude: list[str] | None = None,
+        profile: str | Profile | None = None,
+    ) -> TransferPlan:
+        """Compute the :class:`TransferPlan` ``upload()`` would execute.
+
+        Reads from S3 and the local filesystem but performs no transfers,
+        no deletes, and no directory creation. Raises ``ValueError`` if
+        ``remote`` is not an ``s3://`` URL. A missing local source yields
+        an empty ``to_copy``.
+        """
+        if not _is_s3_path(remote):
+            raise ValueError(f"plan_upload requires an s3:// URL, got: {remote}")
+        effective_profile = self._effective_profile(profile, remote)
+        source = (
+            self.options.cache_path_for(remote, effective_profile)
+            if local is None
+            else Path(local).expanduser().resolve()
+        )
+        bucket, prefix = _parse_s3_url(remote)
+        to_copy, to_delete = _compute_sync_diff(
+            _filter_fileinfo(_scan_local(source), exclude),
+            _filter_fileinfo(self._scan_s3(bucket, prefix, effective_profile), exclude),
+        )
+        copies: list[tuple[str, str]] = []
+        for info in to_copy:
+            if info.is_dir:
+                continue
+            s3_key = _make_s3_key(prefix, info)
+            src = source / info.relative_path if info.relative_path else source
+            copies.append((str(src), f"s3://{bucket}/{s3_key}"))
+        deletes: list[str] = []
+        for info in to_delete:
+            if info.is_dir:
+                continue
+            s3_key = _make_s3_key(prefix, info)
+            deletes.append(f"s3://{bucket}/{s3_key}")
+        return TransferPlan(to_copy=copies, to_delete=deletes)
+
     def sync(
         self,
         remote: str,
@@ -559,8 +672,21 @@ def _final_sync(self, had_error: bool = False) -> None:
         with self._lock:
             uploads = list(self._uploads.values())
         if had_error:
+            # The mirror() context is already unwinding with the caller's
+            # exception. If this cleanup sync also fails, swallow it so the
+            # original exception stays the visible cause — a TransferError
+            # from cleanup is logged but must not mask the failure that
+            # triggered cleanup. The clean-exit path (had_error=False)
+            # still propagates so the CLI exits non-zero on failure.
             uploads = [u for u in uploads if u.sync_on_error]
-        self._sync_uploads(uploads)
+            try:
+                self._sync_uploads(uploads)
+            except TransferError as exc:
+                logger.error(
+                    "Cleanup sync after error failed; original exception preserved: %s", exc
+                )
+        else:
+            self._sync_uploads(uploads)
 
     def _sync_uploads(self, registrations: Iterable[_UploadRegistration]) -> None:
         tasks: list[tuple[str, Path, bool, list[str] | None, Profile | None]] = []
diff --git a/pos3/cli.py b/pos3/cli.py
index b457cb6..4066968 100644
--- a/pos3/cli.py
+++ b/pos3/cli.py
@@ -24,13 +24,8 @@
 
 from . import (
     TransferError,
-    _compute_sync_diff,
-    _filter_fileinfo,
     _is_s3_path,
-    _make_s3_key,
-    _parse_s3_url,
     _require_active_mirror,
-    _scan_local,
     download,
     ls,
     mirror,
@@ -166,53 +161,31 @@ def _cmd_upload(args: argparse.Namespace) -> int:
 
 
 def _print_download_plan(args: argparse.Namespace) -> None:
-    mirror_obj = _require_active_mirror()
-    profile = mirror_obj._effective_profile(args.profile, args.url)
-    local_path = (
-        mirror_obj.options.cache_path_for(args.url, profile)
-        if args.local is None
-        else Path(args.local).expanduser().resolve()
-    )
-    bucket, prefix = _parse_s3_url(args.url)
-    to_copy, to_delete = _compute_sync_diff(
-        _filter_fileinfo(mirror_obj._scan_s3(bucket, prefix, profile), args.exclude),
-        _filter_fileinfo(_scan_local(local_path), args.exclude),
+    plan = _require_active_mirror().plan_download(
+        args.url,
+        local=args.local,
+        exclude=args.exclude,
+        profile=args.profile,
     )
-    # Skip synthesized directory entries: only file-level actions matter for the user.
-    for info in to_copy:
-        if info.is_dir:
-            continue
-        s3_key = _make_s3_key(prefix, info)
-        dst = local_path / info.relative_path if info.relative_path else local_path
-        print(f"download: s3://{bucket}/{s3_key} to {dst}")
+    for src, dst in plan.to_copy:
+        print(f"download: {src} to {dst}")
     if args.delete:
-        for info in to_delete:
-            if info.is_dir:
-                continue
-            target = local_path / info.relative_path if info.relative_path else local_path
+        for target in plan.to_delete:
             print(f"delete: {target}")
 
 
 def _print_upload_plan(args: argparse.Namespace, source: Path) -> None:
-    mirror_obj = _require_active_mirror()
-    profile = mirror_obj._effective_profile(args.profile, args.url)
-    bucket, prefix = _parse_s3_url(args.url)
-    to_copy, to_delete = _compute_sync_diff(
-        _filter_fileinfo(_scan_local(source), args.exclude),
-        _filter_fileinfo(mirror_obj._scan_s3(bucket, prefix, profile), args.exclude),
+    plan = _require_active_mirror().plan_upload(
+        args.url,
+        local=str(source),
+        exclude=args.exclude,
+        profile=args.profile,
     )
-    for info in to_copy:
-        if info.is_dir:
-            continue
-        s3_key = _make_s3_key(prefix, info)
-        local = source / info.relative_path if info.relative_path else source
-        print(f"upload: {local} to s3://{bucket}/{s3_key}")
+    for src, dst in plan.to_copy:
+        print(f"upload: {src} to {dst}")
     if args.delete:
-        for info in to_delete:
-            if info.is_dir:
-                continue
-            s3_key = _make_s3_key(prefix, info)
-            print(f"delete: s3://{bucket}/{s3_key}")
+        for target in plan.to_delete:
+            print(f"delete: {target}")
 
 
 _COMMANDS = {
diff --git a/tests/test_s3.py b/tests/test_s3.py
index a0b17d6..18f1624 100644
--- a/tests/test_s3.py
+++ b/tests/test_s3.py
@@ -715,6 +715,141 @@ def test_sync_conflicts(self, mock_boto_client):
                     s3.sync("s3://bucket/data", interval=None)
 
 
+class TestPlan:
+    """plan_download / plan_upload return what a real call would do, without
+    transferring, deleting, or creating directories."""
+
+    @patch(BOTO3_PATCH_TARGET)
+    def test_plan_download_lists_files_to_copy(self, mock_boto_client):
+        paginate = [
+            {
+                "Contents": [
+                    {"Key": "data/file.txt", "Size": 5},
+                    {"Key": "data/sub/nested.txt", "Size": 7},
+                ]
+            }
+        ]
+        mock_s3 = _setup_s3_mock(mock_boto_client, paginate)
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            local_dir = Path(tmpdir) / "dst"
+            with s3.mirror(cache_root=tmpdir, show_progress=False) as _:
+                from pos3 import _require_active_mirror
+
+                plan = _require_active_mirror().plan_download(
+                    "s3://bucket/data", local=str(local_dir)
+                )
+
+        sources = [src for src, _ in plan.to_copy]
+        assert "s3://bucket/data/file.txt" in sources
+        assert "s3://bucket/data/sub/nested.txt" in sources
+        # No real transfer happened.
+        mock_s3.download_file.assert_not_called()
+        # Directory entries are filtered out — file-level only.
+        assert all(not src.endswith("/") for src in sources)
+
+    @patch(BOTO3_PATCH_TARGET)
+    def test_plan_download_lists_orphans_in_to_delete(self, mock_boto_client):
+        paginate = [{"Contents": [{"Key": "data/keep.txt", "Size": 5}]}]
+        _setup_s3_mock(mock_boto_client, paginate)
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            local_dir = Path(tmpdir) / "data"
+            local_dir.mkdir()
+            (local_dir / "keep.txt").write_bytes(b"12345")
+            orphan = local_dir / "orphan.txt"
+            orphan.write_text("x")
+
+            with s3.mirror(cache_root=tmpdir, show_progress=False):
+                from pos3 import _require_active_mirror
+
+                plan = _require_active_mirror().plan_download(
+                    "s3://bucket/data", local=str(local_dir)
+                )
+
+            # Dry-plan is read-only.
+            assert orphan.exists()
+
+        assert str(orphan) in plan.to_delete
+
+    @patch(BOTO3_PATCH_TARGET)
+    def test_plan_upload_lists_files_to_copy(self, mock_boto_client):
+        mock_s3 = _setup_s3_mock(mock_boto_client)
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            src = Path(tmpdir) / "src"
+            src.mkdir()
+            (src / "file.txt").write_text("content")
+
+            with s3.mirror(cache_root=tmpdir, show_progress=False):
+                from pos3 import _require_active_mirror
+
+                plan = _require_active_mirror().plan_upload(
+                    "s3://bucket/data", local=str(src)
+                )
+
+        destinations = [dst for _, dst in plan.to_copy]
+        assert destinations == ["s3://bucket/data/file.txt"]
+        mock_s3.upload_file.assert_not_called()
+
+    def test_plan_download_rejects_non_s3_url(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            with s3.mirror(cache_root=tmpdir, show_progress=False):
+                from pos3 import _require_active_mirror
+
+                with pytest.raises(ValueError, match="s3:// URL"):
+                    _require_active_mirror().plan_download("/local/path")
+
+    def test_plan_upload_rejects_non_s3_url(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            with s3.mirror(cache_root=tmpdir, show_progress=False):
+                from pos3 import _require_active_mirror
+
+                with pytest.raises(ValueError, match="s3:// URL"):
+                    _require_active_mirror().plan_upload("/local/path", local=tmpdir)
+
+
+class TestMirrorConstructorIsSideEffectFree:
+    def test_constructing_mirror_does_not_create_cache_root(self):
+        """Constructing a Mirror (entering pos3.mirror()) must not mkdir the
+        cache root — dry-run and planning paths rely on this."""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            cache_root = Path(tmpdir) / "nested" / "cache" / "root"
+            assert not cache_root.exists()
+
+            with s3.mirror(cache_root=str(cache_root), show_progress=False):
+                # Just entering the context must not have created cache_root.
+                assert not cache_root.exists()
+
+
+class TestFinalSyncPreservesOriginalException:
+    @patch(BOTO3_PATCH_TARGET)
+    def test_app_exception_survives_failed_cleanup_sync(self, mock_boto_client):
+        """When the mirror body raises AND a sync_on_error upload's cleanup
+        sync also fails, the user must see the app exception, not the
+        TransferError from cleanup."""
+        mock_s3 = _setup_s3_mock(mock_boto_client)
+        mock_s3.upload_file.side_effect = RuntimeError("cleanup upload failed")
+
+        class AppError(Exception):
+            pass
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            output = Path(tmpdir) / "output"
+            output.mkdir()
+            (output / "data.txt").write_text("content")
+
+            with pytest.raises(AppError, match="the real failure"):
+                with s3.mirror(cache_root=tmpdir, show_progress=False):
+                    s3.upload(
+                        "s3://bucket/output",
+                        local=output,
+                        interval=None,
+                        sync_on_error=True,
+                    )
+                    raise AppError("the real failure")
+
+
 class TestLs:
     def test_ls_local_non_recursive(self):
         """Test non-recursive listing excludes nested items."""

From 8479ea52897ae6d9ca23f10caedbb5623ca2c605 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Thu, 28 May 2026 15:26:41 +0000
Subject: [PATCH 08/16] Normalize URL in plan_download / plan_upload before
 parsing
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

A trailing slash on the input URL — e.g. pos3 download -n s3://bucket/data/
— caused plan output to emit s3://bucket/data//file.txt because
_parse_s3_url("s3://bucket/data/") returns prefix "data/" and
_make_s3_key("data/", info) then appends another slash. Mirror.download
already normalizes via _normalize_s3_url before parsing, so the dry-run
plan was misstating exact keys vs what a real run would transfer.

Fix: in plan_download and plan_upload, call _parse_s3_url on
_normalize_s3_url(remote) instead of on the raw URL. Single new line
per method.

Tests: test_plan_download_normalizes_trailing_slash_in_url asserts
sources == ["s3://bucket/data/file.txt"]; the upload variant also
covers the to_delete branch since it goes through _make_s3_key too.

https://claude.ai/code/session_01QoQaZ2G6FsNpSZyStqTXas
---
 pos3/__init__.py |  9 +++++++--
 tests/test_s3.py | 42 ++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 49 insertions(+), 2 deletions(-)

diff --git a/pos3/__init__.py b/pos3/__init__.py
index 5d7fd5b..0f3f60e 100644
--- a/pos3/__init__.py
+++ b/pos3/__init__.py
@@ -506,7 +506,10 @@ def plan_download(
             if local is None
             else Path(local).expanduser().resolve()
         )
-        bucket, prefix = _parse_s3_url(remote)
+        # Normalize before parsing so a trailing slash on the input URL does
+        # not double up in the reconstructed S3 keys — keeps plan output
+        # byte-identical with what a real download() would actually transfer.
+        bucket, prefix = _parse_s3_url(_normalize_s3_url(remote))
         to_copy, to_delete = _compute_sync_diff(
             _filter_fileinfo(self._scan_s3(bucket, prefix, effective_profile), exclude),
             _filter_fileinfo(_scan_local(local_path), exclude),
@@ -548,7 +551,9 @@ def plan_upload(
             if local is None
             else Path(local).expanduser().resolve()
         )
-        bucket, prefix = _parse_s3_url(remote)
+        # Normalize before parsing so a trailing slash on the input URL does
+        # not double up in the reconstructed S3 keys.
+        bucket, prefix = _parse_s3_url(_normalize_s3_url(remote))
         to_copy, to_delete = _compute_sync_diff(
             _filter_fileinfo(_scan_local(source), exclude),
             _filter_fileinfo(self._scan_s3(bucket, prefix, effective_profile), exclude),
diff --git a/tests/test_s3.py b/tests/test_s3.py
index 18f1624..89ae827 100644
--- a/tests/test_s3.py
+++ b/tests/test_s3.py
@@ -808,6 +808,48 @@ def test_plan_upload_rejects_non_s3_url(self):
                 with pytest.raises(ValueError, match="s3:// URL"):
                     _require_active_mirror().plan_upload("/local/path", local=tmpdir)
 
+    @patch(BOTO3_PATCH_TARGET)
+    def test_plan_download_normalizes_trailing_slash_in_url(self, mock_boto_client):
+        """A trailing slash on the input URL must not produce s3://bucket/data//file.txt
+        in the plan — Mirror.download normalizes before parsing and plan_* must agree."""
+        paginate = [{"Contents": [{"Key": "data/file.txt", "Size": 5}]}]
+        _setup_s3_mock(mock_boto_client, paginate)
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            local_dir = Path(tmpdir) / "dst"
+            with s3.mirror(cache_root=tmpdir, show_progress=False):
+                from pos3 import _require_active_mirror
+
+                plan = _require_active_mirror().plan_download(
+                    "s3://bucket/data/", local=str(local_dir)
+                )
+
+        sources = [src for src, _ in plan.to_copy]
+        assert sources == ["s3://bucket/data/file.txt"]
+        assert all("//" not in src.replace("s3://", "") for src in sources)
+
+    @patch(BOTO3_PATCH_TARGET)
+    def test_plan_upload_normalizes_trailing_slash_in_url(self, mock_boto_client):
+        paginate = [{"Contents": [{"Key": "data/orphan.txt", "Size": 5}]}]
+        _setup_s3_mock(mock_boto_client, paginate)
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            src = Path(tmpdir) / "src"
+            src.mkdir()
+            (src / "file.txt").write_text("content")
+
+            with s3.mirror(cache_root=tmpdir, show_progress=False):
+                from pos3 import _require_active_mirror
+
+                plan = _require_active_mirror().plan_upload(
+                    "s3://bucket/data/", local=str(src)
+                )
+
+        destinations = [dst for _, dst in plan.to_copy]
+        assert destinations == ["s3://bucket/data/file.txt"]
+        # And the delete list, which also goes through _make_s3_key for upload.
+        assert plan.to_delete == ["s3://bucket/data/orphan.txt"]
+
 
 class TestMirrorConstructorIsSideEffectFree:
     def test_constructing_mirror_does_not_create_cache_root(self):

From 5334ec1f4e1c1a883c4b200415303a40611d38d9 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Thu, 28 May 2026 15:49:14 +0000
Subject: [PATCH 09/16] Fix ls on exact S3 object keys
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

ls() unconditionally appended "/" to a non-empty key before calling
_scan_s3 / _list_s3_objects. But _list_s3_objects already has the right
"try as exact object first, fall back to directory" logic — it does a
head_object on the raw key and only adds "/" after a 404. Forcing "/"
up front bypassed that probe, so `pos3 ls s3://bucket/results.json`
returned nothing for an existing object (head_object skipped, list with
prefix "results.json/" finds nothing).

Drop the forced slash. Add a single-object branch to the ls loop:
_scan_s3 yields FileInfo(relative_path="", is_dir=False) for an exact
key match, which we now emit as the input URL. Directory-listing
behavior is unchanged: when the key is a real prefix, head_object 404s,
_list_s3_objects appends "/" and lists, and we hit the existing
relative_path-based reconstruction.

The spurious-prefix concern from the old comment ("droid/recovery"
matching "droid/recovery_towels") is already covered by
_list_s3_objects' post-404 append.

Test: test_ls_single_object asserts `pos3 ls s3://bucket/results.json`
prints exactly that URL on stdout when head_object returns 200.

https://claude.ai/code/session_01QoQaZ2G6FsNpSZyStqTXas
---
 pos3/__init__.py  | 36 +++++++++++++++++++++++-------------
 tests/test_cli.py | 16 ++++++++++++++++
 2 files changed, 39 insertions(+), 13 deletions(-)

diff --git a/pos3/__init__.py b/pos3/__init__.py
index 0f3f60e..f3e0119 100644
--- a/pos3/__init__.py
+++ b/pos3/__init__.py
@@ -602,21 +602,31 @@ def ls(self, prefix: str, recursive: bool = False, profile: str | Profile | None
         if _is_s3_path(prefix):
             normalized = _normalize_s3_url(prefix)
             bucket, key = _parse_s3_url(normalized)
-            # Ensure directory-like listing by appending '/' to avoid spurious prefix matches
-            if key:
-                key = key + "/"
+            # Don't force a trailing "/" here. _list_s3_objects probes the
+            # key as an exact object first (via head_object) and only falls
+            # back to directory listing with a trailing "/" on 404. Forcing
+            # the slash here suppresses the exact-key probe, so
+            # `pos3 ls s3://bucket/results.json` would return nothing for an
+            # existing object. The "droid/recovery" vs "droid/recovery_towels"
+            # spurious-prefix case is already covered there.
             items = []
             for info in self._scan_s3(bucket, key, effective_profile):
-                if info.relative_path:
-                    # Skip nested items if not recursive
-                    if not recursive and "/" in info.relative_path:
-                        continue
-                    # Reconstruct the full S3 key
-                    if key:
-                        s3_key = key.rstrip("/") + "/" + info.relative_path
-                    else:
-                        s3_key = info.relative_path
-                    items.append(f"s3://{bucket}/{s3_key}")
+                if not info.relative_path:
+                    # Empty relative_path means either the root directory
+                    # marker (skip) or that ``key`` was the exact object key
+                    # — yield the input URL as a single-line result.
+                    if not info.is_dir:
+                        items.append(f"s3://{bucket}/{key}")
+                    continue
+                # Skip nested items if not recursive
+                if not recursive and "/" in info.relative_path:
+                    continue
+                # Reconstruct the full S3 key
+                if key:
+                    s3_key = key.rstrip("/") + "/" + info.relative_path
+                else:
+                    s3_key = info.relative_path
+                items.append(f"s3://{bucket}/{s3_key}")
             return items
         else:
             display_path = Path(prefix).expanduser()
diff --git a/tests/test_cli.py b/tests/test_cli.py
index 1118279..2f077f3 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -68,6 +68,22 @@ def test_ls_local_path(self, capsys):
         assert rc == 0
         assert str(base / "file.txt") in captured.out
 
+    @patch(BOTO3_PATCH_TARGET)
+    def test_ls_single_object(self, mock_boto_client, capsys):
+        """`pos3 ls s3://bucket/file.json` on an exact object key must return
+        the object URL, not an empty list. ls() used to force a trailing
+        slash, suppressing the head_object exact-key probe."""
+        mock_s3 = _setup_s3_mock(mock_boto_client)
+        # Override the default 404: this key IS an exact S3 object.
+        mock_s3.head_object.side_effect = None
+        mock_s3.head_object.return_value = {"ContentLength": 42}
+
+        rc = main(["ls", "s3://bucket/results.json"])
+
+        captured = capsys.readouterr()
+        assert rc == 0
+        assert captured.out.strip() == "s3://bucket/results.json"
+
 
 class TestCliDownload:
     @patch(BOTO3_PATCH_TARGET)

From d5e60b62d85d0eba72727712c86184a28b8202da Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Thu, 28 May 2026 15:56:34 +0000
Subject: [PATCH 10/16] Fix download() silently no-op'ing on exact S3 object
 keys
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

When the URL named a single object instead of a prefix (head_object
hit), _scan_s3 yielded TWO FileInfos with relative_path="":

  1) FileInfo("", N, is_dir=False)  — the file
  2) FileInfo("", 0, is_dir=True)   — the unconditional root-dir marker

_compute_sync_diff builds a dict keyed only on relative_path, so the
dir marker overwrote the file. _perform_download then mkdir'd the
destination via _put_locally's is_dir branch and never called
download_file. download() returned a path to that empty directory and
the CLI exited 0 — caller's `data_file=$(pos3 download s3://b/x.json)`
got a directory where the object should have been.

Fix: in _scan_s3, track whether we already emitted a file at
relative_path="" (the exact-object case) and suppress the redundant
root marker in that case. Directory listings are unaffected — for
prefix listings _list_s3_objects appends "/" so no listed key matches
prefix exactly, has_root_file stays False, the symmetry-with-_scan_local
root marker is still emitted.

Reproduced before fix:

  >>> with mirror(...): download('s3://bucket/results.json', local=...)
  download_file called: False  # !!
  local exists: True (as a directory)

After fix:

  >>> download_file called: True
  >>> args: ('bucket', 'results.json', '/.../results.json')

Tests:
  - test_download_single_object_calls_download_file (test_s3.py): API
    level, asserts download_file called with the right (bucket, key,
    dst).
  - test_download_single_object_calls_download_file (test_cli.py): CLI
    level, asserts rc==0, download_file called, and the local path is
    printed to stdout as the success contract.

https://claude.ai/code/session_01QoQaZ2G6FsNpSZyStqTXas
---
 pos3/__init__.py  | 11 ++++++++++-
 tests/test_cli.py | 17 +++++++++++++++++
 tests/test_s3.py  | 28 ++++++++++++++++++++++++++++
 3 files changed, 55 insertions(+), 1 deletion(-)

diff --git a/pos3/__init__.py b/pos3/__init__.py
index f3e0119..b3b1a8d 100644
--- a/pos3/__init__.py
+++ b/pos3/__init__.py
@@ -859,6 +859,13 @@ def _scan_s3(self, bucket: str, prefix: str, profile: Profile | None = None) ->
         logger.debug("Scanning S3: s3://%s/%s", bucket, prefix)
         seen_dirs: set[str] = set()
         has_content = False
+        # When the prefix matches an exact object (head_object hit in
+        # _list_s3_objects), we emit a file FileInfo with relative_path="".
+        # In that case the root-dir marker below would collide with it in
+        # _compute_sync_diff's dict-by-relative_path and silently win,
+        # causing download() to mkdir the local path instead of fetching
+        # the object. Track this to suppress the redundant marker.
+        has_root_file = False
 
         for obj in self._list_s3_objects(bucket, prefix, profile):
             has_content = True
@@ -871,6 +878,8 @@ def _scan_s3(self, bucket: str, prefix: str, profile: Profile | None = None) ->
                     yield FileInfo(relative_path=relative, size=0, is_dir=True)
                     seen_dirs.add(relative)
             else:
+                if relative == "":
+                    has_root_file = True
                 yield FileInfo(relative_path=relative, size=obj["Size"], is_dir=False)
 
                 if "/" in relative:
@@ -881,7 +890,7 @@ def _scan_s3(self, bucket: str, prefix: str, profile: Profile | None = None) ->
                             yield FileInfo(relative_path=dir_path, size=0, is_dir=True)
                             seen_dirs.add(dir_path)
 
-        if has_content:
+        if has_content and not has_root_file:
             yield FileInfo(
                 relative_path="", size=0, is_dir=True
             )  # Yield root directory marker for symmetry with _scan_local
diff --git a/tests/test_cli.py b/tests/test_cli.py
index 2f077f3..9f01462 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -170,6 +170,23 @@ def test_download_exclude_multiple_patterns(self, mock_boto_client):
         downloaded_key = mock_s3.download_file.call_args_list[0][0][1]
         assert "file.txt" in downloaded_key
 
+    @patch(BOTO3_PATCH_TARGET)
+    def test_download_single_object_calls_download_file(self, mock_boto_client, capsys, tmp_path):
+        """`pos3 download s3://bucket/results.json` must actually fetch the
+        object, not silently mkdir the destination and exit 0."""
+        mock_s3 = _setup_s3_mock(mock_boto_client)
+        mock_s3.head_object.side_effect = None
+        mock_s3.head_object.return_value = {"ContentLength": 42, "Size": 42}
+
+        local = tmp_path / "results.json"
+        rc = main(["download", "s3://bucket/results.json", "--local", str(local)])
+
+        assert rc == 0
+        assert mock_s3.download_file.call_count == 1
+        # Stdout still emits the local path on success.
+        captured = capsys.readouterr()
+        assert captured.out.strip() == str(local)
+
     @patch(BOTO3_PATCH_TARGET)
     def test_download_unknown_profile_returns_error(self, mock_boto_client, capsys):
         _setup_s3_mock(mock_boto_client)
diff --git a/tests/test_s3.py b/tests/test_s3.py
index 89ae827..7c1d53b 100644
--- a/tests/test_s3.py
+++ b/tests/test_s3.py
@@ -851,6 +851,34 @@ def test_plan_upload_normalizes_trailing_slash_in_url(self, mock_boto_client):
         assert plan.to_delete == ["s3://bucket/data/orphan.txt"]
 
 
+class TestSingleObjectDownload:
+    """Downloading an exact S3 object key (not a prefix) must actually fetch
+    the file. _scan_s3 used to emit a root directory marker that collided
+    with the file in _compute_sync_diff's dict, causing download() to mkdir
+    the local path instead of calling download_file."""
+
+    @patch(BOTO3_PATCH_TARGET)
+    def test_download_single_object_calls_download_file(self, mock_boto_client):
+        mock_s3 = Mock()
+        mock_boto_client.return_value = mock_s3
+        # head_object 200 → _list_s3_objects yields the exact object.
+        mock_s3.head_object.return_value = {"ContentLength": 42, "Size": 42}
+        mock_paginator = Mock()
+        mock_s3.get_paginator.return_value = mock_paginator
+        mock_paginator.paginate.return_value = [{"Contents": []}]
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            local = Path(tmpdir) / "results.json"
+            with s3.mirror(cache_root=tmpdir, show_progress=False):
+                s3.download("s3://bucket/results.json", local=str(local))
+
+        assert mock_s3.download_file.call_count == 1
+        args = mock_s3.download_file.call_args[0]
+        assert args[0] == "bucket"
+        assert args[1] == "results.json"
+        assert args[2] == str(local)
+
+
 class TestMirrorConstructorIsSideEffectFree:
     def test_constructing_mirror_does_not_create_cache_root(self):
         """Constructing a Mirror (entering pos3.mirror()) must not mkdir the

From ef302d64d4446fb654714c9e28d3b23a4b4f3505 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Thu, 28 May 2026 17:32:18 +0000
Subject: [PATCH 11/16] Preserve trailing-slash directory intent in ls()
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

ls() normalized the input URL via _normalize_s3_url, which strips
trailing slashes. So `pos3 ls s3://bucket/data/` became a lookup of
key="data" — and if an object exactly named `data` also existed
alongside the `data/` "directory", head_object('data') hit in
_list_s3_objects and the single-object branch silently won, returning
only `s3://bucket/data` and hiding the directory contents the user
clearly asked for.

Fix: in ls(), use _parse_s3_url directly instead of normalizing first.
The trailing slash carries user intent ("treat as a directory prefix")
that _list_s3_objects already respects (it skips head_object when the
key ends in /). Other ls flows are unaffected: bucket-root (empty key)
and exact-object (no slash) both still behave the same.

Reproduced before fix:
  ls s3://bucket/data/  →  s3://bucket/data            (wrong)
After fix:
  ls s3://bucket/data/  →  s3://bucket/data/file.txt   (correct)
  ls s3://bucket/data   →  s3://bucket/data            (still correct,
                                                       single-object case)

Test: test_ls_trailing_slash_forces_directory_listing sets head_object
to return 200 (an exact 'data' object also exists) AND paginate to
return data/file.txt, then asserts ls s3://bucket/data/ yields only
the directory-content line.

https://claude.ai/code/session_01QoQaZ2G6FsNpSZyStqTXas
---
 pos3/__init__.py  | 11 +++++++++--
 tests/test_cli.py | 24 ++++++++++++++++++++++++
 2 files changed, 33 insertions(+), 2 deletions(-)

diff --git a/pos3/__init__.py b/pos3/__init__.py
index b3b1a8d..59e273e 100644
--- a/pos3/__init__.py
+++ b/pos3/__init__.py
@@ -600,8 +600,15 @@ def ls(self, prefix: str, recursive: bool = False, profile: str | Profile | None
         effective_profile = self._effective_profile(profile, prefix)
 
         if _is_s3_path(prefix):
-            normalized = _normalize_s3_url(prefix)
-            bucket, key = _parse_s3_url(normalized)
+            # Use _parse_s3_url directly, NOT _normalize_s3_url: the latter
+            # strips trailing slashes, but `s3://bucket/data/` vs
+            # `s3://bucket/data` carries user intent here — the trailing
+            # slash means "treat as a directory prefix". _list_s3_objects
+            # respects that ("if key and not key.endswith('/')") so we
+            # preserve it. Otherwise a key collision (an exact object named
+            # `data` plus a `data/` directory) would silently hide the
+            # directory contents.
+            bucket, key = _parse_s3_url(prefix)
             # Don't force a trailing "/" here. _list_s3_objects probes the
             # key as an exact object first (via head_object) and only falls
             # back to directory listing with a trailing "/" on 404. Forcing
diff --git a/tests/test_cli.py b/tests/test_cli.py
index 9f01462..b5f79e5 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -68,6 +68,30 @@ def test_ls_local_path(self, capsys):
         assert rc == 0
         assert str(base / "file.txt") in captured.out
 
+    @patch(BOTO3_PATCH_TARGET)
+    def test_ls_trailing_slash_forces_directory_listing(self, mock_boto_client, capsys):
+        """`pos3 ls s3://bucket/data/` must list the data/ contents even if
+        an object exactly named 'data' also exists. ls() used to strip the
+        trailing slash via _normalize_s3_url, letting head_object('data')
+        win and hide the directory contents."""
+        mock_s3 = _setup_s3_mock(
+            mock_boto_client,
+            [{"Contents": [{"Key": "data/file.txt", "Size": 5}]}],
+        )
+        # Pretend an object exactly named 'data' ALSO exists — without the
+        # fix, head_object('data') would return this and the directory
+        # listing would be skipped.
+        mock_s3.head_object.side_effect = None
+        mock_s3.head_object.return_value = {"ContentLength": 100}
+
+        rc = main(["ls", "s3://bucket/data/"])
+
+        captured = capsys.readouterr()
+        assert rc == 0
+        lines = captured.out.strip().splitlines()
+        # Should be the directory contents, not the exact-key 'data' object.
+        assert lines == ["s3://bucket/data/file.txt"]
+
     @patch(BOTO3_PATCH_TARGET)
     def test_ls_single_object(self, mock_boto_client, capsys):
         """`pos3 ls s3://bucket/file.json` on an exact object key must return

From 8353ac8ed2c71fecfbf92b197037ff9be5c94f0f Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Thu, 28 May 2026 17:39:52 +0000
Subject: [PATCH 12/16] Preserve trailing-slash scan intent in
 plan_download/plan_upload
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Same root issue as the ls fix in ef302d6, now in the plan_* methods.
After the earlier double-slash fix (8479ea5) both planners parsed the
*normalized* URL, which stripped the user's trailing slash before
_scan_s3 → _list_s3_objects. If both `data` (exact object) and `data/`
(directory) existed, head_object('data') hit the exact-key branch and
the plan reported the single object instead of the directory contents
the user asked for with the trailing slash.

But we can't simply drop the normalization — the earlier fix used it
to keep _make_s3_key from producing `data//file.txt`.

Resolution: split the prefix. Use the raw _parse_s3_url(remote) for the
S3 scan (so _list_s3_objects sees the trailing slash and skips
head_object) and the normalized _parse_s3_url(_normalize_s3_url(remote))
for output-key construction (so reconstructed URLs don't double up the
slash). _scan_s3 strips len(scan_prefix) then lstrip("/") so
FileInfo.relative_path is identical either way.

Reproduced before fix (with head_object('data') 200 AND paginate of
data/file.txt):
  download -n s3://bucket/data/   →  download: s3://bucket/data to /tmp/x
After:
  download -n s3://bucket/data/   →  download: s3://bucket/data/file.txt
                                              to /tmp/x/file.txt
  download -n s3://bucket/data    →  download: s3://bucket/data to /tmp/x
                                       (still correct — single object)

Test: test_plan_download_trailing_slash_forces_directory_listing
mocks both head_object('data') 200 and a data/file.txt in the
paginate; asserts plan.to_copy emits the directory-content URL.

https://claude.ai/code/session_01QoQaZ2G6FsNpSZyStqTXas
---
 pos3/__init__.py | 31 +++++++++++++++++++------------
 tests/test_s3.py | 26 ++++++++++++++++++++++++++
 2 files changed, 45 insertions(+), 12 deletions(-)

diff --git a/pos3/__init__.py b/pos3/__init__.py
index 59e273e..86d96c8 100644
--- a/pos3/__init__.py
+++ b/pos3/__init__.py
@@ -506,19 +506,24 @@ def plan_download(
             if local is None
             else Path(local).expanduser().resolve()
         )
-        # Normalize before parsing so a trailing slash on the input URL does
-        # not double up in the reconstructed S3 keys — keeps plan output
-        # byte-identical with what a real download() would actually transfer.
-        bucket, prefix = _parse_s3_url(_normalize_s3_url(remote))
+        # Two prefixes on purpose: the raw one preserves the user's
+        # trailing-slash intent so _list_s3_objects skips its exact-key
+        # head_object probe (matters when both `data` and `data/...` exist);
+        # the normalized one feeds _make_s3_key so reconstructed output keys
+        # don't get a double slash. _scan_s3 strips len(scan_prefix) then
+        # lstrip("/"), so the same FileInfo.relative_path drops out either
+        # way.
+        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(self._scan_s3(bucket, prefix, effective_profile), exclude),
+            _filter_fileinfo(self._scan_s3(scan_bucket, scan_prefix, effective_profile), exclude),
             _filter_fileinfo(_scan_local(local_path), exclude),
         )
         copies: list[tuple[str, str]] = []
         for info in to_copy:
             if info.is_dir:
                 continue
-            s3_key = _make_s3_key(prefix, info)
+            s3_key = _make_s3_key(out_prefix, info)
             dst = local_path / info.relative_path if info.relative_path else local_path
             copies.append((f"s3://{bucket}/{s3_key}", str(dst)))
         deletes: list[str] = []
@@ -551,25 +556,27 @@ def plan_upload(
             if local is None
             else Path(local).expanduser().resolve()
         )
-        # Normalize before parsing so a trailing slash on the input URL does
-        # not double up in the reconstructed S3 keys.
-        bucket, prefix = _parse_s3_url(_normalize_s3_url(remote))
+        # Two prefixes on purpose — see plan_download for the rationale. The
+        # raw prefix preserves trailing-slash intent for the S3 scan; the
+        # normalized one builds collision-free output keys.
+        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(source), exclude),
-            _filter_fileinfo(self._scan_s3(bucket, prefix, effective_profile), exclude),
+            _filter_fileinfo(self._scan_s3(scan_bucket, scan_prefix, effective_profile), exclude),
         )
         copies: list[tuple[str, str]] = []
         for info in to_copy:
             if info.is_dir:
                 continue
-            s3_key = _make_s3_key(prefix, info)
+            s3_key = _make_s3_key(out_prefix, info)
             src = source / info.relative_path if info.relative_path else source
             copies.append((str(src), f"s3://{bucket}/{s3_key}"))
         deletes: list[str] = []
         for info in to_delete:
             if info.is_dir:
                 continue
-            s3_key = _make_s3_key(prefix, info)
+            s3_key = _make_s3_key(out_prefix, info)
             deletes.append(f"s3://{bucket}/{s3_key}")
         return TransferPlan(to_copy=copies, to_delete=deletes)
 
diff --git a/tests/test_s3.py b/tests/test_s3.py
index 7c1d53b..f47548c 100644
--- a/tests/test_s3.py
+++ b/tests/test_s3.py
@@ -828,6 +828,32 @@ def test_plan_download_normalizes_trailing_slash_in_url(self, mock_boto_client):
         assert sources == ["s3://bucket/data/file.txt"]
         assert all("//" not in src.replace("s3://", "") for src in sources)
 
+    @patch(BOTO3_PATCH_TARGET)
+    def test_plan_download_trailing_slash_forces_directory_listing(self, mock_boto_client):
+        """`pos3 download -n s3://bucket/data/` must plan the directory
+        contents even if an exact object `data` also exists. Pre-fix,
+        plan_download normalized away the slash, head_object('data') won,
+        and the plan reported the exact-object copy instead of `data/*`."""
+        mock_s3 = _setup_s3_mock(
+            mock_boto_client,
+            [{"Contents": [{"Key": "data/file.txt", "Size": 5}]}],
+        )
+        # Exact 'data' object ALSO exists — without preserving the slash,
+        # head_object('data') would win and shadow the directory contents.
+        mock_s3.head_object.side_effect = None
+        mock_s3.head_object.return_value = {"ContentLength": 100}
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            with s3.mirror(cache_root=tmpdir, show_progress=False):
+                from pos3 import _require_active_mirror
+
+                plan = _require_active_mirror().plan_download(
+                    "s3://bucket/data/", local=str(Path(tmpdir) / "dst")
+                )
+
+        sources = [src for src, _ in plan.to_copy]
+        assert sources == ["s3://bucket/data/file.txt"]
+
     @patch(BOTO3_PATCH_TARGET)
     def test_plan_upload_normalizes_trailing_slash_in_url(self, mock_boto_client):
         paginate = [{"Contents": [{"Key": "data/orphan.txt", "Size": 5}]}]

From 64f4c6c026cc5ff0c3ebc3d9033aa9c40bca6dea Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Thu, 28 May 2026 17:50:07 +0000
Subject: [PATCH 13/16] Preserve trailing-slash intent in real download/upload
 paths
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Two related P2s flagged by codex:

1. The dry-run/plan fixes from 8353ac8 preserved the user's trailing
   slash, but the real Mirror.download → _perform_download and the
   _sync_uploads paths still normalized the URL before scanning. When
   both `data` (exact object) and `data/...` existed, head_object('data')
   won and download() pulled the wrong target — silently — into the
   destination, exiting 0.

2. plan_upload() with a missing local source returned to_delete = every
   remote object, but real _sync_uploads short-circuits any registration
   whose local_path.exists() is False (no transfers AND no deletes).
   Plan would claim "would delete all remote data" for an action that
   would in fact delete nothing.

Fixes:

- _perform_download: now uses the dual-prefix pattern (raw for scan,
  normalized for output keys). Mirror.download passes the raw `remote`
  through instead of `normalized`. Registration code is unchanged
  (registration is keyed on normalized form, which is the right
  identity for dedup).

- _UploadRegistration grows a `raw_remote` field (excluded from __eq__).
  Mirror.upload populates it from the user's input. _sync_uploads uses
  it for the dual-prefix pattern, falling back to `remote` if not set
  (defensive default of "" preserves backwards compat).

- plan_upload short-circuits to TransferPlan(to_copy=[], to_delete=[])
  when `source` does not exist — matches what real _sync_uploads would
  do for that registration.

Tests added under TestTrailingSlashRealTransfers:
  - test_download_trailing_slash_transfers_directory_contents
  - test_upload_trailing_slash_scans_directory_for_delete
  - test_plan_upload_empty_when_source_missing

Each one pretends both the exact key and the directory contents exist;
asserts the correct one is acted on (or, for plan_upload missing source,
that no action is planned at all).

https://claude.ai/code/session_01QoQaZ2G6FsNpSZyStqTXas
---
 pos3/__init__.py | 50 ++++++++++++++++++++++-------
 tests/test_s3.py | 83 ++++++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 121 insertions(+), 12 deletions(-)

diff --git a/pos3/__init__.py b/pos3/__init__.py
index 86d96c8..3179314 100644
--- a/pos3/__init__.py
+++ b/pos3/__init__.py
@@ -254,7 +254,7 @@ def __eq__(self, other):
 
 @dataclass
 class _UploadRegistration:
-    remote: str
+    remote: str  # normalized form — used as the registration / dedup key
     local_path: Path
     interval: int | None
     delete: bool
@@ -262,6 +262,11 @@ class _UploadRegistration:
     exclude: list[str] | None
     profile: Profile | 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
+    # _list_s3_objects and scans the directory as the user meant. NOT
+    # compared in __eq__ — only the normalized form determines identity.
+    raw_remote: str = ""
 
     def __eq__(self, other):
         if not isinstance(other, _UploadRegistration):
@@ -399,7 +404,9 @@ def download(
 
         if need_download:
             try:
-                self._perform_download(normalized, local_path, delete, exclude, effective_profile)
+                # Pass the raw URL — _perform_download preserves trailing
+                # slashes for scanning while normalizing for output keys.
+                self._perform_download(remote, local_path, delete, exclude, effective_profile)
             except Exception as exc:
                 registration.error = exc
                 registration.ready.set()
@@ -468,6 +475,7 @@ def upload(
             exclude=exclude,
             profile=effective_profile,
             last_sync=0,
+            raw_remote=remote,
         )
 
         with self._lock:
@@ -556,6 +564,14 @@ def plan_upload(
             if local is None
             else Path(local).expanduser().resolve()
         )
+        # Mirror real upload() behavior: _sync_uploads skips any registration
+        # whose local_path does not exist (does nothing — not even deletes).
+        # Without this short-circuit, _scan_local yields nothing while
+        # _scan_s3 yields remote objects, so _compute_sync_diff would
+        # falsely report every remote object as "would delete" — a plan the
+        # real call would never execute.
+        if not source.exists():
+            return TransferPlan(to_copy=[], to_delete=[])
         # Two prefixes on purpose — see plan_download for the rationale. The
         # raw prefix preserves trailing-slash intent for the S3 scan; the
         # normalized one builds collision-free output keys.
@@ -723,7 +739,11 @@ def _sync_uploads(self, registrations: Iterable[_UploadRegistration]) -> None:
             if registration.local_path.exists():
                 tasks.append(
                     (
-                        registration.remote,
+                        # raw_remote preserves the user's trailing slash so
+                        # _list_s3_objects skips its exact-key head_object
+                        # probe; out-key building below normalizes to avoid
+                        # double slashes.
+                        registration.raw_remote or registration.remote,
                         registration.local_path,
                         registration.delete,
                         registration.exclude,
@@ -740,19 +760,20 @@ def _sync_uploads(self, registrations: Iterable[_UploadRegistration]) -> None:
 
         for remote, local_path, delete, exclude, profile in tasks:
             logger.debug("Syncing upload: %s from %s (delete=%s)", remote, local_path, delete)
-            bucket, prefix = _parse_s3_url(remote)
+            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(bucket, prefix, profile), exclude),
+                _filter_fileinfo(self._scan_s3(scan_bucket, scan_prefix, profile), exclude),
             )
 
             for info in to_copy:
-                s3_key = _make_s3_key(prefix, info)
+                s3_key = _make_s3_key(out_prefix, info)
                 to_put.append((info, local_path, bucket, s3_key, profile))
                 total_bytes += info.size
 
             for info in to_delete if delete else []:
-                s3_key = _make_s3_key(prefix, info)
+                s3_key = _make_s3_key(out_prefix, info)
                 to_remove.append((bucket, s3_key, profile))
 
         if to_put:
@@ -790,16 +811,21 @@ def _perform_download(
         exclude: list[str] | None,
         profile: Profile | None = None,
     ) -> None:
-        bucket, prefix = _parse_s3_url(remote)
+        # Dual-prefix: raw remote preserves the user's trailing slash so
+        # _list_s3_objects skips its exact-key head_object probe (matters
+        # when both `data` and `data/` exist); normalized remote feeds
+        # _make_s3_key so output keys don't get double slashes.
+        scan_bucket, scan_prefix = _parse_s3_url(remote)
+        bucket, out_prefix = _parse_s3_url(_normalize_s3_url(remote))
         logger.debug(
             "Performing download: s3://%s/%s to %s (delete=%s)",
-            bucket,
-            prefix,
+            scan_bucket,
+            scan_prefix,
             local_path,
             delete,
         )
         to_copy, to_delete = _compute_sync_diff(
-            _filter_fileinfo(self._scan_s3(bucket, prefix, profile), exclude),
+            _filter_fileinfo(self._scan_s3(scan_bucket, scan_prefix, profile), exclude),
             _filter_fileinfo(_scan_local(local_path), exclude),
         )
 
@@ -808,7 +834,7 @@ def _perform_download(
         total_bytes = 0
 
         for info in to_copy:
-            s3_key = _make_s3_key(prefix, info)
+            s3_key = _make_s3_key(out_prefix, info)
             to_put.append((info, bucket, s3_key, local_path))
             total_bytes += info.size
 
diff --git a/tests/test_s3.py b/tests/test_s3.py
index f47548c..3e0198c 100644
--- a/tests/test_s3.py
+++ b/tests/test_s3.py
@@ -877,6 +877,89 @@ def test_plan_upload_normalizes_trailing_slash_in_url(self, mock_boto_client):
         assert plan.to_delete == ["s3://bucket/data/orphan.txt"]
 
 
+class TestTrailingSlashRealTransfers:
+    """The real (non-dry-run) download() and upload() paths must preserve
+    the user's trailing-slash intent. Mirror.download / _sync_uploads used
+    to normalize the URL before scanning, letting head_object('data') win
+    over the requested data/ directory listing."""
+
+    @patch(BOTO3_PATCH_TARGET)
+    def test_download_trailing_slash_transfers_directory_contents(self, mock_boto_client):
+        mock_s3 = Mock()
+        mock_boto_client.return_value = mock_s3
+        # Both: an exact 'data' object AND objects under 'data/' exist.
+        mock_s3.head_object.return_value = {"ContentLength": 100}
+        mock_paginator = Mock()
+        mock_s3.get_paginator.return_value = mock_paginator
+        mock_paginator.paginate.return_value = [
+            {"Contents": [{"Key": "data/file.txt", "Size": 5}]}
+        ]
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            local = Path(tmpdir) / "dst"
+            with s3.mirror(cache_root=tmpdir, show_progress=False):
+                s3.download("s3://bucket/data/", local=str(local), delete=False)
+
+        # The directory content must be the one downloaded, not the exact
+        # 'data' object that head_object would have picked up.
+        keys = [c[0][1] for c in mock_s3.download_file.call_args_list]
+        assert keys == ["data/file.txt"]
+        assert "data" not in keys
+
+    @patch(BOTO3_PATCH_TARGET)
+    def test_upload_trailing_slash_scans_directory_for_delete(self, mock_boto_client):
+        mock_s3 = Mock()
+        mock_boto_client.return_value = mock_s3
+        # Exact 'data' object AND an orphan under 'data/' both exist.
+        mock_s3.head_object.return_value = {"ContentLength": 100}
+        mock_paginator = Mock()
+        mock_s3.get_paginator.return_value = mock_paginator
+        mock_paginator.paginate.return_value = [
+            {"Contents": [{"Key": "data/orphan.txt", "Size": 5}]}
+        ]
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            source = Path(tmpdir) / "src"
+            source.mkdir()
+            (source / "file.txt").write_text("x")
+
+            with s3.mirror(cache_root=tmpdir, show_progress=False):
+                s3.upload(
+                    "s3://bucket/data/",
+                    local=str(source),
+                    interval=None,
+                    delete=True,
+                )
+
+        # The directory orphan must be the one deleted, not the exact
+        # 'data' object.
+        deleted_keys = [c[1]["Key"] for c in mock_s3.delete_object.call_args_list]
+        assert "data/orphan.txt" in deleted_keys
+        assert "data" not in deleted_keys
+
+    @patch(BOTO3_PATCH_TARGET)
+    def test_plan_upload_empty_when_source_missing(self, mock_boto_client):
+        """Real _sync_uploads skips registrations whose local_path doesn't
+        exist (no transfers, no deletes). plan_upload must mirror that
+        instead of reporting every remote object as 'would delete'."""
+        _setup_s3_mock(
+            mock_boto_client,
+            [{"Contents": [{"Key": "data/orphan.txt", "Size": 5}]}],
+        )
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            missing = Path(tmpdir) / "does-not-exist"
+            with s3.mirror(cache_root=tmpdir, show_progress=False):
+                from pos3 import _require_active_mirror
+
+                plan = _require_active_mirror().plan_upload(
+                    "s3://bucket/data", local=str(missing)
+                )
+
+        assert plan.to_copy == []
+        assert plan.to_delete == []
+
+
 class TestSingleObjectDownload:
     """Downloading an exact S3 object key (not a prefix) must actually fetch
     the file. _scan_s3 used to emit a root directory marker that collided

From 7c32fa24fa36cb0b450a639fa32f347f72345302 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Sat, 30 May 2026 10:03:52 +0000
Subject: [PATCH 14/16] CLI: catch botocore errors + route unknown-arg errors
 through subparsers
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Two related UX gaps:

1. main() only caught (ValueError, TransferError). _scan_s3 →
   _list_s3_objects calls head_object/paginate before any worker future
   can wrap a failure in TransferError, and _list_s3_objects re-raises
   non-404 ClientErrors directly. So access-denied, missing-bucket,
   throttling, or expired-credential errors escaped main() and surfaced
   as a Python traceback, contradicting the documented
     pos3 <cmd>: <message>\nexit 1
   contract that callers like `data_dir=$(pos3 download …)` rely on.

   Catch botocore's BotoCoreError and ClientError alongside the existing
   ValueError / TransferError. Message format includes the exception
   class name so the user can tell auth errors from network errors
   without a traceback.

2. argparse routes "unrecognized arguments" through the TOP-LEVEL parser's
   error(). So `pos3 download s3://b/k --dry_run` (underscore typo) gave:

     usage: pos3 [-h] {ls,download,upload} ...
     pos3: error: unrecognized arguments: --dry_run

   which doesn't show the user that download has a -n / --dry-run flag.
   They had to separately run `pos3 download --help` to find it.

   Use parse_known_args + look up the chosen subparser via the
   _SubParsersAction so the SUBCOMMAND's usage is what gets printed:

     usage: pos3 download [-h] [--local PATH] [--delete] [--exclude PATTERN] [-n]
                          [--profile NAME]
                          url
     pos3 download: error: unrecognized arguments: --dry_run

Tests:
  - TestCliBotoErrors covers ls, real download, and dry-run download
    when head_object raises ClientError(403). Each asserts rc == 1, a
    pos3 <cmd>: prefix on stderr, and (for download) captured.out == "".
  - TestCliEntry.test_unknown_flag_uses_subcommand_usage asserts the
    subcommand usage line appears and the top-level
    `{ls,download,upload}` group does not.
  - test_upload_uploads_existing_local_source extended to assert
    captured.out == "" — the upload counterpart to download's "exactly
    one line = local path on stdout" contract.

https://claude.ai/code/session_01QoQaZ2G6FsNpSZyStqTXas
---
 pos3/cli.py       | 25 ++++++++++++++-
 tests/test_cli.py | 77 ++++++++++++++++++++++++++++++++++++++++++++++-
 2 files changed, 100 insertions(+), 2 deletions(-)

diff --git a/pos3/cli.py b/pos3/cli.py
index 4066968..66a0e02 100644
--- a/pos3/cli.py
+++ b/pos3/cli.py
@@ -22,6 +22,8 @@
 import sys
 from pathlib import Path
 
+from botocore.exceptions import BotoCoreError, ClientError
+
 from . import (
     TransferError,
     _is_s3_path,
@@ -197,12 +199,33 @@ def _print_upload_plan(args: argparse.Namespace, source: Path) -> None:
 
 def main(argv: list[str] | None = None) -> int:
     parser = _build_parser()
-    args = parser.parse_args(argv)
+    # Use parse_known_args so we can re-emit the error through the SUBPARSER
+    # for the chosen command. argparse's default routes "unrecognized
+    # arguments" to the top-level parser, which prints
+    #   usage: pos3 [-h] {ls,download,upload} ...
+    # — useless when the user typo'd a subcommand flag (e.g. `--dry_run`)
+    # because they can't see the flags the subcommand actually exposes.
+    namespace, leftover = parser.parse_known_args(argv)
+    if leftover:
+        subparsers_action = next(
+            a for a in parser._actions if isinstance(a, argparse._SubParsersAction)
+        )
+        sub = subparsers_action.choices.get(namespace.command, parser)
+        sub.error(f"unrecognized arguments: {' '.join(leftover)}")
+    args = namespace
     try:
         return _COMMANDS[args.command](args)
     except (ValueError, TransferError) as exc:
         print(f"pos3 {args.command}: {exc}", file=sys.stderr)
         return 1
+    except (BotoCoreError, ClientError) as exc:
+        # boto3/botocore failures from S3 calls outside _process_futures —
+        # access denied, missing bucket, expired creds, throttling, etc.
+        # _scan_s3 (called by ls and the pre-transfer scan in download /
+        # upload / plan_*) re-raises non-404 ClientErrors directly, so they
+        # would otherwise escape main() and surface as a Python traceback.
+        print(f"pos3 {args.command}: {type(exc).__name__}: {exc}", file=sys.stderr)
+        return 1
 
 
 if __name__ == "__main__":
diff --git a/tests/test_cli.py b/tests/test_cli.py
index b5f79e5..bcdbd34 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -269,7 +269,7 @@ def test_upload_errors_when_source_missing(self, mock_boto_client, capsys):
         mock_boto_client.return_value.upload_file.assert_not_called()
 
     @patch(BOTO3_PATCH_TARGET)
-    def test_upload_uploads_existing_local_source(self, mock_boto_client):
+    def test_upload_uploads_existing_local_source(self, mock_boto_client, capsys):
         mock_s3 = _setup_s3_mock(mock_boto_client)
 
         with tempfile.TemporaryDirectory() as tmpdir:
@@ -279,8 +279,13 @@ def test_upload_uploads_existing_local_source(self, mock_boto_client):
 
             rc = main(["upload", "s3://bucket/data", "--local", str(src)])
 
+        captured = capsys.readouterr()
         assert rc == 0
         assert mock_s3.upload_file.call_count >= 1
+        # upload's success path is silent on stdout: no path, no progress.
+        # Progress bars and logs go to stderr. This is the counterpart to
+        # download's "exactly one line = local path" contract.
+        assert captured.out == ""
 
     @patch(BOTO3_PATCH_TARGET)
     def test_upload_default_source_is_cache_path(self, mock_boto_client):
@@ -492,3 +497,73 @@ def test_no_subcommand_exits_with_error(self, capsys):
         with pytest.raises(SystemExit) as exc:
             main([])
         assert exc.value.code != 0
+
+    def test_unknown_flag_uses_subcommand_usage(self, capsys):
+        """Typoing a subcommand flag (e.g. `--dry_run` instead of `--dry-run`)
+        must produce the SUBCOMMAND's usage, not the top-level one, so the
+        user can see which flags actually exist for what they ran."""
+        with pytest.raises(SystemExit) as exc:
+            main(["download", "s3://bucket/key", "--dry_run"])
+        assert exc.value.code == 2
+        captured = capsys.readouterr()
+        # The error must address the subcommand, not the top-level parser.
+        assert "pos3 download" in captured.err
+        # And it must show download's actual flags so the user can spot `-n`.
+        assert "[-n]" in captured.err or "--dry-run" in captured.err
+        # Sanity: NOT the top-level usage.
+        assert "{ls,download,upload}" not in captured.err
+
+
+class TestCliBotoErrors:
+    @patch(BOTO3_PATCH_TARGET)
+    def test_ls_handles_client_error(self, mock_boto_client, capsys):
+        """A non-404 ClientError from _list_s3_objects (e.g. 403 access
+        denied) must produce the `pos3 ls: ...` error + exit 1, not a
+        Python traceback."""
+        mock_s3 = Mock()
+        mock_boto_client.return_value = mock_s3
+        mock_s3.head_object.side_effect = ClientError(
+            {"Error": {"Code": "403", "Message": "Forbidden"}}, "HeadObject"
+        )
+
+        rc = main(["ls", "s3://bucket/key"])
+
+        captured = capsys.readouterr()
+        assert rc == 1
+        assert "pos3 ls:" in captured.err
+        assert "ClientError" in captured.err
+        assert "403" in captured.err
+
+    @patch(BOTO3_PATCH_TARGET)
+    def test_download_handles_client_error_during_scan(self, mock_boto_client, capsys, tmp_path):
+        """ClientError raised from _scan_s3 (the pre-transfer scan inside
+        Mirror.download) must be caught by main(), not escape as a
+        traceback."""
+        mock_s3 = Mock()
+        mock_boto_client.return_value = mock_s3
+        mock_s3.head_object.side_effect = ClientError(
+            {"Error": {"Code": "403", "Message": "Forbidden"}}, "HeadObject"
+        )
+
+        rc = main(["download", "s3://bucket/data", "--local", str(tmp_path / "dst")])
+
+        captured = capsys.readouterr()
+        assert rc == 1
+        assert "pos3 download:" in captured.err
+        assert captured.out == ""
+
+    @patch(BOTO3_PATCH_TARGET)
+    def test_dry_run_handles_client_error_during_plan(self, mock_boto_client, capsys, tmp_path):
+        """plan_download propagates ClientError too — it goes through
+        _scan_s3 the same way."""
+        mock_s3 = Mock()
+        mock_boto_client.return_value = mock_s3
+        mock_s3.head_object.side_effect = ClientError(
+            {"Error": {"Code": "403", "Message": "Forbidden"}}, "HeadObject"
+        )
+
+        rc = main(["download", "-n", "s3://bucket/data", "--local", str(tmp_path / "dst")])
+
+        captured = capsys.readouterr()
+        assert rc == 1
+        assert "pos3 download:" in captured.err

From b3e7a840e21f1c9f2f0f9983ddb0cedd35ccbe62 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Sat, 30 May 2026 10:11:30 +0000
Subject: [PATCH 15/16] Expose plan API at the module level
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Codex's P2: I'd advertised Mirror.plan_download / plan_upload as
public API in the README and CHANGELOG, but they only lived on the
private _Mirror class. Module-level wrappers existed for download,
upload, sync, and ls, but not for the planning entry points; neither
TransferPlan nor TransferError were in __all__. Users following the
README would have had to reach into pos3._require_active_mirror() —
which is itself underscored.

Fix the surface, not just the docs:

- Add module-level pos3.plan_download(remote, ...) and
  pos3.plan_upload(remote, ...) wrappers, same shape as the existing
  pos3.download / pos3.upload (require an active mirror context, delegate
  to the active Mirror instance).
- Expand __all__ to include plan_download, plan_upload, TransferError,
  and TransferPlan.
- CLI now imports and calls these module-level wrappers (dogfood).
- README's dry-run section gets a Python example using
  pos3.plan_download, no longer references the private Mirror class.
- CHANGELOG entry updated accordingly.

TestPlanPublicAPI adds two tests: one asserting callability and __all__
membership for both wrappers, TransferPlan, and TransferError; another
exercising pos3.plan_download end-to-end via the public API.

Full suite: 136 passed.

https://claude.ai/code/session_01QoQaZ2G6FsNpSZyStqTXas
---
 CHANGELOG.md     | 10 ++++++----
 README.md        | 16 ++++++++++++----
 pos3/__init__.py | 50 +++++++++++++++++++++++++++++++++++++++++++++++-
 pos3/cli.py      |  6 ++++--
 tests/test_s3.py | 31 ++++++++++++++++++++++++++++++
 5 files changed, 102 insertions(+), 11 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index b8b292b..0eaecfe 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -25,11 +25,13 @@
     local-path passthrough still works in code; the CLI rejects non-S3
     inputs with a clear error so a typo can't silently succeed. `ls` is
     unchanged and still accepts both forms.
-- `pos3.TransferPlan` dataclass plus `Mirror.plan_download(remote, ...)` and
-  `Mirror.plan_upload(remote, ...)` methods: compute the set of
+- `pos3.TransferPlan` dataclass plus module-level `pos3.plan_download(remote, ...)`
+  and `pos3.plan_upload(remote, ...)` wrappers: compute the set of
   `(source, destination)` copies and target deletes a real call would
-  perform, without performing any of them. The CLI's `-n` / `--dry-run`
-  is implemented on top of these.
+  perform, without performing any of them. Same calling pattern as
+  `pos3.download` / `pos3.upload` — use them inside a `with pos3.mirror():`
+  block. The CLI's `-n` / `--dry-run` is implemented on top of these.
+- `TransferError` and `TransferPlan` are now in `pos3.__all__`.
 
 ### Changed
 - Per-object transfer failures now raise `pos3.TransferError` instead of
diff --git a/README.md b/README.md
index a7deaa0..e3d5477 100644
--- a/README.md
+++ b/README.md
@@ -148,10 +148,18 @@ interactive shell use; pass `--delete` explicitly to mirror file removals.
 
 `-n` / `--dry-run` is accepted on `download` and `upload` (not `ls`). It
 prints per-file plan lines to stdout in `aws s3 sync --dryrun` style and
-performs no transfers, no deletes, and no local directory creation. The
-underlying `Mirror.plan_download` / `Mirror.plan_upload` methods are
-public — call them directly to inspect what `download()` / `upload()`
-*would* do.
+performs no transfers, no deletes, and no local directory creation.
+
+The same plan is available from Python via `pos3.plan_download` and
+`pos3.plan_upload`, each returning a `pos3.TransferPlan` with
+`to_copy: list[tuple[str, str]]` and `to_delete: list[str]`:
+
+```python
+with pos3.mirror():
+    plan = pos3.plan_download("s3://bucket/dataset/", local="./data/")
+    for src, dst in plan.to_copy:
+        print(f"would download {src} → {dst}")
+```
 
 `download` and `upload` require an `s3://` URL; non-S3 inputs are rejected
 with a non-zero exit. `ls` still accepts both `s3://` prefixes and local
diff --git a/pos3/__init__.py b/pos3/__init__.py
index 3179314..ab5eaec 100644
--- a/pos3/__init__.py
+++ b/pos3/__init__.py
@@ -1169,4 +1169,52 @@ def ls(prefix: str, recursive: bool = False, profile: str | Profile | None = Non
     return mirror_obj.ls(prefix, recursive, profile)
 
 
-__all__ = ["mirror", "with_mirror", "download", "upload", "sync", "ls", "register_profile", "Profile", "_parse_s3_url"]
+def plan_download(
+    remote: str,
+    local: str | Path | None = None,
+    exclude: list[str] | None = None,
+    profile: str | Profile | None = None,
+) -> TransferPlan:
+    """Return the :class:`TransferPlan` ``download()`` would execute.
+
+    Side-effect free: reads from S3 and the local filesystem but performs
+    no transfers, deletes, or directory creation. Raises ``ValueError`` if
+    ``remote`` is not an ``s3://`` URL.
+    """
+    mirror_obj = _require_active_mirror()
+    return mirror_obj.plan_download(remote, local, exclude, profile)
+
+
+def plan_upload(
+    remote: str,
+    local: str | Path | None = None,
+    exclude: list[str] | None = None,
+    profile: str | Profile | None = None,
+) -> TransferPlan:
+    """Return the :class:`TransferPlan` ``upload()`` would execute.
+
+    Side-effect free: reads from S3 and the local filesystem but performs
+    no transfers, deletes, or directory creation. Raises ``ValueError`` if
+    ``remote`` is not an ``s3://`` URL. Returns an empty plan when the
+    local source does not exist (matches real upload() behavior, which
+    skips registrations with a missing local_path).
+    """
+    mirror_obj = _require_active_mirror()
+    return mirror_obj.plan_upload(remote, local, exclude, profile)
+
+
+__all__ = [
+    "mirror",
+    "with_mirror",
+    "download",
+    "upload",
+    "sync",
+    "ls",
+    "plan_download",
+    "plan_upload",
+    "register_profile",
+    "Profile",
+    "TransferError",
+    "TransferPlan",
+    "_parse_s3_url",
+]
diff --git a/pos3/cli.py b/pos3/cli.py
index 66a0e02..7378496 100644
--- a/pos3/cli.py
+++ b/pos3/cli.py
@@ -31,6 +31,8 @@
     download,
     ls,
     mirror,
+    plan_download,
+    plan_upload,
     upload,
 )
 from .profiles import _resolve_profile, _url_profile
@@ -163,7 +165,7 @@ def _cmd_upload(args: argparse.Namespace) -> int:
 
 
 def _print_download_plan(args: argparse.Namespace) -> None:
-    plan = _require_active_mirror().plan_download(
+    plan = plan_download(
         args.url,
         local=args.local,
         exclude=args.exclude,
@@ -177,7 +179,7 @@ def _print_download_plan(args: argparse.Namespace) -> None:
 
 
 def _print_upload_plan(args: argparse.Namespace, source: Path) -> None:
-    plan = _require_active_mirror().plan_upload(
+    plan = plan_upload(
         args.url,
         local=str(source),
         exclude=args.exclude,
diff --git a/tests/test_s3.py b/tests/test_s3.py
index 3e0198c..cd58b12 100644
--- a/tests/test_s3.py
+++ b/tests/test_s3.py
@@ -715,6 +715,37 @@ def test_sync_conflicts(self, mock_boto_client):
                     s3.sync("s3://bucket/data", interval=None)
 
 
+class TestPlanPublicAPI:
+    """The plan API is callable via the public pos3.plan_download /
+    pos3.plan_upload module-level wrappers, the same way pos3.download /
+    pos3.upload are. Users following the README must not need to reach
+    into pos3._require_active_mirror()."""
+
+    def test_plan_download_and_upload_are_module_level(self):
+        # Sanity: they exist on the package surface.
+        assert callable(s3.plan_download)
+        assert callable(s3.plan_upload)
+        # And in __all__ so `from pos3 import *` includes them.
+        assert "plan_download" in s3.__all__
+        assert "plan_upload" in s3.__all__
+        assert "TransferPlan" in s3.__all__
+        assert "TransferError" in s3.__all__
+
+    @patch(BOTO3_PATCH_TARGET)
+    def test_plan_download_via_public_wrapper(self, mock_boto_client):
+        paginate = [{"Contents": [{"Key": "data/file.txt", "Size": 5}]}]
+        _setup_s3_mock(mock_boto_client, paginate)
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            local_dir = Path(tmpdir) / "dst"
+            with s3.mirror(cache_root=tmpdir, show_progress=False):
+                plan = s3.plan_download("s3://bucket/data", local=str(local_dir))
+
+        assert isinstance(plan, s3.TransferPlan)
+        sources = [src for src, _ in plan.to_copy]
+        assert "s3://bucket/data/file.txt" in sources
+
+
 class TestPlan:
     """plan_download / plan_upload return what a real call would do, without
     transferring, deleting, or creating directories."""

From d891dd0dae15d30767d4094b22b884ec0b0f1010 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Sat, 30 May 2026 10:18:05 +0000
Subject: [PATCH 16/16] Broaden cleanup-sync catch in
 _final_sync(had_error=True)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The earlier fix for sync_on_error cleanup masking caught only
TransferError, but _sync_uploads can fail before any worker future
exists — _scan_s3 → _list_s3_objects calls head_object/paginate
first, and a non-404 ClientError (403, expired creds, throttling,
missing bucket) propagates straight through the scan iterator. So
cleanup-time S3 scan failures would still replace the user's
application exception, which is exactly the regression the
had_error=True path was supposed to avoid.

The had_error=True path is fundamentally "best-effort, do no more
harm" — anything that escapes _sync_uploads during cleanup must be
logged and swallowed so the original exception stays the visible
cause. Broaden the catch to Exception. Comment in the source spells
out the failure modes (TransferError from workers, ClientError /
BotoCoreError from the pre-worker scan, OSError on the local fs).

Clean-exit path (had_error=False) is unchanged — still propagates so
one-shot CLI / library callers see definitive failures.

Test: test_app_exception_survives_scan_client_error_in_cleanup sets
mock_s3.head_object.side_effect = ClientError(403), registers a
sync_on_error=True upload, raises AppError inside the mirror() body,
and asserts AppError propagates (not ClientError). Pre-fix this would
have failed.

https://claude.ai/code/session_01QoQaZ2G6FsNpSZyStqTXas
---
 pos3/__init__.py | 23 ++++++++++++++++-------
 tests/test_s3.py | 33 +++++++++++++++++++++++++++++++++
 2 files changed, 49 insertions(+), 7 deletions(-)

diff --git a/pos3/__init__.py b/pos3/__init__.py
index ab5eaec..121256b 100644
--- a/pos3/__init__.py
+++ b/pos3/__init__.py
@@ -718,17 +718,26 @@ def _final_sync(self, had_error: bool = False) -> None:
             uploads = list(self._uploads.values())
         if had_error:
             # The mirror() context is already unwinding with the caller's
-            # exception. If this cleanup sync also fails, swallow it so the
-            # original exception stays the visible cause — a TransferError
-            # from cleanup is logged but must not mask the failure that
-            # triggered cleanup. The clean-exit path (had_error=False)
-            # still propagates so the CLI exits non-zero on failure.
+            # exception. Anything that goes wrong in cleanup must be
+            # logged and swallowed so the original exception stays the
+            # visible cause. The exception type can be:
+            #   - TransferError (worker put/delete failure, after
+            #     _sync_uploads built futures)
+            #   - ClientError / BotoCoreError (_scan_s3 → head_object /
+            #     paginate, BEFORE any worker future exists)
+            #   - OSError on the local fs, etc.
+            # Catching broad Exception here is the right call: the
+            # `had_error=True` path is fundamentally "best-effort, do no
+            # more harm." The clean-exit path (had_error=False) still
+            # propagates so one-shot callers see definitive failures.
             uploads = [u for u in uploads if u.sync_on_error]
             try:
                 self._sync_uploads(uploads)
-            except TransferError as exc:
+            except Exception as exc:
                 logger.error(
-                    "Cleanup sync after error failed; original exception preserved: %s", exc
+                    "Cleanup sync after error failed; original exception preserved: %s: %s",
+                    type(exc).__name__,
+                    exc,
                 )
         else:
             self._sync_uploads(uploads)
diff --git a/tests/test_s3.py b/tests/test_s3.py
index cd58b12..10736a8 100644
--- a/tests/test_s3.py
+++ b/tests/test_s3.py
@@ -1059,6 +1059,39 @@ class AppError(Exception):
                     )
                     raise AppError("the real failure")
 
+    @patch(BOTO3_PATCH_TARGET)
+    def test_app_exception_survives_scan_client_error_in_cleanup(self, mock_boto_client):
+        """Cleanup-time _scan_s3 can raise ClientError (e.g. 403) BEFORE
+        any worker future is created — that path bypasses TransferError.
+        The cleanup catch must be broad enough to preserve the app
+        exception in this case too."""
+        mock_s3 = Mock()
+        mock_boto_client.return_value = mock_s3
+        # _scan_s3 → _list_s3_objects calls head_object first. A 403 here
+        # propagates through the scan iterator, not through a worker future,
+        # so the previous TransferError-only catch would have unmasked it.
+        mock_s3.head_object.side_effect = ClientError(
+            {"Error": {"Code": "403", "Message": "Forbidden"}}, "HeadObject"
+        )
+
+        class AppError(Exception):
+            pass
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            output = Path(tmpdir) / "output"
+            output.mkdir()
+            (output / "data.txt").write_text("content")
+
+            with pytest.raises(AppError, match="the real failure"):
+                with s3.mirror(cache_root=tmpdir, show_progress=False):
+                    s3.upload(
+                        "s3://bucket/output",
+                        local=output,
+                        interval=None,
+                        sync_on_error=True,
+                    )
+                    raise AppError("the real failure")
+
 
 class TestLs:
     def test_ls_local_non_recursive(self):