DataOps-platform/docs/superpowers/plans/2026-07-16-platform-contract-and-consistency.md

Platform Contract and Consistency Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.

Goal: Establish versioned database migrations, stable business identifiers, API contract checks and a reusable outbox before adding cross-store features.

Architecture: PostgreSQL stores migration history and transactional outbox records. Domain commands write local state and events in one transaction; independent idempotent consumers update Neo4j, n8n or the knowledge index. OpenAPI route inventory is regenerated in CI and reviewed as a contract diff.

Tech Stack: Flask, SQLAlchemy, PostgreSQL 16, Alembic, pytest, Neo4j, Docker Compose.

---

Task 1: Introduce an Alembic migration chain

Files:

• Modify: requirements.txt
• Create: alembic.ini
• Create: migrations/env.py
• Create: migrations/versions/20260716_01_baseline.py

• Create: tests/test_database_migrations.py

• [ ] Write a test that creates an empty PostgreSQL test database, runs alembic upgrade head twice and asserts the six current tables plus alembic_version exist.

• [ ] Add Alembic and configure env.py to read SQLALCHEMY_DATABASE_URI; never embed credentials.

• [ ] Build a stamped baseline that represents current database/*.sql without dropping pre-existing production objects.

• [ ] Verify clean install, idempotent re-run and alembic downgrade -1 behavior in a disposable database.

Task 2: Add stable governance object identifiers

Files:

• Create: migrations/versions/20260716_02_governance_uids.py
• Create: app/core/common/identifiers.py
• Modify: app/core/business_domain/business_domain.py
• Modify: app/core/meta_data/meta_data.py
• Modify: app/core/data_flow/dataflows.py

• Create: tests/test_governance_identifiers.py

• [ ] Test that new BusinessDomain, DataMeta and DataFlow objects receive UUIDv7-compatible string IDs and retries preserve the same ID.

• [ ] Add uid uniqueness constraints/indexes to Neo4j initialization and backfill logic that never uses id(n) as the durable external key.

• [ ] Keep legacy numeric IDs readable during migration, but return uid in all new response contracts.

• [ ] Add a reconciliation command that reports objects without uid before enforcing required IDs.

Task 3: Implement the transactional outbox

Files:

• Create: migrations/versions/20260716_03_outbox.py
• Create: app/core/events/outbox.py
• Create: app/core/events/consumer.py
• Create: app/commands/process_outbox.py

• Create: tests/test_outbox.py

• [ ] Test atomic command/event persistence, duplicate delivery, retry backoff, dead-letter transition and consumer restart.

• [ ] Create outbox_events(event_id UUID, aggregate_type, aggregate_id, event_type, payload JSONB, status, attempts, available_at, published_at, last_error) with unique event_id.

• [ ] Implement SELECT ... FOR UPDATE SKIP LOCKED batching and handler idempotency storage.

• [ ] Expose backlog, oldest-event age and failed-event counts through internal health data without exposing payload secrets.

Task 4: Enforce the API inventory in CI

Files:

• Modify: scripts/generate_openapi.py
• Modify: tests/test_architecture_artifacts.py
• Create: .github/workflows/contracts.yml

• Modify: docs/architecture/OPENAPI.yaml

• [ ] Add a test that regenerates OpenAPI into a temporary path and byte-compares it with the committed contract.

• [ ] Fail CI when a Flask route changes without regenerating the contract.

• [ ] Add operation-level request/response schemas incrementally, beginning with system auth, workflow versioning and workbench endpoints.

• [ ] Require contract review for removed or incompatible operations.

Task 5: Prove cross-store failure behavior

Files:

• Create: tests/integration/test_cross_store_consistency.py
• Modify: deploy/docker/docker-compose.yml

• Modify: deploy/docker/README.md

• [ ] Run integration tests against the local stack with Neo4j/n8n temporarily unavailable after PostgreSQL commit.

• [ ] Assert the API reports accepted/processing, the event remains retryable, duplicate delivery causes no duplicate object, and recovery drains the backlog.

• [ ] Document operator commands for retry, dead-letter inspection and reconciliation.

• [ ] Run the complete test/build/smoke suite before enabling dependent feature plans.