[eric] workflows: at most 3 run at once, extra due fires queue for a slot and give up honestly, so 30 saved workflows can never become 30 live agents

backend/apps/workflows/executor.py

@@ -7,6 +7,7 @@ routing, retries, and history all aligned with the rest of the app.
 """
 
 import asyncio
+import time
 import logging
 from datetime import datetime, timedelta, timezone
 from typing import Optional
@@ -21,6 +22,13 @@ logger = logging.getLogger(__name__)
 
 # In-process map: workflow_id -> currently running run id. Prevents two overlapping fires for the same workflow (e.g. cron tick races a manual Run button) without serializing across the whole executor.
 _running: dict[str, str] = {}
+# Global admission on top of the per-workflow guard: every run is a full agent, and an agent's
+# browsers and apps are exempt from the renderer budget by design (sleeping a working agent's browser
+# blinds it), so the ONLY thing bounding total pressure is how many runs exist at once. A library of
+# 30 workflows whose schedules drift into alignment must queue, not stampede.
+MAX_CONCURRENT_RUNS = 3
+ADMISSION_WAIT_S = 600.0
+ADMISSION_POLL_S = 2.0
 _running_lock = asyncio.Lock()
 
 
@@ -211,6 +219,25 @@ async def execute(
     # the scheduler, an agent tool, an invoke, a retry, the Run Now route, or a stale in-flight handle.
     # Guarding this per call site left every unguarded caller able to fire it, which is the field report
     # of a toggled-off workflow running itself. Turn it back on to run it.
+    # Wait for a global slot BEFORE the off-means-off guard, so the guard runs on fresh state after
+    # a possibly long wait (a workflow paused while queueing still gets refused, not run).
+    p_admit_start = time.monotonic()
+    while len(_running) >= MAX_CONCURRENT_RUNS:
+        if time.monotonic() - p_admit_start >= ADMISSION_WAIT_S:
+            p_busy = WorkflowRun(
+                workflow_id=wf.id, status="skipped",
+                error=f"{MAX_CONCURRENT_RUNS} workflows already running; gave up after {int(ADMISSION_WAIT_S)}s",
+                scheduled_for=scheduled_for, started_at=datetime.now(),
+                finished_at=datetime.now(), triggered_by=triggered_by,
+            )
+            p_wf_now = storage.get_workflow(wf.id)
+            if p_wf_now is not None and p_wf_now.deleted_at is None:
+                try:
+                    storage.record_run(p_busy)
+                except Exception:
+                    logger.debug("could not record the admission-skip row", exc_info=True)
+            return p_busy
+        await asyncio.sleep(ADMISSION_POLL_S)
     p_live = storage.get_workflow(wf.id)
     p_refusal = None
     if p_live is None:

backend/tests/test_workflow_run_admission.py

@@ -0,0 +1,102 @@
+"""No stampede: a due fire beyond the global cap queues, and gives up honestly, never silently.
+
+Every workflow run is a full agent, and a working agent's browsers and apps are exempt from the
+renderer budget on purpose (sleeping them blinds the agent). So the ONLY bound on total pressure is
+how many runs exist at once, and before this the scheduler would happily start one agent per due
+workflow: thirty accumulated workflows drifting into schedule alignment meant thirty agents.
+"""
+
+import asyncio
+
+import pytest
+
+from backend.apps.workflows import executor, storage
+from backend.apps.workflows.models import Workflow, WorkflowStep
+
+
+def p_make() -> Workflow:
+    return Workflow(
+        title="queued",
+        steps=[WorkflowStep(text="do it")],
+        schedule={"enabled": True, "kind": "interval", "every_minutes": 45},
+    )
+
+
+@pytest.fixture(autouse=True)
+def p_isolated(tmp_path, monkeypatch):
+    monkeypatch.setattr(storage, "DATA_DIR", str(tmp_path / "workflows"))
+    monkeypatch.setattr(storage, "RUNS_DIR", str(tmp_path / "workflows" / "runs"))
+    monkeypatch.setattr(storage, "PAUSED_FILE", str(tmp_path / "workflows" / "paused.json"))
+    monkeypatch.setattr(storage, "MISSED_FILE", str(tmp_path / "workflows" / "missed.json"))
+    monkeypatch.setattr(storage, "_workflow_cache", {})
+    monkeypatch.setattr(storage, "_runs_cache", {})
+    monkeypatch.setattr(storage, "_missed_cache", [])
+    monkeypatch.setattr(storage, "p_deleted_ids", set(), raising=False)
+    monkeypatch.setattr(storage, "_cache_loaded", True)
+    monkeypatch.setattr(executor, "_running", {})
+    yield
+
+
+def test_at_the_cap_a_fire_waits_then_skips_with_an_honest_row(monkeypatch):
+    monkeypatch.setattr(executor, "MAX_CONCURRENT_RUNS", 2)
+    monkeypatch.setattr(executor, "ADMISSION_WAIT_S", 0.05)
+    monkeypatch.setattr(executor, "ADMISSION_POLL_S", 0.01)
+    monkeypatch.setattr(executor, "_running", {"other-a": "r1", "other-b": "r2"})
+
+    wf = storage.save_workflow(p_make())
+    run = asyncio.run(executor.execute(wf, triggered_by="schedule"))
+
+    assert run.status == "skipped"
+    assert "already running" in (run.error or "")
+    rows = storage.list_runs(wf.id)
+    assert len(rows) == 1 and rows[0].status == "skipped", "the give-up must be visible in History"
+
+
+def test_a_freed_slot_lets_the_queued_fire_proceed(monkeypatch):
+    """The wait is a queue, not a rejection: the moment a slot frees, the run goes ahead and reaches
+    the normal guard path (here: refused as paused, which proves it got past admission)."""
+    monkeypatch.setattr(executor, "MAX_CONCURRENT_RUNS", 1)
+    monkeypatch.setattr(executor, "ADMISSION_WAIT_S", 5.0)
+    monkeypatch.setattr(executor, "ADMISSION_POLL_S", 0.01)
+    monkeypatch.setattr(executor, "_running", {"other-a": "r1"})
+
+    wf = storage.save_workflow(p_make())
+    live = storage.get_workflow(wf.id)
+    live.schedule.enabled = False
+    storage.save_workflow(live)
+
+    async def scenario():
+        async def free_slot_soon():
+            await asyncio.sleep(0.05)
+            executor._running.clear()
+        asyncio.ensure_future(free_slot_soon())
+        return await executor.execute(wf, triggered_by="schedule")
+
+    run = asyncio.run(scenario())
+    assert run.status == "skipped"
+    assert "paused" in (run.error or "").lower(), (
+        f"expected the post-wait guard to refuse the paused workflow, got {run.error!r}"
+    )
+
+
+def test_the_guard_runs_on_state_AFTER_the_wait(monkeypatch):
+    """A workflow deleted while queueing must be refused, not run: admission sits before the
+    off-means-off guard precisely so the guard sees post-wait truth."""
+    monkeypatch.setattr(executor, "MAX_CONCURRENT_RUNS", 1)
+    monkeypatch.setattr(executor, "ADMISSION_WAIT_S", 5.0)
+    monkeypatch.setattr(executor, "ADMISSION_POLL_S", 0.01)
+    monkeypatch.setattr(executor, "_running", {"other-a": "r1"})
+
+    wf = storage.save_workflow(p_make())
+
+    async def scenario():
+        async def delete_then_free():
+            await asyncio.sleep(0.05)
+            storage.delete_workflow(wf.id)
+            executor._running.clear()
+        asyncio.ensure_future(delete_then_free())
+        return await executor.execute(wf, triggered_by="schedule")
+
+    run = asyncio.run(scenario())
+    assert run.status == "skipped"
+    assert "delete" in (run.error or "").lower()