From 52056c191195fb879ab80dc2c2dad31c91cd8306 Mon Sep 17 00:00:00 2001 From: ciregenz Date: Tue, 28 Jul 2026 13:12:03 -0700 Subject: [PATCH] [eric] events: universal sources (agent-check any condition, custom push via /api/events/ingest) --- backend/apps/events/adapters/agent_check.py | 136 ++++++++++++++ .../apps/events/{ => adapters}/file_watch.py | 0 .../apps/events/{ => adapters}/web_watch.py | 0 backend/apps/events/events.py | 72 ++++++++ backend/apps/events/models.py | 27 ++- backend/apps/events/poll_loop.py | 26 ++- backend/main.py | 3 +- backend/tests/test_event_triggers.py | 6 +- backend/tests/test_event_universal.py | 160 ++++++++++++++++ .../pages/Workflows/app/EventTriggerRow.tsx | 169 +++++++++++++++++ .../pages/Workflows/app/EventTriggersCard.tsx | 173 +++++------------- frontend/src/shared/state/workflowsSlice.ts | 16 +- 12 files changed, 644 insertions(+), 144 deletions(-) create mode 100644 backend/apps/events/adapters/agent_check.py rename backend/apps/events/{ => adapters}/file_watch.py (100%) rename backend/apps/events/{ => adapters}/web_watch.py (100%) create mode 100644 backend/apps/events/events.py create mode 100644 backend/tests/test_event_universal.py create mode 100644 frontend/src/app/pages/Workflows/app/EventTriggerRow.tsx diff --git a/backend/apps/events/adapters/agent_check.py b/backend/apps/events/adapters/agent_check.py new file mode 100644 index 00000000..fa8e99d1 --- /dev/null +++ b/backend/apps/events/adapters/agent_check.py @@ -0,0 +1,136 @@ +"""The universal poll adapter: a real agent session verifies ANY +natural-language condition on an interval, with the same tools, MCP gate, and +admission bounds as a normal chat turn. Contract: the agent ends its reply +with EVENT:/NO_EVENT + STATE: lines; state round-trips through the cursor so +"since the last check" means something. Check sessions are plumbing, not chat +history: they're deleted after each check so they can't pollute history or +the pattern miner.""" + +import asyncio +import hashlib +import time +from typing import Dict, List, Optional, Tuple + +from typeguard import typechecked + +from backend.apps.events.models import AgentCheckSource, Event + +CHECK_TIMEOUT_S = 240.0 + + +@typechecked +def build_check_prompt(check: str, state: str) -> str: + return ( + "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" + "Use your tools as needed, then END your reply in EXACTLY this format (as the final lines):\n" + "EVENT: \n" + "STATE: \n" + "If nothing new happened, instead end with:\n" + "NO_EVENT\n" + "STATE: \n" + "On the baseline check (no previous state) always reply NO_EVENT and just record STATE." + ) + + +@typechecked +def parse_check_reply(text: str) -> Tuple[Optional[str], str]: + """(event line or None, state). Last occurrence wins so earlier prose echoing + the format can't fake a verdict. Raises when the contract is missing entirely.""" + event_line: Optional[str] = None + state = "" + saw_verdict = False + for line in text.splitlines(): + s = line.strip() + if s.upper().startswith("EVENT:"): + event_line = s[len("EVENT:"):].strip() + saw_verdict = True + elif s.upper() == "NO_EVENT": + event_line = None + saw_verdict = True + elif s.upper().startswith("STATE:"): + state = s[len("STATE:"):].strip() + if not saw_verdict: + raise ValueError("check agent returned no EVENT/NO_EVENT verdict") + return event_line, state + + +async def p_await_reply(session_id: str) -> str: + from backend.apps.agents.agent_manager import agent_manager + + deadline = time.monotonic() + CHECK_TIMEOUT_S + while time.monotonic() < deadline: + sess = agent_manager.sessions.get(session_id) + if sess is None or getattr(sess, "status", None) in ("completed", "error", "stopped"): + break + await asyncio.sleep(0.5) + sess = agent_manager.sessions.get(session_id) + if sess is None: + raise RuntimeError("check session vanished") + status = getattr(sess, "status", None) + if status == "running": + try: + await agent_manager.stop_agent(session_id) + except Exception: + pass + raise RuntimeError(f"check agent timed out after {int(CHECK_TIMEOUT_S)}s") + if status != "completed": + raise RuntimeError(f"check agent ended with status {status}") + for m in reversed(getattr(sess, "messages", []) or []): + if getattr(m, "role", None) == "assistant" and isinstance(getattr(m, "content", None), str) and m.content.strip(): + return m.content + raise RuntimeError("check agent produced no reply") + + +async def run_check_turn(model: str, prompt: str) -> str: + """One ephemeral agent turn; the session file is deleted afterward.""" + from backend.apps.agents.agent_manager import agent_manager + from backend.apps.agents.core.models import AgentConfig + from backend.apps.agents.manager.session.session_store import delete_session_file + + session = await agent_manager.launch_agent(AgentConfig(name="Event check", model=model, mode="agent")) + try: + await agent_manager.send_message(session.id, prompt) + return await p_await_reply(session.id) + finally: + try: + await agent_manager.close_session(session.id) + except Exception: + pass + try: + delete_session_file(session.id) + except Exception: + pass + + +@typechecked +async def agent_check(source: AgentCheckSource, cursor: Dict) -> Tuple[List[Event], Dict]: + check = source.check.strip() + if not check: + return [], cursor + from backend.apps.settings.settings import load_settings + + baselined = bool(cursor.get("baselined")) and cursor.get("check") == check + prev_state = str(cursor.get("state") or "") if baselined else "" + model = source.model.strip() or (getattr(load_settings(), "default_model", None) or "sonnet") + reply = await run_check_turn(model, build_check_prompt(check, prev_state)) + event_line, state = parse_check_reply(reply) + new_cursor: Dict = {"baselined": True, "check": check, "state": state or prev_state} + if cursor.get("last_event_digest"): + new_cursor["last_event_digest"] = cursor["last_event_digest"] + # Baseline never fires, even if the model ignores the instruction. + if event_line is None or not event_line.strip() or not baselined: + return [], new_cursor + digest = hashlib.sha256(event_line.strip().encode()).hexdigest()[:16] + if cursor.get("last_event_digest") == digest: + # The agent re-reported the identical event (state parroting); once is enough. + return [], new_cursor + new_cursor["last_event_digest"] = digest + return [Event( + source="agent", + event_type="check_event", + summary=event_line.strip()[:300], + dedup_key=digest, + payload={"check": check, "state": state}, + )], new_cursor diff --git a/backend/apps/events/file_watch.py b/backend/apps/events/adapters/file_watch.py similarity index 100% rename from backend/apps/events/file_watch.py rename to backend/apps/events/adapters/file_watch.py diff --git a/backend/apps/events/web_watch.py b/backend/apps/events/adapters/web_watch.py similarity index 100% rename from backend/apps/events/web_watch.py rename to backend/apps/events/adapters/web_watch.py diff --git a/backend/apps/events/events.py b/backend/apps/events/events.py new file mode 100644 index 00000000..deac0f63 --- /dev/null +++ b/backend/apps/events/events.py @@ -0,0 +1,72 @@ +"""Routes for the event engine. POST /api/events/ingest is the universal push +entry: any script, webhook forwarder, macOS Shortcut, or MCP can feed a +workflow's custom trigger. Ingested events ride the SAME dispatcher path as +polled ones, so coalescing, the predicate, and the rate cap all still apply. +The engine's lifecycle itself rides workflows_lifespan; this SubApp is +routes-only. Auth: the per-install bearer token gates this like every other +localhost API route.""" + +from contextlib import asynccontextmanager +from typing import Dict +from uuid import uuid4 + +from fastapi import HTTPException +from pydantic import BaseModel, Field + +from backend.apps.events import dispatcher, stores +from backend.apps.events.models import Event +from backend.config.Apps import SubApp + + +@asynccontextmanager +async def events_lifespan(): + yield + + +events = SubApp("events", events_lifespan) + +MAX_SEEN_KEYS = 300 + + +class IngestBody(BaseModel): + workflow_id: str + trigger_id: str + summary: str + event_type: str = "custom" + # Same key twice = delivered once; lets webhook retries stay idempotent. + dedup_key: str = "" + payload: Dict = Field(default_factory=dict) + + +@events.router.post("/ingest") +async def ingest_event(body: IngestBody): + from backend.apps.workflows import storage + + wf = storage.get_workflow(body.workflow_id) + if wf is None or wf.deleted_at is not None: + raise HTTPException(status_code=404, detail="Workflow not found") + trigger = next((t for t in wf.event_triggers if t.id == body.trigger_id), None) + if trigger is None: + raise HTTPException(status_code=404, detail="Trigger not found") + if trigger.source.kind != "custom": + raise HTTPException(status_code=409, detail="Trigger is not a custom (ingest) source") + if not trigger.enabled: + raise HTTPException(status_code=409, detail="Trigger is disabled") + summary = body.summary.strip()[:300] + if not summary: + raise HTTPException(status_code=400, detail="summary is required") + dedup_key = body.dedup_key.strip() or uuid4().hex + cursor = stores.load_cursor(trigger.id) + seen = [str(k) for k in (cursor.get("seen") or [])] + if dedup_key in seen: + return {"ok": True, "queued": 0, "deduped": True} + seen.append(dedup_key) + stores.save_cursor(trigger.id, {"seen": seen[-MAX_SEEN_KEYS:]}) + await dispatcher.ingest(wf.id, trigger, [Event( + source="custom", + event_type=(body.event_type.strip() or "custom")[:60], + summary=summary, + dedup_key=dedup_key, + payload=body.payload, + )]) + return {"ok": True, "queued": 1, "deduped": False} diff --git a/backend/apps/events/models.py b/backend/apps/events/models.py index b97f3c80..f99e6b11 100644 --- a/backend/apps/events/models.py +++ b/backend/apps/events/models.py @@ -52,8 +52,33 @@ class WebWatchSource(BaseModel): return max(60, min(v, 86400)) +class AgentCheckSource(BaseModel): + """The universal poll source: a real agent verifies any natural-language + condition on an interval, so anything an agent can observe (with tools, + MCPs, the web, the filesystem) becomes a trigger.""" + kind: Literal["agent"] = "agent" + # What event to look for, in the user's words ("a new episode of X dropped"). + check: str = "" + # Empty = the app's default model; each poll is a real (short) agent turn. + model: str = "" + poll_seconds: int = 900 + + @field_validator("poll_seconds") + @classmethod + def p_clamp_poll(cls, v: int) -> int: + # Each poll costs a real agent turn; 60s floor keeps a typo from burning money. + return max(60, min(v, 86400)) + + +class CustomEventSource(BaseModel): + """The universal push source: never polled; events arrive only via + POST /api/events/ingest, so any script, webhook forwarder, Shortcut, or + MCP can feed this trigger.""" + kind: Literal["custom"] = "custom" + + EventSourceConfig = Annotated[ - Union[FileWatchSource, WebWatchSource], + Union[FileWatchSource, WebWatchSource, AgentCheckSource, CustomEventSource], Field(discriminator="kind"), ] diff --git a/backend/apps/events/poll_loop.py b/backend/apps/events/poll_loop.py index c4cccab3..7e7cd218 100644 --- a/backend/apps/events/poll_loop.py +++ b/backend/apps/events/poll_loop.py @@ -10,16 +10,19 @@ import time from typing import Awaitable, Callable, Dict, List, Optional, Set, Tuple from backend.apps.events import dispatcher, stores -from backend.apps.events.file_watch import file_watch -from backend.apps.events.models import Event, EventLogEntry, EventTriggerConfig -from backend.apps.events.web_watch import web_watch +from backend.apps.events.adapters.agent_check import agent_check +from backend.apps.events.adapters.file_watch import file_watch +from backend.apps.events.adapters.web_watch import web_watch +from backend.apps.events.models import CustomEventSource, Event, EventLogEntry, EventTriggerConfig from backend.apps.workflows.models import Workflow logger = logging.getLogger(__name__) +# "custom" is deliberately absent: those triggers are push-only via /api/events/ingest. ADAPTERS: Dict[str, Callable[..., Awaitable[Tuple[List[Event], Dict]]]] = { "file": file_watch, "web": web_watch, + "agent": agent_check, } p_loop_task: Optional["asyncio.Task"] = None @@ -45,14 +48,17 @@ def reset_state() -> None: p_inflight.clear() -def p_live_triggers() -> List[Tuple[Workflow, EventTriggerConfig]]: +def p_live_triggers() -> List[Tuple[Workflow, EventTriggerConfig, float]]: + """(workflow, trigger, poll_seconds) for every pollable trigger; push-only sources are excluded here.""" from backend.apps.workflows import storage - out: List[Tuple[Workflow, EventTriggerConfig]] = [] + out: List[Tuple[Workflow, EventTriggerConfig, float]] = [] for wf in storage.list_workflows(): for trig in wf.event_triggers: - if trig.enabled and trig.source.kind in ADAPTERS: - out.append((wf, trig)) + source = trig.source + if not trig.enabled or isinstance(source, CustomEventSource) or source.kind not in ADAPTERS: + continue + out.append((wf, trig, float(source.poll_seconds))) return out @@ -84,9 +90,9 @@ def tick() -> None: if storage.get_paused(): return now = time.monotonic() - for wf, trig in p_live_triggers(): + 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 + float(trig.source.poll_seconds) + p_next_poll[trig.id] = now + poll_seconds p_inflight.add(trig.id) asyncio.create_task(p_poll_one(wf.id, trig)) @@ -94,7 +100,7 @@ def tick() -> None: def p_seconds_until_next() -> float: now = time.monotonic() soonest: Optional[float] = None - for _, trig in p_live_triggers(): + for _, trig, _ in p_live_triggers(): nxt = p_next_poll.get(trig.id, now) if soonest is None or nxt < soonest: soonest = nxt diff --git a/backend/main.py b/backend/main.py index 7c16164f..dadf4f21 100644 --- a/backend/main.py +++ b/backend/main.py @@ -47,11 +47,12 @@ from backend.apps.agents.proxy.anthropic_proxy import anthropic_proxy from backend.apps.agents.core.openai_passthrough import openai_passthrough from backend.apps.workflows.workflows import workflows from backend.apps.patterns.patterns import patterns +from backend.apps.events.events import events 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, output_versions, dashboards, swarm, service, subscription, auth, web, onboarding, anthropic_proxy, workflows, patterns, openai_passthrough]) +main_app = MainApp([health, agents, skills, tools_lib, modes, settings, mcp_registry, skill_registry, outputs, output_versions, dashboards, swarm, service, subscription, auth, web, onboarding, anthropic_proxy, workflows, patterns, events, openai_passthrough]) app = main_app.app # Generate per-install auth token BEFORE we bind the HTTP port. By the time any request lands, the token file exists. See backend/auth.py. diff --git a/backend/tests/test_event_triggers.py b/backend/tests/test_event_triggers.py index 3e361eb1..c4b9208b 100644 --- a/backend/tests/test_event_triggers.py +++ b/backend/tests/test_event_triggers.py @@ -50,7 +50,7 @@ def test_trigger_config_clamps_and_round_trips(make_wf): def test_file_watch_baselines_then_diffs(tmp_path): - from backend.apps.events.file_watch import file_watch + from backend.apps.events.adapters.file_watch import file_watch watch_dir = tmp_path / "watched" watch_dir.mkdir() @@ -73,7 +73,7 @@ def test_file_watch_baselines_then_diffs(tmp_path): def test_file_watch_caps_burst(tmp_path): - from backend.apps.events import file_watch as fw + from backend.apps.events.adapters import file_watch as fw watch_dir = tmp_path / "burst" watch_dir.mkdir() @@ -99,7 +99,7 @@ def p_fake_pages(monkeypatch, pages: list[str]): def test_web_watch_baseline_dedup_change_error(monkeypatch): - from backend.apps.events.web_watch import web_watch + from backend.apps.events.adapters.web_watch import web_watch source = WebWatchSource(url="https://example.com/reserve", watch_for="a reservation opening") p_fake_pages(monkeypatch, [ diff --git a/backend/tests/test_event_universal.py b/backend/tests/test_event_universal.py new file mode 100644 index 00000000..f1270206 --- /dev/null +++ b/backend/tests/test_event_universal.py @@ -0,0 +1,160 @@ +"""The universal event sources: the agent-check adapter's contract (verdict +parsing, baseline suppression, identical-event dedup) and the /api/events +ingest route (validation, idempotent dedup keys, dispatcher hand-off), plus +the guarantee that custom triggers are never polled. + +Run: + cd backend && .venv/bin/python -m pytest tests/test_event_universal.py -v +""" + +from __future__ import annotations + +import asyncio + +import pytest +from fastapi import HTTPException + +from backend.apps.events.models import AgentCheckSource, CustomEventSource, Event, EventTriggerConfig + + +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")) + dispatcher.stop() + poll_loop.reset_state() + yield + dispatcher.stop() + poll_loop.reset_state() + + +def test_parse_check_reply_contract(): + from backend.apps.events.adapters.agent_check import parse_check_reply + + event, state = parse_check_reply("I looked around.\nEVENT: A new episode dropped\nSTATE: latest is ep 42") + assert event == "A new episode dropped" + assert state == "latest is ep 42" + + event, state = parse_check_reply("Nothing changed.\nNO_EVENT\nSTATE: still ep 41") + assert event is None + assert state == "still ep 41" + + # Prose that echoes the format earlier can't fake a verdict; the LAST verdict wins. + event, state = parse_check_reply("The format is EVENT: like this.\nNO_EVENT\nSTATE: s") + assert event is None + + with pytest.raises(ValueError): + parse_check_reply("I could not complete the check.") + + +def test_agent_check_baseline_then_event_then_dedup(monkeypatch): + from backend.apps.events.adapters import agent_check as ac + + replies = iter([ + "EVENT: ignores the baseline rule\nSTATE: seen ep 41", # model misbehaves on baseline + "EVENT: Episode 42 is out\nSTATE: seen ep 42", + "EVENT: Episode 42 is out\nSTATE: seen ep 42", # same event re-reported + ]) + prompts: list[str] = [] + + async def p_fake_turn(model, prompt): + prompts.append(prompt) + return next(replies) + + monkeypatch.setattr(ac, "run_check_turn", p_fake_turn) + source = AgentCheckSource(check="a new episode of the show dropped", poll_seconds=60) + + events, cursor = p_run(ac.agent_check(source, {})) + assert events == [] # baseline never fires, even when the model reports EVENT + + events, cursor = p_run(ac.agent_check(source, cursor)) + assert len(events) == 1 + assert events[0].summary == "Episode 42 is out" + assert "seen ep 41" in prompts[1] # previous state round-tripped into the prompt + + events, cursor = p_run(ac.agent_check(source, cursor)) + assert events == [] # identical event line reported again fires once, not forever + + +def p_custom_wf(make_wf): + from backend.apps.workflows import storage + + trig = EventTriggerConfig(source=CustomEventSource(), coalesce_seconds=0) + wf = make_wf(event_triggers=[trig]) + storage.save_workflow(wf) + return wf, trig + + +def test_ingest_validates_and_dispatches(make_wf, monkeypatch): + from backend.apps.events import dispatcher + from backend.apps.events.events import IngestBody, ingest_event + + wf, trig = p_custom_wf(make_wf) + delivered: list[Event] = [] + + async def p_fake_ingest(workflow_id, trigger, events, persist=True): + delivered.extend(events) + + monkeypatch.setattr(dispatcher, "ingest", p_fake_ingest) + + body = IngestBody(workflow_id=wf.id, trigger_id=trig.id, summary="Order #123 landed", dedup_key="order-123") + res = p_run(ingest_event(body)) + assert res == {"ok": True, "queued": 1, "deduped": False} + assert len(delivered) == 1 + assert delivered[0].source == "custom" + + # Same dedup_key again = idempotent, not a second run. + res = p_run(ingest_event(body)) + assert res["deduped"] is True + assert len(delivered) == 1 + + with pytest.raises(HTTPException) as e: + p_run(ingest_event(IngestBody(workflow_id="nope", trigger_id=trig.id, summary="x"))) + assert e.value.status_code == 404 + with pytest.raises(HTTPException) as e: + p_run(ingest_event(IngestBody(workflow_id=wf.id, trigger_id="nope", summary="x"))) + assert e.value.status_code == 404 + with pytest.raises(HTTPException) as e: + p_run(ingest_event(IngestBody(workflow_id=wf.id, trigger_id=trig.id, summary=" "))) + assert e.value.status_code == 400 + + +def test_ingest_refuses_non_custom_and_disabled(make_wf): + from backend.apps.events.events import IngestBody, ingest_event + from backend.apps.events.models import FileWatchSource + from backend.apps.workflows import storage + + file_trig = EventTriggerConfig(source=FileWatchSource(path="/tmp/x")) + off_trig = EventTriggerConfig(source=CustomEventSource(), enabled=False) + wf = make_wf(event_triggers=[file_trig, off_trig]) + storage.save_workflow(wf) + + with pytest.raises(HTTPException) as e: + p_run(ingest_event(IngestBody(workflow_id=wf.id, trigger_id=file_trig.id, summary="x"))) + assert e.value.status_code == 409 + with pytest.raises(HTTPException) as e: + p_run(ingest_event(IngestBody(workflow_id=wf.id, trigger_id=off_trig.id, summary="x"))) + assert e.value.status_code == 409 + + +def test_custom_triggers_are_never_polled(make_wf): + from backend.apps.events import poll_loop, stores + + wf, trig = p_custom_wf(make_wf) + + async def scenario(): + poll_loop.tick() + await asyncio.sleep(0.1) + + p_run(scenario()) + # No poll ran: no cursor written (beyond none), no log entries, no errors. + assert stores.read_log(wf.id) == [] + assert stores.load_cursor(trig.id) == {} diff --git a/frontend/src/app/pages/Workflows/app/EventTriggerRow.tsx b/frontend/src/app/pages/Workflows/app/EventTriggerRow.tsx new file mode 100644 index 00000000..ae43f430 --- /dev/null +++ b/frontend/src/app/pages/Workflows/app/EventTriggerRow.tsx @@ -0,0 +1,169 @@ +// One trigger row inside the Event Triggers panel: folder watch, page watch, +// agent check (any natural-language condition), or custom push (ingest API). +// Text fields commit onBlur so typing doesn't PATCH per keystroke. + +import React from 'react'; +import type { CSSProperties } from 'react'; +import { API_BASE } from '@/shared/config'; +import type { EventTriggerConfig, Workflow } from '@/shared/state/workflowsSlice'; +import { useWC, track, knob } from './uiKit'; + +const WEB_POLL_CHOICES: Array<[number, string]> = [[60, 'every minute'], [300, 'every 5 min'], [900, 'every 15 min'], [3600, 'hourly']]; +const AGENT_POLL_CHOICES: Array<[number, string]> = [[300, 'every 5 min'], [900, 'every 15 min'], [3600, 'hourly'], [21600, 'every 6 hours'], [86400, 'daily']]; + +const KIND_LABELS: Record = { + file: 'Folder / file watch', + web: 'Web page watch', + agent: 'Agent check', + custom: 'Custom (push)', +}; + +interface RowProps { + workflow: Workflow; + trigger: EventTriggerConfig; + onMutate: (mut: (t: EventTriggerConfig) => EventTriggerConfig) => void; + onRemove: () => void; +} + +const EventTriggerRow: React.FC = ({ workflow, trigger, onMutate, onRemove }) => { + const WC = useWC(); + const t = trigger; + const src = t.source; + + const fieldStyle: CSSProperties = { + width: '100%', boxSizing: 'border-box', height: 30, background: WC.raised, + border: `1px solid rgba(${WC.inkRGB},0.12)`, borderRadius: 8, padding: '0 9px', + fontSize: 12.5, color: WC.ink, + }; + const labelStyle: CSSProperties = { fontSize: 11.5, color: WC.muted, marginBottom: 4, display: 'block' }; + const ghostBtn: CSSProperties = { + height: 26, padding: '0 8px', borderRadius: 7, border: `1px solid rgba(${WC.inkRGB},0.12)`, + cursor: 'pointer', fontSize: 11.5, fontWeight: 600, background: WC.raised, color: WC.ink3, + }; + + const pollSelect = (value: number, choices: Array<[number, string]>, onChange: (v: number) => void) => ( + + ); + + return ( +
+
+ {KIND_LABELS[src.kind] ?? src.kind} +
+
onMutate((x) => ({ ...x, enabled: !x.enabled }))} + style={{ display: 'flex', alignItems: 'center', gap: 6, cursor: 'pointer' }} + > + {t.enabled ? 'On' : 'Off'} +
+
+ +
+
+ + {src.kind === 'file' && ( +
+ Watch this folder or file + { + const path = e.target.value.trim(); + if (path !== src.path) onMutate((x) => ({ ...x, source: { ...src, path } })); + }} + /> +
+ )} + + {src.kind === 'web' && ( + <> +
+ Page URL + { + const url = e.target.value.trim(); + if (url !== src.url) onMutate((x) => ({ ...x, source: { ...src, url } })); + }} + /> +
+
+
+ Watching for + { + const watchFor = e.target.value.trim(); + if (watchFor !== src.watch_for) onMutate((x) => ({ ...x, source: { ...src, watch_for: watchFor } })); + }} + /> +
+
+ Check + {pollSelect(src.poll_seconds, WEB_POLL_CHOICES, (v) => onMutate((x) => ({ ...x, source: { ...src, poll_seconds: v } })))} +
+
+ + )} + + {src.kind === 'agent' && ( +
+
+ What counts as the event? An agent checks with its tools. + { + const check = e.target.value.trim(); + if (check !== src.check) onMutate((x) => ({ ...x, source: { ...src, check } })); + }} + /> +
+
+ {pollSelect(src.poll_seconds, AGENT_POLL_CHOICES, (v) => onMutate((x) => ({ ...x, source: { ...src, poll_seconds: v } })))} +
+
+ )} + + {src.kind === 'custom' && ( +
+ Anything can push events here (scripts, webhooks, Shortcuts): + 
+            {`POST ${API_BASE}/events/ingest\n{"workflow_id": "${workflow.id}", "trigger_id": "${t.id}", "summary": "what happened", "dedup_key": "optional-id"}`}
+
+
+ )} + +
+ Only when (optional, plain English) + { + const predicate = e.target.value.trim(); + if (predicate !== t.predicate) onMutate((x) => ({ ...x, predicate })); + }} + /> +
+
+ ); +}; + +export default EventTriggerRow; diff --git a/frontend/src/app/pages/Workflows/app/EventTriggersCard.tsx b/frontend/src/app/pages/Workflows/app/EventTriggersCard.tsx index f78a0a12..7249c456 100644 --- a/frontend/src/app/pages/Workflows/app/EventTriggersCard.tsx +++ b/frontend/src/app/pages/Workflows/app/EventTriggersCard.tsx @@ -1,24 +1,38 @@ -// "When something happens" panel beside the Schedule card: watch a folder/file -// or a web page, optionally filter with a plain-English condition, and see the -// trigger's recent activity ("saw X, skipped because Y") so a quiet trigger is -// debuggable instead of mysterious. +// "When something happens" panel beside the Schedule card. Four source kinds +// cover the universe: folder/file watch, web page watch, agent check (any +// natural-language condition), and custom push (anything can POST an event). +// The Recent-activity feed makes a quiet trigger debuggable instead of +// mysterious ("saw X, skipped because Y"). import React, { useCallback, useEffect, useState } from 'react'; import type { CSSProperties } from 'react'; import { API_BASE } from '@/shared/config'; -import type { EventTriggerConfig, Workflow, WorkflowEventLogEntry } from '@/shared/state/workflowsSlice'; -import { useWC, FONT_SERIF, track, knob } from './uiKit'; +import type { EventSourceConfig, EventTriggerConfig, Workflow, WorkflowEventLogEntry } from '@/shared/state/workflowsSlice'; +import EventTriggerRow from './EventTriggerRow'; +import { useWC, FONT_SERIF } from './uiKit'; import { useWorkflowPatch } from './useWorkflowPatch'; -const WEB_POLL_CHOICES: Array<[number, string]> = [[60, 'every minute'], [300, 'every 5 min'], [900, 'every 15 min'], [3600, 'hourly']]; +type TriggerKind = 'file' | 'web' | 'agent' | 'custom'; -function newTrigger(kind: 'file' | 'web'): EventTriggerConfig { +const ADD_CHOICES: Array<[TriggerKind, string]> = [ + ['file', '+ Folder'], + ['web', '+ Page'], + ['agent', '+ Agent check'], + ['custom', '+ Custom'], +]; + +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 }; + return { kind: 'custom' }; +} + +function newTrigger(kind: TriggerKind): EventTriggerConfig { return { id: crypto.randomUUID().replace(/-/g, ''), enabled: true, - source: kind === 'file' - ? { kind: 'file', path: '', poll_seconds: 15 } - : { kind: 'web', url: '', watch_for: '', poll_seconds: 300 }, + source: newSource(kind), predicate: '', coalesce_seconds: kind === 'file' ? 30 : 0, max_fires_per_hour: 6, @@ -35,10 +49,6 @@ const EventTriggersCard: React.FC<{ workflow: Workflow }> = ({ workflow }) => { patch(workflow, { event_triggers: next }); }, [patch, workflow]); - const updateTrigger = useCallback((id: string, mut: (t: EventTriggerConfig) => EventTriggerConfig) => { - patchTriggers(triggers.map((t) => (t.id === id ? mut(t) : t))); - }, [patchTriggers, triggers]); - // Activity poll while the panel is mounted; local endpoint, cheap. useEffect(() => { if (triggers.length === 0) return; @@ -56,15 +66,10 @@ const EventTriggersCard: React.FC<{ workflow: Workflow }> = ({ workflow }) => { }, [workflow.id, triggers.length]); const ghostBtn: CSSProperties = { - height: 26, padding: '0 10px', borderRadius: 7, border: `1px solid rgba(${WC.inkRGB},0.12)`, - cursor: 'pointer', fontSize: 11.5, fontWeight: 600, background: WC.raised, color: WC.ink3, + height: 26, padding: '0 8px', borderRadius: 7, border: `1px solid rgba(${WC.inkRGB},0.12)`, + cursor: 'pointer', fontSize: 11, fontWeight: 600, background: WC.raised, color: WC.ink3, whiteSpace: 'nowrap', }; - const fieldStyle: CSSProperties = { - width: '100%', boxSizing: 'border-box', height: 30, background: WC.raised, - border: `1px solid rgba(${WC.inkRGB},0.12)`, borderRadius: 8, padding: '0 9px', - fontSize: 12.5, color: WC.ink, - }; - const labelStyle: CSSProperties = { fontSize: 11.5, color: WC.muted, marginBottom: 4, display: 'block' }; + const labelStyle: CSSProperties = { fontSize: 11.5, color: WC.muted, marginBottom: 7, display: 'block' }; const dotColor = (kind: WorkflowEventLogEntry['kind']): string => { if (kind === 'fired') return WC.accent; @@ -74,122 +79,34 @@ const EventTriggersCard: React.FC<{ workflow: Workflow }> = ({ workflow }) => { return (
-
- Event triggers -
- - +
+ Event triggers +
+ {ADD_CHOICES.map(([kind, label]) => ( + + ))}
{triggers.length === 0 && ( - Run this workflow when something happens: a file lands in a folder, or a page you care about changes. + Run this workflow when something happens: a file lands, a page changes, an agent spots any condition you describe, or anything pushes an event in. )} - {triggers.map((t) => { - const src = t.source; - return ( -
-
- - {src.kind === 'file' ? 'Folder / file watch' : 'Web page watch'} - -
-
updateTrigger(t.id, (x) => ({ ...x, enabled: !x.enabled }))} - style={{ display: 'flex', alignItems: 'center', gap: 6, cursor: 'pointer' }} - > - {t.enabled ? 'On' : 'Off'} -
-
- -
-
- - {src.kind === 'file' ? ( -
- Watch this folder or file - { - const path = e.target.value.trim(); - if (path !== src.path) updateTrigger(t.id, (x) => ({ ...x, source: { ...src, path } })); - }} - /> -
- ) : ( - <> -
- Page URL - { - const url = e.target.value.trim(); - if (url !== src.url) updateTrigger(t.id, (x) => ({ ...x, source: { ...src, url } })); - }} - /> -
-
-
- Watching for - { - const watchFor = e.target.value.trim(); - if (watchFor !== src.watch_for) updateTrigger(t.id, (x) => ({ ...x, source: { ...src, watch_for: watchFor } })); - }} - /> -
-
- Check - -
-
- - )} - -
- Only when (optional, plain English) - { - const predicate = e.target.value.trim(); - if (predicate !== t.predicate) updateTrigger(t.id, (x) => ({ ...x, predicate })); - }} - /> -
-
- ); - })} + {triggers.map((t) => ( + patchTriggers(triggers.map((x) => (x.id === t.id ? mut(x) : x)))} + onRemove={() => patchTriggers(triggers.filter((x) => x.id !== t.id))} + /> + ))} {triggers.length > 0 && log.length > 0 && (
- Recent activity + Recent activity {log.slice(0, 6).map((e, i) => (
diff --git a/frontend/src/shared/state/workflowsSlice.ts b/frontend/src/shared/state/workflowsSlice.ts index 20b4aed2..e21d4804 100644 --- a/frontend/src/shared/state/workflowsSlice.ts +++ b/frontend/src/shared/state/workflowsSlice.ts @@ -42,7 +42,21 @@ export interface WebWatchSource { poll_seconds: number; } -export type EventSourceConfig = FileWatchSource | WebWatchSource; +export interface AgentCheckSource { + kind: 'agent'; + /** Any natural-language condition; a real agent verifies it each poll with its full tool surface. */ + check: string; + /** Empty = the app's default model. */ + model: string; + poll_seconds: number; +} + +export interface CustomEventSource { + /** Push-only: events arrive via POST /api/events/ingest from any script/webhook/Shortcut. */ + kind: 'custom'; +} + +export type EventSourceConfig = FileWatchSource | WebWatchSource | AgentCheckSource | CustomEventSource; export interface EventTriggerConfig { id: string;