diff --git a/sdk/python/docs/api-reference.md b/sdk/python/docs/api-reference.md index f003a28511d8..fdb402a3648f 100644 --- a/sdk/python/docs/api-reference.md +++ b/sdk/python/docs/api-reference.md @@ -152,6 +152,8 @@ attempt. API-key login completes synchronously and does not return a handle. - `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` @@ -160,6 +162,8 @@ attempt. API-key login completes synchronously and does not return a handle. - `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,6 +188,15 @@ phase-less assistant message item. Use `turn(...)` when you need low-level turn control (`stream()`, `steer()`, `interrupt()`) before collecting the turn result. +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. + +Closing a goal stream releases its SDK routing state. It does not pause the +persisted goal or stop work already running on the server. + ## Sandbox Use `sandbox=` consistently on thread lifecycle methods and turns: diff --git a/sdk/python/docs/getting-started.md b/sdk/python/docs/getting-started.md index fb2c88b4c331..06d899d5d930 100644 --- a/sdk/python/docs/getting-started.md +++ b/sdk/python/docs/getting-started.md @@ -72,6 +72,17 @@ with Codex() as codex: Use `Thread.turn(...)` when you need a `TurnHandle` for streaming, steering, or interrupting an active turn. +For a longer objective on an idle, persisted thread, run a dedicated goal +operation: + +```python +result = thread.run_goal("Improve the benchmark coverage in this repository.") +``` + +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 Use one enum for the initial thread and later turn overrides: diff --git a/sdk/python/examples/16_goal_turns/async.py b/sdk/python/examples/16_goal_turns/async.py new file mode 100644 index 000000000000..e93c0a0ff8aa --- /dev/null +++ b/sdk/python/examples/16_goal_turns/async.py @@ -0,0 +1,26 @@ +import sys +from pathlib import Path + +_EXAMPLES_ROOT = Path(__file__).resolve().parents[1] +if str(_EXAMPLES_ROOT) not in sys.path: + sys.path.insert(0, str(_EXAMPLES_ROOT)) + +from _bootstrap import ensure_local_sdk_src, runtime_config + +ensure_local_sdk_src() + +import asyncio + +from openai_codex import AsyncCodex + + +async def main() -> None: + async with AsyncCodex(config=runtime_config()) as codex: + thread = await codex.thread_start() + goal = await thread.start_goal("Improve the benchmark coverage in this repository.") + result = await goal.run() + print(result.final_response) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/sdk/python/examples/16_goal_turns/sync.py b/sdk/python/examples/16_goal_turns/sync.py new file mode 100644 index 000000000000..a1749c784832 --- /dev/null +++ b/sdk/python/examples/16_goal_turns/sync.py @@ -0,0 +1,17 @@ +import sys +from pathlib import Path + +_EXAMPLES_ROOT = Path(__file__).resolve().parents[1] +if str(_EXAMPLES_ROOT) not in sys.path: + sys.path.insert(0, str(_EXAMPLES_ROOT)) + +from _bootstrap import ensure_local_sdk_src, runtime_config + +ensure_local_sdk_src() + +from openai_codex import Codex + +with Codex(config=runtime_config()) as codex: + thread = codex.thread_start() + result = thread.run_goal("Improve the benchmark coverage in this repository.") + print(result.final_response) diff --git a/sdk/python/examples/README.md b/sdk/python/examples/README.md index 8efeb75b14c3..bf04e64b8e5b 100644 --- a/sdk/python/examples/README.md +++ b/sdk/python/examples/README.md @@ -89,3 +89,5 @@ python examples/01_quickstart_constructor/async.py - separate `steer()` and `interrupt()` demos with concise summaries - `15_login_and_account/` - browser-login handle lifecycle, cancellation, and account inspection +- `16_goal_turns/` + - run a longer autonomous objective as one logical turn diff --git a/sdk/python/src/openai_codex/api.py b/sdk/python/src/openai_codex/api.py index c77ba3acf8cf..e2014e0c738b 100644 --- a/sdk/python/src/openai_codex/api.py +++ b/sdk/python/src/openai_codex/api.py @@ -75,6 +75,21 @@ ) from .models import InitializeResponse, JsonObject, Notification +_MAX_THREAD_GOAL_OBJECTIVE_CHARS = 4_000 + + +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") + if len(objective) > _MAX_THREAD_GOAL_OBJECTIVE_CHARS: + raise ValueError( + f"goal objective must be at most {_MAX_THREAD_GOAL_OBJECTIVE_CHARS} characters" + ) + return objective + class Codex: """Synchronous client for creating threads and running Codex turns. @@ -574,6 +589,10 @@ def run( 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, @@ -611,6 +630,19 @@ def turn( # 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) + handle = TurnHandle(self._client, self.id, turn_id) + handle._goal = _GoalTurnHandleAdapter( + client=self._client, + state=state, + thread_id=self.id, + logical_turn_id=turn_id, + ) + return handle + 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) @@ -662,6 +694,11 @@ async def run( 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, @@ -704,6 +741,20 @@ async def turn( # END GENERATED: AsyncThread.flat_methods + async def start_goal(self, objective: str) -> AsyncTurnHandle: + """Activate a persisted goal and return its async logical turn handle.""" + objective = _normalize_goal_objective(objective) + await self._codex._ensure_initialized() + state, turn_id = await self._codex._client.start_goal_operation(self.id, objective) + handle = AsyncTurnHandle(self._codex, self.id, turn_id) + handle._goal = _AsyncGoalTurnHandleAdapter( + client=self._codex._client, + state=state, + thread_id=self.id, + logical_turn_id=turn_id, + ) + return handle + 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/tests/test_app_server_goal_turns.py b/sdk/python/tests/test_app_server_goal_turns.py new file mode 100644 index 000000000000..af9dc8a067aa --- /dev/null +++ b/sdk/python/tests/test_app_server_goal_turns.py @@ -0,0 +1,386 @@ +import asyncio + +import pytest +from app_server_harness import ( + AppServerHarness, + ev_assistant_message, + ev_completed_with_usage, + ev_function_call, + ev_response_created, + sse, +) +from app_server_helpers import ( + agent_message_texts, + agent_message_texts_from_items, + streaming_response, +) + +from openai_codex import AsyncCodex, Codex +from openai_codex.api import _MAX_THREAD_GOAL_OBJECTIVE_CHARS +from openai_codex.errors import InvalidRequestError +from openai_codex.generated.notification_registry import notification_turn_id +from openai_codex.generated.v2_all import ( + AgentMessageDeltaNotification, + ThreadGoalGetResponse, + ThreadGoalStatus, + TurnCompletedNotification, + TurnStatus, +) + + +def _enqueue_completed_goal( + harness: AppServerHarness, + prefix: str, + *, + initial_text: str, + final_text: str, +) -> None: + harness.responses.enqueue_sse( + sse( + [ + ev_response_created(f"{prefix}-initial"), + ev_assistant_message(f"msg-{prefix}-initial", initial_text), + ev_completed_with_usage( + f"{prefix}-initial", + input_tokens=3, + cached_input_tokens=1, + output_tokens=2, + reasoning_output_tokens=1, + total_tokens=5, + ), + ] + ) + ) + harness.responses.enqueue_sse( + sse( + [ + ev_response_created(f"{prefix}-complete-tool"), + ev_function_call( + f"call-{prefix}-complete", + "update_goal", + '{"status":"complete"}', + ), + ev_completed_with_usage( + f"{prefix}-complete-tool", + input_tokens=4, + cached_input_tokens=1, + output_tokens=1, + reasoning_output_tokens=1, + total_tokens=5, + ), + ] + ) + ) + harness.responses.enqueue_sse( + sse( + [ + ev_response_created(f"{prefix}-final"), + ev_assistant_message(f"msg-{prefix}-final", final_text), + ev_completed_with_usage( + f"{prefix}-final", + input_tokens=5, + cached_input_tokens=0, + output_tokens=3, + reasoning_output_tokens=0, + total_tokens=8, + ), + ] + ) + ) + + +def _continuation_text(request) -> str: + return "\n".join(request.message_input_texts("user")) + + +def test_sync_goal_run_aggregates_automatic_continuation(tmp_path) -> None: + """The public result should cover the initial and automatic continuation turns.""" + with AppServerHarness(tmp_path) as harness: + _enqueue_completed_goal( + harness, + "sync-run", + initial_text="Initial pass complete.", + final_text="Goal complete.", + ) + + with Codex(config=harness.app_server_config()) as codex: + thread = codex.thread_start() + 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 + assert { + "id_is_present": bool(result.id), + "status": result.status, + "messages": agent_message_texts_from_items(result.items), + "final_response": result.final_response, + "usage": usage, + "timing": ( + result.started_at is not None, + result.completed_at is not None, + result.duration_ms == max(0, result.completed_at - result.started_at) * 1000 + if result.started_at is not None and result.completed_at is not None + else False, + ), + "continuation_has_objective": "\nImprove benchmark coverage\n" + in _continuation_text(requests[0]), + } == { + "id_is_present": True, + "status": TurnStatus.completed, + "messages": ["Initial pass complete.", "Goal complete."], + "final_response": "Goal complete.", + "usage": { + "last": { + "cachedInputTokens": 0, + "inputTokens": 5, + "outputTokens": 3, + "reasoningOutputTokens": 0, + "totalTokens": 8, + }, + "modelContextWindow": 258_400, + "total": { + "cachedInputTokens": 2, + "inputTokens": 12, + "outputTokens": 6, + "reasoningOutputTokens": 2, + "totalTokens": 18, + }, + }, + "timing": (True, True, True), + "continuation_has_objective": True, + } + + +def test_goal_stream_exposes_one_logical_lifecycle(tmp_path) -> None: + """Continuation boundaries should not leak through the public stream.""" + with AppServerHarness(tmp_path) as harness: + harness.responses.enqueue_sse( + streaming_response( + "stream-initial", + "msg-stream-initial", + ["initial ", "pass"], + ) + ) + harness.responses.enqueue_sse( + sse( + [ + ev_response_created("stream-complete-tool"), + ev_function_call( + "call-stream-complete", + "update_goal", + '{"status":"complete"}', + ), + ev_completed_with_usage( + "stream-complete-tool", + input_tokens=1, + cached_input_tokens=0, + output_tokens=1, + reasoning_output_tokens=0, + total_tokens=2, + ), + ] + ) + ) + harness.responses.enqueue_sse( + streaming_response( + "stream-final", + "msg-stream-final", + ["goal ", "complete"], + ) + ) + + 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"}] + routed_ids = [ + turn_id for event in events if (turn_id := notification_turn_id(event.payload)) is not None + ] + assert { + "lifecycle": [event.method for event in lifecycle], + "routed_ids": sorted(set(routed_ids)), + "deltas": [ + event.payload.delta + for event in events + if isinstance(event.payload, AgentMessageDeltaNotification) + ], + "messages": agent_message_texts(events), + "completion_statuses": [ + event.payload.turn.status + for event in events + if isinstance(event.payload, TurnCompletedNotification) + ], + } == { + "lifecycle": ["turn/started", "turn/completed"], + "routed_ids": [turn.id], + "deltas": ["initial ", "pass", "goal ", "complete"], + "messages": ["initial pass", "goal complete"], + "completion_statuses": [TurnStatus.completed], + } + + +def test_goal_can_complete_within_the_initial_server_turn(tmp_path) -> None: + """Completing the goal before turn end should not create a continuation turn.""" + with AppServerHarness(tmp_path) as harness: + harness.responses.enqueue_sse( + sse( + [ + ev_response_created("single-turn-complete-tool"), + ev_function_call( + "call-single-turn-complete", + "update_goal", + '{"status":"complete"}', + ), + ev_completed_with_usage( + "single-turn-complete-tool", + input_tokens=1, + cached_input_tokens=0, + output_tokens=1, + reasoning_output_tokens=0, + total_tokens=2, + ), + ] + ) + ) + harness.responses.enqueue_assistant_message( + "Completed without a continuation.", + response_id="single-turn-final", + ) + + 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: + turn.steer("Keep working") + with pytest.raises(InvalidRequestError) as interrupt_error: + turn.interrupt() + + assert { + "result": (result.id, result.status, result.final_response), + "messages": agent_message_texts_from_items(result.items), + "request_count": len(requests), + "inactive_errors": [steer_error.value.message, interrupt_error.value.message], + } == { + "result": (turn.id, TurnStatus.completed, "Completed without a continuation."), + "messages": ["Completed without a continuation."], + "request_count": 2, + "inactive_errors": ["no active turn to steer", "no active turn to interrupt"], + } + + +def test_goal_replaces_an_existing_persisted_goal(tmp_path) -> None: + """Goal mode should replace a resumable persisted goal before model work begins.""" + with AppServerHarness(tmp_path) as harness: + _enqueue_completed_goal( + harness, + "replacement", + initial_text="Replacement started.", + final_text="Replacement complete.", + ) + + with Codex(config=harness.app_server_config()) as codex: + thread = codex.thread_start() + previous = codex._client.thread_goal_set( + thread.id, + objective="Keep the old benchmark objective", + status=ThreadGoalStatus.paused, + ).goal + result = thread.run_goal("Publish the replacement objective") + persisted = codex._client.request( + "thread/goal/get", + {"threadId": thread.id}, + response_model=ThreadGoalGetResponse, + ).goal + requests = harness.responses.wait_for_requests(3) + + assert { + "previous": (previous.objective, previous.status), + "result": (result.status, result.final_response), + "persisted": ( + persisted.objective if persisted else None, + persisted.status if persisted else None, + persisted.token_budget if persisted else None, + ), + "continuation_has_replacement": ( + "Publish the replacement objective" in _continuation_text(requests[0]) + ), + "continuation_has_previous": ( + "Keep the old benchmark objective" in _continuation_text(requests[0]) + ), + } == { + "previous": ("Keep the old benchmark objective", ThreadGoalStatus.paused), + "result": (TurnStatus.completed, "Replacement complete."), + "persisted": ( + "Publish the replacement objective", + ThreadGoalStatus.complete, + None, + ), + "continuation_has_replacement": True, + "continuation_has_previous": False, + } + + +def test_goal_length_validation_preserves_an_existing_goal(tmp_path) -> None: + """An invalid replacement should fail before clearing the stored goal.""" + with AppServerHarness(tmp_path) as harness: + with Codex(config=harness.app_server_config()) as codex: + thread = codex.thread_start() + previous = codex._client.thread_goal_set( + thread.id, + objective="Keep the existing objective", + status=ThreadGoalStatus.paused, + ).goal + with pytest.raises(ValueError) as error: + thread.start_goal("x" * (_MAX_THREAD_GOAL_OBJECTIVE_CHARS + 1)) + persisted = codex._client.request( + "thread/goal/get", + {"threadId": thread.id}, + response_model=ThreadGoalGetResponse, + ).goal + requests = harness.responses.requests() + + assert { + "error": str(error.value), + "persisted": persisted, + "model_requests": requests, + } == { + "error": "goal objective must be at most 4000 characters", + "persisted": previous, + "model_requests": [], + } + + +def test_async_goal_run_matches_sync_logical_result(tmp_path) -> None: + """The async public API should aggregate the same real continuation lifecycle.""" + + async def scenario() -> None: + with AppServerHarness(tmp_path) as harness: + _enqueue_completed_goal( + harness, + "async-run", + initial_text="Async initial pass.", + final_text="Async goal complete.", + ) + + async with AsyncCodex(config=harness.app_server_config()) as codex: + thread = await codex.thread_start() + result = await thread.run_goal("Finish the async goal") + requests = harness.responses.wait_for_requests(3) + + 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[0]) + ), + } == { + "status": TurnStatus.completed, + "messages": ["Async initial pass.", "Async goal complete."], + "final_response": "Async goal complete.", + "continuation_has_objective": True, + } + + asyncio.run(scenario()) diff --git a/sdk/python/tests/test_public_api_signatures.py b/sdk/python/tests/test_public_api_signatures.py index c26ac90f6b1f..8e010eebc254 100644 --- a/sdk/python/tests/test_public_api_signatures.py +++ b/sdk/python/tests/test_public_api_signatures.py @@ -185,6 +185,45 @@ def test_turn_input_methods_accept_string_shortcut() -> None: ) +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 { + "goal_methods": { + fn: ( + list(inspect.signature(fn).parameters), + inspect.signature(fn).parameters["objective"].annotation, + inspect.signature(fn).return_annotation, + ) + for fn in goal_methods + }, + "ordinary_goal_parameter": { + fn: "goal" in inspect.signature(fn).parameters for fn in ordinary_methods + }, + "handle_constructors": { + handle: list(inspect.signature(handle).parameters) + for handle in (TurnHandle, AsyncTurnHandle) + }, + } == { + "goal_methods": { + fn: (["self", "objective"], "str", return_type) + for fn, return_type in goal_methods.items() + }, + "ordinary_goal_parameter": dict.fromkeys(ordinary_methods, False), + "handle_constructors": { + TurnHandle: ["_client", "thread_id", "id"], + AsyncTurnHandle: ["_codex", "thread_id", "id"], + }, + } + + def test_root_exports_approval_mode() -> None: """The root package should expose the high-level approval mode enum.""" assert [(mode.name, mode.value) for mode in ApprovalMode] == [ @@ -228,6 +267,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()} == (