feat(watchers): interval-polling watcher engine with watermark dedup

cron/scheduler.py

@@ -1724,6 +1724,20 @@ def _process_job(job: dict) -> bool:
         except Exception as _e:
             logger.debug("Post-tick MCP orphan cleanup failed: %s", _e)
 
+        # Piggyback watcher tick — runs after cron jobs so cron stays the
+        # priority path.  Watchers have their own file lock and are safe to
+        # invoke from any tick.  Best-effort: if the watcher subsystem is
+        # missing (e.g. mid-migration), cron keeps working.
+        try:
+            from watchers.engine import tick as _watcher_tick
+            _watcher_outcomes = _watcher_tick(adapters=adapters, loop=loop, verbose=verbose)
+            if verbose and _watcher_outcomes:
+                logger.info("%d watcher(s) polled", len(_watcher_outcomes))
+        except ImportError:
+            pass
+        except Exception as _wexc:
+            logger.warning("Watcher tick failed (non-fatal): %s", _wexc)
+
         return sum(_results)
     finally:
         if fcntl:

hermes_cli/main.py

@@ -5239,12 +5239,19 @@ def cmd_cron(args):
 
 
 def cmd_webhook(args):
-    """Webhook subscription management."""
+    """Entry point for 'hermes webhook' command."""
     from hermes_cli.webhook import webhook_command
 
     webhook_command(args)
 
 
+def cmd_watch(args):
+    """Entry point for 'hermes watch' command."""
+    from hermes_cli.watchers import watch_command
+
+    watch_command(args)
+
+
 def cmd_slack(args):
     """Slack integration helpers.
 
@@ -8070,6 +8077,7 @@ def _coalesce_session_name_args(argv: list) -> list:
         "plugins",
         "acp",
         "webhook",
+        "watch",
         "memory",
         "dump",
         "debug",
@@ -9265,6 +9273,83 @@ def main():
 
     webhook_parser.set_defaults(func=cmd_webhook)
 
+    # =========================================================================
+    # watch command — interval-polling watchers (pull-based sibling of webhook)
+    # =========================================================================
+    watch_parser = subparsers.add_parser(
+        "watch",
+        help="Manage interval-polling watchers (HTTP JSON, RSS, GitHub, ...)",
+        description=(
+            "Watchers poll an external source on an interval, detect new items via "
+            "watermark-based dedup, and deliver the result (verbatim in --deliver-only "
+            "mode, or as prompt context to a short-lived agent otherwise)."
+        ),
+    )
+    watch_subparsers = watch_parser.add_subparsers(dest="watch_action")
+
+    w_add = watch_subparsers.add_parser(
+        "add", aliases=["subscribe"], help="Create or update a watcher"
+    )
+    w_add.add_argument("name", help="Watcher name (lowercase alphanumerics + '-'/'_')")
+    w_add.add_argument(
+        "--provider",
+        required=True,
+        help="Provider name: http_json | rss | github | <custom>",
+    )
+    w_add.add_argument(
+        "--interval", type=int, default=300, help="Poll interval in seconds (default: 300)"
+    )
+    w_add.add_argument("--prompt", default="", help="Prompt template (supports {items_json}, {name}, {count})")
+    w_add.add_argument("--skills", default="", help="Comma-separated skill names to load")
+    w_add.add_argument(
+        "--deliver",
+        default="origin",
+        help="Delivery target: origin | local | multi | all | telegram | telegram:-100:17 | ... (see 'hermes cron' docs)",
+    )
+    w_add.add_argument(
+        "--deliver-only",
+        action="store_true",
+        help="Skip the agent — deliver the rendered prompt directly. Zero LLM cost.",
+    )
+    w_add.add_argument("--url", default="", help="Shortcut for --arg url=<URL> (http_json / rss)")
+    w_add.add_argument("--repo", default="", help="Shortcut for --arg repo=owner/name (github)")
+    w_add.add_argument(
+        "--scope",
+        default="",
+        help="github scope: issues | pulls | releases | commits (default: issues)",
+    )
+    w_add.add_argument(
+        "--arg",
+        action="append",
+        default=[],
+        help="Additional provider config as key=value (repeatable)",
+    )
+    w_add.add_argument(
+        "--config",
+        default="",
+        help="Full provider config as a JSON object (merged with --arg)",
+    )
+
+    w_list = watch_subparsers.add_parser("list", aliases=["ls"], help="List all watchers")
+    w_list.add_argument("--verbose", action="store_true", help="Include last_error detail")
+
+    w_rm = watch_subparsers.add_parser("remove", aliases=["rm"], help="Remove a watcher")
+    w_rm.add_argument("name", help="Watcher name to remove")
+
+    w_run = watch_subparsers.add_parser(
+        "run", help="Fire one watcher once, out of band (respects dedup)"
+    )
+    w_run.add_argument("name", help="Watcher name to run")
+
+    w_reset = watch_subparsers.add_parser(
+        "reset", help="Clear a watcher's watermark — next run treats it as a first poll"
+    )
+    w_reset.add_argument("name", help="Watcher name to reset")
+
+    watch_subparsers.add_parser("tick", help="Poll every due watcher (ad-hoc)")
+
+    watch_parser.set_defaults(func=cmd_watch)
+
     # =========================================================================
     # kanban command — multi-profile collaboration board
     # =========================================================================

hermes_cli/watchers.py

@@ -0,0 +1,215 @@
+"""hermes watch — manage polling watchers from the CLI.
+
+Usage:
+    hermes watch add <name> --provider <p> --url <url> [--interval 300] ...
+    hermes watch list
+    hermes watch remove <name>
+    hermes watch run <name>          # fire the watcher once, out of band
+    hermes watch reset <name>        # clear the watermark; next run replays all
+
+Watchers poll an external source on an interval, detect new items via
+watermark-based dedup, and deliver the result (verbatim or via agent).
+Subscriptions persist to ~/.hermes/watchers.json.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+import time
+from typing import Any, Dict, List
+
+from hermes_constants import display_hermes_home
+from watchers.engine import run_watcher, tick
+from watchers.providers import PROVIDERS
+from watchers.store import (
+    WatcherSubscription,
+    delete_watcher,
+    get_watcher,
+    list_watchers,
+    load_watermark,
+    save_watcher,
+    save_watermark,
+)
+
+
+_NAME_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$")
+
+
+def _parse_kv_pairs(items: List[str], *, label: str) -> Dict[str, str]:
+    """Parse a list of ``k=v`` strings into a dict; errors out cleanly."""
+    out: Dict[str, str] = {}
+    for item in items or []:
+        if "=" not in item:
+            raise ValueError(
+                f"--{label} expects key=value pairs; got {item!r}"
+            )
+        k, v = item.split("=", 1)
+        out[k.strip()] = v.strip()
+    return out
+
+
+def _parse_config_json(raw: str) -> Dict[str, Any]:
+    raw = (raw or "").strip()
+    if not raw:
+        return {}
+    try:
+        parsed = json.loads(raw)
+    except json.JSONDecodeError as e:
+        raise ValueError(f"--config must be valid JSON: {e}") from e
+    if not isinstance(parsed, dict):
+        raise ValueError(f"--config must be a JSON object; got {type(parsed).__name__}")
+    return parsed
+
+
+def watch_command(args) -> None:
+    """Entry point for 'hermes watch' subcommand."""
+    sub = getattr(args, "watch_action", None)
+
+    if not sub:
+        print("Usage: hermes watch {add|list|remove|run|reset}")
+        print("Run 'hermes watch --help' for details.")
+        return
+
+    if sub in ("add", "subscribe"):
+        _cmd_add(args)
+    elif sub in ("list", "ls"):
+        _cmd_list(args)
+    elif sub in ("remove", "rm"):
+        _cmd_remove(args)
+    elif sub == "run":
+        _cmd_run(args)
+    elif sub == "reset":
+        _cmd_reset(args)
+    elif sub == "tick":
+        _cmd_tick(args)
+    else:
+        print(f"Unknown watch subcommand: {sub}")
+
+
+def _cmd_add(args) -> None:
+    name = (args.name or "").strip().lower().replace(" ", "-")
+    if not _NAME_RE.match(name):
+        print(f"Error: name must be lowercase alphanumerics + '-'/'_' (got {args.name!r})")
+        return
+
+    provider = (args.provider or "").lower()
+    if provider not in PROVIDERS:
+        print(f"Error: unknown provider {args.provider!r}.")
+        print(f"       Known providers: {sorted(PROVIDERS)}")
+        return
+
+    # Config assembly: --config JSON + individual --arg flags merged.
+    try:
+        cfg = _parse_config_json(getattr(args, "config", "") or "")
+        cfg.update(_parse_kv_pairs(getattr(args, "arg", None) or [], label="arg"))
+    except ValueError as e:
+        print(f"Error: {e}")
+        return
+
+    # Convenience flags — providers that commonly need these get dedicated args.
+    if getattr(args, "url", None):
+        cfg.setdefault("url", args.url)
+    if getattr(args, "repo", None):
+        cfg.setdefault("repo", args.repo)
+    if getattr(args, "scope", None):
+        cfg.setdefault("scope", args.scope)
+
+    existing = get_watcher(name)
+    sub = WatcherSubscription(
+        name=name,
+        provider=provider,
+        config=cfg,
+        interval_seconds=int(args.interval),
+        prompt=args.prompt or "",
+        skills=[s.strip() for s in (args.skills or "").split(",") if s.strip()],
+        deliver=args.deliver or "origin",
+        deliver_only=bool(args.deliver_only),
+        enabled=True,
+        created_at=(existing.created_at if existing else time.time()),
+    )
+    save_watcher(sub)
+
+    verb = "Updated" if existing else "Added"
+    print(f"{verb} watcher '{name}':")
+    print(f"  provider       = {sub.provider}")
+    print(f"  interval       = {sub.interval_seconds}s")
+    print(f"  deliver        = {sub.deliver}")
+    print(f"  deliver_only   = {sub.deliver_only}")
+    print(f"  config         = {json.dumps(sub.config, indent=2)}")
+    if not existing:
+        print()
+        print(f"The watermark file will be created at {display_hermes_home()}/watchers/{name}.watermark.json")
+        print("on first successful poll.  The first poll records baseline state and")
+        print("does NOT fire; only items that appear AFTER the baseline are delivered.")
+
+
+def _cmd_list(args) -> None:
+    subs = list_watchers()
+    if not subs:
+        print("No watchers registered.")
+        print()
+        print("Add one with: hermes watch add <name> --provider <p> --url <url>")
+        return
+
+    print(f"{'NAME':<20} {'PROVIDER':<12} {'INTERVAL':<10} {'LAST RUN':<21} {'EVENTS':<7} {'STATUS'}")
+    for sub in sorted(subs, key=lambda s: s.name):
+        last_run = (
+            time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(sub.last_run_at))
+            if sub.last_run_at
+            else "never"
+        )
+        status = "disabled" if not sub.enabled else ("error" if sub.last_error else "ok")
+        print(
+            f"{sub.name:<20} {sub.provider:<12} {sub.interval_seconds!s:<10} "
+            f"{last_run:<21} {sub.last_event_count:<7} {status}"
+        )
+        if sub.last_error and getattr(args, "verbose", False):
+            print(f"    ↳ last_error: {sub.last_error}")
+
+
+def _cmd_remove(args) -> None:
+    name = args.name.strip().lower()
+    if delete_watcher(name):
+        print(f"Removed watcher '{name}' and cleared its watermark.")
+    else:
+        print(f"No watcher named '{name}'.")
+
+
+def _cmd_run(args) -> None:
+    name = args.name.strip().lower()
+    sub = get_watcher(name)
+    if not sub:
+        print(f"No watcher named '{name}'.")
+        return
+    print(f"Polling watcher '{name}' ({sub.provider})...")
+    outcome = run_watcher(sub)
+    sub.last_run_at = time.time()
+    sub.last_error = outcome.get("error")
+    sub.last_event_count = outcome.get("new_events", 0)
+    save_watcher(sub)
+    print(f"  status      = {outcome.get('status')}")
+    print(f"  new events  = {outcome.get('new_events')}")
+    if outcome.get("error"):
+        print(f"  error       = {outcome['error']}")
+
+
+def _cmd_reset(args) -> None:
+    name = args.name.strip().lower()
+    if not get_watcher(name):
+        print(f"No watcher named '{name}'.")
+        return
+    save_watermark(name, {})
+    print(f"Cleared watermark for '{name}'. Next poll will treat it as a first run.")
+
+
+def _cmd_tick(args) -> None:
+    """Fire every due watcher once and print outcomes (CLI-only ad-hoc poll)."""
+    outcomes = tick(verbose=True)
+    if not outcomes:
+        print("No watchers due.")
+        return
+    for o in outcomes:
+        flag = "!" if o["status"] != "ok" else " "
+        err = f"  [{o['error']}]" if o.get("error") else ""
+        print(f" {flag} {o['name']:<20} {o['status']:<8} events={o['new_events']}{err}")

tests/watchers/__init__.py

@@ -0,0 +1 @@
+"""Marker file so pytest discovers the watchers test package."""