Document pinned schema generation helpers

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Ahmed Ibrahim
2026-05-09 10:20:34 +03:00
parent b7635f4d77
commit 242ca6d8fd
6 changed files with 30 additions and 2 deletions

View File

@@ -35,10 +35,12 @@ def python_runtime_root() -> Path:
def sdk_pyproject_path() -> Path:
"""Return the SDK pyproject file that owns package pins and versions."""
return sdk_root() / "pyproject.toml"
def schema_bundle_path(schema_dir: Path) -> Path:
"""Return the aggregate v2 schema bundle emitted by the runtime binary."""
return schema_dir / "codex_app_server_protocol.v2.schemas.json"
@@ -82,6 +84,7 @@ def current_sdk_version() -> str:
def pinned_runtime_version() -> str:
"""Read the exact runtime package pin used for schema generation."""
pyproject_text = sdk_pyproject_path().read_text()
match = re.search(r"(?ms)^dependencies = \[(.*?)\]$", pyproject_text)
if match is None:
@@ -102,6 +105,7 @@ def pinned_runtime_version() -> str:
def pinned_runtime_codex_path() -> Path:
"""Return the bundled Codex binary from the installed pinned runtime wheel."""
expected_version = pinned_runtime_version()
try:
installed_version = importlib.metadata.version(RUNTIME_DISTRIBUTION_NAME)
@@ -534,6 +538,7 @@ def _annotate_schema(value: Any, base: str | None = None) -> None:
def generate_schema_from_pinned_runtime(schema_dir: Path) -> Path:
"""Generate app-server schemas by invoking the installed pinned runtime binary."""
codex_path = pinned_runtime_codex_path()
if schema_dir.exists():
shutil.rmtree(schema_dir)
@@ -552,6 +557,7 @@ def generate_schema_from_pinned_runtime(schema_dir: Path) -> Path:
def _normalized_schema_bundle_text(schema_dir: Path) -> str:
"""Normalize the schema bundle before feeding it to the Python type generator."""
schema = json.loads(schema_bundle_path(schema_dir).read_text())
definitions = schema.get("definitions", {})
if isinstance(definitions, dict):
@@ -565,6 +571,7 @@ def _normalized_schema_bundle_text(schema_dir: Path) -> str:
def generate_v2_all(schema_dir: Path) -> None:
"""Regenerate the Pydantic v2 protocol model module from runtime schemas."""
out_path = sdk_root() / "src" / "codex_app_server" / "generated" / "v2_all.py"
out_dir = out_path.parent
old_package_dir = out_dir / "v2_all"
@@ -611,6 +618,7 @@ def generate_v2_all(schema_dir: Path) -> None:
def _notification_specs(schema_dir: Path) -> list[tuple[str, str]]:
"""Map each server notification method to its generated payload model class."""
server_notifications = json.loads(
(schema_dir / "ServerNotification.json").read_text()
)
@@ -653,7 +661,7 @@ def _notification_turn_id_specs(
schema_dir: Path,
specs: list[tuple[str, str]],
) -> tuple[list[str], list[str]]:
"""Classify generated notification payloads by where the turn id lives."""
"""Classify notification payloads by where their turn id is carried."""
server_notifications = json.loads(
(schema_dir / "ServerNotification.json").read_text()
)
@@ -690,7 +698,7 @@ def _type_tuple_source(class_names: list[str]) -> str:
def generate_notification_registry(schema_dir: Path) -> None:
"""Regenerate notification models and routing metadata from generated schemas."""
"""Regenerate notification dispatch metadata from the runtime notification schema."""
out = (
sdk_root()
/ "src"
@@ -824,6 +832,7 @@ def _camel_to_snake(name: str) -> str:
def _load_public_fields(
module_name: str, class_name: str, *, exclude: set[str] | None = None
) -> list[PublicFieldSpec]:
"""Load generated model fields used to render the ergonomic public methods."""
exclude = exclude or set()
if module_name == "codex_app_server.generated.v2_all":
module = _load_generated_v2_all_module()
@@ -851,6 +860,7 @@ def _load_public_fields(
def _load_generated_v2_all_module() -> types.ModuleType:
"""Import the freshly generated v2_all module without importing package init."""
module_name = "_codex_app_server_generated_v2_all_for_artifacts"
sys.modules.pop(module_name, None)
module_path = sdk_root() / "src" / "codex_app_server" / "generated" / "v2_all.py"
@@ -1072,6 +1082,7 @@ def _render_async_thread_block(
def generate_public_api_flat_methods() -> None:
"""Regenerate the public convenience methods from generated protocol models."""
src_dir = sdk_root() / "src"
public_api_path = src_dir / "codex_app_server" / "api.py"
if not public_api_path.exists():
@@ -1141,6 +1152,7 @@ def generate_public_api_flat_methods() -> None:
def generate_types_from_schema_dir(schema_dir: Path) -> None:
"""Regenerate every SDK artifact derived from an existing schema directory."""
# v2_all is the authoritative generated surface.
generate_v2_all(schema_dir)
generate_notification_registry(schema_dir)
@@ -1148,6 +1160,7 @@ def generate_types_from_schema_dir(schema_dir: Path) -> None:
def generate_types() -> None:
"""Generate schemas from the pinned runtime and then refresh SDK artifacts."""
with tempfile.TemporaryDirectory(prefix="codex-python-schema-") as td:
schema_dir = generate_schema_from_pinned_runtime(Path(td) / "schema")
generate_types_from_schema_dir(schema_dir)

View File

@@ -16,6 +16,7 @@ ROOT = Path(__file__).resolve().parents[1]
def _load_update_script_module():
"""Load the maintenance script as a module so tests exercise real helpers."""
script_path = ROOT / "scripts" / "update_sdk_artifacts.py"
spec = importlib.util.spec_from_file_location("update_sdk_artifacts", script_path)
if spec is None or spec.loader is None:
@@ -27,6 +28,7 @@ def _load_update_script_module():
def _load_runtime_setup_module():
"""Load runtime setup without importing the SDK package under test."""
runtime_setup_path = ROOT / "_runtime_setup.py"
spec = importlib.util.spec_from_file_location("_runtime_setup", runtime_setup_path)
if spec is None or spec.loader is None:
@@ -40,11 +42,13 @@ def _load_runtime_setup_module():
def test_generation_has_single_maintenance_entrypoint_script() -> None:
"""Keep artifact workflows routed through one script instead of side entrypoints."""
scripts = sorted(p.name for p in (ROOT / "scripts").glob("*.py"))
assert scripts == ["update_sdk_artifacts.py"]
def test_generate_types_wires_all_generation_steps() -> None:
"""The type generation command should refresh every schema-derived artifact."""
source = (ROOT / "scripts" / "update_sdk_artifacts.py").read_text()
tree = ast.parse(source)
@@ -74,6 +78,7 @@ def test_generate_types_wires_all_generation_steps() -> None:
def _load_runtime_schema_bundle(tmp_path: Path) -> dict:
"""Ask the pinned runtime package for a real schema bundle used by tests."""
script = _load_update_script_module()
schema_dir = script.generate_schema_from_pinned_runtime(tmp_path / "schema")
return json.loads(script.schema_bundle_path(schema_dir).read_text())
@@ -82,6 +87,7 @@ def _load_runtime_schema_bundle(tmp_path: Path) -> dict:
def test_schema_normalization_only_flattens_string_literal_oneofs(
tmp_path: Path,
) -> None:
"""Schema normalization should only flatten the enum-shaped oneOf variants."""
script = _load_update_script_module()
schema = _load_runtime_schema_bundle(tmp_path)
definitions = schema["definitions"]
@@ -107,6 +113,7 @@ def test_schema_normalization_only_flattens_string_literal_oneofs(
def test_python_codegen_schema_annotation_adds_stable_variant_titles(
tmp_path: Path,
) -> None:
"""Schema annotations should give generated protocol classes stable names."""
script = _load_update_script_module()
schema = _load_runtime_schema_bundle(tmp_path)
script._annotate_schema(schema)

View File

@@ -50,6 +50,7 @@ def test_generated_v2_bundle_has_single_shared_plan_type_definition() -> None:
def test_thread_resume_response_accepts_auto_review_reviewer() -> None:
"""Generated response models should keep accepting the auto review enum value."""
response = ThreadResumeResponse.model_validate(
{
"approvalPolicy": "on-request",
@@ -66,6 +67,7 @@ def test_thread_resume_response_accepts_auto_review_reviewer() -> None:
"id": "thread-1",
"modelProvider": "openai",
"preview": "",
# The pinned runtime schema requires the session id on threads.
"sessionId": "session-1",
"source": "cli",
"status": {"type": "idle"},

View File

@@ -15,6 +15,7 @@ GENERATED_TARGETS = [
def _snapshot_target(root: Path, rel_path: Path) -> dict[str, bytes] | bytes | None:
"""Capture one generated artifact so regeneration drift is easy to compare."""
target = root / rel_path
if not target.exists():
return None
@@ -29,6 +30,7 @@ def _snapshot_target(root: Path, rel_path: Path) -> dict[str, bytes] | bytes | N
def _snapshot_targets(root: Path) -> dict[str, dict[str, bytes] | bytes | None]:
"""Capture all checked-in generated artifacts before and after regeneration."""
return {
str(rel_path): _snapshot_target(root, rel_path)
for rel_path in GENERATED_TARGETS
@@ -36,6 +38,7 @@ def _snapshot_targets(root: Path) -> dict[str, dict[str, bytes] | bytes | None]:
def test_generated_files_are_up_to_date():
"""Regenerating from the pinned runtime package should leave artifacts unchanged."""
before = _snapshot_targets(ROOT)
# Regenerate contract artifacts via the pinned runtime package, not a local

View File

@@ -82,6 +82,7 @@ def _item_completed_notification(
text: str = "final text",
phase: MessagePhase | None = None,
) -> Notification:
"""Build a realistic completed-item notification accepted by generated models."""
item: dict[str, object] = {
"id": "item-1",
"text": text,
@@ -93,6 +94,7 @@ def _item_completed_notification(
method="item/completed",
payload=ItemCompletedNotification.model_validate(
{
# The pinned runtime schema requires completion timestamps.
"completedAtMs": 1,
"item": item,
"threadId": thread_id,

View File

@@ -54,6 +54,7 @@ def test_package_includes_py_typed_marker() -> None:
def test_generated_public_signatures_are_snake_case_and_typed() -> None:
"""Generated convenience methods should expose typed Pythonic keyword names."""
expected = {
Codex.thread_start: [
"approval_policy",