mirror of
https://github.com/pewdiepie-archdaemon/odysseus.git
synced 2026-07-15 12:58:04 +00:00
Cancel background tasks when Odysseus becomes active
This commit is contained in:
@@ -97,6 +97,21 @@ def _has_recent_browser_activity(now: float | None = None) -> bool:
|
|||||||
return ((now if now is not None else time.monotonic()) - _LAST_BROWSER_ACTIVITY) < ttl
|
return ((now if now is not None else time.monotonic()) - _LAST_BROWSER_ACTIVITY) < ttl
|
||||||
|
|
||||||
|
|
||||||
|
def has_foreground_activity(now: float | None = None) -> bool:
|
||||||
|
"""Return True when foreground browser/model work should stop background jobs.
|
||||||
|
|
||||||
|
This is intentionally narrower than `wait_for_interactive_quiet`: active
|
||||||
|
request tracking is good for delaying task startup, but a running task
|
||||||
|
should not cancel itself just because the UI polls a passive endpoint.
|
||||||
|
Browser heartbeats and active chat streams are the durable "user is here"
|
||||||
|
signals.
|
||||||
|
"""
|
||||||
|
if not _enabled():
|
||||||
|
return False
|
||||||
|
t = now if now is not None else time.monotonic()
|
||||||
|
return _has_recent_browser_activity(t) or _has_active_chat_stream()
|
||||||
|
|
||||||
|
|
||||||
def _has_active_chat_stream() -> bool:
|
def _has_active_chat_stream() -> bool:
|
||||||
"""Best-effort check for foreground model work that outlives HTTP requests.
|
"""Best-effort check for foreground model work that outlives HTTP requests.
|
||||||
|
|
||||||
|
|||||||
+39
-4
@@ -825,6 +825,27 @@ class TaskScheduler:
|
|||||||
# previous llm/research run's model. The executors set it once the
|
# previous llm/research run's model. The executors set it once the
|
||||||
# model is resolved.
|
# model is resolved.
|
||||||
self._last_run_model = None
|
self._last_run_model = None
|
||||||
|
foreground_cancel = {"hit": False}
|
||||||
|
foreground_monitor = None
|
||||||
|
if gate_foreground:
|
||||||
|
current_task = asyncio.current_task()
|
||||||
|
|
||||||
|
async def _cancel_if_foreground_active():
|
||||||
|
# Give the just-finished quiet gate a tiny grace window,
|
||||||
|
# then keep enforcing "background means background" while
|
||||||
|
# a long email/LLM action is already running.
|
||||||
|
await asyncio.sleep(1.0)
|
||||||
|
from src.interactive_gate import has_foreground_activity
|
||||||
|
while True:
|
||||||
|
await asyncio.sleep(1.0)
|
||||||
|
if has_foreground_activity():
|
||||||
|
foreground_cancel["hit"] = True
|
||||||
|
logger.info("Task '%s' interrupted because Odysseus became active", task.name)
|
||||||
|
if current_task:
|
||||||
|
current_task.cancel()
|
||||||
|
return
|
||||||
|
|
||||||
|
foreground_monitor = asyncio.create_task(_cancel_if_foreground_active())
|
||||||
try:
|
try:
|
||||||
if task_type == "action":
|
if task_type == "action":
|
||||||
result, success = await self._execute_action(task, run_id=run_id)
|
result, success = await self._execute_action(task, run_id=run_id)
|
||||||
@@ -864,15 +885,22 @@ class TaskScheduler:
|
|||||||
db.commit()
|
db.commit()
|
||||||
return
|
return
|
||||||
except asyncio.CancelledError:
|
except asyncio.CancelledError:
|
||||||
logger.info("Task '%s' stopped by user", task.name)
|
msg = (
|
||||||
|
"Paused because Odysseus became active"
|
||||||
|
if foreground_cancel.get("hit")
|
||||||
|
else "Stopped by user"
|
||||||
|
)
|
||||||
|
logger.info("Task '%s' %s", task.name, msg)
|
||||||
run_obj = db.query(TaskRun).filter(TaskRun.id == run_id).first()
|
run_obj = db.query(TaskRun).filter(TaskRun.id == run_id).first()
|
||||||
if run_obj:
|
if run_obj:
|
||||||
run_obj.status = "aborted"
|
run_obj.status = "aborted"
|
||||||
run_obj.error = "Stopped by user"
|
run_obj.error = msg
|
||||||
run_obj.result = run_obj.result or "Stopped by user"
|
run_obj.result = run_obj.result or msg
|
||||||
run_obj.finished_at = _utcnow()
|
run_obj.finished_at = _utcnow()
|
||||||
task.last_run = _utcnow()
|
task.last_run = _utcnow()
|
||||||
if (task.trigger_type or "schedule") == "schedule":
|
if foreground_cancel.get("hit"):
|
||||||
|
task.next_run = _utcnow() + timedelta(minutes=15)
|
||||||
|
elif (task.trigger_type or "schedule") == "schedule":
|
||||||
task.next_run = compute_next_run(
|
task.next_run = compute_next_run(
|
||||||
task.schedule, task.scheduled_time,
|
task.schedule, task.scheduled_time,
|
||||||
task.scheduled_day, task.scheduled_date,
|
task.scheduled_day, task.scheduled_date,
|
||||||
@@ -907,6 +935,13 @@ class TaskScheduler:
|
|||||||
task.next_run = None
|
task.next_run = None
|
||||||
db.commit()
|
db.commit()
|
||||||
return
|
return
|
||||||
|
finally:
|
||||||
|
if foreground_monitor and not foreground_monitor.done():
|
||||||
|
foreground_monitor.cancel()
|
||||||
|
try:
|
||||||
|
await foreground_monitor
|
||||||
|
except asyncio.CancelledError:
|
||||||
|
pass
|
||||||
|
|
||||||
run.finished_at = _utcnow()
|
run.finished_at = _utcnow()
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user