diff --git a/backend/apps/agents/agents.py b/backend/apps/agents/agents.py index 8c48f640..e1688c4c 100644 --- a/backend/apps/agents/agents.py +++ b/backend/apps/agents/agents.py @@ -69,10 +69,18 @@ async def list_sessions(dashboard_id: str = ""): @agents.router.get("/activity") async def agent_activity(): - """How many agent tasks are live right now. Drives the desktop's idle-update gate so a - silent update-on-idle never lands on top of a running agent.""" + """How many agent tasks are live right now, plus seconds until the next scheduled + workflow fires. Drives the desktop's idle-update gate so a silent update-on-idle + never lands on top of a running agent or right before a scheduled run.""" active = sum(1 for t in agent_manager.tasks.values() if not t.done()) - return {"active": active} + try: + # Local import: workflows pulls in agent machinery, a module-level import here would cycle. + from backend.apps.workflows.scheduler import seconds_to_next_fire + next_run_in_s = seconds_to_next_fire() + except Exception: + # Fail open (None = no block): a broken lookahead must never wedge updates forever; the agents gate still protects running work. + next_run_in_s = None + return {"active": active, "next_run_in_s": next_run_in_s} @agents.router.get("/sessions/{session_id}") async def get_session(session_id: str): diff --git a/backend/apps/service/service.py b/backend/apps/service/service.py index d6784403..e107543f 100644 --- a/backend/apps/service/service.py +++ b/backend/apps/service/service.py @@ -21,8 +21,10 @@ import platform from collections import Counter from contextlib import asynccontextmanager from datetime import datetime +from typing import Literal, Optional from fastapi import Body +from pydantic import BaseModel, ConfigDict from backend.config.Apps import SubApp from backend.config.paths import SESSIONS_DIR @@ -512,6 +514,25 @@ async def post_event(body: dict): return {"ok": True} +class UpdaterEventBody(BaseModel): + model_config = ConfigDict(validate_assignment=True) + kind: Literal["idle_install"] + staged_version: Optional[str] = None + + +@service.router.post("/updater-event") +async def post_updater_event(body: UpdaterEventBody): + """Electron main reports updater milestones (today just the evergreen idle install) so fleet convergence shows up in analytics logs instead of being inferred.""" + from backend.apps.service.analytics.client import get_analytics_client + client = get_analytics_client() + if client is not None: + try: + client.logs.write(tag="updater", subtag=body.kind, data={"staged_version": body.staged_version or "", "app_version": APP_VERSION}) + except Exception: + pass + return {"ok": True} + + @service.router.get("/spool/count") async def spool_count(): from backend.apps.service import buffer diff --git a/backend/apps/workflows/scheduler.py b/backend/apps/workflows/scheduler.py index 37c8f25c..5bb1f303 100644 --- a/backend/apps/workflows/scheduler.py +++ b/backend/apps/workflows/scheduler.py @@ -322,10 +322,11 @@ async def _fire(wf: Workflow, scheduled_for: Optional[datetime]) -> None: logger.exception("scheduler fire failed for workflow=%s", wf.id) -def _seconds_until_next() -> float: - # While globally paused, _tick no-ops and never rolls next_run_at forward, so an overdue slot would otherwise spin this loop at the 1s floor. Resume calls kick(), so idling the full interval here costs nothing. +def seconds_to_next_fire() -> Optional[float]: + """Seconds until the soonest enabled scheduled workflow fires; None when nothing is + queued or scheduling is globally paused. Also feeds the desktop's idle-update gate.""" if storage.get_paused(): - return 60.0 + return None now_utc = datetime.now(timezone.utc) soonest: Optional[datetime] = None for wf in storage.list_workflows(): @@ -337,8 +338,15 @@ def _seconds_until_next() -> float: if soonest is None or nra < soonest: soonest = nra if soonest is None: + return None + return max(0.0, (soonest - now_utc).total_seconds()) + + +def _seconds_until_next() -> float: + # While globally paused, _tick no-ops and never rolls next_run_at forward, so an overdue slot would otherwise spin this loop at the 1s floor. Resume calls kick(), so idling the full interval here costs nothing. + delta = seconds_to_next_fire() + if delta is None: return 60.0 - delta = (soonest - now_utc).total_seconds() return max(1.0, min(delta, 60.0)) diff --git a/backend/tests/test_idle_update_gate.py b/backend/tests/test_idle_update_gate.py new file mode 100644 index 00000000..be7c2ae2 --- /dev/null +++ b/backend/tests/test_idle_update_gate.py @@ -0,0 +1,63 @@ +"""API surface of the desktop idle-update gate: the /agents/activity lookahead +fields Electron polls before a silent install, and the /service/updater-event +breadcrumb it fires when one happens.""" + +from __future__ import annotations + +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest +from pydantic import ValidationError + + +@pytest.mark.asyncio +async def test_activity_reports_active_and_next_run(monkeypatch): + from backend.apps.agents import agents as agents_module + monkeypatch.setattr(agents_module.agent_manager, "tasks", {}) + monkeypatch.setattr("backend.apps.workflows.scheduler.seconds_to_next_fire", lambda: 123.0) + out = await agents_module.agent_activity() + assert out == {"active": 0, "next_run_in_s": 123.0} + + +@pytest.mark.asyncio +async def test_activity_lookahead_fails_open(monkeypatch): + from backend.apps.agents import agents as agents_module + + def boom(): + raise RuntimeError("lookahead broke") + + monkeypatch.setattr(agents_module.agent_manager, "tasks", {}) + monkeypatch.setattr("backend.apps.workflows.scheduler.seconds_to_next_fire", boom) + out = await agents_module.agent_activity() + assert out["next_run_in_s"] is None + + +@pytest.mark.asyncio +async def test_updater_event_writes_analytics_log(monkeypatch): + from backend.apps.service import service as service_module + from backend.apps.service.analytics import client as analytics_client_module + logs = Mock() + monkeypatch.setattr(analytics_client_module, "get_analytics_client", lambda: SimpleNamespace(logs=logs)) + body = service_module.UpdaterEventBody(kind="idle_install", staged_version="1.5.9") + out = await service_module.post_updater_event(body) + assert out == {"ok": True} + kw = logs.write.call_args.kwargs + assert kw["tag"] == "updater" + assert kw["subtag"] == "idle_install" + assert kw["data"]["staged_version"] == "1.5.9" + + +@pytest.mark.asyncio +async def test_updater_event_survives_missing_client(monkeypatch): + from backend.apps.service import service as service_module + from backend.apps.service.analytics import client as analytics_client_module + monkeypatch.setattr(analytics_client_module, "get_analytics_client", lambda: None) + out = await service_module.post_updater_event(service_module.UpdaterEventBody(kind="idle_install")) + assert out == {"ok": True} + + +def test_updater_event_kind_is_constrained(): + from backend.apps.service.service import UpdaterEventBody + with pytest.raises(ValidationError): + UpdaterEventBody(kind="anything_else") diff --git a/backend/tests/test_workflows_semantics.py b/backend/tests/test_workflows_semantics.py index 4077254f..8c544775 100644 --- a/backend/tests/test_workflows_semantics.py +++ b/backend/tests/test_workflows_semantics.py @@ -888,3 +888,52 @@ def test_escalation_noop_for_single_tier(): run = WorkflowRun(workflow_id=wf.id, status="success") escalation.schedule(wf, run) assert escalation.status(run.id) is None + + +# --- idle-update gate lookahead ---------------------------------------------- + +def test_seconds_to_next_fire_none_when_nothing_queued(): + from backend.apps.workflows import storage + from backend.apps.workflows.scheduler import seconds_to_next_fire + storage.init() + assert seconds_to_next_fire() is None + + +def test_seconds_to_next_fire_reports_soonest_enabled_only(): + from backend.apps.workflows import storage + from backend.apps.workflows.scheduler import seconds_to_next_fire + storage.init() + soon = _make_wf() + soon.next_run_at = datetime.now(timezone.utc) + timedelta(minutes=10) + storage.save_workflow(soon) + later = _make_wf() + later.next_run_at = datetime.now(timezone.utc) + timedelta(hours=3) + storage.save_workflow(later) + disabled = _make_wf() + disabled.schedule.enabled = False + disabled.next_run_at = datetime.now(timezone.utc) + timedelta(minutes=1) + storage.save_workflow(disabled) + got = seconds_to_next_fire() + assert got is not None + assert 9 * 60 < got <= 10 * 60 + + +def test_seconds_to_next_fire_clamps_overdue_to_zero(): + from backend.apps.workflows import storage + from backend.apps.workflows.scheduler import seconds_to_next_fire + storage.init() + wf = _make_wf() + wf.next_run_at = datetime.now(timezone.utc) - timedelta(minutes=5) + storage.save_workflow(wf) + assert seconds_to_next_fire() == 0.0 + + +def test_seconds_to_next_fire_none_while_paused(monkeypatch): + from backend.apps.workflows import storage + from backend.apps.workflows.scheduler import seconds_to_next_fire + storage.init() + wf = _make_wf() + wf.next_run_at = datetime.now(timezone.utc) + timedelta(minutes=1) + storage.save_workflow(wf) + monkeypatch.setattr(storage, "get_paused", lambda: True) + assert seconds_to_next_fire() is None diff --git a/electron/main.js b/electron/main.js index f6debeac..87c6a258 100644 --- a/electron/main.js +++ b/electron/main.js @@ -1633,18 +1633,40 @@ function setupAutoUpdater() { // the button uses. Deliberately conservative so it can never land on top of a live task. const IDLE_INSTALL_MIN_IDLE_S = 30 * 60; const IDLE_INSTALL_MIN_UPTIME_MS = 2 * 60 * 60 * 1000; + const IDLE_INSTALL_WORKFLOW_LOOKAHEAD_S = 15 * 60; const _idleInstallStart = Date.now(); - const _backendActiveAgents = () => new Promise((resolve) => { - if (!backendPort) return resolve(-1); + const _backendActivity = () => new Promise((resolve) => { + if (!backendPort) return resolve(null); const req = http.request({ hostname: '127.0.0.1', port: backendPort, path: '/api/agents/activity', method: 'GET', headers: { ...(authToken ? { Authorization: `Bearer ${authToken}` } : {}) }, timeout: 4000, }, (res) => { let d = ''; res.on('data', (c) => (d += c)); - res.on('end', () => { try { resolve(Number(JSON.parse(d).active)); } catch (_) { resolve(-1); } }); + res.on('end', () => { + try { + const j = JSON.parse(d); + resolve({ active: Number(j.active), nextRunInS: j.next_run_in_s == null ? null : Number(j.next_run_in_s) }); + } catch (_) { resolve(null); } + }); }); - req.on('error', () => resolve(-1)); - req.on('timeout', () => { req.destroy(); resolve(-1); }); + req.on('error', () => resolve(null)); + req.on('timeout', () => { req.destroy(); resolve(null); }); + req.end(); + }); + // Breadcrumb so fleet convergence is queryable in analytics; bounded + best-effort, the install never waits on it failing. + const _reportIdleInstall = () => new Promise((resolve) => { + if (!backendPort) return resolve(); + const payload = JSON.stringify({ + kind: 'idle_install', + staged_version: (cachedUpdateStatus && cachedUpdateStatus.info && cachedUpdateStatus.info.version) || null, + }); + const req = http.request({ + hostname: '127.0.0.1', port: backendPort, path: '/api/service/updater-event', method: 'POST', + headers: { 'Content-Type': 'application/json', ...(authToken ? { Authorization: `Bearer ${authToken}` } : {}) }, timeout: 2000, + }, (res) => { res.resume(); res.on('end', resolve); }); + req.on('error', resolve); + req.on('timeout', () => { req.destroy(); resolve(); }); + req.write(payload); req.end(); }); setInterval(async () => { @@ -1652,9 +1674,12 @@ function setupAutoUpdater() { if (isInstallingUpdate || !cachedUpdateStatus || cachedUpdateStatus.status !== 'downloaded') return; if (Date.now() - _idleInstallStart < IDLE_INSTALL_MIN_UPTIME_MS) return; if (powerMonitor.getSystemIdleTime() < IDLE_INSTALL_MIN_IDLE_S) return; - const active = await _backendActiveAgents(); - if (active !== 0) return; // unknown (-1) or busy -> stay put, never interrupt a task - console.log('[updater] staged update + machine idle + no agents; applying silently'); + const act = await _backendActivity(); + if (!act || act.active !== 0) return; // unknown or busy -> stay put, never interrupt a task + // A scheduled workflow fires soon; restarting now would race it. Let it run, catch the next idle window. + if (act.nextRunInS != null && act.nextRunInS < IDLE_INSTALL_WORKFLOW_LOOKAHEAD_S) return; + console.log('[updater] staged update + machine idle + no agents + no imminent workflow; applying silently'); + try { await _reportIdleInstall(); } catch (_) {} installDownloadedUpdate(); } catch (_) { /* a heartbeat must never throw */ } }, 5 * 60 * 1000);