Generate Python SDK types from repository app-server schemas (#44032)

## Why

Keep Python protocol models aligned with the checked-in app-server schemas and preserve reviewed generated artifacts when staging SDK releases.

## What changed

- Generate SDK types from the schema directory configured in `pyproject.toml`, with a `--schema-dir` override, instead of invoking the pinned runtime binary.
- Refresh Python artifacts through `just write-app-server-schema` for standard repository exports. Skip SDK updates for scratch and experimental exports.
- Regenerate protocol models and notification dispatch, deriving the known payload union from the registry so `Notification.payload` covers every registered event.
- Explicitly allowlist convenience API parameters so new protocol fields do not silently expand method signatures. Preserve existing approval path wrappers.
- Stage SDK releases using checked-in generated files without regenerating them.

## Testing

Add coverage for schema selection, refresh gating and failure handling, release artifact preservation, notification payload typing, and approval path compatibility. Update the generation drift test to use repository schemas.

GitOrigin-RevId: fab350b07cf170258b91fbafa8da384aeb2e3bd7
This commit is contained in:
Ahmed Ibrahim
2026-09-09 03:27:31 +00:00
committed by copyberry
parent fe52d795c9
commit 45134c0463
11 changed files with 2437 additions and 602 deletions
@@ -1,22 +1,28 @@
from __future__ import annotations
from pathlib import Path
from typing import get_type_hints
import pytest
from openai_codex.client import CodexClient, _params_dict
from openai_codex.generated.notification_registry import notification_turn_id
from openai_codex.generated.v2_all import (
AbsolutePathBuf,
AccountRateLimitsUpdatedNotification,
AccountUpdatedNotification,
AgentMessageDeltaNotification,
ApplyPatchGuardianApprovalReviewAction,
ApprovalsReviewer,
AuthRecoveryNotification,
CommandGuardianApprovalReviewAction,
GetAccountResponse,
PlanType,
ReasoningEffort,
ReasoningEffortOption,
ThreadForkParams,
ThreadListParams,
ThreadQueueChangedNotification,
ThreadResumeResponse,
ThreadStartParams,
ThreadTokenUsageUpdatedNotification,
@@ -30,6 +36,31 @@ from openai_codex.types import ThreadSource
ROOT = Path(__file__).resolve().parents[1]
@pytest.mark.parametrize(
("model", "fields"),
[
(
CommandGuardianApprovalReviewAction,
{"type": "command", "command": "pwd", "source": "shell"},
),
(
ApplyPatchGuardianApprovalReviewAction,
{"type": "applyPatch", "files": [AbsolutePathBuf("/workspace/file")]},
),
],
)
def test_approval_review_paths_preserve_existing_wrappers(model, fields) -> None:
action = model(cwd=AbsolutePathBuf("/workspace"), **fields)
expected = {
**fields,
"cwd": "/workspace",
}
if "files" in expected:
expected["files"] = ["/workspace/file"]
assert action.model_dump(mode="json") == expected
assert isinstance(action.cwd, AbsolutePathBuf)
def test_generated_params_models_are_snake_case_and_dump_by_alias() -> None:
params = ThreadListParams(search_term="needle", limit=5)
@@ -209,6 +240,44 @@ def test_unknown_notifications_fall_back_to_unknown_payloads() -> None:
assert event.payload.params["msg"] == {"type": "turn_aborted"}
@pytest.mark.parametrize(
("method", "params", "expected"),
[
(
"modelProvider/authRecoveryCompleted",
{
"provider": "openai",
"message": "Authentication recovered",
"threadId": "thread-1",
"turnId": "turn-1",
},
AuthRecoveryNotification(
provider="openai",
message="Authentication recovered",
thread_id="thread-1",
turn_id="turn-1",
),
),
(
"thread/queue/changed",
{"threadId": "thread-1"},
ThreadQueueChangedNotification(thread_id="thread-1"),
),
("warning", {"message": "heads up"}, WarningNotification(message="heads up")),
(
"future/notification",
{"newField": "value"},
UnknownNotification(params={"newField": "value"}),
),
],
)
def test_decoded_notifications_match_the_declared_payload_type(method, params, expected) -> None:
event = CodexClient()._coerce_notification(method, params)
assert event == Notification(method=method, payload=expected)
assert isinstance(event.payload, get_type_hints(Notification)["payload"])
def test_invalid_notification_payload_falls_back_to_unknown() -> None:
client = CodexClient()
event = client._coerce_notification("thread/tokenUsage/updated", {"threadId": "missing"})