[4/6] Expose dedicated Python goal operations

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:

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:

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())

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)

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

sdk/python/src/openai_codex/api.py

@@ -75,6 +75,21 @@ from .generated.v2_all import (
 )
 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 @@ 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,
@@ -611,6 +630,19 @@ class Thread:
 
     # 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 @@ 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,
@@ -704,6 +741,20 @@ class AsyncThread:
 
     # 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()

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": "<objective>\nImprove benchmark coverage\n</objective>"
+        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": 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] * len(routed_ids),
+        "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())

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()} == (