fix: make corporate actions atomic and recoverable

tests/release/test_release_contracts.py

@@ -17,6 +17,10 @@ def test_branch_verification_has_no_registry_write_authority_or_path():
     for forbidden in ("packages: write","docker login","docker push","REGISTRY_TOKEN","image_release.py publish"):
         assert forbidden not in text
     assert "uv sync --extra dev --locked" in text and "uv run pytest -q" in text
+    assert "Verify corporate-action atomicity and guarded recovery in PostgreSQL" in text
+    assert "tests/test_corporate_action_atomic_recovery_postgres.py" in text
+    assert "${POSTGRES_PASSWORD}@127.0.0.1:5432/postgres" in text
+    assert "***@" not in text
 
 def test_tag_publication_is_strict_single_build_handoff_without_latest():
     text=PUBLISH.read_text(); verify,publish=text.split("\n  publish:",1)
@@ -58,8 +62,8 @@ def test_release_version_and_migration_contract_has_current_head_0031():
         for path in (REPO / "app/db/alembic/versions").glob("*.py")
         if path.name != "__init__.py"
     )
-    assert 'version = "0.3.2"' in project
-    assert 'app_version: str = "0.3.2"' in config
+    assert 'version = "0.3.3"' in project
+    assert 'app_version: str = "0.3.3"' in config
     assert len(migrations) == 31
     assert migrations[-1].startswith("20260913_0031_")
 
@@ -85,7 +89,7 @@ def test_operations_separate_publication_deployment_migration_and_db_rollback():
         "Image rollback is **not** database rollback",
         "read-only root filesystem",
         "10001:10001",
-        "PortfolioHub v0.3.2 container contract",
+        "PortfolioHub v0.3.3 container contract",
         "v0.2.1 includes migration `0030`",
         "trex-web v0.2.3 displays `strategy_nick`",
     ):

tests/test_corporate_action_atomic_recovery_postgres.py

@@ -0,0 +1,306 @@
+"""Real PostgreSQL regressions for atomic corporate-action writes and recovery."""
+from __future__ import annotations
+
+import os
+from datetime import date
+from decimal import Decimal
+from urllib.parse import unquote, urlparse
+from uuid import uuid4
+
+import psycopg
+import pytest
+from psycopg import sql
+
+from app.core.config import Settings
+from app.db.migrations import upgrade_to_head
+from app.domain.corporate_actions import canonical_payload_hash
+from app.domain.models import CorporateAction
+from app.repositories.postgres import PortfolioDBConfig, PostgresPortfolioRepository
+from app.schemas.corporate_actions import CorporateActionRequest
+
+_LOCAL_DSN_ENV = "PORTFOLIOHUB_TEST_POSTGRES_DSN"
+_LOOPBACK_HOSTS = {"127.0.0.1", "localhost", "::1"}
+_PORTFOLIO_ID = "3573f4d6-cc27-4f91-82d9-6330e5250d20"
+_CASH = Decimal("400060.62490000")
+_TOTAL_COST = Decimal("1577211.97990000")
+
+
+def _maintenance_dsn() -> str:
+    dsn = os.environ.get(_LOCAL_DSN_ENV)
+    if not dsn:
+        pytest.skip(f"{_LOCAL_DSN_ENV} is required for real PostgreSQL integration tests")
+    parsed = urlparse(dsn)
+    if parsed.scheme not in {"postgres", "postgresql"}:
+        pytest.fail("test DSN must use postgres/postgresql")
+    if parsed.hostname not in _LOOPBACK_HOSTS or parsed.path.lstrip("/") != "postgres":
+        pytest.fail("tests require a loopback PostgreSQL maintenance database named postgres")
+    return dsn
+
+
+def _settings(dsn: str, database: str) -> Settings:
+    parsed = urlparse(dsn)
+    return Settings(
+        db_enabled=True,
+        timescaledb_host=parsed.hostname,
+        timescaledb_port=parsed.port or 5432,
+        timescaledb_name=database,
+        timescaledb_user=unquote(parsed.username or ""),
+        timescaledb_password=unquote(parsed.password or ""),
+        timescaledb_sslmode="disable",
+        timescaledb_connect_timeout_sec=3,
+        timescaledb_application_name="corporate-action-atomic-test",
+    )
+
+
+def _repo(settings: Settings) -> PostgresPortfolioRepository:
+    return PostgresPortfolioRepository(
+        PortfolioDBConfig(
+            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,
+            connect_timeout_sec=settings.active_db_connect_timeout_sec,
+            application_name=settings.active_db_application_name,
+        )
+    )
+
+
+@pytest.fixture()
+def repository() -> PostgresPortfolioRepository:
+    maintenance_dsn = _maintenance_dsn()
+    database = f"portfolio_hub_ca_{uuid4().hex}"
+    with psycopg.connect(maintenance_dsn, autocommit=True) as conn, conn.cursor() as cur:
+        cur.execute(sql.SQL("CREATE DATABASE {}").format(sql.Identifier(database)))
+    settings = _settings(maintenance_dsn, database)
+    upgrade_to_head(settings)
+    repo = _repo(settings)
+    try:
+        yield repo
+    finally:
+        repo.close()
+        with psycopg.connect(maintenance_dsn, autocommit=True) as conn, 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)))
+
+
+def _seed_daekyo(repo: PostgresPortfolioRepository, *, quantity: Decimal) -> str:
+    user_id = "ca-recovery-user"
+    with repo._connect().cursor() as cur:
+        cur.execute("INSERT INTO users (id, username) VALUES (%s, %s)", (user_id, user_id))
+        cur.execute(
+            "INSERT INTO assets (symbol, markets) VALUES ('019685', ARRAY['KRX']) RETURNING asset_id"
+        )
+        asset_id = str(cur.fetchone()[0])
+        cur.execute(
+            """
+            INSERT INTO portfolios (
+                portfolio_id, name, owner_user_id, cash_balance, asset_quantity,
+                asset_price, profit_buffer, asset_id, broker, status
+            ) VALUES (%s, '대교우B', %s, %s, %s, 643, 0, %s, 'kis', 'enabled')
+            """,
+            (_PORTFOLIO_ID, user_id, _CASH, quantity, asset_id),
+        )
+        # The fee makes the authoritative BUY-ledger cost exactly 1,577,211.9799.
+        cur.execute(
+            """
+            INSERT INTO portfolio_ledger (
+                portfolio_id, recorded_at, side, source, quantity, price, fee_amount, tax_amount,
+                cash_balance_after, asset_quantity_after, total_value_after,
+                profit_buffer_after, strategy_set_id, note, executed_at
+            ) VALUES (%s, '2026-09-09T00:00:00Z', 'buy', 'broker', 1452, 1086, 339.9799, 0,
+                      %s, 1452, %s, 0, 'runtime', 'seed', '2026-09-09T00:00:00Z')
+            """,
+            (_PORTFOLIO_ID, _CASH, _CASH + Decimal(1452) * Decimal(643)),
+        )
+    return asset_id
+
+
+def _request(*, resume_partial: bool = False) -> CorporateActionRequest:
+    return CorporateActionRequest(
+        effective_date=date(2026, 9, 10),
+        quantity_before=Decimal(1452),
+        quantity_after=Decimal(726),
+        evidence_ref="KIS 019685 2026-09-10",
+        resume_partial=resume_partial,
+    )
+
+
+def _action(asset_id: str, request: CorporateActionRequest, key: str) -> CorporateAction:
+    digest = canonical_payload_hash(request)
+    return CorporateAction(
+        str(uuid4()),
+        _PORTFOLIO_ID,
+        asset_id,
+        request.action_type,
+        request.effective_date,
+        request.quantity_before,
+        request.quantity_after,
+        Decimal(0),
+        Decimal(0),
+        Decimal(0),
+        _CASH,
+        Decimal(0),
+        request.evidence_ref,
+        key,
+        digest,
+        0,
+        "ca-recovery-user",
+    )
+
+
+def _record(repo: PostgresPortfolioRepository, asset_id: str, request: CorporateActionRequest, key: str):
+    action = _action(asset_id, request, key)
+    return repo.record_corporate_action(
+        _PORTFOLIO_ID,
+        action=action,
+        idempotency_key=key,
+        payload_hash=action.payload_hash,
+        resume_partial=request.resume_partial,
+    )
+
+
+def _state(repo: PostgresPortfolioRepository) -> tuple:
+    with repo._connect().cursor() as cur:
+        cur.execute(
+            "SELECT asset_quantity, asset_price, cash_balance, profit_buffer "
+            "FROM portfolios WHERE portfolio_id=%s",
+            (_PORTFOLIO_ID,),
+        )
+        portfolio_row = cur.fetchone()
+        assert portfolio_row is not None
+        portfolio = tuple(Decimal(str(value)) for value in portfolio_row)
+        cur.execute(
+            "SELECT COUNT(*) FROM portfolio_ledger "
+            "WHERE portfolio_id=%s AND source='corporate_action'",
+            (_PORTFOLIO_ID,),
+        )
+        ledger_count = cur.fetchone()[0]
+        cur.execute(
+            "SELECT COUNT(*) FROM portfolio_corporate_actions WHERE portfolio_id=%s",
+            (_PORTFOLIO_ID,),
+        )
+        action_count = cur.fetchone()[0]
+    return (*portfolio, ledger_count, action_count)
+
+
+def test_stock_consolidation_persists_reference_price_total_value_and_audit(repository):
+    asset_id = _seed_daekyo(repository, quantity=Decimal(1452))
+
+    saved = _record(repository, asset_id, _request(), "ca-normal")
+
+    assert saved.average_price_after == Decimal("2172.46829187")
+    assert saved.total_cost_basis == _TOTAL_COST
+    assert _state(repository) == (
+        Decimal("726.00000000"),
+        Decimal("1286.00000000"),
+        _CASH,
+        Decimal("0E-8"),
+        1,
+        1,
+    )
+    with repository._connect().cursor() as cur:
+        cur.execute(
+            "SELECT price, total_value_after FROM portfolio_ledger "
+            "WHERE portfolio_id=%s AND source='corporate_action'",
+            (_PORTFOLIO_ID,),
+        )
+        assert cur.fetchone() == (
+            Decimal("1286.00000000"),
+            Decimal("1333696.62490000"),
+        )
+
+
+def test_ledger_failure_rolls_back_portfolio_quantity_and_reference_price(repository):
+    asset_id = _seed_daekyo(repository, quantity=Decimal(1452))
+    with repository._connect().cursor() as cur:
+        cur.execute(
+            """
+            CREATE FUNCTION reject_corporate_action_ledger() RETURNS trigger AS $$
+            BEGIN
+                IF NEW.source = 'corporate_action' THEN
+                    RAISE EXCEPTION 'injected ledger failure';
+                END IF;
+                RETURN NEW;
+            END;
+            $$ LANGUAGE plpgsql
+            """
+        )
+        cur.execute(
+            "CREATE TRIGGER reject_ca_ledger BEFORE INSERT ON portfolio_ledger "
+            "FOR EACH ROW EXECUTE FUNCTION reject_corporate_action_ledger()"
+        )
+
+    with pytest.raises(psycopg.Error, match="injected ledger failure"):
+        _record(repository, asset_id, _request(), "ca-failure")
+
+    assert _state(repository) == (
+        Decimal("1452.00000000"),
+        Decimal("643.00000000"),
+        _CASH,
+        Decimal("0E-8"),
+        0,
+        0,
+    )
+
+
+def test_audit_failure_rolls_back_portfolio_and_inserted_ledger(repository):
+    asset_id = _seed_daekyo(repository, quantity=Decimal(1452))
+    before = _state(repository)
+    with repository._connect().cursor() as cur:
+        cur.execute(
+            """
+            CREATE FUNCTION reject_corporate_action_audit() RETURNS trigger AS $$
+            BEGIN
+                RAISE EXCEPTION 'injected corporate-action audit failure';
+            END;
+            $$ LANGUAGE plpgsql
+            """
+        )
+        cur.execute(
+            "CREATE TRIGGER reject_ca_audit BEFORE INSERT ON portfolio_corporate_actions "
+            "FOR EACH ROW EXECUTE FUNCTION reject_corporate_action_audit()"
+        )
+
+    with pytest.raises(psycopg.Error, match="injected corporate-action audit failure"):
+        _record(repository, asset_id, _request(), "ca-audit-failure")
+
+    assert _state(repository) == before
+
+
+def test_explicit_resume_repairs_only_verified_quantity_only_partial_state(repository):
+    asset_id = _seed_daekyo(repository, quantity=Decimal(726))
+    request = _request(resume_partial=True)
+
+    first = _record(repository, asset_id, request, "ca-recovery")
+    replay = _record(repository, asset_id, request, "ca-recovery")
+
+    assert first.idempotent_replay is False
+    assert replay.idempotent_replay is True
+    assert replay.ledger_entry_id == first.ledger_entry_id
+    assert _state(repository) == (
+        Decimal("726.00000000"),
+        Decimal("1286.00000000"),
+        _CASH,
+        Decimal("0E-8"),
+        1,
+        1,
+    )
+
+    with pytest.raises(ValueError, match="CORPORATE_ACTION_RECOVERY_ALREADY_RECORDED"):
+        _record(repository, asset_id, request, "ca-recovery-different-key")
+    assert _state(repository)[-2:] == (1, 1)
+
+
+def test_resume_rejects_a_normal_unapplied_snapshot(repository):
+    asset_id = _seed_daekyo(repository, quantity=Decimal(1452))
+
+    with pytest.raises(ValueError, match="CORPORATE_ACTION_RECOVERY_STATE_MISMATCH"):
+        _record(repository, asset_id, _request(resume_partial=True), "ca-invalid-recovery")
+
+    assert _state(repository)[-2:] == (0, 0)

tests/test_issue34_regressions.py

@@ -44,6 +44,23 @@ def test_canonical_hash_same_payload_and_changed_payload_differ():
     assert canonical_payload_hash(a) != canonical_payload_hash(b)
 
 
+def test_default_resume_flag_preserves_pre_recovery_idempotency_hash():
+    request = CorporateActionRequest(
+        effective_date=date(2026, 1, 2),
+        quantity_before=Decimal(10),
+        quantity_after=Decimal(5),
+        evidence_ref="ev",
+    )
+    legacy_payload = request.model_dump(mode="json", exclude={"resume_partial"})
+    explicit_default_payload = {**legacy_payload, "resume_partial": False}
+
+    assert canonical_payload_hash(request) == canonical_payload_hash(legacy_payload)
+    assert canonical_payload_hash(explicit_default_payload) == canonical_payload_hash(legacy_payload)
+    assert canonical_payload_hash(
+        request.model_copy(update={"resume_partial": True})
+    ) != canonical_payload_hash(legacy_payload)
+
+
 def test_runtime_schema_has_one_corporate_action_ddl_and_replay_mapping_starts_at_18():
     source = Path("app/repositories/postgres.py").read_text()
     assert source.count("CREATE TABLE IF NOT EXISTS portfolio_corporate_actions") == 1

tests/test_issue34_runtime_blockers.py

@@ -7,7 +7,7 @@ import pytest
 from fastapi.testclient import TestClient
 
 from app.api import v1_portfolios
-from app.domain.models import CorporateAction
+from app.domain.models import CorporateAction, Portfolio
 from app.main import create_app
 from app.repositories.postgres import PostgresPortfolioRepository
 
@@ -20,6 +20,46 @@ class MissingPortfolioRepo:
         pass
 
 
+class ResumeCaptureRepo:
+    def __init__(self):
+        self.resume_partial = None
+
+    def get(self, portfolio_id):
+        return Portfolio(
+            portfolio_id=portfolio_id,
+            broker="kis",
+            status="enabled",
+            name="대교우B",
+            symbol="019685",
+            asset_id="asset-1",
+            owner_user_id="u1",
+            cash_balance=400060.6249,
+            asset_quantity=726,
+            price=643,
+            profit_buffer=0,
+        )
+
+    def record_corporate_action(
+        self,
+        portfolio_id,
+        *,
+        action,
+        idempotency_key,
+        payload_hash,
+        resume_partial=False,
+    ):
+        self.resume_partial = resume_partial
+        action.average_price_before = Decimal("1086.23414594")
+        action.average_price_after = Decimal("2172.46829187")
+        action.total_cost_basis = Decimal("1577211.97990000")
+        action.ledger_entry_id = 7
+        action.recorded_at = datetime(2026, 9, 13, tzinfo=UTC)
+        return action
+
+    def close(self):
+        pass
+
+
 def test_create_corporate_action_maps_postgres_missing_portfolio_to_404(monkeypatch):
     app = create_app()
     repo = MissingPortfolioRepo()
@@ -35,6 +75,31 @@ def test_create_corporate_action_maps_postgres_missing_portfolio_to_404(monkeypa
     assert response.json()["detail"] == "Portfolio not found"
 
 
+def test_corporate_action_api_passes_explicit_resume_mode(monkeypatch):
+    app = create_app()
+    repo = ResumeCaptureRepo()
+    monkeypatch.setattr(v1_portfolios, "_get_repo", lambda: repo)
+    app.dependency_overrides[v1_portfolios.require_trade_write] = lambda: type(
+        "Auth", (), {"user_id": "u1", "permission": "trade_write"}
+    )()
+
+    response = TestClient(app).post(
+        "/v1/portfolios/pf-1/corporate-actions",
+        headers={"Idempotency-Key": "resume-1"},
+        json={
+            "effective_date": "2026-09-10",
+            "quantity_before": "1452",
+            "quantity_after": "726",
+            "evidence_ref": "KIS 019685 2026-09-10",
+            "resume_partial": True,
+        },
+    )
+
+    assert response.status_code == 201
+    assert response.json()["quantity_after"] == "726"
+    assert repo.resume_partial is True
+
+
 def _action(asset_id="asset-1"):
     return CorporateAction("ca-1", "pf-1", asset_id, "stock_consolidation", date(2026, 1, 1), Decimal(2), Decimal(1), Decimal(0), Decimal(0), Decimal(0), Decimal(10), Decimal(0), "ev", "idem", "hash", 0)
 
@@ -47,7 +112,12 @@ def _repo_for_row(asset_id):
     conn.__enter__.return_value = conn
     conn.cursor.return_value.__enter__.return_value = cur
     repo._connect = MagicMock(return_value=conn)
-    cur.fetchone.side_effect = [(Decimal(2), Decimal(10), Decimal(0), asset_id, "enabled"), None, None, None, (7, date(2026, 1, 1))]
+    cur.fetchone.side_effect = [
+        (Decimal(2), Decimal(10), Decimal(0), asset_id, "enabled", Decimal(643)),
+        None,
+        None,
+        (7, date(2026, 1, 1)),
+    ]
     cur.fetchall.return_value = [("buy", Decimal(2), Decimal(0), Decimal(0), Decimal(0), datetime(2025, 12, 31, tzinfo=UTC))]
     return repo
 
@@ -74,8 +144,7 @@ def test_record_corporate_action_accepts_daekyo_rounding_residue():
     conn.cursor.return_value.__enter__.return_value = cur
     repo._connect = MagicMock(return_value=conn)
     cur.fetchone.side_effect = [
-        (Decimal(1452), Decimal(10), Decimal(0), "asset-1", "enabled"),
-        None,
+        (Decimal(1452), Decimal(10), Decimal(0), "asset-1", "enabled", Decimal(643)),
         None,
         None,
         (7, datetime(2026, 9, 13, tzinfo=UTC)),