fix(migrations): harden BAT-025 PostgreSQL verification

app/db/migrations.py

@@ -16,6 +16,10 @@ def get_alembic_config(settings: Settings) -> Config:
     config = Config(str(project_root / "alembic.ini"))
     config.set_main_option("script_location", str(project_root / "migrations"))
     config.set_main_option("sqlalchemy.url", settings.active_database_url.replace("%", "%%"))
+    # migrations/env.py must use this explicit settings object instead of the
+    # process-global cached Settings.  This keeps programmatic callers (notably
+    # isolated integration tests and runtime bootstrap) on their requested DB.
+    config.attributes["portfoliohub_settings"] = settings
     return config
 
 
@@ -29,7 +33,13 @@ def bootstrap_or_upgrade_to_head(settings: Settings) -> str:
         return "head"
 
     if _has_legacy_portfolio_schema(settings):
-        command.stamp(get_alembic_config(settings), "head")
+        # Runtime ensure_schema() can create the BAT-025 columns before Alembic is
+        # introduced.  Stamp only the revision immediately before BAT-025, then run
+        # the real upgrade so 0028 backfills a nullable pre-created `source` and
+        # converges constraints/defaults without replaying the older DDL chain.
+        config = get_alembic_config(settings)
+        command.stamp(config, "20260628_0027")
+        command.upgrade(config, "head")
         return "head"
 
     upgrade_to_head(settings)

docs/batches/bat-025_20260715_portfolio-ledger-alembic-migration/implementation.md

@@ -2,47 +2,54 @@
 
 - Date: 2026-07-15
 - Assignee: dev-impl
-- Scope: REQ-057 portfolio_ledger 수동 장부 메타데이터 명시 Alembic migration
+- Scope: REQ-057 `portfolio_ledger` 수동 장부 메타데이터의 명시 Alembic migration
+- Implementation state: remediation complete; independent QC pending
 
-## Implemented Changes
+## Baseline and remediation provenance
 
-1. Created new Alembic revision `20260715_0028` (`migrations/versions/20260715_0028_portfolio_ledger_manual_metadata.py`)
+- Initial implementation baseline: `f6195ff8090d8761d467e15f44255642ca350e80`.
-   - `down_revision`: `20260628_0027`
+- The previously recorded SHA `850e970` does **not** identify the reviewed implementation and is superseded by the baseline above.
-   - Added additive columns:
+- The remediation below is deliberately **uncommitted** in this worktree. No push, PR, main merge, production migration, deployment, or service activation was performed.
-     - `source TEXT NOT NULL DEFAULT 'broker'`
-     - `target_ledger_entry_id BIGINT NULL`
-     - `confirmation_evidence TEXT NULL`
-     - `actor_user_id TEXT NULL`
-   - Added idempotent runtime compatibility logic:
-     - If `portfolio_ledger` table missing: no-op
-     - If column already exists: update `source IS NULL` → `'broker'`, enforce `NOT NULL` + `DEFAULT`
-     - Add missing columns only when absent
-   - `downgrade` is forward-only no-op
 
-2. Added migration-focused tests `tests/test_bat025_portfolio_ledger_manual_metadata_migration.py`
+## Implemented changes
-   - Covers:
-     - fresh schema add-all-columns behavior
-     - precreated schema backfill + source constraint enforcement
-     - missing table no-op
-     - forward-only downgrade no-op
 
-3. Added operational runbook
+1. Explicit revision `20260715_0028` remains the REQ-057 head.
+   - Parent: `20260628_0027`.
+   - Adds `source`, `target_ledger_entry_id`, `confirmation_evidence`, and `actor_user_id` additively.
+   - Handles runtime-precreated columns by backfilling nullable `source`, then converging its default and `NOT NULL` constraint.
+   - Downgrade remains a forward-only no-op; no column deletion was added.
+
+2. Remediation of actual PostgreSQL migration-chain/runtime behavior.
+   - `migrations/versions/20260605_0025_ledger_consolidation.py`: makes the identity `ledger_entry_id` explicitly `nullable=False`. PostgreSQL identity columns are implicitly non-null; the prior explicit trailing `NULL` was incompatible with fresh real-PostgreSQL migration.
+   - `app/db/migrations.py`: programmatic Alembic configuration now carries the caller-provided `Settings`. For a legacy/runtime-precreated schema, bootstrap stamps `20260628_0027` and then executes `upgrade head`, so revision `0028` performs its required source backfill/default/constraint convergence instead of being skipped by a direct head stamp.
+   - `migrations/env.py`: consumes the explicit settings when supplied and preserves existing application loggers when Alembic runs in-process.
+
+3. Real isolated PostgreSQL integration coverage.
+   - `tests/test_bat025_portfolio_ledger_manual_metadata_migration.py` creates a UUID-named temporary database only through an explicitly supplied loopback `postgres` maintenance DSN.
+   - The test rejects non-loopback hosts and any maintenance database other than `postgres`; each test terminates connections and drops only its own UUID-named database.
+   - Coverage includes pre-0028 upgrade/data preservation, runtime-precreated nullable `source` convergence and Alembic current-head state, plus representative broker/manual ledger writes after upgrade.
+
+4. Existing operation runbook is retained.
    - `docs/runbooks/bat-025-req057-portfolio-ledger-manual-metadata-runbook.md`
+   - It remains a runbook only; no production command in it was executed.
 
-## Preflight Evidence
+## Executed evidence
 
-- `uv run alembic heads`
+All commands below used the local Docker PostgreSQL service only. The DSN was supplied from an environment variable and is intentionally omitted here.
-  - output: `20260715_0028 (head)`
-- `uv run alembic history --verbose`
-  - confirms `20260715_0028` parent is `20260628_0027` and it is the current tail of the migration chain.
-- `PYTHONPATH=. uv run pytest -q -o addopts='' tests/test_bat025_portfolio_ledger_manual_metadata_migration.py`
-  - output: `4 passed in 0.24s`
 
-## Evidence notes
+- `BAT025_LOCAL_POSTGRES_DSN=[REDACTED local loopback maintenance DSN] PYTHONPATH=. uv run --extra dev --with httpx2 python -m pytest -q --override-ini addopts='' -p no:cacheprovider tests/test_bat025_portfolio_ledger_manual_metadata_migration.py`
+  - Result: **3 passed in 6.10s**.
+  - Verifies real 0027→0028 data preservation/source backfill, runtime-precreated compatibility/current revision, and broker/manual write regression in throwaway DBs.
+- `PYTHONPATH=. uv run --extra dev --with httpx2 python -m pytest -q --override-ini addopts='' -p no:cacheprovider`
+  - Result: **597 passed, 11 warnings in 41.41s**.
+  - Warnings are pre-existing Starlette `HTTP_422_UNPROCESSABLE_ENTITY` deprecations in endpoint tests; no test failed.
+- `uv run --extra dev ruff check app/db/migrations.py migrations/env.py migrations/versions/20260605_0025_ledger_consolidation.py tests/test_bat025_portfolio_ledger_manual_metadata_migration.py`
+  - Result: **All checks passed**.
+- `python3 -m compileall -q app migrations` and `git diff --check`
+  - Result: exit 0.
 
-- Migration is idempotent by design: no-op if `portfolio_ledger` is missing, and for pre-created runtimes it backfills `source` and enforces `NOT NULL DEFAULT 'broker'`.
+## Limits / non-claims
-- No DB upgrade was executed in this scope (code-only + evidence-only in repo).
 
-## Commit
+- This evidence is for isolated local PostgreSQL only; it is not an operational DB upgrade.
-
+- Lock contention, backup restore, production access control, deployment, service activation, and M5 release remain unexecuted.
-- Commit SHA: `850e970`
+- This log is developer evidence only. It does not mark BAT-025 verified or approved; independent QC remains required.

docs/batches/bat-025_20260715_portfolio-ledger-alembic-migration/verification.md

@@ -0,0 +1,48 @@
+# BAT-025 Developer Self-Verification
+
+- Status: **draft — independent QC pending**
+- Batch: BAT-025
+- Requirement: REQ-057
+- Date: 2026-07-15
+- Scope boundary: local isolated PostgreSQL evidence only. No production DB, deployment, service activation, push, merge, or release action was performed.
+
+## Acceptance mapping
+
+| REQ-057 acceptance area | Evidence | Result |
+| --- | --- | --- |
+| Explicit `0028` revision is the chain head | `alembic_version` queried in a throwaway PostgreSQL DB after upgrade | Pass — current revision `20260715_0028` |
+| Existing broker/manual business data survives upgrade | Real `20260628_0027 → head` upgrade after a representative pre-existing ledger row is inserted; row fields and metadata are asserted after upgrade | Pass |
+| `source` is backfilled/defaulted/not-null | Real upgrade asserts preserved row `source='broker'`; schema inspection checks text type, non-null constraint, and default | Pass |
+| Runtime-precreated metadata schema converges | Real test creates all four columns with nullable `source`, invokes bootstrap, and asserts source backfill/default/not-null plus `current=head` | Pass |
+| Representative broker/manual writes still work | Post-upgrade synthetic repository writes assert broker source and manual metadata fields | Pass |
+| Earlier fresh PostgreSQL chain can reach 0028 | `0025` identity-column PostgreSQL incompatibility corrected and exercised by fresh throwaway DB migration path | Pass |
+| API/manual-ledger regression | `tests/test_v1_portfolios_manual_ledger.py`, `tests/test_v1_portfolios_trades.py`, `tests/test_v1_portfolios_ledger.py` | Pass — 33 tests |
+| Whole-repository regression | pytest with project dev extra plus temporary TestClient dependency overlay | Pass — 597 tests, 11 warnings |
+| Lint/compile/diff hygiene | Ruff, compileall, `git diff --check` | Pass |
+
+## Executed commands and results
+
+- `BAT025_LOCAL_POSTGRES_DSN=[REDACTED local loopback maintenance DSN] PYTHONPATH=. uv run --extra dev --with httpx2 python -m pytest -q --override-ini addopts='' -p no:cacheprovider tests/test_bat025_portfolio_ledger_manual_metadata_migration.py`
+  - **3 passed in 6.10s**.
+  - Each test creates and removes a UUID-named database on local loopback PostgreSQL; it refuses non-loopback hosts and non-`postgres` maintenance databases.
+- `PYTHONPATH=. uv run --extra dev --with httpx2 python -m pytest -q --override-ini addopts='' -p no:cacheprovider`
+  - **597 passed, 11 warnings in 41.41s**.
+  - Warnings are Starlette deprecations for `HTTP_422_UNPROCESSABLE_ENTITY`; they are not BAT-025 failures.
+- `uv run --extra dev ruff check app/db/migrations.py migrations/env.py migrations/versions/20260605_0025_ledger_consolidation.py tests/test_bat025_portfolio_ledger_manual_metadata_migration.py`
+  - **All checks passed**.
+- `python3 -m compileall -q app migrations` and `git diff --check`
+  - exit 0.
+
+## Evidence boundaries and blockers
+
+- The local container is an isolated test target, not a staging or production database.
+- No production backup/restore, lock-duration measurement under production load, migration run against an operational DB, deploy, or service activation was attempted.
+- The test command uses a transient `httpx2` overlay because this environment's resolved Starlette TestClient requires it while the project declares `httpx`; no dependency source change was made for this batch.
+- Independent `qc-reviewer` must rerun the tests and inspect the uncommitted diff before this document can become a QC verdict.
+
+## Gate result
+
+- Developer self-verification: **ready for independent QC review**.
+- Independent QC: **not yet executed**.
+- `confirm-batch-verification`: **not eligible yet**.
+- M5 release / operational migration: **not started and remains blocked**.

migrations/env.py

@@ -10,13 +10,14 @@ from app.core.config import get_settings
 config = context.config
 
 if config.config_file_name is not None:
-    fileConfig(config.config_file_name)
+    # Alembic may run in-process (bootstrap/tests); preserve application loggers.
+    fileConfig(config.config_file_name, disable_existing_loggers=False)
 
 target_metadata = None
 
 
 def _configure_database_url() -> str:
-    settings = get_settings()
+    settings = config.attributes.get("portfoliohub_settings") or get_settings()
     database_url = settings.active_database_url
     config.set_main_option("sqlalchemy.url", database_url.replace("%", "%%"))
     return database_url

migrations/versions/20260605_0025_ledger_consolidation.py

@@ -18,7 +18,9 @@ depends_on: str | Sequence[str] | None = None
 
 
 def upgrade() -> None:
-    op.add_column("portfolio_ledger", sa.Column("ledger_entry_id", sa.BigInteger(), sa.Identity(always=False), nullable=True))
+    # PostgreSQL identity columns are implicitly NOT NULL.  ``nullable=True``
+    # renders a conflicting trailing ``NULL`` clause on a real PostgreSQL DB.
+    op.add_column("portfolio_ledger", sa.Column("ledger_entry_id", sa.BigInteger(), sa.Identity(always=False), nullable=False))
     op.add_column("portfolio_ledger", sa.Column("order_id", sa.Text(), nullable=True))
     op.add_column("portfolio_ledger", sa.Column("execution_id", sa.Text(), nullable=True))
     op.add_column("portfolio_ledger", sa.Column("side", sa.Text(), nullable=True))

tests/test_bat025_portfolio_ledger_manual_metadata_migration.py

@@ -1,133 +1,273 @@
-"""BAT-025: portfolio_ledger manual metadata migration behavior tests.
+"""Real PostgreSQL integration tests for BAT-025 / REQ-057.
 
-Focus areas:
+These tests deliberately require an explicit local maintenance DSN and create a
-- fresh schema: add required columns safely
+unique throwaway database per test.  They reject non-loopback hosts and any
-- runtime-precreated schema: source backfill + source constraints 보정
+non-``postgres`` maintenance database so test execution cannot target a shared
-- forward-only downgrade policy
+or production application database.
 """
 
 from __future__ import annotations
 
-from typing import Any
+import os
-from pathlib import Path
+from datetime import datetime, timezone
-from unittest.mock import MagicMock, patch
+from urllib.parse import unquote, urlparse
+from uuid import uuid4
 
-import importlib.util
+import psycopg
+import pytest
+from psycopg import sql
+
+from app.core.config import Settings
+from app.db.migrations import (
+    bootstrap_or_upgrade_to_head,
+    get_alembic_config,
+    upgrade_to_head,
+)
+from app.repositories.postgres import PortfolioDBConfig, PostgresPortfolioRepository
+
+_LOCAL_DSN_ENV = "BAT025_LOCAL_POSTGRES_DSN"
+_LOOPBACK_HOSTS = {"127.0.0.1", "localhost", "::1"}
+_HEAD = "20260715_0028"
+_PRE_BAT025 = "20260628_0027"
 
 
-_MIGRATION_PATH = Path(__file__).resolve().parents[1] / "migrations" / "versions" / "20260715_0028_portfolio_ledger_manual_metadata.py"
+def _maintenance_dsn() -> str:
-_MIGRATION_NAME = "m_20260715_0028_portfolio_ledger_manual_metadata"
+    dsn = os.environ.get(_LOCAL_DSN_ENV)
-_spec = importlib.util.spec_from_file_location(_MIGRATION_NAME, _MIGRATION_PATH)
+    if not dsn:
-assert _spec is not None and _spec.loader is not None
+        pytest.skip(f"{_LOCAL_DSN_ENV} is required for real PostgreSQL integration tests")
-_migration = importlib.util.module_from_spec(_spec)
+    parsed = urlparse(dsn)
-_spec.loader.exec_module(_migration)
+    if parsed.scheme not in {"postgres", "postgresql"}:
+        pytest.fail("BAT-025 test DSN must use postgres/postgresql")
+    if parsed.hostname not in _LOOPBACK_HOSTS:
+        pytest.fail("BAT-025 tests refuse a non-loopback PostgreSQL host")
+    if parsed.path.lstrip("/") != "postgres":
+        pytest.fail("BAT-025 test DSN must connect to local maintenance database 'postgres'")
+    if not parsed.username:
+        pytest.fail("BAT-025 test DSN must include a local database user")
+    return dsn
 
 
-class _FakeInspector:
+def _settings_for_database(maintenance_dsn: str, database: str) -> Settings:
-    def __init__(self, columns: list[str]) -> None:
+    parsed = urlparse(maintenance_dsn)
-        self._columns = columns
+    return Settings(
-
+        db_enabled=True,
-    def get_table_names(self) -> list[str]:
+        timescaledb_host=parsed.hostname,
-        return ["portfolio_ledger"]
+        timescaledb_port=parsed.port or 5432,
-
+        timescaledb_name=database,
-    def get_columns(self, table_name: str) -> list[dict[str, Any]]:
+        timescaledb_user=unquote(parsed.username or ""),
-        assert table_name == "portfolio_ledger"
+        timescaledb_password=unquote(parsed.password or ""),
-        return [{"name": name} for name in self._columns]
+        timescaledb_sslmode="disable",
+        timescaledb_connect_timeout_sec=3,
+        timescaledb_application_name="bat025-real-postgres-test",
+    )
 
 
-class _MissingLedgerInspector(_FakeInspector):
+@pytest.fixture()
-    def get_table_names(self) -> list[str]:
+def isolated_settings() -> Settings:
-        return ["other_table"]
+    maintenance_dsn = _maintenance_dsn()
+    database = f"portfolio_hub_bat025_{uuid4().hex}"
+    with psycopg.connect(maintenance_dsn, autocommit=True) as conn:
+        with conn.cursor() as cur:
+            cur.execute(sql.SQL("CREATE DATABASE {}").format(sql.Identifier(database)))
+    try:
+        yield _settings_for_database(maintenance_dsn, database)
+    finally:
+        with psycopg.connect(maintenance_dsn, autocommit=True) as conn:
+            with conn.cursor() as cur:
+                cur.execute(
+                    "SELECT pg_terminate_backend(pid) FROM pg_stat_activity "
+                    "WHERE datname = %s AND pid <> pg_backend_pid()",
+                    (database,),
+                )
+                cur.execute(sql.SQL("DROP DATABASE IF EXISTS {}").format(sql.Identifier(database)))
 
 
-class _FakeResult:
+def _repo(settings: Settings) -> PostgresPortfolioRepository:
-    def __init__(self, scalar_value: Any = None) -> None:
+    return PostgresPortfolioRepository(
-        self._scalar_value = scalar_value
+        PortfolioDBConfig(
-
+            host=settings.active_db_host,
-    def scalar(self) -> Any:
+            port=settings.active_db_port,
-        return self._scalar_value
+            dbname=settings.active_db_name,
+            user=settings.active_db_user,
+            password=settings.active_db_password,
+            sslmode=settings.active_db_sslmode,
+            connect_timeout_sec=settings.active_db_connect_timeout_sec,
+            application_name=settings.active_db_application_name,
+        )
+    )
 
 
-class _FakeBind:
+def _current_revision(settings: Settings) -> str:
-    def __init__(self) -> None:
+    with psycopg.connect(
-        self.queries: list[str] = []
+        host=settings.active_db_host,
-
+        port=settings.active_db_port,
-    def execute(self, statement: object, *args: object, **kwargs: object) -> _FakeResult:
+        dbname=settings.active_db_name,
-        self.queries.append(str(statement).strip())
+        user=settings.active_db_user,
-        return _FakeResult(0)
+        password=settings.active_db_password,
+        sslmode=settings.active_db_sslmode,
+    ) as conn:
+        with conn.cursor() as cur:
+            cur.execute("SELECT version_num FROM alembic_version")
+            return str(cur.fetchone()[0])
 
 
-def _sql_contains(queries: list[str], needle: str) -> bool:
+def _metadata_columns(settings: Settings) -> dict[str, tuple[str, str, str | None]]:
-    return any(needle in q for q in queries)
+    with psycopg.connect(
+        host=settings.active_db_host,
+        port=settings.active_db_port,
+        dbname=settings.active_db_name,
+        user=settings.active_db_user,
+        password=settings.active_db_password,
+        sslmode=settings.active_db_sslmode,
+    ) as conn:
+        with conn.cursor() as cur:
+            cur.execute(
+                """
+                SELECT column_name, data_type, is_nullable, column_default
+                FROM information_schema.columns
+                WHERE table_schema = 'public' AND table_name = 'portfolio_ledger'
+                  AND column_name IN ('source', 'target_ledger_entry_id',
+                                      'confirmation_evidence', 'actor_user_id')
+                """
+            )
+            return {row[0]: (row[1], row[2], row[3]) for row in cur.fetchall()}
 
 
-def test_upgrade_adds_all_columns_on_fresh_portfolio_ledger() -> None:
+def _seed_portfolio(repo: PostgresPortfolioRepository, suffix: str) -> str:
-    fake_bind = _FakeBind()
+    portfolio_id = f"bat025-{suffix}"
-    fake_inspector = _FakeInspector(columns=[])
+    user_id = f"bat025-user-{suffix}"
-
+    symbol = f"BAT025-{suffix}"
-    with (
+    with repo._connect().cursor() as cur:
-        patch.object(_migration.op, "get_bind", return_value=fake_bind),
+        cur.execute(
-        patch.object(_migration, "inspect", return_value=fake_inspector),
+            "INSERT INTO users (id, username) VALUES (%s, %s)",
-        patch.object(_migration.op, "add_column") as add_column,
+            (user_id, user_id),
-        patch.object(_migration.op, "alter_column") as alter_column,
+        )
-    ):
+        cur.execute(
-        _migration.upgrade()
+            "INSERT INTO assets (symbol, markets) VALUES (%s, ARRAY[]::text[]) RETURNING asset_id",
-
+            (symbol,),
-        assert add_column.call_count == 4
+        )
-        called = [call.args[1].name for call in add_column.call_args_list]
+        asset_id = cur.fetchone()[0]
-        assert called == [
+        cur.execute(
-            "source",
+            """
-            "target_ledger_entry_id",
+            INSERT INTO portfolios (
-            "confirmation_evidence",
+                portfolio_id, name, owner_user_id, cash_balance, asset_quantity, asset_price,
-            "actor_user_id",
+                profit_buffer, asset_id, broker, status
-        ]
+            ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
-        alter_column.assert_not_called()
+            """,
+            (portfolio_id, portfolio_id, user_id, 1_000_000, 0, 0, 0, asset_id, "bat025", "enabled"),
+        )
+    return portfolio_id
 
 
-def test_upgrade_preserves_data_and_enforces_source_when_precreated() -> None:
+def test_upgrade_from_pre_0028_preserves_business_data_and_backfills_source(
-    fake_bind = _FakeBind()
+    isolated_settings: Settings,
-    fake_inspector = _FakeInspector(columns=[
+) -> None:
-        "source",
+    """A real 0027 -> head upgrade preserves a representative ledger row."""
-        "target_ledger_entry_id",
+    from alembic import command
-        "confirmation_evidence",
-        "actor_user_id",
-    ])
 
-    with (
+    command.upgrade(get_alembic_config(isolated_settings), _PRE_BAT025)
-        patch.object(_migration.op, "get_bind", return_value=fake_bind),
+    repo = _repo(isolated_settings)
-        patch.object(_migration, "inspect", return_value=fake_inspector),
+    portfolio_id = _seed_portfolio(repo, "preserve")
-        patch.object(_migration.op, "add_column") as add_column,
+    recorded_at = datetime(2026, 7, 15, tzinfo=timezone.utc)
-        patch.object(_migration.op, "alter_column") as alter_column,
+    with repo._connect().cursor() as cur:
-    ):
+        cur.execute(
-        _migration.upgrade()
+            """
+            INSERT INTO portfolio_ledger (
+                portfolio_id, recorded_at, side, price, quantity, fee_amount, tax_amount,
+                cash_balance_after, asset_quantity_after, total_value_after,
+                profit_buffer_after, strategy_set_id, note, executed_at
+            ) VALUES (%s, %s, 'buy', 123.45, 7, 1, 0, 998_000, 7, 998_864.15, 0,
+                      'business-strategy', 'preserve-me', %s)
+            """,
+            (portfolio_id, recorded_at, recorded_at),
+        )
+        cur.execute("SELECT COUNT(*) FROM portfolio_ledger")
+        before_count = cur.fetchone()[0]
 
-        add_column.assert_not_called()
+    upgrade_to_head(isolated_settings)
-        alter_column.assert_called_once()
+
-        alter_kwargs = alter_column.call_args.kwargs
+    assert _current_revision(isolated_settings) == _HEAD
-        assert str(alter_kwargs["existing_type"]).upper().startswith("TEXT")
+    assert _metadata_columns(isolated_settings) == {
-        assert alter_kwargs["server_default"].text == "'broker'"
+        "source": ("text", "NO", "'broker'::text"),
-        assert alter_kwargs["nullable"] is False
+        "target_ledger_entry_id": ("bigint", "YES", None),
-        assert _sql_contains(fake_bind.queries, "UPDATE portfolio_ledger SET source = 'broker' WHERE source IS NULL")
+        "confirmation_evidence": ("text", "YES", None),
+        "actor_user_id": ("text", "YES", None),
+    }
+    with repo._connect().cursor() as cur:
+        cur.execute(
+            "SELECT COUNT(*), side, price::text, quantity::text, note, source "
+            "FROM portfolio_ledger GROUP BY side, price, quantity, note, source"
+        )
+        count, side, price, quantity, note, source = cur.fetchone()
+    assert count == before_count
+    assert (side, price, quantity, note, source) == ("buy", "123.45000000", "7.00000000", "preserve-me", "broker")
+    repo.close()
 
 
-def test_upgrade_noop_when_portfolio_ledger_missing() -> None:
+def test_runtime_precreated_schema_with_null_source_converges_without_duplicate_ddl(
-    fake_bind = _FakeBind()
+    isolated_settings: Settings,
-    fake_inspector = _MissingLedgerInspector(columns=[])
+) -> None:
+    """ensure_schema-first DB stamps 0027 then truly executes 0028 to head."""
+    repo = _repo(isolated_settings)
+    repo.ensure_schema()
+    portfolio_id = _seed_portfolio(repo, "runtime")
+    with repo._connect().cursor() as cur:
+        cur.execute("ALTER TABLE portfolio_ledger ALTER COLUMN source DROP NOT NULL")
+        cur.execute(
+            """
+            INSERT INTO portfolio_ledger (
+                portfolio_id, side, price, quantity, fee_amount, tax_amount,
+                cash_balance_after, asset_quantity_after, total_value_after,
+                profit_buffer_after, strategy_set_id, note, source
+            ) VALUES (%s, 'buy', 10, 1, 0, 0, 990, 1, 1000, 0, 'runtime', 'null-source', NULL)
+            """,
+            (portfolio_id,),
+        )
 
-    with (
+    assert bootstrap_or_upgrade_to_head(isolated_settings) == "head"
-        patch.object(_migration.op, "get_bind", return_value=fake_bind),
+    assert _current_revision(isolated_settings) == _HEAD
-        patch.object(_migration, "inspect", return_value=fake_inspector),
+    assert _metadata_columns(isolated_settings)["source"] == ("text", "NO", "'broker'::text")
-        patch.object(_migration.op, "add_column") as add_column,
+    with repo._connect().cursor() as cur:
-        patch.object(_migration.op, "alter_column") as alter_column,
+        cur.execute("SELECT source FROM portfolio_ledger WHERE portfolio_id = %s", (portfolio_id,))
-    ):
+        assert cur.fetchone()[0] == "broker"
-        _migration.upgrade()
+    repo.close()
-
-        add_column.assert_not_called()
-        alter_column.assert_not_called()
-        assert fake_bind.queries == []
 
 
-def test_downgrade_is_forward_only_noop() -> None:
+def test_broker_and_manual_ledger_writes_regress_after_real_upgrade(
-    with patch.object(_migration.op, "get_bind", return_value=MagicMock()):
+    isolated_settings: Settings,
-        result = _migration.downgrade()
+) -> None:
-    assert result is None
+    """Synthetic broker and manual writes use the migrated columns on real PostgreSQL."""
+    upgrade_to_head(isolated_settings)
+    repo = _repo(isolated_settings)
+    portfolio_id = _seed_portfolio(repo, "writes")
+
+    broker, created = repo.insert_ledger_trade(
+        portfolio_id=portfolio_id,
+        order_id="bat025-order",
+        execution_id="bat025-execution",
+        side="buy",
+        quantity=2,
+        price=100,
+        executed_at=datetime.now(timezone.utc),
+    )
+    manual = repo.record_manual_ledger(
+        portfolio_id,
+        action="INVEST",
+        amount=50,
+        actor_user_id="bat025-user",
+        confirmation_evidence="synthetic-local-fixture",
+    )
+
+    assert created is True
+    assert broker.source == "broker"
+    assert manual.source == "manual"
+    assert manual.actor_user_id == "bat025-user"
+    assert manual.confirmation_evidence == "synthetic-local-fixture"
+    with repo._connect().cursor() as cur:
+        cur.execute(
+            "SELECT source, actor_user_id, confirmation_evidence FROM portfolio_ledger "
+            "WHERE ledger_entry_id = %s",
+            (manual.ledger_entry_id,),
+        )
+        assert cur.fetchone() == ("manual", "bat025-user", "synthetic-local-fixture")
+    repo.close()