diff --git a/backend/apps/agents/agent_manager.py b/backend/apps/agents/agent_manager.py
index 15d8155b..431e73d7 100644
--- a/backend/apps/agents/agent_manager.py
+++ b/backend/apps/agents/agent_manager.py
@@ -1061,6 +1061,23 @@ 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,
@@ -1069,20 +1086,15 @@ 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 without hallucinating.
+ # so it can answer day-of-week questions and pick sensible
+ # cadences ("every Friday afternoon") without hallucinating.
try:
from zoneinfo import ZoneInfo
- # 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"
+ from backend.apps.workflows.storage import _resolve_host_tz_name
+ tz_name = _resolve_host_tz_name()
now_local = datetime.now(ZoneInfo(tz_name))
tz_abbr = now_local.strftime("%Z") or tz_name
time_ctx = (
@@ -1180,6 +1192,27 @@ 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
new file mode 100644
index 00000000..5a75a603
--- /dev/null
+++ b/backend/apps/agents/schedule_mcp_server.py
@@ -0,0 +1,412 @@
+#!/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 52be0543..b62c608f 100644
--- a/backend/apps/dashboards/models.py
+++ b/backend/apps/dashboards/models.py
@@ -53,12 +53,17 @@ class NotePosition(BaseModel):
class DashboardLayout(BaseModel):
- # extra="allow" so any keys the FE sends (or legacy on-disk layouts
- # carry) round-trip without Pydantic stripping them.
+ # 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.
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
new file mode 100644
index 00000000..e69de29b
diff --git a/backend/apps/workflows/audit.py b/backend/apps/workflows/audit.py
new file mode 100644
index 00000000..4ea43a33
--- /dev/null
+++ b/backend/apps/workflows/audit.py
@@ -0,0 +1,95 @@
+"""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
new file mode 100644
index 00000000..a0ab7536
--- /dev/null
+++ b/backend/apps/workflows/escalation.py
@@ -0,0 +1,90 @@
+"""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
new file mode 100644
index 00000000..28073db5
--- /dev/null
+++ b/backend/apps/workflows/executor.py
@@ -0,0 +1,334 @@
+"""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
new file mode 100644
index 00000000..9819b17f
--- /dev/null
+++ b/backend/apps/workflows/models.py
@@ -0,0 +1,164 @@
+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
new file mode 100644
index 00000000..20034872
--- /dev/null
+++ b/backend/apps/workflows/notifier.py
@@ -0,0 +1,55 @@
+"""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
new file mode 100644
index 00000000..d05a5be3
--- /dev/null
+++ b/backend/apps/workflows/scheduler.py
@@ -0,0 +1,352 @@
+"""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
new file mode 100644
index 00000000..86d80352
--- /dev/null
+++ b/backend/apps/workflows/storage.py
@@ -0,0 +1,197 @@
+"""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
new file mode 100644
index 00000000..cb1b4634
--- /dev/null
+++ b/backend/apps/workflows/workflows.py
@@ -0,0 +1,807 @@
+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 f4c76cf4..7e17a007 100644
--- a/backend/main.py
+++ b/backend/main.py
@@ -32,11 +32,12 @@ 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])
+main_app = MainApp([health, agents, skills, tools_lib, modes, settings, mcp_registry, skill_registry, outputs, dashboards, service, subscription, auth, web, anthropic_proxy, workflows])
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
new file mode 100644
index 00000000..3723041a
--- /dev/null
+++ b/backend/tests/test_schedule_e2e.py
@@ -0,0 +1,277 @@
+"""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
new file mode 100644
index 00000000..79a887e7
--- /dev/null
+++ b/backend/tests/test_workflows_semantics.py
@@ -0,0 +1,430 @@
+"""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 1fb765ef..7c8679e9 100644
--- a/electron/main.js
+++ b/electron/main.js
@@ -34,6 +34,7 @@ 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') {
@@ -729,6 +730,10 @@ function markBackendReady() {
if (backendReady) return;
backendReady = true;
_backendReadyResolve();
+ try {
+ workflowsLifecycle.setBackend({ port: backendPort, token: authToken });
+ workflowsLifecycle.startPolling();
+ } catch (_) {}
}
function getAuthTokenFilePath() {
@@ -1229,8 +1234,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 renderer
- // fetches get a real token instead of '' (else they 401).
+ // Load the token before marking ready, same as prod, so the workflow
+ // poller's setBackend() gets a real token instead of '' (else it 401s).
await loadAuthToken();
markBackendReady();
} else {
@@ -1626,6 +1631,10 @@ app.on('before-quit', async (event) => {
try {
await postShutdownAllApps(10000);
} 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();
});
@@ -1752,6 +1761,11 @@ 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/workflowsLifecycle.js b/electron/workflowsLifecycle.js
new file mode 100644
index 00000000..7b79b6ee
--- /dev/null
+++ b/electron/workflowsLifecycle.js
@@ -0,0 +1,208 @@
+// 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 7ca528ed..666873b7 100644
--- a/frontend/src/app/pages/Dashboard/DashboardToolbar.tsx
+++ b/frontend/src/app/pages/Dashboard/DashboardToolbar.tsx
@@ -34,6 +34,9 @@ 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';
@@ -173,6 +176,7 @@ 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);
@@ -420,7 +424,7 @@ const DashboardToolbar = React.forwardRef(
padding: isExpanded ? '6px' : '5px',
userSelect: 'none' as const,
overflow: inputOpen || newAgentBounce || historyOpen ? 'visible' : 'hidden',
- // historyOpen: width owned by the inline history list; leave undefined so framer-motion measures intrinsic size.
+ // historyOpen: width owned by SchedulePopover; leave undefined so framer-motion measures intrinsic size.
width: viewPickerOpen ? 580 : historyOpen ? undefined : isExpanded ? 540 : undefined,
}}
>
@@ -444,57 +448,34 @@ const DashboardToolbar = React.forwardRef(
/>
) : historyOpen ? (
- // Past-chat search list. Fixed-size bordered surface (matches the
- // toolbar popover footprint) with a search input + scrollable
- // results; clicking a row resumes that chat.
-
-
-
-
- setHistoryQuery(e.target.value)}
- placeholder="Search past chats..."
- sx={{ flex: 1, fontSize: '0.85rem', color: c.text.primary, fontFamily: c.font.sans, '& input::placeholder': { color: c.text.ghost, opacity: 1 } }}
- />
-
-
- {historySearch.results.length === 0 && !historySearch.loading && (
-
- {historyQuery ? 'No matching chats' : 'No chat history yet'}
-
- )}
- {historySearch.results.map((entry) => (
- handleHistorySelect(entry.id)}
- sx={{ display: 'flex', alignItems: 'center', gap: 1, px: 1.5, py: 0.9, cursor: 'pointer', '&:hover': { bgcolor: c.bg.elevated } }}
- >
-
- {entry.name}
-
-
- {formatRelativeTime(entry.closed_at)}
-
-
- ))}
-
-
-
+