Merge pull request #18 from fourblocks2020/feat/bat-021-req047-step2-finalize

BAT-021: REQ-047 Step2 마무리 — daemon-scoped PK 동반수정 + 실DB 검증 + 런북

app/repositories/postgres.py

@@ -406,16 +406,20 @@ class PostgresPortfolioRepository:
                 ON daemons (owner_user_id)
                 """
             )
+            # NOTE(REQ-047 Step2 / BAT-021): daemon-scoped PK 전환 이후 최종 스키마.
+            # 이 CREATE TABLE IF NOT EXISTS는 Step2 적용 환경에서 테이블이 이미 존재해 스킵된다.
+            # Step2 미적용 환경에서는 alembic revision(Step1→Step2)을 먼저 적용해야 한다.
             cur.execute(
                 """
                 CREATE TABLE IF NOT EXISTS broker_credentials (
-                    user_id TEXT NOT NULL,
+                    daemon_id TEXT NOT NULL,
                     broker TEXT NOT NULL,
+                    user_id TEXT NOT NULL,
                     account_ref TEXT NOT NULL,
                     encrypted_payload TEXT NOT NULL,
                     credential_version INTEGER NOT NULL DEFAULT 1,
                     last_rotated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
-                    PRIMARY KEY (user_id, broker)
+                    PRIMARY KEY (daemon_id, broker)
                 )
                 """
             )
@@ -465,16 +469,17 @@ class PostgresPortfolioRepository:
 
     @staticmethod
     def _row_to_broker_credential(row: Any) -> BrokerCredential:
+        """Step2(REQ-047) 이후 컬럼 순서: daemon_id, user_id, broker, account_ref, encrypted_payload, credential_version, last_rotated_at."""
         if row is None:
             raise ValueError("브로커 인증 정보를 찾을 수 없습니다")
         return BrokerCredential(
-            user_id=row[0],
+            daemon_id=row[0],
-            broker=row[1],
+            user_id=row[1],
-            account_ref=row[2],
+            broker=row[2],
-            encrypted_payload=row[3],
+            account_ref=row[3],
-            credential_version=int(row[4]) if len(row) > 4 else 1,
+            encrypted_payload=row[4],
-            last_rotated_at=row[5] if len(row) > 5 else None,
+            credential_version=int(row[5]) if len(row) > 5 else 1,
-            daemon_id=row[6] if len(row) > 6 else None,
+            last_rotated_at=row[6] if len(row) > 6 else None,
         )
 
     @staticmethod
@@ -747,40 +752,23 @@ class PostgresPortfolioRepository:
             cur.execute("SELECT 1 FROM users WHERE id = %s", (user_id,))
             if cur.fetchone() is None:
                 raise ValueError("사용자를 찾을 수 없습니다")
-            # NOTE(REQ-046): daemon_id 컬럼은 REQ-047 Step1 마이그레이션 이후 존재.
+            # NOTE(REQ-047 Step2 / BAT-021): PK 전환 이후 ON CONFLICT (daemon_id, broker).
-            # Step1 적용 전 환경에서는 daemon_id 컬럼이 없어 INSERT 실패 가능.
+            # daemon_id는 NOT NULL(Step2 제약). Step2 미적용 환경에서는 이 코드를 배포하지 마라.
-            # Step2(PK 전환) 이후에는 ON CONFLICT 절도 (daemon_id, broker)로 변경 필요.
+            # 배포 순서: Step1 alembic upgrade → backfill CLI → Step2 alembic upgrade → 코드 배포.
-            # 현재는 기존 PK (user_id, broker) ON CONFLICT 유지하고 daemon_id를 컬럼에 포함.
+            cur.execute(
-            try:
+                """
-                cur.execute(
+                INSERT INTO broker_credentials (daemon_id, user_id, broker, account_ref, encrypted_payload)
-                    """
+                VALUES (%s, %s, %s, %s, %s)
-                    INSERT INTO broker_credentials (user_id, broker, account_ref, encrypted_payload, daemon_id)
+                ON CONFLICT (daemon_id, broker) DO UPDATE
-                    VALUES (%s, %s, %s, %s, %s)
+                SET account_ref = EXCLUDED.account_ref,
-                    ON CONFLICT (user_id, broker) DO UPDATE
+                    encrypted_payload = EXCLUDED.encrypted_payload,
-                    SET account_ref = EXCLUDED.account_ref,
+                    user_id = EXCLUDED.user_id,
-                        encrypted_payload = EXCLUDED.encrypted_payload,
+                    credential_version = broker_credentials.credential_version + 1,
-                        daemon_id = EXCLUDED.daemon_id,
+                    last_rotated_at = NOW()
-                        credential_version = broker_credentials.credential_version + 1,
+                RETURNING daemon_id, user_id, broker, account_ref, encrypted_payload, credential_version, last_rotated_at
-                        last_rotated_at = NOW()
+                """,
-                    RETURNING user_id, broker, account_ref, encrypted_payload, credential_version, last_rotated_at, daemon_id
+                (daemon_id, user_id, broker, account_ref, encrypted_payload),
-                    """,
+            )
-                    (user_id, broker, account_ref, encrypted_payload, daemon_id),
-                )
-            except Exception:
-                # Fallback: daemon_id 컬럼이 아직 없는 환경(Step1 미적용)
-                cur.execute(
-                    """
-                    INSERT INTO broker_credentials (user_id, broker, account_ref, encrypted_payload)
-                    VALUES (%s, %s, %s, %s)
-                    ON CONFLICT (user_id, broker) DO UPDATE
-                    SET account_ref = EXCLUDED.account_ref,
-                        encrypted_payload = EXCLUDED.encrypted_payload,
-                        credential_version = broker_credentials.credential_version + 1,
-                        last_rotated_at = NOW()
-                    RETURNING user_id, broker, account_ref, encrypted_payload, credential_version, last_rotated_at
-                    """,
-                    (user_id, broker, account_ref, encrypted_payload),
-                )
             row = cur.fetchone()
         return self._row_to_broker_credential(row)
 
@@ -790,43 +778,30 @@ class PostgresPortfolioRepository:
         daemon_id: str | None = None,
     ) -> list[BrokerCredential]:
         self.ensure_schema()
+        # NOTE(REQ-047 Step2 / BAT-021): daemon_id 컬럼은 Step2 이후 항상 존재(NOT NULL).
+        # Step2 미적용 환경에서는 이 코드를 배포하지 마라.
         with self._connect().cursor() as cur:
-            # NOTE(REQ-046): daemon_id 컬럼은 REQ-047 Step1 이후 존재.
+            if daemon_id is not None:
-            # daemon_id 컬럼이 없는 환경(Step1 미적용)에서는 6컬럼 조회 fallback.
-            try:
-                if daemon_id is not None:
-                    cur.execute(
-                        """
-                        SELECT user_id, broker, account_ref, encrypted_payload, credential_version, last_rotated_at, daemon_id
-                        FROM broker_credentials
-                        WHERE user_id = %s AND daemon_id = %s
-                        ORDER BY broker ASC
-                        """,
-                        (user_id, daemon_id),
-                    )
-                else:
-                    cur.execute(
-                        """
-                        SELECT user_id, broker, account_ref, encrypted_payload, credential_version, last_rotated_at, daemon_id
-                        FROM broker_credentials
-                        WHERE user_id = %s
-                        ORDER BY broker ASC
-                        """,
-                        (user_id,),
-                    )
-                rows = cur.fetchall()
-            except Exception:
-                # Fallback: daemon_id 컬럼이 아직 없는 환경(Step1 미적용)
                 cur.execute(
                     """
-                    SELECT user_id, broker, account_ref, encrypted_payload, credential_version, last_rotated_at
+                    SELECT daemon_id, user_id, broker, account_ref, encrypted_payload, credential_version, last_rotated_at
+                    FROM broker_credentials
+                    WHERE user_id = %s AND daemon_id = %s
+                    ORDER BY broker ASC
+                    """,
+                    (user_id, daemon_id),
+                )
+            else:
+                cur.execute(
+                    """
+                    SELECT daemon_id, user_id, broker, account_ref, encrypted_payload, credential_version, last_rotated_at
                     FROM broker_credentials
                     WHERE user_id = %s
                     ORDER BY broker ASC
                     """,
                     (user_id,),
                 )
-                rows = cur.fetchall()
+            rows = cur.fetchall()
         return [self._row_to_broker_credential(row) for row in rows]
 
     def update_broker_credential_daemon_id(

docs/batches/bat-021_20260628_req047-step2-integration-and-runbook/design.md

@@ -0,0 +1,135 @@
+# BAT-021 Design
+
+## 변경 설계
+
+### T-1/T-2/T-3: `app/repositories/postgres.py` 수정
+
+#### ensure_schema() broker_credentials (T-1)
+
+현재 `ensure_schema()`의 `broker_credentials` CREATE TABLE은:
+```sql
+CREATE TABLE IF NOT EXISTS broker_credentials (
+    user_id TEXT NOT NULL,
+    broker TEXT NOT NULL,
+    account_ref TEXT NOT NULL,
+    encrypted_payload TEXT NOT NULL,
+    credential_version INTEGER NOT NULL DEFAULT 1,
+    last_rotated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    PRIMARY KEY (user_id, broker)
+)
+```
+
+Step2 이후 최종 스키마:
+- `daemon_id TEXT NOT NULL` 추가
+- `PK = (daemon_id, broker)`
+- `user_id` 컬럼 유지 (NOT NULL — 신규 INSERT는 항상 user_id 포함)
+
+갱신 후:
+```sql
+CREATE TABLE IF NOT EXISTS broker_credentials (
+    daemon_id TEXT NOT NULL,
+    broker TEXT NOT NULL,
+    user_id TEXT NOT NULL,
+    account_ref TEXT NOT NULL,
+    encrypted_payload TEXT NOT NULL,
+    credential_version INTEGER NOT NULL DEFAULT 1,
+    last_rotated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    PRIMARY KEY (daemon_id, broker)
+)
+```
+
+`ensure_schema()`는 `IF NOT EXISTS`를 사용하므로 테이블이 이미 존재하면 CREATE를 스킵한다.
+Step2 마이그레이션이 적용된 DB에서는 이 정의와 실제 스키마가 정합된다.
+Step1만 적용된 DB(Step2 미적용)에서 `ensure_schema()`를 호출하면 테이블은 이미 존재하므로 스킵되며 영향 없다.
+
+Step2 이후 추가될 인덱스:
+```sql
+CREATE INDEX IF NOT EXISTS idx_broker_credentials_daemon_broker
+ON broker_credentials (daemon_id, broker)
+```
+이 인덱스는 PK 자체이므로 별도 인덱스는 불필요. Step1에서 만든 `idx_broker_credentials_daemon_id` 인덱스는 그대로 유지.
+
+#### upsert_broker_credential (T-2)
+
+현재:
+- Primary INSERT: `ON CONFLICT (user_id, broker)` — Step2 이후 PK가 바뀌어 동작 안 함
+- Fallback: `ON CONFLICT (user_id, broker)` without daemon_id — Step1 미적용 환경용
+
+Step2 이후 세계에서:
+- daemon_id는 항상 NOT NULL (Step2 PK 제약)
+- ON CONFLICT는 `(daemon_id, broker)` 로 변경
+- Fallback 불필요 → 제거
+
+단, BAT-020의 fallback try/except는 `psycopg` 트랜잭션 abort 이슈가 있었다:
+트랜잭션이 abort 상태에서 fallback 쿼리를 실행하면 `InFailedSqlTransaction` 오류가 발생.
+이 batch에서 fallback을 제거하면서 이 이슈도 동시에 해소된다.
+
+갱신 후 `upsert_broker_credential`:
+```python
+cur.execute(
+    """
+    INSERT INTO broker_credentials (daemon_id, user_id, broker, account_ref, encrypted_payload)
+    VALUES (%s, %s, %s, %s, %s)
+    ON CONFLICT (daemon_id, broker) DO UPDATE
+    SET account_ref = EXCLUDED.account_ref,
+        encrypted_payload = EXCLUDED.encrypted_payload,
+        user_id = EXCLUDED.user_id,
+        credential_version = broker_credentials.credential_version + 1,
+        last_rotated_at = NOW()
+    RETURNING daemon_id, user_id, broker, account_ref, encrypted_payload, credential_version, last_rotated_at
+    """,
+    (daemon_id, user_id, broker, account_ref, encrypted_payload),
+)
+```
+
+RETURNING 절에서 `daemon_id` 컬럼 순서 변경: `_row_to_broker_credential`의 현재 row 매핑 확인 필요.
+
+현재 `_row_to_broker_credential`:
+```python
+return BrokerCredential(
+    user_id=row[0],   # user_id
+    broker=row[1],    # broker
+    account_ref=row[2],
+    encrypted_payload=row[3],
+    credential_version=int(row[4]),
+    last_rotated_at=row[5],
+    daemon_id=row[6],  # optional
+)
+```
+
+RETURNING 컬럼 순서를 유지해야 한다. 새 RETURNING:
+`daemon_id(0), user_id(1), broker(2), account_ref(3), encrypted_payload(4), credential_version(5), last_rotated_at(6)`
+→ `_row_to_broker_credential` 매핑도 함께 수정 필요.
+
+`list_broker_credentials`도 같은 방향으로 정렬:
+`SELECT daemon_id, user_id, broker, account_ref, encrypted_payload, credential_version, last_rotated_at`
+
+`_row_to_broker_credential` 갱신:
+```python
+return BrokerCredential(
+    daemon_id=row[0],
+    user_id=row[1],
+    broker=row[2],
+    account_ref=row[3],
+    encrypted_payload=row[4],
+    credential_version=int(row[5]),
+    last_rotated_at=row[6],
+)
+```
+
+#### list_broker_credentials fallback 제거 (T-3)
+
+Step2 이후 daemon_id 컬럼이 항상 존재하므로 try/except fallback 제거.
+
+### 영향 범위
+
+- `app/repositories/postgres.py` — ensure_schema, upsert_broker_credential, list_broker_credentials, _row_to_broker_credential
+- `tests/` — bat021 신규 테스트, 기존 FakeRepo 의존 테스트 영향 없음 (FakeRepo는 실제 DB 쿼리 미사용)
+
+### 하위 호환성
+
+Step2 PK 전환 이후에만 이 코드가 올바르게 동작한다.
+Step2 미적용 DB에서는 `ON CONFLICT (daemon_id, broker)` 제약이 없어 오류 발생 가능.
+런북에서 Step2 적용 이전에 새 코드를 배포하지 말 것을 명시한다.
+
+배포 순서: Step1 → backfill → Step2 → 새 코드 배포.

docs/batches/bat-021_20260628_req047-step2-integration-and-runbook/implementation.md

@@ -0,0 +1,60 @@
+# BAT-021 Implementation
+
+## 변경 파일
+
+### `app/repositories/postgres.py`
+
+**T-1: ensure_schema() broker_credentials CREATE TABLE 갱신**
+
+구 정의 (Step1 이전):
+```sql
+PRIMARY KEY (user_id, broker)
+-- daemon_id 없음
+```
+
+신 정의 (Step2 이후):
+```sql
+daemon_id TEXT NOT NULL,
+broker TEXT NOT NULL,
+user_id TEXT NOT NULL,
+account_ref TEXT NOT NULL,
+encrypted_payload TEXT NOT NULL,
+credential_version INTEGER NOT NULL DEFAULT 1,
+last_rotated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+PRIMARY KEY (daemon_id, broker)
+```
+
+**T-2: `_row_to_broker_credential` Step2 컬럼 순서 적용**
+
+구 매핑: `row[0]=user_id, row[1]=broker, ..., row[6]=daemon_id`
+
+신 매핑: `row[0]=daemon_id, row[1]=user_id, row[2]=broker, row[3]=account_ref, row[4]=encrypted_payload, row[5]=credential_version, row[6]=last_rotated_at`
+
+**T-3: `upsert_broker_credential` ON CONFLICT 수정 + fallback 제거**
+
+- `ON CONFLICT (user_id, broker)` → `ON CONFLICT (daemon_id, broker)`
+- `RETURNING` 컬럼 순서: `daemon_id, user_id, broker, account_ref, encrypted_payload, credential_version, last_rotated_at`
+- `try/except` fallback 완전 제거 (트랜잭션 abort 이슈 동시 해소)
+
+**T-4: `list_broker_credentials` fallback 제거 + SELECT 컬럼 순서 수정**
+
+- `try/except` fallback 완전 제거
+- `SELECT daemon_id, user_id, broker, ...` 순서로 변경
+
+### `tests/test_bat020_broker_credential_domain.py`
+
+- `test_row_to_broker_credential_with_daemon_id`: 행 순서를 Step2 컬럼 순서로 수정 (AC-3 갱신)
+- `test_row_to_broker_credential_without_daemon_id`: Step2 5컬럼 기본값 테스트로 갱신 (AC-4 갱신)
+- `_make_rows_for_user`: Step2 컬럼 순서로 갱신
+
+### 신규 파일
+
+- `tests/test_bat021_step2_ensure_schema_and_upsert.py`: 12개 신규 테스트
+- `docs/runbooks/bat-021-req047-step2-migration-runbook.md`
+- `docs/batches/bat-021_20260628_req047-step2-integration-and-runbook/` (index, planning, design, implementation, verification)
+
+## 테스트 count (구현 후)
+
+- 기존: 509 passed
+- 신규: 12 추가 (test_bat021_step2_ensure_schema_and_upsert.py)
+- 합계: 521 passed

docs/batches/bat-021_20260628_req047-step2-integration-and-runbook/index.md

@@ -0,0 +1,28 @@
+# BAT-021 Index
+
+- Profile: standard
+- Status: in-progress
+- Included REQ: REQ-047
+- Discovery: dcy-010
+- Created: 2026-06-28
+
+## Scope Summary
+
+REQ-047 Step2 마무리 batch.
+
+BAT-020(PR #17, d4acfd4)에서 Step1/Step2 마이그레이션 revision 코드가 작성됐으나
+실 PostgreSQL 통합검증이 미완이고, Step2 적용 후 동반 코드 수정 2건이 TODO로 잔존했다.
+
+이 batch는:
+1. `ensure_schema()` `broker_credentials` CREATE TABLE을 최종 daemon-scoped 스키마로 갱신
+2. `upsert_broker_credential` `ON CONFLICT (user_id, broker)` → `ON CONFLICT (daemon_id, broker)` 수정 + fallback 트랜잭션 abort 이슈 정리
+3. dev/test PostgreSQL에서 Step1 → backfill → Step2 → downgrade 통합검증 (AC-1~7)
+4. 운영 적용 런북 작성
+5. 신규/갱신 단위 테스트 추가
+
+## Documents
+
+- [planning.md](./planning.md)
+- [design.md](./design.md)
+- [implementation.md](./implementation.md)
+- [verification.md](./verification.md)

docs/batches/bat-021_20260628_req047-step2-integration-and-runbook/planning.md

@@ -0,0 +1,34 @@
+# BAT-021 Planning
+
+## DoR 확인
+
+- [x] REQ-047: `Approved` 상태 확인
+- [x] BAT-020 코드 머지 (d4acfd4) — Step1/Step2 revision 존재
+- [x] dev PostgreSQL (Docker) 가동 확인
+- [x] 동반 수정 TODO 2건 식별 완료 (verification.md 미해결 사항 1, 2)
+
+## 작업 분해
+
+| # | 작업 | 파일 | 비고 |
+| --- | --- | --- | --- |
+| T-1 | `ensure_schema()` broker_credentials 갱신 | `app/repositories/postgres.py` | daemon_id NOT NULL, PK=(daemon_id,broker) |
+| T-2 | `upsert_broker_credential` ON CONFLICT 수정 | `app/repositories/postgres.py` | (daemon_id,broker) + fallback 제거 |
+| T-3 | list_broker_credentials fallback 제거 | `app/repositories/postgres.py` | Step2 이후 컬럼 항상 존재 |
+| T-4 | Step1 alembic upgrade 실 DB 검증 | dev PostgreSQL | AC-1, AC-2 |
+| T-5 | backfill CLI 실행 (테스트 데이터) | pfhub CLI | AC(REQ-048) |
+| T-6 | Step2 alembic upgrade 실 DB 검증 | dev PostgreSQL | AC-3, AC-4, AC-5 |
+| T-7 | downgrade 실 DB 검증 (Step2 → Step1 → base) | dev PostgreSQL | AC-6, AC-7 |
+| T-8 | 단위 테스트 추가/갱신 | `tests/test_bat021_*.py` | ensure_schema 정합, ON CONFLICT 행동 |
+| T-9 | 운영 런북 작성 | `docs/runbooks/bat-021-runbook.md` | 백업→Step1→backfill→Step2→verify→rollback |
+| T-10 | batch 문서 완성 | docs/batches/bat-021 | implementation.md, verification.md |
+
+## Design Gate
+
+Design 필요: 코드 수정 범위가 명확(T-1/T-2/T-3 — 두 함수 내 SQL 수정)하나
+fallback 패턴 제거 및 ensure_schema 스키마 정합성 확인이 필요하므로 design.md를 작성한다.
+
+## 위험
+
+- R-1: Step2 upgrade 중 NULL 행 있으면 revision 코드가 RuntimeError 발생 → 사전에 backfill 완료 확인 필수
+- R-2: `ensure_schema()` 변경 후 Step2 미적용 DB에서 실행 시 NOT NULL/PK 불일치 → ensure_schema는 Step2 적용 후 동작 환경에서만 호출 (런북에 명시)
+- R-3: fallback 제거 시 Step1 미적용 DB에서 upsert 호출 시 오류 → 이 batch는 Step2 이후 세계를 목표로 하므로 Step1 이전 환경 지원 불필요 (런북에 명시)

docs/batches/bat-021_20260628_req047-step2-integration-and-runbook/verification.md

@@ -0,0 +1,164 @@
+# BAT-021 Verification
+
+## pytest 실행 결과
+
+```
+python3 -m pytest tests/ --tb=short
+521 passed, 11 warnings in 26.67s
+```
+
+기존 테스트: 509 passed (BAT-020 기준)
+신규 테스트: 12 added (test_bat021_step2_ensure_schema_and_upsert.py)
+기존 테스트 회귀: 0건
+
+---
+
+## REQ-047 AC 검증 (실 PostgreSQL)
+
+### 검증 환경
+
+- Docker: `portfolio-hub-timescaledb` (timescale/timescaledb:latest-pg16)
+- DB: `portfolio_hub` (127.0.0.1:5432)
+- 시작 alembic 상태: 20260601_0024 (ensure_schema()로 스키마 구성, alembic 미동기 상태)
+- stamp 조치: `alembic stamp 20260605_0025` (DB 실제 스키마 = 0025 적용 상태 반영)
+
+### AC-1: Step1 후 daemon_id TEXT NULL 컬럼 존재, PK (user_id,broker) 유지
+
+```
+alembic upgrade 20260628_0026
+```
+
+결과:
+```
+daemon_id  | text | nullable=YES
+PK: broker_credentials_pkey PRIMARY KEY btree (user_id, broker)
+```
+
+**PASS**
+
+### AC-2: Step1 후 기존 데이터 daemon_id=NULL 보존
+
+```sql
+INSERT INTO broker_credentials ... VALUES ('test-user-1', 'kis', ...)
+SELECT daemon_id FROM broker_credentials WHERE user_id='test-user-1';
+-- 결과: NULL
+```
+
+**PASS**
+
+### AC-3: Step2 후 PK (daemon_id, broker) 전환, daemon_id NOT NULL
+
+```
+# backfill: UPDATE broker_credentials SET daemon_id='test-daemon-1' WHERE daemon_id IS NULL
+alembic upgrade 20260628_0027
+```
+
+결과:
+```
+daemon_id  | text | nullable=NO
+PK: broker_credentials_pkey PRIMARY KEY btree (daemon_id, broker)
+```
+
+**PASS**
+
+### AC-4: Step2 후 user_id 컬럼 유지
+
+```sql
+SELECT column_name FROM information_schema.columns
+WHERE table_name='broker_credentials' AND column_name='user_id';
+-- 결과: user_id 존재
+```
+
+**PASS**
+
+### AC-5: Step2 후 ensure_schema() 정합
+
+```python
+repo.ensure_schema()  # 오류 없음
+# DB 조회 결과:
+# daemon_id: NOT NULL, PK=(daemon_id, broker), user_id 존재
+```
+
+**PASS**
+
+### AC-6: Step1 downgrade — daemon_id 컬럼 제거, PK (user_id,broker) 복원
+
+```
+alembic downgrade 20260605_0025
+```
+
+결과:
+```
+broker_credentials: daemon_id 컬럼 없음
+PK: broker_credentials_pkey PRIMARY KEY btree (user_id, broker)
+```
+
+**PASS**
+
+### AC-7: Step2 downgrade 안전장치 — 중복 (user_id,broker) 존재 시 RuntimeError
+
+```
+# 2개 daemon이 같은 user/broker 사용하도록 데이터 구성 후 downgrade 시도
+alembic downgrade 20260628_0026
+```
+
+결과:
+```
+RuntimeError: downgrade 불가: 1개 (user_id, broker) 조합이 중복됩니다.
+같은 user의 여러 daemon이 동일 broker를 사용하고 있어
+(user_id, broker) PK로 복원 시 데이터 손실이 발생합니다.
+수동으로 데이터를 정리한 후 downgrade하세요.
+```
+
+**PASS**
+
+---
+
+## 동반 코드 검증
+
+### upsert_broker_credential (ON CONFLICT + row mapper)
+
+```python
+cred = repo.upsert_broker_credential(
+    'bat021-test-user', 'kis',
+    account_ref='BAT021-ACCOUNT',
+    encrypted_payload='bat021-encrypted-payload',
+    daemon_id='bat021-daemon-1'
+)
+# 결과: daemon_id='bat021-daemon-1', user_id='bat021-test-user', broker='kis', account_ref='BAT021-ACCOUNT'
+cred2 = repo.upsert_broker_credential(..., account_ref='BAT021-ACCOUNT-UPDATED')
+# 결과: version=2, account_ref='BAT021-ACCOUNT-UPDATED' (idempotent update OK)
+```
+
+**PASS**
+
+### list_broker_credentials (Step2 컬럼 순서)
+
+```python
+creds = repo.list_broker_credentials('bat021-test-user', daemon_id='bat021-daemon-1')
+# 결과: 1건, daemon_id='bat021-daemon-1', broker='kis'
+```
+
+**PASS**
+
+---
+
+## Step2 NULL 사전 검증 (REQ-048 미수행 시 거부)
+
+```
+# daemon_id IS NULL 행 1건 존재 상태에서 Step2 upgrade 시도
+RuntimeError: broker_credentials에 daemon_id IS NULL인 행이 1건 있습니다.
+REQ-048 backfill-daemon-id CLI로 모든 행을 백필한 후 다시 실행하세요.
+```
+
+**PASS**
+
+---
+
+## 미해결/위험 사항
+
+없음 — 이 batch의 TODO 2건 모두 해소됨:
+1. ensure_schema() broker_credentials 갱신: **완료**
+2. upsert ON CONFLICT (user_id,broker) → (daemon_id,broker) + fallback 제거: **완료**
+
+실 운영 DB 적용은 런북에 따라 사장님 결재 후 운영자가 수행한다.

docs/batches/index.md

@@ -27,4 +27,5 @@
 | BAT-017 | standard | released | REQ-039, REQ-040, REQ-041, REQ-042 | dcy-007 | [bat-017](./bat-017_20260618_trade-history-and-pnl-api/) |
 | BAT-018 | batch-lite | release-candidate | REQ-043 | dcy-008 | [bat-018](./bat-018_20260625_holdings-cost-basis-endpoint/) |
 | BAT-019 | standard | confirmed | REQ-044, REQ-045 | dcy-009 | [bat-019](./bat-019_20260627_assets-read-permission-and-daemon-list/) |
-| BAT-020 | standard | draft-batch-verification | REQ-046, REQ-047, REQ-048, REQ-049, REQ-050, REQ-051 | dcy-010 | [bat-020](./bat-020_20260628_daemon-scoped-broker-credential/) |
+| BAT-020 | standard | release-candidate | REQ-046, REQ-047, REQ-048, REQ-049, REQ-050, REQ-051 | dcy-010 | [bat-020](./bat-020_20260628_daemon-scoped-broker-credential/) |
+| BAT-021 | standard | in-progress | REQ-047 | dcy-010 | [bat-021](./bat-021_20260628_req047-step2-integration-and-runbook/) |

docs/runbooks/bat-021-req047-step2-migration-runbook.md

@@ -0,0 +1,193 @@
+# 운영 적용 런북: REQ-047 Step2 — broker_credentials PK 전환 (daemon_id, broker)
+
+- Batch: BAT-021
+- Discovery: dcy-010
+- Date: 2026-06-28
+- 결재 게이트: 사장님 결재 필수 (비가역 PK 변경)
+
+## 전제 조건
+
+- REQ-046 ~ REQ-051 코드(BAT-020 PR #17, d4acfd4)가 배포돼 있어야 한다.
+- BAT-021 코드(ensure_schema 갱신, ON CONFLICT 수정)는 **Step2 적용 완료 이후** 배포한다(아래 배포 순서 준수).
+- 마이그레이션 전 DB 백업이 완료돼 있어야 한다.
+
+## 배포 순서 개요
+
+```
+백업 → Step1 upgrade → backfill CLI → NULL 0건 확인 → Step2 upgrade(결재) → 코드 배포 → 동작 확인 → 비상 시 롤백
+```
+
+---
+
+## 1단계: DB 백업
+
+운영 DB에 마이그레이션을 적용하기 전 전체 스냅샷을 백업한다.
+
+```bash
+# 예: pg_dump 사용
+pg_dump -h <DB_HOST> -U <DB_USER> -d <DB_NAME> -F c -f backup_pre_bat021_$(date +%Y%m%d%H%M%S).dump
+```
+
+---
+
+## 2단계: Step1 alembic upgrade (daemon_id NULL 컬럼 추가)
+
+```bash
+export TIMESCALEDB_HOST=<DB_HOST>
+export TIMESCALEDB_PORT=<DB_PORT>
+export TIMESCALEDB_NAME=<DB_NAME>
+export TIMESCALEDB_USER=<DB_USER>
+export TIMESCALEDB_PASSWORD=<DB_PASS>
+export PORTFOLIOHUB_DB_ENABLED=true
+
+alembic upgrade 20260628_0026
+```
+
+예상 출력:
+```
+INFO  [alembic.runtime.migration] Running upgrade <prev> -> 20260628_0026, broker_credentials: daemon_id TEXT NULL 컬럼 추가 (REQ-047 Step1)
+```
+
+### 검증
+
+```sql
+-- broker_credentials에 daemon_id TEXT NULL 컬럼이 존재해야 함
+\d broker_credentials
+-- 또는
+SELECT column_name, is_nullable FROM information_schema.columns
+WHERE table_name = 'broker_credentials' AND column_name = 'daemon_id';
+-- 기대: daemon_id | YES (nullable)
+
+-- 기존 PK (user_id, broker) 유지 확인
+SELECT constraint_name, constraint_type FROM information_schema.table_constraints
+WHERE table_name = 'broker_credentials' AND constraint_type = 'PRIMARY KEY';
+-- 기대: broker_credentials_pkey | PRIMARY KEY
+
+-- 기존 데이터 daemon_id=NULL 보존 확인
+SELECT COUNT(*) FROM broker_credentials;  -- 기존 행 수 유지
+```
+
+---
+
+## 3단계: backfill CLI — daemon_id 매핑
+
+모든 기존 broker_credentials 행에 daemon_id를 설정한다.
+
+```bash
+# 사용자별로 실행 (daemon_id는 해당 user의 daemon id)
+pfhub credential backfill-daemon-id --user <username> --daemon-id <daemon_slug>
+
+# --dry-run으로 먼저 확인
+pfhub credential backfill-daemon-id --user <username> --daemon-id <daemon_slug> --dry-run
+```
+
+### 검증: NULL 0건 확인
+
+```sql
+-- 이 쿼리가 0을 반환해야 Step2를 진행할 수 있다
+SELECT COUNT(*) FROM broker_credentials WHERE daemon_id IS NULL;
+-- 기대: 0
+```
+
+**경고: NULL이 1건이라도 있으면 Step2 upgrade가 RuntimeError로 거부된다. 반드시 0건 확인 후 진행.**
+
+---
+
+## 4단계: Step2 alembic upgrade (PK 전환) — 사장님 결재 필요
+
+이 단계는 비가역 PK 변경이다. **사장님 결재 후에만 실행한다.**
+
+```bash
+alembic upgrade 20260628_0027
+```
+
+예상 출력:
+```
+INFO  [alembic.runtime.migration] Running upgrade 20260628_0026 -> 20260628_0027, broker_credentials: PK (user_id,broker) → (daemon_id,broker) 전환 (REQ-047 Step2)
+```
+
+### 검증: Step2 완료 확인
+
+```sql
+-- PK가 (daemon_id, broker)로 변경됐는지 확인
+SELECT constraint_name FROM information_schema.table_constraints
+WHERE table_name = 'broker_credentials' AND constraint_type = 'PRIMARY KEY';
+-- 기대: broker_credentials_pkey
+
+SELECT pg_get_constraintdef(oid)
+FROM pg_constraint
+WHERE conname = 'broker_credentials_pkey';
+-- 기대: PRIMARY KEY (daemon_id, broker)
+
+-- daemon_id NOT NULL 확인
+SELECT column_name, is_nullable FROM information_schema.columns
+WHERE table_name = 'broker_credentials' AND column_name = 'daemon_id';
+-- 기대: daemon_id | NO (not nullable)
+
+-- user_id 컬럼 유지 확인
+SELECT column_name FROM information_schema.columns
+WHERE table_name = 'broker_credentials' AND column_name = 'user_id';
+-- 기대: user_id 존재
+```
+
+---
+
+## 5단계: 코드 배포 (BAT-021 코드 포함)
+
+Step2 alembic upgrade가 완료된 후에 코드를 배포한다.
+
+포함 변경:
+- `app/repositories/postgres.py`:
+  - `ensure_schema()` broker_credentials: daemon_id NOT NULL, PK=(daemon_id, broker)
+  - `_row_to_broker_credential()`: Step2 컬럼 순서 (daemon_id 선두)
+  - `upsert_broker_credential()`: ON CONFLICT (daemon_id, broker), fallback 제거
+  - `list_broker_credentials()`: fallback 제거, Step2 컬럼 순서
+
+---
+
+## 6단계: 동작 확인
+
+```bash
+# pfhub CLI로 인증정보 조회/등록 테스트
+pfhub credential list --user <username> --daemon-id <daemon_id>
+
+# 서비스 헬스체크
+curl http://<SERVICE_HOST>/health
+```
+
+---
+
+## 롤백 절차
+
+### Step2 롤백 (alembic downgrade)
+
+Step2 downgrade는 (user_id, broker) 중복이 없는 경우에만 가능하다.
+
+```bash
+# Step2 → Step1 상태로 복원
+alembic downgrade 20260628_0026
+```
+
+**경고**: 같은 user의 여러 daemon이 동일 broker를 사용하는 경우 downgrade가 RuntimeError로 거부된다.
+이 경우 수동 데이터 정리 후 downgrade하거나 백업에서 복원한다.
+
+### Step1 롤백
+
+```bash
+# Step1 → 이전 상태로 복원 (daemon_id 컬럼 제거)
+alembic downgrade 20260605_0025
+```
+
+### DB 복원 (최후 수단)
+
+```bash
+pg_restore -h <DB_HOST> -U <DB_USER> -d <DB_NAME> -c backup_pre_bat021_<TIMESTAMP>.dump
+```
+
+---
+
+## 주의사항
+
+- Step2 이전에 BAT-021 코드를 배포하면 `ON CONFLICT (daemon_id, broker)` 절이 구 스키마와 맞지 않아 오류 발생.
+- 배포 순서: Step1 → backfill → Step2 → 코드 배포 순서를 엄수한다.
+- `ensure_schema()`는 `IF NOT EXISTS`를 사용하므로 Step2 적용 후 실행 시 테이블이 이미 존재해 스킵됨 (신규 인스턴스 초기화 시에만 적용됨).

tests/test_bat020_broker_credential_domain.py

@@ -42,37 +42,42 @@ def test_broker_credential_daemon_id_set() -> None:
 
 
 # ---------------------------------------------------------------------------
-# REQ-046 AC-3: _row_to_broker_credential 7컬럼(daemon_id 포함) 처리
+# REQ-046 AC-3 / BAT-021: _row_to_broker_credential Step2 컬럼 순서 검증
+# Step2(REQ-047) 이후 컬럼 순서: (daemon_id, user_id, broker, account_ref, encrypted_payload, credential_version, last_rotated_at)
 # ---------------------------------------------------------------------------
 
 def test_row_to_broker_credential_with_daemon_id() -> None:
-    """REQ-046 AC-3: 7컬럼 행(daemon_id 포함) 변환 시 daemon_id 세팅."""
+    """REQ-046 AC-3 / BAT-021: Step2 컬럼 순서(daemon_id 선두) 행 변환 검증."""
     from app.repositories.postgres import PostgresPortfolioRepository
 
     now = datetime.now(timezone.utc)
-    row = ("user-1", "kis", "12345678", "encrypted", 2, now, "daemon-xyz")
+    # Step2 이후 RETURNING/SELECT 순서: daemon_id, user_id, broker, account_ref, encrypted_payload, credential_version, last_rotated_at
+    row = ("daemon-xyz", "user-1", "kis", "12345678", "encrypted", 2, now)
     cred = PostgresPortfolioRepository._row_to_broker_credential(row)
+    assert cred.daemon_id == "daemon-xyz"
     assert cred.user_id == "user-1"
     assert cred.broker == "kis"
     assert cred.account_ref == "12345678"
     assert cred.encrypted_payload == "encrypted"
     assert cred.credential_version == 2
     assert cred.last_rotated_at == now
-    assert cred.daemon_id == "daemon-xyz"
 
 
 # ---------------------------------------------------------------------------
-# REQ-046 AC-4: _row_to_broker_credential 6컬럼(daemon_id 없음) 하위호환
+# REQ-046 AC-4 / BAT-021: _row_to_broker_credential credential_version 기본값
+# Step2 이후 daemon_id는 항상 존재. 6컬럼(credential_version 없음) 기본값 1 확인.
 # ---------------------------------------------------------------------------
 
 def test_row_to_broker_credential_without_daemon_id() -> None:
-    """REQ-046 AC-4: 6컬럼 행(daemon_id 없음) 변환 시 daemon_id=None (하위호환)."""
+    """REQ-046 AC-4 / BAT-021: credential_version 없는 5컬럼 행 → credential_version 기본값 1."""
     from app.repositories.postgres import PostgresPortfolioRepository
 
-    now = datetime.now(timezone.utc)
+    # Step2 이후 컬럼 순서: daemon_id, user_id, broker, account_ref, encrypted_payload (5컬럼)
-    row = ("user-1", "kis", "12345678", "encrypted", 1, now)
+    row = ("daemon-abc", "user-1", "kis", "12345678", "encrypted")
     cred = PostgresPortfolioRepository._row_to_broker_credential(row)
-    assert cred.daemon_id is None
+    assert cred.daemon_id == "daemon-abc"
+    assert cred.credential_version == 1
+    assert cred.last_rotated_at is None
 
 
 # ---------------------------------------------------------------------------
@@ -109,11 +114,11 @@ class _FakeConn:
 
 
 def _make_rows_for_user(user_id: str):
-    """daemon_id "d1", "d2" 각각 1건씩."""
+    """Step2 컬럼 순서: (daemon_id, user_id, broker, account_ref, encrypted_payload, credential_version, last_rotated_at)."""
     now = datetime.now(timezone.utc)
     return [
-        (user_id, "kis", "ref-d1", "enc-d1", 1, now, "d1"),
+        ("d1", user_id, "kis", "ref-d1", "enc-d1", 1, now),
-        (user_id, "kis2", "ref-d2", "enc-d2", 1, now, "d2"),
+        ("d2", user_id, "kis2", "ref-d2", "enc-d2", 1, now),
     ]
 
 

tests/test_bat021_step2_ensure_schema_and_upsert.py

@@ -0,0 +1,252 @@
+"""BAT-021: REQ-047 Step2 동반 코드 수정 단위 테스트.
+
+검증 범위:
+- ensure_schema() broker_credentials CREATE TABLE: daemon-scoped PK 정의
+- _row_to_broker_credential: Step2 컬럼 순서 (daemon_id 선두)
+- upsert_broker_credential: ON CONFLICT (daemon_id, broker), fallback 제거
+- list_broker_credentials: fallback 제거, Step2 컬럼 순서
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from app.domain.models import BrokerCredential
+from app.repositories.postgres import PortfolioDBConfig, PostgresPortfolioRepository
+
+
+# ---------------------------------------------------------------------------
+# 공통 픽스처
+# ---------------------------------------------------------------------------
+
+def _make_config() -> PortfolioDBConfig:
+    return PortfolioDBConfig(
+        host="localhost",
+        port=5432,
+        dbname="test",
+        user="test",
+        password="test",
+        sslmode="disable",
+    )
+
+
+def _make_repo() -> PostgresPortfolioRepository:
+    return PostgresPortfolioRepository(_make_config())
+
+
+# ---------------------------------------------------------------------------
+# ensure_schema broker_credentials 스키마 정의 검증
+# ---------------------------------------------------------------------------
+
+def test_ensure_schema_broker_credentials_ddl_contains_daemon_id_pk() -> None:
+    """BAT-021: ensure_schema()의 broker_credentials CREATE TABLE에 daemon_id NOT NULL과
+    PRIMARY KEY (daemon_id, broker)가 포함되어 있다."""
+    import inspect
+    import app.repositories.postgres as repo_module
+
+    source = inspect.getsource(repo_module)
+    # CREATE TABLE에 daemon_id NOT NULL 포함 확인
+    assert "daemon_id TEXT NOT NULL" in source, "ensure_schema broker_credentials에 daemon_id NOT NULL 없음"
+    # PRIMARY KEY (daemon_id, broker) 포함 확인
+    assert "PRIMARY KEY (daemon_id, broker)" in source, "ensure_schema broker_credentials PK가 (daemon_id,broker) 아님"
+
+
+def test_ensure_schema_broker_credentials_no_old_pk_user_id_broker() -> None:
+    """BAT-021: ensure_schema()의 broker_credentials CREATE TABLE에 구 PK (user_id, broker)가 없다."""
+    import inspect
+    import app.repositories.postgres as repo_module
+
+    source = inspect.getsource(repo_module)
+    # ensure_schema 함수 영역만 검사: broker_credentials CREATE TABLE
+    # "PRIMARY KEY (user_id, broker)" 패턴이 없어야 함
+    assert "PRIMARY KEY (user_id, broker)" not in source, "ensure_schema에 구 PK (user_id,broker) 잔존"
+
+
+# ---------------------------------------------------------------------------
+# _row_to_broker_credential Step2 컬럼 순서
+# ---------------------------------------------------------------------------
+
+def test_row_to_broker_credential_step2_7col_canonical_order() -> None:
+    """BAT-021: Step2 표준 7컬럼 행 변환 — daemon_id 선두."""
+    now = datetime.now(timezone.utc)
+    # Step2: (daemon_id, user_id, broker, account_ref, encrypted_payload, credential_version, last_rotated_at)
+    row = ("daemon-abc", "user-1", "kis", "ACC-001", "enc-payload", 3, now)
+    cred = PostgresPortfolioRepository._row_to_broker_credential(row)
+
+    assert cred.daemon_id == "daemon-abc"
+    assert cred.user_id == "user-1"
+    assert cred.broker == "kis"
+    assert cred.account_ref == "ACC-001"
+    assert cred.encrypted_payload == "enc-payload"
+    assert cred.credential_version == 3
+    assert cred.last_rotated_at == now
+
+
+def test_row_to_broker_credential_step2_5col_no_version() -> None:
+    """BAT-021: 5컬럼 행(credential_version 없음) — 기본값 1, last_rotated_at=None."""
+    row = ("daemon-xyz", "user-2", "kis2", "ACC-002", "enc-2")
+    cred = PostgresPortfolioRepository._row_to_broker_credential(row)
+
+    assert cred.daemon_id == "daemon-xyz"
+    assert cred.user_id == "user-2"
+    assert cred.broker == "kis2"
+    assert cred.credential_version == 1
+    assert cred.last_rotated_at is None
+
+
+# ---------------------------------------------------------------------------
+# upsert_broker_credential: ON CONFLICT (daemon_id, broker)
+# ---------------------------------------------------------------------------
+
+def test_upsert_broker_credential_uses_daemon_id_conflict_target() -> None:
+    """BAT-021: upsert_broker_credential SQL에 ON CONFLICT (daemon_id, broker) 포함 확인."""
+    import inspect
+    import app.repositories.postgres as repo_module
+
+    # upsert_broker_credential 메서드 소스 추출
+    source = inspect.getsource(
+        PostgresPortfolioRepository.upsert_broker_credential
+    )
+    assert "ON CONFLICT (daemon_id, broker)" in source, \
+        "upsert_broker_credential에 ON CONFLICT (daemon_id, broker) 없음"
+
+
+def test_upsert_broker_credential_no_old_conflict_target() -> None:
+    """BAT-021: upsert_broker_credential SQL에 구 ON CONFLICT (user_id, broker) 없음."""
+    import inspect
+
+    source = inspect.getsource(
+        PostgresPortfolioRepository.upsert_broker_credential
+    )
+    assert "ON CONFLICT (user_id, broker)" not in source, \
+        "upsert_broker_credential에 구 ON CONFLICT (user_id,broker) 잔존"
+
+
+def test_upsert_broker_credential_no_fallback_try_except() -> None:
+    """BAT-021: upsert_broker_credential에 fallback try/except 없음 (트랜잭션 abort 이슈 해소)."""
+    import inspect
+
+    source = inspect.getsource(
+        PostgresPortfolioRepository.upsert_broker_credential
+    )
+    # Fallback try/except는 "Fallback" 주석과 함께 있었음 — 완전 제거 확인
+    assert "Fallback: daemon_id" not in source, \
+        "upsert_broker_credential에 Fallback try/except 잔존"
+
+
+def test_upsert_broker_credential_mock_returns_step2_row() -> None:
+    """BAT-021: upsert_broker_credential mock 기반 — Step2 RETURNING 순서(daemon_id 선두)로 BrokerCredential 생성."""
+    now = datetime.now(timezone.utc)
+    # Step2 RETURNING 순서: daemon_id, user_id, broker, account_ref, encrypted_payload, credential_version, last_rotated_at
+    mock_row = ("daemon-test", "user-test", "kis", "REF-001", "enc-test", 1, now)
+
+    repo = _make_repo()
+
+    mock_cursor = MagicMock()
+    mock_cursor.__enter__ = lambda s: mock_cursor
+    mock_cursor.__exit__ = MagicMock(return_value=False)
+    # user existence check
+    mock_cursor.fetchone.side_effect = [MagicMock(), mock_row]  # user exists, then credential row
+
+    mock_conn = MagicMock()
+    mock_conn.cursor.return_value = mock_cursor
+    repo._connection = mock_conn
+
+    with patch.object(repo, "ensure_schema"):
+        cred = repo.upsert_broker_credential(
+            "user-test", "kis",
+            account_ref="REF-001",
+            encrypted_payload="enc-test",
+            daemon_id="daemon-test",
+        )
+
+    assert cred.daemon_id == "daemon-test"
+    assert cred.user_id == "user-test"
+    assert cred.broker == "kis"
+    assert cred.account_ref == "REF-001"
+
+
+# ---------------------------------------------------------------------------
+# list_broker_credentials: fallback 제거, Step2 컬럼 순서
+# ---------------------------------------------------------------------------
+
+def test_list_broker_credentials_no_fallback_try_except() -> None:
+    """BAT-021: list_broker_credentials에 fallback try/except 없음."""
+    import inspect
+
+    source = inspect.getsource(
+        PostgresPortfolioRepository.list_broker_credentials
+    )
+    assert "Fallback: daemon_id" not in source, \
+        "list_broker_credentials에 Fallback try/except 잔존"
+
+
+def test_list_broker_credentials_step2_select_order() -> None:
+    """BAT-021: list_broker_credentials SELECT 쿼리에 daemon_id 선두 컬럼 순서."""
+    import inspect
+
+    source = inspect.getsource(
+        PostgresPortfolioRepository.list_broker_credentials
+    )
+    # daemon_id가 SELECT에서 user_id보다 앞에 있어야 함
+    daemon_pos = source.find("SELECT daemon_id")
+    assert daemon_pos != -1, "list_broker_credentials SELECT에 daemon_id 선두 없음"
+
+
+def test_list_broker_credentials_mock_returns_step2_rows() -> None:
+    """BAT-021: list_broker_credentials mock — Step2 컬럼 순서로 BrokerCredential 생성."""
+    now = datetime.now(timezone.utc)
+    # Step2 컬럼 순서: (daemon_id, user_id, broker, account_ref, encrypted_payload, credential_version, last_rotated_at)
+    mock_rows = [
+        ("daemon-1", "user-1", "kis", "REF-A", "enc-a", 1, now),
+        ("daemon-2", "user-1", "kis2", "REF-B", "enc-b", 1, now),
+    ]
+
+    repo = _make_repo()
+
+    mock_cursor = MagicMock()
+    mock_cursor.__enter__ = lambda s: mock_cursor
+    mock_cursor.__exit__ = MagicMock(return_value=False)
+    mock_cursor.fetchall.return_value = mock_rows
+
+    mock_conn = MagicMock()
+    mock_conn.cursor.return_value = mock_cursor
+    repo._connection = mock_conn
+
+    with patch.object(repo, "ensure_schema"):
+        results = repo.list_broker_credentials("user-1")
+
+    assert len(results) == 2
+    assert results[0].daemon_id == "daemon-1"
+    assert results[0].broker == "kis"
+    assert results[1].daemon_id == "daemon-2"
+    assert results[1].broker == "kis2"
+
+
+def test_list_broker_credentials_daemon_id_filter_step2() -> None:
+    """BAT-021: list_broker_credentials daemon_id 필터 — Step2 컬럼 순서."""
+    now = datetime.now(timezone.utc)
+    mock_rows = [
+        ("daemon-1", "user-1", "kis", "REF-A", "enc-a", 1, now),
+    ]
+
+    repo = _make_repo()
+
+    mock_cursor = MagicMock()
+    mock_cursor.__enter__ = lambda s: mock_cursor
+    mock_cursor.__exit__ = MagicMock(return_value=False)
+    mock_cursor.fetchall.return_value = mock_rows
+
+    mock_conn = MagicMock()
+    mock_conn.cursor.return_value = mock_cursor
+    repo._connection = mock_conn
+
+    with patch.object(repo, "ensure_schema"):
+        results = repo.list_broker_credentials("user-1", daemon_id="daemon-1")
+
+    assert len(results) == 1
+    assert results[0].daemon_id == "daemon-1"
+    assert results[0].broker == "kis"