手で走らせた回を、ワーカーの待ち行列からも外す

画面の「今すぐ実行」は行列を通さずその場で走らせるので、その回が行列で待っていた
場合に残ったままになっていた。あとでワーカーが起きたときにもう一度同じ回が流れ、
枠を 1 回ぶん余計に食う。

起こせたら行列からも外す。流している最中の塊に居ても外す（押されたのは
その回そのものなので、どちらに居るかは関係が無い）。

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: claude

CLAUDE.md

@@ -470,6 +470,10 @@ GeoNames 全世界地名辞典 = `geonames`(いずれも 348 言語版・195 か
         人に直させる価値が無い)
       - **収集を消したら行列からも外す**(`collect.remove`)。消えた収集を抱えた
         ままだと、そのワーカーは起こそうとして 404 を踏み続ける
+      - **画面の「今すぐ実行」は行列を通さず、その場で走らせる**(相手はワーカーが
+        選ぶまま)。**起こせたら行列からも外す**(`start_collection_bake` →
+        `workers.done`)—— 外さないと、待っていたぶんがあとでもう一度流れて、
+        枠を 1 回ぶん余計に食う。行列と塊のどちらに居ても外す
       - **時計を待たずに起こせる**(`main.wake_worker` / 画面の「今すぐ起こす」)。
         **枠が明いているうちに回しておきたい、が普通に起きる** —— 次の起動まで待つと、
         待っているあいだに他の依頼が枠を食う(実測で、外からの 1 回が 5 時間枠を

app/main.py

@@ -440,6 +440,10 @@ def start_collection_bake(name: str, sweep: str | None = None) -> dict:
 
     **どの巡回を起こしたかは定義側に控える**(`pending_sweep`)。取り込みは収集の
     名前しか運べないので、素材を作る側はそこを読む。
+
+    **起こせたら、ワーカーの待ち行列からも外す**(`workers.done`)。画面の
+    「今すぐ実行」は行列を通さずその場で走らせるので、外さないと**待っていた
+    ぶんがあとでもう一度流れる**(枠を 1 回ぶん余計に食う)。
     """
     from app.views.admin import TRIGGER_URL, trigger_run
 
@@ -457,6 +461,10 @@ def start_collection_bake(name: str, sweep: str | None = None) -> dict:
     # (押した巡回ではないものが走る)
     collect.mark_pending(name, this.name)
     trigger_run(name)
+    # **行列に居たなら外す。** どの道で走ったかに関わらず、その回はもう走っている
+    if worker := getattr(this, "worker", ""):
+        with suppress(Exception):
+            workers.done(worker, name, this.name)
     # **起こせたときだけ予定を進める** —— 混んでいて断られたのに次回へ送ると、
     # その回は黙って飛ばされる(trigger_run が例外にするのでここへは来ない)
     return collect.to_public(collect.mark_started(name, this.name))

app/workers.py

@@ -312,11 +312,17 @@ def last_at(name: str) -> str:
 
 
 def done(name: str, collection: str, sweep: str) -> None:
-    """流し終えた 1 件を塊から外す。"""
+    """流し終えた 1 件を外す。**塊からも行列からも**。
+
+    **行列のほうも外すのは、待っているあいだに手で走らされることがあるから**
+    (画面の「今すぐ実行」)。残しておくと、そのワーカーが起きたときにもう一度
+    同じ回が流れる —— 枠を 1 回ぶん余計に食う。
+    """
     state = _queue_all()
     slot = _slot(state, name)
     made = _entry(collection, sweep)
-    slot["batch"] = [e for e in slot["batch"] if not _same(made, e)]
+    for key in ("queue", "batch"):
+        slot[key] = [e for e in slot[key] if not _same(made, e)]
     _queue_save(state)
 
 

tests/test_workers.py

@@ -496,6 +496,39 @@ async def broken(_asked, _messages):
         assert not workers.full_until("antigravity")
 
 
+class TestRunningItByHandInstead:
+    """画面の「今すぐ実行」は**行列を通さず、その場で走らせる**。
+
+    行列に居たぶんは、その場で外す —— 外さないと、あとでワーカーが起きたときに
+    もう一度同じ回が流れて、枠を 1 回ぶん余計に食う。
+    """
+
+    def test_a_hand_run_takes_it_out_of_the_queue(self, enabled, monkeypatch):
+        from app import collect, main
+
+        monkeypatch.setenv("CHIEZO_NOTES_DIR", str(enabled / "corpus"))
+        monkeypatch.setattr("app.views.admin.TRIGGER_URL", "http://trigger")
+        monkeypatch.setattr("app.views.admin.trigger_run", lambda _name: None)
+        workers.save([workers.Worker("精査", (workers.Step("codex"),))])
+        collect.create("news", prompt="p", interval_minutes=60,
+                       sweeps=[{"name": "ざっと", "worker": "精査"}])
+        workers.enqueue("精査", "news", "ざっと", "2026-01-01T00:00:00+00:00")
+        assert workers.queued("精査")
+
+        main.start_collection_bake("news", "ざっと")
+
+        assert workers.queued("精査") == []
+
+    def test_the_queue_is_cleared_from_either_list(self, enabled):
+        """流している最中の塊に居ても外す(押されたのはその回そのもの)。"""
+        workers.enqueue("精査", "news", "ざっと", "2026-01-01T00:00:00+00:00")
+        workers.claim("精査", 1, "2026-01-01T00:00:00+00:00")
+
+        workers.done("精査", "news", "ざっと")
+
+        assert workers.queued("精査") == []
+
+
 class TestWakingItByHand:
     """時計を待たずに 1 本流す(`main.wake_worker` / 画面の「今すぐ起こす」)。