# 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. - [x] **Step 2: Verify RED** Run: ```bash 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. - [x] **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. - [x] **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. - [x] **Step 2: Verify RED** Run: ```bash 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. - [x] **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. - [x] **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. - [x] **Step 2: Verify RED** Run: ```bash 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. - [x] **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. - [x] **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: ```bash .venv/bin/python scripts/generate_openapi.py PYTHONPATH=. .venv/bin/pytest -q tests/test_architecture_artifacts.py ``` - [x] **Step 2: Update status documents** Record engineering evidence and leave the two real enterprise source connections as externally blocked until customer inputs are supplied. - [x] **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.