feat(bat-022): daemon CRUD admin API + self-service credential management (REQ-052/053/054)

- POST/PATCH/deactivate/activate /v1/admin/daemons — admin-only daemon CRUD (REQ-053)
- GET/PUT/DELETE /v1/daemons/{daemon_id}/credential — user-owned scoped credential management (REQ-052)
- Owner re-assignment auto-invalidates BrokerCredential via update_daemon_owner() (REQ-053 AC-4)
- Deactivation auto-revokes service tokens via revoke_tokens_by_daemon_id() (REQ-022 reuse, REQ-053 AC-6)
- DELETE /v1/admin/daemons/{id} → 405 (hard delete not provided, REQ-053 AC-8)
- All responses exclude app_key/app_secret; Fernet encryption at rest (REQ-054)
- Service-token → 403 on credential endpoints (REQ-052 AC-10)
- Admin ownership scope enforced: admin cannot write other user's credential (REQ-052 AC-9)
- REQ-051 regression: 4 service-token guard tests pass (REQ-054 AC-7)
- 44 new tests, 565 total passed (521 baseline + 44 new), 0 failures

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

app/repositories/portfolio_repository.py

@@ -114,6 +114,21 @@ class PortfolioRepository(Protocol):
 
     def list_service_tokens_with_null_daemon(self) -> list[AuthToken]: ...
 
+    def get_broker_credential_by_daemon(
+        self,
+        daemon_id: str,
+        broker: str,
+    ) -> BrokerCredential | None: ...
+
+    def delete_credentials_by_daemon_id(self, daemon_id: str) -> int: ...
+
+    def update_daemon_owner(
+        self,
+        daemon_id: str,
+        *,
+        owner_user_id: str,
+    ) -> Daemon: ...
+
     def get(self, portfolio_id: str) -> Portfolio: ...
 
     def get_portfolio_by_name_and_user(self, owner_user_id: str, name: str) -> Portfolio | None: ...

app/repositories/postgres.py

@@ -2343,6 +2343,81 @@ class PostgresPortfolioRepository:
             )
             return cur.rowcount
 
+    def get_broker_credential_by_daemon(
+        self,
+        daemon_id: str,
+        broker: str,
+    ) -> BrokerCredential | None:
+        """Return a single BrokerCredential by (daemon_id, broker) PK, or None if absent (REQ-052)."""
+        self.ensure_schema()
+        with self._connect().cursor() as cur:
+            cur.execute(
+                """
+                SELECT daemon_id, user_id, broker, account_ref, encrypted_payload, credential_version, last_rotated_at
+                FROM broker_credentials
+                WHERE daemon_id = %s AND broker = %s
+                """,
+                (daemon_id, broker),
+            )
+            row = cur.fetchone()
+        if row is None:
+            return None
+        return self._row_to_broker_credential(row)
+
+    def delete_credentials_by_daemon_id(self, daemon_id: str) -> int:
+        """Delete ALL broker credentials for daemon_id (owner re-assignment invalidation — REQ-053 AC-4).
+
+        Returns number of rows deleted.
+        """
+        self.ensure_schema()
+        with self._connect().cursor() as cur:
+            cur.execute(
+                "DELETE FROM broker_credentials WHERE daemon_id = %s",
+                (daemon_id,),
+            )
+            return cur.rowcount
+
+    def update_daemon_owner(
+        self,
+        daemon_id: str,
+        *,
+        owner_user_id: str,
+    ) -> 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.
+        """
+        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
+
     def list_service_tokens_with_null_daemon(self) -> list[AuthToken]:
         self.ensure_schema()
         with self._connect().cursor() as cur: