fix: Integrations 401 auth + Jobs API speed optimization

emergent2api/backends/integrations.py

@@ -1,9 +1,18 @@
-"""Integrations API backend: proxy to integrations.emergentagent.com/llm."""
+"""Integrations API backend: proxy to integrations.emergentagent.com/llm.
+
+The Integrations endpoint is an Anthropic-compatible proxy that requires either:
+  1. A per-account integration_key (provisioned by Emergent for Pro/paying users)
+  2. An explicit ANTHROPIC_API_KEY env var
+  3. Falls back to the Supabase JWT (which won't work for free accounts)
+
+The key is sent via ``x-api-key`` header (Anthropic convention).
+"""
 from __future__ import annotations
 
 import asyncio
 import json
 import logging
+import os
 from typing import Any, AsyncGenerator
 
 import httpx
@@ -16,10 +25,26 @@ logger = logging.getLogger("emergent2api.integrations")
 
 class IntegrationsBackend(EmergentBackend):
     """Proxy to Emergent's OpenAI-compatible integrations endpoint.
-    
+
     Requires a US-based proxy since the endpoint has IP restrictions.
+    Auth: ``x-api-key`` header with the account's ``integration_key``.
     """
 
+    def _resolve_api_key(self, account: dict[str, Any]) -> str:
+        """Pick the best API key for the Integrations endpoint.
+
+        Priority:
+        1. ``integration_key`` stored per-account (DB column)
+        2. ``ANTHROPIC_API_KEY`` env var (global fallback)
+        3. Supabase JWT (will 401 for free accounts, but tried as last resort)
+        """
+        if account.get("integration_key"):
+            return account["integration_key"]
+        env_key = os.environ.get("ANTHROPIC_API_KEY", "")
+        if env_key:
+            return env_key
+        return account["jwt"]
+
     async def chat(
         self,
         messages: list[dict[str, Any]],
@@ -28,15 +53,20 @@ class IntegrationsBackend(EmergentBackend):
         account: dict[str, Any],
         stream: bool = False,
     ) -> AsyncGenerator[dict[str, Any], None]:
-        jwt = account["jwt"]
+        api_key = self._resolve_api_key(account)
         url = f"{settings.integrations_base}/v1/chat/completions"
 
-        body = {
+        body: dict[str, Any] = {
             "model": model,
             "messages": messages,
             "stream": stream,
         }
 
+        headers = {
+            "x-api-key": api_key,
+            "Content-Type": "application/json",
+        }
+
         proxy = settings.proxy or None
         transport = httpx.AsyncHTTPTransport(proxy=proxy) if proxy else None
 
@@ -47,13 +77,7 @@ class IntegrationsBackend(EmergentBackend):
             ) as client:
                 if stream:
                     async with client.stream(
-                        "POST",
-                        url,
-                        json=body,
-                        headers={
-                            "Authorization": f"Bearer {jwt}",
-                            "Content-Type": "application/json",
-                        },
+                        "POST", url, json=body, headers=headers,
                     ) as resp:
                         if resp.status_code != 200:
                             err = await resp.aread()
@@ -79,14 +103,7 @@ class IntegrationsBackend(EmergentBackend):
 
                         yield {"type": "done", "content": full_text, "thinking": ""}
                 else:
-                    resp = await client.post(
-                        url,
-                        json=body,
-                        headers={
-                            "Authorization": f"Bearer {jwt}",
-                            "Content-Type": "application/json",
-                        },
-                    )
+                    resp = await client.post(url, json=body, headers=headers)
                     if resp.status_code != 200:
                         yield {"type": "error", "content": f"Integrations API: HTTP {resp.status_code} {resp.text[:200]}"}
                         return

emergent2api/backends/jobs.py

@@ -64,21 +64,21 @@ class JobsBackend(EmergentBackend):
         confirmed_end = False
         has_received_anything = False
         job_completed = False
-        max_wait_cycles = 600
+        last_change_time = 0.0
+        max_wait_cycles = 900
         start_time = time.time()
 
-        await asyncio.sleep(8.0)
+        await asyncio.sleep(3.0)
 
-        for _ in range(max_wait_cycles):
+        for cycle in range(max_wait_cycles):
             result = await asyncio.to_thread(self._poll_trajectory, jwt, ref_id)
             if result is None:
-                await self._dynamic_sleep(consecutive_unchanged)
+                await self._dynamic_sleep(consecutive_unchanged, has_received_anything)
                 consecutive_unchanged += 1
                 continue
 
             data_items = result.get("data", [])
 
-            # Check for explicit completion signal in trajectory items
             for item in data_items:
                 status = (item.get("traj_payload") or {}).get("status", "")
                 if status in ("completed", "done", "finished"):
@@ -107,6 +107,7 @@ class JobsBackend(EmergentBackend):
                 consecutive_unchanged = 0
                 confirmed_end = False
                 has_received_anything = True
+                last_change_time = time.time()
                 if stream and delta:
                     yield {"type": "text", "content": delta}
             else:
@@ -116,18 +117,25 @@ class JobsBackend(EmergentBackend):
                     logger.info("Stream completed (job status=completed)")
                     break
 
-                # Require more stability cycles and longer confirmation for complex queries
-                stable_threshold = 15 if len(full_text) < 500 else 10
-                if has_received_anything and full_text and consecutive_unchanged >= stable_threshold:
+                stable_secs = time.time() - last_change_time if last_change_time else 0
+                text_len = len(full_text)
+                if text_len < 50:
+                    stable_limit = 15.0
+                elif text_len < 500:
+                    stable_limit = 25.0
+                else:
+                    stable_limit = 45.0
+                if has_received_anything and full_text and stable_secs >= stable_limit:
                     if not confirmed_end:
-                        logger.debug("Content stable, confirming end...")
+                        logger.debug(f"Content stable for {stable_secs:.0f}s (len={text_len}), confirming...")
                         confirmed_end = True
-                        await asyncio.sleep(8.0)
+                        confirm_wait = 5.0 if text_len < 200 else 10.0
+                        await asyncio.sleep(confirm_wait)
                         continue
-                    logger.info("Stream completed (double-confirmed stable)")
+                    logger.info(f"Stream completed (stable {stable_secs:.0f}s, {text_len} chars)")
                     break
 
-            await self._dynamic_sleep(consecutive_unchanged)
+            await self._dynamic_sleep(consecutive_unchanged, has_received_anything)
 
         elapsed = time.time() - start_time
         logger.info(f"Job finished: {len(full_text)} chars in {elapsed:.1f}s")
@@ -142,10 +150,15 @@ class JobsBackend(EmergentBackend):
         }
 
     @staticmethod
-    async def _dynamic_sleep(consecutive_unchanged: int) -> None:
-        if consecutive_unchanged == 0:
-            await asyncio.sleep(1.0)
+    async def _dynamic_sleep(consecutive_unchanged: int, has_content: bool) -> None:
+        """Adaptive sleep: poll fast early, back off when stable."""
+        if not has_content:
+            await asyncio.sleep(0.8)
+        elif consecutive_unchanged == 0:
+            await asyncio.sleep(0.5)
         elif consecutive_unchanged < 3:
+            await asyncio.sleep(1.0)
+        elif consecutive_unchanged < 8:
             await asyncio.sleep(2.0)
         else:
             await asyncio.sleep(3.0)

emergent2api/database.py

@@ -41,6 +41,13 @@ BEGIN
     ) THEN
         ALTER TABLE emergent_accounts ADD COLUMN use_count INTEGER DEFAULT 0;
     END IF;
+
+    IF NOT EXISTS (
+        SELECT 1 FROM information_schema.columns
+        WHERE table_name = 'emergent_accounts' AND column_name = 'integration_key'
+    ) THEN
+        ALTER TABLE emergent_accounts ADD COLUMN integration_key TEXT DEFAULT '';
+    END IF;
 END $$;
 """
 

emergent2api/routes/admin.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 import io
 import json
 import logging
+import os
 import secrets
 import zipfile
 from typing import Any
@@ -313,6 +314,7 @@ async def get_config(request: Request):
         "poll_interval": db_config.get("poll_interval", str(settings.poll_interval)),
         "poll_timeout": db_config.get("poll_timeout", str(settings.poll_timeout)),
         "admin_password": db_config.get("admin_password", settings.admin_password),
+        "anthropic_key": db_config.get("anthropic_key", os.environ.get("ANTHROPIC_API_KEY", "")),
     }
 
 
@@ -322,7 +324,7 @@ async def update_config(request: Request):
     if err:
         return err
     body = await request.json()
-    allowed = {"api_key", "backend", "proxy", "poll_interval", "poll_timeout", "admin_password"}
+    allowed = {"api_key", "backend", "proxy", "poll_interval", "poll_timeout", "admin_password", "anthropic_key"}
     saved = []
     for key, value in body.items():
         if key in allowed:
@@ -335,6 +337,8 @@ async def update_config(request: Request):
                 settings.proxy = str(value)
             elif key == "admin_password":
                 settings.admin_password = str(value)
+            elif key == "anthropic_key":
+                os.environ["ANTHROPIC_API_KEY"] = str(value)
             saved.append(key)
     return {"saved": saved}
 

emergent2api/static/admin/index.html

@@ -234,9 +234,10 @@ tr:hover td{background:rgba(255,255,255,.02)}
     </div>
     <div class="fld">
       <label>Backend</label>
-      <select id="cfg-backend"><option value="jobs">Jobs API（无 IP 限制）</option><option value="integrations">Integrations API（更快，需代理）</option></select>
+      <select id="cfg-backend"><option value="jobs">Jobs API（无 IP 限制，推荐）</option><option value="integrations">Integrations API（需 Anthropic Key + 代理）</option></select>
     </div>
-    <div class="fld"><label>Proxy</label><input id="cfg-proxy" placeholder="http://user:pass@host:port" spellcheck="false"><div class="hint">Integrations 后端必填；Jobs 可选</div></div>
+    <div class="fld"><label>Proxy</label><input id="cfg-proxy" placeholder="http://user:pass@host:port" spellcheck="false"><div class="hint">Integrations 后端必填（需美国 IP）；Jobs 可选</div></div>
+    <div class="fld"><label>ANTHROPIC_API_KEY</label><input id="cfg-anthropic_key" placeholder="sk-ant-... (Integrations 后端专用)" spellcheck="false"><div class="hint">仅 Integrations 后端需要；可留空使用 Jobs 后端</div></div>
   </div>
   <div class="card">
     <h3>Admin</h3>
@@ -443,9 +444,10 @@ async function loadConfig(){
   const r=await api('/config');if(!r)return;const d=await r.json();
   $('cfg-api_key').value=d.api_key||'';$('cfg-backend').value=d.backend||'jobs';
   $('cfg-proxy').value=d.proxy||'';$('cfg-admin_password').value=d.admin_password||'';
+  $('cfg-anthropic_key').value=d.anthropic_key||'';
 }
 async function saveConfig(){
-  const b={api_key:$('cfg-api_key').value,backend:$('cfg-backend').value,proxy:$('cfg-proxy').value,admin_password:$('cfg-admin_password').value};
+  const b={api_key:$('cfg-api_key').value,backend:$('cfg-backend').value,proxy:$('cfg-proxy').value,admin_password:$('cfg-admin_password').value,anthropic_key:$('cfg-anthropic_key').value};
   const r=await api('/config',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify(b)});
   if(r){const d=await r.json();toast(`已保存: ${d.saved.join(', ')}`)}
 }