Webhook receiver: Plane deliveries trigger a tick (PAD-7)

agentd webhook-serve (agentd/webhook.py, stdlib): verifies X-Plane-Signature
(HMAC-SHA256 hex of the body, constant-time; 401 on missing/bad), routes by
data.project to a configured tenant (others 202), coalesces deliveries within
3 s into one tick, runs one more tick when deliveries arrive during a tick,
and retries every 15 s for up to 2 min while the lock is held (tick command
exit 75: flock -E 75 in the cron scripts, cli tick on Busy). Ticks run in a
systemd scope so restarting the receiver never kills one. Log:
<log_dir>/webhook.log; GET /healthz.

Config: webhook section (key names only, PLANE_WEBHOOK_SECRET scrubbed from
child envs like the Plane tokens), notify: {kind: none} as a no-op hook
called on every state move; the unused per-project webhook_secret_key is
gone (Plane webhooks are workspace-level). systemd unit for the LXC and
tools/plane-webhook.py to create/rotate the Plane webhook as the workspace
admin (secret straight into the Vault, never printed).

agentd/cli.py

@@ -38,6 +38,14 @@ def cmd_tick(cfg, a):
         d.tick(project=a.project)
     except Busy as exc:
         print("skip: %s" % exc)
+        sys.exit(75)   # same code as the cron script's flock conflict: the webhook receiver retries later
+
+
+def cmd_webhook_serve(cfg, a):
+    if not cfg.webhook:
+        sys.exit("no webhook section in %s" % cfg.path)
+    from .webhook import serve
+    serve(cfg, a.bind, a.port)
 
 
 def cmd_new(cfg, a, pl=None, quiet=False):
@@ -159,6 +167,9 @@ def cmd_check_config(cfg, a):
         print("terminal %-10s %-13s %-8s executors=%s token %s: %s" % (
             name, t["transport"], t["os"], ",".join(t["executors"]), env, "set" if os.environ.get(env) else "MISSING"))
     print("cron/exec secret args: %s" % " ".join(cfg.secret_env_args()))
+    if cfg.webhook:
+        k = cfg.webhook["secret_key"]
+        print("webhook-serve secret arg: --secret-env %s=%s (%s)" % (k, k, "set" if os.environ.get(k) else "not in this env"))
 
 
 def parser():
@@ -205,6 +216,10 @@ def parser():
                    help="do not tick here; only wait for the scheduled dispatcher (use when it runs on another host)")
     s.add_argument("--as", dest="as_")
     s.set_defaults(fn=cmd_smoke)
+    s = sub.add_parser("webhook-serve", help="Plane webhook receiver: each delivery triggers a tick (agentd/webhook.py)")
+    s.add_argument("--bind", help="address (default: webhook.bind)")
+    s.add_argument("--port", type=int, help="port (default: webhook.port)")
+    s.set_defaults(fn=cmd_webhook_serve)
     sub.add_parser("check-config", help="validate the config and list token env vars").set_defaults(fn=cmd_check_config)
     return ap
 

agentd/config.py

@@ -148,6 +148,8 @@ class Config:
         self.executors = data["executors"]
         self.projects = data["projects"]
         self.agentd = data.get("agentd", {})
+        self.webhook = data.get("webhook", {})
+        self.notify = data.get("notify") or {"kind": "none"}
 
     @property
     def state_dir(self):
@@ -176,9 +178,10 @@ class Config:
         return os.environ.get(self.token_env(terminal))
 
     def secret_keys(self):
-        """Every env var name that carries a secret (Plane tokens, webhook secrets): never handed to executors."""
+        """Every env var name that carries a secret (Plane tokens, the webhook secret): never handed to children."""
         keys = {t["token_key"] for t in self.terminals.values()}
-        keys |= {p["webhook_secret_key"] for p in self.projects.values() if p.get("webhook_secret_key")}
+        if self.webhook.get("secret_key"):
+            keys.add(self.webhook["secret_key"])
         return keys
 
     def scrubbed_env(self):

agentd/core.py

@@ -458,6 +458,13 @@ class Agentd:
         client.set_state(ctx.pid, iid, state, users)
         if iid in ctx.issues:  # keep this tick's view consistent
             ctx.issues[iid].update(state=state, assignees=users)
+        self.notify(ctx, iid, state_key, user)
+
+    def notify(self, ctx, iid, event, user):
+        """Notification hook (design §7 phase 3), called on every state move (event = todo/waiting/review/...).
+        Config `notify: {kind: none}` is the only kind today: a no-op. An ntfy sender would go here."""
+        if self.cfg.notify.get("kind", "none") == "none":
+            return
 
     def post(self, ctx, client, iid, rec, text, pre=None):
         c = client.post_comment(ctx.pid, iid, text_to_html(text, pre))

agentd/webhook.py

@@ -0,0 +1,238 @@
+"""Plane webhook receiver (PAD-7): a delivery for a tenant project triggers a dispatcher tick right away; the
+2-minute cron tick stays as the fallback. Stdlib only.
+
+POST <path> (default /plane):
+  - X-Plane-Signature = hex HMAC-SHA256 of the request body with the webhook secret (Plane signs
+    json.dumps(payload) and sends the same bytes); missing or wrong -> 401, nothing runs.
+  - body {"event": "issue"|"issue_comment", "action": ..., "data": {..., "project": <uuid>}, "activity": ...};
+    data.project not a configured tenant (or absent, e.g. on delete) -> 202, nothing runs.
+  - otherwise -> 200 and the Trigger schedules a tick.
+GET /healthz -> 200 JSON (pending / running / last tick), no secrets.
+
+Trigger: deliveries within `debounce_seconds` coalesce into one tick; deliveries that arrive while a tick runs cause
+one more tick after it (the running tick may have read Plane before the change). The tick command exits 75 when the
+cron lock (or agentd's own state lock) is held: then it is retried every `retry_seconds` for up to
+`retry_window_seconds`; after that the next cron tick picks the change up. Every delivery and tick is logged to
+<log_dir>/webhook.log."""
+import hashlib
+import hmac
+import json
+import os
+import subprocess
+import threading
+import time
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+from . import ROOT
+
+BUSY = 75             # tick command exit code: lock held by another tick (flock -E 75, cli tick on Busy)
+MAX_BODY = 2 << 20    # Plane payloads are a few KB
+
+
+def sign(secret, body):
+    return hmac.new(secret.encode("utf-8"), body, hashlib.sha256).hexdigest()
+
+
+def verify(secret, body, signature):
+    """Constant-time check of the hex signature against the raw body; falls back to Plane's own serialisation
+    (json.dumps of the parsed payload) in case a proxy re-encoded the body."""
+    if not secret or not signature:
+        return False
+    sig = signature.strip().lower().encode("ascii", "replace")
+    if hmac.compare_digest(sign(secret, body).encode(), sig):
+        return True
+    try:
+        canonical = json.dumps(json.loads(body)).encode("utf-8")
+    except ValueError:
+        return False
+    return canonical != body and hmac.compare_digest(sign(secret, canonical).encode(), sig)
+
+
+def project_of(payload):
+    data = payload.get("data") if isinstance(payload, dict) else None
+    pid = (data or {}).get("project") or (data or {}).get("project_id")
+    return pid.get("id") if isinstance(pid, dict) else pid
+
+
+def route(payload, projects):
+    """Configured project name for the delivery, or None."""
+    pid = project_of(payload)
+    return next((name for name, p in projects.items() if pid and p["id"] == pid), None)
+
+
+class Logger:
+    def __init__(self, path):
+        self.path = path
+        self.lock = threading.Lock()
+
+    def __call__(self, msg):
+        line = time.strftime("%Y-%m-%dT%H:%M:%S") + " " + msg + "\n"
+        with self.lock:
+            if self.path:
+                os.makedirs(os.path.dirname(self.path), exist_ok=True)
+                with open(self.path, "a", encoding="utf-8") as fh:
+                    fh.write(line)
+            else:
+                print(line, end="", flush=True)
+
+
+class Trigger:
+    """Debounced tick trigger. `run()` executes one tick and returns its exit code (BUSY = lock held)."""
+
+    def __init__(self, run, debounce=3.0, retry=15.0, retry_window=120.0, log=print, clock=time.monotonic):
+        self.run, self.debounce, self.retry, self.retry_window = run, debounce, retry, retry_window
+        self.log, self.clock = log, clock
+        self.cv = threading.Condition()
+        self.due = None           # clock() when the next tick starts; None = nothing pending
+        self.reasons = []         # deliveries folded into the pending tick
+        self.busy_since = None    # first BUSY of the current retry chain
+        self.running = False
+        self.last = {}
+        self.ticks = 0
+
+    def kick(self, reason):
+        with self.cv:
+            if self.due is None:
+                self.due = self.clock() + self.debounce
+            self.reasons.append(reason)
+            self.cv.notify()
+
+    def status(self):
+        with self.cv:
+            return {"pending": self.due is not None, "queued": len(self.reasons), "running": self.running,
+                    "retrying_since": self.busy_since and round(self.clock() - self.busy_since, 1), "ticks": self.ticks,
+                    "last": dict(self.last)}
+
+    def start(self):
+        threading.Thread(target=self.serve_forever, name="tick-trigger", daemon=True).start()
+        return self
+
+    def serve_forever(self):
+        while True:
+            self.step()
+
+    def step(self, wait=None):
+        """Wait until a tick is due (or `wait` seconds pass), run it, handle the result. True if a tick ran."""
+        with self.cv:
+            end = None if wait is None else self.clock() + wait
+            while self.due is None or self.due > self.clock():
+                timeout = None if self.due is None else self.due - self.clock()
+                if end is not None:
+                    left = end - self.clock()
+                    if left <= 0:
+                        return False
+                    timeout = left if timeout is None else min(timeout, left)
+                self.cv.wait(timeout)
+            reasons, self.reasons, self.due, self.running = self.reasons, [], None, True
+        start = self.clock()
+        self.log("tick start (%d deliveries: %s)" % (len(reasons), ", ".join(reasons[:5]) + (" …" if len(reasons) > 5 else "")))
+        try:
+            rc = self.run()
+        except Exception as exc:   # a broken runner must not kill the trigger thread
+            rc = "error %r" % (exc,)
+        took = round(self.clock() - start, 1)
+        with self.cv:
+            self.running = False
+            self.ticks += 1
+            self.last = {"rc": rc, "seconds": took, "deliveries": len(reasons), "at": time.strftime("%Y-%m-%dT%H:%M:%S")}
+            if rc != BUSY:
+                self.busy_since = None
+                self.log("tick done rc=%s in %.1fs" % (rc, took))
+                return True
+            now = self.clock()
+            self.busy_since = self.busy_since or start
+            if now - self.busy_since + self.retry > self.retry_window:
+                self.log("tick busy for %.0fs, giving up; the next cron tick covers it" % (now - self.busy_since))
+                self.busy_since = None
+                return True
+            self.reasons = reasons + self.reasons
+            self.due = min(self.due, now + self.retry) if self.due is not None else now + self.retry
+            self.log("tick busy (another tick holds the lock), retry in %gs" % self.retry)
+            return True
+
+
+def command_runner(cmd, env, timeout=1000):
+    """Run the tick command (it logs to cron.log itself); returns its exit code."""
+    def run():
+        try:
+            return subprocess.run(cmd, env=env, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
+                                  stderr=subprocess.DEVNULL, timeout=timeout).returncode
+        except subprocess.TimeoutExpired:
+            return "timeout"
+    return run
+
+
+def make_handler(secret, projects, trigger, log, path="/plane"):
+    class Handler(BaseHTTPRequestHandler):
+        server_version = "plane-agentd-webhook"
+
+        def log_message(self, fmt, *args):   # deliveries are logged by us, one line each
+            pass
+
+        def reply(self, code, obj):
+            body = json.dumps(obj).encode()
+            self.send_response(code)
+            self.send_header("Content-Type", "application/json")
+            self.send_header("Content-Length", str(len(body)))
+            self.end_headers()
+            self.wfile.write(body)
+
+        def do_GET(self):
+            if self.path.split("?")[0] == "/healthz":
+                return self.reply(200, dict(ok=True, **trigger.status()))
+            self.reply(404, {"error": "not found"})
+
+        def do_POST(self):
+            if self.path.split("?")[0] != path:
+                return self.reply(404, {"error": "not found"})
+            delivery = self.headers.get("X-Plane-Delivery", "-")
+            try:
+                n = int(self.headers.get("Content-Length", ""))
+            except ValueError:
+                n = -1
+            if not 0 <= n <= MAX_BODY:
+                log("delivery=%s from=%s decision=400 bad length" % (delivery, self.client_address[0]))
+                return self.reply(400, {"error": "bad length"})
+            body = self.rfile.read(n)
+            sig = self.headers.get("X-Plane-Signature")
+            if not verify(secret, body, sig):
+                log("delivery=%s from=%s decision=401 %s signature" % (
+                    delivery, self.client_address[0], "bad" if sig else "missing"))
+                return self.reply(401, {"error": "signature"})
+            try:
+                payload = json.loads(body)
+            except ValueError:
+                log("delivery=%s decision=400 bad json" % delivery)
+                return self.reply(400, {"error": "json"})
+            event = "%s/%s" % (payload.get("event") or self.headers.get("X-Plane-Event"), payload.get("action"))
+            name = route(payload, projects)
+            what = "delivery=%s event=%s project=%s item=%s" % (
+                delivery, event, name or project_of(payload) or "-", (payload.get("data") or {}).get("id", "-"))
+            if not name:
+                log(what + " decision=202 ignored (not a tenant project)")
+                return self.reply(202, {"ignored": "project"})
+            trigger.kick(delivery)
+            log(what + " decision=tick")
+            self.reply(200, {"tick": "scheduled"})
+
+    return Handler
+
+
+def serve(cfg, bind=None, port=None):
+    """`agentd webhook-serve`: the secret comes from the environment (agent-access.py exec --secret-env)."""
+    w = cfg.webhook
+    secret = os.environ.get(w["secret_key"])
+    if not secret:
+        raise SystemExit("webhook secret %s not in the environment; run under agent-access.py exec --secret-env %s=%s"
+                         % (w["secret_key"], w["secret_key"], w["secret_key"]))
+    log = Logger(str(cfg.log_dir / "webhook.log"))
+    cmd = list(w["tick_command"])
+    if "/" in cmd[0] and not os.path.isabs(cmd[0]):
+        cmd[0] = str(ROOT / cmd[0])   # relative script path = repo
+    run = command_runner(cmd, cfg.scrubbed_env(), w.get("tick_timeout_seconds", 1000))
+    trigger = Trigger(run, w.get("debounce_seconds", 3), w.get("retry_seconds", 15), w.get("retry_window_seconds", 120),
+                      log=log).start()
+    bind, port = bind or w.get("bind", "127.0.0.1"), port or w.get("port", 8790)
+    httpd = ThreadingHTTPServer((bind, port), make_handler(secret, cfg.projects, trigger, log, w.get("path", "/plane")))
+    log("listening on http://%s:%s%s (tenants: %s)" % (bind, port, w.get("path", "/plane"), ", ".join(cfg.projects)))
+    httpd.serve_forever()