diff --git a/backend/apps/events/adapters/agent_check.py b/backend/apps/events/adapters/agent_check.py index 0b0554f4..1739aa5b 100644 --- a/backend/apps/events/adapters/agent_check.py +++ b/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: \n" "STATE: \n" diff --git a/backend/apps/events/dispatcher.py b/backend/apps/events/dispatcher.py index 518e0c83..8a56ba63 100644 --- a/backend/apps/events/dispatcher.py +++ b/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() diff --git a/backend/apps/events/poll_loop.py b/backend/apps/events/poll_loop.py index b154b328..9b7695b4 100644 --- a/backend/apps/events/poll_loop.py +++ b/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)) diff --git a/backend/apps/events/stores.py b/backend/apps/events/stores.py index 169746ab..3207ec61 100644 --- a/backend/apps/events/stores.py +++ b/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): diff --git a/frontend/src/app/pages/Workflows/app/DetailView.tsx b/frontend/src/app/pages/Workflows/app/DetailView.tsx index f102d7f4..8bbc6e74 100644 --- a/frontend/src/app/pages/Workflows/app/DetailView.tsx +++ b/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; diff --git a/frontend/src/app/pages/Workflows/app/LeftRail.tsx b/frontend/src/app/pages/Workflows/app/LeftRail.tsx index 11625702..782bd052 100644 --- a/frontend/src/app/pages/Workflows/app/LeftRail.tsx +++ b/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 }) => {
{filtered.map((w) => { const active = isScheduleActive(w.schedule); + const watching = !active && hasLiveTriggers(w); const isSel = nav.mode === 'detail' && w.id === nav.selectedId; return (
= ({ 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' }} > -
+
{(t) =>
{t}
}
- {active ? describeSchedule(w.schedule) : 'Paused'} + {active ? describeSchedule(w.schedule) : watching ? 'Watching for events' : 'Paused'}
{/* 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. */} diff --git a/frontend/src/app/pages/Workflows/app/model.ts b/frontend/src/app/pages/Workflows/app/model.ts index d9662d35..bc46f668 100644 --- a/frontend/src/app/pages/Workflows/app/model.ts +++ b/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)}`; }