diff --git a/docs/plans/board-successor-procedure.md b/docs/plans/board-successor-procedure.md new file mode 100644 index 000000000..a8e08cc36 --- /dev/null +++ b/docs/plans/board-successor-procedure.md @@ -0,0 +1,60 @@ +# Board successor procedure (LANES-20261022 Lane 4) + +Status: proposal, draft PR. Lane 4 of +[LANES-20261022](https://github.com/d6g8k5htny-coder/main/issues/475#issuecomment-6346276856) +(correction [6346280672](https://github.com/d6g8k5htny-coder/main/issues/475#issuecomment-6346280672)), +pickup [475/6346309051](https://github.com/d6g8k5htny-coder/main/issues/475#issuecomment-6346309051). +This is an engineering procedure with no scientific effect. Credit 0, OBL OPEN. + +## When a board or session is succeeded + +A board is succeeded when it fills up (#419 stopped accepting comments at 2,500) +or is retired. A session is succeeded when an agent hands its work to another. + +1. **Post the successor pointer on the old board** and the predecessor pointer as + the first comment, or in the body, of the new one. Each names the other by URL. +2. **Carry open items forward explicitly.** List every unresolved item from the + predecessor by a stable id and its source URL. An item leaves the list only + when it is marked resolved with the URL of the comment that resolved it. + Nothing is silently dropped. +3. **Ownership does not move with the board.** As `AGENTS.md` says, moving the + board does not transfer, renew or release any claim. An owner is recorded only + with the URL of its own PICKUP or assignment comment. Ownership and + acknowledgement are never inferred from silence, mentions or summaries. +4. **Update `AGENTS.md`** only through a normal reviewed PR. + +Board rules carry over unchanged (PICKUP before reading, RESULT/RELEASE, finite leases, nonauthor review, merge commits only, and no merge without the integrator rules), and a predecessor or successor link `url` must be an http(s) URL. + +## Records and the checker + +Records follow node-status interface v0 (Lane 2, `records/node_status/.json`) +with three optional handoff fields: `predecessor` and `successor`, each +`{node_id, url}` or null, and `carried_forward`, a list of +`{id, text, source_url, status: open|resolved, resolution_url}`. +Those three shapes are unchanged. A carried item's `text` is required and non-empty. +Its `source_url` is any GitHub issue or comment URL. `resolution_url`, owner +`evidence_url`, and `released_url` share one comment-URL pattern: +`https://github.com/d6g8k5htny-coder/(main|Math-)/(issues|pull)/[1-9]\d*#(issuecomment-[1-9]\d*|pullrequestreview-[1-9]\d*)`. +An owner with `released_url` is not a current owner. It still needs a non-empty id and an `evidence_url` matching that pattern, and `released_url` must match it too. Owner keys are only `id`, `evidence_url`, and `released_url`. `refresh_error`, when present, is a string or null. + +`python -B -S tools/check_board_links.py records/node_status` reads the records +and fails, with exit status 1, when: + +- a predecessor or successor link is not reciprocated, points to a missing record, or its `url` is not an http(s) URL; +- an owner lacks an id or a comment URL as its evidence, has a key other than `id`, `evidence_url`, and `released_url`, or a `released_url` does not match that pattern; +- `carried_forward` is not a list, an entry is not an object, an id is not a non-empty string or is duplicated, a key is outside `id`, `text`, `source_url`, `status`, and `resolution_url`, `text` is missing or empty, `source_url` is not a GitHub issue or comment URL, or a resolution URL does not match that pattern; +- `refresh_error` is present and is neither a string nor null; +- an open predecessor item is missing from the successor's `carried_forward`, or an + item is marked resolved without a resolution URL. + +It checks link structure only. It does not fetch or judge the linked comments. +Tests: `python -B -m pytest tests/test_board_links.py`. The module also runs under +`python -B -S -m unittest tests.test_board_links`. + +## Boundaries + +- Not a duplicate of #489 (offline board-capacity probe). Capacity output from #489 + is consumed only after #489 merges, and is not used here. +- Not a duplicate of #487 (H02 active-board routing test). +- No records for real boards are added in this PR. Writing them needs owners' + actual PICKUP URLs, which are not inferred here. diff --git a/tests/test_board_links.py b/tests/test_board_links.py new file mode 100644 index 000000000..49bb32dd6 --- /dev/null +++ b/tests/test_board_links.py @@ -0,0 +1,331 @@ +"""Synthetic fixtures for tools/check_board_links.py (LANES-20261022 Lane 4). + +The records below are invented test data. They make no ownership, +acknowledgement or custody claim about any real board or agent. +""" + +import contextlib +import copy +import importlib.util +import io +import json +from pathlib import Path +import subprocess +import sys +import tempfile +import unittest + +ROOT = Path(__file__).resolve().parents[1] +TOOL = ROOT / "tools/check_board_links.py" +_spec = importlib.util.spec_from_file_location("check_board_links", TOOL) +cbl = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(cbl) + +PICKUP = "https://github.com/d6g8k5htny-coder/main/issues/2#issuecomment-200" +RESOLVED = "https://github.com/d6g8k5htny-coder/main/issues/2#issuecomment-201" +REVIEW = "https://github.com/d6g8k5htny-coder/Math-/pull/3#pullrequestreview-300" + +BASE = { + "board-old": { + "schema_version": 0, "node_id": "board-old", "owner": None, + "predecessor": None, + "successor": {"node_id": "board-new", "url": "https://github.com/example/repo/issues/2"}, + "carried_forward": [ + {"id": "item-a", "text": "open review", "source_url": PICKUP, "status": "open"}, + {"id": "item-b", "text": "done", "source_url": PICKUP, "status": "resolved", + "resolution_url": RESOLVED}, + ], + "scientific_effect": "NONE", + }, + "board-new": { + "schema_version": 0, "node_id": "board-new", + "owner": {"id": "agent-x", "evidence_url": PICKUP}, + "predecessor": {"node_id": "board-old", "url": "https://github.com/example/repo/issues/1"}, + "successor": None, + "carried_forward": [ + {"id": "item-a", "text": "open review", "source_url": PICKUP, "status": "open"}, + ], + "scientific_effect": "NONE", + }, +} + + +class BoardLinks(unittest.TestCase): + def run_check(self, records): + with tempfile.TemporaryDirectory() as d: + for node, rec in records.items(): + Path(d, f"{node}.json").write_text(json.dumps(rec), encoding="utf-8") + return cbl.check(d) + + def records(self): + return copy.deepcopy(BASE) + + def assert_rejected(self, records, needle): + with tempfile.TemporaryDirectory() as d: + for node, rec in records.items(): + Path(d, f"{node}.json").write_text(json.dumps(rec), encoding="utf-8") + errs = cbl.check(d) + self.assertTrue(any(needle in e for e in errs), errs) + with contextlib.redirect_stdout(io.StringIO()): + code = cbl.main(["check_board_links.py", d]) + self.assertEqual(code, 1, errs) + + def test_consistent_chain_passes(self): + self.assertEqual(self.run_check(self.records()), []) + + def test_missing_reciprocal_link_fails(self): + r = self.records() + r["board-new"]["predecessor"] = None + errs = self.run_check(r) + self.assertTrue(any("does not link back" in e for e in errs), errs) + + def test_link_to_absent_record_fails(self): + r = self.records() + del r["board-new"] + errs = self.run_check(r) + self.assertTrue(any("has no record" in e for e in errs), errs) + + def test_owner_without_pickup_url_fails(self): + for bad in ({"id": "agent-x"}, {"id": "agent-x", "evidence_url": ""}, + {"id": "agent-x", "evidence_url": "https://github.com/example/repo/issues/2"}): + r = self.records() + r["board-new"]["owner"] = bad + errs = self.run_check(r) + self.assertTrue(any("no PICKUP" in e for e in errs), (bad, errs)) + + def test_null_owner_is_not_inferred(self): + r = self.records() + r["board-new"]["owner"] = None + self.assertEqual(self.run_check(r), []) + + def test_open_item_dropped_by_successor_fails(self): + r = self.records() + r["board-new"]["carried_forward"] = [] + errs = self.run_check(r) + self.assertTrue(any("'item-a' not carried forward" in e for e in errs), errs) + + def test_carried_open_items_are_preserved(self): + r = self.records() + self.assertEqual(self.run_check(r), []) + new = r["board-new"]["carried_forward"] + self.assertEqual([(i["id"], i["status"]) for i in new], [("item-a", "open")]) + self.assertEqual(new[0]["text"], r["board-old"]["carried_forward"][0]["text"]) + + def test_resolved_without_resolution_url_fails(self): + r = self.records() + r["board-new"]["carried_forward"][0] = {"id": "item-a", "status": "resolved"} + errs = self.run_check(r) + self.assertTrue(any("without a resolution_url" in e for e in errs), errs) + + def test_carried_item_without_source_url_fails(self): + for mutate in ( + lambda item: item.pop("source_url"), + lambda item: item.__setitem__("source_url", ""), + lambda item: item.__setitem__("source_url", "https://example.com/notes"), + ): + r = self.records() + mutate(r["board-new"]["carried_forward"][0]) + self.assert_rejected(r, "source_url") + + def test_comment_url_pattern(self): + rejected = ( + "https://github.com/example/main/issues/2#issuecomment-200", + "https://github.com/d6g8k5htny-coder/other/issues/2#issuecomment-200", + "https://github.com/d6g8k5htny-coder/main/issues/2", + ) + for url in (PICKUP, REVIEW): + r = self.records() + r["board-new"]["owner"]["evidence_url"] = url + for node in r: + for item in r[node]["carried_forward"]: + item["source_url"] = url + if item.get("status") == "resolved": + item["resolution_url"] = url + self.assertEqual(self.run_check(r), [], url) + for url in rejected: + r = self.records() + r["board-new"]["owner"]["evidence_url"] = url + self.assert_rejected(r, "no PICKUP") + r = self.records() + r["board-new"]["carried_forward"][0]["source_url"] = url + self.assertEqual(self.run_check(r), [], url) + r = self.records() + r["board-old"]["carried_forward"][1]["resolution_url"] = url + self.assert_rejected(r, "resolution_url") + + def test_zero_leading_comment_ids_fail(self): + rejected = ( + "https://github.com/d6g8k5htny-coder/main/issues/0#issuecomment-200", + "https://github.com/d6g8k5htny-coder/main/pull/02#issuecomment-200", + "https://github.com/d6g8k5htny-coder/main/issues/2#issuecomment-0", + "https://github.com/d6g8k5htny-coder/Math-/pull/3#issuecomment-0200", + "https://github.com/d6g8k5htny-coder/main/pull/3#pullrequestreview-0", + "https://github.com/d6g8k5htny-coder/Math-/issues/4#pullrequestreview-01", + ) + for url in rejected: + r = self.records() + r["board-new"]["owner"]["evidence_url"] = url + self.assert_rejected(r, "no PICKUP") + r = self.records() + r["board-new"]["carried_forward"][0]["source_url"] = url + self.assertEqual(self.run_check(r), [], url) + r = self.records() + r["board-old"]["carried_forward"][1]["resolution_url"] = url + self.assert_rejected(r, "resolution_url") + + def test_trailing_newline_in_url_fails(self): + r = self.records() + r["board-new"]["owner"]["evidence_url"] = PICKUP + "\n" + self.assert_rejected(r, "no PICKUP") + r = self.records() + r["board-new"]["carried_forward"][0]["source_url"] = PICKUP + "\n" + self.assert_rejected(r, "source_url") + r = self.records() + r["board-old"]["carried_forward"][1]["resolution_url"] = RESOLVED + "\n" + self.assert_rejected(r, "resolution_url") + + def test_carried_item_without_text_fails(self): + for mutate in ( + lambda item: item.pop("text"), + lambda item: item.__setitem__("text", ""), + ): + r = self.records() + mutate(r["board-new"]["carried_forward"][0]) + self.assert_rejected(r, "text") + + def test_refresh_error_must_be_string_or_null(self): + r = self.records() + r["board-new"]["refresh_error"] = 5 + self.assert_rejected(r, "refresh_error must be a string or null") + for ok in (None, "unread"): + r = self.records() + r["board-new"]["refresh_error"] = ok + self.assertEqual(self.run_check(r), [], ok) + + def test_released_owner_missing_id_fails(self): + r = self.records() + r["board-new"]["owner"] = {"evidence_url": PICKUP, "released_url": REVIEW} + self.assert_rejected(r, "without an id") + + def test_released_owner_missing_evidence_url_fails(self): + r = self.records() + r["board-new"]["owner"] = {"id": "agent-old", "released_url": REVIEW} + self.assert_rejected(r, "no PICKUP") + + def test_owner_extra_key_fails(self): + r = self.records() + r["board-new"]["owner"] = { + "id": "agent-x", "evidence_url": PICKUP, "note": "extra", + } + self.assert_rejected(r, "only id, evidence_url, and released_url") + + def test_allowed_owner_keys_pass(self): + for owner in ( + {"id": "agent-x", "evidence_url": PICKUP}, + {"id": "agent-old", "evidence_url": PICKUP, "released_url": REVIEW}, + ): + r = self.records() + r["board-new"]["owner"] = owner + self.assertEqual(self.run_check(r), [], owner) + + def test_released_owner_with_evidence_passes(self): + r = self.records() + r["board-new"]["owner"] = { + "id": "agent-old", "evidence_url": PICKUP, "released_url": REVIEW, + } + self.assertEqual(self.run_check(r), []) + + def test_malformed_released_url_fails(self): + for bad in ( + "https://github.com/example/main/issues/2#issuecomment-200", + "https://github.com/d6g8k5htny-coder/other/pull/3#pullrequestreview-300", + "https://github.com/d6g8k5htny-coder/main/issues/2", + "", + ): + r = self.records() + r["board-new"]["owner"]["released_url"] = bad + self.assert_rejected(r, "released_url") + + def test_carried_item_not_dict_fails(self): + r = self.records() + r["board-old"]["carried_forward"].append("stray") + self.assert_rejected(r, "is not a dict") + + def test_carried_forward_dict_instead_of_list_fails(self): + r = self.records() + r["board-old"]["carried_forward"] = { + "id": "item-a", "text": "open review", "source_url": PICKUP, "status": "open", + } + self.assert_rejected(r, "is not a list") + + def test_duplicate_item_id_fails(self): + r = self.records() + r["board-old"]["carried_forward"].append( + {"id": "item-a", "text": "different text", "source_url": PICKUP, "status": "open"}) + self.assert_rejected(r, "duplicate carried_forward id") + + def test_link_url_not_http_fails(self): + for node, field in (("board-old", "successor"), ("board-new", "predecessor")): + for bad in ("not-a-url", "ftp://example.com/board", ""): + r = self.records() + r[node][field]["url"] = bad + self.assert_rejected(r, "not an http(s) URL") + r = self.records() + r["board-old"]["successor"]["url"] = "http://example.com/board" + r["board-new"]["predecessor"]["url"] = "http://example.com/old" + self.assertEqual(self.run_check(r), []) + + def test_malformed_link_fails(self): + r = self.records() + r["board-old"]["successor"] = "board-new" + self.assert_rejected(r, "malformed") + + def test_owner_without_id_fails(self): + r = self.records() + r["board-new"]["owner"] = {"evidence_url": PICKUP} + self.assert_rejected(r, "without an id") + + def test_carried_item_id_must_be_string(self): + r = self.records() + r["board-new"]["carried_forward"].append( + {"id": 5, "text": "extra", "source_url": PICKUP, "status": "open"}) + self.assert_rejected(r, "without an id") + + def test_carried_item_extra_key_fails(self): + r = self.records() + r["board-new"]["carried_forward"][0]["extra"] = "no" + self.assert_rejected(r, "only id, text, source_url, status, and resolution_url") + + def test_carried_item_without_id_fails(self): + r = self.records() + r["board-new"]["carried_forward"].append( + {"text": "extra", "source_url": PICKUP, "status": "open"}) + self.assert_rejected(r, "without an id") + + def test_bad_item_status_fails(self): + r = self.records() + r["board-new"]["carried_forward"].append( + {"id": "item-c", "text": "nope", "source_url": PICKUP, "status": "closed"}) + self.assert_rejected(r, "has status") + + def test_duplicate_node_id_fails(self): + r = self.records() + r["board-old-again"] = copy.deepcopy(r["board-old"]) + self.assert_rejected(r, "duplicate node_id") + + def test_cli_exit_status_and_no_writes(self): + with tempfile.TemporaryDirectory() as d: + r = self.records() + r["board-old"]["successor"] = None + for node, rec in r.items(): + Path(d, f"{node}.json").write_text(json.dumps(rec), encoding="utf-8") + before = sorted(p.name for p in Path(d).iterdir()) + out = subprocess.run([sys.executable, "-B", "-S", str(TOOL), d], + capture_output=True, text=True) + self.assertEqual(out.returncode, 1, out.stdout + out.stderr) + self.assertFalse(json.loads(out.stdout)["ok"]) + self.assertEqual(sorted(p.name for p in Path(d).iterdir()), before) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_board_links_raw_json.py b/tests/test_board_links_raw_json.py new file mode 100644 index 000000000..5621fe616 --- /dev/null +++ b/tests/test_board_links_raw_json.py @@ -0,0 +1,110 @@ +"""Raw-file regression proposal: duplicate JSON names must not erase handoffs. + +All records are synthetic. Tests run the actual checker CLI in both modes; +no network calls and no real ownership/custody assertions are made. +""" +import hashlib +import json +from pathlib import Path +import subprocess +import sys +import tempfile +import unittest + +ROOT = Path(__file__).resolve().parents[1] +TOOL = ROOT / "tools/check_board_links.py" +PICKUP = "https://github.com/d6g8k5htny-coder/main/issues/2#issuecomment-200" +RELEASE = "https://github.com/d6g8k5htny-coder/main/issues/2#pullrequestreview-201" +ITEM = {"id": "hold-1", "text": "Unresolved prerequisite", + "source_url": "https://github.com/example/repo/issues/1", "status": "open"} + + +def raw_object(pairs): + # Intentionally allow duplicate names. json.dumps(dict(...)) would erase them. + return "{" + ",".join(json.dumps(k) + ":" + json.dumps(v) for k, v in pairs) + "}" + + +class RawBoardJSON(unittest.TestCase): + def run_files(self, texts): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + for name, text in texts.items(): + (root / name).write_text(text, encoding="utf-8") + before = {p.name: hashlib.sha256(p.read_bytes()).hexdigest() for p in root.iterdir()} + outputs = [] + for mode in ([], ["-O"]): + command = [sys.executable, "-B", "-S", *mode, str(TOOL), directory] + result = subprocess.run(command, capture_output=True, text=True, timeout=10) + self.assertEqual(result.stderr, "", command) + payload = json.loads(result.stdout) + self.assertEqual(payload["scientific_effect"], "NONE") + outputs.append((result.returncode, result.stdout, payload)) + self.assertEqual(before, {p.name: hashlib.sha256(p.read_bytes()).hexdigest() + for p in root.iterdir()}) + self.assertEqual(outputs[0], outputs[1], "normal/optimized checker behavior differs") + return outputs[0][0], outputs[0][2] + + def assert_duplicate(self, text, key): + code, out = self.run_files({"record.json": text}) + self.assertEqual(code, 1, out) + self.assertFalse(out["ok"], out) + self.assertTrue(any("duplicate JSON key: " + key in e for e in out["errors"]), out) + + def test_dropped_open_item_cannot_be_hidden(self): + old = {"node_id": "old", "owner": None, + "successor": {"node_id": "new", "url": "https://example.com/new"}, + "carried_forward": [ITEM]} + new = {"node_id": "new", "owner": None, + "predecessor": {"node_id": "old", "url": "https://example.com/old"}, + "carried_forward": []} + code, out = self.run_files({"a.json": json.dumps(old), "b.json": json.dumps(new)}) + self.assertEqual(code, 1) + self.assertTrue(any("not carried forward" in e for e in out["errors"]), out) + raw = json.dumps(old)[:-1] + ', "carried_forward": []}' + code, out = self.run_files({"a.json": raw, "b.json": json.dumps(new)}) + self.assertEqual(code, 1, out) + self.assertTrue(any("duplicate JSON key: carried_forward" in e for e in out["errors"]), out) + + def test_owner_cannot_be_hidden_by_later_null(self): + self.assert_duplicate(raw_object([("node_id", "n"), ("owner", {"id": "unbacked"}), + ("owner", None)]), "owner") + + def test_nested_owner_evidence_cannot_be_replaced(self): + owner = raw_object([("id", "agent"), ("evidence_url", "not-a-comment"), + ("evidence_url", PICKUP)]) + self.assert_duplicate('{"node_id":"n","owner":' + owner + '}', "evidence_url") + + def test_nested_item_status_cannot_be_replaced(self): + item = raw_object([*ITEM.items(), ("status", "open")]) + self.assert_duplicate('{"node_id":"n","carried_forward":[' + item + ']}', "status") + + def test_escaped_names_are_compared_after_decoding(self): + self.assert_duplicate('{"node_id":"n","owner":{"id":"unbacked"},"own\\u0065r":null}', "owner") + + def test_equal_valued_duplicate_is_still_ambiguous(self): + self.assert_duplicate('{"node_id":"n","node_id":"n","owner":null}', "node_id") + + def test_array_nested_names_are_checked(self): + # A valid resolved item must retain exactly one resolution_url key. + item = raw_object([*dict(ITEM, status="resolved").items(), + ("resolution_url", "invalid"), ("resolution_url", RELEASE)]) + self.assert_duplicate('{"node_id":"n","carried_forward":[' + item + ']}', "resolution_url") + + def test_separate_objects_can_reuse_the_same_names(self): + a = {"node_id": "a", "owner": {"id": "agent", "evidence_url": PICKUP}, + "successor": {"node_id": "b", "url": "https://example.com/b"}, "carried_forward": [ITEM]} + b = {"node_id": "b", "owner": None, + "predecessor": {"node_id": "a", "url": "https://example.com/a"}, "carried_forward": [ITEM]} + code, out = self.run_files({"a.json": json.dumps(a), "b.json": json.dumps(b)}) + self.assertEqual((code, out), (0, {"ok": True, "errors": [], "scientific_effect": "NONE"})) + + def test_other_file_errors_are_not_silenced(self): + code, out = self.run_files({"a.json": '{"node_id":"a","owner":null,"owner":null}', + "b.json": '{"node_id":"b","owner":{"id":"unbacked"}}'}) + self.assertEqual(code, 1, out) + self.assertTrue(any("duplicate JSON key: owner" in e for e in out["errors"]), out) + self.assertTrue(any("no PICKUP" in e for e in out["errors"]), out) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/check_board_links.py b/tools/check_board_links.py new file mode 100644 index 000000000..3effe65a2 --- /dev/null +++ b/tools/check_board_links.py @@ -0,0 +1,234 @@ +#!/usr/bin/env python3 +"""Read-only check of board and session handoff links (LANES-20261022 Lane 4). + +Input: a directory of JSON records shaped like node-status interface v0 +(records/node_status/.json) with three optional handoff fields: + + predecessor: {"node_id": str, "url": http(s) URL} | null + successor: {"node_id": str, "url": http(s) URL} | null + carried_forward: [{"id": str, "text": str, "source_url": str, + "status": "open" | "resolved", + "resolution_url": str (required when resolved)}] + +Owner evidence, released_url, and resolution_url use COMMENT_URL below. +carried_forward source_url uses the carried-item pattern from interface v0: +any GitHub issue or comment URL. + +Checks (each failure is reported; the exit status is 1 if any): + 1. reciprocity: A.successor -> B requires B.predecessor -> A, and the reverse; + a link to a node with no record is a failure, not a skip; a link url that + is not http(s) is a failure; + 2. ownership: a non-null owner needs an id and an evidence_url matching + COMMENT_URL (the PICKUP or assignment). An owner with released_url is not + a current owner, and still needs that id and evidence_url; released_url + itself must match COMMENT_URL. Owner keys are closed to id, evidence_url, + and released_url. Ownership is never inferred, and moving a board does + not move it; + 3. carry-forward: carried_forward must be a list of objects with unique + string ids, non-empty text, and a source_url. Item keys are only id, + text, source_url, status, and resolution_url. Every predecessor item + that is open must + appear, by id, in the successor's carried_forward, either still open or + resolved with a resolution_url. Items may not be silently dropped. + 4. refresh_error, when present, is a string or null. + +The tool reads files only. It makes no network call and writes nothing. +It checks link structure, not the truth of any linked comment, and has no +scientific effect. +""" + +import json +import re +import sys +from pathlib import Path +from urllib.parse import urlparse + +COMMENT_URL = re.compile( + r"^https://github\.com/d6g8k5htny-coder/(main|Math-)/(issues|pull)/[1-9]\d*#(issuecomment-[1-9]\d*|pullrequestreview-[1-9]\d*)$" +) +CARRIED_SOURCE_URL = re.compile( + r"^https://github\.com/[\w.-]+/[\w.-]+/(?:issues|pull)/\d+" + r"(?:#(?:issuecomment|pullrequestreview)-\d+)?$") + + +def _http_url(url): + if not isinstance(url, str) or not url or any(ch.isspace() for ch in url): + return False + parsed = urlparse(url) + return parsed.scheme.lower() in ("http", "https") and bool(parsed.netloc) + + +def _comment_url(url): + return isinstance(url, str) and COMMENT_URL.fullmatch(url) is not None + + +def _carried_source_url(url): + return isinstance(url, str) and CARRIED_SOURCE_URL.fullmatch(url) is not None + + +def _unique_pairs(pairs): + """Reject duplicate object names before handoff checks, including nested ones.""" + result = {} + for key, value in pairs: + if key in result: + raise ValueError(f"duplicate JSON key: {key}") + result[key] = value + return result + + +def load_records(directory): + records, errors = {}, [] + for path in sorted(Path(directory).glob("*.json")): + try: + rec = json.loads(path.read_text(encoding="utf-8"), object_pairs_hook=_unique_pairs) + except (OSError, ValueError) as exc: + errors.append(f"{path.name}: unreadable JSON ({exc})") + continue + node = rec.get("node_id") if isinstance(rec, dict) else None + if not isinstance(node, str) or not node: + errors.append(f"{path.name}: missing node_id") + continue + if node in records: + errors.append(f"{path.name}: duplicate node_id {node!r}") + continue + records[node] = rec + return records, errors + + +def _link(rec, field): + link = rec.get(field) + if link is None: + return None + if not isinstance(link, dict) or not isinstance(link.get("node_id"), str): + return "" + return link["node_id"] + + +def check_reciprocal(records): + errors = [] + for node, rec in sorted(records.items()): + for field, back in (("successor", "predecessor"), ("predecessor", "successor")): + link = rec.get(field) + target = _link(rec, field) + if target is None: + continue + if target == "": + errors.append(f"{node}: malformed {field} link") + elif target not in records: + errors.append(f"{node}: {field} {target!r} has no record") + elif _link(records[target], back) != node: + errors.append(f"{node}: {field} {target!r} does not link back as {back}") + if isinstance(link, dict) and not _http_url(link.get("url")): + errors.append(f"{node}: {field} url is not an http(s) URL") + return errors + + +def check_ownership(records): + errors = [] + for node, rec in sorted(records.items()): + owner = rec.get("owner") + if owner is None: + continue + if not isinstance(owner, dict): + errors.append(f"{node}: owner without an id") + continue + extra = sorted(set(owner) - {"id", "evidence_url", "released_url"}) + if extra: + errors.append(f"{node}: owner allows only id, evidence_url, and released_url") + evidence = owner.get("evidence_url") + if "evidence_url" not in owner or not isinstance(evidence, str) or evidence == "": + errors.append(f"{node}: owner {owner.get('id')!r} has no PICKUP/assignment comment url") + continue + if not isinstance(owner.get("id"), str) or owner.get("id") == "": + errors.append(f"{node}: owner without an id") + if not _comment_url(evidence): + errors.append(f"{node}: owner {owner.get('id')!r} has no PICKUP/assignment comment url") + if "released_url" in owner and not _comment_url(owner.get("released_url")): + errors.append(f"{node}: owner released_url is not a comment URL") + return errors + + +def _carried(node, rec, errors): + raw = rec.get("carried_forward") + if raw is None: + raw = [] + if not isinstance(raw, list): + errors.append(f"{node}: carried_forward is not a list") + return {} + by_id = {} + for item in raw: + if not isinstance(item, dict): + errors.append(f"{node}: carried_forward item is not a dict") + continue + extra = sorted(set(item) - {"id", "text", "source_url", "status", "resolution_url"}) + if extra: + errors.append( + f"{node}: carried_forward item allows only id, text, source_url, status, and resolution_url") + item_id = item.get("id") + if not isinstance(item_id, str) or item_id == "": + errors.append(f"{node}: carried_forward item without an id") + elif item_id in by_id: + errors.append(f"{node}: duplicate carried_forward id {item_id!r}") + else: + by_id[item_id] = item + text = item.get("text") + if not isinstance(text, str) or text == "": + errors.append(f"{node}: item {item_id!r} has no text") + source = item.get("source_url") + if not isinstance(source, str) or not source: + errors.append(f"{node}: item {item_id!r} has no source_url") + elif not _carried_source_url(source): + errors.append(f"{node}: item {item_id!r} source_url is not a GitHub issue or comment URL") + if not item_id: + continue + if item.get("status") == "resolved" and not _comment_url(item.get("resolution_url")): + errors.append(f"{node}: item {item_id!r} resolved without a resolution_url") + elif item.get("status") not in ("open", "resolved"): + errors.append(f"{node}: item {item_id!r} has status {item.get('status')!r}") + return by_id + + +def check_carried_forward(records): + errors = [] + parsed = {} + for node, rec in sorted(records.items()): + parsed[node] = _carried(node, rec, errors) + for node, rec in sorted(records.items()): + succ = _link(rec, "successor") + if not succ or succ not in records: + continue + carried = parsed[succ] + for item_id, item in parsed[node].items(): + if item_id and item.get("status") == "open" and item_id not in carried: + errors.append(f"{node}: open item {item_id!r} not carried forward to {succ!r}") + return errors + + +def check_refresh_error(records): + errors = [] + for node, rec in sorted(records.items()): + if "refresh_error" in rec and rec["refresh_error"] is not None and not isinstance(rec["refresh_error"], str): + errors.append(f"{node}: refresh_error must be a string or null") + return errors + + +def check(directory): + records, errors = load_records(directory) + return (errors + check_reciprocal(records) + check_ownership(records) + + check_carried_forward(records) + check_refresh_error(records)) + + +def main(argv): + if len(argv) != 2: + print("usage: check_board_links.py RECORDS_DIR", file=sys.stderr) + return 2 + if not Path(argv[1]).is_dir(): + print(f"not a directory: {argv[1]}", file=sys.stderr) + return 2 + errors = check(argv[1]) + print(json.dumps({"ok": not errors, "errors": errors, "scientific_effect": "NONE"}, indent=1)) + return 1 if errors else 0 + + +if __name__ == "__main__": + sys.exit(main(sys.argv))