feat: add group cooldown and parallel actions

src/account_engine.py

@@ -311,6 +311,17 @@ class Engine:
             return "failed", {"code": result["status_code"]}
         return "unknown", {"code": "SDK_RESULT_UNCONFIRMED"}
 
+    async def perform_batch(self, session, tasks):
+        results = await asyncio.gather(
+            *(self.perform(session, task) for task in tasks), return_exceptions=True
+        )
+        return [
+            ("unknown", {"code": "INTERRUPTED_CHECK_MANUALLY"})
+            if isinstance(result, BaseException)
+            else result
+            for result in results
+        ]
+
     async def worker_loop(self, ident):
         session = None
         while ident in self.desired:
@@ -349,23 +360,17 @@ class Engine:
                     self.state(ident, f"等待动作间隔（{interval} 秒）")
                     await self.pause(ident, 1)
                     continue
-                task = self.store.claim(ident)
-                if task:
-                    self.state(ident, "正在执行任务")
-                    started = time.monotonic()
-                    try:
-                        status, result = await self.perform(session, task)
-                    except Exception:
-                        status, result = (
-                            "unknown",
-                            {"code": "INTERRUPTED_CHECK_MANUALLY"},
-                        )
-                    self.store.finish(
-                        task["id"],
-                        status,
-                        result,
-                        round((time.monotonic() - started) * 1000),
+                tasks = self.store.claim_batch(ident)
+                if tasks:
+                    self.state(
+                        ident,
+                        "正在并行执行关注和私信" if len(tasks) > 1 else "正在执行任务",
                     )
+                    started = time.monotonic()
+                    results = await self.perform_batch(session, tasks)
+                    elapsed = round((time.monotonic() - started) * 1000)
+                    for task, (status, result) in zip(tasks, results, strict=True):
+                        self.store.finish(task["id"], status, result, elapsed)
                     self.last_write[ident] = time.monotonic()
                 else:
                     self.state(ident, "等待组内任务")

src/account_store.py

@@ -13,6 +13,9 @@ from follow_user import validate_uid
 
 
 def validate_rule(rule):
+    rule = dict(rule)
+    rule["require_follow"] = False
+    rule.setdefault("cooldown", 14400)
     kinds = rule.get("kinds", [])
     if not isinstance(kinds, list) or any(
         k not in ("digg", "follow", "comment", "general_notice") for k in kinds
@@ -27,10 +30,10 @@ def validate_rule(rule):
         raise ValueError("启用规则必须选择通知类型及动作")
     if rule["dm"] and not rule["text"].strip():
         raise ValueError("私信正文不能为空")
-    if rule["require_follow"] and not (rule["follow"] and rule["dm"]):
-        raise ValueError("先关注成功再私信必须同时启用两个动作")
     if type(rule.get("interval")) is not int or not 5 <= rule["interval"] <= 86400:
         raise ValueError("执行间隔需为 5..86400 秒")
+    if type(rule["cooldown"]) is not int or not 0 <= rule["cooldown"] <= 31536000:
+        raise ValueError("同 UID 冷却需为 0..31536000 秒")
     return rule
 
 
@@ -42,6 +45,7 @@ DEFAULT_RULE = {
     "require_follow": False,
     "text": "",
     "interval": 30,
+    "cooldown": 14400,
 }
 
 
@@ -91,7 +95,11 @@ class Store:
           status TEXT NOT NULL DEFAULT 'pending' CHECK(status IN ('pending','running','succeeded','failed','unknown','cancelled')),
           result TEXT NOT NULL DEFAULT '', updated REAL NOT NULL,
           UNIQUE(event,worker,target,action));
-        CREATE UNIQUE INDEX IF NOT EXISTS worker_single_running ON tasks(worker) WHERE status='running';
+        CREATE TABLE IF NOT EXISTS cooldowns (
+          source TEXT NOT NULL REFERENCES accounts(id), target TEXT NOT NULL,
+          last_at REAL NOT NULL, event INTEGER NOT NULL REFERENCES events(id),
+          PRIMARY KEY(source,target));
+        DROP INDEX IF EXISTS worker_single_running;
         """)
         self.migrate_accounts()
         if "deleted" not in {
@@ -110,6 +118,9 @@ class Store:
                 self.db.execute(
                     "ALTER TABLE inbox ADD COLUMN next_retry_at REAL NOT NULL DEFAULT 0"
                 )
+            for row in self.db.execute("SELECT id,rule FROM accounts").fetchall():
+                rule = validate_rule(decode(row["rule"]))
+                self.db.execute("UPDATE accounts SET rule=? WHERE id=?", (json.dumps(rule), row["id"]))
         self.db.executescript("""
         CREATE TRIGGER IF NOT EXISTS owner_insert BEFORE INSERT ON accounts
         WHEN NEW.owner IS NOT NULL AND NOT EXISTS(SELECT 1 FROM accounts WHERE id=NEW.owner AND role='main')
@@ -304,7 +315,7 @@ class Store:
     def set_rule(self, ident, rule):
         if self.account(ident)["role"] != "main":
             raise ValueError("只有大号可配置规则")
-        validate_rule(rule)
+        rule = validate_rule(rule)
         self.db.execute(
             "UPDATE accounts SET rule=? WHERE id=?", (json.dumps(rule), ident)
         )
@@ -316,8 +327,9 @@ class Store:
             通知类型=",".join(rule["kinds"]),
             关注=rule["follow"],
             私信=rule["dm"],
-            先关注=rule["require_follow"],
+            双动作="关注和私信并行" if rule["follow"] and rule["dm"] else "单动作",
             动作间隔秒=rule["interval"],
+            同组UID冷却秒=rule["cooldown"],
         )
 
     def move(self, worker, owner):
@@ -587,37 +599,64 @@ class Store:
                         target=target,
                     )
                     continue
-                dependency = None
+                active = self.db.execute(
+                    "SELECT status FROM tasks WHERE source=? AND target=? AND status IN ('pending','running','unknown') ORDER BY id DESC LIMIT 1",
+                    (source, target),
+                ).fetchone()
+                if active:
+                    skipped += 1
+                    self.log(
+                        "冷却跳过",
+                        "同组同 UID 已有待执行、执行中或待核对任务，不重复排队",
+                        source=source,
+                        event=event["id"],
+                        target=target,
+                        现有状态=active["status"],
+                    )
+                    continue
+                cooldown = rule.get("cooldown", 14400)
+                previous = self.db.execute(
+                    "SELECT last_at FROM cooldowns WHERE source=? AND target=?",
+                    (source, target),
+                ).fetchone()
+                now = time.time()
+                if cooldown and previous and previous["last_at"] + cooldown > now:
+                    skipped += 1
+                    self.log(
+                        "冷却跳过",
+                        "同组同 UID 仍在冷却期，本通知不创建任务",
+                        source=source,
+                        event=event["id"],
+                        target=target,
+                        剩余秒=max(1, round(previous["last_at"] + cooldown - now)),
+                    )
+                    continue
                 for action in ("follow", "dm"):
                     if not rule[action]:
                         continue
                     cur = self.db.execute(
-                        "INSERT INTO tasks(event,source,worker,target,action,params,dependency,updated) VALUES (?,?,?,?,?,?,?,?)",
+                        "INSERT INTO tasks(event,source,worker,target,action,params,dependency,updated) VALUES (?,?,?,?,?,?,NULL,?)",
                         (
                             event["id"],
                             source,
                             worker["id"],
                             target,
                             action,
-                            event["rule"],
-                            dependency if rule["require_follow"] else None,
-                            time.time(),
+                            json.dumps(rule),
+                            now,
                         ),
                     )
                     created += 1
                     self.log(
                         "任务入队",
-                        "已创建任务，等待执行",
+                        "已创建任务，等待执行；同一通知的关注和私信将并行",
                         source=source,
                         worker=worker["id"],
                         event=event["id"],
                         task=cur.lastrowid,
                         target=target,
                         action=action,
-                        前置任务=dependency if rule["require_follow"] else None,
                     )
-                    if action == "follow":
-                        dependency = cur.lastrowid
             self.log(
                 "分配完成",
                 "分配流程已完成；不代表动作已执行",
@@ -632,7 +671,7 @@ class Store:
             )
         self.db.execute("UPDATE accounts SET rr=? WHERE id=?", (rr, source))
 
-    def claim(self, worker):
+    def _claim(self, worker, batch):
         with self.transaction():
             account = self.account(worker)
             if (
@@ -640,14 +679,14 @@ class Store:
                 or account["owner"] is None
                 or account["uid"] is None
             ):
-                return None
+                return []
             owner = self.account(account["owner"])
             if owner["uid"] is None or not decode(owner["rule"])["enabled"]:
-                return None
+                return []
             if self.db.execute(
                 "SELECT 1 FROM tasks WHERE worker=? AND status='running'", (worker,)
             ).fetchone():
-                return None
+                return []
             cancelled = self.db.execute(
                 """UPDATE tasks SET status='cancelled',result='前置关注未成功',updated=?
                 WHERE worker=? AND status='pending' AND dependency IN
@@ -657,20 +696,20 @@ class Store:
             for row in cancelled:
                 self.log(
                     "任务取消",
-                    "前置关注未成功，取消依赖步骤，不重发",
+                    "前置关注未成功，取消旧版依赖步骤，不重发",
                     worker=worker,
                     source=owner["id"],
                     task=row["id"],
                     event=row["event"],
                     action=row["action"],
                 )
-            task = self.db.execute(
+            first = self.db.execute(
                 """SELECT t.* FROM tasks t WHERE worker=? AND source=? AND status='pending'
                 AND (dependency IS NULL OR EXISTS(SELECT 1 FROM tasks d WHERE d.id=t.dependency AND d.status='succeeded'))
                 ORDER BY id LIMIT 1""",
                 (worker, account["owner"]),
             ).fetchone()
-            if task is None:
+            if first is None:
                 blocked = self.db.execute(
                     "SELECT t.id,t.dependency,d.status FROM tasks t JOIN tasks d ON d.id=t.dependency WHERE t.worker=? AND t.source=? AND t.status='pending' ORDER BY t.id LIMIT 1",
                     (worker, account["owner"]),
@@ -679,7 +718,7 @@ class Store:
                 if key and self._dependency_wait.get(worker) != key:
                     self.log(
                         "依赖等待",
-                        "前置关注尚未确认成功；待核对不会自动重发",
+                        "旧版前置关注尚未确认成功；待核对不会自动重发",
                         worker=worker,
                         source=owner["id"],
                         task=blocked["id"],
@@ -687,23 +726,65 @@ class Store:
                         前置状态=blocked["status"],
                     )
                 self._dependency_wait[worker] = key
-                return None
+                return []
             self._dependency_wait.pop(worker, None)
-            self.db.execute(
+            tasks = [first]
+            if batch and first["dependency"] is None:
+                tasks = self.db.execute(
+                    """SELECT * FROM tasks WHERE worker=? AND source=? AND event=? AND target=?
+                    AND status='pending' AND dependency IS NULL ORDER BY id""",
+                    (worker, first["source"], first["event"], first["target"]),
+                ).fetchall()
+            now = time.time()
+            self.db.executemany(
                 "UPDATE tasks SET status='running',updated=? WHERE id=?",
-                (time.time(), task["id"]),
+                [(now, task["id"]) for task in tasks],
             )
-            self.log(
-                "任务开始",
-                "已领取任务，即将发起动作；中断后需人工核对，不自动重发",
-                source=task["source"],
-                worker=worker,
-                event=task["event"],
-                task=task["id"],
-                target=task["target"],
-                action=task["action"],
-            )
-            return dict(task)
+            cooldown = decode(first["params"]).get("cooldown", 14400)
+            if cooldown:
+                self.db.execute(
+                    """INSERT INTO cooldowns(source,target,last_at,event) VALUES (?,?,?,?)
+                    ON CONFLICT(source,target) DO UPDATE SET last_at=excluded.last_at,event=excluded.event""",
+                    (first["source"], first["target"], now, first["event"]),
+                )
+                self.log(
+                    "冷却开始",
+                    "任务已实际领取；从现在起组内所有小号共享该 UID 冷却",
+                    source=first["source"],
+                    worker=worker,
+                    event=first["event"],
+                    target=first["target"],
+                    冷却秒=cooldown,
+                )
+            if len(tasks) > 1:
+                self.log(
+                    "并行批次",
+                    "同一通知的关注和私信同时开始，互不依赖且不等待动作间隔",
+                    source=first["source"],
+                    worker=worker,
+                    event=first["event"],
+                    target=first["target"],
+                    任务数=len(tasks),
+                )
+            for task in tasks:
+                self.log(
+                    "任务开始",
+                    "已领取任务，即将发起动作；中断后需人工核对，不自动重发",
+                    source=task["source"],
+                    worker=worker,
+                    event=task["event"],
+                    task=task["id"],
+                    target=task["target"],
+                    action=task["action"],
+                )
+            return [dict(task) for task in tasks]
+
+    def claim(self, worker):
+        tasks = self._claim(worker, False)
+        return tasks[0] if tasks else None
+
+    def claim_batch(self, worker):
+        return self._claim(worker, True)
 
     def finish(self, task_id, status, result, elapsed_ms=None):
         if status not in ("succeeded", "failed", "unknown"):

src/accounts_app.py

@@ -421,12 +421,11 @@ class Window(QMainWindow):
             return
         rule = decode(account["rule"])
         win, form, buttons = dialog("自动操作规则 — " + account["name"], self)
-        enabled, follow, dm, require = [QCheckBox() for _ in range(4)]
+        enabled, follow, dm = [QCheckBox() for _ in range(3)]
         for widget, key, label in [
             (enabled, "enabled", "启用自动操作"),
             (follow, "follow", "执行关注"),
             (dm, "dm", "发送私信"),
-            (require, "require_follow", "必须关注成功后才发送私信"),
         ]:
             widget.setChecked(rule[key])
             form.addRow(label, widget)
@@ -447,9 +446,15 @@ class Window(QMainWindow):
         interval = QSpinBox()
         interval.setRange(5, 86400)
         interval.setValue(rule["interval"])
-        form.addRow("小号动作间隔（秒）", interval)
+        form.addRow("不同任务批次间隔（秒）", interval)
+        cooldown = QSpinBox()
+        cooldown.setRange(0, 525600)
+        cooldown.setSpecialValueText("关闭")
+        cooldown.setSuffix(" 分钟")
+        cooldown.setValue(rule.get("cooldown", 14400) // 60)
+        form.addRow("同组同 UID 冷却", cooldown)
         warning = QLabel(
-            "每条通知只分配给一个小号，组内轮询。已排队任务保留原规则快照；关闭规则会暂停执行。\n启用后只处理新通知，结果不确定不会重发；请确保行为获得授权并符合平台规则。"
+            "每条通知只分配给一个小号，组内轮询。同一 UID 的冷却由整个大号组共享，默认 240 分钟；冷却从任务实际开始时计算。\n关注和私信同时开启时会并行执行、结果互不依赖，不等待批次间隔。已排队任务保留原规则快照；结果不确定不会重发。"
         )
         warning.setWordWrap(True)
         form.addRow(warning)
@@ -459,10 +464,11 @@ class Window(QMainWindow):
                 "enabled": enabled.isChecked(),
                 "follow": follow.isChecked(),
                 "dm": dm.isChecked(),
-                "require_follow": require.isChecked(),
+                "require_follow": False,
                 "kinds": [k for k, v in kinds.items() if v.isChecked()],
                 "text": text.toPlainText(),
                 "interval": interval.value(),
+                "cooldown": cooldown.value() * 60,
             }
             try:
                 validate_rule(value)

src/test_accounts.py

@@ -29,7 +29,7 @@ def rule(**overrides):
         "enabled": True,
         "follow": True,
         "dm": True,
-        "require_follow": True,
+        "require_follow": False,
         "text": "offline-only-test",
         **overrides,
     }
@@ -61,6 +61,11 @@ class StoreTest(unittest.TestCase):
             validate_rule(rule(text=""))
         with self.assertRaises(ValueError):
             validate_rule(rule(interval=True))
+        with self.assertRaises(ValueError):
+            validate_rule(rule(cooldown=True))
+        normalized = validate_rule({k: v for k, v in rule(require_follow=True).items() if k != "cooldown"})
+        self.assertFalse(normalized["require_follow"])
+        self.assertEqual(normalized["cooldown"], 14400)
         self.assertFalse(json.loads(self.store.account(self.other)["rule"])["enabled"])
         with self.assertRaises(ValueError):
             self.store.add("worker", "坏UID", "01")
@@ -134,23 +139,68 @@ class StoreTest(unittest.TestCase):
         with self.assertRaises(sqlite3.IntegrityError):
             self.store.add("worker", "不能归属小号", owner=self.worker)
 
-    def test_dedup_round_robin_and_dependency(self):
+    def test_dedup_round_robin_and_parallel_batch(self):
         second = self.store.add("worker", "第二小号", "202", self.main)
-        for nid in ("111", "111", "112", "113"):
-            self.store.ingest(self.main, notice("101", nid))
+        for nid, target in (("111", "901"), ("111", "901"), ("112", "902"), ("113", "903")):
+            self.store.ingest(self.main, notice("101", nid, target))
         tasks = sorted(self.store.tasks(), key=lambda t: t["id"])
         self.assertEqual(len(tasks), 6)
         self.assertEqual(
             [tasks[i]["worker"] for i in (0, 2, 4)], [self.worker, second, self.worker]
         )
-        first = self.store.claim(self.worker)
-        assert first is not None
-        self.assertEqual(first["action"], "follow")
+        batch = self.store.claim_batch(self.worker)
+        self.assertEqual({task["action"] for task in batch}, {"follow", "dm"})
+        self.assertEqual({task["status"] for task in self.store.tasks() if task["id"] in {t["id"] for t in batch}}, {"running"})
         self.assertIsNone(self.store.claim(self.worker))
-        self.store.finish(first["id"], "succeeded", {})
-        second_task = self.store.claim(self.worker)
-        assert second_task is not None
-        self.assertEqual(second_task["action"], "dm")
+        self.store.finish(batch[0]["id"], "failed", {})
+        self.store.finish(batch[1]["id"], "unknown", {})
+        next_batch = self.store.claim_batch(self.worker)
+        self.assertEqual({task["target"] for task in next_batch}, {"903"})
+
+    def test_group_cooldown_persists_is_shared_and_other_groups_are_independent(self):
+        second = self.store.add("worker", "第二小号", "202", self.main)
+        other_worker = self.store.add("worker", "其他组小号", "203", self.other)
+        self.store.set_rule(self.other, rule())
+        with patch("account_store.time.time", return_value=100):
+            self.store.ingest(self.main, notice("101", "111", "999"))
+            self.assertEqual(
+                self.store.db.execute("SELECT count(*) FROM cooldowns").fetchone()[0], 0
+            )
+            # A second notification before execution cannot duplicate the queued bundle.
+            self.store.ingest(self.main, notice("101", "112", "999"))
+            self.assertEqual(len(self.store.tasks()), 2)
+            batch = self.store.claim_batch(self.worker)
+            self.assertEqual(len(batch), 2)
+            self.assertEqual(
+                self.store.db.execute("SELECT last_at FROM cooldowns WHERE source=? AND target='999'", (self.main,)).fetchone()[0], 100
+            )
+            for task in batch:
+                self.store.finish(task["id"], "succeeded", {})
+            # The same UID in a different main-account group is independent.
+            self.store.ingest(self.other, notice("102", "211", "999"))
+            self.assertEqual(len([t for t in self.store.tasks() if t["worker"] == other_worker]), 2)
+        self.store.close()
+        self.store = Store(self.temp.name)
+        with patch("account_store.time.time", return_value=200):
+            self.store.ingest(self.main, notice("101", "113", "999"))
+            self.assertEqual(len([t for t in self.store.tasks() if t["source"] == self.main]), 2)
+        with patch("account_store.time.time", return_value=14501):
+            self.store.ingest(self.main, notice("101", "114", "999"))
+            new_tasks = [t for t in self.store.tasks() if t["source"] == self.main]
+            self.assertEqual(len(new_tasks), 4)
+            self.assertEqual({t["worker"] for t in new_tasks[:2]}, {second})
+
+    def test_zero_cooldown_allows_later_event_after_first_bundle_finishes(self):
+        self.store.set_rule(self.main, rule(cooldown=0))
+        self.store.ingest(self.main, notice("101", "111", "999"))
+        batch = self.store.claim_batch(self.worker)
+        for task in batch:
+            self.store.finish(task["id"], "succeeded", {})
+        self.store.ingest(self.main, notice("101", "112", "999"))
+        self.assertEqual(len(self.store.tasks()), 4)
+        self.assertEqual(
+            self.store.db.execute("SELECT count(*) FROM cooldowns").fetchone()[0], 0
+        )
 
     def test_move_cancels_only_pending_and_is_atomic(self):
         self.store.ingest(self.main, notice("101"))
@@ -169,28 +219,27 @@ class StoreTest(unittest.TestCase):
         self.store.move(self.worker, None)
         self.assertIsNone(self.store.account(self.worker)["owner"])
 
-    def test_restart_running_becomes_unknown_never_resends(self):
+    def test_restart_running_batch_becomes_unknown_never_resends(self):
         self.store.ingest(self.main, notice("101"))
-        first = self.store.claim(self.worker)
-        assert first is not None
+        batch = self.store.claim_batch(self.worker)
+        self.assertEqual(len(batch), 2)
         self.store.close()
         self.store = Store(self.temp.name)
         self.store.recover()
-        self.assertIsNone(self.store.claim(self.worker))
-        self.store.resolve(first["id"], True)
-        second_task = self.store.claim(self.worker)
-        assert second_task is not None
-        self.assertEqual(second_task["action"], "dm")
+        self.assertEqual({t["status"] for t in self.store.tasks()}, {"unknown"})
+        self.assertEqual(self.store.claim_batch(self.worker), [])
+        for task in batch:
+            self.store.resolve(task["id"], True)
+        self.assertEqual(self.store.claim_batch(self.worker), [])
 
-    def test_unknown_failed_and_no_auto_retry(self):
+    def test_parallel_results_are_independent_and_never_resend(self):
         self.store.ingest(self.main, notice("101"))
-        first = self.store.claim(self.worker)
-        assert first is not None
-        self.store.finish(first["id"], "failed", {})
-        self.assertIsNone(self.store.claim(self.worker))
-        self.assertEqual(
-            {t["status"] for t in self.store.tasks()}, {"failed", "cancelled"}
-        )
+        batch = self.store.claim_batch(self.worker)
+        self.assertEqual(len(batch), 2)
+        self.store.finish(batch[0]["id"], "failed", {})
+        self.store.finish(batch[1]["id"], "unknown", {})
+        self.assertEqual(self.store.claim_batch(self.worker), [])
+        self.assertEqual({t["status"] for t in self.store.tasks()}, {"failed", "unknown"})
 
     def test_disabled_and_cross_group_identity(self):
         with self.assertRaises(ValueError):
@@ -236,6 +285,21 @@ class StoreTest(unittest.TestCase):
         ).fetchone()
         self.assertEqual(tuple(row), ("pending", 20, 400))
 
+    def test_rule_migration_adds_default_cooldown_and_disables_old_dependency(self):
+        legacy = rule(require_follow=True)
+        legacy.pop("cooldown")
+        self.store.db.execute(
+            "UPDATE accounts SET rule=? WHERE id=?", (json.dumps(legacy), self.main)
+        )
+        self.store.close()
+        self.store = Store(self.temp.name)
+        migrated = json.loads(self.store.account(self.main)["rule"])
+        self.assertEqual(migrated["cooldown"], 14400)
+        self.assertFalse(migrated["require_follow"])
+        self.assertIsNotNone(
+            self.store.db.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name='cooldowns'").fetchone()
+        )
+
     def test_legacy_inbox_migration_preserves_pending_ids(self):
         self.store.record_push(self.main, ["111"])
         self.store.db.executescript("""
@@ -663,26 +727,41 @@ class EngineTest(unittest.IsolatedAsyncioTestCase):
         worker.page.is_closed = lambda: False
         self.engine.sessions[self.main] = main
         self.engine.open = AsyncMock(return_value=worker)
-        entered = asyncio.Event()
+        both_entered = asyncio.Event()
         release = asyncio.Event()
+        entered = 0
 
-        async def action(*args):
-            entered.set()
+        async def action(kind, *args):
+            nonlocal entered
+            entered += 1
+            if entered == 2:
+                both_entered.set()
             await release.wait()
-            return {"status": "succeeded"}
+            return {"status": "succeeded"} if kind == "follow" else {
+                "success": True,
+                "message": {"client_id": "client", "server_id": "server"},
+            }
 
-        worker.follow.side_effect = action
+        async def follow_action(*args, **kwargs):
+            return await action("follow", *args)
+
+        async def dm_action(*args, **kwargs):
+            return await action("dm", *args)
+
+        worker.follow.side_effect = follow_action
+        worker.im.side_effect = dm_action
         runner = asyncio.create_task(self.engine.worker_loop(self.worker))
         self.engine.runners[self.worker] = runner
-        await asyncio.wait_for(entered.wait(), 2)
+        await asyncio.wait_for(both_entered.wait(), 2)
         stopping = asyncio.create_task(self.engine.stop())
         await asyncio.sleep(0.02)
         self.assertFalse(stopping.done())
         release.set()
         await asyncio.wait_for(stopping, 2)
         tasks = self.engine.store.tasks()
-        self.assertEqual({t["status"] for t in tasks}, {"succeeded", "pending"})
-        worker.im.assert_not_called()
+        self.assertEqual({t["status"] for t in tasks}, {"succeeded"})
+        self.assertEqual(worker.follow.await_count, 1)
+        self.assertEqual(worker.im.await_count, 1)
 
 
 if __name__ == "__main__":

src/test_activity.py

@@ -127,17 +127,14 @@ def test_business_stages_skips_dependencies_and_rollback():
             assert "1234567890123456789" not in all_logs(audit)
             store.record_push(main, ["2"])
             store.ingest(main, notification("2"))
-            task = store.claim(worker)
-            assert task is not None
-            store.finish(task["id"], "unknown", {"code": "UNCONFIRMED"}, 45)
-            assert store.claim(worker) is None
-            before = audit.sequence
-            assert store.claim(worker) is None
-            assert audit.sequence == before, "相同依赖等待不应反复刷日志"
-            store.resolve(task["id"], True)
-            task = store.claim(worker)
-            assert task is not None and task["action"] == "dm"
-            store.finish(task["id"], "succeeded", {"status_code": 0}, 80)
+            batch = store.claim_batch(worker)
+            assert {task["action"] for task in batch} == {"follow", "dm"}
+            follow = next(task for task in batch if task["action"] == "follow")
+            dm = next(task for task in batch if task["action"] == "dm")
+            store.finish(follow["id"], "unknown", {"code": "UNCONFIRMED"}, 45)
+            store.finish(dm["id"], "succeeded", {"status_code": 0}, 80)
+            assert store.claim_batch(worker) == []
+            store.resolve(follow["id"], True)
             text = all_logs(audit)
             for stage in (
                 "通知接收",
@@ -150,7 +147,8 @@ def test_business_stages_skips_dependencies_and_rollback():
                 "任务入队",
                 "任务开始",
                 "任务结果",
-                "依赖等待",
+                "冷却开始",
+                "并行批次",
                 "人工核对",
             ):
                 assert f"[{stage}]" in text