DataOps-platform/docs/superpowers/plans/2026-07-23-ai-data-rule-production-line-implementation.md

AI Data Rule Production Line 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: Build the governed path from natural-language data standards and rules, through immutable DataFlow production-line assembly, to Data Factory deployment on Kestra and the DataOps Runner.

Architecture: Data standards, reusable rules, DataFlow versions, and Data Factory deployments remain separate versioned aggregates while sharing one RuleSpec compiler and artifact model. AI produces schema-constrained candidates; deterministic validators, tests, policy, and immutable hashes decide what can be published. A released DataFlowVersion resolves StandardVersion and RuleVersion references into a ProductionLinePackage and WorkflowSpec; only Data Factory may deploy and activate it.

Tech Stack: Flask 2.3, SQLAlchemy 2, Alembic/PostgreSQL JSONB, OpenAI-compatible LLM client, existing WorkflowSpec/Kestra compiler, DataOps Runner, Vue 2/Vuetify, pytest, Docker Compose.

2026-07-23 implementation status

This iteration delivers a deployable governed foundation:

• completed: closed RuleSpec/StandardSpec/DataFlowSpec contracts and hashes;
• completed: natural-language authoring agent with schema-constrained output, ambiguity/confidence gate, model/prompt/context/candidate evidence;
• completed: immutable PostgreSQL schema and restored contiguous migration history through 20260723_110;
• completed: published StandardVersion expansion, fixed RuleVersion/plan references, deterministic ProductionLinePackage and WorkflowSpec generation;
• completed: governed validate/interpret/preview APIs and separate read/edit/publish/release permissions;
• completed: Runner lookup by binding/version/hash, fail-closed status checks, SQL pushdown and quality-check dispatch;
• completed: AI authoring on Data Standard and Data Flow surfaces, plus an explicit Data Factory deployment readiness gate;
• completed: immutable RuleVersion and StandardVersion repository commands, separate create/publish permissions, and PostgreSQL-backed APIs;
• completed: every AI interpretation persists model, prompt, context, candidate, confidence, ambiguity, decision, and correlation hashes;
• completed: server-side DataFlow release loads only published assets, expands StandardVersion clauses, generates deterministic RuleSpec plans, fixes component bindings, and persists an immutable ProductionLinePackage;
• completed: both authoring surfaces can create governed versions, while Data Factory reports release readiness separately from activation readiness;
• completed: full pytest, frontend production build, fresh-database migration, Docker migration, health and route-authentication acceptance.

The following M3/M5 runtime activation work is deliberately not represented as available capability yet:

• Polars/CEL/SQLGlot compilers, signed generated-code artifacts, violation sample storage, and production run evidence;
• deployment-time binding of concrete data sources, tables/artifacts, engine dialect, and resources;
• Data Factory canary, activate, disable, supersede, and rollback commands;
• live Data Factory deployment actions and a successful LLM-provider canary.

/api/rules/capabilities now returns production_line_release=true and data_factory_activation=false. The Vue deployment page exposes this distinction: a production line may be released, but cannot be activated until the M3 runtime adapter and M5 canary/rollback evidence are complete.

---

Delivery milestones

Milestone	Tasks	Deployable outcome
M1 — governed contracts	1–3	Rule/Standard/DataFlow schemas, immutable persistence, AI candidate validation
M2 — production-line control plane	4–5	Standards and rules resolve into a released ProductionLinePackage and API
M3 — deterministic execution	6	rule.apply/quality.check compile to Kestra and fail closed in Runner
M4 — product experience	7	Data Standard, Data Flow, and Data Factory pages expose their separate duties
M5 — acceptance and rollout	8	Migration, API, canary, rollback, and Docker evidence

File map

• app/core/data_rules/contracts.py: closed schemas, normalization, canonical hashing.
• app/core/data_rules/authoring.py: schema-constrained AI candidate generation and bounded repair.
• app/core/data_rules/repository.py: immutable PostgreSQL versions and catalog reads.
• app/core/data_rules/production_line.py: standard expansion, conflict checks, package resolution.
• app/api/data_rules/routes.py: rule/standard validation, authoring, publishing, and resolve APIs.
• app/core/orchestration/spec.py: registered rule.apply and quality.check workflow nodes.
• app/runner/rules.py: published-plan lookup and deterministic rule execution adapter.
• migrations/versions/20260723_110_ai_data_rules.py: rule, standard, DataFlow version, artifact, deployment, and run tables.
• frontend/src/views/dataGovernance/dataStandard/: natural-language standard authoring and read-only executable artifacts.
• frontend/src/views/dataGovernance/dataProcess/: StandardVersion/RuleVersion production-line assembly.
• frontend/src/views/dataFactory/workflow/: released-line deployment, canary, activate, and rollback.
• tests/core/data_rules/: domain contract, authoring, repository, and resolver tests.
• tests/integration/test_data_rule_production_line.py: PostgreSQL-backed end-to-end control-plane acceptance.

Task 1: Closed rule, standard, and production-line contracts

Files:

• Create: app/core/data_rules/__init__.py
• Create: app/core/data_rules/contracts.py

• Create: tests/core/data_rules/test_contracts.py

• [ ] Step 1: Write failing tests for valid RuleSpec normalization and stable hashing

def test_rule_spec_is_closed_normalized_and_hash_stable():
    normalized = validate_rule_spec(valid_rule_spec())
    assert normalized["steps"][0]["op"] == "normalize_text"
    assert rule_spec_hash(valid_rule_spec()) == rule_spec_hash(reordered_rule_spec())

• Step 2: Run the contract test and verify missing module failure

Run: PYTHONPATH=. .venv/bin/pytest -q tests/core/data_rules/test_contracts.py

Expected: FAIL because app.core.data_rules.contracts does not exist.

• Step 3: Implement closed validators and canonical hashes

RULE_OPS = {
    "cast", "normalize_text", "regex_replace", "fill_null", "filter",
    "derive", "map_values", "assert", "deduplicate", "aggregate",
    "lookup_join", "mask",
}

def rule_spec_hash(value):
    normalized = validate_rule_spec(value)
    canonical = json.dumps(
        normalized, sort_keys=True, separators=(",", ":"), ensure_ascii=False
    )
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()

The validators must close unknown fields, require UUIDv7 identifiers, reject secret material and arbitrary source code, bound arrays and strings, enforce unique step/component IDs, and reject unsupported component types.

• Step 4: Verify contract tests pass

Run: PYTHONPATH=. .venv/bin/pytest -q tests/core/data_rules/test_contracts.py

Expected: all contract tests pass.

Task 2: Schema-constrained AI authoring candidate

Files:

• Create: app/core/data_rules/authoring.py

• Create: tests/core/data_rules/test_authoring.py

• [ ] Step 1: Write failing tests for standard and flow authoring

def test_authoring_validates_model_output_before_returning_candidate():
    model = FakeModel({"candidate_type": "rule", "rule_spec": valid_rule_spec()})
    candidate = RuleAuthoringAgent(model=model).interpret(
        source_text="手机号去空格后必须为11位数字",
        authoring_surface="data_standard",
        context={"input_schema_ref": "bd:customer:v1"},
    )
    assert candidate["source_text"].startswith("手机号")
    assert candidate["candidate_hash"]

• Step 2: Verify tests fail because the authoring agent is absent

Run: PYTHONPATH=. .venv/bin/pytest -q tests/core/data_rules/test_authoring.py

Expected: FAIL on missing RuleAuthoringAgent.

• Step 3: Implement deterministic candidate validation and model metadata

class RuleAuthoringAgent:
    def interpret(self, *, source_text, authoring_surface, context):
        raw = self.model.generate(
            messages=build_rule_messages(source_text, authoring_surface, context),
            response_schema=RULE_CANDIDATE_SCHEMA,
            timeout_seconds=self.timeout_seconds,
        )
        candidate = validate_rule_candidate(json.loads(raw))
        return attach_generation_evidence(candidate, source_text, self.model, context)

The model receives only bounded, redacted context. The returned candidate records provider, model, prompt version, context hash, candidate hash, assumptions, ambiguities, and confidence. Invalid output fails closed; no SQL, Python, publication, or production tool is invoked by the model.

• Step 4: Verify offline and invalid-output tests pass

Run: PYTHONPATH=. .venv/bin/pytest -q tests/core/data_rules/test_authoring.py

Expected: valid candidates pass; malformed/ambiguous output is rejected or marked clarification_required.

Task 3: Immutable PostgreSQL control-plane schema

Files:

• Create: migrations/versions/20260723_110_ai_data_rules.py
• Modify: tests/test_database_migrations.py

• Create: tests/test_data_rule_schema.py

• [ ] Step 1: Write failing migration contract tests

EXPECTED_RULE_TABLES = {
    "data_rules", "data_rule_versions", "rule_generation_runs",
    "data_standards", "data_standard_versions", "standard_rule_bindings",
    "dataflow_versions", "dataflow_component_bindings",
    "rule_execution_plans", "rule_artifacts", "dataflow_deployments",
    "rule_runs", "rule_violation_samples",
}

• Step 2: Run tests and verify the migration/table assertions fail

Run: PYTHONPATH=. .venv/bin/pytest -q tests/test_data_rule_schema.py

Expected: FAIL because revision 20260723_110 is absent.

• Step 3: Add forward-only immutable version tables and constraints

The migration must use PostgreSQL UUID/JSONB/TIMESTAMPTZ, immutable version status checks, unique version numbers, fixed StandardVersion→RuleVersion and DataFlowVersion→component foreign keys, artifact hashes, AI audit hashes, deployment environment/status checks, and run/violation evidence. Downgrade preserves business data and performs no destructive drop.

• Step 4: Verify static migration contracts

Run: PYTHONPATH=. .venv/bin/pytest -q tests/test_data_rule_schema.py tests/test_database_migrations.py

Expected: all non-integration migration tests pass.

Task 4: Production-line resolver and WorkflowSpec generation

Files:

• Create: app/core/data_rules/production_line.py
• Create: tests/core/data_rules/test_production_line.py
• Modify: app/core/orchestration/spec.py

• Modify: tests/core/orchestration/test_spec.py

• [ ] Step 1: Write failing tests for StandardVersion expansion

package = resolve_production_line(
    dataflow_spec=valid_dataflow_spec(),
    standard_versions={standard_id: published_standard(rule_id)},
    rule_versions={rule_id: published_rule(valid_rule_spec())},
)
assert package["standard_version_ids"] == [standard_id]
assert package["rule_version_ids"] == [rule_id]
assert package["workflow_spec"]["nodes"][0]["type"] == "quality.check"

• Step 2: Verify tests fail on missing resolver and node types

Run: PYTHONPATH=. .venv/bin/pytest -q tests/core/data_rules/test_production_line.py tests/core/orchestration/test_spec.py

Expected: FAIL because the resolver and rule nodes are not registered.

• Step 3: Implement fail-closed expansion and deterministic package hash

The resolver must require published referenced versions, expand standard.enforce, deduplicate identical rule versions, preserve standard clause provenance, reject conflicting writes to the same target, generate acyclic nodes/edges, and produce sorted StandardVersion/RuleVersion lists plus a canonical package hash.

• Step 4: Register governed workflow node contracts

rule.apply and quality.check must require component_binding_id, rule_version_id, and execution_plan_hash. rule.apply writes require an idempotency strategy. Neither node accepts inline RuleSpec, SQL, Python source, credentials, URLs, or artifact bodies.

• Step 5: Verify resolver and WorkflowSpec tests pass

Run: PYTHONPATH=. .venv/bin/pytest -q tests/core/data_rules/test_production_line.py tests/core/orchestration/test_spec.py

Expected: all tests pass.

Task 5: Repository and governed API

Files:

• Create: app/core/data_rules/repository.py
• Create: app/api/data_rules/__init__.py
• Create: app/api/data_rules/routes.py
• Modify: app/__init__.py
• Modify: app/core/system/permissions.py
• Create: tests/core/data_rules/test_repository.py

• Create: tests/test_data_rule_api.py

• [ ] Step 1: Write failing API tests

response = client.post(
    "/api/rules/production-lines/resolve",
    json={"dataflow_spec": valid_dataflow_spec()},
    headers=editor_token_headers,
)
assert response.status_code == 200
assert response.get_json()["data"]["package_hash"]

• Step 2: Verify tests fail because the blueprint is absent

Run: PYTHONPATH=. .venv/bin/pytest -q tests/test_data_rule_api.py

Expected: FAIL with 404.

• Step 3: Implement immutable catalog reads/writes

Repository operations create stable identities and draft versions, publish only validated versions, load all referenced StandardVersion/RuleVersion rows in bounded queries, and create released DataFlowVersion/component rows transactionally. Published and released rows are never updated in place.

• Step 4: Implement /api/rules control-plane routes and RBAC

Add capabilities, interpret, validate, create-version, publish, resolve, and release routes. Use existing response envelopes and rules:read/edit/publish, standards:*, dataflows:* permissions. API errors are bounded and do not echo model output, secrets, generated code, or raw data.

• Step 5: Verify repository and API tests pass

Run: PYTHONPATH=. .venv/bin/pytest -q tests/core/data_rules/test_repository.py tests/test_data_rule_api.py tests/test_permission_matrix.py

Expected: all tests pass with 401/403/409 cases covered.

Task 6: Runner plan execution boundary

Files:

• Create: app/runner/rules.py
• Modify: app/runner/bootstrap.py
• Modify: tests/runner/test_nodes.py
• Create: tests/runner/test_rules.py

• Modify: tests/core/orchestration/test_kestra_compiler.py

• [ ] Step 1: Write failing tests for signed plan lookup

result = RulePlanExecutor(repository, adapters={"quality_check": adapter}).execute(
    governed_node(), {}, write_authorized=False
)
assert result["status"] == "pass"
assert repository.requested == [(binding_id, plan_hash)]

• Step 2: Verify tests fail because no rule executor is registered

Run: PYTHONPATH=. .venv/bin/pytest -q tests/runner/test_rules.py

Expected: FAIL on missing executor.

• Step 3: Implement plan lookup and adapter dispatch

Runner loads only a published plan by component binding and exact hash, verifies artifact digest/status, dispatches only a registered backend adapter, and refuses inline plans. Initial adapters support deterministic quality checks and precompiled SQL/Polars handler names; generated Python remains disabled until signed-artifact sandbox tests exist.

• Step 4: Verify Kestra output contains only immutable references

Run: PYTHONPATH=. .venv/bin/pytest -q tests/runner/test_rules.py tests/core/orchestration/test_kestra_compiler.py

Expected: tests pass and compiled YAML contains version/hash references but no rule text, source code, credentials, or data.

Task 7: Three product surfaces

Files:

• Modify: frontend/src/views/dataGovernance/dataStandard/components/edit.vue
• Modify: frontend/src/views/dataGovernance/dataStandard/index.vue
• Modify: frontend/src/views/dataGovernance/dataProcess/components/edit.vue
• Modify: frontend/src/views/dataFactory/workflow/WorkflowList.vue
• Modify: frontend/src/api/dataGovernance.js
• Modify: frontend/src/api/dataFactory.js

• Create: tests/test_data_rule_frontend_contract.py

• [ ] Step 1: Write failing frontend contract tests

The test asserts that Data Standard uses natural-language authoring and read-only executable artifacts, Data Process includes StandardVersion and RuleVersion components plus release, and Data Factory lists released DataFlowVersion deployments plus canary/activate/rollback. It also asserts the legacy direct dataStandardCodeGenerate save path is absent.

• Step 2: Verify frontend contract tests fail

Run: PYTHONPATH=. .venv/bin/pytest -q tests/test_data_rule_frontend_contract.py

Expected: FAIL on missing API calls and component labels.

• Step 3: Implement APIs and focused Vue components

Use existing Vuetify patterns. Keep AI interpretation, assumptions, ambiguities, sample results, immutable versions, and impact visible. Do not render credentials, raw sensitive rows, editable generated source, or Data Factory rule-edit controls.

• Step 4: Verify frontend tests and build

Run: PYTHONPATH=. .venv/bin/pytest -q tests/test_data_rule_frontend_contract.py

Run: npm --prefix frontend run build

Expected: tests and production build pass.

Task 8: PostgreSQL, API, and Docker acceptance

Files:

• Create: tests/integration/test_data_rule_production_line.py
• Modify: tests/test_local_docker_contract.py

• Modify: deploy/docker/README.md

• [ ] Step 1: Write opt-in integration acceptance

The test creates a rule and standard, publishes immutable versions, resolves/releases a DataFlowVersion, creates a test DataFlowDeployment, verifies all hashes and foreign keys, and proves a new StandardVersion does not mutate the released production line.

• Step 2: Run targeted unit and contract suite

Run:

PYTHONPATH=. .venv/bin/pytest -q \
  tests/core/data_rules \
  tests/core/orchestration \
  tests/runner \
  tests/test_data_rule_schema.py \
  tests/test_data_rule_api.py \
  tests/test_database_migrations.py \
  tests/test_local_docker_contract.py

Expected: all tests pass; PostgreSQL integration may skip unless explicitly enabled.

• Step 3: Build and start the isolated Docker stack

Run: docker compose -f deploy/docker/docker-compose.yml up -d --build postgres neo4j minio minio-init kestra runner backend frontend

Expected: all requested services reach healthy/completed state.

• Step 4: Run migration and live API acceptance

Run: docker compose -f deploy/docker/docker-compose.yml exec backend alembic -c alembic.ini upgrade head

Run the authenticated live API smoke test for rule validation, production-line resolve, backend health, and runner health.

Expected: migration head is 20260723_110; API returns immutable package hashes; health endpoints are successful.

• Step 5: Capture final evidence

Run: docker compose -f deploy/docker/docker-compose.yml ps

Run: git diff --check

Expected: containers are healthy, tests/build are green, and diff check reports no whitespace errors.

Self-review

• Spec coverage: Tasks 1–2 cover natural-language AI generation; Task 3 covers all immutable aggregates; Tasks 4–5 cover standard/rule assembly into a DataFlow production line; Task 6 covers governed execution; Task 7 covers all three product surfaces; Task 8 covers local Docker deployment and acceptance.
• Boundary check: Data Standard and Data Flow generate/design versions; Data Factory deploys released DataFlowVersion only.
• Security check: model output never directly executes; WorkflowSpec and Runner carry immutable references and hashes only.
• Migration check: published/released/deployed evidence is forward-preserved; no destructive downgrade.