diff --git a/backend_api_python/Dockerfile.ai-isolated b/backend_api_python/Dockerfile.ai-isolated
new file mode 100644
index 000000000..a5cfd0058
--- /dev/null
+++ b/backend_api_python/Dockerfile.ai-isolated
@@ -0,0 +1,10 @@
+FROM aq-qd-result-003:1173a0b
+
+USER root
+WORKDIR /app
+COPY . /app
+RUN chown -R quantdinger:quantdinger /app
+USER quantdinger
+
+ENTRYPOINT ["/app/docker-entrypoint.sh"]
+CMD ["gunicorn", "--config", "gunicorn_config.py", "run:app"]
diff --git a/backend_api_python/app/routes/strategy.py b/backend_api_python/app/routes/strategy.py
index f52a12c4e..678817cf6 100644
--- a/backend_api_python/app/routes/strategy.py
+++ b/backend_api_python/app/routes/strategy.py
@@ -69,6 +69,27 @@ def _request_lang(default: str = "zh-CN") -> str:
     return lang or default
 
 
+@strategy_blp.route("/strategies/ai-workspace/account-models", methods=["GET"])
+@login_required
+def list_strategy_workspace_account_models():
+    """Return the safe model catalog owned by the local Codex login."""
+    try:
+        from app.services.chatgpt_account import CodexAppServerClient
+
+        models = CodexAppServerClient(timeout_seconds=30).list_models()
+        return _ok({
+            "authMode": "chatgpt_account",
+            "connected": True,
+            "models": [
+                {"id": item.slug, "name": item.display_name}
+                for item in models
+            ],
+        })
+    except Exception as exc:
+        logger.warning("ChatGPT account model discovery failed: %s", exc)
+        return _error("strategyV2.chatgptAccountUnavailable", 503)
+
+
 def _strategy_ai_text(key: str, lang: str = "zh-CN") -> str:
     texts = {
         STRATEGY_CANDIDATE_MESSAGE_KEY: (
@@ -348,7 +369,9 @@ def generate_strategy():
         from app.services.billing_service import get_billing_service
         from app.services.llm import LLMService
 
-        llm = LLMService()
+        auth_mode = str(payload.get("authMode") or "api_key").strip().lower()
+        requested_model = str(payload.get("model") or "").strip() or None
+        llm = LLMService(auth_mode=auth_mode)
         if not llm.is_configured():
             return _error("strategyV2.llmNotConfigured")
         asset_type = normalize_asset_type(payload.get("assetType") or payload.get("asset_type"))
@@ -629,7 +652,9 @@ def run_strategy_workspace_turn():
         context = payload.get("context") if isinstance(payload.get("context"), dict) else {}
         requested_mode = str(payload.get("interactionMode") or "auto")
         generation_mode = str(payload.get("generationMode") or "authoring").strip().lower()
-        llm = LLMService()
+        auth_mode = str(payload.get("authMode") or "api_key").strip().lower()
+        requested_model = str(payload.get("model") or "").strip() or None
+        llm = LLMService(auth_mode=auth_mode)
         if not llm.is_configured():
             return _error("strategyV2.llmNotConfigured")
         intent_workspace = None
@@ -710,7 +735,7 @@ def run_strategy_workspace_turn():
             logger.info("strategy discussion context budget=%s", budget)
             answer = str(llm.call_llm_api(
                 messages=messages,
-                model=llm.get_default_model(),
+                model=requested_model or llm.get_default_model(),
                 temperature=0.2,
                 use_json_mode=False,
             ) or "").strip()
@@ -780,7 +805,7 @@ def run_strategy_workspace_turn():
         else:
             generated = llm.call_llm_api(
                 messages=messages,
-                model=llm.get_code_generation_model(),
+                model=requested_model or llm.get_code_generation_model(),
                 temperature=0.2 if use_patch_response else 0.35,
                 use_json_mode=use_patch_response,
             )
@@ -802,7 +827,7 @@ def run_strategy_workspace_turn():
                     full_messages[-1] = {"role": "user", "content": full_request}
                     fallback = llm.call_llm_api(
                         messages=full_messages,
-                        model=llm.get_code_generation_model(),
+                        model=requested_model or llm.get_code_generation_model(),
                         temperature=0.25,
                         use_json_mode=False,
                     )
diff --git a/backend_api_python/app/services/chatgpt_account.py b/backend_api_python/app/services/chatgpt_account.py
new file mode 100644
index 000000000..197eb12ee
--- /dev/null
+++ b/backend_api_python/app/services/chatgpt_account.py
@@ -0,0 +1,372 @@
+"""Official Codex app-server transport for a local ChatGPT account.
+
+The transport never reads or copies OAuth credentials.  Authentication and
+refresh remain owned by the locally installed Codex CLI.  QuantDinger talks to
+the documented app-server JSON-RPC protocol over stdio and deliberately does
+not fall back to an API-key provider.
+"""
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import subprocess
+import tempfile
+import threading
+import time
+import urllib.error
+import urllib.request
+from dataclasses import dataclass
+from typing import Any, Callable, Iterable
+
+
+class ChatGPTAccountError(RuntimeError):
+    """Safe, credential-free account transport failure."""
+
+
+@dataclass(frozen=True)
+class ChatGPTAccountModel:
+    slug: str
+    display_name: str
+
+
+def _message_input(messages: Iterable[dict[str, Any]]) -> str:
+    """Render complete QD history into one bounded app-server turn."""
+    rendered: list[str] = []
+    for message in messages:
+        role = str(message.get("role") or "user").strip().lower()
+        if role not in {"system", "developer", "user", "assistant"}:
+            continue
+        content = message.get("content")
+        if isinstance(content, list):
+            content = "\n".join(
+                str(item.get("text") or "") if isinstance(item, dict) else str(item)
+                for item in content
+            )
+        text = str(content or "").strip()
+        if text:
+            rendered.append(f"[{role.upper()}]\n{text}")
+    if not rendered:
+        raise ChatGPTAccountError("ChatGPT account request has no text input")
+    return "\n\n".join(rendered)
+
+
+class CodexAppServerClient:
+    """One-request client for the official local Codex app-server."""
+
+    def __init__(
+        self,
+        *,
+        executable: str | None = None,
+        timeout_seconds: float = 180.0,
+        popen_factory: Callable[..., subprocess.Popen[str]] = subprocess.Popen,
+    ) -> None:
+        self.executable = executable or os.getenv("CODEX_CLI_PATH", "codex")
+        self.timeout_seconds = max(5.0, float(timeout_seconds))
+        self._popen_factory = popen_factory
+
+    def _bridge_request(self, path: str, payload: dict[str, Any] | None = None) -> Any:
+        base_url = os.getenv("CHATGPT_ACCOUNT_BRIDGE_URL", "").strip().rstrip("/")
+        if not base_url:
+            return None
+        token = os.getenv("CHATGPT_ACCOUNT_BRIDGE_TOKEN", "").strip()
+        if not token:
+            raise ChatGPTAccountError("ChatGPT account bridge token is not configured")
+        body = None if payload is None else json.dumps(payload).encode("utf-8")
+        request = urllib.request.Request(
+            f"{base_url}{path}",
+            data=body,
+            method="GET" if body is None else "POST",
+            headers={
+                "Authorization": f"Bearer {token}",
+                "Content-Type": "application/json",
+            },
+        )
+        try:
+            with urllib.request.urlopen(request, timeout=self.timeout_seconds) as response:
+                return json.loads(response.read().decode("utf-8"))
+        except (urllib.error.URLError, ValueError) as exc:
+            raise ChatGPTAccountError("ChatGPT account bridge request failed") from exc
+
+    def _command(self) -> list[str]:
+        resolved = shutil.which(self.executable) if not os.path.isabs(self.executable) else self.executable
+        if not resolved or not os.path.exists(resolved):
+            raise ChatGPTAccountError(
+                "ChatGPT account mode requires the official Codex CLI on this local host"
+            )
+        return [
+            resolved,
+            "app-server",
+            "--listen",
+            "stdio://",
+            "-c",
+            "mcp_servers={}",
+            "-c",
+            "analytics.enabled=false",
+        ]
+
+    def _run(
+        self,
+        *,
+        requests: list[dict[str, Any]],
+        notification_handler: Callable[[dict[str, Any]], None] | None = None,
+    ) -> dict[int, dict[str, Any]]:
+        with tempfile.TemporaryDirectory(
+            prefix="quantdinger-chatgpt-", ignore_cleanup_errors=True
+        ) as isolated_cwd:
+            process = self._popen_factory(
+                self._command(),
+                cwd=isolated_cwd,
+                stdin=subprocess.PIPE,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                text=True,
+                encoding="utf-8",
+                errors="replace",
+                bufsize=1,
+                env={**os.environ, "NO_COLOR": "1"},
+            )
+            if process.stdin is None or process.stdout is None:
+                process.kill()
+                raise ChatGPTAccountError("Codex app-server stdio is unavailable")
+
+            responses: dict[int, dict[str, Any]] = {}
+            read_error: list[BaseException] = []
+
+            def read_stdout() -> None:
+                try:
+                    for line in process.stdout:
+                        try:
+                            payload = json.loads(line)
+                        except json.JSONDecodeError:
+                            continue
+                        if isinstance(payload.get("id"), int):
+                            responses[int(payload["id"])] = payload
+                        elif notification_handler:
+                            notification_handler(payload)
+                except BaseException as exc:  # pragma: no cover - OS pipe failure
+                    read_error.append(exc)
+
+            reader = threading.Thread(target=read_stdout, daemon=True)
+            reader.start()
+            try:
+                for payload in requests:
+                    process.stdin.write(json.dumps(payload, separators=(",", ":")) + "\n")
+                    process.stdin.flush()
+                deadline = time.monotonic() + self.timeout_seconds
+                expected_ids = {int(item["id"]) for item in requests if isinstance(item.get("id"), int)}
+                while not expected_ids.issubset(responses):
+                    if read_error:
+                        raise ChatGPTAccountError("Codex app-server output failed")
+                    if process.poll() is not None:
+                        raise ChatGPTAccountError("Codex app-server stopped before completing the request")
+                    if time.monotonic() >= deadline:
+                        raise ChatGPTAccountError("Codex app-server request timed out")
+                    time.sleep(0.02)
+                return responses
+            finally:
+                try:
+                    process.stdin.close()
+                except Exception:
+                    pass
+                if process.poll() is None:
+                    process.terminate()
+                    try:
+                        process.wait(timeout=3)
+                    except subprocess.TimeoutExpired:
+                        process.kill()
+                reader.join(timeout=1)
+
+    @staticmethod
+    def _raise_rpc_error(response: dict[str, Any], operation: str) -> None:
+        if response.get("error"):
+            error = response["error"]
+            code = error.get("code") if isinstance(error, dict) else "unknown"
+            raise ChatGPTAccountError(f"{operation} failed (code={code})")
+
+    @staticmethod
+    def _initialize_request() -> dict[str, Any]:
+        return {
+            "jsonrpc": "2.0",
+            "id": 1,
+            "method": "initialize",
+            "params": {
+                "clientInfo": {
+                    "name": "quantdinger",
+                    "title": "QuantDinger",
+                    "version": "5.7",
+                }
+            },
+        }
+
+    def list_models(self) -> list[ChatGPTAccountModel]:
+        bridged = self._bridge_request("/v1/models")
+        if bridged is not None:
+            return [
+                ChatGPTAccountModel(slug=str(item["id"]), display_name=str(item.get("name") or item["id"]))
+                for item in bridged.get("models", [])
+                if item.get("id")
+            ]
+        responses = self._run(
+            requests=[
+                self._initialize_request(),
+                {"jsonrpc": "2.0", "method": "initialized", "params": {}},
+                {
+                    "jsonrpc": "2.0",
+                    "id": 2,
+                    "method": "model/list",
+                    "params": {"includeHidden": False},
+                },
+            ]
+        )
+        self._raise_rpc_error(responses[1], "ChatGPT account initialization")
+        self._raise_rpc_error(responses[2], "ChatGPT account model discovery")
+        data = (responses[2].get("result") or {}).get("data") or []
+        models: list[ChatGPTAccountModel] = []
+        for item in data:
+            slug = str(item.get("model") or item.get("slug") or "").strip()
+            if slug:
+                models.append(
+                    ChatGPTAccountModel(
+                        slug=slug,
+                        display_name=str(item.get("displayName") or item.get("display_name") or slug),
+                    )
+                )
+        return models
+
+    def complete(
+        self,
+        messages: list[dict[str, Any]],
+        *,
+        model: str | None = None,
+        on_delta: Callable[[str], None] | None = None,
+    ) -> str:
+        bridged = self._bridge_request(
+            "/v1/complete",
+            {"messages": messages, "model": model},
+        )
+        if bridged is not None:
+            result = str(bridged.get("text") or "").strip()
+            if not result:
+                raise ChatGPTAccountError("ChatGPT account bridge returned an empty response")
+            if on_delta:
+                on_delta(result)
+            return result
+        text_parts: list[str] = []
+        completed: list[str] = []
+
+        def notification(payload: dict[str, Any]) -> None:
+            method = payload.get("method")
+            params = payload.get("params") or {}
+            if method == "item/agentMessage/delta":
+                delta = str(params.get("delta") or "")
+                if delta:
+                    text_parts.append(delta)
+                    if on_delta:
+                        on_delta(delta)
+            elif method == "turn/completed":
+                turn = params.get("turn") or {}
+                completed.append(str(turn.get("status") or ""))
+
+        thread_params: dict[str, Any] = {
+            "approvalPolicy": "never",
+            "sandbox": "read-only",
+            "ephemeral": True,
+            "developerInstructions": (
+                "Return text only. Do not call tools, read files, access MCP servers, "
+                "or execute commands. Follow the supplied QuantDinger conversation."
+            ),
+        }
+        if model:
+            thread_params["model"] = model
+        with tempfile.TemporaryDirectory(
+            prefix="quantdinger-chatgpt-", ignore_cleanup_errors=True
+        ) as isolated_cwd:
+            process = self._popen_factory(
+                self._command(), cwd=isolated_cwd, stdin=subprocess.PIPE,
+                stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
+                encoding="utf-8", errors="replace", bufsize=1,
+                env={**os.environ, "NO_COLOR": "1"},
+            )
+            if process.stdin is None or process.stdout is None:
+                process.kill()
+                raise ChatGPTAccountError("Codex app-server stdio is unavailable")
+            responses: dict[int, dict[str, Any]] = {}
+            condition = threading.Condition()
+
+            def read_stdout() -> None:
+                for line in process.stdout:
+                    try:
+                        payload = json.loads(line)
+                    except json.JSONDecodeError:
+                        continue
+                    with condition:
+                        if isinstance(payload.get("id"), int):
+                            responses[int(payload["id"])] = payload
+                        else:
+                            notification(payload)
+                        condition.notify_all()
+
+            reader = threading.Thread(target=read_stdout, daemon=True)
+            reader.start()
+
+            def send(payload: dict[str, Any]) -> None:
+                process.stdin.write(json.dumps(payload, separators=(",", ":")) + "\n")
+                process.stdin.flush()
+
+            def wait_for(predicate: Callable[[], bool], error: str) -> None:
+                deadline = time.monotonic() + self.timeout_seconds
+                with condition:
+                    while not predicate():
+                        if process.poll() is not None:
+                            raise ChatGPTAccountError("Codex app-server stopped before completing the request")
+                        remaining = deadline - time.monotonic()
+                        if remaining <= 0:
+                            raise ChatGPTAccountError(error)
+                        condition.wait(min(remaining, 0.25))
+
+            try:
+                send(self._initialize_request())
+                wait_for(lambda: 1 in responses, "ChatGPT account initialization timed out")
+                self._raise_rpc_error(responses[1], "ChatGPT account initialization")
+                send({"jsonrpc": "2.0", "method": "initialized", "params": {}})
+                send({"jsonrpc": "2.0", "id": 2, "method": "thread/start", "params": thread_params})
+                wait_for(lambda: 2 in responses, "ChatGPT account thread start timed out")
+                self._raise_rpc_error(responses[2], "ChatGPT account thread start")
+                thread = (responses[2].get("result") or {}).get("thread") or {}
+                thread_id = str(thread.get("id") or "")
+                if not thread_id:
+                    raise ChatGPTAccountError("Codex app-server returned no thread id")
+                send({
+                    "jsonrpc": "2.0", "id": 3, "method": "turn/start",
+                    "params": {
+                        "threadId": thread_id,
+                        "input": [{"type": "text", "text": _message_input(messages)}],
+                    },
+                })
+                wait_for(lambda: 3 in responses, "ChatGPT account turn start timed out")
+                self._raise_rpc_error(responses[3], "ChatGPT account turn start")
+                wait_for(lambda: bool(completed), "ChatGPT account turn timed out")
+                if completed[-1] != "completed":
+                    raise ChatGPTAccountError("ChatGPT account turn did not complete successfully")
+            finally:
+                try:
+                    process.stdin.close()
+                except Exception:
+                    pass
+                if process.poll() is None:
+                    process.terminate()
+                    try:
+                        process.wait(timeout=3)
+                    except subprocess.TimeoutExpired:
+                        process.kill()
+                reader.join(timeout=1)
+        result = "".join(text_parts).strip()
+        if not result:
+            raise ChatGPTAccountError("ChatGPT account returned an empty response")
+        return result
+
+
+def account_mode_enabled() -> bool:
+    return os.getenv("LLM_AUTH_MODE", "api_key").strip().lower() == "chatgpt_account"
+
diff --git a/backend_api_python/app/services/llm.py b/backend_api_python/app/services/llm.py
index 3563b4800..7007fcca2 100644
--- a/backend_api_python/app/services/llm.py
+++ b/backend_api_python/app/services/llm.py
@@ -6,6 +6,9 @@
 """
 import json
 import os
+import queue
+import shutil
+import threading
 import requests
 from typing import Dict, Any, Optional, List
 from enum import Enum
@@ -170,7 +173,7 @@ class LLMProvider(Enum):
 class LLMService:
     """LLM provider wrapper with multi-provider support."""
 
-    def __init__(self, provider: str = None):
+    def __init__(self, provider: str = None, auth_mode: str = None):
         """
         Initialize LLM service.
 
@@ -178,6 +181,13 @@ def __init__(self, provider: str = None):
             provider: Override the default provider (openrouter, openai, google, deepseek, grok, atlascloud, custom, minimax)
         """
         self._provider_override = provider
+        self.auth_mode = str(
+            auth_mode
+            or load_addon_config().get('llm', {}).get('auth_mode')
+            or os.getenv('LLM_AUTH_MODE', 'api_key')
+        ).strip().lower()
+        if self.auth_mode not in {'api_key', 'chatgpt_account'}:
+            raise ValueError(f"Unsupported LLM auth mode: {self.auth_mode}")
         # Populated by call_llm_api() so callers can report what actually answered;
         # a model-level fallback is otherwise only visible in provider dashboards.
         self.last_model_used = ""
@@ -263,6 +273,8 @@ def get_base_url(self, provider: LLMProvider = None) -> str:
 
     def get_default_model(self, provider: LLMProvider = None) -> str:
         """Get default model for the specified provider."""
+        if self.auth_mode == 'chatgpt_account':
+            return os.getenv('CHATGPT_ACCOUNT_MODEL', 'gpt-6-luna').strip() or 'gpt-6-luna'
         p = provider or self.provider
         config = load_addon_config()
         
@@ -276,6 +288,8 @@ def get_default_model(self, provider: LLMProvider = None) -> str:
 
     def get_code_generation_model(self, provider: LLMProvider = None) -> str:
         """Get model for AI code generation; fallback to provider default when unset."""
+        if self.auth_mode == 'chatgpt_account':
+            return self.get_default_model(provider)
         model = os.getenv('AI_CODE_GEN_MODEL', '').strip()
         if model:
             return model
@@ -297,6 +311,13 @@ def get_max_tokens(self) -> int:
 
     def is_configured(self, provider: LLMProvider = None) -> bool:
         """Return whether the provider has enough configuration to make a request."""
+        if self.auth_mode == 'chatgpt_account':
+            bridge_url = os.getenv('CHATGPT_ACCOUNT_BRIDGE_URL', '').strip()
+            bridge_token = os.getenv('CHATGPT_ACCOUNT_BRIDGE_TOKEN', '').strip()
+            if bridge_url:
+                return bool(bridge_token)
+            executable = os.getenv('CODEX_CLI_PATH', 'codex').strip() or 'codex'
+            return bool(shutil.which(executable) or (os.path.isabs(executable) and os.path.exists(executable)))
         p = provider or self.provider
         if (self.get_api_key(p) or "").strip():
             return True
@@ -1203,6 +1224,17 @@ def call_llm_api(self, messages: list, model: str = None, temperature: float = 0
             2. Otherwise, use the configured LLM_PROVIDER with normalized model name
             3. Fall back to provider's default model if model name is incompatible
         """
+        if self.auth_mode == 'chatgpt_account':
+            from app.services.chatgpt_account import CodexAppServerClient
+            requested_model = model or self.get_default_model()
+            result = CodexAppServerClient(
+                timeout_seconds=timeout_seconds or 180,
+            ).complete(messages, model=requested_model)
+            self.last_model_used = requested_model
+            self.last_provider_used = 'chatgpt_account'
+            self.model_fallback_used = False
+            return result
+
         cfg = load_addon_config()
         explicit_provider_name = str(
             cfg.get('llm', {}).get('provider') or os.getenv('LLM_PROVIDER', '')
@@ -1430,6 +1462,36 @@ def call_llm_api(self, messages: list, model: str = None, temperature: float = 0
 
     def stream_llm_api(self, messages: list, model: str = None, temperature: float = 0.7):
         """Stream LLM response deltas for providers with OpenAI-compatible streaming."""
+        if self.auth_mode == 'chatgpt_account':
+            from app.services.chatgpt_account import CodexAppServerClient
+            requested_model = model or self.get_default_model()
+            deltas: queue.Queue[object] = queue.Queue()
+            sentinel = object()
+
+            def run_account_turn() -> None:
+                try:
+                    CodexAppServerClient().complete(
+                        messages,
+                        model=requested_model,
+                        on_delta=deltas.put,
+                    )
+                except BaseException as exc:
+                    deltas.put(exc)
+                finally:
+                    deltas.put(sentinel)
+
+            threading.Thread(target=run_account_turn, daemon=True).start()
+            while True:
+                item = deltas.get()
+                if item is sentinel:
+                    break
+                if isinstance(item, BaseException):
+                    raise item
+                yield str(item)
+            self.last_model_used = requested_model
+            self.last_provider_used = 'chatgpt_account'
+            self.model_fallback_used = False
+            return
         p = self.provider
         api_key = (self.get_api_key(p) or "").strip()
         base_url = (self.get_base_url(p) or "").strip()
diff --git a/backend_api_python/scripts/chatgpt_account_bridge.py b/backend_api_python/scripts/chatgpt_account_bridge.py
new file mode 100644
index 000000000..5ddcc5c94
--- /dev/null
+++ b/backend_api_python/scripts/chatgpt_account_bridge.py
@@ -0,0 +1,85 @@
+"""Loopback-only bridge from an isolated QD container to local Codex app-server.
+
+The bearer secret is ephemeral and supplied through the environment.  The
+bridge never exposes or reads the Codex OAuth credential store.
+"""
+from __future__ import annotations
+
+import importlib.util
+import json
+import os
+import sys
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+
+
+MODULE_PATH = Path(__file__).parents[1] / "app" / "services" / "chatgpt_account.py"
+SPEC = importlib.util.spec_from_file_location("qd_chatgpt_account_bridge_transport", MODULE_PATH)
+assert SPEC and SPEC.loader
+MODULE = importlib.util.module_from_spec(SPEC)
+sys.modules[SPEC.name] = MODULE
+SPEC.loader.exec_module(MODULE)
+Client = MODULE.CodexAppServerClient
+
+
+class Handler(BaseHTTPRequestHandler):
+    server_version = "QuantDingerChatGPTBridge/1"
+
+    def _authorized(self) -> bool:
+        expected = os.environ["CHATGPT_ACCOUNT_BRIDGE_TOKEN"]
+        return self.headers.get("Authorization") == f"Bearer {expected}"
+
+    def _json(self, status: int, payload: dict) -> None:
+        encoded = json.dumps(payload, ensure_ascii=False).encode("utf-8")
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json; charset=utf-8")
+        self.send_header("Content-Length", str(len(encoded)))
+        self.end_headers()
+        self.wfile.write(encoded)
+
+    def do_GET(self) -> None:  # noqa: N802
+        if not self._authorized():
+            self._json(401, {"error": "unauthorized"})
+            return
+        if self.path != "/v1/models":
+            self._json(404, {"error": "not_found"})
+            return
+        models = Client(timeout_seconds=30).list_models()
+        self._json(200, {"models": [{"id": item.slug, "name": item.display_name} for item in models]})
+
+    def do_POST(self) -> None:  # noqa: N802
+        if not self._authorized():
+            self._json(401, {"error": "unauthorized"})
+            return
+        if self.path != "/v1/complete":
+            self._json(404, {"error": "not_found"})
+            return
+        length = int(self.headers.get("Content-Length") or 0)
+        if length <= 0 or length > 256_000:
+            self._json(413, {"error": "invalid_request_size"})
+            return
+        try:
+            payload = json.loads(self.rfile.read(length).decode("utf-8"))
+            messages = payload.get("messages")
+            if not isinstance(messages, list):
+                raise ValueError("messages")
+            text = Client(timeout_seconds=180).complete(messages, model=payload.get("model"))
+            self._json(200, {"text": text})
+        except Exception:
+            self._json(502, {"error": "account_transport_failed"})
+
+    def log_message(self, _format: str, *_args) -> None:
+        return
+
+
+def main() -> None:
+    if not os.getenv("CHATGPT_ACCOUNT_BRIDGE_TOKEN", "").strip():
+        raise SystemExit("CHATGPT_ACCOUNT_BRIDGE_TOKEN is required")
+    os.environ.pop("CHATGPT_ACCOUNT_BRIDGE_URL", None)
+    host = os.getenv("CHATGPT_ACCOUNT_BRIDGE_HOST", "127.0.0.1")
+    port = int(os.getenv("CHATGPT_ACCOUNT_BRIDGE_PORT", "5137"))
+    ThreadingHTTPServer((host, port), Handler).serve_forever()
+
+
+if __name__ == "__main__":
+    main()
diff --git a/backend_api_python/scripts/verify_chatgpt_strategy_pipeline.py b/backend_api_python/scripts/verify_chatgpt_strategy_pipeline.py
new file mode 100644
index 000000000..ed0057710
--- /dev/null
+++ b/backend_api_python/scripts/verify_chatgpt_strategy_pipeline.py
@@ -0,0 +1,150 @@
+"""Two-phase isolated proof for ChatGPT account -> native Strategy V2.
+
+Generate runs on the Windows host with the local Codex login. Backtest runs in
+an ephemeral ``--network none`` QuantDinger image. The JSON handoff is hashed
+and contains no credential material.
+"""
+from __future__ import annotations
+
+import argparse
+import hashlib
+import importlib.util
+import json
+import math
+import re
+import sys
+from pathlib import Path
+
+
+ROOT = Path(__file__).parents[1]
+TEMPLATE_SQL = ROOT / "migrations" / "strategy_v2_templates.sql"
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+
+def sha256_text(value: str) -> str:
+    return hashlib.sha256(value.encode("utf-8")).hexdigest()
+
+
+def load_transport():
+    path = ROOT / "app" / "services" / "chatgpt_account.py"
+    spec = importlib.util.spec_from_file_location("qd_pipeline_chatgpt", path)
+    assert spec and spec.loader
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+def template_code() -> str:
+    sql = TEMPLATE_SQL.read_text(encoding="utf-8")
+    match = re.search(
+        r"\('strategy_v2_double_ma'.*?\$(?P<tag>[a-z]+)\$(?P<code>.*?)\$(?P=tag)\$",
+        sql,
+        re.DOTALL,
+    )
+    if not match:
+        raise RuntimeError("strategy_v2_double_ma template not found")
+    return match.group("code").strip()
+
+
+def strip_fence(value: str) -> str:
+    match = re.search(r"```(?:python)?\s*(.*?)```", value, re.DOTALL | re.IGNORECASE)
+    return (match.group(1) if match else value).strip()
+
+
+def generate(path: Path, model: str) -> None:
+    original = template_code()
+    spec_text = (
+        "Development-only synthetic SMA revision: preserve every Strategy API V2 contract and "
+        "change only the fast_period default from 10 to 12. Return the complete source only."
+    )
+    transport = load_transport()
+    client = transport.CodexAppServerClient(timeout_seconds=180)
+    discussion = client.complete(
+        [{"role": "user", "content": "In one short sentence, confirm you can revise a supplied Strategy API V2 source without using tools."}],
+        model=model,
+    )
+    revised = strip_fence(client.complete(
+        [
+            {"role": "system", "content": "Return Python source only. Do not call tools and do not invent a different API."},
+            {"role": "user", "content": f"# Change specification\n{spec_text}\n\n# Existing verified source\n{original}"},
+        ],
+        model=model,
+    ))
+    if not revised or revised == original:
+        raise RuntimeError("model did not produce a revised candidate")
+    payload = {
+        "task_id": "AQ-QD-AUTH-001",
+        "model": model,
+        "discussion": discussion,
+        "spec": spec_text,
+        "spec_hash": sha256_text(spec_text),
+        "base_code_hash": sha256_text(original),
+        "candidate_code": revised,
+        "candidate_code_hash": sha256_text(revised),
+    }
+    path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
+    print(json.dumps({key: payload[key] for key in ("model", "spec_hash", "base_code_hash", "candidate_code_hash")}))
+
+
+def backtest(path: Path, result_path: Path) -> None:
+    import pandas as pd
+    from app.services.strategy_v2 import StrategyV2BacktestRunner, compile_strategy_v2
+
+    payload = json.loads(path.read_text(encoding="utf-8"))
+    code = str(payload["candidate_code"])
+    if sha256_text(code) != payload["candidate_code_hash"]:
+        raise RuntimeError("candidate hash mismatch")
+    program = compile_strategy_v2(code)
+    index = pd.date_range("2025-01-01", periods=420, freq="4h")
+    closes = [100 + (i * 0.025) + 8.0 * math.sin(i / 9.0) for i in range(len(index))]
+    frame = pd.DataFrame({
+        "open": [x * 0.998 for x in closes],
+        "high": [x * 1.006 for x in closes],
+        "low": [x * 0.994 for x in closes],
+        "close": closes,
+        "volume": [1_000_000.0] * len(index),
+    }, index=index)
+    frames = {instrument.key: frame.copy() for instrument in program.manifest.universe.instruments}
+    result = StrategyV2BacktestRunner(
+        code=code,
+        frames=frames,
+        initial_capital=100_000,
+        commission=0.001,
+        slippage=0.0005,
+    ).run()
+    run_id = f"AQ-QD-AUTH-001-{payload['candidate_code_hash'][:12]}"
+    evidence = {
+        "run_id": run_id,
+        "engine": result.get("engine"),
+        "manifest": result.get("manifest"),
+        "trade_count": len(result.get("trades") or []),
+        "equity_points": len(result.get("equityCurve") or []),
+        "commission": 0.001,
+        "slippage": 0.0005,
+        "network": "none",
+        "persisted_to_production": False,
+        "candidate_code_hash": payload["candidate_code_hash"],
+    }
+    result_path.write_text(json.dumps(evidence, ensure_ascii=False, indent=2), encoding="utf-8")
+    print(json.dumps(evidence, ensure_ascii=False))
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("phase", choices=("generate", "backtest"))
+    parser.add_argument("--artifact", required=True, type=Path)
+    parser.add_argument("--result", type=Path)
+    parser.add_argument("--model", default="gpt-6-luna")
+    args = parser.parse_args()
+    if args.phase == "generate":
+        generate(args.artifact, args.model)
+    else:
+        if not args.result:
+            parser.error("--result is required for backtest")
+        backtest(args.artifact, args.result)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/backend_api_python/tests/isolated/test_chatgpt_account_transport.py b/backend_api_python/tests/isolated/test_chatgpt_account_transport.py
new file mode 100644
index 000000000..86a982e6c
--- /dev/null
+++ b/backend_api_python/tests/isolated/test_chatgpt_account_transport.py
@@ -0,0 +1,105 @@
+from __future__ import annotations
+
+import json
+import importlib.util
+import sys
+from pathlib import Path
+
+import pytest
+
+_MODULE_PATH = Path(__file__).parents[2] / "app" / "services" / "chatgpt_account.py"
+_SPEC = importlib.util.spec_from_file_location("chatgpt_account_under_test", _MODULE_PATH)
+assert _SPEC and _SPEC.loader
+_MODULE = importlib.util.module_from_spec(_SPEC)
+sys.modules[_SPEC.name] = _MODULE
+_SPEC.loader.exec_module(_MODULE)
+ChatGPTAccountError = _MODULE.ChatGPTAccountError
+CodexAppServerClient = _MODULE.CodexAppServerClient
+_message_input = _MODULE._message_input
+
+
+class _Pipe:
+    def __init__(self, lines=()):
+        self.lines = iter(lines)
+        self.writes = []
+
+    def __iter__(self):
+        return self
+
+    def __next__(self):
+        return next(self.lines)
+
+    def write(self, value):
+        self.writes.append(value)
+
+    def flush(self):
+        return None
+
+    def close(self):
+        return None
+
+
+class _Process:
+    def __init__(self, lines):
+        self.stdin = _Pipe()
+        self.stdout = _Pipe(line + "\n" for line in lines)
+        self.stderr = _Pipe()
+        self.returncode = None
+
+    def poll(self):
+        return self.returncode
+
+    def terminate(self):
+        self.returncode = 0
+
+    def wait(self, timeout=None):
+        self.returncode = 0
+        return 0
+
+    def kill(self):
+        self.returncode = -9
+
+
+def test_message_input_preserves_complete_history():
+    rendered = _message_input([
+        {"role": "system", "content": "Return JSON."},
+        {"role": "user", "content": "Build SMA."},
+        {"role": "assistant", "content": "Draft one."},
+        {"role": "user", "content": "Change 20 to 30."},
+    ])
+    assert "[SYSTEM]\nReturn JSON." in rendered
+    assert rendered.endswith("[USER]\nChange 20 to 30.")
+
+
+def test_account_mode_never_uses_api_key_fallback(monkeypatch):
+    monkeypatch.setenv("LLM_AUTH_MODE", "chatgpt_account")
+    assert _MODULE.account_mode_enabled() is True
+
+
+def test_rpc_errors_are_redacted():
+    with pytest.raises(ChatGPTAccountError, match=r"code=-32001") as exc:
+        CodexAppServerClient._raise_rpc_error(
+            {"error": {"code": -32001, "message": "secret bearer token"}},
+            "model discovery",
+        )
+    assert "secret" not in str(exc.value)
+
+
+def test_model_list_uses_only_safe_catalog_fields(monkeypatch):
+    process = _Process([
+        json.dumps({"jsonrpc": "2.0", "id": 1, "result": {}}),
+        json.dumps({
+            "jsonrpc": "2.0",
+            "id": 2,
+            "result": {"data": [{"model": "gpt-test", "displayName": "GPT Test", "token": "never expose"}]},
+        }),
+    ])
+    monkeypatch.setattr("shutil.which", lambda value: __file__)
+    client = CodexAppServerClient(
+        timeout_seconds=5,
+        popen_factory=lambda *args, **kwargs: process,
+    )
+    models = client.list_models()
+    assert [(item.slug, item.display_name) for item in models] == [("gpt-test", "GPT Test")]
+    assert "never expose" not in repr(models)
+
diff --git a/docs/AQ_QD_AUTH_001_IMPLEMENTATION_RESULT.md b/docs/AQ_QD_AUTH_001_IMPLEMENTATION_RESULT.md
new file mode 100644
index 000000000..f8499356f
--- /dev/null
+++ b/docs/AQ_QD_AUTH_001_IMPLEMENTATION_RESULT.md
@@ -0,0 +1,73 @@
+# AQ-QD-AUTH-001 implementation result
+
+## Scope and source basis
+
+- Task: isolated QuantDinger AI Collaborate strategy generation through a local ChatGPT account.
+- QuantDinger base: `462cdf9e9c3131296865678f4f7c782e56cc8672` (`origin/main`).
+- QuantDinger-Vue base: `622b7d0b085dbcd721978a7f1c03b06dbc04edfb` (`origin/main`).
+- AdaptiveQuant governance base: `d4bb3dfb6636a3ec51b416e939649df591baf492` (`origin/main`).
+- Approved design: `AQ_QD_CHATGPT_LOCAL_AI_INTEGRATION_ADR_20261025_CN.md` (isolated development only).
+
+No deployed QuantDinger container, production database, broker configuration, trading permission, RiskGate policy, Hermes installation, or TradingAgents installation was changed.
+
+## Implementation
+
+The existing `LLMService`, Strategy API V2 AI workspace, compilation and native backtest engine are reused.
+
+- `api_key` remains the explicit legacy mode.
+- `chatgpt_account` is an explicit, fail-closed mode and never falls back to an API key.
+- Authentication and refresh remain owned by the official local Codex application. QuantDinger does not read, copy or persist the OAuth credential.
+- The official Codex app-server JSON-RPC interface supplies the account model catalog and inference stream.
+- A loopback bearer-protected bridge is included only for the isolated Docker development topology, because a Linux container cannot safely inherit the Windows Codex credential store.
+- The Vue AI Collaborate workspace exposes the two modes and an account-model selector.
+
+## Runtime evidence
+
+Account discovery on the test machine returned seven account models. A bounded `gpt-6-luna` inference returned the expected fixed marker and did not use an API key fallback.
+
+The automated development-only pipeline performed:
+
+1. model discussion;
+2. model revision of an existing Strategy API V2 source (`fast_period` 10 to 12);
+3. native Strategy V2 compilation;
+4. an ephemeral `--network none` historical backtest with a deterministic synthetic OHLCV fixture;
+5. evidence serialization outside production storage.
+
+Evidence identity:
+
+- SPEC SHA-256: `c351251ab22e35c0dced754bfc6fbb944099eec07f1fd2ac7ce2d961e057033f`
+- base code SHA-256: `ae4c4ae314240a96360251ee85586c284403559caa2e23632ffe179a614ca272`
+- revised candidate SHA-256: `61351905357839e9390e8fdbc6f6148bee621b6e232d7b13ac5f4e9244c9e767`
+- isolated evidence Run ID: `AQ-QD-AUTH-001-613519053578`
+- engine: `quantdinger-strategy-api-v2`
+- result: 13 closed trades, 420 equity points
+- explicit commission: `0.001`
+- explicit slippage: `0.0005`
+- persistence: `false`
+
+This Run ID is a content-derived isolated-proof identifier, not a production QuantDinger database run ID. The synthetic result is not strategy-performance evidence and is not eligible for paper or live promotion.
+
+The browser UI was verified at `http://127.0.0.1:8126/#/strategy-ide?tab=script&draft=1&assetType=script` for authenticated access, explicit mode selection and account model discovery. The first end-to-end UI submission exposed a route wiring defect (`LLMService()` ignored the selected mode); that defect was corrected. The post-fix authenticated browser request then completed at `/api/strategies/ai-workspace/turn` with HTTP 200. The temporary stack was stopped after validation; the URL is therefore the tested development URL, not a permanently deployed service.
+
+## Tests
+
+- Focused account-transport checks: four assertions passed under the repository venv when `TEMP`/`TMP` used the writable `D:\QuanT\.tmp` path.
+- Python compile check: passed for the account transport, `LLMService`, route, bridge and proof script.
+- Vue focused test: the new explicit-auth-mode test passed.
+- Vue full unit command (`node --test tests/unit/*.test.mjs`): 514 passed, 1 unrelated environment failure. The existing `factorLibraryLocalization.test.mjs` hard-codes a missing `D:\QuanT\.worktrees\quantdinger_backend` checkout.
+- Vue production build: passed, 2,397 modules transformed; existing bundle-size warning remains.
+- Repository `npm run test:unit` is incompatible with Node 24 on Windows because it passes a directory instead of a file glob.
+
+## Security boundary and limitations
+
+- No secret, token, account identifier or generated browser JWT is stored in Git or this report.
+- The isolated backtest had no network namespace and no production persistence.
+- The browser development stack is not a production deployment.
+- Starting the complete QD application also started credentialless background loops and generated repeated failed local Kafka connection attempts. No strategy was running and no order was placed. The two task-scoped containers and their temporary network were removed after validation.
+- The account model list proves local account capability, not a universal guarantee that every model invocation is included in every ChatGPT plan. The UI deliberately does not perform API-key fallback.
+- A separate, post-fix browser rerun is still required for full UI acceptance.
+- A production deployment would require an approved credential-boundary design, service lifecycle, audit policy and explicit acceptance; this change does not authorize it.
+
+## Acceptance status
+
+`CONDITIONAL PASS`: native API-mode behavior is preserved, the corrected browser submission returned HTTP 200, and real account model discovery/inference plus the automated no-copy Strategy V2 compile/backtest pipeline are verified. Full production adoption remains blocked on an approved credential-boundary deployment design and a clean background-service profile for the isolated web stack.