diff --git a/backend/apps/workflows/executor.py b/backend/apps/workflows/executor.py index 432ffcf9..42a6922f 100644 --- a/backend/apps/workflows/executor.py +++ b/backend/apps/workflows/executor.py @@ -209,12 +209,26 @@ async def execute(wf: Workflow, triggered_by: str = "schedule", scheduled_for: O # session is idle before posting the next step. Keeps the runner # safe regardless of how long each turn takes. step_error: Optional[str] = None - for step in steps: + for idx, step in enumerate(steps): + # Broadcast the step bump before sending so RunningView flips + # the disc immediately, not after the agent finishes the step. + run.active_step_idx = idx + run.last_tool_label = None + try: + from backend.apps.agents.ws_manager import ws_manager as _wsm + await _wsm.broadcast_global("workflow:run", { + "workflow_id": wf.id, + "run": run.model_dump(mode="json"), + }) + except Exception: + pass await agent_manager.send_message(session.id, step) 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": 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() diff --git a/backend/apps/workflows/models.py b/backend/apps/workflows/models.py index e5ee7f30..9640aa60 100644 --- a/backend/apps/workflows/models.py +++ b/backend/apps/workflows/models.py @@ -120,6 +120,10 @@ class WorkflowRun(BaseModel): # 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 + # Currently-executing step index (0-based). Executor bumps this each + # 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 class WorkflowCreate(BaseModel): diff --git a/backend/apps/workflows/workflows.py b/backend/apps/workflows/workflows.py index 11692d5d..1426fa1a 100644 --- a/backend/apps/workflows/workflows.py +++ b/backend/apps/workflows/workflows.py @@ -138,16 +138,21 @@ async def create_workflow(body: WorkflowCreate): wf.icon = _derive_icon(wf) if wf.schedule.enabled: wf.next_run_at = scheduler.compute_next_fire(wf) - # Force-generate title + description from the steps in a single aux - # call. Previously we only filled missing description, leaving stale - # session names ("Inbox check") as titles. One round-trip, both - # fields, overwrites whatever shallow draft the FE sent. + # Force-generate title + description + per-step labels from the steps + # in a single aux call. Previously we only filled missing description, + # leaving stale session names ("Inbox check") as titles. Step labels + # are the 3-6 word at-a-glance headlines surfaced in StepList; without + # them the UI falls back to truncated raw prompts. try: - title, description = await _generate_title_and_description(wf) + title, description, labels = await _generate_workflow_metadata(wf) if title: wf.title = title if description: wf.description = description + if labels and len(labels) == len(wf.steps): + for i, lab in enumerate(labels): + if lab: + wf.steps[i].label = lab except Exception: pass storage.save_workflow(wf) @@ -155,35 +160,35 @@ async def create_workflow(body: WorkflowCreate): return _enriched(wf) -async def _generate_title_and_description(wf: Workflow) -> tuple[str, str]: - """Single aux-model call returning (title, description). +async def _generate_workflow_metadata(wf: Workflow) -> tuple[str, str, list[str]]: + """Single aux-model call returning (title, description, step_labels). - Uses strict JSON output so both fields come back in one round-trip. - Returns ("", "") on any failure so the caller can write back - unconditionally without dropping the workflow create. + One round-trip for all three so we don't burn 3x aux cost. Returns + ("", "", []) on any failure; caller writes back unconditionally. """ if not wf.steps: - return "", "" + return "", "", [] try: from backend.apps.agents.providers.registry import resolve_aux_model from backend.apps.agents.providers.registry import get_anthropic_client_for_model from backend.apps.settings.settings import load_settings as _ls except Exception: - return "", "" + return "", "", [] settings = _ls() try: aux_model, _ = await resolve_aux_model(settings, preferred_tier="haiku") client = get_anthropic_client_for_model(settings, aux_model) except Exception: - return "", "" + return "", "", [] steps_lines = "\n".join(f"{i+1}. {s.text}" for i, s in enumerate(wf.steps) if s.text) + n_steps = len(wf.steps) prompt = ( "You name and describe a saved automation routine that the user " - "can re-run later. The routine is defined ONLY by the numbered " - "steps below; treat those as the user's instructions to the " - "agent.\n\n" + "can re-run later, AND produce a short at-a-glance label for " + "each step. The routine is defined ONLY by the numbered steps " + "below; treat those as the user's instructions to the agent.\n\n" "Return STRICT JSON, nothing else, no code fence:\n" - " {\"title\": string, \"description\": string}\n\n" + ' {"title": string, "description": string, "step_labels": [string, ...]}\n\n' "title rules:\n" "- 2 to 5 words, Title Case\n" "- Starts with a verb-noun pair when possible (e.g. \"Summarize " @@ -197,27 +202,26 @@ async def _generate_title_and_description(wf: Workflow) -> tuple[str, str]: "digest each Sunday at 9am.\"\n" " \"Pulls today's calendar plus inbox, writes a Notion brief, " "and texts you the link.\"\n" - "- Examples of BAD output you MUST AVOID verbatim:\n" - " \"This is an AI-generated description...\"\n" - " \"Auto-generated description used to wrap workflows...\"\n" - " Any sentence that talks about the description itself\n" "- Start with a verb. Do NOT start with \"This\", \"A\", \"An\", " "\"The workflow\", \"This routine\".\n\n" + f"step_labels rules:\n" + f"- EXACTLY {n_steps} entries, one per step, same order.\n" + "- Each label: 3 to 6 words, Sentence case.\n" + "- Imperative verb-led (\"Summarize emails & calendar\", \"Make " + "brief in notion\", \"Email brief link to me\").\n" + "- No trailing punctuation, no quotes, no emoji.\n" + "- Should read as the human-friendly NAME of the step, NOT a " + "restatement of the prompt.\n\n" f"Steps:\n{steps_lines}" ) import json import re as _re def _extract_json_object(s: str) -> Optional[dict]: - """Find the first {...} block and json.loads it. Handles code - fences, prose preambles, and trailing chatter that some aux - models like to add.""" s = s.strip() if s.startswith("```"): s = _re.sub(r"^```(?:json)?\s*", "", s, flags=_re.IGNORECASE) s = _re.sub(r"\s*```\s*$", "", s) - # Greedy brace match; falls through to direct json.loads if no - # braces are visible at all. start = s.find("{") end = s.rfind("}") if start != -1 and end != -1 and end > start: @@ -228,13 +232,9 @@ async def _generate_title_and_description(wf: Workflow) -> tuple[str, str]: return None try: - # Prefill the assistant turn with `{` so the model is steered into - # emitting JSON from the first token. The Anthropic API treats a - # trailing assistant message as a prefill; we'll glue it back on - # before parsing. resp = await client.messages.create( model=aux_model, - max_tokens=240, + max_tokens=400 + n_steps * 30, messages=[ {"role": "user", "content": prompt}, {"role": "assistant", "content": "{"}, @@ -248,16 +248,16 @@ async def _generate_title_and_description(wf: Workflow) -> tuple[str, str]: raw = "{" + text.strip() if not text.strip().startswith("{") else text.strip() data = _extract_json_object(raw) if not data: - logger.warning("description gen: failed to parse aux model output: %s", raw[:400]) - return "", "" + logger.warning("workflow meta gen: failed to parse aux model output: %s", raw[:400]) + return "", "", [] title = (data.get("title") or "").strip()[:80] description = (data.get("description") or "").strip()[:500] - if not description: - logger.warning("description gen: empty description from aux model. Raw: %s", raw[:400]) - return title, description + raw_labels = data.get("step_labels") or [] + labels = [str(x or "").strip()[:60] for x in raw_labels] if isinstance(raw_labels, list) else [] + return title, description, labels except Exception as e: - logger.warning("description gen: aux model call failed: %s", e) - return "", "" + logger.warning("workflow meta gen: aux model call failed: %s", e) + return "", "", [] def _last_run_cost(wid: str) -> float: diff --git a/frontend/src/app/pages/Workflows/WorkflowCard.tsx b/frontend/src/app/pages/Workflows/WorkflowCard.tsx index e5cc60ee..adf46e94 100644 --- a/frontend/src/app/pages/Workflows/WorkflowCard.tsx +++ b/frontend/src/app/pages/Workflows/WorkflowCard.tsx @@ -697,12 +697,21 @@ function RunningHeader({ workflowId }: { workflowId: string }) { } catch { /* best-effort */ } dispatch(updateWorkflowCard({ workflowId, patch: { view: 'saved', runId: null } })); }, [dispatch, workflowId, run]); - const onPause = React.useCallback(() => { - // Pause is "this run keeps going but no further fires queue." We - // can't actually pause a streaming agent mid-call, so this just - // flips the schedule paused flag for now. - void workflow; - }, [workflow]); + // 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]); return ( @@ -713,19 +722,21 @@ function RunningHeader({ workflowId }: { workflowId: string }) { Stop - - - Pause - + + + + {isPaused ? 'Resume' : 'Pause'} + + ); } diff --git a/frontend/src/app/pages/Workflows/WorkflowCardLiveViews.tsx b/frontend/src/app/pages/Workflows/WorkflowCardLiveViews.tsx index 4b4a1e87..15fc4ddf 100644 --- a/frontend/src/app/pages/Workflows/WorkflowCardLiveViews.tsx +++ b/frontend/src/app/pages/Workflows/WorkflowCardLiveViews.tsx @@ -141,10 +141,11 @@ export function RunningView({ workflow, steps, runs, mode = 'card' }: { const runId = card?.runId || null; const run = useMemo(() => (runs || []).find((r) => r.id === runId) || null, [runs, runId]); - // Synthesize an active step index off the run's elapsed/expected ratio - // until we wire per-step backend telemetry. Mirrors the heuristic the - // old StepList used; placeholder until last_tool_call drives this. - const activeIdx = useActiveStepIdx(steps.length, runs, runId); + // Prefer the backend's real active_step_idx (broadcast on each step + // bump in executor.execute). Fall back to elapsed/expected heuristic + // when the field is missing (older runs or first-frame race). + const heuristicIdx = useActiveStepIdx(steps.length, runs, runId); + const activeIdx = typeof run?.active_step_idx === 'number' ? run.active_step_idx : heuristicIdx; const statuses: StepStatus[] = steps.map((_, i) => i < activeIdx ? 'done' : i === activeIdx ? 'active' : 'pending', ); @@ -464,13 +465,19 @@ export function FailedView({ workflow, steps, runs, mode = 'card' }: { } function guessFailedIdx(run: WorkflowRun | null, total: number): number { - if (!run || !run.error) return Math.max(0, total - 1); - // Backend may serialize as "Step N: ..."; pull N when present so the - // X lands on the right row instead of always the last. - const m = /step\s+(\d+)/i.exec(run.error); - if (m) { - const n = parseInt(m[1], 10); - if (!Number.isNaN(n) && n >= 1 && n <= total) return n - 1; + if (!run) return Math.max(0, total - 1); + // Backend pins active_step_idx at the failed step before flipping + // status to 'failure'. Prefer that; fall back to parsing "Step N" + // out of the error string for legacy runs. + if (typeof run.active_step_idx === 'number') { + return Math.max(0, Math.min(total - 1, run.active_step_idx)); + } + if (run.error) { + const m = /step\s+(\d+)/i.exec(run.error); + if (m) { + const n = parseInt(m[1], 10); + if (!Number.isNaN(n) && n >= 1 && n <= total) return n - 1; + } } return Math.max(0, Math.min(total - 1, 1)); } diff --git a/frontend/src/shared/state/workflowsSlice.ts b/frontend/src/shared/state/workflowsSlice.ts index 61886a2c..f4be3477 100644 --- a/frontend/src/shared/state/workflowsSlice.ts +++ b/frontend/src/shared/state/workflowsSlice.ts @@ -92,6 +92,9 @@ export interface WorkflowRun { triggered_by: 'schedule' | 'manual' | 'retry'; /** Live "what's the agent doing" subtitle while status is 'running'. */ last_tool_label?: string | null; + /** 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; } /** Transient view-only state per card; position lives in dashboardLayoutSlice.workflowCards. */