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

@@ -161,6 +161,8 @@ class Agentd:
     def _reset_caches(self):
         self._clients, self._transports, self._executors, self._ready = {}, {}, {}, {}
         self._down = set()   # terminals whose relay failed during this tick
+        self._busy = {}      # terminal -> why terminal_ready() said no
+        self._skipped = []   # Todo items not dispatched this tick, with the reason (heartbeat "skipped")
 
     def mark_down(self, term, exc):
         if term not in self._down:
@@ -207,13 +209,33 @@ class Agentd:
                     self.log("%s tick error: %r" % (name, exc))
                 finally:
                     self.store.save(data)
-            self.store.heartbeat({"errors": errors, "down": sorted(self._down), "inflight": [
+            if not only_issue and self.note_skips(data, [project] if project else list(self.cfg.projects)):
+                self.store.save(data)
+            self.store.heartbeat({"errors": errors, "down": sorted(self._down), "skipped": self._skipped, "inflight": [
                 {"project": p, "run": r["inflight"]["run_id"], "terminal": r["inflight"]["terminal"]}
                 for p, ps in data["projects"].items() for r in ps["issues"].values() if r.get("inflight")],
                 "launching": [{"project": p, "run": r["launching"]["run_id"], "terminal": r["launching"]["terminal"]}
                               for p, ps in data["projects"].items() for r in ps["issues"].values() if r.get("launching")]})
             return data
 
+    def note_skips(self, data, projects):
+        """Log a skip reason only when it is new for that item (the cron log stays quiet on repeat ticks).
+        Persisted as data["skips"] {item: {project, terminal, reason}}; returns True when that changed."""
+        prev = data.get("skips", {})
+        now = {k: v for k, v in prev.items() if v.get("project") not in projects}
+        for s in self._skipped:
+            entry = {"project": s["project"], "terminal": s["terminal"], "reason": s["reason"]}
+            if prev.get(s["item"]) != entry:
+                self.log("%s not dispatched (%s): %s" % (s["item"], s["terminal"] or "-", s["reason"]))
+            now[s["item"]] = entry
+        changed = now != prev
+        data["skips"] = now
+        return changed
+
+    def skip(self, ctx, issue, term, reason):
+        self._skipped.append({"project": ctx.name, "item": ctx.key(issue.get("sequence_id")), "terminal": term,
+                              "reason": reason})
+
     def loop(self, interval):
         from .store import Busy
         while True:
@@ -436,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))
@@ -444,21 +473,24 @@ class Agentd:
 
     # ---------------------------------------------------------------- (b) pick / start
     def pick(self, ctx, data, pstate, only_issue=None):
-        load, groups = Counter(), set()
-        for p in data["projects"].values():
+        load, groups, holders = Counter(), {}, {}   # groups / holders: group / terminal -> keys of the runs holding it
+        for pname, p in data["projects"].items():
+            prefix = self.cfg.projects.get(pname, {}).get("key_prefix", pname)
             for r in p["issues"].values():
                 fl = r.get("inflight") or r.get("launching")   # an unconfirmed launch holds its slot
                 if fl:
                     load[fl["terminal"]] += 1
+                    holders.setdefault(fl["terminal"], []).append("%s-%s" % (prefix, r.get("seq")))
                     if fl.get("group"):
-                        groups.add(fl["group"])
+                        groups.setdefault(fl["group"], []).append("%s-%s" % (prefix, r.get("seq")))
         for g, info in list(data["paused_groups"].items()):
             if info["project"] == ctx.name:
                 issue = ctx.issues.get(info["issue"])
                 if issue is None or issue.get("state") != ctx.state_id("waiting"):
                     del data["paused_groups"][g]
                     self.log("serialize group %s resumed (%s left Waiting)" % (g, info.get("key")))
-        groups |= set(data["paused_groups"])
+        for g, info in data["paused_groups"].items():
+            groups.setdefault(g, []).append("%s in %s (paused)" % (info.get("key"), ctx.cfg["states"]["waiting"]))
         todo, running, now = ctx.state_id("todo"), ctx.state_id("running"), self.now()
         cands = []
         for iid, issue in ctx.issues.items():
@@ -477,20 +509,28 @@ class Agentd:
         for issue in sorted(cands, key=lambda i: i.get("sequence_id", 0)):
             route = self.route(ctx, issue)
             if isinstance(route, str):
-                if route:
-                    self.log("%s not dispatched: %s" % (ctx.key(issue.get("sequence_id")), route))
+                users = [ctx.member_names.get(a, a) for a in issue.get("assignees") or []]
+                self.skip(ctx, issue, None, route or "no route for assignee %s" % (",".join(users) or "(none)"))
                 continue
             term, group = route["terminal"], route["group"]
+            cap = self.cfg.terminals[term].get("concurrency", 1)
             if term in self._down:
+                self.skip(ctx, issue, term, "relay down")
                 continue
-            if load[term] >= self.cfg.terminals[term].get("concurrency", 1) or (group and group in groups):
+            if load[term] >= cap:
+                self.skip(ctx, issue, term, "concurrency %d/%d (%s)" % (load[term], cap, ", ".join(holders.get(term, [])) or "-"))
+                continue
+            if group and group in groups:
+                self.skip(ctx, issue, term, "serialize_group %s held by %s" % (group, ", ".join(groups[group])))
                 continue
             if not self.terminal_ready(term, load[term]):
+                self.skip(ctx, issue, term, self._busy.get(term) or "relay down")
                 continue
             try:
                 self.start(ctx, data, pstate, issue, route)
             except Exception as exc:
                 self.log("%s start failed on %s: %r" % (ctx.key(issue.get("sequence_id")), term, exc))
+                self.skip(ctx, issue, term, "start failed: %s" % (str(exc)[:200] or type(exc).__name__))
                 if isinstance(exc, TransportError):
                     self.mark_down(term, exc)
                 if not pstate["issues"].get(issue["id"], {}).get("launching"):
@@ -498,8 +538,9 @@ class Agentd:
                 # the launch call failed but may have started the run: its marker keeps the slot until probed
             self.store.save(data)  # a launched run must be on disk before anything else can fail
             load[term] += 1
+            holders.setdefault(term, []).append(ctx.key(issue.get("sequence_id")))
             if group:
-                groups.add(group)
+                groups.setdefault(group, []).append(ctx.key(issue.get("sequence_id")))
 
     def terminal_ready(self, term, our_load):
         """Optional guard against processes we did not start (e.g. a manual codex session)."""
@@ -515,7 +556,7 @@ class Agentd:
                     ready = False
                     break
                 elif count > 0 and our_load == 0:
-                    self.log("%s busy (%d %s running that agentd did not start)" % (term, count, name))
+                    self._busy[term] = "busy_process %s (%d running, not started by agentd)" % (name, count)
                     ready = False
             self._ready[term] = ready
         return self._ready[term]

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()

cli/agentd

@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-"""Entry point: agentd tick|new|show|comment|state|assign|smoke|check-config (see agentd/cli.py)."""
+"""Entry point: agentd tick|new|show|comment|state|assign|smoke|webhook-serve|check-config (see agentd/cli.py)."""
 import sys
 from pathlib import Path
 

config/agentd.example.yaml

@@ -13,6 +13,18 @@ agentd:
   comment_max_chars: 6000
   cli_terminal: oregon                  # identity used by `agentd new/comment/state/assign`
 
+# Plane webhook receiver (`agentd webhook-serve`, docs/runbook.md "Webhook"); leave the section out to run on cron only.
+webhook:
+  bind: 127.0.0.1
+  port: 8790
+  path: /plane
+  secret_key: PLANE_WEBHOOK_SECRET    # Vault key name of the Plane webhook secret (workspace-level webhook)
+  tick_command: [cron/agentd-tick.sh] # exits 75 when another tick holds the lock (flock -E 75)
+  debounce_seconds: 3
+  retry_seconds: 15
+  retry_window_seconds: 120
+notify: {kind: none}                  # notification hook: only `none` (no-op) exists today
+
 terminals:
   win10:
     transport: warpgate-ssh
@@ -72,7 +84,6 @@ projects:
   GWENT:
     id: 79cb5a60-aa4f-4138-9864-1dd70426e665
     key_prefix: GW
-    webhook_secret_key: PLANE_WEBHOOK_GWENT
     states: {backlog: Backlog, todo: Todo, running: In Progress, waiting: Waiting, review: Review, done: Done}
     reviewer: oregon                    # Plane user that gets [review] items
     answerer: oregon                    # Plane user that gets [question]/[blocked] and failures (default: reviewer)

config/agentd.yaml

@@ -13,6 +13,21 @@ agentd:
   comment_max_chars: 6000
   cli_terminal: oregon
 
+# Plane webhook receiver (`agentd webhook-serve`, systemd plane-agentd-webhook.service; docs/runbook.md "Webhook").
+# One workspace-level Plane webhook (work items + comments) -> POST http://10.77.10.249:8790/plane -> one tick.
+webhook:
+  bind: 10.77.10.249                  # LAN only (Plane VM 10.77.10.68 is on the same /24)
+  port: 8790
+  path: /plane
+  secret_key: PLANE_WEBHOOK_SECRET    # Vault key name; the value is injected by agent-access.py exec --secret-env
+  # the cron script holds `flock -n -E 75`; --scope moves the tick out of the receiver's cgroup, so restarting the
+  # receiver never kills a running tick
+  tick_command: [systemd-run, --scope, --quiet, --collect, /opt/plane-agentd/cron/agentd-tick-lxc.sh]
+  debounce_seconds: 3                 # deliveries within this window -> one tick
+  retry_seconds: 15                   # lock busy -> retry every 15 s ...
+  retry_window_seconds: 120           # ... for up to 2 min; then the cron tick covers it
+notify: {kind: none}                  # notification hook, no-op (no ntfy service); core calls it on every state move
+
 terminals:
   win10:
     transport: warpgate-ssh
@@ -85,4 +100,3 @@ projects:
     timeout_hours: 4
     max_rounds_per_thread: 8
     max_progress_rounds: 12
-    # notify (ntfy) is phase 3: left out on purpose

config/schema.json

@@ -36,6 +36,30 @@
         "cli_terminal": {"type": "string"}
       }
     },
+    "webhook": {
+      "type": "object",
+      "required": ["secret_key", "tick_command"],
+      "additionalProperties": false,
+      "properties": {
+        "bind": {"type": "string"},
+        "port": {"type": "integer", "minimum": 1},
+        "path": {"type": "string"},
+        "secret_key": {"type": "string"},
+        "tick_command": {"type": "array", "minItems": 1, "items": {"type": "string"}},
+        "tick_timeout_seconds": {"type": "integer", "minimum": 10},
+        "debounce_seconds": {"type": "number", "minimum": 0},
+        "retry_seconds": {"type": "number", "minimum": 1},
+        "retry_window_seconds": {"type": "number", "minimum": 0}
+      }
+    },
+    "notify": {
+      "type": "object",
+      "additionalProperties": false,
+      "properties": {
+        "kind": {"enum": ["none"]},
+        "on": {"$ref": "#/definitions/strings"}
+      }
+    },
     "terminals": {"type": "object", "minProperties": 1, "additionalProperties": {"$ref": "#/definitions/terminal"}},
     "executors": {"type": "object", "minProperties": 1, "additionalProperties": {"$ref": "#/definitions/executor"}},
     "projects": {"type": "object", "minProperties": 1, "additionalProperties": {"$ref": "#/definitions/project"}}
@@ -101,7 +125,6 @@
       "properties": {
         "id": {"type": "string"},
         "key_prefix": {"type": "string"},
-        "webhook_secret_key": {"type": "string"},
         "states": {
           "type": "object",
           "required": ["todo", "running", "waiting", "review", "done"],

cron/agentd-tick-lxc.sh

@@ -2,7 +2,9 @@
 # LXC agentd-01 cron entry (*/2, root): one agentd tick under server-access exec (Plane tokens only in the
 # process env), with a lock. The outer flock skips a tick while the previous one still runs; agentd also holds
 # its own state lock. Logs: /var/log/plane-agentd/{agentd.log,cron.log}; state: /var/lib/plane-agentd.
-exec flock -n /var/lib/plane-agentd/cron.lock \
+# Also the webhook receiver's tick command (agentd/webhook.py): exit 75 = lock held (flock -E 75, or agentd's own
+# state lock), the receiver then retries.
+exec flock -n -E 75 /var/lib/plane-agentd/cron.lock \
   timeout 900 python3 -I /root/.local/share/server-access-client/agent-access.py \
   --config https://nas.tail9531b.ts.net:8443/identity/agent.json exec \
   --secret-env PLANE_OREGON_API_TOKEN=PLANE_OREGON_API_TOKEN \

cron/agentd-tick.sh

@@ -2,7 +2,7 @@
 # oregon cron entry (*/2) -- ROLLBACK ONLY since PAD-11: the dispatcher runs on the LXC agentd-01
 # (cron/agentd-tick-lxc.sh). Never run both: two dispatchers double-dispatch. See docs/runbook.md "回滚到 oregon".
 # One agentd tick under server-access exec (Plane tokens in the process env only), with a lock.
-exec flock -n /home/ubuntu/.gwent-board/agentd.lock \
+exec flock -n -E 75 /home/ubuntu/.gwent-board/agentd.lock \
   timeout 900 python3 -I /home/ubuntu/.local/share/server-access-client/agent-access.py \
   --config https://nas.tail9531b.ts.net:8443/identity/agent.json exec \
   --secret-env PLANE_OREGON_API_TOKEN=PLANE_OREGON_API_TOKEN \

systemd/plane-agentd-webhook.service

@@ -0,0 +1,24 @@
+# Plane webhook receiver on the LXC agentd-01 (docs/runbook.md "Webhook", docs/deploy-lxc.md §8).
+# The webhook secret is injected by agent-access.py exec (Vault key PLANE_WEBHOOK_SECRET) into the process env only.
+# Install: cp systemd/plane-agentd-webhook.service /etc/systemd/system/ && systemctl daemon-reload &&
+#          systemctl enable --now plane-agentd-webhook
+[Unit]
+Description=plane-agentd Plane webhook receiver (triggers dispatcher ticks)
+Wants=network-online.target
+After=network-online.target
+StartLimitIntervalSec=0
+
+[Service]
+Type=simple
+Environment=HOME=/root
+WorkingDirectory=/opt/plane-agentd
+ExecStart=/usr/bin/python3 -I /root/.local/share/server-access-client/agent-access.py \
+  --config https://nas.tail9531b.ts.net:8443/identity/agent.json exec \
+  --secret-env PLANE_WEBHOOK_SECRET=PLANE_WEBHOOK_SECRET --timeout 315360000 -- \
+  /usr/bin/python3 /opt/plane-agentd/cli/agentd --config /opt/plane-agentd/config/agentd.yaml \
+  webhook-serve --bind 10.77.10.249 --port 8790
+Restart=always
+RestartSec=5
+
+[Install]
+WantedBy=multi-user.target

tests/test_skips.py

@@ -0,0 +1,71 @@
+import json
+import unittest
+
+from tests.fakes import Harness
+
+
+class FakeProbe:
+    def __init__(self, counts):
+        self.counts = counts
+
+    def process_count(self, name):
+        return self.counts.get(name, 0)
+
+
+class SkipReasonTest(unittest.TestCase):
+    """Why a Todo item was not dispatched: heartbeat "skipped", logged once per new reason."""
+
+    def setUp(self):
+        self.h = Harness()
+        self.p = self.h.plane
+        self.lines = []
+        self.h.d.log = lambda msg: self.lines.append(msg)
+
+    def tearDown(self):
+        self.h.close()
+
+    def skipped(self):
+        hb = json.loads((self.h.d.store.dir / "heartbeat.json").read_text())
+        return {s["item"]: (s["terminal"], s["reason"]) for s in hb["skipped"]}
+
+    def not_dispatched(self):
+        return [l for l in self.lines if "not dispatched" in l]
+
+    def test_reasons_in_heartbeat_and_logged_once(self):
+        self.p.add_issue(1)                                            # starts on win10
+        self.p.add_issue(2)                                            # win10 full
+        self.p.add_issue(3, assignee="kappiee")                        # no terminal for kappiee
+        self.p.add_issue(4, assignee="oregon", labels=["needs-game"])  # starts on oregon, holds the game group
+        self.p.add_issue(5, assignee="oregon", labels=["needs-game"])  # group held
+        self.h.tick()
+        self.assertEqual(self.skipped(), {
+            "GW-2": ("win10", "concurrency 1/1 (GW-1)"),
+            "GW-3": (None, "no route for assignee kappiee"),
+            "GW-5": ("oregon", "serialize_group game held by GW-4"),
+        })
+        self.assertEqual(len(self.not_dispatched()), 3)
+        self.h.tick()   # same reasons: heartbeat still has them, the log does not repeat them
+        self.assertEqual(len(self.skipped()), 3)
+        self.assertEqual(len(self.not_dispatched()), 3)
+
+    def test_busy_process_and_relay_down(self):
+        self.h.cfg.terminals["win10"]["busy_processes"] = ["codex"]
+        probe = FakeProbe({"codex": 1})
+        self.h.d.transport_factory = lambda term: probe
+        self.p.add_issue(1)
+        self.h.tick()
+        self.assertEqual(self.skipped(), {"GW-1": ("win10", "busy_process codex (1 running, not started by agentd)")})
+        probe.counts = {}
+        self.h.tick()   # leftover codex.exe gone: dispatched, nothing skipped
+        self.assertEqual(self.skipped(), {})
+        self.assertEqual(self.p.state_of("i1"), "In Progress")
+        self.p.add_issue(2, assignee="oregon")
+        self.h.cfg.terminals["oregon"]["busy_processes"] = ["claude"]
+        probe.counts = {"claude": None}   # probe fails: relay down
+        probe.process_count = lambda name: None
+        self.h.tick()
+        self.assertEqual(self.skipped()["GW-2"], ("oregon", "relay down"))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_transports.py

@@ -38,7 +38,7 @@ class LocalSecretsTest(unittest.TestCase):
             data = config_data()
             data["agentd"]["state_dir"] = str(Path(tmp) / "state")
             data["terminals"]["oregon"]["repo"] = tmp
-            data["projects"]["GWENT"]["webhook_secret_key"] = "HOOK_GW"
+            data["webhook"] = {"secret_key": "HOOK_GW", "tick_command": ["true"]}
             cfg = Config(data)
             self.assertEqual(cfg.secret_keys(), {"TOK_OR", "TOK_W10", "HOOK_GW"})
             t = LocalTransport(cfg, "oregon")

tests/test_webhook.py

@@ -0,0 +1,184 @@
+import json
+import threading
+import time
+import unittest
+import urllib.error
+import urllib.request
+from http.server import ThreadingHTTPServer
+
+from agentd.webhook import BUSY, Trigger, make_handler, route, sign, verify
+
+SECRET = "plane_wh_test"
+PROJECTS = {"GWENT": {"id": "proj-gw"}, "PAD": {"id": "proj-pad"}}
+
+
+def payload(project="proj-gw", event="issue", action="updated"):
+    return {"event": event, "action": action, "webhook_id": "w1", "workspace_id": "ws", "workspace_slug": "kappiee",
+            "data": {"id": "i1", "project": project}, "activity": {"field": "state"}}
+
+
+def plane_body(p):
+    return json.dumps(p).encode()   # what Plane signs (json.dumps(payload)) and sends
+
+
+class SignatureTest(unittest.TestCase):
+    def test_ok_bad_missing(self):
+        body = plane_body(payload())
+        self.assertTrue(verify(SECRET, body, sign(SECRET, body)))
+        self.assertTrue(verify(SECRET, body, sign(SECRET, body).upper()))
+        self.assertFalse(verify(SECRET, body, sign("other", body)))
+        self.assertFalse(verify(SECRET, body + b" ", sign(SECRET, body + b"x")))
+        self.assertFalse(verify(SECRET, body, None))
+        self.assertFalse(verify(SECRET, body, ""))
+        self.assertFalse(verify("", body, sign("", body)))       # no secret configured: reject everything
+        self.assertFalse(verify(SECRET, body, "zzé"))         # non-hex garbage
+
+    def test_reencoded_body_matches_plane_serialisation(self):
+        p = payload()
+        compact = json.dumps(p, separators=(",", ":")).encode()
+        self.assertTrue(verify(SECRET, compact, sign(SECRET, plane_body(p))))
+        self.assertFalse(verify(SECRET, b"not json", sign(SECRET, plane_body(p))))
+
+
+class RouteTest(unittest.TestCase):
+    def test_route(self):
+        self.assertEqual(route(payload(), PROJECTS), "GWENT")
+        self.assertEqual(route(payload("proj-pad", "issue_comment", "created"), PROJECTS), "PAD")
+        self.assertIsNone(route(payload("proj-other"), PROJECTS))
+        self.assertIsNone(route({"event": "issue", "action": "deleted", "data": {"id": "i1"}}, PROJECTS))
+        self.assertIsNone(route({"event": "issue"}, PROJECTS))
+        self.assertEqual(route({"data": {"project": {"id": "proj-gw"}}}, PROJECTS), "GWENT")
+
+
+class FakeTrigger:
+    def __init__(self):
+        self.kicks = []
+
+    def kick(self, reason):
+        self.kicks.append(reason)
+
+    def status(self):
+        return {"pending": bool(self.kicks)}
+
+
+class HandlerTest(unittest.TestCase):
+    def setUp(self):
+        self.trigger, self.lines = FakeTrigger(), []
+        self.httpd = ThreadingHTTPServer(("127.0.0.1", 0), make_handler(SECRET, PROJECTS, self.trigger, self.lines.append))
+        threading.Thread(target=self.httpd.serve_forever, daemon=True).start()
+        self.url = "http://127.0.0.1:%d" % self.httpd.server_address[1]
+
+    def tearDown(self):
+        self.httpd.shutdown()
+        self.httpd.server_close()
+
+    def post(self, body, sig, path="/plane"):
+        headers = {"Content-Type": "application/json", "X-Plane-Delivery": "d-1", "X-Plane-Event": "issue"}
+        if sig is not None:
+            headers["X-Plane-Signature"] = sig
+        req = urllib.request.Request(self.url + path, data=body, headers=headers, method="POST")
+        try:
+            with urllib.request.urlopen(req, timeout=5) as r:
+                return r.status
+        except urllib.error.HTTPError as e:
+            return e.code
+
+    def test_signed_tenant_delivery_triggers(self):
+        body = plane_body(payload())
+        self.assertEqual(self.post(body, sign(SECRET, body)), 200)
+        self.assertEqual(self.trigger.kicks, ["d-1"])
+        self.assertIn("project=GWENT", self.lines[-1])
+        self.assertIn("decision=tick", self.lines[-1])
+
+    def test_bad_or_missing_signature_is_401_and_triggers_nothing(self):
+        body = plane_body(payload())
+        self.assertEqual(self.post(body, sign("wrong", body)), 401)
+        self.assertEqual(self.post(body, None), 401)
+        self.assertEqual(self.trigger.kicks, [])
+        self.assertIn("401 missing", self.lines[-1])
+
+    def test_other_project_is_202(self):
+        body = plane_body(payload("proj-other"))
+        self.assertEqual(self.post(body, sign(SECRET, body)), 202)
+        self.assertEqual(self.trigger.kicks, [])
+
+    def test_wrong_path_and_healthz(self):
+        body = plane_body(payload())
+        self.assertEqual(self.post(body, sign(SECRET, body), "/other"), 404)
+        with urllib.request.urlopen(self.url + "/healthz", timeout=5) as r:
+            self.assertEqual((r.status, json.loads(r.read())["ok"]), (200, True))
+
+
+class FakeRunner:
+    def __init__(self, codes=(), hold=0.0):
+        self.codes, self.hold, self.calls = list(codes), hold, 0
+        self.started = threading.Event()
+
+    def __call__(self):
+        self.calls += 1
+        self.started.set()
+        time.sleep(self.hold)
+        return self.codes.pop(0) if self.codes else 0
+
+
+class TriggerTest(unittest.TestCase):
+    def make(self, runner, **kw):
+        kw = dict(dict(debounce=0.2, retry=0.1, retry_window=1.0), **kw)
+        self.lines = []
+        return Trigger(runner, log=self.lines.append, **kw)
+
+    def test_burst_coalesces_into_one_tick(self):
+        run = FakeRunner()
+        t = self.make(run)
+        for i in range(5):
+            t.kick("d%d" % i)
+        self.assertTrue(t.step(wait=2))
+        self.assertEqual(run.calls, 1)
+        self.assertIn("5 deliveries", self.lines[0])
+        self.assertFalse(t.step(wait=0.4))   # nothing left
+        self.assertEqual(run.calls, 1)
+
+    def test_debounce_waits_before_ticking(self):
+        run = FakeRunner()
+        t = self.make(run, debounce=0.3)
+        t0 = time.monotonic()
+        t.kick("d1")
+        t.step(wait=2)
+        self.assertGreaterEqual(time.monotonic() - t0, 0.29)
+
+    def test_delivery_during_a_tick_causes_one_follow_up(self):
+        run = FakeRunner(hold=0.3)
+        t = self.make(run, debounce=0.05).start()
+        t.kick("d1")
+        self.assertTrue(run.started.wait(2))
+        t.kick("d2")      # arrives while the first tick runs
+        t.kick("d3")
+        time.sleep(1.0)
+        self.assertEqual(run.calls, 2)
+        self.assertEqual(t.status()["pending"], False)
+
+    def test_busy_lock_is_retried_until_the_tick_runs(self):
+        run = FakeRunner(codes=[BUSY, BUSY, 0])
+        t = self.make(run, debounce=0.0)
+        t.kick("d1")
+        for _ in range(3):
+            self.assertTrue(t.step(wait=2))
+        self.assertEqual(run.calls, 3)
+        self.assertEqual(sum("retry in" in l for l in self.lines), 2)
+        self.assertIn("1 deliveries: d1", self.lines[-2])   # the delivery is carried into the retry
+        self.assertIsNone(t.status()["retrying_since"])
+        self.assertFalse(t.step(wait=0.3))
+
+    def test_busy_gives_up_after_the_window(self):
+        run = FakeRunner(codes=[BUSY] * 10)
+        t = self.make(run, debounce=0.0, retry=0.1, retry_window=0.35)
+        t.kick("d1")
+        while t.step(wait=1):
+            pass
+        self.assertTrue(any("giving up" in l for l in self.lines))
+        self.assertLessEqual(run.calls, 5)
+        self.assertFalse(t.status()["pending"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tools/plane-webhook.py

@@ -0,0 +1,121 @@
+#!/usr/bin/env python3
+"""Plane workspace webhooks, as the workspace admin kappiee (session login; the internal /api/workspaces/<slug>/webhooks/
+endpoint needs admin). Never prints secrets: a new or regenerated secret goes straight into the Vault as
+PLANE_WEBHOOK_SECRET (then restart plane-agentd-webhook). Run under
+  agent-access.py --config <identity> exec --secret-env PLANE_KAPPIEE_PASSWORD=PLANE_KAPPIEE_PASSWORD -- python3 -I tools/plane-webhook.py ...
+usage: list | create URL (inactive; work item + comment events) | patch ID key=value... (e.g. is_active=true)
+       | regenerate ID | delete ID | logs ID [n]"""
+import http.cookiejar, json, os, ssl, sys, urllib.parse, urllib.request
+
+BASE = "https://plane.kappiee.com"
+WS = "kappiee"
+jar = http.cookiejar.CookieJar()
+op = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar))
+HDR = {"Referer": BASE + "/", "Origin": BASE, "User-Agent": "plane-agentd-admin"}
+
+
+def csrf():
+    return next((c.value for c in jar if c.name == "csrftoken"), "")
+
+
+def call(method, path, body=None, form=None):
+    h = dict(HDR)
+    data = None
+    if form is not None:
+        data = urllib.parse.urlencode(form).encode()
+        h["Content-Type"] = "application/x-www-form-urlencoded"
+    elif body is not None:
+        data = json.dumps(body).encode()
+        h["Content-Type"] = "application/json"
+    if method != "GET":
+        h["X-CSRFToken"] = csrf()
+    req = urllib.request.Request(BASE + path, data=data, headers=h, method=method)
+    try:
+        with op.open(req, timeout=30) as r:
+            raw = r.read()
+            return r.status, raw, r.geturl()
+    except urllib.error.HTTPError as e:
+        return e.code, e.read(), path
+
+
+def login():
+    st, raw, _ = call("GET", "/auth/get-csrf-token/")
+    tok = json.loads(raw).get("csrf_token") or csrf()
+    st, raw, url = call("POST", "/auth/sign-in/", form={"csrfmiddlewaretoken": tok, "email": "kappiee@kappiee.com",
+                                                        "password": os.environ["PLANE_KAPPIEE_PASSWORD"]})
+    ok = any(c.name.startswith("session") for c in jar)
+    print("login", st, "session cookie" if ok else "NO SESSION", urllib.parse.urlparse(url).path)
+    if not ok:
+        sys.exit(1)
+
+
+def redact(d):
+    return {k: ("<%d chars>" % len(v) if k == "secret_key" and v else v) for k, v in d.items()}
+
+
+def vault_store(name, value):
+    ident = json.load(urllib.request.urlopen("https://nas.tail9531b.ts.net:8443/identity/agent.json", timeout=30))
+    v = ident["vault"]
+    ctx = ssl.create_default_context()
+    ctx.check_hostname = False
+    ctx.verify_mode = ssl.CERT_NONE
+    req = urllib.request.Request(v["url"].rstrip("/") + "/v1/credentials", method="POST",
+                                 data=json.dumps({"vault": "default", "credentials": {name: value}}).encode(),
+                                 headers={"Authorization": "Bearer " + v["token"], "X-Vault": "default",
+                                          "Content-Type": "application/json"})
+    with urllib.request.urlopen(req, context=ctx, timeout=30) as r:
+        print("vault store", name, r.status)
+
+
+def store_secret(w):
+    if w.get("secret_key"):
+        vault_store("PLANE_WEBHOOK_SECRET", w["secret_key"])
+
+
+login()
+P = "/api/workspaces/%s/webhooks/" % WS
+cmd = sys.argv[1]
+if cmd == "list":
+    st, raw, _ = call("GET", P)
+    print("list", st)
+    if st == 200:
+        for w in json.loads(raw):
+            print(json.dumps(redact(w)))
+    else:
+        print(raw[:300])
+elif cmd == "create":
+    body = {"url": sys.argv[2], "is_active": False, "project": False, "issue": True, "module": False, "cycle": False,
+            "issue_comment": True}
+    st, raw, _ = call("POST", P, body)
+    print("create", st)
+    if st in (200, 201):
+        w = json.loads(raw)
+        store_secret(w)
+        print(json.dumps(redact(w)))
+    else:
+        print(raw[:500])
+elif cmd == "patch":
+    body = {}
+    for kv in sys.argv[3:]:
+        k, v = kv.split("=", 1)
+        body[k] = {"true": True, "false": False}.get(v, v)
+    st, raw, _ = call("PATCH", P + sys.argv[2] + "/", body)
+    print("patch", st, json.dumps(redact(json.loads(raw))) if st == 200 else raw[:500])
+elif cmd == "regenerate":
+    st, raw, _ = call("POST", P + sys.argv[2] + "/regenerate/", {})
+    print("regenerate", st)
+    if st in (200, 201):
+        w = json.loads(raw)
+        store_secret(w)
+        print(json.dumps(redact(w)))
+    else:
+        print(raw[:500])
+elif cmd == "delete":
+    st, raw, _ = call("DELETE", P + sys.argv[2] + "/")
+    print("delete", st, raw[:200])
+elif cmd == "logs":
+    st, raw, _ = call("GET", "/api/workspaces/%s/webhook-logs/%s/" % (WS, sys.argv[2]))
+    print("logs", st)
+    if st == 200:
+        for l in json.loads(raw)[: int(sys.argv[3]) if len(sys.argv) > 3 else 10]:
+            print(l.get("created_at"), l.get("event_type"), l.get("request_method"), l.get("response_status"), (l.get("response_body") or "")[:120], l.get("retry_count"))