← Files Meetings (Beta)ARCHIVED FILE
scripts/native_runtime.py
72.1 KB · Oct 9, 2026 · 12:23 UTC
#!/usr/bin/env python3
"""Locate, verify, and launch the ChatGPT Meetings companion.
The signed companion is shipped beside the plugin scripts, then materialized
into an immutable, profile-isolated Codex-owned generation before local-dev or
production launch. Recordings, upload state, and the private control bridge stay
in Application Support; a
running executable never depends on a prunable plugin-cache directory.
The signed plugin-local artifact remains the distribution authority. Mutable
recording, upload, and control state never becomes an alternate runtime cache.
"""
from __future__ import annotations
import ctypes
import hashlib as hashlib
try:
import fcntl
except ImportError: # Windows has no POSIX descriptor/lock module.
fcntl = None
import json
import os
import platform
import plistlib
import re
import stat
import subprocess
import sys
import threading
import time as time
import uuid as uuid
from collections.abc import Mapping, Sequence
from contextlib import contextmanager as contextmanager
from contextvars import ContextVar
from dataclasses import dataclass
from datetime import datetime
from enum import Enum
from pathlib import Path
from typing import TYPE_CHECKING, Callable, Final, Literal, Protocol
from typing import Iterator as Iterator
from meetings_sentry import (
MCP_SENTRY_FAILURE_DIAGNOSTIC_VALUES,
MCP_SENTRY_HANDOFF_FAILURE_REASONS,
MCP_SENTRY_MATERIALIZE_FAILURE_REASONS,
MCP_SENTRY_MATERIALIZE_VERIFICATION_PHASES,
McpSentryDiagnostics,
report_mcp_error,
report_recovered_handoff,
)
from native_runtime_types import (
NativeArtifactFingerprint,
PlatformRuntimeSpec,
PublicNativeRuntimeStatus,
)
from recording_control_action_contract import (
ControlPlatform,
ControlPlatformWire,
)
from recording_control_stream_contract import (
CONTROL_STREAM_DARWIN_CAPABILITIES,
CONTROL_STREAM_WINDOWS_CAPABILITIES,
CONTROL_STREAM_WINDOWS_REQUIRED_UPDATE_CAPABILITIES,
)
from runtime_config import RUNTIME_CONFIG
CONTROL_CAPABILITIES_BY_PLATFORM: Final[dict[str, tuple[str, ...]]] = {
ControlPlatform.DARWIN_UNIVERSAL.value: CONTROL_STREAM_DARWIN_CAPABILITIES,
ControlPlatform.DARWIN_ARM64.value: CONTROL_STREAM_DARWIN_CAPABILITIES,
ControlPlatform.DARWIN_X64.value: CONTROL_STREAM_DARWIN_CAPABILITIES,
ControlPlatform.WINDOWS_X64.value: CONTROL_STREAM_WINDOWS_CAPABILITIES,
ControlPlatform.WINDOWS_ARM64.value: CONTROL_STREAM_WINDOWS_CAPABILITIES,
}
class ControlCapability(str, Enum):
"""Known generated live-stream capability; unknown values grant no authority."""
CONTROL_LOCAL_STREAM_V1 = "control.local-stream.v1"
CAPTURE_MICROPHONE_V1 = "capture.microphone.v1"
CAPTURE_SYSTEM_AUDIO_V1 = "capture.system-audio.v1"
CAPTURE_CONTEXT_V1 = "capture.context.v1"
RECORDING_LOCAL_WAV_V1 = "recording.local-wav.v1"
RECORDING_STOP_V1 = "recording.stop.v1"
RECORDING_PAUSE_RESUME_V1 = "recording.pause-resume.v1"
RECORDING_SESSION_FENCED_CONTROLS_V1 = "recording.session-fenced-controls.v1"
RECORDING_RECEIPT_V1 = "recording.receipt.v1"
UI_FLOATING_OVERLAY_V1 = "ui.floating-overlay.v1"
UPLOAD_OUTBOX_V1 = "upload.outbox.v1"
UPLOAD_POST_STOP_V1 = "upload.post-stop.v1"
UPLOAD_MANUAL_RETRY_V1 = "upload.manual-retry.v1"
UPLOAD_MANUAL_RETRY_TARGETED_V1 = "upload.manual-retry.targeted.v1"
HEADLESS_SETTINGS_V1 = "headless.settings.v1"
HEADLESS_SYNC_HOOK_V1 = "headless.sync-hook.v1"
HEADLESS_NOTE_DETAIL_V1 = "headless.note-detail.v1"
LOGS_VIEWER_V1 = "logs.viewer.v1"
AUTH_CODEX_READINESS_V1 = "auth.codex-readiness.v1"
WIDGET_SNAPSHOT_V1 = "widget.snapshot.v1"
WIDGET_RECENT_RECORDING_V1 = "widget.recent-recording.v1"
HOME_CACHE_V1 = "home.cache.v1"
HOME_CACHE_REFRESH_V1 = "home.cache.refresh.v1"
CALENDAR_AUTOMATION_SYNC_V1 = "calendar.automation-sync.v1"
LIFECYCLE_UPDATE_HANDOFF_V1 = "lifecycle.update-handoff.v1"
LIFECYCLE_UPDATE_HANDOFF_WORK_FENCE_V1 = "lifecycle.update-handoff-work-fence.v1"
LIFECYCLE_UPDATE_HANDOFF_RESUMABLE_OUTBOX_V1 = "lifecycle.update-handoff-resumable-outbox.v1"
LIFECYCLE_UPDATE_HANDOFF_RESUMABLE_OUTBOX_V2 = "lifecycle.update-handoff-resumable-outbox.v2"
LIFECYCLE_CAPTURE_AWARE_SIGTERM_V1 = "lifecycle.capture-aware-sigterm.v1"
UNKNOWN = "unknown"
@classmethod
def _missing_(cls, value: object) -> ControlCapability | None:
return cls.UNKNOWN if isinstance(value, str) else None
def supported_on_platform(self, platform: ControlPlatform) -> bool:
if self is type(self).UNKNOWN or platform is ControlPlatform.UNKNOWN:
return False
return self.value in CONTROL_CAPABILITIES_BY_PLATFORM.get(platform.value, ())
class _WindowsCtypesApi(Protocol):
def WinDLL(self, name: str, *, use_last_error: bool) -> ctypes.CDLL: ...
# Linux Pyright stubs intentionally omit this Windows-only ctypes member.
if TYPE_CHECKING:
_windows_ctypes: _WindowsCtypesApi
else:
_windows_ctypes = ctypes
APP_NAME = RUNTIME_CONFIG.app_name
# One immutable runtime config binds the development checkout or production
# artifact to its matching bundle ID, state root, control target, and signer.
BUNDLE_ID = RUNTIME_CONFIG.bundle_identifier
APP_SUPPORT_DIRECTORY_NAME = RUNTIME_CONFIG.app_support_directory
TEAM_IDENTIFIER = RUNTIME_CONFIG.team_identifier
PLUGIN_FRAMEWORK_SYMLINKS = RUNTIME_CONFIG.framework_symlinks
OFFICIAL_CACHE_PUBLISHERS = frozenset(RUNTIME_CONFIG.official_cache_publishers)
PLUGIN_BUNDLE_CROSS_PROCESS_LOCK_TIMEOUT_SECONDS = 10.0
STABLE_RUNTIME_LOCK_TIMEOUT_SECONDS = 10.0
STABLE_RUNTIME_DIRECTORY_NAME = "chatgpt-meetings"
DEVELOPMENT_STABLE_RUNTIME_DIRECTORY_NAME = "chatgpt-meetings-dev"
DEVELOPMENT_SIGNING_AUTHORITY = "ChatGPT Meetings Local Development"
STABLE_RUNTIME_MANIFEST_MAX_BYTES = 16 * 1024
STABLE_RUNTIME_SCHEMA_VERSION = 1
VERIFIED_CAM_DISTRIBUTION_RELATIVE_PATH = Path("native/verified-cam-distribution.json")
NATIVE_BUILD_RECEIPT_NAME = "meetings-build-receipt.json"
LEGACY_NATIVE_BUILD_RECEIPT_NAME = "sushi-build-receipt.json"
NATIVE_BUILD_RECEIPT_NAMES = (NATIVE_BUILD_RECEIPT_NAME, LEGACY_NATIVE_BUILD_RECEIPT_NAME)
PLATFORM_OVERRIDE_ENV = "CHATGPT_MEETINGS_NATIVE_PLATFORM"
E2E_ISOLATED_ENV = "CHATGPT_MEETINGS_E2E_ISOLATED"
# Shared app configuration (including Sentry and hosted policy) deliberately
# recognizes a separate gate. Only synthesize it for the already-validated
# direct-executable E2E launch below; an ambient value must never be enough to
# activate the native isolation path.
CODEX_E2E_ISOLATED_ENV = "CHATGPT_MEETINGS_CODEX_E2E_ISOLATED"
LEGACY_CODEX_E2E_ISOLATED_ENV = "SUSHI_CODEX_MEETINGS_E2E_ISOLATED"
E2E_ROOT_ENV = "CHATGPT_MEETINGS_E2E_ROOT"
E2E_NATIVE_ROOT_ENV = "CHATGPT_MEETINGS_E2E_NATIVE_ROOT"
E2E_CAPTURE_MODE_ENV = "CHATGPT_MEETINGS_E2E_CAPTURE_MODE"
E2E_LAUNCH_REPORT_FD_ENV = "CHATGPT_MEETINGS_E2E_LAUNCH_REPORT_FD"
E2E_LAUNCH_REPORT_NONCE_ENV = "CHATGPT_MEETINGS_E2E_LAUNCH_REPORT_NONCE"
E2E_LAUNCH_REPORT_KIND = "chatgpt-meetings-e2e-launch-report"
E2E_LIVE_PROCESS_REGISTRY_RELATIVE_PATH = Path("tmp/live-processes.json")
E2E_LIVE_PROCESS_REGISTRY_MAX_BYTES = 32 * 1_024
E2E_LIVE_PROCESS_REGISTRY_MAX_ENTRIES = 64
E2E_LIVE_PROCESS_REGISTRY_LOCK_TIMEOUT_SECONDS = 2.0
E2E_COMPANION_EXEC_GATE_ARGUMENT = "--e2e-companion-exec-gate-v1"
E2E_STDIO_EXEC_GATE_ARGUMENT = "--e2e-stdio-exec-gate-v1"
E2E_STDIO_PROFILE_PROTOTYPE = "prototype"
E2E_STDIO_PROFILE_PRODUCTION = "production"
E2E_STDIO_PROFILES = frozenset({E2E_STDIO_PROFILE_PROTOTYPE, E2E_STDIO_PROFILE_PRODUCTION})
E2E_DETERMINISTIC_CAPTURE_MODE = "deterministic-rust-v1"
E2E_ROOT_MARKER = ".chatgpt-meetings-e2e-root-v1"
E2E_ROOT_MARKER_CONTENT = b"chatgpt-meetings-e2e-v1\n"
_E2E_REGISTERED_CHILD_ANCHORS_LOCK = threading.Lock()
# launch_current returns only a PID, so dropping its local Popen would let
# Popen.__del__ poll and reap a fast-exited group leader before the probe uses
# the reported handle. Retain successful E2E children for the MCP lifetime;
# the shared registry's 64-entry contract keeps this supervisor table bounded.
_E2E_REGISTERED_CHILD_ANCHORS: dict[int, subprocess.Popen[bytes]] = {}
_WINDOWS_PENDING_CHILDREN_LOCK = threading.Lock()
_MAXIMUM_WINDOWS_PENDING_CHILDREN = 16
PLUGIN_ROOT = Path(__file__).resolve().parents[1]
MAXIMUM_ARCHIVE_BYTES = 256 * 1024 * 1024
MAXIMUM_MANIFEST_BYTES = 128 * 1024
MAXIMUM_PLUGIN_FRAMEWORK_ENTRIES = 4096
MAXIMUM_PLUGIN_RESOURCE_ENTRIES = 100_000
MAXIMUM_DARWIN_MACHO_SLICES = 64
MAXIMUM_STABLE_RUNTIME_ENTRIES = 100_000
DARWIN_MACHO_CPU_TYPES = {
"darwin-arm64": 0x0100000C,
"darwin-x64": 0x01000007,
}
WINDOWS_PLUGIN_RUNTIME_DESCRIPTOR_RELATIVE_PATH = Path(".codex-plugin/windows-runtime.json")
WINDOWS_PRODUCTION_AZURE_ACCOUNT_NAME = "oaisidekickuploads"
WINDOWS_PRODUCTION_AZURE_CONTAINER_NAME = "chatgpt-meetings"
WINDOWS_PRODUCTION_BUNDLE_RELATIVE_PATH = Path("native/windows-production-bundle.json")
WINDOWS_PRODUCTION_LOCK_RELATIVE_PATH = Path("native/cam-windows-artifact.lock.json")
OPEN_EXIT_CHECK_TIMEOUT_SECONDS = 0.25
WINDOWS_CHILD_EXIT_CHECK_TIMEOUT_SECONDS = 0.25
WINDOWS_LAUNCH_ATTEMPT_LIMIT = 3
WINDOWS_LAUNCH_OWNER_CONFIRM_TIMEOUT_SECONDS = 5.0
WINDOWS_SIGNATURE_VERIFICATION_TIMEOUT_SECONDS = 10.0
COMPANION_CA_ENVIRONMENT_KEYS = ("CODEX_CA_CERTIFICATE", "SSL_CERT_FILE")
WINDOWS_COMPANION_ENVIRONMENT_KEYS = (
"SYSTEMDRIVE",
"SYSTEMROOT",
"WINDIR",
"PROGRAMDATA",
"ALLUSERSPROFILE",
"COMMONPROGRAMFILES",
"COMMONPROGRAMFILES(X86)",
"COMMONPROGRAMW6432",
"LOCALAPPDATA",
"APPDATA",
"USERPROFILE",
"TEMP",
"TMP",
*COMPANION_CA_ENVIRONMENT_KEYS,
)
GITHUB_RELEASE_COMPONENT = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]*\Z")
SHA256_HEX_PATTERN = re.compile(r"[a-f0-9]{64}\Z")
WINDOWS_REQUIRED_UPDATE_CAPABILITIES = frozenset(
CONTROL_STREAM_WINDOWS_REQUIRED_UPDATE_CAPABILITIES
)
def stable_runtime_directory_name() -> str:
"""Keep the local-development native owner isolated from production."""
if RUNTIME_CONFIG.flavor == "development" and RUNTIME_CONFIG.default_mcp_profile == "local-dev":
return DEVELOPMENT_STABLE_RUNTIME_DIRECTORY_NAME
return STABLE_RUNTIME_DIRECTORY_NAME
class NativeRuntimeError(RuntimeError):
pass
class NativeRuntimeResourceOperation(Enum):
"""Observed filesystem boundaries; never inferred from a path or message."""
RESOLVE_ARTIFACT_ROOT = ("resolve_artifact_root", "artifact_root")
RESOLVE_NATIVE_ARTIFACT = ("resolve_native_artifact", "native_artifact")
READ_BUNDLE_METADATA = ("read_bundle_metadata", "bundle_metadata")
def __init__(self, operation: str, role: str) -> None:
self.operation = operation
self.role = role
class NativeRuntimeResourceError(NativeRuntimeError):
"""Preserve a resource boundary without granting retry or replacement authority."""
def __init__(self, operation: NativeRuntimeResourceOperation) -> None:
super().__init__("native runtime resource is unavailable")
self.operation = operation
class NativeRuntimeRegistrationUnavailable(NativeRuntimeError):
"""A previously observed installation path disappeared during verification.
This grants only a delayed recheck, never authority to launch a different
image or terminate a running owner. The role is a fixed code, not a path.
"""
def __init__(
self,
role: Literal["plugin_root", "family_root", "registration_manifest", "registered_plugin"],
) -> None:
super().__init__("ChatGPT Meetings plugin registration is unavailable")
self.role = role
class NativeRuntimeArtifactMissing(NativeRuntimeError):
"""The expected native child was absent beneath an existing plugin root."""
def __init__(self) -> None:
super().__init__("plugin-bundled native artifact is unavailable")
class NativeRuntimeLockBusy(NativeRuntimeError):
"""Another operation still holds a verified native runtime or bundle lock."""
class NativeRuntimeQuitRequired(NativeRuntimeError):
"""The proven older companion cannot cooperatively leave for an update.
This is deliberately a typed local boundary instead of forwarding the
control client's exception text. MCP can turn it into one fixed recovery
state without leaking descriptor paths or implementation details.
"""
class NativeRuntimeUpdateDeferred(NativeRuntimeQuitRequired):
"""A proven cooperative owner is busy and may safely yield later."""
class NativeRuntimeLaunchPending(NativeRuntimeError):
"""Launch cannot safely retry while an exact spawned child remains live.
Foreground callers still receive an error. Initialize bootstrap uses this
marker only to adopt its existing poll/cooldown lane instead of spawning a
second process while the first child may still converge or exit.
"""
class NativeRuntimeLaunchPendingQuitRequired(
NativeRuntimeQuitRequired,
NativeRuntimeLaunchPending,
):
"""A pending exact child raced a proven non-cooperative older owner."""
class NativeRuntimeLaunchPendingUpdateDeferred(
NativeRuntimeUpdateDeferred,
NativeRuntimeLaunchPending,
):
"""A pending exact child raced a proven busy cooperative older owner."""
@dataclass
class _WindowsPendingChild:
process: subprocess.Popen[bytes]
observing: bool
codex_launch_context: tuple[str, str] | None = None
_WindowsPendingChildKey = tuple[
str,
str,
str,
]
_WINDOWS_PENDING_CHILDREN: dict[_WindowsPendingChildKey, _WindowsPendingChild] = {}
class _ProcBSDInfo(ctypes.Structure):
_fields_ = [
("pbi_flags", ctypes.c_uint32),
("pbi_status", ctypes.c_uint32),
("pbi_xstatus", ctypes.c_uint32),
("pbi_pid", ctypes.c_uint32),
("pbi_ppid", ctypes.c_uint32),
("pbi_uid", ctypes.c_uint32),
("pbi_gid", ctypes.c_uint32),
("pbi_ruid", ctypes.c_uint32),
("pbi_rgid", ctypes.c_uint32),
("pbi_svuid", ctypes.c_uint32),
("pbi_svgid", ctypes.c_uint32),
("rfu_1", ctypes.c_uint32),
("pbi_comm", ctypes.c_char * 16),
("pbi_name", ctypes.c_char * 32),
("pbi_nfiles", ctypes.c_uint32),
("pbi_pgid", ctypes.c_uint32),
("pbi_pjobc", ctypes.c_uint32),
("e_tdev", ctypes.c_uint32),
("e_tpgid", ctypes.c_uint32),
("pbi_nice", ctypes.c_int32),
("pbi_start_tvsec", ctypes.c_uint64),
("pbi_start_tvusec", ctypes.c_uint64),
]
from native_runtime_e2e import (
_discard_pre_exec_e2e_child as _discard_pre_exec_e2e_child,
)
from native_runtime_e2e import (
_e2e_companion_exec_gate as _e2e_companion_exec_gate,
)
from native_runtime_e2e import (
_e2e_exec_gate as _e2e_exec_gate,
)
from native_runtime_e2e import (
_e2e_stdio_exec_gate as _e2e_stdio_exec_gate,
)
from native_runtime_e2e import (
_initialize_e2e_live_process_registry as _initialize_e2e_live_process_registry,
)
from native_runtime_e2e import (
_live_process_registry_entry as _live_process_registry_entry,
)
from native_runtime_e2e import (
_mutate_e2e_live_process_registry as _mutate_e2e_live_process_registry,
)
from native_runtime_e2e import (
_process_start_identity as _process_start_identity,
)
from native_runtime_e2e import (
_publish_e2e_launch_report as _publish_e2e_launch_report,
)
from native_runtime_e2e import (
_read_live_process_registry_at as _read_live_process_registry_at,
)
from native_runtime_e2e import (
_register_e2e_live_process as _register_e2e_live_process,
)
from native_runtime_e2e import (
_require_e2e_pipe_descriptor as _require_e2e_pipe_descriptor,
)
from native_runtime_e2e import (
_retain_registered_e2e_child as _retain_registered_e2e_child,
)
from native_runtime_e2e import (
_spawn_registered_e2e_companion as _spawn_registered_e2e_companion,
)
from native_runtime_e2e import (
_strict_live_process_registry as _strict_live_process_registry,
)
from native_runtime_e2e import (
_terminate_unreported_e2e_child as _terminate_unreported_e2e_child,
)
from native_runtime_e2e import (
_write_live_process_registry_at as _write_live_process_registry_at,
)
_PUBLIC_RUNTIME_COMPONENT = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,79}\Z")
_NATIVE_ALPHA_RELEASE_VERSION = re.compile(
r"(0|[1-9][0-9]{0,8})\.(0|[1-9][0-9]{0,8})\.(0|[1-9][0-9]{0,8})"
r"-alpha\.([1-9][0-9]{0,8})\Z"
)
_BUILD_TIMESTAMP_UTC_PATTERN = re.compile(
r"[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}Z\Z"
)
_PUBLIC_RUNTIME_UPDATE_ERROR = "Native update check is unavailable"
_NATIVE_RUNTIME_EVENT_COMPONENT = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}\Z")
_NATIVE_RUNTIME_EVENT_GENERATION = re.compile(r"[a-z0-9-]+-[a-f0-9]{64}\Z")
_NATIVE_RUNTIME_HANDOFF_DISPOSITIONS = frozenset({"fallback_started", "start_failed"})
_NATIVE_RUNTIME_HANDOFF_OPERATIONS = frozenset({"new_note", "stop"})
_NATIVE_RUNTIME_RECOVERY_KINDS = frozenset(
{
"fallback_connection_failed",
"owner_changed",
"quit_required",
"start_retry_required",
"update_deferred",
}
)
_NATIVE_RUNTIME_SENTRY_FAILURES = frozenset(
{
("materialize", "copy"),
("materialize", "verify"),
("materialize", "publish"),
("handoff", "handoff"),
("recording", "connection"),
("recording", "handoff"),
("recording", "runtime"),
}
)
@dataclass
class BootstrapHandoffReporting:
"""Capture one private handoff failure until bootstrap proves its outcome."""
diagnostics: McpSentryDiagnostics | None = None
recovered: bool = False
_BOOTSTRAP_HANDOFF_REPORTING: ContextVar[BootstrapHandoffReporting | None] = ContextVar(
"meetings_bootstrap_handoff_reporting", default=None
)
@contextmanager
def capture_bootstrap_handoff_failure(
episode: BootstrapHandoffReporting,
) -> Iterator[None]:
"""Scope deferred handoff telemetry to exactly one verified launch call."""
token = _BOOTSTRAP_HANDOFF_REPORTING.set(episode)
try:
yield
finally:
_BOOTSTRAP_HANDOFF_REPORTING.reset(token)
def report_bootstrap_handoff_outcome(episode: BootstrapHandoffReporting) -> None:
"""Emit one final handoff notice or error after launch context is cleared."""
if episode.diagnostics is None:
return
try:
if episode.recovered:
report_recovered_handoff(
active_owner_version=episode.diagnostics.get("active_owner_version"),
target_owner_version=episode.diagnostics.get("target_owner_version"),
handoff_reason=episode.diagnostics.get("handoff_reason"),
)
else:
report_mcp_error("handoff", "handoff", **episode.diagnostics)
except Exception:
# Diagnostics never interfere with verified owner recovery.
pass
def log_native_runtime_event(
stage: str,
outcome: str,
*,
platform: object = None,
version: object = None,
build_timestamp: object = None,
generation: object = None,
previous_generation: object = None,
attempt: object = None,
error_kind: object = None,
trigger: object = None,
reason: object = None,
active_owner_version: object = None,
target_owner_version: object = None,
handoff_disposition: object = None,
recovery_kind: object = None,
operation: object = None,
verification_phase: object = None,
failure_operation: object = None,
resource_role: object = None,
os_error_code: object = None,
pid: object = None,
duration_ms: object = None,
) -> None:
"""Emit a small, path-free companion update lifecycle envelope to stderr.
Args:
stage: Bounded lifecycle stage name.
outcome: Bounded lifecycle outcome name.
platform: Untrusted platform diagnostic to include when safe.
version: Untrusted runtime version diagnostic to include when safe.
build_timestamp: Untrusted packaged UTC timestamp to include when valid.
generation: Untrusted active generation identifier to include when safe.
previous_generation: Untrusted prior generation identifier to include when safe.
attempt: Untrusted bounded retry count to include when valid.
error_kind: Untrusted failure category to include when safe.
trigger: Untrusted lifecycle trigger to include when safe.
reason: Untrusted bounded reason code to include when safe.
active_owner_version: Untrusted existing-owner version diagnostic.
target_owner_version: Untrusted successor-owner version diagnostic.
handoff_disposition: Untrusted handoff result diagnostic.
recovery_kind: Untrusted recovery strategy diagnostic.
operation: Untrusted recording operation diagnostic.
verification_phase: Untrusted bounded materialization verification boundary.
failure_operation: Untrusted bounded Stop failure operation.
resource_role: Untrusted bounded Stop failure resource role.
os_error_code: Untrusted bounded Stop operating-system error code.
pid: Optional local process identifier when safely bounded.
duration_ms: Optional safely bounded operation duration in milliseconds.
Returns:
None. Invalid or private diagnostics are omitted.
"""
if (
not isinstance(stage, str)
or _NATIVE_RUNTIME_EVENT_COMPONENT.fullmatch(stage) is None
or not isinstance(outcome, str)
or _NATIVE_RUNTIME_EVENT_COMPONENT.fullmatch(outcome) is None
):
return
is_native_owner_handoff_failure = stage == "handoff" and error_kind == "handoff"
handoff_reason = (
reason if type(reason) is str and reason in MCP_SENTRY_HANDOFF_FAILURE_REASONS else None
)
if (stage, error_kind) == ("materialize", "verify") and (
type(reason) is not str or reason not in MCP_SENTRY_MATERIALIZE_FAILURE_REASONS
):
reason = None
event: dict[str, str | int] = {"stage": stage, "outcome": outcome}
for key, value in (
("platform", platform),
("version", version),
("errorKind", error_kind),
("trigger", trigger),
("reason", handoff_reason if is_native_owner_handoff_failure else reason),
):
if isinstance(value, str) and _NATIVE_RUNTIME_EVENT_COMPONENT.fullmatch(value):
event[key] = value
timestamp = safe_build_timestamp_utc(build_timestamp)
if timestamp is not None:
event["buildTimestamp"] = timestamp
for key, value in (
("generation", generation),
("previousGeneration", previous_generation),
):
if isinstance(value, str) and _NATIVE_RUNTIME_EVENT_GENERATION.fullmatch(value):
event[key] = value
if type(attempt) is int and 0 <= attempt <= 1_000:
event["attempt"] = attempt
if type(pid) is int and 1 <= pid <= 2_147_483_647:
event["pid"] = pid
if type(duration_ms) is int and 0 <= duration_ms <= 3_600_000:
event["durationMs"] = duration_ms
diagnostics = McpSentryDiagnostics()
is_recording_owner_failure = (
stage == "recording"
and isinstance(error_kind, str)
and error_kind in {"handoff", "connection"}
)
if (stage, error_kind) == ("materialize", "verify") and isinstance(reason, str):
diagnostics["failure_reason"] = reason
if (
(stage, error_kind) == ("materialize", "verify")
and outcome == "failed"
and type(verification_phase) is str
and verification_phase in MCP_SENTRY_MATERIALIZE_VERIFICATION_PHASES
):
event["verificationPhase"] = verification_phase
diagnostics["verification_phase"] = verification_phase
if is_recording_owner_failure or is_native_owner_handoff_failure:
if is_native_owner_handoff_failure and handoff_reason is not None:
diagnostics["handoff_reason"] = handoff_reason
if (
is_native_owner_handoff_failure
and target_owner_version is None
and isinstance(version, str)
and _NATIVE_ALPHA_RELEASE_VERSION.fullmatch(version) is not None
):
target_owner_version = version
for event_key, value in (
("activeOwnerVersion", active_owner_version),
("targetOwnerVersion", target_owner_version),
):
if (
isinstance(value, str)
and value.lower() != "unknown"
and _PUBLIC_RUNTIME_COMPONENT.fullmatch(value)
and (
not is_native_owner_handoff_failure
or _NATIVE_ALPHA_RELEASE_VERSION.fullmatch(value) is not None
)
):
event[event_key] = value
if event_key == "activeOwnerVersion":
diagnostics["active_owner_version"] = value
else:
diagnostics["target_owner_version"] = value
if (
isinstance(handoff_disposition, str)
and handoff_disposition in _NATIVE_RUNTIME_HANDOFF_DISPOSITIONS
):
event["handoffDisposition"] = handoff_disposition
diagnostics["handoff_disposition"] = handoff_disposition
if isinstance(recovery_kind, str) and recovery_kind in _NATIVE_RUNTIME_RECOVERY_KINDS:
event["recoveryKind"] = recovery_kind
diagnostics["recovery_kind"] = recovery_kind
if (
is_recording_owner_failure
and isinstance(operation, str)
and operation in _NATIVE_RUNTIME_HANDOFF_OPERATIONS
and (operation != "stop" or error_kind == "connection")
):
event["operation"] = operation
diagnostics["operation"] = operation
if (stage, error_kind, operation) == ("recording", "connection", "stop"):
failure_candidates: tuple[
tuple[Literal["failure_operation", "resource_role", "os_error_code"], object], ...
] = (
("failure_operation", failure_operation),
("resource_role", resource_role),
("os_error_code", os_error_code),
)
for key, candidate in failure_candidates:
if candidate is not None:
value = (
candidate
if type(candidate) is str
and candidate in MCP_SENTRY_FAILURE_DIAGNOSTIC_VALUES[key]
else "unknown"
)
event[key] = value
diagnostics[key] = value
if (
outcome == "failed"
and isinstance(error_kind, str)
and (stage, error_kind) in _NATIVE_RUNTIME_SENTRY_FAILURES
):
episode = _BOOTSTRAP_HANDOFF_REPORTING.get() if is_native_owner_handoff_failure else None
if episode is not None:
episode.diagnostics = diagnostics.copy()
else:
report_mcp_error(stage, error_kind, **diagnostics)
try:
print(
f"ChatGPT Meetings native runtime: {json.dumps(event, sort_keys=True)}",
file=sys.stderr,
flush=True,
)
except (OSError, TypeError, ValueError):
# A closed parent stderr pipe must not turn a valid update or recovery
# into a failed MCP request.
pass
def safe_build_timestamp_utc(value: object) -> str | None:
"""Accept only the immutable UTC build stamp format emitted by packaging.
Args:
value: Untrusted candidate build timestamp.
Returns:
The validated UTC timestamp, or None when the candidate is malformed.
"""
if not isinstance(value, str) or not _BUILD_TIMESTAMP_UTC_PATTERN.fullmatch(value):
return None
try:
datetime.strptime(value, "%Y-%m-%dT%H:%M:%SZ")
except ValueError:
return None
return value
def public_runtime_capabilities(platform_key: object) -> tuple[str, ...]:
"""Return generated portable capabilities for any supported native platform.
Args:
platform_key: Untrusted candidate platform identifier.
Returns:
All supported generated feature capabilities, or no capabilities for an
unknown native platform.
"""
if not isinstance(platform_key, str):
return ()
if platform_key not in PLATFORM_RUNTIME_SPECS:
return ()
return tuple(
capability.value
for capability in ControlCapability
if capability
not in {
ControlCapability.UNKNOWN,
ControlCapability.CALENDAR_AUTOMATION_SYNC_V1,
ControlCapability.HOME_CACHE_REFRESH_V1,
}
)
def public_runtime_status(status: Mapping[str, object]) -> PublicNativeRuntimeStatus:
"""Return the runtime facts safe to expose through MCP/model payloads.
RuntimeManager keeps absolute paths internally because descriptor binding,
signature verification, and LaunchServices need them. Public status only
needs coarse install/version/capability facts; copying that projection here
prevents a future caller from accidentally returning app/artifact paths or
release-feed URLs.
Args:
status: Verified private runtime facts to sanitize for public projection.
Returns:
A path-free, capability-filtered runtime status payload.
"""
raw = status
version = raw.get("version")
build_timestamp = safe_build_timestamp_utc(raw.get("buildTimestamp"))
platform_key = raw.get("platform")
artifact_kind = raw.get("artifactKind")
source = raw.get("source")
capabilities = raw.get("capabilities")
allowed_capabilities = public_runtime_capabilities(platform_key)
projected: PublicNativeRuntimeStatus = {
"installed": raw.get("installed") is True,
"version": (
version
if isinstance(version, str) and _PUBLIC_RUNTIME_COMPONENT.fullmatch(version)
else None
),
"buildTimestamp": build_timestamp,
"platform": (
platform_key
if isinstance(platform_key, str) and _PUBLIC_RUNTIME_COMPONENT.fullmatch(platform_key)
else None
),
"artifactKind": (
artifact_kind
if isinstance(artifact_kind, str) and _PUBLIC_RUNTIME_COMPONENT.fullmatch(artifact_kind)
else None
),
"capabilities": (
[
capability
for capability in capabilities
if isinstance(capability, str) and capability in allowed_capabilities
]
if isinstance(capabilities, list)
else []
),
"source": "plugin-bundled" if source == "plugin-bundled" else None,
}
release_version = raw.get("releaseVersion")
release_match = (
_NATIVE_ALPHA_RELEASE_VERSION.fullmatch(release_version)
if isinstance(release_version, str)
else None
)
if (
raw.get("installed") is True
and source == "plugin-bundled"
and isinstance(platform_key, str)
and platform_key in PLATFORM_RUNTIME_SPECS
and release_match is not None
and isinstance(release_version, str)
and isinstance(version, str)
and (
version
in {
".".join(release_match.group(index) for index in (1, 2, 3)),
release_version,
}
or platform_key.startswith("windows-")
and version == "plugin-bundled"
)
):
projected["releaseVersion"] = release_version
updated = raw.get("updated")
if isinstance(updated, bool):
projected["updated"] = updated
launching = raw.get("launching")
if isinstance(launching, bool):
projected["launching"] = launching
if raw.get("updateError") == _PUBLIC_RUNTIME_UPDATE_ERROR:
projected["updateError"] = _PUBLIC_RUNTIME_UPDATE_ERROR
return projected
@dataclass(frozen=True)
class E2EIsolation:
"""Private roots passed to one permission-free companion proof.
This is intentionally unavailable unless the explicit E2E bit is set.
The normal MCP/runtime path never accepts an arbitrary HOME or native
Application Support override from its environment.
"""
root: Path
home: Path
codex_home: Path
native_root: Path
control_root: Path
temp_root: Path
required_codex_launch_context: tuple[str, str] | None = None
def child_environment(
self,
*,
capture_mode: str | None = None,
) -> dict[str, str]:
"""Return a tiny allowlist instead of inheriting credentials."""
if capture_mode not in {None, E2E_DETERMINISTIC_CAPTURE_MODE}:
raise NativeRuntimeError("E2E capture mode is invalid")
environment = {
"PATH": "/usr/bin:/bin:/usr/sbin:/sbin",
"HOME": str(self.home),
"CFFIXED_USER_HOME": str(self.home),
"CODEX_HOME": str(self.codex_home),
"TMPDIR": f"{self.temp_root}/",
E2E_ISOLATED_ENV: "1",
CODEX_E2E_ISOLATED_ENV: "1",
LEGACY_CODEX_E2E_ISOLATED_ENV: "1",
E2E_ROOT_ENV: str(self.root),
E2E_NATIVE_ROOT_ENV: str(self.native_root),
"CHATGPT_MEETINGS_E2E_SUPPRESS_BOOTSTRAP": "1",
"CHATGPT_MEETINGS_CONTROL_ROOT": str(self.control_root),
}
for key in ("LANG", "LC_ALL", "__CF_USER_TEXT_ENCODING"):
value = os.environ.get(key)
if value:
environment[key] = value
if capture_mode is not None:
environment[E2E_CAPTURE_MODE_ENV] = capture_mode
if self.required_codex_launch_context is not None:
context = _validated_codex_launch_context(platform="macos")
if context is None or context != self.required_codex_launch_context:
raise NativeRuntimeError("required Codex launch context changed or is unavailable")
cli_path, codex_home = context
if codex_home != str(self.codex_home) or not Path(cli_path).is_relative_to(self.root):
raise NativeRuntimeError("required Codex launch context is outside E2E isolation")
# This is a private fixture helper, never an ambient user CLI or
# credential path. Preserve the exact helper through the exec gate.
environment["CODEX_BIN"] = cli_path
environment["CODEX_CLI_PATH"] = cli_path
return environment
def windows_companion_environment(
source: Mapping[str, str] | None = None,
) -> dict[str, str]:
"""Return the Windows paths and CA settings required by the native companion.
The detached companion resolves its own Codex authentication from validated
executable/home command-line paths. Keep the environment positive and fixed
so credentials, proxy configuration, and test-only ChatGPT Meetings
overrides cannot cross the process boundary.
"""
values = os.environ if source is None else source
return {key: value for key in WINDOWS_COMPANION_ENVIRONMENT_KEYS if (value := values.get(key))}
def darwin_companion_ca_arguments() -> list[str]:
"""Forward explicit CA settings through LaunchServices before app arguments."""
arguments: list[str] = []
for key in COMPANION_CA_ENVIRONMENT_KEYS:
value = os.environ.get(key)
if value is not None:
arguments.extend(("--env", f"{key}={value}"))
return arguments
@dataclass(frozen=True)
class _PluginBundleVerification:
"""One successful process-local proof for a plugin-shipped app bundle.
Status and widget-open polls run frequently while the UI is visible. A
full codesign --verify --deep on every one of those polls makes the first
impression feel stuck even though the signed versioned plugin bundle has
not changed. This cache is deliberately only a status-path acceleration:
* it is keyed by the resolved plugin path plus identities for the bundle
metadata, executable, helper, and signature seal;
* failed validations are never cached;
* launch always asks for a fresh full verification before it can hand off
or execute anything.
A resource changed without changing one of the cheap identities can at
worst make a status row briefly stale in this process. It cannot authorize
execution because launch_current bypasses this cache.
"""
fingerprint: NativeArtifactFingerprint
version: str
build_timestamp: str | None
executable_sha256: str
stable_status_sentinel: NativeArtifactFingerprint | None = None
_PLUGIN_BUNDLE_VERIFICATION_CACHE: dict[tuple[str, str, str, bool], _PluginBundleVerification] = {}
_PLUGIN_BUNDLE_VERIFICATION_LOCK = threading.RLock()
def _darwin_runtime_spec(
platform_key: ControlPlatformWire,
manifest_keys: tuple[ControlPlatformWire, ...],
) -> PlatformRuntimeSpec:
return PlatformRuntimeSpec(
platform_key=platform_key,
manifest_keys=manifest_keys,
artifact_kind="macos-app-bundle",
artifact_name=APP_NAME,
artifact_path_kind="directory",
identity_kind="bundleIdentifier",
identity_value=BUNDLE_ID,
capabilities=CONTROL_CAPABILITIES_BY_PLATFORM[platform_key],
launch_strategy="macos-open-background",
)
def _windows_runtime_spec(platform_key: ControlPlatformWire) -> PlatformRuntimeSpec:
return PlatformRuntimeSpec(
platform_key=platform_key,
manifest_keys=(platform_key,),
artifact_kind="windows-executable",
artifact_name="ChatGPT Meetings.exe",
artifact_path_kind="file",
identity_kind="executableName",
identity_value="ChatGPT Meetings.exe",
capabilities=CONTROL_CAPABILITIES_BY_PLATFORM[platform_key],
launch_strategy="windows-detached-process",
)
PLATFORM_RUNTIME_SPECS: dict[str, PlatformRuntimeSpec] = {
ControlPlatform.DARWIN_UNIVERSAL.value: _darwin_runtime_spec(
ControlPlatform.DARWIN_UNIVERSAL.value, (ControlPlatform.DARWIN_UNIVERSAL.value,)
),
# Prefer one honest universal artifact, then accept only the matching legacy
# architecture key. An Intel host must never select arm64.
ControlPlatform.DARWIN_ARM64.value: _darwin_runtime_spec(
ControlPlatform.DARWIN_ARM64.value,
(ControlPlatform.DARWIN_UNIVERSAL.value, ControlPlatform.DARWIN_ARM64.value),
),
ControlPlatform.DARWIN_X64.value: _darwin_runtime_spec(
ControlPlatform.DARWIN_X64.value,
(ControlPlatform.DARWIN_UNIVERSAL.value, ControlPlatform.DARWIN_X64.value),
),
ControlPlatform.WINDOWS_X64.value: _windows_runtime_spec(ControlPlatform.WINDOWS_X64.value),
ControlPlatform.WINDOWS_ARM64.value: _windows_runtime_spec(ControlPlatform.WINDOWS_ARM64.value),
}
def _is_windows_host() -> bool:
"""Whether filesystem/launch calls are executing on Windows itself."""
return os.name == "nt"
def _is_windows_spec(spec: PlatformRuntimeSpec) -> bool:
return spec.platform_key in (
ControlPlatform.WINDOWS_X64.value,
ControlPlatform.WINDOWS_ARM64.value,
)
def host_platform_key() -> ControlPlatformWire | None:
"""Return a feed key for a host we know how to name."""
machine = platform.machine().lower()
if sys.platform == "darwin" and machine in {"arm64", "aarch64"}:
return ControlPlatform.DARWIN_ARM64.value
if sys.platform == "darwin" and machine in {"x86_64", "amd64"}:
return ControlPlatform.DARWIN_X64.value
if os.name == "nt":
# RuntimeInformation.OSArchitecture in install.ps1 selects the native
# Windows OS architecture. A 32/64-bit emulated Codex process reports
# its process architecture through platform.machine(), so consult the
# native WOW64 marker first and keep install/runtime selection aligned
# on ARM64 Windows.
native_machine = (
os.environ.get("PROCESSOR_ARCHITEW6432", "").strip().lower()
or os.environ.get("PROCESSOR_ARCHITECTURE", "").strip().lower()
or machine
)
if native_machine in {"amd64", "x86_64"}:
return ControlPlatform.WINDOWS_X64.value
if native_machine in {"arm64", "aarch64"}:
return ControlPlatform.WINDOWS_ARM64.value
return None
def runtime_spec_for(platform_key: str) -> PlatformRuntimeSpec:
try:
return PLATFORM_RUNTIME_SPECS[platform_key]
except KeyError as exc:
raise NativeRuntimeError(
f"native runtime platform {platform_key!r} is unsupported"
) from exc
def configured_platform_spec() -> PlatformRuntimeSpec:
"""Select the host adapter, with an explicit test/future-build override.
Windows hosts select the EXE adapter; unsupported hosts fail closed unless
a test explicitly selects a known platform.
"""
configured = os.environ.get(PLATFORM_OVERRIDE_ENV, "").strip()
if configured:
return runtime_spec_for(configured)
detected = host_platform_key()
if detected:
return runtime_spec_for(detected)
raise NativeRuntimeError("native runtime host platform is unsupported")
def _private_e2e_directory(path: Path) -> Path:
"""Create one private directory without accepting symlink traversal."""
if not path.is_absolute() or path != Path(os.path.normpath(str(path))):
raise NativeRuntimeError("E2E isolation root is unsafe")
try:
resolved_parent = path.parent.resolve(strict=True)
except OSError as exc:
raise NativeRuntimeError("E2E isolation root is unsafe") from exc
if resolved_parent / path.name != path:
raise NativeRuntimeError("E2E isolation root is unsafe")
created = False
try:
path.mkdir(mode=0o700, exist_ok=False)
created = True
except FileExistsError:
pass
except OSError as exc:
raise NativeRuntimeError("E2E isolation root is unavailable") from exc
try:
if created:
os.chmod(path, 0o700)
metadata = path.lstat()
except OSError as exc:
raise NativeRuntimeError("E2E isolation root is unavailable") from exc
if (
path.is_symlink()
or not stat.S_ISDIR(metadata.st_mode)
or metadata.st_uid != os.getuid()
or stat.S_IMODE(metadata.st_mode) != 0o700
):
raise NativeRuntimeError("E2E isolation root is unsafe")
return path
def require_e2e_root_marker(root: Path) -> None:
marker = root / E2E_ROOT_MARKER
flags = os.O_RDONLY
if hasattr(os, "O_NOFOLLOW"):
flags |= os.O_NOFOLLOW
descriptor = -1
try:
descriptor = os.open(marker, flags)
metadata = os.fstat(descriptor)
content = os.read(descriptor, len(E2E_ROOT_MARKER_CONTENT) + 1)
except OSError as exc:
raise NativeRuntimeError("E2E isolation root is unclaimed") from exc
finally:
if descriptor >= 0:
os.close(descriptor)
if (
not stat.S_ISREG(metadata.st_mode)
or metadata.st_uid != os.getuid()
or stat.S_IMODE(metadata.st_mode) != 0o600
or metadata.st_size != len(E2E_ROOT_MARKER_CONTENT)
or content != E2E_ROOT_MARKER_CONTENT
):
raise NativeRuntimeError("E2E isolation root is unclaimed")
def configured_e2e_isolation() -> E2EIsolation | None:
"""Resolve the dev/E2E-only private state tree, or stay on production paths.
The one accepted input is a private absolute proof root. HOME, CODEX_HOME,
the native state root, and the control root are derived under it so a
caller cannot point one surface back at a user's real state.
"""
if os.environ.get(E2E_ISOLATED_ENV) != "1":
return None
if allow_unsigned_test_app():
raise NativeRuntimeError("E2E isolation requires a signed native app")
if fcntl is None or not hasattr(os, "getuid"):
raise NativeRuntimeError("POSIX E2E isolation is unavailable on Windows")
configured = os.environ.get(E2E_ROOT_ENV, "").strip()
if not configured:
raise NativeRuntimeError("E2E isolation root is required")
configured_root = Path(configured).expanduser()
if not configured_root.is_absolute() or configured_root.is_symlink():
raise NativeRuntimeError("E2E isolation root is unsafe")
try:
canonical_root = configured_root.resolve(strict=False)
except OSError as exc:
raise NativeRuntimeError("E2E isolation root is unsafe") from exc
if not canonical_root.is_dir():
raise NativeRuntimeError("E2E isolation root is unclaimed")
require_e2e_root_marker(canonical_root)
root = _private_e2e_directory(canonical_root)
home = _private_e2e_directory(root / "home")
codex_home = _private_e2e_directory(root / "codex-home")
native_root = _private_e2e_directory(root / "native")
control_root = _private_e2e_directory(native_root / "LocalMeetingsControl")
temp_root = _private_e2e_directory(root / "tmp")
return E2EIsolation(
root=root,
home=home,
codex_home=codex_home,
native_root=native_root,
control_root=control_root,
temp_root=temp_root,
)
def configured_e2e_capture_mode(
isolation: E2EIsolation | None,
) -> str | None:
"""Accept the fake selector only inside a fully validated isolation."""
if isolation is None:
return None
value = os.environ.get(E2E_CAPTURE_MODE_ENV, "").strip()
if not value:
return None
if value != E2E_DETERMINISTIC_CAPTURE_MODE:
raise NativeRuntimeError("E2E capture mode is invalid")
return value
def allow_unsigned_test_app() -> bool:
return os.environ.get("CHATGPT_MEETINGS_NATIVE_ALLOW_UNSIGNED") == "1"
def _valid_authenticode_identity(subject: object, thumbprint: object) -> bool:
return (
isinstance(subject, str)
and bool(subject)
and len(subject.encode("utf-8")) <= 512
and not any(ord(character) < 0x20 or ord(character) == 0x7F for character in subject)
and isinstance(thumbprint, str)
and re.fullmatch(r"[0-9A-Fa-f]{40}", thumbprint) is not None
)
def _load_app_metadata(
app_path: Path,
spec: PlatformRuntimeSpec | None = None,
) -> tuple[dict[str, object], Path, Path]:
"""Read the cheap structural proof shared by status and full validation."""
selected_spec = spec or configured_platform_spec()
if not selected_spec.platform_key.startswith("darwin-"):
raise NativeRuntimeError(
f"native runtime adapter for {selected_spec.platform_key} is not implemented"
)
info_path = app_path / "Contents" / "Info.plist"
try:
with info_path.open("rb") as handle:
info = plistlib.load(handle)
except (OSError, plistlib.InvalidFileException) as exc:
raise NativeRuntimeResourceError(
NativeRuntimeResourceOperation.READ_BUNDLE_METADATA
) from exc
if not isinstance(info, dict) or info.get("CFBundleIdentifier") != selected_spec.identity_value:
raise NativeRuntimeError("native app bundle identifier does not match")
executable_name = info.get("CFBundleExecutable")
if not isinstance(executable_name, str) or not executable_name:
raise NativeRuntimeError("native app executable metadata is missing")
executable = app_path / "Contents" / "MacOS" / executable_name
helper = app_path / "Contents" / "Resources" / "native" / "system_audio_capture"
if not executable.is_file() or not os.access(executable, os.X_OK):
raise NativeRuntimeError("native app executable is missing")
if not helper.is_file() or not os.access(helper, os.X_OK):
raise NativeRuntimeError("native system audio helper is missing")
for key in ("NSMicrophoneUsageDescription", "NSAudioCaptureUsageDescription"):
value = info.get(key)
if not isinstance(value, str) or not value.strip():
raise NativeRuntimeError(f"native app is missing {key}")
return dict(info), executable, helper
def _load_windows_executable(
executable_path: Path,
spec: PlatformRuntimeSpec | None = None,
) -> Path:
"""Validate the cheap structural identity for a Windows companion."""
selected_spec = spec or configured_platform_spec()
if not _is_windows_spec(selected_spec):
raise NativeRuntimeError(
f"native runtime adapter for {selected_spec.platform_key} is not implemented"
)
if (
executable_path.is_symlink()
or not executable_path.is_file()
or executable_path.name != selected_spec.identity_value
):
raise NativeRuntimeError("native executable identity does not match")
expected_machine = {
"windows-x64": 0x8664, # IMAGE_FILE_MACHINE_AMD64
"windows-arm64": 0xAA64, # IMAGE_FILE_MACHINE_ARM64
}[selected_spec.platform_key]
if _windows_pe_machine(executable_path) != expected_machine:
raise NativeRuntimeError("native executable architecture does not match platform")
return executable_path
def _require_darwin_macho_architecture(
executable_path: Path,
spec: PlatformRuntimeSpec,
) -> None:
"""Require one real thin/fat Mach-O slice for the selected Darwin host."""
expected_cpu = DARWIN_MACHO_CPU_TYPES.get(spec.platform_key)
if expected_cpu is None:
# The explicit universal test/local profile predates host-slice
# selection. Production host_platform_key always resolves arm64/x64.
return
try:
with executable_path.open("rb") as handle:
file_size = os.fstat(handle.fileno()).st_size
prefix = handle.read(8)
if len(prefix) != 8:
raise NativeRuntimeError("native app Mach-O header is malformed")
magic = prefix[:4]
if magic == b"\xcf\xfa\xed\xfe":
if file_size < 32:
raise NativeRuntimeError("native app Mach-O header is malformed")
cpu_type = int.from_bytes(prefix[4:8], "little")
if cpu_type != expected_cpu:
raise NativeRuntimeError(
"native app executable architecture does not match platform"
)
return
if magic not in {b"\xca\xfe\xba\xbe", b"\xca\xfe\xba\xbf"}:
raise NativeRuntimeError("native app Mach-O header is malformed")
is_fat64 = magic == b"\xca\xfe\xba\xbf"
count = int.from_bytes(prefix[4:8], "big")
entry_size = 32 if is_fat64 else 20
table_end = 8 + count * entry_size
if count <= 0 or count > MAXIMUM_DARWIN_MACHO_SLICES or table_end > file_size:
raise NativeRuntimeError("native app Mach-O architecture table is malformed")
entries = handle.read(count * entry_size)
if len(entries) != count * entry_size:
raise NativeRuntimeError("native app Mach-O architecture table is malformed")
matching_slices: list[tuple[int, int]] = []
for index in range(count):
entry = entries[index * entry_size : (index + 1) * entry_size]
cpu_type = int.from_bytes(entry[:4], "big")
if is_fat64:
offset = int.from_bytes(entry[8:16], "big")
size = int.from_bytes(entry[16:24], "big")
else:
offset = int.from_bytes(entry[8:12], "big")
size = int.from_bytes(entry[12:16], "big")
if offset < table_end or size < 32 or offset > file_size - size:
raise NativeRuntimeError("native app Mach-O architecture slice is malformed")
if cpu_type == expected_cpu:
matching_slices.append((offset, size))
if len(matching_slices) != 1:
raise NativeRuntimeError(
"native app executable architecture does not match platform"
)
offset, _size = matching_slices[0]
handle.seek(offset)
selected_header = handle.read(8)
if (
len(selected_header) != 8
or selected_header[:4] != b"\xcf\xfa\xed\xfe"
or int.from_bytes(selected_header[4:8], "little") != expected_cpu
):
raise NativeRuntimeError("native app Mach-O architecture slice is malformed")
except NativeRuntimeError:
raise
except OSError as exc:
raise NativeRuntimeError("native app executable architecture is unavailable") from exc
def _windows_pe_machine(executable_path: Path) -> int:
"""Read the bounded DOS/PE prefix needed to bind an EXE to its CPU."""
try:
with executable_path.open("rb") as handle:
file_size = os.fstat(handle.fileno()).st_size
dos_header = handle.read(64)
if len(dos_header) != 64 or dos_header[:2] != b"MZ":
raise NativeRuntimeError("native executable PE header is malformed")
pe_offset = int.from_bytes(dos_header[0x3C:0x40], "little")
if pe_offset < 64 or pe_offset > file_size - 6:
raise NativeRuntimeError("native executable PE header is malformed")
handle.seek(pe_offset)
pe_prefix = handle.read(6)
except OSError as exc:
raise NativeRuntimeError("native executable identity is unavailable") from exc
if len(pe_prefix) != 6 or pe_prefix[:4] != b"PE\x00\x00":
raise NativeRuntimeError("native executable PE header is malformed")
return int.from_bytes(pe_prefix[4:6], "little")
def _version_from_info(info: Mapping[str, object]) -> str:
for key in ("CFBundleShortVersionString", "CFBundleVersion"):
value = info.get(key)
if isinstance(value, str) and value.strip():
return value.strip()
# The bundle is still identity/signature checked by validate_app. Older
# local fixtures did not stamp a marketing version, so keep their status
# useful without inventing a version from an untrusted path.
return "plugin-bundled"
def _plugin_bundle_version(info: Mapping[str, object], app_path: Path) -> str:
"""Recover the full alpha version only when signed build fields agree."""
version = _version_from_info(info)
match = re.fullmatch(
r"([0-9]+\.[0-9]+\.[0-9]+)-alpha\.([1-9][0-9]*)",
app_path.parent.name,
)
if match is None:
return version
build = info.get("CFBundleVersion")
if version == match.group(1) and str(build) == match.group(2):
return match.group(0)
return version
def _plugin_bundle_release_version(info: Mapping[str, object]) -> str | None:
"""Expose a signed alpha label only when all stamped bundle fields agree."""
release_version = info.get("ChatGPTMeetingsVersion")
marketing_version = info.get("CFBundleShortVersionString")
build_number = info.get("CFBundleVersion")
if (
not isinstance(release_version, str)
or not isinstance(marketing_version, str)
or not isinstance(build_number, str)
):
return None
match = _NATIVE_ALPHA_RELEASE_VERSION.fullmatch(release_version)
if (
match is None
or marketing_version != ".".join(match.group(index) for index in (1, 2, 3))
or build_number != match.group(4)
):
return None
return release_version
def _build_timestamp_from_info(info: Mapping[str, object]) -> str | None:
return safe_build_timestamp_utc(info.get("ChatGPTMeetingsBuildTimestampUTC"))
from native_runtime_artifacts import (
_framework_entry_identity as _framework_entry_identity,
)
from native_runtime_artifacts import (
_native_build_receipt_path as _native_build_receipt_path,
)
from native_runtime_artifacts import (
_optional_path_identity as _optional_path_identity,
)
from native_runtime_artifacts import (
_path_identity as _path_identity,
)
from native_runtime_artifacts import (
_plugin_bundle_cross_process_lock as _plugin_bundle_cross_process_lock,
)
from native_runtime_artifacts import (
_plugin_bundle_fingerprint as _plugin_bundle_fingerprint,
)
from native_runtime_artifacts import (
_plugin_executable_file_identity as _plugin_executable_file_identity,
)
from native_runtime_artifacts import (
_plugin_executable_path as _plugin_executable_path,
)
from native_runtime_artifacts import (
_plugin_framework_identities as _plugin_framework_identities,
)
from native_runtime_artifacts import (
_plugin_resource_identities as _plugin_resource_identities,
)
from native_runtime_artifacts import (
_plugin_verification_cache_key as _plugin_verification_cache_key,
)
from native_runtime_artifacts import (
_read_strict_local_json as _read_strict_local_json,
)
from native_runtime_artifacts import (
_regular_file_identity_timestamp_matches as _regular_file_identity_timestamp_matches,
)
from native_runtime_artifacts import (
_require_windows_safe_artifact_path as _require_windows_safe_artifact_path,
)
from native_runtime_artifacts import (
_stable_artifact_status_sentinel as _stable_artifact_status_sentinel,
)
from native_runtime_artifacts import (
_verified_cam_distribution_manifest as _verified_cam_distribution_manifest,
)
from native_runtime_artifacts import (
_windows_executable_launch_guard as _windows_executable_launch_guard,
)
from native_runtime_artifacts import (
_windows_plugin_runtime_descriptor_path as _windows_plugin_runtime_descriptor_path,
)
from native_runtime_artifacts import (
_windows_plugin_verification_manifest as _windows_plugin_verification_manifest,
)
from native_runtime_artifacts import (
_windows_production_lock_asset_is_valid as _windows_production_lock_asset_is_valid,
)
from native_runtime_artifacts import (
_windows_production_verification_manifest as _windows_production_verification_manifest,
)
from native_runtime_artifacts import (
_windows_registration_launch_guard as _windows_registration_launch_guard,
)
from native_runtime_artifacts import (
_windows_write_delete_launch_guard as _windows_write_delete_launch_guard,
)
from native_runtime_artifacts import (
restore_plugin_framework_symlinks as restore_plugin_framework_symlinks,
)
from native_runtime_artifacts import (
sha256_regular_file as sha256_regular_file,
)
from native_runtime_artifacts import (
uses_production_windows_bundle as uses_production_windows_bundle,
)
from native_runtime_artifacts import (
validate_app as validate_app,
)
from native_runtime_artifacts import (
validate_handoff_app as validate_handoff_app,
)
from native_runtime_stable import (
_activate_stable_generation as _activate_stable_generation,
)
from native_runtime_stable import (
_cleanup_stable_runtime_generations as _cleanup_stable_runtime_generations,
)
from native_runtime_stable import (
_fsync_stable_runtime_directory as _fsync_stable_runtime_directory,
)
from native_runtime_stable import (
_fsync_stable_runtime_payload as _fsync_stable_runtime_payload,
)
from native_runtime_stable import (
_materialize_stable_generation as _materialize_stable_generation,
)
from native_runtime_stable import (
_official_cache_singleton_is_active as _official_cache_singleton_is_active,
)
from native_runtime_stable import (
_official_cache_version_candidates as _official_cache_version_candidates,
)
from native_runtime_stable import (
_private_runtime_directory as _private_runtime_directory,
)
from native_runtime_stable import (
_read_active_stable_generation as _read_active_stable_generation,
)
from native_runtime_stable import (
_read_stable_runtime_manifest as _read_stable_runtime_manifest,
)
from native_runtime_stable import (
_require_stable_artifact_receipt as _require_stable_artifact_receipt,
)
from native_runtime_stable import (
_require_windows_private_runtime_acl as _require_windows_private_runtime_acl,
)
from native_runtime_stable import (
_require_windows_safe_lexical_runtime_parent as _require_windows_safe_lexical_runtime_parent,
)
from native_runtime_stable import (
_resolve_plugin_handoff_for_launch as _resolve_plugin_handoff_for_launch,
)
from native_runtime_stable import (
_stable_generation_artifact as _stable_generation_artifact,
)
from native_runtime_stable import (
_stable_generation_name as _stable_generation_name,
)
from native_runtime_stable import (
_stable_runtime_family_lock as _stable_runtime_family_lock,
)
from native_runtime_stable import (
_stable_source_verification_manifest as _stable_source_verification_manifest,
)
from native_runtime_stable import (
_strict_stable_runtime_manifest as _strict_stable_runtime_manifest,
)
from native_runtime_stable import (
_validated_stable_generation_manifest as _validated_stable_generation_manifest,
)
from native_runtime_stable import (
_validated_stable_windows_distribution_binding as _validated_stable_windows_distribution_binding,
)
from native_runtime_stable import (
_verified_stable_artifact_metadata as _verified_stable_artifact_metadata,
)
from native_runtime_stable import (
_windows_generation_sha256 as _windows_generation_sha256,
)
from native_runtime_stable import (
_windows_source_licenses_sha256 as _windows_source_licenses_sha256,
)
from native_runtime_stable import (
_windows_third_party_licenses_sha256 as _windows_third_party_licenses_sha256,
)
from native_runtime_stable import (
_write_stable_runtime_manifest as _write_stable_runtime_manifest,
)
from native_runtime_stable import (
plugin_artifact_path as plugin_artifact_path,
)
from native_runtime_stable import (
plugin_cache_family_root as plugin_cache_family_root,
)
from native_runtime_stable import (
require_canonical_plugin_registration as require_canonical_plugin_registration,
)
from native_runtime_stable import (
stable_runtime_artifact_root as stable_runtime_artifact_root,
)
from native_runtime_stable import (
stable_runtime_root as stable_runtime_root,
)
from native_runtime_stable import (
stable_runtime_verification_manifest as stable_runtime_verification_manifest,
)
def process_start_identity(pid: int) -> tuple[int, int]:
"""Read one live process birth identity through its existing verified seam."""
return _process_start_identity(pid)
def _reap_exited_windows_pending_children_locked() -> None:
for key, entry in list(_WINDOWS_PENDING_CHILDREN.items()):
if entry.process.poll() is None:
continue
_WINDOWS_PENDING_CHILDREN.pop(key, None)
def _require_windows_owner_matches_verified_image(
app_path: str,
family_root: Path,
executable_sha256: str,
) -> None:
"""Reject a live same-path child that belongs to another verified image."""
expected_path = str(Path(app_path))
expected_family = str(family_root)
with _WINDOWS_PENDING_CHILDREN_LOCK:
_reap_exited_windows_pending_children_locked()
conflicts = any(
key_path == expected_path
and key_family == expected_family
and key_sha256 != executable_sha256
for key_path, key_family, key_sha256 in _WINDOWS_PENDING_CHILDREN
)
if conflicts:
raise NativeRuntimeLaunchPending(
"authenticated Windows owner cannot be bound to the freshly verified image"
)
def _claim_or_spawn_windows_pending_child(
key: _WindowsPendingChildKey,
app_path: str,
authorize_spawn: Callable[[], None],
*,
environment: Mapping[str, str],
creation_flags: int | None,
required_codex_launch_context: tuple[str, str] | None = None,
) -> _WindowsPendingChild:
"""Claim one exact child observer or spawn it under a bounded registry."""
key_app_path, _key_family_root, expected_executable_sha256 = key
if app_path != key_app_path:
raise NativeRuntimeError("Windows pending child identity does not match")
with _WINDOWS_PENDING_CHILDREN_LOCK:
_reap_exited_windows_pending_children_locked()
entry = _WINDOWS_PENDING_CHILDREN.get(key)
if entry is not None:
if (
required_codex_launch_context is not None
and entry.codex_launch_context != required_codex_launch_context
):
raise NativeRuntimeLaunchPending(
"pending ChatGPT Meetings launch has different Codex context"
)
if required_codex_launch_context is not None:
_windows_codex_launch_arguments(
required_codex_launch_context=required_codex_launch_context
)
if entry.observing:
raise NativeRuntimeLaunchPending(
"ChatGPT Meetings launch is already being observed"
)
entry.observing = True
return entry
if len(_WINDOWS_PENDING_CHILDREN) >= _MAXIMUM_WINDOWS_PENDING_CHILDREN:
raise NativeRuntimeLaunchPending(
"ChatGPT Meetings pending launch capacity is unavailable"
)
# Handoff resolution may block while another owner exits. Revalidate
# canonical activation under the same registry lock that admits the
# only local Popen, immediately before creating a new process.
authorize_spawn()
codex_arguments = (
_windows_codex_launch_arguments(
required_codex_launch_context=required_codex_launch_context
)
if required_codex_launch_context is not None
else _windows_codex_launch_arguments()
)
arguments = [
app_path,
"--expected-executable-sha256",
expected_executable_sha256,
*codex_arguments,
]
if creation_flags is not None:
process = subprocess.Popen(
arguments,
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
env=environment,
creationflags=creation_flags,
)
else:
# Test/future-build overrides can exercise the portable branch on
# non-Windows hosts without passing a Windows-only creation flag.
process = subprocess.Popen(
arguments,
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
env=environment,
start_new_session=True,
)
entry = _WindowsPendingChild(
process=process,
observing=True,
codex_launch_context=required_codex_launch_context,
)
_WINDOWS_PENDING_CHILDREN[key] = entry
return entry
def _release_windows_pending_child_observer(
key: _WindowsPendingChildKey,
entry: _WindowsPendingChild,
*,
keep_if_alive: bool,
) -> None:
with _WINDOWS_PENDING_CHILDREN_LOCK:
if _WINDOWS_PENDING_CHILDREN.get(key) is not entry:
return
return_code = entry.process.poll()
if keep_if_alive and return_code is None:
entry.observing = False
return
_WINDOWS_PENDING_CHILDREN.pop(key, None)
from native_runtime_manager import RuntimeManager as RuntimeManager
def main(argv: Sequence[str]) -> int:
"""Dispatch a native-runtime test gate.
Args:
argv: Command-line arguments after the executable name.
Returns:
The selected gate's exit status, or 64 for unsupported arguments.
"""
if argv and argv[0] == E2E_COMPANION_EXEC_GATE_ARGUMENT:
return _e2e_companion_exec_gate(list(argv[1:]))
if argv and argv[0] == E2E_STDIO_EXEC_GATE_ARGUMENT:
return _e2e_stdio_exec_gate(list(argv[1:]))
return 64
_CodexLaunchContextUnavailableReason = Literal[
"configured_path_unsupported",
"helper_unavailable",
"helper_launcher_manifest",
"helper_metadata_unavailable",
"helper_not_regular",
"helper_not_executable",
"home_unavailable",
"context_not_absolute",
"resolved_path_unsupported",
]
def _validated_codex_launch_context(
*, platform: Literal["windows", "macos"]
) -> tuple[str, str] | None:
"""Resolve bounded executable/home context without forwarding credentials."""
# control_client imports native_runtime, while codex_auth_client imports
# control_client. Keep this reverse edge local so module initialization
# cannot create a cycle. Resolving the helper only inspects it; it never
# executes a candidate while the native companion is starting.
from codex_auth_client import (
CodexAuthError,
_active_codex_home,
_is_macos_dotslash_path,
_resolve_codex_path,
)
def unavailable(reason: _CodexLaunchContextUnavailableReason) -> None:
# This resolver runs only while preparing a native launch. Never attach
# a path or caught exception: either can contain credential material.
log_native_runtime_event(
"codex_launch_context", "unavailable", platform=platform, reason=reason
)
reason: _CodexLaunchContextUnavailableReason = "configured_path_unsupported"
try:
configured_home = os.environ.get("CODEX_HOME", "")
raw_context_paths = (
os.environ.get("CODEX_BIN", ""),
os.environ.get("CODEX_CLI_PATH", ""),
configured_home,
)
if any(
len(value.encode("utf-8")) > 4096
or value != value.strip()
or any(
ord(character) < 32 or 127 <= ord(character) <= 159 or character in "\u2028\u2029"
for character in value
)
for value in raw_context_paths
):
return unavailable("configured_path_unsupported")
reason = "helper_unavailable"
candidate = _resolve_codex_path(None, prefer_platform_helpers=True)
if sys.platform == "darwin" and _is_macos_dotslash_path(candidate):
return unavailable("helper_launcher_manifest")
reason = "helper_metadata_unavailable"
cli_path = Path(candidate).expanduser().resolve(strict=True)
metadata = cli_path.stat()
if not stat.S_ISREG(metadata.st_mode):
return unavailable("helper_not_regular")
if not os.access(cli_path, os.X_OK):
return unavailable("helper_not_executable")
reason = "home_unavailable"
codex_home = _active_codex_home(configured_home or None)
except (AttributeError, CodexAuthError, OSError, RuntimeError, ValueError):
return unavailable(reason)
cli_value = str(cli_path)
home_value = str(codex_home)
if not cli_path.is_absolute() or not codex_home.is_absolute():
return unavailable("context_not_absolute")
try:
unsafe_resolved_path = any(
len(value.encode("utf-8")) > 4096
or value != value.strip()
or any(
ord(character) < 32 or 127 <= ord(character) <= 159 or character in "\u2028\u2029"
for character in value
)
for value in (cli_value, home_value)
)
except UnicodeError:
return unavailable("resolved_path_unsupported")
if unsafe_resolved_path:
return unavailable("resolved_path_unsupported")
return cli_value, home_value
def _windows_codex_launch_arguments(
*, required_codex_launch_context: tuple[str, str] | None = None
) -> list[str]:
"""Give the Windows companion validated context for its own auth owner."""
if sys.platform != "win32" and os.name != "nt":
if required_codex_launch_context is not None:
raise NativeRuntimeError("required Codex launch context is unavailable")
return []
context = _validated_codex_launch_context(platform="windows")
if required_codex_launch_context is not None and context != required_codex_launch_context:
raise NativeRuntimeError("required Codex launch context changed or is unavailable")
if context is None:
return []
cli_value, home_value = context
log_native_runtime_event("codex_launch_context", "available", platform="windows")
return ["--codex-cli-path", cli_value, "--codex-home", home_value]
def _darwin_codex_launch_arguments(
*, required_codex_launch_context: tuple[str, str] | None = None
) -> list[str]:
"""Carry only validated initiating-Codex context across LaunchServices."""
if sys.platform != "darwin":
if required_codex_launch_context is not None:
raise NativeRuntimeError("required Codex launch context is unavailable")
return []
context = _validated_codex_launch_context(platform="macos")
if required_codex_launch_context is not None and context != required_codex_launch_context:
raise NativeRuntimeError("required Codex launch context changed or is unavailable")
if context is None:
return []
cli_value, home_value = context
launcher_pid = os.getppid()
if not 0 < launcher_pid <= 2_147_483_647:
log_native_runtime_event(
"codex_launch_context",
"unavailable",
platform="macos",
reason="launcher_pid_unavailable",
)
if required_codex_launch_context is not None:
raise NativeRuntimeError("required Codex launcher process is unavailable")
return []
log_native_runtime_event("codex_launch_context", "available", platform="macos")
return [
"--args",
"--codex-cli-path",
cli_value,
"--codex-home",
home_value,
"--codex-launcher-pid",
str(launcher_pid),
]
SHA-256: 50ee460d85e94da883ed25426fd30fa002cf01fac7072909eb91d008df7c8675