From a09a8a7eac050d5e7d511ef92917704eb8cedd1d Mon Sep 17 00:00:00 2001 From: ciregenz Date: Thu, 21 May 2026 23:21:09 -0700 Subject: [PATCH] [eric] workflows: stop endpoint + last_tool_label live subtitle for RunningView --- backend/apps/workflows/executor.py | 56 +++++++++++++++++++ backend/apps/workflows/models.py | 8 +++ backend/apps/workflows/workflows.py | 48 ++++++++++++++++ .../pages/Workflows/WorkflowCardLiveViews.tsx | 8 +-- frontend/src/shared/state/workflowsSlice.ts | 2 + 5 files changed, 118 insertions(+), 4 deletions(-) diff --git a/backend/apps/workflows/executor.py b/backend/apps/workflows/executor.py index 85d9c5a2..432ffcf9 100644 --- a/backend/apps/workflows/executor.py +++ b/backend/apps/workflows/executor.py @@ -154,6 +154,56 @@ async def execute(wf: Workflow, triggered_by: str = "schedule", scheduled_for: O run.session_id = session.id storage.record_run(run) + # Background poller: surface the latest tool-call name as a + # live "what's the agent doing" subtitle on the workflow:run + # ws event. Cheap enough to run at 1.5s cadence; nothing else + # is watching session.messages from here. Cancelled in the + # finally block alongside _running cleanup. + async def _watch_tool_calls() -> None: + last_seen = "" + while True: + try: + await asyncio.sleep(1.5) + sess = agent_manager.sessions.get(session.id) + if not sess: + return + msgs = getattr(sess, "messages", []) or [] + label = "" + for m in reversed(msgs): + if getattr(m, "role", None) != "tool_call": + continue + content = getattr(m, "content", None) + # Content can be a string, a dict with "name", or + # a list of blocks. Pick the first tool_use name. + if isinstance(content, list): + for b in content: + if isinstance(b, dict) and b.get("type") == "tool_use": + label = str(b.get("name") or "") + break + elif isinstance(content, dict): + label = str(content.get("name") or "") + elif isinstance(content, str): + label = content[:60] + if label: + break + if label and label != last_seen: + last_seen = label + run.last_tool_label = label + try: + from backend.apps.agents.ws_manager import ws_manager + await ws_manager.broadcast_global("workflow:run", { + "workflow_id": wf.id, + "run": run.model_dump(mode="json"), + }) + except Exception: + pass + except asyncio.CancelledError: + return + except Exception: + return + + watcher_task = asyncio.create_task(_watch_tool_calls()) + # Send each step sequentially. agent_manager.send_message is a no-op # while a prior turn is still streaming, so we await until the # session is idle before posting the next step. Keeps the runner @@ -208,6 +258,12 @@ async def execute(wf: Workflow, triggered_by: str = "schedule", scheduled_for: O "last_run_at": run.finished_at, }) finally: + # Cancel the tool-call watcher before we tear the session down so + # the next poll doesn't race close_session. + try: + watcher_task.cancel() # type: ignore[name-defined] + except Exception: + pass # Close the workflow's agent session so closed_at is set and the # run shows up in chat history (get_history sorts by closed_at; # sessions with closed_at=None sort to the bottom and fall off diff --git a/backend/apps/workflows/models.py b/backend/apps/workflows/models.py index 4003deb6..e5ee7f30 100644 --- a/backend/apps/workflows/models.py +++ b/backend/apps/workflows/models.py @@ -66,6 +66,10 @@ class WorkflowStep(BaseModel): label: Optional[str] = None +def _empty_str_default() -> str: + return "" + + class Workflow(BaseModel): # validate_assignment is load-bearing for the PATCH /workflows/{id} path # (workflows.py:update_workflow setattr's raw dicts from body.model_dump @@ -112,6 +116,10 @@ class WorkflowRun(BaseModel): error: Optional[str] = None cost_usd: float = 0.0 triggered_by: Literal["schedule", "manual", "retry"] = "schedule" + # Last tool-call label observed on the underlying agent session while + # the workflow is running. Surfaced under the active step in RunningView + # (Image #40) so the user can tell the run is still making progress. + last_tool_label: Optional[str] = None class WorkflowCreate(BaseModel): diff --git a/backend/apps/workflows/workflows.py b/backend/apps/workflows/workflows.py index c3198ae2..11692d5d 100644 --- a/backend/apps/workflows/workflows.py +++ b/backend/apps/workflows/workflows.py @@ -607,6 +607,54 @@ 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 + 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, + }) + try: + from backend.apps.agents.ws_manager import ws_manager + await ws_manager.broadcast_global("workflow:run", { + "workflow_id": target_wf_id, + "run": target_run.model_dump(mode="json"), + }) + except Exception: + pass + return {"ok": True} + + @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/frontend/src/app/pages/Workflows/WorkflowCardLiveViews.tsx b/frontend/src/app/pages/Workflows/WorkflowCardLiveViews.tsx index 2b15fadb..4b4a1e87 100644 --- a/frontend/src/app/pages/Workflows/WorkflowCardLiveViews.tsx +++ b/frontend/src/app/pages/Workflows/WorkflowCardLiveViews.tsx @@ -151,10 +151,10 @@ export function RunningView({ workflow, steps, runs, mode = 'card' }: { const completeCount = statuses.filter((s) => s === 'done').length; const total = steps.length; - // Tool-call subtitle for the active step. The backend will emit this on - // workflow:run as `last_tool_label` once slice 5 lands; until then we - // surface a soft placeholder so the row never feels empty. - const activeSubtitle = (run as unknown as { last_tool_label?: string })?.last_tool_label || null; + // Tool-call subtitle for the active step. Backend polls the session's + // messages at 1.5s cadence and broadcasts on workflow:run as the agent + // makes new tool calls. See executor.py _watch_tool_calls. + const activeSubtitle = run?.last_tool_label || null; const activeDuration = formatLiveDuration(run); const isLinked = mode === 'sidecar-linked' && card?.sidecarKind === 'watching'; diff --git a/frontend/src/shared/state/workflowsSlice.ts b/frontend/src/shared/state/workflowsSlice.ts index 5f2670f9..61886a2c 100644 --- a/frontend/src/shared/state/workflowsSlice.ts +++ b/frontend/src/shared/state/workflowsSlice.ts @@ -90,6 +90,8 @@ export interface WorkflowRun { error: string | null; cost_usd: number; triggered_by: 'schedule' | 'manual' | 'retry'; + /** Live "what's the agent doing" subtitle while status is 'running'. */ + last_tool_label?: string | null; } /** Transient view-only state per card; position lives in dashboardLayoutSlice.workflowCards. */