diff --git a/.github/pyrightconfig.json b/.github/pyrightconfig.json
new file mode 100644
index 0000000..6ceb2c0
--- /dev/null
+++ b/.github/pyrightconfig.json
@@ -0,0 +1,15 @@
+{
+ "typeCheckingMode": "strict",
+ "pythonVersion": "3.13",
+ "include": [
+ "scripts/*.py"
+ ],
+ "reportMissingTypeStubs": "error",
+ "reportUnnecessaryTypeIgnoreComment": "error",
+ "reportImplicitOverride": "error",
+ "exclude": [
+ "**/__pycache__"
+ ],
+ "venvPath": "..",
+ "venv": ".venv"
+}
diff --git a/.github/scripts/check-codeql-sarif.py b/.github/scripts/check-codeql-sarif.py
index 8705490..7177466 100644
--- a/.github/scripts/check-codeql-sarif.py
+++ b/.github/scripts/check-codeql-sarif.py
@@ -2,55 +2,107 @@
# SPDX-License-Identifier: Apache-2.0
"""Reject missing/malformed CodeQL evidence before enforcing the severity gate."""
+
import json
import math
import os
from pathlib import Path
+from typing import TypedDict
+
+RuleProperties = TypedDict("RuleProperties", {"security-severity": str | float | None}, total=False)
-def check(directory, language, threshold):
- if not math.isfinite(threshold) or not 0 <= threshold <= 10:
- raise ValueError("Invalid security severity threshold")
- canonical = {"java-kotlin": "java", "javascript-typescript": "javascript", "c-cpp": "cpp"}.get(language, language)
- files = sorted(Path(directory).glob("*.sarif"))
- if not canonical or not any(path.name == canonical + ".sarif" for path in files):
- raise ValueError("Missing expected CodeQL SARIF report for " + language)
- blocking = []
- for path in files:
- report = json.loads(path.read_text(encoding="utf-8"))
- runs = report.get("runs")
- if report.get("version") != "2.1.0" or not isinstance(runs, list) or not runs:
- raise ValueError("Invalid or empty SARIF runs: " + path.name)
- for run in runs:
- driver = run.get("tool", {}).get("driver", {})
- results = run.get("results")
- if driver.get("name") != "CodeQL" or not isinstance(driver.get("rules"), list) or not isinstance(results, list):
- raise ValueError("Missing CodeQL tool/rules/results evidence: " + path.name)
- rules = {}
- for component in [driver, *run.get("tool", {}).get("extensions", [])]:
- for rule in component.get("rules", []):
- rules[rule["id"]] = rule
- if not rules:
- raise ValueError("CodeQL report has no analyzed rule metadata: " + path.name)
- for invocation in run.get("invocations", []):
- if invocation.get("executionSuccessful") is False:
- raise ValueError("CodeQL invocation failed: " + path.name)
- for result in results:
- rule_id = result.get("ruleId") or result.get("rule", {}).get("id")
- if rule_id not in rules:
- raise ValueError("Result references missing rule metadata: " + path.name)
- if result.get("suppressions"):
- continue
- severity = float(rules[rule_id].get("properties", {}).get("security-severity", 0) or 0)
- if not math.isfinite(severity) or not 0 <= severity <= 10:
- raise ValueError("Invalid rule severity: " + rule_id)
- if severity >= threshold:
- blocking.append((severity, rule_id))
- for severity, rule_id in sorted(blocking, reverse=True):
- print(f"::error title=CodeQL security finding::{rule_id} ({severity})")
- print(f"Validated {len(files)} SARIF report(s); {len(blocking)} blocking result(s)")
- return bool(blocking)
+class Rule(TypedDict):
+ """Keep the rule identity and optional severity metadata used by the gate."""
+
+ id: str
+ properties: RuleProperties
+
+
+class Component(TypedDict, total=False):
+ """Describe CodeQL driver or extension rule metadata."""
+
+ name: str
+ rules: list[Rule]
+
+
+class Tool(TypedDict, total=False):
+ """Group the driver and optional extension rule catalogs."""
+
+ driver: Component
+ extensions: list[Component]
+
+
+class Result(TypedDict, total=False):
+ """Describe only the result fields consumed by the severity gate."""
+
+ ruleId: str
+ rule: dict[str, str]
+ suppressions: list[object]
+
+
+class Run(TypedDict, total=False):
+ """Retain invocation success and findings for one analyzed language."""
+
+ tool: Tool
+ results: list[Result]
+ invocations: list[dict[str, bool]]
+
+
+class Report(TypedDict, total=False):
+ """Represent SARIF fields while retaining the explicit runtime validation."""
+
+ version: str
+ runs: list[Run]
+
+
+MAX_SEVERITY = 10
+
+
+def check(directory: str | Path, language: str, threshold: float) -> bool: # noqa: PLR0912 -- Keep the ordered evidence checks together; none may be skipped.
+ """Validate complete CodeQL evidence and report findings at the threshold."""
+ if not math.isfinite(threshold) or not 0 <= threshold <= MAX_SEVERITY:
+ raise ValueError("Invalid security severity threshold")
+ canonical = {"java-kotlin": "java", "javascript-typescript": "javascript", "c-cpp": "cpp"}.get(language, language)
+ files = sorted(Path(directory).glob("*.sarif"))
+ if not canonical or not any(path.name == canonical + ".sarif" for path in files):
+ raise ValueError("Missing expected CodeQL SARIF report for " + language)
+ blocking: list[tuple[float, str]] = []
+ for path in files:
+ report: Report = json.loads(path.read_text(encoding="utf-8"))
+ runs = report.get("runs")
+ if report.get("version") != "2.1.0" or not isinstance(runs, list) or not runs:
+ raise ValueError("Invalid or empty SARIF runs: " + path.name)
+ for run in runs:
+ driver = run.get("tool", {}).get("driver", {})
+ results = run.get("results")
+ if driver.get("name") != "CodeQL" or not isinstance(driver.get("rules"), list) or not isinstance(results, list):
+ raise ValueError("Missing CodeQL tool/rules/results evidence: " + path.name)
+ rules: dict[str, Rule] = {}
+ for component in [driver, *run.get("tool", {}).get("extensions", [])]:
+ for rule in component.get("rules", []):
+ rules[rule["id"]] = rule
+ if not rules:
+ raise ValueError("CodeQL report has no analyzed rule metadata: " + path.name)
+ for invocation in run.get("invocations", []):
+ if invocation.get("executionSuccessful") is False:
+ raise ValueError("CodeQL invocation failed: " + path.name)
+ for result in results:
+ rule_id = result.get("ruleId") or result.get("rule", {}).get("id")
+ if rule_id not in rules:
+ raise ValueError("Result references missing rule metadata: " + path.name)
+ if result.get("suppressions"):
+ continue
+ severity = float(rules[rule_id].get("properties", {}).get("security-severity", 0) or 0)
+ if not math.isfinite(severity) or not 0 <= severity <= MAX_SEVERITY:
+ raise ValueError("Invalid rule severity: " + rule_id)
+ if severity >= threshold:
+ blocking.append((severity, rule_id))
+ for severity, rule_id in sorted(blocking, reverse=True):
+ print(f"::error title=CodeQL security finding::{rule_id} ({severity})")
+ print(f"Validated {len(files)} SARIF report(s); {len(blocking)} blocking result(s)")
+ return bool(blocking)
if __name__ == "__main__":
- raise SystemExit(1 if check(os.environ["SARIF_DIR"], os.environ["EXPECTED_LANGUAGE"], float(os.environ["THRESHOLD"])) else 0)
+ raise SystemExit(1 if check(os.environ["SARIF_DIR"], os.environ["EXPECTED_LANGUAGE"], float(os.environ["THRESHOLD"])) else 0)
diff --git a/.github/scripts/check-maven-inventory.py b/.github/scripts/check-maven-inventory.py
index ffaac49..616f308 100644
--- a/.github/scripts/check-maven-inventory.py
+++ b/.github/scripts/check-maven-inventory.py
@@ -2,127 +2,172 @@
# SPDX-License-Identifier: Apache-2.0
"""Fail closed unless scanners cover Maven's selected UI-host runtime graph."""
+
import argparse
-from collections import Counter
import json
-from pathlib import Path
import re
+from collections import Counter
+from pathlib import Path
+from typing import TypedDict, cast
from urllib.parse import unquote, urlsplit
+class Bom(TypedDict, total=False):
+ """Describe the selected CycloneDX coordinate fields."""
+
+ bomFormat: str
+ components: list[dict[str, str]]
+ metadata: dict[str, dict[str, str]]
+
+
+class TrivyReport(TypedDict, total=False):
+ """Describe package coordinates, leaving runtime shape checks intact."""
+
+ Results: list[dict[str, list[dict[str, str]]]]
+
+
+class OsvReport(TypedDict, total=False):
+ """Keep result bodies untrusted until their explicit validation."""
+
+ results: list[object]
+
+
+class OsvResult(TypedDict, total=False):
+ """Describe source identity and packages after checking the object shape."""
+
+ source: dict[str, str]
+ packages: list[object]
+
+
+class OsvEntry(TypedDict, total=False):
+ """Describe the coordinate and optional vulnerability array being checked."""
+
+ package: dict[str, str]
+ vulnerabilities: object
+
+
REQUIRED = {
- "com.kingsrook.qqq:qqq-frontend-next",
- "com.kingsrook.qqq:qqq-backend-core",
- "com.kingsrook.qqq:qqq-backend-module-filesystem",
- "com.google.guava:guava",
- "com.h2database:h2",
+ "com.kingsrook.qqq:qqq-frontend-next",
+ "com.kingsrook.qqq:qqq-backend-core",
+ "com.kingsrook.qqq:qqq-backend-module-filesystem",
+ "com.google.guava:guava",
+ "com.h2database:h2",
}
REVIEWED_H2_VERSION = "2.4.240"
OMITTED = object()
-def resolved(text):
- packages = set()
- for line in text.splitlines():
- line = re.sub(r"\x1b\[[0-9;]*m", "", line).strip().split(" -- ", 1)[0].strip()
- if not line or line == "The following files have been resolved:":
- continue
- parts = line.split(":")
- if len(parts) not in (5, 6) or parts[-1] not in ("compile", "runtime"):
- raise ValueError(f"Unexpected Maven runtime coordinate: {line}")
- if not all(parts) or parts[2] != "jar":
- raise ValueError(f"Invalid runtime JAR coordinate: {line}")
- name = parts[0] + ":" + parts[1]
- if parts[0].startswith(("org.junit", "org.opentest4j", "org.apiguardian")):
- raise ValueError("Test library in runtime inventory")
- packages.add((name, parts[-2]))
- if not REQUIRED.issubset({name for name, _ in packages}):
- raise ValueError("Maven inventory is empty or lacks required UI-host dependencies")
- # The pinned dependency-review license exception matcher ignores versions.
- # Keep any H2 metadata exception limited to the reviewed, unmodified runtime.
- if {version for name, version in packages if name == "com.h2database:h2"} != {REVIEWED_H2_VERSION}:
- raise ValueError("H2 runtime must match reviewed version " + REVIEWED_H2_VERSION)
- return packages
+def resolved(text: str) -> set[tuple[str, str]]:
+ """Read Maven's selected runtime JARs and enforce the reviewed H2 pin."""
+ packages: set[tuple[str, str]] = set()
+ for raw_line in text.splitlines():
+ line = re.sub(r"\x1b\[[0-9;]*m", "", raw_line).strip().split(" -- ", 1)[0].strip()
+ if not line or line == "The following files have been resolved:":
+ continue
+ parts = line.split(":")
+ if len(parts) not in (5, 6) or parts[-1] not in ("compile", "runtime"):
+ raise ValueError(f"Unexpected Maven runtime coordinate: {line}")
+ if not all(parts) or parts[2] != "jar":
+ raise ValueError(f"Invalid runtime JAR coordinate: {line}")
+ name = parts[0] + ":" + parts[1]
+ if parts[0].startswith(("org.junit", "org.opentest4j", "org.apiguardian")):
+ raise ValueError("Test library in runtime inventory")
+ packages.add((name, parts[-2]))
+ if not REQUIRED.issubset({name for name, _ in packages}):
+ raise ValueError("Maven inventory is empty or lacks required UI-host dependencies")
+ # The pinned dependency-review license exception matcher ignores versions.
+ # Keep any H2 metadata exception limited to the reviewed, unmodified runtime.
+ if {version for name, version in packages if name == "com.h2database:h2"} != {REVIEWED_H2_VERSION}:
+ raise ValueError("H2 runtime must match reviewed version " + REVIEWED_H2_VERSION)
+ return packages
-def components(report):
- if not isinstance(report, dict) or report.get("bomFormat") != "CycloneDX":
- raise ValueError("Expected CycloneDX inventory")
- packages = set()
- for component in report.get("components", []):
- purl = urlsplit(component.get("purl", ""))
- if purl.scheme != "pkg" or not purl.path.startswith("maven/"):
- raise ValueError("Missing Maven package URL")
- coordinate, version = purl.path.removeprefix("maven/").rsplit("@", 1)
- group, artifact = coordinate.split("/", 1)
- packages.add((unquote(group) + ":" + unquote(artifact), unquote(version)))
- if not packages:
- raise ValueError("Empty CycloneDX inventory")
- return packages
+def components(report: object) -> set[tuple[str, str]]:
+ """Extract exact Maven coordinates from the scanner's CycloneDX inventory."""
+ typed_report = cast(Bom, report)
+ if not isinstance(report, dict) or typed_report.get("bomFormat") != "CycloneDX":
+ raise ValueError("Expected CycloneDX inventory")
+ packages: set[tuple[str, str]] = set()
+ for component in typed_report.get("components", []):
+ purl = urlsplit(component.get("purl", ""))
+ if purl.scheme != "pkg" or not purl.path.startswith("maven/"):
+ raise ValueError("Missing Maven package URL")
+ coordinate, version = purl.path.removeprefix("maven/").rsplit("@", 1)
+ group, artifact = coordinate.split("/", 1)
+ packages.add((unquote(group) + ":" + unquote(artifact), unquote(version)))
+ if not packages:
+ raise ValueError("Empty CycloneDX inventory")
+ return packages
-def check(expected, bom, trivy=OMITTED):
- actual = components(bom)
- if actual != expected:
- raise ValueError(f"Maven/SBOM mismatch: missing={sorted(expected - actual)}, extra={sorted(actual - expected)}")
- if trivy is not OMITTED:
- if not isinstance(trivy, dict) or not isinstance(trivy.get("Results"), list):
- raise ValueError("Expected Trivy report object with Results array")
- scanned = {(p["Name"], p["Version"]) for r in trivy.get("Results", []) for p in r.get("Packages", [])}
- # Trivy also reports the root project represented by SBOM metadata.
- root = bom.get("metadata", {}).get("component", {})
- allowed = expected | {(root.get("group", "") + ":" + root.get("name", ""), root.get("version", ""))}
- if not expected.issubset(scanned) or not scanned.issubset(allowed):
- raise ValueError(f"Maven/Trivy mismatch: missing={sorted(expected - scanned)}, extra={sorted(scanned - allowed)}")
- return len(expected)
+def check(expected: set[tuple[str, str]], bom: object, trivy: object = OMITTED) -> int:
+ """Require exact SBOM coverage and, when supplied, selected Trivy coverage."""
+ actual = components(bom)
+ if actual != expected:
+ raise ValueError(f"Maven/SBOM mismatch: missing={sorted(expected - actual)}, extra={sorted(actual - expected)}")
+ if trivy is not OMITTED:
+ typed_trivy = cast(TrivyReport, trivy)
+ if not isinstance(trivy, dict) or not isinstance(typed_trivy.get("Results"), list):
+ raise ValueError("Expected Trivy report object with Results array")
+ scanned = {(p["Name"], p["Version"]) for r in typed_trivy.get("Results", []) for p in r.get("Packages", [])}
+ # Trivy also reports the root project represented by SBOM metadata.
+ root = cast(Bom, bom).get("metadata", {}).get("component", {})
+ allowed = expected | {(root.get("group", "") + ":" + root.get("name", ""), root.get("version", ""))}
+ if not expected.issubset(scanned) or not scanned.issubset(allowed):
+ raise ValueError(f"Maven/Trivy mismatch: missing={sorted(expected - scanned)}, extra={sorted(scanned - allowed)}")
+ return len(expected)
-def check_osv(expected, report):
- if not isinstance(report, dict) or not isinstance(report.get("results"), list):
- raise ValueError("Expected OSV report object with results array")
- # OSV 2.6.0 queries CycloneDX purls but omits Maven groups in JSON names:
- # https://github.com/google/osv-scanner/issues/3099
- # Fail on ambiguous display coordinates instead of claiming full identity proof.
- projected = Counter((name.split(":", 1)[1], version) for name, version in expected)
- if any(count != 1 for count in projected.values()):
- raise ValueError("Ambiguous Maven coordinates in pinned OSV report format")
- scanned = Counter()
- for result in report["results"]:
- if not isinstance(result, dict) or not isinstance(result.get("source"), dict):
- raise ValueError("Missing OSV source")
- if result["source"].get("type") != "sbom" or not isinstance(result["source"].get("path"), str):
- raise ValueError("Expected OSV SBOM source")
- if not isinstance(result.get("packages"), list):
- raise ValueError("Missing OSV package inventory; use --all-packages")
- for entry in result["packages"]:
- if not isinstance(entry, dict) or not isinstance(entry.get("package"), dict):
- raise ValueError("Malformed OSV package")
- package = entry["package"]
- if package.get("ecosystem") != "Maven" or not all(isinstance(package.get(k), str) and package[k] for k in ("name", "version")):
- raise ValueError("Invalid OSV Maven package coordinate")
- if "vulnerabilities" in entry and not isinstance(entry["vulnerabilities"], list):
- raise ValueError("Malformed OSV vulnerabilities")
- scanned[(package["name"], package["version"])] += 1
- if scanned != projected:
- raise ValueError("OSV --all-packages inventory differs from selected Maven graph")
- return len(expected)
+def check_osv(expected: set[tuple[str, str]], report: object) -> int:
+ """Reject missing, ambiguous, malformed or partial OSV package evidence."""
+ typed_report = cast(OsvReport, report)
+ if not isinstance(report, dict) or not isinstance(typed_report.get("results"), list):
+ raise ValueError("Expected OSV report object with results array")
+ # OSV 2.6.0 queries CycloneDX purls but omits Maven groups in JSON names:
+ # https://github.com/google/osv-scanner/issues/3099
+ # Fail on ambiguous display coordinates instead of claiming full identity proof.
+ projected = Counter((name.split(":", 1)[1], version) for name, version in expected)
+ if any(count != 1 for count in projected.values()):
+ raise ValueError("Ambiguous Maven coordinates in pinned OSV report format")
+ scanned: Counter[tuple[str, str]] = Counter()
+ for raw_result in cast(list[object], typed_report.get("results")):
+ result = cast(OsvResult, raw_result)
+ if not isinstance(raw_result, dict) or not isinstance(result.get("source"), dict):
+ raise ValueError("Missing OSV source")
+ source = cast(dict[str, str], result.get("source"))
+ if source.get("type") != "sbom" or not isinstance(source.get("path"), str):
+ raise ValueError("Expected OSV SBOM source")
+ if not isinstance(result.get("packages"), list):
+ raise ValueError("Missing OSV package inventory; use --all-packages")
+ for raw_entry in cast(list[object], result.get("packages")):
+ entry = cast(OsvEntry, raw_entry)
+ if not isinstance(raw_entry, dict) or not isinstance(entry.get("package"), dict):
+ raise ValueError("Malformed OSV package")
+ package = cast(dict[str, str], entry.get("package"))
+ if package.get("ecosystem") != "Maven" or not all(isinstance(package.get(k), str) and package[k] for k in ("name", "version")):
+ raise ValueError("Invalid OSV Maven package coordinate")
+ if "vulnerabilities" in entry and not isinstance(entry["vulnerabilities"], list):
+ raise ValueError("Malformed OSV vulnerabilities")
+ scanned[(package["name"], package["version"])] += 1
+ if scanned != projected:
+ raise ValueError("OSV --all-packages inventory differs from selected Maven graph")
+ return len(expected)
-def main():
- parser = argparse.ArgumentParser()
- parser.add_argument("resolved", type=Path)
- parser.add_argument("bom", type=Path)
- parser.add_argument("trivy", type=Path, nargs="?")
- parser.add_argument("--osv", type=Path)
- args = parser.parse_args()
- expected = resolved(args.resolved.read_text(encoding="utf-8"))
- count = check(expected,
- json.loads(args.bom.read_text(encoding="utf-8")),
- json.loads(args.trivy.read_text(encoding="utf-8")) if args.trivy else OMITTED)
- if args.osv:
- check_osv(expected, json.loads(args.osv.read_text(encoding="utf-8")))
- print(f"Verified {count} Maven runtime dependencies, including the UI frontend.")
+def main() -> None:
+ """Validate the exact evidence files supplied by the required CI job."""
+ parser = argparse.ArgumentParser()
+ parser.add_argument("resolved", type=Path)
+ parser.add_argument("bom", type=Path)
+ parser.add_argument("trivy", type=Path, nargs="?")
+ parser.add_argument("--osv", type=Path)
+ args = parser.parse_args()
+ expected = resolved(args.resolved.read_text(encoding="utf-8"))
+ count = check(expected, json.loads(args.bom.read_text(encoding="utf-8")), json.loads(args.trivy.read_text(encoding="utf-8")) if args.trivy else OMITTED)
+ if args.osv:
+ check_osv(expected, json.loads(args.osv.read_text(encoding="utf-8")))
+ print(f"Verified {count} Maven runtime dependencies, including the UI frontend.")
if __name__ == "__main__":
- main()
+ main()
diff --git a/.github/scripts/check-script-quality.py b/.github/scripts/check-script-quality.py
new file mode 100644
index 0000000..18385f0
--- /dev/null
+++ b/.github/scripts/check-script-quality.py
@@ -0,0 +1,112 @@
+# Copyright 2026 Kingsrook, LLC
+# SPDX-License-Identifier: Apache-2.0
+
+"""Fail closed unless every tracked owned script is covered by, and really analyzed by, the quality tools."""
+
+import json
+import shutil
+import subprocess
+import sys
+from pathlib import Path, PurePosixPath
+from typing import Any
+
+PYTHON_SUFFIXES = (".py", ".pyi", ".pyw")
+SCRIPT_PATTERN = ".github/scripts/*.py"
+PYRIGHT_CONFIG = ".github"
+JAVASCRIPT_SUFFIXES = (".js", ".cjs", ".mjs", ".jsx", ".ts", ".mts", ".cts", ".tsx")
+# Microsoft's official npm Pyright, restored from package-lock.json by `npm ci --ignore-scripts`.
+PYRIGHT_ENTRY = Path(__file__).resolve().parents[2] / "node_modules" / "pyright" / "index.js"
+
+
+def tracked_files(root: Path) -> list[str]:
+ """List tracked and not-yet-ignored new paths so no source file can silently bypass the gates."""
+ # Fixed Git inventory operation; the root is an argument, never shell input.
+ output = subprocess.run(["git", "-C", str(root), "ls-files", "-z", "--cached", "--others", "--exclude-standard"], check=True, capture_output=True).stdout.decode("utf-8") # noqa: S603, S607 -- Resolve Git from the developer/CI PATH; no shell or user-supplied command.
+ return [name for name in output.split("\u0000") if name]
+
+
+def python_scope(tracked: list[str]) -> list[str]:
+ """Return the tracked Python sources, rejecting any outside the one checked script root."""
+ candidates = sorted(name for name in tracked if name.endswith(PYTHON_SUFFIXES))
+ outside = [name for name in candidates if not PurePosixPath(name).full_match(SCRIPT_PATTERN)]
+ if outside:
+ raise ValueError("Tracked Python outside the checked script root: " + ", ".join(outside))
+ if not candidates:
+ raise ValueError("No owned Python scripts selected")
+ return candidates
+
+
+def javascript_scope(root: Path, tracked: list[str]) -> list[str]:
+ """Return tracked JavaScript/TypeScript sources, rejecting any outside the tsconfig include list."""
+ candidates = sorted(name for name in tracked if name.endswith(JAVASCRIPT_SUFFIXES))
+ tsconfig = root / "tsconfig.json"
+ if not tsconfig.is_file():
+ if candidates:
+ raise ValueError("Tracked JavaScript/TypeScript without a tsconfig.json: " + ", ".join(candidates))
+ return []
+ includes: list[str] = json.loads(tsconfig.read_text(encoding="utf-8"))["include"]
+ outside = [name for name in candidates if not any(PurePosixPath(name).full_match(pattern) for pattern in includes)]
+ if outside:
+ raise ValueError("Tracked JavaScript/TypeScript outside the tsconfig include list: " + ", ".join(outside))
+ if not candidates:
+ raise ValueError("No owned JavaScript sources selected")
+ return candidates
+
+
+def run_pyright(root: Path, config: str) -> dict[str, Any]:
+ """Run the locked npm Pyright through Node with JSON output and return its full report."""
+ node = shutil.which("node")
+ if node is None:
+ raise ValueError("Node.js is required to run the locked Pyright")
+ if not PYRIGHT_ENTRY.is_file():
+ raise ValueError(f"Locked Pyright is missing; run npm ci --ignore-scripts ({PYRIGHT_ENTRY})")
+ # Resolved Node and locked entry script; the config is an argument, never shell input.
+ completed = subprocess.run([node, str(PYRIGHT_ENTRY), "-p", config, "--outputjson"], cwd=root, check=False, capture_output=True, text=True, encoding="utf-8") # noqa: S603 -- Resolved Node and locked entry script; no shell.
+ try:
+ report: dict[str, Any] = json.loads(completed.stdout[completed.stdout.index("{") :])
+ int(report["summary"]["filesAnalyzed"]) # Fail here when the summary shape is missing.
+ report["exitCode"] = completed.returncode
+ except (ValueError, KeyError) as error:
+ raise ValueError(f"Pyright produced no JSON summary (exit {completed.returncode}): {completed.stdout}{completed.stderr}") from error
+ return report
+
+
+def check_pyright_report(report: dict[str, Any], expected: int) -> None:
+ """Require the analyzed-file count to match the selected sources, with no errors or warnings."""
+ summary = report["summary"]
+ analyzed = summary["filesAnalyzed"]
+ if analyzed == 0 or analyzed != expected:
+ raise ValueError(f"Pyright analyzed {analyzed} files, expected {expected}")
+ if summary["errorCount"] or summary["warningCount"]:
+ details = [f"{item['file']}: {item['message']}" for item in report["generalDiagnostics"]]
+ raise ValueError("Pyright reported findings:\n" + "\n".join(details))
+ if report["exitCode"] != 0:
+ raise ValueError(f"Pyright exited {report['exitCode']} despite a clean summary")
+
+
+def run_ruff(root: Path, arguments: list[str], sources: list[str]) -> None:
+ """Run Ruff on the explicit source list so no selected file can be skipped."""
+ # Fixed interpreter module invocation; paths are arguments, never shell input.
+ if subprocess.run([sys.executable, "-m", "ruff", *arguments, "--no-cache", *sources], cwd=root, check=False).returncode != 0: # noqa: S603 -- Fixed interpreter and module; no shell.
+ raise ValueError("Ruff " + " ".join(arguments) + " failed")
+
+
+def main() -> None:
+ """Verify scope, then Ruff format/lint and strict Pyright with a positive analyzed count."""
+ root = Path(__file__).resolve().parents[2]
+ tracked = tracked_files(root)
+ sources = python_scope(tracked)
+ javascript = javascript_scope(root, tracked)
+ run_ruff(root, ["format", "--check"], sources)
+ run_ruff(root, ["check"], sources)
+ report = run_pyright(root, PYRIGHT_CONFIG)
+ check_pyright_report(report, len(sources))
+ print(f"Script quality verified: {len(sources)} Python scripts analyzed by Ruff and Pyright, {len(javascript)} JavaScript files in tsconfig scope")
+
+
+if __name__ == "__main__":
+ try:
+ main()
+ except (ValueError, OSError, subprocess.CalledProcessError) as error:
+ print(str(error), file=sys.stderr)
+ sys.exit(1)
diff --git a/.github/scripts/check-source-headers.py b/.github/scripts/check-source-headers.py
index 2392c7e..e98fd8f 100644
--- a/.github/scripts/check-source-headers.py
+++ b/.github/scripts/check-source-headers.py
@@ -19,14 +19,14 @@ def selected(name: str, policy: dict[str, Any]) -> bool:
return any(fnmatch.fnmatchcase(name, pattern) for pattern in policy["includes"])
-def leading_comments(text: str, style: str) -> str:
+def leading_comments(text: str, style: str) -> str: # noqa: PLR0912 -- Each recognized leading-comment syntax has a distinct fail-closed branch.
"""Read comments only, preserving required interpreter/XML directive ordering."""
text = text.removeprefix("\ufeff")
if text.startswith("", text)
if style != "xml" or declaration is None:
raise ValueError("Invalid XML declaration")
- text = text[declaration.end():]
+ text = text[declaration.end() :]
elif text.startswith("#!"):
if style not in ("hash", "powershell", "slash"):
raise ValueError("Unexpected interpreter directive")
@@ -35,7 +35,7 @@ def leading_comments(text: str, style: str) -> str:
if style != "powershell":
raise ValueError("Unexpected PowerShell directive")
text = text.partition("\n")[2]
- comments = []
+ comments: list[str] = []
while text.strip():
text = text.lstrip()
if text.startswith(" str:
elif style == "slash" and text.startswith("//"):
line, _, text = text.partition("\n")
comments.append(line[2:])
- elif ((style == "slash" and text.startswith("/*"))
- or (style == "xml" and text.startswith(""), "powershell": ("<#", "#>")}[style]
- body, separator, rest = text[len(opening):].partition(closing)
+ body, separator, rest = text[len(opening) :].partition(closing)
if not separator:
raise ValueError("Unterminated leading comment")
comments.append(body)
@@ -72,19 +70,22 @@ def check_notice(name: str, text: str, policy: dict[str, Any]) -> None:
approved = [item for item in policy["notices"] if any(fnmatch.fnmatchcase(name, pattern) for pattern in item["paths"])]
for notice in approved:
holder = re.escape(notice["holder"])
- copyright_pattern = rf"(?m)^Copyright\s+(?:\([cC]\)\s*)?\d{{4}}(?:[-–]\d{{4}})?\.?\s+{holder}(?:\s+\(KofTwentyTwo\))?\.?(?:\s+Licensed under the MIT License \(see LICENSE\)\.)?$"
+ copyright_pattern = rf"(?m)^Copyright\s+(?:\([cC]\)\s*)?\d{{4}}(?:[-\u2013]\d{{4}})?\.?\s+{holder}(?:\s+\(KofTwentyTwo\))?\.?(?:\s+Licensed under the MIT License \(see LICENSE\)\.)?$"
if re.search(copyright_pattern, comments) is None:
continue
spdx = re.findall(r"(?m)^SPDX-License-Identifier:\s*(\S+)\s*$", comments)
if spdx and any(value != notice["license"] for value in spdx):
continue
- apache = all(phrase in comments for phrase in (
- "Licensed under the Apache License, Version 2.0",
- "you may not use this file except in compliance with the License.",
- "https://www.apache.org/licenses/LICENSE-2.0",
- "WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND",
- "limitations under the License.",
- ))
+ apache = all(
+ phrase in comments
+ for phrase in (
+ "Licensed under the Apache License, Version 2.0",
+ "you may not use this file except in compliance with the License.",
+ "https://www.apache.org/licenses/LICENSE-2.0",
+ "WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND",
+ "limitations under the License.",
+ )
+ )
if suffix == ".java" and not apache:
continue
if notice["license"] in spdx or (notice["license"] == "Apache-2.0" and apache):
@@ -94,8 +95,9 @@ def check_notice(name: str, text: str, policy: dict[str, Any]) -> None:
def check_repository(root: Path, policy: dict[str, Any]) -> int:
"""Check tracked sources without scanning build outputs or executing source."""
- tracked = subprocess.run(["git", "-C", str(root), "ls-files", "-z"], check=True, capture_output=True).stdout.decode("utf-8").split("\u0000")
- failures = []
+ # Fixed Git inventory operation; paths are arguments, never shell input.
+ tracked = subprocess.run(["git", "-C", str(root), "ls-files", "-z"], check=True, capture_output=True).stdout.decode("utf-8").split("\u0000") # noqa: S603, S607 -- Resolve Git from the developer/CI PATH; no shell or user-supplied command.
+ failures: list[str] = []
count = 0
for name in tracked:
if not name or not selected(name, policy):
diff --git a/.github/scripts/test-maven-inventory.py b/.github/scripts/test-maven-inventory.py
index 8bd5c77..981908a 100644
--- a/.github/scripts/test-maven-inventory.py
+++ b/.github/scripts/test-maven-inventory.py
@@ -2,107 +2,117 @@
# SPDX-License-Identifier: Apache-2.0
"""Synthetic regressions for missing, incorrect and empty scanner inventories."""
+
import copy
import importlib.util
-from pathlib import Path
-import sys
-import subprocess
-import tempfile
import json
+import subprocess
+import sys
+import tempfile
import unittest
-
+from pathlib import Path
+from typing import Any, cast, override
sys.dont_write_bytecode = True
spec = importlib.util.spec_from_file_location("inventory", Path(__file__).with_name("check-maven-inventory.py"))
+assert spec is not None and spec.loader is not None
inventory = importlib.util.module_from_spec(spec)
spec.loader.exec_module(inventory)
class InventoryTests(unittest.TestCase):
- def setUp(self):
- self.expected = {(name, "2.4.240" if name == "com.h2database:h2" else "1.0") for name in inventory.REQUIRED} | {("org.jspecify:jspecify", "1.0.0")}
- self.text = "The following files have been resolved:\n" + "\n".join(
- f" {name}:jar:{version}:runtime\x1b[36m -- module example\x1b[m" for name, version in self.expected)
- self.bom = {"bomFormat": "CycloneDX", "components": [
- {"purl": "pkg:maven/" + name.replace(":", "/") + "@" + version + "?type=jar"}
- for name, version in self.expected]}
- self.trivy = {"Results": [{"Packages": [{"Name": n, "Version": v} for n, v in self.expected], "Vulnerabilities": []}]}
+ """Exercise selected dependency coverage using isolated synthetic reports."""
- def test_complete_graph_without_findings_passes(self):
- self.assertEqual(self.expected, inventory.resolved(self.text))
- self.assertEqual(6, inventory.check(self.expected, self.bom, self.trivy))
+ @override
+ def setUp(self) -> None:
+ """Create isolated synthetic evidence with cleanup after each test."""
+ self.expected = {(name, "2.4.240" if name == "com.h2database:h2" else "1.0") for name in inventory.REQUIRED} | {("org.jspecify:jspecify", "1.0.0")}
+ self.text = "The following files have been resolved:\n" + "\n".join(f" {name}:jar:{version}:runtime\x1b[36m -- module example\x1b[m" for name, version in self.expected)
+ self.bom: dict[str, Any] = {"bomFormat": "CycloneDX", "components": [{"purl": "pkg:maven/" + name.replace(":", "/") + "@" + version + "?type=jar"} for name, version in self.expected]}
+ self.trivy: dict[str, Any] = {"Results": [{"Packages": [{"Name": n, "Version": v} for n, v in self.expected], "Vulnerabilities": []}]}
- def test_missing_or_unreviewed_h2_version_fails(self):
- without_h2 = "\n".join(line for line in self.text.splitlines() if "com.h2database:h2:" not in line)
- for text in (without_h2, self.text.replace("2.4.240", "2.3.232"), self.text.replace("2.4.240", "2.4.241"),
- self.text + "\ncom.h2database:h2:jar:2.3.232:runtime"):
- with self.subTest(text=text), self.assertRaises(ValueError):
- inventory.resolved(text)
+ def test_complete_graph_without_findings_passes(self) -> None:
+ """Verify complete graph without findings passes."""
+ self.assertEqual(self.expected, inventory.resolved(self.text))
+ self.assertEqual(6, inventory.check(self.expected, self.bom, self.trivy))
- def test_missing_frontend_and_empty_maven_fail(self):
- for text in ("", "\n".join(line for line in self.text.splitlines() if "qqq-frontend-next" not in line)):
- with self.assertRaises(ValueError):
- inventory.resolved(text)
+ def test_missing_or_unreviewed_h2_version_fails(self) -> None:
+ """Verify missing or unreviewed h2 version fails."""
+ without_h2 = "\n".join(line for line in self.text.splitlines() if "com.h2database:h2:" not in line)
+ for text in (without_h2, self.text.replace("2.4.240", "2.3.232"), self.text.replace("2.4.240", "2.4.241"), self.text + "\ncom.h2database:h2:jar:2.3.232:runtime"):
+ with self.subTest(text=text), self.assertRaises(ValueError):
+ inventory.resolved(text)
- def test_test_scope_or_test_jar_fails(self):
- for extra in ("example:test:jar:1:test", "org.junit.jupiter:junit-jupiter:jar:5:runtime"):
- with self.assertRaises(ValueError):
- inventory.resolved(self.text + "\n" + extra)
+ def test_missing_frontend_and_empty_maven_fail(self) -> None:
+ """Verify missing frontend and empty maven fail."""
+ for text in ("", "\n".join(line for line in self.text.splitlines() if "qqq-frontend-next" not in line)):
+ with self.assertRaises(ValueError):
+ inventory.resolved(text)
- def test_empty_malformed_or_missing_component_fails(self):
- for bom in ({}, {"bomFormat": "CycloneDX", "components": []}, {"bomFormat": "CycloneDX", "components": [{}]},
- {"bomFormat": "CycloneDX", "components": self.bom["components"][1:]}):
- with self.assertRaises(ValueError):
- inventory.check(self.expected, bom)
+ def test_test_scope_or_test_jar_fails(self) -> None:
+ """Verify test scope or test jar fails."""
+ for extra in ("example:test:jar:1:test", "org.junit.jupiter:junit-jupiter:jar:5:runtime"):
+ with self.assertRaises(ValueError):
+ inventory.resolved(self.text + "\n" + extra)
- def test_wrong_selected_version_fails(self):
- bad = copy.deepcopy(self.trivy)
- for package in bad["Results"][0]["Packages"]:
- if package["Name"] == "org.jspecify:jspecify":
- package["Version"] = "1.0.1"
- with self.assertRaises(ValueError):
- inventory.check(self.expected, self.bom, bad)
+ def test_empty_malformed_or_missing_component_fails(self) -> None:
+ """Verify empty malformed or missing component fails."""
+ for bom in cast(tuple[object, ...], ({}, {"bomFormat": "CycloneDX", "components": []}, {"bomFormat": "CycloneDX", "components": [{}]}, {"bomFormat": "CycloneDX", "components": self.bom["components"][1:]})):
+ with self.assertRaises(ValueError):
+ inventory.check(self.expected, bom)
- def test_empty_or_dropped_trivy_package_fails(self):
- for report in (None, {}, {"Results": []}, {"Results": [{"Packages": self.trivy["Results"][0]["Packages"][1:]}]}):
- with self.assertRaises(ValueError):
- inventory.check(self.expected, self.bom, report)
+ def test_wrong_selected_version_fails(self) -> None:
+ """Verify wrong selected version fails."""
+ bad = copy.deepcopy(self.trivy)
+ for package in bad["Results"][0]["Packages"]:
+ if package["Name"] == "org.jspecify:jspecify":
+ package["Version"] = "1.0.1"
+ with self.assertRaises(ValueError):
+ inventory.check(self.expected, self.bom, bad)
- def test_cli_supplied_null_trivy_fails(self):
- with tempfile.TemporaryDirectory() as temporary:
- directory = Path(temporary)
- (directory / "resolved.txt").write_text(self.text, encoding="utf-8")
- (directory / "bom.json").write_text(json.dumps(self.bom), encoding="utf-8")
- (directory / "trivy.json").write_text("null", encoding="utf-8")
- result = subprocess.run([sys.executable, str(Path(inventory.__file__)),
- str(directory / "resolved.txt"), str(directory / "bom.json"),
- str(directory / "trivy.json")], capture_output=True, text=True, check=False)
- self.assertNotEqual(0, result.returncode)
- self.assertIn("Expected Trivy report object", result.stderr)
+ def test_empty_or_dropped_trivy_package_fails(self) -> None:
+ """Verify empty or dropped trivy package fails."""
+ for report in cast(tuple[object, ...], (None, {}, {"Results": []}, {"Results": [{"Packages": self.trivy["Results"][0]["Packages"][1:]}]})):
+ with self.assertRaises(ValueError):
+ inventory.check(self.expected, self.bom, report)
- def osv(self):
- return {"results": [{"source": {"path": "/src/printing.cdx.json", "type": "sbom"}, "packages": [
- {"package": {"name": name.split(":", 1)[1], "version": version, "ecosystem": "Maven"}}
- for name, version in self.expected]}]}
+ def test_cli_supplied_null_trivy_fails(self) -> None:
+ """Verify cli supplied null trivy fails."""
+ with tempfile.TemporaryDirectory() as temporary:
+ directory = Path(temporary)
+ (directory / "resolved.txt").write_text(self.text, encoding="utf-8")
+ (directory / "bom.json").write_text(json.dumps(self.bom), encoding="utf-8")
+ (directory / "trivy.json").write_text("null", encoding="utf-8")
+ assert inventory.__file__ is not None
+ result = subprocess.run([sys.executable, str(Path(inventory.__file__)), str(directory / "resolved.txt"), str(directory / "bom.json"), str(directory / "trivy.json")], capture_output=True, text=True, check=False) # noqa: S603 -- Execute this interpreter and the fixed sibling guard using synthetic paths, without a shell.
+ self.assertNotEqual(0, result.returncode)
+ self.assertIn("Expected Trivy report object", result.stderr)
- def test_osv_all_clean_packages_pass_without_vulnerabilities(self):
- self.assertEqual(6, inventory.check_osv(self.expected, self.osv()))
+ def osv(self) -> dict[str, Any]:
+ """Build a clean OSV fixture whose identity matches the selected graph."""
+ return {"results": [{"source": {"path": "/src/printing.cdx.json", "type": "sbom"}, "packages": [{"package": {"name": name.split(":", 1)[1], "version": version, "ecosystem": "Maven"}} for name, version in self.expected]}]}
- def test_osv_missing_malformed_or_partial_evidence_fails(self):
- partial = self.osv()
- partial["results"][0]["packages"].pop()
- wrong = self.osv()
- wrong["results"][0]["packages"][0]["package"]["version"] = "wrong"
- duplicate = self.osv()
- duplicate["results"][0]["packages"].append(duplicate["results"][0]["packages"][0])
- for report in (None, {}, [], {"results": None}, {"results": []}, {"results": [{}]}, partial, wrong, duplicate):
- with self.subTest(report=report), self.assertRaises(ValueError):
- inventory.check_osv(self.expected, report)
+ def test_osv_all_clean_packages_pass_without_vulnerabilities(self) -> None:
+ """Verify osv all clean packages pass without vulnerabilities."""
+ self.assertEqual(6, inventory.check_osv(self.expected, self.osv()))
- def test_osv_ambiguous_group_projection_fails(self):
- with self.assertRaises(ValueError):
- inventory.check_osv(self.expected | {("other:jspecify", "1.0.0")}, self.osv())
+ def test_osv_missing_malformed_or_partial_evidence_fails(self) -> None:
+ """Verify osv missing malformed or partial evidence fails."""
+ partial = self.osv()
+ partial["results"][0]["packages"].pop()
+ wrong = self.osv()
+ wrong["results"][0]["packages"][0]["package"]["version"] = "wrong"
+ duplicate = self.osv()
+ duplicate["results"][0]["packages"].append(duplicate["results"][0]["packages"][0])
+ for report in cast(tuple[object, ...], (None, {}, [], {"results": None}, {"results": []}, {"results": [{}]}, partial, wrong, duplicate)):
+ with self.subTest(report=report), self.assertRaises(ValueError):
+ inventory.check_osv(self.expected, report)
+
+ def test_osv_ambiguous_group_projection_fails(self) -> None:
+ """Verify osv ambiguous group projection fails."""
+ with self.assertRaises(ValueError):
+ inventory.check_osv(self.expected | {("other:jspecify", "1.0.0")}, self.osv())
if __name__ == "__main__":
- unittest.main()
+ unittest.main()
diff --git a/.github/scripts/test-scan-gates.py b/.github/scripts/test-scan-gates.py
index c8494eb..2b173dd 100644
--- a/.github/scripts/test-scan-gates.py
+++ b/.github/scripts/test-scan-gates.py
@@ -2,91 +2,106 @@
# SPDX-License-Identifier: Apache-2.0
"""Exercise the exact CI guards using synthetic evidence only."""
+
import contextlib
import copy
-import io
import importlib.util
+import io
import json
-from pathlib import Path
import sys
import tempfile
import unittest
-
+from pathlib import Path
+from types import ModuleType
+from typing import Any, override
sys.dont_write_bytecode = True
-def load(name):
- spec = importlib.util.spec_from_file_location(name, Path(__file__).with_name(name + ".py"))
- module = importlib.util.module_from_spec(spec)
- spec.loader.exec_module(module)
- return module
+def load(name: str) -> ModuleType:
+ """Load the real sibling guard for synthetic evidence tests."""
+ spec = importlib.util.spec_from_file_location(name, Path(__file__).with_name(name + ".py"))
+ assert spec is not None and spec.loader is not None
+ module = importlib.util.module_from_spec(spec)
+ spec.loader.exec_module(module)
+ return module
codeql = load("check-codeql-sarif")
class SarifTests(unittest.TestCase):
- def setUp(self):
- self.temporary = tempfile.TemporaryDirectory()
- self.addCleanup(self.temporary.cleanup)
- self.directory = Path(self.temporary.name)
- self.report = {"version": "2.1.0", "runs": [{"tool": {"driver": {"name": "CodeQL", "rules": [{"id": "synthetic", "properties": {"security-severity": "8.0"}}]}}, "results": []}]}
+ """Reject incomplete SARIF evidence and preserve severity enforcement."""
- def save(self, report):
- (self.directory / "java.sarif").write_text(json.dumps(report), encoding="utf-8")
+ @override
+ def setUp(self) -> None:
+ """Create isolated synthetic evidence with cleanup after each test."""
+ self.temporary = tempfile.TemporaryDirectory()
+ self.addCleanup(self.temporary.cleanup)
+ self.directory = Path(self.temporary.name)
+ self.report: dict[str, Any] = {"version": "2.1.0", "runs": [{"tool": {"driver": {"name": "CodeQL", "rules": [{"id": "synthetic", "properties": {"security-severity": "8.0"}}]}}, "results": []}]}
- def test_clean_report_with_zero_findings_passes(self):
- self.save(self.report)
- self.assertFalse(codeql.check(self.directory, "java-kotlin", 7))
+ def save(self, report: object) -> None:
+ """Write the test body as SARIF without validating it first."""
+ (self.directory / "java.sarif").write_text(json.dumps(report), encoding="utf-8")
- def test_missing_report_or_wrong_language_fails(self):
- with self.assertRaises(ValueError):
- codeql.check(self.directory, "java-kotlin", 7)
- self.save(self.report)
- with self.assertRaises(ValueError):
- codeql.check(self.directory, "csharp", 7)
+ def test_clean_report_with_zero_findings_passes(self) -> None:
+ """Verify clean report with zero findings passes."""
+ self.save(self.report)
+ self.assertFalse(codeql.check(self.directory, "java-kotlin", 7))
- def test_empty_or_malformed_report_fails(self):
- for text in ["", "{", "null", "[]"]:
- with self.subTest(text=text):
- (self.directory / "java.sarif").write_text(text)
- with self.assertRaises((ValueError, AttributeError)):
- codeql.check(self.directory, "java-kotlin", 7)
+ def test_missing_report_or_wrong_language_fails(self) -> None:
+ """Verify missing report or wrong language fails."""
+ with self.assertRaises(ValueError):
+ codeql.check(self.directory, "java-kotlin", 7)
+ self.save(self.report)
+ with self.assertRaises(ValueError):
+ codeql.check(self.directory, "csharp", 7)
- def test_missing_runs_tool_rules_or_results_fails(self):
- variants = [{"version": "2.1.0"}, {"version": "2.1.0", "runs": []}]
- for field in ["tool", "results"]:
- bad = copy.deepcopy(self.report)
- del bad["runs"][0][field]
- variants.append(bad)
- bad = copy.deepcopy(self.report)
- bad["runs"][0]["tool"]["driver"]["rules"] = []
- variants.append(bad)
- for bad in variants:
- with self.subTest(report=bad):
- self.save(bad)
- with self.assertRaises(ValueError):
- codeql.check(self.directory, "java-kotlin", 7)
+ def test_empty_or_malformed_report_fails(self) -> None:
+ """Verify empty or malformed report fails."""
+ for text in ["", "{", "null", "[]"]:
+ with self.subTest(text=text):
+ (self.directory / "java.sarif").write_text(text)
+ with self.assertRaises((ValueError, AttributeError)):
+ codeql.check(self.directory, "java-kotlin", 7)
- def test_high_finding_blocks_and_unknown_rule_fails(self):
- self.report["runs"][0]["results"] = [{"ruleId": "synthetic"}]
- self.save(self.report)
- with contextlib.redirect_stdout(io.StringIO()):
- self.assertTrue(codeql.check(self.directory, "java-kotlin", 7))
- self.report["runs"][0]["results"] = [{"ruleId": "missing"}]
- self.save(self.report)
- with self.assertRaises(ValueError):
- codeql.check(self.directory, "java-kotlin", 7)
+ def test_missing_runs_tool_rules_or_results_fails(self) -> None:
+ """Verify missing runs tool rules or results fails."""
+ variants: list[dict[str, Any]] = [{"version": "2.1.0"}, {"version": "2.1.0", "runs": []}]
+ for field in ["tool", "results"]:
+ bad = copy.deepcopy(self.report)
+ del bad["runs"][0][field]
+ variants.append(bad)
+ bad = copy.deepcopy(self.report)
+ bad["runs"][0]["tool"]["driver"]["rules"] = []
+ variants.append(bad)
+ for bad in variants:
+ with self.subTest(report=bad):
+ self.save(bad)
+ with self.assertRaises(ValueError):
+ codeql.check(self.directory, "java-kotlin", 7)
- def test_failed_invocation_and_invalid_threshold_fail(self):
- self.report["runs"][0]["invocations"] = [{"executionSuccessful": False}]
- self.save(self.report)
- with self.assertRaises(ValueError):
- codeql.check(self.directory, "java-kotlin", 7)
- with self.assertRaises(ValueError):
- codeql.check(self.directory, "java-kotlin", float("nan"))
+ def test_high_finding_blocks_and_unknown_rule_fails(self) -> None:
+ """Verify high finding blocks and unknown rule fails."""
+ self.report["runs"][0]["results"] = [{"ruleId": "synthetic"}]
+ self.save(self.report)
+ with contextlib.redirect_stdout(io.StringIO()):
+ self.assertTrue(codeql.check(self.directory, "java-kotlin", 7))
+ self.report["runs"][0]["results"] = [{"ruleId": "missing"}]
+ self.save(self.report)
+ with self.assertRaises(ValueError):
+ codeql.check(self.directory, "java-kotlin", 7)
+
+ def test_failed_invocation_and_invalid_threshold_fail(self) -> None:
+ """Verify failed invocation and invalid threshold fail."""
+ self.report["runs"][0]["invocations"] = [{"executionSuccessful": False}]
+ self.save(self.report)
+ with self.assertRaises(ValueError):
+ codeql.check(self.directory, "java-kotlin", 7)
+ with self.assertRaises(ValueError):
+ codeql.check(self.directory, "java-kotlin", float("nan"))
if __name__ == "__main__":
- unittest.main()
+ unittest.main()
diff --git a/.github/scripts/test-script-quality.py b/.github/scripts/test-script-quality.py
new file mode 100644
index 0000000..71a12d9
--- /dev/null
+++ b/.github/scripts/test-script-quality.py
@@ -0,0 +1,238 @@
+# Copyright 2026 Kingsrook, LLC
+# SPDX-License-Identifier: Apache-2.0
+
+"""Prove the quality gates reject real violations, using private temporary fixtures and the locked tools."""
+
+import importlib.util
+import json
+import shutil
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from types import ModuleType
+from typing import Any, override
+from unittest import mock
+
+sys.dont_write_bytecode = True
+
+ROOT = Path(__file__).resolve().parents[2]
+GOOD_PYTHON = '"""Probe module."""\n\n\ndef add(left: int, right: int) -> int:\n """Add two integers."""\n return left + right\n'
+UNUSED_IMPORT_PYTHON = '"""Probe module."""\n\nimport os\n\n\ndef add(left: int, right: int) -> int:\n """Add two integers."""\n return left + right\n'
+TWO_SPACE_PYTHON = '"""Probe module."""\n\n\ndef add(left: int, right: int) -> int:\n """Add two integers."""\n return left + right\n'
+INCOMPATIBLE_PYTHON = '"""Probe module."""\n\nVALUE: int = "text"\n'
+UNANNOTATED_PYTHON = '"""Probe module."""\n\n\ndef echo(value):\n """Return the value."""\n return value\n'
+GOOD_JAVASCRIPT = "export function addNumbers(left, right)\n{\n return left + right;\n}\n"
+BAD_LAYOUT_JAVASCRIPT = "export function addNumbers(left, right) {\n return left + right\n}\n"
+GOOD_TYPED_JAVASCRIPT = "/** @type {number} */\nexport const value = 1;\n"
+INCOMPATIBLE_JAVASCRIPT = '/** @type {number} */\nexport const value = "text";\n'
+
+
+def load(name: str) -> ModuleType:
+ """Load the real sibling guard for tests."""
+ spec = importlib.util.spec_from_file_location(name, Path(__file__).with_name(name + ".py"))
+ assert spec is not None and spec.loader is not None
+ module = importlib.util.module_from_spec(spec)
+ spec.loader.exec_module(module)
+ return module
+
+
+guard = load("check-script-quality")
+
+
+def run(command: list[str], cwd: Path, stdin: str | None = None) -> subprocess.CompletedProcess[str]:
+ """Run one real tool invocation without a shell and capture its diagnostics."""
+ return subprocess.run(command, cwd=cwd, input=stdin, check=False, capture_output=True, text=True, encoding="utf-8") # noqa: S603 -- Fixed tool paths and private temporary arguments; no shell.
+
+
+def ruff(arguments: list[str], source: Path) -> subprocess.CompletedProcess[str]:
+ """Run the locked Ruff with the repository configuration on one private fixture."""
+ return run([sys.executable, "-m", "ruff", *arguments, "--no-cache", "--config", str(ROOT / "ruff.toml"), str(source)], source.parent)
+
+
+class ScopeTests(unittest.TestCase):
+ """Reject tracked sources that fall outside the explicit tool roots."""
+
+ def test_python_outside_script_root_is_rejected(self) -> None:
+ """Reject a tracked Python file outside the script root instead of letting it disappear."""
+ for stray in ("tools/helper.py", ".github/scripts/nested/helper.py", "helper.pyi"):
+ with self.subTest(stray=stray), self.assertRaisesRegex(ValueError, "outside the checked script root"):
+ guard.python_scope([".github/scripts/check-a.py", stray])
+
+ def test_python_scope_returns_sorted_script_root_and_rejects_empty(self) -> None:
+ """Select only the direct script root, sorted, and reject selecting nothing."""
+ self.assertEqual([".github/scripts/a.py", ".github/scripts/b.py"], guard.python_scope([".github/scripts/b.py", "README.md", ".github/scripts/a.py"]))
+ with self.assertRaisesRegex(ValueError, "No owned Python"):
+ guard.python_scope(["README.md"])
+
+ def test_javascript_requires_tsconfig_include_scope(self) -> None:
+ """Reject tracked JavaScript outside tsconfig include patterns, including .js/.cjs/.ts."""
+ with tempfile.TemporaryDirectory() as name:
+ root = Path(name)
+ (root / "tsconfig.json").write_text(json.dumps({"include": ["tools/**/*.mjs", "eslint.config.mjs"]}), encoding="utf-8")
+ self.assertEqual(["eslint.config.mjs", "tools/a/b.mjs", "tools/c.mjs"], guard.javascript_scope(root, ["tools/c.mjs", "tools/a/b.mjs", "eslint.config.mjs", "README.md"]))
+ for stray in ("src/z.js", "tools/w.cjs", "tools/v.ts", "tools/u.js"):
+ with self.subTest(stray=stray), self.assertRaisesRegex(ValueError, "outside the tsconfig include list"):
+ guard.javascript_scope(root, ["tools/c.mjs", stray])
+ with self.assertRaisesRegex(ValueError, "No owned JavaScript"):
+ guard.javascript_scope(root, ["README.md"])
+
+ def test_javascript_without_tsconfig_is_rejected(self) -> None:
+ """Reject JavaScript in a repository that has no tsconfig."""
+ with tempfile.TemporaryDirectory() as name:
+ root = Path(name)
+ self.assertEqual([], guard.javascript_scope(root, ["README.md"]))
+ with self.assertRaisesRegex(ValueError, "without a tsconfig.json"):
+ guard.javascript_scope(root, ["tools/a.mjs"])
+
+
+class PyrightCountTests(unittest.TestCase):
+ """Reject Pyright reports that analyzed too few files or found anything."""
+
+ @staticmethod
+ def report(analyzed: int, errors: int = 0, warnings: int = 0) -> dict[str, Any]:
+ """Build the exact Pyright JSON shape consumed by the guard."""
+ return {"summary": {"filesAnalyzed": analyzed, "errorCount": errors, "warningCount": warnings}, "generalDiagnostics": [{"file": "x.py", "message": "finding"}] * (errors + warnings), "exitCode": 1 if errors or warnings else 0}
+
+ def test_nonzero_exit_fails_even_with_clean_summary(self) -> None:
+ """Fail a nonzero Pyright exit even when every count is clean."""
+ clean = self.report(4)
+ clean["exitCode"] = 1
+ with self.assertRaisesRegex(ValueError, "exited 1"):
+ guard.check_pyright_report(clean, 4)
+
+ def test_count_must_match_selected_sources(self) -> None:
+ """Fail zero or mismatched analyzed counts even with no diagnostics."""
+ guard.check_pyright_report(self.report(4), 4)
+ for analyzed in (0, 3, 5):
+ with self.subTest(analyzed=analyzed), self.assertRaisesRegex(ValueError, "analyzed"):
+ guard.check_pyright_report(self.report(analyzed), 4)
+
+ def test_errors_and_warnings_fail(self) -> None:
+ """Fail on errors and on warnings at the exact count."""
+ for report in (self.report(4, errors=1), self.report(4, warnings=1)):
+ with self.subTest(report=report), self.assertRaisesRegex(ValueError, "reported findings"):
+ guard.check_pyright_report(report, 4)
+
+
+class RuffProbeTests(unittest.TestCase):
+ """Run the real locked Ruff on bounded private fixtures."""
+
+ def test_compliant_fixture_passes_and_violations_fail(self) -> None:
+ """Pass a compliant control and reject unused imports and wrong indentation."""
+ with tempfile.TemporaryDirectory() as name:
+ probe = Path(name) / "probe.py"
+ probe.write_text(GOOD_PYTHON, encoding="utf-8", newline="\n")
+ self.assertEqual(0, ruff(["check"], probe).returncode)
+ self.assertEqual(0, ruff(["format", "--check"], probe).returncode)
+ probe.write_text(UNUSED_IMPORT_PYTHON, encoding="utf-8", newline="\n")
+ unused = ruff(["check", "--output-format", "concise"], probe)
+ self.assertNotEqual(0, unused.returncode)
+ self.assertIn("F401", unused.stdout)
+ probe.write_text(TWO_SPACE_PYTHON, encoding="utf-8", newline="\n")
+ indented = ruff(["format", "--check"], probe)
+ self.assertNotEqual(0, indented.returncode)
+ self.assertIn("would be reformatted", (indented.stdout + indented.stderr).lower())
+
+
+class PyrightProbeTests(unittest.TestCase):
+ """Run the real locked Pyright with the repository strict settings on private projects."""
+
+ @staticmethod
+ def project(root: Path, include: list[str] | None = None) -> None:
+ """Write the repository Pyright config, without its venv location, into a private project."""
+ config: dict[str, Any] = json.loads((ROOT / ".github/pyrightconfig.json").read_text(encoding="utf-8"))
+ del config["venvPath"], config["venv"]
+ if include is not None:
+ config["include"] = include
+ (root / "pyrightconfig.json").write_text(json.dumps(config), encoding="utf-8")
+
+ def test_compliant_control_passes_and_strict_violations_fail(self) -> None:
+ """Analyze the control once with no findings and reject incompatible and unannotated code."""
+ with tempfile.TemporaryDirectory() as name:
+ root = Path(name)
+ self.project(root)
+ scripts = root / "scripts"
+ scripts.mkdir()
+ probe = scripts / "probe.py"
+ probe.write_text(GOOD_PYTHON, encoding="utf-8", newline="\n")
+ clean = guard.run_pyright(root, ".")
+ self.assertEqual(0, clean["exitCode"])
+ guard.check_pyright_report(clean, 1)
+ clean["exitCode"] = 1
+ with self.assertRaisesRegex(ValueError, "exited 1"):
+ guard.check_pyright_report(clean, 1)
+ probe.write_text(INCOMPATIBLE_PYTHON, encoding="utf-8", newline="\n")
+ report = guard.run_pyright(root, ".")
+ self.assertIn("reportAssignmentType", {item.get("rule") for item in report["generalDiagnostics"]})
+ with self.assertRaisesRegex(ValueError, "reported findings"):
+ guard.check_pyright_report(report, 1)
+ probe.write_text(UNANNOTATED_PYTHON, encoding="utf-8", newline="\n")
+ strict = guard.run_pyright(root, ".")
+ self.assertIn("reportMissingParameterType", {item.get("rule") for item in strict["generalDiagnostics"]})
+
+ def test_missing_locked_pyright_or_node_is_rejected(self) -> None:
+ """Reject a missing locked Pyright entry or Node instead of silently skipping the type gate."""
+ with tempfile.TemporaryDirectory() as name:
+ root = Path(name)
+ with mock.patch.object(guard, "PYRIGHT_ENTRY", root / "missing.js"), self.assertRaisesRegex(ValueError, "Locked Pyright is missing"):
+ guard.run_pyright(root, ".")
+ with mock.patch.object(guard.shutil, "which", return_value=None), self.assertRaisesRegex(ValueError, "Node.js is required"):
+ guard.run_pyright(root, ".")
+
+ def test_zero_analyzed_files_cannot_pass(self) -> None:
+ """Reject an empty include and a hidden-directory root config in the wrapper."""
+ with tempfile.TemporaryDirectory() as name:
+ root = Path(name)
+ scripts = root / ".github" / "scripts"
+ scripts.mkdir(parents=True)
+ (scripts / "probe.py").write_text(GOOD_PYTHON, encoding="utf-8", newline="\n")
+ for include in (["nothing/*.py"], [".github/scripts/*.py"]):
+ self.project(root, include)
+ with self.subTest(include=include), self.assertRaises(ValueError):
+ guard.check_pyright_report(guard.run_pyright(root, "."), 1)
+
+
+class JavaScriptProbeTests(unittest.TestCase):
+ """Run the real locked ESLint and TypeScript on bounded private input."""
+
+ @override
+ def setUp(self) -> None:
+ """Skip only where tsconfig is absent, because the scope guard forbids JavaScript there."""
+ if not (ROOT / "tsconfig.json").is_file():
+ self.skipTest("No JavaScript tooling in this repository")
+ node = shutil.which("node")
+ self.assertIsNotNone(node)
+ self.node = str(node)
+
+ def eslint(self, text: str) -> subprocess.CompletedProcess[str]:
+ """Lint text as an in-scope tool file using the repository configuration."""
+ return run([self.node, str(ROOT / "node_modules/eslint/bin/eslint.js"), "--stdin", "--stdin-filename", "tools/probe.mjs", "--max-warnings", "0", "--format", "json"], ROOT, text)
+
+ def test_eslint_rejects_wrong_layout(self) -> None:
+ """Pass a compliant control and reject K&R braces, two-space indent and a missing semicolon."""
+ good = self.eslint(GOOD_JAVASCRIPT)
+ self.assertEqual(0, good.returncode, good.stdout + good.stderr)
+ bad = self.eslint(BAD_LAYOUT_JAVASCRIPT)
+ self.assertNotEqual(0, bad.returncode)
+ rules = {message["ruleId"] for message in json.loads(bad.stdout)[0]["messages"]}
+ self.assertLessEqual({"@stylistic/brace-style", "@stylistic/semi", "@stylistic/indent"}, rules)
+
+ def test_checked_javascript_rejects_incompatible_assignment(self) -> None:
+ """Reject a real incompatible assignment under the repository strict checkJs settings."""
+ with tempfile.TemporaryDirectory() as name:
+ root = Path(name)
+ (root / "tsconfig.json").write_text(json.dumps({"extends": (ROOT / "tsconfig.json").as_posix(), "compilerOptions": {"typeRoots": [(ROOT / "node_modules/@types").as_posix()]}, "include": ["probe.mjs"]}), encoding="utf-8")
+ tsc = [self.node, str(ROOT / "node_modules/typescript/bin/tsc"), "-p", str(root / "tsconfig.json")]
+ (root / "probe.mjs").write_text(GOOD_TYPED_JAVASCRIPT, encoding="utf-8", newline="\n")
+ good = run(tsc, root)
+ self.assertEqual(0, good.returncode, good.stdout + good.stderr)
+ (root / "probe.mjs").write_text(INCOMPATIBLE_JAVASCRIPT, encoding="utf-8", newline="\n")
+ bad = run(tsc, root)
+ self.assertNotEqual(0, bad.returncode)
+ self.assertIn("TS2322", bad.stdout)
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/.github/scripts/test-source-headers.py b/.github/scripts/test-source-headers.py
index 8a04b44..20102c8 100644
--- a/.github/scripts/test-source-headers.py
+++ b/.github/scripts/test-source-headers.py
@@ -11,7 +11,6 @@ import unittest
from pathlib import Path
from typing import Any
-
SPEC = importlib.util.spec_from_file_location("headers", Path(__file__).with_name("check-source-headers.py"))
assert SPEC is not None and SPEC.loader is not None
HEADERS = importlib.util.module_from_spec(SPEC)
@@ -32,8 +31,7 @@ limitations under the License."""
def fixture_policy(holder: str = "James Maes", license_id: str = "MIT") -> dict[str, Any]:
"""Use the production styles with a small explicit owned fixture scope."""
- return {"includes": ["src/*"], "exclusions": [], "styles": POLICY["styles"],
- "notices": [{"holder": holder, "license": license_id, "paths": ["src/*"]}]}
+ return {"includes": ["src/*"], "exclusions": [], "styles": POLICY["styles"], "notices": [{"holder": holder, "license": license_id, "paths": ["src/*"]}]}
class HeaderTests(unittest.TestCase):
@@ -51,12 +49,23 @@ class HeaderTests(unittest.TestCase):
with self.subTest(name=name):
HEADERS.check_notice(name, text, fixture_policy())
+ def test_year_range_separator_is_hyphen_or_en_dash_only(self) -> None:
+ """Accept a hyphen or en dash between years and reject any other separator."""
+ for separator, accepted in (("-", True), (chr(0x2013), True), ("/", False), (chr(0x2014), False)):
+ text = f"# Copyright 2025{separator}2026 James Maes\n# SPDX-License-Identifier: MIT"
+ with self.subTest(separator=separator):
+ if accepted:
+ HEADERS.check_notice("src/a.py", text, fixture_policy())
+ else:
+ with self.assertRaises(ValueError):
+ HEADERS.check_notice("src/a.py", text, fixture_policy())
+
def test_required_first_lines_and_bom(self) -> None:
"""Accept shebangs, PowerShell requirements and XML declarations first."""
for name, prefix, body in (
("src/a.sh", "#!/bin/sh\n", "# " + MIT.replace("\n", "\n# ")),
("src/a.ps1", "#Requires -Version 7.4\n", "# " + MIT.replace("\n", "\n# ")),
- ("src/a.xml", "\n", ""),
+ ("src/a.xml", '\n', ""),
("src/a.mjs", "#!/usr/bin/env node\n", "/*" + MIT + "*/"),
):
with self.subTest(name=name):
@@ -67,21 +76,20 @@ class HeaderTests(unittest.TestCase):
for name, text in (
("src/a.sh", "# " + MIT.replace("\n", "\n# ") + "\n#!/bin/sh"),
("src/a.ps1", "# " + MIT.replace("\n", "\n# ") + "\n#Requires -Version 7.4"),
- ("src/a.xml", ""),
+ ("src/a.xml", "'),
):
with self.subTest(name=name), self.assertRaises(ValueError):
HEADERS.check_notice(name, text, fixture_policy())
def test_code_strings_and_late_comments_do_not_count(self) -> None:
"""Reject docstrings and comments after executable declarations."""
- for text in ("notice = \"" + MIT.replace("\n", "\\n") + "\"",
- "\"\"\"" + MIT + "\"\"\"", "import os\n# " + MIT.replace("\n", "\n# ")):
+ for text in ('notice = "' + MIT.replace("\n", "\\n") + '"', '"""' + MIT + '"""', "import os\n# " + MIT.replace("\n", "\n# ")):
with self.subTest(text=text), self.assertRaises(ValueError):
HEADERS.check_notice("src/a.py", text, fixture_policy())
def test_wrong_syntax_and_unterminated_comments_fail(self) -> None:
"""Notice text alone is insufficient without a complete language comment."""
- for text in ("# " + MIT, "/*" + MIT, "string text = \"" + MIT + "\";"):
+ for text in ("# " + MIT, "/*" + MIT, 'string text = "' + MIT + '";'):
with self.subTest(text=text), self.assertRaises(ValueError):
HEADERS.check_notice("src/a.cs", text, fixture_policy())
with self.assertRaises(ValueError):
@@ -89,8 +97,7 @@ class HeaderTests(unittest.TestCase):
def test_wrong_holder_or_license_and_conflicting_spdx_fail(self) -> None:
"""Do not conflate repository licenses or accept conflicting identifiers."""
- for text in (MIT.replace("James Maes", "Someone Else"), MIT.replace("MIT", "Apache-2.0"),
- MIT + "\nSPDX-License-Identifier: Apache-2.0", MIT.splitlines()[0]):
+ for text in (MIT.replace("James Maes", "Someone Else"), MIT.replace("MIT", "Apache-2.0"), MIT + "\nSPDX-License-Identifier: Apache-2.0", MIT.splitlines()[0]):
with self.subTest(text=text), self.assertRaises(ValueError):
HEADERS.check_notice("src/a.cs", "/*" + text + "*/", fixture_policy())
@@ -103,8 +110,7 @@ class HeaderTests(unittest.TestCase):
def test_java_requires_full_apache_block(self) -> None:
"""An abbreviated SPDX-only Java notice cannot replace scoped convention."""
policy = fixture_policy("Kingsrook, LLC", "Apache-2.0")
- for notice in ("Copyright 2026 Kingsrook, LLC\nSPDX-License-Identifier: Apache-2.0",
- APACHE.replace("limitations under the License.", "")):
+ for notice in ("Copyright 2026 Kingsrook, LLC\nSPDX-License-Identifier: Apache-2.0", APACHE.replace("limitations under the License.", "")):
with self.subTest(notice=notice), self.assertRaises(ValueError):
HEADERS.check_notice("src/A.java", "/*" + notice + "*/", policy)
@@ -121,9 +127,7 @@ class HeaderTests(unittest.TestCase):
"""New scripts enter the guard; data, generated and upstream files do not."""
self.assertTrue(HEADERS.selected(".github/scripts/new-tool.py", POLICY))
self.assertTrue(HEADERS.selected(".github/workflows/new-check.yml", POLICY))
- for name in (".mvn/wrapper/MavenWrapperDownloader.java", "mvnw", "schemas/new.json",
- "src/main/resources/openapi/new.yaml", "src/test/resources/openapi/old.yaml",
- "src/generated/Example.cs", "src/Example/obj/Generated.cs", "packages.lock.json"):
+ for name in (".mvn/wrapper/MavenWrapperDownloader.java", "mvnw", "schemas/new.json", "src/main/resources/openapi/new.yaml", "src/test/resources/openapi/old.yaml", "src/generated/Example.cs", "src/Example/obj/Generated.cs", "packages.lock.json"):
with self.subTest(name=name):
self.assertFalse(HEADERS.selected(name, POLICY))
self.assertTrue(all(item["reason"].strip() for item in POLICY["exclusions"]))
@@ -140,11 +144,11 @@ class HeaderTests(unittest.TestCase):
"""A tracked new source fails while untracked generated output is ignored."""
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
- subprocess.run(["git", "init", "-q", str(root)], check=True)
+ subprocess.run(["git", "init", "-q", str(root)], check=True) # noqa: S603, S607 -- Fixed Git command in the owned temporary test directory.
root.joinpath("src").mkdir()
source = root / "src/a.py"
source.write_text("print(1)", encoding="utf-8")
- subprocess.run(["git", "-C", str(root), "add", "--", "src/a.py"], check=True)
+ subprocess.run(["git", "-C", str(root), "add", "--", "src/a.py"], check=True) # noqa: S603, S607 -- Fixed synthetic path; never a shell command.
with self.assertRaisesRegex(ValueError, "src/a.py"):
HEADERS.check_repository(root, fixture_policy())
source.write_text("# " + MIT.replace("\n", "\n# ") + "\nprint(1)", encoding="utf-8")
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 7a41bce..d1dec72 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -29,13 +29,30 @@ jobs:
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
with:
repository: QRun-IO/qqq-print-protocol
- ref: f3cbc02e9b0fb4fdfefaa59d976c04df747cc589
+ ref: a26c18c90c7d6102bbe26eb50cbb90fcd1bd46e3
path: qqq-print-protocol
persist-credentials: false
- uses: actions/setup-java@be666c2fcd27ec809703dec50e508c2fdc7f6654 # v5.2.0
with:
distribution: temurin
java-version: '21'
+ - uses: actions/setup-node@6044e13b5dc448c55e2357c09f80417699197238 # v6.2.0
+ with:
+ node-version: '24'
+ package-manager-cache: false
+ - name: Restore locked Pyright
+ run: npm ci --ignore-scripts --no-audit --no-fund
+ - name: Set up uv
+ uses: astral-sh/setup-uv@07a047e608db84b3d8c06cd9a7c7dcc9c355c18e # v10.3.2
+ with:
+ version: '0.13.6'
+ enable-cache: false
+ - name: Restore locked Python tools
+ run: uv sync --locked
+ - name: Verify script quality gates and rejection paths
+ run: |
+ uv run --locked --no-sync python .github/scripts/test-script-quality.py
+ uv run --locked --no-sync python .github/scripts/check-script-quality.py
- name: Verify owned source notices and rejection paths
run: |
python3 .github/scripts/test-source-headers.py
diff --git a/.github/workflows/security.yml b/.github/workflows/security.yml
index 8da4dcc..626350e 100644
--- a/.github/workflows/security.yml
+++ b/.github/workflows/security.yml
@@ -174,7 +174,7 @@ jobs:
persist-credentials: false
- name: Set up uv
- uses: astral-sh/setup-uv@c18668ad3cf93ea998bef934396af7bb5c839dc7 # v10.2.0
+ uses: astral-sh/setup-uv@07a047e608db84b3d8c06cd9a7c7dcc9c355c18e # v10.3.2
with:
enable-cache: false
diff --git a/.gitignore b/.gitignore
index e47bfdc..3f84bc3 100644
--- a/.gitignore
+++ b/.gitignore
@@ -12,3 +12,9 @@ dist/
*.iml
.DS_Store
Thumbs.db
+
+# Local script-quality environments and tool caches.
+.venv/
+node_modules/
+.ruff_cache/
+__pycache__/
diff --git a/.python-version b/.python-version
new file mode 100644
index 0000000..24ee5b1
--- /dev/null
+++ b/.python-version
@@ -0,0 +1 @@
+3.13
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 469ae87..ecc0deb 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,8 +2,45 @@
## Unreleased (0.4.0-SNAPSHOT)
-Printing extension **2.1.0** (additive; the agent contract is unchanged).
+Printing extension **2.3.0** (additive; wire payload v1 and the agent subprotocol are
+unchanged). 2.1.0 added the record, table-list and text print processes; 2.2.0 added
+`printingRenderToPdf`, the header and footer values on every document and content
+sections; 2.3.0 adds the least-privilege operator catalog/submission/request lookup,
+reported Windows queue-default media (`printer-media-v1`) and the opt-in
+retained-source facade. Clients should offer a feature when its process is in the
+metadata, not by comparing versions. An agent that does not advertise
+`printer-media-v1` stays a three-capability peer and cannot claim a job carrying a
+media selection, whatever extension version it discovers.
+- Realtime shutdown now waits, with a bound, for channel workers that are already
+ running before the host's durable state is released. Previously the executor was
+ only told to stop, so a late drain could query a stopped file database and
+ recreate it empty (found as a temporary-directory cleanup error in a full test
+ run).
+
+Observed locally on the merged candidate: full `-Pui-demo verify` 189 reported,
+188 executed, one intentional opt-in workload skip, zero failures/errors, zero
+Checkstyle violations; five consecutive focused realtime runs passed. Hosted CI on
+the final signed head, publication and consumer qualification have not run; no
+physical output is claimed.
+
+- The app's header and footer on every page of every document (record sheets,
+ table lists, text and content pages): app name, title or breadcrumb, server,
+ user and printed-at at the top; the client's footer (e.g. `qqq ui 1.0.1 ·
+ server`), a list's filter and sort, and "Page n of m" at the bottom. Clients
+ send `headerApp`, `headerTitle`, `headerServer`, `headerUser`, `headerPrintedAt`,
+ `footerLeft` and `footerNote`; missing ones default to the instance branding,
+ the session user, the current instant and "QQQ printing".
+- Content documents: `printingPrintText` and `printingRenderToPdf` (`TEXT`) take
+ `sections` (JSON `[{title, lines}]`), a screen's panels without navigation or
+ borders, rendered as headings with their lines wrapped to the page width.
+- Save as PDF: `printingRenderToPdf` renders the document a print would carry -
+ a record sheet (`kind` `RECORD`, `tableName`, `recordId`), a table list (`LIST`,
+ the `printingPrintTableList` values) or a text page (`TEXT`) - and returns it to
+ the caller as the standard process download (`downloadFileName`,
+ `serverFilePath`) instead of queueing a job. No printer is needed; reads enforce
+ the caller's permissions, and a list over the limit says so in
+ `printingMessage`. Clients offer it as "Save as PDF…" in their printer pickers.
- Print anything: `PrintingRecordProcesses.registerAll(instance)` adds **Print** to
every table that can be read one record at a time (call it after the host's
tables are in the instance). Hidden tables, tables without a primary key or
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index 89a49f8..044c510 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -9,3 +9,14 @@ Changes belong in a pull request with a concise behavior description and actual
Keep the QBit compatible with QQQ 4.1+ and resolve runtime services through the current QQQ instance. Preserve tenant/device isolation and the distinction between local exposure (`available`) and server approval (`enabled`). Application callers must authorize destinations before submitting or emitting an effect.
Maintainers create reviewed version tags. CI produces JAR/POM/checksum assets and publishes GitHub prereleases for hyphenated versions. Maven Central publication, deployment and physical printer tests are separate operations; this workflow does not perform them.
+
+## Script quality
+
+Owned `.github/scripts/*.py` follow the KofTwentyTwo standard at revision `fbfc8d2cf3b5e16ae9750b04cf9de1794f1e7152` (config adapted from its `python` profile): three-space indentation, Ruff 0.17.3 format/lint (locked in `uv.lock`) and strict Pyright 1.1.414 (Microsoft's npm package, a development-only dependency locked in `package-lock.json` and run through Node 24). Restore with `uv sync --locked` and `npm ci --ignore-scripts`, then run:
+
+```text
+uv run --locked --no-sync python .github/scripts/test-script-quality.py
+uv run --locked --no-sync python .github/scripts/check-script-quality.py
+```
+
+`check-script-quality.py` fails for any tracked Python outside `.github/scripts/` or any JavaScript/TypeScript (this repository has no tsconfig), and requires Pyright's analyzed-file count to equal the selected scripts. Pyright must run as `pyright -p .github`: it silently skips dot directories from a root config. Documented exceptions: CLI `print` only in `check-*.py`; synthetic test assertions and boundary constants; narrow inline `noqa` comments for fixed no-shell Git/interpreter calls and ordered fail-closed validators. Do not loosen a rule or add a type waiver to pass.
diff --git a/README.md b/README.md
index db3e486..e72ba0a 100644
--- a/README.md
+++ b/README.md
@@ -141,7 +141,7 @@ JSONObject job = printing.submit(new JSONObject()
String printJobId = job.getString("jobId");
```
-For terminal and web clients the QBit also offers ready-made print processes: `PrintingRecordProcesses.registerAll(instance)` (after all host tables are added) puts **Print** on every readable table, rendering a record sheet from the table's sections and child lists; `printingPrintTableList` prints a grid of any table (columns, filter, sort, page or all rows, paper and orientation); `printingPrintText` prints text such as a screen capture. Opt a table out with `PrintingTableSettings().withPrintable(false)`. Each returns the queued job id in `printingJobIds`.
+For terminal and web clients the QBit also offers ready-made print processes: `PrintingRecordProcesses.registerAll(instance)` (after all host tables are added) puts **Print** on every readable table, rendering a record sheet from the table's sections and child lists; `printingPrintTableList` prints a grid of any table (columns, filter, sort, page or all rows, paper and orientation); `printingPrintText` prints text such as a screen capture. Opt a table out with `PrintingTableSettings().withPrintable(false)`. Each returns the queued job id in `printingJobIds`. `printingRenderToPdf` (extension 2.2.0) renders the same documents to a PDF download for the caller instead ("Save as PDF"), and every document carries the app header and footer the client sends (`headerApp`, `headerTitle`, `headerServer`, `headerUser`, `headerPrintedAt`, `footerLeft`, `footerNote`).
The returned ID identifies durable queue acceptance. Check job status asynchronously. The [protocol v1 specification](https://github.com/QRun-IO/qqq-print-protocol/blob/f977bd01b62ceb7539190fd2861809537fcf504c/README.md) defines HTTP endpoints, JSON fields, integrity checks and lifecycle semantics.
diff --git a/docs/OPERATOR-CATALOG.md b/docs/OPERATOR-CATALOG.md
new file mode 100644
index 0000000..3edd1cf
--- /dev/null
+++ b/docs/OPERATOR-CATALOG.md
@@ -0,0 +1,67 @@
+# Operator catalog and queue media
+
+This source adds native printing extension 2.3.0 (2.2.0 is the rendering extension and carries no media negotiation). Publication and Windows media
+agent qualification are pending. Existing administrator and device routes keep
+their permissions; this is an additive operator API, not a device bearer API.
+
+Hosts grant `printing_operator` and the `printing_tenant` security key. The native
+`printingSubmit` process remains the domain write permission. Current device and
+printer table/record security, credential scopes and client-channel ceilings apply
+on every operation. Generic job/command history, table writes, enrollment and
+device management remain administrator-only. Hosts may add destination locks.
+Native callbacks cannot replace the initiating actor, session, instance, service
+or row-security context. Submission rechecks current process, credential, channel
+and destination authority after loading/storage callbacks and before commit.
+The backend process is hidden and has no base64-entry form. Echo obtains its user
+selections from the catalog; backend metadata also references native printer and
+format possible-value sources.
+
+- `GET /qqq/v1/qbit/printing/operator/printers?limit=50&after=...` returns at most
+ 200 authorized approved/exposed printers, in stable printer-row-ID order. The
+ opaque next cursor advances over at most `limit` candidate rows even if parent
+ device security hides every candidate. Continue until `nextCursor` is null.
+- `POST /qqq/v1/qbit/printing/operator/jobs` accepts `requestId`, `printerRowId`,
+ `title`, `format` (`PDF` or `ZPL`), `contentBase64` and optional `options`. The
+ native printer row determines the device and opaque Windows queue identity.
+ `requestId` is 1–128 characters from letters, digits, `.`, `_`, `:`, `-`.
+ The authenticated actor and tenant scope the stable key; callers cannot supply
+ actor identity or an internal idempotency key. The response is the existing job
+ receipt after durable commit. Changed requests conflict instead of reprinting.
+- `GET /qqq/v1/qbit/printing/operator/requests/{requestId}` returns only that
+ actor's exact accepted request, or 404. It does not enqueue, read document bytes,
+ or initialize settings. Current destination read authority remains required;
+ later loss of exposure alone does not erase the acceptance receipt.
+ Lookup checks the original accepted native printer-row ID. Deleting that row
+ and creating another with the same device/queue pair does not transfer history.
+
+Each catalog entry has `printerRowId`, `deviceId`, `printerId`, effective `name`,
+`deviceName`, `status`, `formats`, `mediaProfiles`, and optional `description` and
+`location`. These names are presentation; the stable row identity selects a queue.
+
+Media profiles report actual Windows queue defaults. Their typed facts are `id`,
+`revision` (lowercase SHA-256), `name`, `widthMicrons`, `heightMicrons`, `dpiX`,
+`dpiY`, `source: WINDOWS_QUEUE_DEFAULT`, `applyMode: DRIVER_DEFAULT`. Unknown or
+unsupported driver facts produce an empty list. The initial client reports one
+configured default; the wire bound is 32. Dimensions are 1–2,000,000 microns and
+DPI is 1–9600. Names are at most 512 characters. Unknown additive reported facts
+are ignored when canonicalizing; known required facts remain validated. Arbitrary
+page-size values are never interpreted as a Windows stock change.
+Profile and selection IDs share the protocol's 128-character URL-safe ID bound.
+
+Select exact reported media with `options.mediaProfile: {id, revision}`. The
+selection is immutable job intent and accepts no extra fields. New submissions
+reject stale/missing profiles before admission/storage writes; exact accepted
+replays remain accepted. The capable Windows agent rechecks its actual current
+default before spooling. Selected-media dispatch requires the current agent
+capability `printer-media-v1`, including after reconnect/downgrade. Legacy
+inventory without profiles clears stale media facts. Omitted selection preserves
+ordinary existing queue behavior. Paper-width/height overrides remain unsupported,
+including when a media selection is supplied.
+
+Generated/retained source authorization belongs to the configured source owner.
+The retained facade shares this operator permission and destination boundary; it
+preserves its atomic accepted source/actor/request binding. Barcode fit and
+encoding remain the renderer's job.
+
+`QUEUED` is durable acceptance; `SPOOLED` means spooler acceptance. Neither is
+physical output confirmation. These tests use synthetic bytes and no real printer.
diff --git a/docs/RETAINED-PRINT-FACADE.md b/docs/RETAINED-PRINT-FACADE.md
new file mode 100644
index 0000000..68b36af
--- /dev/null
+++ b/docs/RETAINED-PRINT-FACADE.md
@@ -0,0 +1,114 @@
+# Retained print acceptance (additive candidate)
+
+This optional Java integration preserves the existing native generation print
+process and the existing printing queue. It adds no device wire fields or REST
+routes. It is disabled until the host supplies
+`PrintingQBitConfig.withRetainedSourceAuthority`. Current `printing_operator`
+(or administrator) authority and native tenant/device/printer record checks
+remain mandatory. Submission also requires the shared `printingSubmit` process
+permission. This grants no device management or generic job-history access.
+
+## Public contract
+
+`RetainedPrintFacade.Request(requestId, sourceLocator, printerRecordId, title,
+optionsJson)` contains no actor or asserted source binding. Request IDs use
+`[A-Za-z0-9._:-]{1,128}`, including owner namespaces such as `echo:`.
+`submit` derives the signed-in native actor and returns an immutable
+`Acceptance`; it includes the exact source, job/event identities, tenant,
+initiating actor, destination, title and normalized options JSON, including any
+`mediaProfile` ID/revision. `find` authorizes the same complete request and
+returns that acceptance, null for observed absence, or a conflict for an
+unresolved upload reservation. Reuse the same request identity after a lost
+response. Never convert an unavailable lookup into absence or create another
+intent automatically.
+
+The host-configured `RetainedPrintSourceAuthority.resolve(locator, requestId)`
+must authorize the current native actor, source association and currently
+eligible immutable version. It returns `SourceReference(owner, opaqueReference,
+sha256, sizeBytes, mediaType)`. `opaqueReference` is at most 4000 Java
+characters, stable and canonical under that owner's rules; printing does not
+parse owner-specific fields. The optional Record Files adapter must encode its
+complete version/source/generation-request/initiating-actor/tenant tuple and
+compare generation's committed receipt. Accepted MIME types are
+`application/pdf` and `application/vnd.zebra-zpl`; the current configured
+PDF/ZPL limits apply before the byte read, within the existing integer-length
+storage ceiling. `read(reference, requestId)` must enforce the same current
+permissions and return only those exact bytes. Hash/length verification remains
+in printing's existing submission path. Do not supply arbitrary source
+authorities per request or reconstruct a user session.
+
+Source admission remains the owner's policy. An explicitly configured published
+template may qualify generated output without a per-label release dialog;
+external-upload quarantine is unchanged. Printing performs no automatic release
+and creates no human-release audit. `resolve` must be read-only and must not
+invoke a generation reconciliation operation that can generate or dispatch work.
+
+Source/destination checks surround owner callbacks and precede acceptance.
+Submission refuses an ambient QQQ transaction: queue acceptance is independently
+committed and cannot be rolled back by a caller's transaction. Submission also
+enforces the native job-table credential and channel mutation ceilings; lookup
+uses the corresponding read ceilings without granting generic table access.
+`find` calls no byte reader, submission or event writer, and its option
+normalization cannot initialize settings. It reauthorizes the source and
+destination records, then compares the exact accepted intent; a later exposure,
+media inventory or dispatch-limit change does not reinterpret that historical
+intent. A later permission loss may deny returning a receipt whose job already
+committed; it never authorizes a replacement job.
+
+## Persistence and replay
+
+Existing `printing_job` gains nullable `retained_actor_id`,
+`retained_binding_json`, and `accepted_event_id`. Existing `printing_event`
+gains nullable `event_origin`. `initialize` applies additive upgrades; old
+jobs/events remain unbound/unclassified. None of these internal fields are added
+to the device `JOB_COLUMNS` DTO or native table metadata.
+
+Submission and exact lookup both use the existing tenant admission lock, so an
+in-flight inline acceptance cannot be mistaken for absence. The unique
+`(tenant_id, idempotency_key)` serialize submissions. Retained keys use
+`retained:` plus the request ID. Ordinary/bound collisions refuse. Complete
+normalized intent includes actor, owner reference, content hash/length/media,
+destination, title, options and tenant. The existing storage reservation/orphan
+path is reused; it may commit an upload reservation before writing object
+storage. The final job, retained binding, and server acceptance event commit
+together. Failed acceptance-event insertion rolls back the job. Rollback or
+orphan-cleanup failures are suppressed onto the original failure rather than
+replacing it. Existing initialization widens legacy 300-character titles to
+1024; the bound title preserves exact whitespace and Unicode and is compared on
+replay. Acceptance-event JSON has `type=ACCEPTED`, `origin=SERVER_ACCEPTANCE`,
+`status=QUEUED`; it does not claim device receipt or physical output.
+
+This is atomic acceptance binding inside printing, not a distributed transaction
+over the retained owner's authorization database. Deployment still owns durable
+database/storage settings, bounded provider calls and current session policy.
+
+## Committed observations
+
+`observation(requestId, eventId)` reads only an exact committed event joined to
+the accepted job. New ordinary device events are stamped `DEVICE_OBSERVATION`;
+pre-upgrade unclassified events and server acceptance events cannot be returned
+as device evidence. The current native operator must have current device/printer
+row authority and satisfy the job credential/channel read ceiling. Before
+returning any source metadata, printing calls the configured owner's
+`authorizeObservation(source, requestId)` under the current reporter session,
+then rechecks the current ceiling and destination authority. Its default
+implementation refuses; adapters must explicitly implement their current
+source-observation policy. This callback does not grant content access or
+impersonate the initiating actor. This is an operator-authorized observation
+lookup, not device enrollment or permission to forge reporter identity. The
+optional RF authority adapter must preserve initiating actor separately from
+current reporting actor and map only supported stored states. Neither `SPOOLED`
+nor `SPOOLER_COMPLETED` proves physical output.
+
+## Verification scope
+
+The tests use real H2 transactions and native QQQ row authorization with
+synthetic printers. They never contact a device or emit physical output.
+Coverage includes commit-then-throw acknowledgment loss, concurrent exact
+replay, held-admission-lock lookup, mutation ceilings, owner observation
+refusal, callback revocation, exact content hash/length, configured document
+bounds, title upgrade/replay, read-only historical lookup and original-error
+preservation when rollback fails. Qualification of this adopted source is
+recorded in the session notes; the older peer candidate's test result is not
+evidence for this revision. Cross-QBit integration remains a separate gate. No
+release, install, shared-cache substitution or runtime migration has occurred.
diff --git a/docs/SESSION-STATE.md b/docs/SESSION-STATE.md
index 5f6a5af..7925632 100644
--- a/docs/SESSION-STATE.md
+++ b/docs/SESSION-STATE.md
@@ -1,5 +1,58 @@
# Session state — 2026-10-31
+## Echo operator/media and retained-source candidate — 2026-11-01
+
+The isolated Echo candidate adds native printing extension **2.3.0** (after the protected-main 2.2.0 rendering extension) on the
+unchanged product `0.4.0-SNAPSHOT` and QQQ `4.1.0-RC.20` dependency tuple.
+`printing_operator` permits the authorized printer/media catalog and exact-actor
+submission/reconciliation without device administration or general job history.
+Reported Windows queue-default media is revision-bound; clients without the current
+`printer-media-v1` capability cannot claim jobs carrying a media selection.
+
+The opt-in retained-source facade shares that boundary and commits immutable
+source/actor/destination/intent plus its server acceptance event inside the existing
+queue transaction. Its owner adapter is required and default-deny; read-only lookup
+never generates content or submits a replacement job. See
+[operator catalog](OPERATOR-CATALOG.md) and
+[retained acceptance](RETAINED-PRINT-FACADE.md) for exact contracts and limits.
+
+Protected main advanced to a rendering release at printing 2.2.0 (header/footer
+chrome, content sections, render-to-PDF) while this candidate was qualified, so the
+operator/media/retained feature was merged onto it and moved to 2.3.0. The earlier
+183-test gate is historical evidence of the pre-merge source only.
+
+Observed after the merge, locally: the OpenAPI contract was regenerated with the
+existing explicit update switch and its 1.0.0 compatibility baseline hash was
+unchanged. A sandbox full verify reported 188 tests with 11 loopback setup errors and
+no assertion failures (environmental). The first actual-user full verify reported 188
+tests, no assertion failures and **one error**:
+`PrintingRealtimeTest.mediaCapabilityIsReplacedOnEveryAuthenticatedHello` failed
+only when JUnit removed its temporary directory. Cause: `PrintingRealtimeRoutes.close()`
+shut its executor down without waiting, so a channel drain still running after the
+host stopped queried a deleted file database and recreated it empty. The fix waits
+(bounded) for running drains, with regression
+`closeWaitsForChannelWorkersAlreadyRunning`. After it, five consecutive focused
+`PrintingRealtimeTest` runs passed (4 tests each, no empty-database warnings) and the
+full `-Pui-demo verify` passed: **189 reported, 188 executed, one intentional opt-in
+workload skip, zero failures/errors, zero Checkstyle violations**; the source-header
+guard over 146 tracked files and staged whitespace checks passed. The earlier failing
+logs are preserved. This is local evidence only: hosted CI on the signed candidate
+head, publication and cross-QBit consumer qualification are pending. The suite runs
+against synthetic H2 fixtures; no live database, connector, printer or shared
+artifact has been changed. The
+experimental API 1.10 used by Echo is a separate owner-controlled tuple and has not
+been substituted for this candidate's qualified RC.20 dependencies. Publication,
+cross-QBit consumer qualification and any physical output remain separate gates.
+
+Native destination and storage callbacks cannot carry stale actor, session,
+credential, channel or row authority into admission, replay or commit. Receipt
+lookup authorizes the original stored printer-row ID; a replacement row with the
+same device/queue pair cannot inherit earlier acceptance history. The regressions
+reproduce both prior paths, verify denial and rollback, and preserve authorized
+read-only observation after the original queue loses exposure. Media IDs are
+bounded to the shared 128-character contract; long device display names remain
+supported without truncation.
+
## Local Docker development — 2026-11-01
The optional native-auth probe now shares production Core discovery and the
diff --git a/docs/TODO.md b/docs/TODO.md
index 958bdf9..9ff71b8 100644
--- a/docs/TODO.md
+++ b/docs/TODO.md
@@ -1,5 +1,29 @@
# Follow-up
+## Script quality gates
+
+- [x] Pin Ruff and strict Pyright with a locked restore; reject Python outside `.github/scripts/` and zero-file Pyright runs; negative tool probes pass locally.
+- [ ] Observe the new `verify` job steps pass on hosted CI before relying on them.
+
+## Echo integration candidate
+
+- [x] Implement authorized operator catalog, native submission/process permission,
+ exact actor/request lookup and reported driver-default media revisions.
+- [x] Integrate the opt-in atomic retained-source facade using the same queue,
+ current authority, immutable acceptance and exact device-event observation.
+- [x] Close native callback authority changes and preserve the exact accepted
+ printer-row identity during read-only request reconciliation.
+- [x] Complete local combined verification on the merged 2.3.0 candidate: 189
+ reported, 188 executed, one opt-in workload skip, zero failures/errors and
+ Checkstyle violations (the earlier 183-test gate was pre-merge). Preserve exact
+ source, draft protocol fixtures and frozen runtime hashes.
+- [ ] Pass hosted CI on the final signed candidate head; local results do not
+ replace it.
+- [ ] Qualify Echo/Generation's exact core tuple and configured source authority;
+ do not silently adopt experimental core or bypass external-upload admission.
+- [ ] Publish the reviewed additive contract and coordinate client/runtime adoption
+ before owner-operated end-to-end physical acceptance.
+
## Current local development qualification
- [x] Reuse production discovery and the shared major-2 gate in the optional
diff --git a/package-lock.json b/package-lock.json
new file mode 100644
index 0000000..452783d
--- /dev/null
+++ b/package-lock.json
@@ -0,0 +1,51 @@
+{
+ "name": "@qrun-io/qbit-printing-script-quality",
+ "version": "0.0.0",
+ "lockfileVersion": 3,
+ "requires": true,
+ "packages": {
+ "": {
+ "name": "@qrun-io/qbit-printing-script-quality",
+ "version": "0.0.0",
+ "license": "Apache-2.0",
+ "devDependencies": {
+ "pyright": "1.1.414"
+ },
+ "engines": {
+ "node": ">=24"
+ }
+ },
+ "node_modules/fsevents": {
+ "version": "2.3.3",
+ "resolved": "https://registry.npmjs.org/fsevents/-/fsevents-2.3.3.tgz",
+ "integrity": "sha512-5xoDfX+fL7faATnagmWPpbFtwh/R77WmMMqqHGS65C3vvB0YHrgF+B1YmZ3441tMj5n63k0212XNoJwzlhffQw==",
+ "dev": true,
+ "hasInstallScript": true,
+ "license": "MIT",
+ "optional": true,
+ "os": [
+ "darwin"
+ ],
+ "engines": {
+ "node": "^8.16.0 || ^10.6.0 || >=11.0.0"
+ }
+ },
+ "node_modules/pyright": {
+ "version": "1.1.414",
+ "resolved": "https://registry.npmjs.org/pyright/-/pyright-1.1.414.tgz",
+ "integrity": "sha512-FPZZb51jepDX4eP7TEYDeNFtmE3WgwkkEcJpvH3/QmUSsj0EAy3LXu+xB4T/FWDejtsXlCfFr/rbWFjwuwuXww==",
+ "dev": true,
+ "license": "MIT",
+ "bin": {
+ "pyright": "index.js",
+ "pyright-langserver": "langserver.index.js"
+ },
+ "engines": {
+ "node": ">=14.0.0"
+ },
+ "optionalDependencies": {
+ "fsevents": "~2.3.3"
+ }
+ }
+ }
+}
diff --git a/package.json b/package.json
new file mode 100644
index 0000000..6542f49
--- /dev/null
+++ b/package.json
@@ -0,0 +1,14 @@
+{
+ "name": "@qrun-io/qbit-printing-script-quality",
+ "version": "0.0.0",
+ "private": true,
+ "description": "Development-only script quality tooling for QBit Printing; no runtime dependency.",
+ "license": "Apache-2.0",
+ "engines": {
+ "node": ">=24"
+ },
+ "devDependencies": {
+ "pyright": "1.1.414"
+ },
+ "packageManager": "npm@11.19.1"
+}
diff --git a/pyproject.toml b/pyproject.toml
new file mode 100644
index 0000000..4f63273
--- /dev/null
+++ b/pyproject.toml
@@ -0,0 +1,13 @@
+# Copyright 2026 Kingsrook, LLC
+# SPDX-License-Identifier: Apache-2.0
+
+[project]
+name = "qbit-printing-script-quality"
+version = "0.0.0"
+requires-python = ">=3.13"
+
+[dependency-groups]
+dev = ["ruff==0.17.3"]
+
+[tool.uv]
+package = false
diff --git a/ruff.toml b/ruff.toml
new file mode 100644
index 0000000..286fa55
--- /dev/null
+++ b/ruff.toml
@@ -0,0 +1,56 @@
+# Copyright 2026 Kingsrook, LLC
+# SPDX-License-Identifier: Apache-2.0
+# Based on K22 fbfc8d2cf3b5e16ae9750b04cf9de1794f1e7152.
+# KofTwentyTwo Python lint + format configuration (K22-CODE-PY). Copy to the repo root
+# (or merge into [tool.ruff] in pyproject.toml). `ruff format --check` and `ruff check`
+# run in CI and pre-commit with zero findings allowed.
+target-version = "py313"
+include = [".github/scripts/*.py"]
+extend-exclude = [".venv", "node_modules", "dist", "target", "bin", "obj"]
+# Kingsrook: no practical line-length limit (ruff's maximum is 320).
+line-length = 320
+# Kingsrook: 3-space indentation. Ruff's formatter supports it fully; pycodestyle's
+# "multiple of four" rules (E111/E114) are therefore not selected.
+indent-width = 3
+
+[format]
+quote-style = "double"
+indent-style = "space"
+line-ending = "lf"
+docstring-code-format = true
+
+[lint]
+select = [
+ "F", # pyflakes
+ "E4", "E7", "E9", "W", # pycodestyle errors that are layout-independent
+ "I", # isort: no wildcard imports, ordered
+ "N", # pep8-naming
+ "UP", # pyupgrade
+ "B", # bugbear
+ "S", # bandit security
+ "BLE", # blind except
+ "T20", # print statements: use logging (Kingsrook logging rule)
+ "ERA", # commented-out code (Kingsrook: no zombie code)
+ "FIX", # FIXME/TODO must be resolved or tracked
+ "TD", # TODO format: author and issue link
+ "PTH", # pathlib over os.path
+ "RET", "SIM", "C4", "PIE", "PL", "RUF",
+ "ANN", # type annotations on every function
+ "D", # docstrings on modules, classes, functions
+ "LOG", "G", # logging hygiene
+]
+ignore = [
+ "D203", # conflicts with D211
+ "D213", # conflicts with D212
+ "TD003", # issue link is enforced by review, not regex
+]
+
+[lint.per-file-ignores]
+".github/scripts/check-*.py" = ["T201"] # CLI diagnostics are the public stdout contract, not application logging.
+".github/scripts/test-*.py" = ["S101", "PLR2004"] # Synthetic assertions and explicit boundary values.
+
+[lint.pydocstyle]
+convention = "pep257"
+
+[lint.flake8-tidy-imports]
+ban-relative-imports = "all"
diff --git a/src/main/java/com/kingsrook/qbits/printing/PrintingApiSchemas.java b/src/main/java/com/kingsrook/qbits/printing/PrintingApiSchemas.java
index e84015b..ebf594c 100644
--- a/src/main/java/com/kingsrook/qbits/printing/PrintingApiSchemas.java
+++ b/src/main/java/com/kingsrook/qbits/printing/PrintingApiSchemas.java
@@ -48,11 +48,19 @@ final class PrintingApiSchemas
add(schemas, "EnrollmentRequest", fields("enrollmentCode", text(128), "deviceName", text(1024), "agentVersion", text(100)), "enrollmentCode", "deviceName", "agentVersion");
add(schemas, "Enrollment", fields("protocolVersion", number(), "deviceId", id(), "deviceToken", text(128).withDescription("Private device proof, returned once; never log."), "pollIntervalSeconds", number()), "protocolVersion", "deviceId", "deviceToken", "pollIntervalSeconds");
add(schemas, "Profile", fields("name", text(512).withNullable(true), "description", text(4096).withNullable(true), "location", text(512).withNullable(true)));
+ add(schemas, "MediaSelection", fields("id", id(), "revision", text(64)), "id", "revision");
+ add(schemas, "MediaProfile", fields("id", id(), "revision", text(64), "name", text(512), "widthMicrons", number(), "heightMicrons", number(), "dpiX", number(), "dpiY", number(),
+ "source", enumeration("WINDOWS_QUEUE_DEFAULT"), "applyMode", enumeration("DRIVER_DEFAULT")), "id", "revision", "name", "widthMicrons", "heightMicrons", "dpiX", "dpiY", "source", "applyMode");
+ add(schemas, "OperatorPrinter", fields("printerRowId", id(), "deviceId", id(), "printerId", id(), "name", text(1024), "deviceName", text(1024), "description", text(4096), "location", text(512), "status", text(40), "formats", array(enumeration("PDF", "ZPL")), "mediaProfiles", array(ref("MediaProfile"))),
+ "printerRowId", "deviceId", "printerId", "name", "deviceName", "status", "formats", "mediaProfiles");
+ add(schemas, "OperatorCatalog", fields("printers", array(ref("OperatorPrinter")), "nextCursor", text(200).withNullable(true)), "printers", "nextCursor");
+ add(schemas, "OperatorSubmit", fields("requestId", text(128), "printerRowId", id(), "title", text(1024), "format", enumeration("PDF", "ZPL"), "contentBase64", text(Integer.MAX_VALUE), "options", ref("Options")),
+ "requestId", "printerRowId", "title", "format", "contentBase64");
add(schemas, "HeartbeatRequest", fields("agentVersion", text(100), "profile", ref("Profile")), "agentVersion");
add(schemas, "Heartbeat", fields("serverTime", time()), "serverTime");
- add(schemas, "Printer", fields("printerId", id(), "name", text(1024), "driverName", text(1024), "portName", text(1024), "status", enumeration("READY", "OFFLINE", "PAUSED", "ERROR", "UNKNOWN"), "formats", array(enumeration("PDF", "ZPL")), "displayName", text(512), "description", text(4096), "location", text(512)), "printerId", "name", "formats");
+ add(schemas, "Printer", fields("printerId", id(), "name", text(1024), "driverName", text(1024), "portName", text(1024), "status", enumeration("READY", "OFFLINE", "PAUSED", "ERROR", "UNKNOWN"), "formats", array(enumeration("PDF", "ZPL")), "mediaProfiles", array(ref("MediaProfile")), "displayName", text(512), "description", text(4096), "location", text(512)), "printerId", "name", "formats");
add(schemas, "InventoryRequest", fields("printers", array(ref("Printer"))), "printers");
- add(schemas, "Options", fields("copies", number(), "scale", enumeration("ACTUAL_SIZE", "FIT"), "paperWidthMm", new Schema().withType(Type.NUMBER), "paperHeightMm", new Schema().withType(Type.NUMBER)));
+ add(schemas, "Options", fields("copies", number(), "scale", enumeration("ACTUAL_SIZE", "FIT"), "mediaProfile", ref("MediaSelection"), "paperWidthMm", new Schema().withType(Type.NUMBER), "paperHeightMm", new Schema().withType(Type.NUMBER)));
add(schemas, "Job", fields("jobId", id(), "deviceId", id(), "printerId", id(), "title", text(1024), "format", enumeration("PDF", "ZPL"), "contentSha256", text(64), "contentLength", number(), "options", ref("Options"), "createdAt", time(), "status", enumeration("QUEUED", "CLAIMED", "RECEIVED", "SUBMITTING", "SPOOLED", "SPOOLER_COMPLETED", "FAILED", "OUTCOME_UNKNOWN"), "message", text(2048), "spoolerJobId", id()), "jobId", "deviceId", "printerId", "title", "format", "contentSha256", "contentLength", "options", "createdAt", "status");
add(schemas, "Claim", fields("job", ref("Job").withNullable(true)), "job");
add(schemas, "Jobs", fields("jobs", array(ref("Job"))), "jobs");
@@ -60,7 +68,7 @@ final class PrintingApiSchemas
add(schemas, "TestRequest", fields("requestId", id(), "printerId", id(), "format", enumeration("PDF", "ZPL"), "testProfile", text(40)), "requestId", "printerId", "format");
add(schemas, "RetryRequest", fields("requestId", id(), "confirmDuplicateRisk", bool()), "requestId");
add(schemas, "DesiredState", fields("revision", number(), "dispatchPaused", bool(), "connectionState", enumeration("CONNECTED", "DISCONNECTED")), "revision", "dispatchPaused", "connectionState");
- add(schemas, "ControlPollRequest", fields("controlVersion", number(), "agentVersion", text(100), "capabilities", array(enumeration("REFRESH_INVENTORY", "CANCEL_SPOOL")), "lastAppliedRevision", number()), "controlVersion", "agentVersion", "capabilities", "lastAppliedRevision");
+ add(schemas, "ControlPollRequest", fields("controlVersion", number(), "agentVersion", text(100), "capabilities", array(enumeration("REFRESH_INVENTORY", "CANCEL_SPOOL", "printer-media-v1")), "lastAppliedRevision", number()), "controlVersion", "agentVersion", "capabilities", "lastAppliedRevision");
add(schemas, "Command", fields("commandId", id(), "requestId", id(), "action", enumeration("REFRESH_INVENTORY", "CANCEL_SPOOL"), "createdAt", time(), "expiresAt", time(), "jobId", id()), "commandId", "requestId", "action", "createdAt", "expiresAt");
add(schemas, "ControlPoll", fields("controlVersion", number(), "serverTime", time(), "pollIntervalSeconds", number(), "desiredState", ref("DesiredState"), "commands", array(ref("Command"))), "controlVersion", "serverTime", "pollIntervalSeconds", "desiredState", "commands");
add(schemas, "ReceiptRequest", fields("receiptId", id(), "outcome", enumeration("INVENTORY_REFRESHED", "CANCEL_REQUESTED", "ALREADY_ABSENT", "OWNERSHIP_MISMATCH", "UNSUPPORTED", "FAILED", "EXPIRED", "OUTCOME_UNKNOWN"), "completedAt", time(), "message", text(1024)), "receiptId", "outcome", "completedAt");
diff --git a/src/main/java/com/kingsrook/qbits/printing/PrintingChrome.java b/src/main/java/com/kingsrook/qbits/printing/PrintingChrome.java
new file mode 100644
index 0000000..dde6b33
--- /dev/null
+++ b/src/main/java/com/kingsrook/qbits/printing/PrintingChrome.java
@@ -0,0 +1,149 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025. Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.printing;
+
+
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import com.kingsrook.qqq.backend.core.context.QContext;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
+
+
+/*******************************************************************************
+ ** The header and footer every printed page carries, whatever the document:
+ **
+ ** · ·
+ **
+ ** ...
+ **