feat: internalize DataFlow cleaning pipelines - #41
Conversation
|
Caution The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased. |
|
Warning Review limit reached
Next review available in: 56 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (11)
📝 WalkthroughWalkthroughAdds deterministic data operators and SHA-pinned DataFlow compatibility, integrates them into workflow compilation, execution, import, demand drafting, capabilities, and Canvas configuration, and adds comprehensive unit, compatibility, integration, fixture, and documentation coverage. ChangesData operator runtime
Estimated code review effort: 5 (Critical) | ~120 minutes Sequence Diagram(s)sequenceDiagram
participant Canvas
participant WorkflowAPI
participant RuntimeRegistry
participant DataOperator
Canvas->>WorkflowAPI: submit workflow with operatorId, packVersion, and config
WorkflowAPI->>RuntimeRegistry: validate catalog kind and resolve runtime metadata
RuntimeRegistry->>DataOperator: resolve operator and pack
WorkflowAPI->>DataOperator: execute candidate items
DataOperator-->>WorkflowAPI: output items, metrics, lineage, and rejected IDs
WorkflowAPI-->>Canvas: workflow events and results
Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
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 |
|
✅ Health: 7.8 📋 At a glance 🚨 Change risk: 9.9/10 (high)
📊 Full report · ⭐ Star Repowise · 📥 Install bot · Last updated 2026-07-25 07:19 UTC |
|
@coderabbitai review |
✅ Action performedReview finished.
|
- text.rule-filter: reject configs with != 1 field instead of silently
ignoring extra fields (fail-closed, matches compat single-input_key scope)
- text.deduplicate: treat missing/non-string fields as empty text instead
of raising KeyError; reject empty field lists
- node-internals: resolve data-operator internals via explicit
operatorId->kind map instead of operatorId.split(".")[1]
- ci: run tests/compat + tests/integration in backend job; add frontend
workflow-checks job (data-operator projection + workflow contracts)
- tests: cover multi-field rejection, empty-field rejection, missing-field
dedup
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 8
🧹 Nitpick comments (21)
tests/unit/test_data_operators.py (2)
323-330: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winEmpty-batch loop never exercises the 1.1.0 specs.
execute_data_operator(spec.operator_id, [])omitspack_version, so the three v1.1.0 specs resolve back to the legacy binding and are asserted twice instead of once each. Passingpack_version=spec.pack_versioncloses that gap.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/unit/test_data_operators.py` around lines 323 - 330, Update test_empty_batch_has_consistent_metrics_for_every_operator to pass each spec’s pack_version into execute_data_operator alongside spec.operator_id and the empty input, ensuring the v1.1.0 specifications resolve to their intended bindings.
50-73: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winBoth tests couple to
COMPAT_OPERATOR_DEFINITIONS[0]ordering while the config at Line 68 hardcodestext.cleankeys. If the definition order changes, Line 68'sfields/operationsconfig becomes an unsupported-configValueErrorrather than a meaningful failure. Also, theoperatorId/operator_iddual-key lookup is dead defence —tests/unit/test_dataflow_compat.py(Line 43) asserts the camelCase keys.♻️ Select the definition explicitly
- definition = COMPAT_OPERATOR_DEFINITIONS[0] - operator_id = definition.get("operatorId", definition.get("operator_id")) - pack_version = definition.get("packVersion", definition.get("pack_version")) + definition = next( + item + for item in COMPAT_OPERATOR_DEFINITIONS + if item["operatorId"] == "text.clean" + ) + operator_id = definition["operatorId"] + pack_version = definition["packVersion"]🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/unit/test_data_operators.py` around lines 50 - 73, Update test_versioned_registry_keeps_legacy_default_and_resolves_compat_exactly and test_execute_reports_the_resolved_pack_version to select the compatibility definition explicitly by its expected operator ID or another stable identifier, rather than using COMPAT_OPERATOR_DEFINITIONS[0]. Use the definition’s asserted camelCase keys directly, and ensure the execute_data_operator input matches the selected operator’s supported configuration instead of hardcoding text.clean fields and operations.tests/unit/test_dataflow_compat.py (1)
214-224: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueNaming collision hurts readability. The
rejectedparameter (input text) is shadowed conceptually byrejected_ids; renaming torejecting_text/passing_textmakes the assertion pair obvious.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/unit/test_dataflow_compat.py` around lines 214 - 224, Rename the test_phase2_rules_preserve_strict_and_inclusive_boundaries parameters passing and rejected to passing_text and rejecting_text, and update their uses in the _candidate calls. Keep the output and rejected_ids assertions unchanged.tests/integration/test_dataflow_operator_pipeline_api.py (3)
547-567: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winUse
itertools.pairwisefor the consecutive-node edges. Also silences Ruff B905/RUF007.♻️ Proposed change
+from itertools import pairwise @@ - for source, target in zip(ordered_ids, ordered_ids[1:]) + for source, target in pairwise(ordered_ids)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/integration/test_dataflow_operator_pipeline_api.py` around lines 547 - 567, Update the consecutive-node edge construction in the ordered_ids flow to use itertools.pairwise instead of zip(ordered_ids, ordered_ids[1:]), adding the required import if needed; preserve the existing source/target port mapping and edge output.Source: Linters/SAST tools
1006-1024: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDuplicated operator-id list. The same seven node ids are already spelled out at Lines 645-653; extract a module-level constant so the determinism test and the pipeline test cannot drift apart when the chain changes.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/integration/test_dataflow_operator_pipeline_api.py` around lines 1006 - 1024, Extract the shared seven operator IDs from the existing list near the determinism test into a module-level constant, then update partials to filter against that constant instead of an inline set. Reuse the same constant in both tests so changes to the operator chain remain synchronized.
124-341: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚖️ Poor tradeoffManifest snapshot duplicates the registry lock in
tests/compat/dataflow/test_pinned_compatibility.py(Lines 172-322). Both encode every operator'sconfigKeysverbatim, so any registry change requires two synchronized edits. Consider asserting the HTTP projection againstlist_data_operator_specs()here and keeping the literal snapshot in one place.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/integration/test_dataflow_operator_pipeline_api.py` around lines 124 - 341, Replace the duplicated _EXPECTED_OPERATOR_MANIFEST literal in the dataflow pipeline API test with an assertion derived from list_data_operator_specs(), reusing the registry-backed manifest data already maintained by the pinned compatibility test. Preserve validation of the HTTP response projection, including operator metadata and configKeys, while removing the second manually synchronized snapshot.tests/compat/dataflow/test_upstream_oracle.py (3)
69-84: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winNew test files trip Ruff rules with no test-scoped exemption. Both findings stem from one root cause: the repo's Ruff configuration applies production-oriented rules (
S6xx,RUF001) totests/, so deliberate test data and dev-only subprocess calls fail lint.
tests/compat/dataflow/test_upstream_oracle.py#L69-L84: silenceS603/S607for the fixed-argvgitcalls, or exempttests/fromflake8-banditrules.tests/unit/test_dataflow_compat.py#L80-L85: silenceRUF001for the intentional fullwidth＆,；,？fixtures.Prefer a single
per-file-ignoresentry fortests/**over scattered# noqacomments.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/compat/dataflow/test_upstream_oracle.py` around lines 69 - 84, Update the Ruff configuration with a single tests/** per-file-ignores entry covering S603, S607, and RUF001. This should exempt the fixed-argument git subprocess calls in tests/compat/dataflow/test_upstream_oracle.py:69-84 and the intentional fullwidth-character fixtures in tests/unit/test_dataflow_compat.py:80-85; do not add scattered noqa comments.Source: Linters/SAST tools
176-177: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
getattrfailures escape the skip guard._import_classraisesAttributeError(notImportError) when the module exists but the pinned class was renamed, turning a stale-checkout situation into an error instead of a skip/clear failure. Consider catching(ImportError, AttributeError).Also applies to: 231-232
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/compat/dataflow/test_upstream_oracle.py` around lines 176 - 177, Update the exception handler around _import_class in the affected test sections to catch both ImportError and AttributeError, so missing modules and renamed pinned classes are handled by the existing pytest.skip path. Preserve the current skip message and behavior.
69-84: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueRuff S603/S607 will flag these
gitcalls. Argv is fixed and the root comes from a developer-supplied env var, so the ast-grep "command from incoming request" hint is a false positive — but the Ruff findings still need either a per-file ignore or atests/exclusion in the Ruff config to keep lint green.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/compat/dataflow/test_upstream_oracle.py` around lines 69 - 84, Suppress Ruff S603/S607 for the fixed-argument git subprocess calls in test_upstream_oracle.py, using a narrowly scoped per-file ignore or an equivalent tests/ exclusion in the Ruff configuration. Keep the existing subprocess behavior unchanged and ensure lint passes without broadening unrelated suppressions.Source: Linters/SAST tools
tests/compat/dataflow/test_pinned_compatibility.py (1)
373-386: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueDuplicate lock assertions. Lines 374-375 restate what
test_fixture_is_auditable_and_pinned_to_the_supported_upstream_revisionalready asserts (lines 59, 64); the phase-2 value here is the blocklist digests and the 34-alias set union. Trimming keeps failures pointing at one place.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/compat/dataflow/test_pinned_compatibility.py` around lines 373 - 386, The test_phase2_manifest_locks_alias_count_and_new_source_asset_digests test redundantly asserts the compatibility alias count and upstream file count already covered by test_fixture_is_auditable_and_pinned_to_the_supported_upstream_revision. Remove those duplicate assertions while retaining the blocklist digest checks and the 34-alias source ID set-union validation.tests/integration/test_workflow_patch_api.py (1)
828-828: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueMutating the parametrized dict in place.
nodeis the module-level dict built at collection time; assigningnode["id"]mutates shared state across invocations. It's idempotent here, but adding the id in the parametrize data (ornode = {**node, "id": "unsupported"}) keeps the case data immutable.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/integration/test_workflow_patch_api.py` at line 828, Update the parametrized test case around node so it does not mutate the shared module-level dictionary in place; add the unsupported id during parameter construction or create a new dictionary copy with the id before use, preserving the existing test behavior.backend/workflow/data_operators.py (2)
810-822: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
symbolRatiopredicate has a redundant clause.
not c.isalnum() and not c.isalpha()—isalnum()is already true for every alphabetic char, so theisalpha()term never changes the result. Dropping it makes the intent (non-alphanumeric = symbol) clearer.♻️ Simplify the symbol predicate
- symbols = [c for c in non_space if not c.isalnum() and not c.isalpha()] + symbols = [c for c in non_space if not c.isalnum()]🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/workflow/data_operators.py` around lines 810 - 822, Update the symbols comprehension in _statistics to classify characters using only the non-alphanumeric predicate, removing the redundant alphabetic check while preserving the existing symbolRatio calculation.
465-480: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valuePII redaction regexes are duplicated between
_text_refineand_clean_text.The email and phone patterns are byte-identical in both places. Extracting them into module-level compiled constants keeps the two operators from drifting apart as the patterns evolve.
Also applies to: 523-527
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/workflow/data_operators.py` around lines 465 - 480, Extract the duplicated email and phone redaction patterns used by _text_refine and _clean_text into shared module-level compiled regex constants. Update both functions to reuse those constants while preserving the existing substitution behavior and replacement tokens.backend/workflow/demand_assembler.py (2)
264-267: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winUse
itertools.pairwisefor the successive-pair walk (Ruff B905/RUF007).The neighbouring
zipat Line 211 already passesstrict=True; this one has no explicitstrict=, which Ruff flags, andpairwiseexpresses the intent directly.♻️ Proposed fix
- for source_id, target_id in zip( - operator_node_ids, - operator_node_ids[1:], - ): + for source_id, target_id in pairwise(operator_node_ids):Add the import:
from itertools import pairwise🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/workflow/demand_assembler.py` around lines 264 - 267, Replace the successive-pair loop over operator_node_ids with itertools.pairwise, adding the pairwise import and preserving the existing source_id/target_id iteration behavior.Source: Linters/SAST tools
307-415: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winKeyword heuristic is fine, but the operator descriptors are now a second source of truth for pack versions.
"1.1.0" if dataflow_compat else "1.0.0"hardcodes pack versions that also live indataflow_compat.COMPAT_PACK_VERSIONanddata_operators._LEGACY_VERSION. Importing those constants keeps drafting in sync when a pack version bumps.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/workflow/demand_assembler.py` around lines 307 - 415, Update _data_operators_for_need to reuse the canonical pack-version constants from dataflow_compat.COMPAT_PACK_VERSION and data_operators._LEGACY_VERSION instead of hardcoding "1.1.0" and "1.0.0" in the clean, deduplicate, and rule-filter descriptors. Preserve the existing dataflow_compat selection behavior while importing and referencing those shared constants.backend/workflow/runtime_registry.py (1)
345-374: 📐 Maintainability & Code Quality | 🔵 Trivial | 🏗️ Heavy liftData-operator resolution rules and the legacy pack-version default are duplicated across three layers. The same "read operatorId → default packVersion to
1.0.0→ resolve spec → classify unknown/unsupported/kind-mismatch" logic exists in the compiler validator, the runtime resolver, and the tracer executor, so a rule change must be edited in three places to stay consistent.
backend/workflow/runtime_registry.py#L345-L374: extract the shared resolution/classification into one helper (ideally exported fromdata_operators) and call it here.backend/workflow/compiler.py#L388-L413: consume that helper instead of re-derivingexpected_kind, the packVersion default, and the error codes.backend/workflow/opencli_hda_tracer.py#L97-L98: import the legacy pack-version default rather than redefining_LEGACY_DATA_OPERATOR_PACK_VERSIONa third time.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/workflow/runtime_registry.py` around lines 345 - 374, Centralize data-operator resolution and error classification in a shared helper, preferably exported from data_operators, covering operatorId parsing, the 1.0.0 legacy pack-version default, spec resolution, and unknown/unsupported-version/kind-mismatch outcomes. Update _resolve_data_operator_node in backend/workflow/runtime_registry.py (345-374) and the compiler validation flow in backend/workflow/compiler.py (388-413) to use this helper instead of duplicating the rules; update backend/workflow/opencli_hda_tracer.py (97-98) to import the shared legacy pack-version constant and remove its local definition.backend/workflow/external_importer.py (1)
197-215: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winReconstructing the source-id prefix duplicates
dataflow_compat.
f"dataflow@{DATAFLOW_COMPAT_SHA}::"is built here twice whiledataflow_compatalready owns_SOURCE_PREFIX(and theDATAFLOW_ALIAS_SOURCE_IDSmap). Exporting that prefix and importing it keeps the format in one place.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/workflow/external_importer.py` around lines 197 - 215, Update _dataflow_source_id to reuse the exported source-prefix constant from dataflow_compat instead of rebuilding f"dataflow@{DATAFLOW_COMPAT_SHA}::" locally. Import that constant and use it both when reconstructing source_id and when validating its prefix, while preserving the existing error behavior.backend/workflow/capability_projection.py (1)
545-555: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAnnotate the spec index with its real type.
dict[str, list[object]]makesspec.operator_id/spec.kinduntyped accesses onobject; a type checker cannot verify this helper. ImportDataOperatorSpecalongsidelist_data_operator_specsand usedict[str, list[DataOperatorSpec]].🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/workflow/capability_projection.py` around lines 545 - 555, Update the spec index in the workflow capability projection to use the concrete DataOperatorSpec type: import DataOperatorSpec alongside list_data_operator_specs and change specs_by_kind from dict[str, list[object]] to dict[str, list[DataOperatorSpec]], preserving the existing grouping and sorting logic.backend/workflow/opencli_hda_tracer.py (1)
524-547: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winSwallowed exception leaves no server-side diagnostics.
The failed event intentionally carries only
errorType(no message), which is good for not leaking candidate data, but the exception itself is then discarded entirely — nothing is logged. A structuredlogger.exception(...)here keeps the operator failure debuggable without widening what the trace exposes.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@backend/workflow/opencli_hda_tracer.py` around lines 524 - 547, In the data-operator exception handler around _binding_id and WorkflowRunBlockReason, add a structured logger.exception call before continuing so the original exception and traceback are retained server-side. Keep reason.details and the emitted failed event limited to the existing errorType and non-sensitive fields, without exposing exc details in the trace.frontend/components/flow/inspector.tsx (1)
295-335: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winEvery valid JSON keystroke takes an undo snapshot.
Each time the textarea content becomes valid JSON,
updateParameterField→updateParameterInterfaceFieldruns, and that action unconditionally callsget().takeSnapshot()beforeset()(store.ts lines 999-1001). Typing out a JSON object will therefore push many intermediate snapshots onto the undo stack (one per moment the text is valid), making undo effectively useless for this field compared to a single edit action.Consider only committing/snapshotting on blur (or debouncing the commit), while still updating
jsonDrafts/jsonErrorslive for immediate validation feedback.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@frontend/components/flow/inspector.tsx` around lines 295 - 335, The JSON Textarea’s onChange currently calls updateParameterField for every valid intermediate value, creating an undo snapshot per keystroke. Keep updating jsonDrafts and jsonErrors during onChange for live validation, but defer updateParameterField and its snapshot to onBlur, committing the latest valid parsed value once while preserving invalid-input feedback.frontend/scripts/test-data-operator-nodes.mjs (1)
1-311: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winSolid test coverage; consider adding a case for nodes without
ui.catalogId.This harness thoroughly exercises the catalog-created data-operator flow (operator selection, JSON config parsing, version pinning, draft clearing). All node fixtures here go through
createWorkflowNodeFromCatalog, soui.catalogIdis always one of the four"intelligence.data.*"ids. Adding a case for a node withparams.operatorId/configset directly (no matchingui.catalogId— e.g. simulating a DataFlow-imported node) would have caught the gating/kind-inference inconsistencies flagged infrontend/lib/flow/store.tsandfrontend/lib/workflow/node-internals.ts.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@frontend/scripts/test-data-operator-nodes.mjs` around lines 1 - 311, Extend the harness with an imported node whose params.operatorId and params.config are set directly and whose ui has no matching catalogId, simulating a DataFlow-imported node. Use importWorkflowProject and applyWorkflowCapabilities to verify the node is recognized as a data operator without catalog metadata, and assert parameter-interface updates and inferred node kind behave consistently with catalog-created nodes.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@backend/workflow/capability_projection.py`:
- Around line 602-604: Update the operatorIds construction in the capability
projection to remove duplicate operator IDs while preserving their first-seen
order, matching the deduplicated packs behavior. Keep the operators collection
unchanged and use the existing operators sequence as the source.
In `@backend/workflow/compiler.py`:
- Around line 441-466: The fallback classification around resolve_data_operator
must detect operator IDs registered in any pack version, not only the legacy
version. Replace the legacy-only probe with the existing all-version registry
lookup, such as list_data_operator_specs(), and classify the result as
unsupported_data_operator_version when the ID exists in another version;
preserve unknown_data_operator only when no registered version contains
operator_id.
In `@backend/workflow/dataflow_compat.py`:
- Around line 926-932: Update the lineStartWithBulletpoint branch around
_paragraphs and ratio to handle an empty paragraphs result before dividing by
len(paragraphs), returning the same safe outcome used by the related empty-input
checks. Preserve the existing bullet detection and threshold behavior for
non-empty paragraphs.
- Around line 631-636: Update the field extraction in the deduplication loop
around `_normalized` so missing or non-string field values are handled
defensively instead of indexing `normalized` directly. Reuse the module’s
established `_unsupported(...)` behavior or coerce invalid values to an empty
string, ensuring malformed candidates do not raise `KeyError` or
`AttributeError` while preserving normal string handling.
In `@backend/workflow/runtime_registry.py`:
- Around line 391-403: Validate the merged operator configuration in the
compilation flow around config construction, using the available
spec.config_keys to reject any keys outside the supported set before runtime
execution. Apply the same validation in _validate_data_operator_node, preserve
operatorId/packVersion/config handling, and report unsupported parameters as
compile-time errors rather than allowing execute_data_operator to fail later.
In `@docs/dataflow-compatibility-matrix.md`:
- Around line 28-31: The “Tokenizer or statistical text” row conflates runnable
StopWordFilter behavior with unavailable tokenizer-backed refiners. Update its
examples to distinguish basic StopWordFilter support when use_tokenizer=false
from unsupported tokenizer-backed stop-word, stemming, and lemmatization
implementations.
In `@docs/dataflow-operator-packs.md`:
- Around line 22-33: Clarify the documentation around text.deduplicate to
distinguish the v1 native SimHash capability from the pinned
builtin.text-cleaning@1.1.0 compatibility profile. State that SimHash is
supported natively in v1 but excluded from the 1.1.0 pinned compatibility import
contract, consistent with SimHashDeduplicateFilter being unavailable while
HashDeduplicateFilter and NgramHashDeduplicateFilter are mapped.
In `@tests/compat/dataflow/test_pinned_compatibility.py`:
- Around line 573-589: Split the combined assertion in the compatibility test
into separate seam-specific checks. If translate_dataflow_alias is the intended
fail-closed gate, keep only its ValueError assertion and remove the unreachable
execute_data_operator call; otherwise, assert translation succeeds separately
and wrap execute_data_operator in its own pytest.raises block.
---
Nitpick comments:
In `@backend/workflow/capability_projection.py`:
- Around line 545-555: Update the spec index in the workflow capability
projection to use the concrete DataOperatorSpec type: import DataOperatorSpec
alongside list_data_operator_specs and change specs_by_kind from dict[str,
list[object]] to dict[str, list[DataOperatorSpec]], preserving the existing
grouping and sorting logic.
In `@backend/workflow/data_operators.py`:
- Around line 810-822: Update the symbols comprehension in _statistics to
classify characters using only the non-alphanumeric predicate, removing the
redundant alphabetic check while preserving the existing symbolRatio
calculation.
- Around line 465-480: Extract the duplicated email and phone redaction patterns
used by _text_refine and _clean_text into shared module-level compiled regex
constants. Update both functions to reuse those constants while preserving the
existing substitution behavior and replacement tokens.
In `@backend/workflow/demand_assembler.py`:
- Around line 264-267: Replace the successive-pair loop over operator_node_ids
with itertools.pairwise, adding the pairwise import and preserving the existing
source_id/target_id iteration behavior.
- Around line 307-415: Update _data_operators_for_need to reuse the canonical
pack-version constants from dataflow_compat.COMPAT_PACK_VERSION and
data_operators._LEGACY_VERSION instead of hardcoding "1.1.0" and "1.0.0" in the
clean, deduplicate, and rule-filter descriptors. Preserve the existing
dataflow_compat selection behavior while importing and referencing those shared
constants.
In `@backend/workflow/external_importer.py`:
- Around line 197-215: Update _dataflow_source_id to reuse the exported
source-prefix constant from dataflow_compat instead of rebuilding
f"dataflow@{DATAFLOW_COMPAT_SHA}::" locally. Import that constant and use it
both when reconstructing source_id and when validating its prefix, while
preserving the existing error behavior.
In `@backend/workflow/opencli_hda_tracer.py`:
- Around line 524-547: In the data-operator exception handler around _binding_id
and WorkflowRunBlockReason, add a structured logger.exception call before
continuing so the original exception and traceback are retained server-side.
Keep reason.details and the emitted failed event limited to the existing
errorType and non-sensitive fields, without exposing exc details in the trace.
In `@backend/workflow/runtime_registry.py`:
- Around line 345-374: Centralize data-operator resolution and error
classification in a shared helper, preferably exported from data_operators,
covering operatorId parsing, the 1.0.0 legacy pack-version default, spec
resolution, and unknown/unsupported-version/kind-mismatch outcomes. Update
_resolve_data_operator_node in backend/workflow/runtime_registry.py (345-374)
and the compiler validation flow in backend/workflow/compiler.py (388-413) to
use this helper instead of duplicating the rules; update
backend/workflow/opencli_hda_tracer.py (97-98) to import the shared legacy
pack-version constant and remove its local definition.
In `@frontend/components/flow/inspector.tsx`:
- Around line 295-335: The JSON Textarea’s onChange currently calls
updateParameterField for every valid intermediate value, creating an undo
snapshot per keystroke. Keep updating jsonDrafts and jsonErrors during onChange
for live validation, but defer updateParameterField and its snapshot to onBlur,
committing the latest valid parsed value once while preserving invalid-input
feedback.
In `@frontend/scripts/test-data-operator-nodes.mjs`:
- Around line 1-311: Extend the harness with an imported node whose
params.operatorId and params.config are set directly and whose ui has no
matching catalogId, simulating a DataFlow-imported node. Use
importWorkflowProject and applyWorkflowCapabilities to verify the node is
recognized as a data operator without catalog metadata, and assert
parameter-interface updates and inferred node kind behave consistently with
catalog-created nodes.
In `@tests/compat/dataflow/test_pinned_compatibility.py`:
- Around line 373-386: The
test_phase2_manifest_locks_alias_count_and_new_source_asset_digests test
redundantly asserts the compatibility alias count and upstream file count
already covered by
test_fixture_is_auditable_and_pinned_to_the_supported_upstream_revision. Remove
those duplicate assertions while retaining the blocklist digest checks and the
34-alias source ID set-union validation.
In `@tests/compat/dataflow/test_upstream_oracle.py`:
- Around line 69-84: Update the Ruff configuration with a single tests/**
per-file-ignores entry covering S603, S607, and RUF001. This should exempt the
fixed-argument git subprocess calls in
tests/compat/dataflow/test_upstream_oracle.py:69-84 and the intentional
fullwidth-character fixtures in tests/unit/test_dataflow_compat.py:80-85; do not
add scattered noqa comments.
- Around line 176-177: Update the exception handler around _import_class in the
affected test sections to catch both ImportError and AttributeError, so missing
modules and renamed pinned classes are handled by the existing pytest.skip path.
Preserve the current skip message and behavior.
- Around line 69-84: Suppress Ruff S603/S607 for the fixed-argument git
subprocess calls in test_upstream_oracle.py, using a narrowly scoped per-file
ignore or an equivalent tests/ exclusion in the Ruff configuration. Keep the
existing subprocess behavior unchanged and ensure lint passes without broadening
unrelated suppressions.
In `@tests/integration/test_dataflow_operator_pipeline_api.py`:
- Around line 547-567: Update the consecutive-node edge construction in the
ordered_ids flow to use itertools.pairwise instead of zip(ordered_ids,
ordered_ids[1:]), adding the required import if needed; preserve the existing
source/target port mapping and edge output.
- Around line 1006-1024: Extract the shared seven operator IDs from the existing
list near the determinism test into a module-level constant, then update
partials to filter against that constant instead of an inline set. Reuse the
same constant in both tests so changes to the operator chain remain
synchronized.
- Around line 124-341: Replace the duplicated _EXPECTED_OPERATOR_MANIFEST
literal in the dataflow pipeline API test with an assertion derived from
list_data_operator_specs(), reusing the registry-backed manifest data already
maintained by the pinned compatibility test. Preserve validation of the HTTP
response projection, including operator metadata and configKeys, while removing
the second manually synchronized snapshot.
In `@tests/integration/test_workflow_patch_api.py`:
- Line 828: Update the parametrized test case around node so it does not mutate
the shared module-level dictionary in place; add the unsupported id during
parameter construction or create a new dictionary copy with the id before use,
preserving the existing test behavior.
In `@tests/unit/test_data_operators.py`:
- Around line 323-330: Update
test_empty_batch_has_consistent_metrics_for_every_operator to pass each spec’s
pack_version into execute_data_operator alongside spec.operator_id and the empty
input, ensuring the v1.1.0 specifications resolve to their intended bindings.
- Around line 50-73: Update
test_versioned_registry_keeps_legacy_default_and_resolves_compat_exactly and
test_execute_reports_the_resolved_pack_version to select the compatibility
definition explicitly by its expected operator ID or another stable identifier,
rather than using COMPAT_OPERATOR_DEFINITIONS[0]. Use the definition’s asserted
camelCase keys directly, and ensure the execute_data_operator input matches the
selected operator’s supported configuration instead of hardcoding text.clean
fields and operations.
In `@tests/unit/test_dataflow_compat.py`:
- Around line 214-224: Rename the
test_phase2_rules_preserve_strict_and_inclusive_boundaries parameters passing
and rejected to passing_text and rejecting_text, and update their uses in the
_candidate calls. Keep the output and rejected_ids assertions unchanged.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 70948026-fbd2-407a-91ac-bdca59d16124
📒 Files selected for processing (32)
backend/api/v1/workflows.pybackend/schemas/workflow.pybackend/workflow/capability_projection.pybackend/workflow/compiler.pybackend/workflow/data_operators.pybackend/workflow/dataflow_compat.pybackend/workflow/demand_assembler.pybackend/workflow/external_importer.pybackend/workflow/node_registry.pybackend/workflow/opencli_hda_tracer.pybackend/workflow/runtime_contracts.pybackend/workflow/runtime_registry.pydocs/dataflow-compatibility-matrix.mddocs/dataflow-operator-packs.mdfrontend/components/flow/inspector.tsxfrontend/lib/flow/store.tsfrontend/lib/flow/types.tsfrontend/lib/workflow/capabilities.tsfrontend/lib/workflow/node-catalog.tsfrontend/lib/workflow/node-contracts.tsfrontend/lib/workflow/node-internals.tsfrontend/lib/workflow/parameter-interface.tsfrontend/lib/workflow/schema.tsfrontend/scripts/test-data-operator-nodes.mjstests/compat/dataflow/test_pinned_compatibility.pytests/compat/dataflow/test_upstream_oracle.pytests/fixtures/dataflow/pinned_f62aa134_golden.jsontests/fixtures/dataflow/pinned_f62aa134_phase2_golden.jsontests/integration/test_dataflow_operator_pipeline_api.pytests/integration/test_workflow_patch_api.pytests/unit/test_data_operators.pytests/unit/test_dataflow_compat.py
| flat_config = { | ||
| key: value | ||
| for key, value in node.params.items() | ||
| if key not in {"operatorId", "packVersion", "config"} | ||
| } | ||
| nested_config = node.params.get("config") | ||
| config: Any = ( | ||
| {**flat_config, **nested_config} | ||
| if isinstance(nested_config, dict) | ||
| else flat_config | ||
| if nested_config is None | ||
| else nested_config | ||
| ) |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Flat params leak into the operator config and only fail at run time.
Every param other than operatorId/packVersion/config is merged into config, and execute_data_operator rejects unknown keys with ValueError. So an authoring-time stray param (or a UI field written into params) compiles cleanly and then fails the run with Unsupported config for …. Since spec.config_keys is already available here (and in _validate_data_operator_node), validating the merged keys at compile time would surface this as a compile error instead of a failed run.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@backend/workflow/runtime_registry.py` around lines 391 - 403, Validate the
merged operator configuration in the compilation flow around config
construction, using the available spec.config_keys to reject any keys outside
the supported set before runtime execution. Apply the same validation in
_validate_data_operator_node, preserve operatorId/packVersion/config handling,
and report unsupported parameters as compile-time errors rather than allowing
execute_data_operator to fail later.
| with pytest.raises(ValueError, match="dataflow_operator_unsupported"): | ||
| invocation = translate_dataflow_alias( | ||
| DATAFLOW_ALIAS_SOURCE_IDS[alias], | ||
| init_config, | ||
| {"input_key": "content"}, | ||
| ) | ||
| execute_data_operator( | ||
| invocation.operator_id, | ||
| [ | ||
| { | ||
| "candidateId": f"{alias}-tokenizer", | ||
| "normalizedData": {"content": "plain input"}, | ||
| } | ||
| ], | ||
| invocation.config, | ||
| pack_version=invocation.pack_version, | ||
| ) |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Two calls inside one pytest.raises block makes the second unreachable.
If translate_dataflow_alias raises (which it will for use_tokenizer: True if the fail-closed check lives in translation), execute_data_operator is never invoked, so the test silently asserts nothing about the executor path. Pin down which seam is expected to fail closed.
♻️ Split the two seams
- with pytest.raises(ValueError, match="dataflow_operator_unsupported"):
- invocation = translate_dataflow_alias(
- DATAFLOW_ALIAS_SOURCE_IDS[alias],
- init_config,
- {"input_key": "content"},
- )
- execute_data_operator(
- invocation.operator_id,
- [
- {
- "candidateId": f"{alias}-tokenizer",
- "normalizedData": {"content": "plain input"},
- }
- ],
- invocation.config,
- pack_version=invocation.pack_version,
- )
+ invocation = translate_dataflow_alias(
+ DATAFLOW_ALIAS_SOURCE_IDS[alias],
+ init_config,
+ {"input_key": "content"},
+ )
+ with pytest.raises(ValueError, match="dataflow_operator_unsupported"):
+ execute_data_operator(
+ invocation.operator_id,
+ [
+ {
+ "candidateId": f"{alias}-tokenizer",
+ "normalizedData": {"content": "plain input"},
+ }
+ ],
+ invocation.config,
+ pack_version=invocation.pack_version,
+ )If translation is the intended gate, drop the executor call instead.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| with pytest.raises(ValueError, match="dataflow_operator_unsupported"): | |
| invocation = translate_dataflow_alias( | |
| DATAFLOW_ALIAS_SOURCE_IDS[alias], | |
| init_config, | |
| {"input_key": "content"}, | |
| ) | |
| execute_data_operator( | |
| invocation.operator_id, | |
| [ | |
| { | |
| "candidateId": f"{alias}-tokenizer", | |
| "normalizedData": {"content": "plain input"}, | |
| } | |
| ], | |
| invocation.config, | |
| pack_version=invocation.pack_version, | |
| ) | |
| invocation = translate_dataflow_alias( | |
| DATAFLOW_ALIAS_SOURCE_IDS[alias], | |
| init_config, | |
| {"input_key": "content"}, | |
| ) | |
| with pytest.raises(ValueError, match="dataflow_operator_unsupported"): | |
| execute_data_operator( | |
| invocation.operator_id, | |
| [ | |
| { | |
| "candidateId": f"{alias}-tokenizer", | |
| "normalizedData": {"content": "plain input"}, | |
| } | |
| ], | |
| invocation.config, | |
| pack_version=invocation.pack_version, | |
| ) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/compat/dataflow/test_pinned_compatibility.py` around lines 573 - 589,
Split the combined assertion in the compatibility test into separate
seam-specific checks. If translate_dataflow_alias is the intended fail-closed
gate, keep only its ValueError assertion and remove the unreachable
execute_data_operator call; otherwise, assert translation succeeds separately
and wrap execute_data_operator in its own pytest.raises block.
- dataflow_compat: guard lineStartWithBulletpoint against whitespace-only text (ZeroDivisionError), matching the lineEndWithEllipsis guard - capability_projection: dedupe operatorIds across pack versions while preserving order - compiler: probe all registered pack versions when classifying unknown_data_operator vs unsupported_data_operator_version - tests/compat: split translate/execute seams so the tokenizer fail-closed assertion pins the executor path instead of hiding behind translation - docs: clarify runnable StopWordFilter (use_tokenizer=false) vs tokenizer-backed processing, and SimHash v1-native vs pinned 1.1.0 scope
- test_dataflow_operator_pipeline_api: operatorIds assertion now expects the deduped projection (order-preserving unique ids) - ci: frontend job needs pnpm 10 — frontend/pnpm-workspace.yaml is a pnpm 10 config store (allowBuilds, no packages field), pnpm 9 aborts
|
Caution The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased. |
1 participant
What changed
Why
OpenCLI Admin needs to reproduce DataFlow-style data preparation and cleaning pipelines without requiring DataFlow or pandas at application runtime, while avoiding fake implementations for model-, retrieval-, OCR-, or document-dependent operators.
Validation
2026 passed, 13 skipped(90.42% Python coverage)tsc --noEmitpassed