DataOps-platform/docs/superpowers/plans/2026-07-29-wp03-enterprise-datasource-ingestion.md

WP-03 Enterprise Data Source Ingestion 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: Connect the existing PostgreSQL/MySQL data-source runtime to the data-research ingestion control plane so a configured enterprise database can be registered, collected repeatedly, diagnosed, retried, and traced to an immutable catalog snapshot and field evidence.

Architecture: Keep app/core/data_source as the credential-safe connection boundary and app/core/data_research as the ingestion control plane. Database collection runs synchronously behind an explicit job execution endpoint for the demonstration release, while persisted job state, attempt numbers, catalog snapshots, and evidence make later worker extraction possible without changing the API contract.

Tech Stack: Flask, SQLAlchemy, PostgreSQL JSONB, Neo4j-backed data-source definitions, Vue 2, Vuetify, pytest.

Global Constraints

• Only PostgreSQL and MySQL database catalog collection are in WP-03.
• Production sources use the existing read-only metadata_collection connection purpose.
• Credentials and connection strings must never enter ingestion source config, job parameters, snapshots, evidence, API responses, or logs.
• A repeated canonical create reuses the existing job; a deliberate new collection uses force_rerun.
• A retry preserves prior attempts and starts a new numbered attempt.
• Real enterprise connectivity remains an external acceptance gate until the customer supplies network access, read-only accounts, scope, and data dictionaries.
• Work continues on codex/dataops-phase1-equipment-governance; no push or deployment is authorized.

---

Task 1: Database Source Registration

Files:

• Create: app/core/data_research/sources.py
• Modify: app/core/data_research/repository.py
• Modify: app/api/data_development/routes.py
• Test: tests/data_research/test_database_source_registration.py
• Test: tests/data_research/test_development_api.py

Interfaces:

• Consumes: DataSourceConnectionManager.definitions.get(uid) and IngestionSource.

• Produces: DatabaseSourceRegistrationService.ensure(data_source_uid, actor_uid).

• [x] Step 1: Write failing tests

Cover successful PostgreSQL/MySQL registration, idempotent refresh, disabled or unsupported definitions, secret-free stored config, and automatic registration before a catalog_collect job is created.

• Step 2: Verify RED

Run:

PYTHONPATH=. .venv/bin/pytest -q tests/data_research/test_database_source_registration.py tests/data_research/test_development_api.py

Expected: failure because the registration service and route integration do not exist.

• Step 3: Implement the minimal registration boundary

Store only database type, database name, and default schema. Reuse the external data-source UID as the ingestion source UID so jobs, connection pools, audit records, and evidence share one stable identifier.

• Step 4: Verify GREEN

Run the Task 1 test command and expect all tests to pass.

Task 2: Attempt-Aware Catalog Execution and Evidence

Files:

• Create: app/core/data_research/catalog/execution.py
• Modify: app/core/data_research/catalog/models.py
• Modify: app/core/data_research/ingestion.py
• Modify: app/core/data_research/models.py
• Modify: app/core/data_research/repository.py
• Modify: app/core/data_research/artifacts.py
• Modify: app/models/data_research.py
• Create: migrations/versions/20260729_280_catalog_ingestion_execution.py
• Test: tests/data_research/test_catalog_execution.py
• Test: tests/data_research/test_ingestion_service.py
• Test: tests/data_research/test_ingestion_models.py
• Test: tests/test_database_migrations.py

Interfaces:

• Consumes: CatalogCollectionService.collect(data_source_uid, CatalogScope) and IngestionService.transition.

• Produces: CatalogIngestionExecutor.execute(job_uid), CatalogSnapshotRecord, and SqlAlchemyCatalogSnapshotRepository.

• [x] Step 1: Write failing tests

Cover attempt incrementing, failure-stage capture, sanitized errors, valid scope parsing, successful state progression, immutable per-attempt snapshots, field evidence locators, failed collection, retry, and same-attempt resume.

• Step 2: Verify RED

Run:

PYTHONPATH=. .venv/bin/pytest -q tests/data_research/test_catalog_execution.py tests/data_research/test_ingestion_service.py tests/data_research/test_ingestion_models.py tests/test_database_migrations.py

Expected: failure because execution, snapshot persistence, and attempt fields do not exist.

• Step 3: Implement the minimal execution path

Move jobs through queued → extracting → normalizing → matching → awaiting_review, increment attempts at extraction start, persist one catalog snapshot per job attempt, and persist one evidence fragment per collected field. On failure, persist the sanitized diagnostic and the stage that failed before returning an error.

• Step 4: Verify GREEN

Run the Task 2 test command and expect all tests to pass.

Task 3: Management API and Operator UI

Files:

• Modify: app/api/data_development/routes.py
• Modify: app/core/system/permissions.py
• Modify: frontend/src/api/dataDevelopment.js
• Modify: frontend/src/views/dataGovernance/development/ingestion.vue
• Modify: frontend/src/views/dataGovernance/development/tasks.vue
• Test: tests/data_research/test_development_api.py
• Test: tests/data_research/test_development_frontend_contract.py
• Test: tests/test_permission_matrix.py

Interfaces:

• Produces: POST /api/development/v1/ingestion-jobs/{job_uid}/execute, GET /api/development/v1/ingestion-jobs/{job_uid}/catalog-snapshots, and GET /api/development/v1/ingestion-jobs/{job_uid}/evidence.

• [x] Step 1: Write failing tests

Cover permission classification, execution response, snapshot/evidence responses, secret-free serialization, selectable configured data sources, deliberate rerun, attempt/failure diagnostics, and retry controls.

• Step 2: Verify RED

Run:

PYTHONPATH=. .venv/bin/pytest -q tests/data_research/test_development_api.py tests/data_research/test_development_frontend_contract.py tests/test_permission_matrix.py

Expected: failure because the endpoints and UI contract do not exist.

• Step 3: Implement API and UI

Use the existing configured data-source list rather than requiring users to type a UID. Execute a newly created database catalog job explicitly, expose progress and diagnostic fields in the task list, and keep evidence/snapshot reads available to viewers.

• Step 4: Verify GREEN

Run the Task 3 test command and expect all tests to pass.

Task 4: Contracts, Status Ledger, and Release Gate

Files:

• Modify: docs/architecture/OPENAPI.yaml
• Modify: docs/architecture/DATA_MODEL.md
• Modify: docs/FUNCTION_MODULE_CENSUS_20260726.md
• Modify: docs/DATAOPS_PHASE1_3_MONTH_WORK_PLAN_20260729.md

Interfaces:

• Produces: generated API inventory, current data model, and an evidence-backed WP-03 status that separates engineering completion from enterprise connectivity acceptance.

• [x] Step 1: Regenerate and verify contracts

Run:

.venv/bin/python scripts/generate_openapi.py
PYTHONPATH=. .venv/bin/pytest -q tests/test_architecture_artifacts.py

• Step 2: Update status documents

Record engineering evidence and leave the two real enterprise source connections as externally blocked until customer inputs are supplied.

• Step 3: Run complete verification

Run the focused integration tests, full backend suite, frontend production build, migration upgrade, browser flow, and git diff --check.

Verification evidence:

• Backend: 860 passed, 41 skipped, and 59 subtests passed.
• Real local database integration: PostgreSQL and MySQL both passed.
• Frontend: production build completed with zero errors; existing dependency, bundle-size, CSS ordering, and console-statement warnings remain.
• Migration: local Docker database is at 20260729_280 (head).
• Browser: configured PostgreSQL source produced one table, two field evidence rows, and an awaiting_review task with attempt count 1; browser console reported zero errors and zero warnings.
• Release subset: every WP-03 backend file copied into deployment/app matches the canonical file under app; the historical full release-copy reconciliation remains WP-13 scope.

• Hygiene: git diff --check completed without findings.

• [x] Step 4: Commit

Create one independently reversible WP-03 engineering commit. Do not push.