From e4c17bfc9843a5df8c0628ba5cf3075a9adb796e Mon Sep 17 00:00:00 2001
From: Paul Trowbridge <paul@hptrow.me>
Date: Wed, 19 Aug 2026 17:32:24 -0400
Subject: [PATCH] Add `pipekit run-group` for triggering a group from outside

engine.run_group already sequences a group's enabled members by run_order,
continues past individual failures, and aggregates status -- but nothing
outside the process could reach it. The scheduler called it on cron, and
the web UI exposed it as a form POST that returns 303 + a background task.
Neither is usable from an external orchestrator: driving the web route
means scraping HTML to poll for completion.

This wraps it as a synchronous CLI command, mirroring `pipekit run` for
modules, so callers get an exit code and a summary on stdout.

LockBusy maps to exit 75 (EX_TEMPFAIL) rather than 1, so a caller can
distinguish "another run holds the lock, retry later" from a genuine
failure. This intentionally diverges from `pipekit run`, which returns 1
for both.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 pipekit/cli.py | 27 +++++++++++++++++++++++++++
 1 file changed, 27 insertions(+)

diff --git a/pipekit/cli.py b/pipekit/cli.py
index b94f2ce..6d7a9e1 100644
--- a/pipekit/cli.py
+++ b/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)")