diff --git a/.serena/memories/backend/storage.md b/.serena/memories/backend/storage.md
index 101c77702a..6ab862a311 100644
--- a/.serena/memories/backend/storage.md
+++ b/.serena/memories/backend/storage.md
@@ -86,7 +86,8 @@ Since `put-object!` uses backend-specific operations (`impl/resolve-backend` + `
 | `file-thumbnail` | File grid thumbnails in `file_thumbnail.media_id`. | Yes | Authentication required | Reference scan. |
 | `profile` | User and team profile photos. References: `profile.photo_id` and `team.photo_id`. | Yes | Authentication required | Reference scan. |
 | `organization` | Organization logos uploaded by the Nitrate management API. | Yes | Public | No reference scan. A touched object is deleted. |
-| `tempfile` | Export files, chunked-upload chunks, and temporary font downloads. | No | Authentication required | No reference scan. A touched object uses a two-hour deletion delay. |
+| `tempfile` | Export files and temporary font downloads. | No | Authentication required | No reference scan. A touched object uses a two-hour deletion delay. |
+| `upload-session` | Chunked-upload chunks. References: `upload_session_chunk.object_id` and `upload_session_chunk.session_id` (both NO ACTION DEFERRABLE: restrict semantics, procedural deletion). | No | Authentication required | No reference scan. A touched object is deleted after the delay; `gc-deleted` removes mappings before rows. |
 | `file-data` | Encoded file data when `file-data-backend` is `storage`. Reference metadata has `storage-ref-id`, `file-id`, and the `file_data` row ID. | Yes | Authentication required | Reference scan. |
 | `file-data-fragment` | Compatibility value for file-data fragments. The current backend has no dedicated producer for this bucket. | No current write semantics | Public | No touched-object collector case. |
 | `file-change` | Compatibility value for file changes. Current snapshots store data in `file_data`, not this bucket. | No current write semantics | Authentication required | No touched-object collector case. |
@@ -95,7 +96,7 @@ Since `put-object!` uses backend-specific operations (`impl/resolve-backend` + `
 - `file-media-object` is the default bucket for old rows without bucket metadata.
 - Do not assign a new bucket without adding its access and cleanup behavior.
 - The touched-object collector raises an internal error for an unknown bucket.
-- It supports `file-media-object`, `team-font-variant`, `file-object-thumbnail`, `file-thumbnail`, `profile`, `file-data`, `tempfile`, and `organization`.
+- It supports `file-media-object`, `team-font-variant`, `file-object-thumbnail`, `file-thumbnail`, `profile`, `file-data`, `tempfile`, `upload-session`, and `organization`.
 - It does not support `file-data-fragment` or `file-change`.
 
 ## Access Rules
diff --git a/backend/src/app/main.clj b/backend/src/app/main.clj
index 04e52592a7..a9469f5eb9 100644
--- a/backend/src/app/main.clj
+++ b/backend/src/app/main.clj
@@ -395,7 +395,6 @@
      :offload-file-data  (ig/ref :app.tasks.offload-file-data/handler)
      :tasks-gc           (ig/ref :app.tasks.tasks-gc/handler)
      :telemetry          (ig/ref :app.tasks.telemetry/handler)
-     :upload-session-gc  (ig/ref :app.tasks.upload-session-gc/handler)
      :storage-gc-deleted (ig/ref ::sto.gc-deleted/handler)
      :storage-gc-touched (ig/ref ::sto.gc-touched/handler)
      :storage-pending-gc (ig/ref ::sto.pending-gc/handler)
@@ -434,9 +433,6 @@
    :app.tasks.tasks-gc/handler
    {::db/pool (ig/ref ::db/pool)}
 
-   :app.tasks.upload-session-gc/handler
-   {::db/pool (ig/ref ::db/pool)}
-
    :app.tasks.objects-gc/handler
    {::db/pool     (ig/ref ::db/pool)
     ::sto/storage (ig/ref ::sto/storage)}
@@ -569,9 +565,6 @@
      {:cron #penpot/cron "0 0 0 * * ?" ;; daily
       :task :tasks-gc}
 
-     {:cron #penpot/cron "0 0 0 * * ?" ;; daily
-      :task :upload-session-gc}
-
      {:cron #penpot/cron "0 0 2 * * ?" ;; daily
       :task :file-gc-scheduler}
 
diff --git a/backend/src/app/migrations.clj b/backend/src/app/migrations.clj
index ff65057bff..fad1f91168 100644
--- a/backend/src/app/migrations.clj
+++ b/backend/src/app/migrations.clj
@@ -502,7 +502,10 @@
     :fn (mg/resource "app/migrations/sql/0152-rename-version-and-add-indexes-to-server-error-report.sql")}
 
    {:name "0153-add-storage-object-status-and-deletion-attempts"
-    :fn (mg/resource "app/migrations/sql/0153-add-storage-object-status-and-deletion-attempts.sql")}])
+    :fn (mg/resource "app/migrations/sql/0153-add-storage-object-status-and-deletion-attempts.sql")}
+
+   {:name "0154-add-upload-session-chunk-table"
+    :fn (mg/resource "app/migrations/sql/0154-add-upload-session-chunk-table.sql")}])
 
 (defn apply-migrations!
   [pool name migrations]
diff --git a/backend/src/app/migrations/sql/0154-add-upload-session-chunk-table.sql b/backend/src/app/migrations/sql/0154-add-upload-session-chunk-table.sql
new file mode 100644
index 0000000000..4e7e8694d3
--- /dev/null
+++ b/backend/src/app/migrations/sql/0154-add-upload-session-chunk-table.sql
@@ -0,0 +1,61 @@
+--- Add the upload_session_chunk table, a deleted_at marker to upload_session,
+--- and make the upload_session.profile_id foreign key non-deleting.
+
+--- Each row maps one chunk of a chunked-upload session to the storage_object
+--- row that holds its bytes. Both foreign keys are ON DELETE NO ACTION
+--- DEFERRABLE on purpose: neither the session nor the storage object can be
+--- removed while a mapping row exists. Only objects-gc removes mappings
+--- (for consumed, stalled and profile-purge sessions), always before the
+--- session row, touching the chunk objects so storage GC reclaims them.
+--- NO ACTION is identical to RESTRICT in normal
+--- (immediate) operation; only the deferrability differs, which tooling
+--- such as the backend test fixture relies on
+--- (SET CONSTRAINTS ALL DEFERRED).
+---
+--- object_id is nullable: the mapping row is inserted first (reserving the
+--- slot under the UNIQUE(session_id, chunk_index) constraint inside a
+--- transaction that locks the session), and object_id is set once the blob
+--- has been written outside the transaction. A mapping with NULL object_id
+--- and no in-flight upload behind it means that upload died mid-flight; the
+--- client then starts a new session (sessions are ephemeral).
+
+CREATE TABLE upload_session_chunk (
+  id          uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+
+  created_at  timestamptz NOT NULL DEFAULT now(),
+
+  session_id  uuid NOT NULL REFERENCES upload_session(id) ON DELETE NO ACTION DEFERRABLE,
+  object_id   uuid NULL REFERENCES storage_object(id) ON DELETE NO ACTION DEFERRABLE,
+  chunk_index integer NOT NULL,
+
+  UNIQUE (session_id, chunk_index)
+);
+
+--- No standalone index on session_id: the UNIQUE(session_id, chunk_index)
+--- btree already serves session_id-only lookups and the session FK check
+--- via its leftmost column.
+
+CREATE INDEX upload_session_chunk__object_id__idx
+    ON upload_session_chunk(object_id);
+
+--- The deleted_at column marks a session as consumed: assemble-chunks sets it
+--- instead of deleting the row, and objects-gc physically purges consumed
+--- sessions (as well as stalled ones that were never assembled).
+
+ALTER TABLE upload_session
+  ADD COLUMN deleted_at timestamptz NULL DEFAULT NULL;
+
+CREATE INDEX upload_session__deleted_at_created_at__idx
+    ON upload_session(deleted_at, created_at);
+
+--- The profile foreign key moves from CASCADE to NO ACTION DEFERRABLE:
+--- sessions must be purged procedurally (objects-gc drains the sessions of
+--- profiles pending purge before the profile row is deleted), so a profile
+--- can no longer disappear with live chunk mappings behind it.
+
+ALTER TABLE upload_session
+  DROP CONSTRAINT upload_session_profile_id_fkey;
+
+ALTER TABLE upload_session
+  ADD CONSTRAINT upload_session_profile_id_fkey
+  FOREIGN KEY (profile_id) REFERENCES profile(id) ON DELETE NO ACTION DEFERRABLE;
diff --git a/backend/src/app/rpc/commands/media.clj b/backend/src/app/rpc/commands/media.clj
index dbf4a8cb9b..e728fedb29 100644
--- a/backend/src/app/rpc/commands/media.clj
+++ b/backend/src/app/rpc/commands/media.clj
@@ -339,7 +339,6 @@
 
 ;; --- Chunked Upload: Upload a single chunk
 
-(declare ^:private get-upload-chunk)
 (declare ^:private check-upload-chunk-slot)
 
 (def ^:private schema:upload-chunk
@@ -353,11 +352,27 @@
    [:session-id ::sm/uuid]
    [:index      ::sm/int]])
 
+(def ^:private sql:link-upload-session-chunk
+  "UPDATE upload_session_chunk
+      SET object_id = ?
+    WHERE session_id = ?
+      AND chunk_index = ?
+      AND object_id IS NULL")
+
+(defn- link-upload-session-chunk!
+  "Links a stored blob to its reserved (session, index) slot. Returns the
+  number of updated mappings (0 when the mapping vanished concurrently:
+  the session was consumed or purged after the reserve)."
+  [pool object-id session-id index]
+  (-> (db/exec-one! pool [sql:link-upload-session-chunk object-id session-id index])
+      (db/get-update-count)))
+
 (sv/defmethod ::upload-chunk
   {::doc/added "2.17"
    ::sm/params schema:upload-chunk
    ::sm/result schema:upload-chunk-result}
-  [cfg {:keys [::rpc/profile-id session-id index content]}]
+  [{:keys [::db/pool] :as cfg}
+   {:keys [::rpc/profile-id session-id index content] :as _params}]
   (let [session (db/tx-run! cfg check-upload-chunk-slot session-id profile-id index content)]
     (l/trc :hint "upload-chunk"
            :session-id session-id
@@ -365,23 +380,62 @@
            :size (:size content)
            :path (:path content))
 
+    ;; NOTE: the blob is written outside any transaction on purpose (see
+    ;; mem:backend/storage): a failed write must never mingle with the
+    ;; mapping transaction. If the write or the link below fails, the
+    ;; reserved mapping is removed and the error propagates, so the client
+    ;; retries the index in the same session. If the process dies between
+    ;; the reserve and the link, a NULL mapping is left behind and the
+    ;; client starts a new session (sessions are ephemeral).
     (let [storage (sto/resolve cfg)
-          data    (sto/content (:path content))]
-      (sto/put-object! storage
-                       {::sto/content      data
-                        ::sto/deduplicate? false
-                        ::sto/touch        true
-                        :content-type      (:mtype content)
-                        :bucket            sto/tempfile-bucket
-                        :upload-id         (str session-id)
-                        :chunk-index       index}))
+          data    (sto/content (:path content))
+          object  (try
+                    (sto/put-object! storage
+                                     {::sto/content      data
+                                      ::sto/deduplicate? false
+                                      ::sto/touched-at   (ct/in-future {:hours 1})
+                                      :content-type      (:mtype content)
+                                      :bucket            sto/upload-session-bucket})
+                    (catch Throwable cause
+                      (db/delete! pool :upload-session-chunk
+                                  {:session-id session-id :chunk-index index})
+                      (throw cause)))
+          linked  (try
+                    (link-upload-session-chunk! pool (:id object) session-id index)
+                    (catch Throwable cause
+                      ;; The blob was stored but the link failed: drop the
+                      ;; reservation so the client can retry the index in
+                      ;; the same session; the orphaned object stays
+                      ;; touched so touched-gc reclaims it.
+                      (db/delete! pool :upload-session-chunk
+                                  {:session-id session-id :chunk-index index})
+                      (throw cause)))]
+      (when (zero? linked)
+        ;; The mapping vanished concurrently (session consumed or purged
+        ;; after the reserve); the orphaned object stays touched so
+        ;; touched-gc reclaims it.
+        (ex/raise :type :not-found
+                  :code :object-not-found
+                  :hint "upload session no longer available"
+                  :session-id session-id))))
 
-    {:session-id session-id
-     :index      index}))
+  {:session-id session-id
+   :index      index})
 
 (defn- check-upload-chunk-slot
+  "Reserves the (session, index) slot: locks the session row, runs all
+  validations and inserts the mapping with a NULL object_id, all in one
+  transaction. Concurrent uploads of the same session serialize on the
+  session lock, so the UNIQUE(session_id, chunk_index) constraint can
+  never fire."
   [{:keys [::db/conn]} session-id profile-id index content]
   (let [session (db/get conn :upload-session {:id session-id :profile-id profile-id} {::db/for-update true})]
+    (when (:deleted-at session)
+      (ex/raise :type :not-found
+                :code :object-not-found
+                :hint "upload session already consumed"
+                :session-id session-id))
+
     (when (or (neg? index) (>= index (:total-chunks session)))
       (ex/raise :type :validation
                 :code :invalid-chunk-index
@@ -399,40 +453,39 @@
                 :size (:size content)
                 :max-size (cf/get :upload-max-chunk-size)))
 
-    (when (get-upload-chunk conn session-id index)
+    ;; NOTE: a mapping with NULL object_id also counts as occupied: either
+    ;; its upload is still in flight, or its process died mid-flight and
+    ;; the client must start a new session. Known failures (write or link)
+    ;; remove the reservation above, so they stay retryable in the same
+    ;; session.
+    (when (db/get* conn :upload-session-chunk {:session-id session-id :chunk-index index})
       (ex/raise :type :validation
-                :code :duplicate-chunk-index
-                :hint "chunk index already uploaded for this session"
+                :code :chunk-already-exists
+                :hint "chunk already uploaded for this session and index"
                 :session-id session-id
                 :index index))
 
+    (db/insert! conn :upload-session-chunk
+                {:session-id  session-id
+                 :object-id   nil
+                 :chunk-index index})
+
     session))
 
 ;; --- Chunked Upload: shared helpers
 
-(def ^:private sql:get-upload-chunks
-  "SELECT id, size, (metadata->>'~:chunk-index')::integer AS chunk_index
-     FROM storage_object
-    WHERE (metadata->>'~:upload-id') = ?::text
-      AND deleted_at IS NULL
-      AND status = 'valid'
-    ORDER BY (metadata->>'~:chunk-index')::integer ASC")
+(def ^:private sql:get-upload-session-chunks
+  "SELECT so.id, so.size
+     FROM upload_session_chunk AS usc
+     JOIN storage_object AS so ON (so.id = usc.object_id)
+    WHERE usc.session_id = ?
+      AND so.deleted_at IS NULL
+      AND so.status = 'valid'
+    ORDER BY usc.chunk_index ASC")
 
 (defn- get-upload-chunks
   [conn session-id]
-  (db/exec! conn [sql:get-upload-chunks (str session-id)]))
-
-(def ^:private sql:get-upload-chunk
-  "SELECT id
-     FROM storage_object
-    WHERE (metadata->>'~:upload-id') = ?::text
-      AND (metadata->>'~:chunk-index')::integer = ?
-      AND deleted_at IS NULL
-    LIMIT 1")
-
-(defn- get-upload-chunk
-  [conn session-id index]
-  (db/exec-one! conn [sql:get-upload-chunk (str session-id) index]))
+  (db/exec! conn [sql:get-upload-session-chunks session-id]))
 
 (defn- concat-chunks
   "Reads all chunk storage objects in order and writes them to a single
@@ -452,34 +505,47 @@
   conforming to `media.v/schema:upload` with `:filename`, `:path` and
   `:size`.
 
-  Raises a :validation/:missing-chunks error when the stored chunk
-  indices do not form exactly the `0..total-chunks` range recorded in
-  the session row (wrong count, gaps or duplicates).
-  Raises :not-found when the session does not belong to `profile-id`.
-  Deletes the session row from `upload_session` on success."
+  Raises a :validation/:missing-chunks error when the number of stored
+  chunks does not match `:total-chunks` recorded in the session row.
+  Raises :not-found when the session does not belong to `profile-id` or
+  was already consumed. Marks the session row as consumed (`deleted_at`);
+  the chunk mappings stay until the objects-gc task purges them (touching
+  the chunk objects so storage GC reclaims them), and the session row is
+  purged afterwards."
   [{:keys [::db/conn] :as cfg} profile-id session-id]
-  (let [session (db/get conn :upload-session {:id session-id :profile-id profile-id})
-        chunks  (get-upload-chunks conn session-id)
-        indices (sort (map :chunk-index chunks))]
+  (let [session (db/get conn :upload-session {:id session-id :profile-id profile-id})]
+    (when (:deleted-at session)
+      (ex/raise :type :not-found
+                :code :object-not-found
+                :hint "upload session already consumed"
+                :session-id session-id))
 
-    (when (or (not= (count chunks) (:total-chunks session))
-              (not= indices (range (:total-chunks session))))
-      (ex/raise :type :validation
-                :code :missing-chunks
-                :hint "stored chunks do not match expected total"
-                :session-id session-id
-                :expected   (:total-chunks session)
-                :found      (count chunks)))
+    (let [chunks (get-upload-chunks conn session-id)]
 
-    (let [storage (sto/resolve cfg ::db/reuse-conn true)
-          path    (concat-chunks storage chunks)
-          size    (reduce #(+ %1 (:size %2)) 0 chunks)]
+      (when (not= (count chunks) (:total-chunks session))
+        (ex/raise :type :validation
+                  :code :missing-chunks
+                  :hint "number of stored chunks does not match expected total"
+                  :session-id session-id
+                  :expected   (:total-chunks session)
+                  :found      (count chunks)))
 
-      (db/delete! conn :upload-session {:id session-id})
+      (let [storage (sto/resolve cfg ::db/reuse-conn true)
+            path    (concat-chunks storage chunks)
+            size    (reduce #(+ %1 (:size %2)) 0 chunks)]
 
-      {:filename "upload"
-       :path     path
-       :size     size})))
+        ;; NOTE: the session row is only marked (deleted_at) here; the
+        ;; chunk mappings stay until the objects-gc task removes them
+        ;; (before the session row, as the NO ACTION foreign keys
+        ;; require) while touching the chunk objects.
+        (db/update! conn :upload-session
+                    {:deleted-at (ct/now)}
+                    {:id session-id}
+                    {::db/return-keys false})
+
+        {:filename "upload"
+         :path     path
+         :size     size}))))
 
 ;; --- Chunked Upload: Assemble all chunks into a final media object
 
diff --git a/backend/src/app/rpc/quotes.clj b/backend/src/app/rpc/quotes.clj
index 119f2a4b26..1239ce6f71 100644
--- a/backend/src/app/rpc/quotes.clj
+++ b/backend/src/app/rpc/quotes.clj
@@ -535,7 +535,8 @@
 (def ^:private sql:get-upload-sessions-per-profile
   "SELECT count(*) AS total
      FROM upload_session
-    WHERE profile_id = ?")
+    WHERE profile_id = ?
+      AND deleted_at IS NULL")
 
 (defmethod check-quote ::upload-sessions-per-profile
   [{:keys [::profile-id ::target] :as quote}]
diff --git a/backend/src/app/storage.clj b/backend/src/app/storage.clj
index 0f35b0a54a..71dabca288 100644
--- a/backend/src/app/storage.clj
+++ b/backend/src/app/storage.clj
@@ -42,6 +42,10 @@
   "Bucket name for temporary file uploads (10-minute expiry)."
   "tempfile")
 
+(def upload-session-bucket
+  "Bucket name for chunked-upload chunks."
+  "upload-session")
+
 (def valid-buckets
   #{"file-media-object"
     "team-font-variant"
@@ -50,6 +54,7 @@
     "profile"
     "organization"
     tempfile-bucket
+    upload-session-bucket
     "file-data"
     "file-data-fragment"
     "file-change"})
@@ -211,7 +216,8 @@
     (if-some [hit (when (and (::deduplicate? params)
                              (:hash mdata)
                              (:bucket mdata)
-                             (not= tempfile-bucket (:bucket mdata)))
+                             (not= tempfile-bucket (:bucket mdata))
+                             (not= upload-session-bucket (:bucket mdata)))
                     (get-database-object-by-hash pool backend
                                                  (:bucket mdata)
                                                  (:hash mdata)))]
diff --git a/backend/src/app/storage/gc_deleted.clj b/backend/src/app/storage/gc_deleted.clj
index ab49030041..d9dc908e0a 100644
--- a/backend/src/app/storage/gc_deleted.clj
+++ b/backend/src/app/storage/gc_deleted.clj
@@ -57,6 +57,18 @@
     (-> (db/exec-one! conn [sql:delete-sobjects ids])
         (db/get-update-count))))
 
+(def ^:private sql:delete-upload-session-chunks
+  "DELETE FROM upload_session_chunk
+    WHERE object_id = ANY(?::uuid[])")
+
+(defn- delete-upload-session-chunks!
+  "Remove the chunk mappings for the given storage object ids. This must run
+  before the storage_object rows are deleted: the upload_session_chunk
+  foreign keys are ON DELETE NO ACTION."
+  [conn ids]
+  (let [ids (db/create-array conn "uuid" ids)]
+    (db/exec-one! conn [sql:delete-upload-session-chunks ids])))
+
 (def ^:private sql:increment-attempts-and-defer
   "UPDATE storage_object
       SET deletion_attempts = deletion_attempts + 1,
@@ -105,10 +117,21 @@
                :backend (name backend-id)))
 
       (when (seq ok-ids)
+        ;; NOTE: the chunk mappings must be removed before the
+        ;; storage_object rows (NO ACTION foreign keys). It only affects
+        ;; objects of the upload-session bucket; for any other bucket the
+        ;; delete matches no rows.
+        (delete-upload-session-chunks! conn ok-ids)
         (delete-sobjects! conn ok-ids))
 
       (when (seq fail-ids)
         (increment-attempts-and-defer! conn fail-ids)
+        ;; NOTE: same NO ACTION ordering as above: the give-up DELETE below
+        ;; removes storage_object rows, so chunk mappings must go first.
+        ;; Deferred objects keep their rows; only the mapping of a
+        ;; permanently given-up object disappears early, and that object is
+        ;; already deleted-marked.
+        (delete-upload-session-chunks! conn fail-ids)
         (let [given-up (delete-give-up! conn fail-ids)]
           (when (pos? (db/get-update-count given-up))
             (l/wrn :hint "giving up on orphan blob after max attempts"
diff --git a/backend/src/app/storage/gc_touched.clj b/backend/src/app/storage/gc_touched.clj
index af16af5f82..260739b02f 100644
--- a/backend/src/app/storage/gc_touched.clj
+++ b/backend/src/app/storage/gc_touched.clj
@@ -155,14 +155,15 @@
 (defn- process-bucket!
   [conn bucket objects]
   (cond
-    (= bucket "file-media-object")     (process-objects! conn has-file-media-object-refs? bucket objects)
-    (= bucket "team-font-variant")     (process-objects! conn has-team-font-variant-refs? bucket objects)
-    (= bucket "file-object-thumbnail") (process-objects! conn has-file-object-thumbnails-refs? bucket objects)
-    (= bucket "file-thumbnail")        (process-objects! conn has-file-thumbnails-refs? bucket objects)
-    (= bucket "profile")               (process-objects! conn has-profile-refs? bucket objects)
-    (= bucket "file-data")             (process-objects! conn has-file-data-refs? bucket objects)
-    (= bucket sto/tempfile-bucket)     (process-objects! conn (constantly false) sto/tempfile-bucket objects)
-    (= bucket "organization")          (process-objects! conn (constantly false) bucket objects)
+    (= bucket "file-media-object")       (process-objects! conn has-file-media-object-refs? bucket objects)
+    (= bucket "team-font-variant")       (process-objects! conn has-team-font-variant-refs? bucket objects)
+    (= bucket "file-object-thumbnail")   (process-objects! conn has-file-object-thumbnails-refs? bucket objects)
+    (= bucket "file-thumbnail")          (process-objects! conn has-file-thumbnails-refs? bucket objects)
+    (= bucket "profile")                 (process-objects! conn has-profile-refs? bucket objects)
+    (= bucket "file-data")               (process-objects! conn has-file-data-refs? bucket objects)
+    (= bucket sto/tempfile-bucket)       (process-objects! conn (constantly false) sto/tempfile-bucket objects)
+    (= bucket sto/upload-session-bucket) (process-objects! conn (constantly false) sto/upload-session-bucket objects)
+    (= bucket "organization")            (process-objects! conn (constantly false) bucket objects)
     :else
     (ex/raise :type :internal
               :code :unexpected-unknown-reference
diff --git a/backend/src/app/tasks/objects_gc.clj b/backend/src/app/tasks/objects_gc.clj
index 399bc50beb..daf5619ae6 100644
--- a/backend/src/app/tasks/objects_gc.clj
+++ b/backend/src/app/tasks/objects_gc.clj
@@ -16,6 +16,51 @@
    [app.tasks.delete-object :as dobj]
    [integrant.core :as ig]))
 
+(def ^:private sql:get-upload-sessions
+  "SELECT us.id
+     FROM upload_session AS us
+    WHERE (us.deleted_at IS NOT NULL
+           AND us.deleted_at <= ?)
+       OR (us.deleted_at IS NULL
+           AND us.created_at <= ?)
+       OR EXISTS (SELECT 1
+                    FROM profile AS p
+                   WHERE p.id = us.profile_id
+                     AND p.deleted_at IS NOT NULL
+                     AND p.deleted_at <= ?)
+    ORDER BY us.created_at ASC
+    LIMIT ?
+      FOR UPDATE OF us
+     SKIP LOCKED")
+
+(def ^:private sql:delete-session-chunks
+  "DELETE FROM upload_session_chunk
+    WHERE session_id = ?
+    RETURNING object_id")
+
+(defn- delete-upload-sessions!
+  "Purges consumed upload sessions (marked by assemble-chunks), stalled
+  sessions (never assembled within max-age) and sessions owned by profiles
+  pending purge. Referenced storage objects are touched so the storage GC
+  reclaims them with its usual delay; chunk mappings are removed before the
+  session row (NO ACTION foreign keys)."
+  [{:keys [::db/conn ::timestamp ::chunk-size ::sto/storage] :as cfg}]
+  (let [stalled-threshold (ct/minus timestamp {:hours 1})]
+    (->> (db/plan conn [sql:get-upload-sessions timestamp stalled-threshold timestamp chunk-size]
+                  {:fetch-size 5})
+         (reduce (fn [total {:keys [id]}]
+                   (l/trc :obj "upload-session" :id (str id))
+
+                   ;; Remove the chunk mappings, marking as touched all
+                   ;; related storage objects in a single round-trip.
+                   (doseq [{:keys [object-id]} (db/exec! conn [sql:delete-session-chunks id])]
+                     (some->> object-id (sto/touch-object! storage)))
+
+                   (let [affected (-> (db/delete! conn :upload-session {:id id})
+                                      (db/get-update-count))]
+                     (+ total affected)))
+                 0))))
+
 (def ^:private sql:get-profiles
   "SELECT id, photo_id FROM profile
     WHERE deleted_at IS NOT NULL
@@ -292,7 +337,11 @@
                0)))
 
 (def ^:private deletion-proc-vars
-  [#'delete-profiles!
+  ;; NOTE: upload sessions go first: deleting a profile cascades to its
+  ;; sessions, which would hit the upload_session_chunk NO ACTION foreign key
+  ;; while mappings still exist.
+  [#'delete-upload-sessions!
+   #'delete-profiles!
    #'delete-file-media-objects!
    #'delete-file-object-thumbnails!
    #'delete-file-thumbnails!
diff --git a/backend/src/app/tasks/upload_session_gc.clj b/backend/src/app/tasks/upload_session_gc.clj
deleted file mode 100644
index b5a6a1c078..0000000000
--- a/backend/src/app/tasks/upload_session_gc.clj
+++ /dev/null
@@ -1,41 +0,0 @@
-;; This Source Code Form is subject to the terms of the Mozilla Public
-;; License, v. 2.0. If a copy of the MPL was not distributed with this
-;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
-;;
-;; Copyright (c) KALEIDOS SUBSIDIARY SL
-
-(ns app.tasks.upload-session-gc
-  "A maintenance task that deletes stalled (incomplete) upload sessions.
-
-  An upload session is considered stalled when it was created more than
-  `max-age` ago without being completed (i.e. the session row still
-  exists because `assemble-chunks` was never called to clean it up).
-  The default max-age is 1 hour."
-  (:require
-   [app.common.logging :as l]
-   [app.common.time :as ct]
-   [app.db :as db]
-   [integrant.core :as ig]))
-
-(def ^:private sql:delete-stalled-sessions
-  "DELETE FROM upload_session
-    WHERE created_at < ?::timestamptz")
-
-(defmethod ig/assert-key ::handler
-  [_ params]
-  (assert (db/pool? (::db/pool params)) "expected a valid database pool"))
-
-(defmethod ig/expand-key ::handler
-  [k v]
-  {k (merge {::max-age (ct/duration {:hours 1})} v)})
-
-(defmethod ig/init-key ::handler
-  [_ {:keys [::max-age] :as cfg}]
-  (fn [_]
-    (db/tx-run! cfg
-                (fn [{:keys [::db/conn]}]
-                  (let [threshold (ct/minus (ct/now) max-age)
-                        result    (-> (db/exec-one! conn [sql:delete-stalled-sessions threshold])
-                                      (db/get-update-count))]
-                    (l/debug :hint "task finished" :deleted result)
-                    {:deleted result})))))
diff --git a/backend/test/backend_tests/rpc_font_test.clj b/backend/test/backend_tests/rpc_font_test.clj
index eb9f6cc69d..90dabf62a0 100644
--- a/backend/test/backend_tests/rpc_font_test.clj
+++ b/backend/test/backend_tests/rpc_font_test.clj
@@ -158,12 +158,15 @@
 
     (binding [ct/*clock* (ct/fixed-clock (ct/in-future {:days 8}))]
       (let [res (th/run-task! :objects-gc {})]
-        (t/is (= 2 (:processed res)))))
+        ;; processed = 4: the 2 font variants plus the 2 consumed upload sessions
+        (t/is (= 4 (:processed res)))))
 
     (binding [ct/*clock* (ct/fixed-clock (ct/in-future {:days 8 :hours 3}))]
       (let [res (th/run-task! :storage-gc-touched {})]
         (t/is (= 0 (:freeze res)))
-        (t/is (= 6 (:delete res)))))))
+        ;; deleted = 8: the 6 font objects plus the 2 chunk objects touched
+        ;; by objects-gc when purging the consumed sessions
+        (t/is (= 8 (:delete res)))))))
 
 (t/deftest font-deletion-2
   (let [prof    (th/create-profile* 1 {:is-active true})
@@ -224,12 +227,15 @@
 
     (binding [ct/*clock* (ct/fixed-clock (ct/in-future {:days 8}))]
       (let [res (th/run-task! :objects-gc {})]
-        (t/is (= 1 (:processed res)))))
+        ;; processed = 3: the font plus the 2 consumed upload sessions
+        (t/is (= 3 (:processed res)))))
 
     (binding [ct/*clock* (ct/fixed-clock (ct/in-future {:days 8 :hours 3}))]
       (let [res (th/run-task! :storage-gc-touched {})]
         (t/is (= 0 (:freeze res)))
-        (t/is (= 3 (:delete res)))))))
+        ;; deleted = 5: the 3 font objects plus the 2 chunk objects touched
+        ;; by objects-gc when purging the consumed sessions
+        (t/is (= 5 (:delete res)))))))
 
 (t/deftest font-deletion-3
   (let [prof    (th/create-profile* 1 {:is-active true})
@@ -271,12 +277,15 @@
     ;; objects-gc at days 8, then storage-gc-touched at days 8 + 3h
     (binding [ct/*clock* (ct/fixed-clock (ct/in-future {:days 8}))]
       (let [res (th/run-task! :objects-gc {})]
-        (t/is (= 1 (:processed res)))))
+        ;; processed = 3: the font variant plus the 2 consumed upload sessions
+        (t/is (= 3 (:processed res)))))
 
     (binding [ct/*clock* (ct/fixed-clock (ct/in-future {:days 8 :hours 3}))]
       (let [res (th/run-task! :storage-gc-touched {})]
         (t/is (= 0 (:freeze res)))
-        (t/is (= 3 (:delete res)))))))
+        ;; deleted = 5: the 3 font objects plus the 2 chunk objects touched
+        ;; by objects-gc when purging the consumed sessions
+        (t/is (= 5 (:delete res)))))))
 
 (t/deftest input-sanitization-1
   (with-mocks [mock {:target 'app.rpc.quotes/check! :return nil}]
diff --git a/backend/test/backend_tests/rpc_media_test.clj b/backend/test/backend_tests/rpc_media_test.clj
index 5a416b3ba4..8a735639c9 100644
--- a/backend/test/backend_tests/rpc_media_test.clj
+++ b/backend/test/backend_tests/rpc_media_test.clj
@@ -7,6 +7,7 @@
 (ns backend-tests.rpc-media-test
   (:require
    [app.common.uuid :as uuid]
+   [app.db :as db]
    [app.http.client :as http]
    [app.media :as media]
    [app.rpc :as-alias rpc]
@@ -532,7 +533,7 @@
                   :index           0
                   :content         mfile})
 
-    ;; First assemble succeeds; session row is deleted afterwards
+    ;; First assemble succeeds; session row is marked as consumed afterwards
     (let [out1 (th/command! {::th/type        :assemble-file-media-object
                              ::rpc/profile-id (:id prof)
                              :session-id      session-id
@@ -545,7 +546,7 @@
       (t/is (= media-id (:id (:result out1)))))
 
     ;; Second assemble with the same session-id must fail because the
-    ;; session row has been deleted after the first assembly
+    ;; session row has been marked as consumed after the first assembly
     (let [out2 (th/command! {::th/type        :assemble-file-media-object
                              ::rpc/profile-id (:id prof)
                              :session-id      session-id
@@ -681,45 +682,6 @@
       (t/is (= :validation (-> out :error ex-data :type)))
       (t/is (= :missing-chunks (-> out :error ex-data :code))))))
 
-(t/deftest chunked-upload-assemble-rejects-duplicate-indices
-  ;; assemble-chunks must validate the index SET, not just the count: a
-  ;; session declaring 2 chunks but storing [0,0] must fail instead of
-  ;; assembling a corrupt file. Chunks are written at the storage level
-  ;; because upload-chunk itself now rejects the second index.
-  (let [prof        (th/create-profile* 1)
-        _           (th/create-project* 1 {:profile-id (:id prof)
-                                           :team-id (:default-team-id prof)})
-        file        (th/create-file* 1 {:profile-id (:id prof)
-                                        :project-id (:default-project-id prof)
-                                        :is-shared false})
-        session-id  (create-session! prof 2)
-        storage     (:app.storage/storage th/*system*)
-        source-path (th/tempfile "backend_tests/test_files/sample.jpg")
-        chunks      (split-file-into-chunks source-path 312043)
-        put-chunk!  (fn [idx]
-                      (let [mfile (make-chunk-mfile (first chunks) "image/jpeg")]
-                        (sto/put-object! storage
-                                         {::sto/content      (sto/content (:path mfile))
-                                          ::sto/deduplicate? false
-                                          ::sto/touch        true
-                                          :content-type      "image/jpeg"
-                                          :bucket            sto/tempfile-bucket
-                                          :upload-id         (str session-id)
-                                          :chunk-index       idx})))]
-    (put-chunk! 0)
-    (put-chunk! 0)
-
-    (let [out (th/command! {::th/type        :assemble-file-media-object
-                            ::rpc/profile-id (:id prof)
-                            :session-id      session-id
-                            :file-id         (:id file)
-                            :is-local        true
-                            :name            "dupe-indices"
-                            :mtype           "image/jpeg"})]
-      (t/is (some? (:error out)))
-      (t/is (= :validation (-> out :error ex-data :type)))
-      (t/is (= :missing-chunks (-> out :error ex-data :code))))))
-
 (t/deftest chunked-upload-duplicate-then-assemble
   ;; A rejected duplicate must leave the first chunk intact: upload 0,
   ;; re-upload 0 (rejected), then assemble succeeds with the original size.
@@ -748,7 +710,8 @@
                             :index           0
                             :content         (make-chunk-mfile (first chunks) mtype)})]
       (t/is (some? (:error out)))
-      (t/is (= :duplicate-chunk-index (-> out :error ex-data :code))))
+      (t/is (= :validation (-> out :error ex-data :type)))
+      (t/is (= :chunk-already-exists (-> out :error ex-data :code))))
 
     (let [out (th/command! {::th/type        :assemble-file-media-object
                             ::rpc/profile-id (:id prof)
@@ -791,7 +754,8 @@
                             :index           0
                             :content         (make-chunk-mfile (nth chunks 0) mtype)})]
       (t/is (some? (:error out)))
-      (t/is (= :duplicate-chunk-index (-> out :error ex-data :code))))
+      (t/is (= :validation (-> out :error ex-data :type)))
+      (t/is (= :chunk-already-exists (-> out :error ex-data :code))))
 
     (let [out (th/command! {::th/type        :upload-chunk
                             ::rpc/profile-id (:id prof)
@@ -800,11 +764,11 @@
                             :content         (make-chunk-mfile (nth chunks 1) mtype)})]
       (t/is (nil? (:error out))))
 
-    ;; The live store holds exactly the two distinct indices: the
+    ;; The mapping table holds exactly the two distinct indices: the
     ;; rejected duplicate stored nothing.
-    (let [rows (th/db-exec! ["SELECT (metadata->>'~:chunk-index')::integer AS idx FROM storage_object WHERE (metadata->>'~:upload-id') = ?::text AND deleted_at IS NULL ORDER BY idx"
-                             (str session-id)])]
-      (t/is (= [0 1] (mapv :idx rows))))))
+    (let [rows (th/db-exec! ["SELECT chunk_index FROM upload_session_chunk WHERE session_id = ? ORDER BY chunk_index"
+                             session-id])]
+      (t/is (= [0 1] (mapv :chunk-index rows))))))
 
 (t/deftest chunked-upload-session-not-found
   (let [prof       (th/create-profile* 1)
@@ -843,6 +807,48 @@
       (t/is (= :max-quote-reached (-> out :error ex-data :code)))
       (t/is (= "upload-chunks-per-session" (-> out :error ex-data :target))))))
 
+(t/deftest chunked-upload-consumed-session-frees-quota
+  ;; Consumed sessions must not count against the sessions-per-profile
+  ;; quota: with the limit set to 1, assembling a session frees the slot
+  ;; for a new one.
+  (with-mocks [mock {:target 'app.config/get
+                     :return (th/config-get-mock
+                              {:quotes-upload-sessions-per-profile 1})}]
+    (let [prof        (th/create-profile* 1)
+          _           (th/create-project* 1 {:profile-id (:id prof)
+                                             :team-id (:default-team-id prof)})
+          file        (th/create-file* 1 {:profile-id (:id prof)
+                                          :project-id (:default-project-id prof)
+                                          :is-shared false})
+          source-path (th/tempfile "backend_tests/test_files/sample.jpg")
+          mfile       {:filename "sample.jpg"
+                       :path     source-path
+                       :mtype    "image/jpeg"
+                       :size     312043}
+          session-id  (create-session! prof 1)
+          upload-out  (th/command! {::th/type        :upload-chunk
+                                    ::rpc/profile-id (:id prof)
+                                    :session-id      session-id
+                                    :index           0
+                                    :content         mfile})]
+      (t/is (nil? (:error upload-out)))
+
+      (let [assemble-out (th/command! {::th/type        :assemble-file-media-object
+                                       ::rpc/profile-id (:id prof)
+                                       :session-id      session-id
+                                       :file-id         (:id file)
+                                       :is-local        true
+                                       :name            "assembled-image"
+                                       :mtype           "image/jpeg"})]
+        (t/is (nil? (:error assemble-out))))
+
+      ;; the consumed session frees the quota slot
+      (let [out (th/command! {::th/type        :create-upload-session
+                              ::rpc/profile-id (:id prof)
+                              :total-chunks    1})]
+        (t/is (nil? (:error out)))
+        (t/is (uuid? (:session-id (:result out))))))))
+
 (t/deftest chunked-upload-invalid-total-chunks
   ;; total-chunks must be at least 1; zero and negative values are rejected
   ;; with a :validation error.
@@ -892,41 +898,6 @@
       (t/is (= :validation (-> out :error ex-data :type)))
       (t/is (= :invalid-chunk-index (-> out :error ex-data :code))))))
 
-(t/deftest chunked-upload-duplicate-index-rejected
-  ;; Uploading the same chunk index twice into one session must fail:
-  ;; the second call raises :validation / :duplicate-chunk-index and
-  ;; stores nothing, so one session+index keeps at most one object.
-  (let [prof        (th/create-profile* 1)
-        session-id  (create-session! prof 1)
-        source-path (th/tempfile "backend_tests/test_files/sample.jpg")
-        chunks      (split-file-into-chunks source-path 312043)
-        mtype       "image/jpeg"
-        mfile1      (make-chunk-mfile (first chunks) mtype)
-        mfile2      (make-chunk-mfile (first chunks) mtype)]
-
-    ;; First upload succeeds
-    (let [out (th/command! {::th/type        :upload-chunk
-                            ::rpc/profile-id (:id prof)
-                            :session-id      session-id
-                            :index           0
-                            :content         mfile1})]
-      (t/is (nil? (:error out))))
-
-    ;; Second upload of the same index must be rejected
-    (let [out (th/command! {::th/type        :upload-chunk
-                            ::rpc/profile-id (:id prof)
-                            :session-id      session-id
-                            :index           0
-                            :content         mfile2})]
-      (t/is (some? (:error out)))
-      (t/is (= :validation (-> out :error ex-data :type)))
-      (t/is (= :duplicate-chunk-index (-> out :error ex-data :code))))
-
-    ;; Exactly one live object stored for that session/index
-    (let [rows (th/db-exec! ["SELECT id FROM storage_object WHERE (metadata->>'~:upload-id') = ?::text AND (metadata->>'~:chunk-index') = '0' AND deleted_at IS NULL"
-                             (str session-id)])]
-      (t/is (= 1 (count rows))))))
-
 (t/deftest chunked-upload-chunk-too-large
   ;; Chunks larger than the configured cap must be rejected with
   ;; :validation / :chunk-too-large before anything is stored, while a
@@ -951,9 +922,8 @@
         (t/is (= :chunk-too-large (-> out :error ex-data :code))))
 
       ;; Nothing stored for the rejected chunk
-      (let [rows (th/db-exec! ["SELECT id FROM storage_object WHERE (metadata->>'~:upload-id') = ?::text AND deleted_at IS NULL"
-                               (str session-id)])]
-        (t/is (= 0 (count rows))))
+      (t/is (= 0 (:count (th/db-exec-one! ["SELECT count(*) FROM upload_session_chunk WHERE session_id = ?"
+                                           session-id]))))
 
       ;; A chunk exactly at the cap still uploads fine
       (let [out (th/command! {::th/type        :upload-chunk
@@ -984,6 +954,220 @@
         (t/is (= :restriction (-> out :error ex-data :type)))
         (t/is (= :max-quote-reached (-> out :error ex-data :code)))))))
 
+;; --- upload_session_chunk mapping tests ---
+
+(t/deftest chunked-upload-creates-chunk-mapping
+  ;; Uploading a chunk creates a row in upload_session_chunk pointing to the
+  ;; storage object, and the object itself carries no session metadata.
+  (let [prof        (th/create-profile* 1)
+        session-id  (create-session! prof 1)
+        source-path (th/tempfile "backend_tests/test_files/sample.jpg")
+        mfile       {:filename "sample.jpg"
+                     :path     source-path
+                     :mtype    "image/jpeg"
+                     :size     312043}
+        out         (th/command! {::th/type        :upload-chunk
+                                  ::rpc/profile-id (:id prof)
+                                  :session-id      session-id
+                                  :index           0
+                                  :content         mfile})]
+    (t/is (nil? (:error out)))
+
+    (let [row (th/db-exec-one! ["select session_id, object_id, chunk_index from upload_session_chunk where session_id = ?"
+                                session-id])]
+      (t/is (= session-id (:session-id row)))
+      (t/is (= 0 (:chunk-index row)))
+
+      (let [storage (:app.storage/storage th/*system*)
+            obj     (sto/get-object storage (:object-id row))]
+        (t/is (sto/object? obj))
+        (t/is (= "upload-session" (-> obj meta :bucket)))
+        (t/is (nil? (-> obj meta :upload-id)))
+        (t/is (nil? (-> obj meta :chunk-index)))))))
+
+(t/deftest chunked-upload-duplicate-index-fails
+  ;; Re-uploading an already stored index fails with
+  ;; :validation/:chunk-already-exists and creates no new storage object.
+  (let [prof        (th/create-profile* 1)
+        session-id  (create-session! prof 1)
+        source-path (th/tempfile "backend_tests/test_files/sample.jpg")
+        mfile       {:filename "sample.jpg"
+                     :path     source-path
+                     :mtype    "image/jpeg"
+                     :size     312043}
+        out1        (th/command! {::th/type        :upload-chunk
+                                  ::rpc/profile-id (:id prof)
+                                  :session-id      session-id
+                                  :index           0
+                                  :content         mfile})]
+    (t/is (nil? (:error out1)))
+
+    (let [before (:count (th/db-exec-one! ["select count(*) from storage_object"]))
+          out2   (th/command! {::th/type        :upload-chunk
+                               ::rpc/profile-id (:id prof)
+                               :session-id      session-id
+                               :index           0
+                               :content         mfile})]
+      (t/is (some? (:error out2)))
+      (t/is (= :validation (-> out2 :error ex-data :type)))
+      (t/is (= :chunk-already-exists (-> out2 :error ex-data :code)))
+      (t/is (= before (:count (th/db-exec-one! ["select count(*) from storage_object"])))))))
+
+(t/deftest chunked-upload-null-reservation-blocks-retry
+  ;; A reserved slot with NULL object_id (an upload that died between the
+  ;; reserve and the link) counts as occupied: retrying the index in the
+  ;; same session fails with :validation/:chunk-already-exists and stores
+  ;; nothing, so the client must start a new session.
+  (let [prof        (th/create-profile* 1)
+        session-id  (create-session! prof 1)
+        source-path (th/tempfile "backend_tests/test_files/sample.jpg")
+        mfile       {:filename "sample.jpg"
+                     :path     source-path
+                     :mtype    "image/jpeg"
+                     :size     312043}]
+    (th/db-exec! ["insert into upload_session_chunk (session_id, chunk_index, object_id) values (?, ?, null)"
+                  session-id 0])
+    (let [before (:count (th/db-exec-one! ["select count(*) from storage_object"]))
+          out    (th/command! {::th/type        :upload-chunk
+                               ::rpc/profile-id (:id prof)
+                               :session-id      session-id
+                               :index           0
+                               :content         mfile})]
+      (t/is (some? (:error out)))
+      (t/is (= :validation (-> out :error ex-data :type)))
+      (t/is (= :chunk-already-exists (-> out :error ex-data :code)))
+      (t/is (= before (:count (th/db-exec-one! ["select count(*) from storage_object"])))))))
+
+(t/deftest chunked-upload-link-failure-releases-slot
+  ;; When the link UPDATE fails after a successful blob write, the
+  ;; reservation is removed so the client can retry the index in the same
+  ;; session; the orphaned blob stays touched for touched-gc.
+  (let [prof        (th/create-profile* 1)
+        session-id  (create-session! prof 1)
+        source-path (th/tempfile "backend_tests/test_files/sample.jpg")
+        mfile       {:filename "sample.jpg"
+                     :path     source-path
+                     :mtype    "image/jpeg"
+                     :size     312043}
+        orig        @#'app.rpc.commands.media/link-upload-session-chunk!
+        failed?     (atom false)]
+    (with-mocks [mock {:target 'app.rpc.commands.media/link-upload-session-chunk!
+                       :return (fn [pool object-id session-id index]
+                                 (if (compare-and-set! failed? false true)
+                                   (throw (ex-info "link boom" {}))
+                                   (orig pool object-id session-id index)))}]
+      (let [out (th/command! {::th/type        :upload-chunk
+                              ::rpc/profile-id (:id prof)
+                              :session-id      session-id
+                              :index           0
+                              :content         mfile})]
+        (t/is (some? (:error out))))
+      ;; the failed link left no reservation behind
+      (t/is (= 0 (:count (th/db-exec-one! ["select count(*) from upload_session_chunk where session_id = ?"
+                                           session-id]))))
+      ;; retrying the same index in the same session succeeds
+      (let [out (th/command! {::th/type        :upload-chunk
+                              ::rpc/profile-id (:id prof)
+                              :session-id      session-id
+                              :index           0
+                              :content         mfile})]
+        (t/is (nil? (:error out)))
+        (t/is (= 1 (:count (th/db-exec-one! ["select count(*) from upload_session_chunk where session_id = ?"
+                                             session-id]))))))))
+
+(t/deftest chunked-upload-to-consumed-session-fails
+  ;; Once assembled, the session is consumed: uploading another chunk fails
+  ;; with :not-found and the session row stays, marked with deleted_at.
+  (let [prof        (th/create-profile* 1)
+        _           (th/create-project* 1 {:profile-id (:id prof)
+                                           :team-id (:default-team-id prof)})
+        file        (th/create-file* 1 {:profile-id (:id prof)
+                                        :project-id (:default-project-id prof)
+                                        :is-shared false})
+        session-id  (create-session! prof 1)
+        source-path (th/tempfile "backend_tests/test_files/sample.jpg")
+        mfile       {:filename "sample.jpg"
+                     :path     source-path
+                     :mtype    "image/jpeg"
+                     :size     312043}
+        out1        (th/command! {::th/type        :upload-chunk
+                                  ::rpc/profile-id (:id prof)
+                                  :session-id      session-id
+                                  :index           0
+                                  :content         mfile})]
+    (t/is (nil? (:error out1)))
+
+    (let [assemble-out (th/command! {::th/type        :assemble-file-media-object
+                                     ::rpc/profile-id (:id prof)
+                                     :session-id      session-id
+                                     :file-id         (:id file)
+                                     :is-local        true
+                                     :name            "assembled-image"
+                                     :mtype           "image/jpeg"})]
+      (t/is (nil? (:error assemble-out))))
+
+    ;; chunk mappings stay until objects-gc purges them, session row
+    ;; stays marked as consumed
+    (t/is (= 1 (:count (th/db-exec-one! ["select count(*) from upload_session_chunk where session_id = ?"
+                                         session-id]))))
+    (t/is (some? (:deleted-at (th/db-exec-one! ["select deleted_at from upload_session where id = ?"
+                                                session-id]))))
+
+    ;; uploading to the consumed session fails without creating an object
+    (let [before (:count (th/db-exec-one! ["select count(*) from storage_object"]))
+          out    (th/command! {::th/type        :upload-chunk
+                               ::rpc/profile-id (:id prof)
+                               :session-id      session-id
+                               :index           0
+                               :content         mfile})]
+      (t/is (some? (:error out)))
+      (t/is (= :not-found (-> out :error ex-data :type)))
+      (t/is (= :object-not-found (-> out :error ex-data :code)))
+      (t/is (= before (:count (th/db-exec-one! ["select count(*) from storage_object"])))))))
+
+(defn- sql-state-of
+  "Runs thunk (a db statement) and returns the SQLState of the raised
+  SQLException, or nil when no error is raised."
+  [thunk]
+  (try
+    (thunk)
+    nil
+    (catch java.sql.SQLException cause
+      (.getSQLState cause))))
+
+(t/deftest upload-session-chunk-restrict-blocks-direct-deletes
+  ;; With a live mapping row, deleting the storage object or the session
+  ;; directly violates the RESTRICT foreign keys (SQLState 23503).
+  (let [prof        (th/create-profile* 1)
+        session-id  (create-session! prof 1)
+        source-path (th/tempfile "backend_tests/test_files/sample.jpg")
+        mfile       {:filename "sample.jpg"
+                     :path     source-path
+                     :mtype    "image/jpeg"
+                     :size     312043}
+        out         (th/command! {::th/type        :upload-chunk
+                                  ::rpc/profile-id (:id prof)
+                                  :session-id      session-id
+                                  :index           0
+                                  :content         mfile})]
+    (t/is (nil? (:error out)))
+
+    (let [object-id (:object-id (th/db-exec-one! ["select object_id from upload_session_chunk where session_id = ?"
+                                                  session-id]))]
+      (t/is (= "23503" (sql-state-of #(th/db-exec! ["delete from storage_object where id = ?"
+                                                    object-id]))))
+      (t/is (= "23503" (sql-state-of #(th/db-exec! ["delete from upload_session where id = ?"
+                                                    session-id]))))
+      ;; the profile cannot disappear either while its session is live
+      ;; (profile_id FK is NO ACTION DEFERRABLE; purge goes through
+      ;; objects-gc). The deletion_protection rule is disabled here so the
+      ;; statement reaches the FK check.
+      (t/is (= "23503" (sql-state-of #(db/transact! th/*pool*
+                                                    (fn [conn]
+                                                      (db/exec-one! conn ["SET LOCAL rules.deletion_protection TO off"])
+                                                      (db/exec! conn ["delete from profile where id = ?"
+                                                                      (:id prof)])))))))))
+
 ;; --- Clone File Media Object BOLA tests ---
 
 (defn- create-storage-object!
diff --git a/backend/test/backend_tests/storage_test.clj b/backend/test/backend_tests/storage_test.clj
index cc34773e67..519c21dafb 100644
--- a/backend/test/backend_tests/storage_test.clj
+++ b/backend/test/backend_tests/storage_test.clj
@@ -292,8 +292,9 @@
                        {:id (:id result-2)})
 
         ;; run the objects gc task for permanent deletion
+        ;; (processed = 2: the consumed upload session plus the font variant)
         (let [res (th/run-task! :objects-gc {})]
-          (t/is (= 1 (:processed res))))
+          (t/is (= 2 (:processed res))))
 
         ;; revert touched state to all storage objects
 
@@ -817,8 +818,8 @@
 
     ;; mark all the chunks of this session as pending (simulates rows that
     ;; were never promoted)
-    (th/db-exec! ["update storage_object set status = 'pending' where (metadata->>'~:upload-id') = ?"
-                  (str session-id)])
+    (th/db-exec! ["update storage_object set status = 'pending' where id in (select object_id from upload_session_chunk where session_id = ?)"
+                  session-id])
 
     ;; assembling fails because no chunk is visible anymore
     (let [assemble-out (th/command! {::th/type :assemble-file-media-object
@@ -830,6 +831,143 @@
                                      :mtype "image/jpeg"})]
       (t/is (some? (:error assemble-out))))))
 
+(t/deftest upload-session-stalled-purge-lifecycle
+  ;; Full lifecycle of a stalled session: objects-gc purges the session and
+  ;; its mappings while touching the objects, touched-gc marks them deleted
+  ;; and deleted-gc removes rows and blobs.
+  (let [prof       (th/create-profile* 1)
+        _          (th/create-project* 1 {:profile-id (:id prof)
+                                          :team-id (:default-team-id prof)})
+        _          (th/create-file* 1 {:profile-id (:id prof)
+                                       :project-id (:default-project-id prof)
+                                       :is-shared false})
+        mfile      {:filename "chunk"
+                    :path (th/tempfile "backend_tests/test_files/sample.jpg")
+                    :mtype "image/jpeg"
+                    :size 312043}
+        session-id (-> (th/command! {::th/type :create-upload-session
+                                     ::rpc/profile-id (:id prof)
+                                     :total-chunks 1})
+                       :result :session-id)
+        out        (th/command! {::th/type :upload-chunk
+                                 ::rpc/profile-id (:id prof)
+                                 :session-id session-id
+                                 :index 0
+                                 :content mfile})]
+
+    (t/is (nil? (:error out)))
+    (t/is (= 1 (:count (th/db-exec-one! ["select count(*) from upload_session_chunk where session_id = ?"
+                                         session-id]))))
+
+    ;; backdate the session so it counts as stalled
+    (th/db-exec! ["update upload_session set created_at = now() - interval '2 hours' where id = ?"
+                  session-id])
+
+    ;; objects-gc purges session and mappings, touching the objects
+    (let [res (th/run-task! :objects-gc {})]
+      (t/is (= 1 (:processed res))))
+    (t/is (= 0 (:count (th/db-exec-one! ["select count(*) from upload_session where id = ?"
+                                         session-id]))))
+    (t/is (= 0 (:count (th/db-exec-one! ["select count(*) from upload_session_chunk where session_id = ?"
+                                         session-id]))))
+    (t/is (= 1 (:count (th/db-exec-one! ["select count(*) from storage_object where touched_at is not null"]))))
+
+    ;; touched-gc marks the orphaned object as deleted
+    (let [res (binding [ct/*clock* (ct/fixed-clock (ct/in-future {:hours 3}))]
+                (th/run-task! :storage-gc-touched {}))]
+      (t/is (= 0 (:freeze res)))
+      (t/is (= 1 (:delete res))))
+
+    ;; deleted-gc removes the row and the blob (clock past the mark time)
+    (let [res (binding [ct/*clock* (ct/fixed-clock (ct/in-future {:hours 4}))]
+                (th/run-task! :storage-gc-deleted {}))]
+      (t/is (= 1 (:deleted res))))
+    (t/is (= 0 (:count (th/db-exec-one! ["select count(*) from storage_object"]))))))
+
+(t/deftest upload-session-consumed-purge
+  ;; An assembled session is marked as consumed and objects-gc purges it
+  ;; right away, without waiting for the stalled threshold.
+  (let [prof       (th/create-profile* 1)
+        _          (th/create-project* 1 {:profile-id (:id prof)
+                                          :team-id (:default-team-id prof)})
+        file       (th/create-file* 1 {:profile-id (:id prof)
+                                       :project-id (:default-project-id prof)
+                                       :is-shared false})
+        mfile      {:filename "chunk"
+                    :path (th/tempfile "backend_tests/test_files/sample.jpg")
+                    :mtype "image/jpeg"
+                    :size 312043}
+        session-id (-> (th/command! {::th/type :create-upload-session
+                                     ::rpc/profile-id (:id prof)
+                                     :total-chunks 1})
+                       :result :session-id)
+        out        (th/command! {::th/type :upload-chunk
+                                 ::rpc/profile-id (:id prof)
+                                 :session-id session-id
+                                 :index 0
+                                 :content mfile})]
+
+    (t/is (nil? (:error out)))
+
+    (let [assemble-out (th/command! {::th/type :assemble-file-media-object
+                                     ::rpc/profile-id (:id prof)
+                                     :session-id session-id
+                                     :file-id (:id file)
+                                     :is-local true
+                                     :name "assembled-image"
+                                     :mtype "image/jpeg"})]
+      (t/is (nil? (:error assemble-out))))
+
+    ;; mappings stay, session row stays marked as consumed; objects-gc
+    ;; purges both
+    (t/is (= 1 (:count (th/db-exec-one! ["select count(*) from upload_session_chunk where session_id = ?"
+                                         session-id]))))
+    (t/is (some? (:deleted-at (th/db-exec-one! ["select deleted_at from upload_session where id = ?"
+                                                session-id]))))
+
+    ;; objects-gc purges the consumed session immediately
+    (let [res (th/run-task! :objects-gc {})]
+      (t/is (= 1 (:processed res))))
+    (t/is (= 0 (:count (th/db-exec-one! ["select count(*) from upload_session where id = ?"
+                                         session-id]))))
+    (t/is (= 0 (:count (th/db-exec-one! ["select count(*) from upload_session_chunk where session_id = ?"
+                                         session-id]))))))
+
+(t/deftest upload-session-profile-purge
+  ;; Sessions owned by a profile pending purge are drained first, so the
+  ;; profile delete (which cascades to its sessions) never hits the chunk
+  ;; RESTRICT foreign keys.
+  (let [prof       (th/create-profile* 1)
+        mfile      {:filename "chunk"
+                    :path (th/tempfile "backend_tests/test_files/sample.jpg")
+                    :mtype "image/jpeg"
+                    :size 312043}
+        session-id (-> (th/command! {::th/type :create-upload-session
+                                     ::rpc/profile-id (:id prof)
+                                     :total-chunks 1})
+                       :result :session-id)
+        out        (th/command! {::th/type :upload-chunk
+                                 ::rpc/profile-id (:id prof)
+                                 :session-id session-id
+                                 :index 0
+                                 :content mfile})]
+    (t/is (nil? (:error out)))
+
+    ;; soft-delete the profile; the live session is neither consumed nor stalled
+    (th/db-update! :profile {:deleted-at (ct/now)} {:id (:id prof)})
+
+    (th/run-task! :objects-gc {})
+
+    ;; session and mappings are gone, profile row deletes cleanly
+    (t/is (= 0 (:count (th/db-exec-one! ["select count(*) from upload_session where id = ?"
+                                         session-id]))))
+    (t/is (= 0 (:count (th/db-exec-one! ["select count(*) from upload_session_chunk where session_id = ?"
+                                         session-id]))))
+    (t/is (= 0 (:count (th/db-exec-one! ["select count(*) from profile where id = ?"
+                                         (:id prof)]))))
+    ;; and the chunk object was touched for the storage GC
+    (t/is (= 1 (:count (th/db-exec-one! ["select count(*) from storage_object where touched_at is not null"]))))))
+
 (defn- fake-s3-backend
   []
   {::sto/type         :s3