Add dedicated Python goal operations

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

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

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

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:

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