feat(entitlement): execute_product 接权益生效

- 资源模块 execute_product_service 返回 entitlement(账号:有效期+工作空间配额; 存储:容量GB)
- 产品层 execute_product 调 _grant_subscription 写 product_subscription(订阅+配额,幂等)

product_management/core.py

@@ -1332,7 +1332,7 @@ class ProductManager:
 
     async def execute_product(self, product_id=None, product_code=None,
                                request_data=None, user_id=None, user_org_id=None):
-        """执行产品服务（通过资源模块接口）。"""
+        """执行产品服务：资源侧执行 + 订阅/配额权益生效。"""
         iface, product, err = await self._get_product_interface(
             product_id, product_code)
         if err:
@@ -1343,6 +1343,8 @@ class ProductManager:
             return {'success': False, 'message': '资源模块未实现 execute_product_service',
                     'status': 'FAILED'}
 
+        product = product or {}
+
         ref_id, err = self._get_ref_id(product)
         if err:
             return {'success': False, 'message': err, 'status': 'FAILED'}
@@ -1353,7 +1355,67 @@ class ProductManager:
         if not user_org_id:
             user_org_id = self._get_current_org_id()
 
-        return await fn(ref_id, user_id, user_org_id, request_data or {})
+        result = await fn(ref_id, user_id, user_org_id, request_data or {})
+        if not isinstance(result, dict) or not result.get('success'):
+            return result
+
+        # 权益生效：资源模块返回 entitlement 时，写 product_subscription（订阅/配额）
+        ent = result.get('entitlement')
+        if ent:
+            try:
+                result['subscription_id'] = await self._grant_subscription(
+                    product, ent, user_id, user_org_id)
+            except Exception as e:
+                result['message'] = (result.get('message', '') or '') + f'；订阅写入失败: {e}'
+        return result
+
+    async def _grant_subscription(self, product, entitlement, user_id, user_org_id):
+        """写 product_subscription：订阅有效期 + 配额（账号工作空间数/存储容量GB）。
+
+        幂等：同一产品+机构已有活跃订阅则更新有效期与配额，不重复建。
+        """
+        dbname = self._get_dbname()
+        now = time.strftime('%Y-%m-%d %H:%M:%S')
+        today = datetime.date.today()
+        days = int(entitlement.get('duration_days', 0) or 0)
+        end_date = (today + datetime.timedelta(days=days)).isoformat() if days else today.isoformat()
+
+        quota_total = 0.0
+        quota_unit = entitlement.get('quota_unit', '')
+        if entitlement.get('workspace_max'):
+            quota_total = float(entitlement['workspace_max'])
+            quota_unit = quota_unit or 'workspace'
+        elif entitlement.get('storage_gb'):
+            quota_total = float(entitlement['storage_gb'])
+            quota_unit = quota_unit or 'GB'
+        elif entitlement.get('workspace_gb'):
+            quota_total = float(entitlement['workspace_gb'])
+            quota_unit = quota_unit or 'GB'
+
+        sub_id = getID()
+        async with DBPools().sqlorContext(dbname) as sor:
+            existing = await sor.sqlExe(
+                "SELECT id FROM product_subscription WHERE product_id=${pid}$ "
+                "AND user_org_id=${org}$ AND status='1' LIMIT 1",
+                {'pid': product.get('id'), 'org': user_org_id})
+            if existing:
+                sub_id = existing[0].id
+                await sor.U('product_subscription', {
+                    'id': sub_id, 'start_date': today.isoformat(),
+                    'end_date': end_date, 'quota_total': quota_total,
+                    'quota_unit': quota_unit, 'updated_at': now})
+            else:
+                await sor.C('product_subscription', {
+                    'id': sub_id, 'product_id': product.get('id'),
+                    'user_id': user_id, 'user_org_id': user_org_id,
+                    'subscription_type': '1', 'status': '1',
+                    'start_date': today.isoformat(), 'end_date': end_date,
+                    'quota_total': quota_total, 'quota_used': 0,
+                    'quota_unit': quota_unit, 'overflow_mode': '1', 'overflow_rate': 0,
+                    'purchase_price': float(product.get('price', 0) or 0),
+                    'purchase_currency': product.get('currency', 'CNY'),
+                    'created_at': now, 'updated_at': now})
+        return sub_id
 
     async def execute_product_stream(self, product_id=None, product_code=None,
                                       request_data=None, user_id=None,