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