[eric] workflows: pause/trash halts the in-flight run (executor per-step liveness recheck + stop-on-delete)

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

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:

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