From 346e5673de8dd2e2ccc11817c81323dbb16e78f2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E7=B6=A0BG?= Date: Sat, 13 Jun 2026 17:02:35 +0800 Subject: [PATCH] =?UTF-8?q?=F0=9F=90=9B=20fix(dashboard):=20offload=20cron?= =?UTF-8?q?=20profile=20scans?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit (cherry picked from commit 1fded709aaf79ba426ab360ac785fcdfae8229f0) --- hermes_cli/web_server.py | 77 ++++++++++++++----- .../test_web_server_cron_profiles.py | 43 +++++++++++ 2 files changed, 102 insertions(+), 18 deletions(-) diff --git a/hermes_cli/web_server.py b/hermes_cli/web_server.py index 00e487ff664..00931549fe3 100644 --- a/hermes_cli/web_server.py +++ b/hermes_cli/web_server.py @@ -9985,8 +9985,7 @@ def _find_cron_job_profile(job_id: str) -> Optional[str]: return None -@app.get("/api/cron/jobs") -async def list_cron_jobs(profile: str = "all"): +def _list_cron_jobs_sync(profile: str = "all"): requested = (profile or "all").strip() if requested.lower() != "all": return _call_cron_for_profile(requested, "list_jobs", True) @@ -10003,8 +10002,17 @@ async def list_cron_jobs(profile: str = "all"): return jobs -@app.get("/api/cron/jobs/{job_id}") -async def get_cron_job(job_id: str, profile: Optional[str] = None): +async def _run_cron_dashboard_io(func, *args, **kwargs): + """Run cron dashboard profile/job I/O outside the FastAPI event loop.""" + return await asyncio.to_thread(func, *args, **kwargs) + + +@app.get("/api/cron/jobs") +async def list_cron_jobs(profile: str = "all"): + return await _run_cron_dashboard_io(_list_cron_jobs_sync, profile) + + +def _get_cron_job_sync(job_id: str, profile: Optional[str] = None): selected = profile or _find_cron_job_profile(job_id) if not selected: raise HTTPException(status_code=404, detail="Job not found") @@ -10014,8 +10022,12 @@ async def get_cron_job(job_id: str, profile: Optional[str] = None): return job -@app.get("/api/cron/jobs/{job_id}/runs") -async def list_cron_job_runs(job_id: str, profile: Optional[str] = None, limit: int = 20): +@app.get("/api/cron/jobs/{job_id}") +async def get_cron_job(job_id: str, profile: Optional[str] = None): + return await _run_cron_dashboard_io(_get_cron_job_sync, job_id, profile) + + +def _list_cron_job_runs_sync(job_id: str, profile: Optional[str] = None, limit: int = 20): """Run sessions produced by a cron job, newest first. Cron runs are stored as ordinary sessions whose id is @@ -10060,8 +10072,12 @@ async def list_cron_job_runs(job_id: str, profile: Optional[str] = None, limit: db.close() -@app.post("/api/cron/jobs") -async def create_cron_job(body: CronJobCreate, profile: str = "default"): +@app.get("/api/cron/jobs/{job_id}/runs") +async def list_cron_job_runs(job_id: str, profile: Optional[str] = None, limit: int = 20): + return await _run_cron_dashboard_io(_list_cron_job_runs_sync, job_id, profile, limit) + + +def _create_cron_job_sync(body: CronJobCreate, profile: str = "default"): try: profile_name, profile_home = _cron_profile_home(profile) script = _normalize_dashboard_cron_script(body.script, profile_home) @@ -10099,6 +10115,11 @@ async def create_cron_job(body: CronJobCreate, profile: str = "default"): raise HTTPException(status_code=400, detail=str(e)) +@app.post("/api/cron/jobs") +async def create_cron_job(body: CronJobCreate, profile: str = "default"): + return await _run_cron_dashboard_io(_create_cron_job_sync, body, profile) + + @app.get("/api/cron/delivery-targets") async def get_cron_delivery_targets(): """Delivery targets the cron dropdown should offer. @@ -10127,8 +10148,7 @@ async def get_cron_delivery_targets(): return {"targets": targets} -@app.put("/api/cron/jobs/{job_id}") -async def update_cron_job(job_id: str, body: CronJobUpdate, profile: Optional[str] = None): +def _update_cron_job_sync(job_id: str, body: CronJobUpdate, profile: Optional[str] = None): selected = profile or _find_cron_job_profile(job_id) if not selected: raise HTTPException(status_code=404, detail="Job not found") @@ -10162,8 +10182,12 @@ async def update_cron_job(job_id: str, body: CronJobUpdate, profile: Optional[st return job -@app.post("/api/cron/jobs/{job_id}/pause") -async def pause_cron_job(job_id: str, profile: Optional[str] = None): +@app.put("/api/cron/jobs/{job_id}") +async def update_cron_job(job_id: str, body: CronJobUpdate, profile: Optional[str] = None): + return await _run_cron_dashboard_io(_update_cron_job_sync, job_id, body, profile) + + +def _pause_cron_job_sync(job_id: str, profile: Optional[str] = None): selected = profile or _find_cron_job_profile(job_id) if not selected: raise HTTPException(status_code=404, detail="Job not found") @@ -10173,8 +10197,12 @@ async def pause_cron_job(job_id: str, profile: Optional[str] = None): return job -@app.post("/api/cron/jobs/{job_id}/resume") -async def resume_cron_job(job_id: str, profile: Optional[str] = None): +@app.post("/api/cron/jobs/{job_id}/pause") +async def pause_cron_job(job_id: str, profile: Optional[str] = None): + return await _run_cron_dashboard_io(_pause_cron_job_sync, job_id, profile) + + +def _resume_cron_job_sync(job_id: str, profile: Optional[str] = None): selected = profile or _find_cron_job_profile(job_id) if not selected: raise HTTPException(status_code=404, detail="Job not found") @@ -10184,8 +10212,12 @@ async def resume_cron_job(job_id: str, profile: Optional[str] = None): return job -@app.post("/api/cron/jobs/{job_id}/trigger") -async def trigger_cron_job(job_id: str, profile: Optional[str] = None): +@app.post("/api/cron/jobs/{job_id}/resume") +async def resume_cron_job(job_id: str, profile: Optional[str] = None): + return await _run_cron_dashboard_io(_resume_cron_job_sync, job_id, profile) + + +def _trigger_cron_job_sync(job_id: str, profile: Optional[str] = None): selected = profile or _find_cron_job_profile(job_id) if not selected: raise HTTPException(status_code=404, detail="Job not found") @@ -10195,8 +10227,12 @@ async def trigger_cron_job(job_id: str, profile: Optional[str] = None): return job -@app.delete("/api/cron/jobs/{job_id}") -async def delete_cron_job(job_id: str, profile: Optional[str] = None): +@app.post("/api/cron/jobs/{job_id}/trigger") +async def trigger_cron_job(job_id: str, profile: Optional[str] = None): + return await _run_cron_dashboard_io(_trigger_cron_job_sync, job_id, profile) + + +def _delete_cron_job_sync(job_id: str, profile: Optional[str] = None): selected = profile or _find_cron_job_profile(job_id) if not selected: raise HTTPException(status_code=404, detail="Job not found") @@ -10209,6 +10245,11 @@ async def delete_cron_job(job_id: str, profile: Optional[str] = None): return {"ok": True} +@app.delete("/api/cron/jobs/{job_id}") +async def delete_cron_job(job_id: str, profile: Optional[str] = None): + return await _run_cron_dashboard_io(_delete_cron_job_sync, job_id, profile) + + def _fire_cron_job_for_profile(profile: str, job_id: str) -> bool: """Run ONE due cron job end-to-end for ``profile`` via the resolved scheduler provider's ``fire_due`` (store CAS claim + ``run_one_job``). diff --git a/tests/hermes_cli/test_web_server_cron_profiles.py b/tests/hermes_cli/test_web_server_cron_profiles.py index f8fa1e008a5..0e6a268326e 100644 --- a/tests/hermes_cli/test_web_server_cron_profiles.py +++ b/tests/hermes_cli/test_web_server_cron_profiles.py @@ -1,5 +1,7 @@ """Regression tests for dashboard cron job profile routing.""" +import threading + import pytest from fastapi import HTTPException @@ -159,6 +161,47 @@ async def test_cron_mutation_without_profile_finds_named_profile_job(isolated_pr assert worker_jobs[0]["enabled"] is False +@pytest.mark.asyncio +async def test_cron_profile_scan_runs_off_event_loop(isolated_profiles, monkeypatch): + from hermes_cli import web_server + + worker_job = web_server._call_cron_for_profile( + "worker_alpha", + "create_job", + prompt="managed by named profile", + schedule="every 1h", + name="thread-offload-job", + ) + + event_loop_thread = threading.get_ident() + profile_scan_threads = [] + worker_threads = [] + original_profile_dicts = web_server._cron_profile_dicts + original_find = web_server._find_cron_job_profile + + def tracking_profile_dicts(): + profile_scan_threads.append(threading.get_ident()) + return original_profile_dicts() + + def tracking_find(job_id): + worker_threads.append(threading.get_ident()) + return original_find(job_id) + + monkeypatch.setattr(web_server, "_cron_profile_dicts", tracking_profile_dicts) + monkeypatch.setattr(web_server, "_find_cron_job_profile", tracking_find) + + jobs = await web_server.list_cron_jobs(profile="all") + paused = await web_server.pause_cron_job(worker_job["id"]) + + assert any(job["id"] == worker_job["id"] for job in jobs) + assert paused["profile"] == "worker_alpha" + assert profile_scan_threads + assert worker_threads + assert all(thread_id != event_loop_thread for thread_id in profile_scan_threads) + assert all(thread_id != event_loop_thread for thread_id in worker_threads) + + + @pytest.mark.asyncio async def test_update_cron_job_normalizes_dashboard_core_fields(isolated_profiles, tmp_path): from hermes_cli import web_server