# 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**

```python
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**

```python
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**

```python
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**

```python
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**

```python
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**

```python
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**

```python
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**

```python
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:

```bash
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.

## M3A–M5 completion addendum（2026-07-24）

本计划的实现继续沿用“数据标准和数据流程共用规则内核，数据工厂只投产数据生产线”
的边界。后续完整执行计划已补齐真实 SQL/Polars I/O、MinIO 制品传递、执行证据、
发布门禁、产品面收敛和数据工厂生命周期：

- [数据规则完整执行计划](2026-07-23-data-rule-execution-completion.md)
- [M5 验收记录](../../validation/DATA_RULE_M5_ACCEPTANCE_20260724.md)
- [运行时运维手册](../../operations/DATA_RULE_RUNTIME_RUNBOOK.md)

验收采用真实 loopback OpenAI-compatible Qwen CPU 模型，不用固定语义输出替代；
容量覆盖 SQL/Polars 的 100K、1M、10M；生成 Python 仍未进入生产激活路径。