Make the DataFlow dataset workflow bounded and compatible with the current API

[haolpku]

Implementation status — 2026-08-03

Implemented and locally validated in 78615c3 on dataflow/fix-dataflow-install-timeout. Review: fork draft PR #1; upstream comparison. GitHub rejected creation of the upstream PR for this account, so the implementation and validation evidence are recorded in this issue comment.

---

Summary

The DataFlow PR & Discussion Dataset Builder workflow needs two related fixes:

1. its pre-agent DataFlow installation is unbounded and can consume GitHub Actions' six-hour job limit; and
2. after a successful installation, the current pipeline does not use DataFlow because its imports and storage calls target an obsolete API.

The workflow can currently produce a dataset only through its pure-Python fallback. A successful package installation alone is therefore not sufficient.

Workflow source: https://github.com/github/gh-aw/blob/main/.github/workflows/dataflow-pr-discussion-dataset.md
Latest affected scheduled run: https://github.com/github/gh-aw/actions/runs/30602686038

Public-run evidence and failure classification

Primary scheduled-run blocker: unbounded dependency installation

In scheduled run #13:

• activation, checkout, experiment-state push, and discussion fetching succeeded;
• Install DataFlow began at approximately 03:54 UTC;
• pip install --quiet open-dataflow repeatedly timed out while resolving/downloading packages including typer-slim, mpmath, and presidio-analyzer;
• GitHub cancelled the operation at approximately 09:53 UTC; and
• PR fetching, Copilot CLI setup, and agent execution were skipped.

Scheduled runs #4–#8, #10, #11, and #13 show the same approximately six-hour cancellation pattern. The workflow's timeout-minutes: 30 is compiled onto the later agent execution step and does not bound the earlier custom install step.

Earlier runs produced artifacts but still did not complete

• Run Add workflow: githubnext/agentics/weekly-research #2 produced a roughly 657 KB dataset_clean.jsonl: https://github.com/github/gh-aw/actions/runs/26617128352
• Run Add workflow: githubnext/agentics/weekly-research #3 produced a roughly 474 KB dataset_clean.jsonl: https://github.com/github/gh-aw/actions/runs/26994527432

Run #2 later failed while initializing memory/dataflow-dataset, because the repository requires signed commits. The memory branch is still absent. This is an independent finalization failure after data production.

Local reproduction against the current package

The workflow's exact pipeline was tested locally with CPython 3.12.7 and open-dataflow==1.0.10.

Installation result

Using uv:

• resolution completed in approximately 2.7 seconds;
• a full cached installation completed in approximately 20 seconds;
• 219 packages were installed; and
• the virtual environment occupied approximately 1.8 GB.

The dependency graph includes Torch, torchvision, torchaudio, Transformers, Gradio, Google Cloud clients, Presidio, spaCy, and other packages unrelated to this workflow's narrow filtering/deduplication use case. This confirms both that uv materially improves resolution and that the installation must still have an explicit timeout and fallback.

Exact current pipeline result

The package imports successfully, but the workflow immediately executes:

storage = FileStorage(first_entry_file_name=INPUT)
len(storage)

FileStorage in 1.0.10 does not implement __len__, producing:

TypeError: object of type 'FileStorage' has no len()

The exception is caught by the workflow and sets storage = None, so every successfully installed 1.0.10 run enters the pure-Python fallback before trying any DataFlow operator.

The referenced modules also no longer exist:

ModuleNotFoundError: No module named 'dataflow.operators.filter'
ModuleNotFoundError: No module named 'dataflow.operators.dedup'

FileStorage also has no save() or iteration API matching the current script's output path.

With a three-record fixture, the exact workflow code did produce one deduplicated output record, but reported:

{
  "input_count": 3,
  "after_length_filter": 2,
  "after_alpha_filter": 2,
  "after_dedup": 1,
  "operators_used": ["fallback_python_pipeline"],
  "fallback_mode": true
}

This verifies the fallback, not the DataFlow path.

Current DataFlow API smoke test

The corresponding current API is exposed from dataflow.operators.general_text:

• CharNumberFilter
• AlphaWordsFilter
• MinHashDeduplicateFilter
• HashDeduplicateFilter

A local smoke pipeline using FileStorage.read() plus CharNumberFilter, AlphaWordsFilter, and HashDeduplicateFilter completed successfully and reduced the same fixture from 3 → 2 → 2 → 1 records.

However, AlphaWordsFilter attempts to download NLTK punkt_tab during construction even with use_tokenizer=False. That hidden network operation must not be allowed to introduce another unbounded wait. Its semantics also measure alphabetic words, whereas the existing fallback measures alphabetic characters.

Root causes

The workflow combines several independent problems:

1. pip install open-dataflow is unbounded, unpinned, fatal, and runs before the documented fallback can be reached.
2. The workflow targets obsolete DataFlow import paths and class names.
3. It treats FileStorage as a sized, iterable object with save(), which is not the current API.
4. A current candidate operator performs an implicit NLTK network download.
5. Reporting can imply that DataFlow processed the dataset even when only the fallback ran.
6. The absent repo-memory branch cannot be initialized by an unsigned Actions commit under current repository rules.

Agentic implementation plan

1. Modify .github/workflows/dataflow-pr-discussion-dataset.md; do not edit the generated lock file manually.
2. Preserve the current virtual-environment location, but replace the unbounded pip command with a bounded, pinned uv installation:
   • bound the uv bootstrap separately;
   • bound the DataFlow installation to no more than 20 minutes;
   • configure explicit HTTP timeout/retry values;
   • pin both installation tooling and open-dataflow to reviewed versions; and
   • make installation failure non-fatal so the existing fallback remains reachable.
3. Validate more than import dataflow. The install gate must smoke-test the exact storage and operator APIs selected by the pipeline.
4. Update the pipeline for the selected current DataFlow version:
   • count records through FileStorage.step().read("dict") or DataFrames, not len(storage);
   • import operators from dataflow.operators.general_text;
   • use current class names and signatures;
   • advance storage steps deliberately after each operator; and
   • materialize the final DataFrame/list to dataset_clean.jsonl instead of calling the nonexistent storage.save().
5. Preserve the current filtering contract:
   • enforce both the existing 50-character minimum and 100,000-character maximum;
   • preserve the 0.25 alphabetic-character-ratio threshold unless a semantic change is explicitly intended; and
   • keep near-deduplication with an exact-hash fallback.
6. Do not allow an implicit NLTK download. Either:
   • retain the deterministic pure-Python alphabetic-character-ratio stage while using DataFlow for supported length/dedup stages; or
   • provision and cache the required NLTK resource under its own timeout before constructing AlphaWordsFilter.
7. Treat fallback as a supported, observable mode rather than as DataFlow success:
   • emit an Actions warning when installation or API validation fails;
   • set fallback_mode from actual execution;
   • record only operators that really ran; and
   • make the Discussion title/body distinguish dataflow, mixed, and fallback execution modes.
8. Resolve repo-memory finalization with a maintainer-approved choice:
   • preferred: seed memory/dataflow-dataset once with a signed commit; or
   • remove repo-memory and its success criterion if trend memory is not needed for this experiment.
9. Preserve the existing caveman_mode experiment and its state. The install/API fix applies before variant execution and must not reset the experiment branch.
10. Run make recompile, review the generated lock file, and run the required strict compile and repository validation without manually triggering this scheduled workflow.

A representative bounded install shape is:

VENV=/tmp/gh-aw/python/venv
python3 -m venv "$VENV"

dataflow_ready=false
if timeout 5m "$VENV/bin/pip" install --quiet "uv==<reviewed-version>" &&
   timeout 20m env UV_HTTP_TIMEOUT=60 UV_HTTP_RETRIES=3 \
     "$VENV/bin/uv" pip install --python "$VENV/bin/python3" \
     "open-dataflow==<validated-version>" &&
   timeout 2m "$VENV/bin/python3" /path/to/exact_dataflow_api_smoke_test.py; then
  dataflow_ready=true
else
  echo "::warning::DataFlow installation or API validation failed; using the pure-Python fallback"
fi

The exact version pin should be validated on the repository's Ubuntu runner. open-dataflow==1.0.10 is a locally verified candidate, not proof of cross-platform runner compatibility.

Acceptance criteria

• No dependency installation, API import, or auxiliary download can occupy the job for six hours.
• A failed or unavailable DataFlow installation reaches the pure-Python fallback and records that mode honestly.
• A successful DataFlow-mode run executes at least one current DataFlow operator; merely importing the package is insufficient.
• The current pipeline no longer calls len(storage), iterates FileStorage, or calls storage.save().
• The selected operator imports and signatures are covered by a small deterministic smoke test.
• No operator performs an unbounded runtime download, including NLTK resources.
• The 50–100,000 character bound, 0.25 alphabetic-character ratio, and deduplication behavior remain explicit and testable.
• pipeline_stats.json and the Discussion report accurately distinguish DataFlow, mixed, and fallback execution.
• The workflow produces a non-empty dataset_clean.jsonl artifact or fails with a specific data-processing error.
• Repo-memory finalization cannot turn an already-produced dataset into an ambiguous DataFlow failure.
• The next scheduled run completes within the intended timeout rather than being cancelled by GitHub Actions' six-hour limit.
• caveman_mode experiment assignment and state remain intact.

Validation checklist

• Unit/smoke fixture: short text, duplicate text, low-alpha text, and valid text.
• Forced install failure: fallback completes and reports fallback mode.
• Successful install: the API smoke gate and at least one DataFlow operator run.
• Output assertions: valid JSONL, expected record counts, deterministic mode/operator reporting.
• make recompile and generated lock-file review.
• gh aw compile dataflow-pr-discussion-dataset --strict.
• make agent-report-progress and normal generated-shell/CI validation.

[haolpku]

Implementation available

The complete fix has now been implemented and locally validated.

• Commit: haolpku@78615c3
• Branch: https://github.com/haolpku/gh-aw/tree/dataflow/fix-dataflow-install-timeout
• Reviewable draft PR in the fork: Fix DataFlow dataset workflow execution haolpku/gh-aw#1
• Upstream comparison: main...haolpku:gh-aw:dataflow/fix-dataflow-install-timeout

An upstream draft PR was attempted with haolpku:dataflow/fix-dataflow-install-timeout as the head, but GitHub rejected CreatePullRequest because this account does not have the required permission on github/gh-aw. The fork is correctly related to github/gh-aw, is public, and the upstream repository allows forks.

Implemented fixes

• Bounded installation: 5-minute uv bootstrap, 20-minute DataFlow install, and 2-minute exact API smoke gate.
• Pinned uv==0.10.4 and open-dataflow==1.0.10.
• Replaced obsolete operator imports with dataflow.operators.general_text.
• Replaced unsupported len(storage) / storage.save() calls with FileStorage.step().read("dict") and explicit JSONL materialization.
• Added a checked-in deterministic pipeline plus unit tests; processing no longer depends on the agent generating Python scripts correctly.
• Uses CharNumberFilter and MinHashDeduplicateFilter, with current HashDeduplicateFilter and deterministic SHA-256 fallback paths.
• Preserves the 50–100,000 character and 0.25 alphabetic-character thresholds.
• Avoids AlphaWordsFilter, so no implicit NLTK resource download occurs.
• Converts pandas NaT/NaN values before JSON serialization; this was found by the first real DataFlow integration run.
• Reports the actual dataflow, fallback, or no_data mode and only operators that ran.
• Migrates weekly trend metadata from repo-memory to 90-day cache-memory, eliminating the unsigned orphan-branch initialization failure under the signed-commit rule.
• Preserves the existing caveman_mode experiment and recompiles the generated lock file.

Validation evidence

• make recompile: all 269 workflows compiled; the target workflow compiled successfully.
• ./gh-aw compile dataflow-pr-discussion-dataset --strict: 0 errors, 0 warnings.
• Python tests: 3 passed.
• Ruff lint/format checks: passed.
• Exact install path (python3 -m venv → pinned uv → pinned DataFlow → exact API smoke): passed.
• End-to-end DataFlow fixture: 4 input → 3 length-valid → 2 alpha-valid → 1 deduplicated record.
• Confirmed executed operators: CharNumberFilter, MinHashDeduplicateFilter.
• Confirmed NLTK data downloads: 0.

The repository-wide make agent-report-progress reaches Go lint but currently reports 50 pre-existing errcheck findings in unrelated Go files. This implementation changes no Go files; the target compile, Python tests, and real DataFlow integration all pass.