tegwick
0203bf553c
Also fixes run lists and scores mixing runs of different packs in one repository. 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
244 lines
11 KiB
Python
244 lines
11 KiB
Python
"""Projects: runbooks that share an environment, with layered attributes, derived state and templates."""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
from . import maturity as _m # noqa: F401 (kept importable for callers)
|
|
from . import run as R
|
|
from .attrs import Doc, Registry, check_value, fold, normalize, set_layers
|
|
from .pack import digest
|
|
|
|
PROJECT_STATES = ("planned", "applying", "applied", "stale")
|
|
|
|
|
|
# ---- validation --------------------------------------------------------------------------------------
|
|
|
|
def validate(reg: Registry, doc: Doc) -> list[str]:
|
|
"""Structural errors plus references that must resolve in the registry."""
|
|
if not doc.ok:
|
|
return list(doc.errors)
|
|
d, errs = doc.data, []
|
|
for s in d.get("attribute_sets", []):
|
|
try:
|
|
set_layers(reg, s)
|
|
except ValueError as e:
|
|
errs.append(str(e))
|
|
targets = d.get("targets", {})
|
|
for tid, t in targets.items():
|
|
for s in t.get("attribute_sets", []):
|
|
try:
|
|
set_layers(reg, s)
|
|
except ValueError as e:
|
|
errs.append(f"target {tid}: {e}")
|
|
packs = [m["pack"] for m in d["runbooks"]]
|
|
keys = [(m["pack"], m.get("target")) for m in d["runbooks"]]
|
|
if len(keys) != len(set(keys)):
|
|
errs.append("a runbook appears twice for the same target")
|
|
for m in d["runbooks"]:
|
|
if m["pack"] not in reg.packs:
|
|
errs.append(f"runbook {m['pack']} not found")
|
|
elif not reg.packs[m["pack"]].ok:
|
|
errs.append(f"runbook {m['pack']} is invalid")
|
|
if m.get("target") and m["target"] not in targets:
|
|
errs.append(f"runbook {m['pack']}: unknown target {m['target']}")
|
|
for a in m.get("after", []):
|
|
if a not in packs:
|
|
errs.append(f"runbook {m['pack']}: after names {a}, which is not a member")
|
|
# acyclic member order
|
|
deps = {m["pack"]: set(m.get("after", [])) for m in d["runbooks"]}
|
|
state: dict[str, int] = {}
|
|
|
|
def visit(n: str, path: list[str]) -> None:
|
|
if state.get(n) == 1:
|
|
errs.append("member order cycle: " + " -> ".join(path[path.index(n):] + [n])); return
|
|
if state.get(n) == 2 or n not in deps:
|
|
return
|
|
state[n] = 1
|
|
for x in sorted(deps[n]):
|
|
visit(x, path + [n])
|
|
state[n] = 2
|
|
|
|
for n in sorted(deps):
|
|
visit(n, [])
|
|
return sorted(set(errs))
|
|
|
|
|
|
# ---- attribute resolution --------------------------------------------------------------------------
|
|
|
|
def _member(d: dict, pack_id: str, target: str | None = None) -> dict:
|
|
for m in d["runbooks"]:
|
|
if m["pack"] == pack_id and (target is None or m.get("target") == target):
|
|
return m
|
|
raise KeyError(pack_id)
|
|
|
|
|
|
def layers_for(reg: Registry, d: dict, member: dict) -> list:
|
|
layers: list = []
|
|
for s in d.get("attribute_sets", []):
|
|
layers += set_layers(reg, s)
|
|
layers.append(("project", normalize(d.get("attributes"))))
|
|
t = d.get("targets", {}).get(member.get("target") or "", None)
|
|
if t:
|
|
for s in t.get("attribute_sets", []):
|
|
layers += set_layers(reg, s)
|
|
layers.append((f"target:{member['target']}", normalize(t.get("attributes"))))
|
|
layers.append((f"member:{member['pack']}", normalize(member.get("attributes"))))
|
|
return layers
|
|
|
|
|
|
def member_attributes(reg: Registry, d: dict, member: dict) -> dict:
|
|
"""What the member runbook's inputs look like in this project.
|
|
|
|
{'rows': [{id, label, value, state, requirement, source, modified, error?}], 'unused': [names]}
|
|
state: nil (no value), set, err (invalid for the runbook). requirement: required unless the attribute
|
|
says optional or the runbook itself has a default.
|
|
"""
|
|
pack = reg.packs[member["pack"]]
|
|
folded = fold(layers_for(reg, d, member))
|
|
defs = [x for x in (pack.data.get("datapoints") or pack.data.get("parameters") or []) if not x.get("produced_by")]
|
|
rows = []
|
|
for x in defs:
|
|
a = folded.get(x["id"], {"value": None, "requirement": None, "source": None, "modified": False})
|
|
value = a["value"]
|
|
row = {"id": x["id"], "label": x.get("label", x["id"]), "value": value, "source": a["source"],
|
|
"modified": a["modified"], "requirement": a["requirement"] or ("optional" if "default" in x else "required")}
|
|
if value is None:
|
|
row["state"] = "nil"
|
|
if "default" in x:
|
|
row["default"] = x["default"]
|
|
else:
|
|
err = check_value(pack, x["id"], value)
|
|
row["state"] = "err" if err else "set"
|
|
if err:
|
|
row["error"] = err
|
|
rows.append(row)
|
|
known = {x["id"] for x in defs}
|
|
return {"rows": rows, "unused": sorted(n for n in folded if n not in known)}
|
|
|
|
|
|
def missing_required(reg: Registry, d: dict) -> list[dict]:
|
|
"""Attributes some member still needs: required, no value, no runbook default."""
|
|
out = []
|
|
for m in d["runbooks"]:
|
|
if m["pack"] not in reg.packs:
|
|
continue
|
|
for r in member_attributes(reg, d, m)["rows"]:
|
|
if r["state"] == "nil" and r["requirement"] == "required" and "default" not in r:
|
|
out.append({"pack": m["pack"], "attribute": r["id"], "label": r["label"]})
|
|
return out
|
|
|
|
|
|
def member_values(reg: Registry, d: dict, member: dict) -> dict[str, str]:
|
|
"""Values to hand to a run: every set attribute (invalid ones are left out and reported as err)."""
|
|
return {r["id"]: (str(r["value"]).lower() if isinstance(r["value"], bool) else str(r["value"]))
|
|
for r in member_attributes(reg, d, member)["rows"] if r["state"] == "set"}
|
|
|
|
|
|
# ---- state ---------------------------------------------------------------------------------------------
|
|
|
|
def _latest_run(pack, project_id: str, target: str | None):
|
|
best = None
|
|
for rid in R.list_runs(pack):
|
|
r = R.load(pack, rid)
|
|
ctx = r.data.get("context", {})
|
|
if ctx.get("project") == project_id and ctx.get("target") == target:
|
|
if best is None or r.data["created"] >= best.data["created"]:
|
|
best = r
|
|
return best
|
|
|
|
|
|
def member_status(reg: Registry, d: dict, member: dict) -> dict:
|
|
"""not-applied | applying | applied | stale | failed, from the latest run at the current pack revision."""
|
|
pack = reg.packs[member["pack"]]
|
|
r = _latest_run(pack, d["id"], member.get("target"))
|
|
if r is None:
|
|
return {"pack": member["pack"], "target": member.get("target"), "status": "not-applied", "run": None}
|
|
eff = R.effective_outcome(r)
|
|
current = r.data.get("pack_digest") == digest(pack)
|
|
if eff == "succeeded":
|
|
status = "applied" if current else "stale"
|
|
elif eff in ("failed", "abandoned"):
|
|
status = "failed"
|
|
else:
|
|
status = "applying"
|
|
return {"pack": member["pack"], "target": member.get("target"), "status": status, "run": r.data["run_id"],
|
|
"deviations": len(r.data.get("deviations", [])), "current_revision": current}
|
|
|
|
|
|
def state(reg: Registry, doc: Doc) -> dict:
|
|
"""Project state: planned | applying | applied | stale, with per-member status and what can start next."""
|
|
d = doc.data
|
|
members = [member_status(reg, d, m) for m in d["runbooks"]]
|
|
st = [m["status"] for m in members]
|
|
if "stale" in st:
|
|
project = "stale"
|
|
elif all(x == "applied" for x in st):
|
|
project = "applied"
|
|
elif all(x == "not-applied" for x in st):
|
|
project = "planned"
|
|
else:
|
|
project = "applying"
|
|
applied = {m["pack"] for m in members if m["status"] == "applied"}
|
|
nxt = [m["pack"] for m, spec in zip(members, d["runbooks"])
|
|
if m["status"] in ("not-applied", "failed", "stale") and set(spec.get("after", [])) <= applied]
|
|
return {"state": project, "members": members, "next": nxt, "needs": missing_required(reg, d)}
|
|
|
|
|
|
# ---- runs from a project ------------------------------------------------------------------------------------
|
|
|
|
def start_run(reg: Registry, doc: Doc, pack_id: str, operator: str, target: str | None = None, mode: str = "manual"):
|
|
"""Create a run of a member runbook with the project's attributes as defaults and the project as context."""
|
|
d = doc.data
|
|
if d["spec"] != "project/v0.1":
|
|
raise R.RunError("a template is not run; instantiate it into a project first")
|
|
errs = validate(reg, doc)
|
|
if errs:
|
|
raise R.RunError(f"project {doc.id} is invalid: {errs[0]}")
|
|
member = _member(d, pack_id, target)
|
|
pack = reg.packs[pack_id]
|
|
ctx = {"project": doc.id, "environment": d["environment"], "target": member.get("target")}
|
|
values = member_values(reg, d, member)
|
|
if pack.spec == "runbook-pack/v0.2":
|
|
from . import run2
|
|
return run2.create(pack, operator, values, mode=mode, context=ctx)
|
|
if mode != "manual":
|
|
raise R.RunError("v0.1 packs run in manual mode only")
|
|
return R.create(pack, operator, values, ctx)
|
|
|
|
|
|
# ---- templates -----------------------------------------------------------------------------------------------
|
|
|
|
def instantiate(reg: Registry, template_id: str, project_id: str, environment: str,
|
|
values: dict[str, str] | None = None, title: str | None = None, root: Path | None = None) -> Path:
|
|
"""Write projects/<id>/project.yaml from a template. Required attributes may stay unset; `state()` lists them."""
|
|
t = reg.templates.get(template_id)
|
|
if t is None or not t.ok:
|
|
raise ValueError(f"no valid template {template_id}")
|
|
if not re.fullmatch(r"[a-z0-9]+([.-][a-z0-9]+)*", project_id):
|
|
raise ValueError("project id must be lower-case letters, digits, dots and hyphens")
|
|
if environment not in R.ENVIRONMENTS:
|
|
raise ValueError(f"environment must be one of {', '.join(R.ENVIRONMENTS)}")
|
|
d = {k: v for k, v in t.data.items() if k not in ("spec", "id", "title")}
|
|
attrs = normalize(d.get("attributes"))
|
|
for k, v in (values or {}).items():
|
|
attrs.setdefault(k, {"value": None, "requirement": None})
|
|
attrs[k]["value"] = v
|
|
out_attrs = {}
|
|
for k, a in attrs.items():
|
|
out_attrs[k] = a["value"] if a.get("value") is not None and not a.get("requirement") and not a.get("note") else \
|
|
{kk: vv for kk, vv in a.items() if vv is not None}
|
|
project = {"spec": "project/v0.1", "id": project_id, "title": title or t.data["title"],
|
|
**{k: v for k, v in d.items() if k not in ("attributes", "environment")}, "environment": environment,
|
|
"template": template_id}
|
|
if out_attrs:
|
|
project["attributes"] = out_attrs
|
|
home = Path(root) if root else t.path.parent.parent
|
|
target = home / "projects" / project_id / "project.yaml"
|
|
if target.exists():
|
|
raise ValueError(f"{target} already exists")
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.write_text(yaml.safe_dump(project, sort_keys=False))
|
|
return target
|