Add a groups JSON API and group-run cancellation

The group pages were browser-only, so external schedulers had no way to
trigger a group or read its outcome. Adds /api/groups, /api/group-runs and
their run/list endpoints under Basic auth, following the same async contract
as POST /modules/{id}/run.

Cancellation needed more than an endpoint. run_group deliberately continues
past module failures, so killing the in-flight jrunner alone would just let
the loop march on to the next member. A group-level flag in engine/cancel.py
is now checked before each member, and engine.request_group_cancel sets that
flag *and* signals the running module -- it lives in runner.py rather than
cancel.py because finding the live run needs the DB, and cancel.py is
deliberately DB-free.

Group final status gains 'cancelled', ranked above 'error' since the error is
usually the terminated jrunner process. group_run.status already permitted the
value; no migration needed.

Also adds a "cancel group" button to the group run page (web) and .pill
styling for cancelled/dry_run, neither of which had a color rule.

Note: /runs/{id}/cancel keeps its existing meaning -- cancel one member and
let the group continue -- which is now a deliberate contrast to cancel group.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

CLAUDE.md

@@ -80,7 +80,9 @@ run_group(group_id)
   → create group_run row (status=running)
   → for each enabled group_member in run_order: call run_module(group_run_id=...)
   → continues past individual module failures (all members run)
-  → final status: dry_run if all dry_run, error if any errored, success otherwise
+  → but a group cancel breaks the loop: remaining members never start
+  → final status: cancelled if cancelled, else dry_run if all dry_run,
+                  error if any errored, success otherwise
   → finish_group_run(status)
 ```
 

pipekit/api/app.py

@@ -19,7 +19,7 @@ import os
 
 from .. import __version__, db, jrunner, repo
 from ..web import mount_web
-from .routes import connections, introspect, modules, runs, system
+from .routes import connections, groups, introspect, modules, runs, system
 
 _log = logging.getLogger("pipekit.web")
 
@@ -167,6 +167,7 @@ def create_app() -> FastAPI:
     app.include_router(system.router)
     app.include_router(connections.router, prefix="/api")
     app.include_router(introspect.router, prefix="/api")
+    app.include_router(groups.router, prefix="/api")
     app.include_router(modules.router, prefix="/api")
     app.include_router(runs.router, prefix="/api")
     mount_web(app)

pipekit/api/routes/groups.py

@@ -0,0 +1,84 @@
+"""Groups + group runs. The JSON counterpart to the web UI's group pages.
+
+The web routes (web/app.py) already expose group runs, but behind session-cookie auth and
+returning HTML/redirects, so they are only usable from a browser. External schedulers need
+Basic auth and JSON -- hence these. Same async contract as POST /modules/{id}/run: the
+group_run_id comes back immediately and the work continues in the background, so callers poll
+GET /group-runs/{id} for the terminal status.
+"""
+
+from __future__ import annotations
+
+from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException
+
+from ... import engine, repo
+from ..auth import require_auth
+
+router = APIRouter(tags=["groups"], dependencies=[Depends(require_auth)])
+
+
+@router.get("/groups")
+def list_groups() -> list[dict]:
+    return repo.list_groups()
+
+
+@router.get("/groups/{group_id}")
+def get_group(group_id: int) -> dict:
+    g = repo.get_group(group_id)
+    if g is None:
+        raise HTTPException(404, f"group id={group_id} not found")
+    return {**g, "members": repo.list_group_members(group_id)}
+
+
+@router.post("/groups/{group_id}/run")
+def run_group(group_id: int, background: BackgroundTasks,
+              dry_run: bool = False) -> dict:
+    """Kick off a group run. Returns group_run_id immediately."""
+    g = repo.get_group(group_id)
+    if g is None:
+        raise HTTPException(404, f"group id={group_id} not found")
+    group_run_id = repo.create_group_run(group_id, triggered_by="api")
+    background.add_task(_run_in_background, group_id, group_run_id, dry_run)
+    return {"group_run_id": group_run_id}
+
+
+def _run_in_background(group_id: int, group_run_id: int, dry_run: bool) -> None:
+    try:
+        engine.run_group(group_id, group_run_id=group_run_id, dry_run=dry_run)
+    except Exception as e:                                           # noqa: BLE001
+        # run_group already records per-module failures; this is the group row itself
+        # failing (or a LockBusy escaping), which must not leave it stuck at 'running'.
+        repo.finish_group_run(group_run_id, status="error")
+        raise e from None
+
+
+@router.get("/group-runs/{group_run_id}")
+def get_group_run(group_run_id: int) -> dict:
+    gr = repo.get_group_run(group_run_id)
+    if gr is None:
+        raise HTTPException(404, f"group_run id={group_run_id} not found")
+    return {**gr, "runs": repo.list_runs_for_group_run(group_run_id)}
+
+
+@router.post("/group-runs/{group_run_id}/cancel")
+def cancel_group_run(group_run_id: int) -> dict:
+    """Stop an in-flight group run.
+
+    Kills the module currently running and prevents the remaining members from
+    starting. The group run lands with status ``cancelled``; the engine's own
+    finally-blocks release the module lock and write the run_log rows, so this
+    handler deliberately does not touch either.
+    """
+    gr = repo.get_group_run(group_run_id)
+    if gr is None:
+        raise HTTPException(404, f"group_run id={group_run_id} not found")
+    if gr["status"] != "running":
+        raise HTTPException(409, f"group run {group_run_id} is {gr['status']}, not running")
+    return engine.request_group_cancel(group_run_id)
+
+
+@router.get("/groups/{group_id}/group-runs")
+def list_group_runs(group_id: int, limit: int = 20) -> list[dict]:
+    if repo.get_group(group_id) is None:
+        raise HTTPException(404, f"group id={group_id} not found")
+    return repo.list_group_runs(group_id, limit=limit)

pipekit/cli.py

@@ -139,6 +139,26 @@ def cmd_run(args) -> int:
     return 0 if outcome.status == "success" else 1
 
 
+def cmd_run_group(args) -> int:
+    group = repo.get_group_by_name(args.group)
+    if group is None:
+        print(f"error: group {args.group!r} not found")
+        return 1
+    try:
+        outcome = engine.run_group(group["id"], dry_run=args.dry_run)
+    except engine.LockBusy as e:
+        print(f"busy: {e}")
+        return 75          # EX_TEMPFAIL — retryable, distinct from a real failure
+
+    tag = "DRY RUN — no jrunner calls made" if args.dry_run else ""
+    print(f"group_run_id={outcome.group_run_id} status={outcome.status} "
+          f"modules={len(outcome.module_outcomes)} {tag}".rstrip())
+    for o in outcome.module_outcomes:
+        line = f"  run_id={o.run_id} status={o.status} rows={o.row_count}"
+        print(f"{line} error={o.error}" if o.error else line)
+    return 0 if outcome.status in ("success", "dry_run") else 1
+
+
 def cmd_serve(args) -> int:
     import uvicorn
     from .api import create_app
@@ -351,6 +371,13 @@ def main(argv: list[str] | None = None) -> int:
                        help="build SQL but do not invoke jrunner")
     p_run.set_defaults(func=cmd_run)
 
+    p_rg = sub.add_parser("run-group",
+                          help="run all enabled modules in a group (synchronous)")
+    p_rg.add_argument("group", help="group name")
+    p_rg.add_argument("--dry-run", action="store_true",
+                      help="build SQL but do not invoke jrunner")
+    p_rg.set_defaults(func=cmd_run_group)
+
     p_exp = sub.add_parser(
         "export", help="dump config (drivers/connections/modules/groups) to text files")
     p_exp.add_argument("--dir", help="target dir (default <repo>/config)")

pipekit/db.py

@@ -48,6 +48,31 @@ def _apply_migrations(conn: sqlite3.Connection) -> None:
     if "last_fired_at" not in sc_cols:
         conn.execute("ALTER TABLE schedule ADD COLUMN last_fired_at TEXT")
 
+    # group_run.status predates 'dry_run'. schema.sql carries the correct CHECK, but
+    # CREATE TABLE IF NOT EXISTS never re-applies it to an existing DB, so a group
+    # dry run raised IntegrityError in finish_group_run. CHECK constraints need a
+    # table rebuild (SQLite has no ALTER for them); run_log got the same treatment
+    # earlier. Idempotent: keyed on the constraint text itself.
+    gr_ddl = conn.execute(
+        "SELECT sql FROM sqlite_master WHERE type='table' AND name='group_run'"
+    ).fetchone()
+    if gr_ddl and "dry_run" not in gr_ddl[0]:
+        conn.executescript("""
+            CREATE TABLE group_run_new (
+                id           INTEGER PRIMARY KEY AUTOINCREMENT,
+                group_id     INTEGER NOT NULL REFERENCES grp(id),
+                started_at   TEXT DEFAULT (datetime('now')),
+                finished_at  TEXT,
+                status       TEXT NOT NULL DEFAULT 'running'
+                             CHECK (status IN ('running','success','error','cancelled','dry_run')),
+                triggered_by TEXT
+            );
+            INSERT INTO group_run_new (id, group_id, started_at, finished_at, status, triggered_by)
+                SELECT id, group_id, started_at, finished_at, status, triggered_by FROM group_run;
+            DROP TABLE group_run;
+            ALTER TABLE group_run_new RENAME TO group_run;
+        """)
+
 
 @contextmanager
 def connect(db_path: Path | None = None):

pipekit/engine/__init__.py

@@ -1,5 +1,6 @@
 from .cancel import request_cancel
-from .runner import GroupRunOutcome, LockBusy, RunOutcome, run_group, run_module
+from .runner import (GroupRunOutcome, LockBusy, RunOutcome, request_group_cancel,
+                     run_group, run_module)
 
 __all__ = ["GroupRunOutcome", "LockBusy", "RunOutcome", "request_cancel",
-           "run_group", "run_module"]
+           "request_group_cancel", "run_group", "run_module"]

pipekit/engine/cancel.py

@@ -17,6 +17,7 @@ import threading
 _lock = threading.Lock()
 _active: dict[int, object] = {}   # run_id -> subprocess.Popen (live jrunner proc)
 _cancelled: set[int] = set()      # run_ids asked to stop
+_cancelled_groups: set[int] = set()   # group_run_ids asked to stop
 
 
 class RunCancelled(Exception):
@@ -29,6 +30,27 @@ def arm(run_id: int) -> None:
         _cancelled.discard(run_id)
 
 
+def arm_group(group_run_id: int) -> None:
+    """Clear any stale group cancel flag before a group run starts."""
+    with _lock:
+        _cancelled_groups.discard(group_run_id)
+
+
+def flag_group(group_run_id: int) -> None:
+    """Mark a group run as cancelled so run_group stops launching members.
+
+    Only sets the flag; terminating the module that is already in flight is the
+    caller's job (it needs the DB to find the live run_id).
+    """
+    with _lock:
+        _cancelled_groups.add(group_run_id)
+
+
+def is_group_cancelled(group_run_id: int) -> bool:
+    with _lock:
+        return group_run_id in _cancelled_groups
+
+
 def register(run_id: int, proc: object) -> None:
     """Record the live subprocess for a run so cancel can reach it."""
     with _lock:

pipekit/engine/runner.py

@@ -38,7 +38,7 @@ class RunOutcome:
 @dataclass
 class GroupRunOutcome:
     group_run_id: int
-    status: str                   # success | error | dry_run
+    status: str                   # success | error | dry_run | cancelled
     module_outcomes: list
 
 
@@ -208,10 +208,17 @@ def run_group(group_id: int, *, dry_run: bool = False,
     if group_run_id is None:
         group_run_id = repo.create_group_run(group_id, triggered_by="manual")
 
+    cancel.arm_group(group_run_id)
     members = [m for m in repo.list_group_members(group_id) if m["module_enabled"]]
     outcomes: list[RunOutcome] = []
+    stopped = False
 
     for member in members:
+        # Unlike a module failure (which the group runs past), a cancel means stop:
+        # don't start any remaining member.
+        if cancel.is_group_cancelled(group_run_id):
+            stopped = True
+            break
         try:
             outcome = run_module(
                 member["module_id"],
@@ -224,7 +231,10 @@ def run_group(group_id: int, *, dry_run: bool = False,
             outcome = RunOutcome(run_id, "error", None, str(e), None, None)
         outcomes.append(outcome)
 
-    if not outcomes:
+    # 'cancelled' outranks 'error': the error is usually the killed jrunner proc.
+    if stopped or any(o.status == "cancelled" for o in outcomes):
+        final = "cancelled"
+    elif not outcomes:
         final = "dry_run" if dry_run else "success"
     elif all(o.status == "dry_run" for o in outcomes):
         final = "dry_run"
@@ -237,6 +247,23 @@ def run_group(group_id: int, *, dry_run: bool = False,
     return GroupRunOutcome(group_run_id, final, outcomes)
 
 
+def request_group_cancel(group_run_id: int) -> dict:
+    """Stop a group run: flag it, and terminate whichever module is in flight.
+
+    Lives here rather than in cancel.py because finding the live run needs the DB.
+    The flag is what stops *remaining* members; the per-run cancel is what stops
+    the current one. Returns the runs that were signalled (may be empty if the
+    group is between members this instant -- the flag still lands).
+    """
+    cancel.flag_group(group_run_id)
+    signalled = [
+        {"run_id": r["id"], "result": cancel.request_cancel(r["id"])}
+        for r in repo.list_runs_for_group_run(group_run_id)
+        if r["status"] == "running"
+    ]
+    return {"group_run_id": group_run_id, "signalled": signalled}
+
+
 def _run_hooks(module_id: int, *, fail_fast: bool, run_on_set: set[str]) -> str:
     """Run hooks whose ``run_on`` is in run_on_set. Returns a text log."""
     hooks = [h for h in repo.list_hooks(module_id) if h["run_on"] in run_on_set]

pipekit/web/app.py

@@ -866,13 +866,31 @@ def _schedules_with_next(schedules: list[dict]) -> list[dict]:
         s = dict(s)
         try:
             cron = croniter(s["cron_expr"], now)
-            s["next_fire_at"] = cron.get_next(datetime).strftime("%Y-%m-%d %H:%M %Z")
+            nxt = cron.get_next(datetime)
+            s["next_fire_at"] = nxt.strftime("%Y-%m-%d %H:%M")
+            s["next_fire_in"] = _humanize_delta((nxt - now).total_seconds())
         except CroniterBadCronError:
             s["next_fire_at"] = "invalid expression"
+            s["next_fire_in"] = None
         result.append(s)
     return result
 
 
+def _humanize_delta(seconds: float) -> str:
+    """'in 3h 12m' style relative label for a positive second count."""
+    secs = int(max(seconds, 0))
+    days, rem = divmod(secs, 86400)
+    hours, rem = divmod(rem, 3600)
+    mins = rem // 60
+    if days:
+        return f"in {days}d {hours}h"
+    if hours:
+        return f"in {hours}h {mins}m"
+    if mins:
+        return f"in {mins}m"
+    return "in <1m"
+
+
 def _sanitize_identifier(name: str) -> str:
     """Lower-case a source column name and replace characters that aren't
     valid in an unquoted identifier with underscores."""
@@ -1338,6 +1356,7 @@ def groups_index(request: Request):
             g["last_run_at"] = None
             g["last_status"] = None
             g["last_duration_s"] = None
+        g["schedules"] = _schedules_with_next(repo.list_schedules_for_group(g["id"]))
     return _templates.TemplateResponse(
         request, "groups.html",
         _ctx(groups=groups),
@@ -1489,6 +1508,28 @@ def group_run_live_fragment(request: Request, group_run_id: int):
     )
 
 
+@_router.post("/group-runs/{group_run_id}/cancel", response_class=HTMLResponse)
+def group_run_cancel(request: Request, group_run_id: int):
+    """Stop the whole group: kill the module in flight, skip the rest.
+
+    Distinct from /runs/{id}/cancel, which cancels one member and lets the group
+    carry on to the next.
+    """
+    group_run = repo.get_group_run(group_run_id)
+    if group_run is None:
+        raise HTTPException(404, f"group run id={group_run_id} not found")
+    if group_run["status"] == "running":
+        engine.request_group_cancel(group_run_id)
+    # Re-read: the engine writes the terminal status from its own thread, so the
+    # fragment may still show 'running' for a tick until the poll catches up.
+    group_run = repo.get_group_run(group_run_id)
+    module_runs = repo.list_runs_for_group_run(group_run_id)
+    return _templates.TemplateResponse(
+        request, "_group_run_live.html",
+        _ctx(group_run=group_run, module_runs=module_runs),
+    )
+
+
 def _save_group_schedules(form, group_id: int) -> None:
     """Process sched_* parallel arrays from a group form POST."""
     from croniter import croniter as _croniter

pipekit/web/static/style.css

@@ -147,6 +147,8 @@ table.grid tr:hover td { background: #1c2128; }
 .pill.running                { color: var(--accent);  }
 .pill.disabled               { color: var(--text-muted); }
 .pill.warning                { color: var(--warning); }
+.pill.cancelled              { color: var(--warning); }
+.pill.dry_run                { color: var(--text-muted); }
 
 /* Group membership tags */
 .tag {
@@ -162,6 +164,8 @@ table.grid tr:hover td { background: #1c2128; }
 }
 .tag:hover { color: var(--accent); }
 
+.muted { color: var(--text-muted); }
+
 /* Labeled key-value rows (used in detail views) */
 dl.keyval {
     display: grid;

pipekit/web/templates/_group_run_live.html

@@ -9,6 +9,12 @@
         <header>
             Module runs
             <span style="margin-left:auto">
+                {% if group_run.status == 'running' %}
+                <button class="btn ghost" style="margin-right:0.5rem"
+                        hx-post="/group-runs/{{ group_run.id }}/cancel"
+                        hx-target="#group-run-live" hx-swap="outerHTML"
+                        hx-confirm="Cancel this group run? The module in flight is stopped and the remaining modules are skipped.">cancel group</button>
+                {% endif %}
                 <span class="pill {{ group_run.status }}">{{ group_run.status }}</span>
                 {% if group_run.duration_s is not none %}&nbsp;· {{ group_run.duration_s | duration }}{% endif %}
             </span>

pipekit/web/templates/group_detail.html

@@ -76,7 +76,7 @@
             {% for s in schedules %}
                 <tr>
                     <td class="mono">{{ s.cron_expr }}</td>
-                    <td class="mono">{% if s.enabled %}{{ s.next_fire_at }}{% else %}—{% endif %}</td>
+                    <td class="mono">{% if s.enabled %}{{ s.next_fire_at }}{% if s.next_fire_in %} <span class="muted">({{ s.next_fire_in }})</span>{% endif %}{% else %}—{% endif %}</td>
                     <td class="mono">{{ s.last_fired_at | localtime }}</td>
                     <td>
                         {% if s.enabled %}

pipekit/web/templates/group_run_detail.html

@@ -31,7 +31,17 @@
      hx-swap="outerHTML"
      {% endif %}>
     <div class="panel">
-        <header>Module runs</header>
+        <header>
+            Module runs
+            {% if group_run.status == 'running' %}
+            <span style="margin-left:auto">
+                <button class="btn ghost"
+                        hx-post="/group-runs/{{ group_run.id }}/cancel"
+                        hx-target="#group-run-live" hx-swap="outerHTML"
+                        hx-confirm="Cancel this group run? The module in flight is stopped and the remaining modules are skipped.">cancel group</button>
+            </span>
+            {% endif %}
+        </header>
         <div class="body tight">
             {% if module_runs %}
             <table class="grid">

pipekit/web/templates/groups.html

@@ -18,6 +18,8 @@
                 <tr>
                     <th>name</th>
                     <th>members</th>
+                    <th>schedule</th>
+                    <th>next run</th>
                     <th>last run</th>
                     <th style="width:7em;text-align:right">duration</th>
                     <th style="width:9em">status</th>
@@ -29,6 +31,21 @@
                 <tr>
                     <td><a href="/groups/{{ g.id }}"><strong>{{ g.name }}</strong></a></td>
                     <td class="mono">{{ g.member_count }}</td>
+                    {% set scheds = g.schedules if g.schedules is iterable and g.schedules is not string else [] %}
+                    <td class="mono">
+                        {% for s in scheds %}
+                        <div{% if not s.enabled %} class="muted"{% endif %}>{{ s.cron_expr }}{% if not s.enabled %} (disabled){% endif %}</div>
+                        {% else %}
+                        <span class="muted">—</span>
+                        {% endfor %}
+                    </td>
+                    <td class="mono">
+                        {% for s in scheds %}
+                        <div>{% if s.enabled %}{{ s.next_fire_at }}{% if s.next_fire_in %} <span class="muted">({{ s.next_fire_in }})</span>{% endif %}{% else %}<span class="muted">—</span>{% endif %}</div>
+                        {% else %}
+                        <span class="muted">—</span>
+                        {% endfor %}
+                    </td>
                     <td class="mono">{{ g.last_run_at | localtime }}</td>
                     <td class="mono" style="text-align:right">{{ g.last_duration_s | duration }}</td>
                     <td>