diff --git a/CHANGELOG.md b/CHANGELOG.md
index 9579e4bcb..ac784312d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -19,6 +19,13 @@ All notable changes to this project are documented in this file. The format is b
   including delegated share management: a role implying `ACTION_SHARE` may list,
   grant and revoke shares on that resource. Default view/edit roles remain
   unchanged; no migration (#707).
+- Tenant-scoped sync file worker contracts (`store_path`, `open_local`, and
+  `presigned_url`), reusing worker settings/pools, upload validation and hashing,
+  with commit compensation and `FileUploaded` publication after commit (#708).
+- **Behaviour change:** S3 server-side uploads (including browser single-request
+  ones) above `direct_upload_threshold_bytes` now use bounded multipart, including
+  unknown-length streams, with size/part bounds and abort on failure; the S3
+  credentials need `s3:AbortMultipartUpload` (#708).
 - Users auth emails support ordered host template directories through the
   restart-required DB setting `mail_template_dirs`, with per-file bundled
   fallback and optional autoescaped HTML alternatives (#712).
@@ -43,6 +50,11 @@ All notable changes to this project are documented in this file. The format is b
   most 32 characters now fail at registration (#707).
 
 ### Fixed
+- `file_storage_stored_file.size_bytes` is now `BIGINT` (migration `f5b1c3e8a207`,
+  which also gives `file_storage` its own branch label): on PostgreSQL it was a
+  32-bit `INTEGER`, so a file of 2 GiB or more failed on insert (#708).
+- A cancelled one-shot upload's metadata write now also deletes the uploaded
+  object (#708).
 - **Behaviour change:** bare UI borders now use the theme border colour in light
   and dark themes, including pseudo-elements. Use `border-current` to retain the
   previous foreground-coloured look; explicit border colour utilities still win (#714).
diff --git a/CLAUDE.md b/CLAUDE.md
index 265690d07..3a8147407 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -313,7 +313,7 @@ development and testing (`session_cookie_secure`).
   four roots (`77162e7b184b` initial schema, `e36ab09c4a92` tenants,
   `f733c7938fd1` sharing, `d263ea727663` the framework's `sm_lease`) and six
   heads (`audit_log`, `tenants`, `keycloak`, `sharing`, `leases`, and the users
-  trunk, which Alembic reports as `api_tokens` and, since #536,
+  trunk, which Alembic reports as `api_tokens`, `users`, `file_storage` and, since #536,
   `background_tasks`). Apply with `upgrade heads` (what `make migrate`
   does); plain `revision --autogenerate` refuses with "Multiple heads", so name
   the branch:
diff --git a/docs/database/migrations.md b/docs/database/migrations.md
index 096f18cfd..c1f54896e 100644
--- a/docs/database/migrations.md
+++ b/docs/database/migrations.md
@@ -21,7 +21,7 @@ The history is **not** a single linear chain. It has several roots and several h
 
 | Head | Owning module | Label Alembic reports |
 |---|---|---|
-| `b31a70971001` | background_tasks (resource references) | `api_tokens`, `background_tasks`, `users` |
+| `f5b1c3e8a207` | file_storage (`size_bytes` BIGINT) | `api_tokens`, `background_tasks`, `file_storage`, `users` |
 | `a7c2e91d4b58` | audit_log | `audit_log` |
 | `f1a9c3d7b2e4` | tenants | `tenants` |
 | `d263ea727663` | framework (`sm_lease`) | `leases` |
@@ -34,11 +34,11 @@ A module's first migration sets `branch_labels = ("<module_name>",)`. Alembic **
 
 ### Known mislabels
 
-The declared `branch_labels` are `audit_log`, `api_tokens`, `background_tasks`, `keycloak`, `tenants`, `sharing`, `leases` and `users` (added by `a625f0b7c912`). The graph carries these defects; they are documented rather than rewritten, because changing labels on revisions that installed databases already applied is not worth the risk:
+The declared `branch_labels` are `audit_log`, `api_tokens`, `background_tasks`, `keycloak`, `tenants`, `sharing`, `leases`, `users` (added by `a625f0b7c912`) and `file_storage` (added by `f5b1c3e8a207`). The graph carries these defects; they are documented rather than rewritten, because changing labels on revisions that installed databases already applied is not worth the risk:
 
 - `51b2058d8afd_api_tokens_initial` sits in the middle of the trunk, so every revision above it (file_storage, background_tasks including the outbox, and the users revisions through `c8e4f1a2b6d9`) reports the `api_tokens` label. The `api_tokens` head is therefore really a users revision.
 - `keycloak` branches off a users revision, so a Keycloak-only install still builds the users schema.
-- `users`, `file_storage`, `background_tasks`, `settings`, `permissions` and `feature_flags` have no label of their own.
+- `settings`, `permissions` and `feature_flags` have no label of their own. `users` and `file_storage` gained labels late (`a625f0b7c912`, `f5b1c3e8a207`), so their earlier revisions stay unlabelled and report `api_tokens`.
 
 Guidance: give an unlabelled module a `branch_labels` on its **next** revision. Adding a label to an existing revision file is technically possible, because `alembic_version` stores revision ids and never labels, but it also relabels every descendant and changes what `<label>@head` resolves to, so treat it as a deliberate, reviewed change rather than a cleanup.
 
diff --git a/docs/modules/file_storage.md b/docs/modules/file_storage.md
index 3e20fb918..7249e742b 100644
--- a/docs/modules/file_storage.md
+++ b/docs/modules/file_storage.md
@@ -285,6 +285,68 @@ Any task that touches `StoredFile` outside a request must run under
 `tenant_context(row.tenant_id)` for per-tenant work, or `all_tenants()` for a
 deliberate platform-wide sweep.
 
+## Synchronous worker API
+
+Install the `[tasks]` extra and configure the background-tasks worker normally.
+`file_storage.contracts` lazily exports `store_path`, `open_local`, and
+`presigned_url` for synchronous task bodies (#708):
+
+```python
+from simple_module_db import tenant_context
+from file_storage.contracts import open_local, presigned_url, store_path
+
+with tenant_context(tenant_id):
+    stored = store_path("output.tif", content_type="image/tiff")
+    with open_local(stored.id) as local_path:
+        process_raster(local_path)
+    url = presigned_url(stored.id, ttl_seconds=300)  # S3-compatible backends
+```
+
+`store_path(source, *, filename=None, content_type="application/octet-stream",
+public=False)` accepts a string/path-like path or binary stream and returns
+`StoredFileOut`. Paths default to their basename; streams use their `name`
+basename or require an explicit filename. Streams are read from their current
+position and remain open. The API reuses MIME, size, quota and SHA-256 checks,
+with an independent transaction. It returns after commit and publishes
+`FileUploaded` on the worker event bus only after a successful commit. Flush or
+commit failures attempt to delete the uploaded object; an uncertain commit
+outcome is checked in a fresh session first. If that check cannot reach the DB,
+the object is preserved for reconciliation. Cleanup failures are logged without
+replacing the original error. Settings are hydrated from the DB
+through the existing worker cache (normally refreshed within 30 seconds), and
+calls reuse the worker thread's event loop and pooled engine.
+
+All three operations inherit the caller's tenant context. A strict multi-tenant
+install refuses unscoped access; missing and cross-tenant IDs behave alike.
+Reads reload metadata by ID and refuse a file recorded against another backend.
+`open_local(file_id)` closes the metadata transaction before downloading, yields
+a path inside a private temporary directory, and removes the directory on
+normal exit, download failure or a consumer exception. Budget temporary disk
+space for the entire downloaded object. `presigned_url(file_id, *,
+ttl_seconds=None)` uses `s3_presign_ttl_seconds` unless overridden with a positive
+TTL; unsupported backends (including filesystem) raise `NotSupportedError`.
+These sync functions are for task bodies, not async handlers running on the
+worker loop itself.
+
+S3 server-side `put` switches to multipart above
+`direct_upload_threshold_bytes`, even for streams whose size is unknown. It
+spools only the threshold prefix and one part at a time (each spills to disk
+above 10 MiB). Part size comes from `multipart_part_size_bytes`, grown for known
+sizes to fit 10,000 parts; unknown-size streams abort if they exceed that bound.
+Declared nonzero sizes must match the bytes read. Multipart is also used above
+S3's single-PUT limit, even if the threshold was configured higher. Stream,
+part-upload and completion failures, including cancellation, attempt abort
+without masking the failure. Worker uploads still honor `max_file_size_bytes`
+(default 100 MiB): raise that existing DB limit for multi-GB outputs.
+`StoredFile.size_bytes` is a `BIGINT` since migration `f5b1c3e8a207` (it was a
+32-bit `INTEGER` on PostgreSQL, which refused files of 2 GiB or more), and the
+S3 backend's `put` goes multipart, so objects beyond 5 GiB work end to end.
+
+A process killed mid-upload cannot run compensation. Its incomplete multipart
+upload is an orphan like an abandoned browser one, and the upload reaper aborts
+it once it is older than twice `upload_session_ttl_seconds` (48 hours by
+default), which no real put runs for.
+
 ## Settings
 
 DB-backed via `register_module_settings`; pydantic defaults seed at boot. Live edits under Files at `/admin/settings/`.
diff --git a/host/migrations/versions/f5b1c3e8a207_file_storage_size_bytes_bigint.py b/host/migrations/versions/f5b1c3e8a207_file_storage_size_bytes_bigint.py
new file mode 100644
index 000000000..171b7f654
--- /dev/null
+++ b/host/migrations/versions/f5b1c3e8a207_file_storage_size_bytes_bigint.py
@@ -0,0 +1,41 @@
+"""file_storage_stored_file.size_bytes: BIGINT (#708)
+
+``size_bytes`` was a 32-bit ``INTEGER`` on Postgres, so storing a file of
+2 GiB or more failed on insert. Server-side puts now go multipart above
+5 GiB (#708), so the column must hold any object size. SQLite's INTEGER is
+already 64-bit; the batch alter is a no-op rebuild there.
+
+file_storage had no label of its own (its table came in with the initial
+schema); this revision gives it one.
+
+Revision ID: f5b1c3e8a207
+Revises: b31a70971001
+Create Date: 2026-10-30 18:00:00.000000
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "f5b1c3e8a207"
+down_revision: str | None = "b31a70971001"
+branch_labels: str | Sequence[str] | None = ("file_storage",)
+depends_on: str | Sequence[str] | None = None
+
+_TABLE = "file_storage_stored_file"
+
+
+def upgrade() -> None:
+    with op.batch_alter_table(_TABLE) as batch:
+        batch.alter_column(
+            "size_bytes", existing_type=sa.Integer(), type_=sa.BigInteger(), existing_nullable=False
+        )
+
+
+def downgrade() -> None:
+    # Fails on Postgres if a stored file is 2 GiB or larger; that is the point.
+    with op.batch_alter_table(_TABLE) as batch:
+        batch.alter_column(
+            "size_bytes", existing_type=sa.BigInteger(), type_=sa.Integer(), existing_nullable=False
+        )
diff --git a/host/tests/test_migration_graph.py b/host/tests/test_migration_graph.py
index 5c0598042..9bac2ad1f 100644
--- a/host/tests/test_migration_graph.py
+++ b/host/tests/test_migration_graph.py
@@ -34,6 +34,7 @@
     "a659221c464b": "file_storage",
     "e5a8c1f27b94": "file_storage",
     "c7f2d9a41e83": "file_storage",
+    "f5b1c3e8a207": "file_storage",  # size_bytes BIGINT (#708)
     "e5f2a8c1d7b3": "core",  # timezone-aware timestamps across modules
     "b5d3f08a6e17": "background_tasks",
     "e878882924e7": "background_tasks",
@@ -58,7 +59,7 @@
 # head revision -> module owning the tables it creates or alters.
 _HEADS = {
     "a7c2e91d4b58": "audit_log",
-    "b31a70971001": "background_tasks",  # also carries ``api_tokens`` and ``users``
+    "f5b1c3e8a207": "file_storage",  # also carries ``api_tokens``, ``background_tasks``, ``users``
     "f1a9c3d7b2e4": "tenants",
     "f6a3b9d2e8c4": "keycloak",
     "f733c7938fd1": "sharing",
@@ -75,6 +76,7 @@
     "bb906e1a1089": "background_tasks",
     "d263ea727663": "leases",
     "a625f0b7c912": "users",
+    "f5b1c3e8a207": "file_storage",
 }
 
 # Known mislabels: revisions that inherit a label from an ancestor owned by a
@@ -99,6 +101,7 @@
     "1f5985daa10b": "api_tokens",  # also inherits ``background_tasks``
     "a625f0b7c912": "api_tokens",  # also inherits ``background_tasks``
     "b31a70971001": "api_tokens",
+    "f5b1c3e8a207": "api_tokens",
     "a626b1c8d023": "api_tokens",  # also inherits ``background_tasks``
     "51b2058d8afd": "api_tokens",  # ancestors of a labelled revision inherit its label too
 }
diff --git a/modules/file_storage/README.md b/modules/file_storage/README.md
index 19384b8f2..03df0efe6 100644
--- a/modules/file_storage/README.md
+++ b/modules/file_storage/README.md
@@ -27,6 +27,24 @@ The extras combine as `simple_module_file_storage[s3,tasks]`. Without
 - Pluggable backend selected by the `backend` setting (`filesystem` default | `s3`); third-party backends can register under any id.
 - Settings (`backend`, `s3_bucket`, `s3_region`, `s3_endpoint_url` for R2/MinIO/etc., `s3_access_key_id`, `s3_secret_access_key`, `s3_presign_ttl_seconds`) are configured from the DB settings store via the admin UI — they are not read from `SM_FILE_STORAGE_*` environment variables at runtime.
 
+## Sync workers
+
+With the `[tasks]` extra installed, synchronous task bodies can import
+`store_path`, `open_local`, and `presigned_url` from `file_storage.contracts`.
+Run calls inside `simple_module_db.tenant_context(tenant_id)`. `store_path`
+accepts a path or binary stream and returns committed `StoredFileOut` metadata;
+`open_local(file_id)` yields a temporary local path and always cleans it up.
+`presigned_url(file_id)` requires a backend with signing support.
+
+Uploads retain the existing MIME, quota, size and hashing rules. S3 writes above
+`direct_upload_threshold_bytes` use bounded multipart. A file path's size is
+known, so parts grow to fit S3's 10,000-part limit; an unknown-size stream uses
+`multipart_part_size_bytes` as is (raise it for streams beyond ~156 GiB). Raise
+the DB's `max_file_size_bytes` for large worker outputs. An
+incomplete multipart upload left by a killed job is aborted by the upload reaper.
+See `docs/modules/file_storage.md` in the repository for transaction and cleanup
+semantics.
+
 ## Usage
 
 From another module, inject the service via its dependency:
diff --git a/modules/file_storage/file_storage/backends/s3.py b/modules/file_storage/file_storage/backends/s3.py
index 119750ebe..8a7f15af5 100644
--- a/modules/file_storage/file_storage/backends/s3.py
+++ b/modules/file_storage/file_storage/backends/s3.py
@@ -9,7 +9,6 @@
 
 from collections.abc import AsyncGenerator, AsyncIterator
 from datetime import datetime
-from tempfile import SpooledTemporaryFile
 from typing import Any
 
 from file_storage import constants
@@ -47,6 +46,8 @@ def __init__(
         region: str,
         client_kwargs: dict[str, Any],
         public_endpoint_url: str = "",
+        multipart_threshold_bytes: int = constants.DEFAULT_DIRECT_UPLOAD_THRESHOLD_BYTES,
+        multipart_part_size_bytes: int = constants.DEFAULT_PART_SIZE_BYTES,
     ) -> None:
         try:
             import aioboto3
@@ -58,6 +59,8 @@ def __init__(
         self.region = region
         self.client_kwargs = client_kwargs
         self.public_endpoint_url = public_endpoint_url
+        self.multipart_threshold_bytes = multipart_threshold_bytes
+        self.multipart_part_size_bytes = multipart_part_size_bytes
 
     def _client(self):
         return self._session.client("s3", **self.client_kwargs)
@@ -70,25 +73,9 @@ async def put(
         content_type: str,
         size: int,
     ) -> None:
-        # Spool to memory up to SPOOL_MAX_SIZE_BYTES, then to a temp file.
-        # put_object needs a seekable file-like; multipart upload would be
-        # the next step for very large files (deferred to v2).
-        with SpooledTemporaryFile(max_size=constants.SPOOL_MAX_SIZE_BYTES) as buf:
-            async for chunk in stream:
-                buf.write(chunk)
-            length = buf.tell()
-            buf.seek(0)
-            try:
-                async with self._client() as client:
-                    await client.put_object(
-                        Bucket=self.bucket,
-                        Key=key,
-                        Body=buf,
-                        ContentType=content_type,
-                        ContentLength=length,
-                    )
-            except Exception as exc:
-                raise StorageBackendError(f"S3 put_object failed: {exc}") from exc
+        from file_storage.backends.s3_put import put_stream
+
+        await put_stream(self, key, stream, content_type, size)
 
     async def get(self, key: str) -> AsyncIterator[bytes]:
         return await _open(self._stream(key, {}))
@@ -233,4 +220,6 @@ def _build(settings: FileStorageSettings) -> S3Backend:
         region=settings.s3_region,
         client_kwargs=client_kwargs,
         public_endpoint_url=settings.s3_public_endpoint_url,
+        multipart_threshold_bytes=settings.direct_upload_threshold_bytes,
+        multipart_part_size_bytes=settings.multipart_part_size_bytes,
     )
diff --git a/modules/file_storage/file_storage/backends/s3_put.py b/modules/file_storage/file_storage/backends/s3_put.py
new file mode 100644
index 000000000..330f6c184
--- /dev/null
+++ b/modules/file_storage/file_storage/backends/s3_put.py
@@ -0,0 +1,121 @@
+"""Bounded server-side S3 uploads, including streams of unknown size (#708)."""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import AsyncIterator
+from tempfile import SpooledTemporaryFile
+from typing import TYPE_CHECKING
+
+from file_storage import constants, upload_support
+from file_storage.contracts.service import Part, StorageBackendError
+from file_storage.upload_support import SizeMismatchError
+
+if TYPE_CHECKING:
+    from file_storage.backends.s3 import S3Backend
+
+logger = logging.getLogger(__name__)
+
+
+async def put_stream(
+    backend: S3Backend, key: str, stream: AsyncIterator[bytes], content_type: str, size: int
+) -> None:
+    """Buffer up to the threshold, then send one bounded part at a time."""
+    if size < 0:
+        raise ValueError("size must be nonnegative (0 means unknown)")
+    threshold = min(backend.multipart_threshold_bytes, constants.MAX_S3_PART_SIZE_BYTES)
+    iterator = aiter(stream)
+    tail: memoryview | None = None
+    with SpooledTemporaryFile(max_size=constants.SPOOL_MAX_SIZE_BYTES) as prefix:
+        if size <= threshold:
+            async for chunk in iterator:
+                take = min(len(chunk), threshold + 1 - prefix.tell())
+                prefix.write(memoryview(chunk)[:take])
+                if prefix.tell() > threshold:
+                    tail = memoryview(chunk)[take:]
+                    break
+            else:
+                length = prefix.tell()
+                if size and length != size:
+                    raise SizeMismatchError(size, length)
+                prefix.seek(0)
+                try:
+                    async with backend._client() as client:
+                        await client.put_object(
+                            Bucket=backend.bucket,
+                            Key=key,
+                            Body=prefix,
+                            ContentType=content_type,
+                            ContentLength=length,
+                        )
+                except Exception as exc:
+                    raise StorageBackendError(f"S3 put_object failed: {exc}") from exc
+                return
+        prefix.seek(0)
+
+        async def body() -> AsyncIterator[bytes]:
+            while chunk := prefix.read(constants.DEFAULT_CHUNK_SIZE):
+                yield chunk
+            if tail is not None:
+                for offset in range(0, len(tail), constants.DEFAULT_CHUNK_SIZE):
+                    yield bytes(tail[offset : offset + constants.DEFAULT_CHUNK_SIZE])
+            async for chunk in iterator:
+                yield chunk
+
+        await _multipart(backend, key, body(), content_type, size)
+
+
+async def _multipart(backend: S3Backend, key, stream, content_type, size) -> None:
+    part_size = upload_support.choose_part_size(size, backend.multipart_part_size_bytes)
+    if part_size > constants.MAX_S3_PART_SIZE_BYTES:
+        raise ValueError("S3 part size exceeds 5 GiB")
+    upload_id = await backend.create_multipart(key, content_type)
+    parts: list[Part] = []
+    seen = 0
+    try:
+        with SpooledTemporaryFile(max_size=constants.SPOOL_MAX_SIZE_BYTES) as buf:
+            async for chunk in stream:
+                seen += len(chunk)
+                if size and seen > size:
+                    raise SizeMismatchError(size, seen)
+                remaining = memoryview(chunk)
+                while remaining:
+                    take = min(len(remaining), part_size - buf.tell())
+                    buf.write(remaining[:take])
+                    remaining = remaining[take:]
+                    if buf.tell() == part_size:
+                        parts.append(await _send(backend, key, upload_id, parts, buf))
+                        buf.seek(0)
+                        buf.truncate()
+            if size and seen != size:
+                raise SizeMismatchError(size, seen)
+            if buf.tell() or not parts:
+                parts.append(await _send(backend, key, upload_id, parts, buf))
+        await backend.complete_multipart(key, upload_id, parts)
+    except BaseException:
+        try:
+            await backend.abort_multipart(key, upload_id)
+        except BaseException:
+            logger.exception("file_storage.server_upload_abort_failed key=%s", key)
+        raise
+
+
+async def _send(backend: S3Backend, key, upload_id, parts, buf) -> Part:
+    number = len(parts) + 1
+    if number > constants.MAX_UPLOAD_PARTS:
+        raise ValueError("S3 upload exceeds maximum part count")
+    length = buf.tell()
+    buf.seek(0)
+    try:
+        async with backend._client() as client:
+            result = await client.upload_part(
+                Bucket=backend.bucket,
+                Key=key,
+                UploadId=upload_id,
+                PartNumber=number,
+                Body=buf,
+                ContentLength=length,
+            )
+    except Exception as exc:
+        raise StorageBackendError(f"S3 upload_part failed: {exc}") from exc
+    return Part(no=number, etag=result["ETag"], size=length)
diff --git a/modules/file_storage/file_storage/constants.py b/modules/file_storage/file_storage/constants.py
index ca213ac4b..4072e2a7b 100644
--- a/modules/file_storage/file_storage/constants.py
+++ b/modules/file_storage/file_storage/constants.py
@@ -171,6 +171,7 @@ class I18nKey:
 DEFAULT_PART_SIZE_BYTES: Final = 16 * MIB
 # S3's hard ceiling; ``part_size`` is grown until the upload fits under it.
 MAX_UPLOAD_PARTS: Final = 10_000
+MAX_S3_PART_SIZE_BYTES: Final = 5 * 1024 * MIB
 # S3 rejects non-final parts below 5 MiB.
 MIN_PART_SIZE_BYTES: Final = 5 * MIB
 DEFAULT_UPLOAD_SESSION_TTL_SECONDS: Final = 24 * 3600
diff --git a/modules/file_storage/file_storage/contracts/__init__.py b/modules/file_storage/file_storage/contracts/__init__.py
index 8a4c2c5a7..8844f81ea 100644
--- a/modules/file_storage/file_storage/contracts/__init__.py
+++ b/modules/file_storage/file_storage/contracts/__init__.py
@@ -1,5 +1,13 @@
 """file_storage contracts — public interface for other modules."""
 
+from __future__ import annotations
+
+from contextlib import AbstractContextManager
+from os import PathLike
+from pathlib import Path
+from typing import BinaryIO
+from uuid import UUID
+
 from file_storage.contracts.events import FileDeleted, FileUploaded
 from file_storage.contracts.schemas import StoredFileListOut, StoredFileOut
 from file_storage.contracts.service import (
@@ -28,4 +36,34 @@
     "StorageNotFoundError",
     "StoredFileListOut",
     "StoredFileOut",
+    "open_local",
+    "presigned_url",
+    "store_path",
 ]
+
+
+def store_path(
+    source: str | PathLike[str] | BinaryIO,
+    *,
+    filename: str | None = None,
+    content_type: str = "application/octet-stream",
+    public: bool = False,
+) -> StoredFileOut:
+    """Store a path or stream in the caller's tenant, in an independent transaction."""
+    from file_storage.worker import store_path as store
+
+    return store(source, filename=filename, content_type=content_type, public=public)
+
+
+def open_local(file_id: UUID) -> AbstractContextManager[Path]:
+    """Download to a private local path for the lifetime of this context."""
+    from file_storage.worker import open_local as local
+
+    return local(file_id)
+
+
+def presigned_url(file_id: UUID, *, ttl_seconds: int | None = None) -> str:
+    """Return a scoped download URL; unsupported backends raise NotSupportedError."""
+    from file_storage.worker import presigned_url as sign
+
+    return sign(file_id, ttl_seconds=ttl_seconds)
diff --git a/modules/file_storage/file_storage/contracts/service.py b/modules/file_storage/file_storage/contracts/service.py
index 7f2958fbc..6d6536f57 100644
--- a/modules/file_storage/file_storage/contracts/service.py
+++ b/modules/file_storage/file_storage/contracts/service.py
@@ -63,7 +63,7 @@ async def put(
         content_type: str,
         size: int,
     ) -> None:
-        """Persist ``stream`` under ``key``. ``size`` is the total byte count."""
+        """Persist ``stream`` under ``key``. ``size`` is the byte count, or 0 if unknown."""
 
     async def get(self, key: str) -> AsyncIterator[bytes]:
         """Return an iterator over the object's bytes in bounded chunks.
diff --git a/modules/file_storage/file_storage/models.py b/modules/file_storage/file_storage/models.py
index 005892d28..ef7531f58 100644
--- a/modules/file_storage/file_storage/models.py
+++ b/modules/file_storage/file_storage/models.py
@@ -39,7 +39,8 @@ class StoredFile(Base, AuditMixin, SoftDeleteMixin, MultiTenantMixin, table=True
     key: str = Field(max_length=512)
     filename: str = Field(max_length=255)
     content_type: str = Field(max_length=128)
-    size_bytes: int = Field()
+    # BIGINT: server-side puts can exceed 2 GiB (#708).
+    size_bytes: int = Field(sa_column=sa.Column(sa.BigInteger(), nullable=False))
     backend: str = Field(max_length=32)
     # Null for multipart uploads until ``file_storage.compute_checksum`` has
     # streamed the object (or forever, with no background_tasks): the bytes
diff --git a/modules/file_storage/file_storage/service.py b/modules/file_storage/file_storage/service.py
index 10bdd15b1..7cdfa839d 100644
--- a/modules/file_storage/file_storage/service.py
+++ b/modules/file_storage/file_storage/service.py
@@ -115,7 +115,8 @@ async def _hashing_stream() -> AsyncIterator[bytes]:
             key,
             _hashing_stream(),
             content_type=content_type,
-            size=0,  # unknown until stream is drained; backends that need it can spool
+            # 0 means unknown: the backend spools, or streams multipart parts.
+            size=upload.size or 0,
         )
 
         # Compensation: on DB failure, drop the just-uploaded object. A
@@ -139,10 +140,10 @@ async def _hashing_stream() -> AsyncIterator[bytes]:
                 self.db.add(row)
                 await self.db.flush()
                 await self.db.refresh(row)
-        except Exception:
+        except BaseException:
             try:
                 await self.backend.delete(key)
-            except Exception:
+            except BaseException:
                 _logger.exception(
                     "file_storage.cleanup_failed key=%s — original upload error follows",
                     key,
diff --git a/modules/file_storage/file_storage/settings.py b/modules/file_storage/file_storage/settings.py
index e16f8c79d..80d47164d 100644
--- a/modules/file_storage/file_storage/settings.py
+++ b/modules/file_storage/file_storage/settings.py
@@ -123,6 +123,8 @@ class FileStorageSettings(DbBackedSettings):
     multipart_part_size_bytes: int = Field(
         default=constants.DEFAULT_PART_SIZE_BYTES,
         ge=constants.MIN_PART_SIZE_BYTES,
+        # S3 refuses parts over 5 GiB; a larger value would fail every multipart put.
+        le=constants.MAX_S3_PART_SIZE_BYTES,
         description="Preferred part size; grown automatically to stay under 10,000 parts.",
     )
     upload_session_ttl_seconds: int = Field(
diff --git a/modules/file_storage/file_storage/tasks.py b/modules/file_storage/file_storage/tasks.py
index 522f99a5e..fa982ae60 100644
--- a/modules/file_storage/file_storage/tasks.py
+++ b/modules/file_storage/file_storage/tasks.py
@@ -44,13 +44,9 @@
 
 
 def _backend_and_settings():
-    from background_tasks import settings_for
+    from file_storage.worker_support import backend_and_settings
 
-    from file_storage.backends import build_backend
-    from file_storage.settings import FileStorageSettings
-
-    settings = settings_for(FileStorageSettings, constants.MODULE_NAME)
-    return build_backend(settings), settings
+    return backend_and_settings()
 
 
 @shared_task(name=constants.Task.REAP_STALE_UPLOADS)
@@ -122,7 +118,9 @@ async def _abort_all(backend, targets: list[tuple[str, str]]) -> set[tuple[str,
     return failed
 
 
-# Keys this module mints: ``<tenant>/<yyyy>/<mm>/<dd>/<32 hex><ext>``.
+# Keys this module mints: ``<tenant>/<yyyy>/<mm>/<dd>/<32 hex><ext>``. A server-side
+# put that went multipart and died is an orphan like a browser one: no real put
+# runs for the 2 x session-TTL age the reaper waits, so it is safe to abort.
 _OWN_KEY = constants.OWN_KEY
 
 
diff --git a/modules/file_storage/file_storage/worker.py b/modules/file_storage/file_storage/worker.py
new file mode 100644
index 000000000..e04de0882
--- /dev/null
+++ b/modules/file_storage/file_storage/worker.py
@@ -0,0 +1,182 @@
+"""Synchronous, tenant-scoped file access for workers (#708).
+
+Uses the background-tasks calling thread's persistent loop and session factory;
+imports of this module are deferred by the public contracts package.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+from collections.abc import Generator
+from contextlib import contextmanager
+from functools import partial
+from pathlib import Path
+from tempfile import TemporaryDirectory
+from typing import BinaryIO
+from uuid import UUID
+
+from background_tasks.outbox_loop import run_on_worker_loop
+from background_tasks.outbox_worker import worker_bus, worker_session_factory
+from fastapi import UploadFile
+from simple_module_db import independent_session
+from simple_module_db.callbacks import register_on_commit
+from starlette.datastructures import Headers
+
+from file_storage.backend_guard import require_backend
+from file_storage.contracts import FileUploaded, NotSupportedError, StoredFileOut
+from file_storage.downloads import presigned_download_url
+from file_storage.models import StoredFile
+from file_storage.scope import owning_tenant
+from file_storage.service import FileStorageService
+from file_storage.worker_support import backend_and_settings
+
+logger = logging.getLogger(__name__)
+
+
+def store_path(
+    source: str | os.PathLike[str] | BinaryIO,
+    *,
+    filename: str | None = None,
+    content_type: str = "application/octet-stream",
+    public: bool = False,
+) -> StoredFileOut:
+    """Store a path or the remaining bytes of a caller-owned stream.
+
+    Anonymous streams require ``filename``. Caller-owned streams stay open.
+    The independent transaction commits before this function returns.
+    """
+    backend, settings = backend_and_settings()
+    factory = worker_session_factory()
+    bus = worker_bus()
+
+    async def store() -> StoredFileOut:
+        out = None
+        committed = False
+
+        def mark_committed() -> None:
+            nonlocal committed
+            committed = True
+
+        try:
+            async with independent_session(factory) as db:
+                owning_tenant(db)  # Fail closed before even opening the source path.
+                service = FileStorageService(db, backend, settings)
+                with _source(source, filename) as (handle, name):
+                    upload = UploadFile(
+                        handle,
+                        filename=name,
+                        size=_remaining_size(handle),
+                        headers=Headers({"content-type": content_type}),
+                    )
+                    out = await service.upload(upload, public=public)
+                register_on_commit(db, mark_committed)
+                register_on_commit(
+                    db,
+                    partial(
+                        bus.publish,
+                        FileUploaded(
+                            file_id=out.id,
+                            key=out.key,
+                            backend=out.backend,
+                            size_bytes=out.size_bytes,
+                            uploaded_by=out.uploaded_by,
+                        ),
+                    ),
+                )
+            return out
+        except BaseException:
+            if out is not None and not committed:
+                await _compensate(factory, backend, out)
+            raise
+
+    return run_on_worker_loop(store)
+
+
+async def _compensate(factory, backend, out: StoredFileOut) -> None:
+    # A lost COMMIT acknowledgement is ambiguous: never delete a durable file.
+    try:
+        async with factory() as db:
+            # Trashed rows keep their object for restore: only a missing row means rollback.
+            row = await db.get(StoredFile, out.id, execution_options={"include_deleted": True})
+    except BaseException:
+        logger.exception("file_storage.worker_commit_check_failed key=%s", out.key)
+        return
+    if row is None:
+        try:
+            await backend.delete(out.key)
+        except BaseException:
+            logger.exception("file_storage.worker_cleanup_failed key=%s", out.key)
+
+
+@contextmanager
+def _source(source, filename) -> Generator[tuple[BinaryIO, str]]:
+    if isinstance(source, (str, os.PathLike)):
+        path = Path(source)
+        with path.open("rb") as handle:
+            yield handle, filename or path.name
+    else:
+        name = filename or getattr(source, "name", None)
+        if not isinstance(name, (str, os.PathLike)) or not os.fspath(name):
+            raise ValueError("filename is required for anonymous streams")
+        yield source, filename or Path(name).name
+
+
+def _remaining_size(handle: BinaryIO) -> int | None:
+    """Bytes left in a real file, so S3 can size its parts; ``None`` for a pipe or buffer.
+
+    Without it a server put is treated as unknown-length, its part size can't
+    grow, and anything past 10,000 parts (156 GiB at the 16 MiB default) fails.
+    """
+    try:
+        return os.fstat(handle.fileno()).st_size - handle.tell()
+    except (AttributeError, OSError, ValueError):
+        return None
+
+
+def _reader(file_id: UUID):
+    backend, settings = backend_and_settings()
+    factory = worker_session_factory()
+
+    async def metadata():
+        async with factory() as db:
+            service = FileStorageService(db, backend, settings)
+            row = await service.get(file_id)
+            require_backend(backend, row)
+            return row
+
+    # Metadata is detached and its transaction closed before any streaming.
+    return backend, settings, run_on_worker_loop(metadata)
+
+
+@contextmanager
+def open_local(file_id: UUID) -> Generator[Path]:
+    """Download into a private temporary directory, removed on every exit."""
+    backend, _, row = _reader(file_id)
+    with TemporaryDirectory(prefix="sm-file-") as directory:
+        path = Path(directory) / ("file" + Path(row.filename).suffix)
+
+        async def download() -> None:
+            body = await backend.get(row.key)
+            try:
+                with path.open("wb") as handle:
+                    async for chunk in body:
+                        handle.write(chunk)
+            finally:
+                close = getattr(body, "aclose", None)
+                if close is not None:
+                    await close()
+
+        run_on_worker_loop(download)
+        yield path
+
+
+def presigned_url(file_id: UUID, *, ttl_seconds: int | None = None) -> str:
+    """Sign a scoped file download, or raise ``NotSupportedError``."""
+    if ttl_seconds is not None and ttl_seconds <= 0:
+        raise ValueError("ttl_seconds must be positive")
+    backend, settings, row = _reader(file_id)
+    if not backend.supports_presigned_url:
+        raise NotSupportedError("The configured backend cannot presign downloads")
+    ttl = settings.s3_presign_ttl_seconds if ttl_seconds is None else ttl_seconds
+    return run_on_worker_loop(lambda: presigned_download_url(backend, row, ttl, attachment=True))
diff --git a/modules/file_storage/file_storage/worker_support.py b/modules/file_storage/file_storage/worker_support.py
new file mode 100644
index 000000000..e8b6b4577
--- /dev/null
+++ b/modules/file_storage/file_storage/worker_support.py
@@ -0,0 +1,13 @@
+"""Hydrated backend construction shared by sync worker APIs and tasks."""
+
+from file_storage import constants
+from file_storage.backends import build_backend
+from file_storage.contracts.service import StorageBackend
+from file_storage.settings import FileStorageSettings
+
+
+def backend_and_settings() -> tuple[StorageBackend, FileStorageSettings]:
+    from background_tasks.worker_settings import settings_for
+
+    settings = settings_for(FileStorageSettings, constants.MODULE_NAME)
+    return build_backend(settings), settings
diff --git a/modules/file_storage/tests/conftest.py b/modules/file_storage/tests/conftest.py
index a6387828a..53839ecd2 100644
--- a/modules/file_storage/tests/conftest.py
+++ b/modules/file_storage/tests/conftest.py
@@ -15,6 +15,7 @@
 import pytest
 from file_storage.backends.filesystem import FilesystemBackend
 from file_storage_uploads_helpers import PART
+from file_storage_worker_helpers import worker_db as worker_db
 from sqlalchemy import event
 from sqlalchemy.engine import Engine
 
diff --git a/modules/file_storage/tests/file_storage_worker_helpers.py b/modules/file_storage/tests/file_storage_worker_helpers.py
new file mode 100644
index 000000000..9d5361ce8
--- /dev/null
+++ b/modules/file_storage/tests/file_storage_worker_helpers.py
@@ -0,0 +1,40 @@
+"""A real worker database and filesystem, without a Celery broker."""
+
+import pytest
+from background_tasks import outbox_worker, sync_db
+from background_tasks.worker_settings import clear_settings_cache
+from file_storage.models import StoredFile
+from simple_module_core.events import EventBus
+from simple_module_db.stored_settings import Setting
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
+
+@pytest.fixture
+def worker_db(tmp_path, monkeypatch):
+    outbox_worker.reset_worker_state()
+    sync_db.dispose_sync_engine()
+    clear_settings_cache()
+    monkeypatch.setattr(sync_db, "discover_audit_callback", lambda: None)
+    engine = create_engine(f"sqlite:///{tmp_path / 'worker.db'}")
+    StoredFile.metadata.create_all(engine)
+    Setting.metadata.create_all(engine)
+    root = tmp_path / "objects"
+    with Session(engine) as db:
+        db.add(Setting(key="file_storage.fs_root_path", value=str(root), value_type="string"))
+        db.commit()
+    sync_db.set_database_url(str(engine.url), tenant_strict=True)
+    bus = EventBus()
+    outbox_worker.use_bus(bus)
+    yield root, engine, bus
+    outbox_worker.reset_worker_state()
+    sync_db.dispose_sync_engine()
+    clear_settings_cache()
+    engine.dispose()
+
+
+def set_setting(engine, name, value, value_type="integer"):
+    with Session(engine) as db:
+        db.add(Setting(key=f"file_storage.{name}", value=value, value_type=value_type))
+        db.commit()
+    clear_settings_cache()
diff --git a/modules/file_storage/tests/test_file_storage_server_put.py b/modules/file_storage/tests/test_file_storage_server_put.py
new file mode 100644
index 000000000..73d8a448a
--- /dev/null
+++ b/modules/file_storage/tests/test_file_storage_server_put.py
@@ -0,0 +1,222 @@
+"""Server-side multipart boundaries without a network service (#708)."""
+
+import asyncio
+from contextlib import asynccontextmanager
+
+import pytest
+from file_storage import constants
+from file_storage.backends import build_backend
+from file_storage.settings import FileStorageSettings
+from file_storage.upload_support import SizeMismatchError
+
+
+class S3Wire:
+    def __init__(self):
+        self.parts = []
+        self.single = None
+        self.completed = None
+        self.aborted = False
+        self.failure = None
+        self.abort_failure = False
+
+    async def put_object(self, **kw):
+        self.single = kw["Body"].read()
+        assert kw["ContentLength"] == len(self.single)
+
+    async def create_multipart_upload(self, **kw):
+        return {"UploadId": "upload"}
+
+    async def upload_part(self, **kw):
+        assert kw["PartNumber"] == len(self.parts) + 1
+        body = kw["Body"]
+        self.parts.append(body if isinstance(body, bytes) else body.read())
+        return {"ETag": str(kw["PartNumber"])}
+
+    async def complete_multipart_upload(self, **kw):
+        if self.failure:
+            raise self.failure
+        self.completed = kw["MultipartUpload"]["Parts"]
+
+    async def abort_multipart_upload(self, **kw):
+        self.aborted = True
+        if self.abort_failure:
+            raise RuntimeError("abort failed")
+
+
+@pytest.fixture
+def wire_backend(monkeypatch):
+    wire = S3Wire()
+    cfg = FileStorageSettings(
+        backend="s3",
+        s3_bucket="bucket",
+        s3_region="us-east-1",
+        direct_upload_threshold_bytes=6,
+    )
+    backend = build_backend(cfg)
+
+    @asynccontextmanager
+    async def client():
+        yield wire
+
+    monkeypatch.setattr(backend, "_client", client)
+    # Tiny, hand-verifiable parts after testing settings propagation separately.
+    backend.multipart_part_size_bytes = 5
+    monkeypatch.setattr(
+        "file_storage.upload_support.choose_part_size", lambda size, preferred: preferred
+    )
+    return wire, backend
+
+
+async def chunks(*values):
+    for value in values:
+        yield value
+
+
+@pytest.mark.parametrize(
+    "payload,multipart",
+    [(b"", False), (b"123456", False), (b"1234567", True), (b"1234567890", True)],
+)
+@pytest.mark.parametrize("known", [False, True])
+async def test_threshold_and_exact_final_part(wire_backend, payload, multipart, known):
+    wire, backend = wire_backend
+    await backend.put(
+        "key",
+        chunks(payload[:3], payload[3:]),
+        content_type="text/plain",
+        size=len(payload) if known else 0,
+    )
+    if multipart:
+        assert b"".join(wire.parts) == payload
+        assert [len(p) for p in wire.parts] == ([5, 2] if len(payload) == 7 else [5, 5])
+        assert wire.single is None
+        assert wire.completed == [{"PartNumber": 1, "ETag": "1"}, {"PartNumber": 2, "ETag": "2"}]
+    else:
+        assert wire.single == payload
+        assert wire.parts == []
+
+
+@pytest.mark.parametrize("failure", [ValueError("source failed"), asyncio.CancelledError()])
+async def test_stream_failure_aborts_and_preserves_error(wire_backend, failure):
+    wire, backend = wire_backend
+    wire.abort_failure = True
+
+    async def broken():
+        yield b"1234567"
+        raise failure
+
+    with pytest.raises(type(failure)) as caught:
+        await backend.put("key", broken(), content_type="text/plain", size=0)
+    assert caught.value is failure
+    assert wire.aborted
+    assert wire.completed is None
+
+
+async def test_completion_failure_aborts(wire_backend):
+    wire, backend = wire_backend
+    wire.failure = RuntimeError("complete failed")
+    wire.abort_failure = True
+    with pytest.raises(Exception, match="complete failed"):
+        await backend.put("key", chunks(b"1234567"), content_type="text/plain", size=0)
+    assert wire.aborted
+
+
+@pytest.mark.parametrize("declared", [2, 8, 12])
+async def test_declared_size_mismatch_is_not_published(wire_backend, declared):
+    wire, backend = wire_backend
+    with pytest.raises(SizeMismatchError):
+        await backend.put("key", chunks(b"1234567890"), content_type="text/plain", size=declared)
+    assert wire.single is None and wire.completed is None
+    assert wire.aborted
+
+
+async def test_part_count_bound_aborts(wire_backend, monkeypatch):
+    wire, backend = wire_backend
+    monkeypatch.setattr(constants, "MAX_UPLOAD_PARTS", 1)
+    with pytest.raises(Exception, match="part"):
+        await backend.put("key", chunks(b"12345678901"), content_type="text/plain", size=0)
+    assert wire.aborted
+    assert len(wire.parts) == 1
+
+
+def test_factory_passes_stored_multipart_settings():
+    backend = build_backend(
+        FileStorageSettings(
+            backend="s3",
+            s3_bucket="bucket",
+            s3_region="us-east-1",
+            direct_upload_threshold_bytes=123,
+            multipart_part_size_bytes=7 * constants.MIB,
+        )
+    )
+    assert backend.multipart_threshold_bytes == 123
+    assert backend.multipart_part_size_bytes == 7 * constants.MIB
+
+
+async def test_unknown_stream_uploads_before_source_is_exhausted(wire_backend):
+    wire, backend = wire_backend
+
+    async def source():
+        yield b"1234567"
+        assert wire.parts == [b"12345"]
+        yield b"890"
+
+    await backend.put("key", source(), content_type="text/plain", size=0)
+    assert b"".join(wire.parts) == b"1234567890"
+
+
+async def test_large_source_chunk_keeps_spools_bounded(wire_backend, monkeypatch):
+    from tempfile import SpooledTemporaryFile
+
+    from file_storage.backends import s3_put
+
+    maximums = []
+
+    class TrackedBuffer:
+        def __init__(self, **kwargs):
+            self.file = SpooledTemporaryFile(**kwargs)  # noqa: SIM115 — closed by wrapper context
+            self.index = len(maximums)
+            maximums.append(0)
+
+        def __getattr__(self, name):
+            return getattr(self.file, name)
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *args):
+            self.file.close()
+
+        def write(self, data):
+            self.file.write(data)
+            maximums[self.index] = max(maximums[self.index], self.file.tell())
+
+    monkeypatch.setattr(s3_put, "SpooledTemporaryFile", TrackedBuffer)
+    wire, backend = wire_backend
+    await backend.put("key", chunks(b"x" * 100), content_type="text/plain", size=0)
+    assert b"".join(wire.parts) == b"x" * 100
+    assert maximums == [7, 5]
+
+
+async def test_failed_part_upload_aborts(wire_backend, monkeypatch):
+    wire, backend = wire_backend
+
+    async def fail(**kwargs):
+        raise RuntimeError("part failed")
+
+    monkeypatch.setattr(wire, "upload_part", fail)
+    with pytest.raises(Exception, match="part failed"):
+        await backend.put("key", chunks(b"1234567"), content_type="text/plain", size=0)
+    assert wire.aborted and wire.completed is None
+
+
+def test_part_size_setting_is_capped_at_the_s3_limit():
+    from pydantic import ValidationError
+
+    with pytest.raises(ValidationError):
+        FileStorageSettings(multipart_part_size_bytes=constants.MAX_S3_PART_SIZE_BYTES + 1)
+    assert (
+        FileStorageSettings(
+            multipart_part_size_bytes=constants.MAX_S3_PART_SIZE_BYTES
+        ).multipart_part_size_bytes
+        == constants.MAX_S3_PART_SIZE_BYTES
+    )
diff --git a/modules/file_storage/tests/test_file_storage_worker.py b/modules/file_storage/tests/test_file_storage_worker.py
new file mode 100644
index 000000000..c6d38f346
--- /dev/null
+++ b/modules/file_storage/tests/test_file_storage_worker.py
@@ -0,0 +1,231 @@
+"""Public synchronous worker file operations (#708)."""
+
+import hashlib
+import stat
+from io import BytesIO
+from uuid import uuid4
+
+import pytest
+from background_tasks import outbox_worker
+from file_storage import contracts
+from file_storage.backend_guard import BackendUnavailableError
+from file_storage.contracts import FileUploaded, NotSupportedError, StoredFileOut
+from file_storage.models import StoredFile
+from file_storage.service import (
+    ContentTypeNotAllowedError,
+    FileTooLargeError,
+    StoredFileNotFoundError,
+)
+from file_storage.upload_support import QuotaExceededError
+from file_storage_worker_helpers import set_setting
+from simple_module_db import tenant_context
+from simple_module_db.tenancy import TenantIsolationError
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+
+def test_contract_exports_worker_functions():
+    for name in ("store_path", "open_local", "presigned_url"):
+        assert callable(getattr(contracts, name, None)), name
+
+
+def test_path_roundtrip_and_after_commit_event(worker_db, tmp_path):
+    _root, engine, bus = worker_db
+    source = tmp_path / "input.tif"
+    source.write_bytes(b"raster")
+    received = []
+
+    async def uploaded(event):
+        # A separate connection sees only committed metadata.
+        with Session(engine) as db:
+            assert db.get(StoredFile, event.file_id) is not None
+        received.append(event)
+
+    bus.subscribe(FileUploaded, uploaded)
+    with tenant_context("alpha"):
+        out = contracts.store_path(source, content_type="image/tiff", public=True)
+        assert isinstance(out, StoredFileOut)
+        assert (out.filename, out.content_type, out.size_bytes, out.public) == (
+            "input.tif",
+            "image/tiff",
+            6,
+            True,
+        )
+        assert out.key.startswith("alpha/")
+        assert out.checksum_sha256 == hashlib.sha256(b"raster").hexdigest()
+        with contracts.open_local(out.id) as path:
+            assert path.read_bytes() == b"raster"
+            assert stat.S_IMODE(path.parent.stat().st_mode) == 0o700
+        assert not path.parent.exists()
+    assert len(received) == 1
+    assert (received[0].file_id, received[0].key, received[0].size_bytes) == (out.id, out.key, 6)
+
+
+def test_stream_current_position_and_ownership(worker_db):
+    stream = BytesIO(b"skipcontents")
+    stream.seek(4)
+    with tenant_context("alpha"):
+        out = contracts.store_path(stream, filename="output.bin")
+        assert out.size_bytes == 8
+        with contracts.open_local(out.id) as path:
+            assert path.read_bytes() == b"contents"
+    assert not stream.closed
+    assert stream.tell() == 12
+
+
+def test_anonymous_stream_needs_filename(worker_db):
+    stream = BytesIO(b"unchanged")
+    with tenant_context("alpha"), pytest.raises(ValueError, match="filename"):
+        contracts.store_path(stream)
+    assert stream.tell() == 0 and not stream.closed
+
+
+def test_unscoped_upload_fails_before_opening_source(worker_db, tmp_path):
+    with pytest.raises(TenantIsolationError):
+        contracts.store_path(tmp_path / "does-not-exist")
+    assert not worker_db[0].exists()
+
+
+@pytest.mark.parametrize("operation", ["open_local", "presigned_url"])
+def test_missing_and_cross_tenant_ids(worker_db, operation):
+    with tenant_context("alpha"):
+        out = contracts.store_path(BytesIO(b"private"), filename="a.bin")
+    for target in (out.id, uuid4()):
+        with tenant_context("beta"), pytest.raises(StoredFileNotFoundError):
+            if operation == "open_local":
+                with contracts.open_local(target):
+                    pass
+            else:
+                contracts.presigned_url(target)
+
+
+@pytest.mark.parametrize(
+    "name,value,content_type,error",
+    [
+        ("max_file_size_bytes", "2", "text/plain", FileTooLargeError),
+        ("allowed_content_types", '["image/png"]', "text/plain", ContentTypeNotAllowedError),
+        ("quota_bytes", "2", "text/plain", QuotaExceededError),
+    ],
+)
+def test_upload_limits_leave_no_objects_or_rows(worker_db, name, value, content_type, error):
+    root, engine, _ = worker_db
+    set_setting(engine, name, value, "json" if name == "allowed_content_types" else "integer")
+    with tenant_context("alpha"), pytest.raises(error):
+        contracts.store_path(BytesIO(b"too large"), filename="a.txt", content_type=content_type)
+    assert not list(root.rglob("*.*"))
+    with Session(engine) as db:
+        assert list(db.scalars(select(StoredFile))) == []
+
+
+@pytest.mark.parametrize("phase", ["flush", "commit"])
+def test_database_failure_compensates(worker_db, monkeypatch, phase):
+    root, engine, bus = worker_db
+    received = []
+
+    async def uploaded(event):
+        received.append(event)
+
+    bus.subscribe(FileUploaded, uploaded)
+    factory = outbox_worker.worker_session_factory()
+
+    async def fail(*args, **kwargs):
+        raise RuntimeError("database failed")
+
+    monkeypatch.setattr(factory.class_, phase, fail)
+    with tenant_context("alpha"), pytest.raises(RuntimeError, match="database failed"):
+        contracts.store_path(BytesIO(b"contents"), filename="a.bin")
+    assert not list(root.rglob("*.bin"))
+    with Session(engine) as db:
+        assert list(db.scalars(select(StoredFile))) == []
+    assert received == []
+
+
+def test_local_context_cleans_on_body_failure(worker_db):
+    with tenant_context("alpha"):
+        out = contracts.store_path(BytesIO(b"contents"), filename="a.bin")
+        with pytest.raises(RuntimeError, match="consumer"), contracts.open_local(out.id) as path:
+            raise RuntimeError("consumer")
+        assert not path.parent.exists()
+
+
+def test_filesystem_presign_is_unsupported(worker_db):
+    with tenant_context("alpha"):
+        out = contracts.store_path(BytesIO(), filename="empty")
+        with pytest.raises(NotSupportedError):
+            contracts.presigned_url(out.id)
+
+
+@pytest.mark.parametrize("operation", ["open_local", "presigned_url"])
+def test_read_refuses_wrong_backend(worker_db, operation):
+    with tenant_context("alpha"):
+        out = contracts.store_path(BytesIO(b"contents"), filename="a.bin")
+    with Session(worker_db[1]) as db:
+        row = db.get(StoredFile, out.id)
+        row.backend = "other"
+        db.commit()
+    with tenant_context("alpha"), pytest.raises(BackendUnavailableError):
+        if operation == "open_local":
+            with contracts.open_local(out.id):
+                pass
+        else:
+            contracts.presigned_url(out.id)
+
+
+def test_worker_context_does_not_leak_between_calls(worker_db):
+    with tenant_context("alpha"):
+        a = contracts.store_path(BytesIO(b"a"), filename="a.bin")
+        factory = outbox_worker.worker_session_factory()
+    with tenant_context("beta"):
+        b = contracts.store_path(BytesIO(b"b"), filename="b.bin")
+        assert factory is outbox_worker.worker_session_factory()
+        with pytest.raises(StoredFileNotFoundError), contracts.open_local(a.id):
+            pass
+    with pytest.raises(TenantIsolationError):
+        contracts.store_path(BytesIO(b"c"), filename="c.bin")
+    assert a.key.startswith("alpha/") and b.key.startswith("beta/")
+
+
+def test_concurrent_callers_keep_tenants_and_pools_separate(worker_db):
+    from concurrent.futures import ThreadPoolExecutor
+
+    def store(tenant):
+        with tenant_context(tenant):
+            out = contracts.store_path(BytesIO(tenant.encode()), filename="result.bin")
+            return out, id(outbox_worker.worker_session_factory())
+
+    with ThreadPoolExecutor(max_workers=2) as executor:
+        results = list(executor.map(store, ["alpha", "beta"]))
+    assert [out.key.split("/")[0] for out, _ in results] == ["alpha", "beta"]
+    assert len({pool for _, pool in results}) == 2
+
+
+def test_contract_import_does_not_load_worker_dependencies():
+    import subprocess
+    import sys
+
+    code = """
+import builtins
+original = builtins.__import__
+def guarded(name, *args, **kwargs):
+    if name == "background_tasks" or name.startswith("background_tasks."):
+        raise AssertionError("contracts imported worker dependencies eagerly")
+    return original(name, *args, **kwargs)
+builtins.__import__ = guarded
+from file_storage.contracts import store_path, open_local, presigned_url
+assert all(callable(fn) for fn in (store_path, open_local, presigned_url))
+"""
+    subprocess.run([sys.executable, "-c", code], check=True, timeout=20)
+
+
+def test_remaining_size_lets_s3_size_parts_for_real_files(tmp_path):
+    import io
+
+    from file_storage.worker import _remaining_size
+
+    path = tmp_path / "out.bin"
+    path.write_bytes(b"x" * 10)
+    with path.open("rb") as handle:
+        handle.read(3)
+        assert _remaining_size(handle) == 7
+    # A buffer or pipe has no size to declare: S3 then streams parts of unknown total.
+    assert _remaining_size(io.BytesIO(b"abc")) is None
diff --git a/modules/file_storage/tests/test_file_storage_worker_failures.py b/modules/file_storage/tests/test_file_storage_worker_failures.py
new file mode 100644
index 000000000..5de01206c
--- /dev/null
+++ b/modules/file_storage/tests/test_file_storage_worker_failures.py
@@ -0,0 +1,178 @@
+"""Worker download, event and compensation failure boundaries (#708)."""
+
+import asyncio
+from concurrent.futures import CancelledError
+from io import BytesIO
+
+import pytest
+from background_tasks import outbox_worker
+from file_storage import contracts
+from file_storage.backends.filesystem import FilesystemBackend
+from file_storage.models import StoredFile
+from file_storage_worker_helpers import set_setting
+from simple_module_db import tenant_context
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+
+def test_cancelled_flush_removes_uploaded_bytes(worker_db, monkeypatch):
+    factory = outbox_worker.worker_session_factory()
+
+    async def fail(*args, **kwargs):
+        raise asyncio.CancelledError()
+
+    monkeypatch.setattr(factory.class_, "flush", fail)
+    with tenant_context("alpha"), pytest.raises(CancelledError):
+        contracts.store_path(BytesIO(b"contents"), filename="a.bin")
+    assert not list(worker_db[0].rglob("*.bin"))
+
+
+def test_failed_compensation_preserves_commit_error(worker_db, monkeypatch):
+    factory = outbox_worker.worker_session_factory()
+
+    async def commit(*args, **kwargs):
+        raise ValueError("commit failed")
+
+    async def delete(*args, **kwargs):
+        raise RuntimeError("cleanup failed")
+
+    monkeypatch.setattr(factory.class_, "commit", commit)
+    monkeypatch.setattr(FilesystemBackend, "delete", delete)
+    with tenant_context("alpha"), pytest.raises(ValueError, match="commit failed"):
+        contracts.store_path(BytesIO(b"contents"), filename="a.bin")
+    with Session(worker_db[1]) as db:
+        assert list(db.scalars(select(StoredFile))) == []
+
+
+def test_download_failure_closes_stream_and_removes_temporary_directory(worker_db, monkeypatch):
+    from tempfile import TemporaryDirectory
+
+    from file_storage import worker
+
+    directories = []
+    closed = []
+
+    def temporary(**kwargs):
+        directory = TemporaryDirectory(**kwargs)
+        directories.append(directory.name)
+        return directory
+
+    async def get(self, key):
+        async def broken():
+            try:
+                yield b"partial"
+                raise ValueError("download failed")
+            finally:
+                closed.append(True)
+
+        return broken()
+
+    with tenant_context("alpha"):
+        out = contracts.store_path(BytesIO(b"contents"), filename="a.bin")
+        monkeypatch.setattr(FilesystemBackend, "get", get)
+        monkeypatch.setattr(worker, "TemporaryDirectory", temporary)
+        with pytest.raises(ValueError, match="download failed"), contracts.open_local(out.id):
+            pytest.fail("partial download was yielded")
+    from pathlib import Path
+
+    assert directories and not Path(directories[0]).exists()
+    assert closed == [True]
+
+
+def test_presign_honors_settings_override_and_download_headers(worker_db, monkeypatch):
+    seen = []
+
+    async def sign(self, key, ttl_seconds, **kwargs):
+        seen.append((key, ttl_seconds, kwargs))
+        return "https://bucket.example/signed"
+
+    monkeypatch.setattr(FilesystemBackend, "supports_presigned_url", True)
+    monkeypatch.setattr(FilesystemBackend, "presigned_get_url", sign)
+    set_setting(worker_db[1], "s3_presign_ttl_seconds", "42")
+    with tenant_context("alpha"):
+        out = contracts.store_path(
+            BytesIO(b"contents"), filename="report.tif", content_type="image/tiff"
+        )
+        assert contracts.presigned_url(out.id) == "https://bucket.example/signed"
+        assert contracts.presigned_url(out.id, ttl_seconds=12) == "https://bucket.example/signed"
+        with pytest.raises(ValueError, match="positive"):
+            contracts.presigned_url(out.id, ttl_seconds=0)
+    assert [ttl for _, ttl, _ in seen] == [42, 12]
+    assert all(key == out.key for key, _, _ in seen)
+    assert seen[0][2] == {
+        "content_type": "image/tiff",
+        "content_disposition": "attachment; filename=\"report.tif\"; filename*=UTF-8''report.tif",
+    }
+
+
+def test_named_handle_infers_basename_and_stays_open(worker_db, tmp_path):
+    source = tmp_path / "named.bin"
+    source.write_bytes(b"named")
+    with source.open("rb") as handle, tenant_context("alpha"):
+        out = contracts.store_path(handle)
+        assert out.filename == "named.bin"
+        assert not handle.closed
+
+
+@pytest.mark.parametrize("operation", ["open_local", "presigned_url"])
+def test_unscoped_reads_fail_closed(worker_db, operation):
+    from simple_module_db.tenancy import TenantIsolationError
+
+    with tenant_context("alpha"):
+        out = contracts.store_path(BytesIO(b"private"), filename="a.bin")
+    with pytest.raises(TenantIsolationError):
+        if operation == "open_local":
+            with contracts.open_local(out.id):
+                pytest.fail("unscoped file was exposed")
+        else:
+            contracts.presigned_url(out.id)
+
+
+def test_download_does_not_hold_metadata_connection(worker_db, monkeypatch):
+    factory = outbox_worker.worker_session_factory()
+    real = FilesystemBackend.get
+    checked_out = []
+
+    async def get(self, key):
+        checked_out.append(factory.kw["bind"].pool.checkedout())
+        return await real(self, key)
+
+    with tenant_context("alpha"):
+        out = contracts.store_path(BytesIO(b"contents"), filename="a.bin")
+        monkeypatch.setattr(FilesystemBackend, "get", get)
+        with contracts.open_local(out.id) as path:
+            assert path.read_bytes() == b"contents"
+    assert checked_out == [0]
+
+
+def test_commit_acknowledgement_failure_keeps_committed_object(worker_db, monkeypatch):
+    factory = outbox_worker.worker_session_factory()
+    real_commit = factory.class_.commit
+
+    async def commit_then_cancel(db):
+        await real_commit(db)
+        raise asyncio.CancelledError()
+
+    monkeypatch.setattr(factory.class_, "commit", commit_then_cancel)
+    with tenant_context("alpha"), pytest.raises(CancelledError):
+        contracts.store_path(BytesIO(b"contents"), filename="a.bin")
+    with Session(worker_db[1]) as db:
+        row = db.scalars(select(StoredFile)).one()
+    assert (worker_db[0] / row.key[:2] / row.key).read_bytes() == b"contents"
+
+
+def test_unverifiable_commit_outcome_preserves_object_and_original_error(worker_db, monkeypatch):
+    factory = outbox_worker.worker_session_factory()
+
+    async def commit(*args):
+        raise ValueError("commit failed")
+
+    async def get(*args, **kwargs):
+        raise RuntimeError("database unavailable")
+
+    monkeypatch.setattr(factory.class_, "commit", commit)
+    monkeypatch.setattr(factory.class_, "get", get)
+    with tenant_context("alpha"), pytest.raises(ValueError, match="commit failed"):
+        contracts.store_path(BytesIO(b"contents"), filename="a.bin")
+    objects = list(worker_db[0].rglob("*.bin"))
+    assert len(objects) == 1 and objects[0].read_bytes() == b"contents"