fix(bat-022): owner 재할당 credential 무효화 원자화 — leak 방지 (QC Major)

update_daemon_owner()에서 owner UPDATE + credential DELETE를 단일
conn.transaction() 블록으로 묶어 원자성 보장. 실패 시 두 연산 모두
롤백되므로 "새 owner + 이전 credential 잔류" leak 상태(dcy-011 OQ-3)
가 불가능해짐. 실행 순서도 DELETE-first(안전 순서)로 고정.

단위 테스트 3종 추가:
- DELETE-before-UPDATE 실행 순서 검증
- UPDATE 실패 시 rollback(commit 미호출) 검증
- 단일 transaction() 블록 안에서 두 연산 실행 검증

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VHtM5Ws1HKCKQ4JGi6GRb5

app/repositories/postgres.py

@@ -2385,38 +2385,48 @@ class PostgresPortfolioRepository:
     ) -> Daemon:
         """Change daemon owner and invalidate all existing BrokerCredentials (REQ-053 AC-4).
 
-        Owner change + credential invalidation are two separate statements (autocommit).
-        Credential deletion failure is logged by caller — daemon update is treated as
-        authoritative.
+        Owner change and credential invalidation are executed atomically in a single
+        transaction (BAT-022 QC Major fix).  If either statement fails the entire
+        transaction is rolled back, preventing the state where the owner has changed but
+        the previous owner's credentials remain (credential leak — dcy-011 OQ-3).
+
+        Execution order inside the transaction:
+          1. DELETE FROM broker_credentials WHERE daemon_id = %s   (credential first)
+          2. UPDATE daemons SET owner_user_id = %s ...             (owner second)
+        This order is also safe if the transaction machinery is somehow bypassed: if
+        only step 1 completes the owner is unchanged and credentials are gone (no leak);
+        if only step 2 completes the old credentials are still present but the caller
+        still owns them (no cross-owner leak).
         """
         self.ensure_schema()
-        with self._connect().cursor() as cur:
-            cur.execute(
-                """
-                SELECT id FROM daemons WHERE id = %s
-                """,
-                (daemon_id,),
-            )
-            if cur.fetchone() is None:
-                raise ValueError("데몬을 찾을 수 없습니다")
-            cur.execute(
-                """
-                UPDATE daemons
-                SET owner_user_id = %s, updated_at = NOW()
-                WHERE id = %s
-                RETURNING id, owner_user_id, daemon_slug, display_name, is_active, created_at, updated_at
-                """,
-                (owner_user_id, daemon_id),
-            )
-            row = cur.fetchone()
-        daemon = self._row_to_daemon(row)
-        # Invalidate broker credentials after owner change
-        with self._connect().cursor() as cur:
-            cur.execute(
-                "DELETE FROM broker_credentials WHERE daemon_id = %s",
-                (daemon_id,),
-            )
-        return daemon
+        conn = self._connect()
+        # Use conn.transaction() (not `with conn:`) so the shared connection stays open
+        # after the block (psycopg3 closes the connection when used as a context manager).
+        with conn.transaction():
+            with conn.cursor() as cur:
+                cur.execute(
+                    "SELECT id FROM daemons WHERE id = %s",
+                    (daemon_id,),
+                )
+                if cur.fetchone() is None:
+                    raise ValueError("데몬을 찾을 수 없습니다")
+                # Step 1: delete credentials first (safe-order)
+                cur.execute(
+                    "DELETE FROM broker_credentials WHERE daemon_id = %s",
+                    (daemon_id,),
+                )
+                # Step 2: update owner
+                cur.execute(
+                    """
+                    UPDATE daemons
+                    SET owner_user_id = %s, updated_at = NOW()
+                    WHERE id = %s
+                    RETURNING id, owner_user_id, daemon_slug, display_name, is_active, created_at, updated_at
+                    """,
+                    (owner_user_id, daemon_id),
+                )
+                row = cur.fetchone()
+        return self._row_to_daemon(row)
 
     def list_service_tokens_with_null_daemon(self) -> list[AuthToken]:
         self.ensure_schema()