diff --git a/ami/main/api/serializers.py b/ami/main/api/serializers.py
index acefa7785..f7eefe0fd 100644
--- a/ami/main/api/serializers.py
+++ b/ami/main/api/serializers.py
@@ -887,6 +887,64 @@ class BulkIdentificationResponseSerializer(serializers.Serializer):
     results = BulkIdentificationResultSerializer(many=True)
 
 
+MAX_UPLOAD_REQUEST_FILES = 1000
+
+
+class UploadRequestFileSerializer(serializers.Serializer):
+    """One file the client wants to upload directly to storage."""
+
+    filename = serializers.CharField(max_length=255)
+    size = serializers.IntegerField()
+    content_type = serializers.CharField(required=False, allow_blank=True, default="")
+    # base64-encoded raw SHA-256 digest; optional and only honoured against real AWS.
+    sha256 = serializers.CharField(required=False, allow_blank=True, default="")
+
+
+class UploadRequestSerializer(serializers.Serializer):
+    """Request body for POST /deployments/{id}/upload-request/."""
+
+    subdir = serializers.CharField(required=False, allow_blank=True, default="")
+    files = UploadRequestFileSerializer(many=True, allow_empty=False)
+
+    def validate_subdir(self, value: str) -> str:
+        if value and (".." in value or value.startswith("/") or "\\" in value or any(ord(c) < 32 for c in value)):
+            raise serializers.ValidationError(
+                "subdir must not contain '..', a leading '/', backslashes, or control characters."
+            )
+        return value
+
+    def validate_files(self, value: list[dict]) -> list[dict]:
+        if len(value) > MAX_UPLOAD_REQUEST_FILES:
+            raise serializers.ValidationError(
+                f"A single request may contain at most {MAX_UPLOAD_REQUEST_FILES} files, got {len(value)}."
+            )
+        return value
+
+
+class UploadRequestUrlSerializer(serializers.Serializer):
+    """A minted presigned PUT URL for one file."""
+
+    filename = serializers.CharField()
+    key = serializers.CharField()
+    url = serializers.CharField()
+    method = serializers.CharField()
+    headers = serializers.DictField(child=serializers.CharField())
+    expires_at = serializers.DateTimeField()
+
+
+class UploadRequestErrorSerializer(serializers.Serializer):
+    """A file that was rejected before any URL was minted."""
+
+    filename = serializers.CharField()
+    code = serializers.CharField()
+    detail = serializers.CharField()
+
+
+class UploadRequestResponseSerializer(serializers.Serializer):
+    urls = UploadRequestUrlSerializer(many=True)
+    errors = UploadRequestErrorSerializer(many=True)
+
+
 class TaxonDetectionsSerializer(DefaultSerializer):
     class Meta:
         model = Detection
diff --git a/ami/main/api/views.py b/ami/main/api/views.py
index d04fa6703..cb71e78f0 100644
--- a/ami/main/api/views.py
+++ b/ami/main/api/views.py
@@ -1,7 +1,10 @@
 import datetime
 import logging
+import pathlib
+import re
 from statistics import mode
 
+import django_filters
 from django.conf import settings
 from django.contrib.postgres.search import TrigramSimilarity
 from django.core import exceptions
@@ -14,7 +17,8 @@
 from django.utils import timezone
 from django_filters.rest_framework import DjangoFilterBackend, FilterSet
 from drf_spectacular.types import OpenApiTypes
-from drf_spectacular.utils import OpenApiParameter, extend_schema
+from drf_spectacular.utils import OpenApiParameter, extend_schema, inline_serializer
+from guardian.shortcuts import get_objects_for_user
 from rest_framework import exceptions as api_exceptions
 from rest_framework import filters, serializers, status, viewsets
 from rest_framework.decorators import action
@@ -38,8 +42,11 @@
 from ami.main.models_future.occurrence import model_agreement_for_project, top_identifiers_for_project
 from ami.ml.models.algorithm import Algorithm
 from ami.ml.serializers import AlgorithmSerializer
+from ami.utils import s3
+from ami.utils.dates import get_image_timestamp_from_filename
+from ami.utils.fields import url_boolean_param
 from ami.utils.requests import get_default_classification_threshold
-from ami.utils.storages import ConnectionTestResult
+from ami.utils.storages import IMAGE_FILE_EXTENSIONS, ConnectionTestResult
 
 from ..models import (
     Classification,
@@ -101,6 +108,8 @@
     TaxonSearchResultSerializer,
     TaxonSerializer,
     TopIdentifiersResponseSerializer,
+    UploadRequestResponseSerializer,
+    UploadRequestSerializer,
 )
 
 logger = logging.getLogger(__name__)
@@ -193,6 +202,30 @@ def get_queryset(self):
             if user:
                 qs = qs.filter_by_user(user)
 
+        # `?role=manager` (or `?writable=true`) narrows to projects the user may
+        # write to. get_objects_for_user on `update_project` covers ProjectManagers
+        # AND owners (owners are auto-assigned ProjectManager, see main/signals.py),
+        # unlike filter_by_user (members) or role group names (fragile).
+        role = self.request.query_params.get("role")
+        writable = url_boolean_param(self.request, "writable", default=False)
+        if role == "manager" or writable:
+            user = self.request.user
+            if not user or not user.is_authenticated:
+                qs = qs.none()
+            elif not user.is_superuser:
+                # accept_global_perms=False is load-bearing. Role groups carry the
+                # model-level permission as well as the per-project one (see
+                # create_roles_for_project), so a user who manages any one project
+                # holds update_project globally, and guardian's default would answer
+                # with every project in the database.
+                writable_ids = get_objects_for_user(
+                    user,
+                    Project.Permissions.UPDATE_PROJECT,
+                    Project,
+                    accept_global_perms=False,
+                ).values("pk")
+                qs = qs.filter(pk__in=writable_ids)
+
         # Annotate "recent activity" fields only when sorting by them, so the
         # default list stays cheap. Each is a correlated subquery returning one
         # row via a covering index, and only one is ever added per request.
@@ -281,12 +314,70 @@ def charts(self, request, pk=None):
                 required=False,
                 type=OpenApiTypes.INT,
             ),
+            OpenApiParameter(
+                name="role",
+                description=(
+                    "Set to `manager` to return only projects the current user can write to "
+                    "(project managers and owners). Equivalent to `writable=true`."
+                ),
+                required=False,
+                type=OpenApiTypes.STR,
+            ),
+            OpenApiParameter(
+                name="writable",
+                description="Set to `true` to return only projects the current user can write to.",
+                required=False,
+                type=OpenApiTypes.BOOL,
+            ),
         ]
     )
     def list(self, request, *args, **kwargs):
         return super().list(request, *args, **kwargs)
 
 
+class DeploymentFilterSet(django_filters.FilterSet):
+    """Expose the exact param names the mobile client sends.
+
+    ``research_site_id`` / ``device_id`` map straight to the FK id columns.
+    (``StationService.resolve`` on the Swift client sends those names.)
+    """
+
+    research_site_id = django_filters.NumberFilter(field_name="research_site_id")
+    device_id = django_filters.NumberFilter(field_name="device_id")
+
+    class Meta:
+        model = Deployment
+        fields = ["research_site_id", "device_id"]
+
+
+def _validate_upload_filename(filename: str, size: int) -> dict | None:
+    """Validate a single upload filename. Return an ``{code, detail}`` dict on
+    rejection, or ``None`` when the file is acceptable to sign.
+
+    These mirror the constraints ``sync_captures`` later imposes so we never
+    mint a URL for a file the sync would silently drop or the DB would reject.
+    """
+    if ".." in filename or filename.startswith("/") or "\\" in filename or any(ord(c) < 32 for c in filename):
+        return {
+            "code": "invalid_filename",
+            "detail": "Filename must not contain '..', a leading '/', backslashes, or control characters.",
+        }
+    if get_image_timestamp_from_filename(filename) is None:
+        return {
+            "code": "unparseable_timestamp",
+            "detail": "Filename has no parseable timestamp; sync would drop it.",
+        }
+    suffix = pathlib.Path(filename).suffix.lower().lstrip(".")
+    if suffix not in IMAGE_FILE_EXTENSIONS:
+        return {
+            "code": "invalid_extension",
+            "detail": f"'{suffix}' is not a supported image extension.",
+        }
+    if size <= 0:
+        return {"code": "invalid_size", "detail": "File size must be greater than zero."}
+    return None
+
+
 class DeploymentViewSet(DefaultViewSet, ProjectMixin):
     """
     A model viewset that uses different serializers
@@ -294,6 +385,7 @@ class DeploymentViewSet(DefaultViewSet, ProjectMixin):
     """
 
     queryset = Deployment.objects.select_related("project", "device", "research_site")
+    filterset_class = DeploymentFilterSet
     ordering_fields = [
         "created_at",
         "updated_at",
@@ -342,6 +434,24 @@ def get_queryset(self) -> QuerySet:
 
         return qs
 
+    @extend_schema(
+        request=None,
+        responses=inline_serializer(
+            name="DeploymentSyncResponse",
+            fields={
+                "job_id": serializers.IntegerField(),
+                "project_id": serializers.IntegerField(),
+            },
+        ),
+        description=(
+            "Queue a background job to sync captures from the deployment's data source.\n\n"
+            "Returns the `job_id` of the enqueued `data_storage_sync` job. Poll its status at "
+            "`GET /api/v2/jobs/{job_id}/` (also filterable via "
+            "`GET /api/v2/jobs/?deployment={id}&job_type_key=data_storage_sync`). The job is "
+            "finished when its `status` reaches one of the three terminal states: "
+            "`SUCCESS`, `FAILURE`, or `REVOKED`."
+        ),
+    )
     @action(detail=True, methods=["post"], name="sync")
     def sync(self, _request, pk=None) -> Response:
         """
@@ -359,6 +469,89 @@ def sync(self, _request, pk=None) -> Response:
         else:
             raise api_exceptions.ValidationError(detail="Deployment must have a data source to sync captures from")
 
+    @extend_schema(request=UploadRequestSerializer, responses=UploadRequestResponseSerializer)
+    @action(detail=True, methods=["post"], url_path="upload-request")
+    def upload_request(self, request, pk=None) -> Response:
+        """
+        Mint short-lived presigned PUT URLs for direct-to-storage capture uploads.
+
+        The client uploads each file straight to the deployment's storage source,
+        then calls ``sync`` to ingest them. Each requested file is validated up
+        front; rejected files come back in ``errors`` (with no URL) while valid
+        files come back in ``urls``. The minted ``key`` is exactly the object Key
+        the subsequent sync will store as ``SourceImage.path``, so re-requesting
+        and re-uploading the same file is idempotent.
+        """
+        deployment: Deployment = self.get_object()
+        if not deployment.data_source:
+            raise api_exceptions.ValidationError(detail="Deployment must have a data source to upload captures to.")
+
+        request_serializer = UploadRequestSerializer(data=request.data)
+        request_serializer.is_valid(raise_exception=True)
+        subdir = request_serializer.validated_data.get("subdir") or ""
+        files = request_serializer.validated_data["files"]
+
+        config = deployment.data_source.config
+        regex = re.compile(deployment.data_source_regex) if deployment.data_source_regex else None
+        expires_in = 60 * 60
+        # Single wall-clock read shared by every URL so the batch expires together.
+        expires_at = timezone.now() + datetime.timedelta(seconds=expires_in)
+
+        urls: list[dict] = []
+        errors: list[dict] = []
+        for file in files:
+            filename = file["filename"]
+            error = _validate_upload_filename(filename, file["size"])
+            if error:
+                errors.append({"filename": filename, **error})
+                continue
+
+            full_key = s3.derive_upload_key(config, filename, [deployment.data_source_subdir, subdir])
+
+            # SourceImage.path is max_length=255; a longer key would fail to ingest.
+            if len(full_key) > 255:
+                errors.append(
+                    {
+                        "filename": filename,
+                        "code": "key_too_long",
+                        "detail": f"Resulting object key is {len(full_key)} chars (max 255).",
+                    }
+                )
+                continue
+
+            # Same semantics as sync's _filter_single_key: don't sign uploads the
+            # deployment's regex would exclude from ingestion.
+            if regex and not regex.search(full_key):
+                errors.append(
+                    {
+                        "filename": filename,
+                        "code": "regex_mismatch",
+                        "detail": "Object key does not match the deployment's data_source_regex.",
+                    }
+                )
+                continue
+
+            url, headers = s3.get_presigned_put_url(
+                config,
+                full_key,
+                content_type=file.get("content_type") or None,
+                checksum_sha256_b64=file.get("sha256") or None,
+                expires_in=expires_in,
+            )
+            urls.append(
+                {
+                    "filename": filename,
+                    "key": full_key,
+                    "url": url,
+                    "method": "PUT",
+                    "headers": headers,
+                    "expires_at": expires_at,
+                }
+            )
+
+        response = UploadRequestResponseSerializer({"urls": urls, "errors": errors})
+        return Response(response.data)
+
     @action(detail=False, methods=["post"], name="sync-all", url_path="sync-all")
     def sync_all(self, request) -> Response:
         """
diff --git a/ami/main/models.py b/ami/main/models.py
index 78e67e275..22fb2c20b 100644
--- a/ami/main/models.py
+++ b/ami/main/models.py
@@ -884,6 +884,16 @@ def data_source_total_size_display(self) -> str:
         else:
             return filesizeformat(self.data_source_total_size)
 
+    def check_custom_permission(self, user, action: str) -> bool:
+        # `upload_request` mints presigned PUT URLs for direct-to-storage capture
+        # uploads; treat it as the write side of syncing this deployment's data
+        # source, so it reuses SYNC_DEPLOYMENT (no new guardian perm / migration).
+        # Without this, get_object() would probe the nonexistent
+        # `upload_request_deployment` perm and 403 every non-superuser. See #1379.
+        if action == "upload_request":
+            return user.has_perm(Project.Permissions.SYNC_DEPLOYMENT, self.project)
+        return super().check_custom_permission(user, action)
+
     def sync_captures(
         self,
         batch_size=1000,
diff --git a/ami/main/tests.py b/ami/main/tests.py
index e8f56c485..e8941ee34 100644
--- a/ami/main/tests.py
+++ b/ami/main/tests.py
@@ -3151,6 +3151,307 @@ def test_per_row_sync_permission_matrix(self):
                 self.assertEqual(response.status_code, expected, f"{role_name} got {response.status_code}")
 
 
+class TestDeploymentUploadRequest(APITestCase):
+    """POST /deployments/{id}/upload-request/ mints presigned PUT URLs.
+
+    Signing is a local operation (no network), so the permission and validation
+    tests run without MinIO; only the E2E flagship needs the live storage stack.
+    """
+
+    def setUp(self):
+        super().setUp()
+        self.project = Project.objects.create(name="Upload Request Project")
+        create_roles_for_project(self.project)
+
+        self.superuser = User.objects.create_superuser(email="super-upload@insectai.org", password="pw")
+        self.pm_user = User.objects.create_user(email="pm-upload@insectai.org", password="pw")
+        self.ml_user = User.objects.create_user(email="ml-upload@insectai.org", password="pw")
+        self.researcher = User.objects.create_user(email="researcher-upload@insectai.org", password="pw")
+        self.basic_user = User.objects.create_user(email="basic-upload@insectai.org", password="pw")
+        self.outsider = User.objects.create_user(email="outsider-upload@insectai.org", password="pw")
+        ProjectManager.assign_user(self.pm_user, self.project)
+        MLDataManager.assign_user(self.ml_user, self.project)
+        Researcher.assign_user(self.researcher, self.project)
+        BasicMember.assign_user(self.basic_user, self.project)
+
+        # Real-AWS-style source (no endpoint_url) so generate_presigned_url signs
+        # locally without contacting any server.
+        self.source = S3StorageSource.objects.create(
+            name="Upload Source",
+            bucket="test-bucket",
+            region="us-east-1",
+            prefix="uploads",
+            access_key="AKIA_TEST",
+            secret_key="secret",
+            project=self.project,
+        )
+        self.deployment = Deployment.objects.create(
+            name="Uploader",
+            project=self.project,
+            data_source=self.source,
+            data_source_subdir="station-1",
+        )
+        self.no_source = Deployment.objects.create(name="No source", project=self.project)
+        self.url = f"/api/v2/deployments/{self.deployment.pk}/upload-request/"
+
+    def _payload(self, filename="20240101T120000_0001.jpg", size=1024, **extra):
+        file = {"filename": filename, "size": size}
+        file.update(extra)
+        return {"files": [file]}
+
+    def test_permission_matrix(self):
+        matrix = [
+            ("superuser", self.superuser, status.HTTP_200_OK),
+            ("ProjectManager", self.pm_user, status.HTTP_200_OK),
+            ("MLDataManager", self.ml_user, status.HTTP_200_OK),
+            ("Researcher", self.researcher, status.HTTP_403_FORBIDDEN),
+            ("BasicMember", self.basic_user, status.HTTP_403_FORBIDDEN),
+            ("outsider", self.outsider, status.HTTP_403_FORBIDDEN),
+        ]
+        for role_name, user, expected in matrix:
+            with self.subTest(role=role_name):
+                self.client.force_authenticate(user)
+                response = self.client.post(self.url, self._payload(), format="json")
+                self.assertEqual(response.status_code, expected, f"{role_name} got {response.status_code}")
+
+    def test_anonymous_denied(self):
+        self.client.force_authenticate(None)
+        response = self.client.post(self.url, self._payload(), format="json")
+        self.assertIn(response.status_code, (status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN))
+
+    def test_no_data_source_returns_400(self):
+        self.client.force_authenticate(self.superuser)
+        url = f"/api/v2/deployments/{self.no_source.pk}/upload-request/"
+        response = self.client.post(url, self._payload(), format="json")
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+
+    def test_valid_file_mints_url(self):
+        self.client.force_authenticate(self.superuser)
+        response = self.client.post(self.url, self._payload(content_type="image/jpeg"), format="json")
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(response.data["errors"], [])
+        self.assertEqual(len(response.data["urls"]), 1)
+        entry = response.data["urls"][0]
+        self.assertEqual(entry["filename"], "20240101T120000_0001.jpg")
+        self.assertEqual(entry["key"], "uploads/station-1/20240101T120000_0001.jpg")
+        self.assertEqual(entry["method"], "PUT")
+        self.assertIn("X-Amz-Signature", entry["url"])
+        self.assertEqual(entry["headers"]["Content-Type"], "image/jpeg")
+        self.assertIn("expires_at", entry)
+
+    def test_deterministic_key(self):
+        self.client.force_authenticate(self.superuser)
+        r1 = self.client.post(self.url, self._payload(), format="json")
+        r2 = self.client.post(self.url, self._payload(), format="json")
+        self.assertEqual(r1.data["urls"][0]["key"], r2.data["urls"][0]["key"])
+
+    def test_unparseable_timestamp_rejected(self):
+        self.client.force_authenticate(self.superuser)
+        response = self.client.post(self.url, self._payload(filename="photo.jpg"), format="json")
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(response.data["urls"], [])
+        self.assertEqual(response.data["errors"][0]["code"], "unparseable_timestamp")
+
+    def test_invalid_extension_rejected(self):
+        self.client.force_authenticate(self.superuser)
+        response = self.client.post(self.url, self._payload(filename="20240101T120000_0001.txt"), format="json")
+        self.assertEqual(response.data["errors"][0]["code"], "invalid_extension")
+        self.assertEqual(response.data["urls"], [])
+
+    def test_path_traversal_rejected(self):
+        self.client.force_authenticate(self.superuser)
+        response = self.client.post(self.url, self._payload(filename="../20240101T120000_0001.jpg"), format="json")
+        self.assertEqual(response.data["errors"][0]["code"], "invalid_filename")
+
+    def test_zero_size_rejected(self):
+        self.client.force_authenticate(self.superuser)
+        response = self.client.post(self.url, self._payload(size=0), format="json")
+        self.assertEqual(response.data["errors"][0]["code"], "invalid_size")
+
+    def test_key_too_long_rejected(self):
+        """
+        The guard is on the object key, not the filename: the storage prefix and the
+        station's subdirectory are prepended before the key has to fit the 255
+        characters SourceImage.path allows. So the filename here is itself legal
+        (the request serializer caps filenames at 255) and only becomes too long
+        once "uploads/station-1/" is in front of it.
+        """
+        self.client.force_authenticate(self.superuser)
+        long_name = "20240101T120000_" + "a" * 230 + ".jpg"
+        self.assertLessEqual(len(long_name), 255)
+        response = self.client.post(self.url, self._payload(filename=long_name), format="json")
+        self.assertEqual(response.data["errors"][0]["code"], "key_too_long")
+        self.assertEqual(response.data["urls"], [])
+
+    def test_regex_mismatch_rejected(self):
+        self.deployment.data_source_regex = r"WILLNOTMATCH"
+        self.deployment.save()
+        self.client.force_authenticate(self.superuser)
+        response = self.client.post(self.url, self._payload(), format="json")
+        self.assertEqual(response.data["errors"][0]["code"], "regex_mismatch")
+        self.assertEqual(response.data["urls"], [])
+
+    def test_too_many_files_returns_400(self):
+        self.client.force_authenticate(self.superuser)
+        files = [{"filename": f"20240101T12000{i:04d}.jpg", "size": 10} for i in range(1001)]
+        response = self.client.post(self.url, {"files": files}, format="json")
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+
+    def test_invalid_subdir_returns_400(self):
+        self.client.force_authenticate(self.superuser)
+        payload = self._payload()
+        payload["subdir"] = "../escape"
+        response = self.client.post(self.url, payload, format="json")
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+
+
+class TestDeploymentUploadRequestE2E(TestCase):
+    """Flagship end-to-end test against live MinIO.
+
+    Mints a PUT URL via the endpoint, uploads bytes straight to storage with
+    ``requests.put``, runs ``sync_captures``, and asserts a SourceImage lands at
+    exactly the minted key. Requires the MinIO Docker stack.
+    """
+
+    def test_mint_put_upload_then_sync(self):
+        import requests
+
+        from ami.utils import s3
+
+        project, deployment = setup_test_project(reuse=False)
+        assert deployment.data_source is not None
+
+        superuser = User.objects.filter(is_superuser=True).first()
+        client = APIClient()
+        client.force_authenticate(superuser)
+
+        filename = "20240101T120000_0001.jpg"
+        client_subdir = f"deployment_{deployment.pk}"
+        url = f"/api/v2/deployments/{deployment.pk}/upload-request/"
+        response = client.post(
+            url,
+            {"subdir": client_subdir, "files": [{"filename": filename, "size": 4, "content_type": "image/jpeg"}]},
+            format="json",
+        )
+        self.assertEqual(response.status_code, status.HTTP_200_OK, response.data)
+        self.assertEqual(response.data["errors"], [])
+        entry = response.data["urls"][0]
+        minted_key = entry["key"]
+
+        # Upload the bytes directly to storage using the minted URL + headers.
+        put_resp = requests.put(entry["url"], data=b"test", headers=entry["headers"])
+        put_resp.raise_for_status()
+
+        # Sanity: the object is listed at exactly the minted key (config.prefix is
+        # already baked into minted_key, so compare against the full listed Key).
+        listed_keys = [obj["Key"] for obj, _ in s3.list_files_paginated(deployment.data_source.config) if obj]
+        self.assertIn(minted_key, listed_keys)
+
+        # Now ingest via the same path the deployment normally syncs through.
+        deployment.sync_captures()
+
+        self.assertTrue(
+            SourceImage.objects.filter(deployment=deployment, path=minted_key).exists(),
+            f"No SourceImage stored at minted key {minted_key}",
+        )
+
+
+class TestDeploymentAndProjectFilters(APITestCase):
+    """A5 filters: deployment research_site_id/device_id, project writable role, site/device project_id."""
+
+    def setUp(self):
+        super().setUp()
+        self.superuser = User.objects.create_superuser(email="super-filters@insectai.org", password="pw")
+        self.client.force_authenticate(self.superuser)
+
+        self.project = Project.objects.create(name="Filters Project")
+        create_roles_for_project(self.project)
+        self.site_a = Site.objects.create(name="Site A", project=self.project)
+        self.site_b = Site.objects.create(name="Site B", project=self.project)
+        self.device_a = Device.objects.create(name="Device A", project=self.project)
+        self.device_b = Device.objects.create(name="Device B", project=self.project)
+        self.dep_a = Deployment.objects.create(
+            name="Dep A", project=self.project, research_site=self.site_a, device=self.device_a
+        )
+        self.dep_b = Deployment.objects.create(
+            name="Dep B", project=self.project, research_site=self.site_b, device=self.device_b
+        )
+
+    def test_deployment_research_site_id_filter(self):
+        response = self.client.get(
+            f"/api/v2/deployments/?project_id={self.project.pk}&research_site_id={self.site_a.pk}"
+        )
+        self.assertEqual(response.status_code, 200)
+        ids = {row["id"] for row in response.data["results"]}
+        self.assertEqual(ids, {self.dep_a.pk})
+
+    def test_deployment_device_id_filter(self):
+        response = self.client.get(f"/api/v2/deployments/?project_id={self.project.pk}&device_id={self.device_b.pk}")
+        self.assertEqual(response.status_code, 200)
+        ids = {row["id"] for row in response.data["results"]}
+        self.assertEqual(ids, {self.dep_b.pk})
+
+    def test_site_project_id_scoping(self):
+        other = Project.objects.create(name="Other Project")
+        Site.objects.create(name="Other Site", project=other)
+        other_site = Site.objects.create(name="Other Site Two", project=other)
+        response = self.client.get(f"/api/v2/deployments/sites/?project_id={self.project.pk}")
+        self.assertEqual(response.status_code, 200)
+        ids = {row["id"] for row in response.data["results"]}
+        self.assertIn(self.site_a.pk, ids)
+        self.assertIn(self.site_b.pk, ids)
+        self.assertNotIn(other_site.pk, ids)
+
+    def test_device_project_id_scoping(self):
+        other = Project.objects.create(name="Other Project 2")
+        Device.objects.create(name="Other Device", project=other)
+        other_device = Device.objects.create(name="Other Device Two", project=other)
+        response = self.client.get(f"/api/v2/deployments/devices/?project_id={self.project.pk}")
+        self.assertEqual(response.status_code, 200)
+        ids = {row["id"] for row in response.data["results"]}
+        self.assertIn(self.device_a.pk, ids)
+        self.assertIn(self.device_b.pk, ids)
+        self.assertNotIn(other_device.pk, ids)
+
+
+class TestProjectWritableFilter(APITestCase):
+    """`?role=manager` / `?writable=true` returns only projects the user can write to."""
+
+    def setUp(self):
+        super().setUp()
+        self.manager = User.objects.create_user(email="mgr-writable@insectai.org", password="pw")
+
+        self.managed = Project.objects.create(name="Managed Project")
+        create_roles_for_project(self.managed)
+        ProjectManager.assign_user(self.manager, self.managed)
+
+        self.readonly = Project.objects.create(name="Readonly Project")
+        create_roles_for_project(self.readonly)
+        BasicMember.assign_user(self.manager, self.readonly)
+
+    def test_role_manager_returns_only_writable(self):
+        self.client.force_authenticate(self.manager)
+        response = self.client.get("/api/v2/projects/?role=manager")
+        self.assertEqual(response.status_code, 200)
+        ids = {row["id"] for row in response.data["results"]}
+        self.assertIn(self.managed.pk, ids)
+        self.assertNotIn(self.readonly.pk, ids)
+
+    def test_writable_true_returns_only_writable(self):
+        self.client.force_authenticate(self.manager)
+        response = self.client.get("/api/v2/projects/?writable=true")
+        ids = {row["id"] for row in response.data["results"]}
+        self.assertIn(self.managed.pk, ids)
+        self.assertNotIn(self.readonly.pk, ids)
+
+    def test_without_role_returns_all_visible(self):
+        self.client.force_authenticate(self.manager)
+        response = self.client.get("/api/v2/projects/")
+        ids = {row["id"] for row in response.data["results"]}
+        self.assertIn(self.managed.pk, ids)
+        self.assertIn(self.readonly.pk, ids)
+
+
 class TestSyncDeploymentBackfillMigration(APITestCase):
     """The 0095 backfill grants OBJECT-LEVEL sync_deployment to existing projects'
     MLDataManager groups, not just a global group permission.
diff --git a/ami/tests/test_storage.py b/ami/tests/test_storage.py
index 8e1e3e0a3..026af9e4d 100644
--- a/ami/tests/test_storage.py
+++ b/ami/tests/test_storage.py
@@ -191,6 +191,73 @@ def test_key_full_uri_without_protocol(self):
         expected = "/test_bucket/test_prefix/subdir/file.txt"
         self.assertEqual(result, expected)
 
+    def test_derive_upload_key_no_subdir(self):
+        result = s3.derive_upload_key(self.config, "20240101T120000_0001.jpg")
+        self.assertEqual(result, "test_prefix/20240101T120000_0001.jpg")
+
+    def test_derive_upload_key_with_subdir(self):
+        result = s3.derive_upload_key(self.config, "20240101T120000_0001.jpg", ["station-1"])
+        self.assertEqual(result, "test_prefix/station-1/20240101T120000_0001.jpg")
+
+    def test_derive_upload_key_combines_deployment_and_client_subdirs(self):
+        result = s3.derive_upload_key(self.config, "20240101T120000_0001.jpg", ["station-1", "2024-01-01"])
+        self.assertEqual(result, "test_prefix/station-1/2024-01-01/20240101T120000_0001.jpg")
+
+    def test_derive_upload_key_ignores_none_and_slash_only_subdirs(self):
+        # data_source_subdir is frequently None or "/" — those must not add empty
+        # path segments (which would break the key / diverge from what sync lists).
+        result = s3.derive_upload_key(self.config, "20240101T120000_0001.jpg", [None, "/", "client"])
+        self.assertEqual(result, "test_prefix/client/20240101T120000_0001.jpg")
+
+    def test_derive_upload_key_does_not_mangle_filename_matching_subdir(self):
+        # Regression guard vs key_with_prefix's split() dedup heuristic: a filename
+        # whose digits contain the subdir string must survive intact. See #1379.
+        result = s3.derive_upload_key(self.config, "20240101T120000_0001.jpg", ["2024"])
+        self.assertEqual(result, "test_prefix/2024/20240101T120000_0001.jpg")
+
+    def test_derive_upload_key_is_deterministic(self):
+        a = s3.derive_upload_key(self.config, "20240101T120000_0001.jpg", ["station-1"])
+        b = s3.derive_upload_key(self.config, "20240101T120000_0001.jpg", ["station-1"])
+        self.assertEqual(a, b)
+
+
+class TestPresignedPutUrl(TestCase):
+    def test_put_url_headers_and_checksum_gating(self):
+        # Real-AWS style config (endpoint_url is None): flexible checksum header
+        # is emitted. generate_presigned_url is a local signing operation, so this
+        # needs no network / MinIO.
+        aws_config = s3.S3Config(
+            endpoint_url=None,
+            access_key_id="AKIA_TEST",
+            secret_access_key="secret",
+            bucket_name="test-bucket",
+            prefix="uploads",
+            region="us-east-1",
+        )
+        url, headers = s3.get_presigned_put_url(
+            aws_config, "uploads/20240101T120000_0001.jpg", content_type="image/jpeg", checksum_sha256_b64="abc123=="
+        )
+        self.assertIn("X-Amz-Signature", url)
+        self.assertEqual(headers["Content-Type"], "image/jpeg")
+        self.assertEqual(headers["x-amz-checksum-sha256"], "abc123==")
+
+    def test_put_url_checksum_suppressed_for_custom_endpoint(self):
+        # MinIO/Swift style config (endpoint_url set): checksum header suppressed
+        # because Swift's s3api rejects x-amz-checksum-sha256. See #1379.
+        minio_config = s3.S3Config(
+            endpoint_url="http://minio:9000",
+            access_key_id="minioadmin",
+            secret_access_key="minioadmin",
+            bucket_name="test-bucket",
+            prefix="uploads",
+        )
+        url, headers = s3.get_presigned_put_url(
+            minio_config, "uploads/20240101T120000_0001.jpg", content_type="image/jpeg", checksum_sha256_b64="abc123=="
+        )
+        self.assertIn("X-Amz-Signature", url)
+        self.assertEqual(headers["Content-Type"], "image/jpeg")
+        self.assertNotIn("x-amz-checksum-sha256", headers)
+
 
 class TestStorageSource(TestCase):
     def setUp(self):
diff --git a/ami/utils/s3.py b/ami/utils/s3.py
index d34442b2e..8516ecc9b 100644
--- a/ami/utils/s3.py
+++ b/ami/utils/s3.py
@@ -662,6 +662,52 @@ def get_presigned_url(config: S3Config, key: str, expires_in: int = 60 * 60 * 24
     return str(url)
 
 
+def get_presigned_put_url(
+    config: S3Config,
+    key: str,
+    content_type: str | None = None,
+    checksum_sha256_b64: str | None = None,
+    expires_in: int = 60 * 60,
+) -> tuple[str, dict[str, str]]:
+    """Generate a presigned PUT URL for a direct-to-storage upload.
+
+    Returns ``(url, headers)`` where ``headers`` are the request headers the
+    client MUST send for the signature to validate. Unlike ``get_presigned_url``
+    (GET) this is deliberately NOT cached: each URL is per-file and short-lived.
+
+    ``checksum_sha256_b64`` is the base64-encoded raw SHA-256 digest (not hex).
+    The AWS flexible-checksum header (``x-amz-checksum-sha256``) is only emitted
+    against real AWS (``endpoint_url is None``); MinIO tolerates it but Swift's
+    s3api rejects it, so it is gated off for any custom endpoint. See #1379.
+    """
+    client = get_s3_client(config)
+    params: dict[str, typing.Any] = {"Bucket": config.bucket_name, "Key": key}
+    headers: dict[str, str] = {}
+    if content_type:
+        params["ContentType"] = content_type
+        headers["Content-Type"] = content_type
+    if checksum_sha256_b64 and config.endpoint_url is None:
+        params["ChecksumSHA256"] = checksum_sha256_b64
+        headers["x-amz-checksum-sha256"] = checksum_sha256_b64
+    url = client.generate_presigned_url("put_object", Params=params, ExpiresIn=expires_in)
+    return str(url), headers
+
+
+def derive_upload_key(config: S3Config, filename: str, subdirs: list[str | None] | None = None) -> str:
+    """Build the full object key for a direct upload.
+
+    Deliberately NOT ``key_with_prefix``: that helper has a
+    ``key.split(subdir, 1)[-1]`` dedup heuristic that mangles filenames
+    containing the subdir string (e.g. subdir ``2024`` + file ``20240101...jpg``).
+    Here we join the config prefix, the (cleaned) subdir parts, and the filename
+    directly so the result is exactly the object Key that ``sync_captures`` will
+    later store as ``SourceImage.path``. See #1379.
+    """
+    parts = [p.strip("/") for p in (subdirs or []) if p and p.strip("/")]
+    full_prefix = make_full_prefix(config, "/".join(parts) or None)
+    return pathlib.Path(full_prefix, filename).as_posix()
+
+
 # Methods to resize all images under a prefix
 def resize_images(config: S3Config, prefix: str, width: int, height: int):
     bucket = get_bucket(config)
diff --git a/docs/claude/INDEX.md b/docs/claude/INDEX.md
index 61f0980d8..7353f67b4 100644
--- a/docs/claude/INDEX.md
+++ b/docs/claude/INDEX.md
@@ -40,6 +40,7 @@ archived.
 | `planning/celery-queue-split-rollout.md` | Rollout plan for the Celery queue split (`feat/celery-queue-split`) |
 | `planning/2026-05-28-captures-processed-filter-design.md` | Design: captures "Processed / Not processed" filter |
 | `planning/2026-05-28-captures-processed-filter-plan.md` | Implementation plan (checkbox tasks) for the captures processed filter |
+| `planning/2026-07-23-mobile-upload-api.md` | Mobile upload API (#1379): presigned PUT `upload-request` endpoint (A1), deployment/project filters (A5), sync-completion docs (A2). Keywords: upload, presigned, S3, deployment, mobile |
 
 ## Archive / session snapshots
 
diff --git a/docs/claude/planning/2026-07-23-mobile-upload-api.md b/docs/claude/planning/2026-07-23-mobile-upload-api.md
new file mode 100644
index 000000000..bddf5cd7a
--- /dev/null
+++ b/docs/claude/planning/2026-07-23-mobile-upload-api.md
@@ -0,0 +1,72 @@
+# Mobile upload API (A1, A5, A2-docs) — #1379
+
+Branch: `feat/mobile-upload-api-a1-a5`. Migration-free. Deploy is the owner's step.
+
+## A1 — `POST /api/v2/deployments/{id}/upload-request/`
+
+Mints short-lived presigned PUT URLs so the mobile client uploads captures
+directly to the deployment's storage source, then calls `sync` to ingest them.
+
+- `ami/utils/s3.py`
+  - `get_presigned_put_url(config, key, content_type, checksum_sha256_b64, expires_in)`
+    → `(url, headers)`. Not cached (per-file, short-lived). The AWS flexible-checksum
+    header (`x-amz-checksum-sha256`) is emitted **only** when `endpoint_url is None`
+    (real AWS); suppressed for MinIO/Swift because Swift's s3api rejects it.
+    `checksum_sha256_b64` is base64 of the raw digest, not hex.
+  - `derive_upload_key(config, filename, subdirs)` — builds the full object key
+    directly (prefix + cleaned subdir parts + filename). Deliberately NOT
+    `key_with_prefix`, whose `split(subdir)` dedup heuristic mangles filenames
+    containing the subdir string. The result is exactly the Key `sync_captures`
+    stores as `SourceImage.path`, so re-request/re-upload is idempotent
+    (uniqueness on `(deployment, path)`).
+- `ami/main/models.py` — `Deployment.check_custom_permission` maps
+  `"upload_request"` → `SYNC_DEPLOYMENT`. **Load-bearing**: a `detail=True` action
+  runs `get_object()` → object permission check, which would otherwise probe the
+  nonexistent `upload_request_deployment` guardian perm and 403 every non-superuser.
+  Reuses an existing perm — no new perm, no migration.
+- `ami/main/api/serializers.py` — `UploadRequest*` request/response serializers.
+  `subdir` and per-file `sha256`/`content_type` optional. `MAX_UPLOAD_REQUEST_FILES = 1000`.
+- `ami/main/api/views.py` — `upload_request` `@action` on `DeploymentViewSet`.
+  Per-file validation → `errors[]` (no URL minted) for: unparseable timestamp,
+  non-image extension, path traversal / bad chars, key length > 255, regex mismatch,
+  size ≤ 0. 400 for no data source, invalid subdir, or > 1000 files.
+
+## A5 — filters
+
+- `DeploymentFilterSet` exposes `research_site_id` / `device_id` (the exact param
+  names the Swift client sends). Set as `filterset_class` on `DeploymentViewSet`.
+- `ProjectViewSet` gains `?role=manager` / `?writable=true` → filters to projects
+  the user can write via `get_objects_for_user(user, "update_project", Project)`
+  (covers ProjectManagers AND owners). Superusers see all; anonymous sees none.
+- Sites/devices `?project_id=` already worked (via `get_active_project`); added
+  confirming tests only.
+
+## A2 — sync completion (docs only)
+
+`@extend_schema` on the existing `sync` action documents the `{job_id, project_id}`
+response, the polling endpoint (`GET /api/v2/jobs/{id}/`), and the THREE terminal
+states: `SUCCESS`, `FAILURE`, `REVOKED`. The ≤1/min auto-sync cadence is a
+separate follow-up PR (not built here).
+
+## Findings
+
+- **HEIC not supported.** `IMAGE_FILE_EXTENSIONS` (`ami/utils/storages.py`) is
+  `jpg/jpeg/png/gif/webp/svg/bmp/ico/tiff/tif` — no `heic`/`heif`. If the mobile
+  client uploads HEIC, `upload-request` rejects it with `invalid_extension` (and
+  sync would drop it too). Add the extensions if HEIC uploads are required.
+- **MinIO checksum:** because the test/MinIO config sets `endpoint_url`, the
+  checksum header is never sent to MinIO — the E2E upload path does not exercise
+  the checksum branch. That branch only activates against real AWS.
+
+## Tests
+
+- `ami/tests/test_storage.py`: `derive_upload_key` unit tests (incl. None/`/`-only
+  subdirs and the filename-matching-subdir regression) and `get_presigned_put_url`
+  header/checksum-gating tests (all offline — local signing, no MinIO).
+- `ami/main/tests.py`:
+  - `TestDeploymentUploadRequest` — permission matrix, no-data-source 400,
+    each validation error code, deterministic key, >1000 files, invalid subdir.
+    Runs offline (fake AWS source; signing is local).
+  - `TestDeploymentUploadRequestE2E` — flagship: mint PUT → `requests.put` →
+    `sync_captures()` → assert `SourceImage.path == minted key`. **Needs MinIO.**
+  - `TestDeploymentAndProjectFilters`, `TestProjectWritableFilter` — A5 filters.