[eric] events: instant file signals (kqueue), SSE stream tier, watcher-attention surface

backend/apps/agents/schedule_mcp_server.py

@@ -294,7 +294,9 @@ TOOLS = [
             "'agent' = ANY other condition; an agent checks it on an interval with its tools "
             "(needs check, a plain sentence like 'a new email from my landlord arrived'; if the "
             "check needs a connected account, list the tool names in mcps); "
-            "'custom' = an outside system will push events to us (returns the endpoint to call). "
+            "'custom' = an outside system will push events to us (returns the endpoint to call); "
+            "'stream' = subscribe to a live Server-Sent Events feed URL (events arrive instantly; "
+            "use contains to keep only matching messages). "
             "Attach to an existing workflow by passing workflow (its id or exact title), or pass "
             "title + steps to create a new one (steps are what the agent DOES when it fires). "
             "Use only_when for a plain-English filter ('only if it mentions Friday'). "
@@ -306,9 +308,10 @@ TOOLS = [
                 "workflow": {"type": "string", "description": "Existing workflow id or exact title to attach the trigger to. Omit when creating a new workflow via title + steps."},
                 "title": {"type": "string", "description": "Name for a NEW workflow (when workflow is omitted)."},
                 "steps": {"type": "array", "items": {"type": "string"}, "description": "What to do when the event fires, as ordered agent instructions. Required when creating a new workflow."},
-                "kind": {"type": "string", "enum": ["file", "web", "agent", "custom"], "description": "What produces the events."},
+                "kind": {"type": "string", "enum": ["file", "web", "agent", "custom", "stream"], "description": "What produces the events."},
                 "path": {"type": "string", "description": "kind=file: the file or folder to watch (~ ok)."},
-                "url": {"type": "string", "description": "kind=web: the page URL to watch."},
+                "url": {"type": "string", "description": "kind=web: the page URL to watch. kind=stream: the SSE feed URL."},
+                "contains": {"type": "string", "description": "kind=stream: only messages containing this substring become events."},
                 "watch_for": {"type": "string", "description": "kind=web: what change matters, in the user's words."},
                 "check": {"type": "string", "description": "kind=agent: the condition to check, one plain sentence."},
                 "mcps": {"type": "array", "items": {"type": "string"}, "description": "kind=agent: connected tool names the check may use (e.g. 'google-workspace'). Only what the user's check actually needs."},
@@ -706,8 +709,8 @@ def p_validate_mcps(mcps: list) -> str:
 def p_build_trigger(args: dict) -> tuple:
     """(trigger dict, error string). Kind-specific validation with actionable errors."""
     kind = args.get("kind") or ""
-    if kind not in ("file", "web", "agent", "custom"):
-        return None, "kind must be one of: file, web, agent, custom."
+    if kind not in ("file", "web", "agent", "custom", "stream"):
+        return None, "kind must be one of: file, web, agent, custom, stream."
     poll_minutes = args.get("poll_minutes")
     poll_seconds = int(float(poll_minutes) * 60) if poll_minutes else TRIGGER_POLL_DEFAULTS.get(kind, 300)
     if kind == "file":
@@ -726,6 +729,10 @@ def p_build_trigger(args: dict) -> tuple:
         if mcp_err:
             return None, mcp_err
         source = {"kind": "agent", "check": args["check"].strip(), "model": "", "mcps": mcps, "poll_seconds": poll_seconds}
+    elif kind == "stream":
+        if not (args.get("url") or "").strip():
+            return None, "kind=stream needs url (the SSE feed to subscribe to)."
+        source = {"kind": "stream", "url": args["url"].strip(), "contains": (args.get("contains") or "").strip()}
     else:
         source = {"kind": "custom"}
     return {
@@ -747,6 +754,8 @@ def p_describe_trigger(t: dict) -> str:
         what = f"page {s.get('url')}" + (f" (watching for: {s.get('watch_for')})" if s.get("watch_for") else "")
     elif kind == "agent":
         what = f"agent check: {s.get('check')}" + (f" [tools: {', '.join(s.get('mcps') or [])}]" if s.get("mcps") else "")
+    elif kind == "stream":
+        what = f"live feed {s.get('url')}" + (f" (containing: {s.get('contains')})" if s.get("contains") else "")
     else:
         what = "custom push events"
     state = "ON" if t.get("enabled") else "off"

backend/apps/events/adapters/file_signal.py

@@ -0,0 +1,65 @@
+"""Instant wake for file triggers on macOS/BSD: a kqueue vnode watch on the
+directory fires the moment an entry is created, renamed, or deleted, and the
+callback just marks the trigger due so the normal diff-based poll runs
+immediately. The diff stays the source of truth (kqueue is only a wake
+signal), the heartbeat poll stays as the fallback, and platforms without
+kqueue simply keep polling. Content edits inside existing files don't touch
+the directory vnode, so those still ride the heartbeat."""
+
+import logging
+import os
+import select
+from typing import Callable, Optional
+
+logger = logging.getLogger(__name__)
+
+
+def start_file_signal(path: str, on_change: Callable[[], None]) -> Optional[Callable[[], None]]:
+    """Returns a stop() when a kqueue watch was installed, None when unsupported."""
+    if not hasattr(select, "kqueue"):
+        return None
+    target = os.path.expanduser(path.strip())
+    if not target or not os.path.exists(target):
+        return None
+    try:
+        import asyncio
+
+        loop = asyncio.get_running_loop()
+        open_flags = getattr(os, "O_EVTONLY", os.O_RDONLY)
+        fd = os.open(target, open_flags)
+        kq = select.kqueue()
+        fflags = (
+            select.KQ_NOTE_WRITE | select.KQ_NOTE_EXTEND | select.KQ_NOTE_ATTRIB
+            | select.KQ_NOTE_RENAME | select.KQ_NOTE_DELETE
+        )
+        event = select.kevent(fd, filter=select.KQ_FILTER_VNODE,
+                              flags=select.KQ_EV_ADD | select.KQ_EV_CLEAR, fflags=fflags)
+        kq.control([event], 0)
+
+        def p_on_readable() -> None:
+            try:
+                kq.control(None, 16, 0)  # drain whatever accumulated; one wake is enough
+            except OSError:
+                return
+            on_change()
+
+        loop.add_reader(kq.fileno(), p_on_readable)
+
+        def stop() -> None:
+            try:
+                loop.remove_reader(kq.fileno())
+            except Exception:
+                pass
+            try:
+                kq.close()
+            except OSError:
+                pass
+            try:
+                os.close(fd)
+            except OSError:
+                pass
+
+        return stop
+    except Exception as e:
+        logger.debug("file signal unavailable for %s: %s", path, e)
+        return None

backend/apps/events/adapters/stream_watch.py

@@ -0,0 +1,102 @@
+"""The held-open streaming tier: subscribe to a Server-Sent Events feed and
+turn its messages into trigger events as they arrive; nothing is transient
+because we read the source's own event log, not snapshots. Events batch
+(count or interval, whichever first) before hitting the dispatcher so a
+firehose can't write pending files per message; the contains-filter drops
+noise before it costs anything. Reconnects back off through the same failure
+bookkeeping polls use, so a dead feed surfaces instead of spinning."""
+
+import asyncio
+import hashlib
+import logging
+import time
+from typing import List, Optional
+
+from backend.apps.events.models import Event, EventTriggerConfig, StreamSource
+
+logger = logging.getLogger(__name__)
+
+BATCH_MAX_EVENTS = 25
+BATCH_MAX_SECONDS = 2.0
+RECONNECT_MAX_SECONDS = 300.0
+DATA_KEEP_CHARS = 2000
+
+
+def parse_sse_data(line_buffer: List[str]) -> Optional[str]:
+    """One SSE event's data payload from its buffered lines; None for keepalives/comments."""
+    data_lines = [ln[5:].lstrip() for ln in line_buffer if ln.startswith("data:")]
+    joined = "\n".join(data_lines).strip()
+    return joined or None
+
+
+def stream_event_from(data: str, contains: str) -> Optional[Event]:
+    if contains.strip() and contains.strip().lower() not in data.lower():
+        return None
+    digest = hashlib.sha256(data.encode()).hexdigest()[:16]
+    return Event(
+        source="stream",
+        event_type="stream_event",
+        summary=data.replace("\n", " ")[:200],
+        dedup_key=f"{digest}:{int(time.time())}",
+        payload={"data": data[:DATA_KEEP_CHARS]},
+    )
+
+
+async def run_stream_source(workflow_id: str, trigger: EventTriggerConfig, source: StreamSource) -> None:
+    """Long-lived task; cancelled by the reconciler when the trigger changes or dies."""
+    import httpx
+
+    from backend.apps.events import dispatcher, stores
+    from backend.apps.events.models import EventLogEntry
+
+    backoff = 1.0
+    while True:
+        batch: List[Event] = []
+        batch_started = time.monotonic()
+
+        async def flush_batch() -> None:
+            nonlocal batch, batch_started
+            if batch:
+                await dispatcher.ingest(workflow_id, trigger, batch)
+                batch = []
+            batch_started = time.monotonic()
+
+        try:
+            async with httpx.AsyncClient(timeout=httpx.Timeout(15.0, read=90.0)) as client:
+                # Some public feeds (Wikimedia among them) 403 requests with no User-Agent.
+                sse_headers = {"Accept": "text/event-stream", "User-Agent": "OpenSwarm-EventTrigger/1.0"}
+                async with client.stream("GET", source.url, headers=sse_headers) as resp:
+                    resp.raise_for_status()
+                    stores.clear_poll_failures(trigger.id)
+                    backoff = 1.0
+                    logger.info("stream connected for trigger %s: %s", trigger.id, source.url)
+                    line_buffer: List[str] = []
+                    async for line in resp.aiter_lines():
+                        if line:
+                            line_buffer.append(line)
+                            continue
+                        data = parse_sse_data(line_buffer)
+                        line_buffer = []
+                        if data:
+                            event = stream_event_from(data, source.contains)
+                            if event:
+                                batch.append(event)
+                        if batch and (len(batch) >= BATCH_MAX_EVENTS or time.monotonic() - batch_started >= BATCH_MAX_SECONDS):
+                            await flush_batch()
+        except asyncio.CancelledError:
+            raise
+        except Exception as e:
+            try:
+                await flush_batch()
+            except Exception:
+                pass
+            failures = stores.record_poll_failure(trigger.id, str(e)[:200])
+            try:
+                stores.append_log(workflow_id, EventLogEntry(
+                    trigger_id=trigger.id, kind="error",
+                    summary=f"Stream dropped: {str(e)[:160]} (failure {failures}; reconnecting in ~{int(backoff)}s)",
+                ))
+            except Exception:
+                pass
+            await asyncio.sleep(backoff)
+            backoff = min(backoff * 2, RECONNECT_MAX_SECONDS)

backend/apps/events/models.py

@@ -84,8 +84,18 @@ class CustomEventSource(BaseModel):
     kind: Literal["custom"] = "custom"
 
 
+class StreamSource(BaseModel):
+    """Held-open subscription to a Server-Sent Events feed: the source's own
+    event log, read live, so nothing is transient. Not polled; a long-lived
+    task owns the connection and reconnects with backoff."""
+    kind: Literal["stream"] = "stream"
+    url: str = ""
+    # Cheap server-side-of-us noise gate: only messages containing this substring become events. Empty = everything.
+    contains: str = ""
+
+
 EventSourceConfig = Annotated[
-    Union[FileWatchSource, WebWatchSource, AgentCheckSource, CustomEventSource],
+    Union[FileWatchSource, WebWatchSource, AgentCheckSource, CustomEventSource, StreamSource],
     Field(discriminator="kind"),
 ]
 

backend/apps/events/poll_loop.py

@@ -12,9 +12,11 @@ from typing import Awaitable, Callable, Dict, List, Optional, Set, Tuple
 
 from backend.apps.events import dispatcher, stores
 from backend.apps.events.adapters.agent_check import agent_check
+from backend.apps.events.adapters.file_signal import start_file_signal
 from backend.apps.events.adapters.file_watch import file_watch
+from backend.apps.events.adapters.stream_watch import run_stream_source
 from backend.apps.events.adapters.web_watch import web_watch
-from backend.apps.events.models import CustomEventSource, Event, EventLogEntry, EventTriggerConfig
+from backend.apps.events.models import CustomEventSource, Event, EventLogEntry, EventTriggerConfig, FileWatchSource, StreamSource
 from backend.apps.workflows.models import Workflow
 
 logger = logging.getLogger(__name__)
@@ -30,23 +32,36 @@ p_loop_task: Optional["asyncio.Task"] = None
 p_wake = asyncio.Event()
 p_next_poll: Dict[str, float] = {}
 p_inflight: Set[str] = set()
+# Held-open live sources (kqueue file signals, SSE streams): trigger_id -> (config signature, stopper).
+p_live_handles: Dict[str, Tuple[str, Callable[[], None]]] = {}
 
 
 def kick() -> None:
     p_wake.set()
 
 
+def live_source_count() -> int:
+    """How many held-open sources (file signals, streams) are currently running."""
+    return len(p_live_handles)
+
+
 def mark_due(trigger_id: str) -> None:
     """Schedule this trigger's next poll immediately (used with kick())."""
     p_next_poll[trigger_id] = 0.0
 
 
 def reset_state() -> None:
-    """Test seam: forget all per-trigger poll bookkeeping."""
+    """Test seam: forget all per-trigger poll bookkeeping and stop live sources."""
     global p_loop_task
     p_loop_task = None
     p_next_poll.clear()
     p_inflight.clear()
+    for _, stop in p_live_handles.values():
+        try:
+            stop()
+        except Exception:
+            pass
+    p_live_handles.clear()
 
 
 def p_live_triggers() -> List[Tuple[Workflow, EventTriggerConfig, float]]:
@@ -57,7 +72,7 @@ def p_live_triggers() -> List[Tuple[Workflow, EventTriggerConfig, float]]:
     for wf in storage.list_workflows():
         for trig in wf.event_triggers:
             source = trig.source
-            if not trig.enabled or isinstance(source, CustomEventSource) or source.kind not in ADAPTERS:
+            if not trig.enabled or isinstance(source, (CustomEventSource, StreamSource)) or source.kind not in ADAPTERS:
                 continue
             out.append((wf, trig, float(source.poll_seconds)))
     return out
@@ -81,7 +96,7 @@ async def p_poll_one(wf: Workflow, trigger: EventTriggerConfig) -> None:
         logger.warning("poll failed for trigger %s (%s): %s", trigger.id, trigger.source.kind, e)
         try:
             # Exponential backoff on repeated failures: a broken site/model can't burn quota at full cadence, and the log says so instead of dying silently.
-            failures = stores.record_poll_failure(trigger.id)
+            failures = stores.record_poll_failure(trigger.id, str(e))
             base = float(getattr(trigger.source, "poll_seconds", 300))
             backoff = min(base * (2 ** min(failures, 5)), 21600.0)
             p_next_poll[trigger.id] = time.monotonic() + backoff
@@ -96,9 +111,49 @@ async def p_poll_one(wf: Workflow, trigger: EventTriggerConfig) -> None:
         p_inflight.discard(trigger.id)
 
 
+def reconcile_live_sources() -> None:
+    """Start/stop held-open sources to match the current trigger set. File signals
+    make the diff poll instant; stream tasks own an SSE connection outright."""
+    from backend.apps.workflows import storage
+
+    want: Dict[str, Tuple[str, Workflow, EventTriggerConfig]] = {}
+    for wf in storage.list_workflows():
+        for trig in wf.event_triggers:
+            if not trig.enabled:
+                continue
+            if isinstance(trig.source, (FileWatchSource, StreamSource)):
+                want[trig.id] = (trig.source.model_dump_json(), wf, trig)
+    for trigger_id in list(p_live_handles.keys()):
+        signature, stop = p_live_handles[trigger_id]
+        if trigger_id not in want or want[trigger_id][0] != signature:
+            try:
+                stop()
+            except Exception:
+                pass
+            del p_live_handles[trigger_id]
+    for trigger_id, (signature, wf, trig) in want.items():
+        if trigger_id in p_live_handles:
+            continue
+        source = trig.source
+        if isinstance(source, FileWatchSource):
+            def p_on_change(tid: str = trigger_id) -> None:
+                mark_due(tid)
+                kick()
+            stop = start_file_signal(source.path, p_on_change)
+            if stop is not None:
+                p_live_handles[trigger_id] = (signature, stop)
+        elif isinstance(source, StreamSource) and source.url.strip():
+            task = asyncio.create_task(run_stream_source(wf.id, trig, source))
+            p_live_handles[trigger_id] = (signature, task.cancel)
+
+
 def tick() -> None:
     from backend.apps.workflows import storage
 
+    try:
+        reconcile_live_sources()
+    except Exception:
+        logger.exception("live-source reconcile error")
     # The global "pause all" switch holds event polling too; the cursor diff catches net changes at resume.
     if storage.get_paused():
         return

backend/apps/events/stores.py

@@ -5,6 +5,7 @@
 """
 
 import os
+import time
 from typing import Dict, List
 
 from typeguard import typechecked
@@ -78,15 +79,20 @@ def recent_fire_count(trigger_id: str, now_epoch: float, window_seconds: float =
 
 
 @typechecked
-def record_poll_failure(trigger_id: str) -> int:
+def record_poll_failure(trigger_id: str, error: str = "") -> int:
     """Returns the new consecutive-failure count."""
     path = os.path.join(HEALTH_DIR, f"{trigger_id}.json")
     raw = read_json_or_none(path) or {}
     count = int(raw.get("consecutive_failures") or 0) + 1
-    atomic_write_json(path, {"consecutive_failures": count})
+    atomic_write_json(path, {"consecutive_failures": count, "last_error": error[:300], "last_failure_epoch": time.time()})
     return count
 
 
+@typechecked
+def read_poll_health(trigger_id: str) -> Dict:
+    return read_json_or_none(os.path.join(HEALTH_DIR, f"{trigger_id}.json")) or {}
+
+
 @typechecked
 def clear_poll_failures(trigger_id: str) -> None:
     path = os.path.join(HEALTH_DIR, f"{trigger_id}.json")

backend/apps/workflows/workflows.py

@@ -771,6 +771,29 @@ async def get_workflow_audit(workflow_id: str, limit: int = 50):
     return {"entries": audit.read_tail(workflow_id, limit=limit)}
 
 
+@workflows.router.get("/triggers/attention")
+async def triggers_attention():
+    """Event triggers failing repeatedly (their watcher needs something from the user)."""
+    from backend.apps.events.stores import read_poll_health
+    items = []
+    for wf in storage.list_workflows():
+        for t in wf.event_triggers:
+            if not t.enabled:
+                continue
+            health = read_poll_health(t.id)
+            failures = int(health.get("consecutive_failures") or 0)
+            if failures >= 3:
+                items.append({
+                    "workflow_id": wf.id,
+                    "workflow_title": wf.title,
+                    "trigger_id": t.id,
+                    "kind": t.source.kind,
+                    "consecutive_failures": failures,
+                    "last_error": str(health.get("last_error") or ""),
+                })
+    return {"attention": items}
+
+
 @workflows.router.get("/{workflow_id}/events")
 async def get_workflow_events(workflow_id: str):
     """Event-trigger activity log, newest first ("saw X, skipped because Y, fired run Z")."""

backend/tests/test_event_stream_tier.py

@@ -0,0 +1,138 @@
+"""The always-observing tier: SSE parsing + filtering, the live-source
+reconciler's start/stop lifecycle (file signals and stream tasks follow the
+trigger set), the kqueue file signal firing on a real directory change, and
+the watcher-attention endpoint surfacing repeated failures.
+
+Run:
+    cd backend && .venv/bin/python -m pytest tests/test_event_stream_tier.py -v
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from backend.apps.events.models import EventTriggerConfig, FileWatchSource, StreamSource
+
+
+def p_run(coro):
+    return asyncio.new_event_loop().run_until_complete(coro)
+
+
+@pytest.fixture(autouse=True)
+def p_events_env(isolated_workflows_data, reset_scheduler_state, monkeypatch, tmp_path):
+    from backend.apps.events import dispatcher, poll_loop, stores
+
+    monkeypatch.setattr(stores, "EVENTS_DIR", str(tmp_path / "events"))
+    monkeypatch.setattr(stores, "CURSORS_DIR", str(tmp_path / "events" / "cursors"))
+    monkeypatch.setattr(stores, "PENDING_DIR", str(tmp_path / "events" / "pending"))
+    monkeypatch.setattr(stores, "LOGS_DIR", str(tmp_path / "events" / "logs"))
+    monkeypatch.setattr(stores, "FIRES_DIR", str(tmp_path / "events" / "fires"))
+    monkeypatch.setattr(stores, "HEALTH_DIR", str(tmp_path / "events" / "health"))
+    dispatcher.stop()
+    poll_loop.reset_state()
+    yield
+    dispatcher.stop()
+    poll_loop.reset_state()
+
+
+def test_sse_parse_and_filter():
+    from backend.apps.events.adapters.stream_watch import parse_sse_data, stream_event_from
+
+    assert parse_sse_data(["data: hello", "data: world"]) == "hello\nworld"
+    assert parse_sse_data([": keepalive comment"]) is None
+    assert parse_sse_data(["event: message"]) is None
+
+    assert stream_event_from('{"title": "Berlin Wall"}', "berlin") is not None
+    assert stream_event_from('{"title": "Paris"}', "berlin") is None
+    e = stream_event_from("x" * 5000, "")
+    assert len(e.summary) <= 200
+    assert len(e.payload["data"]) <= 2000
+
+
+def test_reconciler_starts_and_stops_live_sources(make_wf, monkeypatch, tmp_path):
+    from backend.apps.events import poll_loop
+    from backend.apps.workflows import storage
+
+    started: list[str] = []
+    stopped: list[str] = []
+
+    def p_fake_signal(path, on_change):
+        started.append(path)
+        return lambda: stopped.append(path)
+
+    async def p_fake_stream(workflow_id, trigger, source):
+        await asyncio.sleep(3600)
+
+    monkeypatch.setattr(poll_loop, "start_file_signal", p_fake_signal)
+    monkeypatch.setattr(poll_loop, "run_stream_source", p_fake_stream)
+
+    watch_dir = str(tmp_path / "sig")
+    file_trig = EventTriggerConfig(source=FileWatchSource(path=watch_dir))
+    stream_trig = EventTriggerConfig(source=StreamSource(url="https://feed.example/sse"))
+    wf = make_wf(event_triggers=[file_trig, stream_trig])
+    storage.save_workflow(wf)
+
+    async def scenario():
+        poll_loop.reconcile_live_sources()
+        assert started == [watch_dir]
+        assert poll_loop.live_source_count() == 2  # file signal + stream task
+
+        # Removing the triggers stops both handles.
+        live = storage.get_workflow(wf.id)
+        live.event_triggers = []
+        storage.save_workflow(live)
+        poll_loop.reconcile_live_sources()
+        assert poll_loop.live_source_count() == 0
+        assert stopped == [watch_dir]
+        await asyncio.sleep(0)
+
+    p_run(scenario())
+
+
+def test_kqueue_signal_fires_on_real_change(tmp_path):
+    from backend.apps.events.adapters.file_signal import start_file_signal
+
+    watch_dir = tmp_path / "instant"
+    watch_dir.mkdir()
+
+    async def scenario() -> bool:
+        fired = asyncio.Event()
+        stop = start_file_signal(str(watch_dir), fired.set)
+        if stop is None:
+            pytest.skip("kqueue unavailable on this platform")
+        try:
+            (watch_dir / "new.txt").write_text("x")
+            await asyncio.wait_for(fired.wait(), timeout=2.0)
+            return True
+        finally:
+            stop()
+
+    assert p_run(scenario()) is True
+
+
+def test_attention_endpoint_surfaces_repeat_failures(make_wf):
+    from backend.apps.events import stores
+    from backend.apps.workflows import storage
+    from backend.apps.workflows.workflows import triggers_attention
+
+    trig = EventTriggerConfig(source=StreamSource(url="https://dead.example/sse"))
+    healthy = EventTriggerConfig(source=FileWatchSource(path="/tmp/x"))
+    wf = make_wf(event_triggers=[trig, healthy])
+    storage.save_workflow(wf)
+
+    for _ in range(2):
+        stores.record_poll_failure(trig.id, "connect refused")
+    assert p_run(triggers_attention()) == {"attention": []}  # 2 failures = not yet
+
+    stores.record_poll_failure(trig.id, "connect refused")
+    res = p_run(triggers_attention())
+    assert len(res["attention"]) == 1
+    item = res["attention"][0]
+    assert item["trigger_id"] == trig.id
+    assert item["consecutive_failures"] == 3
+    assert "connect refused" in item["last_error"]
+
+    stores.clear_poll_failures(trig.id)
+    assert p_run(triggers_attention()) == {"attention": []}

frontend/src/app/components/overlays/TriggerHealthToast.tsx

@@ -0,0 +1,68 @@
+// Bottom-left nudge when an event watcher keeps failing (site changed, sign-in
+// needed, feed dead): names the workflow and jumps to its Event triggers panel,
+// where the activity feed says exactly why. A silently dead watcher is the
+// trust-killer this exists to prevent.
+
+import React from 'react';
+import Snackbar from '@mui/material/Snackbar';
+import Alert from '@mui/material/Alert';
+import Button from '@mui/material/Button';
+import IconButton from '@mui/material/IconButton';
+import CloseIcon from '@mui/icons-material/Close';
+import { useClaudeTokens } from '@/shared/styles/ThemeContext';
+import { useAppDispatch, useAppSelector } from '@/shared/hooks';
+import { hideTriggersHealthToast } from '@/shared/state/triggersHealthSlice';
+import { openWorkflowsApp } from '@/shared/state/dashboardLayoutSlice';
+
+export default function TriggerHealthToast() {
+  const c = useClaudeTokens();
+  const dispatch = useAppDispatch();
+  const open = useAppSelector((s) => s.triggersHealth.toastOpen);
+  const items = useAppSelector((s) => s.triggersHealth.items);
+  const first = items[0];
+
+  const onReview = React.useCallback(() => {
+    if (first) dispatch(openWorkflowsApp({ workflowId: first.workflow_id }));
+    dispatch(hideTriggersHealthToast());
+  }, [dispatch, first]);
+
+  const extra = items.length > 1 ? ` (and ${items.length - 1} more watcher${items.length > 2 ? 's' : ''})` : '';
+
+  return (
+    <Snackbar
+      open={open && !!first}
+      autoHideDuration={null}
+      onClose={(event, reason) => { if (reason !== 'clickaway') dispatch(hideTriggersHealthToast()); }}
+      anchorOrigin={{ vertical: 'bottom', horizontal: 'left' }}
+    >
+      <Alert
+        icon={false}
+        severity="warning"
+        sx={{
+          bgcolor: c.bg.surface,
+          color: c.text.primary,
+          border: `1px solid ${c.border.medium}`,
+          maxWidth: 440,
+          '& .MuiAlert-action': { alignItems: 'center', pt: 0 },
+        }}
+        action={
+          <>
+            <Button size="small" onClick={onReview} sx={{ color: c.accent.primary, fontWeight: 700 }}>
+              Review
+            </Button>
+            <IconButton
+              size="small"
+              aria-label="Dismiss"
+              onClick={() => dispatch(hideTriggersHealthToast())}
+              sx={{ color: c.text.muted, ml: 0.25, '&:hover': { color: c.text.primary } }}
+            >
+              <CloseIcon sx={{ fontSize: 16 }} />
+            </IconButton>
+          </>
+        }
+      >
+        {first ? `A watcher on "${first.workflow_title}" keeps failing (${first.consecutive_failures} in a row)${extra}; it may need something from you.` : ''}
+      </Alert>
+    </Snackbar>
+  );
+}

frontend/src/app/pages/Dashboard/canvas/DashboardOverlays.tsx

@@ -8,6 +8,7 @@ import WorkflowRunningToast from '@/app/pages/Workflows/WorkflowRunningToast';
 import MissedRunsToast from '@/app/pages/Workflows/MissedRunsToast';
 import ProviderHealthToast from '@/app/components/overlays/ProviderHealthToast';
 import PatternOfferToast from '@/app/components/overlays/PatternOfferToast';
+import TriggerHealthToast from '@/app/components/overlays/TriggerHealthToast';
 import type { AgentSession } from '@/shared/state/agentsSlice';
 import type {
   CardPosition,
@@ -160,6 +161,9 @@ const DashboardOverlays: React.FC<DashboardOverlaysProps> = ({
 
       {/* Mined-pattern offer: "you do this a lot, want a workflow?" */}
       <PatternOfferToast />
+
+      {/* A watcher that keeps failing probably needs something from the user */}
+      <TriggerHealthToast />
     </>
   );
 };

frontend/src/app/pages/Dashboard/hooks/lifecycle/useDashboardLifecycle.ts

@@ -27,6 +27,7 @@ import { fetchWorkflows, fetchAllRuns, fetchActiveRuns } from '@/shared/state/wo
 import { fetchMissedRuns } from '@/shared/state/missedRunsSlice';
 import { fetchProviderHealth } from '@/shared/state/subscriptionsSlice';
 import { fetchPatternSuggestions } from '@/shared/state/patternsSlice';
+import { fetchTriggersAttention } from '@/shared/state/triggersHealthSlice';
 import { dashboardWs } from '@/shared/ws/WebSocketManager';
 import { initBrowserCommandHandler } from '@/shared/browserCommandHandler';
 import { getKeepAliveBrowserIds } from '@/shared/browserFocus';
@@ -92,7 +93,8 @@ export function useDashboardLifecycle({
     }, 12_000);
     // Mined-pattern offers ride the same once-per-launch gate, staggered after the health pill so nudges don't stack; the ws patterns:suggestions_updated case covers a miner pass finishing later.
     const tPatterns = setTimeout(() => { dispatch(fetchPatternSuggestions()); }, 20_000);
-    return () => { clearTimeout(t); clearTimeout(tPatterns); };
+    const tTriggers = setTimeout(() => { dispatch(fetchTriggersAttention()); }, 30_000);
+    return () => { clearTimeout(t); clearTimeout(tPatterns); clearTimeout(tTriggers); };
   }, [isActive, dispatch]);
 
   // Track dashboard engagement time

frontend/src/app/pages/Workflows/app/EventTriggerRow.tsx

@@ -16,6 +16,7 @@ const KIND_LABELS: Record<string, string> = {
   web: 'Web page watch',
   agent: 'Agent check',
   custom: 'Custom (push)',
+  stream: 'Live feed (SSE)',
 };
 
 interface RowProps {
@@ -137,6 +138,35 @@ const EventTriggerRow: React.FC<RowProps> = ({ workflow, trigger, onMutate, onRe
         </div>
       )}
 
+      {src.kind === 'stream' && (
+        <div style={{ display: 'flex', gap: 8, marginBottom: 8 }}>
+          <div style={{ flex: 1 }}>
+            <span style={labelStyle}>Feed URL (Server-Sent Events)</span>
+            <input
+              style={fieldStyle}
+              defaultValue={src.url}
+              placeholder="https://stream.example.com/events"
+              onBlur={(e) => {
+                const url = e.target.value.trim();
+                if (url !== src.url) onMutate((x) => ({ ...x, source: { ...src, url } }));
+              }}
+            />
+          </div>
+          <div style={{ width: 150, flex: 'none' }}>
+            <span style={labelStyle}>Only lines containing</span>
+            <input
+              style={fieldStyle}
+              defaultValue={src.contains}
+              placeholder="(everything)"
+              onBlur={(e) => {
+                const contains = e.target.value.trim();
+                if (contains !== src.contains) onMutate((x) => ({ ...x, source: { ...src, contains } }));
+              }}
+            />
+          </div>
+        </div>
+      )}
+
       {src.kind === 'custom' && (
         <div style={{ marginBottom: 8 }}>
           <span style={labelStyle}>Anything can push events here (scripts, webhooks, Shortcuts):</span>

frontend/src/app/pages/Workflows/app/EventTriggersCard.tsx

@@ -12,19 +12,21 @@ import EventTriggerRow from './EventTriggerRow';
 import { useWC, FONT_SERIF } from './uiKit';
 import { useWorkflowPatch } from './useWorkflowPatch';
 
-type TriggerKind = 'file' | 'web' | 'agent' | 'custom';
+type TriggerKind = 'file' | 'web' | 'agent' | 'custom' | 'stream';
 
 const ADD_CHOICES: Array<[TriggerKind, string]> = [
   ['file', '+ Folder'],
   ['web', '+ Page'],
   ['agent', '+ Agent check'],
   ['custom', '+ Custom'],
+  ['stream', '+ Live feed'],
 ];
 
 function newSource(kind: TriggerKind): EventSourceConfig {
   if (kind === 'file') return { kind: 'file', path: '', poll_seconds: 15 };
   if (kind === 'web') return { kind: 'web', url: '', watch_for: '', poll_seconds: 300 };
   if (kind === 'agent') return { kind: 'agent', check: '', model: '', poll_seconds: 900 };
+  if (kind === 'stream') return { kind: 'stream', url: '', contains: '' };
   return { kind: 'custom' };
 }
 

frontend/src/shared/state/store.ts

@@ -18,6 +18,7 @@ import subscriptionsReducer from './subscriptionsSlice';
 import workflowsReducer from './workflowsSlice';
 import missedRunsReducer from './missedRunsSlice';
 import patternsReducer from './patternsSlice';
+import triggersHealthReducer from './triggersHealthSlice';
 import onboardingProgressReducer from '@/shared/state/onboardingProgressSlice';
 import onboardingV3Reducer from '@/shared/state/onboardingV3Slice';
 
@@ -42,6 +43,7 @@ export const store = configureStore({
     workflows: workflowsReducer,
     missedRuns: missedRunsReducer,
     patterns: patternsReducer,
+    triggersHealth: triggersHealthReducer,
     onboardingProgress: onboardingProgressReducer,
     onboardingV3: onboardingV3Reducer,
   },

frontend/src/shared/state/triggersHealthSlice.ts

@@ -0,0 +1,45 @@
+import { createSlice, createAsyncThunk } from '@reduxjs/toolkit';
+import { API_BASE } from '@/shared/config';
+
+export interface TriggerAttentionItem {
+  workflow_id: string;
+  workflow_title: string;
+  trigger_id: string;
+  kind: string;
+  consecutive_failures: number;
+  last_error: string;
+}
+
+interface TriggersHealthState {
+  items: TriggerAttentionItem[];
+  toastOpen: boolean;
+}
+
+const initialState: TriggersHealthState = { items: [], toastOpen: false };
+
+export const fetchTriggersAttention = createAsyncThunk(
+  'triggersHealth/fetch',
+  async (): Promise<{ attention: TriggerAttentionItem[] }> => {
+    const r = await fetch(`${API_BASE}/workflows/triggers/attention`);
+    return (await r.json()) as { attention: TriggerAttentionItem[] };
+  },
+);
+
+const triggersHealthSlice = createSlice({
+  name: 'triggersHealth',
+  initialState,
+  reducers: {
+    hideTriggersHealthToast(state) {
+      state.toastOpen = false;
+    },
+  },
+  extraReducers: (builder) => {
+    builder.addCase(fetchTriggersAttention.fulfilled, (state, action) => {
+      state.items = action.payload.attention ?? [];
+      state.toastOpen = state.items.length > 0;
+    });
+  },
+});
+
+export const { hideTriggersHealthToast } = triggersHealthSlice.actions;
+export default triggersHealthSlice.reducer;

frontend/src/shared/state/workflowsSlice.ts

@@ -56,7 +56,15 @@ export interface CustomEventSource {
   kind: 'custom';
 }
 
-export type EventSourceConfig = FileWatchSource | WebWatchSource | AgentCheckSource | CustomEventSource;
+export interface StreamSource {
+  /** Held-open SSE subscription: the source's own event log, read live. */
+  kind: 'stream';
+  url: string;
+  /** Only messages containing this substring become events; empty = everything. */
+  contains: string;
+}
+
+export type EventSourceConfig = FileWatchSource | WebWatchSource | AgentCheckSource | CustomEventSource | StreamSource;
 
 export interface EventTriggerConfig {
   id: string;
@@ -177,7 +185,7 @@ export interface WorkflowRun {
   session_id: string | null;
   error: string | null;
   cost_usd: number;
-  triggered_by: 'schedule' | 'manual' | 'retry';
+  triggered_by: 'schedule' | 'manual' | 'retry' | 'event';
   /** Live "what's the agent doing" subtitle while status is 'running'. */
   last_tool_label?: string | null;
   /** Currently-executing 0-based step index while status is 'running';