Bound DataFlow dataset setup and update the workflow to the current DataFlow API - #49969
Conversation
Co-authored-by: pelikhan <4175913+pelikhan@users.noreply.github.com>
Co-authored-by: pelikhan <4175913+pelikhan@users.noreply.github.com>
There was a problem hiding this comment.
Pull request overview
Bounds DataFlow setup, updates the dataset pipeline to DataFlow 1.0.10, and improves fallback reporting.
Changes:
- Pins and time-bounds dependency installation.
- Updates storage, filtering, deduplication, and execution-mode handling.
- Removes repo-memory finalization and regenerates the workflow lock file.
Show a summary per file
| File | Description |
|---|---|
.github/workflows/dataflow-pr-discussion-dataset.md |
Updates installation, pipeline, fallback, and reporting logic. |
.github/workflows/dataflow-pr-discussion-dataset.lock.yml |
Regenerates the compiled workflow. |
.github/skills/agentic-workflows/SKILL.md |
Adds the observability optimization reference. |
Review details
Tip
Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
- Files reviewed: 3/3 changed files
- Comments generated: 4
- Review effort level: Balanced
| if dataflow_ops_used and python_ops_used: | ||
| stats["execution_mode"] = "mixed" | ||
| elif dataflow_ops_used: | ||
| stats["execution_mode"] = "dataflow" |
There was a problem hiding this comment.
Fixed in 8a6f6ee. python_alpha_ratio_filter is now added directly to operators_used but is not tracked in python_ops_used. Mode is computed solely from dataflow_ops_used vs python_ops_used (the DataFlow-eligible fallback stages), so a run where both DataFlow stages succeed reports dataflow correctly.
| with open(OUTPUT, "w") as fh: | ||
| for record in records_after_dedup: | ||
| fh.write(json.dumps(record, ensure_ascii=False) + "\n") |
There was a problem hiding this comment.
Fixed in 8a6f6ee. Added _DATAFLOW_INTERNAL_KEYS = frozenset({"char_number_filter_label", "minhash_deduplicated_label", "hash_deduplicated_label"}) and strip those keys before writing each record, so the output schema is uniform regardless of execution path.
| CharNumberFilter(threshold=50).run(storage=storage, input_key="text") | ||
| storage.step() # step 1 = length-filter output | ||
| records_after_length = storage.read("dict") | ||
| dataflow_ops_used.append("CharNumberFilter") |
There was a problem hiding this comment.
Fixed in 8a6f6ee. After reading storage.read("dict") from the CharNumberFilter step, an explicit [r for r in records_after_length if len(r.get('text', '')) <= 100_000] cap is applied before proceeding, enforcing the full 50–100,000 character bound on the DataFlow path.
| MinHashDeduplicateFilter(threshold=0.85).run(storage=storage, input_key="text") | ||
| dataflow_ops_used.append("MinHashDeduplicateFilter") |
There was a problem hiding this comment.
Fixed in 8a6f6ee. Removed MinHashDeduplicateFilter entirely from the pipeline, smoke test, and selected_operators. The pipeline now uses only HashDeduplicateFilter, which is deterministic and fully exercised end-to-end by the smoke test. The open-insertion-session inconsistency no longer applies.
|
|
|
🧠 Matt Pocock Skills Reviewer has completed the skills-based review. ✅ |
|
✅ Test Quality Sentinel completed test quality analysis. No test files were added or modified in this PR. Test Quality Sentinel skipped. PR only contains workflow changes (.github/workflows/dataflow-pr-discussion-dataset.md and .lock.yml). |
|
✅ Design Decision Gate 🏗️ completed the design decision gate check. No ADR enforcement needed: PR does not have the implementation label and has 0 new lines of code in business logic directories. |
|
@copilot sous-chef triage: Please refresh the branch if needed and then run the
|
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
There was a problem hiding this comment.
Skills-Based Review 🧠
Applied /diagnosing-bugs — requesting changes on three correctness gaps in the new setup and pipeline.
📋 Key Themes & Highlights
Key Themes
- Smoke test doesn't exercise MinHash:
MinHashDeduplicateFilteris instantiated but.run()is never called, so the smoke gate doesn't validate the operator that is most likely to regress. selected_operatorsis static: The runtime JSON hardcodes all three operators even when some weren't validated, giving the downstream agent a false confidence signal.- Missing upper-bound in DataFlow path:
CharNumberFilter(threshold=50)only enforces the lower bound; the 100,000-character cap is absent from the DataFlow and mixed execution paths.
These three issues (plus the four already flagged by prior review: mode misclassification, label field leakage, missing upper bound in smoke test, and MinHash LSH session ordering) collectively mean the dataflow and mixed execution paths can produce different output from fallback for the same input.
Positive Highlights
- ✅ Explicit timeout bounds on each install step — a clear improvement over the unbounded previous approach.
- ✅ Graceful degradation with
::warning::annotations keeps the workflow useful even when DataFlow is unavailable. - ✅ Three-way execution mode reporting (
dataflow/mixed/fallback) is a good observability addition. - ✅ Avoiding
AlphaWordsFilterto prevent implicit NLTK downloads is a sound defensive choice.
🧠 Reviewed using Matt Pocock's skills by Matt Pocock Skills Reviewer · sonnet46 · 37.5 AIC · ⌖ 12.9 AIC · ⊞ 7.1K
Comment /matt to run again
Comments that could not be inline-anchored
.github/workflows/dataflow-pr-discussion-dataset.md:478
[/diagnosing-bugs] The smoke test instantiates MinHashDeduplicateFilter but never calls .run() — so MinHash is never actually exercised. If MinHash's API is broken, the smoke test still passes and dataflow_ready=true, deferring the failure to the production pipeline.
<details>
<summary>💡 Suggested fix</summary>
Replace the dead assignment with a real round-trip:
storage.write(after_length)
storage.step()
MinHashDeduplicateFilter(threshold=0.85).run(storage=storage, input…
</details>
<details><summary>.github/workflows/dataflow-pr-discussion-dataset.md:499</summary>
**[/diagnosing-bugs]** `selected_operators` in the runtime status file is hardcoded to all three operators regardless of which ones the smoke test actually exercises. This contradicts the PR goal of "record only operators that actually ran" and misleads the production pipeline about what was validated.
<details>
<summary>💡 Suggested fix</summary>
Build `selected_operators` dynamically based on what the smoke test confirmed:
```python
validated = []
if ...: # CharNumberFilter passed
val…
</details>
<details><summary>.github/workflows/dataflow-pr-discussion-dataset.md:750</summary>
**[/diagnosing-bugs]** `CharNumberFilter(threshold=50)` only applies a lower bound. The PR description requires a 100,000-character upper bound, but there is no `max_threshold` or equivalent argument here. Records over 100 K characters pass the DataFlow path but would be caught by the Python fallback — creating an inconsistency between execution modes.
<details>
<summary>💡 Suggested fix</summary>
Check whether `CharNumberFilter` accepts a `max_threshold` keyword and pass it:
```python
CharN…
</details>
There was a problem hiding this comment.
Review: Bound DataFlow dataset setup
The changes are well-structured overall. The timeout-bounded install, pinned open-dataflow==1.0.10, and graceful fallback path are solid improvements.
Existing inline comments already cover the main correctness concerns. One structural note: in the smoke test, MinHashDeduplicateFilter is instantiated but .run() is never called, so a broken .run() API would not be caught at setup time. The graceful try/except in the main pipeline mitigates this.
🧵 Reviewed using Impeccable skills by Impeccable Skills Reviewer · sonnet46 · 45.4 AIC · ⌖ 7.53 AIC · ⊞ 5.4K
…ema, upper-bound filter, MinHash - Fix mode reporting: exclude intentional Python alpha-ratio stage from python_ops_used; mode is now computed from DataFlow-eligible stages only (CharNumberFilter + HashDeduplicateFilter). Pure DataFlow success reports as `dataflow`, partial fallback reports as `mixed`. - Strip DataFlow internal label fields (char_number_filter_label, minhash_deduplicated_label, hash_deduplicated_label) before writing output so the dataset schema is uniform regardless of execution path. - Add explicit upper-bound cap (100,000 chars) after DataFlow CharNumberFilter output; CharNumberFilter(threshold=50) only enforces the lower bound. - Replace MinHashDeduplicateFilter with HashDeduplicateFilter in the pipeline. The 1.0.10 MinHash implementation queries its LSH index while an insertion_session is still open (datasketch inconsistency); HashDeduplicateFilter is deterministic and already validated by the smoke test. Remove MinHash import, instantiation, and reference from the smoke test and selected_operators list. Co-authored-by: gh-aw-bot <259018956+gh-aw-bot@users.noreply.github.com>
🔍 PR TriageCategory: bug · Risk: medium · Total score: 60/100
Bounds/pins DataFlow dataset setup to avoid budget exhaustion and updates pipeline to current DataFlow API with honest execution-mode reporting. CI pending. Recommended action:
|
Done. Addressed all four review issues in commit
Local validation: |
|
@copilot sous-chef triage: Please refresh the branch if needed and then run the skill.
|
All four review threads have been addressed in commit |
|
@copilot sous-chef triage: All review threads appear addressed in the latest branch state; please run the
|
PR finisher summary:
Still needed (human actions):
|
The DataFlow dataset workflow could spend the full Actions job budget in dependency installation, and even successful installs still fell back because the pipeline targeted obsolete DataFlow modules and
FileStoragebehavior. This change bounds setup, validates the exact runtime API up front, and makes the workflow report whether it ran indataflow,mixed, orfallbackmode.Bound and pin the pre-agent DataFlow setup
pip install open-dataflowpath with pinneduvbootstrap + pinnedopen-dataflowinstall under explicit timeouts.Validate the current DataFlow API instead of import-only success
dataflow.utils.storage.FileStoragedataflow.operators.general_text.CharNumberFilterMinHashDeduplicateFilterHashDeduplicateFilterUpdate the pipeline to the current storage/operator model
len(storage), iteratingFileStorage, andstorage.save().FileStorage.step()transitions and materialize output records directly todataset_clean.jsonl.Avoid hidden NLTK downloads
AlphaWordsFilter, which can trigger implicit NLTK fetches.Make execution mode and reporting honest
execution_modereporting:dataflow,mixed, orfallback.run: https://github.com/github/gh-aw/actions/runs/30815680828
Run: https://github.com/github/gh-aw/actions/runs/30820387900
branch-refresh requested by PR Sous Chef run: https://github.com/github/gh-aw/actions/runs/30830929419