diff --git a/docs/REFERENCE.md b/docs/REFERENCE.md index 5955371..34243b4 100644 --- a/docs/REFERENCE.md +++ b/docs/REFERENCE.md @@ -20,6 +20,7 @@ namespace — that is what lets a claim in - [`uphold schema` — the policy file's JSON Schema](#uphold-schema--the-policy-files-json-schema) - [The review tier](#the-review-tier) - [Controls: can a record produce a finding at all?](#controls-can-a-record-produce-a-finding-at-all) + - [Ruling controls](#ruling-controls) - [Design rulings: `[decide]`](#design-rulings-decide) ## Rule shape @@ -4372,6 +4373,71 @@ answering at a prompt passes a longer one. No model client lives here; the reviewer command carries the model, the key and the budget. See [DESIGN.md](DESIGN.md#a-control-is-the-probe-fixture-one-tier-up). +### Ruling controls + +A ruling has a different shape from a review: a question and its options in, +an answer and the record lines it rests on out. The skill that produces one is +tracked here, in the [Agent Skills](https://agentskills.io/specification) +format, at [`skills/design-ruling/SKILL.md`](../skills/design-ruling/SKILL.md). +A ruling control is the question with its ruled answer, written beside the +review controls: + +```toml +# policy/upheld.toml +[[review.ruling]] +question = "Should the command prompt when stdin is not a terminal?" +options = ["(a) prompt (recommended)", "(b) refuse with a non-zero exit"] +answer = "(b) refuse with a non-zero exit" # or: escalate = true, or: none = true +``` + +```sh +uphold_check.py --review --drive --rule [--runs N] [--agree M] [--timeout SECONDS] \ + [--audit-every N --auditor 'COMMAND'] [--seed N] -- ./my-ruler --model x +``` + +Everything after `--` is the ruler, run once per question per run. It reads +`{"document", "question", "options"}` on stdin and prints one JSON object: +`{"answer": "", "citations": [...]}`, +`{"none": "", "citations": [...]}` where no option meets the goal, or +`{"escalate": ""}`. A citation is `{"record", "line", "quote"}`. The +document is every record `[decide] adopt` names, each file numbered line by +line, and the orders; it is read from `uphold check --json`, so a binary that +does not report `decide` is exit `2`. The options are shuffled on every run +and any "(recommended)" is taken off them, so the ruler is not ruling on the +proposer's label; `--seed` repeats a shuffle, and every run prints its seed. + +A run gives the ruling when its kind and option are the control's, an answer +or a `none` cites at least one line, and every citation names an adopted +record, a line its file has, and a quote that line contains. A control holds +when at least `--agree` runs give it (default: all `--runs`, default 3). With +`--auditor`, one run that gave the ruling in every `--audit-every` (default 5), +at least one per control, is handed to that command as `{"document", +"question", "options", "answer" or "none", "citations"}` and nothing else the +ruler printed; it prints `{"supported": true}` or `{"supported": false, +"reason": "..."}`, and an unsupported run stops the control holding. Without +one, the report says the citations' support is unaudited. The skill's +closest-precedent step is not scored: the document carries no issue history, +so a precedent a ruler named could not be checked. + +```text +seed 3 (pass --seed 3 to shuffle the options the same way) +Should the command prompt when stdin is not a terminal?: ruled '(b) refuse ...', held 3/3 (agree 3), audited 1 (0 unsupported) -- holds + given: '(b) refuse with a non-zero exit', '(b) refuse with a non-zero exit', '(b) refuse with a non-zero exit' +pass@1 1.00; pass^3 1/1 control(s) +``` + +Exit `0` when every ruling control holds and `1` when one does not, or when no +record is adopted. What the review document would refuse (an unclaimed record, +the line ceiling) does not stop a ruling run: a ruling is answered from the +adopted records, not from that document. Exit `2` when no ruling control is +declared, when an adopted id is missing from the catalog beside +`uphold_check.py` (a binary and a script from different releases), for a control +with no `question`, fewer than two distinct `options`, an `answer` that is not +one of them, or not exactly one of `answer`, `escalate = true` and +`none = true`, and for a ruler or auditor that exits non-zero, stalls past +`--timeout`, or prints anything but the objects above, an answer that is not an +option it was shown included. + ## Design rulings: `[decide]` A repository's design questions are answered from the records it adopts, and diff --git a/scripts/review.py b/scripts/review.py index 17bb3f1..c4672d8 100644 --- a/scripts/review.py +++ b/scripts/review.py @@ -41,8 +41,11 @@ from __future__ import annotations import json +import random +import re import subprocess from collections.abc import Callable +from pathlib import Path #: Below the bottom of the range in which review quality was observed to #: degrade. A limit set at the top of a range is a limit that reaches the @@ -454,3 +457,363 @@ def render_drive(results: list[dict], runs: int) -> str: f"false positives {rate:.2f} over {len(measured)} control(s) with a misses" ) return "\n".join(lines) + "\n" + + +# --------------------------------------------------------------------------- +# Rulings +# --------------------------------------------------------------------------- +# +# A ruling has a different shape from a review: a question and its options in, +# an answer and the record lines it rests on out. The harness is the same +# argument `drive` makes one tier down -- a ruler that gives the ruled answer +# one run in three is not one anybody can lean on -- so a ruling control is +# asked k times, and holds only where enough runs give its answer. + +#: How many of the k runs must give a ruling control's answer for it to hold. +#: None means all of them. +DEFAULT_AGREE = None + +#: One answer in this many is handed to the auditor, when one is named. +DEFAULT_AUDIT_EVERY = 5 + +#: Taken off an option before a ruler sees it: the label is the proposer's +#: opinion, and a ruler that reads it is ruling on the label. +RECOMMENDED = re.compile(r"\s*\(\s*recommended\s*\)", re.IGNORECASE) + +RULING_PREAMBLE = """\ +# Rulings + +Answer the question from the records below, which this repository adopts, and +from nothing else. Each record is its file, one line per numbered line, at the +revision this repository pins. + +Print one JSON object: + +- `{{"answer": "", "citations": [...]}}` +- `{{"none": "", "citations": [...]}}` +- `{{"escalate": ""}}` when no record below applies, when two that + apply conflict and no order below settles it, or when the records do not + decide between the options. + +A citation is `{{"record": "", "line": , "quote": ""}}`. +An answer cites at least one. + +## Orders + +{orders} +""" + + +class RulerFailed(Exception): + """The ruler or the auditor did not answer in the protocol. Exit 2.""" + + +def present(options: list[str], rng: random.Random) -> list[str]: + """The options as a ruler sees them: shuffled, with "(recommended)" removed.""" + shown = [RECOMMENDED.sub("", option).strip() for option in options] + rng.shuffle(shown) + return shown + + +def ruling_document( + adopted: list[str], orders: list[dict], records: dict[str, dict] +) -> tuple[str, dict[str, list[str]]]: + """The document a ruler reads, and each adopted record's lines. + + The whole file, numbered, rather than the compiled review entry: a ruling + cites a line, and a line is a position in the file at the pinned revision, + which only the file has. + """ + listed = ( + "\n".join( + f"- `{order['prefer']}` over `{order['over']}`: {order['why']}" + for order in orders + ) + or "- (none)" + ) + parts = [RULING_PREAMBLE.format(orders=listed)] + lines: dict[str, list[str]] = {} + for record_id in adopted: + path = records[record_id]["_path"] + try: + text = Path(path).read_text(encoding="utf-8").splitlines() + except (OSError, UnicodeDecodeError) as error: + raise RulerFailed(f"could not read {path}: {error}") from error + lines[record_id] = text + parts.append(f"\n## {record_id} (principles/{record_id}.toml)\n\n") + parts.extend(f"L{number}: {line}\n" for number, line in enumerate(text, 1)) + return "".join(parts), lines + + +def _parsed(command: list[str], payload: dict, timeout: int) -> dict: + """Run a ruler or an auditor once and return the object it printed.""" + try: + completed = subprocess.run( + command, + input=json.dumps(payload, ensure_ascii=False), + capture_output=True, + encoding="utf-8", + timeout=timeout, + check=False, + ) + except OSError as error: + raise RulerFailed(f"could not run {command[0]!r}: {error}") from error + except subprocess.TimeoutExpired as error: + raise RulerFailed( + f"{command[0]!r} did not answer within {timeout}s; pass --timeout " + f"for one that needs longer" + ) from error + except UnicodeDecodeError as error: + raise RulerFailed( + f"{command[0]!r} printed output that is not UTF-8: {error}" + ) from error + if completed.returncode != 0: + tail = completed.stderr.strip().splitlines()[-5:] + raise RulerFailed( + f"{command[0]!r} exited {completed.returncode}" + + (": " + " | ".join(tail) if tail else "") + ) + try: + answered = json.loads(completed.stdout) + except json.JSONDecodeError as error: + raise RulerFailed( + f"{command[0]!r} printed something that is not one JSON object: {error}" + ) from error + if not isinstance(answered, dict): + raise RulerFailed(f"{command[0]!r} printed JSON that is not an object") + return answered + + +def ask_ruler( + ruler: list[str], + document: str, + question: str, + shown: list[str], + timeout: int = DEFAULT_TIMEOUT, +) -> dict: + """Ask once, and return the answer as `{"kind", "value", "citations"}`. + + `kind` is `answer`, `none` or `escalate`. An answer that is not one of the + options shown, or an object holding more or less than one of the three, is + the ruler breaking the protocol, which says nothing about the ruling. + """ + answered = _parsed( + ruler, {"document": document, "question": question, "options": shown}, timeout + ) + kinds = [kind for kind in ("answer", "none", "escalate") if kind in answered] + if len(kinds) != 1 or not set(answered) <= { + "answer", + "none", + "escalate", + "citations", + }: + raise RulerFailed( + f"{ruler[0]!r} printed {sorted(answered)}; one of answer, none or " + f"escalate is expected, with citations beside an answer or a none" + ) + kind = kinds[0] + value = answered[kind] + if not isinstance(value, str) or not value.strip(): + raise RulerFailed(f"{ruler[0]!r} gave an empty {kind}") + if kind == "answer" and value not in shown: + raise RulerFailed( + f"{ruler[0]!r} answered {value!r}, which is not one of the options it " + f"was shown. Print one option exactly as written" + ) + citations = answered.get("citations", []) + if kind == "escalate" and citations: + raise RulerFailed( + f"{ruler[0]!r} escalated and cited; an escalation cites nothing" + ) + if not isinstance(citations, list) or not all( + isinstance(citation, dict) + and set(citation) == {"record", "line", "quote"} + and isinstance(citation["record"], str) + and isinstance(citation["line"], int) + and not isinstance(citation["line"], bool) + and isinstance(citation["quote"], str) + for citation in citations + ): + raise RulerFailed( + f"{ruler[0]!r} printed citations that are not a list of " + f'{{"record", "line", "quote"}} objects' + ) + return {"kind": kind, "value": value, "citations": citations} + + +def unresolved(citations: list[dict], lines: dict[str, list[str]]) -> list[str]: + """Each citation that does not say what it quotes, in the document's files.""" + found = [] + for citation in citations: + record_id, number, quote = ( + citation["record"], + citation["line"], + citation["quote"], + ) + text = lines.get(record_id) + if text is None: + found.append(f"{record_id} is not an adopted record") + elif not 1 <= number <= len(text): + found.append(f"{record_id} has no line {number}") + elif not quote.strip() or quote not in text[number - 1]: + found.append(f"{record_id} L{number} does not say {quote!r}") + return found + + +def audit( + auditor: list[str], + document: str, + question: str, + options: list[str], + given: dict, + timeout: int = DEFAULT_TIMEOUT, +) -> str | None: + """Ask a fresh auditor whether the citations support the answer. + + It is handed the answer and its citations and nothing else the ruler + printed. Returns None where they support it, and the auditor's reason + where they do not. + """ + answered = _parsed( + auditor, + { + "document": document, + "question": question, + "options": options, + given["kind"]: given["value"], + "citations": given["citations"], + }, + timeout, + ) + supported = answered.get("supported") + if not isinstance(supported, bool): + raise RulerFailed(f'{auditor[0]!r} printed no "supported": true or false') + if supported: + return None + reason = answered.get("reason") + return reason if isinstance(reason, str) and reason.strip() else "no reason given" + + +def ruled(control: dict) -> tuple[str, str | None]: + """A control's ruling as `(kind, the option or None)`.""" + if control.get("escalate"): + return "escalate", None + if control.get("none"): + return "none", None + return "answer", RECOMMENDED.sub("", control["answer"]).strip() + + +def drive_rulings( + controls: list[dict], + document: str, + lines: dict[str, list[str]], + runs: int, + rng: random.Random, + ruler: Callable[[str, str, list[str]], dict], + auditor: Callable[[str, str, list[str], dict], str | None] | None = None, + audit_every: int = DEFAULT_AUDIT_EVERY, +) -> list[dict]: + """Ask every ruling control `runs` times and count the runs that held. + + A run holds where its kind and option are the control's and every citation + resolves. Then one held run in `audit_every`, at least one per control + with any, is handed to the auditor, and a run it finds unsupported stops + holding. + """ + results = [] + for control in controls: + kind, option = ruled(control) + held: list[dict] = [] + broken: list[str] = [] + given_kinds: list[str] = [] + for _ in range(runs): + shown = present(control["options"], rng) + given = ruler(document, control["question"], shown) + given_kinds.append( + given["value"] if given["kind"] == "answer" else given["kind"] + ) + if given["kind"] != kind or (kind == "answer" and given["value"] != option): + continue + if given["kind"] in {"answer", "none"} and not given["citations"]: + broken.append(f"a {given['kind']} that cites nothing") + continue + failed = unresolved(given["citations"], lines) + if failed: + broken.extend(failed) + continue + held.append(given) + audited = unsupported = 0 + if auditor is not None: + answered = [given for given in held if given["kind"] != "escalate"] + sample = ( + rng.sample(answered, -(-len(answered) // audit_every)) + if answered + else [] + ) + for given in sample: + audited += 1 + reason = auditor( + document, + control["question"], + [ + RECOMMENDED.sub("", option).strip() + for option in control["options"] + ], + given, + ) + if reason is not None: + unsupported += 1 + broken.append( + f"the auditor found the citations unsupported: {reason}" + ) + held.remove(given) + results.append( + { + "question": control["question"], + "ruled": option if kind == "answer" else kind, + "held": len(held), + "given": given_kinds, + "broken": broken, + "audited": audited, + "unsupported": unsupported, + } + ) + return results + + +def ruling_holds(result: dict, agree: int) -> bool: + """At least `agree` runs gave the ruling, and no audited one was unsupported.""" + return result["held"] >= agree and not result["unsupported"] + + +def render_rulings( + results: list[dict], runs: int, agree: int, seed: int, audited: bool +) -> str: + """One line per control, then pass@1 and pass^k, and the seed to rerun it.""" + lines = [f"seed {seed} (pass --seed {seed} to shuffle the options the same way)"] + for result in results: + verdict = "holds" if ruling_holds(result, agree) else "DOES NOT HOLD" + question = result["question"].splitlines()[0][:72] + audit_note = ( + f", audited {result['audited']} ({result['unsupported']} unsupported)" + if audited + else "" + ) + lines.append( + f"{question}: ruled {result['ruled']!r}, held {result['held']}/{runs} " + f"(agree {agree}){audit_note} -- {verdict}" + ) + given = ", ".join(repr(value) for value in result["given"]) + lines.append(f" given: {given}") + lines.extend(f" - {reason}" for reason in result["broken"]) + if results: + pass_at_1 = sum(result["held"] for result in results) / (len(results) * runs) + pass_k = sum(result["held"] == runs for result in results) + lines.append( + f"pass@1 {pass_at_1:.2f}; pass^{runs} {pass_k}/{len(results)} control(s)" + ) + if not audited: + lines.append( + "no --auditor named, so whether the citations support the answers is unaudited" + ) + return "\n".join(lines) + "\n" diff --git a/skills/design-ruling/SKILL.md b/skills/design-ruling/SKILL.md new file mode 100644 index 0000000..7334821 --- /dev/null +++ b/skills/design-ruling/SKILL.md @@ -0,0 +1,82 @@ +--- +name: design-ruling +description: Answer the open questions in a repository's design issues from the uphold records that repository adopts under [decide], citing each record line at the pinned uphold revision with a quote, or escalate. Use when asked to rule on, answer or decide a design record's questions. +--- + +# Design ruling + +A ruling answers one question in a design issue from the records the +repository adopts, and from nothing else. The owner reviews which records are +adopted and the order between two that conflict; everything else in a ruling +must be traceable to them. + +## 1. Read what the repository rules by + +Run `uphold check --json` from the repository root and read `decide`: + +- `adopt` is the record ids the repository answers design questions from. +- `order` is one `{prefer, over, why}` per adopted pair that conflicts. + +If `uphold check` exits 1 with `design rulings refused ([decide])`, escalate +every question: an adopted conflict with no order has two answers. + +Find the uphold revision the repository pins: the `rev:` of the +`HackingGate/uphold` entry in `.pre-commit-config.yaml`, or the `ref:` of that +remote in `lefthook.yml`. Read each adopted record as +`principles/.toml` at that revision, with `uphold --version` matching it. +A different binary carries a different catalog, and its line numbers are not +the ones the citations will be checked against. + +## 2. Rule on each question + +For each question under the issue's questions heading: + +1. Strip any "(recommended)" from the options and read them in no particular + order. Which option the proposer prefers is not evidence. +2. Find the adopted records whose `applies_when` holds for the question. None + applies: escalate, saying which part of the question no record covers. +3. Where two applicable records conflict, the `order` between them decides. + No order for that pair: escalate. +4. Pick the option the applicable records decide. Cite every record line the + decision rests on: a `claim`, an `applies_when`, a `does_not_mean`, a + `review_questions` item. Where a record's source decides it, cite the + source's `locator` or URL fragment too. +5. Where no option meets the goal the records state, answer that none does, + and cite the line that states the goal. +6. Name the closest ruled precedent: an earlier design issue labelled + `ruling:accepted` whose `Answered:` line cites the same record lines. None: + say so. + +Run the ruling three times, independently, with the options in a different +order each time. All three must give the same answer, or escalate and say they +did not. + +## 3. Write the answer + +Under the question, on its own line: + +```text +Answered: (a). [ L](https://github.com/HackingGate/uphold/blob//principles/.toml#L) ""; precedent: +``` + +Or, to escalate: + +```text +Answered: escalated -- +``` + +The link's revision is the pin, never `main`. The quote follows its link and is +copied from that line, because `ruling-citations-resolve` refuses the body when +the line at that revision does not contain it. Publish with `gh issue edit` or +`gh issue create` so that check runs. Do not change the issue's `ruling:` label: +the owner records the ruling. + +## 4. Score the answer and its citations + +`uphold_check.py --review --drive --rule -- ` asks a ruler each +`[[review.ruling]]` control in `policy/upheld.toml` and scores it; see +docs/REFERENCE.md, "Ruling controls". The ruler reads +`{"document", "question", "options"}` on stdin and prints +`{"answer", "citations"}`, `{"none", "citations"}` or `{"escalate"}`. The +precedent step is not scored: the ruler is handed no issue history to find one +in. diff --git a/tests/test_review.py b/tests/test_review.py index 1070e58..92917b0 100644 --- a/tests/test_review.py +++ b/tests/test_review.py @@ -9,6 +9,7 @@ from __future__ import annotations import json +import random import shutil import subprocess import sys @@ -394,6 +395,354 @@ def reviewer(document: str, change: str, carried: dict) -> set: self.assertNotIn("false positives", report) +class Rulings(unittest.TestCase): + """`--drive --rule`: a ruling control is asked k times and scored. + + The rulers here are functions and small scripts, so no model is involved: + what is tested is the scoring, the shuffle, and the refusals. + """ + + LINES: ClassVar[dict[str, list[str]]] = { + "fail-fast": ['id = "fail-fast"', 'title = "Fail Fast and Explicitly"'] + } + CITED: ClassVar[list[dict]] = [ + {"record": "fail-fast", "line": 1, "quote": "fail-fast"} + ] + CONTROL: ClassVar[dict] = { + "question": "Which?", + "options": ["(a) first (recommended)", "(b) second"], + "answer": "(b) second", + } + + def drive(self, control: dict, ruler, auditor=None, runs: int = 3) -> dict: + (result,) = review_mod.drive_rulings( + [control], "doc", self.LINES, runs, random.Random(7), ruler, auditor + ) + return result + + def always(self, option: str): + return lambda _doc, _question, shown: { + "kind": "answer", + "value": next(value for value in shown if value.startswith(option)), + "citations": self.CITED, + } + + def test_a_ruler_that_always_answers_a_fails_a_control_ruled_b(self): + result = self.drive(self.CONTROL, self.always("(a)")) + self.assertEqual(result["held"], 0) + self.assertFalse(review_mod.ruling_holds(result, 3)) + + def test_a_ruler_that_always_escalates_fails_a_control_with_an_answer(self): + def escalates(*_): + return {"kind": "escalate", "value": "no record applies", "citations": []} + + result = self.drive(self.CONTROL, escalates) + self.assertEqual(result["given"], ["escalate"] * 3) + self.assertFalse(review_mod.ruling_holds(result, 3)) + + def test_a_ruler_giving_the_ruling_every_run_holds(self): + result = self.drive(self.CONTROL, self.always("(b)")) + self.assertEqual(result["held"], 3) + self.assertTrue(review_mod.ruling_holds(result, 3)) + + def test_an_escalation_control_holds_on_escalations_alone(self): + control = {**self.CONTROL, "escalate": True} + del control["answer"] + + def escalates(*_): + return {"kind": "escalate", "value": "unordered conflict", "citations": []} + + self.assertTrue(review_mod.ruling_holds(self.drive(control, escalates), 3)) + + def test_two_runs_in_three_hold_only_where_agree_is_two(self): + answers = iter(["(b)", "(a)", "(b)"]) + + def ruler(doc, question, shown): + return self.always(next(answers))(doc, question, shown) + + result = self.drive(self.CONTROL, ruler) + self.assertEqual(result["held"], 2) + self.assertFalse(review_mod.ruling_holds(result, 3)) + self.assertTrue(review_mod.ruling_holds(result, 2)) + + def test_the_ruler_sees_shuffled_options_without_the_recommendation(self): + seen = [] + + def ruler(doc, question, shown): + seen.append(tuple(shown)) + return self.always("(b)")(doc, question, shown) + + control = {**self.CONTROL, "options": [f"({c}) {c}" for c in "abcdef"]} + control["answer"] = "(b) b" + control["options"][0] += " (Recommended)" + self.drive(control, ruler, runs=5) + self.assertTrue(all("recommended" not in " ".join(s).lower() for s in seen)) + self.assertEqual( + {frozenset(s) for s in seen}, {frozenset(f"({c}) {c}" for c in "abcdef")} + ) + self.assertGreater(len(set(seen)), 1, "the order never changed across runs") + + def test_a_citation_that_does_not_resolve_breaks_the_run(self): + for cited, says in [ + ( + {"record": "least-power", "line": 1, "quote": "x"}, + "not an adopted record", + ), + ({"record": "fail-fast", "line": 9, "quote": "x"}, "has no line 9"), + ({"record": "fail-fast", "line": 2, "quote": "fail-fast"}, "does not say"), + ]: + with self.subTest(cited=cited): + + def ruler(_doc, _question, shown, cited=cited): + return { + "kind": "answer", + "value": "(b) second", + "citations": [cited], + } + + result = self.drive(self.CONTROL, ruler) + self.assertEqual(result["held"], 0) + self.assertIn(says, result["broken"][0]) + + def test_an_answer_that_cites_nothing_breaks_the_run(self): + def ruler(*_): + return {"kind": "answer", "value": "(b) second", "citations": []} + + result = self.drive(self.CONTROL, ruler) + self.assertEqual(result["held"], 0) + self.assertIn("cites nothing", result["broken"][0]) + + def test_a_none_that_cites_nothing_breaks_the_run(self): + control = {**self.CONTROL, "none": True} + del control["answer"] + + def ruler(*_): + return {"kind": "none", "value": "no option works", "citations": []} + + result = self.drive(control, ruler) + self.assertEqual(result["held"], 0) + self.assertIn("a none that cites nothing", result["broken"][0]) + + def test_a_decide_report_with_no_adopt_list_is_two_not_an_empty_policy(self): + settings = {"rulings": [self.CONTROL]} + driving = {"seed": 1, "timeout": 1, "auditor": None, "command": ["false"]} + records = {record["id"]: record for record in load_catalog()} + self.assertEqual( + uphold_check.drive_rulings(driving, settings, {"decide": {}}, records), 2 + ) + + def test_an_adopted_id_the_scripts_catalog_lacks_is_two_rather_than_dropped(self): + settings = {"rulings": [self.CONTROL]} + report = { + "decide": {"adopt": ["fail-fast", "not-in-this-catalog"], "order": []} + } + driving = {"seed": 1, "timeout": 1, "auditor": None, "command": ["false"]} + records = {record["id"]: record for record in load_catalog()} + self.assertEqual( + uphold_check.drive_rulings(driving, settings, report, records), 2 + ) + + def test_one_held_answer_in_audit_every_is_audited_and_an_unsupported_one_fails( + self, + ): + asked = [] + + def auditor(_doc, question, options, given): + asked.append(given) + return "the line is about something else" + + result = self.drive(self.CONTROL, self.always("(b)"), auditor) + self.assertEqual(result["audited"], 1) + self.assertEqual(result["unsupported"], 1) + self.assertEqual(sorted(asked[0]), ["citations", "kind", "value"]) + self.assertFalse(review_mod.ruling_holds(result, 2)) + + def ruler_script(self, body: str) -> list[str]: + tmp = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, tmp) + script = Path(tmp) / "ruler.py" + script.write_text( + "import json, sys\npayload = json.load(sys.stdin)\n" + + textwrap.dedent(body), + encoding="utf-8", + ) + return [sys.executable, str(script)] + + def test_the_ruler_reads_the_document_question_and_options_and_nothing_else(self): + ruler = self.ruler_script( + """ + assert sorted(payload) == ["document", "options", "question"], payload + print(json.dumps({"answer": payload["options"][0], "citations": []})) + """ + ) + given = review_mod.ask_ruler(ruler, "doc", "Which?", ["x", "y"]) + self.assertEqual(given, {"kind": "answer", "value": "x", "citations": []}) + + def test_a_ruler_breaking_the_protocol_is_refused_rather_than_scored(self): + for body in [ + 'print("(a)")', + 'print(json.dumps({"answer": "z"}))', + 'print(json.dumps({"answer": "x", "escalate": "both"}))', + 'print(json.dumps({"escalate": "r", "citations": [{"record": "a", "line": 1, "quote": "q"}]}))', + 'print(json.dumps({"answer": "x", "citations": [{"record": "a"}]}))', + 'sys.exit("no key")', + ]: + with self.subTest(body=body), self.assertRaises(review_mod.RulerFailed): + review_mod.ask_ruler( + self.ruler_script(body), "doc", "Which?", ["x", "y"] + ) + + +class RuleCommand(unittest.TestCase): + """`--review --drive --rule` end to end, with a stub ruler and the engine.""" + + def setUp(self): + self._directory = tempfile.TemporaryDirectory() + self.tmp = Path(self._directory.name) + self.addCleanup(self._directory.cleanup) + carried = {value for record in load_catalog() for value in record["domains"]} + spare = sorted(set(DOMAINS) - carried) + if not spare: + self.skipTest("every domain is carried by a record; none routes nothing") + (self.tmp / "policy").mkdir() + (self.tmp / "policy" / "principles.toml").write_text( + '[decide]\nadopt = ["fail-fast"]\n', encoding="utf-8" + ) + (self.tmp / "policy" / "upheld.toml").write_text( + textwrap.dedent( + f""" + [review] + include_domains = ["{spare[0]}"] + + [[review.ruling]] + question = "Which?" + options = ["(a) first (recommended)", "(b) second"] + answer = "(b) second" + """ + ), + encoding="utf-8", + ) + + def rule(self, ruler: str, *argv: str) -> subprocess.CompletedProcess: + return subprocess.run( + [ + sys.executable, + str(SCRIPT), + "--review", + "--drive", + "--rule", + *argv, + "--", + sys.executable, + "-c", + textwrap.dedent(ruler), + ], + cwd=self.tmp, + capture_output=True, + text=True, + check=False, + ) + + ANSWERS_B = """ + import json, sys + payload = json.load(sys.stdin) + assert "L1: id = \\"fail-fast\\"" in payload["document"], payload["document"] + assert payload["question"] == "Which?", payload["question"] + answer = next(o for o in payload["options"] if o.startswith("(b)")) + cited = [{"record": "fail-fast", "line": 1, "quote": "fail-fast"}] + print(json.dumps({"answer": answer, "citations": cited})) + """ + + @needs_the_engine + def test_a_ruler_giving_the_ruling_holds_and_prints_the_seed(self): + result = self.rule(self.ANSWERS_B, "--seed", "3") + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn("seed 3", result.stdout) + self.assertIn("held 3/3 (agree 3)", result.stdout) + self.assertIn("unaudited", result.stdout) + + @needs_the_engine + def test_what_the_review_document_would_refuse_does_not_block_a_ruling(self): + # No include_domains: every `automatable = "yes"` record is unclaimed + # here, which `--review` refuses. The rulings are scored regardless. + upheld = self.tmp / "policy" / "upheld.toml" + text = upheld.read_text(encoding="utf-8") + upheld.write_text( + "\n".join( + line for line in text.splitlines() if "include_domains" not in line + ), + encoding="utf-8", + ) + review = subprocess.run( + [sys.executable, str(SCRIPT), "--review"], + cwd=self.tmp, + capture_output=True, + text=True, + check=False, + ) + self.assertEqual(review.returncode, 1, review.stderr) + self.assertIn("no rule here claims it", review.stderr) + result = self.rule(self.ANSWERS_B) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn("held 3/3", result.stdout) + + @needs_the_engine + def test_a_ruler_that_always_escalates_is_one(self): + result = self.rule( + 'import sys; sys.stdin.read(); print(\'{"escalate": "no"}\')' + ) + self.assertEqual(result.returncode, 1, result.stderr) + self.assertIn("DOES NOT HOLD", result.stdout) + + @needs_the_engine + def test_an_auditor_that_finds_the_citations_unsupported_is_one(self): + auditor = f'{sys.executable} -c \'print("{{\\"supported\\": false}}")\'' + result = self.rule(self.ANSWERS_B, "--auditor", auditor) + self.assertEqual(result.returncode, 1, result.stderr) + self.assertIn("audited 1 (1 unsupported)", result.stdout) + + def test_rule_options_outside_rule_or_out_of_range_are_usage_errors(self): + for argv in [ + ["--agree", "4"], + ["--auditor", ""], + ["--audit-every", "0"], + ["--seed", "²"], + ["--runs", "²"], + ]: + result = self.rule("print(1)", *argv) + self.assertEqual(result.returncode, 2, argv) + self.assertIn("--drive --rule", result.stderr) + result = subprocess.run( + [ + sys.executable, + str(SCRIPT), + "--review", + "--drive", + "--agree", + "2", + "--", + "true", + ], + cwd=self.tmp, + capture_output=True, + text=True, + check=False, + ) + self.assertEqual(result.returncode, 2) + + def test_a_ruling_control_whose_answer_is_not_an_option_is_two(self): + upheld = self.tmp / "policy" / "upheld.toml" + upheld.write_text( + upheld.read_text(encoding="utf-8").replace( + 'answer = "(b) second"', 'answer = "(c) third"' + ), + encoding="utf-8", + ) + result = self.rule("print(1)") + self.assertEqual(result.returncode, 2, result.stderr) + self.assertIn("not one of its `options`", result.stderr) + + @needs_the_engine class ClaimsThatEnforceNothing(unittest.TestCase): """A claim naming a rule no seam supplies must not silence the review. diff --git a/tests/test_skills.py b/tests/test_skills.py new file mode 100644 index 0000000..67b5cfd --- /dev/null +++ b/tests/test_skills.py @@ -0,0 +1,67 @@ +"""The skills uphold tracks carry the frontmatter the Agent Skills spec requires. + +https://agentskills.io/specification: a skill is a directory holding SKILL.md, +whose YAML frontmatter has `name` (1-64 characters, lowercase letters, digits +and single hyphens, not opening or closing with one, matching the directory) +and `description` (1-1024 characters, non-empty). +""" + +from __future__ import annotations + +import re +import unittest +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +SKILLS = ROOT / "skills" +NAME = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$") + + +def frontmatter(text: str) -> dict[str, str]: + """The top-level `key: value` lines between the opening `---` pair.""" + lines = text.splitlines() + if not lines or lines[0] != "---" or "---" not in lines[1:]: + return {} + block = lines[1 : lines.index("---", 1)] + fields = {} + for line in block: + if line.startswith(" ") or ": " not in line: + continue + key, value = line.split(": ", 1) + # `>` or `|` opens a block scalar whose text is on the lines below, + # which this reader does not parse; reading the indicator as the value + # would pass an empty or over-long description. + if re.fullmatch(r"[|>][0-9+-]*(\s+#.*)?", value.strip()): + raise AssertionError( + f"{key}: a block scalar, which this test does not read" + ) + fields[key] = value + return fields + + +class Skills(unittest.TestCase): + def test_there_is_a_skill_to_check(self): + self.assertTrue(list(SKILLS.glob("*/SKILL.md")), f"no skill under {SKILLS}") + + def test_every_skill_carries_a_name_matching_its_directory_and_a_description(self): + for path in sorted(SKILLS.glob("*/SKILL.md")): + with self.subTest(skill=path.parent.name): + fields = frontmatter(path.read_text(encoding="utf-8")) + name = fields.get("name", "") + self.assertTrue(NAME.match(name) and len(name) <= 64, name) + self.assertEqual(name, path.parent.name) + description = fields.get("description", "").strip() + self.assertTrue(0 < len(description) <= 1024, description) + + def test_a_block_scalar_is_refused_rather_than_read_as_its_indicator(self): + with self.assertRaises(AssertionError): + frontmatter("---\nname: x\ndescription: >\n folded text\n---\n") + + def test_a_directory_under_skills_is_a_skill(self): + for path in sorted(SKILLS.iterdir()): + with self.subTest(path=path.name): + self.assertTrue((path / "SKILL.md").is_file(), path) + + +if __name__ == "__main__": + unittest.main() diff --git a/uphold_check.py b/uphold_check.py index 1d64ab7..e0d1a79 100755 --- a/uphold_check.py +++ b/uphold_check.py @@ -47,6 +47,8 @@ import functools import json +import random +import shlex import subprocess import sys import tomllib @@ -649,6 +651,7 @@ def review_settings(declaration: dict) -> dict: "emit_controls": emit_controls, "exempt": exempt, "controls": review_controls(settings), + "rulings": review_rulings(settings), } @@ -713,6 +716,72 @@ def review_controls(settings: dict) -> list[dict]: return declared +def review_rulings(settings: dict) -> list[dict]: + """Read `[[review.ruling]]`, refusing a control that could not be scored. + + A ruling control is a question, its options, and what the ruling is: + exactly one of `answer` (one of the options), `escalate = true` or + `none = true`. Like `review_controls`, every refusal here is the shape of + the declaration, so every one is exit 2. + """ + declared = settings.get("ruling", []) + if not isinstance(declared, list) or not all( + isinstance(entry, dict) for entry in declared + ): + raise CouldNotLook( + "`review.ruling` must be an array of `[[review.ruling]]` tables, each " + "a question, its options and the ruled answer" + ) + for entry in declared: + question = entry.get("question") + if not isinstance(question, str) or not question.strip(): + raise CouldNotLook("every `[[review.ruling]]` needs a non-empty `question`") + at = f"the ruling control {question.splitlines()[0][:60]!r}" + unknown = sorted( + set(entry) - {"question", "options", "answer", "escalate", "none"} + ) + if unknown: + raise CouldNotLook( + f"{at} declares {', '.join(unknown)}, which no ruling control has. A " + f"ruling control is `question`, `options`, and one of `answer`, " + f"`escalate` or `none`" + ) + options = entry.get("options") + shown = ( + [review_mod.RECOMMENDED.sub("", option).strip() for option in options] + if isinstance(options, list) and all(isinstance(o, str) for o in options) + else None + ) + if ( + shown is None + or len(shown) < 2 + or not all(shown) + or len(set(shown)) != len(shown) + ): + raise CouldNotLook( + f"{at} needs `options`, two or more distinct non-empty strings once " + f'any "(recommended)" is taken off' + ) + ruled = [key for key in ("answer", "escalate", "none") if key in entry] + if len(ruled) != 1: + raise CouldNotLook( + f"{at} needs exactly one of `answer`, `escalate = true` or `none = true`" + ) + if ruled == ["answer"]: + answer = entry["answer"] + if not isinstance(answer, str) or ( + review_mod.RECOMMENDED.sub("", answer).strip() not in shown + ): + raise CouldNotLook( + f"{at} has an `answer` that is not one of its `options`" + ) + elif entry[ruled[0]] is not True: + raise CouldNotLook( + f"{at} has `{ruled[0]}` set to something other than true" + ) + return declared + + def emit_target(root: Path, name: str) -> Path: """Where one `review.emit` name writes to, refusing anything outside the repo. @@ -740,29 +809,67 @@ def emit_target(root: Path, name: str) -> Path: REVIEW_USAGE = ( "usage: uphold_check.py --review " - "[--emit | --check | --drive [--runs N] [--timeout SECONDS] -- REVIEWER...]" + "[--emit | --check | --drive [--runs N] [--timeout SECONDS] -- REVIEWER...\n" + " | --drive --rule [--runs N] [--timeout SECONDS] [--agree M] " + "[--audit-every N --auditor 'COMMAND'] [--seed N] -- RULER...]" ) -def drive_arguments(argv: list[str]) -> tuple[int, int, list[str]] | None: - """`--drive [--runs N] [--timeout SECONDS] -- REVIEWER...`, or None. +def drive_arguments(argv: list[str]) -> dict | None: + """`--drive [--rule] [options] -- COMMAND...`, or None. - Returns (runs, timeout, reviewer). The reviewer is everything after `--`, - verbatim, so its own flags never have to be told apart from these. + The command is everything after `--`, verbatim, so its own flags never + have to be told apart from these. `--auditor` is one string split the way + a shell would, because it is the second command on the line and `--` is + already the first one's. `--agree`, `--audit-every`, `--auditor` and + `--seed` belong to `--rule` alone. """ - values = { + rest = argv[1:] + rule = rest[:1] == ["--rule"] + if rule: + rest = rest[1:] + counts = { "--runs": review_mod.DEFAULT_RUNS, "--timeout": review_mod.DEFAULT_TIMEOUT, } - rest = argv[1:] - while rest[:1] and rest[0] in values: - if len(rest) < 2 or not rest[1].isdigit() or int(rest[1]) < 1: + if rule: + counts.update( + { + "--agree": review_mod.DEFAULT_AGREE, + "--audit-every": review_mod.DEFAULT_AUDIT_EVERY, + "--seed": None, + } + ) + auditor: list[str] | None = None + while rest[:1] and (rest[0] in counts or (rule and rest[0] == "--auditor")): + if len(rest) < 2: + return None + if rest[0] == "--auditor": + try: + auditor = shlex.split(rest[1]) + except ValueError: + return None + if not auditor: + return None + elif rest[1].isdecimal() and (int(rest[1]) >= 1 or rest[0] == "--seed"): + counts[rest[0]] = int(rest[1]) + else: return None - values[rest[0]] = int(rest[1]) rest = rest[2:] if rest[:1] != ["--"] or len(rest) < 2: return None - return values["--runs"], values["--timeout"], rest[1:] + if rule and counts["--agree"] is not None and counts["--agree"] > counts["--runs"]: + return None + return { + "rule": rule, + "runs": counts["--runs"], + "timeout": counts["--timeout"], + "agree": counts.get("--agree") or counts["--runs"], + "audit_every": counts.get("--audit-every"), + "auditor": auditor, + "seed": counts.get("--seed"), + "command": rest[1:], + } def run_review(argv: list[str]) -> int: @@ -801,6 +908,11 @@ def run_review(argv: list[str]) -> int: return 2 records = load_records() + # Ahead of the routing: a ruling is answered from the adopted records, not + # from the compiled review document, so what that document would refuse + # says nothing about whether a ruler can be scored. + if driving is not None and driving["rule"]: + return drive_rulings(driving, settings, report, records) # Partial reports identify the exact claims that held. A refused claim may # name the same rule as a reconciled one, so the rule id alone is insufficient. claimed = { @@ -836,7 +948,11 @@ def run_review(argv: list[str]) -> int: return 1 if driving is not None: - runs, timeout, reviewer = driving + runs, timeout, reviewer = ( + driving["runs"], + driving["timeout"], + driving["command"], + ) carried = {record["id"]: record["title"] for record in for_review} try: results = review_mod.drive( @@ -902,6 +1018,95 @@ def run_review(argv: list[str]) -> int: return 0 +def drive_rulings(driving: dict, settings: dict, report: dict, records: dict) -> int: + """`--drive --rule`: ask the ruler every ruling control, and score it. + + The adopted records and their orders are the engine's answer, read off the + same `uphold check --json` report the routing reads, because `[decide]` is + in the policy and this script does not read the policy. + """ + if not settings["rulings"]: + print( + "uphold review could not look: no `[[review.ruling]]` control is " + "declared, so there is nothing to drive", + file=sys.stderr, + ) + return 2 + decide = report.get("decide") + if not isinstance(decide, dict): + print( + "uphold review could not look: the uphold binary reports no `decide`, so " + "which records this repository adopts cannot be read. Build or install " + "an uphold that reports it", + file=sys.stderr, + ) + return 2 + adopted = decide.get("adopt") + if not isinstance(adopted, list): + print( + "uphold review could not look: the uphold binary reports `decide` " + "without an `adopt` list, so which records this repository adopts " + "cannot be read", + file=sys.stderr, + ) + return 2 + missing = [record_id for record_id in adopted if record_id not in records] + if missing: + # The engine and this script carry different catalogs. Dropping the id + # would hand the ruler fewer records than the repository adopts. + print( + f"uphold review could not look: `[decide] adopt` names " + f"{', '.join(missing)}, which the catalog beside this script does not " + f"carry. Run the uphold_check.py of the same uphold as the binary", + file=sys.stderr, + ) + return 2 + if not adopted: + print( + "- the ruling controls have no adopted record to be answered from. " + "List the records under `[decide] adopt` in the policy", + file=sys.stderr, + ) + return 1 + seed = driving["seed"] + if seed is None: + seed = random.SystemRandom().randrange(2**32) + rng = random.Random(seed) + timeout = driving["timeout"] + auditor_command = driving["auditor"] + try: + document, lines = review_mod.ruling_document( + adopted, decide.get("order", []), records + ) + results = review_mod.drive_rulings( + settings["rulings"], + document, + lines, + driving["runs"], + rng, + lambda doc, question, shown: review_mod.ask_ruler( + driving["command"], doc, question, shown, timeout + ), + None + if auditor_command is None + else lambda doc, question, options, given: review_mod.audit( + auditor_command, doc, question, options, given, timeout + ), + driving["audit_every"], + ) + except review_mod.RulerFailed as error: + print(f"uphold review could not look: {error}", file=sys.stderr) + return 2 + agree = driving["agree"] + print( + review_mod.render_rulings( + results, driving["runs"], agree, seed, auditor_command is not None + ), + end="", + ) + return 0 if all(review_mod.ruling_holds(result, agree) for result in results) else 1 + + def main(argv: list[str]) -> int: records = load_records()