diff --git a/openhands/automation/dispatcher.py b/openhands/automation/dispatcher.py
index 9065a60d..bf800a11 100644
--- a/openhands/automation/dispatcher.py
+++ b/openhands/automation/dispatcher.py
@@ -77,8 +77,9 @@ async def _download_internal_tarball(
     """Download a tarball from storage using the TarballUpload record.
 
     Raises:
-        TarballNotFoundError: If the tarball upload record doesn't exist.
-            This is a permanent error that should disable the automation.
+        TarballNotFoundError: If the tarball upload record doesn't exist, or if
+            the record exists but its storage object is missing. Both are
+            permanent errors that should disable the automation.
         ValueError: If no database session is provided.
     """
     if session is None:
@@ -94,10 +95,19 @@ async def _download_internal_tarball(
             "The tarball may have been deleted."
         )
 
-    from openhands.automation.storage import get_file_store
+    from openhands.automation.storage import ObjectNotFoundError, get_file_store
 
     store = get_file_store()
-    return store.read(upload.storage_path)
+    try:
+        return store.read(upload.storage_path)
+    except ObjectNotFoundError as e:
+        # Only confirmed absence is permanent; transient storage errors raise
+        # plain FileNotFoundError and keep retrying on the next schedule tick.
+        raise TarballNotFoundError(
+            f"Internal tarball object missing from storage at "
+            f"{upload.storage_path!r} for upload {upload_id}. "
+            "Recreate the automation to restore it."
+        ) from e
 
 
 async def _poll_pending_runs(
diff --git a/openhands/automation/git_sync/loop.py b/openhands/automation/git_sync/loop.py
index ec16b3bd..d20e5884 100644
--- a/openhands/automation/git_sync/loop.py
+++ b/openhands/automation/git_sync/loop.py
@@ -58,7 +58,7 @@
     UploadStatus,
 )
 from openhands.automation.schemas import Trigger, validate_command_string
-from openhands.automation.storage import get_file_store
+from openhands.automation.storage import ObjectNotFoundError, get_file_store
 from openhands.automation.utils import utcnow
 from openhands.automation.utils.periodic_loop import run_periodic_loop
 from openhands.automation.utils.service_metadata import (
@@ -357,13 +357,20 @@ async def _stored_tarball_matches(
         return False
 
 
-async def _delete_superseded_upload(session: AsyncSession, tarball_path: str) -> None:
-    """Remove the upload an automation just stopped pointing at.
+async def _delete_superseded_upload(
+    session: AsyncSession, tarball_path: str, pending_storage_deletes: list[str]
+) -> None:
+    """Mark the upload an automation just stopped pointing at for removal.
 
     Without this, every git-side edit left the previous upload row and file
     referenced by nothing, so routine PR edits accumulated a tarball copy per
     merge with no reaper to collect them. Mirrors the cleanup
     `regenerate_preset_prompt_tarball` does for prompt edits.
+
+    Soft-deletes the record in the current transaction; the storage object is
+    only queued on `pending_storage_deletes` and removed after the cycle's
+    commit. Deleting before the commit destroyed the object irreversibly while
+    a rollback revived the record, stranding it pointing at a missing object.
     """
     upload_id = parse_internal_upload_id(tarball_path)
     if upload_id is None:
@@ -375,21 +382,22 @@ async def _delete_superseded_upload(session: AsyncSession, tarball_path: str) ->
     if upload is None or upload.deleted_at is not None:
         return
 
-    file_removed = False
-    try:
-        await asyncio.to_thread(get_file_store().delete, upload.storage_path)
-        file_removed = True
-    except FileNotFoundError:
-        file_removed = True
-    except Exception:
-        logger.exception(
-            "Failed to delete superseded tarball at %s", upload.storage_path
-        )
-    # Soft-delete only once the file is confirmed gone: a failed delete leaves
-    # the record live, so the file stays discoverable for a retry rather than
-    # becoming a hidden orphan.
-    if file_removed:
-        upload.deleted_at = utcnow()
+    upload.deleted_at = utcnow()
+    pending_storage_deletes.append(upload.storage_path)
+
+
+async def _delete_pending_storage_objects(storage_paths: list[str]) -> None:
+    """Best-effort removal of superseded tarball objects after the commit."""
+    if not storage_paths:
+        return
+    file_store = get_file_store()
+    for path in storage_paths:
+        try:
+            await asyncio.to_thread(file_store.delete, path)
+        except ObjectNotFoundError:
+            pass  # already gone
+        except Exception:
+            logger.exception("Failed to delete superseded tarball at %s", path)
 
 
 async def _resolve_tarball_path(
@@ -398,6 +406,7 @@ async def _resolve_tarball_path(
     deserialized: DeserializedAutomation,
     slug: str,
     existing: Automation | None,
+    pending_storage_deletes: list[str],
 ) -> str:
     if deserialized.tarball_bytes is not None:
         if existing is not None and await _stored_tarball_matches(
@@ -417,7 +426,9 @@ async def _resolve_tarball_path(
             slug,
         )
         if existing is not None:
-            await _delete_superseded_upload(session, existing.tarball_path)
+            await _delete_superseded_upload(
+                session, existing.tarball_path, pending_storage_deletes
+            )
         return new_path
 
     tarball_source = fields.get("tarball_source") or {}
@@ -439,6 +450,7 @@ async def _validate_and_resolve_fields(
     fields: dict,
     deserialized: DeserializedAutomation,
     slug: str,
+    pending_storage_deletes: list[str],
     existing: Automation | None = None,
 ) -> dict:
     """Validate automation.yaml fields and resolve a tarball_path, mirroring
@@ -459,7 +471,7 @@ async def _validate_and_resolve_fields(
     )
     timeout = validate_automation_timeout(fields.get("timeout"))
     tarball_path = await _resolve_tarball_path(
-        session, fields, deserialized, slug, existing
+        session, fields, deserialized, slug, existing, pending_storage_deletes
     )
 
     return {
@@ -486,9 +498,10 @@ async def _create_automation_from_git(
     deserialized: DeserializedAutomation,
     dir_files: dict[str, bytes],
     head: str,
+    pending_storage_deletes: list[str],
 ) -> None:
     values = await _validate_and_resolve_fields(
-        session, deserialized.fields, deserialized, slug
+        session, deserialized.fields, deserialized, slug, pending_storage_deletes
     )
     local_user = _get_local_user()
     automation = Automation(
@@ -516,6 +529,7 @@ async def _update_automation_from_git(
     deserialized: DeserializedAutomation,
     dir_files: dict[str, bytes],
     head: str,
+    pending_storage_deletes: list[str],
 ) -> None:
     new_hash = compute_content_hash(dir_files)
     if new_hash == state.content_hash:
@@ -531,7 +545,12 @@ async def _update_automation_from_git(
         )
 
     values = await _validate_and_resolve_fields(
-        session, deserialized.fields, deserialized, state.slug, existing=automation
+        session,
+        deserialized.fields,
+        deserialized,
+        state.slug,
+        pending_storage_deletes,
+        existing=automation,
     )
     for column, value in values.items():
         setattr(automation, column, value)
@@ -612,6 +631,7 @@ async def _import_from_git(
     timeout: float,
     encryption_key: str,
     result: SyncCycleResult,
+    pending_storage_deletes: list[str],
 ) -> None:
     all_dirs = _list_slug_directories(sync_root)
 
@@ -635,6 +655,9 @@ async def _import_from_git(
             continue
 
         dir_files = await asyncio.to_thread(_read_directory_files, directory)
+        # A rolled-back savepoint revives the soft-deleted upload rows queued
+        # inside it, so their storage paths must not stay pending for deletion.
+        pending_snapshot = len(pending_storage_deletes)
         try:
             # One SAVEPOINT per directory: a failure rolls back just its writes
             # (a half-populated Automation, a flushed TarballUpload) and leaves
@@ -648,18 +671,30 @@ async def _import_from_git(
 
                 if state is None:
                     await _create_automation_from_git(
-                        session, slug, deserialized, dir_files, head
+                        session,
+                        slug,
+                        deserialized,
+                        dir_files,
+                        head,
+                        pending_storage_deletes,
                     )
                     result.imported += 1
                 else:
                     await _update_automation_from_git(
-                        session, state, deserialized, dir_files, head
+                        session,
+                        state,
+                        deserialized,
+                        dir_files,
+                        head,
+                        pending_storage_deletes,
                     )
         except (ValidationError, ValueError) as e:
+            del pending_storage_deletes[pending_snapshot:]
             logger.warning(
                 "Skipping invalid automation directory %r from git: %s", slug, e
             )
         except Exception:
+            del pending_storage_deletes[pending_snapshot:]
             # Deliberately broad: anything escaping aborts the whole cycle --
             # no export, no push -- every cycle until someone fixes that one
             # directory by hand. Traceback logged, since reaching here means an
@@ -945,6 +980,11 @@ async def _run_sync_cycle_locked(
 
     result = SyncCycleResult(head=head)
 
+    # Storage paths of superseded uploads soft-deleted during the import; their
+    # objects are only removed after the commit below succeeds, so a rollback
+    # can never revive a record whose object is already gone.
+    pending_storage_deletes: list[str] = []
+
     # Committed *before* the push, not held open across it: SQLite holds its
     # single-writer lock for the whole transaction, and a push can take up to
     # git_sync_git_timeout_seconds.
@@ -970,6 +1010,7 @@ async def _run_sync_cycle_locked(
                 timeout,
                 encryption_key,
                 result,
+                pending_storage_deletes,
             )
 
         # After the import, so automations that arrived from git this cycle
@@ -981,6 +1022,10 @@ async def _run_sync_cycle_locked(
         await _export_dirty_automations(session, sync_root, encryption_key, result)
         await session.commit()
 
+    # Only after the commit: the soft-deletes are durable, so removing the
+    # objects can no longer strand a live record. A commit failure skips this.
+    await _delete_pending_storage_objects(pending_storage_deletes)
+
     # Unconditional, not gated on `exported`: commit_and_push also retries a
     # previous cycle's unpushed commit and pushes to a newly-repointed remote.
     # Gating made those recovery paths unreachable whenever there was nothing
diff --git a/openhands/automation/preset_router.py b/openhands/automation/preset_router.py
index f355b4c8..12d6267f 100644
--- a/openhands/automation/preset_router.py
+++ b/openhands/automation/preset_router.py
@@ -19,7 +19,15 @@
 from pathlib import Path
 from typing import Any
 
-from fastapi import APIRouter, Depends, HTTPException, Request, Response, status
+from fastapi import (
+    APIRouter,
+    BackgroundTasks,
+    Depends,
+    HTTPException,
+    Request,
+    Response,
+    status,
+)
 from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -34,7 +42,7 @@
     TemplateProvenance,
     Trigger,
 )
-from openhands.automation.storage import FileStore, get_file_store
+from openhands.automation.storage import FileStore, ObjectNotFoundError, get_file_store
 from openhands.automation.telemetry import (
     capture_automation_event,
     get_request_telemetry_context,
@@ -298,10 +306,23 @@ def _replace_prompt_in_tarball(tarball_bytes: bytes, new_prompt: str) -> bytes |
     return out_buffer.read()
 
 
+def _delete_storage_object_best_effort(
+    file_store: FileStore, storage_path: str
+) -> None:
+    """Best-effort post-commit removal of a superseded tarball object."""
+    try:
+        file_store.delete(storage_path)
+    except ObjectNotFoundError:
+        pass  # already gone -- nothing to clean up
+    except Exception:
+        logger.exception("Failed to delete superseded tarball at %s", storage_path)
+
+
 async def regenerate_preset_prompt_tarball(
     automation: Automation,
     new_prompt: str,
     session: AsyncSession,
+    background_tasks: BackgroundTasks,
 ) -> str | None:
     """Rebuild a preset automation's tarball with an updated prompt.
 
@@ -313,6 +334,8 @@ async def regenerate_preset_prompt_tarball(
     Reads the automation's current internal-upload tarball, swaps in ``new_prompt``
     (leaving all other files untouched), uploads the result as a new internal upload,
     and returns its ``oh-internal://`` URL for the caller to store on ``tarball_path``.
+    The superseded upload is soft-deleted in the current transaction; its storage
+    object is removed via ``background_tasks`` only after the transaction commits.
 
     Returns ``None`` — leaving the tarball unchanged — when the automation is not a
     regenerable preset: its ``tarball_path`` is an external URL, the referenced upload
@@ -333,7 +356,10 @@ async def regenerate_preset_prompt_tarball(
 
     try:
         current_tarball = file_store.read(source_upload.storage_path)
-    except FileNotFoundError:
+    except ObjectNotFoundError:
+        # Only confirmed absence means "not regenerable"; a transient storage
+        # error must propagate (rolling back the edit) rather than silently
+        # leaving the old prompt baked into the tarball.
         return None
 
     new_tarball = _replace_prompt_in_tarball(current_tarball, new_prompt)
@@ -376,46 +402,32 @@ async def regenerate_preset_prompt_tarball(
             detail=f"Failed to upload regenerated tarball: {e!s}",
         )
 
-    # The old tarball is now superseded. Remove its file and soft-delete the
-    # upload record so repeated prompt edits don't accumulate orphaned storage.
-    # Only soft-delete once the file is confirmed gone: if the delete fails the
-    # record stays live so the still-present file remains discoverable for a
-    # later retry/cleanup instead of becoming a hidden orphan (file on disk,
-    # record marked deleted).
-    old_object_delete_succeeded = False
-    old_object_already_missing = False
-    try:
-        file_store.delete(source_upload.storage_path)
-        old_object_delete_succeeded = True
-    except FileNotFoundError:
-        old_object_already_missing = True
-    except Exception as e:
-        logger.exception(
-            "Failed to delete superseded tarball at %s: %s",
-            source_upload.storage_path,
-            e,
-        )
-    file_removed = old_object_delete_succeeded or old_object_already_missing
-    if file_removed:
-        source_upload.deleted_at = utcnow()
+    # The old tarball is now superseded. Soft-delete its record inside this
+    # transaction, but remove its storage object only after the commit, via a
+    # background task (which runs only for a successful response, after the
+    # function-scoped session has committed -- see update_automation). Deleting
+    # before the commit destroyed the object irreversibly while a rollback
+    # reverted this soft-delete and the tarball_path update, stranding a live
+    # record pointing at a missing object. Worst case now -- a crash between
+    # commit and task -- leaks an orphaned object whose record is already
+    # soft-deleted, which is the recoverable direction.
+    source_upload.deleted_at = utcnow()
+    background_tasks.add_task(
+        _delete_storage_object_best_effort, file_store, source_upload.storage_path
+    )
 
     logger.info(
         "Regenerated preset tarball: automation_id=%s, old_upload_id=%s, "
-        "new_upload_id=%s, old_object_delete_succeeded=%s, "
-        "old_object_already_missing=%s",
+        "new_upload_id=%s",
         automation.id,
         source_upload.id,
         new_upload_id,
-        old_object_delete_succeeded,
-        old_object_already_missing,
         extra={
             "automation_id": str(automation.id),
             "old_upload_id": str(source_upload.id),
             "old_storage_path": source_upload.storage_path,
             "new_upload_id": str(new_upload_id),
             "new_storage_path": storage_path,
-            "old_object_delete_succeeded": old_object_delete_succeeded,
-            "old_object_already_missing": old_object_already_missing,
         },
     )
 
diff --git a/openhands/automation/router.py b/openhands/automation/router.py
index b22b8988..4dc8f7ac 100644
--- a/openhands/automation/router.py
+++ b/openhands/automation/router.py
@@ -6,7 +6,16 @@
 import uuid
 from typing import Any
 
-from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response, status
+from fastapi import (
+    APIRouter,
+    BackgroundTasks,
+    Depends,
+    HTTPException,
+    Query,
+    Request,
+    Response,
+    status,
+)
 from fastapi.responses import RedirectResponse
 from sqlalchemy import func, select, update
 from sqlalchemy.engine import CursorResult
@@ -194,8 +203,14 @@ async def update_automation(
     automation_id: uuid.UUID,
     body: UpdateAutomationRequest,
     request: Request,
+    background_tasks: BackgroundTasks,
     user: AuthenticatedUser = Depends(_require_manage_automations),
-    session: AsyncSession = Depends(get_session),
+    # Function scope commits the session when the handler returns, BEFORE the
+    # response is sent and its background tasks run. With the default request
+    # scope the deferred tarball delete would run before the commit, and a
+    # commit failure would strand a live upload record pointing at an
+    # already-deleted object.
+    session: AsyncSession = Depends(get_session, scope="function"),
 ) -> AutomationResponse:
     """Partially update an automation."""
     auto = await _get_user_automation(session, automation_id, user.user_id, user.org_id)
@@ -223,7 +238,7 @@ async def update_automation(
         and auto.prompt != original_prompt
     ):
         new_tarball_path = await regenerate_preset_prompt_tarball(
-            auto, auto.prompt, session
+            auto, auto.prompt, session, background_tasks
         )
         if new_tarball_path is not None:
             auto.tarball_path = new_tarball_path
diff --git a/openhands/automation/storage/__init__.py b/openhands/automation/storage/__init__.py
index 98d137b7..cac11903 100644
--- a/openhands/automation/storage/__init__.py
+++ b/openhands/automation/storage/__init__.py
@@ -1,5 +1,5 @@
 from openhands.automation.storage.factory import get_file_store
-from openhands.automation.storage.file_store import FileStore
+from openhands.automation.storage.file_store import FileStore, ObjectNotFoundError
 from openhands.automation.storage.google_cloud import (
     FileSizeLimitExceeded,
     GoogleCloudFileStore,
@@ -13,6 +13,7 @@
     "FileSizeLimitExceeded",
     "GoogleCloudFileStore",
     "LocalFileStore",
+    "ObjectNotFoundError",
     "S3FileStore",
     "get_file_store",
 ]
diff --git a/openhands/automation/storage/file_store.py b/openhands/automation/storage/file_store.py
index 108d3388..0cfe0a2d 100644
--- a/openhands/automation/storage/file_store.py
+++ b/openhands/automation/storage/file_store.py
@@ -7,6 +7,17 @@
 BUCKET_PREFIX = "automation"
 
 
+class ObjectNotFoundError(FileNotFoundError):
+    """The storage object at the given path genuinely does not exist.
+
+    Backends raise this only on confirmed absence (S3 404/NoSuchKey, GCS
+    NotFound, a missing local file). Transient and access errors keep raising
+    plain FileNotFoundError, so callers that must distinguish "gone forever"
+    from "unreadable right now" can catch this subclass while existing
+    ``except FileNotFoundError`` handlers keep working unchanged.
+    """
+
+
 class FileStore(ABC):
     """Abstract base class for file storage operations."""
 
@@ -25,7 +36,8 @@ def read(self, path: str) -> bytes:
         """Read and return the contents of the file at the given path.
 
         Raises:
-            FileNotFoundError: If the file does not exist.
+            ObjectNotFoundError: If the file does not exist.
+            FileNotFoundError: For other storage errors (backend-dependent).
         """
         pass
 
@@ -36,7 +48,12 @@ def list(self, path: str) -> list[str]:
 
     @abstractmethod
     def delete(self, path: str) -> None:
-        """Delete the file at the given path."""
+        """Delete the file at the given path.
+
+        Raises:
+            ObjectNotFoundError: If the file does not exist (LocalFileStore is
+                silently idempotent instead).
+        """
         pass
 
     @abstractmethod
diff --git a/openhands/automation/storage/google_cloud.py b/openhands/automation/storage/google_cloud.py
index 60acd0e0..0700a4c9 100644
--- a/openhands/automation/storage/google_cloud.py
+++ b/openhands/automation/storage/google_cloud.py
@@ -6,7 +6,11 @@
 from google.cloud import storage
 from google.cloud.exceptions import NotFound
 
-from openhands.automation.storage.file_store import BUCKET_PREFIX, FileStore
+from openhands.automation.storage.file_store import (
+    BUCKET_PREFIX,
+    FileStore,
+    ObjectNotFoundError,
+)
 
 
 if TYPE_CHECKING:
@@ -92,14 +96,14 @@ def read(self, path: str) -> bytes:
             The file contents as bytes.
 
         Raises:
-            FileNotFoundError: If the file does not exist.
+            ObjectNotFoundError: If the file does not exist.
         """
         full_path = self._prefixed_path(path)
         blob = self.bucket.blob(full_path)
         try:
             return blob.download_as_bytes()
         except NotFound:
-            raise FileNotFoundError(f"File not found: {full_path}")
+            raise ObjectNotFoundError(f"File not found: {full_path}")
 
     def list(self, path: str) -> list[str]:
         """
@@ -127,14 +131,14 @@ def delete(self, path: str) -> None:
                   with "automation/").
 
         Raises:
-            FileNotFoundError: If the file doesn't exist.
+            ObjectNotFoundError: If the file doesn't exist.
         """
         full_path = self._prefixed_path(path)
         blob = self.bucket.blob(full_path)
         try:
             blob.delete()
         except NotFound:
-            raise FileNotFoundError(f"File not found: {full_path}")
+            raise ObjectNotFoundError(f"File not found: {full_path}")
 
     async def write_stream(
         self,
diff --git a/openhands/automation/storage/local.py b/openhands/automation/storage/local.py
index 72e47a23..33ad52b9 100644
--- a/openhands/automation/storage/local.py
+++ b/openhands/automation/storage/local.py
@@ -5,7 +5,7 @@
 from collections.abc import AsyncIterator
 from pathlib import Path
 
-from openhands.automation.storage.file_store import FileStore
+from openhands.automation.storage.file_store import FileStore, ObjectNotFoundError
 from openhands.automation.storage.google_cloud import FileSizeLimitExceeded
 
 
@@ -61,7 +61,7 @@ def read(self, path: str) -> bytes:
         """Read and return the contents of the file at the given path."""
         full_path = self._full_path(path)
         if not full_path.exists():
-            raise FileNotFoundError(f"File not found: {path}")
+            raise ObjectNotFoundError(f"File not found: {path}")
         return full_path.read_bytes()
 
     def list(self, path: str) -> list[str]:
diff --git a/openhands/automation/storage/s3.py b/openhands/automation/storage/s3.py
index 8e0c78fc..e78b044a 100644
--- a/openhands/automation/storage/s3.py
+++ b/openhands/automation/storage/s3.py
@@ -8,7 +8,11 @@
 import boto3
 import botocore.exceptions
 
-from openhands.automation.storage.file_store import BUCKET_PREFIX, FileStore
+from openhands.automation.storage.file_store import (
+    BUCKET_PREFIX,
+    FileStore,
+    ObjectNotFoundError,
+)
 from openhands.automation.storage.google_cloud import FileSizeLimitExceeded
 
 
@@ -192,7 +196,8 @@ def read(self, path: str) -> bytes:
             The file contents as bytes.
 
         Raises:
-            FileNotFoundError: If the file does not exist.
+            ObjectNotFoundError: If the file does not exist.
+            FileNotFoundError: For other S3 errors.
         """
         full_path = self._prefixed_path(path)
         try:
@@ -236,7 +241,8 @@ def delete(self, path: str) -> None:
                   with "automation/").
 
         Raises:
-            FileNotFoundError: If the file doesn't exist or access is denied.
+            ObjectNotFoundError: If the file doesn't exist.
+            FileNotFoundError: For other S3 errors, including access denied.
         """
         full_path = self._prefixed_path(path)
         try:
@@ -271,7 +277,8 @@ def _handle_client_error(
             path: The S3 key/path involved.
 
         Raises:
-            FileNotFoundError: For not-found and access errors (to match FileStore API).
+            ObjectNotFoundError: If the key genuinely does not exist (404/NoSuchKey).
+            FileNotFoundError: For all other errors (to match FileStore API).
         """
         error_code = e.response.get("Error", {}).get("Code")
         error_msg = e.response.get("Error", {}).get("Message", "")
@@ -289,7 +296,7 @@ def _handle_client_error(
                 f"Bucket '{self.bucket_name}' does not exist"
             ) from e
         elif error_code in ("404", "NoSuchKey"):
-            raise FileNotFoundError(f"File not found: {path}") from e
+            raise ObjectNotFoundError(f"File not found: {path}") from e
         elif error_code == "AccessDenied":
             raise FileNotFoundError(
                 f"Access denied to '{self.bucket_name}/{path}'"
diff --git a/tests/test_disable_automation.py b/tests/test_disable_automation.py
index 9a734ad9..86798868 100644
--- a/tests/test_disable_automation.py
+++ b/tests/test_disable_automation.py
@@ -44,6 +44,32 @@ def _create_mock_backend() -> MagicMock:
     return mock_backend
 
 
+def _storage_path(upload_id: uuid.UUID) -> str:
+    """Storage path for a test upload, mirroring the production layout."""
+    return f"uploads/{TEST_ORG_ID}/{TEST_USER_ID}/{upload_id}.tar"
+
+
+async def _create_completed_upload(async_session_factory) -> uuid.UUID:
+    """Persist a live, COMPLETED upload row whose object is not in the store."""
+    from openhands.automation.models import TarballUpload, UploadStatus
+
+    upload_id = uuid.uuid4()
+    async with async_session_factory() as session:
+        session.add(
+            TarballUpload(
+                id=upload_id,
+                user_id=TEST_USER_ID,
+                org_id=TEST_ORG_ID,
+                name="test-upload",
+                status=UploadStatus.COMPLETED,
+                storage_path=_storage_path(upload_id),
+                size_bytes=1024,
+            )
+        )
+        await session.commit()
+    return upload_id
+
+
 def _docker_available() -> bool:
     """Check if Docker is available for testcontainers."""
     try:
@@ -227,6 +253,59 @@ async def test_raises_tarball_not_found_for_missing_upload(
         assert "not found" in str(exc_info.value).lower()
         assert str(fake_upload_id) in str(exc_info.value)
 
+    async def test_raises_tarball_not_found_for_missing_object(
+        self, async_session_factory
+    ):
+        """TarballNotFoundError is raised when the row is live but the object is gone.
+
+        This is the OSS-9505 failure mode: the object vanished from storage
+        while its upload row survived. Confirmed absence is permanent, so it
+        must be reclassified instead of retrying forever.
+        """
+        from openhands.automation.dispatcher import _download_internal_tarball
+        from openhands.automation.storage import ObjectNotFoundError
+
+        upload_id = await _create_completed_upload(async_session_factory)
+        store = MagicMock()
+        store.read.side_effect = ObjectNotFoundError(
+            f"File not found: {_storage_path(upload_id)}"
+        )
+
+        with patch("openhands.automation.storage.get_file_store", return_value=store):
+            async with async_session_factory() as session:
+                with pytest.raises(TarballNotFoundError) as exc_info:
+                    await _download_internal_tarball(upload_id, session)
+
+        message = str(exc_info.value)
+        assert "missing from storage" in message
+        assert _storage_path(upload_id) in message
+        assert isinstance(exc_info.value.__cause__, ObjectNotFoundError)
+
+    async def test_transient_storage_error_is_not_reclassified(
+        self, async_session_factory
+    ):
+        """A transient storage error must not become a permanent dispatch error.
+
+        The storage layer maps transient S3 failures (5xx, throttling) to plain
+        FileNotFoundError. Reclassifying those would permanently disable a
+        healthy automation during a storage outage; they must propagate
+        unchanged so the run fails and retries on the next tick.
+        """
+        from openhands.automation.dispatcher import _download_internal_tarball
+
+        upload_id = await _create_completed_upload(async_session_factory)
+        store = MagicMock()
+        store.read.side_effect = FileNotFoundError(
+            "S3 read failed (ServiceUnavailable): try again"
+        )
+
+        with patch("openhands.automation.storage.get_file_store", return_value=store):
+            async with async_session_factory() as session:
+                with pytest.raises(FileNotFoundError) as exc_info:
+                    await _download_internal_tarball(upload_id, session)
+
+        assert not isinstance(exc_info.value, TarballNotFoundError)
+
 
 @requires_docker
 class TestExecuteRunDisablesAutomation:
diff --git a/tests/test_git_sync.py b/tests/test_git_sync.py
index c8c0008a..8c6f6fe9 100644
--- a/tests/test_git_sync.py
+++ b/tests/test_git_sync.py
@@ -35,6 +35,7 @@
     GIT_SYNC_LAST_COMMIT_KEY,
     GIT_SYNC_LAST_ERROR_AT_KEY,
     GIT_SYNC_LAST_ERROR_KEY,
+    _delete_superseded_upload,
 )
 from openhands.automation.git_sync.serializer import encrypt_file_tree
 from openhands.automation.models import (
@@ -1753,6 +1754,34 @@ async def test_superseded_upload_is_soft_deleted_when_the_tarball_changes(
             assert automation.tarball_path != old_path
             old_upload = await session.get(TarballUpload, old_upload_id)
             assert old_upload.deleted_at is not None
+            # The superseded object itself is removed once the cycle commits.
+            with pytest.raises(FileNotFoundError):
+                file_store.read(old_upload.storage_path)
+
+    async def test_superseding_does_not_delete_the_object_before_commit(
+        self, sqlite_session_factory, file_store
+    ):
+        """Superseding an upload must leave its storage object intact.
+
+        The object is only removed after the cycle's commit. Deleting it here
+        meant a rolled-back cycle revived the upload row pointing at an
+        already-destroyed object, permanently breaking dispatch (OSS-9505).
+        """
+        # Arrange
+        automation_id = await _create_internal_automation(
+            sqlite_session_factory, file_store
+        )
+        pending: list[str] = []
+
+        # Act — supersede the upload inside a transaction that never commits.
+        async with sqlite_session_factory() as session:
+            automation = await session.get(Automation, automation_id)
+            upload_id = parse_internal_upload_id(automation.tarball_path)
+            await _delete_superseded_upload(session, automation.tarball_path, pending)
+
+        # Assert — only the path was queued; the object is still readable.
+        assert pending == [f"uploads/test/{upload_id}.tar"]
+        assert file_store.read(pending[0])
 
 
 class TestEnabledKeyWithNullValue:
diff --git a/tests/test_router.py b/tests/test_router.py
index 9b7be5d6..2ed9c16c 100644
--- a/tests/test_router.py
+++ b/tests/test_router.py
@@ -14,6 +14,7 @@
     UploadStatus,
 )
 from openhands.automation.preset_router import _build_storage_path, _generate_tarball
+from openhands.automation.storage import ObjectNotFoundError
 from openhands.automation.utils import utcnow
 from openhands.automation.utils.tarball_validation import (
     build_internal_url,
@@ -1301,6 +1302,94 @@ async def test_update_prompt_upload_failure_returns_500(
         await async_session.refresh(automation)
         assert automation.tarball_path == original_tarball_path
 
+    async def test_update_prompt_transient_storage_error_fails_the_edit(
+        self, async_client, async_session, preset_store
+    ):
+        """A transient storage error during a prompt edit fails the request.
+
+        Previously it was swallowed as "nothing to regenerate": the PATCH
+        returned 200 and updated the prompt column while the tarball kept the
+        old baked prompt, so the automation silently kept running the previous
+        prompt.
+        """
+        # Arrange — reading the current tarball fails transiently (e.g. S3 5xx).
+        automation = await _seed_prompt_preset_automation(
+            async_session, preset_store, "Original prompt"
+        )
+        preset_store.read = MagicMock(
+            side_effect=FileNotFoundError("S3 read failed (ServiceUnavailable)")
+        )
+
+        # Act & Assert — the request fails (rolling back the edit in
+        # production) instead of succeeding with a stale tarball.
+        with pytest.raises(FileNotFoundError):
+            await async_client.patch(
+                f"/api/automation/v1/{automation.id}",
+                json={"prompt": "Updated prompt"},
+            )
+        preset_store.write_stream.assert_not_called()
+
+    async def test_update_prompt_missing_source_tarball_skips_regeneration(
+        self, async_client, async_session, preset_store
+    ):
+        """A confirmed-missing source tarball leaves the tarball untouched.
+
+        Only genuine absence means "not a regenerable preset": the prompt
+        column still updates, but no new upload is written and the automation
+        keeps its existing tarball reference.
+        """
+        # Arrange — the current tarball object is confirmed absent.
+        automation = await _seed_prompt_preset_automation(
+            async_session, preset_store, "Original prompt"
+        )
+        original_tarball_path = automation.tarball_path
+        preset_store.read = MagicMock(side_effect=ObjectNotFoundError("File not found"))
+
+        # Act
+        response = await async_client.patch(
+            f"/api/automation/v1/{automation.id}",
+            json={"prompt": "Updated prompt"},
+        )
+
+        # Assert
+        assert response.status_code == 200
+        data = response.json()
+        assert data["prompt"] == "Updated prompt"
+        assert data["tarball_path"] == original_tarball_path
+        preset_store.write_stream.assert_not_called()
+
+    async def test_failed_update_does_not_delete_current_tarball(
+        self, async_client, async_session, preset_store
+    ):
+        """A request that fails after regeneration keeps the current tarball.
+
+        The superseded object may only be removed once the transaction commits.
+        Deleting it earlier meant a later failure rolled the automation back to
+        a tarball reference whose object was already destroyed, permanently
+        breaking every future dispatch (OSS-9505).
+        """
+        # Arrange — fail the request after the tarball has been regenerated.
+        automation = await _seed_prompt_preset_automation(
+            async_session, preset_store, "Original prompt"
+        )
+        old_upload_id = parse_internal_upload_id(automation.tarball_path)
+        assert old_upload_id is not None
+        old_storage_path = _build_storage_path(TEST_ORG_ID, TEST_USER_ID, old_upload_id)
+
+        # Act
+        with patch(
+            "openhands.automation.router.mark_git_sync_dirty",
+            AsyncMock(side_effect=RuntimeError("boom")),
+        ):
+            with pytest.raises(RuntimeError):
+                await async_client.patch(
+                    f"/api/automation/v1/{automation.id}",
+                    json={"prompt": "Updated prompt"},
+                )
+
+        # Assert — the object the automation still points at survives.
+        assert old_storage_path in preset_store._storage
+
     async def test_update_automation_timeout(self, async_client, async_session):
         """Can update automation timeout."""
         automation = Automation(
diff --git a/tests/test_storage.py b/tests/test_storage.py
index 7283e7fd..3ef2494d 100644
--- a/tests/test_storage.py
+++ b/tests/test_storage.py
@@ -17,6 +17,7 @@
     FileStore,
     GoogleCloudFileStore,
     LocalFileStore,
+    ObjectNotFoundError,
     S3FileStore,
     get_file_store,
 )
@@ -232,6 +233,44 @@ def test_delete(self):
             mock_bucket.blob.assert_called_once_with("automation/test/path.txt")
             mock_blob.delete.assert_called_once()
 
+    def test_read_not_found(self):
+        """Read raises ObjectNotFoundError when the blob doesn't exist."""
+        from google.cloud.exceptions import NotFound
+
+        settings = make_gcs_settings()
+        with patch("openhands.automation.storage.google_cloud.storage") as mock_storage:
+            mock_client = MagicMock()
+            mock_bucket = MagicMock()
+            mock_blob = MagicMock()
+
+            mock_storage.Client.return_value = mock_client
+            mock_client.bucket.return_value = mock_bucket
+            mock_bucket.blob.return_value = mock_blob
+            mock_blob.download_as_bytes.side_effect = NotFound("blob missing")
+
+            store = GoogleCloudFileStore(settings)
+            with pytest.raises(ObjectNotFoundError, match="File not found"):
+                store.read("test/nonexistent.txt")
+
+    def test_delete_not_found(self):
+        """Delete raises ObjectNotFoundError when the blob doesn't exist."""
+        from google.cloud.exceptions import NotFound
+
+        settings = make_gcs_settings()
+        with patch("openhands.automation.storage.google_cloud.storage") as mock_storage:
+            mock_client = MagicMock()
+            mock_bucket = MagicMock()
+            mock_blob = MagicMock()
+
+            mock_storage.Client.return_value = mock_client
+            mock_client.bucket.return_value = mock_bucket
+            mock_bucket.blob.return_value = mock_blob
+            mock_blob.delete.side_effect = NotFound("blob missing")
+
+            store = GoogleCloudFileStore(settings)
+            with pytest.raises(ObjectNotFoundError, match="File not found"):
+                store.delete("test/nonexistent.txt")
+
     def test_emulator_creates_bucket(self):
         """When using emulator, bucket is created if it doesn't exist."""
         settings = make_gcs_settings(storage_emulator_host="http://localhost:4443")
diff --git a/tests/test_storage_local.py b/tests/test_storage_local.py
index 21423560..80083da6 100644
--- a/tests/test_storage_local.py
+++ b/tests/test_storage_local.py
@@ -12,7 +12,11 @@
 import pytest
 
 from openhands.automation.config import StorageSettings, clear_config_cache
-from openhands.automation.storage import LocalFileStore, get_file_store
+from openhands.automation.storage import (
+    LocalFileStore,
+    ObjectNotFoundError,
+    get_file_store,
+)
 from openhands.automation.storage.google_cloud import (
     BUCKET_PREFIX,
     FileSizeLimitExceeded,
@@ -97,10 +101,10 @@ def test_read_returns_bytes(self, tmp_path: Path):
         assert isinstance(result, bytes)
 
     def test_read_not_found(self, tmp_path: Path):
-        """Read raises FileNotFoundError when file doesn't exist."""
+        """Read raises ObjectNotFoundError when file doesn't exist."""
         store = LocalFileStore(tmp_path)
 
-        with pytest.raises(FileNotFoundError, match="File not found"):
+        with pytest.raises(ObjectNotFoundError, match="File not found"):
             store.read("nonexistent.txt")
 
     def test_list_files(self, tmp_path: Path):
diff --git a/tests/test_storage_s3.py b/tests/test_storage_s3.py
index 854c0771..4ab5b595 100644
--- a/tests/test_storage_s3.py
+++ b/tests/test_storage_s3.py
@@ -13,7 +13,7 @@
 import pytest
 
 from openhands.automation.config import StorageSettings
-from openhands.automation.storage import S3FileStore
+from openhands.automation.storage import ObjectNotFoundError, S3FileStore
 from openhands.automation.storage.google_cloud import (
     BUCKET_PREFIX,
     FileSizeLimitExceeded,
@@ -113,7 +113,7 @@ def test_read_returns_bytes(self):
             )
 
     def test_read_not_found(self):
-        """Read raises FileNotFoundError when key doesn't exist."""
+        """Read raises ObjectNotFoundError when key doesn't exist."""
         settings = make_s3_settings()
         with patch("openhands.automation.storage.s3.boto3") as mock_boto3:
             mock_client = MagicMock()
@@ -124,9 +124,30 @@ def test_read_not_found(self):
             mock_boto3.client.return_value = mock_client
 
             store = S3FileStore(settings)
-            with pytest.raises(FileNotFoundError, match="File not found"):
+            with pytest.raises(ObjectNotFoundError, match="File not found"):
                 store.read("test/nonexistent.txt")
 
+    def test_read_transient_error_is_not_object_not_found(self):
+        """A transient S3 error must not be reported as a confirmed absence.
+
+        Callers treat ObjectNotFoundError as permanent (e.g. the dispatcher
+        disables the automation), so a 5xx from a flapping backend must stay a
+        plain FileNotFoundError.
+        """
+        settings = make_s3_settings()
+        with patch("openhands.automation.storage.s3.boto3") as mock_boto3:
+            mock_client = MagicMock()
+            error_response = {"Error": {"Code": "ServiceUnavailable"}}
+            mock_client.get_object.side_effect = botocore.exceptions.ClientError(
+                error_response, "GetObject"
+            )
+            mock_boto3.client.return_value = mock_client
+
+            store = S3FileStore(settings)
+            with pytest.raises(FileNotFoundError) as exc_info:
+                store.read("test/path.txt")
+            assert not isinstance(exc_info.value, ObjectNotFoundError)
+
     def test_list(self):
         """List files under a prefix, with automation prefix added and stripped."""
         settings = make_s3_settings()
@@ -179,7 +200,7 @@ def test_delete(self):
             )
 
     def test_delete_not_found(self):
-        """Delete raises FileNotFoundError when key doesn't exist."""
+        """Delete raises ObjectNotFoundError when key doesn't exist."""
         settings = make_s3_settings()
         with patch("openhands.automation.storage.s3.boto3") as mock_boto3:
             mock_client = MagicMock()
@@ -190,7 +211,7 @@ def test_delete_not_found(self):
             mock_boto3.client.return_value = mock_client
 
             store = S3FileStore(settings)
-            with pytest.raises(FileNotFoundError, match="File not found"):
+            with pytest.raises(ObjectNotFoundError, match="File not found"):
                 store.delete("test/nonexistent.txt")
 
     def test_endpoint_creates_bucket_when_auto_create_enabled(self):
diff --git a/tests/test_storage_s3_integration.py b/tests/test_storage_s3_integration.py
index 8684a4fd..84d20bff 100644
--- a/tests/test_storage_s3_integration.py
+++ b/tests/test_storage_s3_integration.py
@@ -16,7 +16,7 @@
 from testcontainers.minio import MinioContainer
 
 from openhands.automation.config import StorageSettings
-from openhands.automation.storage import S3FileStore
+from openhands.automation.storage import ObjectNotFoundError, S3FileStore
 from openhands.automation.storage.google_cloud import FileSizeLimitExceeded
 
 
@@ -107,8 +107,8 @@ def test_write_overwrite(self, s3_store):
         assert result == b"new content"
 
     def test_read_nonexistent_file(self, s3_store):
-        """Reading non-existent file raises FileNotFoundError."""
-        with pytest.raises(FileNotFoundError):
+        """Reading non-existent file raises ObjectNotFoundError."""
+        with pytest.raises(ObjectNotFoundError):
             s3_store.read("test/nonexistent.txt")
 
     def test_delete_file(self, s3_store):
@@ -122,8 +122,8 @@ def test_delete_file(self, s3_store):
             s3_store.read(test_path)
 
     def test_delete_nonexistent_file(self, s3_store):
-        """Deleting non-existent file raises FileNotFoundError."""
-        with pytest.raises(FileNotFoundError):
+        """Deleting non-existent file raises ObjectNotFoundError."""
+        with pytest.raises(ObjectNotFoundError):
             s3_store.delete("test/never_existed.txt")
 
     def test_list_files(self, s3_store):