From 506bcb7cbfa913d1da2f27d087d68428baf00189 Mon Sep 17 00:00:00 2001 From: abccodes Date: Thu, 18 Jun 2026 03:53:27 -0700 Subject: [PATCH] [aidan] feat/workflow-runs: add pause, resume, and stop controls for live runs --- backend/apps/workflows/executor.py | 107 +++++++++--- backend/apps/workflows/models.py | 4 + backend/apps/workflows/workflows.py | 140 +++++++++++----- backend/tests/test_workflows_semantics.py | 86 ++++++++++ .../src/app/pages/AgentChat/AgentChat.tsx | 28 +++- .../src/app/pages/Workflows/WorkflowCard.tsx | 65 ++++---- .../pages/Workflows/WorkflowCardLiveViews.tsx | 22 +-- frontend/src/shared/state/workflowsSlice.ts | 155 +++++++++++++----- 8 files changed, 448 insertions(+), 159 deletions(-) diff --git a/backend/apps/workflows/executor.py b/backend/apps/workflows/executor.py index 9361bd42..fb8188a1 100644 --- a/backend/apps/workflows/executor.py +++ b/backend/apps/workflows/executor.py @@ -25,6 +25,30 @@ _running: dict[str, str] = {} _running_lock = asyncio.Lock() +# 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]] = {} + + +def request_stop(run_id: str) -> None: + _run_control[run_id] = "stop" + + +def set_pause_override(run_id: str, paused: bool, ttl_s: float = 5.0) -> None: + """Keep an explicit pause/resume control state authoritative briefly. + + The tool watcher normally derives paused from the agent session status, + but pause/resume endpoints now return before the slower agent_manager call + finishes. This prevents the watcher from broadcasting the pre-control + status during that handoff window. + """ + _run_pause_override[run_id] = (paused, asyncio.get_event_loop().time() + ttl_s) + + def _resolve_system_prompt(wf: Workflow) -> Optional[str]: if wf.use_synced_prompt: return None @@ -222,12 +246,21 @@ async def execute(wf: Workflow, triggered_by: str = "schedule", scheduled_for: O # finally block alongside _running cleanup. async def _watch_tool_calls() -> None: last_seen = "" + last_paused = False while True: try: await asyncio.sleep(1.5) sess = agent_manager.sessions.get(session.id) if not sess: return + now = asyncio.get_event_loop().time() + override = _run_pause_override.get(run.id) + if override and override[1] >= now: + paused_now = override[0] + else: + if override: + _run_pause_override.pop(run.id, None) + paused_now = getattr(sess, "status", None) == "stopped" msgs = getattr(sess, "messages", []) or [] label = "" for m in reversed(msgs): @@ -247,9 +280,13 @@ async def execute(wf: Workflow, triggered_by: str = "schedule", scheduled_for: O label = content[:60] if label: break - if label and label != last_seen: - last_seen = label - run.last_tool_label = label + label_changed = bool(label) and label != last_seen + if label_changed or paused_now != last_paused: + if label_changed: + last_seen = label + run.last_tool_label = label + last_paused = paused_now + run.paused = paused_now try: from backend.apps.agents.core.ws_manager import ws_manager await ws_manager.broadcast_global("workflow:run", { @@ -271,10 +308,16 @@ async def execute(wf: Workflow, triggered_by: str = "schedule", scheduled_for: O # safe regardless of how long each turn takes. step_error: Optional[str] = None for idx, step in enumerate(steps): + if _run_control.get(run.id) == "stop": + step_error = "Stopped by user" + 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 + run.paused = False set_workflow_approval_step(session.id, step.id) try: from backend.apps.agents.core.ws_manager import ws_manager as _wsm @@ -285,15 +328,17 @@ async def execute(wf: Workflow, triggered_by: str = "schedule", scheduled_for: O except Exception: pass await agent_manager.send_message(session.id, step.text) - await _await_session_idle(session.id) - sess_state = agent_manager.sessions.get(session.id) - if sess_state is not None and getattr(sess_state, "status", None) == "error": + disp = await _await_session_idle(session.id, run.id) + if disp == "stopped": + step_error = "Stopped by user" + # Pin active step so FailedView renders the X on the right row. + break + if disp == "error": step_error = "Agent session entered error state" - # Pin active step so FailedView can render the X on the - # right row. error_step_idx == active_step_idx at fail time. break run.finished_at = datetime.now() + run.paused = False sess_state = agent_manager.sessions.get(session.id) if sess_state is not None: run.cost_usd = float(getattr(sess_state, "cost_usd", 0.0) or 0.0) @@ -327,6 +372,7 @@ async def execute(wf: Workflow, triggered_by: str = "schedule", scheduled_for: O run.status = "failure" run.error = str(e)[:500] run.finished_at = datetime.now() + run.paused = False storage.record_run(run) wf.last_run_status = "failure" _persist_run_fields(wf, { @@ -334,6 +380,8 @@ async def execute(wf: Workflow, triggered_by: str = "schedule", scheduled_for: O "last_run_at": run.finished_at, }) finally: + _run_control.pop(run.id, None) + _run_pause_override.pop(run.id, None) # Cancel the tool-call watcher before we tear the session down so # the next poll doesn't race close_session. try: @@ -377,26 +425,47 @@ async def execute(wf: Workflow, triggered_by: str = "schedule", scheduled_for: O return run -async def _await_session_idle(session_id: str, timeout_s: float = 600.0) -> None: - """Block until the agent session reaches a non-running terminal state. +async def _await_session_idle(session_id: str, run_id: Optional[str] = None, timeout_s: float = 600.0) -> str: + """Wait out the current step's agent turn. Returns a disposition: + 'idle' turn finished, advance to the next step + 'error' the agent session errored + 'stopped' the run was manually stopped (full stop) - Polls cheaply (50ms) since the agent_manager doesn't expose a per-session - completion future. Bounded by timeout_s so a stuck step doesn't hang the - runner forever. + For a real run (run_id given) a user PAUSE shows up as the session going + 'stopped' WITHOUT a stop signal; that is not terminal, so we hold here + until Resume or Stop, keeping the step deadline fresh so a long pause + doesn't fail the step. The attended test-run driver passes no run_id and + treats 'stopped' as terminal (no pause/resume there). + + Polls cheaply since agent_manager doesn't expose a per-session completion + future. Bounded by timeout_s so a stuck step can't hang the runner forever. """ from backend.apps.agents.agent_manager import agent_manager + hold_on_pause = run_id is not None deadline = asyncio.get_event_loop().time() + timeout_s while True: + if run_id is not None and _run_control.get(run_id) == "stop": + return "stopped" sess = agent_manager.sessions.get(session_id) if not sess: - return - task = agent_manager.tasks.get(session_id) - if task is not None and task.done(): - return + return "idle" status = getattr(sess, "status", None) - if status in ("completed", "error", "stopped"): - return + if status == "stopped": + if not hold_on_pause: + return "stopped" + # Paused. Hold, and reset the deadline so paused wall-time + # doesn't count against the step timeout. + deadline = asyncio.get_event_loop().time() + timeout_s + await asyncio.sleep(0.1) + continue + if status == "error": + return "error" + if status == "completed": + return "idle" + task = agent_manager.tasks.get(session_id) + if task is not None and task.done() and status not in ("running", "waiting_approval"): + return "idle" if asyncio.get_event_loop().time() > deadline: raise TimeoutError(f"Step exceeded {timeout_s}s on session {session_id}") await asyncio.sleep(0.05) diff --git a/backend/apps/workflows/models.py b/backend/apps/workflows/models.py index 80e4fa4c..8b134891 100644 --- a/backend/apps/workflows/models.py +++ b/backend/apps/workflows/models.py @@ -169,6 +169,10 @@ class WorkflowRun(BaseModel): # time it dispatches a step prompt and broadcasts the run. RunningView # uses this for the disc statuses; estimate fallback only when null. active_step_idx: Optional[int] = None + # True while the user has paused the in-flight agent turn (same mechanic + # as the chat's stop/resume). Rides the workflow:run broadcast so the + # card shows the paused state even when the live chat isn't open. + paused: bool = False class WorkflowCreate(BaseModel): diff --git a/backend/apps/workflows/workflows.py b/backend/apps/workflows/workflows.py index 48729d99..543d5579 100644 --- a/backend/apps/workflows/workflows.py +++ b/backend/apps/workflows/workflows.py @@ -891,9 +891,8 @@ async def test_run_workflow(workflow_id: str, body: dict): for step in step_entries: set_workflow_approval_step(session.id, step.id) await agent_manager.send_message(session.id, step.text) - await executor._await_session_idle(session.id) - sess_state = agent_manager.sessions.get(session.id) - if sess_state is not None and getattr(sess_state, "status", None) == "error": + disp = await executor._await_session_idle(session.id) + if disp == "error": final = "error" return except Exception: @@ -1024,54 +1023,117 @@ async def run_workflow_now(workflow_id: str): return {"run_id": "", "status": None, "error": None} -@workflows.router.post("/runs/{run_id}/stop") -async def stop_run(run_id: str): - """Force-terminate a running workflow's underlying agent session. - - Fired by RunningView's Stop button (Image #40). The run record gets - marked failure with a "stopped by user" error so it surfaces correctly - in History instead of looking like it succeeded. - """ - target_wf_id = None - target_run = None +def _find_active_run(run_id: str): + """Locate a currently-running run by id, returning (workflow_id, run).""" for wf in storage.list_workflows(): for r in storage.list_runs(wf.id, limit=50): if r.id == run_id and r.status == "running": - target_wf_id = wf.id - target_run = r - break - if target_run: - break - if not target_run or not target_wf_id: - raise HTTPException(status_code=404, detail="Run not found or not active") - if target_run.session_id: - try: - from backend.apps.agents.agent_manager import agent_manager - await agent_manager.close_session(target_run.session_id) - except Exception: - logger.exception("stop_run: close_session failed for %s", target_run.session_id) - target_run.status = "failure" - target_run.error = "Stopped by user" - target_run.finished_at = datetime.now() - storage.record_run(target_run) - wf = storage.get_workflow(target_wf_id) - if wf: - _persist_run_fields(wf, { - "last_run_status": "failure", - "last_run_at": target_run.finished_at, - "last_run_id": target_run.id, - }) + return wf.id, r + return None, None + + +async def _broadcast_run(workflow_id: str, run) -> None: try: from backend.apps.agents.core.ws_manager import ws_manager await ws_manager.broadcast_global("workflow:run", { - "workflow_id": target_wf_id, - "run": target_run.model_dump(mode="json"), + "workflow_id": workflow_id, + "run": run.model_dump(mode="json"), }) except Exception: pass + + +@workflows.router.post("/runs/{run_id}/stop") +async def stop_run(run_id: str): + """Fully stop a running workflow, failing it with a manual-stop reason. + + Fired by the running card's Stop button. We signal the executor (which + owns the run's terminal write) and halt the in-flight agent turn now; the + executor marks the run failure "Stopped by user" and closes the session in + its finally block. Signalling instead of writing the row here avoids the + old race where the still-looping executor overwrote the failure. + """ + target_wf_id, target_run = _find_active_run(run_id) + if not target_run or not target_wf_id: + raise HTTPException(status_code=404, detail="Run not found or not active") + executor.request_stop(run_id) + if target_run.session_id: + try: + from backend.apps.agents.agent_manager import agent_manager + await agent_manager.stop_agent(target_run.session_id) + except Exception: + logger.exception("stop_run: stop_agent failed for %s", target_run.session_id) return {"ok": True} +@workflows.router.post("/runs/{run_id}/pause") +async def pause_run(run_id: str): + """Pause the in-flight agent turn, same mechanic as the chat's Stop. + + The executor holds on the current step (see _await_session_idle) until the + matching resume. We flag the run paused so the card reflects it even when + the live chat isn't open. + """ + target_wf_id, target_run = _find_active_run(run_id) + if not target_run or not target_wf_id: + raise HTTPException(status_code=404, detail="Run not found or not active") + target_run.paused = True + executor.set_pause_override(run_id, True) + storage.record_run(target_run) + await _broadcast_run(target_wf_id, target_run) + + async def _stop_agent_for_pause() -> None: + if not target_run.session_id: + return + try: + from backend.apps.agents.agent_manager import agent_manager + await agent_manager.stop_agent(target_run.session_id) + except Exception: + logger.exception("pause_run: stop_agent failed for %s", target_run.session_id) + target_run.paused = False + executor.set_pause_override(run_id, False, ttl_s=0.1) + storage.record_run(target_run) + await _broadcast_run(target_wf_id, target_run) + + asyncio.create_task(_stop_agent_for_pause()) + return {"ok": True, "run": target_run.model_dump(mode="json")} + + +@workflows.router.post("/runs/{run_id}/resume") +async def resume_run(run_id: str): + """Resume a paused run, same mechanic as the chat's Resume Agent Response: + a hidden "continue where you left off" message restarts the current step's + turn. The executor advances once that turn completes. + """ + target_wf_id, target_run = _find_active_run(run_id) + if not target_run or not target_wf_id: + raise HTTPException(status_code=404, detail="Run not found or not active") + target_run.paused = False + executor.set_pause_override(run_id, False) + storage.record_run(target_run) + await _broadcast_run(target_wf_id, target_run) + + async def _send_resume_message() -> None: + if not target_run.session_id: + return + try: + from backend.apps.agents.agent_manager import agent_manager + await agent_manager.send_message( + target_run.session_id, + "Continue where you left off. Start your response EXACTLY with 'Sorry, let me pick up where I left off'", + hidden=True, + ) + except Exception: + logger.exception("resume_run: send_message failed for %s", target_run.session_id) + target_run.paused = True + executor.set_pause_override(run_id, True, ttl_s=0.1) + storage.record_run(target_run) + await _broadcast_run(target_wf_id, target_run) + + asyncio.create_task(_send_resume_message()) + return {"ok": True, "run": target_run.model_dump(mode="json")} + + @workflows.router.get("/{workflow_id}/runs") async def list_workflow_runs(workflow_id: str, limit: int = 50): wf = storage.get_workflow(workflow_id) diff --git a/backend/tests/test_workflows_semantics.py b/backend/tests/test_workflows_semantics.py index 13df556e..491ea7b4 100644 --- a/backend/tests/test_workflows_semantics.py +++ b/backend/tests/test_workflows_semantics.py @@ -23,6 +23,7 @@ import asyncio import json import os import shutil +import sys import tempfile from datetime import datetime, timedelta, timezone from zoneinfo import ZoneInfo @@ -37,6 +38,7 @@ def isolated_data_dir(monkeypatch, tmp_path): starts with empty caches.""" from backend.apps.workflows import storage as _storage from backend.apps.workflows import escalation as _escalation + from backend.apps.workflows import executor as _executor 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")) @@ -47,6 +49,8 @@ def isolated_data_dir(monkeypatch, tmp_path): # Reset escalation registry between tests. _escalation._tasks.clear() _escalation._state.clear() + _executor._run_control.clear() + _executor._run_pause_override.clear() # Also clear audit dir reference; audit.py reads DATA_DIR at import via # module-level expression, so reach in and override the AUDIT_DIR too. from backend.apps.workflows import audit as _audit @@ -65,6 +69,11 @@ def _make_wf(**overrides): return Workflow(**base) +class _NoopDebug: + def __call__(self, *args, **kwargs): + return None + + # --- DST tests --------------------------------------------------------------- def test_dst_spring_forward_weekly(): @@ -319,6 +328,83 @@ def test_paused_flag_persists_and_blocks_tick(): assert before == after +def test_pause_run_returns_confirmed_state_before_agent_stop_finishes(monkeypatch): + async def scenario(): + from backend.apps.workflows import storage + from backend.apps.workflows.models import WorkflowRun + monkeypatch.setitem(sys.modules, "debug", _NoopDebug()) + from backend.apps.workflows import workflows as routes + from backend.apps.agents import agent_manager as agent_manager_module + + wf = _make_wf() + storage.save_workflow(wf) + run = WorkflowRun(workflow_id=wf.id, status="running", session_id="s1", triggered_by="manual") + storage.record_run(run) + broadcasts: list[bool] = [] + + async def fake_broadcast(_workflow_id, updated_run): + broadcasts.append(updated_run.paused) + + stop_started = asyncio.Event() + stop_release = asyncio.Event() + + async def fake_stop_agent(_session_id): + stop_started.set() + await stop_release.wait() + + monkeypatch.setattr(routes, "_broadcast_run", fake_broadcast) + monkeypatch.setattr(agent_manager_module.agent_manager, "stop_agent", fake_stop_agent) + + result = await asyncio.wait_for(routes.pause_run(run.id), timeout=0.05) + assert result["run"]["paused"] is True + assert storage.list_runs(wf.id)[0].paused is True + assert broadcasts[-1] is True + await asyncio.wait_for(stop_started.wait(), timeout=0.05) + stop_release.set() + await asyncio.sleep(0) + + asyncio.run(scenario()) + + +def test_resume_run_returns_confirmed_state_before_resume_message_finishes(monkeypatch): + async def scenario(): + from backend.apps.workflows import storage + from backend.apps.workflows.models import WorkflowRun + monkeypatch.setitem(sys.modules, "debug", _NoopDebug()) + from backend.apps.workflows import workflows as routes + from backend.apps.agents import agent_manager as agent_manager_module + + wf = _make_wf() + storage.save_workflow(wf) + run = WorkflowRun(workflow_id=wf.id, status="running", session_id="s1", triggered_by="manual", paused=True) + storage.record_run(run) + broadcasts: list[bool] = [] + + async def fake_broadcast(_workflow_id, updated_run): + broadcasts.append(updated_run.paused) + + send_started = asyncio.Event() + send_release = asyncio.Event() + + async def fake_send_message(_session_id, _prompt, hidden=False): + assert hidden is True + send_started.set() + await send_release.wait() + + monkeypatch.setattr(routes, "_broadcast_run", fake_broadcast) + monkeypatch.setattr(agent_manager_module.agent_manager, "send_message", fake_send_message) + + result = await asyncio.wait_for(routes.resume_run(run.id), timeout=0.05) + assert result["run"]["paused"] is False + assert storage.list_runs(wf.id)[0].paused is False + assert broadcasts[-1] is False + await asyncio.wait_for(send_started.wait(), timeout=0.05) + send_release.set() + await asyncio.sleep(0) + + asyncio.run(scenario()) + + # --- Escalation -------------------------------------------------------------- def test_escalation_schedules_and_ack_cancels(): diff --git a/frontend/src/app/pages/AgentChat/AgentChat.tsx b/frontend/src/app/pages/AgentChat/AgentChat.tsx index 5b17583a..b885babf 100644 --- a/frontend/src/app/pages/AgentChat/AgentChat.tsx +++ b/frontend/src/app/pages/AgentChat/AgentChat.tsx @@ -269,6 +269,25 @@ const AgentChat: React.FC = ({ sessionId: sessionIdProp, onClose return found?.workflowId ?? null; }); const isStoppableSidecar = !!linkedWorkflowId; + // A live workflow run being watched owns pause/resume from its workflow + // card, so the chat's own "Resume Agent Response" bubble is redundant and + // would go stale against the card's Resume. Suppress it for any workflow-run + // sidecar, not just the fragile exact "watching" value. Test-run sidecars + // keep their chat-level resume behavior. + const isWorkflowRunSidecar = useAppSelector((s) => { + if (!id) return false; + for (const cd of Object.values(s.workflows.openCards)) { + if (cd.sidecarSessionId !== id || cd.sidecarKind === 'testing') continue; + if (cd.runId) { + const run = (s.workflows.runs[cd.workflowId] || []).find((r) => r.id === cd.runId); + if (!run || run.session_id === id) return true; + } + if (cd.sidecarKind === 'watching' || cd.sidecarKind === 'viewing-completed' || cd.sidecarKind === 'viewing-error') return true; + } + return Object.values(s.workflows.runs).some((runs) => + runs.some((r) => r.session_id === id && r.status === 'running'), + ); + }); const testState = useAppSelector((s) => (id ? s.agents.sessions[id]?.workflow_test_state : null) ?? null); const navigate = useNavigate(); const dispatch = useAppDispatch(); @@ -316,6 +335,9 @@ const AgentChat: React.FC = ({ sessionId: sessionIdProp, onClose const [heightVersion, setHeightVersion] = useState(0); const [showScrollButton, setShowScrollButton] = useState(false); const [showResumeBubble, setShowResumeBubble] = useState(false); + useEffect(() => { + if (isWorkflowRunSidecar) setShowResumeBubble(false); + }, [isWorkflowRunSidecar]); const [awaitingResponse, setAwaitingResponse] = useState(false); const [preSendActivityLabel, setPreSendActivityLabel] = useState(null); const [activatingMcp, setActivatingMcp] = useState(null); @@ -471,7 +493,7 @@ const AgentChat: React.FC = ({ sessionId: sessionIdProp, onClose didDispatchQueued = true; } else { if (curr === 'stopped') { - setShowResumeBubble(true); + setShowResumeBubble(!isWorkflowRunSidecar); } } @@ -489,7 +511,7 @@ const AgentChat: React.FC = ({ sessionId: sessionIdProp, onClose if (curr !== 'draft' && !didDispatchQueued) { setAwaitingResponse(false); } - }, [session?.status, mode, modesMap, id, isDraft, dispatch, dispatchMessage]); + }, [session?.status, mode, modesMap, id, isDraft, dispatch, dispatchMessage, isWorkflowRunSidecar]); // Idle reconcile: if the session has been 'running' for 5s with no // WebSocket activity (no new messages, no streaming updates), do a @@ -1872,7 +1894,7 @@ const AgentChat: React.FC = ({ sessionId: sessionIdProp, onClose /> )} - {showResumeBubble && session.status === 'stopped' && ( + {showResumeBubble && session.status === 'stopped' && !isWorkflowRunSidecar && ( s.workflows.runs[workflowId]); const runId = card?.runId || null; const run = (runs || []).find((r) => r.id === runId); + const pendingAction = useAppSelector((s) => runId ? s.workflows.runControlPending[runId] : undefined); + // Hit a run-control endpoint by run id. Keyed off runId (not the run object, + // which can lag right after Run starts) so Stop/Pause never silently no-op. + const postRunAction = React.useCallback(async (action: 'stop' | 'pause' | 'resume') => { + if (!runId || pendingAction) return; + await dispatch(controlWorkflowRun({ runId, action })); + }, [dispatch, runId, pendingAction]); const onStop = React.useCallback(async () => { - if (!run) return; - try { - const { API_BASE, getAuthToken } = await import('@/shared/config'); - const tok = (() => { try { return getAuthToken(); } catch { return ''; } })(); - await fetch(`${API_BASE}/workflows/runs/${encodeURIComponent(run.id)}/stop`, { - method: 'POST', - headers: tok ? { Authorization: `Bearer ${tok}` } : {}, - }); - } catch { /* best-effort */ } - dispatch(updateWorkflowCard({ workflowId, patch: { view: 'saved', runId: null } })); - }, [dispatch, workflowId, run]); - // Pause = "let this run finish, but stop firing future schedules." - // Can't actually pause a streaming agent turn mid-call, so we flip - // schedule.enabled so the scheduler stops queuing the next fire. The - // button label flips to "Resume" while paused; user can re-enable - // without leaving the running view. - const isPaused = !!workflow && !workflow.schedule.enabled && workflow.schedule.runs_count > 0; - const onPauseToggle = React.useCallback(async () => { - if (!workflow) return; - const next = { ...workflow.schedule, enabled: isPaused }; - await dispatch(updateWorkflow({ - id: workflow.id, - patch: { schedule: next as Workflow['schedule'] }, - ifMatch: workflow.updated_at || null, - })); - }, [dispatch, workflow, isPaused]); + await postRunAction('stop'); + }, [postRunAction]); + // Pause/Resume mirror the chat's stop-agent / resume-agent-response on the + // run's own session, so the paused state shows in both the chat and here. + const isPaused = !!run?.paused; + const onPauseToggle = React.useCallback(() => { + void postRunAction(isPaused ? 'resume' : 'pause'); + }, [postRunAction, isPaused]); + const stopPending = pendingAction === 'stop'; + const pausePending = pendingAction === 'pause' || pendingAction === 'resume'; + const controlsDisabled = !!pendingAction; return ( - + aria-disabled={controlsDisabled} + sx={{ display: 'inline-flex', alignItems: 'center', gap: 0.35, fontSize: '0.82rem', fontWeight: 600, px: 1, py: 0.4, color: c.text.secondary, cursor: controlsDisabled ? 'default' : 'pointer', borderRadius: 999, opacity: controlsDisabled && !stopPending ? 0.55 : 1, '&:hover': controlsDisabled ? {} : { color: c.text.primary, bgcolor: c.bg.elevated } }}> + {stopPending ? : } Stop - + - + {pausePending ? : isPaused ? : } {isPaused ? 'Resume' : 'Pause'} diff --git a/frontend/src/app/pages/Workflows/WorkflowCardLiveViews.tsx b/frontend/src/app/pages/Workflows/WorkflowCardLiveViews.tsx index 7fa2a708..86ef2d5a 100644 --- a/frontend/src/app/pages/Workflows/WorkflowCardLiveViews.tsx +++ b/frontend/src/app/pages/Workflows/WorkflowCardLiveViews.tsx @@ -172,22 +172,6 @@ export function RunningView({ workflow, steps, runs, mode = 'card' }: { const isLinked = mode === 'sidecar-linked' && (card?.sidecarKind === 'watching' || card?.sidecarKind === 'testing'); - const onStop = useCallback(async () => { - if (!runId) return; - try { - const { API_BASE, getAuthToken } = await import('@/shared/config'); - const tok = (() => { try { return getAuthToken(); } catch { return ''; } })(); - await fetch(`${API_BASE}/workflows/runs/${encodeURIComponent(runId)}/stop`, { - method: 'POST', - headers: tok ? { Authorization: `Bearer ${tok}` } : {}, - }); - } catch { /* best-effort */ } - }, [runId]); - const onPause = useCallback(() => { - // Pause flips the global paused state; the in-flight run continues but - // future fires queue up behind it. Maps to the existing /pause-all path. - void undefined; - }, []); const openSidecar = useOpenSidecar(workflow.id); const onWatchLive = useCallback(() => { if (run?.session_id) void openSidecar(run.session_id, 'watching'); @@ -231,11 +215,7 @@ export function RunningView({ workflow, steps, runs, mode = 'card' }: { /> )} - {/* Stop / Pause live in the header row, rendered by WorkflowCard. - See header-button overrides in WorkflowCard.tsx for the - per-view replacement of History/Run. */} - - + {/* Stop / Pause live in the header row, rendered by WorkflowCard. */} ); } diff --git a/frontend/src/shared/state/workflowsSlice.ts b/frontend/src/shared/state/workflowsSlice.ts index 5a3d8a8d..ca588700 100644 --- a/frontend/src/shared/state/workflowsSlice.ts +++ b/frontend/src/shared/state/workflowsSlice.ts @@ -113,8 +113,13 @@ export interface WorkflowRun { /** Currently-executing 0-based step index while status is 'running'; * freezes on the failed step when status flips to 'failure'. */ active_step_idx?: number | null; + /** True while the user has paused the in-flight agent turn (chat-style + * stop/resume). Drives the running card's Pause/Resume button. */ + paused?: boolean; } +export type WorkflowRunControlAction = 'pause' | 'resume' | 'stop'; + /** Transient view-only state per card; position lives in dashboardLayoutSlice.workflowCards. */ export interface OpenCard { workflowId: string; @@ -162,9 +167,62 @@ interface State { cloudSmsEnabled: boolean; allRuns: WorkflowRun[]; allRunsLoading: boolean; + runControlPending: Record; } -const initialState: State = { items: {}, runs: {}, openCards: {}, loaded: false, loading: false, paused: false, active: [], cloudSmsEnabled: false, allRuns: [], allRunsLoading: false }; +const initialState: State = { items: {}, runs: {}, openCards: {}, loaded: false, loading: false, paused: false, active: [], cloudSmsEnabled: false, allRuns: [], allRunsLoading: false, runControlPending: {} }; + +function mergeRunIntoState(state: State, r: WorkflowRun) { + const arr = state.runs[r.workflow_id] || []; + const idx = arr.findIndex((x) => x.id === r.id); + const prev = idx >= 0 ? arr[idx] : null; + if (idx >= 0) arr[idx] = r; else arr.unshift(r); + state.runs[r.workflow_id] = arr.slice(0, 100); + // Keep the cross-workflow log (Scheduled tasks history tab) live without a refetch. + const aIdx = state.allRuns.findIndex((x) => x.id === r.id); + if (aIdx >= 0) state.allRuns[aIdx] = r; else state.allRuns.unshift(r); + state.allRuns.sort((a, b) => (a.started_at < b.started_at ? 1 : -1)); + state.allRuns = state.allRuns.slice(0, 200); + const pending = state.runControlPending[r.id]; + if ( + (pending === 'pause' && r.paused) || + (pending === 'resume' && !r.paused) || + (pending === 'stop' && r.status !== 'running') + ) { + delete state.runControlPending[r.id]; + } + const wf = state.items[r.workflow_id]; + if (wf) { + wf.last_run_at = r.finished_at || r.started_at; + wf.last_run_status = r.status === 'skipped' ? wf.last_run_status : (r.status as Workflow['last_run_status']); + wf.last_run_id = r.id; + } + // Auto-flip the card view on run state transitions so the user sees + // Running while it streams, Completed on success, Failed on failure. + // Only nudge from views that the user hasn't actively navigated away + // from (saved / running). Edit, history, scheduling etc. stay put. + const card = state.openCards[r.workflow_id]; + if (card) { + const fromRunnable = card.view === 'saved' || card.view === 'running'; + if (r.status === 'running' && fromRunnable) { + card.view = 'running'; + card.runId = r.id; + } else if (prev && prev.status === 'running' && r.status === 'success' && (card.view === 'running' || card.view === 'saved')) { + card.view = 'completed'; + card.runId = r.id; + } else if (prev && prev.status === 'running' && r.status === 'failure' && (card.view === 'running' || card.view === 'saved')) { + card.view = 'failed'; + card.runId = r.id; + } + // A run that finishes while the user is watching it live becomes a + // "viewing" link so the sibling chat stays open with Stop Viewing, + // not a stale "watching" arrow pointing at a finished run. + if (card.sidecarSessionId && card.sidecarKind === 'watching' && prev && prev.status === 'running') { + if (r.status === 'failure') card.sidecarKind = 'viewing-error'; + else if (r.status === 'success' || r.status === 'ran_late') card.sidecarKind = 'viewing-completed'; + } + } +} export const fetchWorkflows = createAsyncThunk( 'workflows/fetch', @@ -254,6 +312,20 @@ export const runWorkflowNow = createAsyncThunk('workflows/run', async (id: strin }; }); +export const controlWorkflowRun = createAsyncThunk( + 'workflows/controlRun', + async ({ runId, action }: { runId: string; action: WorkflowRunControlAction }) => { + const res = await fetch(`${API}/runs/${encodeURIComponent(runId)}/${action}`, { method: 'POST' }); + if (!res.ok) throw new Error(`${action} failed ${res.status}`); + const data = await res.json(); + return { + runId, + action, + run: (data.run || null) as WorkflowRun | null, + }; + }, +); + export const fetchRuns = createAsyncThunk( 'workflows/runs', async (id: string) => { @@ -328,48 +400,7 @@ const slice = createSlice({ state.openCards[action.payload.newId] = { ...entry, workflowId: action.payload.newId }; }, upsertRun(state, action: { payload: WorkflowRun }) { - const r = action.payload; - const arr = state.runs[r.workflow_id] || []; - const idx = arr.findIndex((x) => x.id === r.id); - const prev = idx >= 0 ? arr[idx] : null; - if (idx >= 0) arr[idx] = r; else arr.unshift(r); - state.runs[r.workflow_id] = arr.slice(0, 100); - // Keep the cross-workflow log (Scheduled tasks history tab) live without a refetch. - const aIdx = state.allRuns.findIndex((x) => x.id === r.id); - if (aIdx >= 0) state.allRuns[aIdx] = r; else state.allRuns.unshift(r); - state.allRuns.sort((a, b) => (a.started_at < b.started_at ? 1 : -1)); - state.allRuns = state.allRuns.slice(0, 200); - const wf = state.items[r.workflow_id]; - if (wf) { - wf.last_run_at = r.finished_at || r.started_at; - wf.last_run_status = r.status === 'skipped' ? wf.last_run_status : (r.status as Workflow['last_run_status']); - wf.last_run_id = r.id; - } - // Auto-flip the card view on run state transitions so the user sees - // Running while it streams, Completed on success, Failed on failure. - // Only nudge from views that the user hasn't actively navigated away - // from (saved / running). Edit, history, scheduling etc. stay put. - const card = state.openCards[r.workflow_id]; - if (card) { - const fromRunnable = card.view === 'saved' || card.view === 'running'; - if (r.status === 'running' && fromRunnable) { - card.view = 'running'; - card.runId = r.id; - } else if (prev && prev.status === 'running' && r.status === 'success' && (card.view === 'running' || card.view === 'saved')) { - card.view = 'completed'; - card.runId = r.id; - } else if (prev && prev.status === 'running' && r.status === 'failure' && (card.view === 'running' || card.view === 'saved')) { - card.view = 'failed'; - card.runId = r.id; - } - // A run that finishes while the user is watching it live becomes a - // "viewing" link so the sibling chat stays open with Stop Viewing, - // not a stale "watching" arrow pointing at a finished run. - if (card.sidecarSessionId && card.sidecarKind === 'watching' && prev && prev.status === 'running') { - if (r.status === 'failure') card.sidecarKind = 'viewing-error'; - else if (r.status === 'success' || r.status === 'ran_late') card.sidecarKind = 'viewing-completed'; - } - } + mergeRunIntoState(state, action.payload); }, toggleExpandedStep(state, action: { payload: { workflowId: string; stepId: string } }) { const card = state.openCards[action.payload.workflowId]; @@ -420,8 +451,46 @@ const slice = createSlice({ delete state.runs[action.payload]; state.allRuns = state.allRuns.filter((r) => r.workflow_id !== action.payload); }) + .addCase(runWorkflowNow.fulfilled, (state, action) => { + // Enter the running view the moment the run kicks off, off the run_id + // the REST call returns. Don't wait for the workflow:run WS event: + // if it's missed or races the view, the Stop/Pause header never shows. + const { id, run_id, status } = action.payload; + const card = state.openCards[id]; + if (!card || !run_id || status !== 'running') return; + if (['saved', 'running', 'completed', 'failed', 'history', 'history_detail'].includes(card.view)) { + card.view = 'running'; + card.runId = run_id; + } + }) + .addCase(controlWorkflowRun.pending, (state, action) => { + state.runControlPending[action.meta.arg.runId] = action.meta.arg.action; + }) + .addCase(controlWorkflowRun.fulfilled, (state, action) => { + if (action.payload.run) { + mergeRunIntoState(state, action.payload.run); + } + if (action.payload.action !== 'stop') { + delete state.runControlPending[action.payload.runId]; + } else if (action.payload.run && action.payload.run.status !== 'running') { + delete state.runControlPending[action.payload.runId]; + } + }) + .addCase(controlWorkflowRun.rejected, (state, action) => { + delete state.runControlPending[action.meta.arg.runId]; + }) .addCase(fetchRuns.fulfilled, (state, action) => { state.runs[action.payload.id] = action.payload.runs; + for (const r of action.payload.runs) { + const pending = state.runControlPending[r.id]; + if ( + (pending === 'pause' && r.paused) || + (pending === 'resume' && !r.paused) || + (pending === 'stop' && r.status !== 'running') + ) { + delete state.runControlPending[r.id]; + } + } }) .addCase(fetchAllRuns.pending, (state) => { state.allRunsLoading = true; }) .addCase(fetchAllRuns.fulfilled, (state, action) => {