From f2ee7e8a1d6d040022485b1d17afb966c47fb017 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Fri, 9 Oct 2026 13:03:17 +0800
Subject: [PATCH] fix(chat): supply host model auth to isolated Codex sessions
Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
.../private-conversation-panel.tsx | 2 +-
.../app-conversation-and-async-inbox-v0.md | 27 ++-
...p-conversation-and-async-inbox-v0.zh-CN.md | 14 +-
loopx/capabilities/native_chat/codex_auth.py | 119 ++++++++++++
loopx/chat_agent.py | 31 ++++
tests/test_chat_codex_auth.py | 173 ++++++++++++++++++
6 files changed, 355 insertions(+), 11 deletions(-)
create mode 100644 loopx/capabilities/native_chat/codex_auth.py
create mode 100644 tests/test_chat_codex_auth.py
diff --git a/apps/presentation/dashboard/src/features/personal-workspace/private-conversation-panel.tsx b/apps/presentation/dashboard/src/features/personal-workspace/private-conversation-panel.tsx
index ef1424b8f8..c97d1d2a01 100644
--- a/apps/presentation/dashboard/src/features/personal-workspace/private-conversation-panel.tsx
+++ b/apps/presentation/dashboard/src/features/personal-workspace/private-conversation-panel.tsx
@@ -123,7 +123,7 @@ export function PrivateConversationPanel() {
{!loadingGroups && !groups.length && !groupError ? {zh ? "此 App 尚无可见群。先将 Bot 加入调试群，再刷新。" : "This App has no visible groups. Add the Bot to a trial group, then refresh."}
: null}
{groupError ? {groupError}
: null}
setGroupRefresh(value => value + 1)}>{zh ? "刷新群列表" : "Refresh groups"}
- {zh ? "新话题需 @此 Bot；回复留在原话题。仅使用选定工作区，不能继承个人管家、全局工具或已有 Agent 会话。需要在独立执行环境登录；连接成功不代表已通过公开群准出。" : "Mention this Bot to start; replies stay in the original topic. Only the selected workspace is available, without personal steward, global tools or existing Agent Sessions. Sign in to the independent execution environment. Connection is not public release qualification."}
+ {zh ? "新话题需 @此 Bot；回复留在原话题。工作区、历史和工具保持隔离，模型认证可复用可信宿主账号。连接成功不代表已通过公开群准出。" : "Mention this Bot to start; replies stay in the original topic. Workspace, history and tools remain isolated; model authentication can use the trusted host account. Connection is not public release qualification."}
: null}
{zh ? "角色" : "Role"} setRole(event.target.value as "project" | "steward")}>
{zh ? "普通项目助手" : "Project assistant"} {zh ? "LoopX 管家（全部已注册工作）" : "LoopX steward (all registered work)"}
diff --git a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md
index caa9eeea2f..90b6e27ece 100644
--- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md
+++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md
@@ -101,14 +101,15 @@ Group contexts always use `workspace_only` filesystem isolation, even when the
owner's private project uses `host_default`. Only the selected read/write workspace
grant is available. Personal portfolio, attached Agent selection, global skills,
MCP, shell profiles and inherited account environment are unavailable. The native
-host must verify the exact permissions profile and workspace root; its independent
-store needs separate login and is never seeded from personal credentials/history.
+host must verify the exact permissions profile and workspace root. Its independent
+store is never seeded from personal credentials/history; model authentication may
+be supplied by the trusted host through the native external-token protocol below.
`/status` exposes the project title, not the host's absolute workspace path.
Core, HTTP, provider-readback and native-protocol fixtures exercise these boundaries.
They are synthetic transport/model evidence, not a live community rollout. The
[community golden queries](../../product/use-cases/community/golden-queries.md)
-still require an independently authenticated public workspace and actual pilot
+still require an isolated public workspace, qualified model authentication and actual pilot
group journeys before either developer group is enabled. Public-source reading
needs its separately qualified scoped tool; this change does not enable unrestricted
network or personal browser access.
@@ -160,10 +161,22 @@ authorized workspace and App/owner binding; topics retain independent threads
without requiring a new login for every message. Default `host_default` Sessions
keep the account's existing native configuration and authentication.
-No authentication or conversation history is copied into the new store. Log in
-through the native Codex flow with `CODEX_HOME` set to the Session's recorded home.
+No authentication or conversation history is copied into the new store. An
+existing project-native login retains its chosen account. Otherwise, when the
+trusted host has native file-based ChatGPT authentication, the adapter supplies
+only its access token and account identifier over private app-server stdio using
+[`chatgptAuthTokens`](https://learn.chatgpt.com/docs/app-server#3c-log-in-with-externally-managed-chatgpt-tokens-chatgptauthtokens).
+This experimental native mode holds tokens in process memory; it does not place
+credentials in the project store, model prompt, tool environment or command line.
+An unauthorized-token callback asks the native host account store to refresh;
+concurrent callbacks reuse an already rotated token. An unavailable host account,
+invalid credential store or account change fails closed with a redacted error.
+Restoring the host account permits a retry without rebinding the conversation.
+Keyring-only and API-key host authentication are not bridged by this adapter.
+Independent login remains available through the native Codex flow with
+`CODEX_HOME` set to the Session's recorded home.
Existing workspace-only Sessions created with a shared home cannot silently
-resume or migrate: choose a new Session explicitly after configuring its login.
+resume or migrate: choose a new Session explicitly with its isolated native home.
Ordinary legacy Sessions keep their existing home and exact-thread resume behavior.
The adapter sends the Core-owned named permissions profile, never a simultaneous
@@ -176,7 +189,7 @@ the canonical Codex executable so a home-directory symlink needs no read grant.
This is a filesystem-tool boundary, not complete community Bot isolation. It does
not authorize group audiences, erase historical context, isolate arbitrary host
dynamic tools or make a checkout containing private files safe to publish. Group
-admission, a clean public workspace, supported independent authentication and live
+admission, a clean public workspace, qualified model authentication and live
privacy/interaction qualification remain required before public enablement; native
context isolation and a successful file probe are prerequisites, not public Bot
acceptance.
diff --git a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md
index 74c23c98d4..6a6cd13ce6 100644
--- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md
+++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md
@@ -74,14 +74,22 @@ listener、队列或模型 runner。一个 App 仍只有一个 binding owner，
群上下文强制使用 `workspace_only`，不受本人私聊 `host_default` 影响。只保留选定
工作区的读写 grant，不提供个人 portfolio、直连 Agent 选择、全局 skills、MCP、
shell profile 或继承的账号环境。原生宿主必须确认确切 permissions profile 和工作区
-根；独立存储需另行登录，不能复制个人凭据或历史。`/status` 只显示项目标题，不展示
+根；独立存储不能复制个人凭据或历史，模型认证可由可信宿主提供。`/status` 只显示项目标题，不展示
宿主绝对工作区路径。
Core、HTTP、provider 读回及原生协议 fixture 验证上述边界，属于合成传输/模型证据。
-[社区黄金查询](../../product/use-cases/community/golden-queries.md) 仍需独立登录的公开
-工作区和真实调试群旅程通过后，才能接入两个正式开发群。公开来源读取另需完成其
+[社区黄金查询](../../product/use-cases/community/golden-queries.md) 仍需隔离的公开工作区、
+模型认证与真实调试群旅程通过后，才能接入两个正式开发群。公开来源读取另需完成其
限定范围工具的验收；本改动不启用无限制网络或个人浏览器。
+已有项目原生登录继续使用其选定账号；否则，可信宿主的原生文件式 ChatGPT 认证
+可通过 Codex 实验性 `chatgptAuthTokens` 接口，仅在私有 stdio 中提供 access token
+和账号标识。短期认证保留在进程内存，不写入项目存储、模型上下文、工具环境或命令行。
+认证失效时，由原生宿主账号存储刷新；并发请求复用已刷新的 token。宿主账号不可用、
+凭据损坏或账号变化时返回脱敏错误，恢复宿主账号后可在原 Session 重试。
+本 adapter 不桥接仅存于 keyring 的认证或 API key；仍可对项目独立执行原生登录。
+该认证路径不扩大工作区、skills、工具或群受众权限，也不代表社区准出已通过。
+
## 普通工作区读写：默认值与撤权检查点
普通项目 Chat 对宿主声明的工作区默认使用 `workspace_write`。Core context owner
diff --git a/loopx/capabilities/native_chat/codex_auth.py b/loopx/capabilities/native_chat/codex_auth.py
new file mode 100644
index 0000000000..adc6b33725
--- /dev/null
+++ b/loopx/capabilities/native_chat/codex_auth.py
@@ -0,0 +1,119 @@
+"""Trusted-host model authentication for an isolated native Codex process.
+
+Only short-lived external tokens cross the private app-server stdio boundary.
+The native account store owns refresh; no credentials, config or history are
+seeded into the project store, tool environment or model context.
+"""
+from __future__ import annotations
+
+import json
+import os
+import queue
+import subprocess
+import threading
+import time
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+
+_locks: dict[Path, threading.Lock] = {}
+_locks_guard = threading.Lock()
+
+
+class CodexHostAuthUnavailable(RuntimeError):
+ def __init__(self) -> None:
+ super().__init__("Trusted-host Codex model authentication is unavailable.")
+
+
+def _native_refresh(codex_bin: str, home: Path, *, timeout_sec: float = 8) -> None:
+ # Account RPCs need no native thread, LoopX Session or model request.
+ # Reuse the native RPC dispatcher and native credential lifecycle rather
+ # than implementing OAuth or keeping a second refresh-token cache.
+ from ...chat_agent import CodexChatAgentSession, _reader
+
+ deadline = time.monotonic() + timeout_sec
+ process = subprocess.Popen(
+ [codex_bin, "app-server", "-c", 'cli_auth_credentials_store="file"',
+ "--listen", "stdio://"],
+ cwd=str(home), env={**os.environ, "CODEX_HOME": str(home)},
+ stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
+ text=True, encoding="utf-8", bufsize=1,
+ )
+ messages: queue.Queue = queue.Queue()
+ session = CodexChatAgentSession(process=process, messages=messages,
+ thread_id="", work_dir=home, response_timeout_sec=timeout_sec)
+ try:
+ threading.Thread(target=_reader, args=(process.stdout, messages), daemon=True).start()
+ session._request("initialize", {"clientInfo": {
+ "name": "loopx_chat", "title": "LoopX Chat", "version": "0.1.0"},
+ "capabilities": {"experimentalApi": True}}, request_id=1)
+ session._notify("initialized", {})
+ session.response_timeout_sec = max(0.01, deadline - time.monotonic())
+ account = session._request("account/read", {"refreshToken": True}, request_id=2)
+ if (account.get("account") or {}).get("type") != "chatgpt":
+ raise CodexHostAuthUnavailable()
+ finally:
+ session.close()
+
+
+@dataclass(repr=False)
+class CodexHostModelAuth:
+ home: Path
+ codex_bin: str
+ _last_token: str | None = field(default=None, repr=False)
+
+ def _read(self) -> dict[str, str]:
+ path = self.home / "auth.json"
+ if path.is_symlink():
+ raise CodexHostAuthUnavailable()
+ data = json.loads(path.read_text(encoding="utf-8"))
+ tokens = data.get("tokens")
+ if data.get("auth_mode") != "chatgpt" or not isinstance(tokens, dict):
+ raise CodexHostAuthUnavailable()
+ access, account = tokens.get("access_token"), tokens.get("account_id")
+ if not all(isinstance(value, str) and value.strip() for value in (access, account)):
+ raise CodexHostAuthUnavailable()
+ # Deliberately exclude ID/refresh tokens, email and all other native data.
+ return {"accessToken": access, "chatgptAccountId": account}
+
+ def read(self, *, refresh: bool = False, previous_account_id: str | None = None) -> dict[str, str]:
+ deadline = time.monotonic() + 8
+ with _locks_guard:
+ lock = _locks.setdefault(self.home.resolve(), threading.Lock())
+ try:
+ if not lock.acquire(timeout=8):
+ raise CodexHostAuthUnavailable()
+ try:
+ current = self._read()
+ if previous_account_id is not None and current["chatgptAccountId"] != previous_account_id:
+ raise CodexHostAuthUnavailable()
+ # Another project or native client may already have refreshed
+ # the same account. Consume its new token instead of rotating
+ # a shared refresh token again for every concurrent callback.
+ if refresh and current["accessToken"] == self._last_token:
+ remaining = deadline - time.monotonic()
+ if remaining <= 0:
+ raise CodexHostAuthUnavailable()
+ _native_refresh(self.codex_bin, self.home, timeout_sec=remaining)
+ current = self._read()
+ if previous_account_id is not None and current["chatgptAccountId"] != previous_account_id:
+ raise CodexHostAuthUnavailable()
+ self._last_token = current["accessToken"]
+ return current
+ finally:
+ lock.release()
+ except Exception:
+ # Native errors and malformed private files must never reach the
+ # Chat transcript, RPC diagnostics or model as exception details.
+ raise CodexHostAuthUnavailable() from None
+
+
+def for_isolated_process(base_home: Path, isolated_home: Path, codex_bin: str) -> CodexHostModelAuth | None:
+ # A separately authenticated project retains its chosen native account.
+ # Shared host auth is a model-only fallback, never a store identity change.
+ if (isolated_home / "auth.json").is_symlink():
+ raise CodexHostAuthUnavailable()
+ if (isolated_home / "auth.json").exists() or not (base_home / "auth.json").exists():
+ return None
+ return CodexHostModelAuth(base_home, codex_bin)
diff --git a/loopx/chat_agent.py b/loopx/chat_agent.py
index 9912ba7796..68050e09ac 100644
--- a/loopx/chat_agent.py
+++ b/loopx/chat_agent.py
@@ -548,6 +548,7 @@ class CodexChatAgentSession:
_message_dispatch_lock: threading.Lock = field(
default_factory=threading.Lock, repr=False
)
+ _host_model_auth: Any = field(default=None, repr=False)
@classmethod
def start(
@@ -716,6 +717,20 @@ def start(
request_id=1,
)
session._notify("initialized", {})
+ if permissions_profile:
+ from .capabilities.native_chat.codex_auth import for_isolated_process
+ try:
+ session._host_model_auth = for_isolated_process(base_home, runtime_home, resolved)
+ if session._host_model_auth is not None:
+ credentials = session._host_model_auth.read()
+ login = session._request("account/login/start", {
+ "type": "chatgptAuthTokens", **credentials}, request_id=4)
+ if login.get("type") != "chatgptAuthTokens":
+ raise ValueError("unexpected native model authentication mode")
+ except Exception:
+ raise session._runtime_error(
+ "Trusted-host Codex model authentication is unavailable. "
+ "Restore the host account and retry this same Session.") from None
read_project_defaults = project_context is not None and bool(resume_thread_id) and (
model is None or reasoning_effort is None
)
@@ -815,6 +830,8 @@ def start(
# for autonomous execution; enabling it here causes conversational messages
# to be treated as continuation ticks instead of the current user task.
session.next_request_id = 4 if read_project_defaults or permissions_profile else 3
+ if session._host_model_auth is not None:
+ session.next_request_id = 5
return session
except _LegacyModelCatalogSchemaError as exc:
session.close()
@@ -913,6 +930,20 @@ def _next_event(self, *, deadline: float) -> dict[str, Any]:
return message
def _check_server_gate(self, message: dict[str, Any]) -> bool:
+ if message.get("id") is not None and message.get("method") == "account/chatgptAuthTokens/refresh" and self._host_model_auth is not None:
+ try:
+ params = message.get("params") or {}
+ if not isinstance(params, dict) or params.get("reason") != "unauthorized":
+ raise ValueError("invalid native refresh request")
+ previous = params.get("previousAccountId")
+ if not isinstance(previous, str) or not previous:
+ raise ValueError("missing native account identity")
+ result = self._host_model_auth.read(refresh=True, previous_account_id=previous)
+ self._write({"id": message["id"], "result": result})
+ except Exception:
+ self._write({"id": message["id"], "error": {
+ "code": -32000, "message": "Trusted-host model authentication unavailable."}})
+ return True
if (
message.get("id") is not None
and message.get("method") == "item/tool/call"
diff --git a/tests/test_chat_codex_auth.py b/tests/test_chat_codex_auth.py
new file mode 100644
index 0000000000..801ed11fd5
--- /dev/null
+++ b/tests/test_chat_codex_auth.py
@@ -0,0 +1,173 @@
+import io
+import json
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+
+import pytest
+
+from loopx import chat_agent
+from loopx.capabilities.native_chat import codex_auth
+from loopx.capabilities.native_chat.project_context import ChatProjectContexts
+
+
+def write_auth(home, access="synthetic-access", account="synthetic-account"):
+ home.mkdir(exist_ok=True)
+ (home / "auth.json").write_text(json.dumps({"auth_mode": "chatgpt",
+ "tokens": {"access_token": access, "account_id": account,
+ "refresh_token": "synthetic-private-refresh", "id_token": "synthetic-private-id"},
+ "unrelated_private_data": "never-forward"}))
+
+
+def test_only_short_lived_model_credentials_cross_the_host_boundary(tmp_path):
+ home, isolated = tmp_path / "account", tmp_path / "project-store"
+ write_auth(home)
+ broker = codex_auth.for_isolated_process(home, isolated, "synthetic-codex")
+ assert broker.read() == {"accessToken": "synthetic-access", "chatgptAccountId": "synthetic-account"}
+ assert "synthetic-access" not in repr(broker)
+ assert not isolated.exists()
+ assert codex_auth.for_isolated_process(tmp_path / "missing", isolated, "synthetic-codex") is None
+ write_auth(isolated, account="separate-project-account")
+ assert codex_auth.for_isolated_process(home, isolated, "synthetic-codex") is None
+
+
+def test_concurrent_refresh_uses_native_account_owner_once(tmp_path, monkeypatch):
+ write_auth(tmp_path)
+ brokers = [codex_auth.CodexHostModelAuth(tmp_path, "synthetic-codex") for _ in range(6)]
+ for broker in brokers:
+ broker.read()
+ calls = []
+ def refresh(binary, home, **kwargs):
+ calls.append((binary, home))
+ write_auth(home, "synthetic-rotated")
+ monkeypatch.setattr(codex_auth, "_native_refresh", refresh)
+ with ThreadPoolExecutor(max_workers=6) as pool:
+ results = list(pool.map(lambda b: b.read(refresh=True, previous_account_id="synthetic-account"), brokers))
+ assert len(calls) == 1
+ assert all(r["accessToken"] == "synthetic-rotated" for r in results)
+
+
+def test_refresh_failure_is_redacted_and_can_retry_without_rebinding(tmp_path, monkeypatch):
+ write_auth(tmp_path)
+ broker = codex_auth.CodexHostModelAuth(tmp_path, "synthetic-codex")
+ broker.read()
+ def fail(*args, **kwargs):
+ raise RuntimeError("synthetic-private-refresh")
+ monkeypatch.setattr(codex_auth, "_native_refresh", fail)
+ with pytest.raises(codex_auth.CodexHostAuthUnavailable) as error:
+ broker.read(refresh=True, previous_account_id="synthetic-account")
+ assert "synthetic-private" not in str(error.value)
+ assert error.value.__suppress_context__
+ monkeypatch.setattr(codex_auth, "_native_refresh", lambda binary, home, **kw: write_auth(home, "synthetic-recovered"))
+ assert broker.read(refresh=True, previous_account_id="synthetic-account")["accessToken"] == "synthetic-recovered"
+ write_auth(tmp_path, "different-token", "different-account")
+ with pytest.raises(codex_auth.CodexHostAuthUnavailable):
+ broker.read(refresh=True, previous_account_id="synthetic-account")
+
+
+@pytest.mark.parametrize("value", [None, {}, {"auth_mode": "apikey"},
+ {"auth_mode": "chatgpt", "tokens": {}},
+ {"auth_mode": "chatgpt", "tokens": {"access_token": False, "account_id": "account"}}])
+def test_invalid_host_auth_never_uses_wider_or_fallback_credentials(tmp_path, value):
+ (tmp_path / "auth.json").write_text(json.dumps(value))
+ with pytest.raises(codex_auth.CodexHostAuthUnavailable):
+ codex_auth.CodexHostModelAuth(tmp_path, "synthetic-codex").read()
+
+
+def test_host_auth_symlink_is_rejected(tmp_path):
+ other = tmp_path / "other"
+ write_auth(other)
+ (tmp_path / "auth.json").symlink_to(other / "auth.json")
+ with pytest.raises(codex_auth.CodexHostAuthUnavailable):
+ codex_auth.CodexHostModelAuth(tmp_path, "synthetic-codex").read()
+
+
+def test_broken_project_auth_symlink_cannot_switch_to_host_account(tmp_path):
+ home, isolated = tmp_path / "host", tmp_path / "project"
+ write_auth(home)
+ isolated.mkdir()
+ (isolated / "auth.json").symlink_to(tmp_path / "missing-private-file")
+ with pytest.raises(codex_auth.CodexHostAuthUnavailable):
+ codex_auth.for_isolated_process(home, isolated, "synthetic-codex")
+
+
+def test_native_callback_refresh_recovers_but_preserves_account_and_request_gates(tmp_path, monkeypatch):
+ from tests.test_chat_agent import _FakeAppServerProcess
+ import queue
+
+ write_auth(tmp_path)
+ broker = codex_auth.CodexHostModelAuth(tmp_path, "synthetic-codex")
+ broker.read()
+ monkeypatch.setattr(codex_auth, "_native_refresh", lambda binary, home, **kw: write_auth(home, "synthetic-rotated"))
+ process = _FakeAppServerProcess()
+ session = chat_agent.CodexChatAgentSession(process=process, messages=queue.Queue(),
+ thread_id="same-thread", work_dir=tmp_path, _host_model_auth=broker)
+ request = {"id": 9, "method": "account/chatgptAuthTokens/refresh",
+ "params": {"reason": "unauthorized", "previousAccountId": "synthetic-account"}}
+ assert session._check_server_gate(request)
+ assert json.loads(process.stdin.getvalue().splitlines()[-1])["result"] == {
+ "accessToken": "synthetic-rotated", "chatgptAccountId": "synthetic-account"}
+ assert session.thread_id == "same-thread"
+ for params in ([], {"reason": "other", "previousAccountId": "synthetic-account"},
+ {"reason": "unauthorized"}):
+ assert session._check_server_gate({**request, "params": params})
+ assert "error" in json.loads(process.stdin.getvalue().splitlines()[-1])
+ with pytest.raises(chat_agent.CodexChatAgentError):
+ session._check_server_gate({"id": 10, "method": "item/commandExecution/requestApproval", "params": {}})
+
+
+def test_native_refresh_never_opens_a_thread_or_passes_credentials_in_argv(tmp_path, monkeypatch):
+ from tests.test_chat_agent import _FakeAppServerProcess
+ process = _FakeAppServerProcess(thread_response={"account": {"type": "chatgpt"}})
+ launches = []
+ monkeypatch.setattr(codex_auth.subprocess, "Popen", lambda command, **kw: launches.append((command, kw)) or process)
+ codex_auth._native_refresh("synthetic-codex", tmp_path)
+ packets = [json.loads(line) for line in process.stdin.getvalue().splitlines()]
+ assert [p["method"] for p in packets] == ["initialize", "initialized", "account/read"]
+ assert packets[-1]["params"] == {"refreshToken": True}
+ assert launches[0][1]["env"]["CODEX_HOME"] == str(tmp_path)
+ assert "synthetic-access" not in str(launches[0][0])
+ assert process.poll() == 0
+
+
+@pytest.mark.parametrize("resume", [False, True])
+def test_isolated_start_and_resume_supply_auth_only_over_private_rpc(tmp_path, monkeypatch, resume):
+ from tests.test_chat_agent import _FakeAppServerProcess
+ home = tmp_path / "host"
+ write_auth(home)
+ workspace = tmp_path / "workspace"
+ workspace.mkdir()
+ context = ChatProjectContexts([workspace], filesystem_scope="workspace_only").available()[0]
+ profile = "loopx_workspace_only_write"
+ process = _FakeAppServerProcess(config_response={"config": {}}, thread_response={
+ "thread": {"id": "thread-loopx-chat"}, "activePermissionProfile": {"id": profile},
+ "runtimeWorkspaceRoots": [str(workspace)]})
+ process.stdout = io.StringIO(json.dumps({"id": 1, "result": {}}) + "\n" +
+ json.dumps({"id": 4, "result": {"type": "chatgptAuthTokens"}}) + "\n" + process.stdout.getvalue().split("\n", 1)[1])
+ real_which = chat_agent.shutil.which
+ monkeypatch.setattr(chat_agent.shutil, "which", lambda name: str(tmp_path / "native-codex") if name == "synthetic-codex" else real_which(name))
+ real_popen = chat_agent.subprocess.Popen
+ launches = []
+ def launch(command, **kwargs):
+ if command[0] != str(tmp_path / "native-codex"):
+ return real_popen(command, **kwargs)
+ launches.append((command, kwargs))
+ return process
+ monkeypatch.setattr(chat_agent.subprocess, "Popen", launch)
+ session = chat_agent.CodexChatAgentSession.start(codex_bin="synthetic-codex", work_dir=workspace,
+ goal_id=None, objective="project", project_context=context, codex_home=home,
+ resume_thread_id="thread-loopx-chat" if resume else None, model="synthetic-model")
+ try:
+ packets = [json.loads(line) for line in process.stdin.getvalue().splitlines()]
+ login = next(p for p in packets if p["method"] == "account/login/start")
+ assert login["params"] == {"type": "chatgptAuthTokens", "accessToken": "synthetic-access", "chatgptAccountId": "synthetic-account"}
+ thread = next(p for p in packets if p["method"] in {"thread/start", "thread/resume"})
+ assert thread["params"]["permissions"] == profile
+ assert "synthetic-access" not in str(thread)
+ assert "synthetic-access" not in str(launches)
+ isolated = Path(launches[0][1]["env"]["CODEX_HOME"])
+ assert isolated != home and not (isolated / "auth.json").exists()
+ assert session.next_request_id == 5
+ session._check_server_gate({"id": 9, "method": "account/chatgptAuthTokens/refresh", "params": {"reason": "unauthorized", "previousAccountId": "wrong-account"}})
+ assert json.loads(process.stdin.getvalue().splitlines()[-1])["error"]["message"] == "Trusted-host model authentication unavailable."
+ finally:
+ session.close()