diff --git a/src/appengine/handlers/fuzzers.py b/src/appengine/handlers/fuzzers.py
index f3c144f02d9..322f711e806 100644
--- a/src/appengine/handlers/fuzzers.py
+++ b/src/appengine/handlers/fuzzers.py
@@ -50,7 +50,11 @@ class Handler(base_handler.Handler):
   def get(self):
     """Handle a get request."""
     fuzzer_logs_bucket = fuzzer_logs.get_bucket()
-    fuzzers = list(data_types.Fuzzer.query().order(data_types.Fuzzer.name))
+    fuzzers = [
+        fuzzer
+        for fuzzer in data_types.Fuzzer.query().order(data_types.Fuzzer.name)
+        if not fuzzer.deleted
+    ]
     jobs = data_handler.get_all_job_type_names()
     corpora = [
         bundle.name for bundle in data_types.DataBundle.query().order(
@@ -297,11 +301,12 @@ def post(self):
     key = helpers.get_integer_key(request)
 
     fuzzer = ndb.Key(data_types.Fuzzer, key).get()
-    if not fuzzer:
+    if not fuzzer or fuzzer.deleted:
       raise helpers.EarlyExitError('Fuzzer not found.', 400)
 
     fuzzer_selection.update_mappings_for_fuzzer(fuzzer, mappings=[])
-    fuzzer.key.delete()
+    fuzzer.deleted = True
+    fuzzer.put()
 
     helpers.log('Deleted fuzzer %s' % fuzzer.name, helpers.MODIFY_OPERATION)
     return self.redirect('/fuzzers')
@@ -316,7 +321,7 @@ def get(self, fuzzer_name):
     helpers.log('LogHandler', fuzzer_name)
     fuzzer = data_types.Fuzzer.query(
         data_types.Fuzzer.name == fuzzer_name).get()
-    if not fuzzer:
+    if not fuzzer or fuzzer.deleted:
       raise helpers.EarlyExitError('Fuzzer not found.', 400)
 
     return self.render('viewer.html', {
diff --git a/src/appengine/handlers/jobs.py b/src/appengine/handlers/jobs.py
index cdd957a532f..b14c11628bc 100644
--- a/src/appengine/handlers/jobs.py
+++ b/src/appengine/handlers/jobs.py
@@ -61,7 +61,9 @@ def _job_to_dict(job):
   # Adding all associated fuzzers with each job.
   fuzzers = data_types.Fuzzer.query()
   result['fuzzers'] = [
-      fuzzer.name for fuzzer in fuzzers if job.name in fuzzer.jobs
+      fuzzer.name
+      for fuzzer in fuzzers
+      if not fuzzer.deleted and job.name in fuzzer.jobs
   ]
   return result
 
@@ -82,7 +84,7 @@ def get_results():
 
   result = {
       'hasMore': has_more,
-      'items': [_job_to_dict(item) for item in items],
+      'items': [_job_to_dict(item) for item in items if not item.deleted],
       'page': page,
       'pageSize': PAGE_SIZE,
       'totalItems': total_items,
@@ -103,7 +105,9 @@ def get(self):
         data_types.JobTemplate.name))
     queues = get_queues()
     fuzzers = [
-        fuzzer.name for fuzzer in data_types.Fuzzer.query(projection=['name'])
+        fuzzer.name
+        for fuzzer in data_types.Fuzzer.query()
+        if not fuzzer.deleted
     ]
     result, params = get_results()
 
@@ -183,6 +187,7 @@ def post(self):
     job.description = description
     job.environment_string = environment_string
     job.templates = templates
+    job.deleted = False
 
     blob_info = self.get_upload()
     if blob_info:
@@ -268,7 +273,7 @@ def post(self):
     """Handle a post request."""
     key = helpers.get_integer_key(request)
     job = ndb.Key(data_types.Job, key).get()
-    if not job:
+    if not job or job.deleted:
       raise helpers.EarlyExitError('Job not found.', 400)
 
     # Delete from fuzzers' jobs' list.
@@ -280,11 +285,14 @@ def post(self):
     # Delete associated fuzzer-job mapping(s).
     query = data_types.FuzzerJob.query()
     query = query.filter(data_types.FuzzerJob.job == job.name)
-    for mapping in ndb_utils.get_all_from_query(query):
-      mapping.key.delete()
+    mappings = list(ndb_utils.get_all_from_query(query))
+    for mapping in mappings:
+      mapping.deleted = True
+    ndb_utils.put_multi(mappings)
 
-    # Delete job.
-    job.key.delete()
+    # Soft delete job.
+    job.deleted = True
+    job.put()
 
     helpers.log('Deleted job %s' % job.name, helpers.MODIFY_OPERATION)
     return self.redirect('/jobs')
@@ -313,7 +321,7 @@ def get(self):
       raise helpers.EarlyExitError('No job name provided.', 400)
 
     job = data_types.Job.query(data_types.Job.name == name).get()
-    if not job:
+    if not job or job.deleted:
       raise helpers.EarlyExitError('Job not found.', 404)
 
     environment = job.get_environment()
diff --git a/src/clusterfuzz/_internal/bot/tasks/commands.py b/src/clusterfuzz/_internal/bot/tasks/commands.py
index 9260fe8ce71..9b681645056 100644
--- a/src/clusterfuzz/_internal/bot/tasks/commands.py
+++ b/src/clusterfuzz/_internal/bot/tasks/commands.py
@@ -304,10 +304,10 @@ def process_command_impl(task_name,
   environment.set_value('CF_TASK_JOB_NAME', job_name)
   if job_name != 'none':
     job = data_types.Job.query(data_types.Job.name == job_name).get()
-    # Job might be removed. In that case, we don't want an exception
+    # Job might be removed or deleted. In that case, we don't want an exception
     # raised and causing this task to be retried by another bot.
-    if not job:
-      logs.error("Job '%s' not found." % job_name)
+    if not job or job.deleted:
+      logs.error("Job '%s' not found or deleted." % job_name)
       return None
 
     if not job.platform:
@@ -429,6 +429,21 @@ def process_command_impl(task_name,
       minimize_fuzzer_override = job_environment.get('MINIMIZE_FUZZER_OVERRIDE')
       fuzzer_name = minimize_fuzzer_override or fuzzer_name
 
+    if task_name == 'fuzz' and fuzzer_name:
+      fuzzer = data_types.Fuzzer.query(
+          data_types.Fuzzer.name == fuzzer_name).get()
+      if not fuzzer or fuzzer.deleted:
+        logs.error("Fuzzer '%s' not found or deleted." % fuzzer_name)
+        return None
+
+      fuzzer_job = data_types.FuzzerJob.query(
+          data_types.FuzzerJob.fuzzer == fuzzer_name,
+          data_types.FuzzerJob.job == job_name).get()
+      if not fuzzer_job or fuzzer_job.deleted:
+        logs.error(f"FuzzerJob mapping for fuzzer '{fuzzer_name}' and job "
+                   f"'{job_name}' not found or deleted.")
+        return None
+
     if fuzzer_name and not environment.is_engine_fuzzer_job(job_name):
       fuzzer = data_types.Fuzzer.query(
           data_types.Fuzzer.name == fuzzer_name).get()
diff --git a/src/clusterfuzz/_internal/bot/tasks/setup.py b/src/clusterfuzz/_internal/bot/tasks/setup.py
index 9e8aab54316..b9f8eefe167 100644
--- a/src/clusterfuzz/_internal/bot/tasks/setup.py
+++ b/src/clusterfuzz/_internal/bot/tasks/setup.py
@@ -590,8 +590,9 @@ def preprocess_update_fuzzer_and_data_bundles(
   """Does preprocessing for calls to update_fuzzer_and_data_bundles in
   uworker_main. Returns a SetupInput object."""
   fuzzer = data_types.Fuzzer.query(data_types.Fuzzer.name == fuzzer_name).get()
-  if not fuzzer:
-    logs.error('No fuzzer exists with name %s.' % fuzzer_name)
+  if not fuzzer or fuzzer.deleted:
+    logs.error(
+        'No fuzzer exists with name %s or fuzzer is deleted.' % fuzzer_name)
     raise errors.InvalidFuzzerError
 
   update_input = uworker_msg_pb2.SetupInput(  # pylint: disable=no-member
diff --git a/src/clusterfuzz/_internal/bot/tasks/utasks/corpus_pruning_task.py b/src/clusterfuzz/_internal/bot/tasks/utasks/corpus_pruning_task.py
index bd803079963..d594519ccfe 100644
--- a/src/clusterfuzz/_internal/bot/tasks/utasks/corpus_pruning_task.py
+++ b/src/clusterfuzz/_internal/bot/tasks/utasks/corpus_pruning_task.py
@@ -28,6 +28,7 @@
 from google.cloud import ndb
 from google.protobuf import timestamp_pb2
 
+from clusterfuzz._internal.base import errors
 from clusterfuzz._internal.base import feature_flags
 from clusterfuzz._internal.base import utils
 from clusterfuzz._internal.bot.fuzzers import engine_common
@@ -1155,10 +1156,8 @@ def _create_backup_urls(fuzz_target: data_types.FuzzTarget,
   corpus_pruning_task_input.dated_backup_signed_url = dated_backup_signed_url
 
 
-def _utask_preprocess(fuzzer_name, job_type, uworker_env):
+def _utask_preprocess(fuzzer_name, job_type, uworker_env, fuzz_target):
   """Runs preprocessing for corpus pruning task."""
-  fuzz_target = data_handler.get_fuzz_target(fuzzer_name)
-
   task_name = f'corpus_pruning_{fuzzer_name}_{job_type}'
 
   # Get status of last execution.
@@ -1243,8 +1242,16 @@ def _utask_preprocess(fuzzer_name, job_type, uworker_env):
 def utask_preprocess(fuzzer_name, job_type, uworker_env):
   """Sets logs context and runs preprocessing for corpus pruning task."""
   fuzz_target = data_handler.get_fuzz_target(fuzzer_name)
+  if not fuzz_target:
+    logs.error(f'FuzzTarget {fuzzer_name} not found.')
+    return None
   with logs.fuzzer_log_context(fuzzer_name, job_type, fuzz_target):
-    return _utask_preprocess(fuzzer_name, job_type, uworker_env)
+    try:
+      return _utask_preprocess(fuzzer_name, job_type, uworker_env, fuzz_target)
+    except errors.InvalidFuzzerError:
+      logs.error(
+          f'Engine fuzzer {fuzz_target.engine} is invalid or no longer exists.')
+      return None
 
 
 _ERROR_HANDLER = uworker_handle_errors.CompositeErrorHandler({
diff --git a/src/clusterfuzz/_internal/cron/batch_fuzzer_jobs.py b/src/clusterfuzz/_internal/cron/batch_fuzzer_jobs.py
index 0d702d7ca1b..e0a4f79bd87 100644
--- a/src/clusterfuzz/_internal/cron/batch_fuzzer_jobs.py
+++ b/src/clusterfuzz/_internal/cron/batch_fuzzer_jobs.py
@@ -29,8 +29,10 @@ def batch_fuzzer_jobs():
   ]
 
   for platform in platforms:
-    fuzzer_jobs = list(
-        data_types.FuzzerJob.query(data_types.FuzzerJob.platform == platform))
+    fuzzer_jobs = [
+        job for job in data_types.FuzzerJob.query(
+            data_types.FuzzerJob.platform == platform) if not job.deleted
+    ]
     fuzzer_jobs.sort(key=lambda item: item.job)
 
     batches_to_remove = {
diff --git a/src/clusterfuzz/_internal/cron/cleanup.py b/src/clusterfuzz/_internal/cron/cleanup.py
index 13a054559b9..c2ae9dc431d 100644
--- a/src/clusterfuzz/_internal/cron/cleanup.py
+++ b/src/clusterfuzz/_internal/cron/cleanup.py
@@ -225,13 +225,40 @@ def cleanup_unused_fuzz_targets_and_jobs():
     if fuzz_target.fully_qualified_name() not in valid_fuzz_targets:
       to_delete.append(fuzz_target.key)
 
-  ndb_utils.delete_multi(to_delete)
+  if to_delete:
+    ndb_utils.delete_multi(to_delete)
   logs.info(
       f'Deleted {num_fuzz_target_jobs_to_delete} FuzzTargetJob entities and '
       f'{len(to_delete) - num_fuzz_target_jobs_to_delete} FuzzTarget entities. '
       f'{len(valid_target_jobs)} valid FuzzTargetJob entities remain.')
 
 
+def cleanup_invalid_fuzzer_jobs():
+  """Clean up FuzzerJob entities that reference non-existent or deleted Fuzzers
+  or Jobs."""
+  valid_fuzzers = {
+      fuzzer.name
+      for fuzzer in ndb_utils.get_all_from_model(data_types.Fuzzer)
+      if fuzzer.name and not fuzzer.deleted
+  }
+  valid_jobs = {
+      job.name
+      for job in ndb_utils.get_all_from_model(data_types.Job)
+      if job.name and not job.deleted
+  }
+
+  to_delete = [
+      fuzzer_job.key
+      for fuzzer_job in ndb_utils.get_all_from_model(data_types.FuzzerJob)
+      if (fuzzer_job.fuzzer not in valid_fuzzers or
+          fuzzer_job.job not in valid_jobs or fuzzer_job.deleted)
+  ]
+
+  if to_delete:
+    ndb_utils.delete_multi(to_delete)
+  logs.info(f'Deleted {len(to_delete)} invalid FuzzerJob entities.')
+
+
 def get_jobs_and_platforms_for_project():
   """Return a map of projects to jobs and platforms map to use for picking top
   crashes."""
@@ -1473,6 +1500,7 @@ def main():
   cleanup_reports_metadata()
   leak_blacklist.cleanup_global_blacklist()
   cleanup_unused_fuzz_targets_and_jobs()
+  cleanup_invalid_fuzzer_jobs()
   cleanup_unused_heartbeats()
   logs.info('Cleanup task finished successfully.')
   return True
diff --git a/src/clusterfuzz/_internal/cron/schedule_fuzz.py b/src/clusterfuzz/_internal/cron/schedule_fuzz.py
index abd34bdd9ea..52ed943aaea 100644
--- a/src/clusterfuzz/_internal/cron/schedule_fuzz.py
+++ b/src/clusterfuzz/_internal/cron/schedule_fuzz.py
@@ -125,6 +125,8 @@ def get_fuzz_tasks(self, num_tasks: int) -> list[tasks.Task]:
     # TODO(metzman): Handle cases where jobs are fuzzed by multiple fuzzers.
     candidates_by_job = {}
     for job in ndb_utils.get_all_from_query(data_types.Job.query()):
+      if job.deleted:
+        continue
       project = projects_by_name.get(job.project)
       base_os_version = None
       if project and project.base_os_version:
@@ -140,8 +142,23 @@ def get_fuzz_tasks(self, num_tasks: int) -> list[tasks.Task]:
     fuzzer_job_query = ndb_utils.get_all_from_query(
         data_types.FuzzerJob.query())
 
-    # TODO(metzman): Refactor this to use richer types and less primitives.
+    deleted_fuzzers = {
+        fuzzer.name
+        for fuzzer in ndb_utils.get_all_from_model(data_types.Fuzzer)
+        if fuzzer.name and fuzzer.deleted
+    }
+
     for fuzzer_job in fuzzer_job_query:
+      if fuzzer_job.deleted:
+        continue
+      if fuzzer_job.job not in candidates_by_job:
+        logs.warning(f'Job {fuzzer_job.job} for FuzzerJob '
+                     f'{fuzzer_job.fuzzer} not found or deleted.')
+        continue
+      if fuzzer_job.fuzzer in deleted_fuzzers:
+        logs.warning(f'Fuzzer {fuzzer_job.fuzzer} for FuzzerJob with job '
+                     f'{fuzzer_job.job} is deleted.')
+        continue
       fuzz_task_candidate = candidates_by_job[fuzzer_job.job].copy()
       fuzz_task_candidate.fuzzer = fuzzer_job.fuzzer
       fuzz_task_candidate.weight = fuzzer_job.actual_weight
@@ -216,7 +233,11 @@ def get_fuzz_tasks(self, num_tasks: int) -> list[tasks.Task]:
 
 def _get_jobs_for_platforms(platforms: list[str]) -> list[data_types.Job]:
   """Returns all jobs for the given platforms."""
-  return list(data_types.Job.query(data_types.Job.platform.IN(platforms)))
+  return [
+      job
+      for job in data_types.Job.query(data_types.Job.platform.IN(platforms))
+      if not job.deleted
+  ]
 
 
 def _get_swarming_jobs():
@@ -269,6 +290,7 @@ def _fill_queue(queue: PubSubTaskQueue, provider: BaseFuzzTaskProvider):
 def _create_candidates_from_jobs(
     jobs: list[data_types.Job]) -> list[FuzzTaskCandidate]:
   """Create candidates from jobs & assign weights to them."""
+  jobs = [job for job in jobs if not job.deleted]
   if not jobs:
     return []
 
@@ -278,8 +300,22 @@ def _create_candidates_from_jobs(
           data_types.FuzzerJob.job.IN(list(jobs_by_name.keys()))))
   fuzz_task_candidates = []
 
+  deleted_fuzzers = {
+      fuzzer.name
+      for fuzzer in ndb_utils.get_all_from_model(data_types.Fuzzer)
+      if fuzzer.name and fuzzer.deleted
+  }
+
   for fuzzer_job in fuzzer_job_query:
-    job = jobs_by_name[fuzzer_job.job]
+    if fuzzer_job.deleted:
+      continue
+    job = jobs_by_name.get(fuzzer_job.job)
+    if not job:
+      logs.warning(f'Job {fuzzer_job.job} not found or deleted.')
+      continue
+    if fuzzer_job.fuzzer in deleted_fuzzers:
+      logs.warning(f'Fuzzer {fuzzer_job.fuzzer} is deleted.')
+      continue
     fuzz_task_candidate = FuzzTaskCandidate(
         job=job.name,
         project=job.project,
diff --git a/src/clusterfuzz/_internal/datastore/data_handler.py b/src/clusterfuzz/_internal/datastore/data_handler.py
index ce48a0c9f95..e19b02255ba 100644
--- a/src/clusterfuzz/_internal/datastore/data_handler.py
+++ b/src/clusterfuzz/_internal/datastore/data_handler.py
@@ -1886,10 +1886,10 @@ def get_all_fuzzer_names_including_children(include_parents=False,
 @memoize.wrap(memoize.Memcache(MEMCACHE_TTL_IN_SECONDS))
 def get_all_job_type_names(project=None):
   """Return all job type names."""
-  query = data_types.Job.query(projection=['name'])
+  query = data_types.Job.query()
   if project:
     query = query.filter(data_types.Job.project == project)
-  return sorted([job.name for job in query])
+  return sorted([job.name for job in query if not job.deleted])
 
 
 def get_coverage_information(fuzzer_name, date, create_if_needed=False):
diff --git a/src/clusterfuzz/_internal/datastore/data_types.py b/src/clusterfuzz/_internal/datastore/data_types.py
index 2d4a0686f2e..c5adf874715 100644
--- a/src/clusterfuzz/_internal/datastore/data_types.py
+++ b/src/clusterfuzz/_internal/datastore/data_types.py
@@ -280,6 +280,7 @@ class Fuzzer(Model):
       'builtin',
       'differential',
       'has_large_testcases',
+      'deleted',
   )
 
   # Created at timestamp. Not set for fuzzers created before this field was
@@ -382,6 +383,9 @@ class Fuzzer(Model):
   # on disk |FUZZ_INPUTS_DISK|, rather than smaller tmpfs one (FUZZ_INPUTS).
   has_large_testcases = ndb.BooleanProperty(default=False)
 
+  # Whether this fuzzer is soft-deleted.
+  deleted = ndb.BooleanProperty(default=False)
+
   def get_config_dict(self):
     """Returns a dict containing the required config to upload a fuzzer."""
 
@@ -1027,6 +1031,9 @@ class Job(Model):
   # value here is the subscription used for receiving reproduction updates.
   external_updates_subscription = ndb.StringProperty()
 
+  # Whether this job is soft-deleted.
+  deleted = ndb.BooleanProperty(default=False)
+
   def is_external(self):
     """Whether this job is external."""
     return (bool(self.external_reproduction_topic) or
@@ -1580,6 +1587,7 @@ class FuzzerJob(Model):
   platform = ndb.StringProperty()
   weight = ndb.FloatProperty(default=1.0)
   multiplier = ndb.FloatProperty(default=1.0)
+  deleted = ndb.BooleanProperty(default=False)
 
   @property
   def actual_weight(self):
diff --git a/src/clusterfuzz/_internal/fuzzing/fuzzer_selection.py b/src/clusterfuzz/_internal/fuzzing/fuzzer_selection.py
index 0d3edfec30b..55cd909d046 100644
--- a/src/clusterfuzz/_internal/fuzzing/fuzzer_selection.py
+++ b/src/clusterfuzz/_internal/fuzzing/fuzzer_selection.py
@@ -46,7 +46,7 @@ def update_mappings_for_fuzzer(fuzzer, mappings=None):
   if mappings:
     jobs = ndb_utils.get_all_from_query(data_types.Job.query().filter(
         data_types.Job.name.IN(mappings)))
-    jobs = {job.name: job for job in jobs}
+    jobs = {job.name: job for job in jobs if not job.deleted}
   else:
     jobs = {}
 
@@ -71,6 +71,7 @@ def update_mappings_for_fuzzer(fuzzer, mappings=None):
     mapping.fuzzer = fuzzer.name
     mapping.job = job_name
     mapping.platform = jobs[job_name].platform
+    mapping.deleted = False
     new_mappings.append(mapping)
 
   ndb_utils.put_multi(new_mappings)
@@ -162,6 +163,12 @@ def get_fuzz_task_payload(platform=None):
   if not mappings:
     return None, None
 
+  mappings = [
+      entity for entity in mappings if not getattr(entity, 'deleted', False)
+  ]
+  if not mappings:
+    return None, None
+
   selected_mappings = mappings
   # The environment variable containing a list of comma-separated jobs.
   # E.g: "libfuzzer_asan_android_host,afl_asan_android_host,..."
diff --git a/src/clusterfuzz/_internal/tests/appengine/handlers/cron/cleanup_test.py b/src/clusterfuzz/_internal/tests/appengine/handlers/cron/cleanup_test.py
index e542fd071b1..2d1ab9aa062 100644
--- a/src/clusterfuzz/_internal/tests/appengine/handlers/cron/cleanup_test.py
+++ b/src/clusterfuzz/_internal/tests/appengine/handlers/cron/cleanup_test.py
@@ -2383,3 +2383,39 @@ def test_add_different_severity(self):
     self.assertNotIn('Security_Severity-High', self.issue.labels)
     self.assertIn('Security_Severity-Medium', self.issue.labels)
     self.assertIn('different from what was assigned', result)
+
+
+@test_utils.with_cloud_emulators('datastore')
+class CleanupInvalidFuzzerJobsTest(unittest.TestCase):
+  """Tests for cleanup_invalid_fuzzer_jobs."""
+
+  def setUp(self):
+    """Set up test entities."""
+    self.fuzzer1 = data_types.Fuzzer(name='fuzzer1').put()
+    self.fuzzer2_deleted = data_types.Fuzzer(name='fuzzer2', deleted=True).put()
+    self.job1 = data_types.Job(name='job1').put()
+    self.job2_deleted = data_types.Job(name='job2', deleted=True).put()
+
+    # Valid mapping
+    self.fj_valid = data_types.FuzzerJob(fuzzer='fuzzer1', job='job1').put()
+    # Invalid: deleted fuzzer
+    self.fj_del_fuzzer = data_types.FuzzerJob(
+        fuzzer='fuzzer2', job='job1').put()
+    # Invalid: deleted job
+    self.fj_del_job = data_types.FuzzerJob(fuzzer='fuzzer1', job='job2').put()
+    # Invalid: non-existent fuzzer
+    self.fj_nonexistent = data_types.FuzzerJob(
+        fuzzer='unknown', job='job1').put()
+    # Invalid: marked deleted
+    self.fj_marked_deleted = data_types.FuzzerJob(
+        fuzzer='fuzzer1', job='job1', deleted=True).put()
+
+  def test_cleanup_invalid_fuzzer_jobs(self):
+    """Test cleanup_invalid_fuzzer_jobs deletes only invalid entities."""
+    cleanup.cleanup_invalid_fuzzer_jobs()
+
+    self.assertIsNotNone(self.fj_valid.get())
+    self.assertIsNone(self.fj_del_fuzzer.get())
+    self.assertIsNone(self.fj_del_job.get())
+    self.assertIsNone(self.fj_nonexistent.get())
+    self.assertIsNone(self.fj_marked_deleted.get())
diff --git a/src/clusterfuzz/_internal/tests/appengine/handlers/cron/schedule_fuzz_test.py b/src/clusterfuzz/_internal/tests/appengine/handlers/cron/schedule_fuzz_test.py
index 1fccc28df1c..051b64cee0f 100644
--- a/src/clusterfuzz/_internal/tests/appengine/handlers/cron/schedule_fuzz_test.py
+++ b/src/clusterfuzz/_internal/tests/appengine/handlers/cron/schedule_fuzz_test.py
@@ -194,6 +194,42 @@ def test_os_version_no_version(self):
     self.assertEqual(task.job, job_name)
     self.assertIsNone(task.extra_info.get('base_os_version'))
 
+  def test_deleted_jobs_and_fuzzers_ignored(self):
+    """Tests that soft-deleted jobs and fuzzers are not scheduled."""
+    data_types.Fuzzer(name='libFuzzer').put()
+    data_types.Fuzzer(name='deleted_fuzzer', deleted=True).put()
+
+    data_types.Job(
+        name='active_job',
+        environment_string='PROJECT_NAME = myproject',
+        platform='LINUX',
+    ).put()
+    data_types.Job(
+        name='deleted_job',
+        environment_string='PROJECT_NAME = myproject',
+        platform='LINUX',
+        deleted=True,
+    ).put()
+
+    data_types.FuzzerJob(
+        job='active_job', platform='LINUX', fuzzer='libFuzzer',
+        weight=1.0).put()
+    data_types.FuzzerJob(
+        job='active_job', platform='LINUX', fuzzer='deleted_fuzzer',
+        weight=1.0).put()
+    data_types.FuzzerJob(
+        job='deleted_job', platform='LINUX', fuzzer='libFuzzer',
+        weight=1.0).put()
+
+    data_types.OssFuzzProject(name='myproject').put()
+
+    provider = schedule_fuzz.OssfuzzFuzzTaskProvider()
+    tasks = provider.get_fuzz_tasks(num_tasks=5)
+    self.assertEqual(len(tasks), 5)
+    for task in tasks:
+      self.assertEqual(task.job, 'active_job')
+      self.assertEqual(task.argument, 'libFuzzer')
+
 
 @test_utils.with_cloud_emulators('datastore')
 class ChromeFuzzTaskSchedulerTest(unittest.TestCase):
@@ -233,3 +269,22 @@ def test_os_version_job_without_version(self):
     self._setup_chrome_entities()
     task = self._run_and_get_task()
     self.assertIsNone(task.extra_info.get('base_os_version'))
+
+  def test_deleted_job_not_scheduled(self):
+    """Tests that soft-deleted chrome jobs are not scheduled."""
+    data_types.Fuzzer(name='libFuzzer').put()
+    data_types.Job(
+        name='deleted_chrome_job',
+        project='chrome',
+        platform='LINUX',
+        deleted=True).put()
+    data_types.FuzzerJob(
+        job='deleted_chrome_job',
+        platform='LINUX',
+        fuzzer='libFuzzer',
+        weight=1.0).put()
+
+    jobs = list(data_types.Job.query())
+    provider = schedule_fuzz.ChromeFuzzTaskProvider(jobs)
+    tasks = provider.get_fuzz_tasks(num_tasks=1)
+    self.assertEqual(len(tasks), 0)
diff --git a/src/clusterfuzz/_internal/tests/appengine/handlers/fuzzers_test.py b/src/clusterfuzz/_internal/tests/appengine/handlers/fuzzers_test.py
index de764fb02d3..0b5e06cf638 100644
--- a/src/clusterfuzz/_internal/tests/appengine/handlers/fuzzers_test.py
+++ b/src/clusterfuzz/_internal/tests/appengine/handlers/fuzzers_test.py
@@ -260,3 +260,55 @@ def test_update_fuzzer_untrusted_failure_non_linux(self):
     # Verify fuzzer was not updated (remained trusted).
     fuzzer = fuzzer.key.get()
     self.assertTrue(fuzzer.trusted)
+
+
+@test_utils.with_cloud_emulators('datastore')
+class DeleteHandlerTest(unittest.TestCase):
+  """Tests for DeleteHandler."""
+
+  def setUp(self):
+    """Set up test environment and handler."""
+    test_helpers.patch(self, [
+        'libs.access.has_access',
+        'libs.auth.get_current_user',
+        'libs.helpers.get_user_email',
+    ])
+    self.mock.has_access.return_value = True
+    self.mock.get_current_user().email = 'admin@example.com'
+    self.mock.get_user_email.return_value = 'admin@example.com'
+
+    flaskapp = flask.Flask('testflask')
+    flaskapp.add_url_rule(
+        '/fuzzers/delete',
+        view_func=fuzzers.DeleteHandler.as_view('/fuzzers/delete'))
+    self.app = webtest.TestApp(flaskapp)
+
+  def test_delete_fuzzer_soft_deletes(self):
+    """Test that deleting a fuzzer marks it deleted and soft-deletes mappings."""
+    fuzzer = data_types.Fuzzer(name='my_fuzzer', jobs=['job1'])
+    fuzzer.put()
+
+    mapping = data_types.FuzzerJob(fuzzer='my_fuzzer', job='job1')
+    mapping.put()
+
+    payload = {
+        'csrf_token': form.generate_csrf_token(),
+        'key': fuzzer.key.id(),
+    }
+    resp = self.app.post_json('/fuzzers/delete', payload)
+    self.assertEqual(302, resp.status_int)
+
+    fuzzer = fuzzer.key.get()
+    self.assertTrue(fuzzer.deleted)
+
+  def test_delete_already_deleted_fuzzer_fails(self):
+    """Test that deleting an already deleted fuzzer returns 400."""
+    fuzzer = data_types.Fuzzer(name='my_fuzzer', deleted=True)
+    fuzzer.put()
+
+    payload = {
+        'csrf_token': form.generate_csrf_token(),
+        'key': fuzzer.key.id(),
+    }
+    resp = self.app.post_json('/fuzzers/delete', payload, expect_errors=True)
+    self.assertEqual(400, resp.status_int)
diff --git a/src/clusterfuzz/_internal/tests/appengine/handlers/jobs_test.py b/src/clusterfuzz/_internal/tests/appengine/handlers/jobs_test.py
index 402339d5eab..bfa274cf76c 100644
--- a/src/clusterfuzz/_internal/tests/appengine/handlers/jobs_test.py
+++ b/src/clusterfuzz/_internal/tests/appengine/handlers/jobs_test.py
@@ -240,3 +240,63 @@ def test_post_empty_fuzzers(self):
 
     self.assertEqual(200, resp.status_int)
     self.mock.update_mappings_for_job.assert_called_with(mock.ANY, [])
+
+
+@test_utils.with_cloud_emulators('datastore')
+class DeleteJobHandlerTest(unittest.TestCase):
+  """Tests for DeleteJobHandler."""
+
+  def setUp(self):
+    """Set up test environment and handler."""
+    test_helpers.patch(self, [
+        'libs.access.has_access',
+        'libs.auth.get_current_user',
+        'libs.helpers.get_user_email',
+    ])
+    self.mock.has_access.return_value = True
+    self.mock.get_current_user().email = 'admin@example.com'
+    self.mock.get_user_email.return_value = 'admin@example.com'
+
+    flaskapp = flask.Flask('testflask')
+    flaskapp.add_url_rule(
+        '/jobs/delete', view_func=jobs.DeleteJobHandler.as_view('/jobs/delete'))
+    self.app = webtest.TestApp(flaskapp)
+
+  def test_delete_job_soft_deletes(self):
+    """Test that deleting a job marks it deleted and soft-deletes mappings."""
+    job = data_types.Job(name='my_job')
+    job.put()
+
+    fuzzer = data_types.Fuzzer(name='my_fuzzer', jobs=['my_job'])
+    fuzzer.put()
+
+    mapping = data_types.FuzzerJob(fuzzer='my_fuzzer', job='my_job')
+    mapping.put()
+
+    payload = {
+        'csrf_token': form.generate_csrf_token(),
+        'key': job.key.id(),
+    }
+    resp = self.app.post_json('/jobs/delete', payload)
+    self.assertEqual(302, resp.status_int)
+
+    job = job.key.get()
+    self.assertTrue(job.deleted)
+
+    mapping = mapping.key.get()
+    self.assertTrue(mapping.deleted)
+
+    fuzzer = fuzzer.key.get()
+    self.assertNotIn('my_job', fuzzer.jobs)
+
+  def test_delete_already_deleted_job_fails(self):
+    """Test that deleting an already deleted job returns 400."""
+    job = data_types.Job(name='my_job', deleted=True)
+    job.put()
+
+    payload = {
+        'csrf_token': form.generate_csrf_token(),
+        'key': job.key.id(),
+    }
+    resp = self.app.post_json('/jobs/delete', payload, expect_errors=True)
+    self.assertEqual(400, resp.status_int)
diff --git a/src/clusterfuzz/_internal/tests/core/bot/tasks/commands_test.py b/src/clusterfuzz/_internal/tests/core/bot/tasks/commands_test.py
index 8968cef7129..db19ae1aa5b 100644
--- a/src/clusterfuzz/_internal/tests/core/bot/tasks/commands_test.py
+++ b/src/clusterfuzz/_internal/tests/core/bot/tasks/commands_test.py
@@ -257,3 +257,45 @@ def test_timeout_overrides(self):
         'FUZZ_TEST_TIMEOUT = 123\nMAX_TESTCASES = 5\n')
     self.assertEqual(9001, environment.get_value('FUZZ_TEST_TIMEOUT'))
     self.assertEqual(42, environment.get_value('MAX_TESTCASES'))
+
+
+@test_utils.with_cloud_emulators('datastore')
+class ProcessCommandImplPoisonedTest(unittest.TestCase):
+  """Tests for process_command_impl dropping poisoned messages."""
+
+  def setUp(self):
+    """Set up test environment."""
+    helpers.patch_environ(self)
+
+  def test_job_missing(self):
+    """Test process_command_impl returns None when job does not exist."""
+    result = commands.process_command_impl('fuzz', 'fuzzer', 'nonexistent_job',
+                                           False, False)
+    self.assertIsNone(result)
+
+  def test_job_deleted(self):
+    """Test process_command_impl returns None when job is deleted."""
+    data_types.Job(name='deleted_job', platform='LINUX', deleted=True).put()
+    data_types.Fuzzer(name='fuzzer').put()
+    data_types.FuzzerJob(fuzzer='fuzzer', job='deleted_job').put()
+    result = commands.process_command_impl('fuzz', 'fuzzer', 'deleted_job',
+                                           False, False)
+    self.assertIsNone(result)
+
+  def test_fuzzer_deleted(self):
+    """Test process_command_impl returns None when fuzzer is deleted."""
+    data_types.Job(name='job', platform='LINUX').put()
+    data_types.Fuzzer(name='fuzzer', deleted=True).put()
+    data_types.FuzzerJob(fuzzer='fuzzer', job='job').put()
+    result = commands.process_command_impl('fuzz', 'fuzzer', 'job', False,
+                                           False)
+    self.assertIsNone(result)
+
+  def test_fuzzer_job_deleted(self):
+    """Test process_command_impl returns None when fuzzer_job is deleted."""
+    data_types.Job(name='job', platform='LINUX').put()
+    data_types.Fuzzer(name='fuzzer').put()
+    data_types.FuzzerJob(fuzzer='fuzzer', job='job', deleted=True).put()
+    result = commands.process_command_impl('fuzz', 'fuzzer', 'job', False,
+                                           False)
+    self.assertIsNone(result)
diff --git a/src/clusterfuzz/_internal/tests/core/bot/tasks/utasks/corpus_pruning_task_test.py b/src/clusterfuzz/_internal/tests/core/bot/tasks/utasks/corpus_pruning_task_test.py
index fcb5495c762..3924786dc04 100644
--- a/src/clusterfuzz/_internal/tests/core/bot/tasks/utasks/corpus_pruning_task_test.py
+++ b/src/clusterfuzz/_internal/tests/core/bot/tasks/utasks/corpus_pruning_task_test.py
@@ -584,3 +584,17 @@ def tearDown(self):
     shutil.rmtree('a')
     shutil.rmtree('c')
     shutil.rmtree(self.temp_dir)
+
+
+@test_utils.with_cloud_emulators('datastore')
+class CorpusPruningPreprocessPoisonedTest(unittest.TestCase):
+  """Tests for corpus pruning preprocess with missing entities."""
+
+  def setUp(self):
+    """Set up test environment."""
+    helpers.patch_environ(self)
+
+  def test_preprocess_fuzz_target_missing(self):
+    """Test utask_preprocess returns None when fuzz target is not found."""
+    result = corpus_pruning_task.utask_preprocess('nonexistent', 'job', {})
+    self.assertIsNone(result)
diff --git a/src/clusterfuzz/_internal/tests/core/datastore/data_handler_test.py b/src/clusterfuzz/_internal/tests/core/datastore/data_handler_test.py
index bcb1adc43a3..dfbaa9ab8b9 100644
--- a/src/clusterfuzz/_internal/tests/core/datastore/data_handler_test.py
+++ b/src/clusterfuzz/_internal/tests/core/datastore/data_handler_test.py
@@ -1246,3 +1246,21 @@ def test_uworker_bypass(self):
     self.assertEqual(result, 'env_value')
     self.mock.get_value.assert_called_once_with('VAR_PATTERN', 'default_val')
     self.mock.query.assert_not_called()
+
+
+@test_utils.with_cloud_emulators('datastore')
+class GetJobsSoftDeleteTest(unittest.TestCase):
+  """Tests for get_all_job_type_names with soft-delete."""
+
+  def setUp(self):
+    """Set up test entities."""
+    helpers.patch_environ(self)
+
+  def test_get_all_job_type_names_excludes_deleted(self):
+    """Test get_all_job_type_names excludes soft-deleted jobs."""
+    data_types.Job(name='active_job').put()
+    data_types.Job(name='deleted_job', deleted=True).put()
+
+    jobs = data_handler.get_all_job_type_names(__memoize_force__=True)
+    self.assertIn('active_job', jobs)
+    self.assertNotIn('deleted_job', jobs)
diff --git a/src/clusterfuzz/_internal/tests/core/fuzzing/fuzzer_selection_test.py b/src/clusterfuzz/_internal/tests/core/fuzzing/fuzzer_selection_test.py
index 4859eb12db6..f0f44470862 100644
--- a/src/clusterfuzz/_internal/tests/core/fuzzing/fuzzer_selection_test.py
+++ b/src/clusterfuzz/_internal/tests/core/fuzzing/fuzzer_selection_test.py
@@ -147,6 +147,21 @@ def test_nonexistent_job(self):
     mappings = _get_job_list_for_fuzzer(fuzzer)
     self.assertCountEqual(['job_1'], mappings)
 
+  def test_soft_deleted_job(self):
+    """Ensure soft-deleted job references are cleaned up from fuzzer.jobs."""
+    data_types.Job(name='deleted_job', deleted=True).put()
+    fuzzer = data_types.Fuzzer(
+        name='soft_delete_fuzzer', jobs=['deleted_job', 'job_1'])
+    fuzzer.put()
+
+    fuzzer_selection.update_mappings_for_fuzzer(fuzzer)
+
+    fuzzer = fuzzer.key.get()
+    self.assertCountEqual(['job_1'], fuzzer.jobs)
+
+    mappings = _get_job_list_for_fuzzer(fuzzer)
+    self.assertCountEqual(['job_1'], mappings)
+
 
 @test_utils.with_cloud_emulators('datastore')
 class UpdateMappingsForJobTest(unittest.TestCase):