diff --git a/.github/workflows/python-quality.yml b/.github/workflows/python-quality.yml
index e73e422..41506df 100644
--- a/.github/workflows/python-quality.yml
+++ b/.github/workflows/python-quality.yml
@@ -10,6 +10,9 @@ on:
       - "*.ipynb"
       - "pyproject.toml"
       - "requirements*.txt"
+      - "data/**"
+      - "data_quality/**"
+      - "docs/**"
       - "examples/**"
       - "tests/**"
   push:
@@ -21,6 +24,9 @@ on:
       - "*.ipynb"
       - "pyproject.toml"
       - "requirements*.txt"
+      - "data/**"
+      - "data_quality/**"
+      - "docs/**"
       - "examples/**"
       - "tests/**"
   workflow_dispatch:
@@ -72,14 +78,14 @@ jobs:
           python -m pip install -e ".[dev,ml,notebook]"
 
       - name: Compile Python sources
-        run: python -m compileall -q main.py examples tests
+        run: python -m compileall -q main.py data_quality examples tests
 
       - name: Run Ruff lint checks
         id: ruff-lint
         continue-on-error: true
         shell: pwsh
         run: |
-          $output = & python -m ruff check main.py examples tests 2>&1
+          $output = & python -m ruff check main.py data_quality examples tests 2>&1
           $exitCode = $LASTEXITCODE
           $output | Tee-Object -FilePath ruff-lint.txt
           exit $exitCode
@@ -89,17 +95,41 @@ jobs:
         continue-on-error: true
         shell: pwsh
         run: |
-          $output = & python -m ruff format --check main.py examples tests 2>&1
+          $output = & python -m ruff format --check main.py data_quality examples tests 2>&1
           $exitCode = $LASTEXITCODE
           $output | Tee-Object -FilePath ruff-format.txt
           exit $exitCode
 
       - name: Run pytest suite
-        run: python -m pytest
+        id: pytest
+        continue-on-error: true
+        shell: pwsh
+        run: |
+          $output = & python -m pytest 2>&1
+          $exitCode = $LASTEXITCODE
+          $output | Tee-Object -FilePath pytest.txt
+          exit $exitCode
 
       - name: Run baseline entry point
         run: python main.py
 
+      - name: Run data-quality workflow
+        run: >-
+          python -m data_quality
+          --input data/raw/training_results.csv
+          --output .ci-output/data-quality
+
+      - name: Verify data-quality control totals
+        shell: pwsh
+        run: |
+          $report = Get-Content ".ci-output/data-quality/quality_report.json" -Raw | ConvertFrom-Json
+          if ($report.input_rows -ne 15) { throw "Unexpected input row count." }
+          if ($report.accepted_rows -ne 8) { throw "Unexpected accepted row count." }
+          if ($report.rejected_rows -ne 7) { throw "Unexpected rejected row count." }
+          if ($report.exact_duplicate_rows_removed -ne 1) {
+              throw "Unexpected exact duplicate count."
+          }
+
       - name: Run optional ML example
         run: python examples/optional/logistic_regression_basics.py
 
@@ -108,6 +138,15 @@ jobs:
           python -c "from pathlib import Path; Path('.ci-output').mkdir(exist_ok=True)"
           jupyter nbconvert --to notebook --execute dataspell_test.ipynb --output environment-check.executed.ipynb --output-dir .ci-output --ExecutePreprocessor.timeout=120
 
+      - name: Upload data-quality outputs
+        if: always()
+        uses: actions/upload-artifact@v7
+        with:
+          name: data-quality-output-${{ matrix.os }}
+          path: .ci-output/data-quality
+          if-no-files-found: ignore
+          retention-days: 3
+
       - name: Upload quality diagnostics
         if: always()
         uses: actions/upload-artifact@v7
@@ -116,16 +155,22 @@ jobs:
           path: |
             ruff-lint.txt
             ruff-format.txt
+            pytest.txt
           if-no-files-found: ignore
           retention-days: 3
 
-      - name: Enforce Ruff quality gate
+      - name: Enforce quality gate
         if: always()
         shell: pwsh
         run: |
           $lintOutcome = "${{ steps.ruff-lint.outcome }}"
           $formatOutcome = "${{ steps.ruff-format.outcome }}"
-
-          if ($lintOutcome -ne "success" -or $formatOutcome -ne "success") {
-              throw "Ruff quality gate failed. Download the quality diagnostics artifact for details."
+          $pytestOutcome = "${{ steps.pytest.outcome }}"
+
+          if (
+              $lintOutcome -ne "success" -or
+              $formatOutcome -ne "success" -or
+              $pytestOutcome -ne "success"
+          ) {
+              throw "Quality gate failed. Download the diagnostics artifact for details."
           }
diff --git a/README.md b/README.md
index 4c8db1d..c4bc633 100644
--- a/README.md
+++ b/README.md
@@ -2,11 +2,11 @@
 
 [![Python quality](https://github.com/DataTideHH/python-data-basics/actions/workflows/python-quality.yml/badge.svg)](https://github.com/DataTideHH/python-data-basics/actions/workflows/python-quality.yml)
 
-**Python 3.12 · pandas · NumPy · matplotlib · pytest · Ruff · Jupyter · GitHub Actions**
+**Python 3.12 · pandas · data quality · pytest · Ruff · Jupyter · GitHub Actions**
 
-This repository is a compact, tested foundation for reproducible Python Data/BI workflows. It demonstrates a clean project environment, deterministic tabular transformations, explicit dependency groups, public notebook hygiene and cross-platform quality checks.
+This repository is a compact, tested foundation for reproducible Python Data/BI workflows. It combines environment setup, deterministic pandas transformations, a small auditable data-quality workflow, notebook hygiene and cross-platform CI.
 
-It is part of my DataTideHH portfolio during the IHK retraining program in Data and Process Analysis. The scope is deliberately bounded: this is a reusable learning baseline, not a production package, a machine-learning showcase or a substitute for the larger analysis projects in the portfolio.
+It is part of my DataTideHH portfolio during the IHK retraining program in Data and Process Analysis. The scope remains deliberately bounded: this is a reusable learning baseline, not a production ETL platform, predictive-model showcase or finished business analysis.
 
 ---
 
@@ -15,34 +15,36 @@ It is part of my DataTideHH portfolio during the IHK retraining program in Data
 | Area | Current implementation |
 |---|---|
 | Python baseline | Python 3.12-compatible environment and deterministic pandas sanity check |
-| Dependency model | Direct runtime, notebook, optional ML and development groups in `pyproject.toml` |
-| Data handling | Small CSV, JSON and public API examples with synthetic or public-safe inputs |
-| Testing | pytest coverage for the baseline transformation, notebook hygiene and optional ML example |
-| Code quality | Ruff linting and formatting checks plus Python bytecode compilation |
-| Notebook hygiene | Cleared outputs, neutral metadata and tests against committed local paths |
-| Continuous integration | Matrix workflow for Ubuntu 24.04 and Windows 2025 with Python 3.12 |
-| Credential safety | Local environments, tokens, secrets and machine-specific files remain excluded |
+| Dependency model | Runtime, notebook, optional ML and development groups in `pyproject.toml` |
+| Data-quality workflow | CSV input, schema checks, type conversion, row-level rejection, cleaning, KPIs and export |
+| Testing | pytest coverage for baseline logic, data-quality rules, notebook hygiene and optional ML |
+| Code quality | Ruff linting and formatting plus Python bytecode compilation |
+| Continuous integration | Ubuntu 24.04 and Windows 2025 matrix with Python 3.12 |
+| Credential safety | Synthetic/public-safe inputs; local environments and secrets excluded |
 
 ## What This Repository Demonstrates
 
-The repository focuses on foundational tasks that recur in Data/BI work:
+The repository focuses on small tasks that recur in Data/BI work:
 
-- create an isolated and reproducible Python environment
+- create an isolated Python environment
 - define direct dependencies separately from development tooling
-- build and validate small pandas transformations
-- parse nested JSON into tabular structures
-- call a public API without embedding credentials
-- keep notebook outputs and local paths out of version control
-- run syntax, lint, formatting and unit checks automatically
-- use the same entry points on Windows, macOS, Linux and GitHub-hosted runners
-
-More complete business analyses remain in separate repositories. This project provides the tested building blocks underneath them.
+- read raw CSV data as text before controlled conversion
+- enforce required columns and explicit value rules
+- preserve rejected records with reason codes
+- normalise identifiers and text fields
+- derive analysis-ready columns
+- aggregate deterministic module KPIs
+- export cleaned data, rejected rows, KPIs and a JSON quality report
+- test positive and negative data-quality cases
+- run the same checks on Windows and Linux
+
+More complete analyses remain in separate repositories. This project provides tested building blocks underneath them.
 
 ---
 
 ## Quick Start
 
-The detailed platform-specific procedure is documented in [`docs/setup.md`](docs/setup.md).
+Detailed setup instructions are in [`docs/setup.md`](docs/setup.md).
 
 ### Windows PowerShell
 
@@ -64,140 +66,191 @@ python -m pip install -r requirements-dev.txt
 python main.py
 ```
 
-On the Intel iMac used for local portfolio work, Python 3.12 is currently available at `/usr/local/bin/python3.12`.
+On the Intel iMac used for local portfolio work, Python 3.12 is available at `/usr/local/bin/python3.12`.
 
-Expected baseline output contains:
+---
 
-- Python implementation and version
-- pandas, NumPy and matplotlib versions
-- a deterministic two-row category summary
-- `Baseline check passed.`
+## Data-Quality Workflow
 
----
+The main portfolio increment in this repository is the workflow in [`data_quality/`](data_quality/).
 
-## Dependency Model
+It processes the synthetic raw file:
 
-`pyproject.toml` is the source of truth.
+```text
+data/raw/training_results.csv
+```
 
-| Installation | Included scope |
-|---|---|
-| `python -m pip install -e .` | pandas, NumPy and matplotlib runtime baseline |
-| `python -m pip install -e ".[notebook]"` | runtime baseline plus Jupyter |
-| `python -m pip install -e ".[ml]"` | runtime baseline plus scikit-learn example |
-| `python -m pip install -e ".[dev]"` | runtime baseline plus pytest and Ruff |
-| `python -m pip install -r requirements.txt` | complete local learning environment |
-| `python -m pip install -r requirements-dev.txt` | complete development and CI-equivalent environment |
+Run it from the repository root:
 
-The requirement files are intentionally small wrappers. They no longer contain a machine-specific freeze of every transitive Jupyter dependency.
+```powershell
+python -m data_quality `
+  --input "data/raw/training_results.csv" `
+  --output ".ci-output/data-quality"
+```
 
----
+Equivalent macOS/Linux command:
 
-## Baseline Entry Point
+```bash
+python -m data_quality \
+  --input data/raw/training_results.csv \
+  --output .ci-output/data-quality
+```
 
-[`main.py`](main.py) performs two bounded checks:
+The workflow writes:
 
-1. reports the active interpreter and direct runtime package versions
-2. creates a deterministic DataFrame, validates its required columns and values, and calculates one summary row per category
+```text
+.ci-output/data-quality/
+├── cleaned_results.csv
+├── rejected_results.csv
+├── module_kpis.csv
+└── quality_report.json
+```
 
-The script exits with an error when:
+### Validation rules
 
-- Python is older than 3.12
-- a required column is missing
-- the value column contains missing values
-- the deterministic transformation returns an unexpected shape
+Required fields:
 
-This keeps the repository focused on an explainable Data/BI baseline rather than presenting a tiny synthetic model as the primary result.
+```text
+result_id
+learner_id
+module
+assessment_date
+score
+max_score
+pass_score
+```
 
----
+Implemented checks include:
 
-## Example Modules
+- required-column validation
+- missing-value detection
+- strict ISO date parsing
+- numeric conversion
+- positive `max_score`
+- non-negative score and pass threshold
+- score not above maximum
+- pass threshold not above maximum
+- exact duplicate removal
+- conflicting duplicate rejection
+- whitespace and identifier normalisation
 
-### CSV and pandas
+Invalid rows are not silently dropped. They are exported with explicit pipe-separated rejection reasons.
 
-[`examples/01_csv_pandas_basics.py`](examples/01_csv_pandas_basics.py) demonstrates reading in-memory CSV data, filtering and grouped aggregation.
+### Derived fields
 
-### JSON normalization
+Accepted records receive:
 
-[`examples/02_json_basics.py`](examples/02_json_basics.py) demonstrates nested dictionaries and lists and converts selected values into a tabular DataFrame.
+- `source_row` for lineage back to the raw CSV
+- `score_percentage`
+- Boolean `passed`
 
-### Public API request
+### KPI output
 
-[`examples/03_api_request_basics.py`](examples/03_api_request_basics.py) calls Open-Meteo for Hamburg using the Python standard library. It uses no API key or token and handles common network failures.
+The workflow aggregates one row per module with:
 
-### Local Ollama request
+- result count
+- distinct learner count
+- average score percentage
+- passed count
+- failed count
+- pass-rate percentage
 
-[`examples/04_ollama_local_api_basics.py`](examples/04_ollama_local_api_basics.py) remains an optional localhost-only JSON request example. It is not part of the automated CI path because it requires a running local Ollama service and an installed model.
+The committed fixture contains 15 raw rows. The expected control totals are:
 
-### Optional logistic regression
+```text
+accepted rows: 8
+rejected rows: 7
+exact duplicate rows removed: 1
+```
 
-[`examples/optional/logistic_regression_basics.py`](examples/optional/logistic_regression_basics.py) contains the former `main.py` model example. It now has an explicit optional dependency group and a clear limitation: the tiny synthetic dataset demonstrates scikit-learn API usage only and does not support a model-quality claim.
+Detailed rules, expected module values and scope boundaries are documented in [`docs/data-quality-workflow.md`](docs/data-quality-workflow.md).
 
-Run it with:
+---
 
-```bash
-python examples/optional/logistic_regression_basics.py
-```
+## Baseline Entry Point
+
+[`main.py`](main.py) remains a separate deterministic environment and pandas sanity check. It reports the active interpreter and direct runtime package versions, validates a tiny DataFrame and calculates a stable category summary.
+
+This keeps environment verification separate from the larger row-level data-quality workflow.
+
+---
+
+## Additional Example Modules
+
+- [`examples/01_csv_pandas_basics.py`](examples/01_csv_pandas_basics.py) — CSV and grouped pandas operations
+- [`examples/02_json_basics.py`](examples/02_json_basics.py) — nested JSON normalisation
+- [`examples/03_api_request_basics.py`](examples/03_api_request_basics.py) — public Open-Meteo request without credentials
+- [`examples/04_ollama_local_api_basics.py`](examples/04_ollama_local_api_basics.py) — optional localhost-only JSON request
+- [`examples/optional/logistic_regression_basics.py`](examples/optional/logistic_regression_basics.py) — bounded scikit-learn API example without a model-quality claim
 
 ---
 
 ## Notebook Hygiene
 
-[`dataspell_test.ipynb`](dataspell_test.ipynb) verifies the project interpreter and core package imports without storing machine-specific evidence.
+[`dataspell_test.ipynb`](dataspell_test.ipynb) verifies core package imports without committing:
+
+- cell outputs
+- execution counts
+- IDE execution timestamps
+- absolute local paths
+- incorrect legacy Python metadata
 
-The committed notebook contains:
+GitHub Actions executes a temporary copy into `.ci-output/` and leaves the committed notebook unchanged.
 
-- no cell outputs
-- no execution counts
-- no IDE execution timestamps
-- no absolute local interpreter path
-- Python 3.12 kernel and language metadata
+---
+
+## Dependency Model
 
-The pytest suite checks these properties. GitHub Actions executes a temporary notebook copy into the ignored `.ci-output/` directory, leaving the committed notebook unchanged.
+`pyproject.toml` is the source of truth.
+
+| Installation | Included scope |
+|---|---|
+| `python -m pip install -e .` | Runtime baseline and data-quality package |
+| `python -m pip install -e ".[notebook]"` | Runtime plus Jupyter |
+| `python -m pip install -e ".[ml]"` | Runtime plus optional scikit-learn example |
+| `python -m pip install -e ".[dev]"` | Runtime plus pytest and Ruff |
+| `python -m pip install -r requirements-dev.txt` | Complete CI-equivalent environment |
+
+The requirement files remain small wrappers rather than machine-specific freezes of every transitive package.
 
 ---
 
 ## Local Quality Checks
 
-Run the same core checks used in CI:
-
 ```bash
-python -m compileall -q main.py examples tests
-python -m ruff check main.py examples tests
-python -m ruff format --check main.py examples tests
+python -m compileall -q main.py data_quality examples tests
+python -m ruff check main.py data_quality examples tests
+python -m ruff format --check main.py data_quality examples tests
 python -m pytest
 python main.py
+python -m data_quality --input data/raw/training_results.csv --output .ci-output/data-quality
 python examples/optional/logistic_regression_basics.py
 ```
 
-Execute the notebook separately:
-
-```bash
-python -c "from pathlib import Path; Path('.ci-output').mkdir(exist_ok=True)"
-jupyter nbconvert --to notebook --execute dataspell_test.ipynb --output environment-check.executed.ipynb --output-dir .ci-output --ExecutePreprocessor.timeout=120
-```
-
 ---
 
 ## Continuous Integration
 
-The workflow in [`.github/workflows/python-quality.yml`](.github/workflows/python-quality.yml) uses a Python 3.12 matrix on:
+The workflow in [`.github/workflows/python-quality.yml`](.github/workflows/python-quality.yml) runs on:
 
 - Ubuntu 24.04
 - Windows 2025
+- Python 3.12
 
-Each job:
+Each matrix job:
 
-1. checks out the repository with read-only contents permission and without persisted credentials
-2. installs the project and all optional quality groups
-3. compiles Python sources
-4. runs Ruff linting
-5. verifies Ruff formatting
-6. runs pytest
-7. executes the baseline entry point
+1. installs the project and optional quality groups
+2. compiles Python sources
+3. runs Ruff lint and format checks
+4. runs pytest
+5. executes `main.py`
+6. executes the complete data-quality workflow
+7. validates the expected row counts
 8. executes the optional ML example
-9. executes the clean notebook into a temporary ignored directory
+9. executes a clean notebook copy
+10. uploads short-lived generated workflow outputs and Ruff diagnostics
 
-The workflow is an automated quality check, not a deployment or release pipeline.
+The workflow is quality assurance, not deployment or release automation.
 
 ---
 
@@ -205,23 +258,20 @@ The workflow is an automated quality check, not a deployment or release pipeline
 
 ```text
 python-data-basics/
-├── .github/
-│   └── workflows/
-│       └── python-quality.yml
+├── .github/workflows/python-quality.yml
+├── data/raw/training_results.csv
+├── data_quality/
+│   ├── __init__.py
+│   ├── __main__.py
+│   └── workflow.py
 ├── docs/
 │   ├── api-json-oauth2-notes.md
+│   ├── data-quality-workflow.md
 │   ├── ollama-local-api-notes.md
 │   └── setup.md
 ├── examples/
-│   ├── __init__.py
-│   ├── 01_csv_pandas_basics.py
-│   ├── 02_json_basics.py
-│   ├── 03_api_request_basics.py
-│   ├── 04_ollama_local_api_basics.py
-│   └── optional/
-│       ├── __init__.py
-│       └── logistic_regression_basics.py
 ├── tests/
+│   ├── test_data_quality_workflow.py
 │   ├── test_main.py
 │   ├── test_notebook_hygiene.py
 │   └── test_optional_ml.py
@@ -230,52 +280,29 @@ python-data-basics/
 ├── pyproject.toml
 ├── requirements.txt
 ├── requirements-dev.txt
-├── .editorconfig
-├── .gitignore
-├── LICENSE
 └── README.md
 ```
 
 ---
 
-## Credentials and Data Safety
-
-Only synthetic learning data and public endpoints belong in this repository.
-
-Excluded content includes:
+## Data and Credential Safety
 
-- `.env` files
-- API keys and client secrets
-- OAuth access and refresh tokens
-- credential downloads
-- personal or customer data
-- local virtual environments
-- IDE metadata and caches
-- executed CI notebook copies
+Only synthetic learning data and public endpoints belong in this repository. The committed training-results file contains no real learners or personal information.
 
-OAuth2 remains conceptual documentation only. Any future authenticated example must use placeholders and local configuration rather than committed credentials.
+Excluded content includes local environments, `.env` files, API keys, OAuth tokens, credential downloads, personal/customer data, IDE metadata, caches and generated workflow outputs.
 
 ---
 
-## Relationship to Other Portfolio Projects
-
-This repository is the tested Python foundation beneath more specific projects:
-
-- [`open-meteo-germany-weather-ranking`](https://github.com/DataTideHH/open-meteo-germany-weather-ranking) — API-to-CSV scoring workflow
-- [`hamburg-district-data-basics`](https://github.com/DataTideHH/hamburg-district-data-basics) — public-data analysis and Power BI preparation
-- [`sql-server-docker-basics`](https://github.com/DataTideHH/sql-server-docker-basics) — SQL Server, relational integrity, star schema and CI
-- [`flask-country-data-api`](https://github.com/DataTideHH/flask-country-data-api) — validated ingestion, persistence and API delivery
-
 ## Current Boundaries
 
 This repository does not claim:
 
-- a production Python package
-- a production API client
+- a production Python package or ETL platform
+- streaming or distributed processing
+- production orchestration or observability
+- regulatory data validation
 - a validated predictive model
-- a large business analysis
-- a finished dashboard
+- a complete business analysis or dashboard
 - deployment or cloud infrastructure
-- support for copying virtual environments between operating systems
 
-The next useful increment is a small, tested data-quality workflow with explicit raw input, validation rules, cleaned output and KPI aggregation.
+The workflow is intentionally small enough to inspect, run, test and explain in an interview or technical review.
diff --git a/data/raw/training_results.csv b/data/raw/training_results.csv
new file mode 100644
index 0000000..ee34a49
--- /dev/null
+++ b/data/raw/training_results.csv
@@ -0,0 +1,16 @@
+result_id,learner_id,module,assessment_date,score,max_score,pass_score
+R001,L001,SQL Basics,2026-07-01,82,100,60
+R002,L002,SQL Basics,2026-07-01,58,100,60
+R003,L003, Python Basics ,2026-07-02,75,100,60
+R003,L003, Python Basics ,2026-07-02,75,100,60
+R004,L004,Python Basics,2026-07-02,,100,60
+R005,L005,Power BI Basics,2026-07-03,110,100,60
+R006,L006,Power BI Basics,not-a-date,70,100,60
+R007,L007,Process Analysis,2026-07-04,65,100,60
+R008,L008,Process Analysis,2026-07-04,45,100,60
+R009,,SQL Basics,2026-07-05,80,100,60
+R010,L010,Data Quality,2026-07-05,80,0,60
+R011,L011,Data Quality,2026-07-05,88,100,60
+R012,L012,Data Quality,2026-07-05,55,100,60
+R013,L013,Data Quality,2026-07-05,80,100,120
+R014,L014,SQL Basics,2026-07-06,92,100,60
diff --git a/data_quality/__init__.py b/data_quality/__init__.py
new file mode 100644
index 0000000..45badb0
--- /dev/null
+++ b/data_quality/__init__.py
@@ -0,0 +1,19 @@
+"""Tested data-quality workflow for small tabular datasets."""
+
+from data_quality.workflow import (
+    DataQualityError,
+    WorkflowResult,
+    build_module_kpis,
+    load_csv,
+    run_workflow,
+    validate_and_clean,
+)
+
+__all__ = [
+    "DataQualityError",
+    "WorkflowResult",
+    "build_module_kpis",
+    "load_csv",
+    "run_workflow",
+    "validate_and_clean",
+]
diff --git a/data_quality/__main__.py b/data_quality/__main__.py
new file mode 100644
index 0000000..41918ab
--- /dev/null
+++ b/data_quality/__main__.py
@@ -0,0 +1,42 @@
+"""Command-line entry point for the data-quality workflow."""
+
+from __future__ import annotations
+
+import argparse
+from pathlib import Path
+
+from data_quality.workflow import DataQualityError, run_workflow
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        description="Validate, clean, aggregate and export a small assessment-result CSV."
+    )
+    parser.add_argument("--input", type=Path, required=True, help="Path to the raw CSV input.")
+    parser.add_argument(
+        "--output",
+        type=Path,
+        required=True,
+        help="Directory for cleaned data, rejected rows, KPIs and the quality report.",
+    )
+    return parser
+
+
+def main() -> int:
+    args = build_parser().parse_args()
+    try:
+        result = run_workflow(args.input, args.output)
+    except DataQualityError as exc:
+        print(f"Data-quality workflow failed: {exc}")
+        return 1
+
+    print("Data-quality workflow completed.")
+    print(f"Input rows: {result.report['input_rows']}")
+    print(f"Accepted rows: {result.report['accepted_rows']}")
+    print(f"Rejected rows: {result.report['rejected_rows']}")
+    print(f"Output directory: {args.output}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/data_quality/workflow.py b/data_quality/workflow.py
new file mode 100644
index 0000000..61abca3
--- /dev/null
+++ b/data_quality/workflow.py
@@ -0,0 +1,220 @@
+"""Reusable validation, cleaning, KPI and export functions for tabular learning data."""
+
+from __future__ import annotations
+
+import json
+from collections import Counter
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+import pandas as pd
+
+REQUIRED_COLUMNS = (
+    "result_id",
+    "learner_id",
+    "module",
+    "assessment_date",
+    "score",
+    "max_score",
+    "pass_score",
+)
+TEXT_COLUMNS = ("result_id", "learner_id", "module", "assessment_date")
+NUMERIC_COLUMNS = ("score", "max_score", "pass_score")
+
+
+class DataQualityError(ValueError):
+    """Raised when the input schema prevents a meaningful workflow run."""
+
+
+@dataclass(frozen=True)
+class WorkflowResult:
+    """In-memory outputs produced by one workflow execution."""
+
+    cleaned: pd.DataFrame
+    rejected: pd.DataFrame
+    module_kpis: pd.DataFrame
+    report: dict[str, Any]
+
+
+def load_csv(path: str | Path) -> pd.DataFrame:
+    """Read a CSV as text so validation controls all later type conversion."""
+
+    input_path = Path(path)
+    if not input_path.is_file():
+        raise DataQualityError(f"Input file does not exist: {input_path}")
+
+    frame = pd.read_csv(input_path, dtype="string", keep_default_na=False)
+    frame.columns = [str(column).strip() for column in frame.columns]
+    return frame
+
+
+def _normalise_text(series: pd.Series) -> pd.Series:
+    return series.astype("string").str.strip()
+
+
+def _append_reason(reasons: dict[int, list[str]], mask: pd.Series, code: str) -> None:
+    for index in mask[mask].index:
+        reasons[int(index)].append(code)
+
+
+def validate_and_clean(frame: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame, dict[str, Any]]:
+    """Validate rows, reject invalid records and derive analysis-ready columns."""
+
+    missing_columns = sorted(set(REQUIRED_COLUMNS) - set(frame.columns))
+    if missing_columns:
+        joined = ", ".join(missing_columns)
+        raise DataQualityError(f"Missing required columns: {joined}")
+
+    unexpected_columns = sorted(set(frame.columns) - set(REQUIRED_COLUMNS))
+    raw = frame.loc[:, REQUIRED_COLUMNS].copy().reset_index(drop=True)
+    raw.insert(0, "source_row", raw.index + 2)
+
+    working = raw.copy()
+    for column in TEXT_COLUMNS:
+        working[column] = _normalise_text(working[column])
+
+    working["result_id"] = working["result_id"].str.upper()
+    working["learner_id"] = working["learner_id"].str.upper()
+    working["module"] = working["module"].str.replace(r"\s+", " ", regex=True)
+
+    reasons: dict[int, list[str]] = {index: [] for index in working.index}
+
+    for column in REQUIRED_COLUMNS:
+        missing_mask = working[column].astype("string").str.strip().eq("")
+        _append_reason(reasons, missing_mask, f"missing_{column}")
+
+    for column in NUMERIC_COLUMNS:
+        working[column] = pd.to_numeric(working[column], errors="coerce")
+        invalid_numeric = working[column].isna() & raw[column].astype("string").str.strip().ne("")
+        _append_reason(reasons, invalid_numeric, f"invalid_{column}")
+
+    parsed_dates = pd.to_datetime(working["assessment_date"], errors="coerce", format="%Y-%m-%d")
+    invalid_dates = parsed_dates.isna() & raw["assessment_date"].astype("string").str.strip().ne("")
+    _append_reason(reasons, invalid_dates, "invalid_assessment_date")
+    working["assessment_date"] = parsed_dates
+
+    _append_reason(reasons, working["max_score"].le(0), "max_score_not_positive")
+    _append_reason(reasons, working["score"].lt(0), "score_below_zero")
+    _append_reason(reasons, working["pass_score"].lt(0), "pass_score_below_zero")
+    _append_reason(
+        reasons,
+        working["score"].gt(working["max_score"]),
+        "score_above_max_score",
+    )
+    _append_reason(
+        reasons,
+        working["pass_score"].gt(working["max_score"]),
+        "pass_score_above_max_score",
+    )
+
+    duplicate_mask = working["result_id"].ne("") & working["result_id"].duplicated(keep=False)
+    for _, group in working[duplicate_mask].groupby("result_id", sort=False):
+        comparison_columns = list(REQUIRED_COLUMNS[1:])
+        unique_variants = group[comparison_columns].astype("string").drop_duplicates()
+        if len(unique_variants) == 1:
+            duplicate_indexes = list(group.index[1:])
+            for index in duplicate_indexes:
+                reasons[int(index)].append("duplicate_exact")
+        else:
+            for index in group.index:
+                reasons[int(index)].append("duplicate_result_id_conflict")
+
+    rejected_indexes = [index for index, row_reasons in reasons.items() if row_reasons]
+    accepted_indexes = [index for index in working.index if index not in rejected_indexes]
+
+    cleaned = working.loc[accepted_indexes, ["source_row", *REQUIRED_COLUMNS]].copy()
+    cleaned["assessment_date"] = cleaned["assessment_date"].dt.strftime("%Y-%m-%d")
+    cleaned["score_percentage"] = cleaned["score"].div(cleaned["max_score"]).mul(100).round(2)
+    cleaned["passed"] = cleaned["score"].ge(cleaned["pass_score"]).astype(bool)
+    cleaned = cleaned.sort_values(["assessment_date", "result_id"], kind="stable").reset_index(
+        drop=True
+    )
+
+    rejected = raw.loc[rejected_indexes].copy()
+    if rejected.empty:
+        rejected["rejection_reasons"] = pd.Series(dtype="string")
+    else:
+        rejected["rejection_reasons"] = [
+            "|".join(sorted(set(reasons[int(index)]))) for index in rejected_indexes
+        ]
+        rejected = rejected.reset_index(drop=True)
+
+    reason_counts = Counter(
+        reason for row_reasons in reasons.values() for reason in set(row_reasons)
+    )
+    report: dict[str, Any] = {
+        "quality_status": "passed_with_rejections" if rejected_indexes else "passed",
+        "input_rows": int(len(raw)),
+        "accepted_rows": int(len(cleaned)),
+        "rejected_rows": int(len(rejected)),
+        "acceptance_rate_percentage": round(len(cleaned) / len(raw) * 100, 2) if len(raw) else 0.0,
+        "exact_duplicate_rows_removed": int(reason_counts.get("duplicate_exact", 0)),
+        "conflicting_duplicate_rows": int(reason_counts.get("duplicate_result_id_conflict", 0)),
+        "unexpected_columns_ignored": unexpected_columns,
+        "rejection_reason_counts": dict(sorted(reason_counts.items())),
+    }
+    return cleaned, rejected, report
+
+
+def build_module_kpis(cleaned: pd.DataFrame) -> pd.DataFrame:
+    """Aggregate analysis-ready rows into one deterministic record per module."""
+
+    expected = {"module", "learner_id", "score_percentage", "passed"}
+    missing = sorted(expected - set(cleaned.columns))
+    if missing:
+        raise DataQualityError(f"Cannot build KPIs; missing cleaned columns: {', '.join(missing)}")
+
+    if cleaned.empty:
+        return pd.DataFrame(
+            columns=[
+                "module",
+                "result_count",
+                "learner_count",
+                "average_score_percentage",
+                "passed_count",
+                "failed_count",
+                "pass_rate_percentage",
+            ]
+        )
+
+    grouped = cleaned.groupby("module", as_index=False, sort=True).agg(
+        result_count=("result_id", "size"),
+        learner_count=("learner_id", "nunique"),
+        average_score_percentage=("score_percentage", "mean"),
+        passed_count=("passed", "sum"),
+    )
+    grouped["failed_count"] = grouped["result_count"] - grouped["passed_count"]
+    grouped["pass_rate_percentage"] = (
+        grouped["passed_count"].div(grouped["result_count"]).mul(100).round(2)
+    )
+    grouped["average_score_percentage"] = grouped["average_score_percentage"].round(2)
+    for column in ("result_count", "learner_count", "passed_count", "failed_count"):
+        grouped[column] = grouped[column].astype("int64")
+    return grouped
+
+
+def run_workflow(input_path: str | Path, output_dir: str | Path) -> WorkflowResult:
+    """Run the complete workflow and write deterministic CSV and JSON outputs."""
+
+    frame = load_csv(input_path)
+    cleaned, rejected, report = validate_and_clean(frame)
+    module_kpis = build_module_kpis(cleaned)
+
+    output_path = Path(output_dir)
+    output_path.mkdir(parents=True, exist_ok=True)
+
+    cleaned.to_csv(output_path / "cleaned_results.csv", index=False, lineterminator="\n")
+    rejected.to_csv(output_path / "rejected_results.csv", index=False, lineterminator="\n")
+    module_kpis.to_csv(output_path / "module_kpis.csv", index=False, lineterminator="\n")
+    (output_path / "quality_report.json").write_text(
+        json.dumps(report, indent=2, sort_keys=True) + "\n",
+        encoding="utf-8",
+    )
+
+    return WorkflowResult(
+        cleaned=cleaned,
+        rejected=rejected,
+        module_kpis=module_kpis,
+        report=report,
+    )
diff --git a/docs/data-quality-workflow.md b/docs/data-quality-workflow.md
new file mode 100644
index 0000000..16323ed
--- /dev/null
+++ b/docs/data-quality-workflow.md
@@ -0,0 +1,111 @@
+# Data-Quality Workflow
+
+## Purpose
+
+This increment demonstrates a small, reproducible Data/BI preparation workflow rather than a production ETL platform. It turns a deliberately imperfect synthetic CSV into four auditable outputs:
+
+1. cleaned assessment results
+2. rejected rows with explicit reason codes
+3. module-level KPIs
+4. a machine-readable quality report
+
+The implementation is intentionally separated into reusable functions so validation and transformation logic can be tested independently from file-system export.
+
+## Input contract
+
+The raw CSV must contain these columns:
+
+| Column | Rule |
+|---|---|
+| `result_id` | Required, unique after exact-duplicate handling |
+| `learner_id` | Required |
+| `module` | Required; surrounding and repeated whitespace is normalised |
+| `assessment_date` | Required ISO date in `YYYY-MM-DD` format |
+| `score` | Numeric, at least zero and not above `max_score` |
+| `max_score` | Numeric and greater than zero |
+| `pass_score` | Numeric, at least zero and not above `max_score` |
+
+Unexpected columns are ignored but listed in `quality_report.json`.
+
+## Validation and cleaning sequence
+
+The workflow performs the following steps in a fixed order:
+
+1. read all CSV fields as text
+2. verify required columns
+3. preserve the original CSV row number as `source_row`
+4. trim text, uppercase identifiers and collapse repeated module whitespace
+5. convert date and numeric fields with invalid values becoming rejection reasons
+6. enforce score and threshold ranges
+7. remove later copies of exact duplicate `result_id` rows
+8. reject every variant of a conflicting duplicate `result_id`
+9. derive `score_percentage` and Boolean `passed`
+10. aggregate deterministic module KPIs
+11. export CSV and JSON outputs
+
+Invalid rows are not silently discarded. They remain visible in `rejected_results.csv` with pipe-separated reason codes.
+
+## Included synthetic issues
+
+`data/raw/training_results.csv` intentionally contains:
+
+- one exact duplicate
+- a missing score
+- a score above the maximum
+- an invalid date
+- a missing learner identifier
+- a non-positive maximum score
+- a pass threshold above the maximum
+- whitespace requiring normalisation
+
+The committed raw data is synthetic and contains no personal information.
+
+## Run locally
+
+From the repository root:
+
+```bash
+python -m data_quality \
+  --input data/raw/training_results.csv \
+  --output .ci-output/data-quality
+```
+
+PowerShell uses the same arguments:
+
+```powershell
+python -m data_quality `
+  --input "data/raw/training_results.csv" `
+  --output ".ci-output/data-quality"
+```
+
+The `.ci-output/` directory is already ignored by Git.
+
+## Expected sample result
+
+The committed fixture contains 15 input rows. The expected workflow result is 8 analysis-ready rows and 7 rejected rows, including the later copy of the exact duplicate.
+
+Expected module KPI control values include:
+
+| Module | Results | Average score | Pass rate |
+|---|---:|---:|---:|
+| Data Quality | 2 | 71.50% | 50.00% |
+| Process Analysis | 2 | 55.00% | 50.00% |
+| Python Basics | 1 | 75.00% | 100.00% |
+| SQL Basics | 3 | 77.33% | 66.67% |
+
+## Output files
+
+| File | Purpose |
+|---|---|
+| `cleaned_results.csv` | Normalised and validated analysis-ready records |
+| `rejected_results.csv` | Original row values plus explicit rejection reasons |
+| `module_kpis.csv` | Result count, learner count, average score, pass/fail counts and pass rate |
+| `quality_report.json` | Row counts, acceptance rate, duplicate metrics and rejection-reason counts |
+
+## Failure behaviour
+
+A missing input file or missing required column stops the workflow with a non-zero exit code. Row-level quality problems do not crash the workflow; they are isolated in the rejection output and summarised in the report.
+
+## Scope boundary
+
+This is a learning-grade quality workflow for small CSV files. It does not claim streaming ingestion, distributed processing, schema evolution, database transactions, orchestration, production observability or regulatory validation.
diff --git a/pyproject.toml b/pyproject.toml
index 449b44d..9c052a3 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "python-data-basics"
-version = "0.1.0"
+version = "0.2.0"
 description = "A compact, tested Python 3.12 foundation for reproducible Data and BI workflows."
 readme = "README.md"
 requires-python = ">=3.12"
@@ -18,6 +18,9 @@ dependencies = [
     "pandas>=2.2,<4",
 ]
 
+[project.scripts]
+python-data-quality = "data_quality.__main__:main"
+
 [project.optional-dependencies]
 notebook = [
     "jupyter>=1.1,<2",
@@ -32,6 +35,7 @@ dev = [
 
 [tool.setuptools]
 py-modules = ["main"]
+packages = ["data_quality"]
 
 [tool.pytest.ini_options]
 addopts = "-ra --strict-config --strict-markers"
@@ -40,7 +44,7 @@ testpaths = ["tests"]
 [tool.ruff]
 target-version = "py312"
 line-length = 100
-extend-exclude = [".ci-output"]
+extend-exclude = [".ci-output", "output"]
 
 [tool.ruff.lint]
 select = ["B", "E4", "E7", "E9", "F", "I", "UP"]
diff --git a/tests/test_data_quality_workflow.py b/tests/test_data_quality_workflow.py
new file mode 100644
index 0000000..ac7f941
--- /dev/null
+++ b/tests/test_data_quality_workflow.py
@@ -0,0 +1,145 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pandas as pd
+import pytest
+
+from data_quality import DataQualityError, build_module_kpis, run_workflow, validate_and_clean
+
+SAMPLE_INPUT = Path("data/raw/training_results.csv")
+
+
+def test_sample_workflow_writes_verified_outputs(tmp_path: Path) -> None:
+    result = run_workflow(SAMPLE_INPUT, tmp_path)
+
+    assert result.report["quality_status"] == "passed_with_rejections"
+    assert result.report["input_rows"] == 15
+    assert result.report["accepted_rows"] == 8
+    assert result.report["rejected_rows"] == 7
+    assert result.report["exact_duplicate_rows_removed"] == 1
+    assert result.report["conflicting_duplicate_rows"] == 0
+    assert result.report["acceptance_rate_percentage"] == 53.33
+
+    assert set(result.cleaned["result_id"]) == {
+        "R001",
+        "R002",
+        "R003",
+        "R007",
+        "R008",
+        "R011",
+        "R012",
+        "R014",
+    }
+    assert result.cleaned["result_id"].is_unique
+    assert result.cleaned["score_percentage"].between(0, 100).all()
+    assert result.cleaned["passed"].dtype == bool
+
+    expected_files = {
+        "cleaned_results.csv",
+        "rejected_results.csv",
+        "module_kpis.csv",
+        "quality_report.json",
+    }
+    assert {path.name for path in tmp_path.iterdir()} == expected_files
+
+    report_text = (tmp_path / "quality_report.json").read_text(encoding="utf-8")
+    assert json.loads(report_text) == result.report
+
+
+def test_module_kpis_are_deterministic(tmp_path: Path) -> None:
+    result = run_workflow(SAMPLE_INPUT, tmp_path)
+    kpis = result.module_kpis.set_index("module")
+
+    assert list(result.module_kpis["module"]) == [
+        "Data Quality",
+        "Process Analysis",
+        "Python Basics",
+        "SQL Basics",
+    ]
+    assert kpis.loc["SQL Basics", "result_count"] == 3
+    assert kpis.loc["SQL Basics", "average_score_percentage"] == 77.33
+    assert kpis.loc["SQL Basics", "pass_rate_percentage"] == 66.67
+    assert kpis.loc["Process Analysis", "failed_count"] == 1
+    assert kpis.loc["Python Basics", "pass_rate_percentage"] == 100.0
+
+
+def test_missing_required_column_raises_clear_error() -> None:
+    frame = pd.DataFrame(
+        {
+            "result_id": ["R001"],
+            "learner_id": ["L001"],
+            "module": ["SQL Basics"],
+            "assessment_date": ["2026-07-01"],
+            "score": ["80"],
+            "max_score": ["100"],
+        }
+    )
+
+    with pytest.raises(DataQualityError, match="pass_score"):
+        validate_and_clean(frame)
+
+
+def test_conflicting_duplicate_result_ids_reject_every_variant() -> None:
+    frame = pd.DataFrame(
+        {
+            "result_id": ["R001", "R001"],
+            "learner_id": ["L001", "L001"],
+            "module": ["SQL Basics", "SQL Basics"],
+            "assessment_date": ["2026-07-01", "2026-07-01"],
+            "score": ["80", "90"],
+            "max_score": ["100", "100"],
+            "pass_score": ["60", "60"],
+        }
+    )
+
+    cleaned, rejected, report = validate_and_clean(frame)
+
+    assert cleaned.empty
+    assert len(rejected) == 2
+    assert rejected["rejection_reasons"].eq("duplicate_result_id_conflict").all()
+    assert report["conflicting_duplicate_rows"] == 2
+
+
+def test_whitespace_and_identifiers_are_normalised() -> None:
+    frame = pd.DataFrame(
+        {
+            "result_id": [" r001 "],
+            "learner_id": [" l001 "],
+            "module": ["  Data   Quality  "],
+            "assessment_date": ["2026-07-01"],
+            "score": ["75"],
+            "max_score": ["100"],
+            "pass_score": ["60"],
+        }
+    )
+
+    cleaned, rejected, report = validate_and_clean(frame)
+
+    assert rejected.empty
+    assert report["quality_status"] == "passed"
+    assert cleaned.loc[0, "result_id"] == "R001"
+    assert cleaned.loc[0, "learner_id"] == "L001"
+    assert cleaned.loc[0, "module"] == "Data Quality"
+    assert cleaned.loc[0, "score_percentage"] == 75.0
+    assert bool(cleaned.loc[0, "passed"]) is True
+
+
+def test_empty_cleaned_frame_has_stable_kpi_schema() -> None:
+    empty = pd.DataFrame(
+        columns=["module", "learner_id", "result_id", "score_percentage", "passed"]
+    )
+
+    kpis = build_module_kpis(empty)
+
+    assert kpis.empty
+    assert list(kpis.columns) == [
+        "module",
+        "result_count",
+        "learner_count",
+        "average_score_percentage",
+        "passed_count",
+        "failed_count",
+        "pass_rate_percentage",
+    ]