diff --git a/apps/presentation/dashboard/src/features/personal-workspace/codex-account-settings.css b/apps/presentation/dashboard/src/features/personal-workspace/codex-account-settings.css new file mode 100644 index 0000000000..1550f76734 --- /dev/null +++ b/apps/presentation/dashboard/src/features/personal-workspace/codex-account-settings.css @@ -0,0 +1,29 @@ +.personal-codex-accounts { display: grid; gap: 20px; padding: 8px 0 24px; color: var(--pw-text); } +.personal-codex-accounts > header { display: flex; flex-wrap: wrap; align-items: baseline; justify-content: space-between; gap: 12px; } +.personal-codex-accounts h2 { margin: 0; font-size: var(--text-heading); line-height: var(--leading-heading); font-weight: 600; } +.personal-codex-accounts p { margin: 0; line-height: var(--leading-prose); } +.personal-codex-accounts header > span, .codex-account-effect, .personal-codex-accounts details, .codex-account-actions > span { color: var(--pw-muted); font-size: var(--text-caption); line-height: var(--leading-caption); } +.codex-account-enable { display: flex; align-items: center; gap: 12px; min-height: 44px; padding: 12px 0; border-top: 1px solid var(--pw-line); border-bottom: 1px solid var(--pw-line); } +.codex-account-enable input { accent-color: var(--pw-text); width: 18px; height: 18px; } +.codex-account-state { display: flex; gap: 8px; align-items: center; font-size: var(--text-body); } +.codex-account-list { display: grid; grid-template-columns: repeat(3, minmax(0, 1fr)); gap: 24px; } +.codex-account-list article { display: flex; flex-direction: column; align-items: flex-start; gap: 16px; padding-top: 16px; border-top: 2px solid var(--pw-line); min-width: 0; } +.codex-account-list article.is-current { border-color: var(--pw-text); } +.codex-account-name { display: flex; align-items: baseline; flex-wrap: wrap; gap: 8px; } +.codex-account-name > span, .codex-account-list article > span { font-size: var(--text-caption); color: var(--pw-muted); } +.codex-account-list dl { display: grid; gap: 12px; margin: 0; } +.codex-account-list dt { color: var(--pw-muted); font-size: var(--text-caption); } +.codex-account-list dd { margin: 4px 0 0; font-size: var(--text-heading); font-variant-numeric: tabular-nums; } +.personal-codex-accounts button { display: inline-flex; align-items: center; justify-content: center; gap: 8px; min-height: 44px; padding: 8px 12px; border: 1px solid var(--pw-line-strong); border-radius: 6px; background: var(--pw-card); color: var(--pw-text); font-size: var(--text-body); cursor: pointer; } +.personal-codex-accounts button:disabled { opacity: .5; cursor: default; } +.personal-codex-accounts button:focus-visible, .personal-codex-accounts input:focus-visible, .personal-codex-accounts summary:focus-visible { outline: 2px solid var(--pw-blue); outline-offset: 3px; } +.codex-account-actions { display: flex; flex-wrap: wrap; align-items: center; gap: 12px; } +.personal-codex-accounts details { border-top: 1px solid var(--pw-line); padding-top: 16px; } +.personal-codex-accounts summary { cursor: pointer; padding: 4px 0; } +.personal-codex-accounts details p { margin: 12px 0; } +.personal-codex-accounts [role="alert"] { display: flex; gap: 8px; align-items: flex-start; color: var(--pw-red); } +@media (max-width: 640px) { + .codex-account-list { grid-template-columns: 1fr; gap: 20px; } + .codex-account-list dl { grid-template-columns: 1fr 1fr; width: 100%; } + .codex-account-list article > button { width: 100%; } +} diff --git a/apps/presentation/dashboard/src/features/personal-workspace/codex-account-settings.tsx b/apps/presentation/dashboard/src/features/personal-workspace/codex-account-settings.tsx new file mode 100644 index 0000000000..743b781f6a --- /dev/null +++ b/apps/presentation/dashboard/src/features/personal-workspace/codex-account-settings.tsx @@ -0,0 +1,161 @@ +import { useCallback, useEffect, useState } from "react"; +import { RefreshCw, CheckCircle2, Circle, AlertCircle } from "lucide-react"; +import { requestJson } from "../../data/chat"; +import { useWorkspaceI18n } from "./i18n"; +import "./codex-account-settings.css"; + +type AccountStatus = { + ok: boolean; available: boolean; code?: string | null; revision?: number; + active_slot?: string | null; selection_revision?: string; action?: string; + target?: string | null; recovery_required?: boolean; launch_failed?: boolean; + policy?: { enabled: boolean; slots: string[]; cooldown_seconds: number; restart_clients?: boolean }; + slots?: string[]; last_checked_at?: number | null; last_switch_at?: number | null; + accounts?: Array<{ slot: string; availability: string; observed_at: number; + primary_used: number | null; secondary_used: number | null; reset_at: number | null; + credits_available: boolean | null }>; +}; + +const copy = { + "zh-CN": { + title: "Codex 原生账号", enabled: "自动接续已启用", disabled: "自动接续已关闭", + enable: "额度耗尽后自动切换", refresh: "检查额度", retry: "重新连接", + effect: "仅使用下方已登记账号。额度耗尽后正常退出 GPT 专用客户端，保存登录态并切换，再重新打开原来的界面。", + effectWait: "仅使用下方已登记账号。额度耗尽后等待客户端关闭，再保存登录态、切换账号并重新打开。", + restart: "允许自动退出并重开 GPT 客户端", restarting: "正在退出 GPT、切换账号并重新打开", + reopen: "重新打开 GPT", other_clients_busy: "还有其他进程占用 GPT 登录；关闭占用后检查额度重试", + client_quit_refused: "GPT 未接受正常退出；本次未换号，自动尝试已暂停", client_quit_timeout: "GPT 尚未退出；本次未换号，自动尝试已暂停", + client_changed: "GPT 进程已变化；已暂停本次切换", handoff_pending: "独立切换进程正在完成接续，请稍候", + handoff_interrupted: "接续进程中断；请检查并恢复本次切换", quota_stale: "额度记录已过期，正在重新检查", + current: "当前", next: "下一账号", slot: "账号", primary: "短周期剩余", secondary: "长周期剩余", unread: "未读取", + select: "切换到此账号", checking: "正在检查额度", ready: "当前账号可用", selected: "账号已切换", + waiting_for_idle: "已选好候选账号，等待 Codex 客户端关闭", selection_changed: "账号选择已变化，已暂停本次切换", + auth_required: "登录需要更新，请在原生 Codex 完成登录", quota_unknown: "暂时无法确认额度，将稍后重试", + cooldown: "切换后冷却中", all_unavailable: "暂时没有可用账号，等待额度恢复", target_unavailable: "所选账号当前不可用", + recovery_required: "上次切换未完成，需要恢复", unavailable: "原生账号暂不可用", adapter_changed: "槽位管理器已更新，请重新连接", + recover: "恢复未完成的切换", rollback: "撤销最近一次切换", available: "可用", exhausted: "额度已用完", + unknown: "额度未知", credits: "另有可用额度", help: "运行与恢复说明", last: "上次检查", + detail: "检测不会调用模型、购买额度或使用重置券。401 登录失效先由原生客户端刷新；其他账号可通过官方 CLI 单独刷新。查询失败不会被判为额度耗尽。自动退出仅针对 GPT 专用实例，拒绝退出或其他 CLI/文件占用会暂停尝试，不强杀进程。独立进程完成切换和重开，即使发起切换的 Codex 已退出。切换保留模型设置和会话历史，不重放已提交工作。自动观察随 LoopX 服务运行。", + extension_uninstalled: "尚未安装 Codex 账号扩展。安装 loopx-codex-provider-routing 后重新启动 LoopX 服务。", + adapter_unavailable: "未发现已安装的 Personal Apps 槽位管理器和 Codex CLI。", + native_unavailable: "无法连接原生槽位管理器，请检查本机安装后重试。", + launchFailed: "账号已切换，但界面未能重新打开。可以从原来的 GPT 入口打开；请勿重复切换。", + failure: "操作未完成。刷新状态后可以重试。", recoveryHint: "恢复会还原未完成事务；撤销会还原最近一次切换。两者都需要客户端关闭，且不会覆盖后续登录修改。", + }, + en: { + title: "Native Codex accounts", enabled: "Automatic handoff enabled", disabled: "Automatic handoff off", + enable: "Switch automatically when quota runs out", refresh: "Check quota", retry: "Reconnect", + effect: "Use only the registered accounts below. After exhaustion, quit the dedicated GPT client normally, save the login, switch accounts, then reopen the same interface.", + effectWait: "Use only the registered accounts below. After exhaustion, wait for clients to close, save the login, switch accounts, then reopen the same interface.", + restart: "Allow automatic GPT client quit and restart", restarting: "Quitting GPT, switching accounts and reopening", + reopen: "Reopen GPT", other_clients_busy: "Another process still owns the GPT login; close it and check quota to retry", + client_quit_refused: "GPT refused normal quit; no account changed and automatic retries are paused", client_quit_timeout: "GPT has not exited; no account changed and automatic retries are paused", + client_changed: "The GPT process changed; this switch was paused", handoff_pending: "The independent handoff is still finishing; please wait", + handoff_interrupted: "The handoff process stopped; inspect and recover this switch", quota_stale: "Quota observations expired; checking again", + current: "Current", next: "Next account", slot: "Account", primary: "Short window left", secondary: "Long window left", unread: "Not read", + select: "Switch to this account", checking: "Checking quota", ready: "Current account is available", selected: "Account switched", + waiting_for_idle: "Candidate selected; waiting for Codex clients to close", selection_changed: "Account selection changed; this switch was paused", + auth_required: "Sign in again in native Codex", quota_unknown: "Quota could not be confirmed; retrying later", + cooldown: "Cooling down after switching", all_unavailable: "No account is available yet; waiting for quota to recover", target_unavailable: "Selected account is unavailable", + recovery_required: "An interrupted switch needs recovery", unavailable: "Native accounts unavailable", adapter_changed: "The slot manager changed; reconnect to it", + recover: "Recover interrupted switch", rollback: "Undo last switch", available: "Available", exhausted: "Quota exhausted", + unknown: "Quota unknown", credits: "Additional allowance available", help: "Operation and recovery", last: "Last checked", + detail: "Checks do not call a model, buy credits or use reset coupons. Native Codex refreshes an expired active login; other accounts can refresh separately through the official CLI. A failed check never means exhausted quota. Normal quit targets only the dedicated GPT instance. A refused quit or another CLI/file owner pauses attempts; processes are never force-killed. An independent process completes switching and reopening even when the calling Codex exits. Model settings and history stay intact; submitted work is never replayed. Observation runs with the LoopX service.", + extension_uninstalled: "Install the optional loopx-codex-provider-routing package, then restart the LoopX service.", + adapter_unavailable: "The installed Personal Apps slot manager and Codex CLI were not found.", + native_unavailable: "Cannot connect to the native slot manager. Check the local installation and retry.", + launchFailed: "The account switched, but the interface could not reopen. Open the original GPT entry; do not repeat the switch.", + failure: "The operation did not finish. Refresh its status before retrying.", recoveryHint: "Recovery restores an interrupted transaction; undo restores the last switch. Both require closed clients and refuse to overwrite later login changes.", + }, +}; + +export function CodexAccountSettings() { + const { locale } = useWorkspaceI18n(); + const c = copy[locale]; + const [status, setStatus] = useState(null); + const [busy, setBusy] = useState(false); + const [error, setError] = useState(false); + const endpoint = "/api/chat/codex-accounts"; + const label = (key: string | null | undefined) => c[key as keyof typeof c] ?? c.unavailable; + const read = useCallback(async () => { + try { + const result = await requestJson(endpoint); + setStatus(result); + setError(false); + } catch { setError(true); } + }, []); + useEffect(() => { + void read(); + const timer = window.setInterval(() => { if (!document.hidden) void read(); }, 5000); + return () => window.clearInterval(timer); + }, [read]); + async function act(operation: string, parameters = {}) { + setBusy(true); + setError(false); + try { + const result = await requestJson(endpoint, { + method: "POST", body: JSON.stringify({ operation, parameters }), + }); + setStatus(result); + } catch { await read(); setError(true); } + finally { setBusy(false); } + } + const enabled = status?.policy?.enabled === true; + const restartClients = status?.policy?.restart_clients === true; + const problem = status?.code || (status?.ok === false ? "native_unavailable" : null); + return
+

{c.title}

{status?.active_slot ? `${c.current} ${status.active_slot.toUpperCase()} · ` : ""}{enabled ? c.enabled : c.disabled}
+ {!status ?

{c.checking}

: null} + {(problem && problem !== status?.action) || status?.available === false ?

{label(problem)}

: null} + {status?.available && status.slots?.length ? <> +

{restartClients ? c.effect : c.effectWait}

+ + +
+ {status.action === "ready" || status.action === "selected" ? + : } + {label(status.recovery_required ? "recovery_required" : status.action)} +
+
+ {status.slots.map(slot => { + const row = status.accounts?.find(a => a.slot === slot); + // Expired observations remain unknown in the UI, just as in policy. + const fresh = row && Date.now() / 1000 - row.observed_at <= 120; + const current = status.active_slot === slot; + const remaining = (used: number | null | undefined) => fresh && used != null ? `${100 - used}%` : "—"; + return
+
{c.slot} {slot.toUpperCase()} + {current ? c.current : status.target === slot ? c.next : label(row?.availability === "auth_required" ? "auth_required" : fresh ? row?.availability : "unknown")}
+
{c.primary}
{remaining(row?.primary_used)}
+
{c.secondary}
{remaining(row?.secondary_used)}
+ {fresh && row.credits_available ? {c.credits} : null} + {!fresh ? {c.unread} : null} + {!current ? : null} +
; + })} +
+ {status.launch_failed ? <>

{c.launchFailed}

: null} +
+ {status.last_checked_at ? {c.last} {new Date(status.last_checked_at * 1000).toLocaleTimeString(locale)} : null} +
+
{c.help}

{c.detail}

{c.recoveryHint}

+
+
+
+ : null} + {error ?

{c.failure}

: null} + {status?.code === "adapter_changed" || error ? : null} +
; +} diff --git a/apps/presentation/dashboard/src/features/personal-workspace/workspace-settings-page.tsx b/apps/presentation/dashboard/src/features/personal-workspace/workspace-settings-page.tsx index f18ae73b3d..b46a65ba8f 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/workspace-settings-page.tsx +++ b/apps/presentation/dashboard/src/features/personal-workspace/workspace-settings-page.tsx @@ -15,6 +15,7 @@ import { AutomationCadenceSettings } from "./automation-cadence-settings"; import { MachineConfigurationSettings } from "./machine-configuration-settings"; import { ConfigurationBackupSettings } from "./configuration-backup-settings"; import { OperatorCredentialSettings } from "./operator-credential-settings"; +import { CodexAccountSettings } from "./codex-account-settings"; import type { PersonalWorkspaceCallbacks, WorkspaceGoal, WorkspaceGoalNotification } from "./personal-workspace-model"; import type { WorkspaceTheme } from "./workspace-theme"; import type { AttentionNotificationSupport } from "./attention-notifications"; @@ -227,6 +228,7 @@ export function WorkspaceSettingsPage({ {tab === "provider" ? (
+
) : null} diff --git a/loopx/chat_completed_todos.py b/loopx/chat_completed_todos.py index 5bcf20f468..335ab14f09 100644 --- a/loopx/chat_completed_todos.py +++ b/loopx/chat_completed_todos.py @@ -81,7 +81,7 @@ class CompletedTodoPages: ttl_seconds = 300 max_cache_bytes = 16 * 1024 * 1024 - def __init__(self): + def __init__(self) -> None: self._snapshots = OrderedDict() self._lock = Lock() diff --git a/loopx/chat_loopx_mode.py b/loopx/chat_loopx_mode.py index acc59af79a..bd1fe7fc3d 100644 --- a/loopx/chat_loopx_mode.py +++ b/loopx/chat_loopx_mode.py @@ -1138,7 +1138,8 @@ def pump_delegation_wakes( class DelegationWakeService: """Cheap local pump hosted by the existing Chat server, beside the return service.""" - def __init__(self, controller, *, goal_context=None, interval=WAKE_INTERVAL_SECONDS): + def __init__(self, controller: Any, *, goal_context: Callable[[dict[str, Any]], dict[str, Any]] | None = None, + interval: float = WAKE_INTERVAL_SECONDS) -> None: self.controller = controller self.goal_context = goal_context self.interval = interval @@ -1147,7 +1148,7 @@ def __init__(self, controller, *, goal_context=None, interval=WAKE_INTERVAL_SECO target=self.run, daemon=True, name="loopx-delegation-wakes" ) - def start(self): + def start(self) -> DelegationWakeService: self.thread.start() return self @@ -1163,6 +1164,6 @@ def run(self): _LOG.warning("Delegation wake receipts unavailable; retrying") self.stop.wait(self.interval) - def close(self): + def close(self) -> None: self.stop.set() self.thread.join(timeout=3) diff --git a/loopx/chat_operator_provider_api.py b/loopx/chat_operator_provider_api.py index 61dae4a837..f862c4d957 100644 --- a/loopx/chat_operator_provider_api.py +++ b/loopx/chat_operator_provider_api.py @@ -37,7 +37,7 @@ class OperatorProviderRequestMixin: server: _OperatorProviderServer - def _send_json(self, payload: dict, status: int = 200) -> None: ... + def _send_json(self, payload: dict, *, status: int = 200) -> None: ... def _send_error( self, message: str, *, status: int, error_code: str diff --git a/loopx/chat_server.py b/loopx/chat_server.py index fcb71dcd47..02432b3763 100644 --- a/loopx/chat_server.py +++ b/loopx/chat_server.py @@ -11,10 +11,13 @@ import webbrowser from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path -from typing import Any, Literal +from typing import TYPE_CHECKING, Any, Literal, cast from urllib.parse import parse_qs, urlparse from .presentation import configuration_api as config_api +from .presentation.codex_accounts_api import CodexAccountService +if TYPE_CHECKING: + from .capabilities.native_chat.transports import ChatConversationTransports from .attached_session_api import AttachedSessionRequestMixin from .zcode_goal_mode.api import ZCodeGoalRequestMixin from .chat import ( @@ -41,7 +44,6 @@ from .chat_status_api import ChatStatusRequestMixin from .chat_runtime import ( STEERING_NOT_DELIVERED_CODES, - TERMINAL_TURN_STATES, ChatRuntimeController, ChatTurnAcceptanceUnavailableError, ) @@ -51,12 +53,12 @@ ) from .chat_session_open import open_chat_session from .chat_ssh_source_api import SshSourceRequestMixin -from .chat_store import ChatSessionStore, normalize_agent_model_id -from .capabilities.native_chat.model_settings import normalize_agent_reasoning_effort +from .chat_store import ChatSessionStore, TERMINAL_TURN_STATES +from .capabilities.native_chat.model_settings import normalize_agent_reasoning_effort, normalize_agent_model_id from .extensions.lark.private_conversation_api import PrivateConversationRequestMixin, PRIVATE_CONVERSATIONS_PATH from .extensions.lark.conversation_identity import observe_lark_conversation_identity from .extensions.lark.private_conversations import LarkPrivateConversations -from .chat_loopx_mode import handle_loopx_request +from .chat_loopx_mode import DelegationWakeService, handle_loopx_request from .capabilities.manager_context.roundtrip import ( ReturnService, project_chat_session_snapshot, @@ -89,14 +91,13 @@ setup_lark_goal_channel, ) from .extensions.lark.goal_topic_connections import list_lark_apps -from .extensions.lark.goal_topic_runtime import LarkGoalTopicRuntimeService +from .extensions.lark.goal_topic_runtime_service import LarkGoalTopicRuntimeService from .extensions.lark.manager_routing import authorized_manager_goal_ids from .extensions.lark.cli_transport import CommandRunner from .extensions.runtime import ( default_extension_state_file, resolve_extension_activation, ) -from .history import load_registry from .chat_completed_todos import CompletedTodoPages, CompletedTodoRequestMixin from .presentation.explore_results_api import ExploreResultsRequestMixin from .presentation.research_analysis_api import RESEARCH_ANALYSIS_PATH, ResearchAnalysisRequestMixin @@ -104,6 +105,7 @@ from .kiro_cli_goal_mode import KIRO_CLI_BIN from .paths import resolve_runtime_root from .release_manifest import release_runtime_identity +from .control_plane.projects.registry_codec import load_registry from .registry import registry_goals, resolve_state_file from .state_projection import build_active_state_structured_projection from .status_server import ( @@ -259,7 +261,7 @@ def build_chat_status_projection( queue_items = queue.get("items") if isinstance(queue, dict) else [] queue_by_goal = { str(item.get("goal_id") or ""): item - for item in queue_items if isinstance(queue_items, list) and isinstance(item, dict) + for item in (queue_items if isinstance(queue_items, list) else []) if isinstance(item, dict) } goals: list[dict[str, Any]] = [] for raw_goal in registry_goals(registry): @@ -285,7 +287,8 @@ def build_chat_status_projection( latest_validation = asset.get("latest_validation") if isinstance(latest_validation, dict) and latest_validation.get("summary"): evidence.insert(0, _compact_text(latest_validation.get("summary"))) - quota = item.get("quota") if isinstance(item, dict) and isinstance(item.get("quota"), dict) else {} + raw_quota = item.get("quota") if isinstance(item, dict) else None + quota = raw_quota if isinstance(raw_quota, dict) else {} goals.append( { "goal_id": goal_id, @@ -441,7 +444,9 @@ class ChatHTTPServer(ThreadingHTTPServer): allow_reuse_address = True registry_path: Path + runtime_root: Path runtime_root_override: str | None + codex_account_service: CodexAccountService scan_roots: list[Path] limit: int selected_goal_id: str | None @@ -460,6 +465,8 @@ class ChatHTTPServer(ThreadingHTTPServer): lark_private_conversations: LarkPrivateConversations ssh_config_path: Path | None goal_subagent_configuration_enabled: bool + conversation_transports: ChatConversationTransports + delegation_wake_service: DelegationWakeService def __init__(self, *args: Any, **kwargs: Any) -> None: # Freeze before serving: an in-place package upgrade must not retag the @@ -469,6 +476,8 @@ def __init__(self, *args: Any, **kwargs: Any) -> None: self.completed_todo_pages = CompletedTodoPages() def server_close(self) -> None: + if hasattr(self, "codex_account_service"): + self.codex_account_service.close() if hasattr(self, "lark_app_setup_manager"): self.lark_app_setup_manager.close() if hasattr(self, "manager_return_service"): @@ -538,7 +547,12 @@ def _send_error( turn_replay_safe: bool = False, todo_receipt: dict[str, Any] | None = None, delivery_state: Literal["not_delivered", "unresolved"] | None = None, + **unsupported: object, ) -> None: + # Legacy mixins expose **kwargs; keep unknown options rejected at the + # concrete boundary instead of silently widening the HTTP contract. + if unsupported: + raise TypeError("unsupported error response options") payload: dict[str, Any] = {"ok": False, "error": _compact_text(message)} if error_code: payload["error_code"] = _compact_text(error_code, limit=80) @@ -630,7 +644,7 @@ def _create_session(self) -> None: unknown = set(body) - {"goal_id", "agent_id", "agent_model_id", "agent_reasoning_effort", "mode", "context_kind", "project_ref"} if unknown: raise ValueError("unknown session field") - goal_id = _compact_text(body.get("goal_id"), limit=160) or self.server.selected_goal_id or "" + goal_id: str | None = _compact_text(body.get("goal_id"), limit=160) or self.server.selected_goal_id or "" # ``agent_id`` is the caller's explicit executor pick and stays empty # when the caller does not make one. Each channel then resolves its # own default through its own owner instead of inheriting whatever @@ -658,10 +672,10 @@ def _create_session(self) -> None: context = self._session_context({"goal_id": None, "channel_id": selected["channel_id"], "project_context": selected["context"]}) else: - registry, goal = self._registry_and_goal(goal_id) + registry, goal = self._registry_and_goal(goal_id or "") context = _goal_public_context(registry, goal) runtime_objective = str(context["objective"] or context["title"]) - project = context["project"] + project = cast(Path, context["project"]) if not project.is_dir(): raise CodexChatAgentError( "The Goal project root is unavailable.", @@ -819,17 +833,18 @@ def _session_turn(self, session_id: str) -> None: self._send_error(str(exc), turn_replay_safe=True, delivery_state="not_delivered") return try: - submit = (self.server.runtime_controller.enqueue_turn if body.get("delivery") == "queue" - else self.server.runtime_controller.submit_turn) - turn, created = submit( - session_id=session_id, - client_turn_id=client_turn_id, - message=message, - attachments=attachments, - work_dir=context["project"], - objective=runtime_objective, - **({"origin": "web"} if body.get("delivery") == "queue" else {}), - ) + if body.get("delivery") == "queue": + turn, created = self.server.runtime_controller.enqueue_turn( + session_id=session_id, client_turn_id=client_turn_id, + message=message, attachments=attachments, + work_dir=cast(Path, context["project"]), objective=runtime_objective, origin="web", + ) + else: + turn, created = self.server.runtime_controller.submit_turn( + session_id=session_id, client_turn_id=client_turn_id, + message=message, attachments=attachments, + work_dir=cast(Path, context["project"]), objective=runtime_objective, + ) if body.get("client_turn_id"): turn_id = str(turn["turn_id"]) self._send_json( @@ -910,7 +925,8 @@ def _session_turn(self, session_id: str) -> None: def _session_snapshot(self, session_id: str) -> None: try: self.server.runtime_controller.recover_exited_turn(session_id) - self._send_json(project_chat_session_snapshot( + snapshot = cast(Callable[..., dict[str, object]], project_chat_session_snapshot) + self._send_json(snapshot( self.server.runtime_root, self.server.chat_store, session_id, registry=self.server.registry_path)) except KeyError: @@ -943,13 +959,13 @@ def _turn_events(self, session_id: str, turn_id: str) -> None: last_event_id = self.headers.get("Last-Event-ID") query = parse_qs(urlparse(self.path).query) if not last_event_id: - last_event_id = (query.get("after") or [None])[0] + last_event_id = (query.get("after") or [""])[0] or None if str(last_event_id or "0") != "0": - turn = self.server.chat_store.load_turn(session_id, turn_id) or {} + prior_turn = self.server.chat_store.load_turn(session_id, turn_id) or {} self.server.chat_store.update_turn( session_id, turn_id, - sse_reconnect_count=int(turn.get("sse_reconnect_count") or 0) + 1, + sse_reconnect_count=int(prior_turn.get("sse_reconnect_count") or 0) + 1, ) self.send_response(200) self.send_header("Content-Type", "text/event-stream; charset=utf-8") @@ -1053,7 +1069,7 @@ def _resume_session(self, session_id: str) -> None: objective = str(context["objective"] or context["title"]) restored = self.server.runtime_controller.resume_session( session_id=session_id, - work_dir=context["project"], + work_dir=cast(Path, context["project"]), objective=objective, ) except KeyError: @@ -1403,6 +1419,7 @@ def _action_transition(self, proposal_id: str, transition: str) -> None: ) def _action_apply(self, proposal_id: str) -> None: + proposal: dict[str, object] | None try: body = self._read_json() if body: @@ -1552,7 +1569,7 @@ def do_GET(self) -> None: "endpoints": self.server.runtime_controller.capabilities(), } ) - get_dispatch = { + get_dispatch: dict[str, Callable[[], None]] = { "/api/chat/todo/detail": self._todo_detail, "/api/chat/completed-todos": self._completed_todos, "/api/chat/goal-results": self._goal_results, @@ -1605,7 +1622,7 @@ def do_POST(self) -> None: path = urlparse(self.path).path if self._dispatch_zcode_goal(path, apply=True): return - post_dispatch = { + post_dispatch: dict[str, Callable[[], None]] = { RESEARCH_ANALYSIS_PATH: self._research_analysis_write, RESEARCH_ANALYSIS_PATH + "/cancel": self._research_analysis_write, CHAT_SESSIONS_PATH: self._create_session, @@ -1732,6 +1749,7 @@ def serve_chat( server = ChatHTTPServer((host, port), ChatRequestHandler) server.registry_path = resolved_registry_path server.runtime_root = runtime_root + server.codex_account_service = CodexAccountService(runtime_root) server.runtime_root_override = resolved_runtime_root_override server.scan_roots = resolved_scan_roots server.limit = limit @@ -1757,18 +1775,24 @@ def serve_chat( ) server.chat_store = ChatSessionStore(runtime_root) server.action_store = ChatActionStore(runtime_root / "chat" / "actions") + def manager_scope(session: dict[str, object]) -> list[str] | None: + if isinstance(session.get("steward_context"), dict): + bindings = server.runtime_controller.project_contexts.conversation_bindings + if bindings is None: + raise ValueError("Steward conversation bindings are unavailable") + return cast(list[str] | None, bindings.steward_scope(session)) + return authorized_manager_goal_ids( + build_lark_goal_topic_runtime_snapshot( + registry_path=server.registry_path, runtime_root_override=server.runtime_root_override, + ), session, runtime_root=runtime_root, registry_path=resolved_registry_path, + ) + server.runtime_controller = ChatRuntimeController( store=server.chat_store, registry_path=resolved_registry_path, project_contexts=ChatProjectContexts(resolved_scan_roots, workspace_grant=project_workspace_grant, filesystem_scope=project_filesystem_scope), - manager_scope_resolver=lambda session: ( - server.runtime_controller.project_contexts.conversation_bindings.steward_scope(session) - if isinstance(session.get("steward_context"), dict) else authorized_manager_goal_ids( - build_lark_goal_topic_runtime_snapshot( - registry_path=server.registry_path, runtime_root_override=server.runtime_root_override, - ), session, runtime_root=runtime_root, registry_path=resolved_registry_path, - )), + manager_scope_resolver=manager_scope, codex_bin=codex_bin, claude_bin=claude_bin, kiro_cli_bin=kiro_cli_bin, @@ -1788,7 +1812,7 @@ def serve_chat( server.lark_private_conversations = private_transport - def _lark_snapshot(): + def _lark_snapshot() -> dict[str, object]: snapshot = build_lark_goal_topic_runtime_snapshot(registry_path=server.registry_path, runtime_root_override=server.runtime_root_override) snapshot["private_profiles"] = private_transport.profiles() @@ -1828,9 +1852,8 @@ def _lark_snapshot(): server.manager_return_service = start_return_service( server, server.runtime_controller.coordination_runtime_root, start_service=False ) - from .chat_loopx_mode import DelegationWakeService - def _wake_goal_context(session): + def _wake_goal_context(session: dict[str, object]) -> dict[str, object]: registry = load_registry(server.registry_path) goal = next( (item for item in registry_goals(registry) if str(item.get("id") or "") == str(session["goal_id"])), diff --git a/loopx/chat_session_open.py b/loopx/chat_session_open.py index 2326a650c1..658a9a517e 100644 --- a/loopx/chat_session_open.py +++ b/loopx/chat_session_open.py @@ -38,7 +38,7 @@ def open_chat_session( objective: str, mode: str, requested_endpoint: str = "", - agent_model_id: str = "", + agent_model_id: str | None = "", agent_reasoning_effort: str | None = None, project_ref: str | None = None, ) -> tuple[dict[str, Any], bool]: diff --git a/loopx/presentation/codex_accounts_api.py b/loopx/presentation/codex_accounts_api.py new file mode 100644 index 0000000000..3d1a5f12bc --- /dev/null +++ b/loopx/presentation/codex_accounts_api.py @@ -0,0 +1,111 @@ +"""Owner-local transport to the optional native Codex accounts provider. + +No account implementation, credentials or selection rules live in Core. +Uninstalled/disabled providers start no observer and leave native Codex alone. +""" +from __future__ import annotations + +from pathlib import Path +import importlib +import threading +from typing import Any, Protocol, cast + +CODEX_ACCOUNTS_PATH = "/api/chat/codex-accounts" + + +class CodexAccountService: + def __init__(self, runtime_root: Path): + self.provider: Any = None + self.watcher: Any = None + self.mutex = threading.Lock() + self.error: str | None = "extension_uninstalled" + try: + provider_module = importlib.import_module("loopx_codex_provider_routing.native") + except ImportError: + return + self.provider = provider_module.NativeAccounts(runtime_root) + self.watcher = provider_module.Watcher(self.provider) + self.error = None + # Rehydrate an explicitly enabled observer after an App restart. The + # absent/default-off path neither creates state nor reads native auth. + try: + if self.provider.path.exists() and self.provider._load()["policy"]["enabled"]: + self.watcher.start() + except Exception: + self.error = "state_unavailable" + + def call(self, operation: str, parameters: dict[str, Any]) -> dict[str, Any]: + if self.provider is None: + return {"ok": True, "available": False, "code": self.error} + try: + if operation == "status": + value = self.provider.status() + elif operation == "configure": + with self.mutex: + if parameters.get("enabled") is False: + self.watcher.stop.set() + value = self.provider.configure(**parameters) + if value["policy"]["enabled"]: + if self.watcher.stop.is_set(): + module = importlib.import_module("loopx_codex_provider_routing.native") + self.watcher = module.Watcher(self.provider) + self.watcher.start() + else: + self.watcher.close() + elif operation == "refresh": + value = self.provider.tick(refresh=True) + elif operation == "switch": + if set(parameters) != {"manual_slot", "expected_selection", "operation_id"}: + raise ValueError("Invalid switch request") + value = self.provider.tick(refresh=True, **parameters) + elif operation in ("recover", "rollback"): + value = self.provider.recover(rollback=operation == "rollback", **parameters) + elif operation == "reopen": + value = self.provider.reopen(**parameters) + else: + raise ValueError("Unsupported operation") + return {"ok": True, **value} + except Exception as error: + # Optional provider diagnostics can include private native paths. + code = getattr(error, "code", "native_unavailable") + cached = {} + try: + cached = self.provider._projection(self.provider._load()) + except Exception: + pass + return {**cached, "ok": False, "available": True, "code": code, "error_code": code} + + def close(self) -> None: + if self.watcher: + self.watcher.close() + + +class _HTTP(Protocol): + def _send_json(self, payload: dict[str, Any], *, status: int = 200) -> None: ... + + def _read_json(self) -> dict[str, Any]: ... + + +class CodexAccountsRequestMixin: + server: Any + + def _codex_accounts_read(self) -> None: + cast(_HTTP, self)._send_json(self.server.codex_account_service.call("status", {})) + + def _codex_accounts_write(self) -> None: + http = cast(_HTTP, self) + try: + body = http._read_json() + if set(body) != {"operation", "parameters"} or not isinstance(body["parameters"], dict): + raise ValueError("Invalid account operation") + operation, parameters = body["operation"], body["parameters"] + fields = {"configure": {"expected_revision", "enabled", "slots", "reconnect", "restart_clients"}, + "switch": {"manual_slot", "expected_selection", "operation_id"}, + "recover": {"expected_revision", "expected_selection"}, "rollback": {"expected_revision", "expected_selection"}, + "reopen": {"expected_selection"}, "refresh": set()} + if operation not in fields or set(parameters) - fields[operation]: + raise ValueError("Invalid account parameters") + result = self.server.codex_account_service.call(operation, parameters) + except (TypeError, ValueError, KeyError): + result = {"ok": False, "code": "invalid_request"} + http._send_json(result, status=200 if result["ok"] else 409) diff --git a/loopx/presentation/configuration_api.py b/loopx/presentation/configuration_api.py index 38fea2a0a4..88fac0b905 100644 --- a/loopx/presentation/configuration_api.py +++ b/loopx/presentation/configuration_api.py @@ -3,6 +3,7 @@ from collections.abc import Callable from . import codex_attached_api as attached_api +from . import codex_accounts_api as accounts_api from . import workspace_preferences_api as preferences_api from . import configuration_backup_api as backup_api from . import state_recovery_api as recovery_api @@ -20,6 +21,7 @@ class ChatConfigurationRequestMixin( + accounts_api.CodexAccountsRequestMixin, attached_api.CodexAttachedRequestMixin, request_api.RequestControlMixin, preferences_api.WorkspacePreferencesRequestMixin, @@ -40,6 +42,7 @@ class ChatConfigurationRequestMixin( def _configuration_get_routes(self) -> dict[str, Callable[[], None]]: return { + accounts_api.CODEX_ACCOUNTS_PATH: self._codex_accounts_read, attached_api.CODEX_ATTACHED_PATH: self._codex_attached_read, attached_api.CODEX_ATTACHED_LIVE_PATH: self._codex_attached_live, request_api.REQUEST_CONTROL_PATH: self._request_control_read, @@ -59,6 +62,7 @@ def _configuration_get_routes(self) -> dict[str, Callable[[], None]]: def _configuration_post_routes(self) -> dict[str, Callable[[], None]]: return { + accounts_api.CODEX_ACCOUNTS_PATH: self._codex_accounts_write, context_api.DECISION_CONTEXT_PATH: self._decision_context_request, attached_api.CODEX_ATTACHED_PATH: self._codex_attached_send, attached_api.CODEX_ATTACHED_OPEN_PATH: self._codex_attached_open, diff --git a/loopx/presentation/goal_direction_api.py b/loopx/presentation/goal_direction_api.py index 752282dce1..651e580adb 100644 --- a/loopx/presentation/goal_direction_api.py +++ b/loopx/presentation/goal_direction_api.py @@ -20,13 +20,13 @@ class GoalDirectionRequestMixin: server: Any path: str - def _send_json(self, value: dict[str, Any], **kwargs: Any) -> None: + def _send_json(self, payload: dict[str, Any], *, status: int = 200) -> None: raise NotImplementedError def _send_error(self, message: str, **kwargs: Any) -> None: raise NotImplementedError - def _read_json(self, **kwargs: Any) -> dict[str, Any]: + def _read_json(self, *, max_bytes: int = 256 * 1024) -> dict[str, Any]: raise NotImplementedError def _direction_route(self, goal_id: Any) -> tuple[Path, dict[str, str]]: diff --git a/loopx/presentation/workspace_preferences_api.py b/loopx/presentation/workspace_preferences_api.py index 2ffc602e99..952ee8854e 100644 --- a/loopx/presentation/workspace_preferences_api.py +++ b/loopx/presentation/workspace_preferences_api.py @@ -47,13 +47,13 @@ class WorkspacePreferencesRequestMixin: server: Any path: str - def _send_json(self, value: dict[str, Any], **kwargs: Any) -> None: + def _send_json(self, payload: dict[str, Any], *, status: int = 200) -> None: raise NotImplementedError def _send_error(self, message: str, **kwargs: Any) -> None: raise NotImplementedError - def _read_json(self, **kwargs: Any) -> dict[str, Any]: + def _read_json(self, *, max_bytes: int = 256 * 1024) -> dict[str, Any]: raise NotImplementedError def _sidebar_order_file(self) -> Path: diff --git a/packages/README.md b/packages/README.md index 3229fab2fd..ba612e0714 100644 --- a/packages/README.md +++ b/packages/README.md @@ -27,8 +27,8 @@ Current co-located distributions, in directory order: LoopX-governed work unit on Ark Managed Agents (Python distribution, no `extension.toml`; selected with `turn run-once --host generic-cli`); - [`loopx-codex-provider-routing`](loopx-codex-provider-routing/README.md): - public-safe Codex App + CPA catalog compilation, qualification and upgrade - planning; + native Codex account slots, allowance observation and automatic handoff at + safe boundaries, with legacy CPA qualification and rollback compatibility; - [`loopx-community-discussion`](loopx-community-discussion/README.md): public-safe community discussion scanner producing compact `discussion_fact_v0` facts; diff --git a/packages/loopx-codex-provider-routing/NATIVE.md b/packages/loopx-codex-provider-routing/NATIVE.md new file mode 100644 index 0000000000..62c530dcfd --- /dev/null +++ b/packages/loopx-codex-provider-routing/NATIVE.md @@ -0,0 +1,178 @@ +# Native Codex accounts + +The optional `loopx-codex-provider-routing` provider replaces the recommended CPA +path with native slots. It has no new Core capability id. Credentials and lifecycle +stay in this package; Core composes the App transport. Python adapts existing +private file/process owners; `native_policy.ts` owns availability and selection. + +## The user task + +Previously an exhausted account required finding another account, closing clients, +selecting it by hand and reopening the GPT interface. In +**Settings → Model provider → Native Codex accounts**, LoopX checks allowance and +selects/reopens when the native owner admits it. Enable automatic handoff and +**Allow automatic GPT client quit and restart** for the complete automatic path. +The page reads back accounts, +windows, waiting cause and recovery. Tokens and identity claims stay private. + +**An open native Codex client is an occupied owner, even between turns.** With +restart permission, the provider requests a normal AppKit quit of only the +inspected GPT instance, waits for exit and native idle, saves/switches the login, +then reopens that same route. It never addresses a shared bundle id or quits a +different route. A refused quit, timeout, new process owner or remaining CLI/file +owner leaves credentials unchanged and pauses automatic retries until a check. +Without this additional permission it continues to wait for closed clients. +It does not implement hot login replacement, force termination, job replay or +resumption of an interrupted turn. If every +account is unavailable, it waits for the observed reset and checks again. + +## Prerequisites and activation + +Install in the same Python environment as LoopX, then restart its App/Chat service: + +```sh +python -m pip install ./packages/loopx-codex-provider-routing +loopx-codex-accounts status +``` + +The native adapter requires POSIX, the installed **Personal Apps** GPT slot owner +(`personal-apps` with adjacent `personal_apps.py` and `personal_accounts.py`), its +registered native GPT route, and the official `codex` CLI. Switching is macOS-only +in that owner. This package adapts that installation; it ships no second account +vault or launcher. Other hosts show unavailable. Node 22.18+ or 24+ runs the +TypeScript selector. Enrol accounts through the native owner's official login. + +Click **Switch automatically when quota runs out** to authorize the listed slots +as destinations. Installation alone grants no account effect. The App observer +runs with its serving LoopX process and rehydrates the saved opt-in on restart. +Client restart is a separate permission, off by default and absent from older +policies. Enabling it also permits a subsequent explicit manual switch to quit +and reopen the GPT instance. Neither permission is enabled by installation. +An operator-owned service independent of the App can run the same observer: + +```sh +loopx-codex-accounts watch +``` + +Do not run a busy shell loop. A private lock serializes hosts; observations carry +next-due times. This provider is independent of Goal compute quota, heartbeat +settlement and the model data plane. CLI mutations take JSON on stdin with +revisions from `status`: + +```sh +loopx-codex-accounts configure <<'JSON' +{"expected_revision":0,"enabled":true,"restart_clients":true,"slots":["a","b"]} +JSON +loopx-codex-accounts refresh +``` + +Explicit `switch` accepts `manual_slot`, `expected_selection` and a unique +`operation_id`; retrying it does not repeat selection. `recover` and `rollback` +require `expected_revision` and `expected_selection`. App buttons use the same +fences. `--runtime-root` selects a custom private runtime; browser requests +cannot supply paths, executables or source files. +`reopen` accepts `expected_selection` and retries only opening a committed +selection. It never repeats the credential switch. +Explicit checking or selection can connect the registered slots with automatic +effects still off. A manual choice may select an exhausted/expired account for +native sign-in or inspection; this never qualifies it for automatic selection. + +## Availability and continuation + +Usage checks use verified HTTPS without model calls, bound response size and +refuse redirects. Schema drift/network failure means unknown, never exhausted. +The active native client owns its login refresh. A distinct inactive account +can refresh through an isolated official App Server; rotated bytes are saved +only to that slot under native locks after identity and selection checks. No +active-home copy is refreshed. + +A rejected inactive refresh is cached against that private login-cache revision; +background checks do not repeatedly retry the same rejected credential. A new +native login or an explicit check permits one new attempt. This refusal stays +visible as a sign-in requirement, rather than aging into usable allowance. + +The selector prefers App Server `rateLimitsByLimitId.codex` over the legacy +bucket. Review/other product buckets cannot establish Codex availability. +Missing/invalid/future-dated or older-than-two-minute samples stay unknown. A +past reset invalidates an old exhausted sample; it does not grant new allowance. +100% subscription use with positive credits stays available; absent credit facts +remain unknown. Login errors, generic 429, query failure and measured exhaustion +are distinct. The observer never buys credits or uses reset coupons. + +Only fresh active-account exhaustion starts automatic selection. Destinations +must be allowlisted and freshly available. Selection follows the configured +ring with a five-minute cooldown. An external native/manual choice wins and +starts a cooldown. Busy clients defer; refused/changed clients and remaining +file owners pause further shutdowns. Other repeated failures stop after three +attempts until an explicit check/reconnect. Resets require new observations. +All-unavailable accounts are checked again at the next reset or within five +minutes; unknown observations retry within one minute. These are bounded usage +reads, not model probes. The managed extension's enable/disable controls its +permissionless qualification protocol; the native host has this separate +explicit opt-in and stop lifecycle. + +An intent is durable before the effect. A restart-authorized switch runs in a +separate POSIX session and writes a private completion receipt, so quitting the +calling Codex App does not kill the switch/restart or its result channel. A +restarted observer waits for that receipt instead of dispatching another switch. +An interrupted worker remains visible and recoverable; an unresolved live +worker cannot be overwritten by reconfiguration or recovery. If shutdown succeeds +but switching fails, the unchanged source App is reopened when native preflight +allows it. An incomplete native journal stays closed for recovery. + +The native owner checks idle and the +selection revision under its own locks and immediately before writes, saves +outgoing token rotations, commits and reads back. A lost reply reconciles +selection without replaying a switch. Interrupted native journals require +existing recovery, which refuses to overwrite later edits. Model settings and +history stay outside the transaction. Reopening failure is reported after +committed selection, without switching again. +The same checks reject expired automatic quota evidence before shutting down or +writing credentials. A manual/native choice during exit or launch wins. + +## Disable, upgrade and rollback + +Turn the switch off to stop automatic effects. CLI `configure` with +`enabled:false` also works offline. Stop a separate `watch` process through its +service manager. Stop/restart the serving process before uninstalling loaded +code. Uninstall leaves Core and native slots intact. +An already dispatched credential transaction finishes and reopens safely even +when the observer is stopped; disabling never discards that pending intent. + +Activation pins the two installed native source files. Updates produce +`adapter_changed`; **Reconnect** reads and pins the installation again. An old +observer never silently adopts new source. Private state contains paths, pins, +policy, content-free observations and an intent, never tokens. + +From CPA, first stop its separately owned service and restore a backed-up native +configuration using its existing operator workflow. Keep the selected model; +review removal of active custom provider/catalog/profile overrides. The native +adapter refuses a non-native default and never edits an App bundle or rewrites +a running home for migration. The [native template](templates/codex-app-config.toml) +illustrates the new route; the [legacy template](templates/legacy-cpa-config.toml) +and explicit CPA operator remain for deliberate rollback. No credential or +history conversion occurs. + +## Qualification boundary + +Tests challenge exhausted/unknown/credit-funded facts, allowlists, manual fences, +duplicates, concurrency, busy deferral, lost replies, stop-before-effect and +recovery with synthetic credentials. Browser evidence must use the packaged +page with the owning backend and a recovery case. Installed native file/process +qualification uses a disposable home and a signed disposable Cocoa App: actual +AppKit normal quit, refusal, native idle/save/switch, source reopen on failure, +restart failure recovery and a detached worker surviving its owned caller exit. +Quota and authentication transport remain synthetic. Read-only live status certifies neither +an actual account swap nor installed adoption or uninterrupted hot switching. + +## Packaged example + +The packaged settings page below uses synthetic account aliases and allowance, +with the actual native file/transaction owner and a disposable Cocoa App. +Usage/OAuth responses are simulated; a real-account cutover is not qualified by +these images. Desktop and narrow views keep the selected account +visible even when the account list continues below the viewport. + +![Packaged desktop account settings](docs/assets/native-accounts-desktop.png) + +[Narrow account settings](docs/assets/native-accounts-phone.png) diff --git a/packages/loopx-codex-provider-routing/OPERATOR.md b/packages/loopx-codex-provider-routing/OPERATOR.md index c63863e062..f5be0942a3 100644 --- a/packages/loopx-codex-provider-routing/OPERATOR.md +++ b/packages/loopx-codex-provider-routing/OPERATOR.md @@ -1,5 +1,8 @@ # Local CPA operator +Legacy opt-in compatibility. New native account installations use +[Native accounts](NATIVE.md); this operator is never selected automatically. + The package now includes `loopx-cpa-operator`, a separately invoked local CLI. The managed extension protocol remains read-only and permission-free. Installing or running that protocol does not enable the operator, discover credentials, diff --git a/packages/loopx-codex-provider-routing/README.md b/packages/loopx-codex-provider-routing/README.md index efc0b244ba..ac92369661 100644 --- a/packages/loopx-codex-provider-routing/README.md +++ b/packages/loopx-codex-provider-routing/README.md @@ -1,9 +1,17 @@ # LoopX Codex Provider Routing -This optional LoopX extension owns the public-safe integration contract for a -single Codex App using CPA as its multi-subscription and multi-provider data -plane. It packages configuration compilation, qualification readback and -upgrade planning without becoming a model proxy or credential authority. +This optional extension integrates **native Codex account slots** with LoopX. +It observes allowance, selects an authorized available account after exhaustion, +and asks the installed native slot owner to switch at an idle boundary. Codex +keeps its official OpenAI login, model settings and history. CPA is no longer +the recommended provider path. + +Install this package in the LoopX runtime, then open **Settings → Model provider +→ Native Codex accounts**. The page discovers slots, enables automatic handoff, +checks allowance, switches explicitly and reads back recovery. The same owner +is available through `loopx-codex-accounts`. See the [native guide](NATIVE.md) +for prerequisites, migration and the safe-boundary limitation. Installation +alone switches nothing. ## Placement @@ -11,16 +19,26 @@ This package is an **extension** that supplies a standalone integration provider. It is not a built-in capability: - Codex App owns task settings, turn snapshots and the session store; -- CPA owns credential selection, online routing, retry and stream adaptation; -- the operator owns OAuth/API credentials, process installation and private - build/rollback receipts; -- LoopX owns this versioned public contract, managed extension lifecycle, - public-safe qualification results and upgrade gates. - -The extension declares no permissions and performs no writes. It does not read +- the installed Personal Apps slot manager owns credential transactions, + idle admission, login-cache preservation and rollback; +- the optional provider owns allowance observation and its default-off policy; + its TypeScript selector is shared by CLI and App; +- LoopX Core supplies optional lifecycle and owner-local presentation transport, + without becoming a credential or model proxy. + +The **managed qualification protocol** declares no permissions and performs no writes. It does not read `HOME`, `CODEX_HOME`, auth files, process tables or network endpoints. Callers must convert local observations into the content-free snapshot schema before -qualification. +qualification. The separately invoked native host reads local credentials, +queries native usage and performs explicitly enabled account effects. It is +never invoked by the permissionless protocol or its doctor. + +## Legacy compatibility + +Existing CPA contracts, fixtures and `loopx-cpa-operator` remain for explicit +legacy use and rollback. They are not started, migrated or selected automatically. +The following operations describe that compatibility protocol; they do not +certify native-slot availability or authorize a native switch. ## Operations diff --git a/packages/loopx-codex-provider-routing/RUNBOOK.md b/packages/loopx-codex-provider-routing/RUNBOOK.md index 9aaae5c116..f480e59710 100644 --- a/packages/loopx-codex-provider-routing/RUNBOOK.md +++ b/packages/loopx-codex-provider-routing/RUNBOOK.md @@ -1,5 +1,9 @@ # Codex App 多订阅与多 Provider 切换 Runbook +> Legacy CPA maintenance only. The recommended native account provider and +> its activation, automatic handoff and recovery are in [Native accounts](NATIVE.md). +> This historical CPA procedure does not describe the current default route. + > 配套文档：需要长期保留两个独立 App / `CODEX_HOME` 时，参见 > [Codex 多 App 隔离与运维最佳实践（中文）](../../docs/guides/codex-multi-app-best-practices.zh-CN.md)。 diff --git a/packages/loopx-codex-provider-routing/docs/assets/native-accounts-desktop.png b/packages/loopx-codex-provider-routing/docs/assets/native-accounts-desktop.png new file mode 100644 index 0000000000..73f9d4dff1 Binary files /dev/null and b/packages/loopx-codex-provider-routing/docs/assets/native-accounts-desktop.png differ diff --git a/packages/loopx-codex-provider-routing/docs/assets/native-accounts-phone.png b/packages/loopx-codex-provider-routing/docs/assets/native-accounts-phone.png new file mode 100644 index 0000000000..195cc44ae5 Binary files /dev/null and b/packages/loopx-codex-provider-routing/docs/assets/native-accounts-phone.png differ diff --git a/packages/loopx-codex-provider-routing/extension.toml b/packages/loopx-codex-provider-routing/extension.toml index e4934583ac..21fac96db0 100644 --- a/packages/loopx-codex-provider-routing/extension.toml +++ b/packages/loopx-codex-provider-routing/extension.toml @@ -1,6 +1,6 @@ schema_version = "loopx_extension_manifest_v0" id = "loopx-codex-provider-routing" -version = "0.11.0" +version = "0.12.0" requires_loopx_api = ">=1,<2" permissions = [] diff --git a/packages/loopx-codex-provider-routing/pyproject.toml b/packages/loopx-codex-provider-routing/pyproject.toml index 21228addb2..45d756e0ae 100644 --- a/packages/loopx-codex-provider-routing/pyproject.toml +++ b/packages/loopx-codex-provider-routing/pyproject.toml @@ -4,14 +4,18 @@ build-backend = "setuptools.build_meta" [project] name = "loopx-codex-provider-routing" -version = "0.11.0" -description = "Public-safe Codex multi-provider routing integration for LoopX" +version = "0.12.0" +description = "Native Codex account slots and exhausted-account handoff for LoopX" readme = "README.md" requires-python = ">=3.11" [project.scripts] loopx-cpa-operator = "loopx_codex_provider_routing.operator:main" loopx-codex-provider-routing = "loopx_codex_provider_routing.cli:main" +loopx-codex-accounts = "loopx_codex_provider_routing.native:main" [tool.setuptools.packages.find] where = ["src"] + +[tool.setuptools.package-data] +loopx_codex_provider_routing = ["*.ts"] diff --git a/packages/loopx-codex-provider-routing/smoke/codex_provider_routing_smoke.py b/packages/loopx-codex-provider-routing/smoke/codex_provider_routing_smoke.py index 1c4c9da09a..b855d75803 100644 --- a/packages/loopx-codex-provider-routing/smoke/codex_provider_routing_smoke.py +++ b/packages/loopx-codex-provider-routing/smoke/codex_provider_routing_smoke.py @@ -29,7 +29,7 @@ def _assert_layered_retry_templates() -> None: - app = (PACKAGE_ROOT / "templates" / "codex-app-config.toml").read_text() + app = (PACKAGE_ROOT / "templates" / "legacy-cpa-config.toml").read_text() local_cpa = app.split("[model_providers.local-cpa]", maxsplit=1)[1] request_retries = re.search(r"(?m)^request_max_retries\s*=\s*(\d+)\s*$", local_cpa) stream_retries = re.search(r"(?m)^stream_max_retries\s*=\s*(\d+)\s*$", local_cpa) diff --git a/packages/loopx-codex-provider-routing/smoke/native-policy.test.mjs b/packages/loopx-codex-provider-routing/smoke/native-policy.test.mjs new file mode 100644 index 0000000000..726116bbb9 --- /dev/null +++ b/packages/loopx-codex-provider-routing/smoke/native-policy.test.mjs @@ -0,0 +1,61 @@ +import assert from "node:assert/strict"; +import { test } from "node:test"; +import { account, decide } from "../src/loopx_codex_provider_routing/native_policy.ts"; + +const now = 1000; +const sample = (slot, used = 10, extra = {}) => ({ slot, observed_at: now, status: "ok", usage: { + rate_limit: { primary_window: { used_percent: used, reset_at: 2000 }, + secondary_window: { used_percent: 20, reset_at: 9000 } }, credits: { has_credits: false, balance: "0" }, ...extra, +} }); +const input = overrides => ({ now, policy: { enabled: true, slots: ["a", "b", "c"], cooldown_seconds: 300 }, + active: "a", observations: [sample("a", 100), sample("b"), sample("c")], + last_switch_at: null, recovery_required: false, login_matches: true, ...overrides }); + +test("only measured exhaustion switches; default-off preserves the selected account", () => { + assert.equal(decide(input({})).target, "b"); + assert.equal(decide(input({ observations: [sample("a"), sample("b")] })).action, "ready"); + assert.equal(decide(input({ policy: { enabled: false, slots: ["b"], cooldown_seconds: 300 } })).action, "disabled"); + for (const status of ["unavailable", "auth_required"]) { + assert.equal(decide(input({ observations: [{ slot: "a", observed_at: now, status }, sample("b")] })).target, null); + } +}); + +test("credits, null and multi-bucket facts retain their meaning", () => { + assert.equal(account(sample("a", 100, { credits: { has_credits: true, balance: "12.0" } }), now).availability, "available"); + assert.equal(account({ ...sample("a"), usage: {} }, now).availability, "unknown"); + assert.equal(account(sample("a", 100, { credits: null }), now).availability, "unknown"); + assert.equal(account({ ...sample("a"), usage: { rateLimitsByLimitId: { review: { primary: { usedPercent: 0 } } }, + rateLimits: { primary: { usedPercent: 5 } } } }, now).availability, "unknown"); + const native = { ...sample("a"), usage: { rateLimitsByLimitId: { codex: { + primary: { usedPercent: 100, resetsAt: 2000 }, secondary: { usedPercent: 100, resetsAt: 9000 }, credits: { hasCredits: false }, + } } } }; + assert.equal(account(native, now).reset_at, 9000); + assert.equal(account({ ...sample("a", 100), observed_at: 1 }, now).availability, "unknown"); + assert.equal(account(sample("a", 100), 2500).availability, "unknown"); +}); + +test("allowlist, manual fence, cooldown and all-unavailable wait precede effects", () => { + assert.equal(decide(input({ policy: { enabled: true, slots: ["a", "c"], cooldown_seconds: 300 } })).target, "c"); + assert.equal(decide(input({ login_matches: false })).action, "selection_changed"); + assert.equal(decide(input({ recovery_required: true })).action, "recovery_required"); + assert.equal(decide(input({ last_switch_at: 900 })).action, "cooldown"); + assert.equal(decide(input({ manual_slot: "outside" })).action, "target_unavailable"); + assert.equal(decide(input({ manual_slot: "b", observations: [sample("a"), { + slot: "b", observed_at: now, status: "auth_required", + }] })).target, "b", "Explicit selection can open native sign-in; automatic selection cannot"); + const unavailable = decide(input({ observations: [sample("a", 100), sample("b", 100), sample("c", 100)] })); + assert.equal(unavailable.action, "all_unavailable"); + assert.equal(unavailable.retry_at, 1300); + assert.equal(decide(input({ observations: [sample("a", 100), { ...sample("b"), observed_at: 0 }] })).target, null); +}); + +test("normal client restart needs its own permission and the quota proof expires before effects", () => { + const allowed = { enabled: true, slots: ["a", "b"], cooldown_seconds: 300, restart_clients: true }; + const result = decide(input({ policy: allowed, observations: [sample("a", 100), { ...sample("b"), observed_at: 950 }] })); + assert.equal(result.restart_clients, true); + assert.equal(result.valid_until, 1070); + assert.equal(decide(input({})).restart_clients, false, "Existing policies do not gain client shutdown authority"); + assert.equal(decide(input({ policy: allowed, observations: [sample("a"), sample("b")] })).restart_clients, false); + assert.equal(decide(input({ policy: allowed, manual_slot: "b" })).valid_until, null, + "An explicit sign-in choice is not an automatic quota decision"); +}); diff --git a/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native.py b/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native.py new file mode 100644 index 0000000000..1542c9b978 --- /dev/null +++ b/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native.py @@ -0,0 +1,432 @@ +"""Opt-in native Codex account host; one policy for CLI and the LoopX App. + +The managed permissionless qualification protocol never imports this module. +Python owns private file/process IO; native_policy.ts owns availability/order. +""" +from __future__ import annotations + +import argparse +from contextlib import contextmanager +import json +import os +from pathlib import Path +import shutil +import subprocess +import sys +import tempfile +import threading +import time +import uuid + +from .native_host import HostError, discover + +SCHEMA = "codex_native_accounts_v1" + + +def private_read(path: Path) -> dict: + if any(p.is_symlink() for p in (path, *path.parents)): + raise HostError("state_untrusted") + if path.stat().st_mode & 0o077 or path.stat().st_uid != os.getuid(): + raise HostError("state_untrusted") + value = json.loads(path.read_text()) + if value.get("schema_version") != SCHEMA: + raise HostError("state_incompatible") + return value + + +def atomic(path: Path, value: dict) -> None: + if any(p.is_symlink() for p in (path, *path.parents)): + raise HostError("state_untrusted") + fd, temporary = tempfile.mkstemp(dir=path.parent) + try: + os.fchmod(fd, 0o600) + with os.fdopen(fd, "w") as stream: + json.dump(value, stream) + stream.flush() + os.fsync(stream.fileno()) + os.replace(temporary, path) + directory = os.open(path.parent, os.O_RDONLY) + try: + os.fsync(directory) + finally: + os.close(directory) + finally: + Path(temporary).unlink(missing_ok=True) + + +def policy(request: dict) -> dict: + node = shutil.which("node") + if not node: + raise HostError("node_unavailable") + result = subprocess.run([node, "--no-warnings", "--experimental-strip-types", + str(Path(__file__).with_name("native_policy_bridge.ts"))], + input=json.dumps(request), capture_output=True, text=True, timeout=10) + if result.returncode or len(result.stdout) > 131072: + raise HostError("policy_unavailable") + return json.loads(result.stdout) + + +def host(adapter: dict, operation: str, **parameters) -> dict: + receipt = parameters.pop("handoff_receipt", None) + command = [sys.executable, "-m", "loopx_codex_provider_routing.native_host"] + request = json.dumps({"adapter": adapter, "operation": operation, **parameters}) + if receipt: + # Closing the calling native App must not close this continuation or + # its result channel. All native sub-operations have bounded waits. + process = subprocess.Popen([*command, "--result-file", receipt], stdin=subprocess.PIPE, + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + text=True, start_new_session=True) + try: + process.communicate(request, timeout=150) + except subprocess.TimeoutExpired: + # Do not kill a credential transaction. A separate reaper owns only + # this child; the receipt fences retries even if its caller exits. + threading.Thread(target=process.communicate, daemon=True).start() + raise HostError("handoff_pending") from None + packet = private_read(Path(receipt)) + if packet.get("ok") is not True: + raise HostError(packet.get("code", "native_unavailable")) + return packet["result"] + result = subprocess.run(command, input=request, capture_output=True, text=True, timeout=150) + if len(result.stdout) > 131072: + raise HostError("native_unavailable") + try: + packet = json.loads(result.stdout) + except ValueError: + raise HostError("native_unavailable") from None + if packet.get("ok") is not True: + raise HostError(packet.get("code", "native_unavailable")) + return packet["result"] + + +class NativeAccounts: + def __init__(self, runtime_root: Path, *, transport=host, clock=time.time): + self.root = Path(runtime_root) / "extensions" / "codex-native-accounts" + self.path = self.root / "state.json" + self.transport, self.clock = transport, clock + self.switch_allowed = lambda: True + + def _load(self) -> dict: + if self.path.exists(): + return private_read(self.path) + return {"schema_version": SCHEMA, "revision": 0, + "policy": {"enabled": False, "slots": [], "cooldown_seconds": 300, "restart_clients": False}, + "observations": [], "last_switch_at": None, "action": "disabled"} + + @contextmanager + def _lock(self, *, wait_seconds: float = 0): + if os.name != "posix": + raise HostError("platform_unavailable") + import fcntl + + if any(p.is_symlink() for p in (self.root, *self.root.parents)): + raise HostError("state_untrusted") + self.root.mkdir(mode=0o700, parents=True, exist_ok=True) + if self.root.stat().st_uid != os.getuid() or self.root.stat().st_mode & 0o077: + raise HostError("state_untrusted") + path = self.root / "lock" + if path.is_symlink(): + raise HostError("state_untrusted") + with open(path, "a", opener=lambda p, flags: os.open(p, flags, 0o600)) as stream: + deadline = time.monotonic() + wait_seconds + while True: + try: + fcntl.flock(stream, fcntl.LOCK_EX | fcntl.LOCK_NB) + break + except BlockingIOError: + if time.monotonic() >= deadline: + raise + time.sleep(0.05) + yield + + def _projection(self, state: dict, native: dict | None = None) -> dict: + native = native or state.get("native", {}) + return {"schema_version": SCHEMA, "available": True, "revision": state["revision"], + "policy": state["policy"], "slots": native.get("slots", []), + "active_slot": native.get("active_slot"), + "selection_revision": native.get("selection_revision"), + "login_matches": native.get("current_login_matches"), + "recovery_required": native.get("recovery_required", False) or state.get("action") == "handoff_interrupted", + "action": state.get("action", "disabled"), "target": state.get("target"), + "retry_at": state.get("retry_at"), "last_checked_at": state.get("last_checked_at"), + "last_switch_at": state.get("last_switch_at"), "accounts": state.get("accounts", []), + "code": state.get("code"), + "launch_failed": state.get("launch_failed", False) and native.get("client_running") is not True, + "client_running": native.get("client_running"), + "automatic_hot_swap": False} + + def status(self) -> dict: + state = self._load() + adapter = state.get("adapter") or discover() + native = self.transport(adapter, "status") + return self._projection(state, native) + + def configure(self, *, expected_revision: int, enabled: bool, slots: list[str], reconnect: bool = False, + restart_clients: bool | None = None) -> dict: + if type(enabled) is not bool or not isinstance(slots, list) or len(slots) > 16 or len(slots) != len(set(slots)): + raise HostError("invalid_policy") + if restart_clients is not None and type(restart_clients) is not bool: + raise HostError("invalid_policy") + # Stop requests fence new effects at the Watcher first, then persist + # after an already dispatched atomic switch finishes. Other mutations + # retain their immediate conflict behavior. + with self._lock(wait_seconds=155 if not enabled else 0): + state = self._load() + if state["revision"] != expected_revision: + raise HostError("configuration_changed") + pending = state.get("intent") or {} + if pending.get("handoff_receipt") and enabled: + raise HostError("handoff_pending") + # Disabling works offline and never needs credentials or a client. + if enabled or reconnect: + adapter = discover() if reconnect or "adapter" not in state else state["adapter"] + native = self.transport(adapter, "status") + if reconnect and not slots and not enabled: + slots = native["slots"] + if not slots or not set(slots) <= set(native["slots"]): + raise HostError("invalid_slots") + state.update(adapter=adapter, native=native) + state.update(policy={"enabled": enabled, "slots": slots, "cooldown_seconds": 300, + "restart_clients": restart_clients if restart_clients is not None + else state["policy"].get("restart_clients", False)}, + revision=state["revision"] + 1, retry_at=None, code=None, + intent=pending if pending.get("handoff_receipt") else None, + action="checking" if enabled else "disabled", failure_count=0) + atomic(self.path, state) + return self._projection(state) + + def tick(self, *, refresh: bool = False, manual_slot: str | None = None, + expected_selection: str | None = None, operation_id: str | None = None) -> dict: + if manual_slot is not None and (not expected_selection or not isinstance(operation_id, str) or not 1 <= len(operation_id) <= 128): + raise HostError("invalid_request") + with self._lock(): + state = self._load() + if operation_id and operation_id == state.get("last_operation_id"): + return self._projection(state) + now = self.clock() + if not state["policy"]["enabled"] and not refresh and not manual_slot and not state.get("intent"): + return self._projection(state) + if not refresh and not manual_slot and now < (state.get("retry_at") or 0): + return self._projection(state) + if state.get("failure_count", 0) >= 3 and not refresh and not manual_slot: + return self._projection(state) + if "adapter" not in state: + # An explicit check/manual choice can connect existing slots + # without turning automatic effects on first. + state["adapter"] = discover() + native = self.transport(state["adapter"], "status") + state.update(native=native, policy={**state["policy"], "slots": native["slots"]}, + revision=state["revision"] + 1) + try: + native = self.transport(state["adapter"], "status") + if expected_selection is not None and expected_selection != native["selection_revision"]: + raise HostError("selection_changed") + previous = state.get("native", {}).get("selection_revision") + intent = state.get("intent") + receipt = None + if intent and intent.get("handoff_receipt"): + path = Path(intent["handoff_receipt"]) + if path.exists(): + receipt = private_read(path) + else: + # A detached switch may still be exiting/reopening the + # native client. Never launch a second switch on retry. + interrupted = self.clock() - intent["started_at"] > 180 + state.update(native=native, action="handoff_interrupted" if interrupted else "restarting", + retry_at=self.clock() + 5, code="handoff_interrupted" if interrupted else None) + atomic(self.path, state) + return self._projection(state) + if receipt.get("ok") is not True: + state["intent"] = None + raise HostError(receipt.get("code", "native_unavailable")) + if intent and native["active_slot"] == intent["target"] and not native["recovery_required"] and native["current_login_matches"] is True: + # An interrupted caller may have lost the successful reply. + # Read back selection; never replay a switch or a model turn. + state.update(native=native, intent=None, last_switch_at=now, action="selected", target=None, + retry_at=now + 300, code=None, failure_count=0, + launch_failed=receipt["result"].get("launch_failed", False) if receipt + else native.get("client_running") is False) + atomic(self.path, state) + return self._projection(state) + state["native"] = native + if not manual_slot and previous and previous != native["selection_revision"]: + # A native/manual choice outside this host wins. Observe it + # after a cooldown instead of immediately undoing it. + state.update(intent=None, last_switch_at=now, action="selection_changed", target=None, + accounts=[], retry_at=now + 300, code=None, failure_count=0) + atomic(self.path, state) + return self._projection(state) + if native["recovery_required"]: + state.update(action="recovery_required", retry_at=None) + atomic(self.path, state) + return self._projection(state) + if native["current_login_matches"] is not True: + raise HostError("selection_changed") + # Inspect the active account even when it is not an authorized + # destination. Candidates stay strictly inside the allowlist. + slots = list(dict.fromkeys([native["active_slot"], *state["policy"]["slots"]])) + failures = {} if refresh or manual_slot else { + row["slot"]: {"revision": row["credential_revision"], "observed_at": row["observed_at"]} + for row in state.get("observations", []) + if row["status"] == "auth_required" and "credential_revision" in row + } + observed = self.transport(state["adapter"], "observe", slots=slots, + expected_revision=native["selection_revision"], auth_failures=failures) + public_observations = [{k: v for k, v in row.items() if k != "credential_revision"} + for row in observed["observations"]] + decision = policy({"now": self.clock(), "policy": state["policy"], "active": native["active_slot"], + "observations": public_observations, "last_switch_at": state["last_switch_at"], + "recovery_required": False, "login_matches": True, + **({"manual_slot": manual_slot} if manual_slot else {})}) + state.update(observations=observed["observations"], accounts=decision["accounts"], + last_checked_at=self.clock(), action=decision["action"], target=decision["target"], + retry_at=decision["retry_at"] or self.clock() + 60, code=None, failure_count=0) + if decision["action"] == "switch": + if not manual_slot and not self.switch_allowed(): + raise HostError("observer_stopped") + # Durable before the effect, with source selection and target. + # Native owner rechecks the same fence inside its transaction. + state["intent"] = {"id": operation_id or uuid.uuid4().hex, + "source_revision": native["selection_revision"], "target": decision["target"]} + effects = {"restart_clients": decision["restart_clients"], "not_after": decision["valid_until"]} + if decision["restart_clients"]: + directory = self.root / "handoffs" + if directory.is_symlink(): + raise HostError("state_untrusted") + directory.mkdir(mode=0o700, exist_ok=True) + receipt_path = directory / (uuid.uuid4().hex + ".json") + state["intent"].update(handoff_receipt=str(receipt_path), started_at=self.clock()) + state["action"] = "restarting" + effects["handoff_receipt"] = str(receipt_path) + atomic(self.path, state) + try: + result = self.transport(state["adapter"], "switch", slot=decision["target"], + expected_revision=native["selection_revision"], reopen=True, **effects) + except Exception as error: + if getattr(error, "code", None) != "handoff_pending": + # A returned failure means this worker has stopped. + # A lost caller instead leaves its durable receipt. + state["intent"] = None if decision["restart_clients"] else state["intent"] + raise + state.update(native=result, intent=None, last_switch_at=self.clock(), action="selected", + target=None, retry_at=self.clock() + 300, launch_failed=result["launch_failed"], failure_count=0) + if operation_id: + state["last_operation_id"] = operation_id + except (HostError, OSError, ValueError, subprocess.SubprocessError) as error: + code = error.code if isinstance(error, HostError) else "native_unavailable" + # Unrelated/remaining owners still require a safe-boundary wait. + # Other repeated failures stop automatic attempts until Refresh. + failures = state.get("failure_count", 0) + (code not in ("waiting_for_idle", "handoff_pending")) + if code in ("other_clients_busy", "client_quit_refused", "client_quit_timeout", "client_changed"): + failures = 3 # Do not repeatedly close/reopen a refused client. + state.update(action=code if code in ("waiting_for_idle", "selection_changed", "adapter_changed", "client_quit_refused", "client_quit_timeout", "client_changed", "handoff_pending") else "unavailable", + code=code, retry_at=self.clock() + (5 if code == "handoff_pending" else min(300, 60 * 2 ** min(failures, 3))), failure_count=failures) + if manual_slot: + state["last_operation_id"] = operation_id + atomic(self.path, state) + return self._projection(state) + + def recover(self, *, rollback: bool, expected_revision: int, expected_selection: str) -> dict: + with self._lock(): + state = self._load() + if state["revision"] != expected_revision or "adapter" not in state: + raise HostError("configuration_changed") + native = self.transport(state["adapter"], "status") + intent = state.get("intent") or {} + if intent.get("handoff_receipt") and not native["recovery_required"] and not rollback: + if native["selection_revision"] != expected_selection: + raise HostError("selection_changed") + path = Path(intent["handoff_receipt"]) + if not path.exists(): + marker = path.with_suffix(".pid.json") + if not marker.exists(): + raise HostError("handoff_pending") + try: + os.kill(private_read(marker)["pid"], 0) + except ProcessLookupError: + pass + else: + raise HostError("handoff_pending") + else: + native = self.transport(state["adapter"], "rollback" if rollback else "recover", expected_revision=expected_selection) + state.update(native=native, intent=None, action="disabled", retry_at=None, code=None, + policy={**state["policy"], "enabled": False}, revision=state["revision"] + 1, failure_count=0) + atomic(self.path, state) + return self._projection(state) + + def reopen(self, *, expected_selection: str) -> dict: + with self._lock(): + state = self._load() + if "adapter" not in state or state.get("intent"): + raise HostError("handoff_pending") + native = self.transport(state["adapter"], "reopen", expected_revision=expected_selection) + state.update(native=native, launch_failed=False, code=None) + atomic(self.path, state) + return self._projection(state) + + +class Watcher: + """Owned by the serving App (or explicit CLI watch), never an import effect.""" + def __init__(self, provider: NativeAccounts): + self.provider = provider + self.stop = threading.Event() + self.thread: threading.Thread | None = None + self.provider.switch_allowed = lambda: not self.stop.is_set() + + def start(self): + if self.thread is None: + self.thread = threading.Thread(target=self.run, daemon=True, name="codex-account-observer") + self.thread.start() + return self + + def run(self): + while not self.stop.is_set(): + delay = 60.0 + try: + value = self.provider.tick() + due = value.get("retry_at") + delay = max(5, min(300, due - time.time())) if due else 60 + except (HostError, OSError, ValueError, subprocess.SubprocessError): + delay = 60 + self.stop.wait(delay) + + def close(self): + self.stop.set() + if self.thread: + self.thread.join(timeout=2) + + +def main() -> int: + parser = argparse.ArgumentParser(prog="loopx-codex-accounts") + parser.add_argument("operation", choices=("status", "configure", "refresh", "switch", "recover", "rollback", "reopen", "watch")) + parser.add_argument("--runtime-root", type=Path, default=Path.home() / ".loopx") + args = parser.parse_args() + provider = NativeAccounts(args.runtime_root) + try: + if args.operation == "watch": + watcher = Watcher(provider) + try: + watcher.run() + except KeyboardInterrupt: + watcher.close() + return 0 + request = {} if args.operation in ("status", "refresh") else json.load(sys.stdin) + if args.operation == "status": + result = provider.status() + elif args.operation == "configure": + result = provider.configure(**request) + elif args.operation in ("refresh", "switch"): + result = provider.tick(refresh=True, **request) + elif args.operation == "reopen": + result = provider.reopen(**request) + else: + result = provider.recover(rollback=args.operation == "rollback", **request) + print(json.dumps({"ok": True, **result})) + return 0 + except Exception as error: + print(json.dumps({"ok": False, "code": error.code if isinstance(error, HostError) else "native_unavailable"})) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native_host.py b/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native_host.py new file mode 100644 index 0000000000..878e7d3b4c --- /dev/null +++ b/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native_host.py @@ -0,0 +1,428 @@ +"""IO adapter to the installed Personal Apps native account owner. + +No copied account writer: the installed owner's locks, idle check, transaction, +readback and recovery remain authoritative. Paths/pins and credentials stay here, +outside policy input and browser projections. Run only through the opt-in host. +""" +from __future__ import annotations + +import hashlib +import importlib +import json +import os +from pathlib import Path +import selectors +import shutil +import subprocess +import sys +import tempfile +import time +import urllib.error +import urllib.request + + +class HostError(Exception): + def __init__(self, code: str): + self.code = code + super().__init__(code) + + +# Address one inspected process, not an App bundle id shared by other routes. +# terminate() requests the normal AppKit quit lifecycle and can be refused. +QUIT = '''ObjC.import("AppKit"); +function run(argv) { + var app; + // Process discovery can precede Launch Services registration. Do not read + // executableURL.path until registration supplies the inspected executable. + var deadline = Date.now() + 5000; + do { + app = $.NSRunningApplication.runningApplicationWithProcessIdentifier(Number(argv[0])); + if (app && app.isTerminated) return; + if (app && app.executableURL && app.isFinishedLaunching) break; + delay(0.05); + } while (Date.now() < deadline); + if (!app || !app.executableURL || !app.isFinishedLaunching) throw Error("Client startup is still incomplete"); + if (ObjC.unwrap(app.executableURL.path) !== argv[1]) throw Error("Client executable changed"); + if (!app.terminate) throw Error("Client refused normal quit"); +}''' + + +def trusted_file(path: Path) -> Path: + path = path.resolve(strict=True) + info = path.stat() + if not path.is_file() or info.st_uid not in (0, os.getuid()) or info.st_mode & 0o022: + raise HostError("adapter_untrusted") + return path + + +def discover() -> dict: + command = shutil.which("personal-apps") + codex = shutil.which("codex") + if not command or not codex: + raise HostError("adapter_unavailable") + source = trusted_file(Path(command)).parent + pins = {name: hashlib.sha256(trusted_file(source / name).read_bytes()).hexdigest() + for name in ("personal_apps.py", "personal_accounts.py")} + return {"source": str(source), "pins": pins, "codex": str(trusted_file(Path(codex)))} + + +class NativeHost: + def __init__(self, adapter: dict, *, user_home: Path | None = None): + source = Path(adapter["source"]) + for name in ("personal_apps.py", "personal_accounts.py"): + if hashlib.sha256(trusted_file(source / name).read_bytes()).hexdigest() != adapter["pins"][name]: + raise HostError("adapter_changed") + self.codex = trusted_file(Path(adapter["codex"])) + # This adapter runs in its own process; module names cannot collide with + # an earlier owner version loaded by another request. + sys.path.insert(0, str(source)) + self.apps = importlib.import_module("personal_apps") + self.accounts = importlib.import_module("personal_accounts") + self.home = user_home or Path.home() + self.spec = self.apps.route(self.home, "gpt") + self.expected = None + self.not_after = None + + def idle(): + if self.not_after is not None and time.time() >= self.not_after: + raise HostError("quota_stale") + try: + self.apps.account_idle(self.home, self.spec) + except (ValueError, OSError, subprocess.SubprocessError): + raise HostError("waiting_for_idle") from None + if self.expected is not None and self.revision() != self.expected: + raise HostError("selection_changed") + + self.manager = self.accounts.Accounts( + self.home, self.spec, idle, lambda: str(self.codex), + lambda selected: self.apps.clean_env(selected, self.home), + ) + + def revision(self) -> str: + # last.json also changes for a manual re-selection of the same slot. + values = [self.accounts.read_private(path) if path.exists() else b"" + for path in (self.manager.active, self.manager.root / "last.json")] + return hashlib.sha256(b"\0".join(values)).hexdigest() + + def status(self) -> dict: + self.apps.check_provider(self.spec, "gpt") + value = self.manager.status() + result = {key: value[key] for key in ( + "slots", "active_slot", "current_login_matches", "recovery_required" + )} | {"selection_revision": self.revision()} + if sys.platform == "darwin": + exe = self.apps.executable(self.apps.app_bundle(self.spec, "gpt", self.home)) + result["client_running"] = self.apps.running(exe, self.spec) is not None + return result + + def close_client(self, expected: str, *, timeout: float = 25) -> bool: + exe = self.apps.executable(self.apps.app_bundle(self.spec, "gpt", self.home)) + pid = self.apps.running(exe, self.spec) + if self.revision() != expected: + raise HostError("selection_changed") + if pid is None: + return False + # Recheck exact ownership immediately before requesting a normal quit. + if self.apps.running(exe, self.spec) != pid: + raise HostError("client_changed") + try: + subprocess.run(["/usr/bin/osascript", "-l", "JavaScript", "-", str(pid), str(exe)], + input=QUIT, text=True, check=True, capture_output=True, timeout=10) + except (OSError, subprocess.SubprocessError): + raise HostError("client_quit_refused") from None + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + current = self.apps.running(exe, self.spec) + if current is None: + return True + if current != pid: + raise HostError("client_changed") + time.sleep(0.1) + raise HostError("client_quit_timeout") + + def reopen(self, expected: str) -> dict: + value = self.status() + if value["selection_revision"] != expected or value["recovery_required"] or value["current_login_matches"] is not True: + raise HostError("selection_changed") + self.apps.open_app(self.home, "gpt", []) + return self.status() + + def switch(self, slot: str, expected: str, *, reopen: bool, restart_clients: bool = False, + not_after: float | None = None) -> dict: + if sys.platform != "darwin": + raise HostError("platform_unavailable") + self.expected = expected + self.not_after = not_after + before = self.status() + if before["selection_revision"] != expected or before["recovery_required"] or before["current_login_matches"] is not True: + raise HostError("selection_changed") + # idle() is rechecked by the native owner under launch/account locks, + # and immediately before its first mutation. This is the CAS fence, + # including against manual selections outside LoopX. + try: + if restart_clients: + if not_after is not None and time.time() >= not_after: + raise HostError("quota_stale") + self.close_client(expected) + # Children may flush and exit just after their App root. An + # unrelated CLI/file owner remains a hold, never a kill target. + deadline = time.monotonic() + 10 + while True: + try: + self.manager.idle() + break + except HostError as error: + if error.code != "waiting_for_idle": + raise + if time.monotonic() >= deadline: + raise HostError("other_clients_busy") from None + time.sleep(0.2) + result = self.manager.switch(slot) + except Exception as error: + # A refused/failed transaction must not unnecessarily leave a + # normally closed source App down. Native preflight blocks opening + # an incomplete credential journal; a later manual choice wins. + if restart_clients and before.get("client_running") and getattr(error, "code", None) != "client_changed": + try: + if self.status().get("client_running") is False: + self.reopen(expected) + except (HostError, OSError, ValueError, subprocess.SubprocessError): + pass + raise + value = self.status() + if value["active_slot"] != slot or value["current_login_matches"] is not True: + raise HostError("readback_failed") + launch_failed = False + if reopen: + try: + self.apps.open_app(self.home, "gpt", []) + except (OSError, ValueError, subprocess.SubprocessError): + launch_failed = True + after = self.status() + if after["selection_revision"] != value["selection_revision"]: + raise HostError("selection_changed") + value = after + return value | {"history_untouched": result["history_untouched"], "launch_failed": launch_failed} + + def recover(self, rollback: bool, expected: str) -> dict: + if sys.platform != "darwin": + raise HostError("platform_unavailable") + self.expected = expected + self.manager.recover(rollback) + return self.status() + + def observe(self, slots: list[str], expected: str, auth_failures: dict | None = None) -> dict: + before = self.status() + if before["selection_revision"] != expected or before["recovery_required"]: + raise HostError("selection_changed") + if not set(slots) <= set(before["slots"]) or len(slots) > 16: + raise HostError("invalid_slots") + rows = [] + for slot in slots: + path = self.manager.auth if slot == before["active_slot"] else self.manager.slot(slot) + try: + raw = self.accounts.read_private(path) + revision = hashlib.sha256(raw).hexdigest() + failed = (auth_failures or {}).get(slot, {}) + if slot != before["active_slot"] and failed.get("revision") == revision: + rows.append({"slot": slot, "observed_at": failed["observed_at"], + "status": "auth_required", "credential_revision": revision}) + continue + usage = read_usage(raw) + except HostError as error: + if error.code == "auth_required" and slot != before["active_slot"]: + try: + usage = self.refresh_inactive(slot, expected) + except (HostError, ValueError, OSError, subprocess.SubprocessError) as refresh_error: + auth_required = not isinstance(refresh_error, HostError) or refresh_error.code == "auth_required" + rows.append({"slot": slot, "observed_at": time.time(), + "status": "auth_required" if auth_required else "unavailable", + **({"credential_revision": revision} if auth_required else {})}) + continue + else: + rows.append({"slot": slot, "observed_at": time.time(), + "status": "auth_required" if error.code == "auth_required" else "unavailable"}) + continue + rows.append({"slot": slot, "observed_at": time.time(), "status": "ok", "usage": usage}) + after = self.status() + if after["selection_revision"] != expected: + raise HostError("selection_changed") + return {"observations": rows, "native": after} + + def refresh_inactive(self, slot: str, expected: str) -> dict: + # Never refresh a copy of the active account: its running native client + # owns refresh-token rotation. Distinct inactive accounts can refresh in + # isolation, serialized with the native launch/enrol/switch owner. + with self.manager.lock(): + status = self.status() + if status["selection_revision"] != expected or slot == status["active_slot"]: + raise HostError("selection_changed") + target = self.manager.slot(slot) + original = self.accounts.read_private(target) + active = self.accounts.read_private(self.manager.auth) + if self.accounts.identity(original) == self.accounts.identity(active): + raise HostError("duplicate_account") + self.accounts.private_dir(self.manager.root / "quota-probes") + with tempfile.TemporaryDirectory(dir=self.manager.root / "quota-probes") as temp: + home = Path(temp) + self.accounts.atomic(home / "auth.json", original) + self.accounts.atomic(home / "config.toml", b'cli_auth_credentials_store="file"\nforced_login_method="chatgpt"\n') + usage = app_server_usage(str(self.codex), self.apps.clean_env(dict(self.spec, home=home), self.home)) + refreshed = self.accounts.read_private(home / "auth.json") + if self.accounts.identity(original) != self.accounts.identity(refreshed): + raise HostError("auth_identity_changed") + if self.accounts.read_private(target) != original or self.revision() != expected: + raise HostError("selection_changed") + if refreshed != original: + self.accounts.atomic(target, refreshed) + if self.accounts.read_private(target) != refreshed: + raise HostError("readback_failed") + return usage + + +class NoRedirect(urllib.request.HTTPRedirectHandler): + def redirect_request(self, *_args, **_kwargs): + raise HostError("quota_unavailable") + + +def quota_fields(value: dict) -> dict: + """Allowlist transport fields before private persistence/TS invocation.""" + def limits(raw): + if not isinstance(raw, dict): + return None + result = {k: raw[k] for k in ("allowed", "limit_reached") if k in raw} + for key in ("primary", "secondary", "primary_window", "secondary_window"): + item = raw.get(key) + if isinstance(item, dict): + result[key] = {k: item[k] for k in ("usedPercent", "resetsAt", "used_percent", "reset_at") if k in item} + if isinstance(raw.get("credits"), dict): + result["credits"] = {k: raw["credits"][k] for k in + ("hasCredits", "has_credits", "balance", "unlimited") if k in raw["credits"]} + return result + + result = {k: limits(value[k]) for k in ("rateLimits", "rate_limit") if k in value} + if isinstance(value.get("rateLimitsByLimitId"), dict): + result["rateLimitsByLimitId"] = {"codex": limits(value["rateLimitsByLimitId"].get("codex"))} + if isinstance(value.get("credits"), dict): + result["credits"] = limits({"credits": value["credits"]})["credits"] + return result + + +def read_usage(raw: bytes) -> dict: + """Native Codex usage observation only. Never send prompts or model calls.""" + try: + data = json.loads(raw) + tokens = data["tokens"] + if data.get("auth_mode") not in (None, "chatgpt") or data.get("OPENAI_API_KEY"): + raise HostError("auth_required") + token, account_id = tokens["access_token"], tokens["account_id"] + if not isinstance(token, str) or not isinstance(account_id, str) or not token or not account_id: + raise HostError("auth_required") + request = urllib.request.Request("https://chatgpt.com/backend-api/wham/usage", headers={ + "Authorization": "Bearer " + token, "chatgpt-account-id": account_id, + "User-Agent": "codex-cli", "Accept": "application/json", + }) + with urllib.request.build_opener(NoRedirect()).open(request, timeout=10) as response: + encoded = response.read(131073) + if len(encoded) > 131072: + raise HostError("quota_unavailable") + value = json.loads(encoded) + if not isinstance(value, dict): + raise HostError("quota_unavailable") + # Discard account identity and unrelated product/organization metadata. + return quota_fields(value) + except urllib.error.HTTPError as error: + raise HostError("auth_required" if error.code in (401, 403) else "quota_unavailable") from None + except (KeyError, TypeError, ValueError, OSError): + raise HostError("quota_unavailable") from None + + +def app_server_usage(binary: str, env: dict) -> dict: + process = subprocess.Popen([binary, "app-server"], env=env, stdin=subprocess.PIPE, + stdout=subprocess.PIPE, stderr=subprocess.DEVNULL) + selector = selectors.DefaultSelector() + assert process.stdout and process.stdin + selector.register(process.stdout, selectors.EVENT_READ) + pending = b"" + deadline = time.monotonic() + 25 + total = 0 + + def call(index, method, params): + nonlocal pending, total + process.stdin.write(json.dumps({"id": index, "method": method, "params": params}).encode() + b"\n") + process.stdin.flush() + while time.monotonic() < deadline: + if not selector.select(max(0, deadline - time.monotonic())): + break + chunk = os.read(process.stdout.fileno(), 65536) + total += len(chunk) + if not chunk or total > 1048576: + break + pending += chunk + while b"\n" in pending: + line, pending = pending.split(b"\n", 1) + packet = json.loads(line) + if packet.get("id") == index: + if "error" in packet: + raise HostError("auth_required") + return packet["result"] + raise HostError("quota_unavailable") + + try: + call(1, "initialize", {"clientInfo": {"name": "loopx_native_accounts", "version": "1"}}) + process.stdin.write(b'{"method":"initialized"}\n') + process.stdin.flush() + call(2, "account/read", {"refreshToken": True}) + return quota_fields(call(3, "account/rateLimits/read", {})) + finally: + selector.close() + process.stdin.close() + try: + process.wait(timeout=2) + except subprocess.TimeoutExpired: + process.terminate() # Only this disposable observer process. + try: + process.wait(timeout=2) + except subprocess.TimeoutExpired: + process.kill() + process.wait() + process.stdout.close() + + +def main() -> int: + if len(sys.argv) == 3 and sys.argv[1] == "--result-file": + from .native import SCHEMA, atomic + atomic(Path(sys.argv[2]).with_suffix(".pid.json"), + {"schema_version": SCHEMA, "pid": os.getpid()}) + try: + request = json.load(sys.stdin) + host = NativeHost(request["adapter"]) + operation = request["operation"] + if operation == "status": + value = host.status() + elif operation == "observe": + value = host.observe(request["slots"], request["expected_revision"], request.get("auth_failures")) + elif operation == "switch": + value = host.switch(request["slot"], request["expected_revision"], reopen=request["reopen"], + restart_clients=request.get("restart_clients", False), not_after=request.get("not_after")) + elif operation == "reopen": + value = host.reopen(request["expected_revision"]) + elif operation in ("recover", "rollback"): + value = host.recover(operation == "rollback", request["expected_revision"]) + else: + raise HostError("unsupported_operation") + packet = {"ok": True, "result": value} + except Exception as error: # Native diagnostics can contain paths/identities. + packet = {"ok": False, "code": error.code if isinstance(error, HostError) else "native_unavailable"} + if len(sys.argv) == 3 and sys.argv[1] == "--result-file": + # A detached worker completes even if its calling Codex App goes away. + # The private receipt lets a restarted observer reconcile, not replay. + from .native import SCHEMA, atomic + atomic(Path(sys.argv[2]), {"schema_version": SCHEMA, **packet}) + else: + print(json.dumps(packet)) + return 0 if packet["ok"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native_policy.ts b/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native_policy.ts new file mode 100644 index 0000000000..74de49ad5f --- /dev/null +++ b/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native_policy.ts @@ -0,0 +1,101 @@ +/** Native account availability and handoff policy. No credentials or effects here. + * These states belong to this optional provider, not the LoopX quota ledger. + */ +export type Availability = "available" | "exhausted" | "unknown" | "auth_required"; +export type Observation = { + slot: string; + observed_at: number; + status: "ok" | "auth_required" | "unavailable"; + usage?: Record; +}; +export type Account = { + slot: string; + availability: Availability; + observed_at: number; + reset_at: number | null; + primary_used: number | null; + secondary_used: number | null; + credits_available: boolean | null; +}; +export type Policy = { enabled: boolean; slots: string[]; cooldown_seconds: number; restart_clients?: boolean }; + +const number = (v: unknown): v is number => typeof v === "number" && Number.isFinite(v); +const percent = (v: unknown): number | null => number(v) && v >= 0 && v <= 100 ? v : null; +const epoch = (v: unknown): number | null => number(v) && v > 0 ? v : null; + +export function account(observation: Observation, now: number): Account { + const result: Account = { slot: observation.slot, availability: "unknown", + observed_at: observation.observed_at, reset_at: null, primary_used: null, + secondary_used: null, credits_available: null }; + if (observation.status === "auth_required") return { ...result, availability: "auth_required" }; + if (!number(observation.observed_at) || now < observation.observed_at || now - observation.observed_at > 120) return result; + if (observation.status !== "ok" || !observation.usage) return result; + const usage = observation.usage; + // App Server's multi-bucket view wins. Review/other product buckets do not + // establish Codex availability. WHAM is the native read-only HTTP fallback. + const limits = usage.rateLimitsByLimitId != null ? usage.rateLimitsByLimitId.codex + : usage.rateLimits ?? usage.rate_limit; + if (!limits || typeof limits !== "object") return result; + const primary = limits.primary ?? limits.primary_window; + const secondary = limits.secondary ?? limits.secondary_window; + const windows = [primary, secondary].filter(v => v != null); + const values = windows.map(v => ({ used: percent(v.usedPercent ?? v.used_percent), + reset: epoch(v.resetsAt ?? v.reset_at) })); + result.primary_used = primary ? percent(primary.usedPercent ?? primary.used_percent) : null; + result.secondary_used = secondary ? percent(secondary.usedPercent ?? secondary.used_percent) : null; + const credits = limits.credits ?? usage.credits; + const balance = typeof credits?.balance === "string" && credits.balance.trim() !== "" + ? Number(credits.balance) : credits?.balance; + result.credits_available = credits?.unlimited === true || credits?.hasCredits === true + || credits?.has_credits === true || (number(balance) && balance > 0) ? true + : credits && (credits.hasCredits === false || credits.has_credits === false || balance === 0) ? false : null; + // A reset passing makes an old 100% sample unknown, never a new allowance. + if (values.some(v => v.used === null || (v.used === 100 && (v.reset === null || v.reset <= now)))) return result; + const blocked = values.filter(v => v.used === 100); + if (blocked.length) { + result.reset_at = Math.max(...blocked.map(v => v.reset!)); + // Subscription exhaustion alone does not prove a credit-funded account is + // unusable. This observer never buys credits or invokes a paid reset. + result.availability = result.credits_available === true ? "available" + : result.credits_available === false ? "exhausted" : "unknown"; + } else if (values.length && limits.allowed !== false && limits.limit_reached !== true) { + result.availability = "available"; + } + return result; +} + +export function decide(input: { + now: number; policy: Policy; active: string | null; observations: Observation[]; + last_switch_at: number | null; recovery_required: boolean; login_matches: boolean; + manual_slot?: string; +}) { + const accounts = input.observations.map(o => account(o, input.now)); + const output = (action: string, target: string | null = null, retry_at: number | null = null) => + ({ action, target, retry_at, accounts, + restart_clients: action === "switch" && input.policy.restart_clients === true, + valid_until: action === "switch" && !input.manual_slot + ? Math.min(...accounts.filter(a => a.slot === input.active || a.slot === target).map(a => a.observed_at + 120)) + : null }); + if (input.recovery_required) return output("recovery_required"); + if (!input.login_matches || !input.active) return output("selection_changed"); + if (!input.manual_slot && !input.policy.enabled) return output("disabled"); + const current = accounts.find(a => a.slot === input.active); + if (!input.manual_slot && current?.availability !== "exhausted") { + return output(current?.availability === "available" ? "ready" + : current?.availability === "auth_required" ? "auth_required" : "quota_unknown", null, input.now + 60); + } + const cooldown = input.last_switch_at == null ? 0 : input.last_switch_at + input.policy.cooldown_seconds; + if (!input.manual_slot && cooldown > input.now) return output("cooldown", null, cooldown); + const allowed = input.policy.slots; + const index = allowed.indexOf(input.active); + const order = [...allowed.slice(index + 1), ...allowed.slice(0, index + 1)]; + const target = (input.manual_slot ? [input.manual_slot] : order).find(slot => + slot !== input.active && allowed.includes(slot) && (input.manual_slot + || accounts.some(a => a.slot === slot && a.availability === "available"))); + if (target) return output("switch", target); + const resets = accounts.filter(a => allowed.includes(a.slot) && a.availability === "exhausted") + .flatMap(a => a.reset_at && a.reset_at > input.now ? [a.reset_at] : []); + const uncertain = accounts.some(a => allowed.includes(a.slot) && a.availability === "unknown"); + return output(input.manual_slot ? "target_unavailable" : "all_unavailable", null, + Math.min(input.now + (uncertain ? 60 : 300), ...(resets.length ? resets : [Infinity]))); +} diff --git a/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native_policy_bridge.ts b/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native_policy_bridge.ts new file mode 100644 index 0000000000..6c5bdea9f1 --- /dev/null +++ b/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native_policy_bridge.ts @@ -0,0 +1,7 @@ +import { decide } from "./native_policy.ts"; +let raw = ""; +for await (const chunk of process.stdin) { + raw += chunk; + if (raw.length > 131072) throw Error("Policy request exceeds limit"); +} +process.stdout.write(JSON.stringify(decide(JSON.parse(raw)))); diff --git a/packages/loopx-codex-provider-routing/templates/codex-app-config.toml b/packages/loopx-codex-provider-routing/templates/codex-app-config.toml index a30ea10d97..467056d817 100644 --- a/packages/loopx-codex-provider-routing/templates/codex-app-config.toml +++ b/packages/loopx-codex-provider-routing/templates/codex-app-config.toml @@ -1,22 +1,6 @@ -# Public-safe field shape only. Replace placeholders in an operator-owned -# private config; this extension never writes a Codex home. -model = "gpt-5.6-sol" -model_provider = "local-cpa" -model_catalog_json = "" -service_tier = "default" - -[features] -# Keep the native feature available for hosts that can render it. The explicit -# fast/ catalog rows do not depend on the desktop Fast button. -fast_mode = true - -[model_providers.local-cpa] -name = "CPA · Codex A → B → text fallback" -base_url = "http://127.0.0.1:/v1" -wire_api = "responses" -requires_openai_auth = false -supports_websockets = false -request_max_retries = 30 -stream_max_retries = 30 -# The catalog compiler keeps ordinary rows on `default`; only generated Fast -# sibling rows carry `default_service_tier = "fast"`. +# Native OpenAI login. Retain the user's chosen model and unrelated settings. +# Account effects are separately enabled in LoopX; this template enables none. +model_provider = "openai" +cli_auth_credentials_store = "file" +forced_login_method = "chatgpt" +# No custom model catalog, proxy endpoint, profile override or model fallback. diff --git a/packages/loopx-codex-provider-routing/templates/legacy-cpa-config.toml b/packages/loopx-codex-provider-routing/templates/legacy-cpa-config.toml new file mode 100644 index 0000000000..05f5f3444a --- /dev/null +++ b/packages/loopx-codex-provider-routing/templates/legacy-cpa-config.toml @@ -0,0 +1,22 @@ +# Legacy CPA field shape only. Replace placeholders in an operator-owned +# private config; this extension never writes a Codex home. +model = "gpt-5.6-sol" +model_provider = "local-cpa" +model_catalog_json = "" +service_tier = "default" + +[features] +# Keep the native feature available for hosts that can render it. The explicit +# fast/ catalog rows do not depend on the desktop Fast button. +fast_mode = true + +[model_providers.local-cpa] +name = "CPA · Codex A → B → text fallback" +base_url = "http://127.0.0.1:/v1" +wire_api = "responses" +requires_openai_auth = false +supports_websockets = false +request_max_retries = 30 +stream_max_retries = 30 +# The catalog compiler keeps ordinary rows on `default`; only generated Fast +# sibling rows carry `default_service_tier = "fast"`. diff --git a/tests/presentation/test_codex_accounts.py b/tests/presentation/test_codex_accounts.py new file mode 100644 index 0000000000..b13d4f6b82 --- /dev/null +++ b/tests/presentation/test_codex_accounts.py @@ -0,0 +1,559 @@ +"""Account handoff outcomes, using the shipped TS decision and private IO store. + +Synthetic native transport isolates model/network/auth effects. Installed native +transaction qualification is a separate opt-in drill, never a live-account test. +""" +import importlib +import json +import os +import plistlib +import shutil +from pathlib import Path +import subprocess +import sys +import threading +import time +from types import SimpleNamespace + +import pytest + +pytestmark = pytest.mark.skipif(os.name != "posix", reason="Native slot effects use the POSIX account owner") + + +@pytest.mark.skipif(sys.platform != "darwin" or shutil.which("swiftc") is None, + reason="Actual AppKit registration requires macOS and Swift") +def test_normal_quit_waits_for_owned_cold_app_registration(native, tmp_path): + """A PID can exist before NSRunningApplication supplies executableURL.""" + bundle = tmp_path / "ColdFixture.app" + executable = bundle / "Contents/MacOS/ColdFixture" + executable.parent.mkdir(parents=True) + (bundle / "Contents/Info.plist").write_bytes(plistlib.dumps({ + "CFBundleExecutable": "ColdFixture", "CFBundleIdentifier": "dev.loopx.cold-fixture", + "CFBundleName": "ColdFixture", "CFBundlePackageType": "APPL", + })) + source = tmp_path / "fixture.swift" + source.write_text('''import Cocoa +Thread.sleep(forTimeInterval: 1.0) +let app = NSApplication.shared +app.setActivationPolicy(.accessory) +app.run() +''') + subprocess.run(["swiftc", str(source), "-framework", "Cocoa", "-o", str(executable)], + check=True, capture_output=True, timeout=30) + host_type = importlib.import_module("loopx_codex_provider_routing.native_host").NativeHost + host = object.__new__(host_type) + host.home, host.spec = tmp_path, {} + host.revision = lambda: "fixture-selection" + host.apps = SimpleNamespace( + app_bundle=lambda *_: bundle, executable=lambda _: executable, + running=lambda *_: process.pid if process.poll() is None else None, + ) + process = subprocess.Popen([str(executable)], start_new_session=True) + try: + assert host.close_client("fixture-selection", timeout=10) is True + assert process.wait(timeout=5) == 0 + finally: + # Only this test's Popen child may be cleaned up after a failed oracle. + if process.poll() is None: + process.terminate() + process.wait(timeout=5) + + +@pytest.fixture +def native(monkeypatch): + monkeypatch.syspath_prepend(str(Path(__file__).resolve().parents[2] / "packages" / + "loopx-codex-provider-routing" / "src")) + module = importlib.import_module("loopx_codex_provider_routing.native") + monkeypatch.setattr(module, "discover", lambda: {"synthetic": True}) + return module + + +class Clock: + now = 1000 + + def __call__(self): + return self.now + + +class NativeTransport: + def __init__(self, clock): + self.clock = clock + self.active = "a" + self.revision = "initial" + self.calls = [] + self.busy = False + self.pending = False + self.bad_quota = False + self.committed_without_reply = False + self.entered = self.release = None + + def __call__(self, adapter, operation, **parameters): + from loopx_codex_provider_routing.native_host import HostError + + self.calls.append((operation, parameters)) + if operation == "status": + return {"slots": ["a", "b"], "active_slot": self.active, "selection_revision": self.revision, + "current_login_matches": True, "recovery_required": self.pending} + if operation == "observe": + if self.entered: + self.entered.set() + assert self.release.wait(10) + assert parameters["expected_revision"] == self.revision + return {"observations": [{"slot": slot, "status": "unavailable" if self.bad_quota else "ok", + "observed_at": self.clock(), "usage": { + "rate_limit": {"primary_window": {"used_percent": 100 if slot == "a" else 12, + "reset_at": 8000}}, + "credits": {"has_credits": False}, + }} for slot in parameters["slots"]]} + if operation == "switch": + if parameters["expected_revision"] != self.revision: + raise HostError("selection_changed") + if self.busy: + raise HostError("waiting_for_idle") + self.active, self.revision = parameters["slot"], "selected" + if self.committed_without_reply: + raise HostError("native_unavailable") + return self(adapter, "status") | {"launch_failed": False, "history_untouched": True} + if operation in ("recover", "rollback"): + assert parameters["expected_revision"] == self.revision + self.pending = False + return self(adapter, "status") + if operation == "reopen": + assert parameters["expected_revision"] == self.revision + return self(adapter, "status") | {"client_running": True} + raise AssertionError(operation) + + +def configured(native, tmp_path): + clock = Clock() + transport = NativeTransport(clock) + provider = native.NativeAccounts(tmp_path, transport=transport, clock=clock) + provider.configure(expected_revision=0, enabled=True, slots=["a", "b"]) + return provider, transport, clock + + +def test_default_off_has_no_native_effect_or_startup_observer(native, tmp_path): + from loopx.presentation.codex_accounts_api import CodexAccountService + + assert not native.NativeAccounts(tmp_path).path.exists() + service = CodexAccountService(tmp_path) + assert service.watcher.thread is None + assert not native.NativeAccounts(tmp_path).root.exists() + service.close() + + +def test_exhaustion_selects_once_and_manual_retry_is_idempotent(native, tmp_path): + provider, transport, clock = configured(native, tmp_path) + result = provider.tick(manual_slot="b", expected_selection="initial", operation_id="once") + assert result["active_slot"] == "b" + assert result["action"] == "selected" + assert provider.tick(manual_slot="b", expected_selection="initial", operation_id="once")["active_slot"] == "b" + assert len([call for call in transport.calls if call[0] == "switch"]) == 1 + assert provider.path.stat().st_mode & 0o077 == 0 + assert "access_token" not in json.dumps(result) + clock.now += 301 + assert provider.tick()["action"] == "ready" + + +def test_unknown_quota_is_not_exhaustion(native, tmp_path): + provider, transport, _ = configured(native, tmp_path) + transport.bad_quota = True + assert provider.tick()["action"] == "quota_unknown" + assert not [call for call in transport.calls if call[0] == "switch"] + + +def test_busy_waits_then_switches_at_safe_boundary(native, tmp_path): + provider, transport, clock = configured(native, tmp_path) + transport.busy = True + result = provider.tick() + assert result["action"] == "waiting_for_idle" and result["active_slot"] == "a" + assert json.loads(provider.path.read_text())["intent"]["target"] == "b" + transport.busy = False + clock.now += 61 + result = provider.tick() + assert result["active_slot"] == "b" and result["action"] == "selected" + assert all(call[0] not in ("kill", "terminate") for call in transport.calls) + + +def test_manual_native_selection_wins_over_queued_automatic_choice(native, tmp_path): + provider, transport, clock = configured(native, tmp_path) + transport.busy = True + provider.tick() + transport.revision = "external-choice" + clock.now += 61 + before = len(transport.calls) + assert provider.tick()["action"] == "selection_changed" + assert all(call[0] != "switch" for call in transport.calls[before:]) + result = provider.tick(manual_slot="b", expected_selection="initial", operation_id="stale") + assert result["action"] == "selection_changed" + assert transport.active == "a" + + +def test_lost_reply_recovers_readback_without_replaying_switch(native, tmp_path): + provider, transport, clock = configured(native, tmp_path) + transport.committed_without_reply = True + assert provider.tick()["action"] == "unavailable" + restarted = native.NativeAccounts(tmp_path, transport=transport, clock=clock) + result = restarted.tick(refresh=True) + assert result["active_slot"] == "b" and result["action"] == "selected" + assert len([call for call in transport.calls if call[0] == "switch"]) == 1 + + +def test_recovery_stops_auto_and_preserves_revision_fence(native, tmp_path): + provider, transport, _ = configured(native, tmp_path) + transport.pending = True + assert provider.tick()["action"] == "recovery_required" + result = provider.recover(rollback=False, expected_revision=1, expected_selection="initial") + assert result["policy"]["enabled"] is False + assert not result["recovery_required"] + with pytest.raises(Exception, match="configuration_changed"): + provider.configure(expected_revision=1, enabled=True, slots=["a", "b"]) + + +def test_two_hosts_cannot_switch_concurrently(native, tmp_path): + provider, transport, clock = configured(native, tmp_path) + transport.entered, transport.release = threading.Event(), threading.Event() + outcome = [] + thread = threading.Thread(target=lambda: outcome.append(provider.tick())) + thread.start() + assert transport.entered.wait(5) + contender = native.NativeAccounts(tmp_path, transport=transport, clock=clock) + try: + with pytest.raises(BlockingIOError): + contender.tick() + finally: + transport.release.set() + thread.join(10) + assert outcome[0]["active_slot"] == "b" + assert len([call for call in transport.calls if call[0] == "switch"]) == 1 + + +def test_disable_and_stop_before_effect_preserve_login(native, tmp_path): + provider, transport, _ = configured(native, tmp_path) + provider.switch_allowed = lambda: False + provider.tick() + assert transport.active == "a" + provider.configure(expected_revision=1, enabled=False, slots=["a", "b"]) + before = len(transport.calls) + assert provider.tick()["action"] == "disabled" + assert len(transport.calls) == before + # Turning automatic effects off does not revoke an explicit manual choice. + result = provider.tick(manual_slot="b", expected_selection="initial", operation_id="manual-after-disable") + assert result["active_slot"] == "b" and result["policy"]["enabled"] is False + + +def test_first_explicit_check_does_not_enable_automatic_switches(native, tmp_path): + clock = Clock() + transport = NativeTransport(clock) + provider = native.NativeAccounts(tmp_path, transport=transport, clock=clock) + result = provider.tick(refresh=True) + assert result["policy"]["enabled"] is False and result["accounts"] + assert not [call for call in transport.calls if call[0] == "switch"] + + +def test_unknown_upstream_fields_do_not_reach_policy_or_persistence(native): + from loopx_codex_provider_routing.native_host import quota_fields + + result = quota_fields({"access_token": "synthetic-only", "email": "person@example.invalid", + "rate_limit": {"primary_window": {"used_percent": 10, "secret": "synthetic-only"}}, + "credits": {"has_credits": False, "refresh_token": "synthetic-only"}}) + assert "synthetic-only" not in json.dumps(result) + assert result["rate_limit"]["primary_window"]["used_percent"] == 10 + + +def test_rejected_inactive_login_is_not_repeatedly_refreshed(native, tmp_path, monkeypatch): + from loopx_codex_provider_routing import native_host as module + + active, inactive = tmp_path / "active", tmp_path / "inactive" + active.write_bytes(b"synthetic-active") + inactive.write_bytes(b"synthetic-expired") + adapter = module.NativeHost.__new__(module.NativeHost) + adapter.manager = SimpleNamespace(auth=active, slot=lambda _: inactive) + adapter.accounts = SimpleNamespace(read_private=lambda path: path.read_bytes()) + adapter.status = lambda: {"slots": ["a", "b"], "active_slot": "a", + "selection_revision": "stable", "recovery_required": False} + calls = [] + + def usage(raw): + if raw == b"synthetic-active": + return {} + raise module.HostError("auth_required") + + def refresh(*args): + calls.append(args) + raise module.HostError("auth_required") + + monkeypatch.setattr(module, "read_usage", usage) + adapter.refresh_inactive = refresh + result = adapter.observe(["a", "b"], "stable") + failed = result["observations"][1] + cached = {"b": {"revision": failed["credential_revision"], "observed_at": failed["observed_at"]}} + assert adapter.observe(["a", "b"], "stable", cached)["observations"][1] == failed + assert len(calls) == 1 + inactive.write_bytes(b"synthetic-replaced-login") + adapter.observe(["a", "b"], "stable", cached) + assert len(calls) == 2 + + +def test_source_pins_and_symlink_state_are_not_silently_adopted(native, tmp_path): + from loopx_codex_provider_routing.native_host import NativeHost, HostError + + (tmp_path / "personal_apps.py").write_text("# synthetic owner\n") + with pytest.raises(HostError, match="adapter_changed"): + NativeHost({"source": str(tmp_path), "pins": {"personal_apps.py": "bad"}, "codex": "unused"}) + provider = native.NativeAccounts(tmp_path) + provider.root.mkdir(parents=True) + provider.path.symlink_to(tmp_path / "personal_apps.py") + with pytest.raises(HostError, match="state_untrusted"): + provider._load() + + +def test_http_adapter_rejects_secret_and_arbitrary_command_fields(native, tmp_path): + from loopx.presentation.codex_accounts_api import CodexAccountsRequestMixin + + class Handler(CodexAccountsRequestMixin): + def _read_json(self): + return {"operation": "configure", "parameters": {"access_token": "synthetic-only"}} + + def _send_json(self, payload, *, status=200): + self.sent, self.code = payload, status + + handler = Handler() + handler._codex_accounts_write() + assert handler.code == 409 and handler.sent["code"] == "invalid_request" + assert "synthetic-only" not in json.dumps(handler.sent) + + +def test_client_restart_requires_explicit_opt_in_and_old_policy_stays_off(native, tmp_path): + provider, transport, _ = configured(native, tmp_path) + assert provider.tick()["action"] == "selected" + first = next(args for op, args in transport.calls if op == "switch") + assert first["restart_clients"] is False and "handoff_receipt" not in first + assert first["not_after"] == 1120 + # Older saved policies without this new permission remain wait-only. + state = provider._load() + state["policy"].pop("restart_clients") + native.atomic(provider.path, state) + assert provider.configure(expected_revision=1, enabled=True, slots=["a", "b"])["policy"]["restart_clients"] is False + with pytest.raises(Exception, match="invalid_policy"): + provider.configure(expected_revision=2, enabled=True, slots=["a", "b"], restart_clients="yes") + + +def test_opted_in_handoff_has_private_receipt_and_no_replayed_switch(native, tmp_path): + provider, transport, clock = configured(native, tmp_path) + provider.configure(expected_revision=1, enabled=True, slots=["a", "b"], restart_clients=True) + original = transport.__call__ + effect = {} + + def pending(adapter, operation, **parameters): + if operation == "switch": + effect.update(parameters) + raise native.HostError("handoff_pending") + return original(adapter, operation, **parameters) + + provider.transport = pending + assert provider.tick()["action"] == "handoff_pending" + assert effect["restart_clients"] is True + assert Path(effect["handoff_receipt"]).parent.stat().st_mode & 0o077 == 0 + clock.now += 61 + assert provider.tick()["action"] == "restarting" + # Only the completed worker receipt can reconcile the pending operation. + transport.active, transport.revision = "b", "selected" + native.atomic(Path(effect["handoff_receipt"]), {"schema_version": native.SCHEMA, + "ok": True, "result": {"launch_failed": True}}) + result = provider.tick(refresh=True) + assert result["action"] == "selected" and result["launch_failed"] + before = len(transport.calls) + provider.transport = transport + assert provider.reopen(expected_selection="selected")["launch_failed"] is False + assert [op for op, _ in transport.calls[before:]] == ["reopen", "status"] + + +def test_unresolved_worker_never_replays_or_reconfigures_an_enabled_handoff(native, tmp_path): + provider, transport, clock = configured(native, tmp_path) + provider.configure(expected_revision=1, enabled=True, slots=["a", "b"], restart_clients=True) + original = transport.__call__ + switches = [] + + def pending(adapter, operation, **parameters): + if operation == "switch": + switches.append(parameters) + raise native.HostError("handoff_pending") + return original(adapter, operation, **parameters) + + provider.transport = pending + provider.tick() + clock.now += 181 + assert provider.tick(refresh=True)["recovery_required"] + assert len(switches) == 1 + with pytest.raises(Exception, match="handoff_pending"): + provider.configure(expected_revision=2, enabled=True, slots=["a", "b"], reconnect=True) + # Disabling cannot discard a worker that may be committing native state. + provider.configure(expected_revision=2, enabled=False, slots=["a", "b"]) + assert provider._load()["intent"]["handoff_receipt"] == switches[0]["handoff_receipt"] + assert provider.tick(refresh=True)["action"] == "handoff_interrupted" + with pytest.raises(Exception, match="handoff_pending"): + provider.recover(rollback=False, expected_revision=3, expected_selection="initial") + # Recovery can retire a proven exited worker without inventing a native + # credential rollback when the worker never began a credential journal. + child = subprocess.Popen([sys.executable, "-c", "pass"]) + child.wait(timeout=5) + marker = Path(switches[0]["handoff_receipt"]).with_suffix(".pid.json") + native.atomic(marker, {"schema_version": native.SCHEMA, "pid": child.pid}) + result = provider.recover(rollback=False, expected_revision=3, expected_selection="initial") + assert result["policy"]["enabled"] is False and provider._load()["intent"] is None + + +def test_refused_client_is_not_repeatedly_closed_in_the_background(native, tmp_path): + provider, transport, clock = configured(native, tmp_path) + provider.configure(expected_revision=1, enabled=True, slots=["a", "b"], restart_clients=True) + calls = [] + + def refused(adapter, operation, **parameters): + if operation == "switch": + calls.append(parameters) + raise native.HostError("client_quit_refused") + return transport(adapter, operation, **parameters) + + provider.transport = refused + assert provider.tick()["action"] == "client_quit_refused" + clock.now += 1000 + provider.tick() + assert len(calls) == 1 and transport.active == "a" + provider.tick(refresh=True) + assert len(calls) == 2 # A deliberate check authorizes one new attempt. + + +def test_disable_waits_for_owned_transaction_then_persists_off(native, tmp_path): + provider, transport, _ = configured(native, tmp_path) + transport.entered, transport.release = threading.Event(), threading.Event() + active = threading.Thread(target=provider.tick) + active.start() + assert transport.entered.wait(5) + result, errors = [], [] + + def disable(): + try: + result.append(provider.configure(expected_revision=1, enabled=False, slots=["a", "b"])) + except Exception as error: + errors.append(error) + + request = threading.Thread(target=disable) + request.start() + assert request.is_alive() + transport.release.set() + active.join(10) + request.join(10) + assert not errors and result[0]["policy"]["enabled"] is False + before = len(transport.calls) + assert provider.tick()["action"] == "disabled" and len(transport.calls) == before + + +def client_host(native, monkeypatch): + from loopx_codex_provider_routing import native_host as module + + monkeypatch.setattr(module.sys, "platform", "darwin") + adapter = module.NativeHost.__new__(module.NativeHost) + state = {"pid": 11, "active": "a", "revision": "initial", "quit": [], "switch": 0, "open": 0} + adapter.home, adapter.spec = Path("/synthetic"), {} + adapter.revision = lambda: state["revision"] + adapter.status = lambda: {"active_slot": state["active"], "selection_revision": state["revision"], + "current_login_matches": True, "recovery_required": False, + "client_running": state["pid"] is not None} + + def idle(): + if state["pid"] is not None: + raise module.HostError("waiting_for_idle") + if adapter.revision() != adapter.expected: + raise module.HostError("selection_changed") + + def switch(slot): + idle() + state.update(active=slot, revision="selected", switch=state["switch"] + 1) + return {"history_untouched": True} + + def reopen(*_): + state.update(pid=22, open=state["open"] + 1) + + def quit(command, **parameters): + state["quit"].append(command) + assert "app.terminate" in parameters["input"] and "forceTerminate" not in parameters["input"] + assert command[-2:] == ["11", "/synthetic/GPT.app/Contents/MacOS/GPT"] + state["pid"] = None + + adapter.manager = SimpleNamespace(switch=switch, idle=idle) + adapter.apps = SimpleNamespace(executable=lambda _: Path("/synthetic/GPT.app/Contents/MacOS/GPT"), + app_bundle=lambda *_: None, running=lambda *_: state["pid"], open_app=reopen) + monkeypatch.setattr(module.subprocess, "run", quit) + return adapter, state, module + + +def test_normal_quit_precedes_native_idle_transaction_and_same_route_reopen(native, monkeypatch): + adapter, state, _ = client_host(native, monkeypatch) + result = adapter.switch("b", "initial", reopen=True, restart_clients=True) + assert state["quit"] and state["switch"] == 1 and state["open"] == 1 + assert result["active_slot"] == "b" and result["client_running"] is True + + +def test_wait_only_mode_does_not_quit_a_client(native, monkeypatch): + adapter, state, _ = client_host(native, monkeypatch) + with pytest.raises(Exception, match="waiting_for_idle"): + adapter.switch("b", "initial", reopen=True) + assert not state["quit"] and state["active"] == "a" + + +def test_quit_refusal_and_timeout_do_not_mutate_credentials(native, monkeypatch): + adapter, state, module = client_host(native, monkeypatch) + + def refuse(*_, **__): + raise module.subprocess.CalledProcessError(1, "synthetic") + + monkeypatch.setattr(module.subprocess, "run", refuse) + with pytest.raises(Exception, match="client_quit_refused"): + adapter.switch("b", "initial", reopen=True, restart_clients=True) + assert state["switch"] == 0 and state["open"] == 0 + monkeypatch.setattr(module.subprocess, "run", lambda *_, **__: None) + with pytest.raises(Exception, match="client_quit_timeout"): + adapter.close_client("initial", timeout=0) + assert state["active"] == "a" + + +def test_manual_selection_and_new_client_prevent_closing_the_wrong_process(native, monkeypatch): + adapter, state, _ = client_host(native, monkeypatch) + with pytest.raises(Exception, match="selection_changed"): + adapter.switch("b", "stale", reopen=True, restart_clients=True) + assert not state["quit"] + owners = iter([11, 33]) + adapter.apps.running = lambda *_: next(owners) + with pytest.raises(Exception, match="client_changed"): + adapter.switch("b", "initial", reopen=True, restart_clients=True) + assert not state["quit"] and state["switch"] == 0 and state["open"] == 0 + + +def test_native_switch_failure_reopens_source_without_replaying_account_selection(native, monkeypatch): + adapter, state, module = client_host(native, monkeypatch) + adapter.manager.switch = lambda _: (_ for _ in ()).throw(module.HostError("readback_failed")) + with pytest.raises(Exception, match="readback_failed"): + adapter.switch("b", "initial", reopen=True, restart_clients=True) + assert state["active"] == "a" and state["open"] == 1 + + +def test_manual_selection_during_quit_wins_before_any_credential_write(native, monkeypatch): + adapter, state, module = client_host(native, monkeypatch) + quit = module.subprocess.run + + def manual_choice(*args, **kwargs): + quit(*args, **kwargs) + state.update(active="c", revision="manual") + + monkeypatch.setattr(module.subprocess, "run", manual_choice) + with pytest.raises(Exception, match="selection_changed"): + adapter.switch("b", "initial", reopen=True, restart_clients=True) + assert state["active"] == "c" and state["switch"] == 0 and state["open"] == 0 + + +def test_expired_automatic_evidence_does_not_even_quit_the_client(native, monkeypatch): + adapter, state, _ = client_host(native, monkeypatch) + with pytest.raises(Exception, match="quota_stale"): + adapter.switch("b", "initial", reopen=True, restart_clients=True, not_after=time.time() - 1) + assert not state["quit"] and not state["switch"]