pre-push(17.3): quick rules as the fast first tier

Before the engine and inside the same deadline, the shipped quick rules run
over the head side of the changed files (push/quick.py) and keep matches on
changed lines or in changed functions. They print as advisory pattern matches,
labeled apart from admitted alerts, are kept in the artifact's quick_tier with
the files and bytes read, and never admit or block. A new branch with no
remote base runs the quick tier over its last commit and says so. The planted
eval in serve-static's new debug.js now surfaces; the planted Python
%-formatted SQL in records does not (a quick-rule coverage gap, no rule
edited). Requirement 9.3.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo

src/openultrasast/push/explain.py

@@ -57,6 +57,7 @@ GLOSSARY: dict[str, str] = {
     "query_too_expensive": "some engine questions were too expensive to answer",
     "regions_truncated": "only the highest-ranked regions were examined (--max-regions)",
     "files_unparsed": "some files could not be parsed by the engine",
+    "engine_off": "engine checks were switched off for this run (--engine off)",
     "cross_partition_semantics_unresolved": "flows between languages in this repository are not followed",
     "resolution": "the pushed ref could not be resolved ({detail})",
     "push_input_failed": "the push input could not be read ({detail})",

src/openultrasast/push/quick.py

@@ -0,0 +1,130 @@
+"""The quick-rule tier of pre-push: the fast first look (task 17.3, Requirement 9.3).
+
+The shipped pattern rules run over the head side of the files a push changes, before the engine and
+inside the same deadline, and keep the matches that fall on a changed line or inside a changed
+function. They are pattern matches: shown as advisory, labeled apart from admitted alerts, never
+admitted and never blocking. Requirement 3.2 still governs every engine candidate.
+"""
+
+from __future__ import annotations
+
+import time
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Literal
+
+from openultrasast.model.contracts import ChangeContext
+
+Basis = Literal["comparison", "last_commit"]
+
+
+@dataclass(frozen=True)
+class QuickTierResult:
+    status: Literal["completed", "not_run", "failed"]
+    basis: Basis
+    files: tuple[str, ...] = ()
+    bytes_read: int = 0
+    findings: tuple[Mapping[str, object], ...] = ()
+    reason: str | None = None
+    seconds: float = 0.0
+    basis_commit: str | None = None
+
+    def to_payload(self) -> dict[str, object]:
+        return {
+            "label": "quick rules: pattern matches on changed lines; advisory, not admitted, never blocking",
+            "status": self.status,
+            "basis": self.basis,
+            "basis_commit": self.basis_commit,
+            "files": list(self.files),
+            "bytes_read": self.bytes_read,
+            "findings": [dict(item) for item in self.findings],
+            "reason": self.reason,
+            "seconds": self.seconds,
+        }
+
+
+def head_spans(context: ChangeContext) -> dict[str, list[tuple[int, int]]]:
+    """Changed head lines per decoded path, for every changed file that still exists in head."""
+    deleted = set(context.deleted_paths)
+    spans: dict[str, list[tuple[int, int]]] = {}
+    for path in context.changed_paths:
+        if path not in deleted:
+            spans.setdefault(context.decode_path(path), [])
+    for span in context.spans:
+        if span.side == "head":
+            spans.setdefault(context.decode_path(span.path), []).append((span.start_line, span.end_line))
+    return spans
+
+
+def run_quick_tier(
+    root: Path, spans: Mapping[str, Sequence[tuple[int, int]]], *, basis: Basis = "comparison", basis_commit: str | None = None
+) -> QuickTierResult:
+    """Pattern rules over the changed head files of a materialized snapshot at `root`.
+
+    Pure and bounded by its caller: it reads only the named files, runs no project code and calls no
+    model. `files`/`bytes_read` prove what it read, so a zero is never an unread input.
+    """
+    from openultrasast.findings import PATTERN_RULES, quick_scan_findings
+    from openultrasast.mapping import analyze_entry_points, attach_reachability_hints
+    from openultrasast.policy import load_policy
+    from openultrasast.preprocess import build_file_target, detect_language
+    from openultrasast.rank import rank_targets
+
+    started = time.monotonic()
+    resolved = root.resolve()
+    paths = []
+    read = 0
+    for name in sorted(spans):
+        path = resolved / name
+        if path.is_symlink() or not path.is_file() or not path.resolve().is_relative_to(resolved):
+            continue
+        if detect_language(path) == "unknown":
+            continue
+        paths.append(path)
+        read += path.stat().st_size
+    targets = [build_file_target(resolved, path) for path in paths]
+    targets = attach_reachability_hints(targets, analyze_entry_points(resolved, targets))
+    rules = {rule.rule_id: rule for rule in PATTERN_RULES}
+    kept: list[Mapping[str, object]] = []
+    for finding in quick_scan_findings(resolved, targets, rank_targets(targets), PATTERN_RULES, load_policy()):
+        if finding.status == "shadow" or finding.line is None:
+            continue
+        changed = spans.get(finding.path, ())
+        where = _changed_scope(finding.line, changed, finding.reachability_evidence)
+        if where is None:
+            continue
+        rule_id = finding.finding_id.split(":", 1)[0]
+        rule = rules.get(rule_id)
+        kept.append(
+            {
+                "rule": rule_id,
+                "title": finding.title,
+                "cwe": rule.cwe if rule is not None else None,
+                "severity": finding.severity,
+                "path": finding.path,
+                "line": finding.line,
+                "function": finding.function_name,
+                "scope": where,
+            }
+        )
+    return QuickTierResult(
+        "completed",
+        basis,
+        tuple(target.path for target in targets),
+        read,
+        tuple(kept),
+        seconds=time.monotonic() - started,
+        basis_commit=basis_commit,
+    )
+
+
+def _changed_scope(line: int, changed: Sequence[tuple[int, int]], hints: Sequence[Mapping[str, object]]) -> str | None:
+    if any(start <= line <= end for start, end in changed):
+        return "changed_line"
+    for hint in hints:
+        start, end = hint.get("line"), hint.get("end_line")
+        inside = isinstance(start, int) and isinstance(end, int) and start <= line <= end
+        if inside and any(a <= end and start <= b for a, b in changed):  # type: ignore[operator]
+            return "changed_function"
+    return None

src/openultrasast/push/report.py

@@ -45,6 +45,8 @@ class PushReport:
     # Labeled M1a recorded-veto evaluation. Present only under the explicit replay flag;
     # it is retained beside the decision and never feeds admission or enforcement.
     experimental: Mapping[str, object] = field(default_factory=dict)
+    # The quick-rule tier (task 17.3): one payload per comparison it ran for; advisory only.
+    quick: tuple[Mapping[str, object], ...] = ()
 
     def __post_init__(self) -> None:
         object.__setattr__(self, "model_assistance", MappingProxyType(dict(self.model_assistance)))
@@ -134,6 +136,7 @@ def _write_artifact(report: PushReport, target: Path, temporary: Path, connectio
                     "timing_scope": "Caller stage measurements; publication and rendering duration is returned in ReportDelivery.",
                     "resolution": report.resolution.to_payload() if report.resolution else None,
                     "experimental": dict(report.experimental),
+                    "quick_tier": [dict(item) for item in report.quick],
                 }
                 json.dump(payload, stream, ensure_ascii=True, allow_nan=False, indent=2)
                 stream.write("\n")
@@ -216,6 +219,37 @@ def _experimental_line(experimental: Mapping[str, object]) -> str:
     )
 
 
+# Quick-rule matches shown per comparison; the artifact keeps every one.
+_QUICK_LINES = 5
+
+
+def _quick_lines(payload: Mapping[str, object]) -> list[str]:
+    """The quick tier in the terminal: advisory pattern matches, never presented as alerts."""
+    files = payload.get("files")
+    count = len(files) if isinstance(files, list) else 0
+    read = f"{count} file(s), {payload.get('bytes_read', 0)} bytes read"
+    lines: list[str] = []
+    if payload.get("basis") == "last_commit":
+        commit = str(payload.get("basis_commit") or "")[:7]
+        lines.append(f"New branch: instead, quick rules checked the files its last commit changed (compared with {commit}).")
+    if payload.get("status") != "completed":
+        return [*lines, f"Quick rules did not run ({_short(str(payload.get('reason') or 'unknown'))})."]
+    findings = payload.get("findings")
+    found = [item for item in findings if isinstance(item, Mapping)] if isinstance(findings, list) else []
+    if not found:
+        return [*lines, f"Quick rules: no pattern match on changed code ({read})."]
+    lines.append(f"Quick rules: {len(found)} pattern match(es) on changed code ({read}). Advisory: not verified by the engine.")
+    for item in found[:_QUICK_LINES]:
+        cwe = f"{item.get('cwe')}, " if item.get("cwe") else ""
+        lines.append(
+            f"- {_short(str(item.get('rule')))} ({cwe}{_short(str(item.get('severity')))}) at "
+            f"{_short(str(item.get('path')), 4096)}:{item.get('line')}: {_short(str(item.get('title')))}"
+        )
+    if len(found) > _QUICK_LINES:
+        lines.append(f"{len(found) - _QUICK_LINES} more quick-rule matches retained in the detailed result.")
+    return lines
+
+
 def render_report(report: PushReport, *, artifact: Path | None, error: str | None = None) -> str:
     lines: list[str] = []
     defects = report.admission.defects
@@ -251,6 +285,8 @@ def render_report(report: PushReport, *, artifact: Path | None, error: str | Non
         )
     if len(defects) > 3:
         lines.append(f"{len(defects) - 3} more actionable defects retained in the detailed result.")
+    for payload in report.quick:
+        lines.extend(_quick_lines(payload))
     if not defects:
         coverage = report.result.coverage_status
         if coverage == "complete_within_scope":

src/openultrasast/push/runner.py

@@ -42,6 +42,7 @@ from openultrasast.push.policy import (
     dependency_gaps,
     semantics_digest,
 )
+from openultrasast.push.quick import Basis, QuickTierResult, head_spans, run_quick_tier
 from openultrasast.push.report import PushReport, ReportDelivery, ScanRecord, _completed_scan, deliver_report, render_report
 from openultrasast.push.reuse import ReusingBackend, declaration_identity, discovery
 from openultrasast.push.snapshot import SnapshotAdapter
@@ -267,6 +268,8 @@ def _analyze(
     resolved: PushComparison | None = None,
     record_vetoes: bool = False,
     declarations: Path | None = None,
+    engine: Literal["inline", "off"] = "inline",
+    quick_basis: Basis = "comparison",
 ) -> PushReport:
     """Run a single explicit comparison; exit policy never stands for coverage.
 
@@ -295,6 +298,7 @@ def _analyze(
     comparison = resolved
     context = None
     experimental: dict[str, object] = {"label": LABEL, "flag": FLAG, "status": "not_evaluated"} if record_vetoes else {}
+    quick: QuickTierResult | None = None
     timings: dict[str, float] = {
         "deadline_seconds": config.deadline_seconds,
         "cancellation_allowance_seconds": config.cancellation_allowance_seconds,
@@ -375,68 +379,82 @@ def _analyze(
                 reasons.extend(context.unresolved_boundaries)
                 reasons.extend(_uncovered_languages(tip.manifest, context))
                 stage_started = _stage(timings, stage, stage_started)
-                stage = "head"
-                head_scan = scan_repository(
-                    tip.root,
-                    head_regions,
-                    backend=head_backend,
-                    budget=scan_budget,
-                    execution_budget=budget,
-                    ranking_mode="evidence",
-                    unit="repository",
-                    population_complete=all(m.complete for m in manifests),
-                    change_context=context,
-                )
-                records.append(ScanRecord(comparison, "head", head_scan))
+                stage = "quick"
+                # The fast first tier: pattern rules over the changed head lines, before the engine
+                # and inside the same deadline (Requirement 9.3). Advisory, never admitted.
+                try:
+                    spans = head_spans(context)
+                    basis_commit = comparison.base_oid if quick_basis == "last_commit" else None
+                    quick = _prepare(lambda: run_quick_tier(tip.root, spans, basis=quick_basis, basis_commit=basis_commit), budget)
+                except Exception as error:  # noqa: BLE001 -- a failed quick tier is a stated gap, never the end of the check
+                    quick = QuickTierResult("failed", quick_basis, reason=type(error).__name__)
+                    reasons.append("quick_rules_failed:" + type(error).__name__)
                 stage_started = _stage(timings, stage, stage_started)
-                stage = "base"
-                delta = compare_targeted_base(
-                    old.root,
-                    head=head_scan,
-                    head_regions=head_regions,
-                    base_regions=base_regions,
-                    context=context,
-                    backend=base_backend,
-                    execution_budget=budget,
-                    scan_budget=scan_budget,
-                    head_semantics=provenance["semantics"],
-                    base_semantics=provenance["semantics"],
-                    cache=cache,
-                    cache_semantics=cache_semantics,
-                )
-                if delta.base_scan is not None:
-                    records.append(ScanRecord(comparison, "base", delta.base_scan))
-                reasons.extend(delta.coverage_reasons)
-                stage_started = _stage(timings, stage, stage_started)
-                stage = "admission"
-                candidates = tuple(
-                    AdmissionCandidate(
-                        c,
-                        CapabilityKey(
-                            c.head_operation.question.language if c.head_operation else "unknown",
-                            "unspecified",
-                            "unspecified",
-                            c.family,
-                            c.head_operation.mechanism if c.head_operation else "taint",
-                            "unreviewed",
-                            delta.semantics,
-                        ),
-                        comparison,
-                        dependency_gaps(c, context=context, head=head_scan, base=delta.base_scan),
+                if engine == "off":
+                    reasons.append("engine_off")
+                if engine == "inline":
+                    stage = "head"
+                    head_scan = scan_repository(
+                        tip.root,
+                        head_regions,
+                        backend=head_backend,
+                        budget=scan_budget,
+                        execution_budget=budget,
+                        ranking_mode="evidence",
+                        unit="repository",
+                        population_complete=all(m.complete for m in manifests),
+                        change_context=context,
                     )
-                    for c in delta.candidates
-                )
-                admission = _prepare(lambda: _admit(candidates, provenance), budget)
-                stage_started = _stage(timings, stage, stage_started)
-                if record_vetoes:
-                    stage, stage_started = "recorded_vetoes", time.monotonic()
-                    assert context is not None
-                    recorded = _prepare(
-                        lambda: _record(head_scan, delta, context, comparison, tuple(reasons), admission, provenance, declarations),
-                        budget,
-                    )
-                    experimental = {**recorded, "status": "evaluated"}
+                    records.append(ScanRecord(comparison, "head", head_scan))
                     stage_started = _stage(timings, stage, stage_started)
+                    stage = "base"
+                    delta = compare_targeted_base(
+                        old.root,
+                        head=head_scan,
+                        head_regions=head_regions,
+                        base_regions=base_regions,
+                        context=context,
+                        backend=base_backend,
+                        execution_budget=budget,
+                        scan_budget=scan_budget,
+                        head_semantics=provenance["semantics"],
+                        base_semantics=provenance["semantics"],
+                        cache=cache,
+                        cache_semantics=cache_semantics,
+                    )
+                    if delta.base_scan is not None:
+                        records.append(ScanRecord(comparison, "base", delta.base_scan))
+                    reasons.extend(delta.coverage_reasons)
+                    stage_started = _stage(timings, stage, stage_started)
+                    stage = "admission"
+                    candidates = tuple(
+                        AdmissionCandidate(
+                            c,
+                            CapabilityKey(
+                                c.head_operation.question.language if c.head_operation else "unknown",
+                                "unspecified",
+                                "unspecified",
+                                c.family,
+                                c.head_operation.mechanism if c.head_operation else "taint",
+                                "unreviewed",
+                                delta.semantics,
+                            ),
+                            comparison,
+                            dependency_gaps(c, context=context, head=head_scan, base=delta.base_scan),
+                        )
+                        for c in delta.candidates
+                    )
+                    admission = _prepare(lambda: _admit(candidates, provenance), budget)
+                    stage_started = _stage(timings, stage, stage_started)
+                    if record_vetoes:
+                        stage, stage_started = "recorded_vetoes", time.monotonic()
+                        assert context is not None
+                        recorded = _prepare(
+                            lambda: _record(head_scan, delta, context, comparison, tuple(reasons), admission, provenance, declarations),
+                            budget,
+                        )
+                        experimental = {**recorded, "status": "evaluated"}
+                        stage_started = _stage(timings, stage, stage_started)
                 stage = "cleanup"
         timings[stage + "_seconds"] = time.monotonic() - stage_started
     except Exception as error:
@@ -478,6 +496,30 @@ def _analyze(
         snapshots=tuple(manifests),
         change_context=context,
         experimental=experimental,
+        quick=(quick.to_payload(),) if quick is not None else (),
+    )
+
+
+def _last_commit_quick(
+    repository: Path, comparison: PushComparison, *, artifact: Path, config: PushConfig, budget: ExecutionBudget
+) -> PushReport | None:
+    """Quick rules over the files the pushed head's last commit changed; None for a root commit."""
+    try:
+        parent = SnapshotAdapter(repository, execution_budget=budget).resolve_replay(comparison.head_oid + "^", comparison.head_oid)
+    except Exception:  # noqa: BLE001 -- a root commit has no parent to compare with
+        return None
+    resolved = PushComparison(comparison.head_oid, parent.base_oid, "new_branch_last_commit", comparison.refs)
+    assert resolved.base_oid is not None
+    return _analyze(
+        repository,
+        base=resolved.base_oid,
+        head=resolved.head_oid,
+        artifact=artifact,
+        config=config,
+        execution_budget=budget,
+        resolved=resolved,
+        engine="off",
+        quick_basis="last_commit",
     )
 
 
@@ -549,6 +591,7 @@ def push(
     budget = execution_budget or ExecutionBudget(started + config.deadline_seconds, config.cancellation_allowance_seconds)
     reports: list[PushReport] = []
     analyses: list[ComparisonAnalysis] = []
+    fallbacks: list[PushReport] = []
     reasons: list[str] = []
     resolution = None
     try:
@@ -562,6 +605,12 @@ def push(
             if comparison.base_oid is None or time.monotonic() >= budget.deadline_monotonic:
                 analyses.append(ComparisonAnalysis(comparison, (), (), "unavailable"))
                 reasons.append("base_comparison_unavailable" if comparison.base_oid is None else "deadline_exhausted")
+                if comparison.base_oid is None:
+                    # A new branch has no remote base. Say what was checked instead: the quick rules
+                    # over the files its last commit changed (Requirement 9.2, 9.3).
+                    fallback = _last_commit_quick(repository, comparison, artifact=artifact, config=config, budget=budget)
+                    if fallback is not None:
+                        fallbacks.append(fallback)
                 continue
             report = _analyze(
                 repository,
@@ -624,6 +673,7 @@ def push(
         resolution=resolution,
         snapshots=tuple(s for r in reports for s in r.snapshots),
         change_contexts=tuple(r.change_context for r in reports if r.change_context is not None),
+        quick=tuple(q for r in (*reports, *fallbacks) for q in r.quick),
     )
     from openultrasast.push.assistance import assist