diff --git a/coworker/automation/scheduler.py b/coworker/automation/scheduler.py index 36c828ed..c2a5487a 100644 --- a/coworker/automation/scheduler.py +++ b/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: diff --git a/tests/test_standing_approvals.py b/tests/test_standing_approvals.py index 50a19727..08814c2b 100644 --- a/tests/test_standing_approvals.py +++ b/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 --------