|
| 1 | +"""Regression for #5536: browser heartbeats abort scheduled tasks even when |
| 2 | +BACKGROUND_TASK_FOREGROUND_GATE is disabled. |
| 3 | +
|
| 4 | +`/api/activity/heartbeat` unconditionally calls |
| 5 | +`stop_background_tasks_for_foreground()`, and that method never consults the |
| 6 | +gate — so setting BACKGROUND_TASK_FOREGROUND_GATE=false silences |
| 7 | +`mark_browser_activity()` / the interactive middleware but heartbeats still |
| 8 | +force-cancel every executing task. Additionally, runs cancelled this way never |
| 9 | +set the in-task `foreground_cancel["hit"]` flag, so Activity records the |
| 10 | +misleading "Stopped by user" instead of "Paused because Odysseus became |
| 11 | +active" and the task misses the 15-minute foreground defer. |
| 12 | +
|
| 13 | +These tests drive the real `TaskScheduler` (no HTTP layer): the executing |
| 14 | +task is a genuine asyncio task registered in `_task_handles`, exactly how the |
| 15 | +heartbeat-triggered stop sees it in production. |
| 16 | +""" |
| 17 | +import asyncio |
| 18 | + |
| 19 | +import pytest |
| 20 | + |
| 21 | + |
| 22 | +def _make_scheduler(): |
| 23 | + from src.task_scheduler import TaskScheduler |
| 24 | + return TaskScheduler(session_manager=None) |
| 25 | + |
| 26 | + |
| 27 | +async def _register_hanging_task(scheduler, task_id): |
| 28 | + """Register a real, cancellable asyncio task as an executing scheduler job.""" |
| 29 | + started = asyncio.Event() |
| 30 | + |
| 31 | + async def _hang(): |
| 32 | + started.set() |
| 33 | + await asyncio.sleep(3600) |
| 34 | + |
| 35 | + handle = asyncio.create_task(_hang()) |
| 36 | + async with scheduler._executing_lock: |
| 37 | + scheduler._executing.add(task_id) |
| 38 | + scheduler._task_handles[task_id] = handle |
| 39 | + await started.wait() |
| 40 | + return handle |
| 41 | + |
| 42 | + |
| 43 | +def test_stop_is_noop_when_gate_disabled(monkeypatch): |
| 44 | + monkeypatch.setenv("BACKGROUND_TASK_FOREGROUND_GATE", "false") |
| 45 | + scheduler = _make_scheduler() |
| 46 | + aborted = [] |
| 47 | + monkeypatch.setattr( |
| 48 | + scheduler, "_mark_run_aborted", |
| 49 | + lambda *a, **k: aborted.append((a, k)) or False, |
| 50 | + ) |
| 51 | + |
| 52 | + async def _scenario(): |
| 53 | + handle = await _register_hanging_task(scheduler, "task-gate-off") |
| 54 | + stopped = await scheduler.stop_background_tasks_for_foreground(reason="browser heartbeat") |
| 55 | + # Give a cancelled handle the chance to actually die before asserting. |
| 56 | + await asyncio.sleep(0) |
| 57 | + alive = not handle.done() |
| 58 | + handle.cancel() |
| 59 | + return stopped, alive |
| 60 | + |
| 61 | + stopped, alive = asyncio.run(_scenario()) |
| 62 | + assert stopped == 0, ( |
| 63 | + "stop_background_tasks_for_foreground must be a no-op when " |
| 64 | + "BACKGROUND_TASK_FOREGROUND_GATE is disabled" |
| 65 | + ) |
| 66 | + assert alive, "executing task must not be cancelled while the gate is disabled" |
| 67 | + assert not aborted, "no run may be marked aborted while the gate is disabled" |
| 68 | + |
| 69 | + |
| 70 | +def test_stop_cancels_when_gate_enabled(monkeypatch): |
| 71 | + monkeypatch.setenv("BACKGROUND_TASK_FOREGROUND_GATE", "true") |
| 72 | + scheduler = _make_scheduler() |
| 73 | + monkeypatch.setattr(scheduler, "_mark_run_aborted", lambda *a, **k: True) |
| 74 | + |
| 75 | + async def _scenario(): |
| 76 | + handle = await _register_hanging_task(scheduler, "task-gate-on") |
| 77 | + stopped = await scheduler.stop_background_tasks_for_foreground(reason="browser heartbeat") |
| 78 | + await asyncio.sleep(0) |
| 79 | + return stopped, handle.cancelled() or handle.done() |
| 80 | + |
| 81 | + stopped, cancelled = asyncio.run(_scenario()) |
| 82 | + assert stopped > 0 |
| 83 | + assert cancelled, "with the gate enabled the executing task must be cancelled" |
| 84 | + |
| 85 | + |
| 86 | +def test_heartbeat_stop_records_foreground_pause_not_user_stop(monkeypatch): |
| 87 | + """Full path through _execute_task_locked: a heartbeat-triggered stop must |
| 88 | + record the foreground-pause message and the 15-minute defer, not the |
| 89 | + misleading 'Stopped by user' + regular reschedule.""" |
| 90 | + monkeypatch.setenv("BACKGROUND_TASK_FOREGROUND_GATE", "true") |
| 91 | + from core.database import SessionLocal, ScheduledTask, TaskRun |
| 92 | + from src.task_scheduler import TaskScheduler, _utcnow |
| 93 | + |
| 94 | + task_id = "task-5536-fg-pause" |
| 95 | + run_id = "run-5536-fg-pause" |
| 96 | + db = SessionLocal() |
| 97 | + db.add(ScheduledTask( |
| 98 | + id=task_id, |
| 99 | + owner="alice", |
| 100 | + name="hang forever", |
| 101 | + task_type="action", |
| 102 | + action="test_hang_5536", |
| 103 | + status="active", |
| 104 | + trigger_type="manual", |
| 105 | + )) |
| 106 | + db.add(TaskRun(id=run_id, task_id=task_id, status="queued", started_at=_utcnow())) |
| 107 | + db.commit() |
| 108 | + db.close() |
| 109 | + |
| 110 | + scheduler = TaskScheduler(session_manager=None) |
| 111 | + entered = asyncio.Event() |
| 112 | + |
| 113 | + async def _hanging_action(task, run_id=None): |
| 114 | + entered.set() |
| 115 | + await asyncio.sleep(3600) |
| 116 | + return "unreachable", True |
| 117 | + |
| 118 | + monkeypatch.setattr(scheduler, "_execute_action", _hanging_action) |
| 119 | + |
| 120 | + async def _scenario(): |
| 121 | + async with scheduler._executing_lock: |
| 122 | + scheduler._executing.add(task_id) |
| 123 | + handle = asyncio.create_task( |
| 124 | + scheduler._execute_task_locked( |
| 125 | + task_id, run_id, release_executing=False, gate_foreground=True, |
| 126 | + ) |
| 127 | + ) |
| 128 | + scheduler._task_handles[task_id] = handle |
| 129 | + await asyncio.wait_for(entered.wait(), timeout=10) |
| 130 | + await scheduler.stop_background_tasks_for_foreground(reason="browser heartbeat") |
| 131 | + try: |
| 132 | + await asyncio.wait_for(handle, timeout=10) |
| 133 | + except asyncio.CancelledError: |
| 134 | + pass |
| 135 | + |
| 136 | + asyncio.run(_scenario()) |
| 137 | + |
| 138 | + db = SessionLocal() |
| 139 | + try: |
| 140 | + run = db.query(TaskRun).filter(TaskRun.id == run_id).first() |
| 141 | + task = db.query(ScheduledTask).filter(ScheduledTask.id == task_id).first() |
| 142 | + assert run.status == "aborted" |
| 143 | + assert run.error == "Paused because Odysseus became active", ( |
| 144 | + f"heartbeat-triggered cancel recorded {run.error!r} — this is a " |
| 145 | + "foreground pause, not a user stop" |
| 146 | + ) |
| 147 | + assert task.next_run is not None, ( |
| 148 | + "a foreground pause must defer the task (15 min), not drop next_run" |
| 149 | + ) |
| 150 | + delta = (task.next_run - _utcnow()).total_seconds() |
| 151 | + assert 13 * 60 < delta <= 16 * 60, f"expected ~15min defer, got {delta}s" |
| 152 | + finally: |
| 153 | + db.close() |
| 154 | + |
| 155 | + |
| 156 | +def test_queued_cancel_does_not_leak_foreground_stop_flag(monkeypatch): |
| 157 | + """A heartbeat stop that lands while the task is still queued behind the |
| 158 | + run semaphore never reaches _execute_task_locked's CancelledError handler. |
| 159 | + The _foreground_stops entry must still be cleaned up, or a later genuine |
| 160 | + user stop of the same task gets misrecorded as a foreground pause and |
| 161 | + silently rescheduled +15 min.""" |
| 162 | + monkeypatch.setenv("BACKGROUND_TASK_FOREGROUND_GATE", "true") |
| 163 | + from core.database import SessionLocal, ScheduledTask |
| 164 | + from src.task_scheduler import TaskScheduler |
| 165 | + |
| 166 | + task_id = "task-5536-queued-leak" |
| 167 | + db = SessionLocal() |
| 168 | + db.add(ScheduledTask( |
| 169 | + id=task_id, |
| 170 | + owner="alice", |
| 171 | + name="queued victim", |
| 172 | + task_type="llm", |
| 173 | + prompt="hang", |
| 174 | + status="active", |
| 175 | + trigger_type="manual", |
| 176 | + )) |
| 177 | + db.commit() |
| 178 | + db.close() |
| 179 | + |
| 180 | + scheduler = TaskScheduler(session_manager=None) |
| 181 | + |
| 182 | + async def _scenario(): |
| 183 | + # Hold the single run slot so the task parks on the semaphore. |
| 184 | + await scheduler._run_semaphore.acquire() |
| 185 | + try: |
| 186 | + async with scheduler._executing_lock: |
| 187 | + scheduler._executing.add(task_id) |
| 188 | + qtask = asyncio.create_task( |
| 189 | + scheduler._execute_task(task_id, release_executing=False) |
| 190 | + ) |
| 191 | + for _ in range(200): |
| 192 | + if scheduler._task_handles.get(task_id) is qtask: |
| 193 | + break |
| 194 | + await asyncio.sleep(0.01) |
| 195 | + else: |
| 196 | + pytest.fail("queued task never registered its handle") |
| 197 | + await asyncio.sleep(0.05) # let it park on the semaphore |
| 198 | + |
| 199 | + stopped = await scheduler.stop_background_tasks_for_foreground( |
| 200 | + reason="browser heartbeat" |
| 201 | + ) |
| 202 | + assert stopped > 0 |
| 203 | + with pytest.raises(asyncio.CancelledError): |
| 204 | + await qtask |
| 205 | + finally: |
| 206 | + scheduler._run_semaphore.release() |
| 207 | + |
| 208 | + asyncio.run(_scenario()) |
| 209 | + assert task_id not in scheduler._foreground_stops, ( |
| 210 | + "queued-cancel path leaked the foreground-stop flag; the next user " |
| 211 | + "stop of this task would be misclassified as a foreground pause" |
| 212 | + ) |
0 commit comments