feat: refine — graded work items to training data with provenance (#14)

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

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)

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",

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",
 ]
 
 

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 <graded.jsonl> --out <dir> — graded colleague work "
+                "items -> sloth-ready chat-schema train/eval JSONL + lineage.json",
+                "refine lineage --dataset <dir> — 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)

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 <path>` (required), `--schema chat` (default + only schema
+today), `--min-rating N` (default `4`), `--split A/B` (default `90/10`),
+`--seed N` (default `0`), `--out <dir>` (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": <request>},
+  {"role": "assistant", "content": <summary>}]}` — 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 `<dataset>/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,
 }