From 840e59cd05f4c90a4b648cfa6ccf4c253a99bf04 Mon Sep 17 00:00:00 2001 From: pewdiepie-archdaemon Date: Mon, 29 Jun 2026 13:52:52 +0000 Subject: [PATCH] Gate background tasks behind foreground activity --- app.py | 12 ++++ src/builtin_actions.py | 5 ++ src/interactive_gate.py | 125 ++++++++++++++++++++++++++++++++++++++++ src/task_endpoint.py | 2 + src/task_scheduler.py | 33 ++++++++++- 5 files changed, 174 insertions(+), 3 deletions(-) create mode 100644 src/interactive_gate.py diff --git a/app.py b/app.py index a7b77ad0c..4a8a12df8 100644 --- a/app.py +++ b/app.py @@ -197,7 +197,19 @@ class _RequestTimeoutMiddleware(_BaseHTTPMiddleware): ) +class _InteractiveActivityMiddleware(_BaseHTTPMiddleware): + async def dispatch(self, request, call_next): + from src.interactive_gate import should_track_interactive_request, track_interactive_request + + path = request.url.path or "" + if not should_track_interactive_request(path, request.method): + return await call_next(request) + async with track_interactive_request(path, request.method): + return await call_next(request) + + app.add_middleware(_RequestTimeoutMiddleware) +app.add_middleware(_InteractiveActivityMiddleware) # ========= AUTH ========= from routes.auth_routes import setup_auth_routes, SESSION_COOKIE diff --git a/src/builtin_actions.py b/src/builtin_actions.py index 3a05969b0..5caa0a691 100644 --- a/src/builtin_actions.py +++ b/src/builtin_actions.py @@ -15,6 +15,7 @@ from src.auth_helpers import owner_filter from core.platform_compat import IS_WINDOWS, find_bash from core.constants import internal_api_base from src.constants import DATA_DIR, DEEP_RESEARCH_DIR, TIDY_CALENDAR_STATE_FILE, EMAIL_URGENCY_CACHE_DIR, COOKBOOK_STATE_FILE +from src.interactive_gate import wait_for_interactive_quiet logger = logging.getLogger(__name__) @@ -146,6 +147,7 @@ async def action_consolidate_memory(owner: str, **kwargs) -> Tuple[str, bool]: "\"drop\":[{\"id\":\"existing id\",\"reason\":\"short reason\"}]}\n\n" f"MEMORIES:\n{json.dumps(items, ensure_ascii=False)}" ) + await wait_for_interactive_quiet("memory consolidation action") raw = await llm_call_async_with_fallback( candidates, messages=[{"role": "user", "content": prompt}], @@ -749,6 +751,7 @@ async def action_classify_events(owner: str, **kwargs) -> Tuple[str, bool]: f"EVENTS: {_json.dumps(items)}" ) try: + await wait_for_interactive_quiet("calendar classification action") raw = await llm_call_async_with_fallback( llm_candidates, messages=[{"role": "user", "content": prompt}], @@ -1023,6 +1026,7 @@ async def action_learn_sender_signatures(owner: str, **kwargs) -> Tuple[str, boo ) try: + await wait_for_interactive_quiet("sender signature action") raw = await llm_call_async_with_fallback( candidates, messages=[{"role": "user", "content": prompt}], @@ -1863,6 +1867,7 @@ async def action_check_email_urgency(owner: str, **kwargs) -> Tuple[str, bool]: f"Snippet:\n{item.get('body','')}\n" ) try: + await wait_for_interactive_quiet("email urgency action") raw = await llm_call_async_with_fallback( candidates, [{"role": "user", "content": prompt}], diff --git a/src/interactive_gate.py b/src/interactive_gate.py new file mode 100644 index 000000000..36f1860e4 --- /dev/null +++ b/src/interactive_gate.py @@ -0,0 +1,125 @@ +"""Foreground activity gate for background work. + +Background tasks are allowed to run only after normal UI/API traffic has +settled. This keeps scheduled jobs and email pollers from competing with the +user opening Odysseus, Cookbook, email, documents, notes, or other panels. +""" + +from __future__ import annotations + +import asyncio +from contextlib import asynccontextmanager +import os +import time + + +_ACTIVE_REQUESTS = 0 +_LAST_ACTIVITY = 0.0 +_COND: asyncio.Condition | None = None + + +def _enabled() -> bool: + return os.getenv("BACKGROUND_TASK_FOREGROUND_GATE", "true").lower() not in {"0", "false", "no", "off"} + + +def _quiet_seconds() -> float: + try: + return max(0.0, float(os.getenv("BACKGROUND_TASK_QUIET_MS", "1500")) / 1000.0) + except Exception: + return 1.5 + + +def _max_wait_seconds() -> float: + """0 means wait indefinitely until the UI is quiet.""" + try: + return max(0.0, float(os.getenv("BACKGROUND_TASK_MAX_WAIT_SECONDS", "0"))) + except Exception: + return 0.0 + + +def _condition() -> asyncio.Condition: + global _COND + if _COND is None: + _COND = asyncio.Condition() + return _COND + + +_PASSIVE_EXACT_PATHS = { + "/api/tasks/notifications", + "/api/research/active", + "/api/email/urgency-state", +} + +_PASSIVE_PREFIXES = ( + "/api/chat/stream_status", + "/api/health", + "/api/prefs", +) + + +def should_track_interactive_request(path: str, method: str = "GET") -> bool: + if not _enabled(): + return False + if (method or "").upper() == "OPTIONS": + return False + if path in _PASSIVE_EXACT_PATHS: + return False + if any(path.startswith(prefix) for prefix in _PASSIVE_PREFIXES): + return False + return True + + +@asynccontextmanager +async def track_interactive_request(path: str = "", method: str = ""): + global _ACTIVE_REQUESTS, _LAST_ACTIVITY + if not _enabled(): + yield + return + + cond = _condition() + async with cond: + _ACTIVE_REQUESTS += 1 + _LAST_ACTIVITY = time.monotonic() + cond.notify_all() + try: + yield + finally: + async with cond: + _ACTIVE_REQUESTS = max(0, _ACTIVE_REQUESTS - 1) + _LAST_ACTIVITY = time.monotonic() + cond.notify_all() + + +async def wait_for_interactive_quiet(label: str = "") -> bool: + """Wait until foreground requests have stopped for the configured window. + + Returns True if the caller had to wait at all. The label is intentionally + only for future logging/debugging so callers can keep their code simple. + """ + if not _enabled(): + return False + + quiet = _quiet_seconds() + max_wait = _max_wait_seconds() + deadline = time.monotonic() + max_wait if max_wait > 0 else None + cond = _condition() + waited = False + + while True: + async with cond: + now = time.monotonic() + quiet_remaining = quiet - (now - _LAST_ACTIVITY) + if _ACTIVE_REQUESTS <= 0 and quiet_remaining <= 0: + return waited + + waited = True + timeout = 0.25 if _ACTIVE_REQUESTS > 0 else min(max(quiet_remaining, 0.05), 0.5) + if deadline is not None: + remaining = deadline - now + if remaining <= 0: + return waited + timeout = min(timeout, remaining) + try: + await asyncio.wait_for(cond.wait(), timeout=timeout) + except asyncio.TimeoutError: + pass diff --git a/src/task_endpoint.py b/src/task_endpoint.py index 6f9a27c09..6b0d20e08 100644 --- a/src/task_endpoint.py +++ b/src/task_endpoint.py @@ -6,6 +6,7 @@ from src.endpoint_resolver import ( resolve_utility_fallback_candidates, ) from src.llm_core import llm_call_async_with_fallback +from src.interactive_gate import wait_for_interactive_quiet def resolve_task_endpoint(fallback_url=None, fallback_model=None, fallback_headers=None, owner=None): @@ -72,4 +73,5 @@ async def task_llm_call_async( ) if not candidates: raise RuntimeError("No LLM endpoint available for background task") + await wait_for_interactive_quiet("background task LLM") return await llm_call_async_with_fallback(candidates, messages=messages, **kwargs) diff --git a/src/task_scheduler.py b/src/task_scheduler.py index 13b07f377..9c7ac754e 100644 --- a/src/task_scheduler.py +++ b/src/task_scheduler.py @@ -735,11 +735,21 @@ class TaskScheduler: try: if bypass_model_slot or not self._task_needs_model_slot(task_id): - await self._execute_task_locked(task_id, run_id, release_executing=release_executing) + await self._execute_task_locked( + task_id, + run_id, + release_executing=release_executing, + gate_foreground=not bypass_model_slot, + ) return async with self._run_semaphore: - await self._execute_task_locked(task_id, run_id, release_executing=release_executing) + await self._execute_task_locked( + task_id, + run_id, + release_executing=release_executing, + gate_foreground=True, + ) except asyncio.CancelledError: # If cancellation happens while queued behind the semaphore, # _execute_task_locked never runs and cannot update the Activity row. @@ -753,7 +763,14 @@ class TaskScheduler: async with self._executing_lock: self._executing.discard(task_id) - async def _execute_task_locked(self, task_id: str, run_id: str, *, release_executing: bool = True): + async def _execute_task_locked( + self, + task_id: str, + run_id: str, + *, + release_executing: bool = True, + gate_foreground: bool = True, + ): from core.database import SessionLocal, ScheduledTask, TaskRun db = SessionLocal() @@ -770,6 +787,14 @@ class TaskScheduler: db.commit() return + if gate_foreground: + waiting = db.query(TaskRun).filter(TaskRun.id == run_id).first() + if waiting and waiting.status == "queued": + waiting.result = "Queued — waiting for Odysseus to be idle…" + db.commit() + from src.interactive_gate import wait_for_interactive_quiet + await wait_for_interactive_quiet(f"scheduled task {task.name}") + # Flip the run from queued → running. Reset started_at to the # actual execution start so queue wait time is visible from # created_at vs started_at if we ever surface that. @@ -1764,6 +1789,8 @@ class TaskScheduler: # behind the primary endpoint so a downed primary won't silently yield # `(no output)`. try: + from src.interactive_gate import wait_for_interactive_quiet + await wait_for_interactive_quiet(f"agent task {task.name}") from src.task_endpoint import resolve_task_candidates _task_fallbacks = resolve_task_candidates( fallback_url=endpoint_url,