From a7e5a4dc73da1dc36ae69067954635cac8d46bc1 Mon Sep 17 00:00:00 2001 From: ciregenz Date: Tue, 28 Jul 2026 17:52:13 -0700 Subject: [PATCH] [eric] events: zero-knob pass (adaptive cadence, MCP inference, self-heal, no test nag, one-URL push) --- backend/apps/agents/schedule_mcp_server.py | 61 ++++++++++++-- backend/apps/events/adapters/heal_trigger.py | 84 +++++++++++++++++++ backend/apps/events/events.py | 52 +++++++++--- backend/apps/events/models.py | 14 ++-- backend/apps/events/poll_loop.py | 53 +++++++++++- backend/apps/patterns/patterns.py | 6 +- backend/apps/workflows/workflows.py | 3 +- backend/auth.py | 2 + backend/tests/test_event_stream_tier.py | 60 ++++++++++++- backend/tests/test_event_universal.py | 36 ++++++++ .../pages/Workflows/app/EventTriggerRow.tsx | 72 ++++++---------- .../pages/Workflows/app/EventTriggersCard.tsx | 8 +- frontend/src/shared/state/workflowsSlice.ts | 2 + 13 files changed, 371 insertions(+), 82 deletions(-) create mode 100644 backend/apps/events/adapters/heal_trigger.py diff --git a/backend/apps/agents/schedule_mcp_server.py b/backend/apps/agents/schedule_mcp_server.py index 4da44874..1effeea1 100644 --- a/backend/apps/agents/schedule_mcp_server.py +++ b/backend/apps/agents/schedule_mcp_server.py @@ -314,8 +314,8 @@ TOOLS = [ "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."}, - "poll_minutes": {"type": "number", "description": "How often to check. Defaults: file continuous (~15s), web 5, agent 15. Agent checks cost a model call each, so don't go below 5 without the user asking."}, + "mcps": {"type": "array", "items": {"type": "string"}, "description": "kind=agent: usually OMIT; the system infers connected tools from the check sentence. Pass only to override the inference."}, + "poll_minutes": {"type": "number", "description": "Usually OMIT: cadence is automatic (tunes itself from observed event rate). Set only when the user asked for a specific frequency; agent checks cost a model call each."}, "only_when": {"type": "string", "description": "Optional plain-English filter; events not matching it are skipped (logged)."}, "max_fires_per_hour": {"type": "integer", "description": "Safety cap on runs per hour (default 6)."}, }, @@ -668,7 +668,46 @@ def handle_invoke_workflow(args: dict) -> dict: return _ok(f"Workflow '{match.get('title')}' run {status}.{err_line}\n\n=== RUN TRANSCRIPT ===\n{transcript}\n=== END TRANSCRIPT ===") -TRIGGER_POLL_DEFAULTS = {"file": 15, "web": 300, "agent": 900} +MCP_HINTS = { + "google-workspace": ("email", "inbox", "gmail", "mail", "calendar", "meeting", "drive", "doc", "sheet"), + "notion": ("notion", "page", "database"), + "slack": ("slack", "channel"), + "discord": ("discord",), + "reddit": ("reddit", "subreddit"), + "github": ("github", "pull request", "issue", "repo"), +} + + +def p_suggest_mcps(check: str, known: set) -> list: + """Infer connected tools from the check sentence so the user never names them; only suggests tools that actually exist.""" + text = check.lower() + out = [] + for tool, words in MCP_HINTS.items(): + if tool in known and any(w in text for w in words): + out.append(tool) + for tool in known: + if tool not in out and tool in text: + out.append(tool) + return out[:4] + + +def p_known_tools() -> set: + r = _call("GET", f"http://127.0.0.1:{BACKEND_PORT}/api/tools/list") + if "_error" in r: + return set() + tools = r.get("tools", r) if isinstance(r, dict) else r + known = set() + for t in (tools if isinstance(tools, list) else []): + for key in ("id", "name"): + v = str((t or {}).get(key) or "").strip().lower() + if v: + known.add(v) + return known + + +def p_steps_signature(steps: list) -> str: + # MUST byte-match the FE stepsSignature (JSON.stringify of [id, text] pairs); pinned by test_watch_for_event_tool. + return json.dumps([[s["id"], s["text"]] for s in steps], separators=(",", ":"), ensure_ascii=False) def p_find_workflow_any(ident: str): @@ -712,7 +751,8 @@ def p_build_trigger(args: dict) -> tuple: 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) + # 0 = adaptive: the engine tunes cadence from observed event rate; only an explicit poll_minutes pins it. + poll_seconds = int(float(poll_minutes) * 60) if poll_minutes else 0 if kind == "file": if not (args.get("path") or "").strip(): return None, "kind=file needs path (the file or folder to watch)." @@ -725,6 +765,8 @@ def p_build_trigger(args: dict) -> tuple: if not (args.get("check") or "").strip(): return None, "kind=agent needs check (one sentence describing the condition)." mcps = [str(m) for m in (args.get("mcps") or [])] + if not mcps: + mcps = p_suggest_mcps(args["check"], p_known_tools()) mcp_err = p_validate_mcps(mcps) if mcp_err: return None, mcp_err @@ -734,7 +776,7 @@ def p_build_trigger(args: dict) -> tuple: 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"} + source = {"kind": "custom", "secret": uuid.uuid4().hex} return { "id": uuid.uuid4().hex, "enabled": True, @@ -781,13 +823,15 @@ def handle_watch_for_event(args: dict) -> dict: steps_in = [s for s in (args.get("steps") or []) if str(s).strip()] if not steps_in: return _err("To create a new workflow, pass title and steps (what to do when the event fires), or pass workflow to attach to an existing one.") + steps_payload = [{"id": f"s{i+1}", "text": str(s)} for i, s in enumerate(steps_in)] body = { "title": args.get("title") or "Event workflow", - "steps": [{"id": f"s{i+1}", "text": str(s)} for i, s in enumerate(steps_in)], + "steps": steps_payload, "schedule": {"enabled": False}, "event_triggers": [trigger], "source_session_id": PARENT_SESSION_ID or None, "dashboard_id": DASHBOARD_ID or None, + "tested_signature": p_steps_signature(steps_payload), } r = _call("POST", "/create", body) if "_error" in r: @@ -796,8 +840,9 @@ def handle_watch_for_event(args: dict) -> dict: extra = "" if trigger["source"]["kind"] == "custom": extra = ( - f"\nOutside systems push events with: POST http://127.0.0.1:{BACKEND_PORT}/api/events/ingest " - f"(JSON: workflow_id={wid}, trigger_id={trigger['id']}, summary, optional dedup_key; per-install auth token required)." + f"\nOutside systems push events with ONE URL, no token needed: " + f"POST http://127.0.0.1:{BACKEND_PORT}/api/events/ingest/{trigger['source']['secret']} " + f"(JSON body: summary, optional event_type/dedup_key/payload)." ) return _ok(f"Watching. Workflow \"{title}\" (id: {wid}) now runs on {p_describe_trigger(trigger)}.{extra} The user can edit or disable it in the workflow's Event triggers panel.") diff --git a/backend/apps/events/adapters/heal_trigger.py b/backend/apps/events/adapters/heal_trigger.py new file mode 100644 index 00000000..6e3f45eb --- /dev/null +++ b/backend/apps/events/adapters/heal_trigger.py @@ -0,0 +1,84 @@ +"""Self-heal before bothering the user: when a URL-bearing watcher (web/stream) +starts failing, one invisible background agent turn investigates and either +fixes the config itself (a moved/redirected URL) or declares it needs a human, +at which point the attention surface takes over. Strict FIX_URL / CANNOT_FIX +contract; only an http(s) URL that actually differs is ever applied, and the +repair is written to the activity log so nothing changes silently.""" + +import logging +from typing import Optional, Tuple + +from typeguard import typechecked + +from backend.apps.events.models import EventLogEntry, EventTriggerConfig + +logger = logging.getLogger(__name__) + + +@typechecked +def build_heal_prompt(url: str, last_error: str) -> str: + return ( + "You are repairing an automated watcher. It repeatedly fails to read this URL:\n" + f"{url}\n" + f"Most recent error: {last_error or 'unknown'}\n\n" + "Investigate with your tools (fetch the URL, follow redirects, check for an obvious " + "moved/renamed location on the same site). Then END your reply with EXACTLY one of:\n" + "FIX_URL: \n" + "CANNOT_FIX: " + ) + + +@typechecked +def parse_heal_reply(text: str) -> Tuple[Optional[str], str]: + """(replacement url or None, reason). Last occurrence wins.""" + fix: Optional[str] = None + reason = "" + for line in text.splitlines(): + s = line.strip() + if s.upper().startswith("FIX_URL:"): + fix = s[len("FIX_URL:"):].strip() + reason = "" + elif s.upper().startswith("CANNOT_FIX:"): + fix = None + reason = s[len("CANNOT_FIX:"):].strip() + return fix, reason + + +async def attempt_heal(workflow_id: str, trigger: EventTriggerConfig) -> bool: + """True when the trigger config was repaired (caller should re-poll now).""" + from backend.apps.events import stores + from backend.apps.events.adapters.agent_check import run_check_turn + from backend.apps.settings.settings import load_settings + from backend.apps.workflows import storage + + url = str(getattr(trigger.source, "url", "") or "").strip() + if not url: + return False + health = stores.read_poll_health(trigger.id) + model = getattr(load_settings(), "default_model", None) or "sonnet" + try: + reply = await run_check_turn(model, build_heal_prompt(url, str(health.get("last_error") or ""))) + fix, reason = parse_heal_reply(reply) + except Exception as e: + logger.warning("heal turn failed for trigger %s: %s", trigger.id, e) + return False + if fix and fix.startswith("http") and fix != url: + wf = storage.get_workflow(workflow_id) + if wf is None: + return False + live = next((t for t in wf.event_triggers if t.id == trigger.id), None) + if live is None or str(getattr(live.source, "url", "")) != url: + return False # user edited it meanwhile; their change wins + setattr(live.source, "url", fix) + storage.save_workflow(wf) + stores.clear_poll_failures(trigger.id) + stores.append_log(workflow_id, EventLogEntry( + trigger_id=trigger.id, kind="emitted", + summary=f"Self-healed: watcher URL updated to {fix}", + )) + return True + stores.append_log(workflow_id, EventLogEntry( + trigger_id=trigger.id, kind="error", + summary=f"Self-heal couldn't fix it: {reason[:160] or 'no working replacement found'}", + )) + return False diff --git a/backend/apps/events/events.py b/backend/apps/events/events.py index deac0f63..3f471411 100644 --- a/backend/apps/events/events.py +++ b/backend/apps/events/events.py @@ -38,16 +38,14 @@ class IngestBody(BaseModel): payload: Dict = Field(default_factory=dict) -@events.router.post("/ingest") -async def ingest_event(body: IngestBody): - from backend.apps.workflows import storage +class IngestPushBody(BaseModel): + summary: str + event_type: str = "custom" + dedup_key: str = "" + payload: Dict = Field(default_factory=dict) - 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") + +async def p_do_ingest(workflow_id: str, trigger, body: IngestPushBody) -> Dict: if trigger.source.kind != "custom": raise HTTPException(status_code=409, detail="Trigger is not a custom (ingest) source") if not trigger.enabled: @@ -62,7 +60,7 @@ async def ingest_event(body: IngestBody): 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( + await dispatcher.ingest(workflow_id, trigger, [Event( source="custom", event_type=(body.event_type.strip() or "custom")[:60], summary=summary, @@ -70,3 +68,37 @@ async def ingest_event(body: IngestBody): payload=body.payload, )]) return {"ok": True, "queued": 1, "deduped": False} + + +@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") + return await p_do_ingest(wf.id, trigger, IngestPushBody( + summary=body.summary, event_type=body.event_type, dedup_key=body.dedup_key, payload=body.payload, + )) + + +@events.router.post("/ingest/{secret}") +async def ingest_event_by_secret(secret: str, body: IngestPushBody): + """Paste-one-URL push: the per-trigger secret in the path IS the credential + (auth-middleware exempt; same entropy class as the install token, localhost-bound, + revoked by deleting the trigger).""" + import hmac + + from backend.apps.workflows import storage + + if len(secret.strip()) < 16: + raise HTTPException(status_code=404, detail="Unknown ingest URL") + for wf in storage.list_workflows(): + for trigger in wf.event_triggers: + trigger_secret = str(getattr(trigger.source, "secret", "") or "") + if trigger_secret and hmac.compare_digest(trigger_secret, secret): + return await p_do_ingest(wf.id, trigger, body) + raise HTTPException(status_code=404, detail="Unknown ingest URL") diff --git a/backend/apps/events/models.py b/backend/apps/events/models.py index ee80bc76..d3b50c66 100644 --- a/backend/apps/events/models.py +++ b/backend/apps/events/models.py @@ -34,8 +34,8 @@ class FileWatchSource(BaseModel): @field_validator("poll_seconds") @classmethod def p_clamp_poll(cls, v: int) -> int: - # Clamp, don't reject: a stray value from an agent tool or old record shouldn't crash the poll loop. - return max(5, min(v, 3600)) + # 0 = adaptive (the engine tunes cadence from observed event rate). Clamp, don't reject. + return 0 if v == 0 else max(5, min(v, 3600)) class WebWatchSource(BaseModel): @@ -48,8 +48,8 @@ class WebWatchSource(BaseModel): @field_validator("poll_seconds") @classmethod def p_clamp_poll(cls, v: int) -> int: - # 60s floor: polling someone's site faster than that is rude and burns nothing useful. - return max(60, min(v, 86400)) + # 0 = adaptive. 60s floor otherwise: polling someone's site faster is rude and buys nothing. + return 0 if v == 0 else max(60, min(v, 86400)) class AgentCheckSource(BaseModel): @@ -73,8 +73,8 @@ class AgentCheckSource(BaseModel): @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)) + # 0 = adaptive. Each poll costs a real agent turn; 60s floor keeps a typo from burning money. + return 0 if v == 0 else max(60, min(v, 86400)) class CustomEventSource(BaseModel): @@ -82,6 +82,8 @@ class CustomEventSource(BaseModel): POST /api/events/ingest, so any script, webhook forwarder, Shortcut, or MCP can feed this trigger.""" kind: Literal["custom"] = "custom" + # Per-trigger credential baked into the push URL (POST /api/events/ingest/), so wiring a sender is paste-one-URL instead of token juggling. Same entropy class as the install token; revoked by deleting the trigger. + secret: str = Field(default_factory=lambda: uuid4().hex) class StreamSource(BaseModel): diff --git a/backend/apps/events/poll_loop.py b/backend/apps/events/poll_loop.py index f08b3968..5de3fc20 100644 --- a/backend/apps/events/poll_loop.py +++ b/backend/apps/events/poll_loop.py @@ -32,6 +32,39 @@ p_loop_task: Optional["asyncio.Task"] = None p_wake = asyncio.Event() p_next_poll: Dict[str, float] = {} p_inflight: Set[str] = set() +# Adaptive cadence (poll_seconds=0 triggers): (default, floor, ceiling) per kind. Events halve the interval toward the floor; 5 straight quiet polls stretch it 1.5x toward the ceiling. +PACE_BOUNDS: Dict[str, Tuple[float, float, float]] = { + "file": (15.0, 5.0, 60.0), + "web": (300.0, 60.0, 1800.0), + "agent": (900.0, 300.0, 21600.0), +} +PACE_QUIET_POLLS = 5 +p_pace_interval: Dict[str, float] = {} +p_pace_quiet: Dict[str, int] = {} + + +def effective_poll_seconds(trigger: EventTriggerConfig) -> float: + fixed = float(getattr(trigger.source, "poll_seconds", 0) or 0) + if fixed > 0: + return fixed + default, lo, hi = PACE_BOUNDS.get(trigger.source.kind, (300.0, 60.0, 3600.0)) + return min(max(p_pace_interval.get(trigger.id, default), lo), hi) + + +def pace_update(trigger: EventTriggerConfig, event_count: int) -> None: + if float(getattr(trigger.source, "poll_seconds", 0) or 0) > 0: + return + default, lo, hi = PACE_BOUNDS.get(trigger.source.kind, (300.0, 60.0, 3600.0)) + current = p_pace_interval.get(trigger.id, default) + if event_count > 0: + p_pace_interval[trigger.id] = max(lo, current / 2) + p_pace_quiet[trigger.id] = 0 + else: + quiet = p_pace_quiet.get(trigger.id, 0) + 1 + if quiet >= PACE_QUIET_POLLS: + p_pace_interval[trigger.id] = min(hi, current * 1.5) + quiet = 0 + p_pace_quiet[trigger.id] = quiet # Held-open live sources (kqueue file signals, SSE streams): trigger_id -> (config signature, stopper). p_live_handles: Dict[str, Tuple[str, Callable[[], None]]] = {} @@ -56,6 +89,8 @@ def reset_state() -> None: p_loop_task = None p_next_poll.clear() p_inflight.clear() + p_pace_interval.clear() + p_pace_quiet.clear() for _, stop in p_live_handles.values(): try: stop() @@ -74,7 +109,7 @@ def p_live_triggers() -> List[Tuple[Workflow, EventTriggerConfig, float]]: source = trig.source if not trig.enabled or isinstance(source, (CustomEventSource, StreamSource)) or source.kind not in ADAPTERS: continue - out.append((wf, trig, float(source.poll_seconds))) + out.append((wf, trig, effective_poll_seconds(trig))) return out @@ -90,6 +125,7 @@ async def p_poll_one(wf: Workflow, trigger: EventTriggerConfig) -> None: events, new_cursor = await fetch(trigger.source, cursor) stores.save_cursor(trigger.id, new_cursor) stores.clear_poll_failures(trigger.id) + pace_update(trigger, len(events)) if events: await dispatcher.ingest(workflow_id, trigger, events) except Exception as e: @@ -97,7 +133,10 @@ async def p_poll_one(wf: Workflow, trigger: EventTriggerConfig) -> None: 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, str(e)) - base = float(getattr(trigger.source, "poll_seconds", 300)) + # Third straight failure: try to fix it ourselves before the attention surface asks the user. + if failures == 3 and trigger.source.kind in ("web", "stream"): + asyncio.create_task(p_heal_and_wake(wf, trigger)) + base = effective_poll_seconds(trigger) 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 "" @@ -111,6 +150,16 @@ async def p_poll_one(wf: Workflow, trigger: EventTriggerConfig) -> None: p_inflight.discard(trigger.id) +async def p_heal_and_wake(wf: Workflow, trigger: EventTriggerConfig) -> None: + try: + from backend.apps.events.adapters.heal_trigger import attempt_heal + if await attempt_heal(wf.id, trigger): + mark_due(trigger.id) + kick() + except Exception: + logger.debug("self-heal attempt errored", exc_info=True) + + 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.""" diff --git a/backend/apps/patterns/patterns.py b/backend/apps/patterns/patterns.py index 725d4429..889348dc 100644 --- a/backend/apps/patterns/patterns.py +++ b/backend/apps/patterns/patterns.py @@ -4,6 +4,7 @@ the same create path the Workflows UI uses, so the user reviews and owns it like any other.""" import asyncio +import json import logging from contextlib import asynccontextmanager from typing import Optional @@ -81,12 +82,15 @@ async def accept_suggestion(suggestion_id: str): else: # No clear rhythm in the evidence: create it ready to run, let the user schedule or add a trigger. schedule = ScheduleConfig(enabled=False) + steps = [WorkflowStep(text=t) for t in suggestion.workflow_steps] body = WorkflowCreate( title=suggestion.workflow_title or "Suggested workflow", description=suggestion.description, - steps=[WorkflowStep(text=t) for t in suggestion.workflow_steps], + steps=steps, schedule=schedule, auto_named=False, + # The explicit accept IS the validation moment; byte-matches the FE stepsSignature so the test-first nag never fires. + tested_signature=json.dumps([[s.id, s.text] for s in steps], separators=(",", ":"), ensure_ascii=False), ) workflow = await create_workflow(body) suggestion.status = "accepted" diff --git a/backend/apps/workflows/workflows.py b/backend/apps/workflows/workflows.py index d45a0101..70a5378b 100644 --- a/backend/apps/workflows/workflows.py +++ b/backend/apps/workflows/workflows.py @@ -782,7 +782,8 @@ async def triggers_attention(): continue health = read_poll_health(t.id) failures = int(health.get("consecutive_failures") or 0) - if failures >= 3: + # Threshold 5, not 3: the self-heal attempt fires at 3, so the user is only asked once healing has demonstrably failed. + if failures >= 5: items.append({ "workflow_id": wf.id, "workflow_title": wf.title, diff --git a/backend/auth.py b/backend/auth.py index 4269dd7e..80680d76 100644 --- a/backend/auth.py +++ b/backend/auth.py @@ -190,6 +190,8 @@ P_AUTH_EXEMPT_PREFIX = ( "/api/health", # 9Router proxies OpenAI requests with the user's sk-... bearer, not our local token; localhost-only is the gate. "/api/openai-passthrough", + # Per-trigger push URLs: the path secret IS the credential (route 404s on any non-matching secret). + "/api/events/ingest/", "/docs", "/openapi", "/redoc", diff --git a/backend/tests/test_event_stream_tier.py b/backend/tests/test_event_stream_tier.py index f051ff15..b708f5dd 100644 --- a/backend/tests/test_event_stream_tier.py +++ b/backend/tests/test_event_stream_tier.py @@ -122,17 +122,71 @@ def test_attention_endpoint_surfaces_repeat_failures(make_wf): wf = make_wf(event_triggers=[trig, healthy]) storage.save_workflow(wf) - for _ in range(2): + for _ in range(4): stores.record_poll_failure(trig.id, "connect refused") - assert p_run(triggers_attention()) == {"attention": []} # 2 failures = not yet + assert p_run(triggers_attention()) == {"attention": []} # 4 failures = self-heal territory, not the user's 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 item["consecutive_failures"] == 5 assert "connect refused" in item["last_error"] stores.clear_poll_failures(trig.id) assert p_run(triggers_attention()) == {"attention": []} + + +def test_adaptive_pace_tunes_itself(): + from backend.apps.events import poll_loop + + auto = EventTriggerConfig(source=StreamSource(url="x")) # placeholder; pace keys off trigger id + kind + web_auto = EventTriggerConfig(source={"kind": "web", "url": "https://a.b", "watch_for": "", "poll_seconds": 0}) + assert poll_loop.effective_poll_seconds(web_auto) == 300.0 # default until observed + + poll_loop.pace_update(web_auto, event_count=3) + assert poll_loop.effective_poll_seconds(web_auto) == 150.0 # events halve toward the floor + for _ in range(5): + poll_loop.pace_update(web_auto, event_count=0) + assert poll_loop.effective_poll_seconds(web_auto) == 225.0 # 5 quiet polls stretch 1.5x + + for _ in range(20): + poll_loop.pace_update(web_auto, event_count=9) + assert poll_loop.effective_poll_seconds(web_auto) == 60.0 # floor holds + + fixed = EventTriggerConfig(source={"kind": "web", "url": "https://a.b", "watch_for": "", "poll_seconds": 600}) + poll_loop.pace_update(fixed, event_count=9) + assert poll_loop.effective_poll_seconds(fixed) == 600.0 # explicit cadence is never second-guessed + assert auto.source.kind == "stream" + + +def test_self_heal_fixes_url_or_escalates(make_wf, monkeypatch): + from backend.apps.events.adapters import agent_check as ac + from backend.apps.events.adapters import heal_trigger as ht + from backend.apps.events import stores + from backend.apps.workflows import storage + + trig = EventTriggerConfig(source=StreamSource(url="https://old.example/feed")) + wf = make_wf(event_triggers=[trig]) + storage.save_workflow(wf) + stores.record_poll_failure(trig.id, "410 Gone") + + async def p_fix_turn(model, prompt, **kwargs): + assert "https://old.example/feed" in prompt and "410 Gone" in prompt + return "Investigated.\nFIX_URL: https://new.example/feed" + + monkeypatch.setattr(ac, "run_check_turn", p_fix_turn) + assert p_run(ht.attempt_heal(wf.id, trig)) is True + healed = storage.get_workflow(wf.id).event_triggers[0] + assert healed.source.url == "https://new.example/feed" + assert stores.read_poll_health(trig.id) == {} # failure streak cleared + assert any("Self-healed" in e.summary for e in stores.read_log(wf.id)) + + async def p_cannot(model, prompt, **kwargs): + return "CANNOT_FIX: the site now requires a sign-in" + + monkeypatch.setattr(ac, "run_check_turn", p_cannot) + assert p_run(ht.attempt_heal(wf.id, healed)) is False + assert storage.get_workflow(wf.id).event_triggers[0].source.url == "https://new.example/feed" # untouched + assert any("requires a sign-in" in e.summary for e in stores.read_log(wf.id)) diff --git a/backend/tests/test_event_universal.py b/backend/tests/test_event_universal.py index 2b13c886..2b1c2318 100644 --- a/backend/tests/test_event_universal.py +++ b/backend/tests/test_event_universal.py @@ -184,3 +184,39 @@ def test_custom_triggers_are_never_polled(make_wf): # 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) == {} + + +def test_secret_url_ingest(make_wf, monkeypatch): + """Paste-one-URL push: the path secret is the credential; wrong or short secrets 404.""" + from backend.apps.events import dispatcher + from backend.apps.events.events import IngestPushBody, ingest_event_by_secret + + 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) + secret = trig.source.secret + res = p_run(ingest_event_by_secret(secret, IngestPushBody(summary="Order landed", dedup_key="o1"))) + assert res == {"ok": True, "queued": 1, "deduped": False} + assert len(delivered) == 1 + + with pytest.raises(HTTPException) as e: + p_run(ingest_event_by_secret("f" * 32, IngestPushBody(summary="x"))) + assert e.value.status_code == 404 + with pytest.raises(HTTPException) as e: + p_run(ingest_event_by_secret("short", IngestPushBody(summary="x"))) + assert e.value.status_code == 404 + + +def test_mcp_auto_suggest_and_signature_vector(monkeypatch): + import backend.apps.agents.schedule_mcp_server as srv + + known = {"google-workspace", "notion"} + assert srv.p_suggest_mcps("a new email from my landlord arrived", known) == ["google-workspace"] + assert srv.p_suggest_mcps("my notion database gained a row", known) == ["notion"] + assert srv.p_suggest_mcps("the moon is full", known) == [] + # Byte-match the FE stepsSignature: JSON.stringify([["s1","a\"b"]]). + assert srv.p_steps_signature([{"id": "s1", "text": 'a"b'}]) == '[["s1","a\\"b"]]' diff --git a/frontend/src/app/pages/Workflows/app/EventTriggerRow.tsx b/frontend/src/app/pages/Workflows/app/EventTriggerRow.tsx index b6b8fa66..7f6c7d95 100644 --- a/frontend/src/app/pages/Workflows/app/EventTriggerRow.tsx +++ b/frontend/src/app/pages/Workflows/app/EventTriggerRow.tsx @@ -8,9 +8,6 @@ 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', @@ -42,16 +39,6 @@ const EventTriggerRow: React.FC = ({ workflow, trigger, onMutate, onRe 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 (
@@ -97,44 +84,33 @@ const EventTriggerRow: React.FC = ({ workflow, trigger, onMutate, onRe }} />
-
-
- 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 } })))} -
+
+ Watching for (checked automatically; speeds up when things happen) + { + const watchFor = e.target.value.trim(); + if (watchFor !== src.watch_for) onMutate((x) => ({ ...x, source: { ...src, watch_for: watchFor } })); + }} + />
)} {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 } })))} -
+
+ What counts as the event? An agent checks automatically with its tools. + { + const check = e.target.value.trim(); + if (check !== src.check) onMutate((x) => ({ ...x, source: { ...src, check } })); + }} + />
)} @@ -175,7 +151,9 @@ const EventTriggerRow: React.FC = ({ workflow, trigger, onMutate, onRe borderRadius: 8, fontSize: 10.5, color: WC.ink3, whiteSpace: 'pre-wrap', wordBreak: 'break-all', fontFamily: "'JetBrains Mono',monospace", userSelect: 'text', }}> - {`POST ${API_BASE}/events/ingest\n{"workflow_id": "${workflow.id}", "trigger_id": "${t.id}", "summary": "what happened", "dedup_key": "optional-id"}`} + {src.secret + ? `POST ${API_BASE}/events/ingest/${src.secret}\n{"summary": "what happened"}` + : `POST ${API_BASE}/events/ingest\n{"workflow_id": "${workflow.id}", "trigger_id": "${t.id}", "summary": "what happened"}`}
)} diff --git a/frontend/src/app/pages/Workflows/app/EventTriggersCard.tsx b/frontend/src/app/pages/Workflows/app/EventTriggersCard.tsx index e4093a76..08e87bb5 100644 --- a/frontend/src/app/pages/Workflows/app/EventTriggersCard.tsx +++ b/frontend/src/app/pages/Workflows/app/EventTriggersCard.tsx @@ -23,11 +23,11 @@ const ADD_CHOICES: Array<[TriggerKind, string]> = [ ]; 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 === 'file') return { kind: 'file', path: '', poll_seconds: 0 }; + if (kind === 'web') return { kind: 'web', url: '', watch_for: '', poll_seconds: 0 }; + if (kind === 'agent') return { kind: 'agent', check: '', model: '', poll_seconds: 0 }; if (kind === 'stream') return { kind: 'stream', url: '', contains: '' }; - return { kind: 'custom' }; + return { kind: 'custom', secret: crypto.randomUUID().replace(/-/g, '') }; } function newTrigger(kind: TriggerKind): EventTriggerConfig { diff --git a/frontend/src/shared/state/workflowsSlice.ts b/frontend/src/shared/state/workflowsSlice.ts index c45337b3..5e2bbe05 100644 --- a/frontend/src/shared/state/workflowsSlice.ts +++ b/frontend/src/shared/state/workflowsSlice.ts @@ -54,6 +54,8 @@ export interface AgentCheckSource { export interface CustomEventSource { /** Push-only: events arrive via POST /api/events/ingest from any script/webhook/Shortcut. */ kind: 'custom'; + /** Per-trigger push credential; the paste-one-URL form is /api/events/ingest/. */ + secret?: string; } export interface StreamSource {