[eric] events: durable rate cap, failure backoff, poll jitter, observe-only checks, Watching label

backend/apps/events/adapters/agent_check.py

@@ -25,6 +25,8 @@ def build_check_prompt(check: str, state: str) -> str:
         "You are an unattended event checker. Determine whether this event has occurred "
         f"since the last check: {check.strip()}\n\n"
         f"Previous check state: {state or 'none; this is the baseline check'}.\n\n"
+        "Observe without consuming: prefer listings, counts, and previews; avoid opening or acting on "
+        "items in ways that change their state (marking read/seen, dismissing) when your tools allow it.\n"
         "Use your tools as needed, then END your reply in EXACTLY this format (as the final lines):\n"
         "EVENT: <one factual line describing what happened>\n"
         "STATE: <one line capturing what you observed, to compare against next time>\n"

backend/apps/events/dispatcher.py

@@ -24,7 +24,6 @@ MAX_BUFFERED_EVENTS = 200
 p_buffers: Dict[str, List[Event]] = {}
 p_workflow_of: Dict[str, str] = {}
 p_flush_tasks: Dict[str, "asyncio.Task"] = {}
-p_fire_times: Dict[str, List[float]] = {}
 
 
 @typechecked
@@ -80,10 +79,8 @@ async def ingest(workflow_id: str, trigger: EventTriggerConfig, events: List[Eve
 
 @typechecked
 def p_recent_fires(trigger_id: str) -> int:
-    cutoff = time.monotonic() - 3600.0
-    times = [t for t in p_fire_times.get(trigger_id, []) if t >= cutoff]
-    p_fire_times[trigger_id] = times
-    return len(times)
+    # Persisted, not in-memory: the cap must survive restarts (the one soak anomaly rode exactly this gap).
+    return stores.recent_fire_count(trigger_id, time.time())
 
 
 @typechecked
@@ -133,7 +130,7 @@ async def p_flush(trigger_id: str) -> None:
             p_log(workflow_id, trigger_id, "skipped", f"{len(snapshot)} event(s) did not match: \"{trigger.predicate.strip()[:80]}\"")
             return
     p_consume(trigger_id, len(snapshot))
-    p_fire_times.setdefault(trigger_id, []).append(time.monotonic())
+    stores.record_fire(trigger_id, time.time())
     asyncio.create_task(p_run_and_log(wf, trigger, snapshot))
 
 
@@ -175,4 +172,3 @@ def stop() -> None:
     p_flush_tasks.clear()
     p_buffers.clear()
     p_workflow_of.clear()
-    p_fire_times.clear()

backend/apps/events/poll_loop.py

@@ -6,6 +6,7 @@ or stall its neighbors."""
 
 import asyncio
 import logging
+import random
 import time
 from typing import Awaitable, Callable, Dict, List, Optional, Set, Tuple
 
@@ -73,14 +74,21 @@ async def p_poll_one(wf: Workflow, trigger: EventTriggerConfig) -> None:
         else:
             events, new_cursor = await fetch(trigger.source, cursor)
         stores.save_cursor(trigger.id, new_cursor)
+        stores.clear_poll_failures(trigger.id)
         if events:
             await dispatcher.ingest(workflow_id, trigger, events)
     except Exception as e:
         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)
+            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
+            note = f" (failure {failures} in a row; next try in ~{int(backoff / 60) or 1}m)" if failures >= 2 else ""
             stores.append_log(workflow_id, EventLogEntry(
                 trigger_id=trigger.id, kind="error",
-                summary=f"Poll failed: {str(e)[:200]}",
+                summary=f"Poll failed: {str(e)[:180]}{note}",
             ))
         except Exception:
             pass
@@ -97,7 +105,8 @@ def tick() -> None:
     now = time.monotonic()
     for wf, trig, poll_seconds in p_live_triggers():
         if p_next_poll.get(trig.id, 0.0) <= now and trig.id not in p_inflight:
-            p_next_poll[trig.id] = now + poll_seconds
+            # Jitter so logged-in polls aren't metronomic (a bot tell) and many triggers spread out.
+            p_next_poll[trig.id] = now + poll_seconds * random.uniform(0.9, 1.1)
             p_inflight.add(trig.id)
             asyncio.create_task(p_poll_one(wf, trig))
 

backend/apps/events/stores.py

@@ -17,8 +17,11 @@ EVENTS_DIR = os.path.join(DATA_ROOT, "events")
 CURSORS_DIR = os.path.join(EVENTS_DIR, "cursors")
 PENDING_DIR = os.path.join(EVENTS_DIR, "pending")
 LOGS_DIR = os.path.join(EVENTS_DIR, "logs")
+FIRES_DIR = os.path.join(EVENTS_DIR, "fires")
+HEALTH_DIR = os.path.join(EVENTS_DIR, "health")
 
 LOG_ENTRIES_MAX = 200
+FIRES_MAX = 100
 
 
 @typechecked
@@ -55,6 +58,42 @@ def save_pending(trigger_id: str, events: List[Event]) -> None:
     atomic_write_json(path, [e.model_dump(mode="json") for e in events])
 
 
+@typechecked
+def record_fire(trigger_id: str, when_epoch: float) -> None:
+    """Persisted so the rate cap survives restarts (and any process confusion)."""
+    path = os.path.join(FIRES_DIR, f"{trigger_id}.json")
+    raw = read_json_or_none(path)
+    arr = [float(x) for x in raw] if isinstance(raw, list) else []
+    arr.append(float(when_epoch))
+    atomic_write_json(path, arr[-FIRES_MAX:])
+
+
+@typechecked
+def recent_fire_count(trigger_id: str, now_epoch: float, window_seconds: float = 3600.0) -> int:
+    raw = read_json_or_none(os.path.join(FIRES_DIR, f"{trigger_id}.json"))
+    if not isinstance(raw, list):
+        return 0
+    cutoff = now_epoch - window_seconds
+    return sum(1 for x in raw if isinstance(x, (int, float)) and float(x) >= cutoff)
+
+
+@typechecked
+def record_poll_failure(trigger_id: 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})
+    return count
+
+
+@typechecked
+def clear_poll_failures(trigger_id: str) -> None:
+    path = os.path.join(HEALTH_DIR, f"{trigger_id}.json")
+    if os.path.exists(path):
+        os.remove(path)
+
+
 @typechecked
 def append_log(workflow_id: str, entry: EventLogEntry) -> None:
     path = os.path.join(LOGS_DIR, f"{workflow_id}.json")
@@ -87,7 +126,7 @@ def sweep_stale_state(live_trigger_ids: List[str], live_workflow_ids: List[str])
     accumulating orphans forever."""
     keep_triggers = set(live_trigger_ids)
     keep_workflows = set(live_workflow_ids)
-    for directory, keep in ((CURSORS_DIR, keep_triggers), (PENDING_DIR, keep_triggers), (LOGS_DIR, keep_workflows)):
+    for directory, keep in ((CURSORS_DIR, keep_triggers), (PENDING_DIR, keep_triggers), (FIRES_DIR, keep_triggers), (HEALTH_DIR, keep_triggers), (LOGS_DIR, keep_workflows)):
         if not os.path.isdir(directory):
             continue
         for fname in os.listdir(directory):

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

@@ -8,7 +8,7 @@ import AgentChat from '@/app/pages/AgentChat/AgentChat';
 import InlineEditableTitle from '@/app/components/InlineEditableTitle';
 import { Typewriter } from '@/app/components/feedback/Animated';
 import { useWC, colorForWorkflow, statusChip } from './uiKit';
-import { isRunning, runContextChip } from './model';
+import { isRunning, runContextChip, hasLiveTriggers } from './model';
 import { useEditAgentSession } from './useEditAgentSession';
 import { useWorkflowPatch } from './useWorkflowPatch';
 import ScheduleCard from './ScheduleCard';
@@ -44,8 +44,9 @@ const DetailView: React.FC<{ workflowId: string; nav: AppNav }> = ({ workflowId
 
   const running = isRunning(workflow, active);
   const enabled = isScheduleActive(workflow.schedule);
-  const status = running ? 'running' : enabled ? 'success' : 'paused';
-  const statusText = running ? 'Running' : enabled ? 'Active' : 'Paused';
+  const watching = !enabled && hasLiveTriggers(workflow);
+  const status = running ? 'running' : enabled || watching ? 'success' : 'paused';
+  const statusText = running ? 'Running' : enabled ? 'Active' : watching ? 'Watching' : 'Paused';
 
   const runNow = () => {
     if (running) return;

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

@@ -3,6 +3,7 @@ import type { CSSProperties } from 'react';
 import { useAppDispatch, useAppSelector } from '@/shared/hooks';
 import { deleteWorkflow } from '@/shared/state/workflowsSlice';
 import { isScheduleActive, describeSchedule } from '@/app/pages/Workflows/scheduleUtils';
+import { hasLiveTriggers } from './model';
 import ShareButton from '@/app/components/share/ShareButton';
 import { colorForWorkflow, useWC } from './uiKit';
 import WorkflowTitle from './WorkflowTitle';
@@ -90,6 +91,7 @@ const LeftRail: React.FC<{ nav: AppNav }> = ({ nav }) => {
       <div style={{ flex: 1, overflowY: 'auto', padding: '0 8px', minHeight: 0 }}>
         {filtered.map((w) => {
           const active = isScheduleActive(w.schedule);
+          const watching = !active && hasLiveTriggers(w);
           const isSel = nav.mode === 'detail' && w.id === nav.selectedId;
           return (
             <div
@@ -99,13 +101,13 @@ const LeftRail: React.FC<{ nav: AppNav }> = ({ nav }) => {
               onMouseLeave={() => setHovered((h) => (h === w.id ? null : h))}
               style={{ display: 'flex', alignItems: 'center', gap: 9, padding: '5px 9px', borderRadius: 8, cursor: 'pointer', background: isSel ? WC.selBg : 'transparent' }}
             >
-              <div style={{ width: 8, height: 8, borderRadius: '50%', flex: 'none', background: colorForWorkflow(w), opacity: active ? 1 : 0.35 }} />
+              <div style={{ width: 8, height: 8, borderRadius: '50%', flex: 'none', background: colorForWorkflow(w), opacity: active || watching ? 1 : 0.35 }} />
               <div style={{ minWidth: 0, flex: 1 }}>
                 <WorkflowTitle value={w.title} animate={w.auto_named !== false}>
                   {(t) => <div style={{ fontSize: 13.5, fontWeight: 600, color: active ? WC.ink : WC.muted, whiteSpace: 'nowrap', overflow: 'hidden', textOverflow: 'ellipsis' }}>{t}</div>}
                 </WorkflowTitle>
                 <div style={{ fontSize: 11, color: WC.muted2, marginTop: 1, whiteSpace: 'nowrap', overflow: 'hidden', textOverflow: 'ellipsis' }}>
-                  {active ? describeSchedule(w.schedule) : 'Paused'}
+                  {active ? describeSchedule(w.schedule) : watching ? 'Watching for events' : 'Paused'}
                 </div>
               </div>
               {/* Faded rather than unmounted on hover-out: ShareButton owns the modal's open state, so unmounting it would close the modal the moment the pointer left the row for the dialog. Also keeps the row from reflowing on hover. */}

frontend/src/app/pages/Workflows/app/model.ts

@@ -68,9 +68,14 @@ export function relativeDayLabel(date: Date, now = new Date()): string {
   return date.toLocaleDateString([], { weekday: 'short', month: 'short', day: 'numeric' });
 }
 
+// A workflow with an enabled event trigger is live even with its schedule off; "Paused" would lie.
+export function hasLiveTriggers(wf: Workflow): boolean {
+  return (wf.event_triggers ?? []).some((t) => t.enabled);
+}
+
 // `next` is the backend-computed next_run_at (authoritative), not a JS reimpl.
 export function nextRunText(wf: Workflow, next: Date | null): string {
-  if (!isScheduleActive(wf.schedule)) return 'Paused';
+  if (!isScheduleActive(wf.schedule)) return hasLiveTriggers(wf) ? 'On event' : 'Paused';
   if (!next) return 'None scheduled';
   return `${relativeDayLabel(next)} at ${clockOf(next)}`;
 }