fix: scheduled task can run twice when an approval lands on a tick

coworker/automation/scheduler.py

@@ -77,9 +77,14 @@ async def _tick(self, *, trigger: str) -> None:
         for task in self.store.due():
             # Spawn, don't await: a run can suspend on a parked approval (standing
             # scoped approvals, §25) and one blocked automation must never stall the
-            # scheduler loop, other due tasks, or self-wake resumption. Overlap is
-            # still guarded inside run_task via _running_ids.
-            spawned = asyncio.create_task(self.run_task(task, trigger=trigger))
+            # scheduler loop, other due tasks, or self-wake resumption. The overlap
+            # guard must be claimed *here*, before the spawn: this due() snapshot
+            # goes stale, and if the in-flight run finishes before a spawned
+            # duplicate gets its first step, a guard checked inside the spawn is
+            # already clear — the task runs twice.
+            if not self._claim(task.id):
+                continue
+            spawned = asyncio.create_task(self._run_claimed(task, trigger=trigger))
             self._spawned.add(spawned)
             spawned.add_done_callback(self._spawned.discard)
         if self.extra_tick is not None:
@@ -88,11 +93,21 @@ async def _tick(self, *, trigger: str) -> None:
             except Exception:
                 logger.exception("scheduler extra_tick (wake resume) failed")
 
+    def _claim(self, task_id: str) -> bool:
+        if task_id in self._running_ids:  # skip-on-overlap
+            logger.info("skipping %s — previous run still going", task_id)
+            return False
+        self._running_ids.add(task_id)
+        return True
+
     async def run_task(self, task: ScheduledTask, *, trigger: str) -> Optional[TaskRun]:
-        if task.id in self._running_ids:  # skip-on-overlap
-            logger.info("skipping %s — previous run still going", task.id)
+        if not self._claim(task.id):
             return None
-        self._running_ids.add(task.id)
+        return await self._run_claimed(task, trigger=trigger)
+
+    async def _run_claimed(
+        self, task: ScheduledTask, *, trigger: str
+    ) -> Optional[TaskRun]:
         try:
             run = await self.runner(task, trigger)
         except Exception as exc:

tests/test_standing_approvals.py

@@ -424,6 +424,43 @@ async def runner(task, trigger):
     await sched.stop()
 
 
+async def test_tick_while_run_is_parked_never_redispatches_it(tmp_path):
+    """The overlap guard is claimed at dispatch, not inside the spawned run.
+
+    Regression: a tick's due() snapshot still lists a task whose run is parked on
+    an approval (next_run only advances on completion). When the guard lived
+    inside the spawn, an approval landing just before that tick let the parked
+    run finish and clear the guard before the duplicate spawn took its first
+    step — the task ran twice (the intermittent `assert 2 == 1` above)."""
+    store = TaskStore(tmp_path / "auto.db")
+    task = _task(title="parked")
+    store.save(task)
+    store._conn.execute("UPDATE scheduled_tasks SET next_run=1.0 WHERE id=?", (task.id,))
+    store._conn.commit()
+
+    gate = asyncio.Event()
+    started = asyncio.Event()
+    calls = 0
+
+    async def runner(t, trigger):
+        nonlocal calls
+        calls += 1
+        started.set()
+        await gate.wait()
+        return TaskRun(task_id=t.id, status="ok", trigger=trigger)
+
+    sched = Scheduler(store, runner, tick_seconds=9999)
+    await sched._tick(trigger="schedule")  # dispatch; the run parks on the gate
+    await started.wait()
+    gate.set()  # the approval lands...
+    await sched._tick(trigger="schedule")  # ...right as the next tick fires
+    for _ in range(5):
+        await asyncio.sleep(0)  # parked run finishes; any duplicate would start now
+    assert calls == 1
+    assert store.get(task.id).run_count == 1
+    await sched.stop()
+
+
 # -- engine events: standing_target on the card, the note on the tool card --------