Merge pull request #25 from a8851625/agent/miniclaw-claude/1dc8c2c3

Fix account-password auth flow and streaming delivery

Author: YuJunZhiXue

backend/api/admin.py

@@ -124,21 +124,31 @@ async def add_account(request: Request):
     except Exception:
         raise HTTPException(400, detail="Invalid JSON body")
 
-    token = data.get("token", "")
-    if not token:
-        raise HTTPException(400, detail="token is required")
+    email = str(data.get("email", "") or "").strip() or f"manual_{int(time.time())}@qwen"
+    password = str(data.get("password", "") or "").strip()
+    token = str(data.get("token", "") or "").strip()
 
     acc = Account(
-        email=data.get("email", f"manual_{int(time.time())}@qwen"),
-        password=data.get("password", ""),
+        email=email,
+        password=password,
         token=token,
         cookies=data.get("cookies", ""),
         username=data.get("username", "")
     )
 
-    is_valid = await client.verify_token(token)
-    if not is_valid:
-        return {"ok": False, "error": "Invalid token (验证失败，请确认Token有效)"}
+    # 兼容两种接入方式：
+    # 1) 直接传 token
+    # 2) 仅传 email+password，由服务端自动登录并获取 token
+    if acc.token:
+        is_valid = await client.verify_token(acc.token)
+        if not is_valid:
+            return {"ok": False, "error": "Invalid token (验证失败，请确认Token有效)"}
+    else:
+        if not acc.email or not acc.password:
+            raise HTTPException(400, detail="token or email+password is required")
+        refreshed = await client.auth_resolver.refresh_token(acc)
+        if not refreshed or not acc.token:
+            return {"ok": False, "error": "Email/password login failed (账号密码登录失败)"}
 
     await pool.add(acc)
     return {"ok": True, "email": acc.email}

backend/api/v1_chat.py

@@ -1,5 +1,6 @@
 from fastapi import APIRouter, HTTPException, Request
 from fastapi.responses import JSONResponse, StreamingResponse
+import asyncio
 import json
 import logging
 import time
@@ -125,63 +126,82 @@ async def chat_completions(request: Request):
 
         if standard_request.stream:
             async def generate():
-                async with app.state.session_locks.hold(session_key):
-                    try:
-                        update_request_context(stream_attempt=1)
-                        translator = OpenAIStreamTranslator(
-                            completion_id=completion_id,
-                            created=created,
-                            model_name=model_name,
-                            client_profile=standard_request.client_profile,
-                            build_final_directive=lambda answer_text: build_tool_directive(
-                                standard_request,
-                                RuntimeAttemptState(answer_text=answer_text),
-                            ),
-                            allowed_tool_names=standard_request.tool_names,
-                        )
-
-                        async def on_delta(evt: dict[str, Any], text_chunk: str | None, tool_calls: list[dict[str, Any]] | None) -> None:
-                            translator.on_delta(evt, text_chunk, tool_calls)
-
-                        result = await run_retryable_completion_bridge(
-                            client=client,
-                            standard_request=standard_request,
-                            prompt=prompt,
-                            users_db=users_db,
-                            token=token,
-                            history_messages=history_messages,
-                            max_attempts=request_max_attempts(standard_request),
-                            usage_delta_factory=build_usage_delta_factory(prompt),
-                            allow_after_visible_output=True,
-                            capture_events=False,
-                            on_delta=on_delta,
-                        )
-                        execution = result.execution
-                        directive = result.directive or build_tool_directive(standard_request, execution.state)
-                        assistant_message = build_openai_assistant_history_message(
-                            execution=execution,
-                            request=standard_request,
-                            directive=directive,
-                        )
-                        await persist_session_turn(
-                            app=app,
-                            request=standard_request,
-                            surface="openai",
-                            execution=execution,
-                            assistant_message=assistant_message,
-                        )
-                        final_finish_reason = "tool_calls" if directive.stop_reason == "tool_use" else execution.state.finish_reason
-                        for chunk in translator.finalize(final_finish_reason):
-                            yield chunk
-                        return
-                    except HTTPException as he:
-                        await clear_invalidated_session_chat(app=app, request=standard_request)
-                        yield f"data: {json.dumps({'error': he.detail})}\n\n"
-                        return
-                    except Exception as e:
-                        await clear_invalidated_session_chat(app=app, request=standard_request)
-                        yield f"data: {json.dumps({'error': str(e)})}\n\n"
-                        return
+                queue: asyncio.Queue[str | None] = asyncio.Queue()
+
+                async def producer() -> None:
+                    async with app.state.session_locks.hold(session_key):
+                        try:
+                            update_request_context(stream_attempt=1)
+                            translator = OpenAIStreamTranslator(
+                                completion_id=completion_id,
+                                created=created,
+                                model_name=model_name,
+                                client_profile=standard_request.client_profile,
+                                build_final_directive=lambda answer_text: build_tool_directive(
+                                    standard_request,
+                                    RuntimeAttemptState(answer_text=answer_text),
+                                ),
+                                allowed_tool_names=standard_request.tool_names,
+                            )
+
+                            async def on_delta(evt: dict[str, Any], text_chunk: str | None, tool_calls: list[dict[str, Any]] | None) -> None:
+                                translator.on_delta(evt, text_chunk, tool_calls)
+                                while translator.pending_chunks:
+                                    await queue.put(translator.pending_chunks.pop(0))
+
+                            result = await run_retryable_completion_bridge(
+                                client=client,
+                                standard_request=standard_request,
+                                prompt=prompt,
+                                users_db=users_db,
+                                token=token,
+                                history_messages=history_messages,
+                                max_attempts=request_max_attempts(standard_request),
+                                usage_delta_factory=build_usage_delta_factory(prompt),
+                                allow_after_visible_output=True,
+                                capture_events=False,
+                                on_delta=on_delta,
+                            )
+                            execution = result.execution
+                            directive = result.directive or build_tool_directive(standard_request, execution.state)
+                            assistant_message = build_openai_assistant_history_message(
+                                execution=execution,
+                                request=standard_request,
+                                directive=directive,
+                            )
+                            await persist_session_turn(
+                                app=app,
+                                request=standard_request,
+                                surface="openai",
+                                execution=execution,
+                                assistant_message=assistant_message,
+                            )
+                            final_finish_reason = "tool_calls" if directive.stop_reason == "tool_use" else (execution.state.finish_reason or "stop")
+                            for chunk in translator.finalize(final_finish_reason):
+                                await queue.put(chunk)
+                        except HTTPException as he:
+                            await clear_invalidated_session_chat(app=app, request=standard_request)
+                            await queue.put(f"data: {json.dumps({'error': he.detail})}\n\n")
+                        except Exception as e:
+                            await clear_invalidated_session_chat(app=app, request=standard_request)
+                            await queue.put(f"data: {json.dumps({'error': str(e)})}\n\n")
+                        finally:
+                            await queue.put(None)
+
+                producer_task = asyncio.create_task(producer())
+                try:
+                    while True:
+                        chunk = await queue.get()
+                        if chunk is None:
+                            break
+                        yield chunk
+                finally:
+                    if not producer_task.done():
+                        producer_task.cancel()
+                        try:
+                            await producer_task
+                        except Exception:
+                            pass
 
             return StreamingResponse(
                 generate(),

backend/services/auth_resolver.py

@@ -1,4 +1,5 @@
 import asyncio
+import hashlib
 import html as html_lib
 import json
 import logging
@@ -736,6 +737,10 @@ class AuthResolver:
     def __init__(self, pool: AccountPool):
         self.pool = pool
 
+    @staticmethod
+    def _sha256_password(password: str) -> str:
+        return hashlib.sha256((password or "").encode("utf-8")).hexdigest()
+
     async def auto_heal_account(self, acc: Account):
         """Background task to refresh token. If successful, marks account valid.
         If refresh fails or account is pending activation, tries to activate via email."""
@@ -775,27 +780,80 @@ async def refresh_token(self, acc: Account) -> bool:
         if not acc.email or not acc.password:
             log.warning(f"[Refresh] 账号 {acc.email} 无密码，无法刷新")
             return False
-            
+
         log.info(f"[Refresh] 正在为 {acc.email} 刷新 token...")
         try:
-            async with _new_browser() as browser:
-                page = await browser.new_page()
-                new_token = await _login_and_get_token(page, acc.email, acc.password, timeout_sec=20)
-                if new_token and new_token != acc.token:
-                    old_prefix = acc.token[:20] if acc.token else "空"
-                    acc.token = new_token
-                    acc.valid = True
-                    await self.pool.save()
-                    log.info(f"[Refresh] {acc.email} token 已更新 ({old_prefix}... → {new_token[:20]}...)")
-                    return True
-                elif new_token == acc.token:
-                    # Token same but might still be valid — mark valid again
-                    acc.valid = True
-                    log.info(f"[Refresh] {acc.email} token 未变化，重新标记有效")
-                    return True
-                else:
-                    log.warning(f"[Refresh] {acc.email} 登录后未获取到token，URL={page.url}")
-                    return False
+            import httpx
+
+            payload = {
+                "email": acc.email,
+                "password": self._sha256_password(acc.password),
+            }
+            headers = {
+                "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
+                              "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
+                "Accept": "application/json, text/plain, */*",
+                "Content-Type": "application/json",
+                "Referer": f"{BASE_URL}/",
+                "Origin": BASE_URL,
+            }
+
+            async with httpx.AsyncClient(timeout=15, follow_redirects=True) as hc:
+                resp = await hc.post(f"{BASE_URL}/api/v1/auths/signin", json=payload, headers=headers)
+
+            if resp.status_code != 200:
+                log.warning(f"[Refresh] {acc.email} HTTP {resp.status_code}，登录失败")
+                return False
+
+            try:
+                data = resp.json()
+            except Exception:
+                log.warning(f"[Refresh] {acc.email} 登录响应不是 JSON")
+                return False
+
+            new_token = str(data.get("token", "") or "").strip()
+            if not new_token:
+                log.warning(f"[Refresh] {acc.email} 登录响应缺少 token 字段")
+                return False
+
+            old_prefix = acc.token[:20] if acc.token else "空"
+            acc.token = new_token
+            acc.valid = True
+            acc.activation_pending = False
+            acc.status_code = "valid"
+            acc.last_error = ""
+            await self.pool.save()
+            log.info(f"[Refresh] {acc.email} token 已更新 ({old_prefix}... → {new_token[:20]}...)")
+            return True
+
         except Exception as e:
-            log.error(f"[Refresh] {acc.email} 刷新异常: {e}")
-            return False
+            log.warning(f"[Refresh] {acc.email} 直连登录失败，回退浏览器流程: {e}")
+            try:
+                async with _new_browser() as browser:
+                    page = await browser.new_page()
+                    new_token = await _login_and_get_token(page, acc.email, acc.password, timeout_sec=20)
+                    if new_token and new_token != acc.token:
+                        old_prefix = acc.token[:20] if acc.token else "空"
+                        acc.token = new_token
+                        acc.valid = True
+                        acc.activation_pending = False
+                        acc.status_code = "valid"
+                        acc.last_error = ""
+                        await self.pool.save()
+                        log.info(f"[Refresh] {acc.email} token 已更新 ({old_prefix}... → {new_token[:20]}...)")
+                        return True
+                    elif new_token == acc.token:
+                        # Token same but might still be valid — mark valid again
+                        acc.valid = True
+                        acc.activation_pending = False
+                        if acc.status_code in ("invalid", "auth_error", "pending_activation"):
+                            acc.status_code = "valid"
+                        acc.last_error = ""
+                        log.info(f"[Refresh] {acc.email} token 未变化，重新标记有效")
+                        return True
+                    else:
+                        log.warning(f"[Refresh] {acc.email} 登录后未获取到token，URL={page.url}")
+                        return False
+            except Exception as browser_err:
+                log.error(f"[Refresh] {acc.email} 刷新异常: {browser_err}")
+                return False