Phase 17: forgelm audit --workers N + determinism contract

CHANGELOG.md

@@ -13,6 +13,34 @@ All notable changes to ForgeLM are documented here.
 > Per-PR CHANGELOG entries below collapse into the v0.5.5 release
 > notes at tag time.
 
+### Added — Wave 2a / Phase 17 — `forgelm audit --workers N` determinism
+
+- **Split-level parallelism for the audit pipeline.**  `--workers N`
+  (default 1) runs each split in its own `multiprocessing.Pool` worker.
+  Speed-up scales with the number of splits — `--workers 3` on a
+  `train` / `validation` / `test` corpus typically yields a near-linear
+  speed-up.  Single-split corpora ignore values >1.
+- **Determinism contract pinned by tests.**  The merge step that
+  builds the final report stays single-threaded so
+  `data_audit_report.json` is byte-identical across worker counts (only
+  `generated_at` differs as expected).  18 tests in
+  `tests/test_data_audit_workers.py` cover: byte-identical JSON for
+  workers={1,2,4}; per-split `lang_sample` equality; identical
+  `pii_summary` / `secrets_summary` / `near_duplicate_summary` /
+  `total_samples` across worker counts; CLI `--workers 0` rejected at
+  parse time; library `audit_dataset(workers=0/-1/bool/str)` raises
+  typed `ValueError`; default-when-omitted equals `--workers 1`;
+  single-split corpus tolerates `workers > 1`.
+- **CLI exposure.**  `forgelm audit --workers N` registered on the
+  audit subparser with a new `_positive_int` argparse type validator
+  (rejects 0 / negatives at parse time).
+- New module-level helper `_process_split_for_pool` in
+  `forgelm/data_audit/_orchestrator.py` so worker pickling stays
+  spawn-method safe.
+- Operator docs updated: `docs/guides/data_audit.md` Run-it section
+  shows the `--workers 4` example, CLI reference includes the flag,
+  and a dedicated explanation of the determinism contract was added.
+
 ### Added — Wave 1 closure (Faz 9, 11, 12, 13, 25, 31, 32 — see PR description)
 
 - **Article 14 staging directory + `forgelm approve` / `forgelm reject` (Faz 9)** —

docs/guides/data_audit-tr.md

@@ -37,6 +37,9 @@ forgelm audit data/large_corpus.jsonl --dedup-method minhash --jaccard-threshold
 
 # Faz 12: opt-in heuristic kalite filtresi (Gopher/C4 stili)
 forgelm audit data/ --quality-filter
+
+# Faz 17: çoklu-split corpora için split-seviyesi paralelizm
+forgelm audit data/ --workers 4
 ```
 
 > **Eski alias:** `forgelm --data-audit PATH` deprecation alias'ı
@@ -485,10 +488,25 @@ forgelm audit PATH \
   [--pii-ml] \
   [--pii-ml-language LANG] \
   [--croissant] \
+  [--workers N] \
   [--output-format {text,json}] \
   [--quiet | --log-level {DEBUG,INFO,WARNING,ERROR}]
 ```
 
+`--workers N` (Faz 17) split-seviyesi paralelizmi açar: her split kendi
+worker process'inde, bir `multiprocessing.Pool` içinde koşar. Varsayılan
+`1` (sıralı, Faz-17-öncesi yol ile byte-identical). Speed-up *split sayısı*
+ile ölçeklenir, satır sayısıyla değil — tek-split corpus'lar 1'den büyük
+değerleri görmezden gelir. Tipik çoklu-split koşular (`train` /
+`validation` / `test`) `--workers 3` ile neredeyse-doğrusal hızlanma görür.
+
+Raporu üreten merge adımı tek-thread'li kalır, dolayısıyla diskteki JSON
+worker sayısından bağımsız **byte-identical**'dir (yalnızca
+`generated_at` wall-clock zaman damgası değişir). Bu invariant
+`tests/test_data_audit_workers.py` tarafından pinlenir; CI gate
+audit-report hash'ini karşılaştırırken `--workers 1`'den `--workers 4`'e
+geçiş yapan bir operatör için çalışmaya devam eder.
+
 `PATH` bir `.jsonl` dosyası veya bir dizin olabilir. `--output`
 varsayılan olarak `./audit/`'tir. `--verbose`, insan-okunabilir özette
 bulgu olmayan split'leri de gösterir (varsayılan, çoklu-split

docs/guides/data_audit.md

@@ -38,6 +38,9 @@ forgelm audit data/large_corpus.jsonl --dedup-method minhash --jaccard-threshold
 
 # Phase 12: opt-in heuristic quality filter (Gopher/C4 style)
 forgelm audit data/ --quality-filter
+
+# Phase 17: split-level parallelism for multi-split corpora
+forgelm audit data/ --workers 4
 ```
 
 > **Legacy alias:** `forgelm --data-audit PATH` keeps working unchanged
@@ -479,6 +482,7 @@ forgelm audit PATH \
   [--pii-ml] \
   [--pii-ml-language LANG] \
   [--croissant] \
+  [--workers N] \
   [--output-format {text,json}] \
   [--quiet | --log-level {DEBUG,INFO,WARNING,ERROR}]
 ```
@@ -503,6 +507,20 @@ is no flag to disable it.
 > `os.replace` — Phase 11.5 hardening so a crashed audit can never leave
 > a half-written report on disk.
 
+`--workers N` (Phase 17) opts into split-level parallelism: each split
+runs in its own worker process inside a `multiprocessing.Pool`.  Default
+is `1` (sequential, byte-identical to the pre-Phase-17 path).  Speed-up
+scales with the number of *splits*, not with row count — a single-split
+corpus ignores values >1.  Typical multi-split runs (`train` /
+`validation` / `test`) see a near-linear speed-up at `--workers 3`.
+
+The merge step that builds the final report stays single-threaded so
+the on-disk JSON is **byte-identical** across worker counts (the only
+field that varies is `generated_at`, the wall-clock timestamp).  This
+invariant is pinned by `tests/test_data_audit_workers.py` so a CI gate
+that compares the audit-report hash across runs continues to work
+when an operator switches from `--workers 1` to `--workers 4`.
+
 The legacy `forgelm --data-audit PATH` flag is preserved as a
 deprecation alias and logs a one-line notice. Behaviour is identical;
 new scripts should use the subcommand.

forgelm/cli/__init__.py

@@ -25,6 +25,7 @@
     _add_common_subparser_flags,  # noqa: F401 — re-export for tests
     _non_negative_float,  # noqa: F401 — re-export for tests
     _non_negative_int,  # noqa: F401 — re-export for tests
+    _positive_int,  # noqa: F401 — re-export for tests
 )
 
 # Config loading + offline flag application (training mode).

forgelm/cli/_argparse_types.py

@@ -21,6 +21,23 @@ def _non_negative_int(value: str) -> int:
     return ivalue
 
 
+def _positive_int(value: str) -> int:
+    """argparse type for flags that must be >= 1 (e.g. ``--workers``).
+
+    Mirrors :func:`_non_negative_int` but rejects 0 so a typo
+    (``--workers 0``) produces an immediate, helpful CLI error instead
+    of getting validated downstream as ``ValueError`` deep inside an
+    audit call.
+    """
+    try:
+        ivalue = int(value)
+    except (TypeError, ValueError) as exc:
+        raise argparse.ArgumentTypeError(f"invalid integer: {value!r}") from exc
+    if ivalue < 1:
+        raise argparse.ArgumentTypeError(f"value must be >= 1, got {ivalue}")
+    return ivalue
+
+
 def _non_negative_float(value: str) -> float:
     """argparse type for ``--jaccard-threshold`` and similar floats.
 

forgelm/cli/_parser.py

@@ -11,7 +11,7 @@
 
 import argparse
 
-from ._argparse_types import _add_common_subparser_flags, _non_negative_float, _non_negative_int
+from ._argparse_types import _add_common_subparser_flags, _non_negative_float, _non_negative_int, _positive_int
 from ._logging import _get_version
 
 
@@ -409,6 +409,21 @@ def _add_audit_subcommand(subparsers) -> None:
             "corpus AND make sure the matching spaCy model is installed."
         ),
     )
+    p.add_argument(
+        "--workers",
+        type=_positive_int,
+        default=1,
+        metavar="N",
+        help=(
+            "Phase 17: number of worker processes for the split-level pipeline "
+            "(default: 1 — sequential, byte-identical to the pre-Phase-17 path). "
+            "Set to 2-4 on multi-split corpora (train / validation / test) for a "
+            "near-linear speed-up.  Speed-up scales with the number of splits, "
+            "not row count — single-split corpora ignore values > 1.  The merge "
+            "step is single-threaded so the audit JSON is byte-identical across "
+            "worker counts (determinism contract pinned by the test suite)."
+        ),
+    )
     _add_common_subparser_flags(p, include_output_format=True)
 
 

forgelm/cli/subcommands/_audit.py

@@ -29,6 +29,7 @@ def _run_data_audit(
     enable_pii_ml: bool = False,
     pii_ml_language: str = "en",
     emit_croissant: bool = False,
+    workers: int = 1,
 ) -> None:
     """Phase 11 / 11.5 / 12 dispatch: dataset quality + governance audit.
 
@@ -59,6 +60,7 @@ def _run_data_audit(
             enable_pii_ml=enable_pii_ml,
             pii_ml_language=pii_ml_language,
             emit_croissant=emit_croissant,
+            workers=workers,
         )
     except OSError as exc:
         # OSError covers FileNotFoundError / PermissionError / ENOSPC /
@@ -148,4 +150,5 @@ def _run_audit_cmd(args, output_format: str) -> None:
         enable_pii_ml=getattr(args, "pii_ml", False),
         pii_ml_language=getattr(args, "pii_ml_language", "en"),
         emit_croissant=getattr(args, "croissant", False),
+        workers=getattr(args, "workers", 1),
     )

forgelm/data_audit/_orchestrator.py

@@ -11,12 +11,13 @@
 
 import json
 import logging
+import multiprocessing
 import os
 import tempfile
 from dataclasses import asdict
 from datetime import datetime, timezone
 from pathlib import Path
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Tuple
 
 from ._croissant import _build_croissant_metadata
 
@@ -179,6 +180,20 @@ def _atomic_write_json(target: Path, payload: Dict[str, Any]) -> None:
                 pass
 
 
+def _process_split_for_pool(
+    args_tuple: Tuple[str, Path, Dict[str, Any]],
+):
+    """Module-level wrapper: dispatches to :func:`_process_split` from a tuple.
+
+    ``multiprocessing.Pool.map`` requires the worker callable + its arguments
+    to be picklable.  Lambdas and ``functools.partial`` over keyword-only
+    args don't pickle cleanly across spawn-method workers, so we accept a
+    single positional tuple and unpack the kwargs inside the worker.
+    """
+    split_name, path, kwargs = args_tuple
+    return _process_split(split_name, path, **kwargs)
+
+
 def audit_dataset(  # NOSONAR — cognitive complexity is inherent to the audit orchestration logic; extraction would fragment cohesive pipeline steps
     source: str,
     *,
@@ -191,6 +206,7 @@ def audit_dataset(  # NOSONAR — cognitive complexity is inherent to the audit
     enable_pii_ml: bool = False,
     pii_ml_language: str = "en",
     emit_croissant: bool = False,
+    workers: int = 1,
 ) -> AuditReport:
     """Run the audit pipeline over a JSONL file or split-keyed directory.
 
@@ -233,6 +249,15 @@ def audit_dataset(  # NOSONAR — cognitive complexity is inherent to the audit
         enable_quality_filter: Phase 12 opt-in flag — when ``True``, run
             the heuristic quality checks (Gopher / C4 / RefinedWeb-style)
             and surface findings under ``quality_summary``.
+        workers: Phase 17 — number of worker processes used for the
+            split-level pipeline.  ``1`` (default) keeps the pre-Phase-17
+            single-process behaviour exactly.  ``> 1`` distributes the
+            per-split processing over ``min(workers, len(splits))``
+            ``multiprocessing.Pool`` workers; the merge step that builds
+            the final report stays single-threaded so the determinism
+            contract (byte-identical JSON across worker counts) is
+            preserved.  Speed-up scales with the number of *splits*, not
+            row count — single-split corpora ignore values >1.
 
     Returns:
         :class:`AuditReport`. JSON-serialize via ``asdict(report)``.
@@ -260,6 +285,9 @@ def audit_dataset(  # NOSONAR — cognitive complexity is inherent to the audit
         # for per-row resilience on pathological strings).
         _require_presidio(language=pii_ml_language)
 
+    if not isinstance(workers, int) or isinstance(workers, bool) or workers < 1:
+        raise ValueError(f"workers must be a positive integer; got {workers!r}.")
+
     splits_paths, resolution_notes = _resolve_input(source)
 
     splits_info: Dict[str, Dict[str, Any]] = {}
@@ -276,18 +304,61 @@ def audit_dataset(  # NOSONAR — cognitive complexity is inherent to the audit
     parse_errors_total = 0
     decode_errors_total = 0
 
-    for split_name, path in splits_paths.items():
-        outcome = _process_split(
-            split_name,
-            path,
-            near_dup_threshold=near_dup_threshold,
-            dedup_method=dedup_method,
-            minhash_jaccard=minhash_jaccard,
-            minhash_num_perm=minhash_num_perm,
-            enable_quality_filter=enable_quality_filter,
-            enable_pii_ml=enable_pii_ml,
-            pii_ml_language=pii_ml_language,
+    # Determinism contract: regardless of ``workers``, every aggregate is
+    # built up by walking the splits in the exact order ``splits_paths``
+    # yields them.  Multiprocessing only changes how the per-split outcomes
+    # are *computed*; the merge step below is single-threaded so byte-
+    # equivalent output is guaranteed across worker counts.  See the
+    # ``test_data_audit_workers.py`` regression suite for the test that
+    # pins this invariant.
+    split_kwargs: Dict[str, Any] = {
+        "near_dup_threshold": near_dup_threshold,
+        "dedup_method": dedup_method,
+        "minhash_jaccard": minhash_jaccard,
+        "minhash_num_perm": minhash_num_perm,
+        "enable_quality_filter": enable_quality_filter,
+        "enable_pii_ml": enable_pii_ml,
+        "pii_ml_language": pii_ml_language,
+    }
+    effective_workers = min(workers, len(splits_paths)) if splits_paths else 1
+    if effective_workers > 1:
+        # Pool.map preserves input order so ``outcomes`` lines up with
+        # ``splits_paths.items()`` 1:1.  We bound the pool to
+        # ``len(splits_paths)`` — extra workers would idle since the unit
+        # of work is one split.
+        #
+        # Wave 2a Round-1 review (F-26-03): pin the start method to
+        # ``"spawn"`` so the determinism contract is platform-independent.
+        # ``Pool()`` defaults to fork on Linux + spawn on macOS / Windows;
+        # Linux CI never exercises the spawn path otherwise.  Spawn also
+        # gives every worker a fresh interpreter (no shared
+        # langdetect.DetectorFactory state, no copied open file handles)
+        # which is exactly what the byte-identical contract requires.
+        logger.info(
+            "audit: spawning %d worker process(es) over %d split(s)",
+            effective_workers,
+            len(splits_paths),
         )
+        ctx = multiprocessing.get_context("spawn")
+        pool_args = [(name, path, split_kwargs) for name, path in splits_paths.items()]
+        with ctx.Pool(effective_workers) as pool:
+            try:
+                outcomes_list = pool.map(_process_split_for_pool, pool_args)
+            except Exception:  # noqa: BLE001 — best-effort: re-raise unchanged so the operator sees the worker traceback through Pool.map's wrapping; we log at ERROR first so the contextual "which split failed" hint lands even when the traceback is opaque.
+                logger.error(
+                    "audit: parallel worker raised inside the pool; the original "
+                    "traceback is wrapped — re-run with --workers 1 to surface it "
+                    "with full context."
+                )
+                raise
+        outcomes_iter = zip(splits_paths.keys(), outcomes_list)
+    else:
+        logger.info("audit: sequential mode (workers=%d, splits=%d)", workers, len(splits_paths))
+        outcomes_iter = (
+            (split_name, _process_split(split_name, path, **split_kwargs)) for split_name, path in splits_paths.items()
+        )
+
+    for split_name, outcome in outcomes_iter:
         splits_info[split_name] = outcome.info
         signatures_by_split[split_name] = outcome.signatures
         total_samples += outcome.row_count