|
|
@@ -0,0 +1,1034 @@
|
|
|
+# Data Rule Execution Completion Implementation Plan
|
|
|
+
|
|
|
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
|
|
|
+
|
|
|
+**Goal:** Complete the governed path from natural-language rules to deterministic compilation, real data execution, violation evidence, and Data Factory canary/activation/rollback.
|
|
|
+
|
|
|
+**Architecture:** Keep DataStandardVersion, RuleVersion, DataFlowVersion, and DataFlowDeployment as separate immutable aggregates. A server-owned schema and dataset resolver binds a released production line to physical data only in Data Factory; a compiler registry emits strictly validated SQLGlot or Polars plans, and the Runner loads plans by binding/version/hash without accepting inline code. Start with a PostgreSQL/MySQL SQL-pushdown vertical slice, then add Polars artifact execution for cross-source pipelines.
|
|
|
+
|
|
|
+**Tech Stack:** Flask 2.3, SQLAlchemy 2, Alembic/PostgreSQL JSONB, SQLGlot, Polars Lazy API, existing DataSourceConnectionManager, DataOps Runner, Kestra adapter/compiler, MinIO, Vue 2/Vuetify, pytest, Docker Compose.
|
|
|
+
|
|
|
+## Global Constraints
|
|
|
+
|
|
|
+- Natural language is an authoring source; no model output executes before deterministic validation, compilation, testing, and publication.
|
|
|
+- Data Standard and Data Flow reuse the same RuleVersion, compiler registry, artifacts, and execution evidence.
|
|
|
+- DataFlowVersion contains fixed StandardVersion and RuleVersion references; runtime never resolves `latest`.
|
|
|
+- Data Factory deploys an immutable production-line package and binds environment/resources/schedule/data sources; it never regenerates rule semantics.
|
|
|
+- Runner accepts only server-published plan identifiers and hashes; no request may contain inline SQL, Python, credentials, or artifact bodies.
|
|
|
+- SQL compilation uses SQLGlot strict dialect handling and fails closed on unsupported semantics.
|
|
|
+- Cross-source execution uses Polars Lazy plans reconstructed from closed JSON operations; serialized Python objects and arbitrary expressions are forbidden.
|
|
|
+- Generated Python remains disabled until signed-artifact, dependency-manifest, and sandbox acceptance is separately complete.
|
|
|
+- Every task follows red-green-refactor TDD and preserves unrelated changes in the existing dirty worktree.
|
|
|
+
|
|
|
+---
|
|
|
+
|
|
|
+## 1. Verified baseline and execution-readiness gap
|
|
|
+
|
|
|
+### 1.1 Already implemented
|
|
|
+
|
|
|
+- Closed RuleSpec, StandardSpec, and DataFlowSpec validation with canonical hashes.
|
|
|
+- Natural-language authoring endpoint with schema-constrained candidates, confidence/ambiguity handling, and generation-run audit rows.
|
|
|
+- Immutable RuleVersion and StandardVersion creation/publication with separate permissions.
|
|
|
+- Server-side DataFlow release that expands standards, fixes version references, persists component bindings/plans, and creates an immutable ProductionLinePackage.
|
|
|
+- Runner-side plan lookup by component binding, rule version, and plan hash.
|
|
|
+- Kestra WorkflowSpec compiler and Kestra engine adapter.
|
|
|
+- Data Standard and Data Flow authoring surfaces plus an explicit Data Factory activation gate.
|
|
|
+- PostgreSQL control-plane integration test, full Python suite, frontend build, and local Docker deployment.
|
|
|
+
|
|
|
+### 1.2 Blocking gaps
|
|
|
+
|
|
|
+| Priority | Gap | Current evidence | Consequence |
|
|
|
+|---|---|---|---|
|
|
|
+| P0 | Released plan and Runner adapter contracts do not match | `compiler.py` emits RuleSpec-shaped `polars_batch` or `quality_check`; `SqlRulePlanAdapter` accepts only `statement`, `parameters`, and `data_source_uid`; Runner registers no `polars_batch` adapter | A released production line cannot execute its generated plans |
|
|
|
+| P0 | No physical dataset binding | DataFlow release accepts schema references/hashes but no source/target `data_source_uid`, table/view/query, artifact, dialect, or write mode | Runtime does not know which data to read or where to write |
|
|
|
+| P0 | Schema hashes are client-submitted | Release API receives `input_schema_hashes` and `output_schema_hash` directly | A caller can claim an unverified schema and release an incompatible plan |
|
|
|
+| P0 | Expressions are opaque strings | `assert`, `derive`, and `filter` expressions are length-checked but not parsed, typed, or function-allowlisted | Cross-backend semantics and injection safety are not established |
|
|
|
+| P0 | Plans are marked `published` before compile/dry-run evidence | `persist_component_plan` inserts plan status `published` during DataFlow release | Publication overstates executability |
|
|
|
+| P1 | No real rule I/O contract between nodes | Kestra sends parameters and task tokens; Runner returns JSON, but production-line nodes do not pass dataset/artifact references downstream | Multi-step transformation cannot transport data safely |
|
|
|
+| P1 | Rule run evidence tables are unused | `rule_runs`, `rule_violation_samples`, and `rule_artifacts` exist only in migration/tests | No row counts, reject/quarantine proof, sample retention, or replay evidence |
|
|
|
+| P1 | AI generation is not linked to the created version | `/interpret` creates an audit row with no RuleVersion; `/rule-versions` does not accept a signed generation receipt | Natural-language intent-to-runtime lineage is incomplete |
|
|
|
+| P1 | No deterministic test/repair/publication gate | No generated test cases, sample dry-run, compatibility matrix, bounded repair loop, or risk policy is called before publish | Human publication can approve syntactically valid but semantically invalid rules |
|
|
|
+| P1 | Data Factory deployment lifecycle is only a table/UI gate | No service/API for draft, disabled deploy, canary, activate, supersede, or rollback | Released lines cannot be safely put into production |
|
|
|
+| P1 | Local Kestra is stopped | Compose reports `dataops-test-kestra-1` exited after PostgreSQL broken-pipe/closed-connection failures | Local end-to-end scheduling acceptance is unavailable |
|
|
|
+| P2 | UI still straddles legacy implementations | Standard retains legacy generated-code field; Data Flow retains free-text `rule`/`rule_spec`; no published asset catalog/assembly UI | Users can create parallel rule representations and cannot reliably assemble versions |
|
|
|
+| P2 | RBAC lacks execution/deployment separation | Rules have read/edit/publish and DataFlow has release; Data Factory deploy/canary/activate/rollback are not independently classified | Operational duties cannot be separated cleanly |
|
|
|
+| P2 | No runtime quality/performance acceptance | Tests stop at control-plane persistence and fake adapters | No evidence for real rows, large batches, restart recovery, tamper rejection, or rollback |
|
|
|
+
|
|
|
+### 1.3 Definition of “complete adaptable execution”
|
|
|
+
|
|
|
+A rule is considered execution-ready only when one acceptance scenario proves all of the following:
|
|
|
+
|
|
|
+1. Natural language produces a RuleCandidate and an audit record.
|
|
|
+2. The candidate is linked to an immutable RuleVersion.
|
|
|
+3. Expressions and operations are parsed, typed against a server-owned SchemaSnapshot, and checked against a backend capability matrix.
|
|
|
+4. The compiler emits a deterministic backend plan and golden tests.
|
|
|
+5. Sample/dry-run evidence passes before RuleVersion and plan publication.
|
|
|
+6. A DataFlowVersion fixes StandardVersion, RuleVersion, SchemaSnapshot, and plan hashes.
|
|
|
+7. Data Factory binds physical source/target datasets, environment, schedule, and resources without changing rule semantics.
|
|
|
+8. Kestra invokes Runner using only immutable IDs, hashes, artifact references, and short-lived task tokens.
|
|
|
+9. Runner executes real data, applies reject/quarantine policy, writes idempotently, and records rule-run evidence.
|
|
|
+10. Canary, activation, monitoring, and rollback succeed, including process/container restart recovery.
|
|
|
+
|
|
|
+---
|
|
|
+
|
|
|
+## 2. Delivery milestones
|
|
|
+
|
|
|
+| Milestone | Scope | Exit criterion | Suggested duration |
|
|
|
+|---|---|---|---|
|
|
|
+| M3A — SQL executable vertical slice | Tasks 1–4 | One PostgreSQL/MySQL rule executes real rows through released plan and records evidence | 2–3 weeks |
|
|
|
+| M3B — Polars and multi-step artifacts | Tasks 5–6 | Cross-source production line passes Parquet artifact references through multiple rule nodes | 2–3 weeks |
|
|
|
+| M4 — governed publication and product UX | Tasks 7–8 | Users select published assets, see compile/dry-run evidence, and legacy duplicate paths are read-only | 1–2 weeks |
|
|
|
+| M5 — Data Factory rollout | Tasks 9–10 | Disabled deploy, canary, activation, rollback, restart, and tamper tests pass in Docker | 2–3 weeks |
|
|
|
+
|
|
|
+For one full-time engineer, allow approximately 10–14 weeks. With two backend engineers, one frontend engineer, and shared QA/DevOps, the four milestones can reasonably fit 7–10 weeks because M3B UI work and M5 infrastructure work can overlap after M3A contracts freeze.
|
|
|
+
|
|
|
+---
|
|
|
+
|
|
|
+## 3. File map
|
|
|
+
|
|
|
+- `app/core/data_rules/execution_contracts.py`: dataset bindings, schema snapshots, execution-plan V2, rule results, and backend capability contracts.
|
|
|
+- `app/core/data_rules/schema_resolver.py`: resolve server-owned schema and physical dataset metadata.
|
|
|
+- `app/core/data_rules/expressions.py`: canonical expression AST, type checker, function allowlist, and backend capability checks.
|
|
|
+- `app/core/data_rules/compilers/base.py`: compiler protocol and deterministic registry.
|
|
|
+- `app/core/data_rules/compilers/sql.py`: strict SQLGlot compiler for PostgreSQL/MySQL.
|
|
|
+- `app/core/data_rules/compilers/polars.py`: closed RuleSpec-to-Polars-plan compiler.
|
|
|
+- `app/core/data_rules/publication.py`: compile, tests, dry-run evidence, risk decision, and publication state machine.
|
|
|
+- `app/core/data_rules/deployment.py`: DataFlowDeployment create/deploy/canary/activate/rollback service.
|
|
|
+- `app/core/data_rules/repository.py`: immutable asset, plan, deployment, and run-evidence persistence.
|
|
|
+- `app/runner/rule_sql.py`: SQL-pushdown execution adapter.
|
|
|
+- `app/runner/rule_polars.py`: Polars Lazy execution adapter.
|
|
|
+- `app/runner/artifacts.py`: bounded Parquet artifact read/write using MinIO references.
|
|
|
+- `app/runner/rule_evidence.py`: rule run metrics, violation samples, and redaction/retention writes.
|
|
|
+- `app/core/orchestration/compilers/kestra.py`: dataset/artifact handoff in Runner task payloads.
|
|
|
+- `app/api/data_rules/routes.py`: catalogs, publication evidence, release, deployment, canary, activation, and rollback APIs.
|
|
|
+- `migrations/versions/20260723_120_rule_execution_runtime.py`: schema snapshots, dataset bindings, compile/test evidence, audit linkage, and deployment transitions.
|
|
|
+- `frontend/src/views/dataGovernance/dataStandard/`: standard clauses, linked RuleVersions, and evidence.
|
|
|
+- `frontend/src/views/dataGovernance/dataProcess/`: production-line assembly from published catalogs.
|
|
|
+- `frontend/src/views/dataFactory/workflow/ProductionLineDeployment.vue`: deployment/canary/activation/rollback operations.
|
|
|
+- `tests/core/data_rules/`: domain, compiler, publication, deployment, and repository tests.
|
|
|
+- `tests/runner/`: SQL/Polars/artifact/evidence adapter tests.
|
|
|
+- `tests/integration/test_data_rule_sql_execution.py`: real PostgreSQL/MySQL SQL path.
|
|
|
+- `tests/integration/test_data_rule_polars_execution.py`: cross-source artifact path.
|
|
|
+- `tests/integration/test_data_rule_factory_lifecycle.py`: Kestra/Runner canary and rollback.
|
|
|
+
|
|
|
+---
|
|
|
+
|
|
|
+### Task 1: Freeze execution, schema, and dataset contracts
|
|
|
+
|
|
|
+**Files:**
|
|
|
+- Create: `app/core/data_rules/execution_contracts.py`
|
|
|
+- Create: `tests/core/data_rules/test_execution_contracts.py`
|
|
|
+- Create: `migrations/versions/20260723_120_rule_execution_runtime.py`
|
|
|
+- Modify: `tests/test_data_rule_schema.py`
|
|
|
+
|
|
|
+**Interfaces:**
|
|
|
+- Produces: `validate_schema_snapshot(value) -> dict`
|
|
|
+- Produces: `validate_dataset_binding(value) -> dict`
|
|
|
+- Produces: `validate_execution_plan_v2(value) -> dict`
|
|
|
+- Produces: `execution_plan_hash(value) -> str`
|
|
|
+- Produces: PostgreSQL tables `data_schema_snapshots`, `dataflow_dataset_bindings`, `rule_compile_evidence`, and `rule_test_evidence`
|
|
|
+
|
|
|
+- [ ] **Step 1: Write failing contract tests**
|
|
|
+
|
|
|
+```python
|
|
|
+def test_execution_plan_requires_fixed_schema_dataset_and_backend():
|
|
|
+ plan = validate_execution_plan_v2(
|
|
|
+ {
|
|
|
+ "schema_version": "2.0",
|
|
|
+ "backend": "sql_pushdown",
|
|
|
+ "compiler_version": "dataops-sqlglot-1.0",
|
|
|
+ "rule_version_id": new_governance_uid(),
|
|
|
+ "input_schema_snapshot_id": new_governance_uid(),
|
|
|
+ "output_schema_snapshot_id": new_governance_uid(),
|
|
|
+ "input_binding_id": new_governance_uid(),
|
|
|
+ "output_binding_id": new_governance_uid(),
|
|
|
+ "operations": [{"kind": "sql", "statement": "SELECT 1"}],
|
|
|
+ }
|
|
|
+ )
|
|
|
+ assert plan["backend"] == "sql_pushdown"
|
|
|
+ assert execution_plan_hash(plan) == execution_plan_hash(dict(reversed(list(plan.items()))))
|
|
|
+
|
|
|
+
|
|
|
+def test_dataset_binding_rejects_credentials_and_unversioned_objects():
|
|
|
+ with pytest.raises(ValueError, match="secret|credential"):
|
|
|
+ validate_dataset_binding(
|
|
|
+ {
|
|
|
+ "data_source_uid": new_governance_uid(),
|
|
|
+ "object_kind": "table",
|
|
|
+ "object_ref": "public.customer",
|
|
|
+ "schema_snapshot_id": new_governance_uid(),
|
|
|
+ "access_mode": "read",
|
|
|
+ "password": "unsafe",
|
|
|
+ }
|
|
|
+ )
|
|
|
+```
|
|
|
+
|
|
|
+- [ ] **Step 2: Run tests and verify the missing-module failure**
|
|
|
+
|
|
|
+Run: `PYTHONPATH=. .venv/bin/pytest -q tests/core/data_rules/test_execution_contracts.py`
|
|
|
+
|
|
|
+Expected: FAIL because `app.core.data_rules.execution_contracts` does not exist.
|
|
|
+
|
|
|
+- [ ] **Step 3: Implement closed contracts**
|
|
|
+
|
|
|
+Use exact enums:
|
|
|
+
|
|
|
+```python
|
|
|
+BACKENDS = {"sql_pushdown", "polars_batch", "quality_check"}
|
|
|
+OBJECT_KINDS = {"table", "view", "query", "parquet_artifact"}
|
|
|
+ACCESS_MODES = {"read", "write", "read_write"}
|
|
|
+PLAN_STATUSES = {"compiled", "tested", "published", "revoked"}
|
|
|
+```
|
|
|
+
|
|
|
+`DatasetBinding` must require `data_source_uid`, `object_kind`, `object_ref`, `schema_snapshot_id`, `access_mode`, `dialect`, and `write_mode`. `ExecutionPlanV2` must reference snapshot/binding IDs rather than embed credentials or connection strings. Canonical hashing must use sorted JSON with compact separators and UTF-8.
|
|
|
+
|
|
|
+- [ ] **Step 4: Add forward-only migration**
|
|
|
+
|
|
|
+Create:
|
|
|
+
|
|
|
+```sql
|
|
|
+CREATE TABLE public.data_schema_snapshots (
|
|
|
+ id UUID PRIMARY KEY,
|
|
|
+ schema_ref VARCHAR(500) NOT NULL,
|
|
|
+ schema_hash CHAR(64) NOT NULL,
|
|
|
+ fields JSONB NOT NULL,
|
|
|
+ source_revision VARCHAR(200) NOT NULL,
|
|
|
+ created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
|
|
+ UNIQUE (schema_ref, schema_hash)
|
|
|
+);
|
|
|
+
|
|
|
+CREATE TABLE public.dataflow_dataset_bindings (
|
|
|
+ id UUID PRIMARY KEY,
|
|
|
+ dataflow_deployment_id UUID NOT NULL REFERENCES public.dataflow_deployments(id),
|
|
|
+ logical_ref VARCHAR(500) NOT NULL,
|
|
|
+ data_source_uid UUID,
|
|
|
+ object_kind VARCHAR(30) NOT NULL,
|
|
|
+ object_ref VARCHAR(1000) NOT NULL,
|
|
|
+ schema_snapshot_id UUID NOT NULL REFERENCES public.data_schema_snapshots(id),
|
|
|
+ dialect VARCHAR(30) NOT NULL,
|
|
|
+ access_mode VARCHAR(20) NOT NULL,
|
|
|
+ write_mode VARCHAR(30),
|
|
|
+ binding_hash CHAR(64) NOT NULL,
|
|
|
+ UNIQUE (dataflow_deployment_id, logical_ref)
|
|
|
+);
|
|
|
+```
|
|
|
+
|
|
|
+Add compile/test evidence tables keyed by `rule_execution_plan_id`, and add `created_by` plus `candidate` to `rule_generation_runs` so a signed generation receipt can be bound to its RuleVersion.
|
|
|
+
|
|
|
+- [ ] **Step 5: Verify contracts and migration**
|
|
|
+
|
|
|
+Run: `PYTHONPATH=. .venv/bin/pytest -q tests/core/data_rules/test_execution_contracts.py tests/test_data_rule_schema.py`
|
|
|
+
|
|
|
+Expected: PASS.
|
|
|
+
|
|
|
+- [ ] **Step 6: Commit**
|
|
|
+
|
|
|
+```bash
|
|
|
+git add app/core/data_rules/execution_contracts.py migrations/versions/20260723_120_rule_execution_runtime.py tests/core/data_rules/test_execution_contracts.py tests/test_data_rule_schema.py
|
|
|
+git commit -m "feat: define governed rule execution contracts"
|
|
|
+```
|
|
|
+
|
|
|
+---
|
|
|
+
|
|
|
+### Task 2: Resolve trusted schemas and physical dataset bindings
|
|
|
+
|
|
|
+**Files:**
|
|
|
+- Create: `app/core/data_rules/schema_resolver.py`
|
|
|
+- Create: `tests/core/data_rules/test_schema_resolver.py`
|
|
|
+- Modify: `app/core/data_rules/repository.py`
|
|
|
+- Modify: `app/api/data_rules/routes.py`
|
|
|
+- Modify: `tests/test_data_rule_api.py`
|
|
|
+
|
|
|
+**Interfaces:**
|
|
|
+- Consumes: `validate_schema_snapshot`, `validate_dataset_binding`
|
|
|
+- Produces: `SchemaResolver.resolve(schema_ref: str) -> dict`
|
|
|
+- Produces: `DatasetBindingService.bind(deployment_id: str, bindings: list[dict], actor_uid: str) -> list[dict]`
|
|
|
+- Changes: production-line release no longer accepts client-authored schema hashes
|
|
|
+
|
|
|
+- [ ] **Step 1: Write failing resolver tests**
|
|
|
+
|
|
|
+```python
|
|
|
+def test_schema_resolver_hashes_server_metadata_not_request_values():
|
|
|
+ metadata = FakeMetadataCatalog(
|
|
|
+ {
|
|
|
+ "bd:customer:v7": {
|
|
|
+ "source_revision": "neo4j:42",
|
|
|
+ "fields": [
|
|
|
+ {"name": "customer_id", "type": "string", "nullable": False},
|
|
|
+ {"name": "mobile", "type": "string", "nullable": True},
|
|
|
+ ],
|
|
|
+ }
|
|
|
+ }
|
|
|
+ )
|
|
|
+ snapshot = SchemaResolver(metadata, Repository()).resolve("bd:customer:v7")
|
|
|
+ assert snapshot["schema_hash"] == canonical_schema_hash(snapshot["fields"])
|
|
|
+ assert snapshot["source_revision"] == "neo4j:42"
|
|
|
+```
|
|
|
+
|
|
|
+Add API coverage proving `input_schema_hashes` and `output_schema_hash` are rejected as unsupported release fields.
|
|
|
+
|
|
|
+- [ ] **Step 2: Run tests and verify failure**
|
|
|
+
|
|
|
+Run: `PYTHONPATH=. .venv/bin/pytest -q tests/core/data_rules/test_schema_resolver.py tests/test_data_rule_api.py -k schema`
|
|
|
+
|
|
|
+Expected: FAIL because the resolver and new release contract are absent.
|
|
|
+
|
|
|
+- [ ] **Step 3: Implement server-owned resolution**
|
|
|
+
|
|
|
+`SchemaResolver` must:
|
|
|
+
|
|
|
+1. Load the schema by stable `schema_ref`.
|
|
|
+2. Normalize fields by name, type, nullability, precision, scale, and timezone.
|
|
|
+3. Hash only normalized fields.
|
|
|
+4. Persist/reuse an immutable snapshot.
|
|
|
+5. Reject missing, duplicate, or unsupported field types.
|
|
|
+
|
|
|
+`DatasetBindingService` must resolve `data_source_uid` through the existing datasource-definition repository and never accept credentials. Verify requested tables/views with a read-only metadata query using the existing pool manager.
|
|
|
+
|
|
|
+- [ ] **Step 4: Remove client schema hashes from release**
|
|
|
+
|
|
|
+Change:
|
|
|
+
|
|
|
+```python
|
|
|
+result = _release_service().release(
|
|
|
+ dataflow_uid=dataflow_uid,
|
|
|
+ dataflow_spec=body["dataflow_spec"],
|
|
|
+ source_text=body["source_text"],
|
|
|
+ created_by=g.current_user["id"],
|
|
|
+)
|
|
|
+```
|
|
|
+
|
|
|
+The release service obtains all snapshots from `SchemaResolver`; dataset bindings remain deployment-time objects.
|
|
|
+
|
|
|
+- [ ] **Step 5: Verify tests**
|
|
|
+
|
|
|
+Run: `PYTHONPATH=. .venv/bin/pytest -q tests/core/data_rules/test_schema_resolver.py tests/test_data_rule_api.py tests/integration/test_data_rule_control_plane.py`
|
|
|
+
|
|
|
+Expected: PASS.
|
|
|
+
|
|
|
+- [ ] **Step 6: Commit**
|
|
|
+
|
|
|
+```bash
|
|
|
+git add app/core/data_rules/schema_resolver.py app/core/data_rules/repository.py app/api/data_rules/routes.py tests/core/data_rules/test_schema_resolver.py tests/test_data_rule_api.py tests/integration/test_data_rule_control_plane.py
|
|
|
+git commit -m "feat: resolve trusted rule schemas and datasets"
|
|
|
+```
|
|
|
+
|
|
|
+---
|
|
|
+
|
|
|
+### Task 3: Parse and type-check canonical rule expressions
|
|
|
+
|
|
|
+**Files:**
|
|
|
+- Create: `app/core/data_rules/expressions.py`
|
|
|
+- Create: `tests/core/data_rules/test_expressions.py`
|
|
|
+- Modify: `app/core/data_rules/contracts.py`
|
|
|
+- Modify: `tests/core/data_rules/test_contracts.py`
|
|
|
+
|
|
|
+**Interfaces:**
|
|
|
+- Produces: `parse_expression(source: str) -> dict`
|
|
|
+- Produces: `type_check_expression(ast: dict, fields: dict[str, str]) -> str`
|
|
|
+- Produces: `backend_support(ast: dict) -> frozenset[str]`
|
|
|
+- Expression functions allowed in V1: `matches`, `lower`, `upper`, `trim`, `length`, `coalesce`, `date`, `timestamp`, `abs`, `round`
|
|
|
+
|
|
|
+- [ ] **Step 1: Write failing expression tests**
|
|
|
+
|
|
|
+```python
|
|
|
+def test_expression_is_parsed_typed_and_backend_capable():
|
|
|
+ ast = parse_expression("matches(mobile, '^[0-9]{11}$') && customer_id != ''")
|
|
|
+ result_type = type_check_expression(
|
|
|
+ ast,
|
|
|
+ {"mobile": "string", "customer_id": "string"},
|
|
|
+ )
|
|
|
+ assert result_type == "boolean"
|
|
|
+ assert backend_support(ast) == frozenset(
|
|
|
+ {"postgresql", "mysql", "polars"}
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.parametrize(
|
|
|
+ "source",
|
|
|
+ [
|
|
|
+ "__import__('os')",
|
|
|
+ "http_get('https://example.com')",
|
|
|
+ "unknown_column == 1",
|
|
|
+ ],
|
|
|
+)
|
|
|
+def test_expression_rejects_code_network_and_unknown_fields(source):
|
|
|
+ with pytest.raises(ValueError):
|
|
|
+ type_check_expression(
|
|
|
+ parse_expression(source),
|
|
|
+ {"mobile": "string"},
|
|
|
+ )
|
|
|
+```
|
|
|
+
|
|
|
+- [ ] **Step 2: Run tests and verify failure**
|
|
|
+
|
|
|
+Run: `PYTHONPATH=. .venv/bin/pytest -q tests/core/data_rules/test_expressions.py`
|
|
|
+
|
|
|
+Expected: FAIL because the expression parser does not exist.
|
|
|
+
|
|
|
+- [ ] **Step 3: Implement the closed V1 expression grammar**
|
|
|
+
|
|
|
+Implement a DataOps-owned lexer and Pratt parser in `expressions.py`; do not execute Python, SQL, JavaScript, or model-produced code. The V1 grammar supports literals, schema-bound identifiers, parentheses, function calls from the fixed allowlist, unary `!`/`-`, comparisons, arithmetic, `&&`, and `||`. The canonical AST is closed JSON with explicit node kinds so SQL and Polars compilers never depend on runtime parser objects.
|
|
|
+
|
|
|
+- [ ] **Step 4: Implement parser, allowlist, and type checker**
|
|
|
+
|
|
|
+Reject:
|
|
|
+
|
|
|
+- unknown identifiers;
|
|
|
+- unknown functions;
|
|
|
+- dynamic field access;
|
|
|
+- network/filesystem/environment functions;
|
|
|
+- mixed numeric/string comparisons without explicit cast;
|
|
|
+- regex patterns exceeding 500 characters;
|
|
|
+- AST depth over 40 and node count over 500.
|
|
|
+
|
|
|
+Update RuleSpec normalization so `expression` is stored as canonical AST plus original display text:
|
|
|
+
|
|
|
+```python
|
|
|
+step["expression_ast"] = parse_expression(step.pop("expression"))
|
|
|
+```
|
|
|
+
|
|
|
+Increment RuleSpec schema version for newly authored rules while retaining a read-only V1 migration parser.
|
|
|
+
|
|
|
+- [ ] **Step 5: Add cross-backend golden semantics**
|
|
|
+
|
|
|
+Cover null, timezone, regex, rounding, Unicode case, and date conversion with the same input/output fixtures for PostgreSQL, MySQL, and Polars.
|
|
|
+
|
|
|
+- [ ] **Step 6: Verify tests**
|
|
|
+
|
|
|
+Run: `PYTHONPATH=. .venv/bin/pytest -q tests/core/data_rules/test_expressions.py tests/core/data_rules/test_contracts.py`
|
|
|
+
|
|
|
+Expected: PASS.
|
|
|
+
|
|
|
+- [ ] **Step 7: Commit**
|
|
|
+
|
|
|
+```bash
|
|
|
+git add app/core/data_rules/expressions.py app/core/data_rules/contracts.py tests/core/data_rules/test_expressions.py tests/core/data_rules/test_contracts.py
|
|
|
+git commit -m "feat: add typed rule expression language"
|
|
|
+```
|
|
|
+
|
|
|
+---
|
|
|
+
|
|
|
+### Task 4: Deliver the SQLGlot executable vertical slice
|
|
|
+
|
|
|
+**Files:**
|
|
|
+- Create: `app/core/data_rules/compilers/__init__.py`
|
|
|
+- Create: `app/core/data_rules/compilers/base.py`
|
|
|
+- Create: `app/core/data_rules/compilers/sql.py`
|
|
|
+- Create: `app/runner/rule_sql.py`
|
|
|
+- Create: `tests/core/data_rules/test_sql_compiler.py`
|
|
|
+- Create: `tests/runner/test_rule_sql.py`
|
|
|
+- Create: `tests/integration/test_data_rule_sql_execution.py`
|
|
|
+- Modify: `app/runner/bootstrap.py`
|
|
|
+- Modify: `app/core/data_rules/release.py`
|
|
|
+
|
|
|
+**Interfaces:**
|
|
|
+- Produces: `RuleCompiler.compile(rule_version, schema, binding, backend) -> dict`
|
|
|
+- Produces: `CompilerRegistry.select(rule_spec, input_binding, output_binding) -> RuleCompiler`
|
|
|
+- Produces: `SqlGlotRulePlanAdapter.execute(plan, node, parameters, write_authorized) -> dict`
|
|
|
+
|
|
|
+- [ ] **Step 1: Write failing compiler tests**
|
|
|
+
|
|
|
+```python
|
|
|
+def test_postgres_compiler_emits_one_strict_parameterized_statement():
|
|
|
+ compiled = SqlGlotRuleCompiler("postgresql").compile(
|
|
|
+ rule_version=published_rule_version(),
|
|
|
+ input_schema=customer_schema(),
|
|
|
+ input_binding=table_binding("raw.customer"),
|
|
|
+ output_binding=table_binding("clean.customer"),
|
|
|
+ )
|
|
|
+ assert compiled["backend"] == "sql_pushdown"
|
|
|
+ assert compiled["plan"]["dialect"] == "postgresql"
|
|
|
+ assert compiled["plan"]["statements"][0]["purpose"] == "write"
|
|
|
+ assert ";" not in compiled["plan"]["statements"][0]["sql"]
|
|
|
+ assert compiled["plan_hash"] == execution_plan_hash(compiled["plan"])
|
|
|
+```
|
|
|
+
|
|
|
+Add tests for PostgreSQL/MySQL quoting, unsupported operator failure, binding mismatch, and hash determinism.
|
|
|
+
|
|
|
+- [ ] **Step 2: Run tests and verify failure**
|
|
|
+
|
|
|
+Run: `PYTHONPATH=. .venv/bin/pytest -q tests/core/data_rules/test_sql_compiler.py tests/runner/test_rule_sql.py`
|
|
|
+
|
|
|
+Expected: FAIL because compiler and adapter are absent.
|
|
|
+
|
|
|
+- [ ] **Step 3: Implement compiler registry and strict SQL compiler**
|
|
|
+
|
|
|
+Support the first executable set:
|
|
|
+
|
|
|
+- `cast`
|
|
|
+- `normalize_text`
|
|
|
+- `regex_replace`
|
|
|
+- `fill_null`
|
|
|
+- `filter`
|
|
|
+- `derive`
|
|
|
+- `assert`
|
|
|
+- `deduplicate`
|
|
|
+- `map_values`
|
|
|
+
|
|
|
+Use SQLGlot AST construction rather than SQL string concatenation. Bind runtime values as parameters. Fail compilation if the source/target dialect cannot preserve semantics or if the rule needs cross-source data.
|
|
|
+
|
|
|
+- [ ] **Step 4: Implement Runner SQL adapter**
|
|
|
+
|
|
|
+The plan schema is:
|
|
|
+
|
|
|
+```python
|
|
|
+{
|
|
|
+ "dialect": "postgresql",
|
|
|
+ "data_source_uid": "...",
|
|
|
+ "statements": [
|
|
|
+ {
|
|
|
+ "purpose": "write",
|
|
|
+ "sql": "INSERT INTO ... SELECT ...",
|
|
|
+ "parameters": {},
|
|
|
+ }
|
|
|
+ ],
|
|
|
+ "result_contract": {
|
|
|
+ "rows_in": "counted",
|
|
|
+ "rows_out": "counted",
|
|
|
+ "rows_rejected": "counted",
|
|
|
+ },
|
|
|
+}
|
|
|
+```
|
|
|
+
|
|
|
+The adapter must verify plan dialect against datasource definition, execute inside one transaction, enforce idempotency, and return row metrics plus commit outcome.
|
|
|
+
|
|
|
+- [ ] **Step 5: Register only matching adapters**
|
|
|
+
|
|
|
+Replace the current `quality_check -> SqlRulePlanAdapter` alias with explicit SQL and quality adapters. Release must fail if `CompilerRegistry` cannot find a registered adapter for every component.
|
|
|
+
|
|
|
+- [ ] **Step 6: Add real PostgreSQL/MySQL integration test**
|
|
|
+
|
|
|
+The test creates source/target tables, inserts good/bad customer rows, releases a fixed rule, executes through `RulePlanExecutor`, and asserts transformed target rows plus rejected counts. It also modifies the stored plan body without updating the hash and asserts fail-closed execution.
|
|
|
+
|
|
|
+- [ ] **Step 7: Verify vertical slice**
|
|
|
+
|
|
|
+Run:
|
|
|
+
|
|
|
+```bash
|
|
|
+TEST_DATABASE_URL=postgresql://dataops:dataops-test-password@127.0.0.1:15432/dataops \
|
|
|
+PYTHONPATH=. .venv/bin/pytest -q \
|
|
|
+tests/core/data_rules/test_sql_compiler.py \
|
|
|
+tests/runner/test_rule_sql.py \
|
|
|
+tests/integration/test_data_rule_sql_execution.py
|
|
|
+```
|
|
|
+
|
|
|
+Expected: PASS for PostgreSQL and MySQL source containers.
|
|
|
+
|
|
|
+- [ ] **Step 8: Commit**
|
|
|
+
|
|
|
+```bash
|
|
|
+git add app/core/data_rules/compilers app/runner/rule_sql.py app/runner/bootstrap.py app/core/data_rules/release.py tests/core/data_rules/test_sql_compiler.py tests/runner/test_rule_sql.py tests/integration/test_data_rule_sql_execution.py
|
|
|
+git commit -m "feat: execute governed SQL rule plans"
|
|
|
+```
|
|
|
+
|
|
|
+---
|
|
|
+
|
|
|
+### Task 5: Add Polars cross-source compilation and artifact I/O
|
|
|
+
|
|
|
+**Files:**
|
|
|
+- Create: `app/core/data_rules/compilers/polars.py`
|
|
|
+- Create: `app/runner/rule_polars.py`
|
|
|
+- Create: `app/runner/artifacts.py`
|
|
|
+- Create: `tests/core/data_rules/test_polars_compiler.py`
|
|
|
+- Create: `tests/runner/test_rule_polars.py`
|
|
|
+- Create: `tests/runner/test_artifacts.py`
|
|
|
+- Create: `tests/integration/test_data_rule_polars_execution.py`
|
|
|
+- Modify: `requirements.txt`
|
|
|
+- Modify: `app/runner/bootstrap.py`
|
|
|
+
|
|
|
+**Interfaces:**
|
|
|
+- Produces: `PolarsRuleCompiler.compile(...) -> dict`
|
|
|
+- Produces: `PolarsRulePlanAdapter.execute(...) -> dict`
|
|
|
+- Produces: `ArtifactStore.read(ref, expected_digest) -> LazyFrame`
|
|
|
+- Produces: `ArtifactStore.write(frame, correlation_id, ttl_seconds) -> dict`
|
|
|
+
|
|
|
+- [ ] **Step 1: Write failing Polars compiler tests**
|
|
|
+
|
|
|
+```python
|
|
|
+def test_polars_compiler_emits_closed_lazy_operations():
|
|
|
+ result = PolarsRuleCompiler().compile(
|
|
|
+ rule_version=published_rule_version(),
|
|
|
+ input_schema=customer_schema(),
|
|
|
+ input_binding=artifact_binding("input.parquet"),
|
|
|
+ output_binding=artifact_binding("output.parquet"),
|
|
|
+ )
|
|
|
+ assert result["backend"] == "polars_batch"
|
|
|
+ assert result["plan"]["operations"][0]["op"] == "normalize_text"
|
|
|
+ assert "pickle" not in json.dumps(result)
|
|
|
+ assert "python" not in json.dumps(result)
|
|
|
+```
|
|
|
+
|
|
|
+- [ ] **Step 2: Run tests and verify failure**
|
|
|
+
|
|
|
+Run: `PYTHONPATH=. .venv/bin/pytest -q tests/core/data_rules/test_polars_compiler.py tests/runner/test_rule_polars.py tests/runner/test_artifacts.py`
|
|
|
+
|
|
|
+Expected: FAIL because compiler/adapter/store are absent.
|
|
|
+
|
|
|
+- [ ] **Step 3: Add pinned Polars dependency**
|
|
|
+
|
|
|
+Pin a tested Polars release and document its license/SBOM entry. Keep the plan JSON independent of Polars internal serialization formats.
|
|
|
+
|
|
|
+- [ ] **Step 4: Implement closed LazyFrame reconstruction**
|
|
|
+
|
|
|
+Map each operation dict to allowlisted `pl.Expr` constructors. Support all SQL vertical-slice operations, then add:
|
|
|
+
|
|
|
+- `aggregate`
|
|
|
+- `lookup_join`
|
|
|
+- `mask` through pre-registered masking policies
|
|
|
+
|
|
|
+Reject any plan field that could reference modules, callables, file paths, or URLs.
|
|
|
+
|
|
|
+- [ ] **Step 5: Implement digest-bound Parquet artifacts**
|
|
|
+
|
|
|
+Store artifacts under a generated server path such as `rules/<correlation_id>/<artifact_id>.parquet`. Return only:
|
|
|
+
|
|
|
+```python
|
|
|
+{
|
|
|
+ "artifact_ref": "minio://dataops-rules/rules/...parquet",
|
|
|
+ "digest": "sha256...",
|
|
|
+ "row_count": 100,
|
|
|
+ "schema_hash": "sha256...",
|
|
|
+ "expires_at": "2026-07-24T00:00:00Z",
|
|
|
+}
|
|
|
+```
|
|
|
+
|
|
|
+Never accept a client-selected bucket/key. Verify digest and schema before reading.
|
|
|
+
|
|
|
+- [ ] **Step 6: Add cross-source integration**
|
|
|
+
|
|
|
+Read customers from PostgreSQL and reference data from MySQL, materialize bounded Parquet inputs, run normalize/join/assert/deduplicate, and verify the output artifact and violation counts.
|
|
|
+
|
|
|
+- [ ] **Step 7: Verify tests**
|
|
|
+
|
|
|
+Run: `PYTHONPATH=. .venv/bin/pytest -q tests/core/data_rules/test_polars_compiler.py tests/runner/test_rule_polars.py tests/runner/test_artifacts.py tests/integration/test_data_rule_polars_execution.py`
|
|
|
+
|
|
|
+Expected: PASS.
|
|
|
+
|
|
|
+- [ ] **Step 8: Commit**
|
|
|
+
|
|
|
+```bash
|
|
|
+git add requirements.txt app/core/data_rules/compilers/polars.py app/runner/rule_polars.py app/runner/artifacts.py app/runner/bootstrap.py tests/core/data_rules/test_polars_compiler.py tests/runner/test_rule_polars.py tests/runner/test_artifacts.py tests/integration/test_data_rule_polars_execution.py
|
|
|
+git commit -m "feat: execute cross-source Polars rule plans"
|
|
|
+```
|
|
|
+
|
|
|
+---
|
|
|
+
|
|
|
+### Task 6: Pass artifacts between nodes and persist rule evidence
|
|
|
+
|
|
|
+**Files:**
|
|
|
+- Create: `app/runner/rule_evidence.py`
|
|
|
+- Create: `tests/runner/test_rule_evidence.py`
|
|
|
+- Modify: `app/runner/rules.py`
|
|
|
+- Modify: `app/runner/api.py`
|
|
|
+- Modify: `app/core/orchestration/compilers/kestra.py`
|
|
|
+- Modify: `app/core/data_rules/repository.py`
|
|
|
+- Modify: `tests/core/orchestration/test_kestra_compiler.py`
|
|
|
+- Modify: `tests/runner/test_rules.py`
|
|
|
+
|
|
|
+**Interfaces:**
|
|
|
+- Produces: `RuleEvidenceWriter.start(...) -> rule_run_id`
|
|
|
+- Produces: `RuleEvidenceWriter.finish(rule_run_id, result) -> None`
|
|
|
+- Runner result includes `output_artifact`, `rows_in`, `rows_out`, `rows_rejected`, `rows_quarantined`, and `commit_outcome`
|
|
|
+
|
|
|
+- [ ] **Step 1: Write failing evidence and handoff tests**
|
|
|
+
|
|
|
+```python
|
|
|
+def test_rule_executor_records_success_and_bounded_violation_sample():
|
|
|
+ result = executor.execute(rule_node(), {"input_artifact": input_ref})
|
|
|
+ assert result["rows_in"] == 3
|
|
|
+ assert result["rows_out"] == 2
|
|
|
+ assert result["rows_quarantined"] == 1
|
|
|
+ assert evidence.finished["status"] == "success"
|
|
|
+ assert evidence.sample["sample_count"] <= 100
|
|
|
+ assert evidence.sample["redaction_policy"] == "rule-violation-default-v1"
|
|
|
+```
|
|
|
+
|
|
|
+Kestra compiler tests must prove downstream nodes receive the upstream `artifact_ref` expression rather than raw rows.
|
|
|
+
|
|
|
+- [ ] **Step 2: Run tests and verify failure**
|
|
|
+
|
|
|
+Run: `PYTHONPATH=. .venv/bin/pytest -q tests/runner/test_rule_evidence.py tests/runner/test_rules.py tests/core/orchestration/test_kestra_compiler.py`
|
|
|
+
|
|
|
+Expected: FAIL.
|
|
|
+
|
|
|
+- [ ] **Step 3: Implement evidence transaction boundaries**
|
|
|
+
|
|
|
+Insert `rule_runs` before adapter execution. On success/failure/cancel, finalize status, timings, counts, plan hash, and correlation ID. Violation samples must be redacted, stored as expiring artifacts, capped at 100 rows by default, and never written to application logs.
|
|
|
+
|
|
|
+- [ ] **Step 4: Add artifact handoff**
|
|
|
+
|
|
|
+Kestra node payloads reference prior task outputs:
|
|
|
+
|
|
|
+```python
|
|
|
+"parameters": {
|
|
|
+ "input_artifact": "{{ outputs.dataops_dag.tasks.<upstream>.body.output_artifact }}"
|
|
|
+}
|
|
|
+```
|
|
|
+
|
|
|
+For SQL pushdown within one datasource, pass a server-created staging dataset binding instead of row JSON.
|
|
|
+
|
|
|
+- [ ] **Step 5: Verify tests**
|
|
|
+
|
|
|
+Run: `PYTHONPATH=. .venv/bin/pytest -q tests/runner/test_rule_evidence.py tests/runner/test_rules.py tests/core/orchestration/test_kestra_compiler.py`
|
|
|
+
|
|
|
+Expected: PASS.
|
|
|
+
|
|
|
+- [ ] **Step 6: Commit**
|
|
|
+
|
|
|
+```bash
|
|
|
+git add app/runner/rule_evidence.py app/runner/rules.py app/runner/api.py app/core/orchestration/compilers/kestra.py app/core/data_rules/repository.py tests/runner/test_rule_evidence.py tests/runner/test_rules.py tests/core/orchestration/test_kestra_compiler.py
|
|
|
+git commit -m "feat: persist rule evidence and artifact handoff"
|
|
|
+```
|
|
|
+
|
|
|
+---
|
|
|
+
|
|
|
+### Task 7: Enforce compile, test, and publication gates
|
|
|
+
|
|
|
+**Files:**
|
|
|
+- Create: `app/core/data_rules/publication.py`
|
|
|
+- Create: `tests/core/data_rules/test_publication.py`
|
|
|
+- Modify: `app/core/data_rules/repository.py`
|
|
|
+- Modify: `app/api/data_rules/routes.py`
|
|
|
+- Modify: `app/core/data_rules/authoring.py`
|
|
|
+- Modify: `tests/test_data_rule_api.py`
|
|
|
+
|
|
|
+**Interfaces:**
|
|
|
+- Produces: `RulePublicationService.validate(version_id, actor_uid) -> dict`
|
|
|
+- Produces: `RulePublicationService.publish(version_id, actor_uid) -> dict`
|
|
|
+- Produces: signed `generation_receipt` from `/interpret`
|
|
|
+
|
|
|
+- [ ] **Step 1: Write failing state-machine tests**
|
|
|
+
|
|
|
+```python
|
|
|
+def test_rule_cannot_publish_without_compile_and_test_evidence():
|
|
|
+ with pytest.raises(ValueError, match="compile and test evidence"):
|
|
|
+ service.publish(version_id, actor_uid)
|
|
|
+
|
|
|
+
|
|
|
+def test_ready_generation_receipt_links_intent_to_rule_version():
|
|
|
+ interpreted = client.post("/api/rules/interpret", json=authoring_request()).get_json()["data"]
|
|
|
+ created = client.post(
|
|
|
+ "/api/rules/rule-versions",
|
|
|
+ json={
|
|
|
+ "source_text": interpreted["source_text"],
|
|
|
+ "rule_spec": interpreted["candidate"]["rule_spec"],
|
|
|
+ "generation_receipt": interpreted["generation_receipt"],
|
|
|
+ },
|
|
|
+ )
|
|
|
+ assert created.get_json()["data"]["generation_run_id"] == interpreted["generation_run_id"]
|
|
|
+```
|
|
|
+
|
|
|
+- [ ] **Step 2: Run tests and verify failure**
|
|
|
+
|
|
|
+Run: `PYTHONPATH=. .venv/bin/pytest -q tests/core/data_rules/test_publication.py tests/test_data_rule_api.py -k 'publish or receipt'`
|
|
|
+
|
|
|
+Expected: FAIL.
|
|
|
+
|
|
|
+- [ ] **Step 3: Implement signed receipt**
|
|
|
+
|
|
|
+Sign `generation_run_id`, actor UID, candidate hash, RuleSpec hash, and expiry using the application secret. On version creation, verify signature, actor, expiry, and RuleSpec hash before linking the generation run.
|
|
|
+
|
|
|
+- [ ] **Step 4: Correct lifecycle states**
|
|
|
+
|
|
|
+New RuleVersions start as `draft`. `validate` performs schema/type/capability compilation and stores compile evidence. A sample/golden dry-run stores test evidence. Only a version with successful evidence may become `validated`; only `validated` may become `published`.
|
|
|
+
|
|
|
+Plans transition `compiled -> tested -> published`. DataFlow release rejects any plan not in `published`.
|
|
|
+
|
|
|
+- [ ] **Step 5: Add bounded AI repair**
|
|
|
+
|
|
|
+Return deterministic validation errors to the authoring model for at most two repair attempts. Persist every candidate hash and attempt. Never repair ambiguity, destructive scope, or permission failures automatically.
|
|
|
+
|
|
|
+- [ ] **Step 6: Verify tests**
|
|
|
+
|
|
|
+Run: `PYTHONPATH=. .venv/bin/pytest -q tests/core/data_rules/test_publication.py tests/test_data_rule_api.py tests/core/data_rules/test_data_rule_repository.py`
|
|
|
+
|
|
|
+Expected: PASS.
|
|
|
+
|
|
|
+- [ ] **Step 7: Commit**
|
|
|
+
|
|
|
+```bash
|
|
|
+git add app/core/data_rules/publication.py app/core/data_rules/repository.py app/api/data_rules/routes.py app/core/data_rules/authoring.py tests/core/data_rules/test_publication.py tests/test_data_rule_api.py tests/core/data_rules/test_data_rule_repository.py
|
|
|
+git commit -m "feat: gate rule publication on compile and tests"
|
|
|
+```
|
|
|
+
|
|
|
+---
|
|
|
+
|
|
|
+### Task 8: Converge Standard and Data Flow product experiences
|
|
|
+
|
|
|
+**Files:**
|
|
|
+- Create: `frontend/src/components/DataRules/RuleCatalogPicker.vue`
|
|
|
+- Create: `frontend/src/components/DataRules/CompilationEvidence.vue`
|
|
|
+- Create: `frontend/src/components/DataRules/ProductionLineAssembler.vue`
|
|
|
+- Modify: `frontend/src/api/dataRules.js`
|
|
|
+- Modify: `frontend/src/components/DataRules/RuleAuthoringPanel.vue`
|
|
|
+- Modify: `frontend/src/views/dataGovernance/dataStandard/components/edit.vue`
|
|
|
+- Modify: `frontend/src/views/dataGovernance/dataProcess/components/edit.vue`
|
|
|
+- Modify: `tests/test_data_rule_frontend_contract.py`
|
|
|
+
|
|
|
+**Interfaces:**
|
|
|
+- Consumes: published rule/standard catalogs and compile/test evidence APIs
|
|
|
+- Produces: DataFlowSpec containing only fixed `rule_version_id` and `standard_version_id`
|
|
|
+
|
|
|
+- [ ] **Step 1: Write failing frontend contract tests**
|
|
|
+
|
|
|
+```python
|
|
|
+def test_dataflow_assembler_uses_published_catalog_ids_without_inline_rules():
|
|
|
+ source = ASSEMBLER.read_text(encoding="utf-8")
|
|
|
+ assert "RuleCatalogPicker" in source
|
|
|
+ assert "standard_version_id" in source
|
|
|
+ assert "rule_version_id" in source
|
|
|
+ assert "rule_spec" not in source
|
|
|
+ assert "free-text rule" not in source
|
|
|
+```
|
|
|
+
|
|
|
+- [ ] **Step 2: Run tests and verify failure**
|
|
|
+
|
|
|
+Run: `PYTHONPATH=. .venv/bin/pytest -q tests/test_data_rule_frontend_contract.py`
|
|
|
+
|
|
|
+Expected: FAIL because new components are absent.
|
|
|
+
|
|
|
+- [ ] **Step 3: Add catalog/evidence APIs and components**
|
|
|
+
|
|
|
+Expose searchable published assets with name, stable UID, version number, schema compatibility, owner, status, impact count, compiler backend, and latest test evidence. Never make raw IDs the only visible selection label.
|
|
|
+
|
|
|
+- [ ] **Step 4: Replace parallel legacy editing**
|
|
|
+
|
|
|
+For Data Standard, replace the required legacy “操作代码” field with read-only generated artifact/evidence display. For Data Flow, replace free-text `rule` plus embedded `rule_spec` with ProductionLineAssembler. Preserve legacy fields read-only for migration and display a clear migration status.
|
|
|
+
|
|
|
+- [ ] **Step 5: Verify frontend**
|
|
|
+
|
|
|
+Run:
|
|
|
+
|
|
|
+```bash
|
|
|
+PYTHONPATH=. .venv/bin/pytest -q tests/test_data_rule_frontend_contract.py
|
|
|
+cd frontend && npm run build
|
|
|
+```
|
|
|
+
|
|
|
+Expected: tests PASS and build completes without errors.
|
|
|
+
|
|
|
+- [ ] **Step 6: Commit**
|
|
|
+
|
|
|
+```bash
|
|
|
+git add frontend/src/components/DataRules frontend/src/api/dataRules.js frontend/src/views/dataGovernance/dataStandard/components/edit.vue frontend/src/views/dataGovernance/dataProcess/components/edit.vue tests/test_data_rule_frontend_contract.py
|
|
|
+git commit -m "feat: assemble production lines from published rule assets"
|
|
|
+```
|
|
|
+
|
|
|
+---
|
|
|
+
|
|
|
+### Task 9: Implement Data Factory deployment, canary, activation, and rollback
|
|
|
+
|
|
|
+**Files:**
|
|
|
+- Create: `app/core/data_rules/deployment.py`
|
|
|
+- Create: `tests/core/data_rules/test_deployment.py`
|
|
|
+- Create: `tests/integration/test_data_rule_factory_lifecycle.py`
|
|
|
+- Modify: `app/core/data_rules/repository.py`
|
|
|
+- Modify: `app/api/data_rules/routes.py`
|
|
|
+- Modify: `app/core/system/permissions.py`
|
|
|
+- Modify: `frontend/src/views/dataFactory/workflow/ProductionLineDeployment.vue`
|
|
|
+- Modify: `frontend/src/api/dataRules.js`
|
|
|
+- Modify: `deploy/docker/kestra/application.yml`
|
|
|
+- Modify: `deploy/docker/docker-compose.yml`
|
|
|
+
|
|
|
+**Interfaces:**
|
|
|
+- Produces: `DataFlowDeploymentService.create(...) -> dict`
|
|
|
+- Produces: `deploy_disabled(deployment_id, actor_uid) -> dict`
|
|
|
+- Produces: `run_canary(deployment_id, inputs, actor_uid) -> dict`
|
|
|
+- Produces: `activate(deployment_id, evidence_id, actor_uid) -> dict`
|
|
|
+- Produces: `rollback(deployment_id, actor_uid) -> dict`
|
|
|
+
|
|
|
+- [ ] **Step 1: Write failing lifecycle tests**
|
|
|
+
|
|
|
+```python
|
|
|
+def test_production_deployment_requires_disabled_deploy_and_passed_canary():
|
|
|
+ deployment = service.create(released_flow_id, production_binding(), actor)
|
|
|
+ disabled = service.deploy_disabled(deployment["id"], actor)
|
|
|
+ with pytest.raises(ValueError, match="passed canary"):
|
|
|
+ service.activate(disabled["id"], None, actor)
|
|
|
+ evidence = service.run_canary(disabled["id"], {"biz_date": "2026-07-23"}, actor)
|
|
|
+ active = service.activate(disabled["id"], evidence["id"], actor)
|
|
|
+ assert active["status"] == "active"
|
|
|
+```
|
|
|
+
|
|
|
+Add rollback tests proving the previous active package/bindings remain available and that rollback deactivates the candidate before restoring the prior version.
|
|
|
+
|
|
|
+- [ ] **Step 2: Run tests and verify failure**
|
|
|
+
|
|
|
+Run: `PYTHONPATH=. .venv/bin/pytest -q tests/core/data_rules/test_deployment.py`
|
|
|
+
|
|
|
+Expected: FAIL.
|
|
|
+
|
|
|
+- [ ] **Step 3: Implement state machine**
|
|
|
+
|
|
|
+Allowed transitions:
|
|
|
+
|
|
|
+```python
|
|
|
+TRANSITIONS = {
|
|
|
+ "draft": {"disabled", "failed"},
|
|
|
+ "disabled": {"canary", "failed"},
|
|
|
+ "canary": {"active", "failed", "rolled_back"},
|
|
|
+ "active": {"superseded", "rolled_back", "failed"},
|
|
|
+ "failed": {"disabled", "rolled_back"},
|
|
|
+ "superseded": {"rolled_back"},
|
|
|
+ "rolled_back": set(),
|
|
|
+}
|
|
|
+```
|
|
|
+
|
|
|
+Compile the released package’s WorkflowSpec with its schedule plan, deploy disabled through `KestraAdapter`, store engine definition/revision/hash, and require passed canary evidence with matching deployment/package/binding hashes before activation.
|
|
|
+
|
|
|
+- [ ] **Step 4: Separate permissions**
|
|
|
+
|
|
|
+Add:
|
|
|
+
|
|
|
+- `dataflows:deploy`
|
|
|
+- `dataflows:canary`
|
|
|
+- `dataflows:activate`
|
|
|
+- `dataflows:rollback`
|
|
|
+- `rules:execute`
|
|
|
+
|
|
|
+Keep authoring, publication, release, and production activation as distinct permissions.
|
|
|
+
|
|
|
+- [ ] **Step 5: Repair and verify Kestra Docker stability**
|
|
|
+
|
|
|
+Use a dedicated Kestra PostgreSQL database/user or stable connection configuration, add a restart policy and health-gated dependencies, then run a 30-minute queue/trigger health soak. The current broken-pipe/closed-connection exit must not recur.
|
|
|
+
|
|
|
+- [ ] **Step 6: Implement Data Factory UI**
|
|
|
+
|
|
|
+Show released package hash, bound datasets/schemas, engine definition, canary evidence, current/previous active version, and explicit deploy/canary/activate/rollback actions based on permissions.
|
|
|
+
|
|
|
+- [ ] **Step 7: Verify lifecycle**
|
|
|
+
|
|
|
+Run: `PYTHONPATH=. .venv/bin/pytest -q tests/core/data_rules/test_deployment.py tests/integration/test_data_rule_factory_lifecycle.py tests/test_permission_matrix.py`
|
|
|
+
|
|
|
+Expected: PASS.
|
|
|
+
|
|
|
+- [ ] **Step 8: Commit**
|
|
|
+
|
|
|
+```bash
|
|
|
+git add app/core/data_rules/deployment.py app/core/data_rules/repository.py app/api/data_rules/routes.py app/core/system/permissions.py frontend/src/views/dataFactory/workflow/ProductionLineDeployment.vue frontend/src/api/dataRules.js deploy/docker/kestra/application.yml deploy/docker/docker-compose.yml tests/core/data_rules/test_deployment.py tests/integration/test_data_rule_factory_lifecycle.py tests/test_permission_matrix.py
|
|
|
+git commit -m "feat: deploy and roll back governed data production lines"
|
|
|
+```
|
|
|
+
|
|
|
+---
|
|
|
+
|
|
|
+### Task 10: Complete end-to-end acceptance and rollout controls
|
|
|
+
|
|
|
+**Files:**
|
|
|
+- Create: `tests/e2e/test_ai_rule_to_data_product.py`
|
|
|
+- Create: `tests/performance/test_rule_execution_capacity.py`
|
|
|
+- Create: `docs/operations/DATA_RULE_RUNTIME_RUNBOOK.md`
|
|
|
+- Modify: `docs/architecture/ADR-006-data-rule-runtime.md`
|
|
|
+- Modify: `docs/architecture/NEXT_ITERATION_ROADMAP.md`
|
|
|
+- Modify: `docs/superpowers/plans/2026-07-23-ai-data-rule-production-line-implementation.md`
|
|
|
+
|
|
|
+**Interfaces:**
|
|
|
+- Consumes: complete authoring, publication, release, execution, evidence, and deployment APIs
|
|
|
+- Produces: reproducible acceptance evidence and operational rollback procedure
|
|
|
+
|
|
|
+- [ ] **Step 1: Add one real-model acceptance scenario**
|
|
|
+
|
|
|
+Configure one approved OpenAI-compatible provider in the test environment and submit:
|
|
|
+
|
|
|
+> 客户手机号去除空格后必须为 11 位数字;格式错误的记录进入隔离区;按 customer_id 去重并保留 updated_at 最新记录。
|
|
|
+
|
|
|
+Assert that the generated candidate contains normalization, assertion/quarantine, and deduplication semantics; then link it to a RuleVersion and retain the model/prompt/context/candidate hashes.
|
|
|
+
|
|
|
+- [ ] **Step 2: Add full data-product E2E**
|
|
|
+
|
|
|
+The E2E test must:
|
|
|
+
|
|
|
+1. Create raw PostgreSQL/MySQL input rows.
|
|
|
+2. Publish RuleVersion and StandardVersion through APIs.
|
|
|
+3. Assemble and release DataFlowVersion.
|
|
|
+4. Bind development datasets in Data Factory.
|
|
|
+5. Deploy disabled to Kestra.
|
|
|
+6. Run canary and verify output rows plus violation evidence.
|
|
|
+7. Activate.
|
|
|
+8. Run a scheduled/manual production execution.
|
|
|
+9. Roll back and prove the prior active version is restored.
|
|
|
+
|
|
|
+- [ ] **Step 3: Add failure and security acceptance**
|
|
|
+
|
|
|
+Cover:
|
|
|
+
|
|
|
+- plan body/hash tampering;
|
|
|
+- schema drift after release;
|
|
|
+- revoked plan;
|
|
|
+- expired artifact;
|
|
|
+- missing/rotated datasource credential;
|
|
|
+- duplicate task-token replay;
|
|
|
+- Runner restart during execution;
|
|
|
+- Kestra restart before/after canary;
|
|
|
+- write failure with known/unknown commit outcome;
|
|
|
+- unauthorized publish/deploy/activate/rollback;
|
|
|
+- violation sample redaction and TTL deletion.
|
|
|
+
|
|
|
+- [ ] **Step 4: Add capacity gates**
|
|
|
+
|
|
|
+Measure SQL pushdown and Polars paths at 100 thousand, 1 million, and 10 million rows where the local machine can support it. Record throughput, peak memory, datasource pool usage, artifact size, and p95 node duration. Fail the gate when memory exceeds configured Runner limits or the pool budget is exceeded.
|
|
|
+
|
|
|
+- [ ] **Step 5: Write operational runbook**
|
|
|
+
|
|
|
+Document:
|
|
|
+
|
|
|
+- dependency and health checks;
|
|
|
+- compile/test evidence lookup;
|
|
|
+- release/deployment identifiers and hashes;
|
|
|
+- canary interpretation;
|
|
|
+- activate/rollback commands through governed APIs;
|
|
|
+- cleanup/TTL behavior;
|
|
|
+- incident triage for model, compiler, Runner, datasource, MinIO, and Kestra failures;
|
|
|
+- explicit rule that Data Factory never edits/recompiles business semantics.
|
|
|
+
|
|
|
+- [ ] **Step 6: Run final acceptance**
|
|
|
+
|
|
|
+Run:
|
|
|
+
|
|
|
+```bash
|
|
|
+PYTHONPATH=. .venv/bin/pytest -q
|
|
|
+cd frontend && npm run build
|
|
|
+cd ..
|
|
|
+docker compose -f deploy/docker/docker-compose.yml up -d --build
|
|
|
+docker compose -f deploy/docker/docker-compose.yml ps
|
|
|
+PYTHONPATH=. .venv/bin/pytest -q tests/e2e/test_ai_rule_to_data_product.py tests/performance/test_rule_execution_capacity.py
|
|
|
+```
|
|
|
+
|
|
|
+Expected:
|
|
|
+
|
|
|
+- all required containers healthy, including Kestra;
|
|
|
+- full Python suite and frontend build pass;
|
|
|
+- real-model authoring canary passes;
|
|
|
+- SQL and Polars execution paths pass;
|
|
|
+- canary, activation, rollback, restart, drift, tamper, and RBAC cases pass;
|
|
|
+- `/api/rules/capabilities` reports `data_factory_activation=true` only after these gates.
|
|
|
+
|
|
|
+- [ ] **Step 7: Commit**
|
|
|
+
|
|
|
+```bash
|
|
|
+git add tests/e2e/test_ai_rule_to_data_product.py tests/performance/test_rule_execution_capacity.py docs/operations/DATA_RULE_RUNTIME_RUNBOOK.md docs/architecture/ADR-006-data-rule-runtime.md docs/architecture/NEXT_ITERATION_ROADMAP.md docs/superpowers/plans/2026-07-23-ai-data-rule-production-line-implementation.md
|
|
|
+git commit -m "test: accept end-to-end governed rule execution"
|
|
|
+```
|
|
|
+
|
|
|
+---
|
|
|
+
|
|
|
+## 4. Recommended execution order
|
|
|
+
|
|
|
+Do not start with the Data Factory UI or arbitrary generated Python. The shortest safe path is:
|
|
|
+
|
|
|
+1. Tasks 1–3 freeze trustworthy execution contracts and expression semantics.
|
|
|
+2. Task 4 proves one real SQL-pushdown production-line node end to end.
|
|
|
+3. Tasks 5–6 add cross-source/multi-node execution and evidence.
|
|
|
+4. Task 7 makes publication status truthful.
|
|
|
+5. Task 8 converges user-facing Standard/Rule/DataFlow editing.
|
|
|
+6. Task 9 opens Data Factory activation only after runtime readiness.
|
|
|
+7. Task 10 supplies operational acceptance and turns the capability flag on.
|
|
|
+
|
|
|
+M3A is the first operational checkpoint. At its completion, DataOps can execute a limited but honest subset of rules. Operators unsupported by a selected backend must remain visibly “not compilable”; they must never fall back to model-generated code or silently change semantics.
|
|
|
+
|
|
|
+## 5. Self-review result
|
|
|
+
|
|
|
+- Spec coverage: natural-language authoring, Data Standard, Data Flow production-line assembly, Data Factory deployment, deterministic code/plan generation, real execution, evidence, canary, activation, and rollback each map to explicit tasks.
|
|
|
+- Boundary coverage: Standard, Rule, DataFlowVersion, and DataFlowDeployment remain separate and linked by immutable IDs/hashes.
|
|
|
+- Runtime coverage: SQL and Polars paths have separate compilers, adapters, and real-data tests.
|
|
|
+- Safety coverage: typed expressions, trusted schema snapshots, plan/hash verification, no inline code, redacted evidence, RBAC separation, drift/tamper/restart tests, and fail-closed capability selection are included.
|
|
|
+- Deferred boundary: generated Python is intentionally excluded from the activation path until a separately accepted signed-artifact supply chain exists.
|