diff --git a/backend/apps/agents/agent_manager.py b/backend/apps/agents/agent_manager.py
index 431e73d7..15d8155b 100644
--- a/backend/apps/agents/agent_manager.py
+++ b/backend/apps/agents/agent_manager.py
@@ -1061,23 +1061,6 @@ class AgentManager:
mcp_registry_ctx = self._build_mcp_registry_summary(session.allowed_tools, session.active_mcps)
global_settings = load_settings()
- # Nudge the agent to surface ScheduleWorkflow proactively when
- # the user's ask looks recurring. The per-tool description
- # carries the full protocol; this is just the "when to think
- # about it" signal so the agent doesn't ignore the surface.
- schedule_ctx = (
- "\n"
- "After completing a substantive task, if the work looks "
- "repeatable (the user said 'every', 'each', 'daily', "
- "'weekly', 'morning', 'before standup', or you just did "
- "the same sequence twice in this session), offer to "
- "schedule it. Use AskUserQuestion to confirm cadence, "
- "then ScheduleWorkflow to create it. Never reach for "
- "crontab, launchctl, or schtasks; always use the native "
- "scheduler so the user can see, pause, and edit it. "
- "Don't ask after trivial one-off requests.\n"
- ""
- )
composed_prompt = self._compose_system_prompt(
global_settings.default_system_prompt,
mode_sys_prompt,
@@ -1086,15 +1069,20 @@ class AgentManager:
browser_ctx,
mcp_registry_ctx,
)
- composed_prompt = (composed_prompt + "\n\n" + schedule_ctx) if composed_prompt else schedule_ctx
# Pin the agent's notion of "now" to the host wall clock + zone
- # so it can answer day-of-week questions and pick sensible
- # cadences ("every Friday afternoon") without hallucinating.
+ # so it can answer day-of-week questions without hallucinating.
try:
from zoneinfo import ZoneInfo
- from backend.apps.workflows.storage import _resolve_host_tz_name
- tz_name = _resolve_host_tz_name()
+ # Best-effort IANA name for the host. Mirrors apps/service/client.py.
+ tz_name = os.environ.get("OPENSWARM_TIMEZONE", "").strip()
+ if not tz_name:
+ try:
+ from tzlocal import get_localzone_name # type: ignore
+ tz_name = get_localzone_name() or ""
+ except Exception:
+ tz_name = ""
+ tz_name = tz_name or "UTC"
now_local = datetime.now(ZoneInfo(tz_name))
tz_abbr = now_local.strftime("%Z") or tz_name
time_ctx = (
@@ -1192,27 +1180,6 @@ class AgentManager:
"type": "stdio",
}
- # Always-on schedule server. Exposes ScheduleWorkflow + CRUD
- # tools so the agent can offer to schedule recurring work via
- # the native scheduler (visible, auditable) rather than reach
- # for cron/launchctl. Tool descriptions tell the agent to
- # AskUserQuestion FIRST to confirm cadence with the user.
- schedule_server_path = os.path.join(
- os.path.dirname(__file__), "schedule_mcp_server.py"
- )
- from backend.auth import get_auth_token as _get_auth_token_sched
- mcp_servers["openswarm-schedule"] = {
- "command": sys.executable,
- "args": [schedule_server_path],
- "env": {
- "OPENSWARM_PORT": os.environ.get("OPENSWARM_PORT", "8324"),
- "OPENSWARM_AUTH_TOKEN": _get_auth_token_sched(),
- "OPENSWARM_PARENT_SESSION_ID": session.id,
- "OPENSWARM_DASHBOARD_ID": session.dashboard_id or "",
- },
- "type": "stdio",
- }
-
# Always-on meta-MCP server. Exposes MCPList / MCPSearch /
# MCPActivate so the model can discover and activate user MCPs at
# runtime. The activation gate (active_mcps filter in
diff --git a/backend/apps/agents/schedule_mcp_server.py b/backend/apps/agents/schedule_mcp_server.py
deleted file mode 100644
index 5a75a603..00000000
--- a/backend/apps/agents/schedule_mcp_server.py
+++ /dev/null
@@ -1,412 +0,0 @@
-#!/usr/bin/env python3
-"""Stdio MCP server exposing scheduled-workflow tools to the agent.
-
-Why this exists: the agent should be able to schedule recurring work on
-the user's behalf, but ALWAYS through the native scheduler (visible,
-auditable, cost-capped) rather than `crontab`. Each tool is a thin
-wrapper around /api/workflows/*. The descriptions are written to nudge
-the agent toward AskUserQuestion-first behavior (confirm cadence with
-the user before calling ScheduleWorkflow).
-"""
-
-import json
-import sys
-import os
-import urllib.request
-import urllib.error
-
-BACKEND_PORT = os.environ.get("OPENSWARM_PORT", "8324")
-BACKEND_AUTH = os.environ.get("OPENSWARM_AUTH_TOKEN", "")
-BACKEND_BASE = f"http://127.0.0.1:{BACKEND_PORT}/api/workflows"
-PARENT_SESSION_ID = os.environ.get("OPENSWARM_PARENT_SESSION_ID", "")
-DASHBOARD_ID = os.environ.get("OPENSWARM_DASHBOARD_ID", "")
-
-
-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]},
- "weekly_monday": {"enabled": True, "repeat_unit": "week", "repeat_every": 1, "hour": 9, "minute": 0, "on_days": [1]},
- "weekly_friday": {"enabled": True, "repeat_unit": "week", "repeat_every": 1, "hour": 17, "minute": 0, "on_days": [5]},
- "monthly_first": {"enabled": True, "repeat_unit": "month", "repeat_every": 1, "hour": 9, "minute": 0, "on_days": []},
-}
-
-
-TOOLS = [
- {
- "name": "ScheduleWorkflow",
- "description": (
- "Create a recurring scheduled workflow for the user. Use this "
- "ONLY after confirming cadence with the user via AskUserQuestion "
- "(do not assume — the user must pick or accept the time). "
- "The workflow runs the listed steps on the schedule and is "
- "visible in the user's Workflows hub. Never use crontab, "
- "launchctl, or schtasks to schedule recurring work; always use "
- "this tool so the user can see, pause, edit, or delete it. "
- "After creating, briefly confirm to the user what was scheduled."
- ),
- "inputSchema": {
- "type": "object",
- "properties": {
- "title": {"type": "string", "description": "Short workflow name shown in the hub and on the dashboard card."},
- "steps": {
- "type": "array",
- "items": {"type": "string"},
- "description": "Ordered list of instructions for the agent to execute on each fire. Each string is one step.",
- },
- "preset": {
- "type": "string",
- "enum": ["daily_morning", "weekdays_morning", "weekly_monday", "weekly_friday", "monthly_first", "custom"],
- "description": "Cadence preset. Use 'custom' to specify your own hour/minute/days.",
- },
- "hour": {"type": "integer", "description": "Hour 0-23 in the user's local time. Required when preset='custom'."},
- "minute": {"type": "integer", "description": "Minute 0/15/30/45. Required when preset='custom'."},
- "repeat_unit": {"type": "string", "enum": ["day", "week", "month"], "description": "Required when preset='custom'."},
- "on_days": {
- "type": "array",
- "items": {"type": "integer"},
- "description": "Weekdays (Sun=0..Sat=6) when preset='custom' and repeat_unit='week'.",
- },
- "source_session_id": {"type": "string", "description": "Optional; the chat session this workflow was created from. Inherits its tool surface."},
- },
- "required": ["title", "steps", "preset"],
- },
- },
- {
- "name": "ListScheduledWorkflows",
- "description": "List the user's scheduled workflows. Use this to find a workflow the user is referring to before editing or deleting it.",
- "inputSchema": {"type": "object", "properties": {}},
- },
- {
- "name": "UpdateScheduledWorkflow",
- "description": "Modify an existing scheduled workflow. Only pass the fields you want to change. Always confirm with the user via AskUserQuestion before making changes that meaningfully alter behavior (cadence, steps, permissions).",
- "inputSchema": {
- "type": "object",
- "properties": {
- "workflow_id": {"type": "string"},
- "title": {"type": "string"},
- "steps": {"type": "array", "items": {"type": "string"}},
- "schedule_enabled": {"type": "boolean", "description": "Quick on/off without changing other schedule fields."},
- "hour": {"type": "integer"},
- "minute": {"type": "integer"},
- "repeat_unit": {"type": "string", "enum": ["day", "week", "month"]},
- "on_days": {"type": "array", "items": {"type": "integer"}},
- },
- "required": ["workflow_id"],
- },
- },
- {
- "name": "DeleteScheduledWorkflow",
- "description": "Permanently delete a scheduled workflow. Cannot be undone. ALWAYS confirm via AskUserQuestion before calling this — the user should pick from a list, not have you guess.",
- "inputSchema": {
- "type": "object",
- "properties": {"workflow_id": {"type": "string"}},
- "required": ["workflow_id"],
- },
- },
- {
- "name": "PauseAllWorkflows",
- "description": "Globally pause every scheduled workflow. In-flight runs finish; future runs are blocked until resumed. Use when the user wants a temporary stop (vacation, debugging) without deleting workflows.",
- "inputSchema": {"type": "object", "properties": {}},
- },
- {
- "name": "ResumeAllWorkflows",
- "description": "Resume scheduled workflows after a previous PauseAllWorkflows.",
- "inputSchema": {"type": "object", "properties": {}},
- },
- {
- "name": "RunWorkflowNow",
- "description": "Trigger an immediate one-off run of a scheduled workflow. The schedule continues to fire on its normal cadence in addition.",
- "inputSchema": {
- "type": "object",
- "properties": {"workflow_id": {"type": "string"}},
- "required": ["workflow_id"],
- },
- },
- {
- "name": "EditWorkflowStep",
- "description": (
- "Edit a single step's prompt text on an existing workflow. Use "
- "when the user has accepted a proposed change during an Edit "
- "Agent conversation; the new prompt replaces the existing one "
- "and persists immediately. The next scheduled run uses the new "
- "version. Always confirm the change with the user before "
- "calling this; AskUserQuestion FIRST if there is any ambiguity."
- ),
- "inputSchema": {
- "type": "object",
- "properties": {
- "workflow_id": {"type": "string", "description": "The workflow to edit."},
- "step_idx": {"type": "integer", "description": "0-based index of the step to modify."},
- "new_text": {"type": "string", "description": "Full replacement prompt text for the step."},
- },
- "required": ["workflow_id", "step_idx", "new_text"],
- },
- },
- {
- "name": "TestWorkflow",
- "description": (
- "Spawn a sibling Test Agent that runs the workflow end-to-end "
- "(with the latest persisted steps) so the user can watch it "
- "work. Use after editing a step to verify the change. The Test "
- "Agent renders as a sibling card on the dashboard with a "
- "'Testing' arrow chip linking back to this workflow."
- ),
- "inputSchema": {
- "type": "object",
- "properties": {
- "workflow_id": {"type": "string", "description": "The workflow to test."},
- },
- "required": ["workflow_id"],
- },
- },
-]
-
-
-def send_response(id_, result=None, error=None):
- msg = {"jsonrpc": "2.0", "id": id_}
- if error is not None:
- msg["error"] = error
- else:
- msg["result"] = result
- sys.stdout.write(json.dumps(msg) + "\n")
- sys.stdout.flush()
-
-
-def _call(method: str, path: str, body=None) -> dict:
- url = BACKEND_BASE + path
- data = json.dumps(body).encode() if body is not None else None
- headers = {"Content-Type": "application/json"}
- if BACKEND_AUTH:
- headers["Authorization"] = f"Bearer {BACKEND_AUTH}"
- req = urllib.request.Request(url, data=data, headers=headers, method=method)
- try:
- with urllib.request.urlopen(req, timeout=30) as resp:
- return json.loads(resp.read().decode() or "null") or {}
- except urllib.error.HTTPError as e:
- body_err = e.read().decode() if e.fp else str(e)
- return {"_error": f"HTTP {e.code}: {body_err}"}
- except Exception as e:
- return {"_error": str(e)}
-
-
-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}
- if preset == "custom":
- return {
- **base,
- "enabled": True,
- "repeat_unit": args.get("repeat_unit", "day"),
- "repeat_every": 1,
- "hour": int(args.get("hour", 9)),
- "minute": int(args.get("minute", 0)),
- "on_days": list(args.get("on_days") or []),
- }
- preset_def = PRESETS.get(preset)
- if not preset_def:
- return {}
- return {**base, **preset_def, "repeat_every": 1}
-
-
-def handle_schedule_workflow(args: dict) -> dict:
- title = args.get("title") or "Scheduled workflow"
- steps_in = args.get("steps") or []
- preset = args.get("preset") or "daily_morning"
- schedule = _build_schedule_from_preset(preset, args)
- if not schedule:
- return _err(f"Unknown preset: {preset}. Use one of: {list(PRESETS.keys()) + ['custom']}.")
- body = {
- "title": title,
- "steps": [{"id": f"s{i+1}", "text": s} for i, s in enumerate(steps_in) if s],
- "schedule": schedule,
- "source_session_id": args.get("source_session_id") or PARENT_SESSION_ID or None,
- }
- r = _call("POST", "/create", body)
- if "_error" in r:
- return _err(r["_error"])
- wid = r.get("id", "")
- nxt = r.get("next_run_at") or "soon"
- return _ok(f"Scheduled \"{title}\" ({preset}). Workflow id: {wid}. Next run: {nxt}. The user can view, pause, or edit it in the Workflows hub.")
-
-
-def handle_list(_args: dict) -> dict:
- r = _call("GET", "/list")
- if "_error" in r:
- return _err(r["_error"])
- ws = r.get("workflows", [])
- if not ws:
- return _ok("No scheduled workflows yet.")
- lines = ["Scheduled workflows:"]
- for w in ws:
- s = w.get("schedule") or {}
- enabled = s.get("enabled")
- unit = s.get("repeat_unit", "?")
- hour = s.get("hour")
- title = w.get("title", "(untitled)")
- wid = w.get("id", "")
- state = "ON" if enabled else "off"
- lines.append(f" - {title} [{state}] {unit} at {hour:02d}:00 (id: {wid})")
- return _ok("\n".join(lines))
-
-
-def handle_update(args: dict) -> dict:
- wid = args.get("workflow_id") or ""
- if not wid:
- return _err("workflow_id is required.")
- cur = _call("GET", f"/{wid}")
- if "_error" in cur:
- return _err(cur["_error"])
- sched = cur.get("schedule") or {}
- patch: dict = {}
- if "title" in args: patch["title"] = args["title"]
- if "steps" in args:
- patch["steps"] = [{"id": f"s{i+1}", "text": s} for i, s in enumerate(args["steps"] or []) if s]
- sched_patch = dict(sched)
- sched_dirty = False
- if "schedule_enabled" in args:
- sched_patch["enabled"] = bool(args["schedule_enabled"])
- sched_dirty = True
- for k in ("hour", "minute", "repeat_unit", "on_days"):
- if k in args:
- sched_patch[k] = args[k]
- sched_dirty = True
- if sched_dirty:
- patch["schedule"] = sched_patch
- if not patch:
- return _ok(f"No changes requested for workflow {wid}.")
- r = _call("PATCH", f"/{wid}", patch)
- if "_error" in r:
- return _err(r["_error"])
- return _ok(f"Updated \"{r.get('title', wid)}\". Next run: {r.get('next_run_at') or 'paused/unscheduled'}.")
-
-
-def handle_delete(args: dict) -> dict:
- wid = args.get("workflow_id") or ""
- if not wid:
- return _err("workflow_id is required.")
- r = _call("DELETE", f"/{wid}")
- if "_error" in r:
- return _err(r["_error"])
- return _ok(f"Deleted workflow {wid}.")
-
-
-def handle_pause_all(_args: dict) -> dict:
- r = _call("POST", "/pause-all")
- if "_error" in r:
- return _err(r["_error"])
- return _ok("All scheduled workflows are paused. In-flight runs will finish; future fires are blocked. Resume with ResumeAllWorkflows.")
-
-
-def handle_resume_all(_args: dict) -> dict:
- r = _call("POST", "/resume-all")
- if "_error" in r:
- return _err(r["_error"])
- return _ok("Scheduled workflows resumed.")
-
-
-def handle_run_now(args: dict) -> dict:
- wid = args.get("workflow_id") or ""
- if not wid:
- return _err("workflow_id is required.")
- r = _call("POST", f"/{wid}/run")
- if "_error" in r:
- return _err(r["_error"])
- if r.get("status") == "skipped":
- return _ok(f"Run was skipped: {r.get('error', 'unknown reason')}.")
- return _ok(f"Run started (run id: {r.get('run_id', '')}). Output will appear in the workflow's History.")
-
-
-def _ok(text: str) -> dict:
- return {"content": [{"type": "text", "text": text}]}
-
-
-def _err(text: str) -> dict:
- return {"content": [{"type": "text", "text": f"Error: {text}"}], "isError": True}
-
-
-def handle_edit_step(args: dict) -> dict:
- wid = args.get("workflow_id") or ""
- if not wid:
- return _err("workflow_id is required.")
- try:
- idx = int(args.get("step_idx"))
- except (TypeError, ValueError):
- return _err("step_idx must be an integer.")
- new_text = (args.get("new_text") or "").strip()
- if not new_text:
- return _err("new_text is required.")
- cur = _call("GET", f"/{wid}")
- if "_error" in cur:
- return _err(cur["_error"])
- steps = cur.get("steps") or []
- if idx < 0 or idx >= len(steps):
- return _err(f"step_idx {idx} out of range (workflow has {len(steps)} steps).")
- new_steps = list(steps)
- new_steps[idx] = {**new_steps[idx], "text": new_text}
- r = _call("PATCH", f"/{wid}", {"steps": new_steps})
- if "_error" in r:
- return _err(r["_error"])
- return _ok(f"Step {idx + 1} updated. The next run uses the new prompt.")
-
-
-def handle_test_workflow(args: dict) -> dict:
- wid = args.get("workflow_id") or ""
- if not wid:
- return _err("workflow_id is required.")
- r = _call("POST", f"/{wid}/test-run", {})
- if "_error" in r:
- return _err(r["_error"])
- sid = r.get("session_id", "")
- return _ok(f"Test Agent spawned (session {sid[:8]}...). It runs the latest workflow on the dashboard with a Testing arrow chip.")
-
-
-HANDLERS = {
- "ScheduleWorkflow": handle_schedule_workflow,
- "ListScheduledWorkflows": handle_list,
- "UpdateScheduledWorkflow": handle_update,
- "DeleteScheduledWorkflow": handle_delete,
- "PauseAllWorkflows": handle_pause_all,
- "ResumeAllWorkflows": handle_resume_all,
- "RunWorkflowNow": handle_run_now,
- "EditWorkflowStep": handle_edit_step,
- "TestWorkflow": handle_test_workflow,
-}
-
-
-def main():
- for line in sys.stdin:
- line = line.strip()
- if not line:
- continue
- try:
- msg = json.loads(line)
- except json.JSONDecodeError:
- continue
- method = msg.get("method")
- id_ = msg.get("id")
- params = msg.get("params", {})
- if method == "initialize":
- send_response(id_, {
- "protocolVersion": "2024-11-05",
- "capabilities": {"tools": {}},
- "serverInfo": {"name": "openswarm-schedule", "version": "1.0.0"},
- })
- elif method == "notifications/initialized":
- pass
- elif method == "tools/list":
- send_response(id_, {"tools": TOOLS})
- elif method == "tools/call":
- tool_name = params.get("name", "")
- arguments = params.get("arguments", {})
- handler = HANDLERS.get(tool_name)
- if handler is None:
- send_response(id_, _err(f"Unknown tool: {tool_name}"))
- else:
- send_response(id_, handler(arguments))
- elif method == "ping":
- send_response(id_, {})
- elif id_ is not None:
- send_response(id_, error={"code": -32601, "message": f"Method not found: {method}"})
-
-
-if __name__ == "__main__":
- main()
diff --git a/backend/apps/dashboards/models.py b/backend/apps/dashboards/models.py
index b62c608f..52be0543 100644
--- a/backend/apps/dashboards/models.py
+++ b/backend/apps/dashboards/models.py
@@ -53,17 +53,12 @@ class NotePosition(BaseModel):
class DashboardLayout(BaseModel):
- # Accept whatever the FE serialises (workflow_cards, configure_panels,
- # workflows_hub etc). Pydantic was silently stripping these because
- # they weren't declared, which made the dashboard re-render WITHOUT
- # the workflow card the user just placed.
+ # extra="allow" so any keys the FE sends (or legacy on-disk layouts
+ # carry) round-trip without Pydantic stripping them.
model_config = ConfigDict(extra="allow")
cards: dict[str, CardPosition] = Field(default_factory=dict)
view_cards: dict[str, ViewCardPosition] = Field(default_factory=dict)
browser_cards: dict[str, BrowserCardPosition] = Field(default_factory=dict)
- workflow_cards: dict = Field(default_factory=dict)
- configure_panels: dict = Field(default_factory=dict)
- workflows_hub: Optional[dict] = None
notes: dict[str, NotePosition] = Field(default_factory=dict)
expanded_session_ids: list[str] = Field(default_factory=list)
diff --git a/backend/apps/workflows/__init__.py b/backend/apps/workflows/__init__.py
deleted file mode 100644
index e69de29b..00000000
diff --git a/backend/apps/workflows/audit.py b/backend/apps/workflows/audit.py
deleted file mode 100644
index 4ea43a33..00000000
--- a/backend/apps/workflows/audit.py
+++ /dev/null
@@ -1,95 +0,0 @@
-"""Append-only audit log for workflow edits.
-
-One JSONL file per workflow at /workflows/audit/.jsonl. We
-diff before/after rather than snapshotting the full record so the file
-stays small even after dozens of edits. Read path tails the file; we don't
-keep this in memory because audits are inspected rarely.
-"""
-
-import json
-import logging
-import os
-from datetime import datetime, timezone
-from threading import Lock
-from typing import Any
-
-from backend.apps.workflows.storage import DATA_DIR
-
-logger = logging.getLogger(__name__)
-
-AUDIT_DIR = os.path.join(DATA_DIR, "audit")
-_io_lock = Lock()
-# Soft cap on bytes per audit file. When exceeded we truncate to the last
-# CAP/2 bytes on next write so attackers (or a runaway PATCH loop) can't
-# fill the disk. 256 KiB is ~2000 edits; we never expect to hit it.
-SOFT_CAP_BYTES = 256 * 1024
-
-
-def _audit_path(wid: str) -> str:
- return os.path.join(AUDIT_DIR, f"{wid}.jsonl")
-
-
-def _diff(before: dict, after: dict) -> dict[str, dict[str, Any]]:
- """Return only the keys whose value changed. Nested dicts are diffed
- shallowly; the schedule/actions/permissions blocks are small so we just
- record the whole sub-dict when any sub-key changes.
- """
- changed: dict[str, dict[str, Any]] = {}
- keys = set(before) | set(after)
- for k in keys:
- b = before.get(k)
- a = after.get(k)
- if b != a:
- changed[k] = {"before": b, "after": a}
- return changed
-
-
-def log_change(wid: str, who: str, before: dict, after: dict) -> None:
- diff = _diff(before, after)
- if not diff:
- return
- entry = {
- "ts": datetime.now(timezone.utc).isoformat(),
- "who": who,
- "diff": diff,
- }
- try:
- with _io_lock:
- os.makedirs(AUDIT_DIR, exist_ok=True)
- path = _audit_path(wid)
- if os.path.exists(path) and os.path.getsize(path) > SOFT_CAP_BYTES:
- # Keep the tail half. Cheap, lossy, prevents pathological
- # disk growth without crashing on a corrupt file.
- with open(path, "rb") as f:
- f.seek(-(SOFT_CAP_BYTES // 2), os.SEEK_END)
- tail = f.read()
- first_nl = tail.find(b"\n")
- tail = tail[first_nl + 1:] if first_nl >= 0 else b""
- with open(path, "wb") as f:
- f.write(tail)
- with open(path, "a") as f:
- f.write(json.dumps(entry) + "\n")
- except Exception:
- logger.debug("audit log_change failed", exc_info=True)
-
-
-def read_tail(wid: str, limit: int = 50) -> list[dict]:
- path = _audit_path(wid)
- if not os.path.exists(path):
- return []
- try:
- with open(path) as f:
- lines = f.readlines()
- except Exception:
- return []
- out: list[dict] = []
- for line in lines[-limit:]:
- line = line.strip()
- if not line:
- continue
- try:
- out.append(json.loads(line))
- except Exception:
- continue
- out.reverse()
- return out
diff --git a/backend/apps/workflows/escalation.py b/backend/apps/workflows/escalation.py
deleted file mode 100644
index a0ab7536..00000000
--- a/backend/apps/workflows/escalation.py
+++ /dev/null
@@ -1,90 +0,0 @@
-"""Server-side escalation timer.
-
-The permission chain in the UI (notify -> text -> call) used to time out
-client-side, which dies the moment the window closes. We move the timer
-here so a run that finishes at 9am can escalate to a real text at 9:05am
-whether or not the user has the app open. The text/call wire-up itself
-still routes through notifier (cloud SMS bridge is wired separately); we
-just own the *when*.
-
-State lives in module-scoped dicts, not on disk. If the backend restarts
-mid-escalation the chain is lost on purpose: the user is already in front
-of an open app at that point (otherwise the backend wouldn't have started)
-and they can ack manually. Persisting escalation state would mean
-re-firing on a stale schedule after a multi-day downtime, which is worse.
-"""
-
-import asyncio
-import logging
-from datetime import datetime, timedelta, timezone
-from typing import Optional
-
-from backend.apps.workflows.models import PermissionTier, Workflow, WorkflowRun
-
-logger = logging.getLogger(__name__)
-
-
-_tasks: dict[str, asyncio.Task] = {} # run_id -> escalation task
-_state: dict[str, dict] = {} # run_id -> {tier_idx, next_at, kind}
-
-
-def _tier_delay_seconds(tier: PermissionTier) -> int:
- """Tier minutes/hours convention matches the FE: text uses minutes,
- call uses hours (the UI label flips with tier.kind). We translate at
- the boundary so the backend math is always in seconds."""
- if tier.kind == "call":
- return max(0, tier.after_minutes) * 3600
- return max(0, tier.after_minutes) * 60
-
-
-def schedule(wf: Workflow, run: WorkflowRun) -> None:
- """Kick off escalation for a finished run. No-op if the workflow has
- only the default notify tier (i.e. nothing to escalate to)."""
- tiers = wf.permissions or []
- if len(tiers) <= 1:
- return
- # Cancel any prior task for this run (defense against a re-fire).
- cancel(run.id)
- task = asyncio.create_task(_runner(wf, run, tiers))
- _tasks[run.id] = task
-
-
-def cancel(run_id: str) -> bool:
- task = _tasks.pop(run_id, None)
- _state.pop(run_id, None)
- if task is None:
- return False
- task.cancel()
- return True
-
-
-def status(run_id: str) -> Optional[dict]:
- return _state.get(run_id)
-
-
-async def _runner(wf: Workflow, run: WorkflowRun, tiers: list[PermissionTier]) -> None:
- from backend.apps.workflows.notifier import send_tier
-
- try:
- # Tier 0 is the initial notify; we don't re-fire it here. Walk
- # 1..N, sleeping the tier's delay before sending. If the user acks
- # via /workflows/runs/{run_id}/ack, the task is cancelled.
- for idx in range(1, len(tiers)):
- tier = tiers[idx]
- delay = _tier_delay_seconds(tier)
- fire_at = datetime.now(timezone.utc) + timedelta(seconds=delay)
- _state[run.id] = {
- "tier_idx": idx,
- "tier_kind": tier.kind,
- "next_at": fire_at.isoformat(),
- }
- await asyncio.sleep(delay)
- try:
- await send_tier(wf, run, tier)
- except Exception:
- logger.exception("escalation send_tier failed run=%s tier=%s", run.id, tier.kind)
- except asyncio.CancelledError:
- pass
- finally:
- _state.pop(run.id, None)
- _tasks.pop(run.id, None)
diff --git a/backend/apps/workflows/executor.py b/backend/apps/workflows/executor.py
deleted file mode 100644
index 28073db5..00000000
--- a/backend/apps/workflows/executor.py
+++ /dev/null
@@ -1,334 +0,0 @@
-"""Run a workflow by launching an agent session and feeding it the steps.
-
-The executor is intentionally thin: it leans entirely on agent_manager's
-existing launch + send_message path so a scheduled run looks identical to
-a manual chat. That keeps the MCP gate, action filtering, provider
-routing, retries, and history all aligned with the rest of the app.
-"""
-
-import asyncio
-import logging
-from datetime import datetime, timedelta, timezone
-from typing import Optional
-
-from backend.apps.agents.core.models import AgentConfig
-from backend.apps.workflows.models import Workflow, WorkflowRun
-from backend.apps.workflows import storage
-
-logger = logging.getLogger(__name__)
-
-
-# In-process map: workflow_id -> currently running run id. Prevents two
-# overlapping fires for the same workflow (e.g. cron tick races a manual
-# Run button) without serializing across the whole executor.
-_running: dict[str, str] = {}
-_running_lock = asyncio.Lock()
-
-
-def _resolve_system_prompt(wf: Workflow) -> Optional[str]:
- if wf.use_synced_prompt:
- return None
- return wf.system_prompt or None
-
-
-def _resolve_allowed_tools(wf: Workflow) -> list[str]:
- if not wf.actions.freeze:
- return []
- return list(wf.actions.configured_sets)
-
-
-def _persist_run_fields(wf: Workflow, run_fields: dict, schedule_runs_count_delta: int = 0) -> None:
- """Merge run-side fields into the current on-disk workflow.
-
- The executor holds the `wf` it was launched with; meanwhile the user
- may have PATCHed unrelated fields (title, schedule, permissions...).
- Saving our captured `wf` would clobber those edits. Re-read the
- authoritative record from storage and only mutate the run-side fields
- we own. If the workflow has been deleted while we ran, silently skip
- the save so we don't resurrect a deleted record.
-
- schedule_runs_count_delta is a small int (0 or 1) that we add to the
- on-disk schedule.runs_count to avoid the same race overwriting an
- in-flight bump on the user's PATCH path.
- """
- fresh = storage.get_workflow(wf.id)
- if fresh is None:
- # Deleted while we ran. Don't resurrect.
- return
- for k, v in run_fields.items():
- setattr(fresh, k, v)
- if schedule_runs_count_delta:
- fresh.schedule.runs_count = fresh.schedule.runs_count + schedule_runs_count_delta
- storage.save_workflow(fresh)
-
-
-def _monthly_spend_so_far(wf: Workflow) -> float:
- """Sum cost_usd across runs of `wf` started in the last 30 days.
-
- Reads the bounded run log (200 rows max per workflow), so this is
- O(history) and runs once per fire. Naive datetimes (legacy rows) are
- treated as host-local then normalized to UTC by Python's astimezone.
- """
- cutoff = datetime.now(timezone.utc) - timedelta(days=30)
- total = 0.0
- for r in storage.list_runs(wf.id, limit=200):
- started = r.started_at
- if started is None:
- continue
- if started.tzinfo is None:
- started = started.astimezone(timezone.utc)
- else:
- started = started.astimezone(timezone.utc)
- if started >= cutoff:
- total += float(r.cost_usd or 0.0)
- return total
-
-
-async def execute(wf: Workflow, triggered_by: str = "schedule", scheduled_for: Optional[datetime] = None) -> WorkflowRun:
- from backend.apps.agents.agent_manager import agent_manager
-
- run = WorkflowRun(
- workflow_id=wf.id,
- status="running",
- scheduled_for=scheduled_for,
- started_at=datetime.now(),
- triggered_by=triggered_by,
- )
-
- # Cost cap pre-check happens before claiming `_running` so a capped
- # workflow doesn't block its own next fire. We still record the run so
- # the user sees it in History with a clear reason.
- if wf.cost_cap_usd_monthly is not None:
- spent = _monthly_spend_so_far(wf)
- if spent >= wf.cost_cap_usd_monthly:
- run.status = "skipped"
- run.error = f"Monthly cost cap reached (${spent:.2f} / ${wf.cost_cap_usd_monthly:.2f})"
- run.finished_at = datetime.now()
- storage.record_run(run)
- _persist_run_fields(wf, {
- "last_run_at": run.finished_at,
- "last_run_status": "skipped",
- "last_run_id": run.id,
- })
- return run
-
- storage.record_run(run)
-
- async with _running_lock:
- if wf.id in _running:
- run.status = "skipped"
- run.error = "Previous run still active"
- run.finished_at = datetime.now()
- storage.record_run(run)
- return run
- _running[wf.id] = run.id
-
- wf.last_run_at = run.started_at
- wf.last_run_status = "running"
- wf.last_run_id = run.id
- _persist_run_fields(wf, {
- "last_run_at": run.started_at,
- "last_run_status": "running",
- "last_run_id": run.id,
- })
-
- session = None
- try:
- steps = [s.text for s in wf.steps if s.text and s.text.strip()]
- if not steps:
- raise ValueError("Workflow has no steps")
-
- config = AgentConfig(
- name=wf.title or "Workflow",
- model=wf.model or "sonnet",
- mode=wf.mode or "agent",
- provider=wf.provider or "anthropic",
- system_prompt=_resolve_system_prompt(wf),
- allowed_tools=_resolve_allowed_tools(wf) or [
- "Read", "Edit", "Write", "Bash", "Glob", "Grep", "AskUserQuestion",
- ],
- dashboard_id=wf.dashboard_id,
- )
-
- session = await agent_manager.launch_agent(config)
- run.session_id = session.id
- storage.record_run(run)
-
- # Background poller: surface the latest tool-call name as a
- # live "what's the agent doing" subtitle on the workflow:run
- # ws event. Cheap enough to run at 1.5s cadence; nothing else
- # is watching session.messages from here. Cancelled in the
- # finally block alongside _running cleanup.
- async def _watch_tool_calls() -> None:
- last_seen = ""
- while True:
- try:
- await asyncio.sleep(1.5)
- sess = agent_manager.sessions.get(session.id)
- if not sess:
- return
- msgs = getattr(sess, "messages", []) or []
- label = ""
- for m in reversed(msgs):
- if getattr(m, "role", None) != "tool_call":
- continue
- content = getattr(m, "content", None)
- # Content can be a string, a dict with "name", or
- # a list of blocks. Pick the first tool_use name.
- if isinstance(content, list):
- for b in content:
- if isinstance(b, dict) and b.get("type") == "tool_use":
- label = str(b.get("name") or "")
- break
- elif isinstance(content, dict):
- label = str(content.get("name") or "")
- elif isinstance(content, str):
- label = content[:60]
- if label:
- break
- if label and label != last_seen:
- last_seen = label
- run.last_tool_label = label
- try:
- from backend.apps.agents.core.ws_manager import ws_manager
- await ws_manager.broadcast_global("workflow:run", {
- "workflow_id": wf.id,
- "run": run.model_dump(mode="json"),
- })
- except Exception:
- pass
- except asyncio.CancelledError:
- return
- except Exception:
- return
-
- watcher_task = asyncio.create_task(_watch_tool_calls())
-
- # Send each step sequentially. agent_manager.send_message is a no-op
- # while a prior turn is still streaming, so we await until the
- # session is idle before posting the next step. Keeps the runner
- # safe regardless of how long each turn takes.
- step_error: Optional[str] = None
- for idx, step in enumerate(steps):
- # Broadcast the step bump before sending so RunningView flips
- # the disc immediately, not after the agent finishes the step.
- run.active_step_idx = idx
- run.last_tool_label = None
- try:
- from backend.apps.agents.core.ws_manager import ws_manager as _wsm
- await _wsm.broadcast_global("workflow:run", {
- "workflow_id": wf.id,
- "run": run.model_dump(mode="json"),
- })
- except Exception:
- pass
- await agent_manager.send_message(session.id, step)
- await _await_session_idle(session.id)
- sess_state = agent_manager.sessions.get(session.id)
- if sess_state is not None and getattr(sess_state, "status", None) == "error":
- step_error = "Agent session entered error state"
- # Pin active step so FailedView can render the X on the
- # right row. error_step_idx == active_step_idx at fail time.
- break
-
- run.finished_at = datetime.now()
- 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)
-
- if step_error is not None:
- run.status = "failure"
- run.error = step_error
- wf.last_run_status = "failure"
- elif scheduled_for is not None and (run.finished_at.replace(tzinfo=None) - scheduled_for.replace(tzinfo=None)).total_seconds() > 300:
- # Started more than 5 minutes after its slot (app was closed,
- # event loop backed up, etc.). Surface in History as ran_late
- # so the user can tell apart "fired on time" from "caught up".
- # Strip tz before the subtraction so a UTC-aware scheduled_for
- # (new code path) and a naive finished_at don't raise.
- run.status = "ran_late"
- wf.last_run_status = "ran_late"
- else:
- run.status = "success"
- wf.last_run_status = "success"
- # Bump runs_count for scheduled fires that reached a terminal state
- # other than "skipped". Manual runs don't count against max_runs.
- 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, {
- "last_run_at": run.finished_at,
- "last_run_status": wf.last_run_status,
- }, 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()
- storage.record_run(run)
- wf.last_run_status = "failure"
- _persist_run_fields(wf, {
- "last_run_status": "failure",
- "last_run_at": run.finished_at,
- })
- finally:
- # Cancel the tool-call watcher before we tear the session down so
- # the next poll doesn't race close_session.
- try:
- watcher_task.cancel() # type: ignore[name-defined]
- except Exception:
- pass
- # Close the workflow's agent session so closed_at is set and the
- # run shows up in chat history (get_history sorts by closed_at;
- # sessions with closed_at=None sort to the bottom and fall off
- # the first page). close_session also drops in-memory state and
- # persists the final snapshot to disk.
- if session is not None:
- try:
- await agent_manager.close_session(session.id)
- except Exception:
- logger.exception("close_session failed for workflow run %s", run.id)
- async with _running_lock:
- _running.pop(wf.id, None)
-
- try:
- from backend.apps.workflows.notifier import notify_run_complete
- await notify_run_complete(wf, run)
- except Exception:
- logger.debug("notifier failed", exc_info=True)
-
- try:
- from backend.apps.agents.core.ws_manager import ws_manager
- await ws_manager.broadcast_global("workflow:run", {
- "workflow_id": wf.id,
- "run": run.model_dump(mode="json"),
- })
- except Exception:
- pass
-
- 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.
-
- 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.
- """
- from backend.apps.agents.agent_manager import agent_manager
-
- deadline = asyncio.get_event_loop().time() + timeout_s
- while True:
- 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
- status = getattr(sess, "status", None)
- if status in ("completed", "error", "stopped"):
- return
- 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
deleted file mode 100644
index 9819b17f..00000000
--- a/backend/apps/workflows/models.py
+++ /dev/null
@@ -1,164 +0,0 @@
-from pydantic import BaseModel, ConfigDict, Field, field_validator
-from typing import Optional, Literal, Any
-from datetime import datetime
-from uuid import uuid4
-
-
-# Each "tier" in the permission chain: notify in app, fall through to text
-# after N minutes if no response, then to call after a further N minutes/hours.
-# Matches images 17 to 19 (Schedule edit). Order in the list = escalation order.
-class PermissionTier(BaseModel):
- kind: Literal["notify", "text", "call"] = "notify"
- after_minutes: int = 0
- phone: Optional[str] = None
-
-
-class ScheduleConfig(BaseModel):
- enabled: bool = False
- # Bounds keep the scheduler from blowing up on malformed input. The
- # FE clamps these too, but defense-in-depth: a misbehaving agent
- # tool, an old JSON file, or a curl-wielding power user shouldn't
- # be able to crash _next_fire_after by passing hour=99.
- repeat_every: int = Field(default=1, ge=1, le=365)
- repeat_unit: Literal["day", "week", "month"] = "week"
- on_days: list[int] = Field(default_factory=list)
- hour: int = Field(default=9, ge=0, le=23)
- minute: int = Field(default=0, ge=0, le=59)
- # IANA zone name (e.g. "America/Los_Angeles") or "local" for legacy
- # records that predate explicit tz. storage._load_all_from_disk coerces
- # "local" to the host zone in memory; we leave it on disk until the
- # user's next save so backup/sync tools don't see spurious churn.
- timezone: str = "local"
- on_missed: Literal["skip", "run_once", "run_all"] = "skip"
- # Optional end conditions. None = forever / unbounded. Schedule auto-
- # disables once either is satisfied; scheduler._tick zeroes out
- # next_run_at and flips enabled=False so the UI reflects reality.
- ends_at: Optional[datetime] = None
- max_runs: Optional[int] = Field(default=None, ge=1)
- runs_count: int = Field(default=0, ge=0)
-
- @field_validator("on_days")
- @classmethod
- def _clean_on_days(cls, v: list[int]) -> list[int]:
- # Backend uses JS-style weekday (Sun=0..Sat=6). Drop entries
- # outside that range so a malformed PATCH can't trip the
- # scheduler later, and dedupe while preserving order.
- seen: set[int] = set()
- out: list[int] = []
- for d in v or []:
- if isinstance(d, int) and 0 <= d <= 6 and d not in seen:
- seen.add(d)
- out.append(d)
- return out
-
-
-class ActionsConfig(BaseModel):
- prevent_unused: bool = False
- freeze: bool = False
- configured_sets: list[str] = Field(default_factory=list)
-
-
-class WorkflowStep(BaseModel):
- id: str = Field(default_factory=lambda: uuid4().hex)
- text: str = ""
- # 3 to 6 word LLM-generated headline shown in the collapsed step row.
- # The full prompt lives in `text`; this is the "at-a-glance" label.
- label: Optional[str] = None
-
-
-def _empty_str_default() -> str:
- return ""
-
-
-class Workflow(BaseModel):
- # validate_assignment is load-bearing for the PATCH /workflows/{id} path
- # (workflows.py:update_workflow setattr's raw dicts from body.model_dump
- # straight onto the cached Workflow). Without coercion the nested
- # schedule/steps/actions/permissions fields become plain dicts in
- # memory, and every downstream call; scheduler tick, executor.execute,
- # subsequent PATCHes; crashes on `.enabled` / `.text`.
- model_config = ConfigDict(validate_assignment=True)
-
- id: str = Field(default_factory=lambda: uuid4().hex)
- title: str = "Untitled workflow"
- description: str = ""
- icon: str = ""
- system_prompt: Optional[str] = None
- use_synced_prompt: bool = True
- steps: list[WorkflowStep] = Field(default_factory=list)
- actions: ActionsConfig = Field(default_factory=ActionsConfig)
- schedule: ScheduleConfig = Field(default_factory=ScheduleConfig)
- permissions: list[PermissionTier] = Field(
- default_factory=lambda: [PermissionTier(kind="notify")]
- )
- source_session_id: Optional[str] = None
- dashboard_id: Optional[str] = None
- model: str = "sonnet"
- mode: str = "agent"
- provider: str = "anthropic"
- created_at: datetime = Field(default_factory=datetime.now)
- updated_at: datetime = Field(default_factory=datetime.now)
- last_run_at: Optional[datetime] = None
- last_run_status: Optional[Literal["success", "failure", "ran_late", "running", "skipped"]] = None
- last_run_id: Optional[str] = None
- next_run_at: Optional[datetime] = None
- cost_cap_usd_monthly: Optional[float] = None
- # Sticky session id for the Edit Agent embedded in the workflow card
- # (Image #38, #48). Optional so older workflows don't fail validation
- # on rehydrate.
- edit_agent_session_id: Optional[str] = None
-
-
-class WorkflowRun(BaseModel):
- id: str = Field(default_factory=lambda: uuid4().hex)
- workflow_id: str
- status: Literal["running", "success", "failure", "ran_late", "skipped"] = "running"
- scheduled_for: Optional[datetime] = None
- started_at: datetime = Field(default_factory=datetime.now)
- finished_at: Optional[datetime] = None
- session_id: Optional[str] = None
- error: Optional[str] = None
- cost_usd: float = 0.0
- triggered_by: Literal["schedule", "manual", "retry"] = "schedule"
- # Last tool-call label observed on the underlying agent session while
- # the workflow is running. Surfaced under the active step in RunningView
- # (Image #40) so the user can tell the run is still making progress.
- last_tool_label: Optional[str] = None
- # Currently-executing step index (0-based). Executor bumps this each
- # time it dispatches a step prompt and broadcasts the run. RunningView
- # uses this for the disc statuses; estimate fallback only when null.
- active_step_idx: Optional[int] = None
-
-
-class WorkflowCreate(BaseModel):
- title: str = "Untitled workflow"
- description: str = ""
- icon: str = ""
- system_prompt: Optional[str] = None
- use_synced_prompt: bool = True
- steps: list[WorkflowStep] = Field(default_factory=list)
- actions: ActionsConfig = Field(default_factory=ActionsConfig)
- schedule: ScheduleConfig = Field(default_factory=ScheduleConfig)
- permissions: Optional[list[PermissionTier]] = None
- source_session_id: Optional[str] = None
- dashboard_id: Optional[str] = None
- model: Optional[str] = None
- mode: Optional[str] = None
- provider: Optional[str] = None
- cost_cap_usd_monthly: Optional[float] = None
-
-
-class WorkflowUpdate(BaseModel):
- title: Optional[str] = None
- description: Optional[str] = None
- icon: Optional[str] = None
- system_prompt: Optional[str] = None
- use_synced_prompt: Optional[bool] = None
- steps: Optional[list[WorkflowStep]] = None
- actions: Optional[ActionsConfig] = None
- schedule: Optional[ScheduleConfig] = None
- permissions: Optional[list[PermissionTier]] = None
- model: Optional[str] = None
- mode: Optional[str] = None
- provider: Optional[str] = None
- cost_cap_usd_monthly: Optional[float] = None
diff --git a/backend/apps/workflows/notifier.py b/backend/apps/workflows/notifier.py
deleted file mode 100644
index 20034872..00000000
--- a/backend/apps/workflows/notifier.py
+++ /dev/null
@@ -1,55 +0,0 @@
-"""Permission/escalation chain notifier.
-
-The notify tier broadcasts a ws event the renderer picks up. The text/call
-tiers route through the cloud SMS bridge once enabled; until it's enabled
-we fall back to an extra ws notify with a `fallback: true` marker so the
-renderer can label it honestly ("Text-me fallback: cloud SMS not wired").
-The *when* of escalation is owned by apps/workflows/escalation.py.
-"""
-
-import logging
-from datetime import datetime
-
-from backend.apps.workflows.models import PermissionTier, Workflow, WorkflowRun
-
-logger = logging.getLogger(__name__)
-
-
-def _base_payload(wf: Workflow, run: WorkflowRun) -> dict:
- return {
- "workflow_id": wf.id,
- "workflow_title": wf.title,
- "run_id": run.id,
- "status": run.status,
- "session_id": run.session_id,
- "started_at": run.started_at.isoformat() if isinstance(run.started_at, datetime) else run.started_at,
- "finished_at": run.finished_at.isoformat() if isinstance(run.finished_at, datetime) else run.finished_at,
- }
-
-
-async def notify_run_complete(wf: Workflow, run: WorkflowRun) -> None:
- from backend.apps.agents.core.ws_manager import ws_manager
- from backend.apps.workflows import escalation
-
- payload = _base_payload(wf, run)
- await ws_manager.broadcast_global("workflow:notify", payload)
-
- # Kick off server-side escalation only if there are additional tiers
- # beyond the default notify. The escalation runner will sleep + call
- # send_tier per tier.
- escalation.schedule(wf, run)
-
-
-async def send_tier(wf: Workflow, run: WorkflowRun, tier: PermissionTier) -> None:
- """Send a single escalation tier. Today the text/call paths fall back
- to an in-app notify with `fallback: true` and the tier kind set so the
- renderer can show "Text-me fallback (cloud SMS not wired)."
- """
- from backend.apps.agents.core.ws_manager import ws_manager
-
- payload = _base_payload(wf, run)
- payload["tier_kind"] = tier.kind
- payload["tier_phone"] = (tier.phone or "")[-4:] if tier.phone else None
- payload["fallback"] = True # flip to False once the cloud SMS bridge is wired
- await ws_manager.broadcast_global("workflow:notify", payload)
- logger.info("workflow tier=%s fallback fired wf=%s run=%s", tier.kind, wf.id, run.id)
diff --git a/backend/apps/workflows/scheduler.py b/backend/apps/workflows/scheduler.py
deleted file mode 100644
index d05a5be3..00000000
--- a/backend/apps/workflows/scheduler.py
+++ /dev/null
@@ -1,352 +0,0 @@
-"""In-process cron-style scheduler.
-
-One long-lived asyncio task wakes on the next-due workflow boundary, fires
-matching workflows, then re-computes. We deliberately avoid one-task-per-
-workflow (turns rescheduling into a thundering re-spawn problem). On
-startup we walk persisted workflows once, decide what to do about missed
-fires via on_missed, and queue each.
-
-Schedule semantics:
- unit=day: fires every repeat_every days at hour:minute
- unit=week: fires on the listed weekday(s) every repeat_every weeks
- unit=month: fires on the original day-of-month every repeat_every months
-
-Wall-clock math runs in the workflow's IANA timezone, then we convert to
-UTC at the boundary. This is the only safe way to honor DST (a "9am
-Monday" schedule must remain 9am local across spring-forward / fall-back).
-Legacy records with timezone="local" are coerced to the host zone in
-memory by storage._load_all_from_disk; the on-disk file is not rewritten
-until the user's next save.
-"""
-
-import asyncio
-import calendar
-import logging
-import os
-from datetime import datetime, timedelta, timezone
-from typing import Optional
-from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
-
-from backend.apps.workflows.models import Workflow, ScheduleConfig
-from backend.apps.workflows import storage, executor
-
-logger = logging.getLogger(__name__)
-
-
-_loop_task: Optional[asyncio.Task] = None
-_wake = asyncio.Event()
-_host_tz_cache: Optional[ZoneInfo] = None
-
-
-def _host_tz() -> ZoneInfo:
- global _host_tz_cache
- if _host_tz_cache is not None:
- return _host_tz_cache
- 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:
- _host_tz_cache = ZoneInfo(name) if name else ZoneInfo("UTC")
- except ZoneInfoNotFoundError:
- _host_tz_cache = ZoneInfo("UTC")
- return _host_tz_cache
-
-
-def _resolve_tz(tz: str) -> ZoneInfo:
- if not tz or tz == "local":
- return _host_tz()
- try:
- return ZoneInfo(tz)
- except ZoneInfoNotFoundError:
- return _host_tz()
-
-
-def _as_utc(dt: Optional[datetime]) -> Optional[datetime]:
- """Normalize an arbitrary stored datetime to aware-UTC.
-
- Pydantic deserializes naive ISO strings as naive datetimes. Treat such
- values as host-local (matches the pre-tz codepath that wrote them) so
- comparisons against datetime.now(timezone.utc) don't raise.
- """
- if dt is None:
- return None
- if dt.tzinfo is None:
- return dt.replace(tzinfo=_host_tz()).astimezone(timezone.utc)
- return dt.astimezone(timezone.utc)
-
-
-def _add_months(dt: datetime, months: int) -> datetime:
- """Add months preserving day-of-month, clamping only if the target month
- is shorter (e.g. Jan 31 + 1mo → Feb 28/29). Wall-clock arithmetic; the
- caller is responsible for tz attachment.
- """
- total = dt.month - 1 + months
- year = dt.year + total // 12
- month = total % 12 + 1
- day = min(dt.day, calendar.monthrange(year, month)[1])
- return dt.replace(year=year, month=month, day=day)
-
-
-def _js_weekday(d: datetime) -> int:
- """Frontend uses JS getDay() convention (Sun=0..Sat=6). Python's
- datetime.weekday() is Mon=0..Sun=6. Wire format stays JS-style so the
- on_days array round-trips between FE and BE without translation in two
- places."""
- return (d.weekday() + 1) % 7
-
-
-def _next_fire_after(sched: ScheduleConfig, ref_utc: datetime) -> Optional[datetime]:
- if not sched.enabled:
- return None
- tz = _resolve_tz(sched.timezone)
- ref_local = ref_utc.astimezone(tz)
- base = ref_local.replace(second=0, microsecond=0)
- candidate = base.replace(hour=sched.hour, minute=sched.minute)
- if candidate <= ref_local:
- candidate = candidate + timedelta(days=1)
-
- if sched.repeat_unit == "day":
- step = max(1, sched.repeat_every)
- # Walk forward in step-day increments until we find a slot strictly
- # after `ref_local`. Cheap because step is small.
- while candidate <= ref_local:
- candidate = candidate + timedelta(days=step)
- return candidate.astimezone(timezone.utc)
-
- if sched.repeat_unit == "week":
- allowed = sched.on_days or [_js_weekday(ref_local)]
- for _ in range(0, 14):
- if _js_weekday(candidate) in allowed and candidate > ref_local:
- return candidate.astimezone(timezone.utc)
- candidate = candidate + timedelta(days=1)
- return candidate.astimezone(timezone.utc)
-
- if sched.repeat_unit == "month":
- target_day = ref_local.day
- step = max(1, sched.repeat_every)
- c = candidate.replace(day=min(target_day, calendar.monthrange(candidate.year, candidate.month)[1]))
- while c <= ref_local:
- c = _add_months(c, step)
- return c.astimezone(timezone.utc)
-
- return None
-
-
-def compute_next_fire(wf: Workflow, ref: Optional[datetime] = None) -> Optional[datetime]:
- ref_utc = _as_utc(ref) if ref is not None else datetime.now(timezone.utc)
- return _next_fire_after(wf.schedule, ref_utc)
-
-
-def fires_in_window(wf: Workflow, days: int = 30) -> int:
- """Count fires from now through `days` days from now. Used by the
- cost-estimate response. Honors end conditions so the projection doesn't
- over-count after ends_at or max_runs. Caps the walk at 1000 fires to
- guard pathological sub-day schedules (none today, but cheap insurance).
- """
- sched = wf.schedule
- if not sched.enabled:
- return 0
- if sched.max_runs is not None and sched.runs_count >= sched.max_runs:
- return 0
- cursor_utc = datetime.now(timezone.utc)
- end_utc = cursor_utc + timedelta(days=days)
- ends_at_utc = _as_utc(sched.ends_at)
- if ends_at_utc is not None and ends_at_utc < end_utc:
- end_utc = ends_at_utc
- remaining_budget = (
- sched.max_runs - sched.runs_count if sched.max_runs is not None else 1000
- )
- count = 0
- while count < min(1000, remaining_budget):
- nxt = _next_fire_after(sched, cursor_utc)
- if nxt is None or nxt > end_utc:
- break
- count += 1
- cursor_utc = nxt
- return count
-
-
-def kick() -> None:
- _wake.set()
-
-
-def _end_condition_hit(wf: Workflow, now_utc: datetime) -> bool:
- s = wf.schedule
- ends_at = _as_utc(s.ends_at)
- if ends_at is not None and now_utc >= ends_at:
- return True
- if s.max_runs is not None and s.runs_count >= s.max_runs:
- return True
- return False
-
-
-def _disable_schedule(wf: Workflow) -> None:
- wf.schedule.enabled = False
- wf.next_run_at = None
- storage.save_workflow(wf)
-
-
-async def _tick() -> None:
- now_utc = datetime.now(timezone.utc)
- if storage.get_paused():
- return
- due: list[Workflow] = []
- for wf in storage.list_workflows():
- if not wf.schedule.enabled:
- continue
- if _end_condition_hit(wf, now_utc):
- _disable_schedule(wf)
- continue
- nra = _as_utc(wf.next_run_at)
- if nra and nra <= now_utc:
- due.append(wf)
-
- for wf in due:
- scheduled_for = _as_utc(wf.next_run_at)
- nxt = _next_fire_after(wf.schedule, now_utc)
- wf.next_run_at = nxt
- storage.save_workflow(wf)
- asyncio.create_task(_fire(wf, scheduled_for=scheduled_for))
-
-
-async def _fire(wf: Workflow, scheduled_for: Optional[datetime]) -> None:
- try:
- await executor.execute(wf, triggered_by="schedule", scheduled_for=scheduled_for)
- except Exception:
- logger.exception("scheduler fire failed for workflow=%s", wf.id)
-
-
-def _seconds_until_next() -> float:
- now_utc = datetime.now(timezone.utc)
- soonest: Optional[datetime] = None
- for wf in storage.list_workflows():
- if not wf.schedule.enabled:
- continue
- nra = _as_utc(wf.next_run_at)
- if nra is None:
- continue
- if soonest is None or nra < soonest:
- soonest = nra
- if soonest is None:
- return 60.0
- delta = (soonest - now_utc).total_seconds()
- return max(1.0, min(delta, 60.0))
-
-
-async def _loop() -> None:
- logger.info("workflow scheduler loop started")
- while True:
- try:
- await _tick()
- except Exception:
- logger.exception("scheduler tick error")
- try:
- await asyncio.wait_for(_wake.wait(), timeout=_seconds_until_next())
- except asyncio.TimeoutError:
- pass
- _wake.clear()
-
-
-def _mark_stuck_runs_failed() -> None:
- """Any run marked 'running' that survives a backend restart is dead.
-
- The owning event loop is gone, so there's no way to resume. Mark it
- failed once at startup instead of letting the History tab show a
- forever-spinning row that misleads the user.
- """
- now = datetime.now()
- for wf in storage.list_workflows():
- for r in storage.list_runs(wf.id, limit=200):
- if r.status == "running":
- storage.update_run(
- r.id,
- status="failure",
- error="OpenSwarm closed before this run finished.",
- finished_at=now,
- )
-
-
-def reconcile_on_startup() -> None:
- """Walk persisted workflows once and resolve missed fires per policy.
-
- Missed-run policies:
- skip -> roll forward to next future fire, ignore missed
- run_once -> if any fires were missed, schedule a single catch-up at now
- run_all -> not actually run_all in v1 (would burn tokens); same as run_once
- but we mark the run.status as ran_late so the UI surfaces it
- """
- now_utc = datetime.now(timezone.utc)
- for wf in storage.list_workflows():
- if not wf.schedule.enabled:
- wf.next_run_at = None
- storage.save_workflow(wf)
- continue
-
- if _end_condition_hit(wf, now_utc):
- _disable_schedule(wf)
- continue
-
- nra = _as_utc(wf.next_run_at)
- missed = bool(nra and nra <= now_utc)
- if missed and wf.schedule.on_missed in ("run_once", "run_all"):
- # Keep next_run_at <= now_utc so the very next tick fires it.
- # Normalize to a UTC-aware value so future comparisons don't
- # trip on naive legacy datetimes.
- wf.next_run_at = nra
- storage.save_workflow(wf)
- else:
- wf.next_run_at = _next_fire_after(wf.schedule, now_utc)
- storage.save_workflow(wf)
-
-
-async def start() -> None:
- global _loop_task
- if _loop_task is not None:
- return
- _mark_stuck_runs_failed()
- reconcile_on_startup()
- _loop_task = asyncio.create_task(_loop())
-
-
-async def stop() -> None:
- global _loop_task
- if _loop_task is None:
- return
- _loop_task.cancel()
- try:
- await _loop_task
- except (asyncio.CancelledError, Exception):
- pass
- _loop_task = None
-
-
-def list_active() -> list[dict]:
- """Snapshot of currently-running workflow runs.
-
- Reads executor._running (workflow_id -> run_id) and joins against the
- workflow cache for titles. Used by GET /workflows/active so the tray
- and the auto-updater veto can both ask "are any runs in flight?"
- without holding the executor lock.
- """
- out: list[dict] = []
- snapshot = dict(executor._running)
- for wid, run_id in snapshot.items():
- wf = storage.get_workflow(wid)
- title = wf.title if wf else ""
- started_at = None
- if wf:
- for r in storage.list_runs(wid, limit=10):
- if r.id == run_id:
- started_at = r.started_at.isoformat() if isinstance(r.started_at, datetime) else r.started_at
- break
- out.append({
- "workflow_id": wid,
- "run_id": run_id,
- "title": title,
- "started_at": started_at,
- })
- return out
diff --git a/backend/apps/workflows/storage.py b/backend/apps/workflows/storage.py
deleted file mode 100644
index 86d80352..00000000
--- a/backend/apps/workflows/storage.py
+++ /dev/null
@@ -1,197 +0,0 @@
-"""On-disk store for workflows + workflow runs.
-
-Layout under DATA_ROOT/workflows/:
- .json workflow record
- runs/.json bounded log (latest N) of runs for that workflow
-
-A separate runs file per workflow keeps history reads O(history size) instead
-of O(total runs across all workflows). The workflow record only carries
-last_run_* / next_run_at summary fields; full history lives in the runs file.
-"""
-
-import json
-import os
-from threading import Lock
-from typing import Optional
-
-from backend.config.paths import DATA_ROOT
-from backend.apps.workflows.models import Workflow, WorkflowRun
-
-DATA_DIR = os.path.join(DATA_ROOT, "workflows")
-RUNS_DIR = os.path.join(DATA_DIR, "runs")
-PAUSED_FILE = os.path.join(DATA_DIR, "paused.json")
-
-_io_lock = Lock()
-_workflow_cache: dict[str, Workflow] = {}
-_runs_cache: dict[str, list[WorkflowRun]] = {}
-_cache_loaded = False
-_paused = False
-
-
-def _resolve_host_tz_name() -> str:
- """Best-effort IANA name for the host. Mirrors apps/service/client.py."""
- 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 = ""
- return name or "UTC"
-
-# Keep this much run history per workflow on disk. Older runs are pruned;
-# the History tab caps at ~20 anyway, and unbounded growth turned the JSON
-# read into a real cost on hot-reload of the schedule page.
-RUNS_PER_WORKFLOW = 200
-
-
-def _ensure_dirs() -> None:
- os.makedirs(DATA_DIR, exist_ok=True)
- os.makedirs(RUNS_DIR, exist_ok=True)
-
-
-def _wf_path(wid: str) -> str:
- return os.path.join(DATA_DIR, f"{wid}.json")
-
-
-def _runs_path(wid: str) -> str:
- return os.path.join(RUNS_DIR, f"{wid}.json")
-
-
-def _load_all_from_disk() -> None:
- global _cache_loaded, _paused
- _ensure_dirs()
- _workflow_cache.clear()
- _runs_cache.clear()
- host_tz = _resolve_host_tz_name()
- for fname in os.listdir(DATA_DIR):
- if not fname.endswith(".json") or fname == "paused.json":
- continue
- try:
- with open(os.path.join(DATA_DIR, fname)) as f:
- wf = Workflow(**json.load(f))
- # Coerce legacy timezone="local" to the host IANA zone in
- # memory only. We don't rewrite the file here so backup/sync
- # tooling doesn't see mtime churn on every startup; the next
- # user-driven save migrates the on-disk record naturally.
- if wf.schedule.timezone == "local":
- wf.schedule.timezone = host_tz
- _workflow_cache[wf.id] = wf
- except Exception:
- continue
- if os.path.exists(RUNS_DIR):
- for fname in os.listdir(RUNS_DIR):
- if not fname.endswith(".json"):
- continue
- wid = fname[:-5]
- try:
- with open(os.path.join(RUNS_DIR, fname)) as f:
- arr = json.load(f)
- _runs_cache[wid] = [WorkflowRun(**r) for r in arr]
- except Exception:
- _runs_cache[wid] = []
- # Load the global pause flag if it's been set previously.
- if os.path.exists(PAUSED_FILE):
- try:
- with open(PAUSED_FILE) as f:
- _paused = bool(json.load(f).get("paused", False))
- except Exception:
- _paused = False
- _cache_loaded = True
-
-
-def init() -> None:
- with _io_lock:
- _load_all_from_disk()
-
-
-def list_workflows() -> list[Workflow]:
- if not _cache_loaded:
- init()
- return list(_workflow_cache.values())
-
-
-def get_workflow(wid: str) -> Optional[Workflow]:
- if not _cache_loaded:
- init()
- return _workflow_cache.get(wid)
-
-
-def save_workflow(wf: Workflow) -> Workflow:
- with _io_lock:
- _ensure_dirs()
- _workflow_cache[wf.id] = wf
- with open(_wf_path(wf.id), "w") as f:
- json.dump(wf.model_dump(mode="json"), f, indent=2)
- return wf
-
-
-def delete_workflow(wid: str) -> bool:
- with _io_lock:
- existed = wid in _workflow_cache
- _workflow_cache.pop(wid, None)
- _runs_cache.pop(wid, None)
- wf_file = _wf_path(wid)
- if os.path.exists(wf_file):
- os.remove(wf_file)
- rf = _runs_path(wid)
- if os.path.exists(rf):
- os.remove(rf)
- return existed
-
-
-def list_runs(wid: str, limit: int = 50) -> list[WorkflowRun]:
- if not _cache_loaded:
- init()
- runs = _runs_cache.get(wid, [])
- return runs[-limit:][::-1]
-
-
-def record_run(run: WorkflowRun) -> WorkflowRun:
- with _io_lock:
- _ensure_dirs()
- arr = _runs_cache.setdefault(run.workflow_id, [])
- # Replace prior entry with same id if we're updating an in-flight run.
- for i, prior in enumerate(arr):
- if prior.id == run.id:
- arr[i] = run
- break
- else:
- arr.append(run)
- # Bound the per-workflow history to keep disk + memory cheap.
- if len(arr) > RUNS_PER_WORKFLOW:
- del arr[: len(arr) - RUNS_PER_WORKFLOW]
- with open(_runs_path(run.workflow_id), "w") as f:
- json.dump([r.model_dump(mode="json") for r in arr], f, indent=2)
- return run
-
-
-def get_paused() -> bool:
- if not _cache_loaded:
- init()
- return _paused
-
-
-def set_paused(value: bool) -> bool:
- global _paused
- with _io_lock:
- _ensure_dirs()
- _paused = bool(value)
- with open(PAUSED_FILE, "w") as f:
- json.dump({"paused": _paused}, f)
- return _paused
-
-
-def update_run(run_id: str, **fields) -> Optional[WorkflowRun]:
- if not _cache_loaded:
- init()
- for arr in _runs_cache.values():
- for i, r in enumerate(arr):
- if r.id == run_id:
- updated = r.model_copy(update=fields)
- arr[i] = updated
- with _io_lock:
- with open(_runs_path(updated.workflow_id), "w") as f:
- json.dump([x.model_dump(mode="json") for x in arr], f, indent=2)
- return updated
- return None
diff --git a/backend/apps/workflows/workflows.py b/backend/apps/workflows/workflows.py
deleted file mode 100644
index cb1b4634..00000000
--- a/backend/apps/workflows/workflows.py
+++ /dev/null
@@ -1,807 +0,0 @@
-import asyncio
-import logging
-from contextlib import asynccontextmanager
-from datetime import datetime
-from typing import Optional
-
-from fastapi import HTTPException, Header, Request
-
-from backend.config.Apps import SubApp
-from backend.apps.workflows.models import (
- Workflow,
- WorkflowCreate,
- WorkflowUpdate,
- WorkflowRun,
-)
-from backend.apps.workflows import storage, scheduler, executor, audit, escalation
-
-logger = logging.getLogger(__name__)
-
-
-def _scan_cron_for_openswarm() -> list[str]:
- """Surface OS-level scheduled-task entries that reference us.
-
- macOS + Linux: read `crontab -l`. Windows: query `schtasks` for any
- task whose command/path contains 'openswarm'. Best-effort across all
- three; any failure (no tool installed, permission denied, parse
- error) just returns []. Surfaced to the FE so the Workflows hub can
- offer a one-click migration banner to convert into native workflows.
- """
- import subprocess
- import platform as _platform
- findings: list[str] = []
- if _platform.system() == "Windows":
- try:
- proc = subprocess.run(
- ["schtasks", "/query", "/fo", "CSV", "/v"],
- capture_output=True, text=True, timeout=4,
- )
- if proc.returncode != 0:
- return []
- for line in (proc.stdout or "").splitlines():
- if "openswarm" in line.lower() and not line.lstrip().startswith('"#'):
- findings.append(line.strip())
- except Exception:
- return []
- return findings
- # macOS + Linux
- try:
- proc = subprocess.run(
- ["crontab", "-l"],
- capture_output=True, text=True, timeout=2,
- )
- if proc.returncode != 0:
- return []
- out = proc.stdout or ""
- return [line.strip() for line in out.splitlines() if "openswarm" in line.lower() and not line.strip().startswith("#")]
- except Exception:
- return []
-
-
-_cron_findings: list[str] = []
-
-
-@asynccontextmanager
-async def workflows_lifespan():
- storage.init()
- await scheduler.start()
- # Cheap one-shot scan for prior cron entries that reference us. We
- # don't migrate automatically; the FE shows a banner with a "Convert
- # to OpenSwarm scheduled tasks" button so the user is in control.
- global _cron_findings
- _cron_findings = _scan_cron_for_openswarm()
- try:
- yield
- finally:
- await scheduler.stop()
-
-
-workflows = SubApp("workflows", workflows_lifespan)
-
-
-def _derive_icon(wf: Workflow) -> str:
- """Cheap icon hint used until proper auto-icon generation lands.
-
- Pull the first emoji from the title, falling back to the first
- letter. Keeps the Search list (image 2 annotation) populated without
- waiting on the LLM-based icon generator.
- """
- title = (wf.title or "").strip()
- for ch in title:
- if ord(ch) > 0x2700:
- return ch
- if title:
- return title[:1].upper()
- return "W"
-
-
-@workflows.router.get("/list")
-async def list_workflows(dashboard_id: Optional[str] = None):
- items = storage.list_workflows()
- if dashboard_id:
- items = [w for w in items if not w.dashboard_id or w.dashboard_id == dashboard_id]
- items.sort(key=lambda w: w.updated_at or w.created_at, reverse=True)
- # Enrich with cost_estimate so calendar tooltips and the WorkflowsHub
- # list don't have to round-trip to GET /workflows/{id} per row. Cheap
- # because fires_in_window walks at most ~30 fires per workflow.
- return {"workflows": [_enriched(w) for w in items]}
-
-
-@workflows.router.post("/create")
-async def create_workflow(body: WorkflowCreate):
- 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.
- # Source-session creates inherit the chat's tool choices so we leave
- # them alone there (the source session itself already vetted the
- # blast radius).
- if body.schedule.enabled and not actions.freeze and not body.source_session_id:
- actions = actions.model_copy(update={"freeze": True})
- wf = Workflow(
- title=body.title,
- description=body.description,
- icon=body.icon,
- system_prompt=body.system_prompt,
- use_synced_prompt=body.use_synced_prompt,
- steps=body.steps,
- actions=actions,
- schedule=body.schedule,
- permissions=body.permissions or [],
- source_session_id=body.source_session_id,
- dashboard_id=body.dashboard_id,
- model=body.model or "sonnet",
- mode=body.mode or "agent",
- provider=body.provider or "anthropic",
- cost_cap_usd_monthly=body.cost_cap_usd_monthly,
- )
- if not wf.icon:
- wf.icon = _derive_icon(wf)
- if wf.schedule.enabled:
- wf.next_run_at = scheduler.compute_next_fire(wf)
- # Force-generate title + description + per-step labels from the steps
- # in a single aux call. Previously we only filled missing description,
- # leaving stale session names ("Inbox check") as titles. Step labels
- # are the 3-6 word at-a-glance headlines surfaced in StepList; without
- # them the UI falls back to truncated raw prompts.
- try:
- title, description, labels = await _generate_workflow_metadata(wf)
- if title:
- wf.title = title
- if description:
- wf.description = description
- if labels and len(labels) == len(wf.steps):
- for i, lab in enumerate(labels):
- if lab:
- wf.steps[i].label = lab
- except Exception:
- pass
- storage.save_workflow(wf)
- scheduler.kick()
- return _enriched(wf)
-
-
-async def _generate_workflow_metadata(wf: Workflow) -> tuple[str, str, list[str]]:
- """Single aux-model call returning (title, description, step_labels).
-
- One round-trip for all three so we don't burn 3x aux cost. Returns
- ("", "", []) on any failure; caller writes back unconditionally.
- """
- if not wf.steps:
- return "", "", []
- try:
- from backend.apps.agents.providers.registry import resolve_aux_model
- from backend.apps.settings.credentials import get_anthropic_client_for_model
- from backend.apps.settings.settings import load_settings as _ls
- except Exception:
- return "", "", []
- settings = _ls()
- try:
- aux_model, _ = await resolve_aux_model(settings, preferred_tier="haiku")
- client = get_anthropic_client_for_model(settings, aux_model)
- except Exception:
- return "", "", []
- steps_lines = "\n".join(f"{i+1}. {s.text}" for i, s in enumerate(wf.steps) if s.text)
- n_steps = len(wf.steps)
- prompt = (
- "You name and describe a saved automation routine that the user "
- "can re-run later, AND produce a short at-a-glance label for "
- "each step. The routine is defined ONLY by the numbered steps "
- "below; treat those as the user's instructions to the agent.\n\n"
- "Return STRICT JSON, nothing else, no code fence:\n"
- ' {"title": string, "description": string, "step_labels": [string, ...]}\n\n'
- "title rules:\n"
- "- 2 to 5 words, Title Case\n"
- "- Starts with a verb-noun pair when possible (e.g. \"Summarize "
- "Daily Emails\")\n"
- "- No emoji, no quotes, no trailing punctuation\n\n"
- "description rules:\n"
- "- 1 to 2 sentences, under 30 words total\n"
- "- Describes the concrete WORK the routine performs for the user, "
- "not metadata about itself. Examples of GOOD output:\n"
- " \"Reads recent Gmail, ranks urgency, and emails you a PDF "
- "digest each Sunday at 9am.\"\n"
- " \"Pulls today's calendar plus inbox, writes a Notion brief, "
- "and texts you the link.\"\n"
- "- Start with a verb. Do NOT start with \"This\", \"A\", \"An\", "
- "\"The workflow\", \"This routine\".\n\n"
- f"step_labels rules:\n"
- f"- EXACTLY {n_steps} entries, one per step, same order.\n"
- "- Each label: 3 to 6 words, Sentence case.\n"
- "- Imperative verb-led (\"Summarize emails & calendar\", \"Make "
- "brief in notion\", \"Email brief link to me\").\n"
- "- No trailing punctuation, no quotes, no emoji.\n"
- "- Should read as the human-friendly NAME of the step, NOT a "
- "restatement of the prompt.\n\n"
- f"Steps:\n{steps_lines}"
- )
- import json
- import re as _re
-
- def _extract_json_object(s: str) -> Optional[dict]:
- s = s.strip()
- if s.startswith("```"):
- s = _re.sub(r"^```(?:json)?\s*", "", s, flags=_re.IGNORECASE)
- s = _re.sub(r"\s*```\s*$", "", s)
- start = s.find("{")
- end = s.rfind("}")
- if start != -1 and end != -1 and end > start:
- s = s[start : end + 1]
- try:
- return json.loads(s)
- except Exception:
- return None
-
- try:
- resp = await client.messages.create(
- model=aux_model,
- max_tokens=400 + n_steps * 30,
- messages=[
- {"role": "user", "content": prompt},
- {"role": "assistant", "content": "{"},
- ],
- )
- text = ""
- if isinstance(resp.content, list):
- for block in resp.content:
- if getattr(block, "type", None) == "text":
- text += getattr(block, "text", "")
- raw = "{" + text.strip() if not text.strip().startswith("{") else text.strip()
- data = _extract_json_object(raw)
- if not data:
- logger.warning("workflow meta gen: failed to parse aux model output: %s", raw[:400])
- return "", "", []
- title = (data.get("title") or "").strip()[:80]
- description = (data.get("description") or "").strip()[:500]
- raw_labels = data.get("step_labels") or []
- labels = [str(x or "").strip()[:60] for x in raw_labels] if isinstance(raw_labels, list) else []
- return title, description, labels
- except Exception as e:
- logger.warning("workflow meta gen: aux model call failed: %s", e)
- return "", "", []
-
-
-def _last_run_cost(wid: str) -> float:
- for r in storage.list_runs(wid, limit=10):
- if r.status in ("success", "ran_late") and r.cost_usd:
- return float(r.cost_usd)
- return 0.0
-
-
-def _enriched(wf: Workflow) -> dict:
- """Serialize a workflow with a cost_estimate block attached.
-
- monthly_usd assumes future fires cost the same as the last successful
- fire. Surfaces honestly as "at last run's cost" in the UI so users
- understand it's a projection, not a quota.
- """
- base = wf.model_dump(mode="json")
- last = _last_run_cost(wf.id)
- fires = scheduler.fires_in_window(wf, days=30)
- base["cost_estimate"] = {
- "monthly_usd": round(last * fires, 4),
- "last_run_usd": round(last, 4),
- "fires_per_month": fires,
- }
- return base
-
-
-@workflows.router.get("/active")
-async def list_active_runs():
- """Snapshot of currently-running workflow runs. Used by the tray and
- the auto-updater veto."""
- return {"active": scheduler.list_active()}
-
-
-@workflows.router.post("/pause-all")
-async def pause_all_schedules():
- storage.set_paused(True)
- scheduler.kick()
- return {"paused": True}
-
-
-@workflows.router.post("/resume-all")
-async def resume_all_schedules():
- storage.set_paused(False)
- scheduler.kick()
- return {"paused": False}
-
-
-@workflows.router.get("/paused")
-async def get_paused_state():
- return {"paused": storage.get_paused()}
-
-
-@workflows.router.get("/cron/findings")
-async def cron_findings():
- """Cron entries we found at startup that reference OpenSwarm. The
- FE renders a one-time banner inviting users to convert them; we
- return the raw lines so the user can verify before migrating."""
- return {"entries": list(_cron_findings)}
-
-
-@workflows.router.get("/cloud/sms/status")
-async def cloud_sms_status():
- """Probe used by the FE to decide whether to show the 'falls back to
- in-app notify' acknowledgement on the text/call tiers. Returns
- enabled=False until the cloud SMS bridge ships."""
- return {"enabled": False}
-
-
-@workflows.router.post("/runs/{run_id}/ack")
-async def ack_run(run_id: str):
- cancelled = escalation.cancel(run_id)
- return {"acked": True, "had_pending_escalation": cancelled}
-
-
-@workflows.router.get("/runs/{run_id}/escalation")
-async def get_run_escalation(run_id: str):
- state = escalation.status(run_id)
- return {"state": state}
-
-
-@workflows.router.get("/{workflow_id}")
-async def get_workflow(workflow_id: str):
- wf = storage.get_workflow(workflow_id)
- if not wf:
- raise HTTPException(status_code=404, detail="Workflow not found")
- return _enriched(wf)
-
-
-@workflows.router.get("/{workflow_id}/audit")
-async def get_workflow_audit(workflow_id: str, limit: int = 50):
- wf = storage.get_workflow(workflow_id)
- if not wf:
- raise HTTPException(status_code=404, detail="Workflow not found")
- return {"entries": audit.read_tail(workflow_id, limit=limit)}
-
-
-@workflows.router.patch("/{workflow_id}")
-async def update_workflow(
- workflow_id: str,
- body: WorkflowUpdate,
- if_match: Optional[str] = Header(default=None, alias="If-Match"),
-):
- wf = storage.get_workflow(workflow_id)
- if not wf:
- raise HTTPException(status_code=404, detail="Workflow not found")
- # Optimistic concurrency: if the client passed If-Match, verify it
- # matches the current updated_at. Stale writes (another window or a
- # mid-edit background fire) get a 409 so the FE can prompt to reload
- # instead of silently clobbering the other actor's changes. Missing
- # header = legacy client, allow through (back-compat with the
- # frontend's pre-409 code path; FE rolls out If-Match immediately).
- if if_match:
- current_stamp = wf.updated_at.isoformat() if hasattr(wf.updated_at, "isoformat") else str(wf.updated_at)
- # Strip quotes a well-behaved HTTP client might add per RFC 7232.
- if if_match.strip().strip('"') != current_stamp:
- raise HTTPException(
- status_code=409,
- detail={
- "error": "stale_update",
- "message": "This workflow changed in another window or by a recent run. Reload and try again.",
- "current_updated_at": current_stamp,
- },
- )
- before = wf.model_dump(mode="json")
- data = body.model_dump(exclude_unset=True)
- for k, v in data.items():
- setattr(wf, k, v)
- wf.updated_at = datetime.now()
- if not wf.icon:
- wf.icon = _derive_icon(wf)
- wf.next_run_at = scheduler.compute_next_fire(wf) if wf.schedule.enabled else None
- storage.save_workflow(wf)
- audit.log_change(wf.id, "user", before, wf.model_dump(mode="json"))
- scheduler.kick()
- return _enriched(wf)
-
-
-@workflows.router.delete("/{workflow_id}")
-async def delete_workflow(workflow_id: str):
- existed = storage.delete_workflow(workflow_id)
- if not existed:
- raise HTTPException(status_code=404, detail="Workflow not found")
- scheduler.kick()
- return {"ok": True}
-
-
-@workflows.router.post("/{workflow_id}/propose-edit")
-async def propose_edit(workflow_id: str, body: dict):
- """Aux-LLM-propose a single-step edit from a natural-language request.
-
- Powers the Edit Agent chat (Image #38). Frontend hands us the user's
- message, the current draft steps, optional failure-context (Fix-with-
- Agent), AND the prior turns so the model has multi-turn memory. We
- respond with a reply string PLUS, optionally, a `step_idx` + `new_text`
- that the FE shows as a proposal card.
- """
- wf = storage.get_workflow(workflow_id)
- if not wf:
- raise HTTPException(status_code=404, detail="Workflow not found")
- message = (body or {}).get("message", "").strip()
- steps_in = (body or {}).get("steps") or []
- context = (body or {}).get("context") or None
- history = (body or {}).get("history") or []
- if not message or not isinstance(steps_in, list):
- raise HTTPException(status_code=400, detail="Missing message or steps")
- try:
- from backend.apps.agents.providers.registry import resolve_aux_model
- from backend.apps.settings.credentials import get_anthropic_client_for_model
- from backend.apps.settings.settings import load_settings as _ls
- except Exception:
- raise HTTPException(status_code=500, detail="Aux model unavailable")
- settings = _ls()
- try:
- aux_model, _ = await resolve_aux_model(settings, preferred_tier="haiku")
- client = get_anthropic_client_for_model(settings, aux_model)
- except Exception:
- raise HTTPException(status_code=500, detail="Aux model unavailable")
- import json, re
- steps_lines = "\n".join(
- f"{i+1}. {(s.get('label') or '').strip() or (s.get('text') or '')[:60]}: {(s.get('text') or '')}"
- for i, s in enumerate(steps_in)
- )
- fix_context = ""
- if context and isinstance(context, dict):
- fs = context.get("failed_step")
- err = context.get("error")
- if fs is not None and err:
- fix_context = (
- f"\n\nFAILURE CONTEXT: Step {int(fs) + 1} failed on the most recent run. "
- f"The error was: {err}\n"
- f"Your proposed edit should specifically address that failure if possible."
- )
- # Build history block so the model remembers prior turns. Each entry
- # is {role, text}; we only carry assistant/user pairs (proposals get
- # summarised inline so the assistant has context for follow-ups).
- history_lines = []
- if isinstance(history, list):
- for h in history[-12:]:
- if not isinstance(h, dict):
- continue
- role = str(h.get("role") or "").strip().lower()
- text = str(h.get("text") or "").strip()
- if role in ("user", "assistant") and text:
- history_lines.append(f"{role.capitalize()}: {text}")
- history_block = ("\n\nPrior conversation:\n" + "\n".join(history_lines)) if history_lines else ""
-
- prompt = (
- "You are an Edit Agent helping the user iterate on a saved automation "
- "workflow. The workflow's current steps are listed below. The user has "
- "asked for a modification.\n\n"
- "Respond with STRICT JSON, no prose, no fence. Schema:\n"
- ' {"reply": string, '
- '"step_idx": int | null, '
- '"new_text": string | null, '
- '"explanation": string | null}\n\n'
- "Rules:\n"
- "- `reply` is a short conversational acknowledgement (1-2 sentences).\n"
- "- If the user is asking a question or for clarification, set step_idx=null and new_text=null.\n"
- "- If the user is asking to change a specific step, set step_idx (0-based) and new_text to the FULL replacement prompt for that step.\n"
- "- `explanation` describes the change in user-facing terms.\n"
- "- Never invent new steps. Never remove steps. Only edit existing ones.\n"
- "- Use prior conversation context to disambiguate follow-ups (e.g. \"yes do that\" should reference the last proposal).\n\n"
- f"Workflow steps:\n{steps_lines}{fix_context}{history_block}\n\n"
- f"User: {message}"
- )
- try:
- resp = await client.messages.create(
- model=aux_model,
- max_tokens=400,
- messages=[
- {"role": "user", "content": prompt},
- {"role": "assistant", "content": "{"},
- ],
- )
- out = ""
- if isinstance(resp.content, list):
- for block in resp.content:
- if getattr(block, "type", None) == "text":
- out += getattr(block, "text", "")
- raw = "{" + out.strip() if not out.strip().startswith("{") else out.strip()
- m = re.search(r"\{.*\}", raw, flags=re.DOTALL)
- if m:
- raw = m.group(0)
- data = json.loads(raw)
- except Exception as e:
- logger.warning("propose-edit: aux LLM failed: %s", e)
- raise HTTPException(status_code=400, detail="Couldn't generate a proposal")
- reply = str(data.get("reply") or "").strip()[:600]
- step_idx = data.get("step_idx")
- new_text = data.get("new_text")
- explanation = str(data.get("explanation") or "").strip()[:600]
- out: dict = {"reply": reply}
- if isinstance(step_idx, int) and 0 <= step_idx < len(steps_in) and isinstance(new_text, str) and new_text.strip():
- out["step_idx"] = step_idx
- out["new_text"] = new_text.strip()
- if explanation:
- out["explanation"] = explanation
- return out
-
-
-@workflows.router.post("/{workflow_id}/edit-agent-session")
-async def edit_agent_session(workflow_id: str):
- """Create (or return existing) Edit Agent session for this workflow.
-
- The Edit Agent is a real agent session that the user chats with to
- iterate on the workflow (Image #38, #48). It has the workflow context
- pre-loaded in its system prompt and the full default tool surface so
- tool calls render as cards in the chat (Image #48: MCP Activation,
- Gmail Query, etc.).
-
- Singleton per workflow: re-entering edit mode reattaches to the same
- session so the conversation persists. Frontend stores the returned
- session_id in the workflow card's openCard state.
- """
- wf = storage.get_workflow(workflow_id)
- if not wf:
- raise HTTPException(status_code=404, detail="Workflow not found")
- # Track the edit-agent session id on the workflow record so the FE
- # can find it after a reload. Persisted under a private namespace
- # field added below; we attach it lazily so existing workflows don't
- # need a migration.
- existing_id = getattr(wf, "edit_agent_session_id", None) or None
- if existing_id:
- from backend.apps.agents.agent_manager import agent_manager
- if existing_id in agent_manager.sessions:
- return {"session_id": existing_id}
- # In-memory miss but on disk it's still valid; fall through to
- # rehydrate via launch_agent OR return id for the FE to fetch.
- return {"session_id": existing_id}
-
- from backend.apps.agents.core.models import AgentConfig
- from backend.apps.agents.agent_manager import agent_manager
- steps_lines = "\n".join(f"{i+1}. {(s.label or '').strip() or (s.text or '')[:60]}\n Prompt: {s.text}" for i, s in enumerate(wf.steps))
- system_prompt = (
- f"You are the Edit Agent for the user's saved workflow \"{wf.title}\" "
- f"(id: {wf.id}). Help the user iterate on it. The workflow's purpose: "
- f"{wf.description or '(unspecified)'}.\n\n"
- f"Current steps:\n{steps_lines}\n\n"
- "How to work:\n"
- "1. When the user describes a change, briefly confirm what you'll do.\n"
- "2. If you need to look at files / search / activate an MCP / etc. to "
- "verify your idea, use your tools.\n"
- "3. Call EditWorkflowStep(workflow_id, step_idx, new_text) to apply a "
- "prompt change to a specific step. The change persists immediately. "
- "Confirm with the user via AskUserQuestion FIRST if there's any "
- "ambiguity about what they want.\n"
- "4. Call TestWorkflow(workflow_id) to spawn a sibling Test Agent that "
- "runs the latest version end-to-end. Use this after a change to verify "
- "it works.\n\n"
- "Be brief in your replies. Don't restate the whole workflow back; the "
- "user can see it. Just confirm what changed and what you're doing."
- )
- config = AgentConfig(
- name=f"Edit Agent: {wf.title}",
- model=wf.model or "sonnet",
- mode=wf.mode or "agent",
- provider=wf.provider or "anthropic",
- system_prompt=system_prompt,
- allowed_tools=[],
- dashboard_id=wf.dashboard_id,
- )
- session = await agent_manager.launch_agent(config)
- try:
- setattr(wf, "edit_agent_session_id", session.id)
- storage.save_workflow(wf)
- except Exception:
- logger.debug("could not persist edit_agent_session_id (legacy schema)", exc_info=True)
- return {"session_id": session.id}
-
-
-@workflows.router.post("/{workflow_id}/test-run")
-async def test_run_workflow(workflow_id: str, body: dict):
- """Spawn a Test Agent session running the (possibly-unsaved) draft.
-
- Powers Image #39: EditAgentView's Test button. Takes an optional
- draft `steps` array overriding the saved workflow's steps so the
- user can validate edits before persisting. The spawned session is
- a normal agent session; nothing is recorded as a WorkflowRun so
- History stays clean. Returns the new session id; the FE wires it
- to the workflow card via setCardSidecar(kind='testing') and the
- dashboard draws the labeled arrow chip between the two cards.
- """
- wf = storage.get_workflow(workflow_id)
- if not wf:
- raise HTTPException(status_code=404, detail="Workflow not found")
- draft_steps = (body or {}).get("steps")
- steps_texts: list[str]
- if isinstance(draft_steps, list) and draft_steps:
- steps_texts = [str(s.get("text") or "") for s in draft_steps if isinstance(s, dict) and s.get("text")]
- else:
- steps_texts = [s.text for s in wf.steps if s.text and s.text.strip()]
- if not steps_texts:
- raise HTTPException(status_code=400, detail="Workflow has no steps to test")
-
- from backend.apps.agents.core.models import AgentConfig
- from backend.apps.agents.agent_manager import agent_manager
- from backend.apps.workflows import executor
-
- config = AgentConfig(
- name=f"{wf.title or 'Workflow'} (test)",
- model=wf.model or "sonnet",
- mode=wf.mode or "agent",
- provider=wf.provider or "anthropic",
- system_prompt=executor._resolve_system_prompt(wf),
- allowed_tools=executor._resolve_allowed_tools(wf) or [
- "Read", "Edit", "Write", "Bash", "Glob", "Grep", "AskUserQuestion",
- ],
- dashboard_id=wf.dashboard_id,
- )
- session = await agent_manager.launch_agent(config)
-
- async def _drive_test() -> None:
- try:
- for step in steps_texts:
- await agent_manager.send_message(session.id, step)
- 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":
- return
- except Exception:
- logger.exception("test-run drive loop failed")
- asyncio.create_task(_drive_test())
-
- return {"session_id": session.id}
-
-
-@workflows.router.post("/{workflow_id}/parse-schedule")
-async def parse_schedule(workflow_id: str, body: dict):
- """Aux-LLM-parse natural language into a ScheduleConfig.
-
- Frontend SchedulingView (Image #49) hits this on submit; the parsed
- config rides back to the user for explicit "Schedule it" confirmation
- before any persistence. Returns the parsed config under {"schedule": ...}.
- """
- wf = storage.get_workflow(workflow_id)
- if not wf:
- raise HTTPException(status_code=404, detail="Workflow not found")
- text = (body or {}).get("text", "").strip()
- if not text:
- raise HTTPException(status_code=400, detail="Missing text")
- try:
- from backend.apps.agents.providers.registry import resolve_aux_model
- from backend.apps.settings.credentials import get_anthropic_client_for_model
- from backend.apps.settings.settings import load_settings as _ls
- except Exception:
- raise HTTPException(status_code=500, detail="Aux model unavailable")
- settings = _ls()
- try:
- aux_model, _ = await resolve_aux_model(settings, preferred_tier="haiku")
- client = get_anthropic_client_for_model(settings, aux_model)
- except Exception:
- raise HTTPException(status_code=500, detail="Aux model unavailable")
- import json, re
- prompt = (
- "Parse the following natural-language schedule into STRICT JSON. "
- "No prose, no fence, no comments. Schema:\n"
- ' {"repeat_unit": "day"|"week"|"month", '
- '"repeat_every": int>=1, '
- '"on_days": [int 0..6, Sunday=0], '
- '"hour": int 0..23, "minute": int 0..59, '
- '"timezone": IANA tz string (default to local)}\n\n'
- "Rules:\n"
- "- If user says weekdays, on_days=[1,2,3,4,5], repeat_unit=week.\n"
- "- If user says weekends, on_days=[0,6], repeat_unit=week.\n"
- "- If user names a single day (e.g. \"Mondays\"), on_days=[1], repeat_unit=week.\n"
- "- If user says daily/everyday, repeat_unit=day, on_days=[].\n"
- "- If no AM/PM, assume PM for 1-7 and AM for 8-12.\n"
- "- timezone: assume system local if not given.\n\n"
- f"Input: {text}"
- )
- try:
- resp = await client.messages.create(
- model=aux_model,
- max_tokens=180,
- messages=[
- {"role": "user", "content": prompt},
- {"role": "assistant", "content": "{"},
- ],
- )
- out = ""
- if isinstance(resp.content, list):
- for block in resp.content:
- if getattr(block, "type", None) == "text":
- out += getattr(block, "text", "")
- raw = "{" + out.strip() if not out.strip().startswith("{") else out.strip()
- m = re.search(r"\{[^{}]*\}", raw, flags=re.DOTALL)
- if m:
- raw = m.group(0)
- data = json.loads(raw)
- except Exception as e:
- logger.warning("parse-schedule: aux LLM failed: %s", e)
- raise HTTPException(status_code=400, detail="Couldn't parse schedule")
- cfg = wf.schedule.model_copy(update={
- "enabled": True,
- "repeat_unit": str(data.get("repeat_unit") or "week"),
- "repeat_every": int(data.get("repeat_every") or 1),
- "on_days": [int(d) for d in (data.get("on_days") or [])],
- "hour": int(data.get("hour") or 9),
- "minute": int(data.get("minute") or 0),
- "timezone": str(data.get("timezone") or wf.schedule.timezone or "UTC"),
- })
- return {"schedule": cfg.model_dump(mode="json")}
-
-
-@workflows.router.post("/{workflow_id}/run")
-async def run_workflow_now(workflow_id: str):
- wf = storage.get_workflow(workflow_id)
- if not wf:
- raise HTTPException(status_code=404, detail="Workflow not found")
- # executor.execute() owns the run record. Don't pre-create a stub here
- # 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"))
-
- # 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
- # short-circuit, _running collision) so the FE can render a toast
- # instead of silently switching to History.
- for _ in range(25):
- for r in storage.list_runs(wf.id, limit=10):
- if r.id not in pre_ids and r.triggered_by == "manual":
- return {
- "run_id": r.id,
- "status": r.status,
- "error": r.error,
- }
- await asyncio.sleep(0.01)
- return {"run_id": "", "status": None, "error": None}
-
-
-@workflows.router.post("/runs/{run_id}/stop")
-async def stop_run(run_id: str):
- """Force-terminate a running workflow's underlying agent session.
-
- Fired by RunningView's Stop button (Image #40). The run record gets
- marked failure with a "stopped by user" error so it surfaces correctly
- in History instead of looking like it succeeded.
- """
- target_wf_id = None
- target_run = None
- for wf in storage.list_workflows():
- for r in storage.list_runs(wf.id, limit=50):
- if r.id == run_id and r.status == "running":
- target_wf_id = wf.id
- target_run = r
- break
- if target_run:
- break
- if not target_run or not target_wf_id:
- raise HTTPException(status_code=404, detail="Run not found or not active")
- if target_run.session_id:
- try:
- from backend.apps.agents.agent_manager import agent_manager
- await agent_manager.close_session(target_run.session_id)
- except Exception:
- logger.exception("stop_run: close_session failed for %s", target_run.session_id)
- target_run.status = "failure"
- target_run.error = "Stopped by user"
- target_run.finished_at = datetime.now()
- storage.record_run(target_run)
- wf = storage.get_workflow(target_wf_id)
- if wf:
- _persist_run_fields(wf, {
- "last_run_status": "failure",
- "last_run_at": target_run.finished_at,
- "last_run_id": target_run.id,
- })
- try:
- from backend.apps.agents.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"),
- })
- except Exception:
- pass
- return {"ok": True}
-
-
-@workflows.router.get("/{workflow_id}/runs")
-async def list_workflow_runs(workflow_id: str, limit: int = 50):
- wf = storage.get_workflow(workflow_id)
- if not wf:
- raise HTTPException(status_code=404, detail="Workflow not found")
- runs = storage.list_runs(workflow_id, limit=limit)
- return {"runs": [r.model_dump(mode="json") for r in runs]}
diff --git a/backend/main.py b/backend/main.py
index 2341ad7a..ad65599c 100644
--- a/backend/main.py
+++ b/backend/main.py
@@ -32,12 +32,11 @@ from backend.apps.subscription.router import subscription
from backend.apps.auth.router import auth
from backend.apps.web.web import web
from backend.apps.agents.proxy.anthropic_proxy import anthropic_proxy
-from backend.apps.workflows.workflows import workflows
from fastapi.middleware.cors import CORSMiddleware
from fastapi import WebSocket, WebSocketDisconnect
import json
-main_app = MainApp([health, agents, skills, tools_lib, modes, settings, mcp_registry, skill_registry, outputs, dashboards, service, subscription, auth, web, anthropic_proxy, workflows])
+main_app = MainApp([health, agents, skills, tools_lib, modes, settings, mcp_registry, skill_registry, outputs, dashboards, service, subscription, auth, web, anthropic_proxy])
app = main_app.app
# Generate per-install auth token BEFORE we bind the HTTP port. By the
diff --git a/backend/tests/test_schedule_e2e.py b/backend/tests/test_schedule_e2e.py
deleted file mode 100644
index 3723041a..00000000
--- a/backend/tests/test_schedule_e2e.py
+++ /dev/null
@@ -1,277 +0,0 @@
-"""End-to-end smoke: does a scheduled workflow actually fire when its
-time hits, with the full scheduler loop running?
-
-Runs the real scheduler.start() loop with the executor mocked so we
-don't need a live agent_manager. Then arms a workflow whose
-next_run_at is one second in the future, waits, and asserts the
-mocked executor was called.
-
-Run:
- cd backend && .venv/bin/python -m pytest tests/test_schedule_e2e.py -v
-"""
-
-from __future__ import annotations
-
-import asyncio
-from datetime import datetime, timedelta, timezone
-from unittest.mock import AsyncMock
-from zoneinfo import ZoneInfo
-
-import pytest
-
-pytestmark = pytest.mark.asyncio
-
-
-@pytest.fixture(autouse=True)
-def isolated_data_dir(monkeypatch, tmp_path):
- from backend.apps.workflows import storage as _storage
- from backend.apps.workflows import escalation as _escalation
- from backend.apps.workflows import audit as _audit
- from backend.apps.workflows import scheduler as _scheduler
- 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"))
- monkeypatch.setattr(_storage, "_workflow_cache", {})
- monkeypatch.setattr(_storage, "_runs_cache", {})
- monkeypatch.setattr(_storage, "_cache_loaded", False)
- monkeypatch.setattr(_storage, "_paused", False)
- monkeypatch.setattr(_audit, "AUDIT_DIR", str(tmp_path / "workflows" / "audit"))
- # Module-level scheduler state survives across tests; reset it so
- # each test gets a fresh _wake Event bound to its own event loop.
- _scheduler._loop_task = None
- _scheduler._wake = asyncio.Event()
- _escalation._tasks.clear()
- _escalation._state.clear()
- yield
-
-
-def _make_wf(**overrides):
- from backend.apps.workflows.models import Workflow, ScheduleConfig, WorkflowStep
- base = dict(
- title="smoke",
- steps=[WorkflowStep(text="hi")],
- schedule=ScheduleConfig(
- enabled=True, repeat_unit="day", repeat_every=1,
- hour=9, minute=0, timezone="America/Los_Angeles",
- ),
- )
- base.update(overrides)
- return Workflow(**base)
-
-
-async def test_loop_fires_due_workflow(monkeypatch):
- """Arm a workflow to fire ~now and assert the executor was actually
- invoked by the scheduler loop within the test window. Note the save
- happens AFTER scheduler.start() so reconcile_on_startup doesn't
- clobber next_run_at."""
- from backend.apps.workflows import storage, scheduler, executor
-
- fired = asyncio.Event()
- captured: dict = {}
-
- async def fake_execute(wf, triggered_by="schedule", scheduled_for=None):
- captured["wf_id"] = wf.id
- captured["triggered_by"] = triggered_by
- captured["scheduled_for"] = scheduled_for
- from backend.apps.workflows.models import WorkflowRun
- run = WorkflowRun(
- workflow_id=wf.id,
- status="success",
- scheduled_for=scheduled_for,
- started_at=datetime.now(timezone.utc),
- finished_at=datetime.now(timezone.utc),
- triggered_by=triggered_by,
- )
- storage.record_run(run)
- fired.set()
- return run
-
- monkeypatch.setattr(executor, "execute", fake_execute)
-
- await scheduler.start()
- try:
- wf = _make_wf()
- wf.next_run_at = datetime.now(timezone.utc) + timedelta(seconds=1)
- storage.save_workflow(wf)
- scheduler.kick() # force immediate tick
- # Wait up to 5s for the fire to land.
- await asyncio.wait_for(fired.wait(), timeout=5.0)
- finally:
- await scheduler.stop()
-
- assert captured.get("wf_id") == wf.id
- assert captured.get("triggered_by") == "schedule"
- runs = storage.list_runs(wf.id, limit=10)
- assert len(runs) == 1
- assert runs[0].status == "success"
- # Scheduler should have rolled next_run_at forward to a future slot.
- after = storage.get_workflow(wf.id)
- assert after.next_run_at is not None
- assert after.next_run_at > datetime.now(timezone.utc)
-
-
-async def test_disabled_workflow_does_not_fire(monkeypatch):
- """Master switch off => loop never invokes the executor even if
- next_run_at is in the past."""
- from backend.apps.workflows import storage, scheduler, executor
-
- fake = AsyncMock()
- monkeypatch.setattr(executor, "execute", fake)
-
- wf = _make_wf()
- wf.schedule.enabled = False
- wf.next_run_at = datetime.now(timezone.utc) - timedelta(seconds=10)
- storage.save_workflow(wf)
-
- await scheduler.start()
- try:
- scheduler.kick()
- await asyncio.sleep(2.0)
- finally:
- await scheduler.stop()
- fake.assert_not_called()
-
-
-async def test_paused_state_blocks_all_fires(monkeypatch):
- """Global pause flag wins over per-workflow enabled state."""
- from backend.apps.workflows import storage, scheduler, executor
-
- fake = AsyncMock()
- monkeypatch.setattr(executor, "execute", fake)
-
- wf = _make_wf()
- wf.next_run_at = datetime.now(timezone.utc) - timedelta(seconds=1)
- storage.save_workflow(wf)
- storage.set_paused(True)
-
- await scheduler.start()
- try:
- scheduler.kick()
- await asyncio.sleep(2.0)
- finally:
- await scheduler.stop()
- fake.assert_not_called()
- storage.set_paused(False)
-
-
-async def test_reconcile_skip_rolls_past_missed(monkeypatch):
- """on_missed='skip' + a missed next_run_at => startup rolls forward
- to the next future fire without queuing a catch-up."""
- from backend.apps.workflows import storage, scheduler
- wf = _make_wf()
- wf.schedule.on_missed = "skip"
- # Stash a missed fire 6 hours ago.
- wf.next_run_at = datetime.now(timezone.utc) - timedelta(hours=6)
- storage.save_workflow(wf)
- scheduler.reconcile_on_startup()
- after = storage.get_workflow(wf.id)
- assert after.next_run_at is not None
- assert after.next_run_at > datetime.now(timezone.utc)
-
-
-async def test_reconcile_run_once_keeps_missed(monkeypatch):
- """on_missed='run_once' => startup leaves next_run_at in the past so
- the first tick fires a catch-up."""
- from backend.apps.workflows import storage, scheduler
- wf = _make_wf()
- wf.schedule.on_missed = "run_once"
- missed = datetime.now(timezone.utc) - timedelta(hours=6)
- wf.next_run_at = missed
- storage.save_workflow(wf)
- scheduler.reconcile_on_startup()
- after = storage.get_workflow(wf.id)
- assert after.next_run_at <= datetime.now(timezone.utc)
-
-
-async def test_create_workflow_schedules_next_fire():
- """POST-like create path: enabled schedule => next_run_at populated
- by compute_next_fire."""
- from backend.apps.workflows.models import Workflow, ScheduleConfig, WorkflowStep
- from backend.apps.workflows import scheduler
- wf = Workflow(
- title="t",
- steps=[WorkflowStep(text="hi")],
- schedule=ScheduleConfig(
- enabled=True, repeat_unit="week", repeat_every=1, on_days=[0],
- hour=9, minute=0, timezone="America/Los_Angeles",
- ),
- )
- nxt = scheduler.compute_next_fire(wf)
- assert nxt is not None
- assert nxt > datetime.now(timezone.utc)
- tz = ZoneInfo("America/Los_Angeles")
- local = nxt.astimezone(tz)
- assert local.weekday() == 6 # Python: Sunday
- assert (local.hour, local.minute) == (9, 0)
-
-
-async def test_next_run_at_advances_after_fire(monkeypatch):
- """After a fire the loop should re-compute next_run_at into the
- future and persist it, so the same fire can't repeat in the same
- minute."""
- from backend.apps.workflows import storage, scheduler, executor
-
- fired = asyncio.Event()
-
- async def fake_execute(wf, triggered_by="schedule", scheduled_for=None):
- from backend.apps.workflows.models import WorkflowRun
- run = WorkflowRun(
- workflow_id=wf.id, status="success", scheduled_for=scheduled_for,
- started_at=datetime.now(timezone.utc), finished_at=datetime.now(timezone.utc),
- triggered_by=triggered_by,
- )
- storage.record_run(run)
- fired.set()
- return run
-
- monkeypatch.setattr(executor, "execute", fake_execute)
-
- await scheduler.start()
- try:
- wf = _make_wf()
- armed_at = datetime.now(timezone.utc) + timedelta(seconds=1)
- wf.next_run_at = armed_at
- storage.save_workflow(wf)
- scheduler.kick()
- await asyncio.wait_for(fired.wait(), timeout=5.0)
- # Give the loop one extra tick to persist next_run_at.
- await asyncio.sleep(0.2)
- finally:
- await scheduler.stop()
-
- after = storage.get_workflow(wf.id)
- assert after.next_run_at is not None
- assert after.next_run_at > armed_at, "scheduler did not advance next_run_at past the slot it just fired"
-
-
-async def test_kick_wakes_loop_before_timeout(monkeypatch):
- """kick() should wake the loop early so manual schedule edits don't
- have to wait a full minute for the next tick boundary."""
- from backend.apps.workflows import storage, scheduler, executor
-
- fired = asyncio.Event()
-
- async def fake_execute(wf, triggered_by="schedule", scheduled_for=None):
- from backend.apps.workflows.models import WorkflowRun
- run = WorkflowRun(
- workflow_id=wf.id, status="success",
- started_at=datetime.now(timezone.utc),
- finished_at=datetime.now(timezone.utc), triggered_by=triggered_by,
- )
- storage.record_run(run)
- fired.set()
- return run
-
- monkeypatch.setattr(executor, "execute", fake_execute)
-
- await scheduler.start()
- try:
- wf = _make_wf()
- wf.next_run_at = datetime.now(timezone.utc) - timedelta(seconds=1)
- storage.save_workflow(wf)
- scheduler.kick()
- # Without kick(), the loop would sleep up to 60s before checking
- # the freshly-saved workflow. With kick, it should fire fast.
- await asyncio.wait_for(fired.wait(), timeout=3.0)
- finally:
- await scheduler.stop()
diff --git a/backend/tests/test_workflows_semantics.py b/backend/tests/test_workflows_semantics.py
deleted file mode 100644
index 79a887e7..00000000
--- a/backend/tests/test_workflows_semantics.py
+++ /dev/null
@@ -1,430 +0,0 @@
-"""Backend semantics tests for the scheduled-tasks fix.
-
-Covers:
- - DST-safe wall-clock math (spring forward + fall back) via zoneinfo
- - End conditions (ends_at + max_runs) auto-disable the schedule
- - Cost cap skips fires with a clear error
- - Freeze-default on for new scheduled non-source-session creates
- - Audit log captures field diffs
- - /workflows/active surfaces in-process running runs
- - Legacy timezone="local" coerced in memory at load
- - Storage paused flag round-trips
- - Month math no longer clamps to day 28
- - Server-side escalation kicks tasks (and ack cancels them)
-
-Run:
- pip install -r backend/requirements.txt -r backend/requirements-dev.txt
- cd backend && python -m pytest tests/test_workflows_semantics.py -v
-"""
-
-from __future__ import annotations
-
-import asyncio
-import json
-import os
-import shutil
-import tempfile
-from datetime import datetime, timedelta, timezone
-from zoneinfo import ZoneInfo
-
-import pytest
-
-
-@pytest.fixture(autouse=True)
-def isolated_data_dir(monkeypatch, tmp_path):
- """Point storage at a fresh tmpdir per test so we never touch a real
- install's workflows data. Reloads in-process module state so each test
- starts with empty caches."""
- from backend.apps.workflows import storage as _storage
- from backend.apps.workflows import escalation as _escalation
- 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"))
- monkeypatch.setattr(_storage, "_workflow_cache", {})
- monkeypatch.setattr(_storage, "_runs_cache", {})
- monkeypatch.setattr(_storage, "_cache_loaded", False)
- monkeypatch.setattr(_storage, "_paused", False)
- # Reset escalation registry between tests.
- _escalation._tasks.clear()
- _escalation._state.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
- monkeypatch.setattr(_audit, "AUDIT_DIR", str(tmp_path / "workflows" / "audit"))
- yield
-
-
-def _make_wf(**overrides):
- from backend.apps.workflows.models import Workflow, ScheduleConfig, WorkflowStep
- base = dict(
- title="t",
- steps=[WorkflowStep(text="hi")],
- schedule=ScheduleConfig(enabled=True, repeat_unit="day", repeat_every=1, hour=9, minute=0, timezone="America/Los_Angeles"),
- )
- base.update(overrides)
- return Workflow(**base)
-
-
-# --- DST tests ---------------------------------------------------------------
-
-def test_dst_spring_forward_weekly():
- """A 2:30am LA weekly Sunday schedule lands on 3:30am LA on the spring-
- forward Sunday (2025-03-09) because the wall clock skips 02:30."""
- from backend.apps.workflows.scheduler import _next_fire_after
- from backend.apps.workflows.models import ScheduleConfig
- tz = ZoneInfo("America/Los_Angeles")
- sched = ScheduleConfig(enabled=True, repeat_unit="week", repeat_every=1, on_days=[0], hour=2, minute=30, timezone="America/Los_Angeles")
- # Saturday 2025-03-08 23:00 LA, asking "what's the next Sunday 2:30?"
- ref_local = datetime(2025, 3, 8, 23, 0, tzinfo=tz)
- nxt = _next_fire_after(sched, ref_local.astimezone(timezone.utc))
- assert nxt is not None
- nxt_local = nxt.astimezone(tz)
- # 02:30 wall-clock on the spring-forward day doesn't exist; zoneinfo
- # resolves it forward to 03:30. The point is the *date* lands on the
- # 9th, not the 8th and not the 16th.
- assert nxt_local.date() == datetime(2025, 3, 9).date()
- assert nxt_local.hour in (2, 3)
-
-
-def test_dst_fall_back_no_double_fire():
- """A 9am LA daily schedule should fire exactly once on the fall-back day
- (2025-11-02) and the next fire is the 3rd, not the 2nd again."""
- from backend.apps.workflows.scheduler import _next_fire_after
- from backend.apps.workflows.models import ScheduleConfig
- tz = ZoneInfo("America/Los_Angeles")
- sched = ScheduleConfig(enabled=True, repeat_unit="day", repeat_every=1, hour=9, minute=0, timezone="America/Los_Angeles")
- ref_local = datetime(2025, 11, 1, 23, 0, tzinfo=tz)
- nxt = _next_fire_after(sched, ref_local.astimezone(timezone.utc))
- assert nxt.astimezone(tz).date() == datetime(2025, 11, 2).date()
- # After firing on the 2nd, the next fire should be the 3rd, not a
- # second 2nd from the duplicated hour.
- after = _next_fire_after(sched, nxt)
- assert after.astimezone(tz).date() == datetime(2025, 11, 3).date()
-
-
-# --- End condition tests -----------------------------------------------------
-
-def test_max_runs_disables_schedule():
- from backend.apps.workflows import storage, scheduler
- wf = _make_wf()
- wf.schedule.max_runs = 2
- wf.schedule.runs_count = 2
- wf.next_run_at = datetime.now(timezone.utc) - timedelta(minutes=1)
- storage.save_workflow(wf)
- asyncio.new_event_loop().run_until_complete(scheduler._tick())
- after = storage.get_workflow(wf.id)
- assert after.schedule.enabled is False
- assert after.next_run_at is None
-
-
-def test_ends_at_disables_schedule():
- from backend.apps.workflows import storage, scheduler
- wf = _make_wf()
- wf.schedule.ends_at = datetime.now(timezone.utc) - timedelta(days=1)
- wf.next_run_at = datetime.now(timezone.utc) - timedelta(minutes=1)
- storage.save_workflow(wf)
- asyncio.new_event_loop().run_until_complete(scheduler._tick())
- after = storage.get_workflow(wf.id)
- assert after.schedule.enabled is False
-
-
-# --- Month-day-31 (formerly clamped to 28) -----------------------------------
-
-def test_month_repeat_no_longer_clamps_to_28():
- """An every-month schedule starting on March 31 should next fire on
- April 30 (last day of April), then May 31, then June 30."""
- from backend.apps.workflows.scheduler import _next_fire_after
- from backend.apps.workflows.models import ScheduleConfig
- tz = ZoneInfo("America/Los_Angeles")
- sched = ScheduleConfig(enabled=True, repeat_unit="month", repeat_every=1, hour=9, minute=0, timezone="America/Los_Angeles")
- ref_local = datetime(2025, 3, 31, 10, 0, tzinfo=tz) # past 9am on the 31st
- nxt = _next_fire_after(sched, ref_local.astimezone(timezone.utc))
- assert nxt.astimezone(tz).date() == datetime(2025, 4, 30).date()
-
-
-# --- Cost cap ----------------------------------------------------------------
-
-def test_cost_cap_skips_with_clear_error(monkeypatch):
- from backend.apps.workflows import storage, executor
- from backend.apps.workflows.models import WorkflowRun
- wf = _make_wf()
- wf.cost_cap_usd_monthly = 1.0
- storage.save_workflow(wf)
- storage.record_run(WorkflowRun(workflow_id=wf.id, status="success", cost_usd=0.6, started_at=datetime.now(timezone.utc), finished_at=datetime.now(timezone.utc)))
- storage.record_run(WorkflowRun(workflow_id=wf.id, status="success", cost_usd=0.6, started_at=datetime.now(timezone.utc), finished_at=datetime.now(timezone.utc)))
-
- async def fake_launch(*a, **k):
- raise AssertionError("agent_manager should not be reached when cost-capped")
-
- # Patch agent_manager.launch_agent so we'd fail loudly if the cap
- # didn't short-circuit before launch.
- from backend.apps.agents import agent_manager
- monkeypatch.setattr(agent_manager.agent_manager, "launch_agent", fake_launch)
-
- run = asyncio.new_event_loop().run_until_complete(executor.execute(wf, triggered_by="schedule"))
- assert run.status == "skipped"
- assert "Monthly cost cap reached" in (run.error or "")
-
-
-# --- Freeze-default for scheduled non-source-session creates ----------------
-
-def test_freeze_defaults_on_for_scheduled_create():
- """POST /workflows/create with schedule.enabled=true and no source
- session should flip actions.freeze=True to keep blast radius small."""
- from backend.apps.workflows.workflows import create_workflow
- from backend.apps.workflows.models import WorkflowCreate, ScheduleConfig, ActionsConfig
- body = WorkflowCreate(
- title="scheduled",
- schedule=ScheduleConfig(enabled=True, repeat_unit="day", repeat_every=1, hour=9, minute=0),
- actions=ActionsConfig(freeze=False, configured_sets=[]),
- )
- result = asyncio.new_event_loop().run_until_complete(create_workflow(body))
- assert result["actions"]["freeze"] is True
-
-
-def test_freeze_not_forced_when_source_session_present():
- """Source-session creates inherit the chat's choices; we don't override."""
- from backend.apps.workflows.workflows import create_workflow
- from backend.apps.workflows.models import WorkflowCreate, ScheduleConfig, ActionsConfig
- body = WorkflowCreate(
- title="from chat",
- source_session_id="sess-1",
- schedule=ScheduleConfig(enabled=True, repeat_unit="day", repeat_every=1, hour=9, minute=0),
- actions=ActionsConfig(freeze=False, configured_sets=[]),
- )
- result = asyncio.new_event_loop().run_until_complete(create_workflow(body))
- assert result["actions"]["freeze"] is False
-
-
-# --- Audit log ---------------------------------------------------------------
-
-def test_audit_log_records_title_change():
- from backend.apps.workflows import audit
- audit.log_change("wf-1", "user", {"title": "old"}, {"title": "new"})
- entries = audit.read_tail("wf-1", limit=10)
- assert len(entries) == 1
- diff = entries[0]["diff"]
- assert diff["title"]["before"] == "old"
- assert diff["title"]["after"] == "new"
-
-
-def test_audit_log_no_op_when_unchanged():
- from backend.apps.workflows import audit
- audit.log_change("wf-2", "user", {"title": "same"}, {"title": "same"})
- assert audit.read_tail("wf-2") == []
-
-
-# --- /workflows/active -------------------------------------------------------
-
-def test_list_active_reflects_running_map():
- from backend.apps.workflows import storage, executor, scheduler
- wf = _make_wf(title="active-test")
- storage.save_workflow(wf)
- from backend.apps.workflows.models import WorkflowRun
- run = WorkflowRun(workflow_id=wf.id, status="running")
- storage.record_run(run)
- executor._running[wf.id] = run.id
- try:
- active = scheduler.list_active()
- assert len(active) == 1
- assert active[0]["workflow_id"] == wf.id
- assert active[0]["title"] == "active-test"
- finally:
- executor._running.pop(wf.id, None)
-
-
-# --- Legacy tz coercion ------------------------------------------------------
-
-def test_legacy_timezone_coerced_on_load(monkeypatch):
- from backend.apps.workflows import storage
- storage._ensure_dirs()
- wf_id = "legacy-wf"
- legacy_blob = {
- "id": wf_id,
- "title": "legacy",
- "schedule": {
- "enabled": False, "repeat_every": 1, "repeat_unit": "week",
- "on_days": [], "hour": 9, "minute": 0, "timezone": "local",
- "on_missed": "skip", "ends_at": None, "max_runs": None, "runs_count": 0,
- },
- }
- with open(os.path.join(storage.DATA_DIR, f"{wf_id}.json"), "w") as f:
- json.dump(legacy_blob, f)
- monkeypatch.setenv("OPENSWARM_TIMEZONE", "America/Los_Angeles")
- monkeypatch.setattr(storage, "_cache_loaded", False)
- loaded = storage.get_workflow(wf_id)
- assert loaded is not None
- # In-memory should be the host zone, not "local".
- assert loaded.schedule.timezone == "America/Los_Angeles"
- # On-disk file should be unchanged (still "local") so we don't churn
- # mtime on every restart.
- with open(os.path.join(storage.DATA_DIR, f"{wf_id}.json")) as f:
- on_disk = json.load(f)
- assert on_disk["schedule"]["timezone"] == "local"
-
-
-# --- Paused flag -------------------------------------------------------------
-
-def test_paused_flag_persists_and_blocks_tick():
- from backend.apps.workflows import storage, scheduler
- wf = _make_wf()
- wf.next_run_at = datetime.now(timezone.utc) - timedelta(minutes=1)
- storage.save_workflow(wf)
- storage.set_paused(True)
- # Reload simulates a backend restart.
- storage._cache_loaded = False
- assert storage.get_paused() is True
- # Tick must not advance next_run_at when paused.
- before = storage.get_workflow(wf.id).next_run_at
- asyncio.new_event_loop().run_until_complete(scheduler._tick())
- after = storage.get_workflow(wf.id).next_run_at
- assert before == after
-
-
-# --- Escalation --------------------------------------------------------------
-
-def test_escalation_schedules_and_ack_cancels():
- from backend.apps.workflows import escalation
- from backend.apps.workflows.models import Workflow, PermissionTier, WorkflowRun, ScheduleConfig
-
- async def runner():
- wf = Workflow(title="t", permissions=[
- PermissionTier(kind="notify"),
- PermissionTier(kind="text", after_minutes=60, phone="+15551234567"),
- ])
- run = WorkflowRun(workflow_id=wf.id, status="success")
- escalation.schedule(wf, run)
- # State should be present immediately.
- await asyncio.sleep(0.01)
- assert escalation.status(run.id) is not None
- # Ack cancels.
- assert escalation.cancel(run.id) is True
- await asyncio.sleep(0.01)
- assert escalation.status(run.id) is None
-
- asyncio.new_event_loop().run_until_complete(runner())
-
-
-def test_executor_merge_does_not_clobber_concurrent_patch():
- """Executor's final save must NOT overwrite unrelated fields that
- were PATCHed while the run was in flight. We simulate this by
- capturing a wf, mutating storage's record directly (acting as the
- PATCH that landed mid-run), then asking the executor's persist
- helper to flush its run-side bookkeeping. The patched fields must
- survive.
- """
- from backend.apps.workflows import storage, executor
- from datetime import datetime
- wf = _make_wf(title="t-orig")
- storage.save_workflow(wf)
- # Simulate a user PATCH mid-run.
- storage._workflow_cache[wf.id].title = "t-patched"
- storage._workflow_cache[wf.id].description = "patched while running"
- storage.save_workflow(storage._workflow_cache[wf.id])
- # Executor uses the stale `wf` it captured before the patch. With
- # the merge helper, the patched fields must remain.
- executor._persist_run_fields(wf, {
- "last_run_at": datetime.now(),
- "last_run_status": "success",
- })
- after = storage.get_workflow(wf.id)
- assert after.title == "t-patched", "title clobbered by executor"
- assert after.description == "patched while running", "description clobbered"
- assert after.last_run_status == "success"
-
-
-def test_executor_delete_during_run_does_not_resurrect():
- """If the workflow was deleted mid-run, executor's persist must
- silently no-op so the deleted record isn't re-written."""
- from backend.apps.workflows import storage, executor
- from datetime import datetime
- wf = _make_wf(title="doomed")
- storage.save_workflow(wf)
- storage.delete_workflow(wf.id)
- executor._persist_run_fields(wf, {
- "last_run_at": datetime.now(),
- "last_run_status": "success",
- }, schedule_runs_count_delta=1)
- assert storage.get_workflow(wf.id) is None
-
-
-def test_patch_if_match_rejects_stale_write():
- """A PATCH with a stale If-Match must return 409. Without If-Match,
- the request still succeeds (legacy clients keep working until they
- roll out the header)."""
- from backend.apps.workflows.workflows import update_workflow
- from backend.apps.workflows.models import WorkflowUpdate
- from backend.apps.workflows import storage
- from fastapi import HTTPException
-
- wf = _make_wf(title="optimistic-test")
- storage.save_workflow(wf)
- stale = "1999-01-01T00:00:00"
-
- async def runner():
- # Stale If-Match → 409.
- try:
- await update_workflow(wf.id, WorkflowUpdate(title="x"), if_match=stale)
- return "no exception"
- except HTTPException as he:
- return he.status_code
- code = asyncio.new_event_loop().run_until_complete(runner())
- assert code == 409, f"stale If-Match should 409, got {code}"
-
- # Fresh If-Match → 200.
- fresh = storage.get_workflow(wf.id)
- fresh_stamp = fresh.updated_at.isoformat()
- async def runner_ok():
- return await update_workflow(wf.id, WorkflowUpdate(title="y"), if_match=fresh_stamp)
- result = asyncio.new_event_loop().run_until_complete(runner_ok())
- assert result["title"] == "y"
-
- # Missing If-Match → legacy path still works.
- async def runner_legacy():
- return await update_workflow(wf.id, WorkflowUpdate(title="z"), if_match=None)
- result = asyncio.new_event_loop().run_until_complete(runner_legacy())
- assert result["title"] == "z"
-
-
-def test_killed_by_restart_message_is_friendly():
- """stuck-run reaper writes a user-facing string, not internal jargon."""
- from backend.apps.workflows import storage, scheduler
- from backend.apps.workflows.models import WorkflowRun
- wf = _make_wf()
- storage.save_workflow(wf)
- storage.record_run(WorkflowRun(workflow_id=wf.id, status="running"))
- scheduler._mark_stuck_runs_failed()
- runs = storage.list_runs(wf.id, limit=10)
- assert any(r.status == "failure" and "OpenSwarm closed" in (r.error or "") for r in runs)
- assert not any("Killed by restart" in (r.error or "") for r in runs)
-
-
-def test_run_endpoint_surfaces_skipped_status():
- """POST /workflows/{id}/run returns the skipped status + error when
- a cost-cap or in-flight collision short-circuits the run."""
- from backend.apps.workflows.workflows import run_workflow_now
- from backend.apps.workflows import storage
- from backend.apps.workflows.models import WorkflowRun
- from datetime import datetime, timezone
- wf = _make_wf(title="cap-immediate")
- wf.cost_cap_usd_monthly = 0.01
- storage.save_workflow(wf)
- # Burn the cap with a single $5 historical run.
- storage.record_run(WorkflowRun(workflow_id=wf.id, status="success", cost_usd=5.0,
- started_at=datetime.now(timezone.utc),
- finished_at=datetime.now(timezone.utc)))
-
- async def runner():
- return await run_workflow_now(wf.id)
- res = asyncio.new_event_loop().run_until_complete(runner())
- assert res.get("status") == "skipped"
- assert "cost cap" in (res.get("error") or "").lower()
-
-
-def test_escalation_noop_for_single_tier():
- from backend.apps.workflows import escalation
- from backend.apps.workflows.models import Workflow, PermissionTier, WorkflowRun
- wf = Workflow(title="t", permissions=[PermissionTier(kind="notify")])
- run = WorkflowRun(workflow_id=wf.id, status="success")
- escalation.schedule(wf, run)
- assert escalation.status(run.id) is None
diff --git a/electron/main.js b/electron/main.js
index 57d1e090..6f30242c 100644
--- a/electron/main.js
+++ b/electron/main.js
@@ -34,7 +34,6 @@ const fs = require('fs');
const getPort = require('get-port');
const http = require('http');
const affiliateTracking = require('./affiliateTracking');
-const workflowsLifecycle = require('./workflowsLifecycle');
// Defender warmup: NSIS runs us with --prewarm right after install so Windows scans the bundled binaries while the user is already watching the installer instead of staring at a slow first launch.
if (process.argv.includes('--prewarm') && process.platform === 'win32') {
@@ -694,10 +693,6 @@ function markBackendReady() {
if (backendReady) return;
backendReady = true;
_backendReadyResolve();
- try {
- workflowsLifecycle.setBackend({ port: backendPort, token: authToken });
- workflowsLifecycle.startPolling();
- } catch (_) {}
}
function getAuthTokenFilePath() {
@@ -1146,8 +1141,8 @@ app.whenReady().then(async () => {
backendPort = parseInt(process.env.OPENSWARM_PORT || '8324', 10);
console.log(`Dev mode: using existing backend on port ${backendPort}`);
emitSplashStatus('Connecting to dev backend…');
- // Load the token before marking ready, same as prod, so the workflow
- // poller's setBackend() gets a real token instead of '' (else it 401s).
+ // Load the token before marking ready, same as prod, so renderer
+ // fetches get a real token instead of '' (else they 401).
await loadAuthToken();
markBackendReady();
} else {
@@ -1530,10 +1525,6 @@ app.on('before-quit', async (event) => {
try {
await postShutdownAllApps(2000);
} catch (_) {}
- // Give in-flight workflow runs up to 30s to land so we don't destroy paid LLM work.
- try {
- await workflowsLifecycle.drainOnQuit(30);
- } catch (_) {}
app.quit();
});
@@ -1657,11 +1648,6 @@ ipcMain.handle('set-allow-prerelease', async (_e, value) => {
ipcMain.handle('install-update', async () => {
if (!autoUpdater) return;
- // Veto while a workflow is in flight; lifecycle poller fires the deferred install once active drains.
- try {
- const vetoed = await workflowsLifecycle.maybeVetoInstall();
- if (vetoed) return { vetoed: true };
- } catch (_) {}
autoUpdater.quitAndInstall(false, true);
});
diff --git a/electron/package.json b/electron/package.json
index 285bd1e9..2b603628 100644
--- a/electron/package.json
+++ b/electron/package.json
@@ -1,6 +1,6 @@
{
"name": "openswarm",
- "version": "1.1.63",
+ "version": "1.1.64",
"description": "OpenSwarm — AI Agent Orchestrator",
"author": "openswarm-ai",
"main": "main.js",
diff --git a/electron/workflowsLifecycle.js b/electron/workflowsLifecycle.js
deleted file mode 100644
index 7b79b6ee..00000000
--- a/electron/workflowsLifecycle.js
+++ /dev/null
@@ -1,208 +0,0 @@
-// Lifecycle helpers that keep scheduled workflows surviving real-world
-// app states (machine sleep, window closed, auto-update). All exports are
-// safe to call before the backend is up; failed fetches return null and
-// callers degrade to "no active runs known."
-
-const { app, powerSaveBlocker, Notification, shell } = require('electron');
-const http = require('http');
-
-let backendPortRef = null;
-let authTokenRef = null;
-let blockerId = null;
-let updaterVetoPending = false;
-let pollTimer = null;
-let lastActiveCount = 0;
-let onActiveChange = () => {};
-
-function setBackend({ port, token }) {
- backendPortRef = port;
- authTokenRef = token;
-}
-
-function setActiveChangeListener(cb) {
- onActiveChange = cb || (() => {});
-}
-
-// Cheap GET to the localhost backend. Resolves null on any error.
-function fetchJson(pathStr) {
- return new Promise((resolve) => {
- if (!backendPortRef) return resolve(null);
- const req = http.request({
- hostname: '127.0.0.1',
- port: backendPortRef,
- path: pathStr,
- method: 'GET',
- headers: authTokenRef ? { Authorization: `Bearer ${authTokenRef}` } : {},
- timeout: 1500,
- }, (res) => {
- let data = '';
- res.on('data', (c) => { data += c; });
- res.on('end', () => {
- try { resolve(JSON.parse(data)); } catch { resolve(null); }
- });
- });
- req.on('error', () => resolve(null));
- req.on('timeout', () => { req.destroy(); resolve(null); });
- req.end();
- });
-}
-
-async function getActive() {
- // Must hit the /api prefix; the bare path 401s and would leave the
- // powerSaveBlocker + updater-veto blind to in-flight runs.
- const res = await fetchJson('/api/workflows/active');
- if (!res || !Array.isArray(res.active)) return [];
- return res.active;
-}
-
-// powerSaveBlocker holds the system awake while at least one workflow is
-// active. Released as soon as the active list goes empty so we don't pin
-// the user's laptop on idle.
-function ensureBlocker(active) {
- if (active && blockerId == null) {
- try { blockerId = powerSaveBlocker.start('prevent-app-suspension'); } catch (_) {}
- } else if (!active && blockerId != null) {
- try { powerSaveBlocker.stop(blockerId); } catch (_) {}
- blockerId = null;
- }
-}
-
-function startPolling() {
- if (pollTimer) return;
- // 5s cadence is the sweet spot: fast enough to release the
- // powerSaveBlocker promptly after a fire, slow enough that the localhost
- // request is invisible in CPU traces.
- pollTimer = setInterval(async () => {
- const active = await getActive();
- const count = active.length;
- ensureBlocker(count > 0);
- if (count !== lastActiveCount) {
- lastActiveCount = count;
- try { onActiveChange(active); } catch (_) {}
- }
- // If the updater queued an install while a run was in flight, fire it
- // the moment the active list drains.
- if (updaterVetoPending && count === 0) {
- updaterVetoPending = false;
- try {
- const { autoUpdater } = require('electron-updater');
- autoUpdater.quitAndInstall(false, true);
- } catch (_) {}
- }
- }, 5000);
-}
-
-function stopPolling() {
- if (pollTimer) {
- clearInterval(pollTimer);
- pollTimer = null;
- }
-}
-
-// Updater veto: if a workflow is running and the user clicks "Install
-// update," queue it instead of quitAndInstall'ing on top of an active
-// run. Returns true if vetoed (caller should display a "queued" banner),
-// false otherwise.
-async function maybeVetoInstall() {
- const active = await getActive();
- if (active.length === 0) return false;
- updaterVetoPending = true;
- return true;
-}
-
-// Drain on quit: give in-flight runs up to QUIT_DRAIN_S to finish before
-// killing the backend. The user-facing tradeoff is a slow quit when busy
-// vs. losing the run; we lean toward "wait" because the run already
-// committed real cost.
-function drainOnQuit(maxSeconds = 30) {
- return new Promise((resolve) => {
- const deadline = Date.now() + maxSeconds * 1000;
- const tick = async () => {
- const active = await getActive();
- if (active.length === 0 || Date.now() > deadline) return resolve();
- setTimeout(tick, 500);
- };
- tick();
- });
-}
-
-// Native OS notification. Falls back silently when Notification isn't
-// supported (some Linux setups, headless test envs). When `actions` is
-// provided AND we're on macOS, attaches button actions so the user can
-// ack/re-run/open without the app taking focus. Routes the chosen
-// outcome back to the renderer via an IPC channel that the renderer's
-// WebSocketManager already listens for.
-function showNativeNotification({ title, body, deepLink, runId, workflowId, actions }) {
- if (!Notification || !Notification.isSupported()) return null;
- try {
- const opts = { title: title || 'OpenSwarm', body: body || '', silent: false };
- const platformActions = Array.isArray(actions) && process.platform === 'darwin'
- ? actions.map((a) => ({ type: 'button', text: a.text }))
- : undefined;
- if (platformActions && platformActions.length) opts.actions = platformActions;
- const n = new Notification(opts);
- const route = (outcome) => {
- try {
- const { BrowserWindow } = require('electron');
- const wins = BrowserWindow.getAllWindows();
- const wc = wins[0]?.webContents;
- if (wc) wc.send('workflow:notification-action', { outcome, runId, workflowId, deepLink });
- } catch (_) {}
- };
- n.on('action', (_event, idx) => {
- const a = (actions || [])[idx];
- if (a) route(a.outcome);
- });
- n.on('click', () => {
- if (deepLink) {
- try { shell.openExternal(deepLink); } catch (_) {}
- }
- route('open');
- });
- n.show();
- return n;
- } catch (_) {
- return null;
- }
-}
-
-// Launch-at-login wrappers. macOS + Windows both honor this; Linux is a
-// no-op in Electron's API.
-function getLoginItem() {
- try {
- const { openAtLogin } = app.getLoginItemSettings();
- return Boolean(openAtLogin);
- } catch (_) { return false; }
-}
-
-function setLoginItem(value) {
- try {
- // openAsHidden is macOS-only; on Windows the equivalent is passing
- // a --hidden arg and having main.js suppress the initial window
- // when the arg is present. Linux uses a .desktop file in
- // ~/.config/autostart/ which Electron writes for us via this same
- // call (no extra plumbing needed).
- const opts = {
- openAtLogin: Boolean(value),
- openAsHidden: true,
- };
- if (process.platform === 'win32') {
- opts.args = ['--hidden'];
- }
- app.setLoginItemSettings(opts);
- return Boolean(value);
- } catch (_) { return false; }
-}
-
-module.exports = {
- setBackend,
- setActiveChangeListener,
- startPolling,
- stopPolling,
- getActive,
- maybeVetoInstall,
- drainOnQuit,
- showNativeNotification,
- getLoginItem,
- setLoginItem,
-};
diff --git a/frontend/src/app/pages/Dashboard/DashboardToolbar.tsx b/frontend/src/app/pages/Dashboard/DashboardToolbar.tsx
index a3df6f96..b14400f5 100644
--- a/frontend/src/app/pages/Dashboard/DashboardToolbar.tsx
+++ b/frontend/src/app/pages/Dashboard/DashboardToolbar.tsx
@@ -34,9 +34,6 @@ import SearchIcon from '@mui/icons-material/Search';
import { motion } from 'framer-motion';
import ChatInput from '@/app/pages/AgentChat/ChatInput';
import type { ContextPath } from '@/app/components/editor/DirectoryBrowser';
-import SchedulePopover from '@/app/pages/Workflows/SchedulePopover';
-import { openWorkflowCard } from '@/shared/state/workflowsSlice';
-import { addWorkflowCard, openWorkflowsHub } from '@/shared/state/dashboardLayoutSlice';
import { useElementSelection } from '@/app/components/editor/ElementSelectionContext';
import { useClaudeTokens } from '@/shared/styles/ThemeContext';
import { useAppDispatch, useAppSelector } from '@/shared/hooks';
@@ -176,7 +173,6 @@ const DashboardToolbar = React.forwardRef(
const [viewSearch, setViewSearch] = useState('');
const [historyOpen, setHistoryOpen] = useState(false);
const [historyQuery, setHistoryQuery] = useState('');
- const [popoverMode, setPopoverMode] = useState<'search' | 'schedule'>('search');
const shortcut = useAppSelector((s) => s.settings.data.new_agent_shortcut);
const outputs = useAppSelector((s) => s.outputs.items);
const historySearch = useAppSelector((s) => s.agents.historySearch);
@@ -449,7 +445,6 @@ const DashboardToolbar = React.forwardRef(
// the render branch below so the popover would be hidden
// behind it otherwise.
if (inputOpen) onCancel();
- setPopoverMode('search');
setHistoryOpen(true);
}}
role="button"
@@ -484,7 +479,7 @@ const DashboardToolbar = React.forwardRef(
padding: isExpanded ? '6px' : '5px',
userSelect: 'none' as const,
overflow: inputOpen || newAgentBounce || historyOpen ? 'visible' : 'hidden',
- // historyOpen: width owned by SchedulePopover; leave undefined so framer-motion measures intrinsic size.
+ // historyOpen: width owned by the inline history list; leave undefined so framer-motion measures intrinsic size.
width: viewPickerOpen ? 580 : historyOpen ? undefined : isExpanded ? 540 : undefined,
}}
>
@@ -508,34 +503,57 @@ const DashboardToolbar = React.forwardRef(
/>
) : historyOpen ? (
-