Fix SDK async file IO, session races and resource cleanup

CHANGELOG.md

@@ -1,5 +1,21 @@
 # Changelog
 
+## Unreleased maintenance
+
+- Reject non-standard JSON number constants and numbers outside the supported finite float range; report bounded integer decoder failures as observable HTTP errors instead of leaking raw parser exceptions.
+- Redact complete URL credentials before truncating diagnostics so a long password cannot expose its prefix; retain the final output limit after replacements.
+- Finish single async HTTP response and transport cleanup before returning cancellation, including interrupted body reads, multipart uploads and repeated cancellation during connection release.
+- Reject unsafe manifest records and Windows junctions before enumerating or hashing release sources, while continuing to exclude build artifacts from extracted-source scans.
+- Close every cluster client even if one close fails; finish async client cleanup before reporting errors or cancellation and drain cancelled health/bulk siblings safely.
+- Validate integer bulk bounds, worker counts, routing weights and failure thresholds before consuming work.
+- Accept valid JSON `null` responses in sync/async requests instead of incorrectly reporting HTTP 200 as malformed JSON.
+- Protect async sessions against delayed authentication, logout and unauthorized responses so stale requests cannot restore a signed-out session or replace/clear a newer account.
+- Move async attachment snapshot, multipart file reads and atomic download writes off the event loop; drain active disk workers before returning cancellation or closing owned files.
+- Never replay multipart file bodies across cluster endpoints, including PUT and keyed POST requests.
+- Cancel and drain sibling operations when an async bulk item is cancelled, so no owned operation continues after bulk returns.
+- Reject non-integer async concurrency values before consuming work; fractional semaphore counts could otherwise bypass the parallelism limit.
+- Keep these fixes separate from the frozen v0.5.2 publication candidate.
+
 ## 0.5.2 — 2026-10-11
 
 - Add lazy synchronous and asynchronous cursor/offset traversal with resumable cursors, strict page validation, cycle detection, item bounds and cancellation propagation.

MANIFEST.sha256

@@ -13,7 +13,7 @@ f84c1be2d6dd0db7db256496aacb82c7e727e601971e04078d89844d265e002f  .gitignore
 02e735b3dfe1c32833eb550b7ff8ffa17f5f2bc3fa1e7bae61a8f5a3883ce398  .python-version
 09094d533067dddbaf48b9c021ec594292ddc875e8f93735e44dcff9ab62bf55  AI_USAGE_POLICY.md
 759da1adabb4d60f41e3b678452af42192ee3a2c9cc40b0d30522153df374708  AUTHORS.md
-05d98f7ab3cc7a9d653c9a9902ca2ab5c3936c0509da776d21880e4fb6270f56  CHANGELOG.md
+6b05c59b9fc679a94d250abe4f78748ef6b49b218a446834d6614019379a7069  CHANGELOG.md
 f3f6f53ecab4081766570e7e0eb8ccce377e18b2f45e0f583e5e602bda1f98eb  CITATION.cff
 ea87ab99ee7120f85aba29aa86365691dbeb12a1c15ace2b07bd4cbf90942af2  CODE_OF_CONDUCT.md
 93bbc8954fe0e1680d786c08e9a7c1c667bdaaf45e3e7663c821f8b817cb32b3  CONTRIBUTING.md
@@ -25,8 +25,8 @@ ce84a06b898c79cd6199e9eac425e40a3b8f7f8a346d6bbdd019580de56d7652  LICENSE-HISTOR
 b6995f71c80ef0c57d9994107bfac4ff84be1e71d45796235656cfe670d124fe  LICENSE-METADATA.json
 8e9754c8f30a66a580e57a0fe61937ee9d0c37c5d56a0087dfb78761586f0c16  NOTICE.md
 c48cc9fa65e5969c12a670a25a7c712c998f0c8898e0de2906d58713ac4acdae  OWNERSHIP.md
-7d1bcaa4e570d431553ec4abd420f5072204f7feba574cee43027e1eee3ef2de  PYTHON_LIBRARY.md
-d14e5e605ba26ce24ec0a85c053e00577f6382be97d133c8b2b2662c43fb6375  README.md
+8819da447016c48fb317854f167cb7267abbe76a3ff41fe0257ebdae052e571c  PYTHON_LIBRARY.md
+e29158a6782790a9aa7c3c00566e45387e9ce5e673c9482a9b898661dff08d28  README.md
 c25e8ed7ed94aa07a17313cacb83369430595a55d52fd377ee83df6ef8c450fb  RELEASE.md
 19d61a429e7d02875a7f1ded000ad06f187f4c6565b720998caec4c77008190a  RELEASE_ASSETS.md
 858ea89e51f4217a45b042979c718d17f9c2481dc1494c0ba4f49e4ff31caa96  RELEASE_CHECKLIST.md
@@ -37,11 +37,12 @@ e1ace8bfefd6c5e38ec0c70dbc059c53350da23a43177329e25dca812ba5bf8d  THIRD_PARTY_NO
 26e0ab83aa1ff6bd0ae3882787a92663aa19db128e1241ffc99375fb9ca0c121  VERSION
 8f004523844f4b429a9ed8e15e5d46e2957a4c9854a772520267c34be6cf7998  contract-tests/test_against_main.py
 80ab7e3be1458df26ec83bb8482d9b879e96b1eec1c476a6af0950dbdc0be164  json_api_forge/__init__.py
+5c1d12bf6337372eac54c08109583c439db3952bafb0008afd0b9c1402cfa2cd  json_api_forge/_io.py
 8ac26b98304b45a83b65cd942bd668bd618ebfb8e2d4e87ecc9b4930d245917c  json_api_forge/_version.py
-1fdcaad555395a478241288fe757bac93ecb28d5da4a5491eaf64c3af5b439a6  json_api_forge/client.py
-879c685c76c87e073bec9a3ff933a9e5103ddc948a172a0b8ea3eabeb2a9081d  json_api_forge/control_plane.py
-4b2333de0eea36838e8641d668563075a6ff2a269c9a4894c0025a4c84c88537  json_api_forge/enterprise.py
-0a72defc02eeea7d93a813bd7c2b2703fef46b10569201d7ea9d16f442a1b7a7  json_api_forge/errors.py
+c850df04c590083a73afc0b66cf1ef2c4eede44c5b38c4de6ab14451f836490a  json_api_forge/client.py
+5bdc5d59bf9eff5f95d8b878a636b32aeab1fd89bae47b82c9be7d05ce39583c  json_api_forge/control_plane.py
+c0f60c048a1512d9fe820055ce3606551c951b8a9a909d0ff999cf3f8829abe2  json_api_forge/enterprise.py
+477f7e0957458e597c2a78b6d4522a8cce0f8ba306f76f4da2dc5d98ac436aa9  json_api_forge/errors.py
 78c04a6bdf39170ce61e9b957aba66d8a6583e9fc87fa0465f58df9095ee9705  json_api_forge/integrations/__init__.py
 7dde121bcbe377560c684cbfc5875f0dc099e4d393eb494ce61da9731d562d54  json_api_forge/integrations/ddm.py
 100f1f5e0bf1172503ca9ea63614d5616a7f46395ac67e3dfc9b20ce10d863ff  json_api_forge/integrations/younglion.py
@@ -50,11 +51,18 @@ c2d0db468c7e41c65791983f512b484cb850c5f65ae725069b7aaab6cc78b3f8  json_api_forge
 0d7c4adff9f102814c0b6b24ca8745d138b3cca96febea98b2cf355de5d02c4d  json_api_forge/pagination.py
 01ba4719c80b6fe911b091a7c05124b64eeece964e09c058ef8f9805daca546b  json_api_forge/py.typed
 d0cf58ccb32c816f0c85cd4a29a28b976343409647e4affcd84932acf98898c2  pyproject.toml
-915a198346437156ef6388e6d3ac4f859478c26d9b48abf5a89e65a30d6e96ed  scripts/check_critical_coverage.py
-11ef6b2ffc520f475010082b370f8518d62a11d3c51619fdcecda4271c01363a  scripts/check_manifest.py
+dbfbc0b31387b5121c4c51fedfbaa147bbab8c789b96ea3d710daa67f7ba2b64  scripts/check_critical_coverage.py
+fc719b85b872e730b347f4674e2aaa3a3bc8bff3dfe361b41e46837dee709062  scripts/check_manifest.py
 7e505ff356cb16900bc9928c97c8a66b1acde8d06ac29e1cc01250353af9031c  tests/__init__.py
+50830855c94d3e55c4bd74d6df786d95875929fbe72653fd85ff98d8c2250d3b  tests/test_async_attachment_io.py
+3c00da7a012803b247ea5dc5820bb8c6448e9798f60e38abb09ff801ce01c546  tests/test_async_session_races.py
+668b74e65c5d3ce0769feef88b5d9aeb389f6b69c444630d1f5929f0d9399e1c  tests/test_async_transport_cleanup.py
+cc6694028ea0148fc4e5dc7877740c1133b21afee030390c9e039b68aa34ce79  tests/test_cluster_lifecycle.py
+ff662164581e2f72607604cb9f09d5c56b88440d171aa0242716ff1d693b36b4  tests/test_manifest_security.py
+61109979cdb492ac7976947d804b3e717871fa367346f3cdfa911382b25c59a3  tests/test_night_cluster.py
 4c99c2621e4c2f215da3cdaf9a77455893cc552354eb5ba6ec86c86d4d3ada4d  tests/test_python_control_plane.py
 45658c58e7c989fd735dd73ee77e2cc3164f0671ff846e1a4af3cd1864d319c9  tests/test_python_enterprise.py
 9959ece37e0df7010bb2c87407e843b53a8d4b5ccfb9693e4fd6a4fb2e8943d0  tests/test_python_library.py
-fd19f7b746b195fddbcd2997b787b68d625f3fe2b0cb9899bcb44570ff563766  tests/test_v051_security.py
+40db1d05ac5607c4bae91b750fb6c431791fbfeb27c090684b46b4908eea345a  tests/test_response_json.py
+1a44d0909278cc2750a80f757631dd2f51d1dc6a3ea206f5ac85f5210f9de7a5  tests/test_v051_security.py
 fa33f2147dd8e704684e752e5f4d0caec848f9c8ffb176a2dc202d8cf1513e4d  tests/test_v052_pagination.py

PYTHON_LIBRARY.md

@@ -43,6 +43,8 @@ Iteration fetches on demand, buffers at most one bounded response and stops at `
 
 Collection iterators and cluster bulk helpers are bounded. The default bulk input ceiling is 10,000 items; raise it only after evaluating memory and downstream capacity.
 
+Multipart file uploads are never retried or failed over automatically, even with an idempotency key. Retry explicitly with a fresh body after checking the upload outcome. Cancelling async bulk work, including cancellation raised by an individual operation, cancels and waits for its sibling operations before returning. Completed server-side effects cannot be rolled back by cancellation.
+
 ## Control-plane security
 
 - Remote HTTP is rejected; `allow_insecure_http=True` is restricted to loopback development endpoints.

README.md

@@ -65,6 +65,8 @@ notes = asyncio.run(main())
 
 Bounded retries are opt-in. The retry/failover policy treats GET, HEAD, OPTIONS, PUT and DELETE as idempotent HTTP methods. Other methods, including POST/PATCH, need an explicit idempotency key before replay is permitted. The server endpoint must actually support idempotency; adding a header alone does not make arbitrary side effects replay-safe.
 
+JSON responses use Python integers and finite floating-point numbers. Non-standard `NaN`/`Infinity` constants, exponent values outside the supported finite float range (for example, `1e999`), and integers rejected by the interpreter's configured digit limit produce `ForgeHTTPError` with the original response status and request ID. Finite numbers and JSON `null` retain their usual Python values.
+
 ## Editor control-plane client
 
 ```python
@@ -82,10 +84,16 @@ The control-plane clients cover founder setup, invitation enrollment, profiles,
 
 Security defaults include HTTPS-only remote URLs, no ambient proxy inheritance, no redirects or retained cookies, strict same-origin paths, exact token formats, bounded response bodies, non-symlink attachment snapshots and atomic downloads. WebRTC call tickets are placed in a URL fragment for the local call client, never in an HTTP query. v0.5.2 uses the revision 2 HTTP signaling call client.
 
+Async attachment snapshots, multipart file reads and atomic download writes run in worker threads so slow storage does not freeze other async tasks. Cancellation waits for an active disk operation to finish before closing its file or returning. If cancellation arrives during the atomic save, the destination may already contain the complete download when cancellation is reported; no background writer remains after that report.
+
+The async control-plane client discards an authentication response with `ForgeSessionError` if the session changed while it was pending, including sign-out, client close or a newer sign-in. An old request's unauthorized/logout response cannot clear a newer session.
+
 ## Cluster and optional integrations
 
 `ForgeCluster` and `AsyncForgeCluster` provide deterministic routing, circuit breakers, bounded bulk work and idempotency-gated failover across multiple Forge endpoints.
 
+Cluster shutdown attempts every client close and reports the first error after cleanup finishes. Async shutdown also finishes cleanup before propagating caller cancellation. Bulk and health operations cancel and drain their owned siblings before returning a cancelled result; worker/item counts and integer routing policies are checked before work starts.
+
 YoungLion/DDM adapters remain optional:
 
 ```bash

json_api_forge/_io.py

@@ -0,0 +1,33 @@
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Callable
+from functools import partial
+from typing import Any, TypeVar
+
+_T = TypeVar("_T")
+
+
+async def run_io(function: Callable[..., _T], *args: Any, **kwargs: Any) -> _T:
+    """Run blocking IO without abandoning owned files when the caller cancels."""
+    worker = asyncio.create_task(asyncio.to_thread(partial(function, *args, **kwargs)))
+    return await wait_owned(worker)
+
+
+async def wait_owned(worker: asyncio.Future[_T]) -> _T:
+    """Finish owned cleanup before returning caller cancellation."""
+    try:
+        return await asyncio.shield(worker)
+    except asyncio.CancelledError:
+        # An owned operation must finish before its context can return or close
+        # its resources, even if cancellation repeats during cleanup.
+        while not worker.done():
+            try:
+                await asyncio.shield(worker)
+            except asyncio.CancelledError:
+                continue
+            except Exception:
+                break
+        if not worker.cancelled():
+            worker.exception()
+        raise

json_api_forge/client.py

@@ -9,11 +9,13 @@
 import unicodedata
 import uuid
 from collections.abc import AsyncIterator, Callable, Iterator, Mapping, Sequence
+from contextlib import asynccontextmanager
 from typing import Any, Literal
 from urllib.parse import quote, unquote, urlsplit
 
 import httpx
 
+from ._io import run_io, wait_owned
 from .errors import ForgeHTTPError, ForgeResponseTooLarge, ForgeTransportError, safe_error_text
 from .models import ForgeResponse, JsonObject, RequestAttempt
 from .options import RetryPolicy
@@ -149,12 +151,27 @@ def _headers(
     return values
 
 
+def _reject_json_constant(value: str) -> Any:
+    raise ValueError(f"Invalid JSON number constant: {value}")
+
+
+def _finite_json_float(value: str) -> float:
+    result = float(value)
+    if not math.isfinite(result):
+        raise ValueError("JSON number exceeds the supported finite float range")
+    return result
+
+
 def _decode_response(response: httpx.Response, content: bytes, *, expect_json: bool) -> ForgeResponse[Any]:
     detail: Any
+    valid_json = True
     try:
-        decoded = json.loads(content) if content else None
-    except (UnicodeDecodeError, json.JSONDecodeError, RecursionError):
+        decoded = json.loads(content, parse_constant=_reject_json_constant, parse_float=_finite_json_float) if content else None
+    except (ValueError, RecursionError):
+        # ValueError also covers the interpreter's bounded integer conversion,
+        # UnicodeDecodeError and JSONDecodeError; keep these HTTP failures observable.
         decoded = None
+        valid_json = False
     if response.is_redirect:
         raise ForgeHTTPError(
             response.status_code,
@@ -175,7 +192,7 @@ def _decode_response(response: httpx.Response, content: bytes, *, expect_json: b
             response.headers.get("X-Request-ID"),
             response.headers.get("Retry-After"),
         )
-    if expect_json and decoded is None and content:
+    if expect_json and not valid_json and content:
         raise ForgeHTTPError(response.status_code, "Server returned non-JSON content", response.headers.get("X-Request-ID"))
     data = decoded if expect_json else content
     return ForgeResponse(
@@ -234,6 +251,55 @@ async def _bounded_content_async(response: httpx.Response, max_response_bytes: i
     return bytes(chunks)
 
 
+class _AsyncFileStream(httpx.AsyncByteStream):
+    """Keep httpx multipart's synchronous file reads off the event loop."""
+
+    def __init__(self, stream: httpx.SyncByteStream):
+        self._chunks = iter(stream)
+
+    async def __aiter__(self) -> AsyncIterator[bytes]:
+        while True:
+            chunk = await run_io(next, self._chunks, None)
+            if chunk is None:
+                break
+            yield chunk
+
+
+class _OwnedResponseStream(httpx.AsyncByteStream):
+    """Finish releasing the HTTP connection even when cancellation repeats."""
+
+    def __init__(self, stream: httpx.AsyncByteStream):
+        self._stream = stream
+
+    async def __aiter__(self) -> AsyncIterator[bytes]:
+        async for chunk in self._stream:
+            yield chunk
+
+    async def aclose(self) -> None:
+        # httpx marks a Response closed before awaiting the underlying stream.
+        # Shield here, including its automatic close at the end of body reads;
+        # a later Response.aclose() cannot resume an interrupted first close.
+        await wait_owned(asyncio.create_task(self._stream.aclose()))
+
+
+@asynccontextmanager
+async def _stream_async(client: httpx.AsyncClient, method: str, path: str, **kwargs: Any) -> AsyncIterator[httpx.Response]:
+    if kwargs.get("files") is None:
+        async with client.stream(method, path, **kwargs) as response:
+            response.stream = _OwnedResponseStream(response.stream)
+            yield response
+        return
+    # Request assembly also seeks/fstats files to compute Content-Length.
+    request = await run_io(client.build_request, method, path, **kwargs)
+    request.stream = _AsyncFileStream(request.stream)
+    response = await client.send(request, stream=True)
+    response.stream = _OwnedResponseStream(response.stream)
+    try:
+        yield response
+    finally:
+        await response.aclose()
+
+
 class ForgeClient:
     """Synchronous, bounded client for one JSON API Forge server."""
 
@@ -509,7 +575,8 @@ async def request(
             try:
                 self._client.cookies.clear()
                 try:
-                    async with self._client.stream(
+                    async with _stream_async(
+                        self._client,
                         normalized_method,
                         normalized_path,
                         json=json_body,
@@ -642,7 +709,7 @@ async def batch_create_items(self, project: str, resource: str, payloads: Sequen
         return await self.request("POST", f"api/{_segment(project)}/v1/{_route(resource)}/_batch", json_body=batch_payload(payloads))
 
     async def aclose(self) -> None:
-        await self._client.aclose()
+        await wait_owned(asyncio.create_task(self._client.aclose()))
 
     async def __aenter__(self) -> AsyncForgeClient:
         return self