Run outcomes, score, explicit exercise recording, and step captures (RBT-WP-0004-T13)

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

rtut/__main__.py

@@ -45,6 +45,13 @@ def _run(pk, a) -> int:
             R.retry(r, a.step); return 0
         if a.action == "skip":
             R.skip(r, a.step, a.note); return 0
+        if a.action == "finalize":
+            R.finalize(r, a.outcome or "", a.note)
+            if r.data["outcome"] == "succeeded" and a.apply:
+                print("recorded:", R.apply_exercise(r))
+            print(r.data["outcome"]); return 0
+        if a.action == "capture":
+            R.add_capture(r, a.step, sys.stdin.read(), a.command); print("captured"); return 0
         if a.action == "feedback":
             R.add_feedback(r, a.rating or "", a.note); print("recorded"); return 0
         old, new = R.exercise_update(r)
@@ -65,7 +72,7 @@ def main(argv: list[str] | None = None) -> int:
         sp = sub.add_parser(name)
         sp.add_argument("root", nargs="?", default=".")
     sp = sub.add_parser("run", help="record a run of a native pack")
-    sp.add_argument("action", choices=["new", "list", "status", "cmd", "start", "verify", "done", "fail", "retry", "skip", "finish", "feedback"])
+    sp.add_argument("action", choices=["new", "list", "status", "cmd", "start", "verify", "done", "fail", "retry", "skip", "finish", "feedback", "finalize", "capture"])
     sp.add_argument("manifest")
     sp.add_argument("run_id", nargs="?")
     sp.add_argument("step", nargs="?")
@@ -73,6 +80,8 @@ def main(argv: list[str] | None = None) -> int:
     sp.add_argument("--param", "-p", action="append", default=[], help="name=value")
     sp.add_argument("--evidence", "-e", action="append", default=[], help="name=value (non-secret)")
     sp.add_argument("--note", default="")
+    sp.add_argument("--outcome", choices=["succeeded", "failed", "abandoned"], help="finalize: how the run ended")
+    sp.add_argument("--command", default="", help="capture: the command you actually ran; output is read from stdin")
     sp.add_argument("--rating", choices=["good", "neutral", "poor"], help="feedback: how the runbook felt")
     sp.add_argument("--yes", action="store_true", help="confirm running a verify command")
     sp.add_argument("--pass", dest="passed", action="store_true", help="manual check: operator confirms it passed")
@@ -80,6 +89,8 @@ def main(argv: list[str] | None = None) -> int:
     sp = sub.add_parser("ui", help="serve the local UI on 127.0.0.1")
     sp.add_argument("roots", nargs="*", default=["."])
     sp.add_argument("--port", type=int, default=8877)
+    sp = sub.add_parser("score", help="how often each runbook was tried and how it went")
+    sp.add_argument("root", nargs="?", default=".")
     sp = sub.add_parser("feedback", help="list the latest feedback of every run of a pack")
     sp.add_argument("manifest")
     sp = sub.add_parser("launch"); sp.add_argument("manifest")
@@ -98,6 +109,13 @@ def main(argv: list[str] | None = None) -> int:
         except KeyboardInterrupt:
             pass
         return 0
+    if a.cmd == "score":
+        from . import run as R
+        for p in _packs(a.root):
+            if p.ok and p.data.get("engine") == "native" and p.spec == "runbook-pack/v0.1":
+                c = R.scorecard(p)
+                print(f"{p.id}\ttried {c['tried']}\tsucceeded {c['succeeded']}\tfailed {c['failed']}\tabandoned {c['abandoned']}\topen {c['open']}\t[{p.data['exercise_status']}]")
+        return 0
     if a.cmd in ("list", "validate"):
         packs = _packs(a.root)
         if not packs:

rtut/run.py

@@ -63,6 +63,12 @@ class Run:
         return all(v["state"] in ("done", "skipped") for v in self.data["steps"].values())
 
 
+
+def _writable(run: Run) -> None:
+    if run.data.get("outcome"):
+        raise RunError(f"run is finished (outcome: {run.data['outcome']}); start a new run")
+
+
 def create(pack: Pack, operator: str, params: dict[str, str] | None = None) -> Run:
     if not pack.ok or pack.data.get("engine") != "native":
         raise RunError("only valid native packs can be run")
@@ -125,6 +131,7 @@ def command_for(run: Run, step_id: str) -> str | None:
 
 
 def start(run: Run, step_id: str) -> None:
+    _writable(run)
     run.step_def(step_id)
     _order(run, step_id)
     st = run.state(step_id)
@@ -140,6 +147,7 @@ def verify(run: Run, step_id: str, confirm: bool = False, manual_ok: bool | None
     Stored: exit code, duration, output size, and whether 'contains' matched.
     Output text is never stored.
     """
+    _writable(run)
     v = run.step_def(step_id)["verify"]
     st = run.state(step_id)
     if st["state"] != "running":
@@ -174,6 +182,7 @@ def verify(run: Run, step_id: str, confirm: bool = False, manual_ok: bool | None
 
 
 def complete(run: Run, step_id: str, evidence: dict[str, str] | None = None, note: str = "") -> None:
+    _writable(run)
     sd = run.step_def(step_id)
     st = run.state(step_id)
     if st["state"] != "running":
@@ -192,6 +201,7 @@ def complete(run: Run, step_id: str, evidence: dict[str, str] | None = None, not
 
 
 def fail(run: Run, step_id: str, note: str) -> None:
+    _writable(run)
     st = run.state(step_id)
     if st["state"] != "running":
         raise RunError(f"step {step_id} is not running")
@@ -200,6 +210,7 @@ def fail(run: Run, step_id: str, note: str) -> None:
 
 
 def retry(run: Run, step_id: str) -> None:
+    _writable(run)
     st = run.state(step_id)
     if st["state"] != "failed":
         raise RunError(f"step {step_id} has not failed")
@@ -209,6 +220,7 @@ def retry(run: Run, step_id: str) -> None:
 
 
 def skip(run: Run, step_id: str, reason: str) -> None:
+    _writable(run)
     sd = run.step_def(step_id)
     if sd.get("risk", "read-only") != "read-only":
         raise RunError("only read-only steps may be skipped")
@@ -262,3 +274,109 @@ def all_feedback(pack: Pack) -> list[dict]:
         if fb:
             out.append({"run_id": rid, "operator": r.data["operator"], **fb})
     return sorted(out, key=lambda x: x["at"], reverse=True)
+
+
+# ---- captures: optional documentation of what was run and what came back --------------------
+
+CAPTURE_MAX_OUTPUT = 20000
+CAPTURE_MAX_COMMAND = 2000
+
+
+def add_capture(run: Run, step_id: str, output: str, command: str = "", source: str = "operator") -> dict:
+    """Keep the command actually run and its output as documentation or for inspection.
+
+    The engine cannot detect every secret. It refuses obvious markers and oversized text;
+    the operator remains responsible for what is pasted.
+    """
+    _writable(run)
+    run.step_def(step_id)
+    st = run.state(step_id)
+    if st["state"] == "todo":
+        raise RunError("start the step before capturing its output")
+    if len(output) > CAPTURE_MAX_OUTPUT or len(command) > CAPTURE_MAX_COMMAND:
+        raise RunError(f"capture too large (output max {CAPTURE_MAX_OUTPUT}, command max {CAPTURE_MAX_COMMAND} characters)")
+    if not output.strip() and not command.strip():
+        raise RunError("nothing to capture")
+    entry = {"command": _clean(command.strip(), "captured command"), "output": _clean(output.rstrip(), "captured output"),
+             "source": source, "at": _now()}
+    st.setdefault("captures", []).append(entry)
+    run.save()
+    return entry
+
+
+def verify_pasted(run: Run, step_id: str, output: str, command: str = "") -> dict:
+    """Check an output-contains step against pasted output, for systems the engine cannot reach."""
+    _writable(run)
+    v = run.step_def(step_id)["verify"]
+    st = run.state(step_id)
+    if st["state"] != "running":
+        raise RunError(f"step {step_id} is not running")
+    if v.get("expect") != "output-contains":
+        raise RunError("only output-contains checks can be verified from pasted output")
+    add_capture(run, step_id, output, command, source="pasted-for-verification")
+    needle = render(v["contains"], run.data["params"], run.pack)
+    result = {"kind": "output-contains", "source": "pasted", "passed": needle in output,
+              "output_bytes": len(output.encode()), "at": _now()}
+    st["verification"] = result
+    run.save()
+    return result
+
+
+# ---- outcome and score ------------------------------------------------------------------------
+
+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."""
+    if run.data.get("outcome"):
+        raise RunError(f"run already finished ({run.data['outcome']})")
+    if outcome not in OUTCOMES:
+        raise RunError(f"outcome must be one of {', '.join(OUTCOMES)}")
+    states = [v["state"] for v in run.data["steps"].values()]
+    if outcome == "succeeded" and any(x != "done" for x in states):
+        raise RunError("a run succeeds only when every step is done (skipped steps do not count)")
+    if outcome == "failed" and "failed" not in states:
+        raise RunError("a run can be finished as failed only when a step has failed; otherwise abandon it")
+    run.data["outcome"] = outcome
+    run.data["finished"] = _now()
+    if note.strip():
+        run.data["outcome_note"] = _clean(note.strip(), "note")
+    run.save()
+
+
+def effective_outcome(run: Run, now: datetime | None = None, stale_days: int = STALE_DAYS) -> str:
+    """finished outcome, or 'open', or 'abandoned' for an open run untouched for stale_days."""
+    if run.data.get("outcome"):
+        return run.data["outcome"]
+    now = now or datetime.now(timezone.utc)
+    last = datetime.fromisoformat(run.data.get("updated") or run.data["created"])
+    return "abandoned" if (now - last).days >= stale_days else "open"
+
+
+def scorecard(pack: Pack, now: datetime | None = None, stale_days: int = STALE_DAYS) -> dict:
+    """How often a runbook was tried and how it went, derived from its run records."""
+    card = {"tried": 0, "succeeded": 0, "failed": 0, "abandoned": 0, "open": 0,
+            "feedback": {"good": 0, "neutral": 0, "poor": 0}}
+    for rid in list_runs(pack):
+        r = load(pack, rid)
+        card["tried"] += 1
+        card[effective_outcome(r, now, stale_days)] += 1
+        fb = latest_feedback(r)
+        if fb:
+            card["feedback"][fb["rating"]] += 1
+    return card
+
+
+def apply_exercise(run: Run, today: date | None = None) -> str:
+    """Write 'exercised <date> by <operator>' into pack.yaml for a succeeded run. Returns the new status."""
+    if run.data.get("outcome") != "succeeded":
+        raise RunError("only a run finished as succeeded can exercise the pack")
+    _, new = exercise_update(run, today)
+    run.pack.path.write_text(new)
+    status = STATUS_LINE.search(new).group(0).split(": ", 1)[1]
+    run.pack.data["exercise_status"] = status
+    run.data["exercise_recorded"] = status
+    run.save()
+    return status

rtut/ui.py

@@ -34,7 +34,7 @@ def page(title: str, body: str) -> bytes:
             "const f=b.closest('form'),n=f.querySelector('[name=note]'),w=f.querySelector('[data-warn]');"
             "if(n&&!n.value.trim()&&!b.dataset.warned){ev.preventDefault();b.dataset.warned='1';w.hidden=false;n.focus();}}));"
             # keep the operator on the current step after every action
-            "const c=document.getElementById(location.hash==='#feedback'?'feedback':'current');if(c)c.scrollIntoView({block:'center'});"
+            "const c=document.getElementById(location.hash==='#feedback'?'feedback':location.hash==='#close'?'close':'current');if(c)c.scrollIntoView({block:'center'});"
             "</script>").encode()
 
 
@@ -45,6 +45,11 @@ def e(s) -> str:
 SMILEYS = (("good", "🙂", "Went well"), ("neutral", "😐", "So-so"), ("poor", "🙁", "Poor"))
 
 
+def score_line(c: dict) -> str:
+    return (f"tried {c['tried']} · succeeded {c['succeeded']} · failed {c['failed']} · "
+            f"abandoned {c['abandoned']} · open {c['open']}")
+
+
 class App:
     def __init__(self, roots: list[Path]):
         self.token = secrets.token_urlsafe(16)
@@ -67,7 +72,8 @@ class App:
                             f"<span class=note>start with: python -m rtut launch {e(p.path)}</span></li>")
             else:
                 rows.append(f"<li><a href='/pack/{i}'>{e(d['title'])}</a> "
-                            f"<span class=pill>{e(d['exercise_status'])}</span></li>")
+                            f"<span class=pill>{e(d['exercise_status'])}</span> "
+                            f"<span class=note>{e(score_line(R.scorecard(p)))}</span></li>")
         return page("Runbooks", f"<h1>Runbooks</h1><ul>{''.join(rows) or '<li>No packs found.</li>'}</ul>")
 
     def pack_page(self, i: int) -> bytes:
@@ -81,11 +87,71 @@ class App:
             + f"</label><br><span class=note>{e(x.get('help', ''))} Never enter secrets.</span></p>"
             for x in d.get("parameters", []))
         return page(d["title"], f"<h1>{e(d['title'])}</h1><p class=pill>{e(d['exercise_status'])}</p>"
-                    f"<p>{e(d.get('outcome', ''))}</p><h2>Prerequisites</h2><ul>{prereq}</ul>"
+                    f"<p>{e(d.get('outcome', ''))}</p><p class=note>{e(score_line(R.scorecard(p)))}</p>"
+                    f"{self.runs_list(i, p)}<h2>Prerequisites</h2><ul>{prereq}</ul>"
                     f"<form method=post action='/run/new/{i}'>{self.hidden()}{fields}"
                     "<p><label>Your name (one word)<br><input name=operator required></label></p>"
                     "<button>Start run</button></form>")
 
+    def runs_list(self, i: int, p) -> str:
+        rows = []
+        for rid in reversed(R.list_runs(p)):
+            r = R.load(p, rid)
+            rows.append(f"<li><a href='/run/{i}/{e(rid)}'>{e(rid)}</a> · {e(r.data['operator'])} · "
+                        f"<span class=pill>{e(R.effective_outcome(r))}</span></li>")
+        return f"<h2>Runs</h2><ul>{''.join(rows)}</ul>" if rows else ""
+
+    def capture_html(self, url: str, s: dict, st: dict, state: str, finished: bool) -> str:
+        saved = ""
+        for c in st.get("captures", []):
+            head = f"$ {c['command']}\n" if c.get("command") else ""
+            saved += f"<pre>{e(head + c['output'][:4000])}</pre><p class=note>captured {e(c['at'][:19])} ({e(c['source'])})</p>"
+        if finished or state == "todo":
+            return f"<details><summary class=note>Captured output</summary>{saved}</details>" if saved else ""
+        paste = ""
+        if state == "running" and s["verify"].get("expect") == "output-contains":
+            paste = f"<button formaction='{url}/verify-pasted' formnovalidate>Check against this output</button>"
+        return (f"<details><summary class=note>Capture command and output (optional)"
+                f"{' - ' + str(len(st.get('captures', []))) + ' saved' if st.get('captures') else ''}</summary>{saved}"
+                f"<form method=post action='{url}/capture'>{self.hidden()}"
+                "<p><input name=command class=grow style='width:100%' placeholder='command you actually ran (optional)'></p>"
+                "<p><textarea name=output rows=5 style='width:100%' placeholder='paste the output'></textarea></p>"
+                f"<p class=row><button>Save capture</button>{paste}</p>"
+                "<p class=note>Kept in the run record as documentation. Never paste secrets: obvious markers are "
+                "refused, but not everything can be detected. Use the check button when the command ran on a system "
+                "this runbook cannot reach.</p></form></details>")
+
+    def close_run_form(self, i: int, rid: str, r, p) -> str:
+        outcome = r.data.get("outcome")
+        if outcome:
+            note = f" - {e(r.data['outcome_note'])}" if r.data.get("outcome_note") else ""
+            box = f"<p><b>Run finished: {e(outcome)}</b>{note}</p>"
+            if outcome == "succeeded":
+                if r.data.get("exercise_recorded"):
+                    box += f"<p>Pack recorded as <b>{e(r.data['exercise_recorded'])}</b>. Commit the change in the owning repository.</p>"
+                else:
+                    old, new = R.exercise_update(r)
+                    line = new.split("exercise_status:", 1)[1].split("\n", 1)[0].strip()
+                    box += (f"<p>Current pack status: <span class=pill>{e(p.data['exercise_status'])}</span>. "
+                            f"Recording this run would set it to <b>{e(line)}</b> in <code>{e(p.path)}</code> "
+                            "(a reviewable edit; commit it in the owning repository).</p>"
+                            f"<form method=post action='/run/{i}/{e(rid)}/exercise'>{self.hidden()}"
+                            "<button>Record pack as exercised</button></form>")
+            return f"<div id=close class=step><h2>Close this run</h2>{box}</div>"
+        states = [v["state"] for v in r.data["steps"].values()]
+        btns = []
+        if r.complete and all(x == "done" for x in states):
+            btns.append("<button name=outcome value=succeeded>Finish: succeeded</button>")
+        if "failed" in states:
+            btns.append("<button name=outcome value=failed onclick=\"return confirm('Finish this run as failed?')\">Finish: failed</button>")
+        btns.append("<button name=outcome value=abandoned onclick=\"return confirm('Abandon this run?')\">Abandon run</button>")
+        return (f"<div id=close class=step><h2>Close this run</h2>"
+                "<p class=note>Finishing records the outcome in the pack's score. A finished run is read-only. "
+                "Runs left open for a week count as abandoned.</p>"
+                f"<form method=post action='/run/{i}/{e(rid)}/finalize'>{self.hidden()}"
+                "<p><input name=note class=grow style='width:100%' placeholder='comment (optional)'></p>"
+                f"<p class=row>{' '.join(btns)}</p></form></div>")
+
     def feedback_form(self, i: int, rid: str, r) -> str:
         fb = R.latest_feedback(r)
         thanks = ""
@@ -111,6 +177,7 @@ class App:
             out.append(f"<p class=bad>{e(msg)}</p>")
         blocked = False
         waiting_for = ""
+        finished = bool(r.data.get("outcome"))
         total = len(d["steps"])
         done_n = sum(1 for x in d["steps"] if r.state(x["id"])["state"] in ("done", "skipped"))
         out.append(f"<p class=eyebrow>{done_n} of {total} steps done</p>"
@@ -163,6 +230,8 @@ class App:
                 act = f"<p class=bad>Failed: {e(st.get('note') or '(no comment)')}</p><form method=post action='{url}/retry'>{self.hidden()}<button>Retry</button></form>"
             elif state == "done" and st.get("note"):
                 act = f"<p class=note>Comment: {e(st['note'])}</p>"
+            if finished and state != "done":
+                act = ""                                  # a finished run is read-only
             if state not in ("done", "skipped") and not blocked:
                 blocked = True
                 waiting_for = f"step {n}, {s['title']}"
@@ -173,12 +242,14 @@ class App:
                        f"<span><b>{n}. {e(s['title'])}</b> "
                        f"<span class=note>[{e(s['owner'])}] {e(s.get('risk', 'read-only'))}</span></span>"
                        f"<span class=pill>{state}</span></span></summary>{block}"
+                       f"{self.capture_html(url, s, st, state, finished)}"
                        f"<p class=note>Done when: {e(s['verify']['done_when'])}</p>{rb}{act}</details>")
         if r.complete:
             out.append("<p><b>All steps done.</b> To record this run as exercising the pack, review and apply: "
                        f"<code>python -m rtut run finish {e(p.path)} {e(rid)} --apply</code></p>")
         if r.complete:
             out.append(self.feedback_form(i, rid, r))
+        out.append(self.close_run_form(i, rid, r, p))
         out.append("<h2>Threat checks</h2><ul>" + "".join(f"<li>{e(t)}</li>" for t in d["threat_checks"]) + "</ul>")
         return page(d["title"], "".join(out))
 
@@ -231,6 +302,15 @@ def make_handler(app: App, port_holder: list):
                     i, rid = int(parts[1]), parts[2]
                     R.add_feedback(R.load(app.packs[i], rid), form.get("rating", ""), form.get("comment", ""))
                     return self._redirect(f"/run/{i}/{rid}#feedback")
+                if len(parts) == 4 and parts[0] == "run" and parts[3] in ("finalize", "exercise"):
+                    i, rid = int(parts[1]), parts[2]
+                    r = R.load(app.packs[i], rid)
+                    if parts[3] == "finalize":
+                        R.finalize(r, form.get("outcome", ""), form.get("note", ""))
+                    else:
+                        R.apply_exercise(r)
+                        app.packs[i] = P.load(app.packs[i].path)      # pick up the new exercise status
+                    return self._redirect(f"/run/{i}/{rid}#close")
                 i, rid, sid, action = int(parts[1]), parts[2], parts[3], parts[4]
                 p = app.packs[i]; r = R.load(p, rid)
                 if action == "start": R.start(r, sid)
@@ -239,6 +319,8 @@ def make_handler(app: App, port_holder: list):
                 elif action == "done":
                     R.complete(r, sid, {k[2:]: v for k, v in form.items() if k.startswith("e_")}, form.get("note", ""))
                 elif action == "fail": R.fail(r, sid, form.get("note", ""))
+                elif action == "capture": R.add_capture(r, sid, form.get("output", ""), form.get("command", ""))
+                elif action == "verify-pasted": R.verify_pasted(r, sid, form.get("output", ""), form.get("command", ""))
                 elif action == "retry": R.retry(r, sid)
                 return self._redirect(f"/run/{i}/{rid}#current")
             except (R.RunError, KeyError, ValueError, IndexError) as ex: