feat(postgres): define driver-neutral migration port

.github/workflows/ci.yml

@@ -114,3 +114,75 @@ jobs:
         run: bash tests/smoke_postgres_container_logging.sh
       - name: Run legacy SQL cleanup integration smoke
         run: bash tests/smoke_legacy_sql_cleanup.sh
+      - name: Set up Python 3.14 for candidate parity
+        uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
+        with:
+          python-version: "3.14"
+      - name: Set up uv for candidate parity
+        uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
+        with:
+          prune-cache: true
+      - name: Install locked project dependencies
+        run: uv sync --locked
+      - name: Download exact pg8000 candidate artifact
+        run: >-
+          python -m pip download --no-deps --only-binary=:all:
+          --dest /tmp/pg8000-candidate pg8000==1.31.5
+      - name: Verify pg8000 candidate artifact digest
+        run: >-
+          echo
+          "0af2c1926b153307639868d2ee5cef6cd3a7d07448e12736989b10e1d491e201  /tmp/pg8000-candidate/pg8000-1.31.5-py3-none-any.whl"
+          | sha256sum --check --strict
+      - name: Install candidate only into the CI environment
+        run: >-
+          uv pip install --python .venv/bin/python
+          /tmp/pg8000-candidate/pg8000-1.31.5-py3-none-any.whl
+      - name: Start candidate PostgreSQL runtime
+        shell: bash
+        run: |
+          candidate_password="$(python -c 'import secrets; print(secrets.token_urlsafe(32))')"
+          echo "::add-mask::$candidate_password"
+          password_file="$(mktemp)"
+          printf '%s' "$candidate_password" > "$password_file"
+          chmod 600 "$password_file"
+          container="pg-llm-batch-pg8000-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}"
+          echo "PG_LLM_BATCH_POSTGRES_PASSWORD=$candidate_password" >> "$GITHUB_ENV"
+          echo "PG8000_CANDIDATE_PASSWORD_FILE=$password_file" >> "$GITHUB_ENV"
+          echo "PG8000_CANDIDATE_CONTAINER=$container" >> "$GITHUB_ENV"
+          docker run --detach --name "$container" \
+            --publish 127.0.0.1:5432:5432 \
+            --mount "type=bind,source=$password_file,target=/run/secrets/postgres_password,readonly" \
+            --env POSTGRES_USER=pgllm \
+            --env POSTGRES_DB=pgllm \
+            --env POSTGRES_PASSWORD_FILE=/run/secrets/postgres_password \
+            pg-llm-batch-postgres:ci
+      - name: Wait for candidate PostgreSQL health contract
+        shell: bash
+        run: |
+          for attempt in $(seq 1 90); do
+            if docker exec "$PG8000_CANDIDATE_CONTAINER" \
+              pg_isready -U pgllm -d pgllm >/dev/null 2>&1; then
+              ready="$(docker exec "$PG8000_CANDIDATE_CONTAINER" \
+                psql -U pgllm -d pgllm -tAc \
+                "SELECT bool_and(is_ready) FROM pg_llm_batch_health_check() WHERE component IN ('database','pg_tiktoken','com_config')" \
+                2>/dev/null || true)"
+              if [ "$ready" = "t" ]; then
+                exit 0
+              fi
+            fi
+            sleep 2
+          done
+          docker logs "$PG8000_CANDIDATE_CONTAINER"
+          exit 1
+      - name: Run real pg8000 candidate PostgreSQL smoke
+        run: uv run --no-sync python tests/smoke_pg8000_candidate_postgres.py
+      - name: Tear down candidate PostgreSQL runtime
+        if: ${{ always() }}
+        shell: bash
+        run: |
+          if [ -n "${PG8000_CANDIDATE_CONTAINER:-}" ]; then
+            docker rm --force "$PG8000_CANDIDATE_CONTAINER" >/dev/null 2>&1 || true
+          fi
+          if [ -n "${PG8000_CANDIDATE_PASSWORD_FILE:-}" ]; then
+            rm -f "$PG8000_CANDIDATE_PASSWORD_FILE"
+          fi

pg_llm_batch/checkpoint_store.py

@@ -18,6 +18,7 @@
     validate_tenant_scope,
 )
 from .exceptions import ConfigError, PgLlmBatchError, ValidationError
+from .postgres_driver_port import PostgresDriverPort
 from .result_streaming import BatchResultCheckpoint
 
 MIGRATION_PATH = (
@@ -85,6 +86,23 @@ def _validated_postgres_dsn(value: Any) -> str:
     return value
 
 
+def _connect_postgres(
+    postgres_dsn: str,
+    postgres_driver: PostgresDriverPort | None,
+) -> Any:
+    """Connect through an injected driver while preserving the legacy default.
+
+    The optional port lets one bounded persistence consumer migrate away from
+    Psycopg without changing its SQL, transaction, tenant, or checkpoint
+    semantics. Until the repository selects and validates a commercial
+    replacement, omitting the port retains the current Psycopg path explicitly.
+    """
+    if postgres_driver is not None:
+        return postgres_driver.connect(postgres_dsn)
+    _require_psycopg()
+    return psycopg.connect(postgres_dsn)
+
+
 def _validated_checkpoint(value: Any, field: str) -> BatchResultCheckpoint:
     """Require one immutable checkpoint whose counters fit PostgreSQL storage."""
     if not isinstance(value, BatchResultCheckpoint):
@@ -171,13 +189,14 @@ def _checkpoint_values(checkpoint: BatchResultCheckpoint) -> tuple[Any, ...]:
 def apply_result_checkpoint_schema(
     postgres_dsn: str,
     migration_path: Optional[str] = None,
+    *,
+    postgres_driver: PostgresDriverPort | None = None,
 ) -> None:
     """Apply the idempotent durable result-checkpoint migration."""
     dsn = _validated_postgres_dsn(postgres_dsn)
-    _require_psycopg()
     path = Path(migration_path) if migration_path else MIGRATION_PATH
     sql = path.read_text(encoding="utf-8")
-    with psycopg.connect(dsn) as conn:
+    with _connect_postgres(dsn, postgres_driver) as conn:
         with conn.cursor() as cur:
             cur.execute(sql)
         conn.commit()
@@ -191,9 +210,11 @@ def __init__(
         postgres_dsn: str,
         *,
         tenant_scope: str = DEFAULT_TENANT_SCOPE,
+        postgres_driver: PostgresDriverPort | None = None,
     ) -> None:
-        """Bind one explicit database and trusted local tenant scope to the store."""
+        """Bind one explicit database, tenant scope, and optional driver port."""
         self.postgres_dsn = _validated_postgres_dsn(postgres_dsn)
+        self._postgres_driver = postgres_driver
         try:
             self.tenant_scope = validate_tenant_scope(tenant_scope)
         except ValidationError as exc:
@@ -210,8 +231,7 @@ def load(
         endpoint_alias: str,
     ) -> Optional[BatchResultCheckpoint]:
         """Load the current checkpoint in one package-owned transaction."""
-        _require_psycopg()
-        with psycopg.connect(self.postgres_dsn) as conn:
+        with _connect_postgres(self.postgres_dsn, self._postgres_driver) as conn:
             with conn.cursor() as cur:
                 return self.load_in_transaction(
                     cur,
@@ -257,8 +277,7 @@ def save(
         expected_previous: Optional[BatchResultCheckpoint] = None,
     ) -> BatchResultCheckpoint:
         """Create or advance a checkpoint in one package-owned transaction."""
-        _require_psycopg()
-        with psycopg.connect(self.postgres_dsn) as conn:
+        with _connect_postgres(self.postgres_dsn, self._postgres_driver) as conn:
             with conn.cursor() as cur:
                 saved = self.save_in_transaction(
                     cur,

pg_llm_batch/cli.py

@@ -33,17 +33,16 @@
 import sys
 import warnings
 from contextlib import ExitStack
+from functools import partial
 from typing import List, Optional
 
-from psycopg import ProgrammingError
-from psycopg.conninfo import conninfo_to_dict
-
 from . import db
 from .batch_api_client import BatchAPIClient, config_credentials_provider
 from .bootstrap import resolve_dsn, resolve_secret_key
 from .config import PostgresConfigStore, SecretStore
 from .exceptions import ConfigError, PgLlmBatchError
 from .health import check_health, serve_healthz
+from .postgres_driver_port import PostgresDriverPort
 from .token_counter import TokenCounter
 
 MAX_SECRET_INPUT_CHARACTERS = 65_536
@@ -73,28 +72,50 @@ def error(self, message: str) -> None:
         super().error(redacted_message)
 
 
-def _validate_cli_dsn(value: str) -> str:
-    """Accept valid libpq selectors while refusing credential-bearing argv data."""
+def _default_postgres_driver() -> PostgresDriverPort:
+    """Load the retained Psycopg adapter only when a CLI parse needs the default."""
+    from .psycopg_driver_adapter import PsycopgDriverAdapter
+
+    return PsycopgDriverAdapter()
+
+
+def _validate_cli_dsn(
+    value: str,
+    *,
+    postgres_driver: PostgresDriverPort | None = None,
+) -> str:
+    """Accept valid PostgreSQL selectors without concrete-driver coupling."""
+    driver = (
+        postgres_driver
+        if postgres_driver is not None
+        else _default_postgres_driver()
+    )
     try:
-        parameters = conninfo_to_dict(value)
-    except ProgrammingError:
-        raise argparse.ArgumentTypeError(
-            "Postgres DSN must be valid libpq connection information"
-        ) from None
+        parameters = driver.parse_conninfo(value)
+    except Exception as exc:
+        if driver.is_invalid_conninfo(exc):
+            raise argparse.ArgumentTypeError(
+                "Postgres DSN must be valid connection information"
+            ) from None
+        raise
     if CLI_DSN_SENSITIVE_PARAMETERS.intersection(parameters):
         raise argparse.ArgumentTypeError(
             "Credential-bearing Postgres DSNs are not accepted in --dsn; "
-            "use libpq secret mechanisms outside process argv"
+            "use PostgreSQL secret mechanisms outside process argv"
         )
     return value
 
 
-def _add_common(parser: argparse.ArgumentParser) -> None:
+def _add_common(
+    parser: argparse.ArgumentParser,
+    *,
+    postgres_driver: PostgresDriverPort | None = None,
+) -> None:
     """Add the shared credential-free ``--dsn`` selector to a subcommand parser."""
     parser.add_argument(
         "--dsn",
         default=None,
-        type=_validate_cli_dsn,
+        type=partial(_validate_cli_dsn, postgres_driver=postgres_driver),
         help=(
             "Credential-free Postgres selector "
             "(else PG_LLM_BATCH_DSN bootstrap env var)"
@@ -186,40 +207,43 @@ def _read_token_input() -> str:
         raise ConfigError("Token input must be valid UTF-8") from None
 
 
-def build_parser() -> argparse.ArgumentParser:
-    """Build the command-line parser and all supported subcommands."""
+def build_parser(
+    *,
+    postgres_driver: PostgresDriverPort | None = None,
+) -> argparse.ArgumentParser:
+    """Build the command-line parser with an injectable PostgreSQL DSN parser."""
     parser = _RedactingArgumentParser(
         prog="pg_llm_batch",
         description="Standalone Postgres LLM batch engine",
     )
     sub = parser.add_subparsers(dest="command", required=True)
 
     p_init = sub.add_parser("init-db", help="Apply batch schema (idempotent)")
-    _add_common(p_init)
+    _add_common(p_init, postgres_driver=postgres_driver)
 
     p_cfg = sub.add_parser("config", help="Manage KV config and secrets")
     cfg_sub = p_cfg.add_subparsers(dest="config_command", required=True)
     p_set = cfg_sub.add_parser("set", help="Set a config value")
-    _add_common(p_set)
+    _add_common(p_set, postgres_driver=postgres_driver)
     p_set.add_argument("category")
     p_set.add_argument("key")
     p_set.add_argument("value")
     p_get = cfg_sub.add_parser("get", help="Get a config value")
-    _add_common(p_get)
+    _add_common(p_get, postgres_driver=postgres_driver)
     p_get.add_argument("category")
     p_get.add_argument("key")
     p_secret = cfg_sub.add_parser(
         "set-secret",
         help="Store a secret from a no-echo prompt or standard input",
     )
-    _add_common(p_secret)
+    _add_common(p_secret, postgres_driver=postgres_driver)
     p_secret.add_argument("secret_key")
 
     p_count = sub.add_parser(
         "count-tokens",
         help="Count bounded UTF-8 stdin content without exposing it in argv",
     )
-    _add_common(p_count)
+    _add_common(p_count, postgres_driver=postgres_driver)
     p_count.add_argument("--model", required=True)
     p_count.add_argument(
         "--stdin",
@@ -229,38 +253,38 @@ def build_parser() -> argparse.ArgumentParser:
     )
 
     p_submit = sub.add_parser("submit", help="Upload payload + create batch job")
-    _add_common(p_submit)
+    _add_common(p_submit, postgres_driver=postgres_driver)
     p_submit.add_argument("--endpoint", required=True, help="Endpoint alias")
     p_submit.add_argument("--file-path", required=True, help="memory://<file_id>")
     p_submit.add_argument("--batch-endpoint", default="/v1/chat/completions")
 
     p_poll = sub.add_parser("poll", help="Poll a batch job status once")
-    _add_common(p_poll)
+    _add_common(p_poll, postgres_driver=postgres_driver)
     p_poll.add_argument("--endpoint", required=True)
     p_poll.add_argument("--batch-id", required=True)
 
     p_wait = sub.add_parser("wait", help="Wait for a terminal batch status")
-    _add_common(p_wait)
+    _add_common(p_wait, postgres_driver=postgres_driver)
     p_wait.add_argument("--endpoint", required=True)
     p_wait.add_argument("--batch-id", required=True)
     p_wait.add_argument("--poll-interval", type=float, default=5.0)
     p_wait.add_argument("--timeout", type=float, default=3600.0)
 
     p_retrieve = sub.add_parser("retrieve", help="Download batch results")
-    _add_common(p_retrieve)
+    _add_common(p_retrieve, postgres_driver=postgres_driver)
     p_retrieve.add_argument("--endpoint", required=True)
     p_retrieve.add_argument("--batch-id", required=True)
 
     p_cancel = sub.add_parser("cancel", help="Cancel a provider batch job")
-    _add_common(p_cancel)
+    _add_common(p_cancel, postgres_driver=postgres_driver)
     p_cancel.add_argument("--endpoint", required=True)
     p_cancel.add_argument("--batch-id", required=True)
 
     p_health = sub.add_parser("health", help="Print readiness report")
-    _add_common(p_health)
+    _add_common(p_health, postgres_driver=postgres_driver)
 
     p_serve = sub.add_parser("serve-healthz", help="Serve GET /healthz")
-    _add_common(p_serve)
+    _add_common(p_serve, postgres_driver=postgres_driver)
     p_serve.add_argument("--host", default="127.0.0.1")
     p_serve.add_argument("--port", type=int, default=8080)