Bound CLS cleanup time and support stopped-instance retries

backend/src/alphapulse/crawler/remove_cls.py

@@ -64,6 +64,15 @@ async def cleanup(connection, execute=False):
                    for table in SOURCE_TABLES})
     counts['news_revisions'] = await connection.scalar(text('''SELECT count(*) FROM news_revisions
         WHERE news_id IN (SELECT id FROM news_items WHERE source_code = ANY(:sources))'''), params)
+    cleanup_indexes = {
+        'cleanup_cls_revisions_raw': ('news_revisions', 'raw_response_id'),
+        'cleanup_cls_economic_raw': ('economic_revisions', 'raw_response_id'),
+        'cleanup_cls_series_raw': ('economic_series_values', 'raw_response_id'),
+        'cleanup_cls_response_body': ('raw_responses', 'body_sha256'),
+    }
+    if execute:
+        for name, (table, column) in cleanup_indexes.items():
+            await connection.execute(text(f'CREATE INDEX {name} ON {table} ({column})'))
     before = await retained_news(connection)
     protected = await protected_records(connection)
     if execute:
@@ -83,6 +92,8 @@ async def cleanup(connection, execute=False):
             WHERE body.sha256 = removed.body_sha256 AND NOT EXISTS
             (SELECT 1 FROM raw_responses response WHERE response.body_sha256 = body.sha256)'''))
         counts['unreferenced_raw_bodies'] = result.rowcount
+        for name in cleanup_indexes:
+            await connection.execute(text('DROP INDEX ' + name))
         if protected != await protected_records(connection):
             raise RuntimeError('保留业务数据校验失败，清理已回滚')
         if before != await retained_news(connection):

scripts/deploy.py

@@ -270,18 +270,20 @@ def remove_cls(build, execute=False):
         if execute:
             current = json.loads((INSTANCE / 'deployment.json').read_text())
             old_command = compose(INSTANCE, current)
-            for _ in range(60):
+            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']:
+                if idle['busy']:
-                    break
+                    run('docker', 'unpause', 'AlphaPulse-dev-trader')
-                time.sleep(5)
+                    raise RuntimeError('Analysis became active; retry after it completes')
-            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:

scripts/tests/test_remove_cls.py

@@ -9,7 +9,7 @@ import deploy
 
 
 class RemoveClsTests(unittest.TestCase):
-    def exercise(self, execute, failure=False):
+    def exercise(self, execute, failure=False, running=True):
         with tempfile.TemporaryDirectory() as temporary:
             directory = Path(temporary)
             build = {'commit': 'a'*40, 'version': '20261104-aaaaaaaa'}
@@ -26,6 +26,7 @@ class RemoveClsTests(unittest.TestCase):
 
             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'):
@@ -54,3 +55,8 @@ class RemoveClsTests(unittest.TestCase):
         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_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))