Lock and seal closed runs; feedback with the closing form; Home headlines link to their sections; drop the back link
Some checks failed
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 3s
Work Records / validate (push) Has been cancelled

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
This commit is contained in:
tegwick 2026-09-29 18:03:12 +02:00
parent b7edb8f023
commit 711c4e1149
13 changed files with 381 additions and 42 deletions

View file

@ -44,10 +44,22 @@ class Run:
def path(self) -> Path:
return runs_dir(self.pack) / f"{self.data['run_id']}.json"
def save(self) -> None:
def save(self, allow: tuple = ()) -> None:
"""Write the record. A closed run is locked: nothing may change except the keys named in `allow`."""
disk = _read_json(self.path)
if disk and disk.get("outcome"):
keys = set(disk) | set(self.data)
changed = {k for k in keys if disk.get(k) != self.data.get(k)} - {"updated"} - set(allow)
if changed:
raise RunError(f"this run is closed and locked ({disk['outcome']}); it cannot be changed "
f"({', '.join(sorted(changed))})")
self.data["updated"] = _now()
self.path.parent.mkdir(parents=True, exist_ok=True)
if self.path.exists():
self.path.chmod(0o644) # a sealed file is read-only; the allowed update reopens it briefly
self.path.write_text(json.dumps(self.data, indent=2, sort_keys=True) + "\n")
if self.data.get("outcome"):
self.path.chmod(0o444)
def step_def(self, step_id: str) -> dict:
if self.pack.spec == "runbook-pack/v0.2": # steps live in stages; alternatives are applied
@ -280,12 +292,19 @@ def exercise_update(run: Run, today: date | None = None) -> tuple[str, str]:
RATINGS = {"good": ":)", "neutral": ":|", "poor": ":("}
def add_feedback(run: Run, rating: str, comment: str = "") -> dict:
"""Record how the runbook felt. Allowed at any time; the latest entry is the current one."""
def _feedback_entry(rating: str, comment: str) -> dict:
if rating not in RATINGS:
raise RunError(f"rating must be one of {', '.join(RATINGS)}")
entry = {"rating": rating, "comment": _clean(comment.strip(), "comment"),
"at": datetime.now(timezone.utc).isoformat(timespec="microseconds")} # sortable even within one second
return {"rating": rating, "comment": _clean(comment.strip(), "comment"),
"at": datetime.now(timezone.utc).isoformat(timespec="microseconds")} # sortable even within one second
def add_feedback(run: Run, rating: str, comment: str = "") -> dict:
"""Record how the runbook felt. Possible while the run is open (the latest entry counts); a closed run is locked,
so give the feedback before closing, or with the closing form."""
if run.data.get("outcome"):
raise RunError("this run is closed and locked: its feedback can no longer be changed")
entry = _feedback_entry(rating, comment)
run.data.setdefault("feedback", []).append(entry)
run.save()
return entry
@ -359,8 +378,37 @@ OUTCOMES = ("succeeded", "failed", "abandoned")
STALE_DAYS = 7
def finalize(run: Run, outcome: str, note: str = "") -> None:
"""Close a run. 'succeeded' needs every step done; 'failed' needs a failed step."""
def _seal_of(data: dict) -> str:
"""Fingerprint of a closed record. Timestamps of allowed later updates and the exercise marker are not part of it."""
import hashlib
body = {k: v for k, v in data.items() if k not in ("seal", "updated", "exercise_recorded")}
return hashlib.sha256(json.dumps(body, sort_keys=True).encode()).hexdigest()
def _read_json(path) -> dict | None:
try:
return json.loads(Path(path).read_text())
except (OSError, ValueError):
return None
def seal_state(run: Run) -> str:
"""'open' | 'sealed' | 'broken' (changed after closing, outside the app) | 'unsealed' (closed before sealing existed).
Judged on what is on disk, not on the in-memory copy."""
disk = _read_json(run.path) or run.data
if not disk.get("outcome"):
return "open"
if not disk.get("seal"):
return "unsealed"
return "sealed" if disk["seal"] == _seal_of(disk) else "broken"
def finalize(run: Run, outcome: str, note: str = "", rating: str | None = None, comment: str = "") -> None:
"""Close a run: its outcome is recorded, its record sealed and locked. An optional rating and comment are
recorded with it, because a closed run can no longer take feedback.
'succeeded' needs every step done; 'failed' needs a failed or pushed-through step."""
if run.data.get("outcome"):
raise RunError(f"run already finished ({run.data['outcome']})")
if outcome not in OUTCOMES:
@ -373,10 +421,14 @@ def finalize(run: Run, outcome: str, note: str = "") -> None:
gstates = [g["state"] for g in run.data.get("gates", {}).values()]
if outcome == "failed" and not ({"failed", "pushed"} & set(states) or {"failed", "pushed"} & set(gstates)):
raise RunError("a run can be finished as failed only when a step or gate has failed or was pushed through; otherwise abandon it")
entry = _feedback_entry(rating, comment) if rating else None # validated before anything changes
run.data["outcome"] = outcome
run.data["finished"] = _now()
if note.strip():
run.data["outcome_note"] = _clean(note.strip(), "note")
if entry:
run.data.setdefault("feedback", []).append(entry)
run.data["seal"] = _seal_of(run.data)
run.save()
@ -462,5 +514,5 @@ def apply_exercise(run: Run, today: date | None = None) -> str:
status = STATUS_LINE.search(new).group(0).split(": ", 1)[1]
run.pack.data["exercise_status"] = status
run.data["exercise_recorded"] = status
run.save()
run.save(allow=("exercise_recorded",)) # the one thing a closed run may still receive
return status