fix: enforce global concurrent-run budget for manual triggers (#4769)

* Initial plan

* fix: enforce global concurrent-run budget for manual triggers

Manual triggers now check count_active_runs() before dispatching and
return a conflict result (409 at the router) when max_concurrent_runs
is already reached, preventing the global cap from being exceeded.

Co-authored-by: WillemJiang <219644+WillemJiang@users.noreply.github.com>

---------

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: WillemJiang <219644+WillemJiang@users.noreply.github.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

backend/app/scheduler/service.py

@@ -183,6 +183,25 @@ class ScheduledTaskService:
                 return self._active_run_conflict_result(execution_thread_id)
             return await self._record_scheduled_skip(task, thread_id=execution_thread_id, now=now, trigger=trigger)
 
+        # Global concurrent-run budget check for manual triggers.  The poller
+        # enforces max_concurrent_runs via count_active_runs() before
+        # claim_due_tasks(); a manual trigger bypasses that path and must apply
+        # the same cap so it cannot push the active count above the limit.
+        # Like the poller's count this is a non-atomic fast path; the partial
+        # unique index uq_scheduled_task_run_active is the atomic arbiter that
+        # rejects a second active insert for the *same task*, but there is no
+        # DB-level constraint that caps the global count, so we treat this as a
+        # best-effort guard consistent with how the poller enforces the budget.
+        if trigger == "manual" and self._max_concurrent_runs > 0:
+            active = await self._task_run_repo.count_active_runs()
+            if active >= self._max_concurrent_runs:
+                return {
+                    "outcome": "conflict",
+                    "task_run_id": None,
+                    "run_id": None,
+                    "thread_id": execution_thread_id,
+                    "error": "global concurrent-run limit reached",
+                }
         if self._multi_instance and trigger == "manual":
             task = await self._task_repo.claim_dispatch_lease(
                 task["id"],

backend/tests/test_scheduled_task_service.py

@@ -1001,3 +1001,63 @@ async def test_malformed_launch_result_still_retains_active_slot():
 
     first_row_id = run_repo.created[0]["run_record_id"]
     assert run_repo.rows[first_row_id]["status"] == "running"
+
+
+@pytest.mark.asyncio
+async def test_manual_trigger_rejected_when_global_budget_exhausted():
+    """Manual trigger must return a conflict when max_concurrent_runs is reached."""
+    launched = []
+
+    async def fake_launch(**kwargs):
+        launched.append(kwargs)
+        return {"run_id": "run-budget", "thread_id": kwargs["thread_id"]}
+
+    row = _once_task_row(task_id="task-budget", status="enabled")
+    row.update({"schedule_type": "cron", "schedule_spec": {"cron": "* * * * *"}, "overlap_policy": "skip"})
+    task_repo = DummyTaskRepo([row])
+    # active_count equals max_concurrent_runs → budget is exhausted
+    run_repo = DummyRunRepo(active_count=3)
+    service = ScheduledTaskService(
+        task_repo=task_repo,
+        task_run_repo=run_repo,
+        launch_run=fake_launch,
+        poll_interval_seconds=5,
+        lease_seconds=120,
+        max_concurrent_runs=3,
+    )
+
+    result = await service.dispatch_task(row, now=datetime.now(UTC), trigger="manual")
+
+    assert result["outcome"] == "conflict"
+    assert "limit" in result["error"]
+    assert launched == []
+    assert run_repo.created is None
+
+
+@pytest.mark.asyncio
+async def test_manual_trigger_proceeds_when_global_budget_available():
+    """Manual trigger must launch when active count is below max_concurrent_runs."""
+    launched = []
+
+    async def fake_launch(**kwargs):
+        launched.append(kwargs)
+        return {"run_id": "run-ok", "thread_id": kwargs["thread_id"]}
+
+    row = _once_task_row(task_id="task-ok", status="enabled")
+    row.update({"schedule_type": "cron", "schedule_spec": {"cron": "* * * * *"}, "overlap_policy": "skip"})
+    task_repo = DummyTaskRepo([row])
+    # active_count is 2, max_concurrent_runs is 3 → one slot left
+    run_repo = DummyRunRepo(active_count=2)
+    service = ScheduledTaskService(
+        task_repo=task_repo,
+        task_run_repo=run_repo,
+        launch_run=fake_launch,
+        poll_interval_seconds=5,
+        lease_seconds=120,
+        max_concurrent_runs=3,
+    )
+
+    result = await service.dispatch_task(row, now=datetime.now(UTC), trigger="manual")
+
+    assert result["outcome"] == "launched"
+    assert len(launched) == 1