diff --git a/CHANGELOG.md b/CHANGELOG.md index d3ac3fd..219d81f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,39 @@ All notable changes to this project will be documented in this file. Format follows [Keep a Changelog](https://keepachangelog.com/). This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [0.10.0] - 2026-07-06 + +### Added + +- `data-refinery refine dataset` — turns a `colleague feedback export` JSONL + stream (one graded work item per line) into a sloth-ready **chat-schema** + train/eval split plus a provenance `lineage.json`. A pure-function pipeline + (`data_refinery/refine/dataset.py`): shape-validate each export line (a + malformed line is skipped, never silently dropped — one diagnostic per bad + line), filter to `status == "ok"` and a graded rating at/above + `--min-rating`, map to the sloth chat schema, dedup same-content examples + (reusing the existing `find_duplicate_groups` same-hash-same-scope + primitive), and split train/eval via a deterministic seeded shuffle so the + two sets are **disjoint by construction**. Every example carries its + per-example provenance (source task id, rating, exported-at timestamp). +- `data-refinery refine lineage` — reads back a `lineage.json` written by + `refine dataset` for inspection. +- `docs/contract.md` Wave 4 — the `refine` consumer contract (agent-first: + `--json` on every verb, results to stdout, diagnostics to stderr, exit `1` + on a missing `--from`/`--dataset` path with a `hint:`). +- New `explain`/`overview`/`learn` catalog entries for `refine dataset` and + `refine lineage`. +- Stdlib-only, no network, no LLM call — the pipeline is unit-testable stage + by stage. + +### Fixed + +- `refine dataset`/`refine lineage` classify a non-UTF-8 input (`--from`) or + `lineage.json` as a structured environment error (exit `2`, a "not valid + UTF-8" `hint:`) instead of a generic unexpected error — `UnicodeDecodeError` + is now caught alongside `OSError` at both read sites (addresses Qodo review + on #15). + ## [0.9.0] - 2026-06-24 ### Added diff --git a/data_refinery/cli/__init__.py b/data_refinery/cli/__init__.py index 3272806..da9a88a 100644 --- a/data_refinery/cli/__init__.py +++ b/data_refinery/cli/__init__.py @@ -68,6 +68,7 @@ def _build_parser() -> argparse.ArgumentParser: from data_refinery.cli._commands import learn as _learn_cmd from data_refinery.cli._commands import overview as _overview_cmd from data_refinery.cli._commands import quality as _quality_cmd + from data_refinery.cli._commands import refine as _refine_group from data_refinery.cli._commands import stack as _stack_group from data_refinery.cli._commands import store as _store_group from data_refinery.cli._commands import whoami as _whoami_cmd @@ -94,6 +95,7 @@ def _build_parser() -> argparse.ArgumentParser: _stack_group.register(sub) _store_group.register(sub) _quality_cmd.register(sub) + _refine_group.register(sub) # Register your own noun groups here: # from data_refinery.cli._commands import my_noun as _my_noun_group # _my_noun_group.register(sub) diff --git a/data_refinery/cli/_commands/learn.py b/data_refinery/cli/_commands/learn.py index 1b13773..df85abe 100644 --- a/data_refinery/cli/_commands/learn.py +++ b/data_refinery/cli/_commands/learn.py @@ -36,6 +36,8 @@ data-refinery dedup Collapse same-hash duplicates (idempotent). data-refinery integrity Check stored hash matches sha256(content). data-refinery freshness Report age/staleness facts from metadata. + data-refinery refine dataset Graded colleague work items -> train/eval JSONL. + data-refinery refine lineage Read back a dataset's lineage.json. Machine-readable output ----------------------- @@ -73,6 +75,14 @@ def _as_json_payload() -> dict[str, object]: {"path": ["dedup"], "summary": "Collapse same-hash duplicates (idempotent)."}, {"path": ["integrity"], "summary": "Check stored hash matches sha256(content)."}, {"path": ["freshness"], "summary": "Report age/staleness facts from metadata."}, + { + "path": ["refine", "dataset"], + "summary": "Graded colleague work items -> sloth-ready train/eval JSONL + lineage.", + }, + { + "path": ["refine", "lineage"], + "summary": "Read back a dataset directory's lineage.json.", + }, ], "exit_codes": { "0": "success", diff --git a/data_refinery/cli/_commands/overview.py b/data_refinery/cli/_commands/overview.py index 2a0964f..416ce28 100644 --- a/data_refinery/cli/_commands/overview.py +++ b/data_refinery/cli/_commands/overview.py @@ -33,6 +33,7 @@ "stack up|down|status — manage the storage substrate (mongo + neo4j)", "store put|get|list — put/get/list opaque envelopes in the store", "validate|dedup|integrity|freshness — data-quality checks over the store", + "refine dataset|lineage — graded colleague work items to sloth-ready training data", ] diff --git a/data_refinery/cli/_commands/refine.py b/data_refinery/cli/_commands/refine.py new file mode 100644 index 0000000..61fecb4 --- /dev/null +++ b/data_refinery/cli/_commands/refine.py @@ -0,0 +1,212 @@ +"""``data-refinery refine`` — graded colleague work items to sloth-ready training data. + +The S6 leg of the colleague/data-refinery/unsloth "refine" arc +(agentculture/data-refinery-cli#14, owner decision recorded on +agentculture/colleague#291): turn a `colleague feedback export` JSONL stream +(one graded work item per line) into a sloth-ready chat-schema train/eval split +plus a provenance ``lineage.json``. Pure library logic lives in +:mod:`data_refinery.refine.dataset`; this module is the CLI mirror, following +the same shell-or-import pattern as ``store``/``validate``/``dedup``/etc. + +Contract (agent-first): ``--json`` on every verb; results to stdout, +diagnostics/errors to stderr; a malformed input line is skipped with **one** +stderr diagnostic plus a counted total — never silently dropped; a missing +``--from``/``--dataset`` path exits ``1`` (user-input error) with a ``hint:``, +never a traceback. Stdlib only: no network, no LLM. +""" + +from __future__ import annotations + +import argparse +from pathlib import Path + +from data_refinery.cli._errors import EXIT_USER_ERROR, CliError +from data_refinery.cli._output import emit_diagnostic, emit_result +from data_refinery.refine import dataset + +_DEFAULT_MIN_RATING = 4 +_DEFAULT_SPLIT = "90/10" +_DEFAULT_SEED = 0 + + +def _add_json_flag(p: argparse.ArgumentParser) -> None: + p.add_argument("--json", action="store_true", help="Emit structured JSON.") + + +def cmd_refine_dataset(args: argparse.Namespace) -> int: + json_mode = bool(getattr(args, "json", False)) + from_path = Path(args.from_path) + if not from_path.is_file(): + raise CliError( + code=EXIT_USER_ERROR, + message=f"--from file not found: {from_path}", + remediation=( + "pass --from pointing at a JSONL file produced by 'colleague feedback export'" + ), + ) + + result = dataset.run_refine_dataset( + from_path, + schema=args.schema, + min_rating=args.min_rating, + split=args.split, + seed=args.seed, + out_dir=Path(args.out), + ) + for diag in result["diagnostics"]: + emit_diagnostic(diag) + + payload = result["payload"] + if json_mode: + emit_result(payload, json_mode=True) + else: + agg = payload["aggregate"] + emit_result( + f"refine dataset: kept {agg['kept_count']} " + f"(train {agg['train_size']} / eval {agg['eval_size']}) from " + f"{agg['input_count']} input line(s) — {agg['malformed_count']} malformed, " + f"{agg['filtered_count']} filtered, {agg['duplicates_removed']} duplicate(s) " + f"removed\nwrote {payload['train_file']}, {payload['eval_file']}, " + f"{payload['lineage_file']}", + json_mode=False, + ) + return 0 + + +def cmd_refine_lineage(args: argparse.Namespace) -> int: + json_mode = bool(getattr(args, "json", False)) + payload = dataset.read_lineage(Path(args.dataset)) + if json_mode: + emit_result(payload, json_mode=True) + else: + agg = payload.get("aggregate", {}) + dist = agg.get("grade_distribution", {}) + lines = [ + f"lineage: {args.dataset}", + f"kept: {agg.get('kept_count')} " + f"(train {agg.get('train_size')} / eval {agg.get('eval_size')})", + f"input: {agg.get('input_count')} " + f"(malformed {agg.get('malformed_count')}, filtered {agg.get('filtered_count')}, " + f"duplicates removed {agg.get('duplicates_removed')})", + f"min_rating: {agg.get('min_rating')}, split: {agg.get('split')}, " + f"seed: {agg.get('seed')}", + "grade distribution: " + ", ".join(f"{k}={v}" for k, v in sorted(dist.items())), + f"train_sha256: {agg.get('train_sha256')}", + f"eval_sha256: {agg.get('eval_sha256')}", + ] + emit_result("\n".join(lines), json_mode=False) + return 0 + + +def _refine_overview(args: argparse.Namespace) -> int: + """`data-refinery refine` with no sub-verb prints the noun's overview.""" + from data_refinery.cli._commands.overview import emit_overview + + sections = [ + { + "title": "Verbs", + "items": [ + "refine dataset --from --out — graded colleague work " + "items -> sloth-ready chat-schema train/eval JSONL + lineage.json", + "refine lineage --dataset — read back a dataset's lineage.json " + "(aggregates + grade distribution)", + ], + }, + { + "title": "Input line shape (colleague `feedback export`)", + "items": [ + "{task_id, request, summary, rating(int 1-5 | null), notes, status, at, " + "stats{steps,files_changed,bytes_written}}", + "one JSON object per line; a malformed line is skipped with a stderr " + "diagnostic, never silently dropped", + ], + }, + { + "title": "Filter + split rules", + "items": [ + "kept iff status == 'ok' AND rating is graded (non-null) AND " + ">= --min-rating (default 4) — ungraded/below-threshold never enters silently", + "deduped by content hash before splitting (reuses the quality dedup primitive)", + "train/eval are disjoint by construction: a deterministic seeded shuffle " + "(--seed, default 0) then a straight slice by --split (default 90/10)", + ], + }, + { + "title": "Conventions", + "items": [ + "every verb supports --json", + "stdlib-only: no network, no LLM", + "--schema chat is the only supported schema today", + ], + }, + ] + emit_overview("data-refinery refine", sections, json_mode=bool(getattr(args, "json", False))) + return 0 + + +def register(sub: argparse._SubParsersAction) -> None: + p = sub.add_parser( + "refine", + help="Turn graded colleague work items into sloth-ready train/eval training data.", + ) + _add_json_flag(p) + p.set_defaults(func=_refine_overview, json=False) + # Propagate the structured-error parser_class to nested verbs. + verb = p.add_subparsers(dest="refine_command", parser_class=type(p)) + + ds = verb.add_parser( + "dataset", + help="Build a train/eval JSONL split + lineage.json from graded colleague work items.", + ) + ds.add_argument( + "--from", + dest="from_path", + required=True, + help="Path to a JSONL file produced by 'colleague feedback export'.", + ) + ds.add_argument( + "--schema", + default="chat", + choices=sorted(dataset.KNOWN_SCHEMAS), + help="Output example schema (default: chat; the only schema supported today).", + ) + ds.add_argument( + "--min-rating", + type=int, + default=_DEFAULT_MIN_RATING, + help=f"Minimum feedback rating to keep (default: {_DEFAULT_MIN_RATING}).", + ) + ds.add_argument( + "--split", + default=_DEFAULT_SPLIT, + help=f"train/eval ratio as A/B (default: {_DEFAULT_SPLIT}).", + ) + ds.add_argument( + "--seed", + type=int, + default=_DEFAULT_SEED, + help=f"Seed for the deterministic train/eval shuffle (default: {_DEFAULT_SEED}).", + ) + ds.add_argument( + "--out", + required=True, + help="Output directory for train.jsonl/eval.jsonl/lineage.json.", + ) + _add_json_flag(ds) + ds.set_defaults(func=cmd_refine_dataset) + + lin = verb.add_parser( + "lineage", + help="Read back a dataset directory's lineage.json (aggregates + grade distribution).", + ) + lin.add_argument( + "--dataset", + required=True, + help="Dataset output directory (as produced by 'refine dataset --out').", + ) + _add_json_flag(lin) + lin.set_defaults(func=cmd_refine_lineage) + + ov = verb.add_parser("overview", help="Describe the refine noun.") + _add_json_flag(ov) + ov.set_defaults(func=_refine_overview) diff --git a/data_refinery/explain/catalog.py b/data_refinery/explain/catalog.py index 0183d3b..f93a49f 100644 --- a/data_refinery/explain/catalog.py +++ b/data_refinery/explain/catalog.py @@ -36,6 +36,8 @@ - `data-refinery dedup` — collapse same-hash-same-scope duplicates (idempotent). - `data-refinery integrity` — check every stored hash matches sha256(content). - `data-refinery freshness` — report age/staleness facts from a metadata timestamp. +- `data-refinery refine dataset` — graded colleague work items to sloth-ready train/eval JSONL. +- `data-refinery refine lineage` — read back a dataset's lineage.json. ## Exit-code policy @@ -266,6 +268,69 @@ """ +_REFINE = """\ +# data-refinery refine + +Turns a `colleague feedback export` JSONL stream (one graded work item per +line) into a sloth-ready **chat**-schema train/eval split plus a provenance +`lineage.json` — the S6 leg of the colleague/data-refinery/unsloth "refine" +arc (agentculture/data-refinery-cli#14). Stdlib only: no network, no LLM. + +## Input line shape + +```json +{"task_id": "...", "request": "...", "summary": "...", + "rating": 4, "notes": "...", "status": "ok", + "at": "2026-07-05T12:00:00+00:00", + "stats": {"steps": 6, "files_changed": 2, "bytes_written": 1345}} +``` + +`rating` is an int `1`-`5` **or** `null` (an ungraded work item — colleague's +own feedback store reads back an ungraded item as `None`, never an error). +`notes`/`stats` are optional. A malformed line (bad JSON, wrong types, an +out-of-range rating) is skipped with **one** stderr diagnostic — never +silently dropped — and counted in `malformed_count`. + +## `data-refinery refine dataset` + + data-refinery refine dataset --from graded.jsonl --out ./dataset --json + +Flags: `--from ` (required), `--schema chat` (default + only schema +today), `--min-rating N` (default `4`), `--split A/B` (default `90/10`), +`--seed N` (default `0`), `--out ` (required). + +- **Filter:** kept iff `status == "ok"` **and** `rating` is graded (non-null) + **and** `>= --min-rating` — an ungraded or below-threshold line is filtered, + never enters the dataset silently. +- **Map:** `{"messages": [{"role": "user", "content": }, + {"role": "assistant", "content": }]}` — matches + `sloth.tune.datasets` chat-schema validation exactly (`sloth validate` + accepts the output). +- **Dedup:** by content hash, before splitting (reuses the same + same-hash-same-scope primitive `data-refinery dedup` runs over the store). +- **Split:** a deterministic seeded shuffle (`--seed`) then a straight slice by + `--split` — train/eval are **disjoint by construction**. + +Writes `train.jsonl`, `eval.jsonl`, and `lineage.json` under `--out`. + +## `data-refinery refine lineage` + + data-refinery refine lineage --dataset ./dataset --json + +Reads back `/lineage.json`: per-example provenance +(`{source: "colleague", task_id, rating, content_sha256, exported_at}`) plus an +aggregate (input/malformed/filtered/duplicate counts, grade distribution, +split sizes, seed, min_rating, and the sha256 of both emitted files). Exits `1` +with a `hint:` if `lineage.json` is absent. + +## Usage + + data-refinery refine dataset --from graded.jsonl --min-rating 4 \\ + --split 90/10 --out ./dataset --json + data-refinery refine lineage --dataset ./dataset --json +""" + + ENTRIES: dict[tuple[str, ...], str] = { (): _ROOT, ("data-refinery",): _ROOT, @@ -293,4 +358,8 @@ ("dedup",): _DEDUP, ("integrity",): _INTEGRITY, ("freshness",): _FRESHNESS, + ("refine",): _REFINE, + ("refine", "dataset"): _REFINE, + ("refine", "lineage"): _REFINE, + ("refine", "overview"): _REFINE, } diff --git a/data_refinery/refine/__init__.py b/data_refinery/refine/__init__.py new file mode 100644 index 0000000..2bd7763 --- /dev/null +++ b/data_refinery/refine/__init__.py @@ -0,0 +1,42 @@ +"""data-refinery's ``refine`` surface — graded colleague work items to training data. + +Turns a `colleague feedback export` JSONL stream (one graded work item per line) +into a sloth-ready chat-schema train/eval split plus a provenance ``lineage.json`` +— the S6 leg of the colleague/data-refinery/unsloth "refine" arc +(agentculture/data-refinery-cli#14). Pure stdlib: no network, no LLM, no third- +party dependency. Importable as a library (``data_refinery.refine.dataset``) and +mirrored by the CLI verbs ``refine dataset`` / ``refine lineage``, the same +shell-or-import pattern as :mod:`data_refinery.store` / :mod:`data_refinery.quality`. +""" + +from __future__ import annotations + +from data_refinery.refine.dataset import ( + KNOWN_SCHEMAS, + ExampleItem, + build_lineage, + dedup_items, + filter_records, + parse_export_jsonl, + parse_split_ratio, + read_lineage, + run_refine_dataset, + split_items, + to_example, + validate_export_line, +) + +__all__ = [ + "KNOWN_SCHEMAS", + "ExampleItem", + "validate_export_line", + "parse_export_jsonl", + "filter_records", + "to_example", + "dedup_items", + "parse_split_ratio", + "split_items", + "build_lineage", + "read_lineage", + "run_refine_dataset", +] diff --git a/data_refinery/refine/dataset.py b/data_refinery/refine/dataset.py new file mode 100644 index 0000000..bf6dd5c --- /dev/null +++ b/data_refinery/refine/dataset.py @@ -0,0 +1,536 @@ +"""Graded colleague work items -> sloth-ready chat-schema train/eval JSONL. + +Pure functions (no I/O beyond the two file-reading/writing entry points at the +bottom), so the pipeline is unit-testable stage by stage and shared by the CLI +verb. Every stage operates on plain dicts/dataclasses — never network, never an +LLM call, stdlib only (``json``/``hashlib`` via +:func:`data_refinery.store.envelope.content_hash`/``random.Random``/``pathlib``). + +Pipeline (mirrors the S6 spec, agentculture/data-refinery-cli#14): + +1. :func:`parse_export_jsonl` — parse + shape-validate each input line + (:func:`validate_export_line`); a malformed line is skipped, never silently + dropped (the caller gets one diagnostic per bad line). +2. :func:`filter_records` — keep only ``status == "ok"`` **and** a graded + (non-null) ``rating >= min_rating``; an ungraded or below-threshold line is + filtered, never enters the dataset silently. +3. :func:`to_example` — map a kept record to the sloth **chat** schema. +4. :func:`dedup_items` — collapse same-content examples, reusing + :func:`data_refinery.quality.checks.find_duplicate_groups` (the same + same-hash-same-scope primitive the ``dedup`` verb uses) over synthetic + per-example :class:`~data_refinery.store.envelope.Envelope` objects. +5. :func:`split_items` — a deterministic seeded shuffle, then a straight slice + by ratio: train and eval are **disjoint by construction** (a slice of a + shuffled list can never share an element). +6. :func:`build_lineage` — the per-example provenance + aggregate record. + +:func:`run_refine_dataset` composes all of the above and writes the three +output files; :func:`read_lineage` is the read-back half for ``refine lineage``. +""" + +from __future__ import annotations + +import json +import random +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from data_refinery.cli._errors import EXIT_ENV_ERROR, EXIT_USER_ERROR, CliError +from data_refinery.quality.checks import find_duplicate_groups +from data_refinery.store.envelope import Envelope, content_hash + +KNOWN_SCHEMAS = frozenset({"chat"}) +_VALID_ROLES = ("user", "assistant") # the two roles `refine dataset` emits +_RATING_BUCKETS = ("1", "2", "3", "4", "5") +_TMP_SUFFIX = ".tmp" + + +# --------------------------------------------------------------------------- +# Stage 1 — parse + shape-validate +# --------------------------------------------------------------------------- + + +def validate_export_line(obj: Any) -> dict[str, Any]: + """Shape-check one parsed ``colleague feedback export`` line. + + Required: ``task_id`` (non-empty str), ``request``/``summary``/``status`` + (str), ``at`` (non-empty str, carried verbatim into lineage as + ``exported_at``), and ``rating`` — present, but its *value* may be an int in + ``1..5`` **or** ``null`` (an ungraded work item; colleague's own feedback + store reads back an ungraded item as ``None``, never an error — this mirrors + that). ``notes``/``stats`` are optional; when present they are type-checked + but never required, since a real export stream may omit either. Returns + ``{valid, errors}`` — never raises (the caller decides skip-vs-count). + """ + if not isinstance(obj, dict): + return {"valid": False, "errors": ["line must be a JSON object"]} + errors: list[str] = [] + errors.extend(_required_field_errors(obj)) + errors.extend(_at_errors(obj)) + errors.extend(_rating_errors(obj)) + errors.extend(_notes_errors(obj)) + errors.extend(_stats_errors(obj.get("stats"))) + return {"valid": not errors, "errors": errors} + + +def _required_field_errors(obj: dict[str, Any]) -> list[str]: + """Shape-check the required ``task_id``/``request``/``summary``/``status`` string fields.""" + errors: list[str] = [] + for key in ("task_id", "request", "summary", "status"): + value = obj.get(key) + if key == "task_id": + if not isinstance(value, str) or not value: + errors.append("task_id: must be a non-empty string") + elif not isinstance(value, str): + errors.append(f"{key}: must be a string") + return errors + + +def _at_errors(obj: dict[str, Any]) -> list[str]: + """Shape-check the required ``at`` timestamp field.""" + if "at" not in obj or not isinstance(obj.get("at"), str) or not obj.get("at"): + return ["at: must be a non-empty string (ISO-8601 timestamp)"] + return [] + + +def _rating_errors(obj: dict[str, Any]) -> list[str]: + """Shape-check the required ``rating`` key (int 1-5, or ``null`` for ungraded).""" + if "rating" not in obj: + return ["rating: required key (int 1-5, or null for ungraded)"] + rating = obj["rating"] + if rating is not None and (not isinstance(rating, int) or isinstance(rating, bool)): + return ["rating: must be an int (1-5) or null"] + if isinstance(rating, int) and not isinstance(rating, bool) and not 1 <= rating <= 5: + return ["rating: must be between 1 and 5"] + return [] + + +def _notes_errors(obj: dict[str, Any]) -> list[str]: + """Shape-check the optional ``notes`` field when present.""" + if obj.get("notes") is not None and not isinstance(obj.get("notes"), str): + return ["notes: must be a string when present"] + return [] + + +def _stats_errors(stats: Any) -> list[str]: + """Shape-check the optional ``stats`` sub-object; split out for readability.""" + if stats is None: + return [] + if not isinstance(stats, dict): + return ["stats: must be an object when present"] + errors: list[str] = [] + for key in ("steps", "files_changed", "bytes_written"): + if key in stats and (not isinstance(stats[key], int) or isinstance(stats[key], bool)): + errors.append(f"stats.{key}: must be an int when present") + return errors + + +def parse_export_jsonl(path: Path) -> tuple[list[tuple[int, dict[str, Any]]], list[dict[str, Any]]]: + """Read *path* line by line; return ``(valid, malformed)``. + + ``valid`` is a list of ``(line_no, record)`` for every shape-valid line (blank + lines are skipped silently — they don't advance the logical line count, same + convention as sloth's own JSONL reader). ``malformed`` is a list of + ``{"line": n, "errors": [...]}`` diagnostics for a bad JSON parse or a + shape-validation failure — a malformed line is skipped, never silently + dropped: the caller surfaces one diagnostic per entry plus the count. + """ + try: + text = path.read_text(encoding="utf-8") + except UnicodeDecodeError as exc: + raise CliError( + code=EXIT_ENV_ERROR, + message=f"{path} is not valid UTF-8: {exc}", + remediation="re-export the file as UTF-8 text", + ) from exc + except OSError as exc: + raise CliError( + code=EXIT_ENV_ERROR, + message=f"cannot read {path}: {exc.strerror or exc}", + remediation="check that the file exists and is readable", + ) from exc + + valid: list[tuple[int, dict[str, Any]]] = [] + malformed: list[dict[str, Any]] = [] + for line_no, raw_line in enumerate(text.splitlines(), start=1): + line = raw_line.strip() + if not line: + continue + try: + obj = json.loads(line) + except json.JSONDecodeError as exc: + malformed.append({"line": line_no, "errors": [f"invalid JSON: {exc.msg}"]}) + continue + result = validate_export_line(obj) + if not result["valid"]: + malformed.append({"line": line_no, "errors": result["errors"]}) + continue + valid.append((line_no, obj)) + return valid, malformed + + +# --------------------------------------------------------------------------- +# Stage 2 — filter (graded + status ok + at/above threshold) +# --------------------------------------------------------------------------- + + +def _is_gradable(record: dict[str, Any], *, min_rating: int) -> bool: + rating = record.get("rating") + if not isinstance(rating, int) or isinstance(rating, bool): + return False # ungraded (null) — never enters silently + return record.get("status") == "ok" and rating >= min_rating + + +def filter_records( + records: list[tuple[int, dict[str, Any]]], *, min_rating: int +) -> tuple[list[tuple[int, dict[str, Any]]], int]: + """Keep only ``status == "ok"`` and a graded ``rating >= min_rating``. + + Returns ``(kept, filtered_out_count)``. An ungraded (``rating: null``) or + below-threshold or non-``"ok"``-status line is counted as filtered, never + enters the dataset silently. + """ + kept = [r for r in records if _is_gradable(r[1], min_rating=min_rating)] + return kept, len(records) - len(kept) + + +# --------------------------------------------------------------------------- +# Stage 3 — map to the sloth chat schema +# --------------------------------------------------------------------------- + + +def to_example(record: dict[str, Any]) -> dict[str, Any]: + """Map one kept record to the sloth **chat** schema. + + Shape matches ``sloth.tune.datasets._validate_chat_record`` exactly: the + record has **only** the ``"messages"`` key (an extra key fails sloth's + ``extra = set(record.keys()) - CHAT_KEYS`` check), a non-empty list of + ``{"role", "content"}`` messages, roles ``"user"`` then ``"assistant"``. + """ + return { + "messages": [ + {"role": "user", "content": record["request"]}, + {"role": "assistant", "content": record["summary"]}, + ] + } + + +# --------------------------------------------------------------------------- +# Stage 4 — dedup (reuses the quality-check primitive) +# --------------------------------------------------------------------------- + + +@dataclass +class ExampleItem: + """One kept, mapped, deduped training example plus its provenance.""" + + task_id: str + rating: int + at: str + example: dict[str, Any] + content_sha256: str + + +def _canonical_json(example: dict[str, Any]) -> str: + """A stable text form of *example* to hash — key order is fixed by construction.""" + return json.dumps(example, sort_keys=True, ensure_ascii=False) + + +def dedup_items( + records: list[tuple[int, dict[str, Any]]], +) -> tuple[list[ExampleItem], int]: + """Map every kept record to an example, then dedup by content hash. + + Reuses :func:`data_refinery.quality.checks.find_duplicate_groups` — the + same same-hash-same-scope primitive the ``dedup`` verb runs over the store — + by wrapping each mapped example in a synthetic, default-scope + :class:`Envelope` (id = the record's position, content = the canonical JSON + of the example). The **first** occurrence of a duplicate content hash is + kept (matching ``find_duplicate_groups``'s own "first id kept" rule); + later occurrences are dropped. Returns ``(kept_items, duplicates_removed)`` + in original input order. + """ + envelopes: list[Envelope] = [] + items: list[ExampleItem] = [] + for i, (_line_no, record) in enumerate(records): + example = to_example(record) + canonical = _canonical_json(example) + envelope = Envelope(id=str(i), content=canonical) + envelopes.append(envelope) + items.append( + ExampleItem( + task_id=record["task_id"], + rating=record["rating"], + at=record["at"], + example=example, + content_sha256=envelope.hash, + ) + ) + + groups = find_duplicate_groups(envelopes) + removed_ids = {rid for group in groups for rid in group["removed"]} + kept = [item for i, item in enumerate(items) if str(i) not in removed_ids] + return kept, len(removed_ids) + + +# --------------------------------------------------------------------------- +# Stage 5 — deterministic disjoint split +# --------------------------------------------------------------------------- + + +def parse_split_ratio(split: str) -> tuple[int, int]: + """Parse ``"A/B"`` into ``(a, b)`` non-negative ints, at least one positive.""" + parts = split.split("/") + if len(parts) != 2: + raise CliError( + code=EXIT_USER_ERROR, + message=f"--split must be 'A/B' (e.g. 90/10), got {split!r}", + remediation="pass two integers separated by '/', e.g. --split 90/10", + ) + try: + a, b = int(parts[0]), int(parts[1]) + except ValueError as exc: + raise CliError( + code=EXIT_USER_ERROR, + message=f"--split must be two integers separated by '/', got {split!r}", + remediation="pass two integers separated by '/', e.g. --split 90/10", + ) from exc + if a < 0 or b < 0 or a + b == 0: + raise CliError( + code=EXIT_USER_ERROR, + message=f"--split values must be non-negative and not both zero, got {split!r}", + remediation="pass two non-negative integers that aren't both zero, e.g. --split 90/10", + ) + return a, b + + +def split_items( + items: list[ExampleItem], *, split: str, seed: int +) -> tuple[list[ExampleItem], list[ExampleItem]]: + """Deterministic seeded shuffle, then slice by ratio. + + Train and eval are **disjoint by construction**: shuffling a list and then + slicing it into two contiguous, non-overlapping parts structurally cannot + place the same element in both halves — there is no separate membership + test that could itself be wrong. Same *items* + same *seed* always yields + the same shuffle order (``random.Random(seed)`` is a pure, seeded PRNG — + nothing wall-clock-derived), so re-running `refine dataset` on the same + input is reproducible. + """ + a, b = parse_split_ratio(split) + order = list(range(len(items))) + # A deterministic, seeded reproducible shuffle for a dataset split, never a + # security/cryptographic use of randomness. + random.Random(seed).shuffle(order) # nosec B311 (Sonar S2245 waived in sonar cfg) + shuffled = [items[i] for i in order] + train_fraction = a / (a + b) + train_size = round(len(shuffled) * train_fraction) + return shuffled[:train_size], shuffled[train_size:] + + +# --------------------------------------------------------------------------- +# Stage 6 — lineage (provenance + aggregate) +# --------------------------------------------------------------------------- + + +def _grade_distribution(records: list[tuple[int, dict[str, Any]]]) -> dict[str, int]: + """Tally ratings across every *valid* parsed line (graded or not). + + Reports the full grading spread the operator has to choose a threshold + against — not just the examples that made the cut — with a zero-filled + bucket for every rating 1-5 so a consumer never has to guess a missing key. + """ + dist = dict.fromkeys(_RATING_BUCKETS, 0) + for _line_no, record in records: + rating = record.get("rating") + if isinstance(rating, int) and not isinstance(rating, bool) and 1 <= rating <= 5: + dist[str(rating)] += 1 + return dist + + +def build_lineage( + *, + train_items: list[ExampleItem], + eval_items: list[ExampleItem], + valid_records: list[tuple[int, dict[str, Any]]], + input_count: int, + malformed_count: int, + filtered_count: int, + duplicates_removed: int, + min_rating: int, + split: str, + seed: int, + train_sha256: str, + eval_sha256: str, +) -> dict[str, Any]: + """Build the ``lineage.json`` payload: per-example provenance + aggregate.""" + examples = [ + { + "source": "colleague", + "task_id": item.task_id, + "rating": item.rating, + "content_sha256": item.content_sha256, + "exported_at": item.at, + } + for item in (*train_items, *eval_items) + ] + aggregate = { + "input_count": input_count, + "malformed_count": malformed_count, + "filtered_count": filtered_count, + "duplicates_removed": duplicates_removed, + "kept_count": len(train_items) + len(eval_items), + "grade_distribution": _grade_distribution(valid_records), + "min_rating": min_rating, + "split": split, + "seed": seed, + "train_size": len(train_items), + "eval_size": len(eval_items), + "train_sha256": train_sha256, + "eval_sha256": eval_sha256, + } + return {"examples": examples, "aggregate": aggregate} + + +# --------------------------------------------------------------------------- +# Orchestration + I/O +# --------------------------------------------------------------------------- + + +def _atomic_write(path: Path, text: str) -> None: + """Write *text* to *path* atomically (temp sibling + ``os.replace``). + + Mirrors :meth:`data_refinery.store.backends.files.FilesBackend._atomic_write` + — a crash leaves either no file or the old one, never a half-written one. + """ + import os + + tmp = path.with_name(path.name + _TMP_SUFFIX) + try: + path.parent.mkdir(parents=True, exist_ok=True) + tmp.write_text(text, encoding="utf-8") + os.replace(tmp, path) + except OSError as exc: + try: + tmp.unlink() + except OSError: # pragma: no cover - best effort cleanup + pass + raise CliError( + code=EXIT_ENV_ERROR, + message=f"could not write {path}: {exc}", + remediation=f"check free space and permissions on {path.parent}", + ) from exc + + +def _serialize_examples(items: list[ExampleItem]) -> str: + return "".join(json.dumps(item.example, ensure_ascii=False) + "\n" for item in items) + + +def run_refine_dataset( + from_path: Path, + *, + schema: str = "chat", + min_rating: int = 4, + split: str = "90/10", + seed: int = 0, + out_dir: Path, +) -> dict[str, Any]: + """Run the full pipeline and write ``train.jsonl``/``eval.jsonl``/``lineage.json``. + + Returns ``{"diagnostics": [str, ...], "payload": {...}}`` — one diagnostic + string per malformed line (for the caller to route to stderr) plus the + result payload (paths + the same aggregate that lands in ``lineage.json``). + """ + if schema not in KNOWN_SCHEMAS: + raise CliError( + code=EXIT_USER_ERROR, + message=f"unknown schema {schema!r}; must be one of {sorted(KNOWN_SCHEMAS)}", + remediation="pass --schema chat (the only schema refine dataset supports today)", + ) + # Validate the split string early (before any I/O) so a bad --split fails + # fast with a clean error, not after writing partial output. + parse_split_ratio(split) + + valid_records, malformed = parse_export_jsonl(from_path) + diagnostics = [ + f"refine: skipping malformed line {m['line']}: {'; '.join(m['errors'])}" for m in malformed + ] + + kept_records, filtered_count = filter_records(valid_records, min_rating=min_rating) + items, duplicates_removed = dedup_items(kept_records) + train_items, eval_items = split_items(items, split=split, seed=seed) + + out_dir = Path(out_dir) + train_path = out_dir / "train.jsonl" + eval_path = out_dir / "eval.jsonl" + lineage_path = out_dir / "lineage.json" + + train_text = _serialize_examples(train_items) + eval_text = _serialize_examples(eval_items) + train_sha256 = content_hash(train_text) + eval_sha256 = content_hash(eval_text) + + lineage = build_lineage( + train_items=train_items, + eval_items=eval_items, + valid_records=valid_records, + input_count=len(valid_records) + len(malformed), + malformed_count=len(malformed), + filtered_count=filtered_count, + duplicates_removed=duplicates_removed, + min_rating=min_rating, + split=split, + seed=seed, + train_sha256=train_sha256, + eval_sha256=eval_sha256, + ) + + _atomic_write(train_path, train_text) + _atomic_write(eval_path, eval_text) + _atomic_write(lineage_path, json.dumps(lineage, ensure_ascii=False, indent=2) + "\n") + + payload = { + "out": str(out_dir.resolve()), + "train_file": str(train_path), + "eval_file": str(eval_path), + "lineage_file": str(lineage_path), + "aggregate": lineage["aggregate"], + } + return {"diagnostics": diagnostics, "payload": payload} + + +def read_lineage(dataset_dir: Path) -> dict[str, Any]: + """Read back ``/lineage.json``. Raises CliError if absent/corrupt.""" + lineage_path = Path(dataset_dir) / "lineage.json" + if not lineage_path.is_file(): + raise CliError( + code=EXIT_USER_ERROR, + message=f"no lineage.json found in {dataset_dir}", + remediation=( + "run 'data-refinery refine dataset --out ' first, or point --dataset " + "at a directory it produced" + ), + ) + try: + text = lineage_path.read_text(encoding="utf-8") + except UnicodeDecodeError as exc: + raise CliError( + code=EXIT_ENV_ERROR, + message=f"{lineage_path} is not valid UTF-8: {exc}", + remediation="the lineage.json is corrupt; re-run 'refine dataset' to regenerate it", + ) from exc + except OSError as exc: + raise CliError( + code=EXIT_ENV_ERROR, + message=f"cannot read {lineage_path}: {exc.strerror or exc}", + remediation="check permissions on the dataset directory", + ) from exc + try: + return json.loads(text) + except json.JSONDecodeError as exc: + raise CliError( + code=EXIT_ENV_ERROR, + message=f"corrupt lineage.json in {dataset_dir}: {exc.msg}", + remediation="regenerate the dataset with 'data-refinery refine dataset'", + ) from exc diff --git a/docs/contract.md b/docs/contract.md index 4f8f4ed..43f0471 100644 --- a/docs/contract.md +++ b/docs/contract.md @@ -6,8 +6,8 @@ shape, exit-code meaning, or the image tag scheme requires a **version bump** in [`pyproject.toml`](../pyproject.toml) (the `version-check` CI job enforces a bump on every PR). -- **Contract version:** `3` (Wave 3 — adds the store-migration endpoint to the - Wave 2 store + data-quality surface). +- **Contract version:** `4` (Wave 4 — adds the `refine` surface: graded + colleague work items to sloth-ready training data, with provenance). - **Package version pinned by a consumer:** see `pyproject.toml` `version`. ## Wave 1 — the storage stack (stable) @@ -207,6 +207,138 @@ layout, so it owns the ignore pattern that tracks it; this keeps the `.gitignore write-path sink (the consumer's prior pythonsecurity:S2083) on the storage owner. Continues issues #8 / #1. +## Wave 4 — `refine`: graded colleague work items to training data (stable) + +Turns a `colleague feedback export` JSONL stream (one graded work item per +line) into a sloth-ready **chat**-schema train/eval split plus a provenance +`lineage.json` (agentculture/data-refinery-cli#14). Stdlib only: no network, +no LLM call is made by data-refinery itself. + +### Input line shape (one JSON object per line) + +```json +{ + "task_id": "t-abc123", + "request": "...", + "summary": "...", + "rating": 4, + "notes": "...", + "status": "ok", + "at": "2026-07-05T12:00:00+00:00", + "stats": {"steps": 6, "files_changed": 2, "bytes_written": 1345} +} +``` + +`task_id`/`request`/`summary`/`status` are strings; `at` is a non-empty string +(carried verbatim into lineage as `exported_at`). `rating` is an int `1`-`5` +**or** `null` — an ungraded work item (colleague's own feedback store reads +back an ungraded item as `None`, never an error; this mirrors that). `notes` +and `stats` are optional; when present their fields are type-checked but never +required. A line that fails this shape check (bad JSON, wrong types, an +out-of-range rating, a non-int `stats` field) is a **malformed** line: it is +skipped with one diagnostic on stderr and counted in `malformed_count` — never +silently dropped. + +### `data-refinery refine dataset` + +```text +data-refinery refine dataset --from [--schema chat] + [--min-rating N] [--split A/B] [--seed N] --out [--json] +``` + +Defaults: `--schema chat` (the only schema supported today), `--min-rating 4`, +`--split 90/10`, `--seed 0`. + +**Filter** (never enters the dataset silently): a line is kept iff +`status == "ok"` **and** `rating` is graded (non-`null`) **and** +`rating >= --min-rating`. + +**Map** (chat schema) — one example per kept line: + +```json +{"messages": [ + {"role": "user", "content": ""}, + {"role": "assistant", "content": ""} +]} +``` + +The emitted record has **only** the `"messages"` key (matching +`sloth.tune.datasets`'s chat-schema validator exactly — `sloth validate` +accepts the output verbatim). + +**Dedup** — examples with identical mapped content (by sha256, the same +same-hash primitive `data-refinery dedup` runs over the store) collapse to one +survivor (first occurrence kept), **before** the split. + +**Split** — a deterministic seeded shuffle (`--seed`) of the deduped examples, +then a straight slice by `--split`'s ratio (`A/(A+B)` to train, the remainder +to eval). Train and eval are **disjoint by construction**: slicing a shuffled +list cannot place the same element in both halves. The same input, the same +`--seed`, and the same flags always reproduce byte-identical output. + +Writes three files under `--out`: + +- `train.jsonl` / `eval.jsonl` — one chat-schema example per line. +- `lineage.json`: + +```json +{ + "examples": [ + {"source": "colleague", "task_id": "t-abc123", "rating": 4, + "content_sha256": "...", "exported_at": "2026-07-05T12:00:00+00:00"} + ], + "aggregate": { + "input_count": 12, "malformed_count": 1, "filtered_count": 3, + "duplicates_removed": 1, "kept_count": 7, + "grade_distribution": {"1": 0, "2": 1, "3": 2, "4": 3, "5": 5}, + "min_rating": 4, "split": "90/10", "seed": 0, + "train_size": 6, "eval_size": 1, + "train_sha256": "...", "eval_sha256": "..." + } +} +``` + +`grade_distribution` tallies every *valid* parsed line's rating (graded or +not, regardless of `--min-rating`) — the full grading spread, zero-filled for +every bucket `1`-`5`. `train_sha256`/`eval_sha256` are `sha256` over the +exact bytes of the emitted `train.jsonl`/`eval.jsonl` (integrity-style, the +same primitive the store's envelope hash uses). + +### `data-refinery refine lineage` + +```text +data-refinery refine lineage --dataset [--json] +``` + +Reads back `/lineage.json` verbatim. Exits `1` with a `hint:` when +`lineage.json` is absent (an "unknown path" user error, same convention as +every other verb); a corrupt `lineage.json` exits `2` (environment fault). + +### Invariants the consumer can rely on + +- **No silent admission** — an ungraded, non-`"ok"`-status, or + below-threshold line never becomes a training example; a malformed line is + always both diagnosed (stderr) and counted (`malformed_count`). +- **Disjoint split** — no `content_sha256` appears in both `train.jsonl` and + `eval.jsonl`. +- **Determinism** — identical `--from`/flags/`--seed` reproduce byte-identical + `train.jsonl`/`eval.jsonl`/`lineage.json`. +- **sloth-ready** — the `chat` schema output matches + `sloth.tune.datasets.validate_dataset(schema="chat")` exactly. + +### Honest limits + +- `--schema chat` is the only schema today; a `task`-schema mapping (colleague + work items don't carry a natural `task`/`input`/`expected_output` split) is + a documented follow-up, not built. +- A dataset small enough that a split ratio rounds one side to zero examples + is not rejected — `refine dataset` reports the true sizes; `sloth validate` + itself rejects an empty JSONL file, so a consumer validating a too-small + split will see that failure downstream, not from `refine`. +- `--min-rating` and `--split` accept any integer inputs `refine dataset` can + reason about (e.g. a `--split` where one side is `0`); there is no floor + enforcing both `train`/`eval` be non-empty. + ## Versioning policy | Change | Requires | diff --git a/pyproject.toml b/pyproject.toml index 2230c67..07ac0e8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "data-refinery-cli" -version = "0.9.0" +version = "0.10.0" description = "Agent and CLI for data quality in storage and retrieval — validating, deduplicating, and checking the integrity and freshness of data as it is stored and fetched. Split out of eidetic-cli so eidetic keeps agent-memory; sibling to daria, the Data Refinery Intelligent Agent." readme = "README.md" license = "Apache-2.0" diff --git a/sonar-project.properties b/sonar-project.properties index ed98873..b97baee 100644 --- a/sonar-project.properties +++ b/sonar-project.properties @@ -16,6 +16,17 @@ sonar.python.version=3.12 # Don't analyze noise. sonar.exclusions=**/__pycache__/**,**/.venv/**,uv.lock,.claude/skills/** +# Suppress the weak-PRNG rule (python:S2245) on the refine dataset shuffle. +# `split_items()` in data_refinery/refine/dataset.py uses +# `random.Random(seed).shuffle(...)` for a *deterministic, reproducible* +# train/eval split — a non-cryptographic use of randomness (the whole point is +# that the same seed reproduces the same split; `secrets` can't be seeded). +# Reviewed safe: the inline `# nosec B311` covers bandit, this covers Sonar +# (inline `# NOSONAR` is not honored for this rule here). +sonar.issue.ignore.multicriteria=e1 +sonar.issue.ignore.multicriteria.e1.ruleKey=python:S2245 +sonar.issue.ignore.multicriteria.e1.resourceKey=**/refine/dataset.py + # Block CI on a failing quality gate. The scanner polls SonarCloud and exits # non-zero on a red gate (coverage regression, new bugs/vulnerabilities). Only # takes effect when the SonarCloud Scan step runs — that step is guarded by diff --git a/tests/test_refine_dataset.py b/tests/test_refine_dataset.py new file mode 100644 index 0000000..093182c --- /dev/null +++ b/tests/test_refine_dataset.py @@ -0,0 +1,441 @@ +"""Tests for ``data-refinery refine dataset`` (library + CLI). + +Sloth's chat-schema rules (`sloth.tune.datasets._validate_chat_record` / +`_validate_chat_message` in the sibling unsloth-cli repo) are asserted here +structurally rather than imported — a test in *this* repo must not depend on a +sibling checkout being present on the test machine / in CI. The rules asserted +below (record has exactly the key ``"messages"``, a non-empty list, each +message exactly ``{"role", "content"}`` with role in +``{"system","user","assistant"}`` and string content) were verified by hand +against the real `validate_dataset` against this module's actual output. +""" + +from __future__ import annotations + +import json + +import pytest + +from data_refinery.cli import main +from data_refinery.refine import dataset + +_VALID_ROLES = {"system", "user", "assistant"} + + +def _assert_sloth_chat_record(record: dict) -> None: + """Structurally mirror sloth's chat-schema validation (see module docstring).""" + assert set(record.keys()) == {"messages"} + messages = record["messages"] + assert isinstance(messages, list) and messages + for msg in messages: + assert set(msg.keys()) == {"role", "content"} + assert msg["role"] in _VALID_ROLES + assert isinstance(msg["content"], str) + + +def _write_jsonl(path, lines: list[dict]) -> None: + path.write_text("".join(json.dumps(line) + "\n" for line in lines), encoding="utf-8") + + +def _line(task_id: str, rating, *, status: str = "ok", **overrides) -> dict: + base = { + "task_id": task_id, + "request": f"request for {task_id}", + "summary": f"summary for {task_id}", + "rating": rating, + "notes": "", + "status": status, + "at": "2026-07-01T00:00:00+00:00", + "stats": {"steps": 1, "files_changed": 1, "bytes_written": 10}, + } + base.update(overrides) + return base + + +# --- pure library: validate_export_line --------------------------------- + + +def test_validate_export_line_accepts_null_rating() -> None: + result = dataset.validate_export_line(_line("t1", None)) + assert result["valid"] is True + + +def test_validate_export_line_rejects_out_of_range_rating() -> None: + result = dataset.validate_export_line(_line("t1", 6)) + assert result["valid"] is False + assert any("rating" in e for e in result["errors"]) + + +def test_validate_export_line_rejects_bool_rating() -> None: + result = dataset.validate_export_line(_line("t1", True)) + assert result["valid"] is False + + +def test_validate_export_line_requires_task_id() -> None: + bad = _line("t1", 5) + bad["task_id"] = "" + result = dataset.validate_export_line(bad) + assert result["valid"] is False + assert any("task_id" in e for e in result["errors"]) + + +def test_validate_export_line_rejects_non_object() -> None: + assert dataset.validate_export_line([1, 2, 3])["valid"] is False + + +def test_validate_export_line_tolerates_missing_stats_and_notes() -> None: + line = _line("t1", 5) + del line["stats"] + del line["notes"] + assert dataset.validate_export_line(line)["valid"] is True + + +def test_validate_export_line_rejects_bad_stats_type() -> None: + line = _line("t1", 5, stats={"steps": "six"}) + result = dataset.validate_export_line(line) + assert result["valid"] is False + assert any("stats.steps" in e for e in result["errors"]) + + +# --- the 3-line fixture (ratings 5,4,2; threshold 4) --------------------- + + +@pytest.fixture +def three_line_fixture(tmp_path): + path = tmp_path / "graded.jsonl" + _write_jsonl( + path, + [ + _line("t1", 5, at="2026-07-01T00:00:00+00:00"), + _line("t2", 4, at="2026-07-02T00:00:00+00:00"), + _line("t3", 2, at="2026-07-03T00:00:00+00:00"), + ], + ) + return path + + +def test_three_line_fixture_keeps_exactly_two_above_threshold(three_line_fixture, tmp_path) -> None: + out = tmp_path / "out" + result = dataset.run_refine_dataset(three_line_fixture, min_rating=4, out_dir=out) + agg = result["payload"]["aggregate"] + assert agg["input_count"] == 3 + assert agg["malformed_count"] == 0 + assert agg["kept_count"] == 2 + assert agg["filtered_count"] == 1 + + lineage = json.loads((out / "lineage.json").read_text()) + task_ids = {ex["task_id"] for ex in lineage["examples"]} + assert task_ids == {"t1", "t2"} + assert "t3" not in task_ids # rating-2 line is absent + + +def test_three_line_fixture_provenance_shape(three_line_fixture, tmp_path) -> None: + out = tmp_path / "out" + dataset.run_refine_dataset(three_line_fixture, min_rating=4, out_dir=out) + lineage = json.loads((out / "lineage.json").read_text()) + for ex in lineage["examples"]: + assert set(ex.keys()) == {"source", "task_id", "rating", "content_sha256", "exported_at"} + assert ex["source"] == "colleague" + assert ex["rating"] >= 4 + + +def test_three_line_fixture_examples_pass_sloth_chat_schema(three_line_fixture, tmp_path) -> None: + out = tmp_path / "out" + dataset.run_refine_dataset(three_line_fixture, min_rating=4, seed=1, split="50/50", out_dir=out) + for name in ("train.jsonl", "eval.jsonl"): + text = (out / name).read_text() + for raw_line in text.splitlines(): + if raw_line.strip(): + _assert_sloth_chat_record(json.loads(raw_line)) + + +def test_three_line_fixture_cli_json_mode(three_line_fixture, tmp_path, capsys) -> None: + out = tmp_path / "out" + rc = main( + [ + "refine", + "dataset", + "--from", + str(three_line_fixture), + "--min-rating", + "4", + "--out", + str(out), + "--json", + ] + ) + assert rc == 0 + payload = json.loads(capsys.readouterr().out) + assert payload["aggregate"]["kept_count"] == 2 + + +# --- malformed lines are skipped, never silently dropped ----------------- + + +def test_malformed_line_skipped_with_diagnostic_and_count(tmp_path, capsys) -> None: + path = tmp_path / "graded.jsonl" + path.write_text( + json.dumps(_line("t1", 5)) + "\n" + "not json at all\n" + + json.dumps({"task_id": "t2"}) + + "\n" # missing required keys + + json.dumps(_line("t3", 4)) + + "\n", + encoding="utf-8", + ) + out = tmp_path / "out" + rc = main(["refine", "dataset", "--from", str(path), "--min-rating", "4", "--out", str(out)]) + assert rc == 0 + err = capsys.readouterr().err + assert err.count("refine: skipping malformed line") == 2 + assert "line 2" in err + assert "line 3" in err + + lineage = json.loads((out / "lineage.json").read_text()) + assert lineage["aggregate"]["malformed_count"] == 2 + assert lineage["aggregate"]["input_count"] == 4 # 2 valid + 2 malformed + + +# --- filter: status + ungraded (null rating) ------------------------------ + + +def test_non_ok_status_is_filtered(tmp_path) -> None: + path = tmp_path / "graded.jsonl" + _write_jsonl(path, [_line("t1", 5, status="error"), _line("t2", 5, status="ok")]) + out = tmp_path / "out" + result = dataset.run_refine_dataset(path, min_rating=4, out_dir=out) + agg = result["payload"]["aggregate"] + assert agg["kept_count"] == 1 + assert agg["filtered_count"] == 1 + + +def test_ungraded_null_rating_is_filtered_not_malformed(tmp_path) -> None: + path = tmp_path / "graded.jsonl" + _write_jsonl(path, [_line("t1", None), _line("t2", 5)]) + out = tmp_path / "out" + result = dataset.run_refine_dataset(path, min_rating=4, out_dir=out) + agg = result["payload"]["aggregate"] + assert agg["malformed_count"] == 0 # a null rating is valid shape, just ungraded + assert agg["filtered_count"] == 1 + assert agg["kept_count"] == 1 + + +# --- dedup by content hash ------------------------------------------------- + + +def test_dedup_collapses_identical_content(tmp_path) -> None: + path = tmp_path / "graded.jsonl" + _write_jsonl( + path, + [ + _line("t1", 5, request="same request", summary="same summary"), + _line("t2", 5, request="same request", summary="same summary"), + _line("t3", 5, request="different request", summary="different summary"), + ], + ) + out = tmp_path / "out" + result = dataset.run_refine_dataset(path, min_rating=4, out_dir=out) + agg = result["payload"]["aggregate"] + assert agg["duplicates_removed"] == 1 + assert agg["kept_count"] == 2 + + lineage = json.loads((out / "lineage.json").read_text()) + assert {ex["task_id"] for ex in lineage["examples"]} == {"t1", "t3"} # first id kept + + +# --- disjoint split property (20-line fixture) ---------------------------- + + +@pytest.fixture +def twenty_line_fixture(tmp_path): + path = tmp_path / "graded.jsonl" + lines = [_line(f"t{i}", 5, request=f"req {i}", summary=f"sum {i}") for i in range(20)] + _write_jsonl(path, lines) + return path + + +def test_disjoint_split_no_shared_content_hash(twenty_line_fixture, tmp_path) -> None: + out = tmp_path / "out" + dataset.run_refine_dataset( + twenty_line_fixture, min_rating=4, split="70/30", seed=7, out_dir=out + ) + lineage = json.loads((out / "lineage.json").read_text()) + agg = lineage["aggregate"] + assert agg["train_size"] + agg["eval_size"] == agg["kept_count"] == 20 + assert agg["train_size"] > 0 and agg["eval_size"] > 0 + + train_requests = { + json.loads(line)["messages"][0]["content"] + for line in (out / "train.jsonl").read_text().splitlines() + if line.strip() + } + eval_requests = { + json.loads(line)["messages"][0]["content"] + for line in (out / "eval.jsonl").read_text().splitlines() + if line.strip() + } + assert train_requests.isdisjoint(eval_requests) + + # also check by the lineage-recorded content_sha256 directly + all_shas = [ex["content_sha256"] for ex in lineage["examples"]] + assert len(all_shas) == len(set(all_shas)) # no duplicate hash slipped through + + +def test_disjoint_split_ratio_produces_expected_sizes(twenty_line_fixture, tmp_path) -> None: + out = tmp_path / "out" + dataset.run_refine_dataset( + twenty_line_fixture, min_rating=4, split="80/20", seed=0, out_dir=out + ) + lineage = json.loads((out / "lineage.json").read_text()) + agg = lineage["aggregate"] + assert agg["train_size"] == 16 + assert agg["eval_size"] == 4 + + +# --- deterministic given the same seed ------------------------------------ + + +def test_deterministic_given_same_seed(twenty_line_fixture, tmp_path) -> None: + out_a = tmp_path / "out_a" + out_b = tmp_path / "out_b" + dataset.run_refine_dataset(twenty_line_fixture, min_rating=4, seed=42, out_dir=out_a) + dataset.run_refine_dataset(twenty_line_fixture, min_rating=4, seed=42, out_dir=out_b) + assert (out_a / "train.jsonl").read_text() == (out_b / "train.jsonl").read_text() + assert (out_a / "eval.jsonl").read_text() == (out_b / "eval.jsonl").read_text() + + +def test_different_seed_can_change_split_order(twenty_line_fixture, tmp_path) -> None: + out_a = tmp_path / "out_a" + out_b = tmp_path / "out_b" + dataset.run_refine_dataset(twenty_line_fixture, min_rating=4, seed=0, out_dir=out_a) + dataset.run_refine_dataset(twenty_line_fixture, min_rating=4, seed=1, out_dir=out_b) + # both runs are internally disjoint and deterministic; they need not match + # each other, but each must reproduce itself (covered above). + assert (out_a / "train.jsonl").exists() and (out_b / "train.jsonl").exists() + + +# --- --split parsing -------------------------------------------------------- + + +def test_parse_split_ratio_rejects_bad_format() -> None: + from data_refinery.cli._errors import CliError + + with pytest.raises(CliError): + dataset.parse_split_ratio("90-10") + + +def test_parse_split_ratio_rejects_both_zero() -> None: + from data_refinery.cli._errors import CliError + + with pytest.raises(CliError): + dataset.parse_split_ratio("0/0") + + +def test_parse_split_ratio_ok() -> None: + assert dataset.parse_split_ratio("90/10") == (90, 10) + + +# --- schema flag ------------------------------------------------------------- + + +def test_unknown_schema_raises_clean_error(three_line_fixture, tmp_path) -> None: + from data_refinery.cli._errors import CliError + + with pytest.raises(CliError): + dataset.run_refine_dataset( + three_line_fixture, schema="task", min_rating=4, out_dir=tmp_path / "out" + ) + + +def test_missing_from_file_cli_exits_1(tmp_path, capsys) -> None: + rc = main( + [ + "refine", + "dataset", + "--from", + str(tmp_path / "does-not-exist.jsonl"), + "--out", + str(tmp_path / "out"), + ] + ) + assert rc == 1 + assert "hint:" in capsys.readouterr().err + + +def test_non_utf8_export_raises_env_error(tmp_path) -> None: + """A non-UTF-8 ``--from`` file is a structured env error, not "unexpected". + + ``UnicodeDecodeError`` is a ``ValueError``, not an ``OSError`` — before the + fix it slipped past ``parse_export_jsonl``'s ``except OSError`` and reached + the dispatcher's catch-all (exit 1, "unexpected"). It is now caught and + re-raised as ``EXIT_ENV_ERROR`` (exit 2) with a UTF-8 remediation. + """ + from data_refinery.cli._errors import EXIT_ENV_ERROR, CliError + + bad = tmp_path / "latin1.jsonl" + bad.write_bytes(b'{"task_id": "caf\xe9 not utf-8"}\n') # lone 0xE9, invalid UTF-8 + with pytest.raises(CliError) as exc_info: + dataset.parse_export_jsonl(bad) + assert exc_info.value.code == EXIT_ENV_ERROR + assert "UTF-8" in exc_info.value.message + + +def test_non_utf8_lineage_raises_env_error(tmp_path) -> None: + """A corrupt (non-UTF-8) ``lineage.json`` is a structured env error too. + + Mirrors :func:`test_non_utf8_export_raises_env_error` for the read-back path + (``read_lineage``), the second site that caught only ``OSError``. + """ + from data_refinery.cli._errors import EXIT_ENV_ERROR, CliError + + (tmp_path / "lineage.json").write_bytes(b'{"aggregate": "\xff\xfe"}\n') + with pytest.raises(CliError) as exc_info: + dataset.read_lineage(tmp_path) + assert exc_info.value.code == EXIT_ENV_ERROR + assert "UTF-8" in exc_info.value.message + + +# --- main()-level wiring: verbs are actually registered ------------------- + + +def test_refine_help_wiring(capsys) -> None: + with pytest.raises(SystemExit) as exc: + main(["refine", "--help"]) + assert exc.value.code == 0 + out = capsys.readouterr().out + assert "dataset" in out and "lineage" in out + + +def test_refine_dataset_help_wiring(capsys) -> None: + with pytest.raises(SystemExit) as exc: + main(["refine", "dataset", "--help"]) + assert exc.value.code == 0 + out = capsys.readouterr().out + assert "--from" in out and "--out" in out and "--split" in out + + +def test_refine_lineage_help_wiring(capsys) -> None: + with pytest.raises(SystemExit) as exc: + main(["refine", "lineage", "--help"]) + assert exc.value.code == 0 + assert "--dataset" in capsys.readouterr().out + + +def test_refine_bare_noun_prints_overview(capsys) -> None: + rc = main(["refine"]) + assert rc == 0 + assert capsys.readouterr().out.strip() + + +def test_refine_overview_json_shape(capsys) -> None: + rc = main(["refine", "overview", "--json"]) + assert rc == 0 + payload = json.loads(capsys.readouterr().out) + assert payload["subject"] == "data-refinery refine" + assert isinstance(payload["sections"], list) and payload["sections"] + + +def test_explain_refine_entry_exists(capsys) -> None: + rc = main(["explain", "refine"]) + assert rc == 0 + assert "refine" in capsys.readouterr().out.lower() diff --git a/tests/test_refine_lineage.py b/tests/test_refine_lineage.py new file mode 100644 index 0000000..e3e400b --- /dev/null +++ b/tests/test_refine_lineage.py @@ -0,0 +1,133 @@ +"""Tests for ``data-refinery refine lineage`` — the lineage.json read-back verb.""" + +from __future__ import annotations + +import json + +import pytest + +from data_refinery.cli import main +from data_refinery.refine import dataset + + +def _write_jsonl(path, lines: list[dict]) -> None: + path.write_text("".join(json.dumps(line) + "\n" for line in lines), encoding="utf-8") + + +def _line(task_id: str, rating, **overrides) -> dict: + base = { + "task_id": task_id, + "request": f"request {task_id}", + "summary": f"summary {task_id}", + "rating": rating, + "notes": "", + "status": "ok", + "at": "2026-07-01T00:00:00+00:00", + "stats": {"steps": 1, "files_changed": 1, "bytes_written": 10}, + } + base.update(overrides) + return base + + +@pytest.fixture +def built_dataset(tmp_path): + graded = tmp_path / "graded.jsonl" + _write_jsonl( + graded, + [_line("t1", 5), _line("t2", 4), _line("t3", 3), _line("t4", 2), _line("t5", 1)], + ) + out = tmp_path / "out" + dataset.run_refine_dataset(graded, min_rating=4, split="50/50", seed=3, out_dir=out) + return out + + +# --- round trip ------------------------------------------------------------- + + +def test_lineage_round_trip_via_library(built_dataset) -> None: + lineage = dataset.read_lineage(built_dataset) + on_disk = json.loads((built_dataset / "lineage.json").read_text()) + assert lineage == on_disk + assert lineage["aggregate"]["kept_count"] == 2 # ratings 5 and 4 only + + +def test_lineage_round_trip_via_cli_json(built_dataset, capsys) -> None: + rc = main(["refine", "lineage", "--dataset", str(built_dataset), "--json"]) + assert rc == 0 + payload = json.loads(capsys.readouterr().out) + assert payload["aggregate"]["min_rating"] == 4 + assert payload["aggregate"]["split"] == "50/50" + assert payload["aggregate"]["seed"] == 3 + assert {ex["task_id"] for ex in payload["examples"]} == {"t1", "t2"} + + +def test_lineage_grade_distribution_over_all_valid_lines(built_dataset) -> None: + lineage = dataset.read_lineage(built_dataset) + # 5 valid lines rated 5,4,3,2,1 -> one each, regardless of the min_rating filter + assert lineage["aggregate"]["grade_distribution"] == { + "1": 1, + "2": 1, + "3": 1, + "4": 1, + "5": 1, + } + + +def test_lineage_text_mode_shows_grade_distribution(built_dataset, capsys) -> None: + rc = main(["refine", "lineage", "--dataset", str(built_dataset)]) + assert rc == 0 + out = capsys.readouterr().out + assert "grade distribution:" in out + assert "train_sha256:" in out + assert "eval_sha256:" in out + + +# --- integrity: the lineage hashes match the emitted files ---------------- + + +def test_lineage_file_hashes_match_emitted_files(built_dataset) -> None: + lineage = dataset.read_lineage(built_dataset) + agg = lineage["aggregate"] + from data_refinery.store.envelope import content_hash + + train_text = (built_dataset / "train.jsonl").read_text() + eval_text = (built_dataset / "eval.jsonl").read_text() + assert agg["train_sha256"] == content_hash(train_text) + assert agg["eval_sha256"] == content_hash(eval_text) + + +# --- absent / corrupt lineage.json ----------------------------------------- + + +def test_lineage_missing_exits_1_with_hint(tmp_path, capsys) -> None: + rc = main(["refine", "lineage", "--dataset", str(tmp_path / "nope")]) + assert rc == 1 + err = capsys.readouterr().err + assert err.startswith("error:") + assert "hint:" in err + + +def test_lineage_missing_json_mode_exits_1(tmp_path, capsys) -> None: + rc = main(["refine", "lineage", "--dataset", str(tmp_path / "nope"), "--json"]) + assert rc == 1 + payload = json.loads(capsys.readouterr().err) + assert payload["code"] == 1 + assert payload["remediation"] + + +def test_lineage_corrupt_json_exits_2(tmp_path, capsys) -> None: + bad_dir = tmp_path / "bad" + bad_dir.mkdir() + (bad_dir / "lineage.json").write_text("{not valid json", encoding="utf-8") + rc = main(["refine", "lineage", "--dataset", str(bad_dir)]) + assert rc == 2 + assert "hint:" in capsys.readouterr().err + + +# --- main()-level wiring ---------------------------------------------------- + + +def test_lineage_verb_registered_end_to_end(built_dataset) -> None: + with pytest.raises(SystemExit) as exc: + main(["refine", "lineage", "--help"]) + assert exc.value.code == 0 diff --git a/uv.lock b/uv.lock index 12b9cb8..6b8119b 100644 --- a/uv.lock +++ b/uv.lock @@ -156,7 +156,7 @@ wheels = [ [[package]] name = "data-refinery-cli" -version = "0.9.0" +version = "0.10.0" source = { editable = "." } [package.optional-dependencies]