diff --git a/CLAUDE.md b/CLAUDE.md index 19a3a57..a8b6f62 100644 --- a/CLAUDE.md +++ b/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) ``` diff --git a/pipekit/api/app.py b/pipekit/api/app.py index 6bfc018..59c97b1 100644 --- a/pipekit/api/app.py +++ b/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) diff --git a/pipekit/api/routes/groups.py b/pipekit/api/routes/groups.py new file mode 100644 index 0000000..656b6c2 --- /dev/null +++ b/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) diff --git a/pipekit/engine/__init__.py b/pipekit/engine/__init__.py index c63cd53..57a3783 100644 --- a/pipekit/engine/__init__.py +++ b/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"] diff --git a/pipekit/engine/cancel.py b/pipekit/engine/cancel.py index b6f0c21..bf5d31e 100644 --- a/pipekit/engine/cancel.py +++ b/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: diff --git a/pipekit/engine/runner.py b/pipekit/engine/runner.py index cea73ad..25a2611 100644 --- a/pipekit/engine/runner.py +++ b/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] diff --git a/pipekit/web/app.py b/pipekit/web/app.py index e93e90d..6549a68 100644 --- a/pipekit/web/app.py +++ b/pipekit/web/app.py @@ -1508,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 diff --git a/pipekit/web/static/style.css b/pipekit/web/static/style.css index 5428053..b10831e 100644 --- a/pipekit/web/static/style.css +++ b/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 { diff --git a/pipekit/web/templates/_group_run_live.html b/pipekit/web/templates/_group_run_live.html index 3eb1105..1047706 100644 --- a/pipekit/web/templates/_group_run_live.html +++ b/pipekit/web/templates/_group_run_live.html @@ -9,6 +9,12 @@
Module runs + {% if group_run.status == 'running' %} + + {% endif %} {{ group_run.status }} {% if group_run.duration_s is not none %} · {{ group_run.duration_s | duration }}{% endif %} diff --git a/pipekit/web/templates/group_run_detail.html b/pipekit/web/templates/group_run_detail.html index a758abd..0816f5a 100644 --- a/pipekit/web/templates/group_run_detail.html +++ b/pipekit/web/templates/group_run_detail.html @@ -31,7 +31,17 @@ hx-swap="outerHTML" {% endif %}>
-
Module runs
+
+ Module runs + {% if group_run.status == 'running' %} + + + + {% endif %} +
{% if module_runs %}