diff --git a/backend/apps/workflows/executor.py b/backend/apps/workflows/executor.py index 357396ab..fee06152 100644 --- a/backend/apps/workflows/executor.py +++ b/backend/apps/workflows/executor.py @@ -35,12 +35,25 @@ def p_ran_late(started_at: datetime, scheduled_for: datetime) -> bool: # run_id -> "stop". Set by the stop endpoint so the executor loop, not the HTTP handler, owns the run's terminal write. Without this the still-running executor task could overwrite a "Stopped by user" failure with success. Pause is NOT in here: it rides the agent session's own "stopped" status, which the step loop waits out (see _await_session_idle). _run_control: dict[str, str] = {} _run_pause_override: dict[str, tuple[bool, float]] = {} +# workflow_id -> session_id of the in-flight run, kept in step with _running so pause/trash can halt a live run even for a workflow that list_workflows() now filters out (deleted). +_running_session: dict[str, str] = {} def request_stop(run_id: str) -> None: _run_control[run_id] = "stop" +def stop_active_run(workflow_id: str) -> Optional[str]: + """Signal the in-flight run for this workflow to stop and return its session id + (None if nothing is running). Lets delete/pause halt a live run NOW instead of only + stopping future fires; reads the workflow_id-keyed maps so it still works after trash.""" + run_id = _running.get(workflow_id) + if not run_id: + return None + _run_control[run_id] = "stop" + return _running_session.get(workflow_id) + + def set_pause_override(run_id: str, paused: bool, ttl_s: float = 5.0) -> None: """Keep an explicit pause/resume control state authoritative briefly. @@ -268,6 +281,7 @@ async def execute( session = await agent_manager.launch_agent(config) run.session_id = session.id + _running_session[wf.id] = session.id storage.record_run(run) # Reuse the user's earlier allow/deny answers so an unattended fire doesn't park on a permission prompt. Scheduled runs prompt for an unseen tool only briefly (30s) before failing; manual/test runs are attended, so keep the roomy window. Sensitive-path prompts are never remembered (handled by the gate); they keep prompting every run. @@ -343,6 +357,14 @@ async def execute( if _run_control.get(run.id) == "stop": step_error = "Stopped by user" break + # Re-read live state each step so pause/trash actually halts an in-flight run at the next step boundary. The scheduler tick + routes only stop FUTURE fires; without this a run keeps stepping after the workflow is trashed (any trigger) or, for a scheduled run, paused. Manual "Run Now" of a paused workflow is deliberately left alone. + fresh_wf = storage.get_workflow(wf.id) + if fresh_wf is None or fresh_wf.deleted_at is not None: + step_error = "Workflow deleted" + break + if triggered_by == "schedule" and not fresh_wf.schedule.enabled: + step_error = "Workflow paused" + break # Broadcast the step bump before sending so RunningView flips the disc immediately, not after the agent finishes the step. Advancing means we're not paused; keep the broadcast authoritative so it never races a stale paused=True from the watcher. run.active_step_idx = idx run.last_tool_label = None @@ -427,6 +449,7 @@ async def execute( logger.exception("close_session failed for workflow run %s", run.id) async with _running_lock: _running.pop(wf.id, None) + _running_session.pop(wf.id, None) try: from backend.apps.workflows.notifier import notify_run_complete diff --git a/backend/apps/workflows/workflows.py b/backend/apps/workflows/workflows.py index c97c95aa..d01a9056 100644 --- a/backend/apps/workflows/workflows.py +++ b/backend/apps/workflows/workflows.py @@ -832,6 +832,19 @@ async def update_workflow( return enriched +async def _stop_in_flight_run(workflow_id: str) -> None: + """Halt a workflow's live run now. The executor's per-step recheck is the backstop; + this stops the current agent turn without waiting a long step out. Works even after + trash (reads executor's workflow-keyed maps, not the deleted-filtered run list).""" + session_id = executor.stop_active_run(workflow_id) + if session_id: + try: + from backend.apps.agents.agent_manager import agent_manager + await agent_manager.stop_agent(session_id) + except Exception: + logger.exception("_stop_in_flight_run: stop_agent failed for %s", session_id) + + @workflows.router.delete("/{workflow_id}") async def delete_workflow(workflow_id: str): """Soft-delete: move to Trash. The record stays on disk with deleted_at @@ -844,6 +857,8 @@ async def delete_workflow(workflow_id: str): wf.schedule.enabled = False wf.next_run_at = None storage.save_workflow(wf) + # A trashed workflow's in-flight run was previously un-stoppable even by hand (the manual Stop path filters deleted); halt it now, with the executor's per-step deleted-recheck as backstop. + await _stop_in_flight_run(workflow_id) # Drop any pending missed fires so a trashed workflow can't haunt the card. stale = [m.id for m in storage.list_missed() if m.workflow_id == workflow_id] if stale: diff --git a/backend/tests/test_scheduled_stop_on_pause_trash.py b/backend/tests/test_scheduled_stop_on_pause_trash.py new file mode 100644 index 00000000..ea2bd7ae --- /dev/null +++ b/backend/tests/test_scheduled_stop_on_pause_trash.py @@ -0,0 +1,133 @@ +"""A paused or trashed workflow must stop its in-flight run, not run to completion. + +The bug: the scheduler stopped FUTURE fires on pause/trash, but an already-running +execution kept stepping (the executor never re-read the workflow's live state), and a +trashed run couldn't even be stopped by hand. These drive executor.execute() end to end +(faked agent) and mutate the workflow mid-run to prove the run halts at the next step. +""" + +from __future__ import annotations + +import asyncio +from datetime import datetime + +import pytest + +from backend.apps.workflows.models import WorkflowStep + + +@pytest.fixture(autouse=True) +def p_wf_env(isolated_workflows_data, reset_scheduler_state): + yield + + +def p_run(coro): + return asyncio.new_event_loop().run_until_complete(coro) + + +def p_three_step_wf(make_wf): + return make_wf(steps=[WorkflowStep(text="step1"), WorkflowStep(text="step2"), WorkflowStep(text="step3")]) + + +def p_mutate_after_first_step(monkeypatch, fake_agent_manager, mutate): + """Wrap the faked send_message so `mutate()` runs right after step1 is sent.""" + from backend.apps.agents import agent_manager as p_am + orig = p_am.agent_manager.send_message + + async def wrapped(session_id, text, hidden=False): + await orig(session_id, text, hidden=hidden) + if text == "step1": + mutate() + + monkeypatch.setattr(p_am.agent_manager, "send_message", wrapped) + + +def test_trashed_scheduled_run_halts_at_next_step(make_wf, fake_agent_manager, monkeypatch): + from backend.apps.workflows import storage, executor + wf = p_three_step_wf(make_wf) + storage.save_workflow(wf) + + def trash(): + w = storage.get_workflow(wf.id) + w.deleted_at = datetime.now() + storage.save_workflow(w) + + p_mutate_after_first_step(monkeypatch, fake_agent_manager, trash) + run = p_run(executor.execute(wf, triggered_by="schedule")) + assert run.status == "failure" + assert run.error == "Workflow deleted" + assert fake_agent_manager.sent_messages == ["step1"] # step2/step3 never sent + + +def test_trashed_manual_run_also_halts(make_wf, fake_agent_manager, monkeypatch): + """A deleted workflow stops for ANY trigger, not just scheduled.""" + from backend.apps.workflows import storage, executor + wf = p_three_step_wf(make_wf) + storage.save_workflow(wf) + + def trash(): + w = storage.get_workflow(wf.id) + w.deleted_at = datetime.now() + storage.save_workflow(w) + + p_mutate_after_first_step(monkeypatch, fake_agent_manager, trash) + run = p_run(executor.execute(wf, triggered_by="manual")) + assert run.error == "Workflow deleted" + assert fake_agent_manager.sent_messages == ["step1"] + + +def test_paused_scheduled_run_halts_at_next_step(make_wf, fake_agent_manager, monkeypatch): + from backend.apps.workflows import storage, executor + wf = p_three_step_wf(make_wf) + storage.save_workflow(wf) + + def pause(): + w = storage.get_workflow(wf.id) + w.schedule.enabled = False + storage.save_workflow(w) + + p_mutate_after_first_step(monkeypatch, fake_agent_manager, pause) + run = p_run(executor.execute(wf, triggered_by="schedule")) + assert run.status == "failure" + assert run.error == "Workflow paused" + assert fake_agent_manager.sent_messages == ["step1"] + + +def test_pause_does_not_halt_a_manual_run(make_wf, fake_agent_manager, monkeypatch): + """Pausing the SCHEDULE must not kill a manual Run Now that's in flight.""" + from backend.apps.workflows import storage, executor + wf = p_three_step_wf(make_wf) + storage.save_workflow(wf) + + def pause(): + w = storage.get_workflow(wf.id) + w.schedule.enabled = False + storage.save_workflow(w) + + p_mutate_after_first_step(monkeypatch, fake_agent_manager, pause) + run = p_run(executor.execute(wf, triggered_by="manual")) + assert run.status == "success" + assert fake_agent_manager.sent_messages == ["step1", "step2", "step3"] # all ran + + +def test_stop_active_run_signals_and_returns_session(make_wf, fake_agent_manager, monkeypatch): + """delete/pause use executor.stop_active_run to halt a live run even after trash.""" + from backend.apps.workflows import storage, executor + wf = p_three_step_wf(make_wf) + storage.save_workflow(wf) + captured: dict = {} + + def grab_then_trash(): + # While the run is live, stop_active_run must find its session and signal stop. + captured["session"] = executor.stop_active_run(wf.id) + + p_mutate_after_first_step(monkeypatch, fake_agent_manager, grab_then_trash) + run = p_run(executor.execute(wf, triggered_by="schedule")) + assert captured["session"] is not None # found the live session + assert run.error == "Stopped by user" # _run_control stop took effect + assert fake_agent_manager.sent_messages == ["step1"] + + +def test_no_running_workflow_returns_none(make_wf, fake_agent_manager): + from backend.apps.workflows import executor + assert executor.stop_active_run("no-such-workflow") is None