The Pipeline

Every nfr-review scan follows the same three-phase pipeline:

Repository ──▶ Collect ──▶ Evaluate ──▶ Report
                  │            │            │
              Evidence      Findings     MD/HTML/
              (typed)     (structured)   CSV/SARIF

This chapter walks through each phase and shows the code that drives it.

Phase 1: Collection

Collectors are parsers that read files from the target repository and emit structured evidence. Each collector focuses on a specific technology or file type:

Collector	What it reads	Evidence it produces
python-ast	.py files	Function signatures, imports, class hierarchies
java-ast	.java files	Class structure, annotations, method signatures
dockerfile	Dockerfile	Base images, exposed ports, multi-stage patterns
k8s-manifest	Kubernetes YAML	Resource limits, probes, security contexts
terraform	.tf files	Resource declarations, provider configs
python-deps	requirements.txt, pyproject.toml	Dependency versions and constraints
repo-structure	Directory tree	File counts, language distribution
adr	docs/adr/	Architecture Decision Records

nfr-review ships with 33 collectors. They run in parallel when --workers > 1.

Evidence shape

Every piece of evidence has a consistent structure:

src/nfr_review/models.py

class Evidence(BaseModel):
    """A single piece of evidence produced by a collector."""

    model_config = ConfigDict(extra="forbid")

    collector_name: str
    collector_version: str
    locator: str
    kind: str
    payload: Any = Field(default_factory=dict)

Evidence model — the output of every collector

The kind field determines which rules will examine this evidence. A rule declares what evidence kinds it consumes, and the engine only dispatches matching evidence.

Fault tolerance

If a collector crashes, nfr-review logs a warning and continues. The failing collector’s evidence is simply absent — rules that needed it are marked as skipped in the report, so you know what wasn’t checked.

Phase 2: Evaluation

Rules examine evidence and yield findings. Each rule focuses on a specific non-functional concern:

• python-star-import checks for from x import * patterns
• dockerfile-latest-tag checks for :latest base images
• k8s-missing-resource-limits checks for pods without CPU/memory limits
• missing-health-check checks for services without /health endpoints

The FieldRule pattern

Most rules extend FieldRule, a typed base class that handles evidence filtering and payload coercion automatically:

src/nfr_review/rules/framework.py

class FieldRule(Generic[P]):
    """Declarative single-evidence-kind rule with typed payload access.

    Subclasses set class attributes and implement ``check()``.
    The base handles selection, skip-if-empty, payload coercion,
    the green all-clear finding, and Finding construction.
    """

    id: str
    band: Band = 1
    collector_name: str
    evidence_kind: str
    payload_type: type[P]
    pattern_tag: str
    required_tech: list[str] = []
    default_confidence: float = 0.9
    all_clear_summary: str = "No issues detected."
    all_clear_recommendation: str = "No action required."

    all_clear_tag: str = ""
    skip_evidence_kind: str = ""
    required_collectors: list[str] = []

    def __init_subclass__(cls, **kw: object) -> None:
        super().__init_subclass__(**kw)
        if not cls.__dict__.get("required_collectors") and hasattr(cls, "collector_name"):
            cls.required_collectors = [cls.collector_name]
        if "id" in cls.__dict__ and cls.id not in rule_registry:
            rule_registry.register(cls.id, cls())

    def check(self, payload: P, ev: Evidence) -> Iterable[Hit]:
        """Yield ``Hit`` objects for one typed payload. Yield nothing when clean."""
        raise NotImplementedError

FieldRule — the typed base class for most rules

A concrete rule only needs to implement check():

src/nfr_review/rules/python_star_import.py

class PythonStarImportRule(FieldRule[PythonAstFilePayload]):
    """Flag wildcard imports that obscure dependencies."""

    id = "python-star-import"
    collector_name = "python-ast"
    evidence_kind = "python-ast-file"
    payload_type = PythonAstFilePayload
    pattern_tag = "star-import"
    default_confidence = 0.95
    all_clear_summary = "No wildcard imports detected."

    def check(self, payload: PythonAstFilePayload, ev: Evidence) -> Iterable[Hit]:
        for imp in payload.imports:
            if imp.is_star:
                yield Hit(
                    rag="amber",
                    summary=f"Star import from {imp.module}",
                    recommendation=(
                        "Use explicit imports to make dependencies"
                        " visible and avoid namespace pollution."
                    ),
                    locator=f"{payload.file_path}:{imp.line}",
                )

A real rule — detecting Python star imports

When a rule finds a problem, it yields a Hit:

src/nfr_review/rules/framework.py

@dataclass(frozen=True, slots=True)
class Hit:
    """What a rule author yields from ``check()``.

    Everything else (rule_id, collector info, default severity, the green
    all-clear) is filled by the base class or ``make_finding``.
    """

    rag: RAG
    summary: str
    recommendation: str
    locator: str
    severity: Severity | None = None
    confidence: float | None = None
    pattern_tag: str | None = None
    content_hash: str = ""

Hit — what a rule yields when it finds a problem

The engine converts each Hit into a Finding (which you saw in Chapter 1) and adds metadata like the collector name and version.

Rule filtering

Not every rule runs on every scan. The engine skips rules when:

• The rule requires a technology not declared in config.tech
• The rule’s required collector failed during Phase 1
• The rule is in config.rules.skip
• config.rules.include_only is set and the rule isn’t in the list

Skipped rules appear in the report with their skip reason — this is deliberate transparency, not a bug.

Phase 3: Reporting

After evaluation, the engine returns a RunResult containing all findings, metadata, and audit information. Renderers then format this into the output you requested:

Format	Flag	Use case
Markdown	(default)	Human review, PR comments
HTML	--html	Self-contained report with diagrams
CSV	--csv	Import into spreadsheets or tracking tools
JSONL	(always produced)	Pipeline integration, programmatic analysis
SARIF	--sarif	GitHub code scanning integration
PDF	--pdf	Executive stakeholders (requires [pdf] extra)

The Markdown report includes:

1. Summary table — category × severity cross-tabulation
2. Design maturity score — 0–100 weighted score (with --score)
3. Diagrams — severity distribution, tech overview (Mermaid)
4. Findings — grouped by RAG status (red / amber / green)
5. Skipped rules — what wasn’t checked and why
6. Dependency findings — third-party code findings (separated from first-party)

The engine orchestration

The Engine.run() method ties all three phases together:

src/nfr_review/engine.py

def run(self, target: Path, config: Config) -> RunResult:
    if not target.exists():
        raise EngineError(f"target does not exist: {target}")
    if not target.is_dir():
        raise EngineError(f"target is not a directory: {target}")

    config = config.model_copy(update={"target": target.resolve()})
    active_collectors = _select_collectors(self._collectors, config.collectors.skip)
    exclude_pats = (
        compile_exclude_patterns(config.exclude_paths) if config.exclude_paths else None
    )

    evidence, warnings, succeeded = self._run_collection_phase(
        active_collectors, target, config, exclude_pats
    )
    findings, rule_results, rules_run, rules_skipped = self._run_rules_phase(
        evidence, config, succeeded
    )

    apply_origin_classification(findings, config.dependency_paths)
    run_metadata = build_run_metadata(target, active_collectors, rules_run, rules_skipped)

    return RunResult(
        findings=findings,
        rule_results=rule_results,
        run_metadata=run_metadata,
        warnings=warnings,
        evidence=evidence,
    )

Engine.run() — the three-phase orchestrator

Key design choices visible in this code:

• Fault tolerance: collector failures don’t abort the run
• Path filtering: evidence from excluded paths is dropped before evaluation
• Origin classification: findings are tagged as first_party or dependency
• Audit trail: metadata records which collectors ran, which rules were skipped, and why

Next

The next chapter dives deeper into collectors — how they work, what evidence they produce, and how to understand their output.