diff --git a/backend/apps/agents/agents.py b/backend/apps/agents/agents.py index 96746f02..d4265849 100644 --- a/backend/apps/agents/agents.py +++ b/backend/apps/agents/agents.py @@ -9,9 +9,15 @@ from fastapi.responses import JSONResponse import asyncio import json import logging +import time logger = logging.getLogger(__name__) +# Soft per-session throttle so the integration suggestion can fire on any turn +# without nagging every message; only stamped when a suggestion actually emits. +MCP_SUGGEST_COOLDOWN_S = 300.0 +p_mcp_suggest_cooldown: dict[str, float] = {} + # Dedup concurrent generate-group-meta calls; collapses the 429 thundering herd by sharing one upstream Future per (session, group). _group_meta_inflight: dict[tuple[str, str], asyncio.Future] = {} @@ -74,24 +80,26 @@ async def send_message(session_id: str, body: dict): raise HTTPException(status_code=400, detail="prompt is required") # Run MCP-suggestion classifier in parallel with the agent launch; fails open. + # Fires on any turn, but a per-session cooldown keeps it from nagging every message. try: - from backend.apps.agents.core.mcp_preflight import run_preflight - from backend.apps.agents.core.ws_manager import ws_manager as _ws + last_suggested = p_mcp_suggest_cooldown.get(session_id, 0.0) + if time.monotonic() - last_suggested >= MCP_SUGGEST_COOLDOWN_S: + from backend.apps.agents.core.mcp_preflight import run_preflight - async def _emit_preflight(): - try: - result = await run_preflight(prompt, task_id=session_id) - if result.get("suggestions") or result.get("is_vague"): - await _ws.send_to_session(session_id, "agent:mcp_suggestions", { - "session_id": session_id, - "suggestions": result.get("suggestions", []), - "is_vague": bool(result.get("is_vague")), - }) - except Exception: - pass + async def _emit_preflight(): + try: + result = await run_preflight(prompt, task_id=session_id) + if result.get("suggestions"): + p_mcp_suggest_cooldown[session_id] = time.monotonic() + await ws_manager.send_to_session(session_id, "agent:mcp_suggestions", { + "session_id": session_id, + "suggestions": result.get("suggestions", []), + "is_vague": bool(result.get("is_vague")), + }) + except Exception: + pass - import asyncio as _asyncio - _asyncio.create_task(_emit_preflight()) + asyncio.create_task(_emit_preflight()) except Exception: pass diff --git a/backend/apps/agents/schedule_mcp_server.py b/backend/apps/agents/schedule_mcp_server.py index 578b362e..b8acd7c9 100644 --- a/backend/apps/agents/schedule_mcp_server.py +++ b/backend/apps/agents/schedule_mcp_server.py @@ -15,6 +15,7 @@ import os import uuid import urllib.request import urllib.error +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError BACKEND_PORT = os.environ.get("OPENSWARM_PORT", "8324") BACKEND_AUTH = os.environ.get("OPENSWARM_AUTH_TOKEN", "") @@ -23,6 +24,20 @@ PARENT_SESSION_ID = os.environ.get("OPENSWARM_PARENT_SESSION_ID", "") DASHBOARD_ID = os.environ.get("OPENSWARM_DASHBOARD_ID", "") +def _local_timezone_name() -> str: + name = os.environ.get("OPENSWARM_TIMEZONE", "").strip() + if not name: + try: + from tzlocal import get_localzone_name # type: ignore + name = get_localzone_name() or "" + except Exception: + name = "" + try: + return (getattr(ZoneInfo(name), "key", None) or "UTC") if name else "UTC" + except ZoneInfoNotFoundError: + return "UTC" + + PRESETS = { "daily_morning": {"enabled": True, "repeat_unit": "day", "repeat_every": 1, "hour": 9, "minute": 0, "on_days": []}, "weekdays_morning": {"enabled": True, "repeat_unit": "week", "repeat_every": 1, "hour": 9, "minute": 0, "on_days": [1, 2, 3, 4, 5]}, @@ -68,7 +83,7 @@ TOOLS = [ "items": {"type": "integer"}, "description": "Weekdays (Sun=0..Sat=6) when preset='custom' and repeat_unit='week'.", }, - "timezone": {"type": "string", "description": "IANA timezone name (e.g. 'America/Los_Angeles'). Omit to use the user's local zone."}, + "timezone": {"type": "string", "description": "IANA timezone name (e.g. 'America/Los_Angeles'). Omit to use the user's current local zone at scheduling time."}, "source_session_id": {"type": "string", "description": "Optional; the chat session this workflow was created from. Inherits its tool surface."}, }, "required": ["title", "steps", "preset"], @@ -253,7 +268,8 @@ def _call(method: str, path: str, body=None) -> dict: def _build_schedule_from_preset(preset: str, args: dict) -> dict: - base = {"timezone": "local", "on_missed": "skip", "ends_at": None, "max_runs": None, "runs_count": 0} + local_tz = _local_timezone_name() + base = {"timezone": args.get("timezone") or local_tz, "on_missed": "skip", "ends_at": None, "max_runs": None, "runs_count": 0} if preset == "custom": return { **base, @@ -263,7 +279,6 @@ def _build_schedule_from_preset(preset: str, args: dict) -> dict: "hour": int(args.get("hour", 9)), "minute": int(args.get("minute", 0)), "on_days": list(args.get("on_days") or []), - "timezone": args.get("timezone") or "local", } preset_def = PRESETS.get(preset) if not preset_def: diff --git a/backend/apps/settings/settings.py b/backend/apps/settings/settings.py index bd501e1f..7c8294ef 100644 --- a/backend/apps/settings/settings.py +++ b/backend/apps/settings/settings.py @@ -3,6 +3,7 @@ import os import tempfile import time import logging +from datetime import datetime, timezone from contextlib import asynccontextmanager from fastapi import HTTPException, Query, UploadFile, File from fastapi.responses import JSONResponse @@ -292,6 +293,21 @@ async def put_app_theme_override(body: AppThemeOverridePayload): return {"ok": True, "mode": current.app_template_theme_override} +class DismissMcpSuggestionPayload(BaseModel): + ids: list[str] + + +@settings.router.put("/dismiss-mcp-suggestion") +async def put_dismiss_mcp_suggestion(body: DismissMcpSuggestionPayload): + """MERGE dismissed integration suggestions; the general PUT replaces the whole object and would blank secrets.""" + current = load_settings() + now = datetime.now(timezone.utc).isoformat() + for tool_id in body.ids: + current.dismissed_mcp_suggestions[tool_id] = now + await save_settings_async(current) + return {"ok": True, "settings": current.model_dump()} + + @settings.router.get("/default-system-prompt") async def get_default_system_prompt(): return {"default_system_prompt": DEFAULT_SYSTEM_PROMPT} diff --git a/backend/apps/workflows/executor.py b/backend/apps/workflows/executor.py index 9361bd42..146d78cf 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 @@ -55,7 +79,11 @@ def p_make_remember_approval(workflow_id: str): return p_remember_approval -def p_persist_step_tool_usage(workflow_id: str, step_usage: dict[str, dict[str, bool]]) -> None: +def p_persist_step_tool_usage( + workflow_id: str, + step_usage: dict[str, dict[str, bool]], + tested_signature: Optional[str] = None, +) -> None: fresh = storage.get_workflow(workflow_id) if fresh is None: return @@ -68,6 +96,8 @@ def p_persist_step_tool_usage(workflow_id: str, step_usage: dict[str, dict[str, for sid, tools in (step_usage or {}).items() if sid in live_ids and isinstance(tools, dict) } + if tested_signature is not None: + fresh.tested_signature = tested_signature storage.save_workflow(fresh) try: from backend.apps.agents.core.ws_manager import ws_manager @@ -126,7 +156,12 @@ def _monthly_spend_so_far(wf: Workflow) -> float: return total -async def execute(wf: Workflow, triggered_by: str = "schedule", scheduled_for: Optional[datetime] = None) -> WorkflowRun: +async def execute( + wf: Workflow, + triggered_by: str = "schedule", + scheduled_for: Optional[datetime] = None, + tested_signature: Optional[str] = None, +) -> WorkflowRun: from backend.apps.agents.agent_manager import ( agent_manager, clear_workflow_approval_memory, @@ -222,12 +257,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 +291,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 +319,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 +339,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) @@ -318,15 +374,19 @@ async def execute(wf: Workflow, triggered_by: str = "schedule", scheduled_for: O runs_delta = 1 if (triggered_by == "schedule" and run.status in ("success", "ran_late", "failure")) else 0 storage.record_run(run) wf.last_run_at = run.finished_at - _persist_run_fields(wf, { + run_fields = { "last_run_at": run.finished_at, "last_run_status": wf.last_run_status, - }, schedule_runs_count_delta=runs_delta) + } + if triggered_by == "manual" and run.status in ("success", "ran_late") and isinstance(tested_signature, str): + run_fields["tested_signature"] = tested_signature + _persist_run_fields(wf, run_fields, schedule_runs_count_delta=runs_delta) except Exception as e: logger.exception("Workflow run failed: %s", e) 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 +394,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 +439,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..b5c3212e 100644 --- a/backend/apps/workflows/models.py +++ b/backend/apps/workflows/models.py @@ -106,6 +106,11 @@ class Workflow(BaseModel): default_factory=lambda: [PermissionTier(kind="notify")] ) source_session_id: Optional[str] = None + # Tool names observed in the source chat when this workflow was generated. + # This preserves conversion context without pretending those calls map to + # generated workflow step ids. Explicit approval decisions still live in + # remembered_approvals and are the only values reused as permissions. + source_tools: list[str] = Field(default_factory=list) dashboard_id: Optional[str] = None model: str = "sonnet" mode: str = "agent" @@ -148,6 +153,11 @@ class Workflow(BaseModel): # the scheduled/unscheduled lists until the first commit clears the flag, # so an in-progress build doesn't litter the sidebar. unsaved: bool = False + # Stable signature of the steps last validated by a test run (or seeded at + # chat conversion). The FE compares it against the current steps before + # scheduling: a mismatch means "edited since you last approved tools" and + # triggers the test-first warning. Computed FE-side so there's one algorithm. + tested_signature: Optional[str] = None class WorkflowRun(BaseModel): @@ -169,6 +179,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): @@ -191,6 +205,7 @@ class WorkflowCreate(BaseModel): mode: Optional[str] = None provider: Optional[str] = None cost_cap_usd_monthly: Optional[float] = None + tested_signature: Optional[str] = None class WorkflowUpdate(BaseModel): diff --git a/backend/apps/workflows/scheduler.py b/backend/apps/workflows/scheduler.py index 7fa620a6..28dd949f 100644 --- a/backend/apps/workflows/scheduler.py +++ b/backend/apps/workflows/scheduler.py @@ -58,6 +58,11 @@ def _host_tz() -> ZoneInfo: return _host_tz_cache +def host_timezone_name() -> str: + """Concrete IANA-ish zone name for schedules created on this host.""" + return getattr(_host_tz(), "key", None) or "UTC" + + def _resolve_tz(tz: str) -> ZoneInfo: if not tz or tz == "local": return _host_tz() @@ -194,6 +199,53 @@ def fires_in_window(wf: Workflow, days: int = 30) -> int: return count +def occurrences_between( + wf: Workflow, + from_utc: datetime, + to_utc: datetime, + cap: int = 5000, +) -> list[datetime]: + """Return scheduled fire instants in [from_utc, to_utc). + + Calendar previews must use the same timezone-aware recurrence engine as + the scheduler. Inputs and outputs are UTC-aware datetimes; callers can + render those absolute instants in any local timezone. + """ + sched = wf.schedule + if not sched.enabled or not is_schedule_configured(sched): + return [] + if sched.max_runs is not None and sched.runs_count >= sched.max_runs: + return [] + start_utc = _as_utc(from_utc) + end_utc = _as_utc(to_utc) + if start_utc is None or end_utc is None or end_utc <= start_utc: + return [] + + created_at = _as_utc(getattr(wf, "created_at", None)) + cursor_utc = start_utc - timedelta(microseconds=1) + if created_at is not None and created_at > cursor_utc: + cursor_utc = created_at + + ends_at = _as_utc(sched.ends_at) + if ends_at is not None: + if ends_at <= start_utc: + return [] + if ends_at < end_utc: + end_utc = ends_at + + remaining = sched.max_runs - sched.runs_count if sched.max_runs is not None else cap + limit = max(0, min(cap, remaining)) + out: list[datetime] = [] + while len(out) < limit: + nxt = _next_fire_after(sched, cursor_utc) + if nxt is None or nxt >= end_utc: + break + if nxt >= start_utc: + out.append(nxt.astimezone(timezone.utc)) + cursor_utc = nxt + return out + + def kick() -> None: _wake.set() diff --git a/backend/apps/workflows/workflows.py b/backend/apps/workflows/workflows.py index 48729d99..785b7f6b 100644 --- a/backend/apps/workflows/workflows.py +++ b/backend/apps/workflows/workflows.py @@ -1,10 +1,10 @@ import asyncio import logging from contextlib import asynccontextmanager -from datetime import datetime +from datetime import datetime, timezone from typing import Optional -from fastapi import HTTPException, Header, Request +from fastapi import HTTPException, Header, Query, Request from backend.config.Apps import SubApp from backend.apps.workflows.models import ( @@ -96,30 +96,79 @@ def _derive_icon(wf: Workflow) -> str: return "W" -def p_source_session_approvals(session_id: Optional[str]) -> dict[str, str]: +def _source_tool_name(value) -> str: + if not isinstance(value, str): + return "" + name = value.strip() + return name if name else "" + + +def _collect_tool_names_from_content(content, out: set[str]) -> None: + if isinstance(content, list): + for item in content: + _collect_tool_names_from_content(item, out) + return + if not isinstance(content, dict): + return + block_type = content.get("type") + if block_type == "tool_use": + name = _source_tool_name(content.get("name") or content.get("tool")) + if name: + out.add(name) + for key in ("tool_name", "tool"): + name = _source_tool_name(content.get(key)) + if name: + out.add(name) + nested = content.get("content") + if nested is not content: + _collect_tool_names_from_content(nested, out) + + +def p_source_session_memory(session_id: Optional[str]) -> tuple[dict[str, str], list[str]]: if not session_id: - return {} + return {}, [] try: from backend.apps.agents.agent_manager import agent_manager sess = agent_manager.sessions.get(session_id) decisions = getattr(sess, "approval_decisions", None) if sess is not None else None + messages = getattr(sess, "messages", None) if sess is not None else None + tool_latencies = getattr(sess, "tool_latencies", None) if sess is not None else None if decisions is None: from backend.apps.agents.manager.session.session_store import _load_session_data data = _load_session_data(session_id) or {} decisions = data.get("approval_decisions") or [] + messages = data.get("messages") or [] + tool_latencies = data.get("tool_latencies") or {} except Exception: - return {} - out: dict[str, str] = {} + return {}, [] + approvals: dict[str, str] = {} + tools: set[str] = set() for entry in decisions or []: if not isinstance(entry, dict): continue + tool = _source_tool_name(entry.get("tool")) + if tool: + tools.add(tool) if entry.get("sensitive_pattern"): continue - tool = str(entry.get("tool") or "") behavior = entry.get("behavior") if tool and behavior in ("allow", "deny"): - out[tool] = behavior - return out + approvals[tool] = behavior + if isinstance(tool_latencies, dict): + for tool in tool_latencies.keys(): + name = _source_tool_name(tool) + if name: + tools.add(name) + for msg in messages or []: + role = getattr(msg, "role", None) if not isinstance(msg, dict) else msg.get("role") + content = getattr(msg, "content", None) if not isinstance(msg, dict) else msg.get("content") + if role == "tool_call": + tool_name = getattr(msg, "tool_name", None) if not isinstance(msg, dict) else msg.get("tool_name") + name = _source_tool_name(tool_name) + if name: + tools.add(name) + _collect_tool_names_from_content(content, tools) + return approvals, sorted(tools) def p_prune_step_tool_usage(wf: Workflow) -> None: @@ -144,13 +193,34 @@ async def list_workflows(dashboard_id: Optional[str] = None): def _normalize_schedule_state(wf: Workflow) -> None: + if wf.schedule.timezone == "local" and wf.schedule.enabled: + wf.schedule.timezone = scheduler.host_timezone_name() if wf.schedule.enabled and not scheduler.is_schedule_configured(wf.schedule): wf.schedule.enabled = False wf.next_run_at = scheduler.compute_next_fire(wf) if wf.schedule.enabled else None +def _has_nonempty_steps(steps: list[WorkflowStep] | None) -> bool: + return any(bool((s.text or "").strip()) for s in (steps or [])) + + +def _parse_calendar_bound(value: str, label: str) -> datetime: + raw = (value or "").strip() + if raw.endswith("Z"): + raw = raw[:-1] + "+00:00" + try: + dt = datetime.fromisoformat(raw) + except ValueError: + raise HTTPException(status_code=400, detail=f"Invalid {label} timestamp") + if dt.tzinfo is None: + raise HTTPException(status_code=400, detail=f"{label} timestamp must include a timezone") + return dt.astimezone(timezone.utc) + + @workflows.router.post("/create") async def create_workflow(body: WorkflowCreate): + if not body.unsaved and not _has_nonempty_steps(body.steps): + raise HTTPException(status_code=400, detail="Workflow must have at least one step") actions = body.actions # Scheduled workflows default to freeze=on for safety. The user can # flip "Full agent access" in the editor with an explicit confirm. @@ -178,7 +248,13 @@ async def create_workflow(body: WorkflowCreate): auto_named=body.auto_named, unsaved=body.unsaved, ) - wf.remembered_approvals = p_source_session_approvals(body.source_session_id) + source_approvals, source_tools = p_source_session_memory(body.source_session_id) + wf.remembered_approvals = source_approvals + wf.source_tools = source_tools + # Convert-from-chat passes the steps signature so the workflow counts as + # already validated (the chat already prompted for permissions); a blank + # "New" create leaves it None so the first schedule warns to test first. + wf.tested_signature = body.tested_signature if not wf.icon: wf.icon = _derive_icon(wf) _normalize_schedule_state(wf) @@ -507,6 +583,30 @@ async def list_all_runs(limit: int = 200): return {"runs": [r.model_dump(mode="json") for r in runs]} +@workflows.router.get("/calendar") +async def list_calendar_events( + from_: str = Query(..., alias="from"), + to: str = Query(...), + dashboard_id: Optional[str] = None, +): + start_utc = _parse_calendar_bound(from_, "from") + end_utc = _parse_calendar_bound(to, "to") + if end_utc <= start_utc: + raise HTTPException(status_code=400, detail="to must be after from") + items = storage.list_workflows() + if dashboard_id: + items = [w for w in items if not w.dashboard_id or w.dashboard_id == dashboard_id] + events: list[dict] = [] + for wf in items: + for fire_at in scheduler.occurrences_between(wf, start_utc, end_utc): + events.append({ + "workflow_id": wf.id, + "fire_at": fire_at.astimezone(timezone.utc).isoformat(), + }) + events.sort(key=lambda e: (e["fire_at"], e["workflow_id"])) + return {"events": events} + + @workflows.router.get("/{workflow_id}") async def get_workflow(workflow_id: str): wf = storage.get_workflow(workflow_id) @@ -746,15 +846,21 @@ async def commit_draft(workflow_id: str): wf = storage.get_workflow(workflow_id) if not wf: raise HTTPException(status_code=404, detail="Workflow not found") - # Clicking Save is the user committing to this workflow, so reveal it in - # the hub (clears the "+ New" build-in-progress flag) even if there's no - # pending draft to flush. - wf.unsaved = False if wf.draft_steps is None: + if not _has_nonempty_steps(wf.steps): + raise HTTPException(status_code=400, detail="Workflow must have at least one step") + # Clicking Save is the user committing to this workflow, so reveal it + # in the hub (clears the "+ New" build-in-progress flag). + wf.unsaved = False await p_end_edit_session(wf) storage.save_workflow(wf) return _enriched(wf) before = wf.model_dump(mode="json") + if not _has_nonempty_steps(wf.draft_steps): + raise HTTPException(status_code=400, detail="Workflow must have at least one step") + # Clicking Save is the user committing to this workflow, so reveal it in + # the hub (clears the "+ New" build-in-progress flag). + wf.unsaved = False wf.steps = wf.draft_steps wf.draft_steps = None await p_relabel_changed_steps(wf, before.get("steps") or []) @@ -818,6 +924,7 @@ async def test_run_workflow(workflow_id: str, body: dict): wf = storage.get_workflow(workflow_id) if not wf: raise HTTPException(status_code=404, detail="Workflow not found") + tested_signature = body.get("signature") if isinstance(body, dict) else None draft_steps = (body or {}).get("steps") step_entries: list[WorkflowStep] if isinstance(draft_steps, list) and draft_steps: @@ -891,9 +998,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: @@ -901,7 +1007,11 @@ async def test_run_workflow(workflow_id: str, body: dict): final = "error" finally: try: - executor.p_persist_step_tool_usage(wf.id, get_workflow_step_usage(session.id)) + executor.p_persist_step_tool_usage( + wf.id, + get_workflow_step_usage(session.id), + tested_signature=tested_signature if isinstance(tested_signature, str) else None, + ) except Exception: logger.exception("test-run step usage persist failed") set_workflow_approval_step(session.id, None) @@ -956,11 +1066,12 @@ async def schedule_agent_session(workflow_id: str): from backend.apps.agents.core.models import AgentConfig from backend.apps.agents.agent_manager import agent_manager now_local = datetime.now().astimezone() + local_tz = scheduler.host_timezone_name() current_dt = now_local.strftime("%A %Y-%m-%d %H:%M %Z") system_prompt = ( f"You are the Scheduling Agent for the user's saved workflow \"{wf.title}\" " f"(id: {wf.id}). Your only job is to set when this workflow runs.\n\n" - f"The current local date and time is {current_dt}. Resolve relative " + f"The current local date and time is {current_dt} in {local_tz}. Resolve relative " "phrasing (\"this month\", \"next Wednesday\", \"this time\") against it.\n\n" "When the user states a cadence, interpret it yourself and call " "UpdateScheduledWorkflow with:\n" @@ -971,7 +1082,7 @@ async def schedule_agent_session(workflow_id: str): " - repeat_every: the interval count (1 unless they say e.g. \"every other\"; " "for repeat_unit=\"minute\" the minimum is 15, e.g. \"every 15 minutes\")\n" " - on_days: weekday indices when repeat_unit=\"week\" (Sun=0, Mon=1, ... Sat=6)\n" - " - timezone: an IANA name only if the user names a specific zone\n\n" + f" - timezone: \"{local_tz}\" unless the user names a different specific zone\n\n" "If no AM/PM is given, assume PM for 1-7 and AM for 8-12. If the cadence " "is genuinely ambiguous, ask ONE short clarifying question first; otherwise " "go straight to the tool call. The user approves or rejects the change in a " @@ -998,7 +1109,7 @@ async def schedule_agent_session(workflow_id: str): @workflows.router.post("/{workflow_id}/run") -async def run_workflow_now(workflow_id: str): +async def run_workflow_now(workflow_id: str, body: Optional[dict] = None): wf = storage.get_workflow(workflow_id) if not wf: raise HTTPException(status_code=404, detail="Workflow not found") @@ -1006,7 +1117,12 @@ async def run_workflow_now(workflow_id: str): # or we end up with two rows per manual fire (one orphan "running" # row from this handler plus the real one from the executor). pre_ids = {r.id for r in storage.list_runs(wf.id, limit=10)} - asyncio.create_task(executor.execute(wf, triggered_by="manual")) + tested_signature = body.get("signature") if isinstance(body, dict) else None + asyncio.create_task(executor.execute( + wf, + triggered_by="manual", + tested_signature=tested_signature if isinstance(tested_signature, str) else None, + )) # Poll briefly for the newly created run id. We also surface the # run's status + error string when it lands quickly (e.g. cost-cap @@ -1024,54 +1140,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..a4b53d3e 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(): @@ -180,6 +189,122 @@ def test_month_repeat_no_longer_clamps_to_28(): assert nxt.astimezone(tz).date() == datetime(2025, 4, 30).date() +def test_calendar_occurrences_use_schedule_timezone_not_viewer_timezone(): + """A 9am New York schedule returns UTC instants. The frontend can then + render those instants in the viewer's current timezone.""" + from backend.apps.workflows import scheduler + from backend.apps.workflows.models import ScheduleConfig + wf = _make_wf( + schedule=ScheduleConfig( + enabled=True, + repeat_unit="day", + repeat_every=1, + hour=9, + minute=0, + timezone="America/New_York", + ) + ) + wf.created_at = datetime(2026, 1, 1, tzinfo=timezone.utc) + start = datetime(2026, 6, 18, 0, 0, tzinfo=timezone.utc) + end = datetime(2026, 6, 20, 0, 0, tzinfo=timezone.utc) + fires = scheduler.occurrences_between(wf, start, end) + assert len(fires) == 2 + assert fires[0].astimezone(ZoneInfo("America/New_York")).hour == 9 + assert fires[0].astimezone(ZoneInfo("America/Los_Angeles")).hour == 6 + + +def test_calendar_occurrences_stay_wall_clock_across_dst(): + from backend.apps.workflows import scheduler + from backend.apps.workflows.models import ScheduleConfig + wf = _make_wf( + schedule=ScheduleConfig( + enabled=True, + repeat_unit="day", + repeat_every=1, + hour=9, + minute=0, + timezone="America/New_York", + ) + ) + wf.created_at = datetime(2025, 1, 1, tzinfo=timezone.utc) + fires = scheduler.occurrences_between( + wf, + datetime(2025, 3, 8, 0, 0, tzinfo=timezone.utc), + datetime(2025, 3, 11, 0, 0, tzinfo=timezone.utc), + ) + ny = ZoneInfo("America/New_York") + locals_ = [f.astimezone(ny) for f in fires] + assert [d.date() for d in locals_] == [ + datetime(2025, 3, 8).date(), + datetime(2025, 3, 9).date(), + datetime(2025, 3, 10).date(), + ] + assert all((d.hour, d.minute) == (9, 0) for d in locals_) + assert [f.hour for f in fires] == [14, 13, 13] + + +def test_calendar_occurrences_honor_end_conditions(): + from backend.apps.workflows import scheduler + from backend.apps.workflows.models import ScheduleConfig + start = datetime(2026, 6, 18, 0, 0, tzinfo=timezone.utc) + end = datetime(2026, 6, 22, 0, 0, tzinfo=timezone.utc) + wf = _make_wf( + schedule=ScheduleConfig( + enabled=True, + repeat_unit="day", + repeat_every=1, + hour=9, + minute=0, + timezone="UTC", + max_runs=3, + runs_count=1, + ends_at=datetime(2026, 6, 21, 0, 0, tzinfo=timezone.utc), + ) + ) + wf.created_at = datetime(2026, 1, 1, tzinfo=timezone.utc) + fires = scheduler.occurrences_between(wf, start, end) + assert [f.date() for f in fires] == [ + datetime(2026, 6, 18).date(), + datetime(2026, 6, 19).date(), + ] + + wf.schedule.enabled = False + assert scheduler.occurrences_between(wf, start, end) == [] + + wf.schedule.enabled = True + wf.schedule.repeat_unit = "week" + wf.schedule.on_days = [] + assert scheduler.occurrences_between(wf, start, end) == [] + + +def test_calendar_endpoint_returns_sorted_utc_events(): + from backend.apps.workflows import storage + from backend.apps.workflows.workflows import list_calendar_events + from backend.apps.workflows.models import ScheduleConfig + wf = _make_wf( + schedule=ScheduleConfig( + enabled=True, + repeat_unit="day", + repeat_every=1, + hour=9, + minute=0, + timezone="America/New_York", + ) + ) + wf.created_at = datetime(2026, 1, 1, tzinfo=timezone.utc) + storage.save_workflow(wf) + + async def runner(): + return await list_calendar_events( + from_="2026-06-18T00:00:00+00:00", + to="2026-06-20T00:00:00+00:00", + ) + + res = asyncio.new_event_loop().run_until_complete(runner()) + assert [e["workflow_id"] for e in res["events"]] == [wf.id, wf.id] + assert res["events"][0]["fire_at"].startswith("2026-06-18T13:00:00") + + # --- Cost cap ---------------------------------------------------------------- def test_cost_cap_skips_with_clear_error(monkeypatch): @@ -234,6 +359,59 @@ def test_freeze_not_forced_when_source_session_present(): assert result["actions"]["freeze"] is False +def test_create_enabled_schedule_normalizes_local_timezone(monkeypatch): + from backend.apps.workflows import scheduler + from backend.apps.workflows.workflows import create_workflow + from backend.apps.workflows.models import WorkflowCreate, ScheduleConfig + monkeypatch.setenv("OPENSWARM_TIMEZONE", "America/Chicago") + monkeypatch.setattr(scheduler, "_host_tz_cache", None) + body = WorkflowCreate( + title="local-tz-create", + schedule=ScheduleConfig( + enabled=True, + repeat_unit="day", + repeat_every=1, + hour=9, + minute=0, + timezone="local", + ), + ) + result = asyncio.new_event_loop().run_until_complete(create_workflow(body)) + assert result["schedule"]["timezone"] == "America/Chicago" + + +def test_enable_schedule_normalizes_local_timezone_and_preserves_concrete_timezone(monkeypatch): + from backend.apps.workflows import storage, scheduler + from backend.apps.workflows.workflows import update_workflow + from backend.apps.workflows.models import WorkflowUpdate, ScheduleConfig + monkeypatch.setenv("OPENSWARM_TIMEZONE", "America/Denver") + monkeypatch.setattr(scheduler, "_host_tz_cache", None) + + wf = _make_wf() + wf.schedule.enabled = False + wf.schedule.timezone = "local" + storage.save_workflow(wf) + + async def enable_runner(): + sched = ScheduleConfig(**wf.schedule.model_dump(mode="json")) + sched.enabled = True + return await update_workflow(wf.id, WorkflowUpdate(schedule=sched), if_match=None) + + enabled = asyncio.new_event_loop().run_until_complete(enable_runner()) + assert enabled["schedule"]["timezone"] == "America/Denver" + + stored = storage.get_workflow(wf.id) + sched = ScheduleConfig(**stored.schedule.model_dump(mode="json")) + sched.hour = 10 + + async def edit_runner(): + return await update_workflow(wf.id, WorkflowUpdate(schedule=sched), if_match=None) + + edited = asyncio.new_event_loop().run_until_complete(edit_runner()) + assert edited["schedule"]["timezone"] == "America/Denver" + assert edited["schedule"]["hour"] == 10 + + # --- Audit log --------------------------------------------------------------- def test_audit_log_records_title_change(): @@ -319,6 +497,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..477e850f 100644 --- a/frontend/src/app/pages/AgentChat/AgentChat.tsx +++ b/frontend/src/app/pages/AgentChat/AgentChat.tsx @@ -6,6 +6,7 @@ import IconButton from '@mui/material/IconButton'; import Tooltip from '@mui/material/Tooltip'; import TextField from '@mui/material/TextField'; import ClickAwayListener from '@mui/material/ClickAwayListener'; +import Fade from '@mui/material/Fade'; import CloseIcon from '@mui/icons-material/Close'; import KeyboardArrowDownIcon from '@mui/icons-material/KeyboardArrowDown'; import KeyboardArrowUpIcon from '@mui/icons-material/KeyboardArrowUp'; @@ -17,7 +18,7 @@ import DragIndicatorIcon from '@mui/icons-material/DragIndicator'; import RestartAltIcon from '@mui/icons-material/RestartAlt'; import { useAppDispatch, useAppSelector } from '@/shared/hooks'; import { friendlyStatusLabel } from '@/shared/statusLabel'; -import { openSettingsModal } from '@/shared/state/settingsSlice'; +import { openSettingsModal, dismissMcpSuggestion } from '@/shared/state/settingsSlice'; import { API_BASE, getAuthToken } from '@/shared/config'; import { sendMessage as sendMessageThunk, @@ -269,6 +270,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,10 +336,16 @@ 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); const [activateError, setActivateError] = useState(null); + // Holds the last non-empty suggestions so the docked banner's exit fade renders + // them instead of going blank the instant the array is cleared. + const mcpSnapshotRef = useRef>([]); const [mode, setMode] = useState('agent'); const [model, setModel] = useState('sonnet'); @@ -471,7 +497,7 @@ const AgentChat: React.FC = ({ sessionId: sessionIdProp, onClose didDispatchQueued = true; } else { if (curr === 'stopped') { - setShowResumeBubble(true); + setShowResumeBubble(!isWorkflowRunSidecar); } } @@ -489,7 +515,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 @@ -915,9 +941,13 @@ const AgentChat: React.FC = ({ sessionId: sessionIdProp, onClose if (id) dispatch(removeCard(id)); }, [linkedWorkflowId, id, dispatch]); - const onTestSaveWorkflow = useCallback(() => { + const onTestSaveWorkflow = useCallback(async () => { if (linkedWorkflowId) { - dispatch(commitDraft(linkedWorkflowId)); + try { + await dispatch(commitDraft(linkedWorkflowId)).unwrap(); + } catch { + return; + } dispatch(updateWorkflowCard({ workflowId: linkedWorkflowId, patch: { view: 'saved' } })); dispatch(setCardSidecar({ workflowId: linkedWorkflowId, sessionId: null, kind: null })); } @@ -1562,123 +1592,6 @@ const AgentChat: React.FC = ({ sessionId: sessionIdProp, onClose }} > - {(session.mcp_suggestions && session.mcp_suggestions.length > 0) && ( - - id && dispatch(clearMcpSuggestions({ sessionId: id }))} - sx={{ - position: 'absolute', - top: 6, - right: 8, - width: 20, - height: 20, - display: 'flex', - alignItems: 'center', - justifyContent: 'center', - fontSize: '1rem', - lineHeight: 1, - color: c.text.muted, - cursor: 'pointer', - borderRadius: 0.75, - '&:hover': { color: c.text.primary, bgcolor: c.bg.elevated }, - }} - > - × - - - Looks like this might need an integration - - - Activating one of these will let the agent answer in a single round-trip. - - - {session.mcp_suggestions.map((s) => ( - - - - {s.title} - - {s.reason && ( - - {s.reason} - - )} - - { - if (activatingMcp) return; - setActivateError(null); - setActivatingMcp(s.id); - try { - const headers: Record = { 'Content-Type': 'application/json' }; - const tok = (() => { try { return getAuthToken(); } catch { return ''; } })(); - if (tok) headers['Authorization'] = `Bearer ${tok}`; - const r = await fetch(`${API_BASE}/mcp-meta/activate`, { - method: 'POST', - headers, - body: JSON.stringify({ - server_name: s.id.toLowerCase().replace(/\s+/g, '-'), - reason: s.reason || 'preflight suggestion', - parent_session_id: session.id, - }), - }); - const body = await r.json().catch(() => ({} as any)); - if (!r.ok) { - setActivateError(`Activation failed (${r.status})`); - } else if (body?.status === 'unknown_server') { - // Not yet connected; jump straight to Actions - // so the user can finish OAuth. Nothing here - // can do it on their behalf. - navigate('/actions'); - } else if (id) { - // Activation succeeded; clear the banner so the user - // gets visual confirmation the click did something. - dispatch(clearMcpSuggestions({ sessionId: id })); - } - } catch (e: any) { - setActivateError(e?.message || 'Activation failed'); - } finally { - setActivatingMcp(null); - } - }} - sx={{ - cursor: activatingMcp === s.id ? 'wait' : 'pointer', - border: `1px solid ${c.border.medium}`, - borderRadius: 1, - px: 1.25, - py: 0.5, - bgcolor: 'transparent', - color: c.text.primary, - opacity: activatingMcp === s.id ? 0.5 : 1, - '&:hover': { bgcolor: activatingMcp ? 'transparent' : c.bg.elevated }, - flexShrink: 0, - }} - > - {activatingMcp === s.id ? 'Activating…' : 'Activate'} - - - ))} - - {activateError && ( - - {activateError} - - )} - - )} {session.context_overflow && (() => { const reason = session.context_overflow.reason; const isAuth = reason === 'openswarm_pro_auth_expired' || reason === 'anthropic_auth_invalid' || reason === 'auth_error'; @@ -1872,7 +1785,7 @@ const AgentChat: React.FC = ({ sessionId: sessionIdProp, onClose /> )} - {showResumeBubble && session.status === 'stopped' && ( + {showResumeBubble && session.status === 'stopped' && !isWorkflowRunSidecar && ( = ({ sessionId: sessionIdProp, onClose ); })()} + {(() => { + const list = session.mcp_suggestions ?? []; + if (list.length) mcpSnapshotRef.current = list; + const display = mcpSnapshotRef.current; + return ( + 0} timeout={{ enter: 200, exit: 220 }} unmountOnExit> + + { + if (!id) return; + dispatch(clearMcpSuggestions({ sessionId: id })); + dispatch(dismissMcpSuggestion(display.map((s) => s.id))); + }} + sx={{ + position: 'absolute', + top: 6, + right: 8, + width: 20, + height: 20, + display: 'flex', + alignItems: 'center', + justifyContent: 'center', + fontSize: '1rem', + lineHeight: 1, + color: c.text.muted, + cursor: 'pointer', + borderRadius: 0.75, + '&:hover': { color: c.text.primary, bgcolor: c.bg.elevated }, + }} + > + × + + + Looks like this might need an integration + + + Activating one of these will let the agent answer in a single round-trip. + + + {display.map((s) => ( + + + + {s.title} + + {s.reason && ( + + {s.reason} + + )} + + { + if (activatingMcp) return; + setActivateError(null); + setActivatingMcp(s.id); + try { + const headers: Record = { 'Content-Type': 'application/json' }; + const tok = (() => { try { return getAuthToken(); } catch { return ''; } })(); + if (tok) headers['Authorization'] = `Bearer ${tok}`; + const r = await fetch(`${API_BASE}/mcp-meta/activate`, { + method: 'POST', + headers, + body: JSON.stringify({ + server_name: s.id.toLowerCase().replace(/\s+/g, '-'), + reason: s.reason || 'preflight suggestion', + parent_session_id: session.id, + }), + }); + const body = await r.json().catch(() => ({} as any)); + if (!r.ok) { + setActivateError(`Activation failed (${r.status})`); + } else if (body?.status === 'unknown_server') { + // Not yet connected; jump straight to Actions + // so the user can finish OAuth. Nothing here + // can do it on their behalf. + navigate('/actions'); + } else if (id) { + // Activation succeeded; clear the banner so the user + // gets visual confirmation the click did something. + dispatch(clearMcpSuggestions({ sessionId: id })); + } + } catch (e: any) { + setActivateError(e?.message || 'Activation failed'); + } finally { + setActivatingMcp(null); + } + }} + sx={{ + cursor: activatingMcp === s.id ? 'wait' : 'pointer', + border: `1px solid ${c.border.medium}`, + borderRadius: 1, + px: 1.25, + py: 0.5, + bgcolor: 'transparent', + color: c.text.primary, + opacity: activatingMcp === s.id ? 0.5 : 1, + '&:hover': { bgcolor: activatingMcp ? 'transparent' : c.bg.elevated }, + flexShrink: 0, + }} + > + {activatingMcp === s.id ? 'Activating…' : 'Activate'} + + + ))} + + {activateError && ( + + {activateError} + + )} + + + ); + })()} {isStoppableSidecar ? ( ) : ( diff --git a/frontend/src/app/pages/Dashboard/geometry/dashboardTethers.ts b/frontend/src/app/pages/Dashboard/geometry/dashboardTethers.ts index ec5a1e22..9b5c9624 100644 --- a/frontend/src/app/pages/Dashboard/geometry/dashboardTethers.ts +++ b/frontend/src/app/pages/Dashboard/geometry/dashboardTethers.ts @@ -55,6 +55,7 @@ export function elbowPath(x1: number, y1: number, x2: number, y2: number): strin } type Anchor = { x: number; y: number; side: 'left' | 'right' | 'top' | 'bottom' }; +type CanvasRect = { x: number; y: number; width: number; height: number }; // Where the ray from a rect's center toward (tx,ty) crosses the rect border. // Pins a tether endpoint to the card edge facing the other card, so it can @@ -69,6 +70,39 @@ function borderPoint(x: number, y: number, w: number, h: number, tx: number, ty: return { x: cx + dx * scale, y: cy + dy * scale }; } +function rectCenter(r: CanvasRect): { x: number; y: number } { + return { x: r.x + r.width / 2, y: r.y + r.height / 2 }; +} + +function selectCardElement(content: HTMLElement, type: 'agent-card' | 'workflow-card', id: string): HTMLElement | null { + const candidates = content.querySelectorAll(`[data-select-type="${type}"]`); + for (const el of Array.from(candidates)) { + if (el.dataset.selectId === id) return el; + } + return null; +} + +function measuredCanvasRect( + contentRef: RefObject, + zoom: number, + type: 'agent-card' | 'workflow-card', + id: string, +): CanvasRect | null { + const content = contentRef.current; + if (!content) return null; + const el = selectCardElement(content, type, id); + if (!el) return null; + const contentRect = content.getBoundingClientRect(); + const elRect = el.getBoundingClientRect(); + const z = zoom || 1; + return { + x: (elRect.left - contentRect.left) / z, + y: (elRect.top - contentRect.top) / z, + width: elRect.width / z, + height: elRect.height / z, + }; +} + interface UseTethersArgs { glowingAgentCards: Record; glowingBrowserCards: Record; @@ -82,6 +116,8 @@ interface UseTethersArgs { liveDragInfo: LiveDragInfo | null; measuredHeightsRef: RefObject>; measuredHeightsTick: number; + contentRef: RefObject; + zoom: number; sessionList: AgentSession[]; } @@ -98,6 +134,8 @@ export function useTethers({ liveDragInfo, measuredHeightsRef, measuredHeightsTick, + contentRef, + zoom, sessionList, }: UseTethersArgs): Tether[] { return useMemo(() => { @@ -357,10 +395,14 @@ export function useTethers({ ? Math.max(EXPANDED_CARD_MIN_H, sidecar.height) : sidecar.height); const wcH = wfHeight(wc); - const srcCx = srcX + wc.width / 2, srcCy = srcY + wcH / 2; - const dstCx = dstX + sidecar.width / 2, dstCy = dstY + dstH / 2; - const a = borderPoint(srcX, srcY, wc.width, wcH, dstCx, dstCy); - const b = borderPoint(dstX, dstY, sidecar.width, dstH, srcCx, srcCy); + const measuredWorkflow = measuredCanvasRect(contentRef, zoom, 'workflow-card', wc.workflow_id); + const measuredSidecar = measuredCanvasRect(contentRef, zoom, 'agent-card', sidecarId); + const workflowRect = measuredWorkflow ?? { x: srcX, y: srcY, width: wc.width, height: wcH }; + const sidecarRect = measuredSidecar ?? { x: dstX, y: dstY, width: sidecar.width, height: dstH }; + const srcCenter = rectCenter(workflowRect); + const dstCenter = rectCenter(sidecarRect); + const a = borderPoint(workflowRect.x, workflowRect.y, workflowRect.width, workflowRect.height, dstCenter.x, dstCenter.y); + const b = borderPoint(sidecarRect.x, sidecarRect.y, sidecarRect.width, sidecarRect.height, srcCenter.x, srcCenter.y); const x1 = a.x, y1 = a.y; const x2 = b.x, y2 = b.y; const pathD = elbowPath(x1, y1, x2, y2); @@ -429,5 +471,5 @@ export function useTethers({ // measuredHeightsTick re-runs the memo once ResizeObserver reports a new // height after a collapse (the ref read is invisible to the dep checker). // eslint-disable-next-line react-hooks/exhaustive-deps - }, [glowingAgentCards, glowingBrowserCards, cards, browserCards, workflowCards, workflowItems, workflowOpenCards, configurePanels, expandedSessionIds, liveDragInfo, measuredHeightsTick, sessionList]); + }, [glowingAgentCards, glowingBrowserCards, cards, browserCards, workflowCards, workflowItems, workflowOpenCards, configurePanels, expandedSessionIds, liveDragInfo, measuredHeightsTick, contentRef, zoom, sessionList]); } diff --git a/frontend/src/app/pages/Dashboard/hooks/state/useDashboardController.ts b/frontend/src/app/pages/Dashboard/hooks/state/useDashboardController.ts index e37e4233..33e57949 100644 --- a/frontend/src/app/pages/Dashboard/hooks/state/useDashboardController.ts +++ b/frontend/src/app/pages/Dashboard/hooks/state/useDashboardController.ts @@ -288,6 +288,8 @@ export function useDashboardController(dashboardId: string, isActive: boolean) { liveDragInfo, measuredHeightsRef, measuredHeightsTick, + contentRef: canvas.contentRef, + zoom: canvas.zoom, sessionList, }); diff --git a/frontend/src/app/pages/Workflows/AddToSchedulePopover.tsx b/frontend/src/app/pages/Workflows/AddToSchedulePopover.tsx index 8cc8a7d2..0c744bc1 100644 --- a/frontend/src/app/pages/Workflows/AddToSchedulePopover.tsx +++ b/frontend/src/app/pages/Workflows/AddToSchedulePopover.tsx @@ -7,6 +7,7 @@ import { useClaudeTokens } from '@/shared/styles/ThemeContext'; import { useAppDispatch } from '@/shared/hooks'; import { addWorkflowCard } from '@/shared/state/dashboardLayoutSlice'; import { openWorkflowCard, type Workflow } from '@/shared/state/workflowsSlice'; +import { needsScheduleTestWarning } from './scheduleUtils'; interface Props { anchorEl: HTMLElement | null; @@ -23,7 +24,11 @@ export default function AddToSchedulePopover({ anchorEl, workflow, onClose }: Pr const makeSchedule = useCallback(() => { if (!workflow) return; dispatch(addWorkflowCard({ workflowId: workflow.id })); - dispatch(openWorkflowCard({ workflowId: workflow.id, view: 'scheduling' })); + // Untested steps: land on the saved card so its Schedule button can warn and + // offer a test run (which needs the card's sidecar context). Otherwise go + // straight to scheduling. + const view = needsScheduleTestWarning(workflow) ? 'saved' : 'scheduling'; + dispatch(openWorkflowCard({ workflowId: workflow.id, view })); onClose(); }, [dispatch, workflow, onClose]); diff --git a/frontend/src/app/pages/Workflows/EditAgentView.tsx b/frontend/src/app/pages/Workflows/EditAgentView.tsx index da5cf65f..ecf722a3 100644 --- a/frontend/src/app/pages/Workflows/EditAgentView.tsx +++ b/frontend/src/app/pages/Workflows/EditAgentView.tsx @@ -20,6 +20,8 @@ import StepList from './StepList'; import AgentChat from '@/app/pages/AgentChat/AgentChat'; import { useOpenSidecar } from './WorkflowCardLiveViews'; import EditAgentSavePopovers, { type SavePhase } from './EditAgentSavePopovers'; +import { runWorkflowTest } from './runWorkflowTest'; +import { needsScheduleTestWarning } from './scheduleUtils'; interface Props { workflow: Workflow; @@ -112,6 +114,8 @@ export default function EditAgentView({ workflow, steps, isFixMode = false, onEd const [saveAnchorEl, setSaveAnchorEl] = useState(null); const [testSessionId, setTestSessionId] = useState(null); const draftSteps = workflow.draft_steps ?? steps; + const canSave = draftSteps.some((s) => (s.text || '').trim().length > 0); + const allowDiscard = !workflow.unsaved; // A draft always exists in edit mode (we snapshot on entry), so only flag // "unsaved" once the draft actually diverges from the committed steps. const hasChanges = workflow.draft_steps != null && JSON.stringify(workflow.draft_steps) !== JSON.stringify(workflow.steps); @@ -134,35 +138,31 @@ export default function EditAgentView({ workflow, steps, isFixMode = false, onEd } catch { /* best-effort */ } }, [testSessionId]); + const onSaveNow = useCallback(async () => { + if (!canSave) return; + setSavePhase('idle'); + try { + await dispatch(commitDraft(workflow.id)).unwrap(); + } catch { + return; + } + toSaved(); + }, [canSave, dispatch, workflow.id, toSaved]); + const onSaveClick = useCallback((e: React.MouseEvent) => { + if (!canSave) return; + // Already validated this exact version? Skip the "test first?" nudge. + if (!needsScheduleTestWarning(workflow)) { void onSaveNow(); return; } setSaveAnchorEl(e.currentTarget); setSavePhase('ask-test'); - }, []); - - const onSaveNow = useCallback(async () => { - setSavePhase('idle'); - await dispatch(commitDraft(workflow.id)); - toSaved(); - }, [dispatch, workflow.id, toSaved]); + }, [canSave, workflow, onSaveNow]); const onRunTest = useCallback(async () => { - try { - const tok = (() => { try { return getAuthToken(); } catch { return ''; } })(); - const res = await fetch(`${API_BASE}/workflows/${encodeURIComponent(workflow.id)}/test-run`, { - method: 'POST', - headers: { 'Content-Type': 'application/json', ...(tok ? { Authorization: `Bearer ${tok}` } : {}) }, - body: JSON.stringify({ steps: draftSteps }), - }); - if (!res.ok) { setSavePhase('idle'); return; } - const data = await res.json(); - const sid = data?.session_id as string | undefined; - if (!sid) { setSavePhase('idle'); return; } - setTestSessionId(sid); - // The Test Agent card now owns the post-test decision (Continue editing / - // Save workflow) in its own footer, so just close this popover. - setSavePhase('idle'); - await openSidecar(sid, 'testing'); - } catch { setSavePhase('idle'); } + setSavePhase('idle'); + // The Test Agent card now owns the post-test decision (Continue editing / + // Save workflow) in its own footer, so just close this popover. + const sid = await runWorkflowTest(workflow.id, draftSteps, openSidecar); + if (sid) setTestSessionId(sid); }, [workflow.id, draftSteps, openSidecar]); const onDiscardClick = useCallback((e: React.MouseEvent) => { @@ -199,18 +199,22 @@ export default function EditAgentView({ workflow, steps, isFixMode = false, onEd · unsaved )} + {allowDiscard && ( + + Discard + + )} - Discard - - Save @@ -218,8 +222,12 @@ export default function EditAgentView({ workflow, steps, isFixMode = false, onEd {stepsOpen && ( - {isFixMode && fixSeed && setFixPrefixExpanded((x) => !x)} />} + {isFixMode && fixSeed && ( + + setFixPrefixExpanded((x) => !x)} /> + + )} )} diff --git a/frontend/src/app/pages/Workflows/ScheduleCalendar.tsx b/frontend/src/app/pages/Workflows/ScheduleCalendar.tsx index 0ac97031..dcfeb12b 100644 --- a/frontend/src/app/pages/Workflows/ScheduleCalendar.tsx +++ b/frontend/src/app/pages/Workflows/ScheduleCalendar.tsx @@ -1,4 +1,4 @@ -import React, { useMemo, useState } from 'react'; +import React, { useEffect, useMemo, useState } from 'react'; import Box from '@mui/material/Box'; import Typography from '@mui/material/Typography'; import Tooltip from '@mui/material/Tooltip'; @@ -7,10 +7,11 @@ import Menu from '@mui/material/Menu'; import MenuItem from '@mui/material/MenuItem'; import { useClaudeTokens } from '@/shared/styles/ThemeContext'; import { useAppDispatch, useAppSelector } from '@/shared/hooks'; +import { API_BASE } from '@/shared/config'; import type { Workflow } from '@/shared/state/workflowsSlice'; import { runWorkflowNow, deleteWorkflow, updateWorkflow, openWorkflowCard } from '@/shared/state/workflowsSlice'; import { addWorkflowCard } from '@/shared/state/dashboardLayoutSlice'; -import { WEEKDAY_FULL, WEEKDAY_LABEL_SHORT, addDays, sameDay, startOfMonthGrid, startOfWeek, fireTimesWithin, formatTime, formatHourLabel, isScheduleActive } from './scheduleUtils'; +import { WEEKDAY_FULL, WEEKDAY_LABEL_SHORT, addDays, sameDay, startOfMonthGrid, startOfWeek, formatTime, formatHourLabel, stepsSignature } from './scheduleUtils'; interface Props { view: 'Week' | 'Month' | 'List'; @@ -24,6 +25,11 @@ interface Props { // starting hour. The scroll container caps the visible window. const HOURS_24 = Array.from({ length: 24 }, (_, i) => i); +interface CalendarEvent { + workflow_id: string; + fire_at: string; +} + export default function ScheduleCalendar({ view, density, onSelectWorkflow, refDate }: Props) { const c = useClaudeTokens(); const dispatch = useAppDispatch(); @@ -34,7 +40,10 @@ export default function ScheduleCalendar({ view, density, onSelectWorkflow, refD const closeMenu = () => setCtxMenu(null); const onRunNow = () => { if (!ctxMenu) return; - dispatch(runWorkflowNow(ctxMenu.workflow.id)); + dispatch(runWorkflowNow({ + id: ctxMenu.workflow.id, + signature: stepsSignature(ctxMenu.workflow.steps), + })); closeMenu(); }; const onPauseToggle = () => { @@ -74,32 +83,74 @@ export default function ScheduleCalendar({ view, density, onSelectWorkflow, refD Delete ); - // refDate is recreated on every render unless the caller memoizes it, - // which then trips the eventsByDay memo every paint. Pin the calendar - // to a day-precision key so the heavy fireTimesWithin loop only re-runs - // when the day or workflow set actually changed. + // refDate is recreated on every render unless the caller memoizes it. + // Pin the calendar to a day-precision key so occurrence fetches only + // change when the visible day, view, or schedule set changes. const today = refDate || new Date(); const dayKey = `${today.getFullYear()}-${today.getMonth()}-${today.getDate()}`; const compact = density === 'compact'; + const range = view === 'Month' ? 35 : view === 'Week' ? 7 : 14; + const rangeStart = useMemo( + () => view === 'Month' ? startOfMonthGrid(today) : view === 'Week' ? startOfWeek(today) : new Date(today.getFullYear(), today.getMonth(), today.getDate()), + // eslint-disable-next-line react-hooks/exhaustive-deps + [view, dayKey], + ); + const rangeEndExclusive = useMemo(() => addDays(rangeStart, range), [rangeStart, range]); + const [calendarEvents, setCalendarEvents] = useState([]); + const [calendarFetchKey, setCalendarFetchKey] = useState(''); + const workflowScheduleKey = workflows + .map((w) => `${w.id}:${w.updated_at}:${w.schedule.enabled}:${w.schedule.timezone}:${w.schedule.repeat_unit}:${w.schedule.repeat_every}:${w.schedule.hour}:${w.schedule.minute}:${w.schedule.on_days.join(',')}:${w.schedule.ends_at || ''}:${w.schedule.max_runs ?? ''}:${w.schedule.runs_count}`) + .sort() + .join('|'); + const fromIso = rangeStart.toISOString(); + const toIso = rangeEndExclusive.toISOString(); + const calendarRequestKey = `${view}:${fromIso}:${toIso}:${workflowScheduleKey}`; + + useEffect(() => { + let cancelled = false; + const ctrl = new AbortController(); + fetch(`${API_BASE}/workflows/calendar?from=${encodeURIComponent(fromIso)}&to=${encodeURIComponent(toIso)}`, { signal: ctrl.signal }) + .then((res) => { + if (!res.ok) throw new Error(`calendar failed ${res.status}`); + return res.json(); + }) + .then((data) => { + if (cancelled) return; + setCalendarEvents((data.events || []) as CalendarEvent[]); + setCalendarFetchKey(calendarRequestKey); + }) + .catch(() => { + if (cancelled) return; + setCalendarEvents([]); + setCalendarFetchKey(calendarRequestKey); + }); + return () => { + cancelled = true; + ctrl.abort(); + }; + }, [fromIso, toIso, calendarRequestKey]); const eventsByDay = useMemo(() => { - const range = view === 'Month' ? 35 : view === 'Week' ? 7 : 14; - const start = view === 'Month' ? startOfMonthGrid(today) : view === 'Week' ? startOfWeek(today) : today; - const end = addDays(start, range - 1); const map = new Map(); - for (const wf of workflows) { - if (!isScheduleActive(wf.schedule)) continue; - const fires = fireTimesWithin(wf, start, end, 60); - for (const d of fires) { - const key = `${d.getFullYear()}-${d.getMonth()}-${d.getDate()}`; - const arr = map.get(key) || []; - arr.push({ workflow: wf, date: d }); - map.set(key, arr); - } + if (calendarFetchKey !== calendarRequestKey) { + return { map, start: rangeStart, end: rangeEndExclusive, key: calendarFetchKey }; } - return { map, start, end }; - // eslint-disable-next-line react-hooks/exhaustive-deps - }, [workflows, view, dayKey]); + const workflowById = new Map(workflows.map((wf) => [wf.id, wf])); + for (const event of calendarEvents) { + const wf = workflowById.get(event.workflow_id); + if (!wf) continue; + const d = new Date(event.fire_at); + if (Number.isNaN(d.getTime())) continue; + const key = `${d.getFullYear()}-${d.getMonth()}-${d.getDate()}`; + const arr = map.get(key) || []; + arr.push({ workflow: wf, date: d }); + map.set(key, arr); + } + for (const arr of map.values()) { + arr.sort((a, b) => a.date.getTime() - b.date.getTime()); + } + return { map, start: rangeStart, end: rangeEndExclusive, key: calendarFetchKey }; + }, [calendarEvents, calendarFetchKey, calendarRequestKey, workflows, rangeStart, rangeEndExclusive]); const SLOT_H = compact ? 32 : 44; const ROW_LABEL = compact ? '0.7rem' : '0.74rem'; diff --git a/frontend/src/app/pages/Workflows/ScheduleTestWarningDialog.tsx b/frontend/src/app/pages/Workflows/ScheduleTestWarningDialog.tsx new file mode 100644 index 00000000..e4e3d6c4 --- /dev/null +++ b/frontend/src/app/pages/Workflows/ScheduleTestWarningDialog.tsx @@ -0,0 +1,48 @@ +import React from 'react'; +import Box from '@mui/material/Box'; +import Typography from '@mui/material/Typography'; +import Dialog from '@mui/material/Dialog'; +import DialogActions from '@mui/material/DialogActions'; +import DialogContent from '@mui/material/DialogContent'; +import DialogTitle from '@mui/material/DialogTitle'; +import { useClaudeTokens } from '@/shared/styles/ThemeContext'; + +interface Props { + open: boolean; + onClose: () => void; + onTestFirst: () => void; + onScheduleAnyway: () => void; +} + +// Shown before scheduling a workflow whose current steps haven't been validated +// by a test run. Scheduled fires can't pause to ask for tool permission, so an +// untested workflow that needs approval would silently fail on its first run. +export default function ScheduleTestWarningDialog({ open, onClose, onTestFirst, onScheduleAnyway }: Props) { + const c = useClaudeTokens(); + return ( + + Test before scheduling? + + + Scheduled runs can't pause to ask for permission. If this workflow uses tools that + need your approval, it could fail when it runs on its own. A quick test run lets you + approve those tools now. + + + + + Schedule anyway + + + Test first + + + + ); +} diff --git a/frontend/src/app/pages/Workflows/ScheduleThisPopover.tsx b/frontend/src/app/pages/Workflows/ScheduleThisPopover.tsx index d02fedde..81edead4 100644 --- a/frontend/src/app/pages/Workflows/ScheduleThisPopover.tsx +++ b/frontend/src/app/pages/Workflows/ScheduleThisPopover.tsx @@ -30,6 +30,30 @@ const PRESETS: Preset[] = [ { label: 'Every month on the 1st', hint: 'Monthly summary, billing report', build: () => ({ enabled: true, repeat_unit: 'month', repeat_every: 1, hour: 9, minute: 0 }) }, ]; +function extractStepsFromSession(session: { messages?: Array<{ role: string; content: unknown; hidden?: boolean }> } | null | undefined): Array<{ id: string; text: string }> { + const out: Array<{ id: string; text: string }> = []; + for (const msg of session?.messages || []) { + if (msg.role !== 'user' || msg.hidden) continue; + const text = typeof msg.content === 'string' + ? msg.content + : Array.isArray(msg.content) + ? msg.content.map((b: any) => (typeof b === 'string' ? b : b?.text || '')).join(' ') + : ''; + const trimmed = text.trim(); + if (trimmed.length < 6) continue; + out.push({ id: `step-${out.length + 1}-${Date.now().toString(36)}`, text: trimmed.slice(0, 400) }); + if (out.length === 3) break; + } + if (out.length === 0 && session?.messages?.length) { + const fallback = session.messages.find((m) => m.role === 'user'); + if (fallback) { + const text = typeof fallback.content === 'string' ? fallback.content : ''; + out.push({ id: `step-1-${Date.now().toString(36)}`, text: text.slice(0, 400) || 'Run the original task' }); + } + } + return out; +} + interface Props { anchorEl: HTMLElement | null; onClose: () => void; @@ -61,6 +85,7 @@ export default function ScheduleThisPopover({ anchorEl, onClose, sessionId, sess const sessionDashboardId = useAppSelector( (s) => sessionId ? s.agents.sessions[sessionId]?.dashboard_id : null, ); + const sourceSession = useAppSelector((s) => sessionId ? s.agents.sessions[sessionId] : null); // Dup-detect: a chat session can only sanely have one schedule attached. // If we find one already, offer "Open existing" instead of silently @@ -82,6 +107,7 @@ export default function ScheduleThisPopover({ anchorEl, onClose, sessionId, sess const result = await dispatch(createWorkflow({ title, source_session_id: sessionId, + steps: extractStepsFromSession(sourceSession), schedule, } as Partial)); if (createWorkflow.fulfilled.match(result)) { @@ -98,7 +124,7 @@ export default function ScheduleThisPopover({ anchorEl, onClose, sessionId, sess } finally { setBusy(false); } - }, [busy, dispatch, sessionId, title, onClose, onCreated]); + }, [busy, dispatch, sessionId, sourceSession, title, onClose, onCreated]); const openCustom = useCallback(() => { // Open a local draft. NO backend create yet — the workflow only diff --git a/frontend/src/app/pages/Workflows/WorkflowCard.tsx b/frontend/src/app/pages/Workflows/WorkflowCard.tsx index 1bba3997..5b03aca6 100644 --- a/frontend/src/app/pages/Workflows/WorkflowCard.tsx +++ b/frontend/src/app/pages/Workflows/WorkflowCard.tsx @@ -5,6 +5,7 @@ import { motion } from 'framer-motion'; import IconButton from '@mui/material/IconButton'; import Tooltip from '@mui/material/Tooltip'; import Snackbar from '@mui/material/Snackbar'; +import CircularProgress from '@mui/material/CircularProgress'; import CloseIcon from '@mui/icons-material/Close'; import HistoryIcon from '@mui/icons-material/HistoryRounded'; import PlayArrowIcon from '@mui/icons-material/PlayArrowRounded'; @@ -14,6 +15,7 @@ import { useClaudeTokens } from '@/shared/styles/ThemeContext'; import { useAppDispatch, useAppSelector } from '@/shared/hooks'; import { closeWorkflowCard, + controlWorkflowRun, createWorkflow, deleteWorkflow, fetchRuns, @@ -45,6 +47,7 @@ import { Typewriter } from '@/app/components/feedback/Animated'; import StopRounded from '@mui/icons-material/StopRounded'; import PauseRounded from '@mui/icons-material/PauseRounded'; import { StatusDot, RunSparkline, LastFiredHint, isStaleSinceLastRun, isRealTitle } from './workflowVisuals'; +import { stepsSignature } from './scheduleUtils'; import { store } from '@/shared/state/store'; import { getAgentWorkTime, fmtSeconds } from '@/shared/agentWorkTime'; @@ -379,18 +382,24 @@ const WorkflowCard: React.FC = ({ const persistDraft = useCallback(async (): Promise => { const d = card?.draft; if (!d || persistingRef.current) return null; + const draftSteps = d.steps || []; + if (!draftSteps.some((s) => (s.text || '').trim().length > 0)) return null; persistingRef.current = true; try { const result = await dispatch(createWorkflow({ title: (d.title as string) || 'New workflow', description: (d.description as string) || '', - steps: (d.steps || []).map((s) => ({ id: s.id, text: s.text })), + steps: draftSteps.map((s) => ({ id: s.id, text: s.text })), source_session_id: (d.source_session_id as string | undefined) || card?.sourceSessionId || null, use_synced_prompt: true, model: defaultModel || (d.model as string), mode: defaultMode || (d.mode as string), + tested_signature: ((d.source_session_id as string | undefined) || card?.sourceSessionId) + ? stepsSignature(draftSteps) + : undefined, } as Partial)); - const wf = (result as unknown as { payload: Workflow }).payload; + if (!createWorkflow.fulfilled.match(result)) return null; + const wf = result.payload as Workflow; if (!wf?.id) return null; dispatch(rekeyOpenCard({ oldId: workflowId, newId: wf.id })); dispatch(rekeyWorkflowCard({ oldId: workflowId, newId: wf.id })); @@ -641,7 +650,7 @@ const WorkflowCard: React.FC = ({ const wf = await persistDraft(); if (!wf) return; dispatch(openWorkflowCardAction({ workflowId: wf.id, sourceSessionId: card?.sourceSessionId || null, view: 'saved', draft: null })); - await dispatch(runWorkflowNow(wf.id)); + await dispatch(runWorkflowNow({ id: wf.id, signature: stepsSignature(wf.steps) })); await dispatch(fetchRuns(wf.id)); } finally { setTimeout(() => setRunStarting(false), 600); @@ -671,7 +680,7 @@ const WorkflowCard: React.FC = ({ if (runStarting) return; setRunStarting(true); try { - const result = await dispatch(runWorkflowNow(workflow.id)); + const result = await dispatch(runWorkflowNow({ id: workflow.id, signature: stepsSignature(workflow.steps) })); await dispatch(fetchRuns(workflow.id)); if (runWorkflowNow.fulfilled.match(result)) { const payload = result.payload; @@ -984,56 +993,51 @@ function RunningHeader({ workflowId }: { workflowId: string }) { const runs = useAppSelector((s) => 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 2ed7bf52..74a7ce37 100644 --- a/frontend/src/app/pages/Workflows/WorkflowCardLiveViews.tsx +++ b/frontend/src/app/pages/Workflows/WorkflowCardLiveViews.tsx @@ -171,22 +171,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'); @@ -230,11 +214,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/app/pages/Workflows/WorkflowCardSubviews.tsx b/frontend/src/app/pages/Workflows/WorkflowCardSubviews.tsx index 1e65c877..6ebc2d03 100644 --- a/frontend/src/app/pages/Workflows/WorkflowCardSubviews.tsx +++ b/frontend/src/app/pages/Workflows/WorkflowCardSubviews.tsx @@ -26,7 +26,10 @@ import { placeCard, removeWorkflowCard } from '@/shared/state/dashboardLayoutSli import { setPendingFocusAgentId } from '@/shared/state/tempStateSlice'; import { CostChip, humanDuration, routingFor, StreakBadge } from './workflowVisuals'; import StepList from './StepList'; -import { isScheduleConfigured } from './scheduleUtils'; +import { isScheduleConfigured, needsScheduleTestWarning, stepsSignature } from './scheduleUtils'; +import ScheduleTestWarningDialog from './ScheduleTestWarningDialog'; +import { runWorkflowTest } from './runWorkflowTest'; +import { useOpenSidecar } from './WorkflowCardLiveViews'; export function statusColor(s: string, c: ReturnType): string { if (s === 'success') return c.status.success; @@ -123,6 +126,7 @@ export function PreviewView({ workflowId, steps, sourceSessionId, initialDraft, const liveDraft = (card?.draft ?? initialDraft ?? {}) as Partial; const title = (liveDraft.title as string) || 'New workflow'; const description = (liveDraft.description as string) || ''; + const canSave = steps.some((s) => (s.text || '').trim().length > 0); // The new workflow runs with the user's configured default model/mode (their // subscription, etc.), falling back to whatever the source chat used. Without // this the backend picks its own default, which surprised users who'd set a @@ -161,6 +165,7 @@ export function PreviewView({ workflowId, steps, sourceSessionId, initialDraft, }, [closeRequestNonce]); const saveWorkflow = useCallback(async (): Promise => { + if (!canSave) return null; const result = await dispatch(createWorkflow({ title, description, @@ -171,11 +176,15 @@ export function PreviewView({ workflowId, steps, sourceSessionId, initialDraft, // happened to run on, so a converted workflow behaves like a fresh chat. model: defaultModel || (liveDraft.model as string), mode: defaultMode || (liveDraft.mode as string), + // Converting a chat carries its prior approvals, so count it as already + // validated for these steps: scheduling won't nag to test first. + tested_signature: sourceSessionId ? stepsSignature(steps) : undefined, } as Partial)); - const wf = (result as unknown as { payload: Workflow }).payload; + if (!createWorkflow.fulfilled.match(result)) return null; + const wf = result.payload as Workflow; if (wf?.id) return wf; return null; - }, [dispatch, title, description, steps, sourceSessionId, liveDraft, defaultModel, defaultMode]); + }, [canSave, dispatch, title, description, steps, sourceSessionId, liveDraft, defaultModel, defaultMode]); const onIgnore = useCallback(async () => { if (busy) return; @@ -183,7 +192,7 @@ export function PreviewView({ workflowId, steps, sourceSessionId, initialDraft, }, [busy]); const onSaveThenSchedule = useCallback(async () => { - if (busy) return; + if (busy || !canSave) return; setBusy(true); try { const wf = await saveWorkflow(); @@ -191,10 +200,10 @@ export function PreviewView({ workflowId, steps, sourceSessionId, initialDraft, } finally { setBusy(false); } - }, [busy, saveWorkflow, onSaved]); + }, [busy, canSave, saveWorkflow, onSaved]); const onSaveDraft = useCallback(async () => { - if (busy) return; + if (busy || !canSave) return; setBusy(true); try { const wf = await saveWorkflow(); @@ -203,7 +212,7 @@ export function PreviewView({ workflowId, steps, sourceSessionId, initialDraft, setBusy(false); setSavePromptOpen(false); } - }, [busy, saveWorkflow, onSaved]); + }, [busy, canSave, saveWorkflow, onSaved]); const onDontSave = useCallback(() => { setSavePromptOpen(false); @@ -252,15 +261,16 @@ export function PreviewView({ workflowId, steps, sourceSessionId, initialDraft, Not now Schedule Workflow @@ -288,8 +298,9 @@ export function PreviewView({ workflowId, steps, sourceSessionId, initialDraft, + onClick={canSave ? onSaveDraft : undefined} + title={canSave ? undefined : 'Add at least one step before saving'} + sx={{ fontSize: '0.84rem', fontWeight: 700, color: '#fff', bgcolor: c.accent.primary, borderRadius: 999, cursor: busy ? 'wait' : canSave ? 'pointer' : 'not-allowed', px: 1.5, py: 0.6, opacity: busy || !canSave ? 0.6 : 1, '&:hover': { filter: 'brightness(1.06)' } }}> Save @@ -343,9 +354,25 @@ export function SavedView({ workflow, steps, runs, activeRunId }: { workflow: Wo const openEditAgent = useCallback(() => { dispatch(updateWorkflowCard({ workflowId: workflow.id, patch: { view: 'edit_agent' } })); }, [dispatch, workflow.id]); + const openSidecar = useOpenSidecar(workflow.id); + const [warnOpen, setWarnOpen] = useState(false); const openScheduling = useCallback(() => { dispatch(updateWorkflowCard({ workflowId: workflow.id, patch: { view: 'scheduling', showScheduleNudge: false } })); }, [dispatch, workflow.id]); + // Gate the schedule action: warn first if the current steps haven't been + // validated by a test run (so an unattended fire won't silently deny a tool). + const requestSchedule = useCallback(() => { + if (needsScheduleTestWarning(workflow)) { setWarnOpen(true); return; } + openScheduling(); + }, [workflow, openScheduling]); + const onTestFirst = useCallback(() => { + setWarnOpen(false); + void runWorkflowTest(workflow.id, workflow.draft_steps ?? workflow.steps, openSidecar); + }, [workflow.id, workflow.draft_steps, workflow.steps, openSidecar]); + const onScheduleAnyway = useCallback(() => { + setWarnOpen(false); + openScheduling(); + }, [openScheduling]); const onToggleStep = useCallback((stepId: string) => { dispatch(toggleExpandedStep({ workflowId: workflow.id, stepId })); }, [dispatch, workflow.id]); @@ -434,7 +461,7 @@ export function SavedView({ workflow, steps, runs, activeRunId }: { workflow: Wo Not now {showNudge ? : ( + setWarnOpen(false)} + onTestFirst={onTestFirst} + onScheduleAnyway={onScheduleAnyway} + /> ); } diff --git a/frontend/src/app/pages/Workflows/WorkflowsHubCard.tsx b/frontend/src/app/pages/Workflows/WorkflowsHubCard.tsx index 3bbe53ac..149599d7 100644 --- a/frontend/src/app/pages/Workflows/WorkflowsHubCard.tsx +++ b/frontend/src/app/pages/Workflows/WorkflowsHubCard.tsx @@ -28,7 +28,7 @@ import Tooltip from '@mui/material/Tooltip'; import { useEffect } from 'react'; import ScheduleCalendar from './ScheduleCalendar'; import AddToSchedulePopover from './AddToSchedulePopover'; -import { WEEKDAY_LABEL, addDays, sameDay, startOfMonthGrid, isWorkflowSchedulable } from './scheduleUtils'; +import { WEEKDAY_LABEL, addDays, sameDay, startOfMonthGrid, isWorkflowSchedulable, stepsSignature } from './scheduleUtils'; import { isRealTitle } from './workflowVisuals'; import { Typewriter } from '@/app/components/feedback/Animated'; @@ -144,6 +144,23 @@ const WorkflowsHubCard: React.FC = ({ const [view, setView] = useState('List'); const [viewOpen, setViewOpen] = useState(false); const [refDate, setRefDate] = useState(new Date()); + + // The hub card lives on the canvas for days at a time, so a refDate frozen + // at mount leaves the calendar stuck on the day it was opened (e.g. still + // showing "yesterday" past midnight). Roll it forward when the day flips, + // but only if the user was parked on today, so manual navigation is left be. + const refDateRef = useRef(refDate); + refDateRef.current = refDate; + useEffect(() => { + let lastToday = new Date(); + const id = window.setInterval(() => { + const now = new Date(); + if (sameDay(now, lastToday)) return; + if (sameDay(refDateRef.current, lastToday)) setRefDate(now); + lastToday = now; + }, 60000); + return () => window.clearInterval(id); + }, []); const [search, setSearch] = useState(''); const [sidebarOpen, setSidebarOpen] = useState(true); // Right-click on a sidebar row opens this menu pinned to the cursor. @@ -522,7 +539,10 @@ const WorkflowsHubCard: React.FC = ({ anchorPosition={sidebarCtxMenu ? { top: sidebarCtxMenu.y, left: sidebarCtxMenu.x } : undefined}> { if (!sidebarCtxMenu) return; - dispatch(runWorkflowNow(sidebarCtxMenu.workflow.id)); + dispatch(runWorkflowNow({ + id: sidebarCtxMenu.workflow.id, + signature: stepsSignature(sidebarCtxMenu.workflow.steps), + })); closeSidebarCtxMenu(); }}>Run now {sidebarCtxMenu && isWorkflowSchedulable(sidebarCtxMenu.workflow) && ( diff --git a/frontend/src/app/pages/Workflows/runWorkflowTest.ts b/frontend/src/app/pages/Workflows/runWorkflowTest.ts new file mode 100644 index 00000000..28e74cc9 --- /dev/null +++ b/frontend/src/app/pages/Workflows/runWorkflowTest.ts @@ -0,0 +1,32 @@ +import { API_BASE, getAuthToken } from '@/shared/config'; +import type { WorkflowStep } from '@/shared/state/workflowsSlice'; +import { stepsSignature } from './scheduleUtils'; + +type OpenSidecar = (sessionId: string, kind: 'testing') => Promise; + +// Kick off a Test Agent run for the given (possibly-draft) steps and wire its +// session into the workflow card's sidecar. The signature rides along so a +// completed test run stamps the workflow as validated (see scheduleUtils + +// the test-run endpoint). Returns the session id, or null if it didn't start. +export async function runWorkflowTest( + workflowId: string, + steps: WorkflowStep[], + openSidecar: OpenSidecar, +): Promise { + try { + const tok = (() => { try { return getAuthToken(); } catch { return ''; } })(); + const res = await fetch(`${API_BASE}/workflows/${encodeURIComponent(workflowId)}/test-run`, { + method: 'POST', + headers: { 'Content-Type': 'application/json', ...(tok ? { Authorization: `Bearer ${tok}` } : {}) }, + body: JSON.stringify({ steps, signature: stepsSignature(steps) }), + }); + if (!res.ok) return null; + const data = await res.json(); + const sid = data?.session_id as string | undefined; + if (!sid) return null; + await openSidecar(sid, 'testing'); + return sid; + } catch { + return null; + } +} diff --git a/frontend/src/app/pages/Workflows/scheduleUtils.ts b/frontend/src/app/pages/Workflows/scheduleUtils.ts index 51ecea0c..53b89d56 100644 --- a/frontend/src/app/pages/Workflows/scheduleUtils.ts +++ b/frontend/src/app/pages/Workflows/scheduleUtils.ts @@ -1,4 +1,4 @@ -import type { Workflow, ScheduleConfig } from '@/shared/state/workflowsSlice'; +import type { Workflow, ScheduleConfig, WorkflowStep } from '@/shared/state/workflowsSlice'; export const WEEKDAY_LABEL = ['S', 'M', 'T', 'W', 'T', 'F', 'S']; export const WEEKDAY_LABEL_SHORT = ['SUN', 'MON', 'TUE', 'WED', 'THU', 'FRI', 'SAT']; @@ -39,6 +39,23 @@ export function isWorkflowSchedulable(workflow: Workflow): boolean { return isScheduleConfigured(workflow.schedule); } +// Stable fingerprint of the steps that actually drive behavior (order + id + +// text). label is just the at-a-glance headline, so it's left out. Computed +// only here so the backend stores exactly what the FE compares: no cross- +// language hashing drift. +export function stepsSignature(steps: WorkflowStep[] | null | undefined): string { + return JSON.stringify((steps || []).map((s) => [s.id, s.text])); +} + +// True when the current steps haven't been validated by a test run (or seeded +// at chat conversion) since they were last edited. Drives the test-first +// warning before scheduling. +export function needsScheduleTestWarning(workflow: Workflow): boolean { + const steps = workflow.draft_steps ?? workflow.steps; + if (!steps || steps.length === 0) return false; + return stepsSignature(steps) !== (workflow.tested_signature ?? ''); +} + export function formatTime(hour: number, minute: number): string { const h12 = ((hour + 11) % 12) + 1; const suffix = hour < 12 ? 'am' : 'pm'; diff --git a/frontend/src/shared/state/settingsSlice.ts b/frontend/src/shared/state/settingsSlice.ts index 90518f83..b951d7fd 100644 --- a/frontend/src/shared/state/settingsSlice.ts +++ b/frontend/src/shared/state/settingsSlice.ts @@ -168,6 +168,19 @@ export const resetSystemPrompt = createAsyncThunk( } ); +export const dismissMcpSuggestion = createAsyncThunk( + 'settings/dismissMcpSuggestion', + async (ids: string[]) => { + const res = await fetch(`${SETTINGS_API}/dismiss-mcp-suggestion`, { + method: 'PUT', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ ids }), + }); + const data = await res.json(); + return data.settings as AppSettings; + } +); + export const browseDirectories = createAsyncThunk( 'settings/browseDirectories', async (path: string) => { @@ -298,6 +311,10 @@ const settingsSlice = createSlice({ state.data = action.payload; state.draft = null; state.draftTab = null; + }) + .addCase(dismissMcpSuggestion.fulfilled, (state, action) => { + state.latestWriteId = action.meta.requestId; + state.data = action.payload; }); }, }); diff --git a/frontend/src/shared/state/workflowsSlice.ts b/frontend/src/shared/state/workflowsSlice.ts index 9da866ba..52d820bd 100644 --- a/frontend/src/shared/state/workflowsSlice.ts +++ b/frontend/src/shared/state/workflowsSlice.ts @@ -91,10 +91,16 @@ export interface Workflow { * unattended scheduled fire doesn't stall on a prompt. tool name -> answer. */ remembered_approvals?: Record; step_tool_usage?: Record>; + /** Tool names observed in the source chat when this workflow was generated. */ + source_tools?: string[]; /** False once the user explicitly renames the workflow; backend may auto-rename while true. */ auto_named?: boolean; /** True while a brand-new "+ New" workflow is still being built and hasn't been saved; hub hides these. */ unsaved?: boolean; + /** Signature of the steps last validated by a test run (or seeded at chat + * conversion). Compared against the current steps to decide whether to warn + * before scheduling. See scheduleUtils.needsScheduleTestWarning. */ + tested_signature?: string | null; } export interface WorkflowRun { @@ -113,8 +119,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; @@ -169,9 +180,73 @@ interface State { allRuns: WorkflowRun[]; allRunsLoading: boolean; runningToast: RunningToast | null; + runControlPending: Record; } -const initialState: State = { items: {}, runs: {}, openCards: {}, loaded: false, loading: false, paused: false, active: [], cloudSmsEnabled: false, allRuns: [], allRunsLoading: false, runningToast: null }; +const initialState: State = { items: {}, runs: {}, openCards: {}, loaded: false, loading: false, paused: false, active: [], cloudSmsEnabled: false, allRuns: [], allRunsLoading: false, runningToast: null, 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]; + // A scheduled run flipping into 'running' fired unattended, so nudge the + // user with a clickable toast. Only on the into-running edge (not every + // tool-label/step bump), and only for schedule (manual runs they kicked + // off themselves don't need a "surprise, it's running" popup). + if (r.status === 'running' && r.triggered_by === 'schedule' && (!prev || prev.status !== 'running')) { + state.runningToast = { + workflowId: r.workflow_id, + runId: r.id, + workflowTitle: state.items[r.workflow_id]?.title || 'Workflow', + }; + } + 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', @@ -249,8 +324,18 @@ export const deleteWorkflow = createAsyncThunk('workflows/delete', async (id: st return id; }); -export const runWorkflowNow = createAsyncThunk('workflows/run', async (id: string) => { - const res = await fetch(`${API}/${id}/run`, { method: 'POST' }); +type RunWorkflowNowArg = string | { id: string; signature?: string | null }; + +export const runWorkflowNow = createAsyncThunk('workflows/run', async (arg: RunWorkflowNowArg) => { + const id = typeof arg === 'string' ? arg : arg.id; + const signature = typeof arg === 'string' ? null : arg.signature; + const res = await fetch(`${API}/${id}/run`, { + method: 'POST', + ...(signature ? { + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ signature }), + } : {}), + }); if (!res.ok) throw new Error(`run failed ${res.status}`); const data = await res.json(); return { @@ -261,6 +346,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) => { @@ -335,62 +434,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'; - } - } - // A scheduled run flipping into 'running' fired unattended, so nudge the - // user with a clickable toast. Only on the into-running edge (not every - // tool-label/step bump), and only for schedule (manual runs they kicked - // off themselves don't need a "surprise, it's running" popup). - if (r.status === 'running' && r.triggered_by === 'schedule' && (!prev || prev.status !== 'running')) { - state.runningToast = { - workflowId: r.workflow_id, - runId: r.id, - workflowTitle: state.items[r.workflow_id]?.title || 'Workflow', - }; - } - }, - dismissRunningToast(state) { - state.runningToast = null; + mergeRunIntoState(state, action.payload); }, toggleExpandedStep(state, action: { payload: { workflowId: string; stepId: string } }) { const card = state.openCards[action.payload.workflowId]; @@ -421,6 +465,9 @@ const slice = createSlice({ delete state.runs[action.payload]; state.allRuns = state.allRuns.filter((r) => r.workflow_id !== action.payload); }, + dismissRunningToast(state) { + state.runningToast = null; + }, }, extraReducers: (builder) => { builder @@ -441,8 +488,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) => { diff --git a/frontend/src/shared/ws/WebSocketManager.ts b/frontend/src/shared/ws/WebSocketManager.ts index 78b517f6..6d41d1aa 100644 --- a/frontend/src/shared/ws/WebSocketManager.ts +++ b/frontend/src/shared/ws/WebSocketManager.ts @@ -28,6 +28,7 @@ import { addBrowserCardFromBackend, markBrowserCardEnding, setBrowserCardPositio import { upsertOutput } from '../state/outputsSlice'; import { displaySessionName } from '../state/sessionDisplay'; import { upsertRun, ackRun, runWorkflowNow, openWorkflowCard, upsertWorkflow, removeWorkflow } from '../state/workflowsSlice'; +import { stepsSignature } from '@/app/pages/Workflows/scheduleUtils'; import { getAuthToken } from '../config'; import { notifyAgentCompletion } from '../notifications'; @@ -946,7 +947,8 @@ import { WS_BASE } from '@/shared/config'; return; } if (outcome === 'rerun') { - store.dispatch(runWorkflowNow(workflowId)); + const wf = store.getState().workflows.items[workflowId]; + store.dispatch(wf ? runWorkflowNow({ id: workflowId, signature: stepsSignature(wf.steps) }) : runWorkflowNow(workflowId)); return; } if (outcome === 'edit' || outcome === 'open') {