Add Python SDK goal turns

sdk/python/src/openai_codex/_goal.py

@@ -0,0 +1,382 @@
+import queue
+import threading
+from collections import deque
+from dataclasses import dataclass, field
+from typing import AsyncIterator, Awaitable, Callable, Iterator
+
+from .generated.notification_registry import notification_turn_id
+from .generated.v2_all import (
+    ItemCompletedNotification,
+    ThreadGoalClearedNotification,
+    ThreadGoalStatus,
+    ThreadGoalUpdatedNotification,
+    ThreadItem,
+    ThreadTokenUsage,
+    ThreadTokenUsageUpdatedNotification,
+    Turn,
+    TurnCompletedNotification,
+    TurnStartedNotification,
+    TurnStatus,
+)
+from .models import Notification, UnknownNotification
+
+
+class _GoalStreamClosed(Exception):
+    """Wake a notification reader after its logical stream closes."""
+
+
+def _terminal_goal_status(status: ThreadGoalStatus | None) -> bool:
+    return status in {
+        ThreadGoalStatus.paused,
+        ThreadGoalStatus.blocked,
+        ThreadGoalStatus.usage_limited,
+        ThreadGoalStatus.budget_limited,
+        ThreadGoalStatus.complete,
+    }
+
+
+@dataclass(slots=True)
+class _GoalOperationState:
+    """Private state for one goal operation exposed as a logical turn."""
+
+    thread_id: str
+    logical_turn_id: str | None = None
+    current_turn_id: str | None = None
+    status: ThreadGoalStatus | None = None
+    started_turn: Turn | None = None
+    completed_turn: Turn | None = None
+    items: list[ThreadItem] = field(default_factory=list)
+    usage: ThreadTokenUsage | None = None
+    interrupted: bool = False
+    interrupt_requested: bool = False
+    cleared: bool = False
+    _condition: threading.Condition = field(default_factory=threading.Condition)
+    _notifications: queue.Queue[Notification | BaseException] = field(default_factory=queue.Queue)
+    _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):
+                self.current_turn_id = payload.turn.id
+                if self.started_turn is None:
+                    self.started_turn = payload.turn
+            elif isinstance(payload, TurnCompletedNotification):
+                self.completed_turn = payload.turn
+                if self.current_turn_id == payload.turn.id:
+                    self.current_turn_id = None
+            elif isinstance(payload, ThreadGoalUpdatedNotification):
+                self.status = payload.goal.status
+            elif isinstance(payload, ThreadGoalClearedNotification):
+                self.cleared = True
+            elif isinstance(payload, ItemCompletedNotification):
+                self.items.append(payload.item)
+            elif isinstance(payload, ThreadTokenUsageUpdatedNotification):
+                self.usage = payload.token_usage
+            if (
+                isinstance(payload, TurnCompletedNotification)
+                and payload.turn.status in {TurnStatus.failed, TurnStatus.interrupted}
+            ) or (
+                self.current_turn_id is None
+                and self.completed_turn is not None
+                and (self.cleared or _terminal_goal_status(self.status))
+            ):
+                self._finished = True
+            self._condition.notify_all()
+        self._notifications.put(notification)
+
+    def fail(self, exc: BaseException) -> None:
+        with self._condition:
+            self._failure = exc
+            self._condition.notify_all()
+        self._notifications.put(exc)
+
+    def next_notification(self) -> Notification:
+        item = self._notifications.get()
+        if isinstance(item, BaseException):
+            raise item
+        return item
+
+    def finish(self) -> None:
+        """Mark the logical operation inactive and wake waiting controls."""
+        with self._condition:
+            self._finished = True
+            self.current_turn_id = None
+            self._condition.notify_all()
+
+    def is_finished(self) -> bool:
+        with self._condition:
+            return self._finished
+
+    def begin_interrupt(self) -> bool:
+        with self._condition:
+            if self._finished:
+                return False
+            self.interrupt_requested = True
+            return True
+
+    def confirm_interrupt(self) -> None:
+        with self._condition:
+            self.interrupted = True
+            self.interrupt_requested = False
+
+    def cancel_interrupt(self) -> None:
+        with self._condition:
+            self.interrupt_requested = False
+
+    def explicit_interrupt(self, status: ThreadGoalStatus | None) -> bool:
+        with self._condition:
+            return self.interrupted or (
+                self.interrupt_requested and status == ThreadGoalStatus.paused
+            )
+
+    def active_turn(self, *, after: str | None = None) -> str | None:
+        """Wait for the current turn, or return None once the goal has ended."""
+        with self._condition:
+            while True:
+                if self._failure is not None:
+                    raise self._failure
+                if self._finished:
+                    return None
+                if self.current_turn_id is not None and self.current_turn_id != after:
+                    return self.current_turn_id
+                if self.cleared or _terminal_goal_status(self.status):
+                    return None
+                self._condition.wait()
+
+    def current_turn(self) -> str | None:
+        """Return the current physical turn without waiting for rollover."""
+        with self._condition:
+            return self.current_turn_id
+
+    def wake_notification_reader(self) -> None:
+        """Release a reader blocked after its stream has been closed."""
+        self._notifications.put(_GoalStreamClosed())
+
+
+def _logical_notification(notification: Notification, logical_turn_id: str) -> Notification:
+    """Return a copy whose turn metadata uses the logical operation id."""
+    payload = notification.payload
+    if isinstance(payload, UnknownNotification):
+        params = dict(payload.params)
+        if isinstance(params.get("turnId"), str):
+            params["turnId"] = logical_turn_id
+        turn = params.get("turn")
+        if isinstance(turn, dict) and isinstance(turn.get("id"), str):
+            params["turn"] = {**turn, "id": logical_turn_id}
+        return Notification(notification.method, UnknownNotification(params))
+
+    turn_id = notification_turn_id(payload)
+    if turn_id is None:
+        return notification
+    if hasattr(payload, "turn_id"):
+        return Notification(
+            notification.method,
+            payload.model_copy(update={"turn_id": logical_turn_id}),
+        )
+    if hasattr(payload, "turn"):
+        logical_turn = payload.turn.model_copy(update={"id": logical_turn_id})
+        return Notification(
+            notification.method,
+            payload.model_copy(update={"turn": logical_turn}),
+        )
+    return notification
+
+
+def _logical_completion(
+    completed: TurnCompletedNotification,
+    *,
+    logical_turn_id: str,
+    started: Turn | None,
+    interrupted: bool,
+) -> TurnCompletedNotification:
+    """Coalesce the final physical completion into one logical completion."""
+    final_turn = completed.turn
+    started_at = started.started_at if started is not None else final_turn.started_at
+    duration_ms = final_turn.duration_ms
+    if started_at is not None and final_turn.completed_at is not None:
+        duration_ms = max(0, final_turn.completed_at - started_at) * 1000
+    updates: dict[str, object] = {
+        "id": logical_turn_id,
+        "started_at": started_at,
+        "duration_ms": duration_ms,
+    }
+    if interrupted:
+        updates["status"] = TurnStatus.interrupted
+    return completed.model_copy(update={"turn": final_turn.model_copy(update=updates)})
+
+
+@dataclass(slots=True)
+class _GoalStreamCursor:
+    """Consume physical goal events as one ordered logical turn stream."""
+
+    state: _GoalOperationState
+    started: Turn | None = None
+    last_completed: TurnCompletedNotification | None = None
+    status: ThreadGoalStatus | None = None
+    active: bool = False
+    cleared: bool = False
+
+    def process(self, notification: Notification) -> tuple[list[Notification], bool]:
+        logical_turn_id = self.state.logical_turn_id
+        if logical_turn_id is None:
+            raise RuntimeError("goal operation has not been bound to a logical turn id")
+
+        payload = notification.payload
+        if isinstance(payload, TurnStartedNotification):
+            self.active = True
+            if self.started is not None:
+                return [], False
+            self.started = payload.turn
+            return [_logical_notification(notification, logical_turn_id)], False
+
+        if isinstance(payload, TurnCompletedNotification):
+            self.active = False
+            self.last_completed = payload
+            if payload.turn.status in {TurnStatus.failed, TurnStatus.interrupted}:
+                self.state.finish()
+                return [self._completion(notification.method, payload)], True
+            if self.status is None and not self.cleared:
+                raise RuntimeError(
+                    "the connected Codex runtime did not activate goal mode for this turn"
+                )
+            if self.cleared or _terminal_goal_status(self.status):
+                self.state.finish()
+                return [self._completion(notification.method, payload)], True
+            return [], False
+
+        events = [_logical_notification(notification, logical_turn_id)]
+        if isinstance(payload, ThreadGoalUpdatedNotification):
+            self.status = payload.goal.status
+            events = []
+        elif isinstance(payload, ThreadGoalClearedNotification):
+            self.cleared = True
+            events = []
+
+        if (
+            not self.active
+            and self.last_completed is not None
+            and (self.cleared or _terminal_goal_status(self.status))
+        ):
+            self.state.finish()
+            events.append(self._completion("turn/completed", self.last_completed))
+            return events, True
+        return events, False
+
+    def _completion(
+        self,
+        method: str,
+        payload: TurnCompletedNotification,
+    ) -> Notification:
+        logical_turn_id = self.state.logical_turn_id
+        if logical_turn_id is None:
+            raise RuntimeError("goal operation has not been bound to a logical turn id")
+        return Notification(
+            method,
+            _logical_completion(
+                payload,
+                logical_turn_id=logical_turn_id,
+                started=self.started,
+                interrupted=self.state.explicit_interrupt(self.status),
+            ),
+        )
+
+
+@dataclass(slots=True)
+class _GoalNotificationStream(Iterator[Notification]):
+    """Closeable synchronous view of one logical goal operation."""
+
+    state: _GoalOperationState
+    next_notification: Callable[[], Notification]
+    unregister: Callable[[], None]
+    _cursor: _GoalStreamCursor = field(init=False)
+    _pending: deque[Notification] = field(default_factory=deque)
+    _closed: bool = False
+
+    def __post_init__(self) -> None:
+        self._cursor = _GoalStreamCursor(self.state)
+
+    def __iter__(self) -> "_GoalNotificationStream":
+        return self
+
+    def __next__(self) -> Notification:
+        if self._closed:
+            raise StopIteration
+        try:
+            while not self._pending:
+                events, completed = self._cursor.process(self.next_notification())
+                self._pending.extend(events)
+                if completed:
+                    self._finish()
+            return self._pending.popleft()
+        except _GoalStreamClosed:
+            self.close()
+            raise StopIteration from None
+        except BaseException:
+            self.close()
+            raise
+
+    def _finish(self) -> None:
+        if self._closed:
+            return
+        self.state.finish()
+        self.state.wake_notification_reader()
+        self.unregister()
+        self._closed = True
+
+    def close(self) -> None:
+        self._finish()
+
+
+@dataclass(slots=True)
+class _AsyncGoalNotificationStream(AsyncIterator[Notification]):
+    """Closeable asynchronous view of one logical goal operation."""
+
+    state: _GoalOperationState
+    next_notification: Callable[[], Awaitable[Notification]]
+    unregister: Callable[[], None]
+    _cursor: _GoalStreamCursor = field(init=False)
+    _pending: deque[Notification] = field(default_factory=deque)
+    _closed: bool = False
+
+    def __post_init__(self) -> None:
+        self._cursor = _GoalStreamCursor(self.state)
+
+    def __aiter__(self) -> "_AsyncGoalNotificationStream":
+        return self
+
+    async def __anext__(self) -> Notification:
+        if self._closed:
+            raise StopAsyncIteration
+        try:
+            while not self._pending:
+                events, completed = self._cursor.process(await self.next_notification())
+                self._pending.extend(events)
+                if completed:
+                    self._finish()
+            return self._pending.popleft()
+        except _GoalStreamClosed:
+            await self.aclose()
+            raise StopAsyncIteration from None
+        except BaseException:
+            await self.aclose()
+            raise
+
+    def _finish(self) -> None:
+        if self._closed:
+            return
+        self.state.finish()
+        self.state.wake_notification_reader()
+        self.unregister()
+        self._closed = True
+
+    async def aclose(self) -> None:
+        self._finish()

sdk/python/src/openai_codex/_message_router.py

@@ -4,8 +4,9 @@ import queue
 import threading
 from collections import deque
 
+from ._goal import _GoalOperationState
 from .errors import CodexError, map_jsonrpc_error
-from .generated.notification_registry import notification_turn_id
+from .generated.notification_registry import notification_thread_id, notification_turn_id
 from .generated.v2_all import AccountLoginCompletedNotification
 from .models import JsonValue, Notification, UnknownNotification
 
@@ -30,6 +31,7 @@ class MessageRouter:
         self._pending_login_notifications: dict[str, deque[Notification]] = {}
         self._turn_notifications: dict[str, queue.Queue[NotificationQueueItem]] = {}
         self._pending_turn_notifications: dict[str, deque[Notification]] = {}
+        self._goal_operations: dict[str, _GoalOperationState] = {}
         self._global_notifications: queue.Queue[NotificationQueueItem] = queue.Queue()
 
     def create_response_waiter(self, request_id: str) -> queue.Queue[ResponseQueueItem]:
@@ -116,6 +118,21 @@ class MessageRouter:
             raise item
         return item
 
+    def register_goal(self, thread_id: str) -> _GoalOperationState:
+        """Register one thread-scoped logical goal operation before it starts."""
+        state = _GoalOperationState(thread_id=thread_id)
+        with self._lock:
+            if thread_id in self._goal_operations:
+                raise RuntimeError(f"thread {thread_id!r} already has an active goal operation")
+            self._goal_operations[thread_id] = state
+        return state
+
+    def unregister_goal(self, state: _GoalOperationState) -> None:
+        """Stop routing notifications to a completed logical goal operation."""
+        with self._lock:
+            if self._goal_operations.get(state.thread_id) is state:
+                self._goal_operations.pop(state.thread_id)
+
     def route_response(self, msg: dict[str, JsonValue]) -> None:
         """Deliver a JSON-RPC response or error to its request waiter."""
 
@@ -157,6 +174,17 @@ class MessageRouter:
             return
 
         turn_id = self._notification_turn_id(notification)
+        thread_id = self._notification_thread_id(notification)
+        if thread_id is not None:
+            with self._lock:
+                goal_state = self._goal_operations.get(thread_id)
+            if goal_state is not None and (
+                turn_id is not None or notification.method.startswith("thread/goal/")
+            ):
+                goal_state.observe(notification)
+                if goal_state.is_finished():
+                    self.unregister_goal(goal_state)
+                return
         if turn_id is None:
             self._global_notifications.put(notification)
             return
@@ -182,6 +210,8 @@ class MessageRouter:
             self._pending_login_notifications.clear()
             turn_queues = list(self._turn_notifications.values())
             self._pending_turn_notifications.clear()
+            goal_operations = list(self._goal_operations.values())
+            self._goal_operations.clear()
         # Put the same transport failure into every queue so no SDK call blocks
         # forever waiting for a response that cannot arrive.
         for waiter in response_waiters:
@@ -190,6 +220,8 @@ class MessageRouter:
             login_queue.put(exc)
         for turn_queue in turn_queues:
             turn_queue.put(exc)
+        for goal_operation in goal_operations:
+            goal_operation.fail(exc)
         self._global_notifications.put(exc)
 
     def _notification_login_id(self, notification: Notification) -> str | None:
@@ -220,3 +252,11 @@ class MessageRouter:
                     return raw_nested_turn_id
             return None
         return notification_turn_id(payload)
+
+    def _notification_thread_id(self, notification: Notification) -> str | None:
+        """Extract thread ids from known generated payloads or raw 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)

sdk/python/src/openai_codex/api.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import re
 from dataclasses import dataclass
 from typing import AsyncIterator, Iterator
 
@@ -9,6 +10,11 @@ from ._approval_mode import (
     _approval_mode_override_settings,
     _approval_mode_settings,
 )
+from ._goal import (
+    _AsyncGoalNotificationStream,
+    _GoalNotificationStream,
+    _GoalOperationState,
+)
 from ._initialize_metadata import validate_initialize_metadata
 from ._inputs import (
     ImageInput as ImageInput,
@@ -40,6 +46,7 @@ from ._run import (
 from ._sandbox import Sandbox as Sandbox, _sandbox_mode, _sandbox_policy
 from .async_client import AsyncCodexClient
 from .client import CodexClient, CodexConfig
+from .errors import InvalidRequestError
 from .generated.v2_all import (
     ApiKeyLoginAccountParams,
     GetAccountParams,
@@ -72,6 +79,19 @@ from .generated.v2_all import (
 from .models import InitializeResponse, JsonObject, Notification
 
 
+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
+
+
+def _inactive_turn_error() -> InvalidRequestError:
+    return InvalidRequestError(-32600, "no active turn to steer")
+
+
+def _inactive_interrupt_error() -> InvalidRequestError:
+    return InvalidRequestError(-32600, "no active turn to interrupt")
+
+
 class Codex:
     """Synchronous client for creating threads and running Codex turns.
 
@@ -541,6 +561,7 @@ class Thread:
         self,
         input: RunInput,
         *,
+        goal: bool = False,
         approval_mode: ApprovalMode | None = None,
         cwd: str | None = None,
         effort: ReasoningEffort | None = None,
@@ -554,6 +575,7 @@ 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,
@@ -575,6 +597,7 @@ class Thread:
         self,
         input: RunInput,
         *,
+        goal: bool = False,
         approval_mode: ApprovalMode | None = None,
         cwd: str | None = None,
         effort: ReasoningEffort | None = None,
@@ -602,8 +625,16 @@ class Thread:
             service_tier=service_tier,
             summary=summary,
         )
-        turn = self._client.turn_start(self.id, wire_input, params=params)
-        return TurnHandle(self._client, self.id, turn.turn.id)
+        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)
 
     # END GENERATED: Thread.flat_methods
 
@@ -629,6 +660,7 @@ class AsyncThread:
         self,
         input: RunInput,
         *,
+        goal: bool = False,
         approval_mode: ApprovalMode | None = None,
         cwd: str | None = None,
         effort: ReasoningEffort | None = None,
@@ -642,6 +674,7 @@ 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,
@@ -663,6 +696,7 @@ class AsyncThread:
         self,
         input: RunInput,
         *,
+        goal: bool = False,
         approval_mode: ApprovalMode | None = None,
         cwd: str | None = None,
         effort: ReasoningEffort | None = None,
@@ -691,12 +725,21 @@ class AsyncThread:
             service_tier=service_tier,
             summary=summary,
         )
-        turn = await self._codex._client.turn_start(
-            self.id,
-            wire_input,
-            params=params,
-        )
-        return AsyncTurnHandle(self._codex, self.id, turn.turn.id)
+        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)
 
     # END GENERATED: AsyncThread.flat_methods
 
@@ -721,9 +764,30 @@ class TurnHandle:
     _client: CodexClient
     thread_id: str
     id: str
+    _goal: _GoalOperationState | None = None
 
     def steer(self, input: RunInput) -> TurnSteerResponse:
         """Send additional input to this active turn."""
+        if self._goal is not None:
+            wire_input = _to_wire_input(_normalize_run_input(input))
+            turn_id = self._goal.active_turn()
+            if turn_id is None:
+                raise _inactive_turn_error()
+            try:
+                response = self._client.turn_steer(self.thread_id, turn_id, wire_input)
+            except InvalidRequestError as exc:
+                if not (
+                    exc.message == "no active turn to steer"
+                    or exc.message.startswith("expected active turn id")
+                ):
+                    raise
+                next_turn_id = _active_turn_id_from_error(exc)
+                if next_turn_id is None:
+                    next_turn_id = self._goal.active_turn(after=turn_id)
+                if next_turn_id is None:
+                    raise _inactive_turn_error() from exc
+                response = self._client.turn_steer(self.thread_id, next_turn_id, wire_input)
+            return response.model_copy(update={"turn_id": self.id})
         return self._client.turn_steer(
             self.thread_id,
             self.id,
@@ -732,23 +796,61 @@ class TurnHandle:
 
     def interrupt(self) -> TurnInterruptResponse:
         """Request interruption of this active turn."""
+        if self._goal is not None:
+            if not self._goal.begin_interrupt():
+                raise _inactive_interrupt_error()
+            try:
+                self._client.pause_goal(self.thread_id)
+            except BaseException:
+                self._goal.cancel_interrupt()
+                raise
+            self._goal.confirm_interrupt()
+            turn_id = self._goal.current_turn()
+            if turn_id is None:
+                return TurnInterruptResponse()
+            try:
+                return self._client.turn_interrupt(self.thread_id, turn_id)
+            except InvalidRequestError as exc:
+                if exc.message == "no active turn to interrupt":
+                    return TurnInterruptResponse()
+                if exc.message.startswith("expected active turn id"):
+                    next_turn_id = _active_turn_id_from_error(exc) or self._goal.current_turn()
+                    if next_turn_id is None or next_turn_id == turn_id:
+                        return TurnInterruptResponse()
+                    try:
+                        return self._client.turn_interrupt(self.thread_id, next_turn_id)
+                    except InvalidRequestError as retry_exc:
+                        if retry_exc.message == "no active turn to interrupt":
+                            return TurnInterruptResponse()
+                        raise
+                raise
         return self._client.turn_interrupt(self.thread_id, self.id)
 
     def stream(self) -> Iterator[Notification]:
         """Yield only notifications routed to this turn handle."""
-        self._client.register_turn_notifications(self.id)
-        try:
-            while True:
-                event = self._client.next_turn_notification(self.id)
-                yield event
-                if (
-                    event.method == "turn/completed"
-                    and isinstance(event.payload, TurnCompletedNotification)
-                    and event.payload.turn.id == self.id
-                ):
-                    break
-        finally:
-            self._client.unregister_turn_notifications(self.id)
+        if self._goal is not None:
+            return _GoalNotificationStream(
+                self._goal,
+                lambda: self._client.next_goal_notification(self._goal),
+                lambda: self._client.unregister_goal_operation(self._goal),
+            )
+
+        def ordinary_stream() -> Iterator[Notification]:
+            self._client.register_turn_notifications(self.id)
+            try:
+                while True:
+                    event = self._client.next_turn_notification(self.id)
+                    yield event
+                    if (
+                        event.method == "turn/completed"
+                        and isinstance(event.payload, TurnCompletedNotification)
+                        and event.payload.turn.id == self.id
+                    ):
+                        break
+            finally:
+                self._client.unregister_turn_notifications(self.id)
+
+        return ordinary_stream()
 
     def run(self) -> TurnResult:
         """Consume the turn stream and return its completed result."""
@@ -766,10 +868,39 @@ class AsyncTurnHandle:
     _codex: AsyncCodex
     thread_id: str
     id: str
+    _goal: _GoalOperationState | None = None
 
     async def steer(self, input: RunInput) -> TurnSteerResponse:
         """Send additional input to this active turn."""
         await self._codex._ensure_initialized()
+        if self._goal is not None:
+            wire_input = _to_wire_input(_normalize_run_input(input))
+            turn_id = await asyncio.to_thread(self._goal.active_turn)
+            if turn_id is None:
+                raise _inactive_turn_error()
+            try:
+                response = await self._codex._client.turn_steer(
+                    self.thread_id,
+                    turn_id,
+                    wire_input,
+                )
+            except InvalidRequestError as exc:
+                if not (
+                    exc.message == "no active turn to steer"
+                    or exc.message.startswith("expected active turn id")
+                ):
+                    raise
+                next_turn_id = _active_turn_id_from_error(exc)
+                if next_turn_id is None:
+                    next_turn_id = await asyncio.to_thread(self._goal.active_turn, after=turn_id)
+                if next_turn_id is None:
+                    raise _inactive_turn_error() from exc
+                response = await self._codex._client.turn_steer(
+                    self.thread_id,
+                    next_turn_id,
+                    wire_input,
+                )
+            return response.model_copy(update={"turn_id": self.id})
         return await self._codex._client.turn_steer(
             self.thread_id,
             self.id,
@@ -779,24 +910,70 @@ class AsyncTurnHandle:
     async def interrupt(self) -> TurnInterruptResponse:
         """Request interruption of this active turn."""
         await self._codex._ensure_initialized()
+        if self._goal is not None:
+            if not self._goal.begin_interrupt():
+                raise _inactive_interrupt_error()
+            try:
+                await self._codex._client.pause_goal(self.thread_id)
+            except BaseException:
+                self._goal.cancel_interrupt()
+                raise
+            self._goal.confirm_interrupt()
+            turn_id = self._goal.current_turn()
+            if turn_id is None:
+                return TurnInterruptResponse()
+            try:
+                return await self._codex._client.turn_interrupt(self.thread_id, turn_id)
+            except InvalidRequestError as exc:
+                if exc.message == "no active turn to interrupt":
+                    return TurnInterruptResponse()
+                if exc.message.startswith("expected active turn id"):
+                    next_turn_id = _active_turn_id_from_error(exc) or self._goal.current_turn()
+                    if next_turn_id is None or next_turn_id == turn_id:
+                        return TurnInterruptResponse()
+                    try:
+                        return await self._codex._client.turn_interrupt(
+                            self.thread_id,
+                            next_turn_id,
+                        )
+                    except InvalidRequestError as retry_exc:
+                        if retry_exc.message == "no active turn to interrupt":
+                            return TurnInterruptResponse()
+                        raise
+                raise
         return await self._codex._client.turn_interrupt(self.thread_id, self.id)
 
-    async def stream(self) -> AsyncIterator[Notification]:
+    def stream(self) -> AsyncIterator[Notification]:
         """Yield only notifications routed to this async turn handle."""
-        await self._codex._ensure_initialized()
-        self._codex._client.register_turn_notifications(self.id)
-        try:
-            while True:
-                event = await self._codex._client.next_turn_notification(self.id)
-                yield event
-                if (
-                    event.method == "turn/completed"
-                    and isinstance(event.payload, TurnCompletedNotification)
-                    and event.payload.turn.id == self.id
-                ):
-                    break
-        finally:
-            self._codex._client.unregister_turn_notifications(self.id)
+        if self._goal is not None:
+
+            async def next_goal_notification() -> Notification:
+                await self._codex._ensure_initialized()
+                return await self._codex._client.next_goal_notification(self._goal)
+
+            return _AsyncGoalNotificationStream(
+                self._goal,
+                next_goal_notification,
+                lambda: self._codex._client.unregister_goal_operation(self._goal),
+            )
+
+        async def ordinary_stream() -> AsyncIterator[Notification]:
+            await self._codex._ensure_initialized()
+            self._codex._client.register_turn_notifications(self.id)
+            try:
+                while True:
+                    event = await self._codex._client.next_turn_notification(self.id)
+                    yield event
+                    if (
+                        event.method == "turn/completed"
+                        and isinstance(event.payload, TurnCompletedNotification)
+                        and event.payload.turn.id == self.id
+                    ):
+                        break
+            finally:
+                self._codex._client.unregister_turn_notifications(self.id)
+
+        return ordinary_stream()
 
     async def run(self) -> TurnResult:
         """Consume the turn stream and return its completed result."""

sdk/python/src/openai_codex/async_client.py

@@ -6,6 +6,7 @@ from typing import AsyncIterator, Callable, ParamSpec, TypeVar
 
 from pydantic import BaseModel
 
+from ._goal import _GoalOperationState
 from .client import CodexClient, CodexConfig
 from .generated.v2_all import (
     AccountLoginCompletedNotification,
@@ -21,6 +22,7 @@ from .generated.v2_all import (
     ThreadCompactStartResponse,
     ThreadForkParams as V2ThreadForkParams,
     ThreadForkResponse,
+    ThreadGoalSetResponse,
     ThreadListParams as V2ThreadListParams,
     ThreadListResponse,
     ThreadReadResponse,
@@ -107,6 +109,18 @@ class AsyncCodexClient:
         """Unregister a turn notification queue on the wrapped sync client."""
         self._sync.unregister_turn_notifications(turn_id)
 
+    def register_goal_operation(self, thread_id: str) -> _GoalOperationState:
+        """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)
+
     async def request(
         self,
         method: str,
@@ -192,14 +206,26 @@ class AsyncCodexClient:
         """Start thread compaction using the wrapped sync client."""
         return await self._call_sync(self._sync.thread_compact, thread_id)
 
+    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 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(self._sync.turn_start, thread_id, input_items, params)
+        return await self._call_sync(
+            self._sync.turn_start,
+            thread_id,
+            input_items,
+            params,
+            goal=goal,
+        )
 
     async def turn_interrupt(self, thread_id: str, turn_id: str) -> TurnInterruptResponse:
         """Interrupt a turn using the wrapped sync client."""
@@ -256,6 +282,10 @@ class AsyncCodexClient:
         """Wait for the next notification routed to one turn."""
         return await self._call_sync(self._sync.next_turn_notification, turn_id)
 
+    async def next_goal_notification(self, state: _GoalOperationState) -> Notification:
+        """Wait for the next notification in a logical goal turn."""
+        return await self._call_sync(self._sync.next_goal_notification, state)
+
     async def wait_for_login_completed(
         self,
         login_id: str,

sdk/python/src/openai_codex/client.py

@@ -10,6 +10,7 @@ from typing import Callable, Iterator, TypeVar
 
 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
@@ -30,6 +31,8 @@ from .generated.v2_all import (
     ThreadCompactStartResponse,
     ThreadForkParams as V2ThreadForkParams,
     ThreadForkResponse,
+    ThreadGoalSetResponse,
+    ThreadGoalStatus,
     ThreadListParams as V2ThreadListParams,
     ThreadListResponse,
     ThreadReadResponse,
@@ -352,6 +355,23 @@ class CodexClient:
         """Return the next routed notification for the requested turn id."""
         return self._router.next_turn_notification(turn_id)
 
+    def register_goal_operation(self, thread_id: str) -> _GoalOperationState:
+        """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)
+
+    def next_goal_notification(self, state: _GoalOperationState) -> Notification:
+        """Wait for the next notification in a logical goal turn."""
+        return state.next_notification()
+
     def account_login_start(
         self,
         params: V2LoginAccountParams | JsonObject,
@@ -452,11 +472,21 @@ class CodexClient:
             response_model=ThreadCompactStartResponse,
         )
 
+    def pause_goal(self, thread_id: str) -> ThreadGoalSetResponse:
+        """Pause the active goal used by a logical goal turn."""
+        return self.request(
+            "thread/goal/set",
+            {"threadId": thread_id, "status": ThreadGoalStatus.paused.value},
+            response_model=ThreadGoalSetResponse,
+        )
+
     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 = {
@@ -464,6 +494,11 @@ 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

sdk/python/src/openai_codex/generated/notification_registry.py

@@ -169,6 +169,53 @@ NESTED_TURN_NOTIFICATION_TYPES: tuple[type[BaseModel], ...] = (
     TurnStartedNotification,
 )
 
+DIRECT_THREAD_ID_NOTIFICATION_TYPES: tuple[type[BaseModel], ...] = (
+    AgentMessageDeltaNotification,
+    CommandExecutionOutputDeltaNotification,
+    ContextCompactedNotification,
+    ErrorNotification,
+    FileChangeOutputDeltaNotification,
+    FileChangePatchUpdatedNotification,
+    GuardianWarningNotification,
+    HookCompletedNotification,
+    HookStartedNotification,
+    ItemCompletedNotification,
+    ItemGuardianApprovalReviewCompletedNotification,
+    ItemGuardianApprovalReviewStartedNotification,
+    ItemStartedNotification,
+    McpToolCallProgressNotification,
+    ModelReroutedNotification,
+    ModelVerificationNotification,
+    PlanDeltaNotification,
+    ReasoningSummaryPartAddedNotification,
+    ReasoningSummaryTextDeltaNotification,
+    ReasoningTextDeltaNotification,
+    ServerRequestResolvedNotification,
+    TerminalInteractionNotification,
+    ThreadArchivedNotification,
+    ThreadClosedNotification,
+    ThreadGoalClearedNotification,
+    ThreadGoalUpdatedNotification,
+    ThreadNameUpdatedNotification,
+    ThreadRealtimeClosedNotification,
+    ThreadRealtimeErrorNotification,
+    ThreadRealtimeItemAddedNotification,
+    ThreadRealtimeOutputAudioDeltaNotification,
+    ThreadRealtimeSdpNotification,
+    ThreadRealtimeStartedNotification,
+    ThreadRealtimeTranscriptDeltaNotification,
+    ThreadRealtimeTranscriptDoneNotification,
+    ThreadSettingsUpdatedNotification,
+    ThreadStatusChangedNotification,
+    ThreadTokenUsageUpdatedNotification,
+    ThreadUnarchivedNotification,
+    TurnCompletedNotification,
+    TurnDiffUpdatedNotification,
+    TurnPlanUpdatedNotification,
+    TurnStartedNotification,
+    WarningNotification,
+)
+
 
 def notification_turn_id(payload: BaseModel) -> str | None:
     """Return the turn id carried by generated notification payload metadata."""
@@ -177,3 +224,10 @@ def notification_turn_id(payload: BaseModel) -> str | None:
     if isinstance(payload, NESTED_TURN_NOTIFICATION_TYPES):
         return payload.turn.id
     return None
+
+
+def notification_thread_id(payload: BaseModel) -> str | None:
+    """Return the thread id carried by generated notification payload metadata."""
+    if isinstance(payload, DIRECT_THREAD_ID_NOTIFICATION_TYPES):
+        return payload.thread_id
+    return None

sdk/python/src/openai_codex/generated/v2_all.py

@@ -6956,6 +6956,12 @@ class TurnStartParams(BaseModel):
         ReasoningEffort | None,
         Field(description="Override the reasoning effort for this turn and subsequent turns."),
     ] = None
+    goal: Annotated[
+        bool | None,
+        Field(
+            description="Replace the thread's active goal with an objective derived from this turn's text input."
+        ),
+    ] = False
     input: list[UserInput]
     model: Annotated[
         str | None, Field(description="Override the model for this turn and subsequent turns.")

sdk/python/src/openai_codex/models.py

@@ -27,6 +27,8 @@ from .generated.v2_all import (
     ReasoningSummaryTextDeltaNotification,
     ReasoningTextDeltaNotification,
     TerminalInteractionNotification,
+    ThreadGoalClearedNotification,
+    ThreadGoalUpdatedNotification,
     ThreadNameUpdatedNotification,
     ThreadStartedNotification,
     ThreadTokenUsageUpdatedNotification,
@@ -70,6 +72,8 @@ NotificationPayload: TypeAlias = (
     | ReasoningTextDeltaNotification
     | TerminalInteractionNotification
     | ThreadNameUpdatedNotification
+    | ThreadGoalClearedNotification
+    | ThreadGoalUpdatedNotification
     | ThreadStartedNotification
     | ThreadTokenUsageUpdatedNotification
     | TurnCompletedNotification