feat(webhook): cron jobs can fire from app events, not just the clock (ChatGPT Work-inspired)

gateway/platforms/webhook.py

@@ -17,6 +17,13 @@
     message that gets delivered.  Use for external push notifications
     (Supabase, monitoring alerts, inter-agent pings) where zero LLM cost
     and sub-second delivery matter more than agent reasoning.
+  - cron_job: fire an existing cron job (by ID or name) on each event
+    instead of starting a fresh webhook agent session.  The rendered
+    prompt becomes transient per-run context for that job; the job's own
+    prompt, skills, model, and delivery settings apply.  Turns any cron
+    job into an event-triggered task (inspired by ChatGPT Work's
+    webhook-triggered scheduled tasks, Aug 2026).  Mutually exclusive
+    with deliver_only.
 
 Security:
   - HMAC secret is required per route (validated at startup)
@@ -275,6 +282,14 @@ async def connect(self, *, is_reconnect: bool = False) -> bool:
             # Validate up-front so misconfiguration surfaces at startup rather
             # than on the first webhook POST.
             if route.get("deliver_only"):
+                if route.get("cron_job"):
+                    raise ValueError(
+                        f"[webhook] Route '{name}' sets both deliver_only and "
+                        f"cron_job. They are mutually exclusive: deliver_only "
+                        f"pushes the rendered template as a message, cron_job "
+                        f"fires an existing cron job (which handles its own "
+                        f"delivery)."
+                    )
                 deliver = route.get("deliver", "log")
                 if not deliver or deliver == "log":
                     raise ValueError(
@@ -861,6 +876,67 @@ async def _handle_webhook(self, request: "web.Request") -> "web.Response":
                 status=200,
             )
 
+        # ── Cron-job trigger mode (cron_job) ────────────────────
+        # Inspired by ChatGPT Work's webhook-triggered scheduled tasks
+        # (Aug 25 2026): an app event (Gmail/Slack/GitHub/anything that can
+        # POST) fires an EXISTING cron job instead of polling on a cadence.
+        # The rendered prompt template becomes transient per-run context
+        # (same rail as cronjob(action='run', prompt=...)); the job's own
+        # stored prompt, skills, model, and delivery settings all apply.
+        # Reuses the same HMAC auth, rate limiting, filters, script, and
+        # idempotency that protect agent-mode routes above.
+        if route_config.get("cron_job"):
+            job_ref = str(route_config["cron_job"])
+            event_context = (
+                f"This run was triggered by webhook event '{event_type}' "
+                f"on route '{route_name}' (not the schedule).\n\n{prompt}"
+            )
+            logger.info(
+                "[webhook] cron-trigger event=%s route=%s job=%s delivery=%s",
+                event_type,
+                route_name,
+                job_ref,
+                delivery_id,
+            )
+
+            async def _fire_cron_job() -> None:
+                try:
+                    from tools.cronjob_tools import execute_job_for_event
+
+                    # The job is a full agent run (minutes); keep it off the
+                    # gateway event loop.
+                    result = await asyncio.to_thread(
+                        execute_job_for_event, job_ref, event_context
+                    )
+                    if not result.get("success"):
+                        logger.warning(
+                            "[webhook] cron-trigger job=%s route=%s did not "
+                            "complete cleanly: %s",
+                            job_ref,
+                            route_name,
+                            result.get("error"),
+                        )
+                except Exception:
+                    logger.exception(
+                        "[webhook] cron-trigger failed job=%s route=%s",
+                        job_ref,
+                        route_name,
+                    )
+
+            task = asyncio.create_task(_fire_cron_job())
+            self._background_tasks.add(task)
+            task.add_done_callback(self._background_tasks.discard)
+            return web.json_response(
+                {
+                    "status": "accepted",
+                    "route": route_name,
+                    "cron_job": job_ref,
+                    "event": event_type,
+                    "delivery_id": delivery_id,
+                },
+                status=202,
+            )
+
         # ── Direct delivery mode (deliver_only) ─────────────────
         # Skip the agent entirely — the rendered prompt IS the message we
         # deliver.  Use case: external services (Supabase, monitoring,

hermes_cli/subcommands/webhook.py

@@ -62,6 +62,15 @@ def build_webhook_parser(subparsers, *, cmd_webhook: Callable) -> None:
         "payload is passed as JSON on stdin; empty stdout, [SILENT], or a "
         "nonzero exit code ignores the webhook.",
     )
+    wh_sub.add_argument(
+        "--cron-job",
+        default="",
+        help="Fire an existing cron job (by ID or name) when this route "
+        "receives an event, instead of starting a fresh agent run. The "
+        "rendered --prompt template is passed to the job as transient "
+        "per-run context; the job's own prompt, skills, and delivery "
+        "settings apply. Mutually exclusive with --deliver-only.",
+    )
 
     webhook_subparsers.add_parser(
         "list", aliases=["ls"], help="List all dynamic subscriptions"

hermes_cli/webhook.py

@@ -182,6 +182,14 @@ def _cmd_subscribe(args):
     }
 
     if getattr(args, "deliver_only", False):
+        if getattr(args, "cron_job", ""):
+            print(
+                "Error: --deliver-only and --cron-job are mutually exclusive. "
+                "--deliver-only pushes the rendered template as a message; "
+                "--cron-job fires an existing cron job (which handles its own "
+                "delivery)."
+            )
+            return
         if route["deliver"] == "log":
             print(
                 "Error: --deliver-only requires --deliver to be a real target "
@@ -190,6 +198,31 @@ def _cmd_subscribe(args):
             return
         route["deliver_only"] = True
 
+    cron_job = (getattr(args, "cron_job", "") or "").strip()
+    if cron_job:
+        # Validate the reference up-front so a typo surfaces here, not on
+        # the first inbound event.
+        try:
+            from cron.jobs import AmbiguousJobReference, resolve_job_ref
+
+            try:
+                job = resolve_job_ref(cron_job)
+            except AmbiguousJobReference as e:
+                print(f"Error: {e}")
+                return
+            if job is None:
+                print(
+                    f"Error: no cron job matches '{cron_job}'. "
+                    "List jobs with: hermes cron list"
+                )
+                return
+            cron_job = job["id"]
+            route["cron_job"] = cron_job
+        except ImportError:
+            # Cron subsystem unavailable — store the reference as-is; the
+            # adapter resolves it per-event.
+            route["cron_job"] = cron_job
+
     script = getattr(args, "script", "") or ""
     if script.strip():
         route["script"] = script.strip()
@@ -213,6 +246,8 @@ def _cmd_subscribe(args):
     print(f"  Deliver: {route['deliver']}")
     if route.get("deliver_only"):
         print("  Mode: direct delivery (no agent, zero LLM cost)")
+    if route.get("cron_job"):
+        print(f"  Mode: cron-job trigger — fires job '{route['cron_job']}' on each event")
     if route.get("prompt"):
         prompt_preview = route["prompt"][:80] + ("..." if len(route["prompt"]) > 80 else "")
         label = "Message" if route.get("deliver_only") else "Prompt"
@@ -238,6 +273,8 @@ def _cmd_list(args):
         deliver = route.get("deliver", "log")
         if route.get("deliver_only"):
             deliver = f"{deliver} (direct — no agent)"
+        if route.get("cron_job"):
+            deliver = f"cron job '{route['cron_job']}'"
         desc = route.get("description", "")
         print(f"  ◆ {name}")
         if desc:

tests/gateway/test_webhook_cron_trigger.py

@@ -0,0 +1,217 @@
+"""Tests for the webhook adapter's ``cron_job`` route mode.
+
+``cron_job`` routes turn an existing cron job into an event-triggered task
+(inspired by ChatGPT Work's webhook-triggered scheduled tasks, Aug 2026):
+an inbound webhook event fires the referenced job through the same
+claimed-run body a manual ``cronjob(action='run')`` uses, instead of
+starting a fresh webhook agent session.
+
+Covers:
+- The referenced job is fired via ``execute_job_for_event`` with the
+  rendered prompt as transient per-run context
+- The normal webhook agent session is NOT started (``handle_message``
+  never called)
+- HTTP returns 202 Accepted immediately
+- Startup validation rejects routes that set both ``cron_job`` and
+  ``deliver_only``
+- ``execute_job_for_event`` resolves refs and fails cleanly on unknowns
+"""
+
+import asyncio
+import json
+from unittest.mock import patch
+
+import pytest
+from aiohttp import web
+from aiohttp.test_utils import TestClient, TestServer
+
+from gateway.config import PlatformConfig
+from gateway.platforms.webhook import WebhookAdapter, _INSECURE_NO_AUTH
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+def _make_adapter(routes, **extra_kw) -> WebhookAdapter:
+    extra = {"host": "127.0.0.1", "port": 0, "routes": routes}
+    extra.update(extra_kw)
+    config = PlatformConfig(enabled=True, extra=extra)
+    return WebhookAdapter(config)
+
+
+def _create_app(adapter: WebhookAdapter) -> web.Application:
+    app = web.Application()
+    app.router.add_post("/webhooks/{route_name}", adapter._handle_webhook)
+    return app
+
+
+async def _drain_background_tasks(adapter: WebhookAdapter) -> None:
+    tasks = list(adapter._background_tasks)
+    if tasks:
+        await asyncio.gather(*tasks, return_exceptions=True)
+
+
+# ===================================================================
+# Core behaviour: event fires the cron job, not a webhook session
+# ===================================================================
+
+class TestCronJobTrigger:
+    @pytest.mark.asyncio
+    async def test_post_fires_job_with_event_context(self):
+        routes = {
+            "pr-feedback": {
+                "secret": _INSECURE_NO_AUTH,
+                "cron_job": "review-sweeper",
+                "prompt": "PR #{number} received feedback: {review.body}",
+            }
+        }
+        adapter = _make_adapter(routes)
+
+        handle_message_calls = []
+
+        async def _capture(event):
+            handle_message_calls.append(event)
+
+        adapter.handle_message = _capture
+
+        fired = []
+
+        def _fake_execute(job_ref, extra_prompt=None):
+            fired.append((job_ref, extra_prompt))
+            return {"claimed": True, "success": True, "error": None}
+
+        app = _create_app(adapter)
+        body = json.dumps(
+            {"number": 7, "review": {"body": "needs tests"}}
+        ).encode()
+
+        with patch(
+            "tools.cronjob_tools.execute_job_for_event",
+            side_effect=_fake_execute,
+        ):
+            async with TestClient(TestServer(app)) as cli:
+                resp = await cli.post(
+                    "/webhooks/pr-feedback",
+                    data=body,
+                    headers={
+                        "Content-Type": "application/json",
+                        "X-GitHub-Delivery": "delivery-cron-1",
+                        "X-GitHub-Event": "pull_request_review",
+                    },
+                )
+                assert resp.status == 202
+                data = await resp.json()
+                assert data["status"] == "accepted"
+                assert data["cron_job"] == "review-sweeper"
+                await _drain_background_tasks(adapter)
+
+        # Job fired exactly once with the rendered prompt as run context
+        assert len(fired) == 1
+        job_ref, extra_prompt = fired[0]
+        assert job_ref == "review-sweeper"
+        assert "PR #7 received feedback: needs tests" in extra_prompt
+        assert "pull_request_review" in extra_prompt  # event provenance
+
+        # No fresh webhook agent session was started
+        assert handle_message_calls == []
+
+    @pytest.mark.asyncio
+    async def test_job_failure_does_not_break_http_response(self):
+        routes = {
+            "flaky": {"secret": _INSECURE_NO_AUTH, "cron_job": "gone-job"}
+        }
+        adapter = _make_adapter(routes)
+        app = _create_app(adapter)
+
+        def _fail(job_ref, extra_prompt=None):
+            return {
+                "claimed": False,
+                "success": False,
+                "error": "Cron job 'gone-job' not found.",
+            }
+
+        with patch(
+            "tools.cronjob_tools.execute_job_for_event", side_effect=_fail
+        ):
+            async with TestClient(TestServer(app)) as cli:
+                resp = await cli.post(
+                    "/webhooks/flaky",
+                    data=b"{}",
+                    headers={
+                        "Content-Type": "application/json",
+                        "X-GitHub-Delivery": "delivery-cron-2",
+                    },
+                )
+                # Fire-and-forget: the POST is accepted even when the job
+                # later fails; the failure is logged, not surfaced.
+                assert resp.status == 202
+                await _drain_background_tasks(adapter)
+
+
+# ===================================================================
+# Startup validation
+# ===================================================================
+
+class TestCronJobRouteValidation:
+    @pytest.mark.asyncio
+    async def test_cron_job_plus_deliver_only_rejected_at_connect(self):
+        routes = {
+            "bad": {
+                "secret": "s3cret",
+                "cron_job": "some-job",
+                "deliver_only": True,
+                "deliver": "telegram",
+            }
+        }
+        adapter = _make_adapter(routes)
+        with pytest.raises(ValueError, match="mutually exclusive"):
+            await adapter.connect()
+
+
+# ===================================================================
+# execute_job_for_event unit behaviour
+# ===================================================================
+
+class TestExecuteJobForEvent:
+    def test_unknown_job_returns_error(self):
+        from tools import cronjob_tools
+
+        with patch.object(cronjob_tools, "resolve_job_ref", return_value=None):
+            result = cronjob_tools.execute_job_for_event("nope")
+        assert result["claimed"] is False
+        assert result["success"] is False
+        assert "not found" in result["error"]
+
+    def test_ambiguous_ref_returns_error(self):
+        from cron.jobs import AmbiguousJobReference
+        from tools import cronjob_tools
+
+        with patch.object(
+            cronjob_tools,
+            "resolve_job_ref",
+            side_effect=AmbiguousJobReference(
+                "x", [{"id": "job-a"}, {"id": "job-b"}]
+            ),
+        ):
+            result = cronjob_tools.execute_job_for_event("x")
+        assert result["claimed"] is False
+        assert result["success"] is False
+        assert "ambiguous" in result["error"].lower()
+
+    def test_resolved_job_fires_with_extra_prompt(self):
+        from tools import cronjob_tools
+
+        job = {"id": "job-123", "name": "sweeper"}
+        with patch.object(
+            cronjob_tools, "resolve_job_ref", return_value=job
+        ), patch.object(
+            cronjob_tools,
+            "_execute_job_now",
+            return_value={"claimed": True, "success": True, "error": None},
+        ) as mock_exec:
+            result = cronjob_tools.execute_job_for_event(
+                "sweeper", extra_prompt="event context"
+            )
+        assert result["success"] is True
+        mock_exec.assert_called_once_with(job, extra_prompt="event context")