diff --git a/docs/node-maintenance.md b/docs/node-maintenance.md
index 5b307f6..4d1fd31 100644
--- a/docs/node-maintenance.md
+++ b/docs/node-maintenance.md
@@ -300,6 +300,24 @@ recovery before the next attempt. Production repair and report integration
remain later delivery slices. Nothing merges automatically.
See the [service reference](node-maintenance-operator.md) for current limits.
+### Reading repair choices
+
+๐ก **Needs review** means a decision or explicit pin needs attention.
+๐ด **Blocked** means the available evidence does not support a compatible fix.
+A check mark identifies a proposed applied fix; the report's validation section
+states whether the project checks passed. Status icons do not indicate severity;
+the severity column reports that separately.
+
+An eligible pin has a folded assessment immediately below its package row. It
+shows the published candidate, checked parent constraints and advisory links.
+**Likely feasible** means that an automatic attempt appears possible. It does
+not mean repair validation passed or that the dependency was upgraded.
+
+The report component can produce the exact selection block bound to a known PR
+head and base. Unbound audit reports do not offer an active repair command. The
+published caller integration and actual dependency repair remain unreleased;
+the harmless pilot's success does not change the package table to upgraded.
+
## Toolchain selection
The basic setup follows project declarations. Explicit or inherited constraints
diff --git a/node/runtime/repair_choices.py b/node/runtime/repair_choices.py
new file mode 100644
index 0000000..d45b2dd
--- /dev/null
+++ b/node/runtime/repair_choices.py
@@ -0,0 +1,195 @@
+"""Bounded repair choices shared by the report and App consent contract."""
+
+import hashlib
+import json
+import re
+
+START = ""
+END = ""
+PACKAGE = r"(?:@[a-z0-9][a-z0-9._-]*/)?[a-z0-9][a-z0-9._-]*"
+VERSION = r"(?:0|[1-9][0-9]*)\.(?:0|[1-9][0-9]*)\.(?:0|[1-9][0-9]*)"
+SHA = r"[a-f0-9]{40}"
+LINE = re.compile(
+ rf"- \[([ xX])\] Allow `({PACKAGE})` `({VERSION}) โ ({VERSION})` by updating its explicit pin\."
+)
+
+
+def _version_tuple(value):
+ if not isinstance(value, str) or not re.fullmatch(VERSION, value):
+ return None
+ return tuple(value.split("."))
+
+
+def _patch_key(version):
+ return len(version[2]), version[2]
+
+
+def validate(items):
+ """Return canonical sorted items; an empty selection is valid."""
+ if not isinstance(items, list) or len(items) > 20:
+ raise ValueError("Select at most twenty fixes")
+ result, names = [], set()
+ for item in items:
+ if not isinstance(item, dict) or set(item) != {"name", "from", "to"}:
+ raise ValueError("Invalid selection fields")
+ name = item["name"]
+ if (
+ not isinstance(name, str)
+ or not re.fullmatch(PACKAGE, name)
+ or len(name) > 214
+ or name in {"node_modules", "favicon.ico"}
+ or name in names
+ ):
+ raise ValueError("Invalid or duplicate package")
+ if any(
+ not isinstance(item[key], str) or not re.fullmatch(VERSION, item[key])
+ for key in ("from", "to")
+ ):
+ raise ValueError("Versions must be exact stable versions")
+ old, new = (item[key].split(".") for key in ("from", "to"))
+ if old[:2] != new[:2] or (len(old[2]), old[2]) >= (len(new[2]), new[2]):
+ raise ValueError("Only increasing same-minor patches are allowed")
+ names.add(name)
+ result.append({"name": name, "from": item["from"], "to": item["to"]})
+ return sorted(result, key=lambda item: item["name"])
+
+
+def parse(body):
+ if not isinstance(body, str) or body.count(START) != 1 or body.count(END) != 1:
+ raise ValueError("Expected one selection block")
+ lines = re.split(r"\r?\n", body)
+ try:
+ start, end = lines.index(START), lines.index(END)
+ except ValueError as error:
+ raise ValueError("Selection markers must occupy their own lines") from error
+ if end <= start:
+ raise ValueError("Selection markers are out of order")
+ all_items, selected = [], []
+ for line in lines[start + 1 : end]:
+ match = LINE.fullmatch(line)
+ if match is None:
+ raise ValueError("Invalid selection line")
+ item = dict(zip(("name", "from", "to"), match.groups()[1:]))
+ all_items.append(item)
+ if match[1].lower() == "x":
+ selected.append(item)
+ # App validation covers unchecked entries and duplicates too. Only checked
+ # entries count against the twenty-selection limit.
+ names = set()
+ for item in all_items:
+ validate([item])
+ if item["name"] in names:
+ raise ValueError("Duplicate package")
+ names.add(item["name"])
+ return validate(selected)
+
+
+def revision_marker(head, base):
+ if any(
+ not isinstance(value, str) or not re.fullmatch(SHA, value)
+ for value in (head, base)
+ ):
+ raise ValueError("Exact head and base revisions are required")
+ return f""
+
+
+def fingerprint(pr, head, base, items):
+ if type(pr) is not int or not 0 < pr <= 9_007_199_254_740_991:
+ raise ValueError("Invalid PR number")
+ revision_marker(head, base)
+ payload = json.dumps(
+ [pr, head, base, validate(items)], separators=(",", ":"), ensure_ascii=False
+ )
+ return hashlib.sha256(payload.encode()).hexdigest()
+
+
+def candidates(data):
+ """Use the audit engine's verified evidence, never a parsed status label."""
+ if not isinstance(data, dict) or data.get("package_manager") != "pnpm":
+ return []
+ result = []
+ for name, finding in sorted(data.get("vulnerabilities", {}).items()):
+ details = finding.get("via", [])
+ if not details or not all(isinstance(detail, dict) for detail in details):
+ continue
+ # The audit engine retains an eligible version set for every advisory.
+ # Intersect those sets before choosing one patch so a lower version from
+ # one advisory cannot hide a later patch that satisfies all advisories.
+ if any("_verifiedPatchedVersions" in detail for detail in details):
+ evidence = []
+ repair_from = []
+ for detail in details:
+ versions = detail.get("_verifiedRepairVersions")
+ source = detail.get("_repairFrom")
+ if not isinstance(versions, list) or not versions:
+ evidence = []
+ break
+ parsed = {_version_tuple(version) for version in versions}
+ if None in parsed:
+ evidence = []
+ break
+ if _version_tuple(source) is None:
+ evidence = []
+ break
+ evidence.append(set(versions))
+ repair_from.append(source)
+ if not evidence or len(set(repair_from)) != 1:
+ continue
+ common = set.intersection(*evidence)
+ old_version = _version_tuple(repair_from[0])
+ eligible = [
+ version
+ for version in common
+ if (parsed_version := _version_tuple(version)) is not None
+ and parsed_version[:2] == old_version[:2]
+ and _patch_key(parsed_version) > _patch_key(old_version)
+ ]
+ if not eligible:
+ continue
+ try:
+ items = validate(
+ [
+ {
+ "name": name,
+ "from": repair_from[0],
+ "to": min(
+ eligible,
+ key=lambda version: _patch_key(_version_tuple(version)),
+ ),
+ }
+ ]
+ )
+ except ValueError:
+ continue
+ result.extend(items)
+ continue
+ return result[:20]
+
+
+def render(items, *, head, base):
+ """Render consent only with the exact revisions of the published PR."""
+ items = validate(items)
+ marker = revision_marker(head, base)
+ if not items:
+ return ""
+ return (
+ "\n## Choose fixes to try\n\n"
+ "These patches appear compatible with the checked parent constraints. "
+ "Repair validation has not run. Selecting a fix permits one attempt; "
+ "it does not approve merging.\n\n"
+ + START
+ + "\n"
+ + "\n".join(
+ f"- [ ] Allow `{item['name']}` `{item['from']} โ {item['to']}` by updating its explicit pin."
+ for item in items
+ )
+ + "\n"
+ + END
+ + "\n"
+ + marker
+ + "\n\n### Run selected fixes\n\n"
+ "Select the fixes above, then post a **new comment** containing only "
+ "`/repair selected` and click GitHub's **Comment** button. "
+ "The bot acknowledges the request and updates the same reply with the result. "
+ "Checkbox edits alone start no work. Nothing merges automatically.\n"
+ )
diff --git a/node/runtime/templates/dependency_audit.py b/node/runtime/templates/dependency_audit.py
index ded2727..19bd2f3 100644
--- a/node/runtime/templates/dependency_audit.py
+++ b/node/runtime/templates/dependency_audit.py
@@ -3,6 +3,7 @@
import argparse
import copy
import hashlib
+import html
import json
import re
import runpy
@@ -17,6 +18,7 @@
LOCAL_PATCHES = runpy.run_path(str(Path(__file__).parents[1] / "local_patches.py"))
NODE_RELEASES = runpy.run_path(str(Path(__file__).parents[1] / "node_releases.py"))
+REPAIR_CHOICES = runpy.run_path(str(Path(__file__).parents[1] / "repair_choices.py"))
SEVERITY_ORDER = {"critical": 0, "high": 1, "moderate": 2, "low": 3, "info": 4}
DIRECTORY_PATTERN = re.compile(r"[A-Za-z0-9_./ -]+")
@@ -1127,9 +1129,13 @@ def published(specifier: str) -> set[str] | None:
vulnerability["fixAvailable"] = False
for detail in vulnerability["via"]:
detail.pop("fixVersion", None)
+ detail.pop("_repairCandidate", None)
+ detail.pop("_repairFrom", None)
+ detail.pop("_verifiedRepairVersions", None)
+ detail.pop("_checkedParentConstraints", None)
+ detail.pop("_verifiedPatchedVersions", None)
if collect_proof:
detail.pop("_verifiedVulnerableVersions", None)
- detail.pop("_verifiedPatchedVersions", None)
patched = detail.get("patchedRange")
vulnerable_range = detail.get("range")
if collect_proof and isinstance(vulnerable_range, str) and vulnerable_range:
@@ -1159,10 +1165,13 @@ def published(specifier: str) -> set[str] | None:
f"No published stable patch matches registry range `{patched}`."
)
continue
- if collect_proof:
- detail["_verifiedPatchedVersions"] = sorted(patches)
+ # Candidate selection needs the same verified publication set that
+ # the audit report used. Keep this evidence internal; the normal
+ # report continues to use fixStatus/fixVersion below.
+ detail["_verifiedPatchedVersions"] = sorted(patches)
vulnerability["fixAvailable"] = True
blocked = []
+ checked_parents = []
unknown = False
compatible_patches = None
for parent, snapshot in lock.get("snapshots", {}).items():
@@ -1191,6 +1200,7 @@ def published(specifier: str) -> set[str] | None:
if not isinstance(constraint, str):
unknown = True
continue
+ checked_parents.append(f"{parent} declares {name}: {constraint}")
intersections = [
published(f"{name}@{parent_range.strip()} {patch_range.strip()}")
for parent_range in constraint.split("||")
@@ -1212,6 +1222,8 @@ def published(specifier: str) -> set[str] | None:
if compatible_patches is None
else compatible_patches & parent_patches
)
+ detail["_checkedParentConstraints"] = sorted(set(checked_parents))
+ parent_compatible_patches = compatible_patches
override = overrides.get(name) if isinstance(overrides, dict) else None
override_status = ""
override_patches = None
@@ -1286,6 +1298,22 @@ def published(specifier: str) -> set[str] | None:
candidates = compatible_patches
if candidates is None:
candidates = override_patches or patches
+ # A repair updates one exact, unqualified override. Its current
+ # value must not constrain the candidate set: that is the pin being
+ # repaired. Parent-compatible published versions remain eligible,
+ # while unknown evidence is withheld below.
+ if (
+ isinstance(override, str)
+ and stable_version(override) is not None
+ and not unknown
+ and not qualified
+ ):
+ repair_versions = parent_compatible_patches
+ if repair_versions is None:
+ repair_versions = patches
+ if repair_versions:
+ detail["_repairFrom"] = override
+ detail["_verifiedRepairVersions"] = sorted(repair_versions)
version = min(candidates, key=lambda v: tuple(map(int, v.split("."))))
if not unknown and not override_status:
detail["fixVersion"] = version
@@ -2403,8 +2431,6 @@ def package_advisory_links(before, after, name, source, resolved, pending):
def audit_fix_reason(detail, source):
status = detail.get("fixStatus", "")
- if "Blocked by retained override" in status:
- return "review", "A retained override blocks the published patch."
if "no single patch satisfies all" in status:
parents = sorted(set(re.findall(r"`([^`]+)` requires `[^`]+`", status)))
if len(parents) == 1:
@@ -2413,6 +2439,8 @@ def audit_fix_reason(detail, source):
f"Blocked by the `{parents[0].rsplit('@', 1)[0]}` version constraint.",
)
return "blocked", "No published patch fits all checked parent constraints."
+ if "Blocked by retained override" in status:
+ return "review", "A retained override blocks the published patch."
if status.startswith("No published stable patch"):
return "blocked", "No published patch matches the fix range."
if status.startswith("No patched version declared"):
@@ -2478,6 +2506,8 @@ def format_audit_summary(
reported_parents=None,
locks=None,
summary=None,
+ repair_options=None,
+ repair_revision=None,
) -> str:
if isinstance(after, str):
return "" if fixes_only else f"โช **Security:** {after}"
@@ -2518,6 +2548,13 @@ def complete(data, rows):
)
fixed_packages, review_packages, blocked_packages = set(), set(), set()
new_packages, unconfirmed_packages = set(), set()
+ repair_options = REPAIR_CHOICES["validate"](repair_options or [])
+ repair_names = {item["name"] for item in repair_options}
+ selection_block = (
+ REPAIR_CHOICES["render"](repair_options, **repair_revision)
+ if repair_revision is not None
+ else ""
+ )
for name in old_groups.keys() | groups.keys():
previous, pending = old_groups.get(name, {}), groups.get(name, {})
fixed = len(resolved_ids.intersection(previous))
@@ -2534,7 +2571,7 @@ def complete(data, rows):
outcomes = [audit_fix_reason(detail, source) for detail in details]
statuses = [detail.get("fixStatus", "") for detail in details]
pinned = any("Blocked by retained override" in status for status in statuses)
- review = any(kind == "review" for kind, _ in outcomes)
+ review = pinned or any(kind == "review" for kind, _ in outcomes)
blocked = any(kind == "blocked" for kind, _ in outcomes) or any(
"no single patch satisfies all" in status for status in statuses
)
@@ -2610,6 +2647,8 @@ def complete(data, rows):
+ ", ".join(f"`{other}`" for other in locations)
+ "."
)
+ if name in repair_names:
+ action = "Likely feasible. Repair validation has not run. " + action
new_count = len(new_ids.intersection(pending))
if new_count:
new_packages.add(name)
@@ -2646,12 +2685,12 @@ def complete(data, rows):
counts.insert(0, "fixes unconfirmed")
if comparable and not pending_count and not unconfirmed_disappeared:
badge, rank, action = "๐ข", 0, "Resolved."
- elif review:
- badge, rank = "๐ ", 1
elif new_count:
badge, rank = "๐ด", 2
elif blocked:
badge, rank = "", 2
+ elif review:
+ badge, rank = "๐ ", 1
else:
badge, rank = "โช", 3
severity_details = details + unconfirmed_details + list(previous.values())
@@ -2678,16 +2717,33 @@ def complete(data, rows):
)
label = {
"๐ข": "โ
Fix proposed. ",
- "๐ ": "๐ Needs review. ",
+ "๐ ": "๐ก Needs review. ",
"๐ด": "๐ด New findings. ",
"โช": "โช Needs verification. ",
- "": "Blocked. ",
+ "": "๐ด Blocked. ",
}[badge]
+ if blocked and review:
+ label += "๐ก Needs review. "
rows.append(
(
(rank, severity, name),
- f"| `{name}` | {version_change} | {severity_label} | {' ยท '.join(counts)} | {label}{action.strip()} | "
- f"{package_advisory_links(before, after, name, source, resolved_ids.intersection(previous) if fixes_only else set(previous), set() if fixes_only else set(pending))} |",
+ [
+ f"`{name}`",
+ version_change,
+ severity_label,
+ " ยท ".join(counts),
+ label + action.strip(),
+ package_advisory_links(
+ before,
+ after,
+ name,
+ source,
+ resolved_ids.intersection(previous)
+ if fixes_only
+ else set(previous),
+ set() if fixes_only else set(pending),
+ ),
+ ],
)
)
@@ -2708,9 +2764,9 @@ def packages(number):
headline += f" ยท ๐ด **New findings in {packages(len(new_packages))}**"
if review_packages:
verb = "needs" if len(review_packages) == 1 else "need"
- headline += f" ยท ๐ **{packages(len(review_packages))} {verb} your review**"
+ headline += f" ยท ๐ก **{packages(len(review_packages))} {verb} your review**"
if blocked_packages:
- headline += f" ยท **Fixes blocked for {packages(len(blocked_packages))}**"
+ headline += f" ยท ๐ด **Fixes blocked for {packages(len(blocked_packages))}**"
if unconfirmed_packages:
verb = "needs" if len(unconfirmed_packages) == 1 else "need"
headline += (
@@ -2724,13 +2780,82 @@ def packages(number):
)
if rows:
unit = "Advisories" if source == "pnpm" else "Affected package entries"
+ if repair_options:
+
+ def cell(value):
+ escaped = html.escape(html.unescape(value))
+ escaped = re.sub(r"`([^`]+)`", r"\1", escaped)
+ return re.sub(
+ r"\[([^\]]+)\]\((https?://[^ )]+)\)",
+ r'\1',
+ escaped,
+ )
+
+ options = {item["name"]: item for item in repair_options}
+ table = [
+ "
| {value} | " + for value in ( + "Package", + "Update", + "Severity", + unit, + "Action / blocker", + "Security evidence", + ) + ) + + "|||||
|---|---|---|---|---|---|
| {cell(value)} | " for value in columns) + + "|||||
Pin repair assessment for ' + + html.escape(name) + + "A compatible published patch was found: The existing pin remains unchanged. Historical rationale has not been collected automatically.
| |||||