Build an AI Data Quality Review Agent for M&E

# AI Data Quality Review Agent
## Goal
Review the fictional monitoring dataset using approved validation rules and produce a traceable data-quality review for human decision-making.
## Source Files
- `source-data/participant-monitoring-data.csv`
- `source-data/data-dictionary.md`
- `DATA_QUALITY_RULES.md`
## Machine-Generated Outputs
- `outputs/automated-check-results.csv`
- `outputs/check-summary.json`
## AI-Assisted Outputs
- `outputs/data-quality-review.md`
- `outputs/issue-log.csv`
- `outputs/human-review-list.md`
- Updated `PROGRESS.md`
## Scope
The agent may read files inside this project folder and run `tools/check_data.py`.
The agent may write only to the `outputs/` folder and `PROGRESS.md`.
## Prohibited Actions
- Do not edit, delete, rename or move files in `source-data/`.
- Do not silently correct the dataset.
- Do not invent validation rules, source evidence or corrected values.
- Do not treat every flag as a confirmed data error.
- Do not publish, email or upload outputs.
- Stop when source verification, data-protection judgement or professional judgement is required.

# Data Quality Rules
## Purpose
These are the human-approved rules for the fictional practice dataset. They are training rules, not universal M&E standards. The AI may explain the rules but must not add, remove, reinterpret or change them without human approval.
## Rules Version
- Version: 2.0
- Reporting period: 2026-04-01 to 2026-06-30
- Maximum data-entry delay: 7 days
## Severity Labels
- **Critical:** The file structure prevents a reliable review, such as a missing or blank header, missing required column, duplicate column name or malformed row.
- **Error:** A value directly violates an approved rule, such as an invalid date, missing required value or value outside the permitted range.
- **Warning:** A value may be acceptable but needs checking, such as late entry, an out-of-period date or extra whitespace.
- **Review:** The value may be valid, but an authorised person must decide how the record may be used. A review flag is not proof of a data error.
## Required Columns
- record_id
- participant_id
- activity_date
- data_entry_date
- sex
- age
- district
- attendance_status
- sessions_attended
- consent_confirmed
- endline_score
- source_document
## Required Values
All fields are required except `endline_score`.
## Format, Validity and Consistency Rules
- Dates must use `YYYY-MM-DD`.
- `record_id` must be unique.
- `sex` must be Female, Male, Other, or Prefer not to say.
- `age` must be a whole number from 15 to 99.
- `district` must be North, Central, or South.
- `attendance_status` must be Attended or Absent.
- `sessions_attended` must be a whole number from 0 to 12.
- A record marked Absent must have zero sessions attended.
- `consent_confirmed` must be Yes or No.
- A value of No triggers an authorised-use review. It is not automatically a data-quality error.
- `endline_score`, when present, must be numeric from 0 to 100.
- `source_document` must not be blank.
## Timeliness Rules
- `data_entry_date` must not be earlier than `activity_date`.
- Data should be entered no more than 7 days after the activity.
## Interpretation Rules
- A script finding is a flag, not an automatic correction.
- Passing these checks does not prove factual accuracy, authenticity, representativeness or absence of bias.
- Accuracy against the original paper or electronic source cannot be confirmed from the CSV alone.
- The reviewer must verify suspected errors against authorised source documentation.
- Missing information must be labelled as missing, not guessed.
- The source CSV must remain unchanged during the review.
## Change-Control Rule
When a human approves a rule change, update the rule text, increase the version number and record the reason in `PROGRESS.md` before rerunning the checker.

# Agent Progress
## Current State
- Status: Ready for first manual test
- Dataset: `source-data/participant-monitoring-data.csv`
- Rules version: 2.0
- Last updated: Not yet run
## Last Run
- Date and time:
- Command used:
- Source SHA-256:
- Rows reviewed:
- Critical findings:
- Errors:
- Warnings:
- Review flags:
- Outputs produced:
## Open Issues
-
## Needs Human Review
-
## Next Run Should
- Read all instruction and rule files.
- Run the deterministic checker.
- Compare the narrative report with `check-summary.json` and the detailed issue file.
- Explain findings without modifying source data.
## Decisions Made
- Source data are read-only.
- Automated findings are flags, not automatic corrections.
- A review flag is not automatically a data error.
- No sensitive or real beneficiary data are permitted in this practice exercise.
## Do Not Repeat
- Do not invent corrected values.
- Do not overwrite the source CSV.
- Do not change validation rules without recorded human approval.

# Loop Instructions
You are running a controlled AI Data Quality Review Agent for a fictional M&E dataset.
## Phase 1: Read Context
1. Read `TASK.md`.
2. Read `DATA_QUALITY_RULES.md`.
3. Read `PROGRESS.md`.
4. Read `source-data/data-dictionary.md`.
5. Confirm that `source-data/participant-monitoring-data.csv` exists.
## Phase 2: Run Deterministic Checks
Run the provided standard-library Python script.
Windows:
`py tools/check_data.py`
macOS or Linux:
`python3 tools/check_data.py`
The script works even if the current terminal directory is not the project root. If neither Python command works, stop and explain that Python is unavailable. Do not install software or packages without human approval.
## Phase 3: Read Machine-Generated Results
Read:
- `outputs/check-summary.json`
- `outputs/automated-check-results.csv`
Use the summary file for exact counts and the CSV for row-level details. Do not estimate counts from memory.
For every important finding:
- Identify the CSV row number and `record_id`.
- Name the field and approved rule.
- Preserve the severity assigned by the checker.
- Distinguish direct rule violations from warnings and governance review flags.
- Do not guess a corrected value.
## Phase 4: Create Human-Readable Outputs
Create or update:
### `outputs/data-quality-review.md`
Include:
- Dataset and run date
- Rules version and source SHA-256
- Scope of checks
- Exact finding counts by severity
- Main issue patterns
- Limitations
- Recommended human follow-up
- Explicit statement that source data were not modified
### `outputs/issue-log.csv`
Use these columns:
`issue_id,severity,row_number,record_id,field,rule,observed_value,issue_description,recommended_follow_up,status`
Set `status` to `Open` unless a human has documented a decision.
### `outputs/human-review-list.md`
Include only findings that require source verification, authorised-use review, methodological judgement or a correction decision.
## Phase 5: Update State
Update `PROGRESS.md` with:
- Run date and command
- Source SHA-256
- Rows reviewed
- Exact counts by severity
- Outputs produced
- Open issues
- Human-review items
- What the next run should do
## Evidence and Safety Rules
- Do not modify source data.
- Do not claim the dataset is accurate because rule checks passed.
- Do not infer a correct value without an authorised source.
- Do not expose or process sensitive personal data in this practice exercise.
- Do not send, publish or upload outputs.
- Stop if files contain real names, contact details, health information, safeguarding data or confidential identifiers.
## Verification Checklist
Before ending, confirm:
- The Python checker ran successfully.
- `outputs/check-summary.json` and `outputs/automated-check-results.csv` exist.
- Counts in the narrative match `check-summary.json` exactly.
- Every narrative finding is traceable to a CSV row and rule.
- Warnings and review flags are not mislabelled as confirmed errors.
- Missing information was not guessed.
- No file in `source-data/` was changed.
- `PROGRESS.md` was updated.
- Human-review items are clearly separated.
If a verification item fails, correct the output once. If it still fails, stop and request human review.

"""Rule-based checks for the fictional EvalCommunity M&E dataset.
The script never edits the source CSV. It writes a detailed issue list and a
machine-readable run summary to the outputs folder. It uses only Python's
standard library and is compatible with Python 3.8+.
"""
import csv
import hashlib
import json
from datetime import datetime, timezone
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
SOURCE = ROOT / "source-data" / "participant-monitoring-data.csv"
ISSUES_OUTPUT = ROOT / "outputs" / "automated-check-results.csv"
SUMMARY_OUTPUT = ROOT / "outputs" / "check-summary.json"
RULES_VERSION = "2.0"
REPORTING_START = datetime.strptime("2026-04-01", "%Y-%m-%d").date()
REPORTING_END = datetime.strptime("2026-06-30", "%Y-%m-%d").date()
MAX_ENTRY_DELAY_DAYS = 7
REQUIRED_COLUMNS = [
    "record_id", "participant_id", "activity_date", "data_entry_date",
    "sex", "age", "district", "attendance_status", "sessions_attended",
    "consent_confirmed", "endline_score", "source_document",
]
OPTIONAL_FIELDS = {"endline_score"}
ALLOWED = {
    "sex": {"Female", "Male", "Other", "Prefer not to say"},
    "district": {"North", "Central", "South"},
    "attendance_status": {"Attended", "Absent"},
    "consent_confirmed": {"Yes", "No"},
}
SEVERITY_ORDER = {"critical": 0, "error": 1, "warning": 2, "review": 3}

def add_issue(issues, row_number, record_id, field, rule, severity, value, message):
    issues.append({
        "row_number": str(row_number),
        "record_id": record_id,
        "field": field,
        "rule": rule,
        "severity": severity,
        "observed_value": value,
        "message": message,
    })

def parse_date(value):
    try:
        return datetime.strptime(value, "%Y-%m-%d").date()
    except ValueError:
        return None

def file_sha256(path):
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(65536), b""):
            digest.update(chunk)
    return digest.hexdigest()

def check_integer(issues, row_number, record_id, field, value, minimum, maximum):
    if not value:
        return None
    try:
        number = int(value)
    except ValueError:
        add_issue(issues, row_number, record_id, field, "integer_value", "error", value,
                  "The value must be a whole number.")
        return None
    if not minimum <= number <= maximum:
        add_issue(issues, row_number, record_id, field, "valid_range", "error", value,
                  "The value must be between {} and {}.".format(minimum, maximum))
    return number

def check_number(issues, row_number, record_id, field, value, minimum, maximum):
    if not value:
        return
    try:
        number = float(value)
    except ValueError:
        add_issue(issues, row_number, record_id, field, "numeric_value", "error", value,
                  "The value must be numeric when provided.")
        return
    if not minimum <= number <= maximum:
        add_issue(issues, row_number, record_id, field, "valid_range", "error", value,
                  "The value must be between {} and {}.".format(minimum, maximum))

def main():
    if not SOURCE.exists():
        raise SystemExit("Source file not found: {}".format(SOURCE))
    ISSUES_OUTPUT.parent.mkdir(parents=True, exist_ok=True)
    issues = []
    row_count = 0
    first_record_row = {}
    try:
        with SOURCE.open("r", encoding="utf-8-sig", newline="") as handle:
            reader = csv.DictReader(handle)
            headers = reader.fieldnames or []
            blank_headers = [str(i + 1) for i, h in enumerate(headers) if not h]
            duplicate_headers = sorted({h for h in headers if h and headers.count(h) > 1})
            missing_headers = [h for h in REQUIRED_COLUMNS if h not in headers]
            extra_headers = [h for h in headers if h and h not in REQUIRED_COLUMNS]
            if not headers:
                add_issue(issues, 1, "", "header", "missing_header", "critical", "",
                          "The CSV does not contain a header row.")
            if blank_headers:
                add_issue(issues, 1, "", "header", "blank_column_names", "critical",
                          ", ".join(blank_headers), "One or more columns have a blank header name.")
            if duplicate_headers:
                add_issue(issues, 1, "", "header", "duplicate_columns", "critical",
                          ", ".join(duplicate_headers), "One or more column names are duplicated.")
            if missing_headers:
                add_issue(issues, 1, "", "header", "required_columns", "critical",
                          ", ".join(missing_headers), "One or more required columns are missing.")
            if extra_headers:
                add_issue(issues, 1, "", "header", "unexpected_columns", "warning",
                          ", ".join(extra_headers), "Unexpected columns are present; confirm whether they are authorised.")
            # Row checks are unsafe when the schema is incomplete or ambiguous.
            if headers and not blank_headers and not duplicate_headers and not missing_headers:
                for row_number, row in enumerate(reader, start=2):
                    row_count += 1
                    if None in row:
                        add_issue(issues, row_number, "", "row", "extra_values", "critical",
                                  " | ".join(row.get(None) or []),
                                  "The row contains more values than the header defines.")
                    raw = {field: row.get(field) or "" for field in REQUIRED_COLUMNS}
                    values = {field: value.strip() for field, value in raw.items()}
                    record_id = values["record_id"]
                    for field, value in raw.items():
                        if value != value.strip():
                            add_issue(issues, row_number, record_id, field,
                                      "leading_or_trailing_whitespace", "warning", value,
                                      "The value contains leading or trailing whitespace.")
                    for field in REQUIRED_COLUMNS:
                        if field not in OPTIONAL_FIELDS and not values[field]:
                            add_issue(issues, row_number, record_id, field, "required_value", "error", "",
                                      "A required value is missing.")
                    if record_id:
                        if record_id in first_record_row:
                            add_issue(issues, row_number, record_id, "record_id", "unique_record_id", "error",
                                      record_id, "Duplicate record_id; first seen on row {}.".format(first_record_row[record_id]))
                        else:
                            first_record_row[record_id] = row_number
                    activity_date = parse_date(values["activity_date"]) if values["activity_date"] else None
                    entry_date = parse_date(values["data_entry_date"]) if values["data_entry_date"] else None
                    if values["activity_date"] and activity_date is None:
                        add_issue(issues, row_number, record_id, "activity_date", "iso_date_format", "error",
                                  values["activity_date"], "Use the YYYY-MM-DD date format.")
                    elif activity_date and not REPORTING_START <= activity_date <= REPORTING_END:
                        add_issue(issues, row_number, record_id, "activity_date", "reporting_period", "warning",
                                  values["activity_date"], "The date falls outside the training reporting period.")
                    if values["data_entry_date"] and entry_date is None:
                        add_issue(issues, row_number, record_id, "data_entry_date", "iso_date_format", "error",
                                  values["data_entry_date"], "Use the YYYY-MM-DD date format.")
                    if activity_date and entry_date:
                        delay = (entry_date - activity_date).days
                        if delay < 0:
                            add_issue(issues, row_number, record_id, "data_entry_date", "logical_date_order", "error",
                                      values["data_entry_date"], "The data-entry date is earlier than the activity date.")
                        elif delay > MAX_ENTRY_DELAY_DAYS:
                            add_issue(issues, row_number, record_id, "data_entry_date", "entry_timeliness", "warning",
                                      values["data_entry_date"],
                                      "Data were entered {} days after the activity; the training rule allows no more than {}.".format(delay, MAX_ENTRY_DELAY_DAYS))
                    for field, allowed_values in ALLOWED.items():
                        value = values[field]
                        if value and value not in allowed_values:
                            add_issue(issues, row_number, record_id, field, "allowed_category", "error", value,
                                      "Use one of: {}.".format(", ".join(sorted(allowed_values))))
                    # This is a governance flag, not proof that the value is erroneous.
                    if values["consent_confirmed"] == "No":
                        add_issue(issues, row_number, record_id, "consent_confirmed", "restricted_use_review", "review",
                                  "No", "The value may be correct, but authorised human review is required before the record is used or shared.")
                    check_integer(issues, row_number, record_id, "age", values["age"], 15, 99)
                    sessions = check_integer(issues, row_number, record_id, "sessions_attended",
                                             values["sessions_attended"], 0, 12)
                    if values["attendance_status"] == "Absent" and sessions not in (None, 0):
                        add_issue(issues, row_number, record_id, "sessions_attended", "attendance_consistency", "error",
                                  values["sessions_attended"], "A record marked Absent should have zero sessions attended.")
                    check_number(issues, row_number, record_id, "endline_score", values["endline_score"], 0, 100)
                if row_count == 0:
                    add_issue(issues, 1, "", "dataset", "empty_dataset", "critical", "",
                              "The CSV contains a header but no data rows.")
    except UnicodeDecodeError:
        raise SystemExit("The CSV is not valid UTF-8. Save it as CSV UTF-8 and try again.")
    except csv.Error as exc:
        raise SystemExit("The CSV could not be parsed: {}".format(exc))
    issues.sort(key=lambda item: (int(item["row_number"]), SEVERITY_ORDER[item["severity"]], item["field"]))
    fields = ["row_number", "record_id", "field", "rule", "severity", "observed_value", "message"]
    with ISSUES_OUTPUT.open("w", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=fields)
        writer.writeheader()
        writer.writerows(issues)
    counts = {name: 0 for name in SEVERITY_ORDER}
    for issue in issues:
        counts[issue["severity"]] += 1
    summary = {
        "checked_file": str(SOURCE.relative_to(ROOT)),
        "source_sha256": file_sha256(SOURCE),
        "rules_version": RULES_VERSION,
        "rows_reviewed": row_count,
        "issues_found": len(issues),
        "counts": counts,
        "generated_at_utc": datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z"),
    }
    SUMMARY_OUTPUT.write_text(json.dumps(summary, indent=2) + "\n", encoding="utf-8")
    print("Checked: {}".format(summary["checked_file"]))
    print("Rows reviewed: {}".format(row_count))
    print("Issues found: {}".format(len(issues)))
    for severity in ("critical", "error", "warning", "review"):
        print("{}: {}".format(severity.title(), counts[severity]))
    print("Results: {}".format(ISSUES_OUTPUT.relative_to(ROOT)))
    print("Summary: {}".format(SUMMARY_OUTPUT.relative_to(ROOT)))
    # A structural failure should be visible to terminals, scripts and CI jobs.
    if counts["critical"]:
        raise SystemExit(2)

if __name__ == "__main__":
    main()