From e502792f495d7099dcb03cb70050ce2efa04b335 Mon Sep 17 00:00:00 2001 From: Ahmed Ibrahim Date: Mon, 8 Jun 2026 12:07:38 -0700 Subject: [PATCH] Add dedicated Python goal operations --- sdk/python/docs/api-reference.md | 20 +- sdk/python/docs/getting-started.md | 11 +- sdk/python/examples/16_goal_turns/async.py | 6 +- sdk/python/examples/16_goal_turns/sync.py | 2 +- sdk/python/scripts/update_sdk_artifacts.py | 51 +--- sdk/python/src/openai_codex/_goal.py | 27 ++- .../src/openai_codex/_message_router.py | 32 +-- sdk/python/src/openai_codex/api.py | 70 +++--- sdk/python/src/openai_codex/async_client.py | 57 ++++- sdk/python/src/openai_codex/client.py | 91 +++++-- sdk/python/tests/app_server_harness.py | 7 +- .../tests/test_app_server_goal_turns.py | 225 +++++++++--------- .../tests/test_public_api_signatures.py | 34 ++- 13 files changed, 375 insertions(+), 258 deletions(-) diff --git a/sdk/python/docs/api-reference.md b/sdk/python/docs/api-reference.md index 0ee6689c64..07b0a2b29a 100644 --- a/sdk/python/docs/api-reference.md +++ b/sdk/python/docs/api-reference.md @@ -150,16 +150,20 @@ attempt. API-key login completes synchronously and does not return a handle. ### Thread -- `run(input: str | Input, *, goal: bool = False, approval_mode=None, cwd=None, effort=None, model=None, output_schema=None, personality=None, sandbox: Sandbox | None = None, service_tier=None, summary=None) -> TurnResult` -- `turn(input: str | Input, *, goal: bool = False, approval_mode=None, cwd=None, effort=None, model=None, output_schema=None, personality=None, sandbox: Sandbox | None = None, service_tier=None, summary=None) -> TurnHandle` +- `run(input: str | Input, *, approval_mode=None, cwd=None, effort=None, model=None, output_schema=None, personality=None, sandbox: Sandbox | None = None, service_tier=None, summary=None) -> TurnResult` +- `turn(input: str | Input, *, approval_mode=None, cwd=None, effort=None, model=None, output_schema=None, personality=None, sandbox: Sandbox | None = None, service_tier=None, summary=None) -> TurnHandle` +- `run_goal(objective: str) -> TurnResult` +- `start_goal(objective: str) -> TurnHandle` - `read(*, include_turns: bool = False) -> ThreadReadResponse` - `set_name(name: str) -> ThreadSetNameResponse` - `compact() -> ThreadCompactStartResponse` ### AsyncThread -- `run(input: str | Input, *, goal: bool = False, approval_mode=None, cwd=None, effort=None, model=None, output_schema=None, personality=None, sandbox: Sandbox | None = None, service_tier=None, summary=None) -> Awaitable[TurnResult]` -- `turn(input: str | Input, *, goal: bool = False, approval_mode=None, cwd=None, effort=None, model=None, output_schema=None, personality=None, sandbox: Sandbox | None = None, service_tier=None, summary=None) -> Awaitable[AsyncTurnHandle]` +- `run(input: str | Input, *, approval_mode=None, cwd=None, effort=None, model=None, output_schema=None, personality=None, sandbox: Sandbox | None = None, service_tier=None, summary=None) -> Awaitable[TurnResult]` +- `turn(input: str | Input, *, approval_mode=None, cwd=None, effort=None, model=None, output_schema=None, personality=None, sandbox: Sandbox | None = None, service_tier=None, summary=None) -> Awaitable[AsyncTurnHandle]` +- `run_goal(objective: str) -> Awaitable[TurnResult]` +- `start_goal(objective: str) -> Awaitable[AsyncTurnHandle]` - `read(*, include_turns: bool = False) -> Awaitable[ThreadReadResponse]` - `set_name(name: str) -> Awaitable[ThreadSetNameResponse]` - `compact() -> Awaitable[ThreadCompactStartResponse]` @@ -184,9 +188,11 @@ phase-less assistant message item. Use `turn(...)` when you need low-level turn control (`stream()`, `steer()`, `interrupt()`) before collecting the turn result. -Pass `goal=True` to either method for an objective that can continue through -multiple internal turns. Streaming, steering, interruption, and the returned -`TurnResult` still present one logical turn with one stable ID. +Use `run_goal(...)` or `start_goal(...)` for an objective that can continue +through multiple internal turns. Goal operations require an idle, persisted +thread and use its existing configuration. Starting one replaces any stored +goal. Streaming, steering, interruption, and the returned `TurnResult` still +present one logical turn with one stable ID. ## Sandbox diff --git a/sdk/python/docs/getting-started.md b/sdk/python/docs/getting-started.md index 78547cde8d..06d899d5d9 100644 --- a/sdk/python/docs/getting-started.md +++ b/sdk/python/docs/getting-started.md @@ -72,15 +72,16 @@ with Codex() as codex: Use `Thread.turn(...)` when you need a `TurnHandle` for streaming, steering, or interrupting an active turn. -For a longer objective, opt into goal mode on the operation producing the -result: +For a longer objective on an idle, persisted thread, run a dedicated goal +operation: ```python -result = thread.run("Improve the benchmark coverage in this repository.", goal=True) +result = thread.run_goal("Improve the benchmark coverage in this repository.") ``` -Goal mode may continue working internally, but it streams and returns as one -logical turn. +A goal may continue working internally, but it streams and returns as one +logical turn. It uses the thread's existing configuration and replaces any +stored goal. ## 4. Choose Sandbox Access diff --git a/sdk/python/examples/16_goal_turns/async.py b/sdk/python/examples/16_goal_turns/async.py index 7a390d6323..e93c0a0ff8 100644 --- a/sdk/python/examples/16_goal_turns/async.py +++ b/sdk/python/examples/16_goal_turns/async.py @@ -17,10 +17,8 @@ from openai_codex import AsyncCodex async def main() -> None: async with AsyncCodex(config=runtime_config()) as codex: thread = await codex.thread_start() - result = await thread.run( - "Improve the benchmark coverage in this repository.", - goal=True, - ) + goal = await thread.start_goal("Improve the benchmark coverage in this repository.") + result = await goal.run() print(result.final_response) diff --git a/sdk/python/examples/16_goal_turns/sync.py b/sdk/python/examples/16_goal_turns/sync.py index d82f59c721..a1749c7848 100644 --- a/sdk/python/examples/16_goal_turns/sync.py +++ b/sdk/python/examples/16_goal_turns/sync.py @@ -13,5 +13,5 @@ from openai_codex import Codex with Codex(config=runtime_config()) as codex: thread = codex.thread_start() - result = thread.run("Improve the benchmark coverage in this repository.", goal=True) + result = thread.run_goal("Improve the benchmark coverage in this repository.") print(result.final_response) diff --git a/sdk/python/scripts/update_sdk_artifacts.py b/sdk/python/scripts/update_sdk_artifacts.py index 683ed97247..b9f0881f57 100755 --- a/sdk/python/scripts/update_sdk_artifacts.py +++ b/sdk/python/scripts/update_sdk_artifacts.py @@ -527,18 +527,6 @@ def _normalized_schema_bundle_text(schema_dir: Path) -> str: schema = json.loads(schema_bundle_path(schema_dir).read_text()) definitions = schema.get("definitions", {}) if isinstance(definitions, dict): - turn_start = definitions.get("TurnStartParams") - if isinstance(turn_start, dict): - properties = turn_start.get("properties") - if isinstance(properties, dict): - properties["goal"] = { - "default": False, - "description": ( - "Replace the thread's active goal with an objective derived " - "from this turn's text input." - ), - "type": "boolean", - } for definition in definitions.values(): if isinstance(definition, dict): _flatten_string_enum_one_of(definition) @@ -1121,7 +1109,6 @@ def _render_thread_block( " self,", " input: RunInput,", " *,", - " goal: bool = False,", *_approval_mode_override_signature_lines(), *_kw_signature_lines(turn_fields), " ) -> TurnHandle:", @@ -1134,16 +1121,8 @@ def _render_thread_block( *_approval_mode_model_arg_lines(), *_model_arg_lines(turn_fields), " )", - " goal_state = self._client.register_goal_operation(self.id) if goal else None", - " try:", - " turn = self._client.turn_start(self.id, wire_input, params=params, goal=goal)", - " except BaseException:", - " if goal_state is not None:", - " self._client.unregister_goal_operation(goal_state)", - " raise", - " if goal_state is not None:", - " self._client.bind_goal_operation(goal_state, turn.turn.id)", - " return TurnHandle(self._client, self.id, turn.turn.id, _goal=goal_state)", + " turn = self._client.turn_start(self.id, wire_input, params=params)", + " return TurnHandle(self._client, self.id, turn.turn.id)", ] return "\n".join(lines) @@ -1156,7 +1135,6 @@ def _render_async_thread_block( " self,", " input: RunInput,", " *,", - " goal: bool = False,", *_approval_mode_override_signature_lines(), *_kw_signature_lines(turn_fields), " ) -> AsyncTurnHandle:", @@ -1170,21 +1148,12 @@ def _render_async_thread_block( *_approval_mode_model_arg_lines(), *_model_arg_lines(turn_fields), " )", - " goal_state = self._codex._client.register_goal_operation(self.id) if goal else None", - " try:", - " turn = await self._codex._client.turn_start(", - " self.id,", - " wire_input,", - " params=params,", - " goal=goal,", - " )", - " except BaseException:", - " if goal_state is not None:", - " self._codex._client.unregister_goal_operation(goal_state)", - " raise", - " if goal_state is not None:", - " self._codex._client.bind_goal_operation(goal_state, turn.turn.id)", - " return AsyncTurnHandle(self._codex, self.id, turn.turn.id, _goal=goal_state)", + " turn = await self._codex._client.turn_start(", + " self.id,", + " wire_input,", + " params=params,", + " )", + " return AsyncTurnHandle(self._codex, self.id, turn.turn.id)", ] return "\n".join(lines) @@ -1226,9 +1195,7 @@ def generate_public_api_flat_methods() -> None: turn_start_fields = _load_public_fields( "openai_codex.generated.v2_all", "TurnStartParams", - # `goal` has a stable bool default and private routing setup, so render - # it explicitly rather than inheriting the generated model default. - exclude={"thread_id", "input", "client_user_message_id", "goal", *approval_fields}, + exclude={"thread_id", "input", "client_user_message_id", *approval_fields}, ) turn_start_fields = _replace_public_sandbox_field(turn_start_fields, wire_name="sandbox_policy") diff --git a/sdk/python/src/openai_codex/_goal.py b/sdk/python/src/openai_codex/_goal.py index b9ef31eb52..4e64211958 100644 --- a/sdk/python/src/openai_codex/_goal.py +++ b/sdk/python/src/openai_codex/_goal.py @@ -1,5 +1,6 @@ import queue import threading +import time from collections import deque from dataclasses import dataclass, field from typing import AsyncIterator, Awaitable, Callable, Iterator @@ -55,17 +56,12 @@ class _GoalOperationState: _failure: BaseException | None = None _finished: bool = False - def bind(self, logical_turn_id: str) -> None: - with self._condition: - self.logical_turn_id = logical_turn_id - if self.current_turn_id is None and self.completed_turn is None: - self.current_turn_id = logical_turn_id - self._condition.notify_all() - def observe(self, notification: Notification) -> None: payload = notification.payload with self._condition: if isinstance(payload, TurnStartedNotification): + if self.logical_turn_id is None: + self.logical_turn_id = payload.turn.id self.current_turn_id = payload.turn.id if self.started_turn is None: self.started_turn = payload.turn @@ -75,6 +71,8 @@ class _GoalOperationState: self.current_turn_id = None elif isinstance(payload, ThreadGoalUpdatedNotification): self.status = payload.goal.status + if self.status == ThreadGoalStatus.active: + self.cleared = False elif isinstance(payload, ThreadGoalClearedNotification): self.cleared = True elif isinstance(payload, ItemCompletedNotification): @@ -93,6 +91,19 @@ class _GoalOperationState: self._condition.notify_all() self._notifications.put(notification) + def wait_for_start(self, timeout: float) -> str | None: + """Wait for the runtime-generated first turn without consuming its event.""" + deadline = time.monotonic() + timeout + with self._condition: + while self.started_turn is None or self.logical_turn_id is None: + if self._failure is not None: + raise self._failure + remaining = deadline - time.monotonic() + if remaining <= 0: + return None + self._condition.wait(remaining) + return self.logical_turn_id + def fail(self, exc: BaseException) -> None: with self._condition: self._failure = exc @@ -256,6 +267,8 @@ class _GoalStreamCursor: events = [_logical_notification(notification, logical_turn_id)] if isinstance(payload, ThreadGoalUpdatedNotification): self.status = payload.goal.status + if self.status == ThreadGoalStatus.active: + self.cleared = False events = [] elif isinstance(payload, ThreadGoalClearedNotification): self.cleared = True diff --git a/sdk/python/src/openai_codex/_message_router.py b/sdk/python/src/openai_codex/_message_router.py index c77e9c9b08..26c7dd5ab6 100644 --- a/sdk/python/src/openai_codex/_message_router.py +++ b/sdk/python/src/openai_codex/_message_router.py @@ -224,22 +224,8 @@ class MessageRouter: goal_operation.fail(exc) self._global_notifications.put(exc) - def _notification_login_id(self, notification: Notification) -> str | None: - """Extract the login attempt id from completion notifications.""" - if notification.method != "account/login/completed": - return None - - payload = notification.payload - if isinstance(payload, AccountLoginCompletedNotification): - return payload.login_id - if isinstance(payload, UnknownNotification): - raw_login_id = payload.params.get("loginId") - if isinstance(raw_login_id, str): - return raw_login_id - return None - def _notification_turn_id(self, notification: Notification) -> str | None: - """Extract routing ids from known generated payloads or raw unknown payloads.""" + """Extract routing ids from generated metadata or raw unknown payloads.""" payload = notification.payload if isinstance(payload, UnknownNotification): raw_turn_id = payload.params.get("turnId") @@ -254,9 +240,23 @@ class MessageRouter: return notification_turn_id(payload) def _notification_thread_id(self, notification: Notification) -> str | None: - """Extract thread ids from known generated payloads or raw payloads.""" + """Extract thread ids from generated metadata or raw unknown payloads.""" payload = notification.payload if isinstance(payload, UnknownNotification): raw_thread_id = payload.params.get("threadId") return raw_thread_id if isinstance(raw_thread_id, str) else None return notification_thread_id(payload) + + def _notification_login_id(self, notification: Notification) -> str | None: + """Extract the login attempt id from completion notifications.""" + if notification.method != "account/login/completed": + return None + + payload = notification.payload + if isinstance(payload, AccountLoginCompletedNotification): + return payload.login_id + if isinstance(payload, UnknownNotification): + raw_login_id = payload.params.get("loginId") + if isinstance(raw_login_id, str): + return raw_login_id + return None diff --git a/sdk/python/src/openai_codex/api.py b/sdk/python/src/openai_codex/api.py index 54852dedd1..a856c83f9c 100644 --- a/sdk/python/src/openai_codex/api.py +++ b/sdk/python/src/openai_codex/api.py @@ -79,6 +79,15 @@ from .generated.v2_all import ( from .models import InitializeResponse, JsonObject, Notification +def _normalize_goal_objective(objective: str) -> str: + if not isinstance(objective, str): + raise TypeError("goal objective must be a string") + objective = objective.strip() + if not objective: + raise ValueError("goal objective must not be empty") + return objective + + def _active_turn_id_from_error(exc: InvalidRequestError) -> str | None: match = re.search(r" but found `?([^`]+)`?$", exc.message) return match.group(1) if match is not None else None @@ -561,7 +570,6 @@ class Thread: self, input: RunInput, *, - goal: bool = False, approval_mode: ApprovalMode | None = None, cwd: str | None = None, effort: ReasoningEffort | None = None, @@ -575,7 +583,6 @@ class Thread: """Run a complete turn and collect its final result.""" turn = self.turn( input, - goal=goal, approval_mode=approval_mode, cwd=cwd, effort=effort, @@ -592,12 +599,15 @@ class Thread: finally: stream.close() + def run_goal(self, objective: str) -> TurnResult: + """Run a persisted goal to completion as one logical turn.""" + return self.start_goal(objective).run() + # BEGIN GENERATED: Thread.flat_methods def turn( self, input: RunInput, *, - goal: bool = False, approval_mode: ApprovalMode | None = None, cwd: str | None = None, effort: ReasoningEffort | None = None, @@ -625,19 +635,17 @@ class Thread: service_tier=service_tier, summary=summary, ) - goal_state = self._client.register_goal_operation(self.id) if goal else None - try: - turn = self._client.turn_start(self.id, wire_input, params=params, goal=goal) - except BaseException: - if goal_state is not None: - self._client.unregister_goal_operation(goal_state) - raise - if goal_state is not None: - self._client.bind_goal_operation(goal_state, turn.turn.id) - return TurnHandle(self._client, self.id, turn.turn.id, _goal=goal_state) + turn = self._client.turn_start(self.id, wire_input, params=params) + return TurnHandle(self._client, self.id, turn.turn.id) # END GENERATED: Thread.flat_methods + def start_goal(self, objective: str) -> TurnHandle: + """Activate a persisted goal and return its logical turn handle.""" + objective = _normalize_goal_objective(objective) + state, turn_id = self._client.start_goal_operation(self.id, objective) + return TurnHandle(self._client, self.id, turn_id, _goal=state) + def read(self, *, include_turns: bool = False) -> ThreadReadResponse: """Read this thread, optionally including its turn history.""" return self._client.thread_read(self.id, include_turns=include_turns) @@ -660,7 +668,6 @@ class AsyncThread: self, input: RunInput, *, - goal: bool = False, approval_mode: ApprovalMode | None = None, cwd: str | None = None, effort: ReasoningEffort | None = None, @@ -674,7 +681,6 @@ class AsyncThread: """Run a complete turn asynchronously and collect its final result.""" turn = await self.turn( input, - goal=goal, approval_mode=approval_mode, cwd=cwd, effort=effort, @@ -691,12 +697,16 @@ class AsyncThread: finally: await stream.aclose() + async def run_goal(self, objective: str) -> TurnResult: + """Run a persisted goal asynchronously as one logical turn.""" + goal = await self.start_goal(objective) + return await goal.run() + # BEGIN GENERATED: AsyncThread.flat_methods async def turn( self, input: RunInput, *, - goal: bool = False, approval_mode: ApprovalMode | None = None, cwd: str | None = None, effort: ReasoningEffort | None = None, @@ -725,24 +735,22 @@ class AsyncThread: service_tier=service_tier, summary=summary, ) - goal_state = self._codex._client.register_goal_operation(self.id) if goal else None - try: - turn = await self._codex._client.turn_start( - self.id, - wire_input, - params=params, - goal=goal, - ) - except BaseException: - if goal_state is not None: - self._codex._client.unregister_goal_operation(goal_state) - raise - if goal_state is not None: - self._codex._client.bind_goal_operation(goal_state, turn.turn.id) - return AsyncTurnHandle(self._codex, self.id, turn.turn.id, _goal=goal_state) + turn = await self._codex._client.turn_start( + self.id, + wire_input, + params=params, + ) + return AsyncTurnHandle(self._codex, self.id, turn.turn.id) # END GENERATED: AsyncThread.flat_methods + async def start_goal(self, objective: str) -> AsyncTurnHandle: + """Activate a persisted goal and return its async logical turn handle.""" + await self._codex._ensure_initialized() + objective = _normalize_goal_objective(objective) + state, turn_id = await self._codex._client.start_goal_operation(self.id, objective) + return AsyncTurnHandle(self._codex, self.id, turn_id, _goal=state) + async def read(self, *, include_turns: bool = False) -> ThreadReadResponse: """Read this thread, optionally including its turn history.""" await self._codex._ensure_initialized() diff --git a/sdk/python/src/openai_codex/async_client.py b/sdk/python/src/openai_codex/async_client.py index 5ea34ead8e..b1f2f6759c 100644 --- a/sdk/python/src/openai_codex/async_client.py +++ b/sdk/python/src/openai_codex/async_client.py @@ -22,7 +22,9 @@ from .generated.v2_all import ( ThreadCompactStartResponse, ThreadForkParams as V2ThreadForkParams, ThreadForkResponse, + ThreadGoalClearResponse, ThreadGoalSetResponse, + ThreadGoalStatus, ThreadListParams as V2ThreadListParams, ThreadListResponse, ThreadReadResponse, @@ -113,10 +115,6 @@ class AsyncCodexClient: """Register a logical goal route on the wrapped sync client.""" return self._sync.register_goal_operation(thread_id) - def bind_goal_operation(self, state: _GoalOperationState, turn_id: str) -> None: - """Bind a logical goal route to its stable turn id.""" - self._sync.bind_goal_operation(state, turn_id) - def unregister_goal_operation(self, state: _GoalOperationState) -> None: """Release one logical goal route.""" self._sync.unregister_goal_operation(state) @@ -206,17 +204,63 @@ class AsyncCodexClient: """Start thread compaction using the wrapped sync client.""" return await self._call_sync(self._sync.thread_compact, thread_id) + async def thread_goal_clear(self, thread_id: str) -> ThreadGoalClearResponse: + """Clear the persisted goal through the wrapped sync client.""" + return await self._call_sync(self._sync.thread_goal_clear, thread_id) + + async def thread_goal_set( + self, + thread_id: str, + *, + objective: str | None = None, + status: ThreadGoalStatus | None = None, + ) -> ThreadGoalSetResponse: + """Create or update a persisted goal through the wrapped sync client.""" + return await self._call_sync( + self._sync.thread_goal_set, + thread_id, + objective=objective, + status=status, + ) + async def pause_goal(self, thread_id: str) -> ThreadGoalSetResponse: """Pause the active goal through the wrapped sync client.""" return await self._call_sync(self._sync.pause_goal, thread_id) + async def start_goal_operation( + self, + thread_id: str, + objective: str, + ) -> tuple[_GoalOperationState, str]: + """Start a logical goal through the wrapped sync client.""" + operation = asyncio.create_task( + asyncio.to_thread( + self._sync.start_goal_operation, + thread_id, + objective, + ) + ) + try: + return await asyncio.shield(operation) + except asyncio.CancelledError: + try: + state, _ = await operation + except BaseException: + pass + else: + try: + await self.pause_goal(thread_id) + except Exception: + pass + state.finish() + self.unregister_goal_operation(state) + raise + async def turn_start( self, thread_id: str, input_items: list[JsonObject] | JsonObject | str, params: V2TurnStartParams | JsonObject | None = None, - *, - goal: bool = False, ) -> TurnStartResponse: """Start a turn using the wrapped sync client.""" return await self._call_sync( @@ -224,7 +268,6 @@ class AsyncCodexClient: thread_id, input_items, params, - goal=goal, ) async def turn_interrupt(self, thread_id: str, turn_id: str) -> TurnInterruptResponse: diff --git a/sdk/python/src/openai_codex/client.py b/sdk/python/src/openai_codex/client.py index c8e0b09191..4ef54108e9 100644 --- a/sdk/python/src/openai_codex/client.py +++ b/sdk/python/src/openai_codex/client.py @@ -13,7 +13,7 @@ from pydantic import BaseModel from ._goal import _GoalOperationState from ._message_router import MessageRouter from ._version import __version__ as SDK_VERSION -from .errors import CodexError, TransportClosedError +from .errors import CodexError, InvalidRequestError, TransportClosedError from .generated.notification_registry import NOTIFICATION_MODELS from .generated.v2_all import ( AccountLoginCompletedNotification, @@ -23,6 +23,7 @@ from .generated.v2_all import ( ChatgptLoginAccountResponse, GetAccountParams as V2GetAccountParams, GetAccountResponse, + IdleThreadStatus, LoginAccountParams as V2LoginAccountParams, LoginAccountResponse, LogoutAccountResponse, @@ -31,6 +32,7 @@ from .generated.v2_all import ( ThreadCompactStartResponse, ThreadForkParams as V2ThreadForkParams, ThreadForkResponse, + ThreadGoalClearResponse, ThreadGoalSetResponse, ThreadGoalStatus, ThreadListParams as V2ThreadListParams, @@ -60,6 +62,7 @@ from .retry import retry_on_overload ModelT = TypeVar("ModelT", bound=BaseModel) ApprovalHandler = Callable[[str, JsonObject | None], JsonObject] RUNTIME_PKG_NAME = "openai-codex-cli-bin" +_GOAL_START_TIMEOUT_S = 30.0 def _params_dict( @@ -359,11 +362,6 @@ class CodexClient: """Register a private thread-scoped route for a logical goal turn.""" return self._router.register_goal(thread_id) - def bind_goal_operation(self, state: _GoalOperationState, turn_id: str) -> None: - """Bind a pending goal route to its stable logical turn id.""" - state.bind(turn_id) - self.unregister_turn_notifications(turn_id) - def unregister_goal_operation(self, state: _GoalOperationState) -> None: """Release routing state for one logical goal turn.""" self._router.unregister_goal(state) @@ -472,21 +470,87 @@ class CodexClient: response_model=ThreadCompactStartResponse, ) - def pause_goal(self, thread_id: str) -> ThreadGoalSetResponse: - """Pause the active goal used by a logical goal turn.""" + def thread_goal_clear(self, thread_id: str) -> ThreadGoalClearResponse: + """Clear the persisted goal for a thread before replacing it.""" + return self.request( + "thread/goal/clear", + {"threadId": thread_id}, + response_model=ThreadGoalClearResponse, + ) + + def thread_goal_set( + self, + thread_id: str, + *, + objective: str | None = None, + status: ThreadGoalStatus | None = None, + ) -> ThreadGoalSetResponse: + """Create or update the persisted goal for a thread.""" + payload: JsonObject = {"threadId": thread_id} + if objective is not None: + payload["objective"] = objective + if status is not None: + payload["status"] = status.value return self.request( "thread/goal/set", - {"threadId": thread_id, "status": ThreadGoalStatus.paused.value}, + payload, response_model=ThreadGoalSetResponse, ) + def pause_goal(self, thread_id: str) -> ThreadGoalSetResponse: + """Pause the active goal used by a logical goal turn.""" + return self.thread_goal_set(thread_id, status=ThreadGoalStatus.paused) + + def start_goal_operation( + self, + thread_id: str, + objective: str, + ) -> tuple[_GoalOperationState, str]: + """Start a logical goal and wait for its runtime-generated first turn.""" + thread = self.thread_read(thread_id).thread + if not isinstance(thread.status.root, IdleThreadStatus): + raise InvalidRequestError( + -32600, + f"thread must be idle before starting a goal: {thread_id}", + ) + if thread.ephemeral or thread.path is None: + raise InvalidRequestError( + -32600, + f"thread must be persisted before starting a goal: {thread_id}", + ) + + self.thread_goal_clear(thread_id) + state = self.register_goal_operation(thread_id) + activated = False + try: + self.thread_goal_set( + thread_id, + objective=objective, + status=ThreadGoalStatus.active, + ) + activated = True + turn_id = state.wait_for_start(_GOAL_START_TIMEOUT_S) + if turn_id is None: + raise CodexError( + "timed out waiting for goal turn to start after " + f"{int(_GOAL_START_TIMEOUT_S)} seconds" + ) + return state, turn_id + except BaseException: + if activated: + try: + self.pause_goal(thread_id) + except Exception: + pass + state.finish() + self.unregister_goal_operation(state) + raise + def turn_start( self, thread_id: str, input_items: list[JsonObject] | JsonObject | str, params: V2TurnStartParams | JsonObject | None = None, - *, - goal: bool = False, ) -> TurnStartResponse: """Start a turn and register its notification queue as early as possible.""" payload = { @@ -494,11 +558,6 @@ class CodexClient: "threadId": thread_id, "input": self._normalize_input_items(input_items), } - params_goal = payload.pop("goal", False) - if not isinstance(params_goal, bool): - raise TypeError("turn/start goal must be a bool") - if goal or params_goal: - payload["goal"] = True started = self.request("turn/start", payload, response_model=TurnStartResponse) self.register_turn_notifications(started.turn.id) return started diff --git a/sdk/python/tests/app_server_harness.py b/sdk/python/tests/app_server_harness.py index 135ea47f0d..d609cfb6e0 100644 --- a/sdk/python/tests/app_server_harness.py +++ b/sdk/python/tests/app_server_harness.py @@ -204,7 +204,7 @@ class MockResponsesServer: class AppServerHarness: - """Test fixture that points an app-server process at MockResponsesServer.""" + """Test fixture that points a pinned runtime app-server at MockResponsesServer.""" def __init__( self, @@ -232,10 +232,9 @@ class AppServerHarness: shutil.rmtree(self.codex_home, ignore_errors=True) shutil.rmtree(self.workspace, ignore_errors=True) - def app_server_config(self, *, codex_bin: str | None = None) -> CodexConfig: - """Build SDK config for an isolated app-server process.""" + def app_server_config(self) -> CodexConfig: + """Build SDK config for an isolated pinned-runtime app-server process.""" return CodexConfig( - codex_bin=codex_bin, cwd=str(self.workspace), env={ "CODEX_HOME": str(self.codex_home), diff --git a/sdk/python/tests/test_app_server_goal_turns.py b/sdk/python/tests/test_app_server_goal_turns.py index 6cb853f5a0..660351a3ae 100644 --- a/sdk/python/tests/test_app_server_goal_turns.py +++ b/sdk/python/tests/test_app_server_goal_turns.py @@ -1,5 +1,4 @@ import asyncio -import os import pytest from app_server_harness import ( @@ -17,30 +16,17 @@ from app_server_helpers import ( streaming_response, ) -from openai_codex import AsyncCodex, Codex, CodexConfig, TextInput +from openai_codex import AsyncCodex, Codex from openai_codex.errors import InvalidRequestError, TransportClosedError from openai_codex.generated.notification_registry import notification_turn_id from openai_codex.generated.v2_all import ( AgentMessageDeltaNotification, ThreadGoalGetResponse, - ThreadGoalSetResponse, ThreadGoalStatus, TurnCompletedNotification, TurnStatus, ) -SOURCE_CODEX_BIN = os.environ.get("CODEX_EXEC_PATH") - -pytestmark = pytest.mark.skipif( - SOURCE_CODEX_BIN is None, - reason="requires CODEX_EXEC_PATH pointing to the checkout-built Codex binary", -) - - -def _source_config(harness: AppServerHarness) -> CodexConfig: - assert SOURCE_CODEX_BIN is not None - return harness.app_server_config(codex_bin=SOURCE_CODEX_BIN) - def _enqueue_completed_goal( harness: AppServerHarness, @@ -189,29 +175,14 @@ def test_sync_goal_run_aggregates_automatic_continuation(tmp_path) -> None: final_text="Goal complete.", ) - with Codex(config=_source_config(harness)) as codex: + with Codex(config=harness.app_server_config()) as codex: thread = codex.thread_start() - turn = thread.turn( - [ - TextInput(" Improve benchmark coverage "), - TextInput("Document the results"), - ], - goal=True, - model="goal-test-model", - output_schema={ - "type": "object", - "properties": {"summary": {"type": "string"}}, - "required": ["summary"], - "additionalProperties": False, - }, - ) - result = turn.run() + result = thread.run_goal(" Improve benchmark coverage ") requests = harness.responses.wait_for_requests(3) usage = result.usage.model_dump(by_alias=True, mode="json") if result.usage else None - first_body = requests[0].body_json() assert { - "id": result.id, + "id_is_present": bool(result.id), "status": result.status, "messages": agent_message_texts_from_items(result.items), "final_response": result.final_response, @@ -223,15 +194,10 @@ def test_sync_goal_run_aggregates_automatic_continuation(tmp_path) -> None: if result.started_at is not None and result.completed_at is not None else False, ), - "initial_input": requests[0].message_input_texts("user")[-2:], - "model": first_body["model"], - "output_schema": first_body["text"]["format"]["schema"], - "continuation_has_objective": ( - "\nImprove benchmark coverage\n\nDocument the results\n" - in _continuation_text(requests[1]) - ), + "continuation_has_objective": "\nImprove benchmark coverage\n" + in _continuation_text(requests[0]), } == { - "id": turn.id, + "id_is_present": True, "status": TurnStatus.completed, "messages": ["Initial pass complete.", "Goal complete."], "final_response": "Goal complete.", @@ -253,14 +219,6 @@ def test_sync_goal_run_aggregates_automatic_continuation(tmp_path) -> None: }, }, "timing": (True, True, True), - "initial_input": [" Improve benchmark coverage ", "Document the results"], - "model": "goal-test-model", - "output_schema": { - "type": "object", - "properties": {"summary": {"type": "string"}}, - "required": ["summary"], - "additionalProperties": False, - }, "continuation_has_objective": True, } @@ -303,8 +261,8 @@ def test_goal_stream_exposes_one_logical_lifecycle(tmp_path) -> None: ) ) - with Codex(config=_source_config(harness)) as codex: - turn = codex.thread_start().turn("Finish the integration suite", goal=True) + with Codex(config=harness.app_server_config()) as codex: + turn = codex.thread_start().start_goal("Finish the integration suite") events = list(turn.stream()) lifecycle = [event for event in events if event.method in {"turn/started", "turn/completed"}] @@ -362,8 +320,8 @@ def test_goal_can_complete_within_the_initial_server_turn(tmp_path) -> None: response_id="single-turn-final", ) - with Codex(config=_source_config(harness)) as codex: - turn = codex.thread_start().turn("Finish in the initial turn", goal=True) + with Codex(config=harness.app_server_config()) as codex: + turn = codex.thread_start().start_goal("Finish in the initial turn") result = turn.run() requests = harness.responses.wait_for_requests(2) with pytest.raises(InvalidRequestError) as steer_error: @@ -394,19 +352,14 @@ def test_goal_replaces_an_existing_persisted_goal(tmp_path) -> None: final_text="Replacement complete.", ) - with Codex(config=_source_config(harness)) as codex: + with Codex(config=harness.app_server_config()) as codex: thread = codex.thread_start() - previous = codex._client.request( - "thread/goal/set", - { - "threadId": thread.id, - "objective": "Keep the old benchmark objective", - "status": ThreadGoalStatus.paused.value, - "tokenBudget": 500, - }, - response_model=ThreadGoalSetResponse, + previous = codex._client.thread_goal_set( + thread.id, + objective="Keep the old benchmark objective", + status=ThreadGoalStatus.paused, ).goal - result = thread.run("Publish the replacement objective", goal=True) + result = thread.run_goal("Publish the replacement objective") persisted = codex._client.request( "thread/goal/get", {"threadId": thread.id}, @@ -415,7 +368,7 @@ def test_goal_replaces_an_existing_persisted_goal(tmp_path) -> None: requests = harness.responses.wait_for_requests(3) assert { - "previous": (previous.objective, previous.status, previous.token_budget), + "previous": (previous.objective, previous.status), "result": (result.status, result.final_response), "persisted": ( persisted.objective if persisted else None, @@ -423,17 +376,13 @@ def test_goal_replaces_an_existing_persisted_goal(tmp_path) -> None: persisted.token_budget if persisted else None, ), "continuation_has_replacement": ( - "Publish the replacement objective" in _continuation_text(requests[1]) + "Publish the replacement objective" in _continuation_text(requests[0]) ), "continuation_has_previous": ( - "Keep the old benchmark objective" in _continuation_text(requests[1]) + "Keep the old benchmark objective" in _continuation_text(requests[0]) ), } == { - "previous": ( - "Keep the old benchmark objective", - ThreadGoalStatus.paused, - 500, - ), + "previous": ("Keep the old benchmark objective", ThreadGoalStatus.paused), "result": (TurnStatus.completed, "Replacement complete."), "persisted": ( "Publish the replacement objective", @@ -456,8 +405,8 @@ def test_goal_steer_targets_an_active_continuation(tmp_path) -> None: final_text="Steered goal complete.", ) - with Codex(config=_source_config(harness)) as codex: - turn = codex.thread_start().turn("Start a goal that needs refinement", goal=True) + with Codex(config=harness.app_server_config()) as codex: + turn = codex.thread_start().start_goal("Start a goal that needs refinement") harness.responses.wait_for_requests(2) steer = turn.steer("Prioritize the edge-case coverage.") result = turn.run() @@ -491,9 +440,9 @@ def test_goal_interrupt_pauses_continuation_and_leaves_thread_usable(tmp_path) - follow_up_text="Ordinary follow-up complete.", ) - with Codex(config=_source_config(harness)) as codex: + with Codex(config=harness.app_server_config()) as codex: thread = codex.thread_start() - turn = thread.turn("Start interruptible goal work", goal=True) + turn = thread.start_goal("Start interruptible goal work") harness.responses.wait_for_requests(2) interrupt = turn.interrupt() interrupted = turn.run() @@ -532,10 +481,10 @@ def test_terminal_goal_failure_stops_continuation_and_releases_routing(tmp_path) response_id="goal-failure-follow-up", ) - with Codex(config=_source_config(harness)) as codex: + with Codex(config=harness.app_server_config()) as codex: thread = codex.thread_start() with pytest.raises(RuntimeError, match="goal model failed"): - thread.run("Fail this goal turn", goal=True) + thread.run_goal("Fail this goal turn") follow_up = thread.run("Run after the goal failure") harness.responses.wait_for_requests(2) @@ -553,7 +502,7 @@ def test_terminal_goal_failure_stops_continuation_and_releases_routing(tmp_path) def test_closing_goal_stream_releases_real_process_routing(tmp_path) -> None: - """Closing a public stream should immediately unregister its logical operation.""" + """Closing a stream should release SDK routing without pausing the persisted goal.""" with AppServerHarness(tmp_path, enable_goals=True) as harness: harness.responses.enqueue_sse( streaming_response( @@ -564,14 +513,26 @@ def test_closing_goal_stream_releases_real_process_routing(tmp_path) -> None: delay_between_events_s=0.5, ) - with Codex(config=_source_config(harness)) as codex: - turn = codex.thread_start().turn("Close this goal stream", goal=True) + with Codex(config=harness.app_server_config()) as codex: + thread = codex.thread_start() + turn = thread.start_goal("Close this goal stream") harness.responses.wait_for_requests(1) stream = turn.stream() stream.close() registered_goals = dict(codex._client._router._goal_operations) + persisted = codex._client.request( + "thread/goal/get", + {"threadId": thread.id}, + response_model=ThreadGoalGetResponse, + ).goal - assert registered_goals == {} + assert { + "registered_goals": registered_goals, + "persisted_status": persisted.status if persisted else None, + } == { + "registered_goals": {}, + "persisted_status": ThreadGoalStatus.active, + } def test_app_server_exit_unblocks_goal_stream_and_releases_routing(tmp_path) -> None: @@ -586,8 +547,8 @@ def test_app_server_exit_unblocks_goal_stream_and_releases_routing(tmp_path) -> delay_between_events_s=0.5, ) - with Codex(config=_source_config(harness)) as codex: - turn = codex.thread_start().turn("Stop the app-server during this goal", goal=True) + with Codex(config=harness.app_server_config()) as codex: + turn = codex.thread_start().start_goal("Stop the app-server during this goal") harness.responses.wait_for_requests(1) process = codex._client._proc assert process is not None @@ -609,29 +570,89 @@ def test_failed_goal_starts_release_routing_without_model_requests(tmp_path) -> response_id="validation-follow-up", ) - with Codex(config=_source_config(harness)) as codex: + with Codex(config=harness.app_server_config()) as codex: thread = codex.thread_start() - with pytest.raises(InvalidRequestError) as empty_error: - thread.turn(" ", goal=True) + with pytest.raises(ValueError) as empty_error: + thread.start_goal(" ") + with pytest.raises(TypeError) as type_error: + thread.start_goal(123) # type: ignore[arg-type] ephemeral = codex.thread_start(ephemeral=True) with pytest.raises(InvalidRequestError) as ephemeral_error: - ephemeral.turn("Persist this goal", goal=True) + ephemeral.start_goal("Persist this goal") follow_up = thread.run("Run after rejected goals") requests = harness.responses.wait_for_requests(1) + registered_goals = dict(codex._client._router._goal_operations) assert { - "errors": [empty_error.value.message, ephemeral_error.value.message], + "errors": [str(empty_error.value), str(type_error.value), ephemeral_error.value.message], "follow_up": (follow_up.status, follow_up.final_response), "request_count": len(requests), + "registered_goals": registered_goals, } == { "errors": [ "goal objective must not be empty", - f"ephemeral thread does not support goals: {ephemeral.id}", + "goal objective must be a string", + f"thread must be persisted before starting a goal: {ephemeral.id}", ], "follow_up": (TurnStatus.completed, "Ordinary turn complete."), "request_count": 1, + "registered_goals": {}, + } + + +def test_disabled_goals_fail_before_model_work_or_routing(tmp_path) -> None: + """A runtime with goals disabled should reject startup without leaking state.""" + with AppServerHarness(tmp_path) as harness: + with Codex(config=harness.app_server_config()) as codex: + thread = codex.thread_start() + with pytest.raises(InvalidRequestError) as error: + thread.start_goal("This goal must not start") + registered_goals = dict(codex._client._router._goal_operations) + requests = harness.responses.requests() + + assert { + "error": error.value.message, + "registered_goals": registered_goals, + "request_count": len(requests), + } == { + "error": "goals feature is disabled", + "registered_goals": {}, + "request_count": 0, + } + + +def test_active_thread_rejects_goal_start_and_keeps_ordinary_turn_usable(tmp_path) -> None: + """Goal startup should require an idle thread without disturbing active work.""" + with AppServerHarness(tmp_path, enable_goals=True) as harness: + harness.responses.enqueue_sse( + streaming_response( + "active-ordinary-turn", + "msg-active-ordinary-turn", + ["ordinary ", "work"], + ), + delay_between_events_s=0.5, + ) + + with Codex(config=harness.app_server_config()) as codex: + thread = codex.thread_start() + ordinary = thread.turn("Keep this ordinary turn active") + harness.responses.wait_for_requests(1) + with pytest.raises(InvalidRequestError) as error: + thread.start_goal("Do not replace active work") + ordinary.interrupt() + result = ordinary.run() + registered_goals = dict(codex._client._router._goal_operations) + + assert { + "error": error.value.message, + "ordinary_result": (result.id, result.status), + "registered_goals": registered_goals, + } == { + "error": f"thread must be idle before starting a goal: {thread.id}", + "ordinary_result": (ordinary.id, TurnStatus.interrupted), + "registered_goals": {}, } @@ -647,30 +668,23 @@ def test_async_goal_run_matches_sync_logical_result(tmp_path) -> None: final_text="Async goal complete.", ) - async with AsyncCodex(config=_source_config(harness)) as codex: + async with AsyncCodex(config=harness.app_server_config()) as codex: thread = await codex.thread_start() - turn = await thread.turn("Finish the async goal", goal=True) - result = await turn.run() + result = await thread.run_goal("Finish the async goal") requests = harness.responses.wait_for_requests(3) - with pytest.raises(InvalidRequestError) as steer_error: - await turn.steer("Keep working") - with pytest.raises(InvalidRequestError) as interrupt_error: - await turn.interrupt() assert { "status": result.status, "messages": agent_message_texts_from_items(result.items), "final_response": result.final_response, "continuation_has_objective": ( - "Finish the async goal" in _continuation_text(requests[1]) + "Finish the async goal" in _continuation_text(requests[0]) ), - "inactive_errors": [steer_error.value.message, interrupt_error.value.message], } == { "status": TurnStatus.completed, "messages": ["Async initial pass.", "Async goal complete."], "final_response": "Async goal complete.", "continuation_has_objective": True, - "inactive_errors": ["no active turn to steer", "no active turn to interrupt"], } asyncio.run(scenario()) @@ -689,12 +703,9 @@ def test_async_goal_steer_targets_an_active_continuation(tmp_path) -> None: final_text="Async steered goal complete.", ) - async with AsyncCodex(config=_source_config(harness)) as codex: + async with AsyncCodex(config=harness.app_server_config()) as codex: thread = await codex.thread_start() - turn = await thread.turn( - "Start an async goal that needs refinement", - goal=True, - ) + turn = await thread.start_goal("Start an async goal that needs refinement") await asyncio.to_thread(harness.responses.wait_for_requests, 2) steer = await turn.steer("Prioritize async edge-case coverage.") result = await turn.run() @@ -732,9 +743,9 @@ def test_async_goal_interrupts_an_active_continuation(tmp_path) -> None: follow_up_text="Async ordinary follow-up complete.", ) - async with AsyncCodex(config=_source_config(harness)) as codex: + async with AsyncCodex(config=harness.app_server_config()) as codex: thread = await codex.thread_start() - turn = await thread.turn("Start async interruptible goal work", goal=True) + turn = await thread.start_goal("Start async interruptible goal work") await asyncio.to_thread(harness.responses.wait_for_requests, 2) interrupt = await turn.interrupt() interrupted = await turn.run() diff --git a/sdk/python/tests/test_public_api_signatures.py b/sdk/python/tests/test_public_api_signatures.py index b735e41733..adf7ac3e7a 100644 --- a/sdk/python/tests/test_public_api_signatures.py +++ b/sdk/python/tests/test_public_api_signatures.py @@ -185,17 +185,29 @@ def test_turn_input_methods_accept_string_shortcut() -> None: ) -def test_turn_result_producers_expose_goal_mode() -> None: - """Goal mode belongs to APIs that produce a turn result or handle.""" - funcs = [Thread.run, Thread.turn, AsyncThread.run, AsyncThread.turn] +def test_dedicated_goal_operations_have_pythonic_signatures() -> None: + """Goal operations should accept only an objective and return existing turn types.""" + goal_methods = { + Thread.run_goal: "TurnResult", + Thread.start_goal: "TurnHandle", + AsyncThread.run_goal: "TurnResult", + AsyncThread.start_goal: "AsyncTurnHandle", + } + ordinary_methods = [Thread.run, Thread.turn, AsyncThread.run, AsyncThread.turn] assert { fn: ( - inspect.signature(fn).parameters["goal"].annotation, - inspect.signature(fn).parameters["goal"].default, + list(inspect.signature(fn).parameters), + inspect.signature(fn).parameters["objective"].annotation, + inspect.signature(fn).return_annotation, ) - for fn in funcs - } == dict.fromkeys(funcs, ("bool", False)) + for fn in goal_methods + } == { + fn: (["self", "objective"], "str", return_type) for fn, return_type in goal_methods.items() + } + assert {fn: "goal" in inspect.signature(fn).parameters for fn in ordinary_methods} == ( + dict.fromkeys(ordinary_methods, False) + ) def test_root_exports_approval_mode() -> None: @@ -241,6 +253,10 @@ def test_curated_public_api_has_builtin_help_documentation() -> None: "thread_resume": Codex.thread_resume, "thread_run": Thread.run, "thread_turn": Thread.turn, + "thread_run_goal": Thread.run_goal, + "thread_start_goal": Thread.start_goal, + "async_thread_run_goal": AsyncThread.run_goal, + "async_thread_start_goal": AsyncThread.start_goal, } assert {name: inspect.getdoc(value) is not None for name, value in documented.items()} == ( @@ -383,7 +399,6 @@ def test_generated_public_signatures_are_snake_case_and_typed() -> None: "thread_source", ], Thread.turn: [ - "goal", "approval_mode", "cwd", "effort", @@ -395,7 +410,6 @@ def test_generated_public_signatures_are_snake_case_and_typed() -> None: "summary", ], Thread.run: [ - "goal", "approval_mode", "cwd", "effort", @@ -460,7 +474,6 @@ def test_generated_public_signatures_are_snake_case_and_typed() -> None: "thread_source", ], AsyncThread.turn: [ - "goal", "approval_mode", "cwd", "effort", @@ -472,7 +485,6 @@ def test_generated_public_signatures_are_snake_case_and_typed() -> None: "summary", ], AsyncThread.run: [ - "goal", "approval_mode", "cwd", "effort",