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:

Natural language produces a RuleCandidate and an audit record.
The candidate is linked to an immutable RuleVersion.
Expressions and operations are parsed, typed against a server-owned SchemaSnapshot, and checked against a backend capability matrix.
The compiler emits a deterministic backend plan and golden tests.
Sample/dry-run evidence passes before RuleVersion and plan publication.
A DataFlowVersion fixes StandardVersion, RuleVersion, SchemaSnapshot, and plan hashes.
Data Factory binds physical source/target datasets, environment, schedule, and resources without changing rule semantics.
Kestra invokes Runner using only immutable IDs, hashes, artifact references, and short-lived task tokens.
Runner executes real data, applies reject/quarantine policy, writes idempotently, and records rule-run evidence.
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

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:

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:

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

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

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:

Load the schema by stable schema_ref.
Normalize fields by name, type, nullability, precision, scale, and timezone.
Hash only normalized fields.
Persist/reuse an immutable snapshot.
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:

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

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

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:

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

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

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:

{
    "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:

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

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

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:

{
    "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

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

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:

"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

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

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

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

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:

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

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

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:

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

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:

Create raw PostgreSQL/MySQL input rows.
Publish RuleVersion and StandardVersion through APIs.
Assemble and release DataFlowVersion.
Bind development datasets in Data Factory.
Deploy disabled to Kestra.
Run canary and verify output rows plus violation evidence.
Activate.
Run a scheduled/manual production execution.
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:

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

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:

Tasks 1–3 freeze trustworthy execution contracts and expression semantics.
Task 4 proves one real SQL-pushdown production-line node end to end.
Tasks 5–6 add cross-source/multi-node execution and evidence.
Task 7 makes publication status truthful.
Task 8 converges user-facing Standard/Rule/DataFlow editing.
Task 9 opens Data Factory activation only after runtime readiness.
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.

2026-07-23-data-rule-execution-completion.md 47 KB Vēsture Neapstrādāts

Data Rule Execution Completion Implementation Plan

Global Constraints

1. Verified baseline and execution-readiness gap

1.1 Already implemented

1.2 Blocking gaps

1.3 Definition of “complete adaptable execution”

2. Delivery milestones

3. File map

Task 1: Freeze execution, schema, and dataset contracts

Task 2: Resolve trusted schemas and physical dataset bindings

Task 3: Parse and type-check canonical rule expressions

Task 4: Deliver the SQLGlot executable vertical slice

Task 5: Add Polars cross-source compilation and artifact I/O

Task 6: Pass artifacts between nodes and persist rule evidence

Task 7: Enforce compile, test, and publication gates

Task 8: Converge Standard and Data Flow product experiences

Task 9: Implement Data Factory deployment, canary, activation, and rollback

Task 10: Complete end-to-end acceptance and rollout controls

4. Recommended execution order

5. Self-review result

2026-07-23-data-rule-execution-completion.md 47 KB

Vēsture Neapstrādāts