Skills › Compound Engineering › ce-resolve-pr-feedback › pending-feedback.py
pending-feedback.py
287 lines · skills/ce-resolve-pr-feedback/scripts/pending-feedback.py at 3449230
1"""Preserve a judged feedback batch. This helper never mutates GitHub."""
3from __future__ import annotations
5import argparse
6import json
7import os
8import re
9import subprocess
10import sys
11import tempfile
12from pathlib import Path
13from urllib.parse import urlsplit
16SCHEMA_VERSION = 1
17VERDICTS = {"fixed", "fixed-differently", "replied", "not-addressing", "declined", "needs-human"}
18KINDS = {"thread", "comment", "review"}
21def require(condition: bool, message: str) -> None:
22 if not condition:
23 raise ValueError(message)
26def string(value: object, name: str, pattern: str | None = None) -> str:
27 require(isinstance(value, str) and bool(value), f"{name} must be a nonempty string")
28 if pattern:
29 require(re.fullmatch(pattern, value) is not None, f"invalid {name}")
30 return value
33def positive_int(value: object, name: str) -> None:
34 require(type(value) is int and value > 0, f"{name} must be a positive integer")
37def object_value(value: object, name: str) -> dict:
38 require(isinstance(value, dict), f"{name} must be an object")
39 return value
42def array(value: object, name: str) -> list:
43 require(isinstance(value, list), f"{name} must be an array")
44 return value
47def decision(value: object) -> None:
48 context = object_value(value, "decision_context")
49 for key in ("quoted_feedback", "investigation", "decision_reason"):
50 string(context.get(key), key)
51 options = array(context.get("options"), "options")
52 require(bool(options), "decision options must not be empty")
53 for option in options:
54 option = object_value(option, "option")
55 string(option.get("option"), "option")
56 string(option.get("tradeoff"), "tradeoff")
57 require("recommendation" in context, "missing recommendation")
58 if context["recommendation"] is not None:
59 string(context["recommendation"], "recommendation")
62def progress(value: object) -> None:
63 value = object_value(value, "progress")
64 require(set(value) <= {"reply_id", "reply_url", "resolved", "applied"}, "unknown progress field")
65 if "reply_id" in value:
66 positive_int(value["reply_id"], "reply_id")
67 if "reply_url" in value:
68 string(value["reply_url"], "reply_url")
69 for key in ("resolved", "applied"):
70 if key in value:
71 require(type(value[key]) is bool, f"{key} must be boolean")
74def validate(record: object) -> dict:
75 record = object_value(record, "record")
76 require(type(record.get("schema_version")) is int and record["schema_version"] == SCHEMA_VERSION, "unsupported schema_version")
77 require(record.get("status") in {"pending", "completed"}, "invalid status")
78 pr = object_value(record.get("pr"), "pr")
79 host = string(pr.get("host"), "host", r"[A-Za-z0-9](?:[A-Za-z0-9.-]*[A-Za-z0-9])?(?::[0-9]+)?")
80 for key in ("base_repo", "head_repo"):
81 string(pr.get(key), key, r"[A-Za-z0-9_-][A-Za-z0-9_.-]*/(?!\.{1,2}$)[A-Za-z0-9_.-]+")
82 positive_int(pr.get("number"), "PR number")
83 expected_url = f"https://{host}/{pr['base_repo']}/pull/{pr['number']}"
84 require(pr.get("url") == expected_url, "PR URL does not match host/base/number")
85 ref = string(pr.get("head_ref"), "head_ref")
86 require(not re.search(r"[\x00-\x20\x7f~^:?*\[\\]", ref) and not any(token in ref for token in ("..", "@{")), "invalid head_ref")
87 require(not ref.startswith(("-", "/", ".")) and not ref.endswith(("/", ".", ".lock")) and "//" not in ref, "invalid head_ref")
88 require("fix_commit" in record, "missing fix_commit")
89 if record["fix_commit"] is not None:
90 string(record["fix_commit"], "fix_commit", r"[0-9a-f]{40}")
91 verification = object_value(record.get("verification"), "verification")
92 require(verification.get("outcome") in {"passed", "pre-existing-failure", "failed", "not-run"}, "invalid verification outcome")
93 for key in ("command", "details"):
94 require(isinstance(verification.get(key), str), f"verification.{key} must be a string")
95 seen = set()
96 for action in array(record.get("actions"), "actions"):
97 action = object_value(action, "action")
98 source = object_value(action.get("source"), "source")
99 require(source.get("kind") in KINDS, "invalid source kind")
100 source_id = string(source.get("id"), "source id", r"[A-Za-z0-9_+=:-]+")
101 key = (source["kind"], source_id)
102 require(key not in seen, "duplicate source identity")
103 seen.add(key)
104 url = urlsplit(string(source.get("url"), "source URL"))
105 require(url.scheme == "https" and url.netloc == host and url.path.rstrip("/") == f"/{pr['base_repo']}/pull/{pr['number']}" and not url.query, "source URL does not belong to PR")
106 string(source.get("body_sha256"), "body_sha256", r"[0-9a-f]{64}")
107 require(action.get("verdict") in VERDICTS, "invalid verdict")
108 string(action.get("reply_body"), "reply_body")
109 require(type(action.get("resolve")) is bool, "resolve must be boolean")
110 require(action["resolve"] == (source["kind"] == "thread" and action["verdict"] != "needs-human"), "resolve must match the source kind and verdict")
111 if source["kind"] == "thread":
112 positive_int(action.get("root_comment_id"), "root_comment_id")
113 string(action.get("thread_id"), "thread_id", r"[A-Za-z0-9_+=:-]+")
114 else:
115 require(action.get("root_comment_id") is None and action.get("thread_id") is None and not action["resolve"], "non-thread action cannot resolve")
116 if action["verdict"] == "needs-human":
117 require(not action["resolve"], "needs-human action cannot resolve")
118 decision(action.get("decision_context"))
119 else:
120 require(action.get("decision_context") is None, "decision_context belongs to needs-human")
121 require("invariant_key" in action, "missing invariant_key")
122 if action["invariant_key"] is not None:
123 string(action["invariant_key"], "invariant_key", r"[A-Za-z0-9._:-]{1,120}")
124 if "progress" in action:
125 progress(action["progress"])
126 if record["status"] == "completed":
127 observed = action.get("progress", {})
128 positive_int(observed.get("reply_id"), "completed action reply_id")
129 require(not action["resolve"] or observed.get("resolved") is True, "completed action must have its required resolution")
130 for tick in array(record.get("body_ticks"), "body_ticks"):
131 tick = object_value(tick, "body tick")
132 original = string(tick.get("original"), "original checklist bullet")
133 require(original.startswith("- [ ] ") and "\n" not in original and "\r" not in original, "invalid checklist bullet")
134 require(tick.get("checked") == original.replace("- [ ] ", "- [x] ", 1), "body tick must only check the saved bullet")
135 if "progress" in tick:
136 progress(tick["progress"])
137 if record["status"] == "completed":
138 require(tick.get("progress", {}).get("applied") is True, "completed body tick must be applied")
139 for residual in array(record.get("residuals"), "residuals"):
140 residual = object_value(residual, "residual")
141 require(residual.get("type") == "needs-human", "invalid residual type")
142 sources = array(residual.get("sources"), "residual sources")
143 require(bool(sources), "residual sources must not be empty")
144 for source in sources:
145 source = object_value(source, "residual source")
146 require(source.get("kind") in KINDS, "invalid residual source kind")
147 string(source.get("id"), "residual source id", r"[A-Za-z0-9_+=:-]+")
148 decision(residual.get("decision_context"))
149 thread_urls = array(residual.get("thread_urls"), "thread_urls")
150 require(not any(source["kind"] == "thread" for source in sources) or bool(thread_urls), "thread residual must carry thread_urls")
151 for url in thread_urls:
152 string(url, "thread URL")
153 return record
156def unique_keys(pairs: list) -> dict:
157 result = {}
158 for key, value in pairs:
159 require(key not in result, f"duplicate JSON key: {key}")
160 result[key] = value
161 return result
164def reject_constant(value: str) -> None:
165 raise ValueError(f"invalid JSON constant: {value}")
168def read_record(path: Path) -> tuple[dict, bytes]:
169 raw = path.read_bytes()
170 record = json.loads(raw.decode("utf-8"), object_pairs_hook=unique_keys,
171 parse_constant=reject_constant)
172 return validate(record), raw
175def prepared_content(record: dict) -> dict:
176 return {
177 "schema_version": record["schema_version"],
178 "pr": record["pr"], "fix_commit": record["fix_commit"], "verification": record["verification"],
179 "actions": [{key: value for key, value in action.items() if key != "progress"} for action in record["actions"]],
180 "body_ticks": [{key: value for key, value in tick.items() if key != "progress"} for tick in record["body_ticks"]],
181 "residuals": record["residuals"],
182 }
185def write_record(path: Path, raw: bytes, *, checkpoint: bool) -> None:
186 if checkpoint:
187 temporary = None
188 try:
189 with tempfile.NamedTemporaryFile(dir=path.parent, prefix=".pending-feedback-", delete=False) as temp:
190 temporary = Path(temp.name)
191 temp.write(raw)
192 temp.flush()
193 os.fsync(temp.fileno())
194 os.replace(temporary, path)
195 finally:
196 if temporary is not None:
197 temporary.unlink(missing_ok=True)
198 else:
199 descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
200 with os.fdopen(descriptor, "wb") as target:
201 target.write(raw)
202 target.flush()
203 os.fsync(target.fileno())
204 read_record(path)
207def github_json(host: str, endpoint: str) -> dict:
208 result = subprocess.run(
209 ["gh", "api", "--hostname", host, "--method", "GET", endpoint],
210 capture_output=True, text=True, check=False,
211 )
212 require(result.returncode == 0, result.stderr.strip() or f"GitHub GET failed: {endpoint}")
213 return object_value(json.loads(result.stdout), "GitHub response")
216def inspect_publication(record: dict) -> dict:
217 proof = {"verified": False, "reason": "", "head_sha": None,
218 "head_repo": None, "head_ref": None, "comparison_status": None}
219 pr = record["pr"]
220 try:
221 fresh = github_json(pr["host"], f"repos/{pr['base_repo']}/pulls/{pr['number']}")
222 require(fresh.get("number") == pr["number"] and fresh.get("html_url") == pr["url"],
223 "fresh PR host/number/URL differs from the saved PR")
224 base = object_value(fresh.get("base"), "fresh base")
225 require(object_value(base.get("repo"), "fresh base repository").get("full_name") == pr["base_repo"],
226 "fresh PR base repository differs from the saved PR")
227 head = object_value(fresh.get("head"), "fresh head")
228 proof["head_repo"] = object_value(head.get("repo"), "fresh head repository").get("full_name")
229 proof["head_ref"] = head.get("ref")
230 proof["head_sha"] = string(head.get("sha"), "fresh head SHA", r"[0-9a-f]{40}")
231 require(proof["head_repo"] == pr["head_repo"] and proof["head_ref"] == pr["head_ref"],
232 "fresh PR head repository/ref differs from the saved PR")
233 commit = record["fix_commit"]
234 if commit is None:
235 proof.update(verified=True, reason="saved batch created no fix commit")
236 return proof
237 comparison = github_json(pr["host"], f"repos/{proof['head_repo']}/compare/{commit}...{proof['head_sha']}")
238 proof["comparison_status"] = comparison.get("status")
239 merge_base = object_value(comparison.get("merge_base_commit"), "comparison merge base")
240 require(proof["comparison_status"] in {"ahead", "identical"} and merge_base.get("sha") == commit,
241 "recorded fix commit is not positively reachable from the fresh PR head")
242 proof.update(verified=True, reason="recorded fix commit is reachable from the fresh PR head")
243 except (OSError, ValueError, TypeError) as error:
244 proof["reason"] = str(error)
245 return proof
248def main() -> None:
249 parser = argparse.ArgumentParser(description=__doc__)
250 subcommands = parser.add_subparsers(dest="command", required=True)
251 preflight = subcommands.add_parser("preflight")
252 preflight.add_argument("--path")
253 for command in ("create", "validate", "checkpoint", "inspect-publication"):
254 command_parser = subcommands.add_parser(command)
255 command_parser.add_argument("--path", required=True)
256 if command in {"create", "checkpoint"}:
257 command_parser.add_argument("--input", required=True)
258 args = parser.parse_args()
259 if args.command == "preflight":
260 path = Path(args.path).absolute() if args.path else Path(tempfile.mkdtemp(prefix="ce-pending-feedback-")) / "pending.json"
261 require(not os.path.lexists(path), "handoff destination already exists")
262 require(path.parent.is_dir(), "handoff parent directory does not exist")
263 with tempfile.TemporaryFile(dir=path.parent):
264 pass
265 print(json.dumps({"handoff": str(path)}))
266 return
267 path = Path(args.path).absolute()
268 if args.command in {"validate", "inspect-publication"}:
269 record, _ = read_record(path)
270 else:
271 record, raw = read_record(Path(args.input))
272 if args.command == "checkpoint":
273 previous, _ = read_record(path)
274 require(prepared_content(previous) == prepared_content(record), "checkpoint cannot replace the prepared batch")
275 write_record(path, raw, checkpoint=args.command == "checkpoint")
276 output = {"handoff": str(path), "record": record}
277 if args.command == "inspect-publication":
278 output["publication"] = inspect_publication(record)
279 print(json.dumps(output, ensure_ascii=False))
282if __name__ == "__main__":
283 try:
284 main()
285 except (OSError, ValueError, TypeError) as error:
286 print(f"pending feedback: {error}", file=sys.stderr)
287 sys.exit(1)