docs: plan datasource connection pool implementation

docs/superpowers/plans/2026-07-18-datasource-connection-pool.md

@@ -0,0 +1,1782 @@
+# Data Source Connection Pool 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 a secure, lazy, per-data-source connection pool framework for PostgreSQL and MySQL while keeping the platform PostgreSQL connection path separate.
+
+**Architecture:** Flask-SQLAlchemy continues to own only the platform control database. External `DataSource` definitions are resolved through a `DataSourceConnectionManager`, which loads encrypted credentials from PostgreSQL, selects an allowlisted adapter, and acquires a connection from a worker-local lazy pool registry. Credential/configuration version changes invalidate old pools; per-source circuit breakers and diagnostics isolate failures.
+
+**Tech Stack:** Flask 2.3, Flask-SQLAlchemy 3.1, SQLAlchemy 2.0 `QueuePool`/`NullPool`, PostgreSQL, MySQL 8, Neo4j, AES-256-GCM from `cryptography`, psycopg2, PyMySQL, Alembic, pytest, Docker Compose.
+
+---
+
+## File and responsibility map
+
+| File | Responsibility |
+|---|---|
+| `app/core/data_source/errors.py` | Stable data-source error classes and safe API codes |
+| `app/core/data_source/redaction.py` | Recursive request and exception redaction |
+| `app/core/data_source/models.py` | Immutable definitions, credentials and pool-status value objects |
+| `app/core/data_source/credentials.py` | AES-GCM codec and PostgreSQL credential repository |
+| `app/core/data_source/definitions.py` | Neo4j `DataSource` definition repository without secrets |
+| `app/core/data_source/adapters/base.py` | Adapter protocol and allowlisted common options |
+| `app/core/data_source/adapters/postgresql.py` | PostgreSQL Engine and connection probe |
+| `app/core/data_source/adapters/mysql.py` | MySQL Engine and connection probe |
+| `app/core/data_source/circuit_breaker.py` | Per-source closed/open/half-open state machine |
+| `app/core/data_source/pool_registry.py` | Worker-local Engine registry, leases, draining and idle eviction |
+| `app/core/data_source/manager.py` | Public `connect(uid, purpose)` context manager |
+| `app/core/data_source/runtime.py` | One manager instance per Gunicorn worker |
+| `app/core/data_source/service.py` | Save/update/delete/test orchestration across PostgreSQL and Neo4j |
+| `app/api/data_source/routes.py` | HTTP boundary; no direct Engine or credential access |
+| `app/commands/report_datasource_credentials.py` | Read-only legacy credential audit |
+| `app/commands/migrate_datasource_credentials.py` | Explicit encryption/backfill and plaintext removal |
+| `app/commands/reconcile_datasource_credentials.py` | Cross-store orphan/stale reference report and repair |
+| `migrations/versions/20260718_50_datasource_credentials.py` | Encrypted credential and audit schema |
+| `tests/core/data_source/` | Unit and contract tests |
+| `tests/integration/test_datasource_pools.py` | PostgreSQL/MySQL pool integration |
+| `tests/integration/test_datasource_pool_failures.py` | Outage, recovery and invalidation acceptance |
+| `frontend/src/views/dataGovernance/dataSource/` | Secret-free PostgreSQL/MySQL management UI |
+
+## Task 1: Add dependencies and environment contract
+
+**Files:**
+- Modify: `requirements.txt`
+- Modify: `app/config/config.py`
+- Modify: `deploy/docker/.env.example`
+- Modify: `deployment/dataops.env`
+- Test: `tests/core/data_source/test_config.py`
+
+- [ ] **Step 1: Write the failing configuration test**
+
+```python
+def test_datasource_pool_defaults_are_bounded(monkeypatch):
+    monkeypatch.delenv("DATASOURCE_POOL_SIZE", raising=False)
+    monkeypatch.delenv("DATASOURCE_MAX_OVERFLOW", raising=False)
+    from app.config.config import BaseConfig
+
+    assert BaseConfig.DATASOURCE_POOL_SIZE == 2
+    assert BaseConfig.DATASOURCE_MAX_OVERFLOW == 3
+    assert BaseConfig.DATASOURCE_POOL_TIMEOUT == 10
+    assert BaseConfig.DATASOURCE_POOL_RECYCLE == 1800
+    assert BaseConfig.DATASOURCE_POOL_IDLE_TTL == 900
+    assert BaseConfig.DATASOURCE_MAX_IDLE_POOLS == 20
+    assert BaseConfig.DATASOURCE_QUERY_TIMEOUT == 30
+
+
+def test_pool_limits_reject_unsafe_values(monkeypatch):
+    monkeypatch.setenv("DATASOURCE_POOL_SIZE", "100")
+    from app.config.config import datasource_pool_settings
+
+    with pytest.raises(ValueError, match="DATASOURCE_POOL_SIZE"):
+        datasource_pool_settings()
+
+
+def test_datasource_override_is_clamped_to_server_limits():
+    settings = datasource_pool_settings(
+        {"pool_size": 4, "max_overflow": 5}
+    )
+    assert settings["pool_size"] == 4
+    assert settings["max_overflow"] == 5
+
+    with pytest.raises(ValueError, match="pool_size"):
+        datasource_pool_settings({"pool_size": 11})
+```
+
+- [ ] **Step 2: Run the test and confirm the red state**
+
+Run:
+
+```bash
+.venv/bin/python -m pytest tests/core/data_source/test_config.py -q
+```
+
+Expected: FAIL because the pool settings do not exist.
+
+- [ ] **Step 3: Pin the required drivers**
+
+Add:
+
+```text
+cryptography==43.0.3
+PyMySQL==1.1.1
+```
+
+- [ ] **Step 4: Add bounded environment parsing**
+
+Add to `app/config/config.py`:
+
+```python
+def _bounded_int(name: str, default: int, minimum: int, maximum: int) -> int:
+    value = int(os.environ.get(name, str(default)))
+    if value < minimum or value > maximum:
+        raise ValueError(f"{name} must be between {minimum} and {maximum}")
+    return value
+
+
+def datasource_pool_settings(overrides: Mapping[str, int] | None = None) -> dict:
+    settings = {
+        "pool_size": _bounded_int("DATASOURCE_POOL_SIZE", 2, 1, 10),
+        "max_overflow": _bounded_int("DATASOURCE_MAX_OVERFLOW", 3, 0, 10),
+        "pool_timeout": _bounded_int("DATASOURCE_POOL_TIMEOUT", 10, 1, 60),
+        "pool_recycle": _bounded_int("DATASOURCE_POOL_RECYCLE", 1800, 60, 86400),
+        "idle_ttl": _bounded_int("DATASOURCE_POOL_IDLE_TTL", 900, 60, 86400),
+        "max_idle_pools": _bounded_int("DATASOURCE_MAX_IDLE_POOLS", 20, 1, 100),
+        "query_timeout": _bounded_int("DATASOURCE_QUERY_TIMEOUT", 30, 1, 300),
+    }
+    return apply_allowlisted_pool_overrides(settings, overrides or {})
+```
+
+Expose the same values on `BaseConfig`, plus:
+
+```python
+DATASOURCE_CREDENTIAL_MASTER_KEY = os.environ.get(
+    "DATASOURCE_CREDENTIAL_MASTER_KEY", ""
+)
+DATASOURCE_CREDENTIAL_KEY_VERSION = os.environ.get(
+    "DATASOURCE_CREDENTIAL_KEY_VERSION", "v1"
+)
+DATASOURCE_CERT_DIR = os.environ.get(
+    "DATASOURCE_CERT_DIR", "/etc/dataops-platform/datasource-certs"
+)
+```
+
+Only `pool_size` and `max_overflow` may be overridden per data source in phase
+one. Reject unknown keys and values outside the same server-side bounds; never
+accept `pool_timeout`, driver names or raw `connect_args` from an API request.
+
+- [ ] **Step 5: Document environment variables without real secrets**
+
+Add template values:
+
+```text
+DATASOURCE_CREDENTIAL_MASTER_KEY=replace-with-base64-32-byte-key
+DATASOURCE_CREDENTIAL_KEY_VERSION=v1
+DATASOURCE_POOL_SIZE=2
+DATASOURCE_MAX_OVERFLOW=3
+DATASOURCE_POOL_TIMEOUT=10
+DATASOURCE_POOL_RECYCLE=1800
+DATASOURCE_POOL_IDLE_TTL=900
+DATASOURCE_MAX_IDLE_POOLS=20
+DATASOURCE_QUERY_TIMEOUT=30
+DATASOURCE_CERT_DIR=/etc/dataops-platform/datasource-certs
+```
+
+- [ ] **Step 6: Run tests**
+
+Run:
+
+```bash
+.venv/bin/python -m pytest tests/core/data_source/test_config.py -q
+```
+
+Expected: PASS.
+
+- [ ] **Step 7: Commit**
+
+```bash
+git add requirements.txt app/config/config.py deploy/docker/.env.example \
+  deployment/dataops.env tests/core/data_source/test_config.py
+git commit -m "feat: define external datasource pool settings"
+```
+
+## Task 2: Create the encrypted credential schema
+
+**Files:**
+- Create: `migrations/versions/20260718_50_datasource_credentials.py`
+- Create: `tests/core/data_source/test_credential_schema.py`
+- Modify: `tests/test_database_migrations.py`
+
+- [ ] **Step 1: Write the failing migration contract test**
+
+```python
+def test_datasource_credential_migration_is_encrypted_and_versioned():
+    source = Path(
+        "migrations/versions/20260718_50_datasource_credentials.py"
+    ).read_text(encoding="utf-8")
+
+    assert "datasource_credentials" in source
+    assert "encrypted_payload" in source
+    assert "nonce" in source
+    assert "key_version" in source
+    assert "credential_version" in source
+    assert "UNIQUE (data_source_uid, credential_version)" in source
+    assert "password VARCHAR" not in source
+```
+
+- [ ] **Step 2: Run the test and confirm failure**
+
+```bash
+.venv/bin/python -m pytest \
+  tests/core/data_source/test_credential_schema.py -q
+```
+
+Expected: FAIL because the migration does not exist.
+
+- [ ] **Step 3: Add the non-destructive migration**
+
+Use `down_revision = "20260716_40"` and create:
+
+```sql
+CREATE TABLE IF NOT EXISTS public.datasource_credentials (
+    id UUID PRIMARY KEY,
+    data_source_uid UUID NOT NULL,
+    credential_version INTEGER NOT NULL CHECK (credential_version > 0),
+    encrypted_payload BYTEA NOT NULL,
+    nonce BYTEA NOT NULL CHECK (octet_length(nonce) = 12),
+    key_version VARCHAR(40) NOT NULL,
+    status VARCHAR(20) NOT NULL DEFAULT 'active'
+        CHECK (status IN ('active', 'retired', 'revoked')),
+    created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    retired_at TIMESTAMPTZ,
+    UNIQUE (data_source_uid, credential_version)
+);
+
+CREATE UNIQUE INDEX IF NOT EXISTS uq_datasource_credential_active
+ON public.datasource_credentials(data_source_uid)
+WHERE status = 'active';
+```
+
+Also create `datasource_credential_audit_events` with event type, UID, version,
+actor UID, success flag, safe detail and timestamp. Do not create password,
+connection URL or raw request columns.
+
+Keep `downgrade()` data-preserving, matching the current migration policy.
+
+- [ ] **Step 4: Extend disposable-database migration verification**
+
+Add `datasource_credentials` and `datasource_credential_audit_events` to the
+post-upgrade expected table set, not to the original baseline migration set.
+
+- [ ] **Step 5: Run schema tests**
+
+```bash
+TEST_POSTGRES_ADMIN_URL=postgresql://dataops:dataops-test-password@localhost:15432/postgres \
+  .venv/bin/python -m pytest \
+  tests/core/data_source/test_credential_schema.py \
+  tests/test_database_migrations.py -q
+```
+
+Expected: all tests PASS; temporary database is dropped.
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add migrations/versions/20260718_50_datasource_credentials.py \
+  tests/core/data_source/test_credential_schema.py \
+  tests/test_database_migrations.py
+git commit -m "feat: add encrypted datasource credentials schema"
+```
+
+## Task 3: Implement AES-GCM credential handling
+
+**Files:**
+- Create: `app/core/data_source/__init__.py`
+- Create: `app/core/data_source/errors.py`
+- Create: `app/core/data_source/models.py`
+- Create: `app/core/data_source/credentials.py`
+- Test: `tests/core/data_source/test_credentials.py`
+
+- [ ] **Step 1: Write failing codec tests**
+
+```python
+def test_credential_round_trip_and_tamper_detection():
+    codec = CredentialCodec(
+        master_key=b"k" * 32,
+        key_version="v1",
+    )
+    sealed = codec.encrypt(
+        data_source_uid="01900000-0000-7000-8000-000000000010",
+        credential_version=1,
+        credential=DataSourceCredential("reader", "secret"),
+    )
+
+    assert b"secret" not in sealed.encrypted_payload
+    assert codec.decrypt(sealed) == DataSourceCredential("reader", "secret")
+
+    tampered = dataclasses.replace(
+        sealed,
+        encrypted_payload=sealed.encrypted_payload[:-1] + b"x",
+    )
+    with pytest.raises(DataSourceCredentialUnavailable):
+        codec.decrypt(tampered)
+
+
+def test_credential_cannot_be_replayed_for_another_datasource():
+    sealed = codec.encrypt(DATA_SOURCE_A, 1, credential)
+    replayed = dataclasses.replace(sealed, data_source_uid=DATA_SOURCE_B)
+
+    with pytest.raises(DataSourceCredentialUnavailable):
+        codec.decrypt(replayed)
+```
+
+- [ ] **Step 2: Verify tests fail**
+
+```bash
+.venv/bin/python -m pytest \
+  tests/core/data_source/test_credentials.py -q
+```
+
+Expected: FAIL with missing modules.
+
+- [ ] **Step 3: Define immutable value objects and stable errors**
+
+```python
+@dataclass(frozen=True)
+class DataSourceCredential:
+    username: str
+    password: str
+    options: Mapping[str, str] = field(default_factory=dict)
+
+
+@dataclass(frozen=True)
+class SealedCredential:
+    id: str
+    data_source_uid: str
+    credential_version: int
+    encrypted_payload: bytes
+    nonce: bytes
+    key_version: str
+    status: str = "active"
+```
+
+Define `DataSourceError` subclasses for every stable phase-one code:
+
+```python
+class DataSourceCredentialUnavailable(DataSourceError):
+    code = "DATASOURCE_CREDENTIAL_UNAVAILABLE"
+    http_status = 503
+```
+
+The complete set is `DATASOURCE_NOT_FOUND`,
+`DATASOURCE_TYPE_UNSUPPORTED`, `DATASOURCE_CONFIGURATION_INVALID`,
+`DATASOURCE_CREDENTIAL_UNAVAILABLE`, `DATASOURCE_CONNECTION_FAILED`,
+`DATASOURCE_POOL_TIMEOUT`, `DATASOURCE_CIRCUIT_OPEN`,
+`DATASOURCE_QUERY_TIMEOUT`, and `DATASOURCE_READ_ONLY_VIOLATION`. Safe public
+messages must not include DBAPI exception text.
+
+- [ ] **Step 4: Implement AES-256-GCM**
+
+Use:
+
+```python
+aad = f"{data_source_uid}:{credential_version}".encode("utf-8")
+nonce = os.urandom(12)
+payload = json.dumps(
+    {
+        "username": credential.username,
+        "password": credential.password,
+        "options": dict(credential.options),
+    },
+    sort_keys=True,
+    separators=(",", ":"),
+).encode("utf-8")
+ciphertext = AESGCM(master_key).encrypt(nonce, payload, aad)
+```
+
+Decode the environment key with `base64.urlsafe_b64decode`; require exactly
+32 bytes. Convert `InvalidTag`, malformed JSON and key mismatch to
+`DataSourceCredentialUnavailable("credential cannot be decrypted")`.
+
+- [ ] **Step 5: Add repository create/get/retire/revoke methods**
+
+Repository methods accept a SQLAlchemy Session and use parameterized `text()`.
+`create_version()` takes an advisory transaction lock on `data_source_uid`,
+increments the version and retires the prior active row in the same transaction.
+
+- [ ] **Step 6: Run tests**
+
+```bash
+.venv/bin/python -m pytest \
+  tests/core/data_source/test_credentials.py -q
+```
+
+Expected: PASS.
+
+- [ ] **Step 7: Commit**
+
+```bash
+git add app/core/data_source tests/core/data_source/test_credentials.py
+git commit -m "feat: encrypt external datasource credentials"
+```
+
+## Task 4: Remove secrets from logs and API serialization
+
+**Files:**
+- Create: `app/core/data_source/redaction.py`
+- Modify: `app/api/data_source/routes.py`
+- Test: `tests/core/data_source/test_redaction.py`
+- Test: `tests/test_datasource_api_security.py`
+
+- [ ] **Step 1: Write failing redaction tests**
+
+```python
+def test_recursive_redaction_covers_connection_secrets():
+    value = {
+        "name": "warehouse",
+        "password": "plain",
+        "options": {
+            "api_key": "key",
+            "url": "postgresql://user:secret@db:5432/app",
+        },
+    }
+
+    assert redact_mapping(value) == {
+        "name": "warehouse",
+        "password": "[redacted]",
+        "options": {
+            "api_key": "[redacted]",
+            "url": "postgresql://[redacted]@db:5432/app",
+        },
+    }
+
+
+def test_datasource_list_never_serializes_credentials(client, viewer_token):
+    response = client.post(
+        "/api/datasource/list",
+        json={},
+        headers=bearer(viewer_token),
+    )
+    text = response.get_data(as_text=True).lower()
+
+    assert "password" not in text
+    assert "encrypted_payload" not in text
+    assert "credential_configured" in text
+```
+
+- [ ] **Step 2: Verify tests fail**
+
+```bash
+.venv/bin/python -m pytest \
+  tests/core/data_source/test_redaction.py \
+  tests/test_datasource_api_security.py -q
+```
+
+Expected: FAIL because request logging and serialization expose sensitive keys.
+
+- [ ] **Step 3: Implement recursive redaction**
+
+Redact case-insensitive keys:
+
+```python
+SENSITIVE_KEYS = {
+    "password",
+    "passwd",
+    "credential",
+    "credentials",
+    "api_key",
+    "token",
+    "authorization",
+    "encrypted_payload",
+    "nonce",
+    "conn_str",
+    "connection_string",
+}
+```
+
+Redact URL userinfo using `urllib.parse.urlsplit()` and return a URL containing
+`[redacted]@host`. Truncate sanitized exception messages to 1000 characters.
+
+- [ ] **Step 4: Remove raw request logging immediately**
+
+Replace every pattern like:
+
+```python
+logger.debug("保存数据源请求数据: %s", json.dumps(data))
+```
+
+with:
+
+```python
+logger.debug("保存数据源请求: %s", redact_mapping(data or {}))
+```
+
+The list response must strip `username`, `password`, `credential_ref`,
+`encrypted_payload` and `nonce`; expose only:
+
+```python
+item["credential_configured"] = bool(item.pop("credential_version", None))
+```
+
+- [ ] **Step 5: Run security tests and source scan**
+
+```bash
+.venv/bin/python -m pytest \
+  tests/core/data_source/test_redaction.py \
+  tests/test_datasource_api_security.py -q
+rg -n "json\\.dumps\\(data|log_data.*password|RETURN n.*password" \
+  app/api/data_source app/core/data_source
+```
+
+Expected: tests PASS and `rg` returns no unsafe logging pattern.
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add app/core/data_source/redaction.py app/api/data_source/routes.py \
+  tests/core/data_source/test_redaction.py \
+  tests/test_datasource_api_security.py
+git commit -m "fix: prevent datasource credential disclosure"
+```
+
+## Task 5: Implement the Neo4j definition repository
+
+**Files:**
+- Create: `app/core/data_source/definitions.py`
+- Test: `tests/core/data_source/test_definitions.py`
+
+- [ ] **Step 1: Write failing repository tests**
+
+```python
+def test_definition_repository_assigns_uid_and_excludes_secrets(fake_graph):
+    repository = DataSourceDefinitionRepository(fake_graph)
+    saved = repository.save(
+        DataSourceDefinition(
+            uid=None,
+            name_en="warehouse",
+            database_type="postgresql",
+            host="source-postgres",
+            port=5432,
+            database="analytics",
+            schema="public",
+            credential_ref="01900000-0000-7000-8000-000000000011",
+            credential_version=1,
+        )
+    )
+
+    assert uuid.UUID(saved.uid).version == 7
+    assert "password" not in fake_graph.last_parameters["properties"]
+    assert "username" not in fake_graph.last_parameters["properties"]
+```
+
+- [ ] **Step 2: Verify failure**
+
+```bash
+.venv/bin/python -m pytest \
+  tests/core/data_source/test_definitions.py -q
+```
+
+Expected: FAIL because the repository does not exist.
+
+- [ ] **Step 3: Implement a parameterized repository**
+
+Support:
+
+```python
+get(uid: str) -> DataSourceDefinition | None
+list(filters: DataSourceFilters) -> list[DataSourceDefinition]
+save(definition: DataSourceDefinition) -> DataSourceDefinition
+delete(uid: str) -> bool
+credential_references() -> dict[str, tuple[str, int]]
+remove_legacy_secret_fields(uid: str) -> None
+```
+
+Use `uid`, not Neo4j internal numeric IDs, for new writes and updates. Keep a
+temporary numeric-ID lookup only inside the migration command.
+
+- [ ] **Step 4: Add uniqueness constraint reconciliation**
+
+Extend `app/commands/reconcile_governance_uids.py` with:
+
+```cypher
+CREATE CONSTRAINT datasource_uid_unique IF NOT EXISTS
+FOR (node:DataSource) REQUIRE node.uid IS UNIQUE
+```
+
+- [ ] **Step 5: Run tests**
+
+```bash
+.venv/bin/python -m pytest \
+  tests/core/data_source/test_definitions.py \
+  tests/test_governance_identifiers.py -q
+```
+
+Expected: PASS.
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add app/core/data_source/definitions.py \
+  app/commands/reconcile_governance_uids.py \
+  tests/core/data_source/test_definitions.py \
+  tests/test_governance_identifiers.py
+git commit -m "feat: add secret-free datasource definitions"
+```
+
+## Task 6: Implement PostgreSQL and MySQL adapters
+
+**Files:**
+- Create: `app/core/data_source/adapters/__init__.py`
+- Create: `app/core/data_source/adapters/base.py`
+- Create: `app/core/data_source/adapters/postgresql.py`
+- Create: `app/core/data_source/adapters/mysql.py`
+- Test: `tests/core/data_source/test_adapters.py`
+
+- [ ] **Step 1: Write failing adapter tests**
+
+```python
+@pytest.mark.parametrize(
+    ("database_type", "drivername"),
+    [
+        ("postgresql", "postgresql+psycopg2"),
+        ("mysql", "mysql+pymysql"),
+    ],
+)
+def test_adapter_uses_allowlisted_driver(database_type, drivername):
+    adapter = adapter_for(database_type)
+    url = adapter.build_url(definition, credential)
+
+    assert url.drivername == drivername
+    assert url.password == "secret"
+
+
+def test_unknown_driver_is_rejected():
+    with pytest.raises(DataSourceTypeUnsupported):
+        adapter_for("sqlite")
+
+
+def test_adapter_rejects_arbitrary_connection_options():
+    with pytest.raises(DataSourceConfigurationInvalid):
+        adapter.validate_options({"driver": "custom", "connect_args": {}})
+
+
+def test_read_purpose_configures_database_read_only_transaction():
+    adapter.configure_transaction(connection, purpose="metadata_preview")
+    assert connection.executed == ["SET TRANSACTION READ ONLY"]
+
+
+def test_connection_test_uses_null_pool_and_disposes(monkeypatch):
+    engine = FakeEngine()
+    monkeypatch.setattr(base, "create_engine", lambda *a, **kw: engine)
+
+    adapter.test_connection(definition, credential)
+
+    assert engine.pool_class is NullPool
+    assert engine.disposed is True
+```
+
+- [ ] **Step 2: Verify failure**
+
+```bash
+.venv/bin/python -m pytest \
+  tests/core/data_source/test_adapters.py -q
+```
+
+Expected: FAIL with missing adapter modules.
+
+- [ ] **Step 3: Define the adapter protocol**
+
+```python
+class DataSourceAdapter(Protocol):
+    database_type: str
+
+    def build_url(
+        self,
+        definition: DataSourceDefinition,
+        credential: DataSourceCredential,
+    ) -> URL: ...
+
+    def connect_args(
+        self,
+        definition: DataSourceDefinition,
+        credential: DataSourceCredential,
+    ) -> dict[str, object]: ...
+
+    def validate_options(self, options: Mapping[str, object]) -> None: ...
+
+    def configure_transaction(
+        self,
+        connection: Connection,
+        purpose: str,
+    ) -> None: ...
+
+    def probe(self, connection: Connection) -> None: ...
+```
+
+- [ ] **Step 4: Implement the allowlisted adapters**
+
+PostgreSQL returns:
+
+```python
+{
+    "connect_timeout": 5,
+    "keepalives": 1,
+    "keepalives_idle": 30,
+    "keepalives_interval": 10,
+    "keepalives_count": 3,
+    "application_name": "dataops-platform-datasource",
+    "options": "-c statement_timeout=30000",
+}
+```
+
+MySQL returns:
+
+```python
+{
+    "connect_timeout": 5,
+    "read_timeout": 30,
+    "write_timeout": 30,
+    "charset": "utf8mb4",
+}
+```
+
+Both execute `text("SELECT 1")` in `probe()`.
+
+Build the PostgreSQL `statement_timeout` value from the validated
+`query_timeout`; keep it platform-controlled. MySQL uses the same bounded value
+for `read_timeout` and `write_timeout`.
+
+Allow only documented TLS settings: PostgreSQL `sslmode`, `sslrootcert`,
+`sslcert`, and `sslkey`; MySQL `ssl_ca`, `ssl_cert`, and `ssl_key`. File paths
+must resolve under the configured certificate directory. Reject unknown option
+keys, URL fragments, frontend driver names, and raw `connect_args`.
+
+For every purpose except `dataflow_write`, configure the database transaction
+as read-only before yielding the connection. PostgreSQL executes
+`SET TRANSACTION READ ONLY`; MySQL executes the equivalent supported by MySQL
+8. Map a database read-only violation to
+`DATASOURCE_READ_ONLY_VIOLATION`. Never retry or replay write transactions.
+
+- [ ] **Step 5: Implement temporary connection testing**
+
+```python
+engine = create_engine(
+    adapter.build_url(definition, credential),
+    poolclass=NullPool,
+    connect_args=adapter.connect_args(definition, credential),
+)
+try:
+    with engine.connect() as connection:
+        adapter.probe(connection)
+finally:
+    engine.dispose()
+```
+
+- [ ] **Step 6: Run adapter tests**
+
+```bash
+.venv/bin/python -m pytest \
+  tests/core/data_source/test_adapters.py -q
+```
+
+Expected: PASS.
+
+- [ ] **Step 7: Commit**
+
+```bash
+git add app/core/data_source/adapters \
+  tests/core/data_source/test_adapters.py requirements.txt
+git commit -m "feat: add PostgreSQL and MySQL datasource adapters"
+```
+
+## Task 7: Build the per-source circuit breaker
+
+**Files:**
+- Create: `app/core/data_source/circuit_breaker.py`
+- Test: `tests/core/data_source/test_circuit_breaker.py`
+
+- [ ] **Step 1: Write failing state-machine tests**
+
+```python
+def test_breaker_opens_after_three_failures_and_recovers():
+    clock = FakeClock()
+    breaker = CircuitBreaker(
+        failure_threshold=3,
+        recovery_seconds=30,
+        clock=clock,
+    )
+
+    breaker.record_failure()
+    breaker.record_failure()
+    breaker.record_failure()
+    assert breaker.state == "open"
+
+    with pytest.raises(DataSourceCircuitOpen):
+        with breaker.probe_permission():
+            pass
+
+    clock.advance(30)
+    with breaker.probe_permission():
+        breaker.record_success()
+
+    assert breaker.state == "closed"
+```
+
+- [ ] **Step 2: Verify failure**
+
+```bash
+.venv/bin/python -m pytest \
+  tests/core/data_source/test_circuit_breaker.py -q
+```
+
+Expected: FAIL because `CircuitBreaker` is undefined.
+
+- [ ] **Step 3: Implement closed/open/half-open**
+
+Use a `threading.Lock`. Only one caller may own the half-open probe. A normal
+query failure does not open the breaker; only connection creation, checkout
+disconnect and probe failures count.
+
+- [ ] **Step 4: Run tests**
+
+```bash
+.venv/bin/python -m pytest \
+  tests/core/data_source/test_circuit_breaker.py -q
+```
+
+Expected: PASS, including concurrent half-open callers.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add app/core/data_source/circuit_breaker.py \
+  tests/core/data_source/test_circuit_breaker.py
+git commit -m "feat: isolate datasource failures with circuit breakers"
+```
+
+## Task 8: Build the worker-local pool registry
+
+**Files:**
+- Create: `app/core/data_source/pool_registry.py`
+- Test: `tests/core/data_source/test_pool_registry.py`
+
+- [ ] **Step 1: Write failing reuse, drain and eviction tests**
+
+```python
+def test_registry_reuses_same_engine_and_invalidates_new_version():
+    registry = PoolRegistry(settings)
+    first = PoolKey(DATA_SOURCE_UID, 1, "hash-a")
+    second = PoolKey(DATA_SOURCE_UID, 2, "hash-b")
+
+    with registry.lease(first, lambda: engine_factory(first)) as engine_a:
+        with registry.lease(first, lambda: engine_factory(first)) as engine_b:
+            assert engine_a is engine_b
+
+    with registry.lease(second, lambda: engine_factory(second)) as engine_c:
+        assert engine_c is not engine_a
+
+    assert engine_a.disposed is True
+
+
+def test_idle_pool_is_evicted_only_after_last_lease_returns():
+    with registry.lease(key, lambda: engine):
+        clock.advance(901)
+        registry.evict_idle()
+        assert engine.disposed is False
+
+    registry.evict_idle()
+    assert engine.disposed is True
+```
+
+- [ ] **Step 2: Verify failure**
+
+```bash
+.venv/bin/python -m pytest \
+  tests/core/data_source/test_pool_registry.py -q
+```
+
+Expected: FAIL because the registry does not exist.
+
+- [ ] **Step 3: Implement pool entries**
+
+```python
+@dataclass
+class PoolEntry:
+    key: PoolKey
+    engine: Engine
+    breaker: CircuitBreaker
+    created_at: float
+    last_used_at: float
+    leases: int = 0
+    draining: bool = False
+    checkout_wait_ms: float = 0.0
+```
+
+Protect the entry map with `threading.RLock`. Do not hold the registry lock
+while executing SQL. Create one Engine per key using:
+
+```python
+create_engine(
+    url,
+    pool_size=settings.pool_size,
+    max_overflow=settings.max_overflow,
+    pool_timeout=settings.pool_timeout,
+    pool_recycle=settings.pool_recycle,
+    pool_pre_ping=True,
+    pool_use_lifo=True,
+    pool_reset_on_return="rollback",
+    connect_args=connect_args,
+)
+```
+
+Resolve `settings` from the server defaults plus the validated per-source
+`pool_size` and `max_overflow` override. A lease for a new key must mark all
+older keys for the same UID as draining.
+
+- [ ] **Step 4: Add invalidation and bounded idle cache**
+
+Implement:
+
+```python
+invalidate(data_source_uid: str, reason: str) -> None
+expire_draining(now: float | None = None) -> int
+evict_idle(now: float | None = None) -> int
+close_all() -> None
+snapshot() -> list[PoolStatus]
+```
+
+When the idle-entry count exceeds `max_idle_pools`, dispose least-recently-used
+idle entries first. `invalidate()` immediately blocks new leases for the old
+key, lets existing leases drain for a bounded grace period, then
+`expire_draining()` force-disposes the Engine. Returning a lease after forced
+disposal must not re-register the old Engine.
+
+- [ ] **Step 5: Add SQLAlchemy event metrics**
+
+Track `connect`, `checkout`, `checkin` and `invalidate`. Store counts only;
+never inspect or log DBAPI connection arguments.
+
+- [ ] **Step 6: Run pool tests**
+
+```bash
+.venv/bin/python -m pytest \
+  tests/core/data_source/test_pool_registry.py -q
+```
+
+Expected: PASS, including a 50-thread lease/release test with no duplicate
+Engine for the same key and no leaked leases.
+
+- [ ] **Step 7: Commit**
+
+```bash
+git add app/core/data_source/pool_registry.py \
+  tests/core/data_source/test_pool_registry.py
+git commit -m "feat: add lazy worker-local datasource pool registry"
+```
+
+## Task 9: Implement the public connection manager
+
+**Files:**
+- Create: `app/core/data_source/manager.py`
+- Create: `app/core/data_source/runtime.py`
+- Test: `tests/core/data_source/test_manager.py`
+- Modify: `gunicorn_config.py`
+
+- [ ] **Step 1: Write failing manager tests**
+
+```python
+def test_manager_resolves_definition_credential_and_adapter():
+    with manager.connect(DATA_SOURCE_UID, purpose="metadata_preview") as connection:
+        connection.execute(text("SELECT 1"))
+
+    assert definitions.requested_uid == DATA_SOURCE_UID
+    assert credentials.requested_version == 4
+    assert adapters.requested_type == "postgresql"
+    assert registry.last_key.credential_version == 4
+
+
+def test_write_requires_explicit_write_purpose(manager, fake_engine):
+    with manager.connect(DATA_SOURCE_UID, purpose="metadata_preview"):
+        pass
+    read_transaction = fake_engine.transactions[-1]
+    assert read_transaction.rolled_back is True
+    assert read_transaction.committed is False
+
+    with manager.connect(DATA_SOURCE_UID, purpose="dataflow_write"):
+        pass
+    write_transaction = fake_engine.transactions[-1]
+    assert write_transaction.committed is True
+```
+
+- [ ] **Step 2: Verify failure**
+
+```bash
+.venv/bin/python -m pytest \
+  tests/core/data_source/test_manager.py -q
+```
+
+Expected: FAIL because `DataSourceConnectionManager` does not exist.
+
+- [ ] **Step 3: Implement the context manager**
+
+```python
+@contextmanager
+def connect(self, data_source_uid: str, purpose: str):
+    definition = self.definitions.get(data_source_uid)
+    if definition is None:
+        raise DataSourceNotFound(data_source_uid)
+    credential = self.credentials.get_active(
+        data_source_uid,
+        definition.credential_version,
+    )
+    adapter = adapter_for(definition.database_type)
+    key = PoolKey(
+        data_source_uid=data_source_uid,
+        credential_version=definition.credential_version,
+        config_fingerprint=definition.connection_fingerprint(),
+    )
+    with self.registry.lease(
+        key,
+        lambda: adapter.create_pooled_engine(definition, credential, self.settings),
+    ) as engine:
+        with engine.connect() as connection:
+            transaction = connection.begin()
+            try:
+                adapter.configure_transaction(connection, purpose)
+                yield ManagedConnection(connection, purpose)
+                if purpose == "dataflow_write":
+                    transaction.commit()
+                else:
+                    transaction.rollback()
+            except Exception:
+                transaction.rollback()
+                raise
+```
+
+`ManagedConnection.execute()` records query duration and translates known
+driver timeout and read-only exceptions to the stable safe errors. Measure
+checkout wait around `engine.connect()`. It must not retry statements or
+transactions.
+
+- [ ] **Step 4: Create one manager per Worker**
+
+`runtime.py` lazily constructs one manager in a process-local variable guarded
+by a lock. Add:
+
+```python
+def close_data_source_runtime() -> None:
+    manager = _manager
+    if manager is not None:
+        manager.close()
+```
+
+Call it from Gunicorn `worker_exit`, not from every Flask request.
+
+- [ ] **Step 5: Run manager tests**
+
+```bash
+.venv/bin/python -m pytest \
+  tests/core/data_source/test_manager.py -q
+```
+
+Expected: PASS.
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add app/core/data_source/manager.py app/core/data_source/runtime.py \
+  tests/core/data_source/test_manager.py gunicorn_config.py
+git commit -m "feat: expose managed datasource connections"
+```
+
+## Task 10: Orchestrate secure data-source lifecycle
+
+**Files:**
+- Create: `app/core/data_source/service.py`
+- Modify: `app/api/data_source/routes.py`
+- Modify: `app/core/system/permissions.py`
+- Test: `tests/test_datasource_lifecycle_api.py`
+
+- [ ] **Step 1: Write failing API lifecycle tests**
+
+```python
+def test_create_stores_secret_only_in_credential_store(
+    client, editor_token, fake_graph, credential_repository
+):
+    response = client.post(
+        "/api/datasource/save",
+        json=POSTGRES_SOURCE_PAYLOAD,
+        headers=bearer(editor_token),
+    )
+
+    assert response.status_code == 201
+    graph_properties = fake_graph.saved_properties
+    assert "password" not in graph_properties
+    assert "username" not in graph_properties
+    assert graph_properties["credential_version"] == 1
+    assert credential_repository.decrypt_latest().password == "source-pass"
+
+
+def test_update_invalidates_old_pool(client, editor_token, pool_registry):
+    response = client.post(
+        "/api/datasource/save",
+        json={**UPDATED_PAYLOAD, "uid": DATA_SOURCE_UID},
+        headers=bearer(editor_token),
+    )
+
+    assert response.status_code == 200
+    assert pool_registry.invalidations == [
+        (DATA_SOURCE_UID, "configuration_changed")
+    ]
+```
+
+- [ ] **Step 2: Verify failure**
+
+```bash
+.venv/bin/python -m pytest \
+  tests/test_datasource_lifecycle_api.py -q
+```
+
+Expected: FAIL because routes still write request properties directly to Neo4j.
+
+- [ ] **Step 3: Implement save/update orchestration**
+
+`DataSourceService.save()` must:
+
+1. Validate only `postgresql` and `mysql`.
+2. Test using Adapter + `NullPool`.
+3. Write an encrypted immutable credential version in a PostgreSQL transaction.
+4. Enqueue `datasource.credential_version_created` with safe definition fields.
+5. Save the secret-free Neo4j definition.
+6. Invalidate the old pool after the definition points to the new version.
+7. Return the sanitized definition.
+
+On Neo4j failure, retain the prior definition and record the newly created
+credential as an orphan candidate for reconciliation.
+
+For an existing source, omitted username/password means “reuse the active
+credential internally”: decrypt it only inside the service, combine any
+explicit replacement fields, and write the next immutable credential version.
+Never send the old credential back to the frontend. Per-source pool overrides
+and TLS fields pass through the server allowlists before connection testing.
+
+- [ ] **Step 4: Implement delete orchestration**
+
+Order:
+
+```python
+registry.invalidate(uid, reason="datasource_deleted")
+definitions.delete(uid)
+credentials.revoke_all(uid, actor_id)
+enqueue_outbox(..., event_type="datasource.deleted", payload={"uid": uid})
+```
+
+If PostgreSQL finalization fails, the reconciliation command must detect a
+deleted graph definition with active credentials.
+
+- [ ] **Step 5: Refactor HTTP routes**
+
+Routes call `DataSourceService` only. Remove imports of `create_engine` and
+`URL` from `app/api/data_source/routes.py`. Use real HTTP status codes.
+
+Keep the existing governance permissions for normal data-source operations:
+
+- viewer: `READ_GOVERNANCE` for list, detail and approved read use;
+- editor: viewer permissions plus `EDIT_GOVERNANCE` for save, update, delete,
+  connection test and dataflow use;
+- admin: editor permissions plus the new
+  `DATASOURCE_POOL_MANAGE = "datasources:pools:manage"` for pool diagnostics
+  and forced invalidation.
+
+Do not grant `DATASOURCE_POOL_MANAGE` to editor or viewer.
+
+- [ ] **Step 6: Run API tests**
+
+```bash
+.venv/bin/python -m pytest \
+  tests/test_datasource_lifecycle_api.py \
+  tests/test_permission_matrix.py \
+  tests/test_datasource_api_security.py -q
+```
+
+Expected: PASS.
+
+- [ ] **Step 7: Regenerate OpenAPI**
+
+```bash
+.venv/bin/python scripts/generate_openapi.py
+.venv/bin/python -m pytest tests/test_architecture_artifacts.py -q
+```
+
+Expected: generated contract matches route source.
+
+- [ ] **Step 8: Commit**
+
+```bash
+git add app/core/data_source/service.py app/api/data_source/routes.py \
+  app/core/system/permissions.py tests/test_datasource_lifecycle_api.py \
+  tests/test_permission_matrix.py tests/test_datasource_api_security.py \
+  docs/architecture/OPENAPI.yaml
+git commit -m "feat: secure datasource lifecycle APIs"
+```
+
+## Task 11: Make the data-source UI secret-free
+
+**Files:**
+- Modify: `frontend/src/views/dataGovernance/dataSource/index.vue`
+- Modify: `frontend/src/views/dataGovernance/dataSource/components/edit.vue`
+- Modify: `frontend/src/api/dataOrigin.js`
+- Create: `tests/test_datasource_frontend_contract.py`
+
+- [ ] **Step 1: Write the failing frontend contract test**
+
+```python
+def test_datasource_ui_does_not_render_or_replay_stored_credentials():
+    listing = Path(
+        "frontend/src/views/dataGovernance/dataSource/index.vue"
+    ).read_text(encoding="utf-8")
+    editor = Path(
+        "frontend/src/views/dataGovernance/dataSource/components/edit.vue"
+    ).read_text(encoding="utf-8")
+
+    assert "value: 'username'" not in listing
+    assert "value: 'param'" not in listing
+    assert "credential_configured" in listing
+    assert "Object.assign(this.formValues, this.itemData)" not in editor
+    assert "datasourceParse" not in editor
+    assert "key: 'param'" not in editor
+    assert "postgresql" in editor
+    assert "mysql" in editor
+```
+
+- [ ] **Step 2: Verify failure**
+
+```bash
+.venv/bin/python -m pytest \
+  tests/test_datasource_frontend_contract.py -q
+```
+
+Expected: FAIL because the existing UI lists username and arbitrary connection
+parameters and copies the full list item into the credential form.
+
+- [ ] **Step 3: Refactor list and identity handling**
+
+Use `uid`, never the Neo4j numeric `id`, for edit and delete. Remove username
+and free-form connection-parameter columns. Show a
+`credential_configured` status chip instead. Keep host, port, database and
+schema visible because they are non-secret definition fields.
+
+- [ ] **Step 4: Refactor create/edit/test payloads**
+
+Use a fixed database-type selector containing only PostgreSQL and MySQL.
+Remove quick connection-string parsing and the free-form `param` field. On
+edit, copy only safe definition fields into `formValues`; initialize username
+and password as empty and show “凭据已配置，留空则沿用当前凭据”. Omit blank
+credential fields from the request.
+
+For a new source, require username and password. For an existing source,
+send its UID so the backend may combine safe field edits with the current
+credential internally and create the next immutable credential version.
+Connection testing follows the same rule: new source uses entered credentials;
+existing source may use UID plus any explicitly entered replacement
+credentials.
+
+- [ ] **Step 5: Remove obsolete API helpers and run verification**
+
+Remove `datasourceParse()` when no other import remains. Keep the backend
+compatibility endpoint out of navigation until its later removal is separately
+verified.
+
+```bash
+.venv/bin/python -m pytest \
+  tests/test_datasource_frontend_contract.py \
+  tests/test_datasource_api_security.py -q
+cd frontend
+npm run build
+```
+
+Expected: tests PASS and the production frontend build exits 0.
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add frontend/src/views/dataGovernance/dataSource/index.vue \
+  frontend/src/views/dataGovernance/dataSource/components/edit.vue \
+  frontend/src/api/dataOrigin.js \
+  tests/test_datasource_frontend_contract.py
+git commit -m "fix: make datasource management UI secret-free"
+```
+
+## Task 12: Add diagnostics and health aggregation
+
+**Files:**
+- Modify: `app/api/data_source/routes.py`
+- Modify: `app/core/system/health.py`
+- Test: `tests/test_datasource_pool_diagnostics.py`
+
+- [ ] **Step 1: Write failing diagnostics tests**
+
+```python
+def test_admin_pool_status_is_safe(client, admin_token, seeded_pool):
+    response = client.get(
+        f"/api/datasource/{DATA_SOURCE_UID}/pool",
+        headers=bearer(admin_token),
+    )
+    body = response.get_json()["data"]
+
+    assert body["data_source_uid"] == DATA_SOURCE_UID
+    assert body["pool_state"] == "healthy"
+    assert "host" not in body
+    assert "username" not in body
+    assert "password" not in response.get_data(as_text=True).lower()
+
+
+def test_platform_health_contains_only_pool_aggregates(client):
+    body = client.get("/api/system/health").get_json()["data"]["components"]
+    assert body["datasource_pools"] == {
+        "active_pool_count": 1,
+        "degraded_pool_count": 0,
+        "open_circuit_count": 0,
+        "checked_out_total": 0,
+        "pool_timeout_total": 0,
+    }
+```
+
+- [ ] **Step 2: Verify failure**
+
+```bash
+.venv/bin/python -m pytest \
+  tests/test_datasource_pool_diagnostics.py -q
+```
+
+Expected: FAIL because diagnostics are not registered.
+
+- [ ] **Step 3: Add admin-only endpoints**
+
+```text
+GET  /api/datasource/pools
+GET  /api/datasource/{data_source_uid}/pool
+POST /api/datasource/{data_source_uid}/pool/invalidate
+```
+
+All require `datasources:pools:manage`; invalidation accepts only a fixed
+reason enum and records an audit event.
+
+- [ ] **Step 4: Add aggregate health**
+
+If the runtime has not created a registry, report zeros without creating any
+data-source connections. Do not perform a network probe from platform health.
+
+- [ ] **Step 5: Run tests**
+
+```bash
+.venv/bin/python -m pytest \
+  tests/test_datasource_pool_diagnostics.py \
+  tests/test_outbox.py -q
+```
+
+Expected: PASS.
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add app/api/data_source/routes.py app/core/system/health.py \
+  tests/test_datasource_pool_diagnostics.py
+git commit -m "feat: expose safe datasource pool diagnostics"
+```
+
+## Task 13: Add legacy credential report, migration and reconciliation
+
+**Files:**
+- Create: `app/commands/report_datasource_credentials.py`
+- Create: `app/commands/migrate_datasource_credentials.py`
+- Create: `app/commands/reconcile_datasource_credentials.py`
+- Test: `tests/core/data_source/test_credential_migration.py`
+
+- [ ] **Step 1: Write failing migration tests**
+
+```python
+def test_report_never_returns_plaintext(fake_graph):
+    report = inspect_legacy_datasources(fake_graph)
+
+    assert report == [
+        {
+            "data_source_uid": DATA_SOURCE_UID,
+            "issues": ["plaintext_credentials"],
+        }
+    ]
+    assert "legacy-password" not in json.dumps(report)
+
+
+def test_migration_removes_plaintext_only_after_round_trip_verification(
+    fake_graph, credential_repository
+):
+    result = migrate_one(DATA_SOURCE_UID, fake_graph, credential_repository)
+
+    assert result.status == "migrated"
+    assert credential_repository.decrypt_latest().password == "legacy-password"
+    assert "password" not in fake_graph.node
+    assert "username" not in fake_graph.node
+```
+
+- [ ] **Step 2: Verify failure**
+
+```bash
+.venv/bin/python -m pytest \
+  tests/core/data_source/test_credential_migration.py -q
+```
+
+Expected: FAIL because commands do not exist.
+
+- [ ] **Step 3: Implement read-only reporting**
+
+Report issue categories only:
+
+```text
+missing_uid
+plaintext_credentials
+missing_credential_reference
+unsupported_database_type
+incomplete_connection_definition
+```
+
+Default output is JSON Lines and excludes host, username and password.
+
+- [ ] **Step 4: Implement explicit migration**
+
+CLI contract:
+
+```bash
+python -m app.commands.migrate_datasource_credentials \
+  --uid <uuid> \
+  --confirm-encrypt-and-remove-plaintext
+```
+
+Without the confirmation flag, exit non-zero without mutation. For each source:
+
+1. Backfill UUIDv7 when missing.
+2. Encrypt username/password.
+3. Decrypt and compare in memory.
+4. Update credential reference.
+5. Remove plaintext Neo4j fields.
+6. Record safe audit event.
+
+- [ ] **Step 5: Implement reconciliation**
+
+Default is report-only. Compare Neo4j references to PostgreSQL credentials and
+report:
+
+```text
+orphaned_credential
+missing_credential
+stale_credential_version
+revoked_credential_referenced
+plaintext_remaining
+```
+
+Repairs require `--repair --uid <uuid>`. Never delete source data or credentials
+in a broad repair run.
+
+- [ ] **Step 6: Run tests**
+
+```bash
+.venv/bin/python -m pytest \
+  tests/core/data_source/test_credential_migration.py -q
+```
+
+Expected: PASS.
+
+- [ ] **Step 7: Commit**
+
+```bash
+git add app/commands/report_datasource_credentials.py \
+  app/commands/migrate_datasource_credentials.py \
+  app/commands/reconcile_datasource_credentials.py \
+  tests/core/data_source/test_credential_migration.py
+git commit -m "feat: migrate legacy datasource credentials safely"
+```
+
+## Task 14: Add isolated PostgreSQL and MySQL acceptance services
+
+**Files:**
+- Modify: `deploy/docker/docker-compose.yml`
+- Create: `deploy/docker/datasources/postgres/init.sql`
+- Create: `deploy/docker/datasources/mysql/init.sql`
+- Modify: `deploy/docker/.env.example`
+- Modify: `tests/test_local_docker_contract.py`
+- Create: `tests/integration/test_datasource_pools.py`
+
+- [ ] **Step 1: Write failing Compose contract tests**
+
+```python
+def test_compose_has_isolated_external_datasources():
+    source = COMPOSE.read_text(encoding="utf-8")
+
+    assert "source-postgres:" in source
+    assert "source-mysql:" in source
+    assert "dataops-test-source-postgres:" in source
+    assert "dataops-test-source-mysql:" in source
+    assert "DATASOURCE_CREDENTIAL_MASTER_KEY:" in source
+```
+
+- [ ] **Step 2: Verify failure**
+
+```bash
+.venv/bin/python -m pytest \
+  tests/test_local_docker_contract.py -q
+```
+
+Expected: FAIL because the services are absent.
+
+- [ ] **Step 3: Add isolated services**
+
+Use pinned major/minor image tags:
+
+```yaml
+source-postgres:
+  image: postgres:16-alpine
+  environment:
+    POSTGRES_DB: source_db
+    POSTGRES_USER: source_reader
+    POSTGRES_PASSWORD: source-postgres-test-password
+  ports: ["25432:5432"]
+  volumes:
+    - dataops-test-source-postgres:/var/lib/postgresql/data
+    - ./datasources/postgres/init.sql:/docker-entrypoint-initdb.d/init.sql:ro
+
+source-mysql:
+  image: mysql:8.4
+  environment:
+    MYSQL_DATABASE: source_db
+    MYSQL_USER: source_reader
+    MYSQL_PASSWORD: source-mysql-test-password
+    MYSQL_ROOT_PASSWORD: source-mysql-root-test-password
+  ports: ["23306:3306"]
+  volumes:
+    - dataops-test-source-mysql:/var/lib/mysql
+    - ./datasources/mysql/init.sql:/docker-entrypoint-initdb.d/init.sql:ro
+```
+
+Both join `dataops-test-net` and define real health checks. Add test-only
+32-byte base64 master key to the backend environment.
+
+- [ ] **Step 4: Seed equivalent acceptance tables**
+
+Both databases create:
+
+```sql
+CREATE TABLE acceptance_customers (
+    customer_id INTEGER PRIMARY KEY,
+    customer_name VARCHAR(100) NOT NULL
+);
+INSERT INTO acceptance_customers VALUES
+    (1, 'Alpha'),
+    (2, 'Beta');
+```
+
+- [ ] **Step 5: Add integration tests**
+
+Test:
+
+- create both definitions through the API;
+- run `SELECT COUNT(*)` through `DataSourceConnectionManager`;
+- run 100 concurrent read operations;
+- assert every result is `2`;
+- assert no entry exceeds `pool_size + max_overflow`;
+- assert all leases return to zero;
+- advance the clock past 15 minutes and prove idle pools are disposed;
+- close and recreate the Worker runtime and prove pools are rebuilt lazily;
+- delete a source and prove no new connection can be acquired.
+
+- [ ] **Step 6: Start services and run integration**
+
+```bash
+docker compose -f deploy/docker/docker-compose.yml up -d \
+  source-postgres source-mysql backend
+
+TEST_DATABASE_URL=postgresql://dataops:dataops-test-password@localhost:15432/dataops \
+  TEST_SOURCE_POSTGRES_URL=postgresql://source_reader:source-postgres-test-password@localhost:25432/source_db \
+  TEST_SOURCE_MYSQL_URL=mysql+pymysql://source_reader:source-mysql-test-password@localhost:23306/source_db \
+  .venv/bin/python -m pytest \
+  tests/integration/test_datasource_pools.py -q
+```
+
+Expected: PASS; Compose reports both sources healthy.
+
+- [ ] **Step 7: Commit**
+
+```bash
+git add deploy/docker/docker-compose.yml deploy/docker/datasources \
+  deploy/docker/.env.example tests/test_local_docker_contract.py \
+  tests/integration/test_datasource_pools.py
+git commit -m "test: add isolated datasource pool stack"
+```
+
+## Task 15: Verify outage, recovery and credential invalidation
+
+**Files:**
+- Create: `tests/integration/test_datasource_pool_failures.py`
+- Modify: `deploy/docker/README.md`
+
+- [ ] **Step 1: Add outage acceptance test**
+
+The test helper must execute these externally controlled phases:
+
+```python
+assert query(POSTGRES_UID).scalar_one() == 2
+assert query(MYSQL_UID).scalar_one() == 2
+
+    compose_stop("source-mysql")
+    wait_for_error(MYSQL_UID, "DATASOURCE_CONNECTION_FAILED", attempts=3)
+    assert query_error(MYSQL_UID).code == "DATASOURCE_CIRCUIT_OPEN"
+    assert query(POSTGRES_UID).scalar_one() == 2
+    assert platform_health()["database"] is True
+
+    compose_start_and_wait_healthy("source-mysql")
+    clock.advance(30)
+    assert query(MYSQL_UID).scalar_one() == 2
+    assert pool_status(MYSQL_UID)["pool_state"] == "healthy"
+```
+
+- [ ] **Step 2: Make the outage test explicitly opt-in and self-restoring**
+
+The test may run Docker Compose commands only when
+`RUN_DATASOURCE_OUTAGE_TEST=1`. Inject a fake monotonic clock into the breaker
+so recovery does not require a 30-second sleep. Wrap the outage in
+`try/finally`; the `finally` block always starts `source-mysql` and waits for
+its Compose health condition. A failed assertion must never leave the service
+stopped.
+
+- [ ] **Step 3: Verify the default skip**
+
+```bash
+.venv/bin/python -m pytest \
+  tests/integration/test_datasource_pool_failures.py -q
+```
+
+Expected: SKIP unless `RUN_DATASOURCE_OUTAGE_TEST=1`.
+
+- [ ] **Step 4: Run outage, recovery and credential-rotation acceptance**
+
+```bash
+RUN_DATASOURCE_OUTAGE_TEST=1 \
+  .venv/bin/python -m pytest \
+  tests/integration/test_datasource_pool_failures.py -q
+```
+
+Within the same test process: establish both pools, stop MySQL, open only the
+MySQL circuit after three failures, prove PostgreSQL and the platform control
+database remain healthy, restart MySQL, advance the injected clock, and prove
+the half-open probe closes the circuit. Then update a test credential version,
+assert the old Engine is disposed, and prove the next query uses the new
+version. Poll the Compose health condition rather than sleeping a fixed
+duration.
+
+- [ ] **Step 5: Document operations**
+
+Add:
+
+- pool defaults and environment variables;
+- report/migrate/reconcile commands;
+- admin diagnostics;
+- safe invalidation;
+- outage recovery;
+- explicit warning that platform PostgreSQL is not managed by this registry.
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add tests/integration/test_datasource_pool_failures.py \
+  deploy/docker/README.md
+git commit -m "test: verify datasource outage and recovery"
+```
+
+## Task 16: Final contract, release sync and complete verification
+
+**Files:**
+- Modify: `docs/architecture/ARCHITECTURE_OVERVIEW.md`
+- Modify: `docs/architecture/DATA_MODEL.md`
+- Modify: `docs/architecture/OPENAPI.yaml`
+- Modify: `deployment/sync_release.sh` only if new source directories are not already copied
+- Test: existing full test suites
+
+- [ ] **Step 1: Update architecture artifacts**
+
+Document:
+
+- control-plane PostgreSQL versus external data-source pools;
+- credential table and Neo4j reference;
+- Worker-local pool multiplication;
+- supported adapters;
+- failure and security boundaries.
+
+- [ ] **Step 2: Regenerate OpenAPI and sync release source**
+
+```bash
+.venv/bin/python scripts/generate_openapi.py
+bash deployment/sync_release.sh
+diff -qr app deployment/app
+diff -qr migrations deployment/migrations
+cmp requirements.txt deployment/requirements.txt
+```
+
+Expected: all diff/cmp commands exit 0.
+
+- [ ] **Step 3: Run full Python verification**
+
+```bash
+TEST_POSTGRES_ADMIN_URL=postgresql://dataops:dataops-test-password@localhost:15432/postgres \
+TEST_DATABASE_URL=postgresql://dataops:dataops-test-password@localhost:15432/dataops \
+TEST_ADMIN_PASSWORD=AdminPass123 \
+  .venv/bin/python -m pytest -q
+```
+
+Expected: zero failures and zero errors.
+
+- [ ] **Step 4: Run frontend build**
+
+```bash
+cd frontend
+npm run build
+```
+
+Expected: build exits 0; pre-existing Vue 2 dependency warnings may remain,
+but no new ESLint errors are allowed.
+
+- [ ] **Step 5: Rebuild and smoke-test the complete stack**
+
+```bash
+docker compose -f deploy/docker/docker-compose.yml up -d --build
+docker compose -f deploy/docker/docker-compose.yml ps
+curl -fsS http://localhost:18183/ >/dev/null
+curl -fsS http://localhost:15500/api/system/health
+```
+
+Expected: platform services, `source-postgres` and `source-mysql` are healthy.
+
+- [ ] **Step 6: Scan for plaintext secrets**
+
+```bash
+rg -n \
+  "password.*logger|logger.*password|postgresql://[^[:space:]]+:[^[:space:]@]+@" \
+  app deployment/app docs/architecture
+```
+
+Expected: no data-source credential disclosure. Test-only Compose credentials
+must remain limited to `deploy/docker`.
+
+- [ ] **Step 7: Commit final artifacts**
+
+```bash
+git add -- \
+  docs/architecture/ARCHITECTURE_OVERVIEW.md \
+  docs/architecture/DATA_MODEL.md \
+  docs/architecture/OPENAPI.yaml \
+  deployment/app/api/data_source/routes.py \
+  deployment/app/commands/migrate_datasource_credentials.py \
+  deployment/app/commands/reconcile_datasource_credentials.py \
+  deployment/app/commands/reconcile_governance_uids.py \
+  deployment/app/commands/report_datasource_credentials.py \
+  deployment/app/config/config.py \
+  deployment/app/core/data_source \
+  deployment/app/core/system/health.py \
+  deployment/app/core/system/permissions.py \
+  deployment/gunicorn_config.py \
+  deployment/migrations/versions/20260718_50_datasource_credentials.py \
+  deployment/requirements.txt
+git commit -m "docs: finalize datasource pool operations"
+```
+
+Before staging, inspect `git diff --name-only -- deployment/` and confirm every
+changed release-copy file is the generated counterpart of a source file from
+Tasks 1–15. Do not stage unrelated pre-existing workspace changes.
+
+- [ ] **Step 8: Request code review**
+
+Use the `requesting-code-review` skill. Review specifically for:
+
+- secret leakage;
+- cross-store partial failure;
+- connection leaks;
+- pool multiplication across Workers;
+- unsafe driver or connect-argument injection;
+- write-transaction replay.

docs/superpowers/specs/2026-07-18-datasource-connection-pool-design.md

@@ -1,7 +1,7 @@
 # 数据源连接池与凭据安全设计
 
 日期：2026-07-18
-状态：已完成讨论，等待书面规格审阅
+状态：已批准，进入实施计划
 
 ## 1. 目标