fix(cli): replay the opened file on upload retries and read redirect bodies

An upload retry re-opened the source path, so a file replaced during a
Retry-After wait could be sent under the original Content-Length and
mtime. The file is now opened once; each attempt rewinds that handle,
and size and mtime come from fstat on it.

A streamed 3xx response was left unread, so describing the failure
raised ResponseNotRead instead of naming the status. Every non-2xx
response is now read before it is handed back.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01APFUk2bjEwMptMQqpRhjea
Signed-off-by: phernandez <paul@basicmachines.co>

Author: phernandez

src/basic_memory/cli/commands/cloud/webdav.py

@@ -20,6 +20,7 @@
 """
 
 import asyncio
+import os
 import re
 import xml.etree.ElementTree as ElementTree
 from collections.abc import AsyncIterator, Callable
@@ -211,8 +212,9 @@ async def _rate_limited_request(
     that makes a retry something other than the same request again.
 
     A successful response body is left unread, so a download can go to disk as it
-    arrives. An error response is read in full: it is small, and the caller's
-    error message quotes it. The response is closed when the block exits.
+    arrives. Any other response (a redirect from a proxy as well as an error) is
+    read in full: it is small, and the caller's error message quotes it. The
+    response is closed when the block exits.
 
     A 429 on the final attempt is yielded rather than raised, so the caller's
     own error handling reports it with the rate-limit detail attached.
@@ -232,20 +234,24 @@ async def _rate_limited_request(
         attempt += 1
 
     try:
-        if response.is_error:
+        if not response.is_success:
             await response.aread()
         yield response
     finally:
         await response.aclose()
 
 
-def _file_chunks(source: Path) -> Callable[[], AsyncIterator[bytes]]:
-    """A body factory that reads ``source`` from the start on every call."""
+def _file_chunks(stream: BinaryIO) -> Callable[[], AsyncIterator[bytes]]:
+    """A body factory that rewinds one open file and reads it on every call.
+
+    The file is opened once by the caller, so a retry replays the file that was
+    validated and measured, even if the path is replaced during a Retry-After wait.
+    """
 
     async def chunks() -> AsyncIterator[bytes]:
-        with source.open("rb") as stream:
-            while chunk := stream.read(_STREAM_CHUNK_BYTES):
-                yield chunk
+        stream.seek(0)
+        while chunk := stream.read(_STREAM_CHUNK_BYTES):
+            yield chunk
 
     return chunks
 
@@ -361,25 +367,28 @@ async def upload_file(
         WebdavError: If the service refuses the upload for any other reason.
     """
     request_path = webdav_path(project, rel_path)
-    stat = source.stat()
-    headers = {
-        "X-OC-Mtime": str(int(stat.st_mtime)),
-        "Content-Length": str(stat.st_size),
-    }
-    if create_only:
-        headers["If-None-Match"] = "*"
+    with source.open("rb") as stream:
+        # Measured from the open handle, so the declared size and mtime describe
+        # exactly the file every attempt sends.
+        stat = os.fstat(stream.fileno())
+        headers = {
+            "X-OC-Mtime": str(int(stat.st_mtime)),
+            "Content-Length": str(stat.st_size),
+        }
+        if create_only:
+            headers["If-None-Match"] = "*"
 
-    try:
-        async with _rate_limited_request(
-            client, "PUT", request_path, content=_file_chunks(source), headers=headers
-        ) as response:
-            # Checked before raise_for_status: a refused precondition is the answer
-            # this call asked for, not a failure.
-            if create_only and response.status_code == httpx.codes.PRECONDITION_FAILED:
-                return False
-            response.raise_for_status()
-    except httpx.HTTPError as exc:
-        raise WebdavError(f"Failed to upload {rel_path}: {_describe(exc)}") from exc
+        try:
+            async with _rate_limited_request(
+                client, "PUT", request_path, content=_file_chunks(stream), headers=headers
+            ) as response:
+                # Checked before raise_for_status: a refused precondition is the
+                # answer this call asked for, not a failure.
+                if create_only and response.status_code == httpx.codes.PRECONDITION_FAILED:
+                    return False
+                response.raise_for_status()
+        except httpx.HTTPError as exc:
+            raise WebdavError(f"Failed to upload {rel_path}: {_describe(exc)}") from exc
 
     return True
 

tests/cli/cloud/test_webdav_client.py

@@ -7,6 +7,7 @@
 
 import io
 import os
+import sys
 from contextlib import asynccontextmanager
 from datetime import datetime, timezone
 from pathlib import Path
@@ -535,6 +536,42 @@ async def handler(request: httpx.Request) -> httpx.Response:
     assert fetched == ["/webdav/research/a#draft.md"]
 
 
+@pytest.mark.asyncio
+async def test_a_redirect_is_reported_with_its_status():
+    """A proxy's 3xx is not a success; the message must name it, not a stream error."""
+
+    async def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(302, text="moved", headers={"Location": "/login"})
+
+    async with _client(handler) as client:
+        with pytest.raises(WebdavError, match="HTTP 302 - moved"):
+            await download_file(client, "research", "a.md", io.BytesIO())
+
+
+# Windows refuses to replace a file another handle holds open, so the race this
+# guards against cannot happen there.
+@pytest.mark.skipif(sys.platform == "win32", reason="cannot replace an open file on Windows")
+@pytest.mark.asyncio
+async def test_upload_retry_sends_the_file_it_opened(recorded_waits, tmp_path):
+    """Replacing the path during a Retry-After wait does not change what is replayed."""
+    source = _local_file(tmp_path, b"original")
+    bodies: list[bytes] = []
+
+    async def handler(request: httpx.Request) -> httpx.Response:
+        bodies.append(request.content)
+        if len(bodies) == 1:
+            replacement = tmp_path / "replacement.md"
+            replacement.write_bytes(b"swapped!")
+            os.replace(replacement, source)
+            return _rate_limited()
+        return httpx.Response(201)
+
+    async with _client(handler) as client:
+        await upload_file(client, "research", "a.md", source=source)
+
+    assert bodies == [b"original", b"original"]
+
+
 @pytest.mark.asyncio
 async def test_upload_file_create_only_sends_the_conditional_header(tmp_path):
     seen: dict[str, object] = {}