Gate background tasks behind foreground activity

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}],

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

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)

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,