tegwick
6d70f4d97d
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: sonnet Assistant-Process: 295952@bnt-lap001 Assistant-Session: e93f64ad-516c-46eb-9666-aad8d300c477
331 lines
15 KiB
Python
331 lines
15 KiB
Python
"""A project run: one run of every member runbook of a project, in `after` order, in one mode.
|
|
|
|
The record lives next to the project (`.local/project-runs/`); its state is derived from the member runs, which carry it in
|
|
their context. It snapshots the project's resolved attributes when it starts, so every member run uses the same values even if
|
|
the project is edited meanwhile. Manual: the operator starts each member, one at a time. Auto-confirm: the engine starts the next
|
|
ready member, ticks its run, closes it when it finishes cleanly and moves on, stopping wherever a person is needed.
|
|
Closing locks and seals the record like a run.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import time
|
|
import uuid
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from . import project as J
|
|
from . import run as R
|
|
from . import run2
|
|
from .attrs import Doc, Registry
|
|
from .run import RunError, _clean, _now, _read_json, _seal_of
|
|
|
|
CONTINUE = ("member-started", "member-finished", "started", "completed", "push-through")
|
|
OUTCOMES = ("succeeded", "failed", "abandoned")
|
|
|
|
|
|
def key_of(member: dict) -> str:
|
|
return member["pack"] + (f"@{member['target']}" if member.get("target") else "")
|
|
|
|
|
|
@dataclass
|
|
class ProjectRun:
|
|
path: Path
|
|
data: dict
|
|
|
|
@property
|
|
def id(self) -> str:
|
|
return self.data["id"]
|
|
|
|
@property
|
|
def closed(self) -> bool:
|
|
return bool(self.data.get("outcome"))
|
|
|
|
def save(self) -> None:
|
|
disk = _read_json(self.path)
|
|
if disk and disk.get("outcome"):
|
|
changed = {k for k in set(disk) | set(self.data) if disk.get(k) != self.data.get(k)} - {"updated"}
|
|
if changed:
|
|
raise RunError(f"this project run is closed and locked ({disk['outcome']}); it cannot be changed")
|
|
self.data["updated"] = _now()
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
if self.path.exists():
|
|
self.path.chmod(0o644)
|
|
self.path.write_text(json.dumps(self.data, indent=2, sort_keys=True) + "\n")
|
|
if self.data.get("outcome"):
|
|
self.path.chmod(0o444)
|
|
|
|
def event(self, text: str, who: str | None = None) -> None:
|
|
self.data["events"].append({"at": _now(), "who": who or self.data["operator"], "text": text})
|
|
|
|
|
|
def runs_dir(doc: Doc) -> Path:
|
|
return Path(doc.path).resolve().parents[2] / ".local" / "project-runs"
|
|
|
|
|
|
def load(reg: Registry, run_id: str) -> ProjectRun:
|
|
for doc in reg.projects.values():
|
|
f = runs_dir(doc) / f"{run_id}.json"
|
|
if f.exists():
|
|
return ProjectRun(f, json.loads(f.read_text()))
|
|
raise RunError(f"no such project run: {run_id}")
|
|
|
|
|
|
def list_runs(reg: Registry, project_id: str | None = None) -> list[ProjectRun]:
|
|
out, seen = [], set()
|
|
for doc in reg.projects.values():
|
|
d = runs_dir(doc)
|
|
if d in seen or not d.exists():
|
|
continue
|
|
seen.add(d)
|
|
for f in sorted(d.glob("*.json")):
|
|
data = _read_json(f)
|
|
if data and (project_id is None or data.get("project") == project_id):
|
|
out.append(ProjectRun(f, data))
|
|
return sorted(out, key=lambda p: p.data["created"], reverse=True)
|
|
|
|
|
|
def active_run(reg: Registry, project_id: str) -> ProjectRun | None:
|
|
return next((p for p in list_runs(reg, project_id) if not p.closed), None)
|
|
|
|
|
|
# ---- create --------------------------------------------------------------------------------------------------------
|
|
|
|
def create(reg: Registry, doc: Doc, operator: str, mode: str = "manual") -> ProjectRun:
|
|
if doc.data.get("spec") != "project/v0.1":
|
|
raise RunError("a template is not run; instantiate it into a project first")
|
|
errs = J.validate(reg, doc)
|
|
if errs:
|
|
raise RunError(f"project {doc.id} is invalid: {errs[0]}")
|
|
if mode not in ("manual", "auto-confirm"):
|
|
raise RunError("mode must be manual or auto-confirm")
|
|
needs = J.missing_required(reg, doc.data)
|
|
if needs:
|
|
raise RunError("attributes still needed: " + ", ".join(sorted({n["attribute"] for n in needs})))
|
|
running = active_run(reg, doc.id)
|
|
if running:
|
|
raise RunError(f"project run {running.id} is still open; close it first")
|
|
operator = _clean(operator.strip(), "operator")
|
|
d = doc.data
|
|
members, snapshot = [], {}
|
|
for spec in d["runbooks"]:
|
|
k = key_of(spec)
|
|
members.append({"key": k, "pack": spec["pack"], "target": spec.get("target"), "after": list(spec.get("after", [])), "run_id": None})
|
|
snapshot[k] = J.member_values(reg, d, spec)
|
|
data = {"spec": "project-run/v0.1", "id": time.strftime("%Y%m%d-%H%M%S-") + uuid.uuid4().hex[:6], "project": doc.id,
|
|
"title": d.get("title", doc.id), "environment": d["environment"], "operator": operator, "mode": mode,
|
|
"created": _now(), "members": members, "snapshot": snapshot, "events": [], "deviations": [], "outcome": None}
|
|
pr = ProjectRun(runs_dir(doc) / f"{data['id']}.json", data)
|
|
pr.event(f"started the project run ({mode} mode)")
|
|
pr.save()
|
|
return pr
|
|
|
|
|
|
# ---- derived view ---------------------------------------------------------------------------------------------------
|
|
|
|
def _member_run(reg: Registry, m: dict):
|
|
if not m.get("run_id"):
|
|
return None
|
|
pack = reg.packs.get(m["pack"])
|
|
try:
|
|
return R.load(pack, m["run_id"]) if pack else None
|
|
except RunError:
|
|
return None
|
|
|
|
|
|
def _done(status: str) -> bool:
|
|
return status in ("succeeded", "pushed")
|
|
|
|
|
|
def view(reg: Registry, pr: ProjectRun) -> dict:
|
|
"""Members with their derived status, what is ready, and overall progress."""
|
|
rows = []
|
|
for m in pr.data["members"]:
|
|
run = _member_run(reg, m)
|
|
row = {**m, "status": "waiting", "progress": None, "deviations": 0, "detail": ""}
|
|
if run is not None:
|
|
d = run.data
|
|
row["deviations"] = len(d.get("deviations", []))
|
|
if d.get("outcome"):
|
|
pushed = any(s["state"] == "pushed" for s in d["steps"].values()) or any(
|
|
g["state"] == "pushed" for g in d.get("gates", {}).values())
|
|
row["status"] = "pushed" if d["outcome"] == "failed" and pushed else d["outcome"]
|
|
else:
|
|
row["status"] = "running"
|
|
if run.pack.spec == "runbook-pack/v0.2":
|
|
ov = run2.overview(run)
|
|
row["progress"] = {"done": ov["progress"]["done"] + ov["progress"]["pushed"], "total": ov["progress"]["total"]}
|
|
ap = d["auto_confirm"]
|
|
if row["status"] == "running" and ap.get("stopped"):
|
|
row["detail"] = f"auto-confirm stopped: {ap['stopped']['reason']}"
|
|
elif row["status"] == "running" and d["mode"] == "auto-confirm" and ap.get("last"):
|
|
row["detail"] = f"{ap['last']['action']} {ap['last'].get('item') or ''} {ap['last'].get('detail', '')}".strip()
|
|
else:
|
|
steps = list(d["steps"].values())
|
|
row["progress"] = {"done": sum(s["state"] in ("done", "skipped") for s in steps), "total": len(steps)}
|
|
rows.append(row)
|
|
by_pack: dict[str, list[dict]] = {}
|
|
for r in rows:
|
|
by_pack.setdefault(r["pack"], []).append(r)
|
|
for r in rows:
|
|
if r["status"] != "waiting":
|
|
continue
|
|
unmet = [a for a in r["after"] if not all(_done(x["status"]) for x in by_pack.get(a, []))]
|
|
r["blocked_by"] = unmet
|
|
r["status"] = "ready" if not unmet else "waiting"
|
|
open_member = next((r for r in rows if r["status"] == "running"), None)
|
|
total = len(rows)
|
|
done = sum(r["status"] in ("succeeded", "pushed") for r in rows)
|
|
return {"members": rows, "open": open_member["key"] if open_member else None, "done": done, "total": total,
|
|
"ready": [r["key"] for r in rows if r["status"] == "ready"],
|
|
"problems": [r["key"] for r in rows if r["status"] in ("failed", "abandoned", "pushed")]}
|
|
|
|
|
|
def _writable(pr: ProjectRun) -> None:
|
|
if pr.closed:
|
|
raise RunError(f"this project run is closed and locked ({pr.data['outcome']})")
|
|
|
|
|
|
# ---- starting members ------------------------------------------------------------------------------------------------------
|
|
|
|
def start_member(reg: Registry, doc: Doc, pr: ProjectRun, key: str, override_reason: str = "", acknowledge_risk: bool = False):
|
|
"""Start the member's run with the snapshot values. One member is open at a time; out of order needs an override."""
|
|
_writable(pr)
|
|
v = view(reg, pr)
|
|
row = next((r for r in v["members"] if r["key"] == key), None)
|
|
if row is None:
|
|
raise RunError(f"no such member: {key}")
|
|
if row["status"] in ("running", "succeeded", "pushed", "failed", "abandoned") and row.get("run_id"):
|
|
raise RunError(f"{key} already has a run ({row['status']})")
|
|
if v["open"]:
|
|
raise RunError(f"finish or abandon the open run of {v['open']} first")
|
|
if row["status"] == "waiting":
|
|
need = ", ".join(row["blocked_by"])
|
|
if not override_reason.strip():
|
|
raise RunError(f"{key} waits for {need}. To start it anyway give a reason and acknowledge the risk; this is recorded.")
|
|
if not acknowledge_risk:
|
|
raise RunError("an override needs the operator to acknowledge the risk")
|
|
if pr.data["mode"] != "manual":
|
|
raise RunError("overrides are made in manual mode")
|
|
pr.data["deviations"].append({"kind": "override", "item": key, "unmet": [f"{a} is not done" for a in row["blocked_by"]],
|
|
"reason": _clean(override_reason.strip(), "reason"), "who": pr.data["operator"], "at": _now()})
|
|
spec = J._member(doc.data, row["pack"], row.get("target"))
|
|
mode = pr.data["mode"] if reg.packs[row["pack"]].spec == "runbook-pack/v0.2" else "manual"
|
|
run = J.start_run(reg, doc, row["pack"], pr.data["operator"], row.get("target"), mode,
|
|
values=pr.data["snapshot"][key], extra_context={"project_run": pr.id})
|
|
for m in pr.data["members"]:
|
|
if m["key"] == key:
|
|
m["run_id"] = run.data["run_id"]
|
|
pr.event(f"started {key}" + (" (override)" if row["status"] == "waiting" else ""))
|
|
pr.save()
|
|
return run
|
|
|
|
|
|
def attributes_changed(reg: Registry, doc: Doc, pr: ProjectRun) -> list[str]:
|
|
"""Member keys whose current project attributes differ from the snapshot this project run started with."""
|
|
out = []
|
|
for spec in doc.data["runbooks"]:
|
|
k = key_of(spec)
|
|
if k in pr.data["snapshot"] and J.member_values(reg, doc.data, spec) != pr.data["snapshot"][k]:
|
|
out.append(k)
|
|
return out
|
|
|
|
|
|
# ---- auto-confirm ----------------------------------------------------------------------------------------------------------------
|
|
|
|
def tick(reg: Registry, doc: Doc, pr: ProjectRun) -> dict:
|
|
"""One auto-confirm decision for the project run."""
|
|
_writable(pr)
|
|
if pr.data["mode"] != "auto-confirm":
|
|
raise RunError("tick needs an auto-confirm project run")
|
|
v = view(reg, pr)
|
|
if v["open"]:
|
|
row = next(r for r in v["members"] if r["key"] == v["open"])
|
|
run = _member_run(reg, row)
|
|
if run.pack.spec != "runbook-pack/v0.2" or run.data["mode"] != "auto-confirm":
|
|
return {"action": "awaiting-operator", "member": row["key"], "detail": "this member is run by you, in manual mode"}
|
|
res = run2.auto_confirm_tick(run)
|
|
run.data["auto_confirm"]["last"] = res
|
|
run.save()
|
|
if res["action"] == "finished":
|
|
try:
|
|
R.finalize(run, "succeeded", "closed by auto-confirm")
|
|
outcome = "succeeded"
|
|
except RunError:
|
|
R.finalize(run, "failed", "closed by auto-confirm: pushed through problems")
|
|
outcome = "failed"
|
|
pr.event(f"{row['key']} finished ({outcome})")
|
|
pr.save()
|
|
return {"action": "member-finished", "member": row["key"], "detail": outcome}
|
|
return {**res, "member": row["key"]}
|
|
if v["ready"]:
|
|
start_member(reg, doc, pr, v["ready"][0])
|
|
return {"action": "member-started", "member": v["ready"][0], "detail": ""}
|
|
if v["done"] == v["total"]:
|
|
return {"action": "finished", "member": None, "detail": "every member is done"}
|
|
stuck = [r for r in v["members"] if r["status"] == "waiting"]
|
|
why = "; ".join(f"{r['key']} waits for {', '.join(r['blocked_by'])}" for r in stuck) or "nothing can start"
|
|
return {"action": "blocked", "member": None, "detail": why}
|
|
|
|
|
|
def run_until_waits(reg: Registry, doc: Doc, pr: ProjectRun, max_ticks: int = 300) -> dict:
|
|
"""Tick until something needs a person, everything is done, or the limit is reached. Returns the last result."""
|
|
last = {"action": "idle", "member": None, "detail": ""}
|
|
for _ in range(max_ticks):
|
|
last = tick(reg, doc, pr)
|
|
if last["action"] not in CONTINUE:
|
|
break
|
|
pr.data["last"] = last
|
|
pr.save()
|
|
return last
|
|
|
|
|
|
def switch_mode(reg: Registry, pr: ProjectRun, mode: str) -> None:
|
|
"""Take over or hand over the project run; open v0.2 member runs follow."""
|
|
_writable(pr)
|
|
if mode not in ("manual", "auto-confirm") or mode == pr.data["mode"]:
|
|
raise RunError(f"the project run is already in {pr.data['mode']} mode" if mode == pr.data["mode"] else "mode must be manual or auto-confirm")
|
|
for m in pr.data["members"]:
|
|
run = _member_run(reg, m)
|
|
if run is not None and not run.data.get("outcome") and run.pack.spec == "runbook-pack/v0.2" and run.data["mode"] != mode:
|
|
run2.switch_mode(run, mode)
|
|
pr.data["mode"] = mode
|
|
pr.event(f"switched to {mode} mode")
|
|
pr.save()
|
|
|
|
|
|
# ---- closing -------------------------------------------------------------------------------------------------------------------------
|
|
|
|
def finalize(reg: Registry, pr: ProjectRun, outcome: str, note: str = "") -> None:
|
|
"""Close the project run, locked and sealed. Open member runs are abandoned with it."""
|
|
_writable(pr)
|
|
if outcome not in OUTCOMES:
|
|
raise RunError(f"outcome must be one of {', '.join(OUTCOMES)}")
|
|
v = view(reg, pr)
|
|
if outcome == "succeeded" and not all(r["status"] == "succeeded" for r in v["members"]):
|
|
raise RunError("a project run succeeds only when every member run succeeded")
|
|
if outcome == "failed" and not v["problems"] and not v["open"]:
|
|
raise RunError("a project run can be finished as failed only when a member failed, was abandoned or pushed through; otherwise abandon it")
|
|
for r in v["members"]:
|
|
if r["status"] == "running":
|
|
run = _member_run(reg, r)
|
|
R.finalize(run, "abandoned", f"project run closed as {outcome}")
|
|
pr.data["outcome"] = outcome
|
|
pr.data["finished"] = _now()
|
|
if note.strip():
|
|
pr.data["outcome_note"] = _clean(note.strip(), "note")
|
|
pr.event(f"closed the project run as {outcome}")
|
|
pr.data["seal"] = _seal_of(pr.data)
|
|
pr.save()
|
|
|
|
|
|
def seal_state(pr: ProjectRun) -> str:
|
|
return R.record_seal_state(_read_json(pr.path) or pr.data)
|
|
|
|
|
|
def times(pr: ProjectRun, now: datetime | None = None) -> dict:
|
|
now = now or datetime.now(timezone.utc)
|
|
start = R._ts(pr.data["created"])
|
|
end = R._ts(pr.data["finished"]) if pr.data.get("finished") else now
|
|
return {"started": pr.data["created"], "finished": pr.data.get("finished"), "elapsed": max(0.0, (end - start).total_seconds())}
|