Fix maintenance deployment locking and failure recovery

Hold the instance lock across CLS cleanup, migration and deployment, and persist maintenance phases for safe retries. Keep stopped deployments stopped during recovery and unpause the trader after preparation failures.

scripts/deploy.py

@@ -153,82 +153,98 @@ def deploy(build, directory=INSTANCE):
     directory = Path(directory).resolve(strict=True)
     require(directory == INSTANCE, 'Unexpected instance directory')
     with locked(directory / '.deploy.lock'):
-        require(not (directory / 'deployment.pending.json').exists(), 'Resolve the previous pending deployment with recover first')
-        if branch_head() != build['commit']:
-            print('Skipped: dev has advanced')
-            return
-        validate_images(build)
-        run('docker', 'pull', SEARXNG_IMAGE)
-        validate_search_image()
-        # Only this read-only schema guard receives the existing application environment.
-        print(run('docker', 'run', '--rm', '-i', '--network', 'dev_default', '--env-file', directory / '.env',
-                  IMAGES['backend'] + ':' + build['version'], 'python', '-', input=SCHEMA), flush=True)
-        previous_file = directory / 'deployment.json'
-        previous = json.loads(previous_file.read_text()) if previous_file.exists() else None
-        old_compose = (directory / 'compose.yml').read_text() if previous else None
-        previous_running = []
-        if previous:
-            old_command = compose(directory, previous)
-            containers = inspect_services(old_command)
-            require(not any(c['State'].get('Paused') for c in containers), 'Recover paused services before deploying')
-            previous_running = [c['Config']['Labels']['com.docker.compose.service'] for c in containers if c['State']['Running']]
-            for _ in range(60) if previous_running else range(0):
-                idle = json.loads(run('docker', 'exec', '-i', 'AlphaPulse-dev-api', 'python', '-', input=IDLE))
-                if not idle['busy']:
-                    break
-                time.sleep(5)
-            else:
-                require(not previous_running, 'Chat or trading analysis did not become idle within five minutes')
-        # Recheck after pulls and waiting; only a current dev commit can mutate the instance.
-        if branch_head() != build['commit']:
-            print('Skipped: dev advanced while preparing deployment')
-            return
-        install_entrypoints(directory)
-        pending = {'build': build, 'previous': previous, 'previous_compose': old_compose,
-                   'previous_running': previous_running, 'started_at': time.time(), 'phase': 'preparing'}
-        save(directory / 'deployment.pending.json', pending)
-        try:
-            if previous and 'trader' in previous_running:
-                run('docker', 'pause', 'AlphaPulse-dev-trader')
-                idle = json.loads(run('docker', 'exec', '-i', 'AlphaPulse-dev-api', 'python', '-', input=IDLE))
-                require(not idle['busy'], 'Analysis became active during preparation; retry after it completes')
-            # Persist the transition before stopping any existing service.
-            save(directory / 'deployment.pending.json', {**pending, 'phase': 'updating'})
-            pending['phase'] = 'updating'
-            if previous:
-                # Docker unpauses a paused container before delivering the stop signal.
-                run(*old_command, 'stop', 'trader')
-                run(*old_command, 'stop', *[service for service in previous_running if service != 'trader'])
-            (directory / 'compose.yml').write_text((ROOT / 'deploy/docker-compose.yml').read_text())
-            search_config = directory / 'searxng.yml'
-            if not search_config.exists():
-                search_config.write_text((ROOT / 'deploy/searxng.yml').read_text().replace('@SEARXNG_SECRET@', secrets.token_hex(32)))
-                search_config.chmod(0o644)
-            command = compose(directory, build)
-            run(*command, 'config', '--quiet')
-            run(*command, 'up', '-d', '--no-build', '--pull', 'never', '--wait', '--wait-timeout', '120', 'api', 'web')
-            run(*command, 'up', '-d', '--no-build', '--pull', 'never', 'market', 'crawler', 'calendar', 'trader', 'news-risk', 'jin10-calendar', 'searxng')
+        deploy_locked(build, directory)
+
+
+def deploy_locked(build, directory, maintenance=None):
+    """Caller holds the instance lock throughout preparation and activation."""
+    require(maintenance is not None or not (directory / 'deployment.pending.json').exists(),
+            'Resolve the previous pending deployment with recover first')
+    current_head = branch_head()
+    if maintenance is not None:
+        require(current_head == build['commit'], 'Maintenance build is stale; services remain stopped')
+    if current_head != build['commit']:
+        print('Skipped: dev has advanced')
+        return
+    validate_images(build)
+    run('docker', 'pull', SEARXNG_IMAGE)
+    validate_search_image()
+    # Only this read-only schema guard receives the existing application environment.
+    print(run('docker', 'run', '--rm', '-i', '--network', 'dev_default', '--env-file', directory / '.env',
+              IMAGES['backend'] + ':' + build['version'], 'python', '-', input=SCHEMA), flush=True)
+    previous_file = directory / 'deployment.json'
+    previous = json.loads(previous_file.read_text()) if previous_file.exists() else None
+    old_compose = (directory / 'compose.yml').read_text() if previous else None
+    previous_running = []
+    if previous:
+        old_command = compose(directory, previous)
+        containers = inspect_services(old_command)
+        require(not any(c['State'].get('Paused') for c in containers), 'Recover paused services before deploying')
+        previous_running = [c['Config']['Labels']['com.docker.compose.service'] for c in containers if c['State']['Running']]
+        for _ in range(60) if previous_running else range(0):
+            idle = json.loads(run('docker', 'exec', '-i', 'AlphaPulse-dev-api', 'python', '-', input=IDLE))
+            if not idle['busy']:
+                break
             time.sleep(5)
-            health(command, build)
-            if previous:
-                save(directory / 'deployment.previous.json', previous)
-                (directory / 'compose.previous.yml').write_text(old_compose)
-            save(previous_file, build)
-            (directory / 'deployment.pending.json').unlink()
-        except BaseException:
-            if pending['phase'] == 'preparing':
-                restore(directory, pending)
-                raise
-            # Never roll back or reset business data. A pending marker blocks subsequent jobs.
-            subprocess.run(['docker', 'compose', '--project-name', 'alphapulse-dev', '--env-file', str(directory / 'images.env'),
-                            '-f', str(directory / 'compose.yml'), 'stop'], capture_output=True)
-            if previous:
-                restore(directory, pending)
+        else:
+            require(not previous_running, 'Chat or trading analysis did not become idle within five minutes')
+    # Recheck after pulls and waiting; only a current dev commit can mutate the instance.
+    current_head = branch_head()
+    if maintenance is not None:
+        require(current_head == build['commit'], 'Maintenance build is stale; services remain stopped')
+    if current_head != build['commit']:
+        print('Skipped: dev advanced while preparing deployment')
+        return
+    install_entrypoints(directory)
+    pending = {'build': build, 'previous': previous, 'previous_compose': old_compose,
+               'previous_running': previous_running, 'started_at': time.time(), 'phase': 'preparing'}
+    if maintenance is not None:
+        pending.update(operation='remove-cls', phase='updating')
+    save(directory / 'deployment.pending.json', pending)
+    try:
+        if previous and 'trader' in previous_running:
+            run('docker', 'pause', 'AlphaPulse-dev-trader')
+            idle = json.loads(run('docker', 'exec', '-i', 'AlphaPulse-dev-api', 'python', '-', input=IDLE))
+            require(not idle['busy'], 'Analysis became active during preparation; retry after it completes')
+        # Persist the transition before stopping any existing service.
+        save(directory / 'deployment.pending.json', {**pending, 'phase': 'updating'})
+        pending['phase'] = 'updating'
+        if previous:
+            # Docker unpauses a paused container before delivering the stop signal.
+            run(*old_command, 'stop', 'trader')
+            run(*old_command, 'stop', *[service for service in previous_running if service != 'trader'])
+        (directory / 'compose.yml').write_text((ROOT / 'deploy/docker-compose.yml').read_text())
+        search_config = directory / 'searxng.yml'
+        if not search_config.exists():
+            search_config.write_text((ROOT / 'deploy/searxng.yml').read_text().replace('@SEARXNG_SECRET@', secrets.token_hex(32)))
+            search_config.chmod(0o644)
+        command = compose(directory, build)
+        run(*command, 'config', '--quiet')
+        run(*command, 'up', '-d', '--no-build', '--pull', 'never', '--wait', '--wait-timeout', '120', 'api', 'web')
+        run(*command, 'up', '-d', '--no-build', '--pull', 'never', 'market', 'crawler', 'calendar', 'trader', 'news-risk', 'jin10-calendar', 'searxng')
+        time.sleep(5)
+        health(command, build)
+        if previous:
+            save(directory / 'deployment.previous.json', previous)
+            (directory / 'compose.previous.yml').write_text(old_compose)
+        save(previous_file, build)
+        (directory / 'deployment.pending.json').unlink()
+    except BaseException:
+        if pending['phase'] == 'preparing':
+            restore(directory, pending)
             raise
-        print('All application services now run the accepted development version tags', flush=True)
+        # Never roll back or reset business data. A pending marker blocks subsequent jobs.
+        subprocess.run(['docker', 'compose', '--project-name', 'alphapulse-dev', '--env-file', str(directory / 'images.env'),
+                        '-f', str(directory / 'compose.yml'), 'stop'], capture_output=True)
+        if previous and maintenance is None:
+            restore(directory, pending)
+        raise
+    print('All application services now run the accepted development version tags', flush=True)
 
 
 def restore(directory, pending):
+    require(pending.get('operation') != 'remove-cls' or pending.get('phase') == 'preparing',
+            'CLS maintenance is incomplete; retry remove-cls with a current accepted build; services must remain stopped')
     previous = pending['previous']
     require(previous is not None, 'No previous deployment exists; inspect the pending first-start failure')
     trader = run('docker', 'ps', '-aq', '--filter', 'name=^/AlphaPulse-dev-trader$')
@@ -249,47 +265,75 @@ def restore(directory, pending):
         validate_search_image()
     (directory / 'compose.yml').write_text(previous_compose)
     command = compose(directory, previous)
-    run(*command, 'up', '-d', '--no-build', '--pull', 'never', '--remove-orphans', '--wait', '--wait-timeout', '120', *pending['previous_running'])
+    if pending['previous_running']:
+        run(*command, 'up', '-d', '--no-build', '--pull', 'never', '--remove-orphans', '--wait', '--wait-timeout', '120', *pending['previous_running'])
+    else:
+        run(*command, 'stop')
     previous_services = re.findall(r'^  ([a-z][a-z0-9-]*):$', pending['previous_compose'], re.M)
-    if set(pending['previous_running']) == set(previous_services):
+    if pending['previous_running'] and set(pending['previous_running']) == set(previous_services):
         health(command, previous, previous_services)
     save(directory / 'deployment.json', previous)
     (directory / 'deployment.pending.json').unlink()
     print('Restored previous images and configuration; business data was retained', flush=True)
 
 
-
 def remove_cls(build, execute=False):
-    """Retire CLS and reset all news AI using an accepted image and the instance lock."""
+    """Retire CLS under one lock, retaining a durable marker until activation succeeds."""
     with locked(INSTANCE / '.deploy.lock'):
-        require(not (INSTANCE / 'deployment.pending.json').exists(), 'Recover the pending deployment first')
+        marker = INSTANCE / 'deployment.pending.json'
+        pending = json.loads(marker.read_text()) if marker.exists() else None
+        require(pending is None or (execute and pending.get('operation') == 'remove-cls'),
+                'Recover the pending deployment first')
         require(branch_head() == build['commit'], 'Expected the current dev build')
         validate_images(build)
         command = ['docker', 'run', '--rm', '-i', '--network', 'dev_default', '--env-file', str(INSTANCE / '.env'),
                    IMAGES['backend'] + ':' + build['version']]
-        if execute:
-            current = json.loads((INSTANCE / 'deployment.json').read_text())
-            old_command = compose(INSTANCE, current)
-            running = any(container['State']['Running'] for container in inspect_services(old_command))
-            if running:
-                for _ in range(60):
+        if not execute:
+            print(run(*command, 'python', '-m', 'alphapulse.crawler.remove_cls'), flush=True)
+            return
+        current = json.loads((INSTANCE / 'deployment.json').read_text())
+        old_command = compose(INSTANCE, current)
+        if pending is None:
+            pending = {'operation': 'remove-cls', 'build': build, 'previous': current,
+                       'phase': 'preparing', 'started_at': time.time()}
+            save(marker, pending)
+        elif pending['build'] != build:
+            # A newer accepted dev build may resume completed destructive stages.
+            pending['build'] = build
+            save(marker, pending)
+        if pending['phase'] == 'preparing':
+            # Preparation recovery only unpauses; destructive stages always remain stopped.
+            try:
+                running = any(container['State']['Running'] for container in inspect_services(old_command))
+                if running:
+                    for _ in range(60):
+                        idle = json.loads(run('docker', 'exec', '-i', 'AlphaPulse-dev-api', 'python', '-', input=IDLE))
+                        if not idle['busy']:
+                            break
+                        time.sleep(5)
+                    else:
+                        raise RuntimeError('Chat or trading analysis remained busy')
+                    run('docker', 'pause', 'AlphaPulse-dev-trader')
                     idle = json.loads(run('docker', 'exec', '-i', 'AlphaPulse-dev-api', 'python', '-', input=IDLE))
-                    if not idle['busy']:
-                        break
-                    time.sleep(5)
-                else:
-                    raise RuntimeError('Chat or trading analysis remained busy')
-                run('docker', 'pause', 'AlphaPulse-dev-trader')
-                idle = json.loads(run('docker', 'exec', '-i', 'AlphaPulse-dev-api', 'python', '-', input=IDLE))
-                if idle['busy']:
-                    run('docker', 'unpause', 'AlphaPulse-dev-trader')
-                    raise RuntimeError('Analysis became active; retry after it completes')
-            run(*old_command, 'stop')
-        print(run(*command, 'python', '-m', 'alphapulse.crawler.remove_cls', *(['--execute'] if execute else [])), flush=True)
-        if execute:
+                    require(not idle['busy'], 'Analysis became active; retry after it completes')
+                save(marker, {**pending, 'phase': 'stopping'})
+            except BaseException:
+                restore(INSTANCE, pending)
+                raise
+            pending['phase'] = 'stopping'
+        # Also stop partially activated candidate services when resuming a failed deployment.
+        run(*old_command, 'stop')
+        if pending['phase'] in ('stopping', 'cleanup'):
+            pending['phase'] = 'cleanup'
+            save(marker, pending)
+            print(run(*command, 'python', '-m', 'alphapulse.crawler.remove_cls', '--execute'), flush=True)
+            pending['phase'] = 'migration'
+            save(marker, pending)
+        if pending['phase'] == 'migration':
             print(run(*command, 'alembic', 'upgrade', 'head'), flush=True)
-    if execute:
-        deploy(build)
+            pending['phase'] = 'deploy-ready'
+            save(marker, pending)
+        deploy_locked(build, INSTANCE, maintenance=pending)
 
 
 def reset_news_ai(build, confirmed=False):

scripts/tests/test_cicd.py

@@ -178,6 +178,20 @@ class NewsResetTests(unittest.TestCase):
 
 
 class RecoveryTests(unittest.TestCase):
+    def test_empty_previous_running_list_keeps_all_services_stopped(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            directory = Path(temporary)
+            ci.save(directory / 'deployment.pending.json', {})
+            previous = {'version': '20261101-aaaaaaaa'}
+            compose = 'services:\n  api:\n  news-labeling:\n'
+            with patch.object(deploy, 'run', return_value='') as run, \
+                    patch.object(deploy, 'validate_images'), patch.object(deploy, 'health') as health:
+                deploy.restore(directory, {'previous': previous, 'previous_compose': compose, 'previous_running': []})
+                self.assertFalse(any('up' in call.args for call in run.call_args_list))
+                self.assertTrue(any('stop' in call.args for call in run.call_args_list))
+                health.assert_not_called()
+            self.assertFalse((directory / 'deployment.pending.json').exists())
+
     def test_rollback_removes_new_calendar_worker_and_checks_previous_services(self):
         with tempfile.TemporaryDirectory() as temporary:
             directory = Path(temporary)

scripts/tests/test_instance.py

@@ -137,7 +137,7 @@ class EntrypointTests(unittest.TestCase):
 
     def test_pending_and_shared_lock(self):
         pending = self.directory / 'deployment.pending.json'
-        pending.write_text('{}')
+        pending.write_text('{"operation":"remove-cls","phase":"cleanup"}')
         self.assertEqual(self.invoke(code=1), [])
         calls = self.invoke('stop')
         self.assertEqual(calls[-1]['args'][-1], 'stop')

scripts/tests/test_remove_cls.py

@@ -1,62 +1,259 @@
 """Exercise retirement orchestration with synthetic configuration only."""
+import fcntl
 import json
 from pathlib import Path
+import sys
 import tempfile
 import unittest
 from unittest.mock import patch
 
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
 import deploy
 
 
 class RemoveClsTests(unittest.TestCase):
-    def exercise(self, execute, failure=False, running=True):
-        with tempfile.TemporaryDirectory() as temporary:
-            directory = Path(temporary)
-            build = {'commit': 'a'*40, 'version': '20261104-aaaaaaaa'}
-            (directory / 'deployment.json').write_text(json.dumps(build))
-            calls = []
+    def setUp(self):
+        temporary = tempfile.TemporaryDirectory()
+        self.addCleanup(temporary.cleanup)
+        self.directory = Path(temporary.name)
+        self.build = {'commit': 'a' * 40, 'version': '20261104-aaaaaaaa'}
+        self.previous = {'commit': 'b' * 40, 'version': '20261103-bbbbbbbb'}
+        self.marker = self.directory / 'deployment.pending.json'
+        deploy.save(self.directory / 'deployment.json', self.previous)
+        (self.directory / 'compose.yml').write_text('services:\n  api:\n  trader:\n  news-labeling:\n')
+        source = self.directory / 'source' / 'deploy'
+        source.mkdir(parents=True)
+        (source / 'docker-compose.yml').write_text('services:\n  api:\n  trader:\n')
+        (source / 'searxng.yml').write_text('secret: @SEARXNG_SECRET@\n')
+        self.calls = []
+        self.running = True
+        self.paused = False
+        self.failure = lambda args, kwargs: None
+        self.enterContext(patch.object(deploy, 'INSTANCE', self.directory))
+        self.enterContext(patch.object(deploy, 'ROOT', source.parent))
+        self.enterContext(patch.object(deploy, 'validate_images'))
+        self.enterContext(patch.object(deploy, 'validate_search_image'))
+        self.enterContext(patch.object(deploy, 'install_entrypoints'))
+        self.lock = self.enterContext(patch.object(deploy, 'locked', wraps=deploy.locked))
+        self.enterContext(patch.object(deploy.time, 'sleep'))
+        self.head = self.enterContext(patch.object(deploy, 'branch_head', return_value=self.build['commit']))
+        self.health = self.enterContext(patch.object(deploy, 'health'))
+        self.enterContext(patch.object(deploy, 'inspect_services', side_effect=self.containers))
+        self.enterContext(patch.object(deploy, 'run', side_effect=self.command))
+        # Failure cleanup bypasses run(); it must also stay inside this fixture.
+        self.enterContext(patch.object(deploy.subprocess, 'run', side_effect=lambda args, **kwargs: self.command(*args)))
 
-            def command(*args, **kwargs):
-                calls.append(args)
-                if kwargs.get('input') == deploy.IDLE:
-                    return '{"busy":0}'
-                if failure and '--execute' in args:
-                    raise RuntimeError('cleanup failed')
-                return '{}'
+    def containers(self, command):
+        return [{'Config': {'Labels': {'com.docker.compose.service': name}},
+                 'State': {'Running': self.running, 'Paused': self.paused if name == 'trader' else False}}
+                for name in ('api', 'trader')]
 
-            with patch.object(deploy, 'INSTANCE', directory), patch.object(deploy, 'validate_images'), \
-                    patch.object(deploy, 'branch_head', return_value=build['commit']), \
-                    patch.object(deploy, 'inspect_services', return_value=[{'State': {'Running': running}}]), \
-                    patch.object(deploy, 'run', side_effect=command), patch.object(deploy, 'deploy') as update:
-                if failure:
-                    with self.assertRaisesRegex(RuntimeError, 'cleanup failed'):
-                        deploy.remove_cls(build, execute)
-                    update.assert_not_called()
-                else:
-                    deploy.remove_cls(build, execute)
-                    self.assertEqual(update.call_count, int(execute))
-            return calls
+    def assert_locked(self):
+        with (self.directory / '.deploy.lock').open('a') as stream:
+            with self.assertRaises(BlockingIOError):
+                fcntl.flock(stream, fcntl.LOCK_EX | fcntl.LOCK_NB)
+
+    def command(self, *args, **kwargs):
+        self.calls.append(args)
+        if '--execute' in args or 'alembic' in args or 'up' in args:
+            self.assert_locked()
+            self.assertTrue(self.marker.exists())
+        self.failure(args, kwargs)
+        if kwargs.get('input') == deploy.IDLE:
+            return '{"busy":0}'
+        if args[:2] == ('docker', 'pause'):
+            self.assertEqual(json.loads(self.marker.read_text())['phase'], 'preparing')
+            self.paused = True
+        if args[:2] == ('docker', 'ps'):
+            return 'trader-id' if self.running else ''
+        if args[:2] == ('docker', 'inspect'):
+            return json.dumps([{'State': {'Paused': self.paused}}])
+        if args[:2] == ('docker', 'unpause'):
+            self.paused = False
+        if 'stop' in args:
+            self.running = self.paused = False
+        if 'up' in args:
+            self.running = True
+        return '{}'
+
+    def fail_command(self, target, message):
+        def fail(args, kwargs):
+            if target in args:
+                raise RuntimeError(message)
+        self.failure = fail
 
     def test_preview_never_stops_migrates_or_deploys(self):
-        calls = self.exercise(False)
-        self.assertEqual(len(calls), 1)
-        self.assertNotIn('--execute', calls[0])
-        self.assertNotIn('alembic', calls[0])
+        deploy.remove_cls(self.build)
+        self.assertEqual(len(self.calls), 1)
+        self.assertNotIn('--execute', self.calls[0])
+        self.assertNotIn('alembic', self.calls[0])
+        self.assertFalse(self.marker.exists())
+        self.health.assert_not_called()
 
-    def test_execute_stops_before_cleanup_and_migration(self):
-        calls = self.exercise(True)
-        stop = next(i for i, args in enumerate(calls) if 'stop' in args)
-        cleanup = next(i for i, args in enumerate(calls) if '--execute' in args)
-        migration = next(i for i, args in enumerate(calls) if 'alembic' in args)
+    def test_execute_holds_lock_through_cleanup_migration_and_activation(self):
+        deploy.remove_cls(self.build, True)
+        stop = next(i for i, args in enumerate(self.calls) if 'stop' in args)
+        cleanup = next(i for i, args in enumerate(self.calls) if '--execute' in args)
+        migration = next(i for i, args in enumerate(self.calls) if 'alembic' in args)
+        activation = next(i for i, args in enumerate(self.calls) if 'up' in args)
         self.assertLess(stop, cleanup)
         self.assertLess(cleanup, migration)
+        self.assertLess(migration, activation)
+        self.assertEqual(self.lock.call_count, 1)
+        self.assertFalse(self.marker.exists())
+        self.assertEqual(json.loads((self.directory / 'deployment.json').read_text()), self.build)
+        self.health.assert_called_once()
 
-    def test_cleanup_failure_leaves_writers_stopped(self):
-        calls = self.exercise(True, True)
-        self.assertTrue(any('stop' in args for args in calls))
-        self.assertFalse(any('up' in args or 'alembic' in args for args in calls))
+    def test_cleanup_failure_blocks_update_recovery_and_keeps_writers_stopped(self):
+        self.fail_command('--execute', 'cleanup failed')
+        with self.assertRaisesRegex(RuntimeError, 'cleanup failed'):
+            deploy.remove_cls(self.build, True)
+        self.assertFalse(self.running)
+        self.assertFalse(any('up' in args or 'alembic' in args for args in self.calls))
+        pending = json.loads(self.marker.read_text())
+        self.assertEqual(pending['phase'], 'cleanup')
+        count = len(self.calls)
+        with self.assertRaisesRegex(RuntimeError, 'pending deployment'):
+            deploy.deploy(self.build, self.directory)
+        with patch.object(sys, 'argv', ['deploy.py', 'rollback']):
+            with self.assertRaisesRegex(RuntimeError, 'pending deployment'):
+                deploy.main()
+        with self.assertRaisesRegex(RuntimeError, 'CLS maintenance is incomplete'):
+            deploy.restore(self.directory, pending)
+        self.assertEqual(len(self.calls), count)
+        self.assertTrue(self.marker.exists())
+
+    def test_cleanup_failure_can_resume_under_the_same_lock(self):
+        self.fail_command('--execute', 'cleanup failed')
+        with self.assertRaises(RuntimeError):
+            deploy.remove_cls(self.build, True)
+        self.failure = lambda args, kwargs: None
+        deploy.remove_cls(self.build, True)
+        self.assertFalse(self.marker.exists())
+        self.assertEqual(sum('--execute' in args for args in self.calls), 2)
+        self.health.assert_called_once()
+
+    def test_migration_retry_does_not_repeat_successful_cleanup(self):
+        self.fail_command('alembic', 'migration failed')
+        with self.assertRaisesRegex(RuntimeError, 'migration failed'):
+            deploy.remove_cls(self.build, True)
+        self.assertEqual(json.loads(self.marker.read_text())['phase'], 'migration')
+        self.assertFalse(self.running)
+        self.failure = lambda args, kwargs: None
+        deploy.remove_cls(self.build, True)
+        self.assertEqual(sum('--execute' in args for args in self.calls), 1)
+        self.assertEqual(sum('alembic' in args for args in self.calls), 2)
+        self.assertFalse(self.marker.exists())
+
+    def test_failed_activation_stops_candidates_without_restoring_old_workers(self):
+        self.health.side_effect = RuntimeError('health failed')
+        with self.assertRaisesRegex(RuntimeError, 'health failed'):
+            deploy.remove_cls(self.build, True)
+        self.assertFalse(self.running)
+        pending = json.loads(self.marker.read_text())
+        self.assertEqual(pending['operation'], 'remove-cls')
+        self.assertEqual(pending['phase'], 'updating')
+        self.assertEqual(json.loads((self.directory / 'deployment.json').read_text()), self.previous)
+        last_up = max(i for i, args in enumerate(self.calls) if 'up' in args)
+        self.assertIn('stop', self.calls[-1])
+        self.assertFalse(any('up' in args for args in self.calls[last_up + 1:]))
+        self.health.side_effect = None
+        deploy.remove_cls(self.build, True)
+        self.assertEqual(sum('--execute' in args for args in self.calls), 1)
+        self.assertEqual(sum('alembic' in args for args in self.calls), 1)
+        self.assertFalse(self.marker.exists())
 
     def test_execute_resumes_with_all_writers_stopped(self):
-        calls = self.exercise(True, running=False)
-        self.assertFalse(any(args[:2] == ('docker', 'pause') for args in calls))
-        self.assertTrue(any('--execute' in args for args in calls))
+        self.running = False
+        deploy.remove_cls(self.build, True)
+        self.assertFalse(any(args[:2] == ('docker', 'pause') for args in self.calls))
+        self.assertTrue(any('--execute' in args for args in self.calls))
+        self.assertFalse(self.marker.exists())
+
+    def test_idle_recheck_exception_unpauses_without_stopping_services(self):
+        checks = 0
+
+        def fail(args, kwargs):
+            nonlocal checks
+            if kwargs.get('input') == deploy.IDLE:
+                checks += 1
+                if checks == 2:
+                    raise RuntimeError('idle check failed')
+        self.failure = fail
+        with self.assertRaisesRegex(RuntimeError, 'idle check failed'):
+            deploy.remove_cls(self.build, True)
+        self.assertTrue(self.running)
+        self.assertFalse(self.paused)
+        self.assertFalse(self.marker.exists())
+        self.assertFalse(any('stop' in args or '--execute' in args for args in self.calls))
+
+    def test_failed_preparation_recovery_preserves_marker_for_explicit_recovery(self):
+        def fail(args, kwargs):
+            if kwargs.get('input') == deploy.IDLE and self.paused:
+                raise RuntimeError('idle check failed')
+            if args[:2] == ('docker', 'unpause'):
+                raise RuntimeError('unpause failed')
+        self.failure = fail
+        with self.assertRaisesRegex(RuntimeError, 'unpause failed'):
+            deploy.remove_cls(self.build, True)
+        self.assertTrue(self.paused)
+        pending = json.loads(self.marker.read_text())
+        self.assertEqual(pending['phase'], 'preparing')
+        self.failure = lambda args, kwargs: None
+        with deploy.locked(self.directory / '.deploy.lock'):
+            deploy.restore(self.directory, pending)
+        self.assertFalse(self.marker.exists())
+        self.assertFalse(self.paused)
+        self.assertTrue(self.running)
+
+    def test_failed_transition_write_unpauses_before_any_stop(self):
+        save = deploy.save
+
+        def fail(path, value):
+            if value.get('phase') == 'stopping':
+                raise OSError('disk failure')
+            save(path, value)
+        with patch.object(deploy, 'save', side_effect=fail):
+            with self.assertRaisesRegex(OSError, 'disk failure'):
+                deploy.remove_cls(self.build, True)
+        self.assertFalse(self.paused)
+        self.assertTrue(self.running)
+        self.assertFalse(self.marker.exists())
+        self.assertFalse(any('stop' in args for args in self.calls))
+
+    def test_initial_marker_failure_never_pauses_or_stops(self):
+        with patch.object(deploy, 'save', side_effect=OSError('disk failure')):
+            with self.assertRaisesRegex(OSError, 'disk failure'):
+                deploy.remove_cls(self.build, True)
+        self.assertFalse(any('stop' in args or 'pause' in args for args in self.calls))
+
+    def test_stale_maintenance_retains_marker_and_can_resume_with_new_accepted_build(self):
+        # The branch advances after cleanup, immediately before candidate preparation.
+        self.head.side_effect = [self.build['commit'], 'c' * 40]
+        with self.assertRaisesRegex(RuntimeError, 'Maintenance build is stale'):
+            deploy.remove_cls(self.build, True)
+        self.assertFalse(self.running)
+        self.assertFalse(any('up' in args for args in self.calls))
+        self.assertEqual(json.loads(self.marker.read_text())['phase'], 'deploy-ready')
+        newer = {'commit': 'c' * 40, 'version': '20261104-cccccccc'}
+        self.head.side_effect = None
+        self.head.return_value = newer['commit']
+        deploy.remove_cls(newer, True)
+        self.assertEqual(sum('--execute' in args for args in self.calls), 1)
+        self.assertEqual(sum('alembic' in args for args in self.calls), 1)
+        self.assertFalse(self.marker.exists())
+        self.assertEqual(json.loads((self.directory / 'deployment.json').read_text()), newer)
+
+    def test_branch_advance_after_candidate_validation_never_skips_successfully(self):
+        self.head.side_effect = [self.build['commit'], self.build['commit'], 'c' * 40]
+        with self.assertRaisesRegex(RuntimeError, 'Maintenance build is stale'):
+            deploy.remove_cls(self.build, True)
+        self.assertFalse(self.running)
+        self.assertTrue(self.marker.exists())
+        self.assertFalse(any('up' in args for args in self.calls))
+
+    def test_other_pending_deployment_blocks_cleanup(self):
+        deploy.save(self.marker, {'phase': 'updating'})
+        with self.assertRaisesRegex(RuntimeError, 'Recover the pending deployment first'):
+            deploy.remove_cls(self.build, True)
+        self.assertEqual(self.calls, [])