diff --git a/docker-compose.gpu-amd.yml b/docker-compose.gpu-amd.yml index 82e22e4409..255d87c37e 100644 --- a/docker-compose.gpu-amd.yml +++ b/docker-compose.gpu-amd.yml @@ -45,6 +45,10 @@ services: - CHROMADB_PORT=8000 - DATABASE_URL=${DATABASE_URL:-sqlite:///./data/app.db} - AUTH_ENABLED=${AUTH_ENABLED:-true} + - BACKGROUND_TASK_FOREGROUND_GATE=${BACKGROUND_TASK_FOREGROUND_GATE:-true} + - BACKGROUND_TASK_QUIET_MS=${BACKGROUND_TASK_QUIET_MS:-1500} + - BACKGROUND_TASK_BROWSER_ACTIVE_SECONDS=${BACKGROUND_TASK_BROWSER_ACTIVE_SECONDS:-45} + - BACKGROUND_TASK_MAX_WAIT_SECONDS=${BACKGROUND_TASK_MAX_WAIT_SECONDS:-0} - LOCALHOST_BYPASS=${LOCALHOST_BYPASS:-false} - ODYSSEUS_ADMIN_USER=${ODYSSEUS_ADMIN_USER:-admin} - ODYSSEUS_ADMIN_PASSWORD=${ODYSSEUS_ADMIN_PASSWORD:-} diff --git a/docker-compose.gpu-nvidia.yml b/docker-compose.gpu-nvidia.yml index 1b551c6699..73a5a31e5f 100644 --- a/docker-compose.gpu-nvidia.yml +++ b/docker-compose.gpu-nvidia.yml @@ -44,6 +44,10 @@ services: - CHROMADB_PORT=8000 - DATABASE_URL=${DATABASE_URL:-sqlite:///./data/app.db} - AUTH_ENABLED=${AUTH_ENABLED:-true} + - BACKGROUND_TASK_FOREGROUND_GATE=${BACKGROUND_TASK_FOREGROUND_GATE:-true} + - BACKGROUND_TASK_QUIET_MS=${BACKGROUND_TASK_QUIET_MS:-1500} + - BACKGROUND_TASK_BROWSER_ACTIVE_SECONDS=${BACKGROUND_TASK_BROWSER_ACTIVE_SECONDS:-45} + - BACKGROUND_TASK_MAX_WAIT_SECONDS=${BACKGROUND_TASK_MAX_WAIT_SECONDS:-0} - LOCALHOST_BYPASS=${LOCALHOST_BYPASS:-false} - ODYSSEUS_ADMIN_USER=${ODYSSEUS_ADMIN_USER:-admin} - ODYSSEUS_ADMIN_PASSWORD=${ODYSSEUS_ADMIN_PASSWORD:-} diff --git a/docker-compose.yml b/docker-compose.yml index cbeec1e372..ebbc65aa86 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -33,6 +33,10 @@ services: - CHROMADB_PORT=8000 - DATABASE_URL=${DATABASE_URL:-sqlite:///./data/app.db} - AUTH_ENABLED=${AUTH_ENABLED:-true} + - BACKGROUND_TASK_FOREGROUND_GATE=${BACKGROUND_TASK_FOREGROUND_GATE:-true} + - BACKGROUND_TASK_QUIET_MS=${BACKGROUND_TASK_QUIET_MS:-1500} + - BACKGROUND_TASK_BROWSER_ACTIVE_SECONDS=${BACKGROUND_TASK_BROWSER_ACTIVE_SECONDS:-45} + - BACKGROUND_TASK_MAX_WAIT_SECONDS=${BACKGROUND_TASK_MAX_WAIT_SECONDS:-0} - LOCALHOST_BYPASS=${LOCALHOST_BYPASS:-false} - ODYSSEUS_ADMIN_USER=${ODYSSEUS_ADMIN_USER:-admin} - ODYSSEUS_ADMIN_PASSWORD=${ODYSSEUS_ADMIN_PASSWORD:-} diff --git a/src/interactive_gate.py b/src/interactive_gate.py index c0f5907fcb..a3667895ce 100644 --- a/src/interactive_gate.py +++ b/src/interactive_gate.py @@ -24,6 +24,11 @@ def _enabled() -> bool: return os.getenv("BACKGROUND_TASK_FOREGROUND_GATE", "true").lower() not in {"0", "false", "no", "off"} +def foreground_gate_enabled() -> bool: + """Whether foreground activity may pause/stop background tasks at all.""" + return _enabled() + + def _quiet_seconds() -> float: try: return max(0.0, float(os.getenv("BACKGROUND_TASK_QUIET_MS", "1500")) / 1000.0) diff --git a/src/task_scheduler.py b/src/task_scheduler.py index d5b1dad625..1e2eaccf39 100644 --- a/src/task_scheduler.py +++ b/src/task_scheduler.py @@ -350,6 +350,9 @@ def __init__(self, session_manager): self._run_semaphore = asyncio.Semaphore(1) self._concurrency_cap = 1 self._task_handles = {} + # Task IDs cancelled by stop_background_tasks_for_foreground; lets the + # CancelledError path distinguish a foreground pause from a user stop. + self._foreground_stops = set() def _set_run_progress(self, run_id: str, message: str): """Persist short live progress text for Activity while a run is active.""" @@ -772,6 +775,10 @@ async def _execute_task(self, task_id: str, *, bypass_model_slot: bool = False, self._defer_immediately_due_task(task_id, delay=timedelta(minutes=15)) raise finally: + # Task-level cleanup so the flag can't leak when the cancel lands + # outside _execute_task_locked's own handler (e.g. while queued + # behind the run semaphore) and misclassify a later user stop. + self._foreground_stops.discard(task_id) handle = self._task_handles.get(task_id) if handle is current: self._task_handles.pop(task_id, None) @@ -945,9 +952,15 @@ async def _cancel_if_foreground_active(): db.commit() return except asyncio.CancelledError: + # A cancel that came from stop_background_tasks_for_foreground + # is a foreground pause too, even though the in-task monitor + # never saw the activity itself. + foreground_hit = ( + foreground_cancel.get("hit") or task_id in self._foreground_stops + ) msg = ( "Paused because Odysseus became active" - if foreground_cancel.get("hit") + if foreground_hit else "Stopped by user" ) logger.info("Task '%s' %s", task.name, msg) @@ -958,7 +971,7 @@ async def _cancel_if_foreground_active(): run_obj.result = run_obj.result or msg run_obj.finished_at = _utcnow() task.last_run = _utcnow() - if foreground_cancel.get("hit"): + if foreground_hit: task.next_run = _utcnow() + timedelta(minutes=15) elif (task.trigger_type or "schedule") == "schedule": task.next_run = compute_next_run( @@ -2247,16 +2260,25 @@ async def stop_background_tasks_for_foreground(self, *, reason: str = "Odysseus user opens or uses Odysseus, foreground interaction wins immediately. Manual force-runs can be restarted by the user; automatic jobs will be deferred by their cancellation path instead of stealing the app. + + Respects BACKGROUND_TASK_FOREGROUND_GATE: with the gate disabled the + user has opted out of foreground preemption entirely, so this is a + no-op (#5536 — browser heartbeats were cancelling scheduled tasks + regardless of the gate). """ + from src.interactive_gate import foreground_gate_enabled + if not foreground_gate_enabled(): + return 0 async with self._executing_lock: task_ids = list(self._executing) stopped = 0 for task_id in task_ids: handle = self._task_handles.get(task_id) if handle and not handle.done(): + self._foreground_stops.add(task_id) handle.cancel() stopped += 1 - if self._mark_run_aborted(task_id): + if self._mark_run_aborted(task_id, message="Paused because Odysseus became active"): stopped += 1 if stopped: logger.info("Stopped %d background scheduler task(s): %s", stopped, reason) diff --git a/tests/test_foreground_gate_heartbeat_stop.py b/tests/test_foreground_gate_heartbeat_stop.py new file mode 100644 index 0000000000..ed1ff92b73 --- /dev/null +++ b/tests/test_foreground_gate_heartbeat_stop.py @@ -0,0 +1,212 @@ +"""Regression for #5536: browser heartbeats abort scheduled tasks even when +BACKGROUND_TASK_FOREGROUND_GATE is disabled. + +`/api/activity/heartbeat` unconditionally calls +`stop_background_tasks_for_foreground()`, and that method never consults the +gate — so setting BACKGROUND_TASK_FOREGROUND_GATE=false silences +`mark_browser_activity()` / the interactive middleware but heartbeats still +force-cancel every executing task. Additionally, runs cancelled this way never +set the in-task `foreground_cancel["hit"]` flag, so Activity records the +misleading "Stopped by user" instead of "Paused because Odysseus became +active" and the task misses the 15-minute foreground defer. + +These tests drive the real `TaskScheduler` (no HTTP layer): the executing +task is a genuine asyncio task registered in `_task_handles`, exactly how the +heartbeat-triggered stop sees it in production. +""" +import asyncio + +import pytest + + +def _make_scheduler(): + from src.task_scheduler import TaskScheduler + return TaskScheduler(session_manager=None) + + +async def _register_hanging_task(scheduler, task_id): + """Register a real, cancellable asyncio task as an executing scheduler job.""" + started = asyncio.Event() + + async def _hang(): + started.set() + await asyncio.sleep(3600) + + handle = asyncio.create_task(_hang()) + async with scheduler._executing_lock: + scheduler._executing.add(task_id) + scheduler._task_handles[task_id] = handle + await started.wait() + return handle + + +def test_stop_is_noop_when_gate_disabled(monkeypatch): + monkeypatch.setenv("BACKGROUND_TASK_FOREGROUND_GATE", "false") + scheduler = _make_scheduler() + aborted = [] + monkeypatch.setattr( + scheduler, "_mark_run_aborted", + lambda *a, **k: aborted.append((a, k)) or False, + ) + + async def _scenario(): + handle = await _register_hanging_task(scheduler, "task-gate-off") + stopped = await scheduler.stop_background_tasks_for_foreground(reason="browser heartbeat") + # Give a cancelled handle the chance to actually die before asserting. + await asyncio.sleep(0) + alive = not handle.done() + handle.cancel() + return stopped, alive + + stopped, alive = asyncio.run(_scenario()) + assert stopped == 0, ( + "stop_background_tasks_for_foreground must be a no-op when " + "BACKGROUND_TASK_FOREGROUND_GATE is disabled" + ) + assert alive, "executing task must not be cancelled while the gate is disabled" + assert not aborted, "no run may be marked aborted while the gate is disabled" + + +def test_stop_cancels_when_gate_enabled(monkeypatch): + monkeypatch.setenv("BACKGROUND_TASK_FOREGROUND_GATE", "true") + scheduler = _make_scheduler() + monkeypatch.setattr(scheduler, "_mark_run_aborted", lambda *a, **k: True) + + async def _scenario(): + handle = await _register_hanging_task(scheduler, "task-gate-on") + stopped = await scheduler.stop_background_tasks_for_foreground(reason="browser heartbeat") + await asyncio.sleep(0) + return stopped, handle.cancelled() or handle.done() + + stopped, cancelled = asyncio.run(_scenario()) + assert stopped > 0 + assert cancelled, "with the gate enabled the executing task must be cancelled" + + +def test_heartbeat_stop_records_foreground_pause_not_user_stop(monkeypatch): + """Full path through _execute_task_locked: a heartbeat-triggered stop must + record the foreground-pause message and the 15-minute defer, not the + misleading 'Stopped by user' + regular reschedule.""" + monkeypatch.setenv("BACKGROUND_TASK_FOREGROUND_GATE", "true") + from core.database import SessionLocal, ScheduledTask, TaskRun + from src.task_scheduler import TaskScheduler, _utcnow + + task_id = "task-5536-fg-pause" + run_id = "run-5536-fg-pause" + db = SessionLocal() + db.add(ScheduledTask( + id=task_id, + owner="alice", + name="hang forever", + task_type="action", + action="test_hang_5536", + status="active", + trigger_type="manual", + )) + db.add(TaskRun(id=run_id, task_id=task_id, status="queued", started_at=_utcnow())) + db.commit() + db.close() + + scheduler = TaskScheduler(session_manager=None) + entered = asyncio.Event() + + async def _hanging_action(task, run_id=None): + entered.set() + await asyncio.sleep(3600) + return "unreachable", True + + monkeypatch.setattr(scheduler, "_execute_action", _hanging_action) + + async def _scenario(): + async with scheduler._executing_lock: + scheduler._executing.add(task_id) + handle = asyncio.create_task( + scheduler._execute_task_locked( + task_id, run_id, release_executing=False, gate_foreground=True, + ) + ) + scheduler._task_handles[task_id] = handle + await asyncio.wait_for(entered.wait(), timeout=10) + await scheduler.stop_background_tasks_for_foreground(reason="browser heartbeat") + try: + await asyncio.wait_for(handle, timeout=10) + except asyncio.CancelledError: + pass + + asyncio.run(_scenario()) + + db = SessionLocal() + try: + run = db.query(TaskRun).filter(TaskRun.id == run_id).first() + task = db.query(ScheduledTask).filter(ScheduledTask.id == task_id).first() + assert run.status == "aborted" + assert run.error == "Paused because Odysseus became active", ( + f"heartbeat-triggered cancel recorded {run.error!r} — this is a " + "foreground pause, not a user stop" + ) + assert task.next_run is not None, ( + "a foreground pause must defer the task (15 min), not drop next_run" + ) + delta = (task.next_run - _utcnow()).total_seconds() + assert 13 * 60 < delta <= 16 * 60, f"expected ~15min defer, got {delta}s" + finally: + db.close() + + +def test_queued_cancel_does_not_leak_foreground_stop_flag(monkeypatch): + """A heartbeat stop that lands while the task is still queued behind the + run semaphore never reaches _execute_task_locked's CancelledError handler. + The _foreground_stops entry must still be cleaned up, or a later genuine + user stop of the same task gets misrecorded as a foreground pause and + silently rescheduled +15 min.""" + monkeypatch.setenv("BACKGROUND_TASK_FOREGROUND_GATE", "true") + from core.database import SessionLocal, ScheduledTask + from src.task_scheduler import TaskScheduler + + task_id = "task-5536-queued-leak" + db = SessionLocal() + db.add(ScheduledTask( + id=task_id, + owner="alice", + name="queued victim", + task_type="llm", + prompt="hang", + status="active", + trigger_type="manual", + )) + db.commit() + db.close() + + scheduler = TaskScheduler(session_manager=None) + + async def _scenario(): + # Hold the single run slot so the task parks on the semaphore. + await scheduler._run_semaphore.acquire() + try: + async with scheduler._executing_lock: + scheduler._executing.add(task_id) + qtask = asyncio.create_task( + scheduler._execute_task(task_id, release_executing=False) + ) + for _ in range(200): + if scheduler._task_handles.get(task_id) is qtask: + break + await asyncio.sleep(0.01) + else: + pytest.fail("queued task never registered its handle") + await asyncio.sleep(0.05) # let it park on the semaphore + + stopped = await scheduler.stop_background_tasks_for_foreground( + reason="browser heartbeat" + ) + assert stopped > 0 + with pytest.raises(asyncio.CancelledError): + await qtask + finally: + scheduler._run_semaphore.release() + + asyncio.run(_scenario()) + assert task_id not in scheduler._foreground_stops, ( + "queued-cancel path leaked the foreground-stop flag; the next user " + "stop of this task would be misclassified as a foreground pause" + ) diff --git a/tests/test_task_scheduler_cancel.py b/tests/test_task_scheduler_cancel.py index 3d399f1444..0b6414f727 100644 --- a/tests/test_task_scheduler_cancel.py +++ b/tests/test_task_scheduler_cancel.py @@ -64,6 +64,7 @@ async def drive(): scheduler._task_handles = {} scheduler._concurrency_cap = 1 scheduler._task_defer_counts = {} + scheduler._foreground_stops = set() await scheduler._run_semaphore.acquire() task = asyncio.create_task(scheduler._execute_task("queued-task"))