Repository navigation
Conversation
Closure plan §7 Phase 17 deliverable. Adds split-level parallelism
to the audit pipeline while pinning byte-identical output across
worker counts via the test suite.
Implementation:
- forgelm/data_audit/_orchestrator.py
- new module-level _process_split_for_pool wrapper so worker
pickling stays spawn-method safe (lambdas + functools.partial
over keyword-only args do not pickle cleanly across spawn)
- audit_dataset(workers: int = 1) parameter:
workers=1 keeps the pre-Phase-17 single-process path exactly,
workers>1 distributes per-split processing over min(workers,
len(splits)) Pool workers; merge step stays single-threaded so
the final report is byte-identical across worker counts.
- Pre-flight validation: workers must be positive int; rejects
bool / non-int / negative with a typed ValueError.
- forgelm/cli/_argparse_types.py
- new _positive_int type validator (rejects 0 / negatives at
argparse parse time, separate from _non_negative_int which
accepts 0).
- forgelm/cli/_parser.py
- --workers N flag on the audit subparser, default 1, type
_positive_int.
- forgelm/cli/subcommands/_audit.py
- workers param threaded through _run_data_audit + _run_audit_cmd
so the CLI flag reaches audit_dataset.
- forgelm/cli/__init__.py
- re-export _positive_int for tests / monkeypatch.
Determinism contract (the whole point of this phase):
data_audit_report.json is byte-identical between --workers 1 and
--workers 4. The only field that varies is generated_at (wall-clock
timestamp). CI gates that compare the audit-report hash across
runs continue to work after operators switch worker counts.
Tests (tests/test_data_audit_workers.py — 18 cases):
- byte-identical JSON for workers in {2, 4} vs sequential baseline
(parametrised, generated_at stripped before compare)
- per-split lang_sample byte-identical (the field most likely to
diverge under non-deterministic ordering)
- pii_summary / pii_severity identical across worker counts
- near_duplicate_summary identical (with injected duplicate rows)
- total_samples identical
- secrets_summary identical (with injected fake AKIA token)
- single-split corpus tolerates workers > 1 (no parallelism
available, must still produce a valid report)
- default-when-omitted equals workers=1 (backwards compatibility)
- workers in {0, -1, -10} raises typed ValueError
- workers="four" / workers=True raises typed ValueError
- CLI: --workers flag advertised in audit --help with default 1
- CLI: --workers 0 rejected at parse time with clear error
- CLI: --workers -2 rejected at parse time
- CLI: omitted --workers produces same on-disk JSON as --workers 1
(subprocess end-to-end smoke)
Documentation:
- docs/guides/data_audit.md
- Run-it section: --workers 4 example next to existing dedup-
method / quality-filter examples
- CLI reference: --workers N added to the flag list
- dedicated section explaining the determinism contract + how the
speed-up scales with number of splits, not row count
- CHANGELOG.md [Unreleased]: full Phase 17 entry
Test count: 1013 -> 1031 (+18). Coverage 67.94% (unchanged within
rounding; new code adds tests + reduces cli/_argparse_types coverage
gap). ruff format/check clean.
Closes: closure-plan-202604300906.md §7 Phase 17 (F-performance-104)
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Reviewer's GuideImplements split-level parallelism for the data audit pipeline via a new --workers CLI flag and workers parameter on audit_dataset, while preserving byte-identical output across worker counts and pinning this determinism with a dedicated test suite, plus associated CLI, docs, and changelog updates. Sequence diagram for forgelm audit --workers N parallel pipeline with deterministic mergesequenceDiagram
actor Operator
participant CLI as ForgelmCLI
participant Argparse as ArgparseParser
participant AuditCmd as AuditSubcommand_run_audit_cmd
participant Orchestrator as audit_dataset
participant Pool as MultiprocessingPool
participant Worker as _process_split_for_pool
Operator->>CLI: forgelm audit data/ --workers 4
CLI->>Argparse: parse args (includes workers)
Argparse-->>CLI: Namespace(workers=4,...)
CLI->>AuditCmd: _run_audit_cmd(args, output_format)
AuditCmd->>AuditCmd: extract workers = getattr(args, workers, 1)
AuditCmd->>Orchestrator: audit_dataset(source, ..., workers=4)
Orchestrator->>Orchestrator: validate workers is positive int
Orchestrator->>Orchestrator: splits_paths, resolution_notes = _resolve_input(source)
Orchestrator->>Orchestrator: compute effective_workers = min(workers, len(splits_paths))
alt effective_workers > 1
Orchestrator->>Pool: create Pool(effective_workers)
Orchestrator->>Pool: map(_process_split_for_pool, pool_args)
Pool->>Worker: _process_split_for_pool(args_tuple)
Worker->>Worker: unpack (split_name, path, kwargs)
Worker->>Orchestrator: _process_split(split_name, path, **kwargs)
Orchestrator-->>Worker: outcome
Worker-->>Pool: outcome
Pool-->>Orchestrator: outcomes_list (order preserved)
Orchestrator->>Orchestrator: outcomes_iter = zip(splits_paths.keys(), outcomes_list)
else sequential path
loop for split_name, path in splits_paths.items()
Orchestrator->>Orchestrator: outcome = _process_split(split_name, path, **kwargs)
Orchestrator-->>Orchestrator: yield split_name, outcome via _sequential_outcomes
end
Orchestrator->>Orchestrator: outcomes_iter = _sequential_outcomes()
end
loop deterministic merge over outcomes_iter
Orchestrator->>Orchestrator: update splits_info, signatures_by_split
Orchestrator->>Orchestrator: accumulate totals (samples, tokens, errors)
end
Orchestrator-->>AuditCmd: AuditReport
AuditCmd->>AuditCmd: _atomic_write_json(target_path, asdict(report))
AuditCmd-->>CLI: write deterministic data_audit_report.json
CLI-->>Operator: exit code 0, report hash stable across worker counts
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
Contributor
|
Important Review skippedAuto reviews are disabled on base/target branches other than the default branch. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Review Summary by QodoAdd split-level parallelism to audit pipeline with determinism contract
WalkthroughsDescription• Adds split-level parallelism to audit pipeline via --workers N flag • Guarantees byte-identical JSON output across worker counts via single-threaded merge • Implements 18 regression tests pinning determinism contract across worker configurations • Introduces _positive_int argparse validator rejecting zero and negative values Diagramflowchart LR
CLI["CLI: --workers N flag"]
Parser["argparse: _positive_int validator"]
Audit["audit_dataset: workers param"]
Validate["Validation: workers >= 1"]
Sequential["workers=1: Sequential path"]
Parallel["workers>1: Pool.map per-split"]
Merge["Single-threaded merge"]
JSON["Byte-identical JSON output"]
CLI --> Parser
Parser --> Audit
Audit --> Validate
Validate --> Sequential
Validate --> Parallel
Sequential --> Merge
Parallel --> Merge
Merge --> JSON
File Changes1. forgelm/cli/__init__.py
|
Code Review by Qodo
1.
|
Up to standards ✅🟢 Issues
|
| Metric | Results |
|---|---|
| Complexity | 48 |
| Duplication | 0 |
NEW Get contextual insights on your PRs based on Codacy's metrics, along with PR and Jira context, without leaving GitHub. Enable AI reviewer
TIP This summary will be updated as you push new changes.
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've found 1 issue, and left some high level feedback:
- The
workersvalidation logic inaudit_datasetduplicates and slightly diverges from the_positive_intargparse type; consider extracting a shared internal helper (that also handles theboolcase explicitly) so CLI and library callers always use exactly the same validation contract. - In
audit_dataset, the sequential branch builds_sequential_outcomes()only to immediately iterate it; simplifying this back to a directforloop (like the pre-Phase-17 code) would reduce indirection and make the control flow easier to follow.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The `workers` validation logic in `audit_dataset` duplicates and slightly diverges from the `_positive_int` argparse type; consider extracting a shared internal helper (that also handles the `bool` case explicitly) so CLI and library callers always use exactly the same validation contract.
- In `audit_dataset`, the sequential branch builds `_sequential_outcomes()` only to immediately iterate it; simplifying this back to a direct `for` loop (like the pre-Phase-17 code) would reduce indirection and make the control flow easier to follow.
## Individual Comments
### Comment 1
<location path="tests/test_data_audit_workers.py" line_range="284" />
<code_context>
+ assert "--workers" in result.stdout
+ assert "default: 1" in result.stdout.lower() or "(default: 1" in result.stdout
+
+ def test_cli_workers_zero_rejected_at_parse_time(self) -> None:
+ """``--workers 0`` exits with argparse usage-error rather than
+ propagating into the audit pipeline."""
</code_context>
<issue_to_address>
**suggestion (testing):** Add a CLI test for non-integer `--workers` to exercise `_positive_int`'s type error path
Current tests cover `--workers 0` and negative values, but `_positive_int` also has a path for non-integer input (`argparse.ArgumentTypeError("invalid integer: ...")`) that isn’t exercised. Please add a CLI test for a non-integer value to cover that branch, for example by running the module with `--workers notanint` and asserting a non-zero exit code and that `"invalid integer"` appears in `stderr`. This will lock down `_positive_int`’s behavior and error messaging.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
| assert "--workers" in result.stdout | ||
| assert "default: 1" in result.stdout.lower() or "(default: 1" in result.stdout | ||
|
|
||
| def test_cli_workers_zero_rejected_at_parse_time(self) -> None: |
There was a problem hiding this comment.
suggestion (testing): Add a CLI test for non-integer --workers to exercise _positive_int's type error path
Current tests cover --workers 0 and negative values, but _positive_int also has a path for non-integer input (argparse.ArgumentTypeError("invalid integer: ...")) that isn’t exercised. Please add a CLI test for a non-integer value to cover that branch, for example by running the module with --workers notanint and asserting a non-zero exit code and that "invalid integer" appears in stderr. This will lock down _positive_int’s behavior and error messaging.
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces split-level parallelism to the forgelm audit pipeline via a new --workers N flag. The implementation utilizes multiprocessing.Pool to process dataset splits in parallel while maintaining a determinism contract that ensures byte-identical JSON reports regardless of the worker count. Key changes include the addition of a _positive_int argparse validator, orchestrator updates to handle process pooling, and a comprehensive new test suite in tests/test_data_audit_workers.py. Feedback was provided to simplify the sequential execution path in the orchestrator by replacing a nested function with a more concise generator expression.
Comment on lines
+334
to
+338
| def _sequential_outcomes() -> Any: | ||
| for split_name, path in splits_paths.items(): | ||
| yield split_name, _process_split(split_name, path, **split_kwargs) | ||
|
|
||
| outcomes_iter = _sequential_outcomes() |
There was a problem hiding this comment.
The nested function _sequential_outcomes is somewhat verbose for a simple sequential iteration. You can achieve the same result more concisely using a generator expression, which improves readability and avoids the overhead of defining a nested function.
Suggested change
| def _sequential_outcomes() -> Any: | |
| for split_name, path in splits_paths.items(): | |
| yield split_name, _process_split(split_name, path, **split_kwargs) | |
| outcomes_iter = _sequential_outcomes() | |
| outcomes_iter = ( | |
| (split_name, _process_split(split_name, path, **split_kwargs)) | |
| for split_name, path in splits_paths.items() | |
| ) |
![qodo-code-review[bot]](https://avatars.githubusercontent.com/in/484649?s=60&v=4){kind=small}
Comment on lines
+323
to
+331
| 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. | ||
| pool_args = [(name, path, split_kwargs) for name, path in splits_paths.items()] | ||
| with multiprocessing.Pool(effective_workers) as pool: | ||
| outcomes_list = pool.map(_process_split_for_pool, pool_args) | ||
| outcomes_iter = zip(splits_paths.keys(), outcomes_list) |
There was a problem hiding this comment.
2. Schema order nondeterminism 🐞 Bug ≡ Correctness
When audit_dataset(workers>1) runs splits in separate processes, schema column ordering can differ from workers=1 because _record_schema_for_dict iterates a frozenset(row.keys()). This can change splits[*].columns / schema_drift_columns ordering and break the PR’s “byte-identical across worker counts” contract for multi-column rows.
Agent Prompt
### Issue description
`audit_dataset(workers>1)` computes per-split `info` in child processes. `_record_schema_for_dict` iterates `frozenset(row.keys())`, making `columns` ordering dependent on per-process hash seed; this can cause `workers=1` vs `workers>1` to produce different JSON.
### Issue Context
The final report uses `outcome.info` verbatim per split, so any ordering differences inside `_audit_split` propagate to the on-disk report.
### Fix Focus Areas
- forgelm/data_audit/_aggregator.py[88-93]
- forgelm/data_audit/_aggregator.py[189-199]
- tests/test_data_audit_workers.py[44-57]
### Suggested change
Keep `frozenset(row.keys())` for counting keysets, but populate `seen_columns` with a deterministic iteration (e.g., `for col in sorted(row.keys()): ...` or `for col in row.keys(): ...`). Add/extend a workers determinism fixture to include multi-column rows and assert `columns` ordering is identical across worker counts.
ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools
Closes the two CRITICAL findings, three HIGH findings, and one
HIGH bot comment.
CRITICAL fixes:
- F-26-01 (CRITICAL): test_lang_sample_byte_identical was vacuous
— both sides returned None because lang_sample is the in-memory
_StreamingAggregator field, never serialised. Renamed to
test_languages_top3_byte_identical and now compares the actual
on-disk languages_top3 list (the persisted derivative operators
and CI gates can see).
- F-26-02 (CRITICAL): test_audit_json_byte_identical_to_sequential
compared parsed dicts (tolerates key ordering / whitespace /
Unicode normalisation / float repr drift), not bytes. Now adds
_file_sha256() helper that strips generated_at textually and
hashes the rest — exactly what an Article 10 governance-bundle CI
gate compares. Dict equality kept as a secondary looser check so
both layers regress-trip simultaneously.
HIGH fixes:
- F-26-03 (HIGH): _orchestrator now pins multiprocessing context to
"spawn" (was: default Pool() = fork on Linux + spawn on macOS /
Windows; Linux CI never exercised the spawn path). Spawn gives
every worker a fresh interpreter — no shared
langdetect.DetectorFactory state, no copied open file handles.
- F-26-04 (HIGH): new TestWorkersWithMinHash exercises the
byte-identical contract on dedup_method="minhash" + workers > 1
(the path operators explicitly recommend for >50K-row corpora).
Skipped when datasketch optional extra is not installed.
- F-26-05 (HIGH): new TestWorkersErrorPropagation —
test_sequential_split_failure_propagates (per-split failure
bubbles up unchanged) + test_parallel_path_does_not_silently_complete
(a missing-file split surfaces inline as read_failed without
hanging or killing the whole report).
- F-26-06 (HIGH): new test_split_iteration_order_pinned asserts
list(report.splits.keys()) == ["train", "validation", "test"] so a
future refactor that swaps the dict for a set or alphabetises
silently breaks the byte-identical contract.
MEDIUM / NIT:
- gemini bot comment: removed nested _sequential_outcomes function
in favour of a single generator expression in the else branch
(cleaner + same semantics).
- Added INFO log on entering parallel path so operators can verify
--workers 4 actually engaged (F-26-08).
- sourcery bot: new test_cli_workers_non_integer_rejected_at_parse_time
exercises the _positive_int "invalid integer" error path.
- qodo bot: docs/guides/data_audit-tr.md updated with --workers
example + dedicated section (was: only EN side updated in original
Phase 17 PR).
DEFERRED (out of this PR's scope):
- F-26-07 (MEDIUM): effective_workers upper-bound — the
min(workers, len(splits_paths)) clamp already caps at split count
(typically 3); 1M-process loophole only triggers if a future
refactor adds per-row parallelism. Logged for that follow-up.
- F-26-09 (MEDIUM): _process_split_for_pool relocation — keeps
spawn-method picklability concerns local to the orchestrator;
splitting would require careful re-export.
- F-26-10 (MEDIUM): fake AKIA test fixture — gitleaks did not flag
in pre-commit; revisited if it ever does.
- F-26-11 (MEDIUM): component-invariant tests calling output_dir —
TestWorkersDeterminism tests already exercise the on-disk path
via _audit_to_canonical_json; component tests are deliberately
in-memory for speed.
- F-26-12..16 LOW/NIT: backlog.
Tests: 1031 → 1035 (+4 net: +6 new tests, -2 vacuous removed).
Skipped: 13 → 14 (datasketch-gated minhash×workers test).
ruff format/check clean. Coverage 67.94% → 67.91% (within rounding).
Closes (review IDs): F-26-01..06 + gemini-sequential + sourcery-non-int
+ qodo-tr-guide
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
|
This was referenced May 2, 2026
Collaborator
Author
Wave 2a Round-1 multi-agent review — fixes pushedAll CRITICAL + HIGH findings from the multi-agent review (and the relevant inline bot comments) have been addressed in the latest commit on this branch. See the commit message for the per-finding mapping (review IDs F-26-*, gemini bot IDs, qodo bot IDs, sourcery bot IDs). Per-PR test counts:
ruff format/check clean across all five PRs. Cross-PR XPR-01 concern (audit-log parser proliferation) closed in PR #25 via the new Items deferred to follow-up PRs (with rationale) are listed under DEFERRED in each commit message. @coderabbitai review |
This was referenced May 2, 2026
Contributor
|
Tip For best results, initiate chat on the files or code changes.
[review] |
8 tasks
Collaborator
Author
|
Superseded by #28 (consolidated Wave 2a integration). All commits from this branch are merged into |
cemililik
added a commit
that referenced
this pull request
Jun 15, 2026
_read_jsonl_split decoded each line twice (replace-decode for text, then strict-decode to infer decode_error). Switch to strict-first with an errors='replace' fallback only on UnicodeDecodeError — behaviour-identical, one decode on the common path. Existing TestJsonlDecodeErrorDetection guards both branches. CodeRabbit #26. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Adds split-level parallelism to the audit pipeline (`--workers N` flag) while pinning byte-identical output across worker counts via the test suite. CI gates that compare the audit-report hash continue to work after operators switch from `--workers 1` to `--workers 4`.
```shell
Sequential (default; pre-Phase-17 behaviour, byte-identical)
forgelm audit data/
Phase 17: split-level parallelism on multi-split corpora
forgelm audit data/ --workers 4
```
Speed-up scales with the number of splits — `--workers 3` on a `train` / `validation` / `test` corpus typically yields a near-linear speed-up.
Determinism contract
The merge step that builds the final report stays single-threaded so `data_audit_report.json` is byte-identical between worker counts. The only field that varies is `generated_at` (wall-clock timestamp). Pinned by 18 tests in `tests/test_data_audit_workers.py`.
Files
Test plan
Risk
Low. Strictly additive: `workers=1` is the default and reproduces the pre-Phase-17 single-process path exactly. The merge step is single-threaded so the on-disk artefact is invariant — operators who pin the report hash in CI continue to pass without changing their gate.
Closes
🤖 Generated with Claude Code
Summary by Sourcery
Introduce configurable split-level parallelism to the data audit pipeline while preserving deterministic, byte-identical reports across worker counts.
New Features:
Enhancements:
Documentation:
Tests: