From f7fbfc81a5955da39af510e1a74cf40f0469c09e Mon Sep 17 00:00:00 2001 From: Ahmed Ibrahim Date: Sun, 7 Jun 2026 16:25:10 -0700 Subject: [PATCH] Add real-process Python goal tests --- .github/workflows/sdk.yml | 12 +- sdk/python/tests/app_server_harness.py | 19 +- .../tests/test_app_server_goal_turns.py | 758 ++++++++++++++++++ sdk/python/tests/test_client_rpc_methods.py | 64 -- sdk/python/tests/test_goal_turns.py | 598 -------------- 5 files changed, 783 insertions(+), 668 deletions(-) create mode 100644 sdk/python/tests/test_app_server_goal_turns.py delete mode 100644 sdk/python/tests/test_goal_turns.py diff --git a/.github/workflows/sdk.yml b/.github/workflows/sdk.yml index 42a44afdb5..3f33d8d483 100644 --- a/.github/workflows/sdk.yml +++ b/.github/workflows/sdk.yml @@ -45,7 +45,7 @@ jobs: runs-on: group: codex-runners labels: codex-linux-x64 - timeout-minutes: 10 + timeout-minutes: 15 environment: name: bazel deployment: false @@ -141,6 +141,16 @@ jobs: set -euo pipefail "${CODEX_EXEC_PATH}" --version + - name: Test Python goal turns against source Codex + shell: bash + run: | + set -euo pipefail + python3 -m venv /tmp/codex-python-sdk-uv + /tmp/codex-python-sdk-uv/bin/python -m pip install uv==0.11.3 + cd sdk/python + /tmp/codex-python-sdk-uv/bin/uv sync --extra dev --frozen + /tmp/codex-python-sdk-uv/bin/uv run --extra dev pytest tests/test_app_server_goal_turns.py + - name: Install dependencies run: pnpm install --frozen-lockfile diff --git a/sdk/python/tests/app_server_harness.py b/sdk/python/tests/app_server_harness.py index cf0c847394..135ea47f0d 100644 --- a/sdk/python/tests/app_server_harness.py +++ b/sdk/python/tests/app_server_harness.py @@ -204,13 +204,20 @@ class MockResponsesServer: class AppServerHarness: - """Test fixture that points a pinned runtime app-server at MockResponsesServer.""" + """Test fixture that points an app-server process at MockResponsesServer.""" - def __init__(self, tmp_path: Path, *, requires_openai_auth: bool = False) -> None: + def __init__( + self, + tmp_path: Path, + *, + requires_openai_auth: bool = False, + enable_goals: bool = False, + ) -> None: self.tmp_path = tmp_path self.codex_home = tmp_path / "codex-home" self.workspace = tmp_path / "workspace" self.requires_openai_auth = requires_openai_auth + self.enable_goals = enable_goals self.responses = MockResponsesServer() def __enter__(self) -> AppServerHarness: @@ -225,9 +232,10 @@ class AppServerHarness: shutil.rmtree(self.codex_home, ignore_errors=True) shutil.rmtree(self.workspace, ignore_errors=True) - def app_server_config(self) -> CodexConfig: - """Build SDK config for an isolated pinned-runtime app-server process.""" + def app_server_config(self, *, codex_bin: str | None = None) -> CodexConfig: + """Build SDK config for an isolated app-server process.""" return CodexConfig( + codex_bin=codex_bin, cwd=str(self.workspace), env={ "CODEX_HOME": str(self.codex_home), @@ -240,6 +248,7 @@ class AppServerHarness: """Write config.toml that routes model calls to the mock server.""" config_toml = self.codex_home / "config.toml" requires_openai_auth = "requires_openai_auth = true\n" if self.requires_openai_auth else "" + goals = "[features]\ngoals = true\n\n" if self.enable_goals else "" config_toml.write_text( f""" model = "mock-model" @@ -248,7 +257,7 @@ sandbox_mode = "read-only" model_provider = "mock_provider" -[model_providers.mock_provider] +{goals}[model_providers.mock_provider] name = "Mock provider for Python SDK tests" base_url = "{self.responses.url}/v1" wire_api = "responses" diff --git a/sdk/python/tests/test_app_server_goal_turns.py b/sdk/python/tests/test_app_server_goal_turns.py new file mode 100644 index 0000000000..4248d7e018 --- /dev/null +++ b/sdk/python/tests/test_app_server_goal_turns.py @@ -0,0 +1,758 @@ +import asyncio +import os + +import pytest +from app_server_harness import ( + AppServerHarness, + ev_assistant_message, + ev_completed_with_usage, + ev_failed, + ev_function_call, + ev_response_created, + sse, +) +from app_server_helpers import ( + agent_message_texts, + agent_message_texts_from_items, + streaming_response, +) + +from openai_codex import AsyncCodex, Codex, CodexConfig, TextInput +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, + prefix: str, + *, + initial_text: str, + final_text: str, +) -> None: + harness.responses.enqueue_sse( + sse( + [ + ev_response_created(f"{prefix}-initial"), + ev_assistant_message(f"msg-{prefix}-initial", initial_text), + ev_completed_with_usage( + f"{prefix}-initial", + input_tokens=3, + cached_input_tokens=1, + output_tokens=2, + reasoning_output_tokens=1, + total_tokens=5, + ), + ] + ) + ) + harness.responses.enqueue_sse( + sse( + [ + ev_response_created(f"{prefix}-complete-tool"), + ev_function_call( + f"call-{prefix}-complete", + "update_goal", + '{"status":"complete"}', + ), + ev_completed_with_usage( + f"{prefix}-complete-tool", + input_tokens=4, + cached_input_tokens=1, + output_tokens=1, + reasoning_output_tokens=1, + total_tokens=5, + ), + ] + ) + ) + harness.responses.enqueue_sse( + sse( + [ + ev_response_created(f"{prefix}-final"), + ev_assistant_message(f"msg-{prefix}-final", final_text), + ev_completed_with_usage( + f"{prefix}-final", + input_tokens=5, + cached_input_tokens=0, + output_tokens=3, + reasoning_output_tokens=0, + total_tokens=8, + ), + ] + ) + ) + + +def _enqueue_steerable_goal( + harness: AppServerHarness, + prefix: str, + *, + initial_text: str, + continuation_chunks: list[str], + final_text: str, +) -> None: + harness.responses.enqueue_assistant_message( + initial_text, + response_id=f"{prefix}-initial", + ) + harness.responses.enqueue_sse( + streaming_response( + f"{prefix}-continuation", + f"msg-{prefix}-continuation", + continuation_chunks, + ), + delay_between_events_s=0.15, + ) + harness.responses.enqueue_sse( + sse( + [ + ev_response_created(f"{prefix}-complete-tool"), + ev_function_call( + f"call-{prefix}-complete", + "update_goal", + '{"status":"complete"}', + ), + ev_completed_with_usage( + f"{prefix}-complete-tool", + input_tokens=1, + cached_input_tokens=0, + output_tokens=1, + reasoning_output_tokens=0, + total_tokens=2, + ), + ] + ) + ) + harness.responses.enqueue_assistant_message( + final_text, + response_id=f"{prefix}-final", + ) + + +def _enqueue_interruptible_goal( + harness: AppServerHarness, + prefix: str, + *, + initial_text: str, + continuation_chunks: list[str], + follow_up_text: str, +) -> None: + harness.responses.enqueue_assistant_message( + initial_text, + response_id=f"{prefix}-initial", + ) + harness.responses.enqueue_sse( + streaming_response( + f"{prefix}-continuation", + f"msg-{prefix}-continuation", + continuation_chunks, + ), + delay_between_events_s=0.2, + ) + harness.responses.enqueue_assistant_message( + follow_up_text, + response_id=f"{prefix}-follow-up", + ) + + +def _continuation_text(request) -> str: + return "\n".join(request.message_input_texts("user")) + + +def test_sync_goal_run_aggregates_automatic_continuation(tmp_path) -> None: + """The public result should cover the initial and automatic continuation turns.""" + with AppServerHarness(tmp_path, enable_goals=True) as harness: + _enqueue_completed_goal( + harness, + "sync-run", + initial_text="Initial pass complete.", + final_text="Goal complete.", + ) + + with Codex(config=_source_config(harness)) 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() + 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, + "status": result.status, + "messages": agent_message_texts_from_items(result.items), + "final_response": result.final_response, + "usage": usage, + "timing": ( + result.started_at is not None, + result.completed_at is not None, + result.duration_ms == max(0, result.completed_at - result.started_at) * 1000 + if result.started_at is not None and result.completed_at is not None + else False, + ), + "initial_input": requests[0].message_input_texts("user")[-2:], + "model": first_body["model"], + "output_schema": first_body["text"]["format"]["schema"], + "continuation_has_objective": ( + "\nImprove benchmark coverage\n\nDocument the results\n" + in _continuation_text(requests[1]) + ), + } == { + "id": turn.id, + "status": TurnStatus.completed, + "messages": ["Initial pass complete.", "Goal complete."], + "final_response": "Goal complete.", + "usage": { + "last": { + "cachedInputTokens": 0, + "inputTokens": 5, + "outputTokens": 3, + "reasoningOutputTokens": 0, + "totalTokens": 8, + }, + "total": { + "cachedInputTokens": 2, + "inputTokens": 12, + "outputTokens": 6, + "reasoningOutputTokens": 2, + "totalTokens": 18, + }, + }, + "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, + } + + +def test_goal_stream_exposes_one_logical_lifecycle(tmp_path) -> None: + """Continuation boundaries should not leak through the public stream.""" + with AppServerHarness(tmp_path, enable_goals=True) as harness: + harness.responses.enqueue_sse( + streaming_response( + "stream-initial", + "msg-stream-initial", + ["initial ", "pass"], + ) + ) + harness.responses.enqueue_sse( + sse( + [ + ev_response_created("stream-complete-tool"), + ev_function_call( + "call-stream-complete", + "update_goal", + '{"status":"complete"}', + ), + ev_completed_with_usage( + "stream-complete-tool", + input_tokens=1, + cached_input_tokens=0, + output_tokens=1, + reasoning_output_tokens=0, + total_tokens=2, + ), + ] + ) + ) + harness.responses.enqueue_sse( + streaming_response( + "stream-final", + "msg-stream-final", + ["goal ", "complete"], + ) + ) + + with Codex(config=_source_config(harness)) as codex: + turn = codex.thread_start().turn("Finish the integration suite", goal=True) + events = list(turn.stream()) + + lifecycle = [event for event in events if event.method in {"turn/started", "turn/completed"}] + routed_ids = [ + turn_id for event in events if (turn_id := notification_turn_id(event.payload)) is not None + ] + assert { + "lifecycle": [event.method for event in lifecycle], + "routed_ids": routed_ids, + "deltas": [ + event.payload.delta + for event in events + if isinstance(event.payload, AgentMessageDeltaNotification) + ], + "messages": agent_message_texts(events), + "completion_statuses": [ + event.payload.turn.status + for event in events + if isinstance(event.payload, TurnCompletedNotification) + ], + } == { + "lifecycle": ["turn/started", "turn/completed"], + "routed_ids": [turn.id] * len(routed_ids), + "deltas": ["initial ", "pass", "goal ", "complete"], + "messages": ["initial pass", "goal complete"], + "completion_statuses": [TurnStatus.completed], + } + + +def test_goal_can_complete_within_the_initial_server_turn(tmp_path) -> None: + """Completing the goal before turn end should not create a continuation turn.""" + with AppServerHarness(tmp_path, enable_goals=True) as harness: + harness.responses.enqueue_sse( + sse( + [ + ev_response_created("single-turn-complete-tool"), + ev_function_call( + "call-single-turn-complete", + "update_goal", + '{"status":"complete"}', + ), + ev_completed_with_usage( + "single-turn-complete-tool", + input_tokens=1, + cached_input_tokens=0, + output_tokens=1, + reasoning_output_tokens=0, + total_tokens=2, + ), + ] + ) + ) + harness.responses.enqueue_assistant_message( + "Completed without a continuation.", + response_id="single-turn-final", + ) + + with Codex(config=_source_config(harness)) as codex: + turn = codex.thread_start().turn("Finish in the initial turn", goal=True) + result = turn.run() + requests = harness.responses.wait_for_requests(2) + with pytest.raises(InvalidRequestError) as steer_error: + turn.steer("Keep working") + with pytest.raises(InvalidRequestError) as interrupt_error: + turn.interrupt() + + assert { + "result": (result.id, result.status, result.final_response), + "messages": agent_message_texts_from_items(result.items), + "request_count": len(requests), + "inactive_errors": [steer_error.value.message, interrupt_error.value.message], + } == { + "result": (turn.id, TurnStatus.completed, "Completed without a continuation."), + "messages": ["Completed without a continuation."], + "request_count": 2, + "inactive_errors": ["no active turn to steer", "no active turn to interrupt"], + } + + +def test_goal_replaces_an_existing_persisted_goal(tmp_path) -> None: + """Goal mode should replace a resumable persisted goal before model work begins.""" + with AppServerHarness(tmp_path, enable_goals=True) as harness: + _enqueue_completed_goal( + harness, + "replacement", + initial_text="Replacement started.", + final_text="Replacement complete.", + ) + + with Codex(config=_source_config(harness)) 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, + ).goal + result = thread.run("Publish the replacement objective", goal=True) + persisted = codex._client.request( + "thread/goal/get", + {"threadId": thread.id}, + response_model=ThreadGoalGetResponse, + ).goal + requests = harness.responses.wait_for_requests(3) + + assert { + "previous": (previous.objective, previous.status, previous.token_budget), + "result": (result.status, result.final_response), + "persisted": ( + persisted.objective if persisted else None, + persisted.status if persisted else None, + persisted.token_budget if persisted else None, + ), + "continuation_has_replacement": ( + "Publish the replacement objective" in _continuation_text(requests[1]) + ), + "continuation_has_previous": ( + "Keep the old benchmark objective" in _continuation_text(requests[1]) + ), + } == { + "previous": ( + "Keep the old benchmark objective", + ThreadGoalStatus.paused, + 500, + ), + "result": (TurnStatus.completed, "Replacement complete."), + "persisted": ( + "Publish the replacement objective", + ThreadGoalStatus.complete, + None, + ), + "continuation_has_replacement": True, + "continuation_has_previous": False, + } + + +def test_goal_steer_targets_an_active_continuation(tmp_path) -> None: + """Steering should reach the active server turn while returning the logical ID.""" + with AppServerHarness(tmp_path, enable_goals=True) as harness: + _enqueue_steerable_goal( + harness, + "steer", + initial_text="Initial work complete.", + continuation_chunks=["continuation ", "work"], + 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) + harness.responses.wait_for_requests(2) + steer = turn.steer("Prioritize the edge-case coverage.") + result = turn.run() + requests = harness.responses.wait_for_requests(4) + + assert { + "steer_id": steer.turn_id, + "result": (result.id, result.status, result.final_response), + "messages": agent_message_texts_from_items(result.items), + "steering_input": requests[2].message_input_texts("user")[-1:], + } == { + "steer_id": turn.id, + "result": (turn.id, TurnStatus.completed, "Steered goal complete."), + "messages": [ + "Initial work complete.", + "continuation work", + "Steered goal complete.", + ], + "steering_input": ["Prioritize the edge-case coverage."], + } + + +def test_goal_interrupt_pauses_continuation_and_leaves_thread_usable(tmp_path) -> None: + """Interrupt should stop the logical goal operation and permit ordinary follow-up work.""" + with AppServerHarness(tmp_path, enable_goals=True) as harness: + _enqueue_interruptible_goal( + harness, + "interrupt", + initial_text="Initial work complete.", + continuation_chunks=["still ", "working"], + follow_up_text="Ordinary follow-up complete.", + ) + + with Codex(config=_source_config(harness)) as codex: + thread = codex.thread_start() + turn = thread.turn("Start interruptible goal work", goal=True) + harness.responses.wait_for_requests(2) + interrupt = turn.interrupt() + interrupted = turn.run() + follow_up = thread.run("Continue with an ordinary turn") + harness.responses.wait_for_requests(3) + requests = harness.responses.requests() + + assert { + "interrupt": interrupt.model_dump(by_alias=True, mode="json"), + "interrupted": (interrupted.id, interrupted.status), + "follow_up": (follow_up.status, follow_up.final_response), + "request_count": len(requests), + "follow_up_input": requests[2].message_input_texts("user")[-1:], + } == { + "interrupt": {}, + "interrupted": (turn.id, TurnStatus.interrupted), + "follow_up": (TurnStatus.completed, "Ordinary follow-up complete."), + "request_count": 3, + "follow_up_input": ["Continue with an ordinary turn"], + } + + +def test_terminal_goal_failure_stops_continuation_and_releases_routing(tmp_path) -> None: + """A failed server turn should end the logical operation without another continuation.""" + with AppServerHarness(tmp_path, enable_goals=True) as harness: + harness.responses.enqueue_sse( + sse( + [ + ev_response_created("goal-terminal-failure"), + ev_failed("goal-terminal-failure", "goal model failed"), + ] + ) + ) + harness.responses.enqueue_assistant_message( + "Recovered with an ordinary turn.", + response_id="goal-failure-follow-up", + ) + + with Codex(config=_source_config(harness)) as codex: + thread = codex.thread_start() + with pytest.raises(RuntimeError, match="goal model failed"): + thread.run("Fail this goal turn", goal=True) + + follow_up = thread.run("Run after the goal failure") + harness.responses.wait_for_requests(2) + requests = harness.responses.requests() + + assert { + "follow_up": (follow_up.status, follow_up.final_response), + "request_count": len(requests), + "follow_up_input": requests[1].message_input_texts("user")[-1:], + } == { + "follow_up": (TurnStatus.completed, "Recovered with an ordinary turn."), + "request_count": 2, + "follow_up_input": ["Run after the goal failure"], + } + + +def test_closing_goal_stream_releases_real_process_routing(tmp_path) -> None: + """Closing a public stream should immediately unregister its logical operation.""" + with AppServerHarness(tmp_path, enable_goals=True) as harness: + harness.responses.enqueue_sse( + streaming_response( + "stream-close", + "msg-stream-close", + ["long ", "running ", "goal"], + ), + 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) + harness.responses.wait_for_requests(1) + stream = turn.stream() + stream.close() + registered_goals = dict(codex._client._router._goal_operations) + + assert registered_goals == {} + + +def test_app_server_exit_unblocks_goal_stream_and_releases_routing(tmp_path) -> None: + """A real transport death should wake the logical reader and clean up routing.""" + with AppServerHarness(tmp_path, enable_goals=True) as harness: + harness.responses.enqueue_sse( + streaming_response( + "transport-exit", + "msg-transport-exit", + ["long ", "running ", "goal"], + ), + 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) + harness.responses.wait_for_requests(1) + process = codex._client._proc + assert process is not None + process.terminate() + process.wait(timeout=5) + + with pytest.raises(TransportClosedError): + turn.run() + registered_goals = dict(codex._client._router._goal_operations) + + assert registered_goals == {} + + +def test_failed_goal_starts_release_routing_without_model_requests(tmp_path) -> None: + """Validation failures should leave the client and persisted thread usable.""" + with AppServerHarness(tmp_path, enable_goals=True) as harness: + harness.responses.enqueue_assistant_message( + "Ordinary turn complete.", + response_id="validation-follow-up", + ) + + with Codex(config=_source_config(harness)) as codex: + thread = codex.thread_start() + with pytest.raises(InvalidRequestError) as empty_error: + thread.turn(" ", goal=True) + + ephemeral = codex.thread_start(ephemeral=True) + with pytest.raises(InvalidRequestError) as ephemeral_error: + ephemeral.turn("Persist this goal", goal=True) + + follow_up = thread.run("Run after rejected goals") + requests = harness.responses.wait_for_requests(1) + + assert { + "errors": [empty_error.value.message, ephemeral_error.value.message], + "follow_up": (follow_up.status, follow_up.final_response), + "request_count": len(requests), + } == { + "errors": [ + "goal objective must not be empty", + f"ephemeral thread does not support goals: {ephemeral.id}", + ], + "follow_up": (TurnStatus.completed, "Ordinary turn complete."), + "request_count": 1, + } + + +def test_async_goal_run_matches_sync_logical_result(tmp_path) -> None: + """The async public API should aggregate the same real continuation lifecycle.""" + + async def scenario() -> None: + with AppServerHarness(tmp_path, enable_goals=True) as harness: + _enqueue_completed_goal( + harness, + "async-run", + initial_text="Async initial pass.", + final_text="Async goal complete.", + ) + + async with AsyncCodex(config=_source_config(harness)) as codex: + thread = await codex.thread_start() + turn = await thread.turn("Finish the async goal", goal=True) + result = await turn.run() + 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]) + ), + "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()) + + +def test_async_goal_steer_targets_an_active_continuation(tmp_path) -> None: + """Async steering should target the continuation and preserve the logical ID.""" + + async def scenario() -> None: + with AppServerHarness(tmp_path, enable_goals=True) as harness: + _enqueue_steerable_goal( + harness, + "async-steer", + initial_text="Async initial work complete.", + continuation_chunks=["async continuation ", "work"], + final_text="Async steered goal complete.", + ) + + async with AsyncCodex(config=_source_config(harness)) as codex: + thread = await codex.thread_start() + turn = await thread.turn( + "Start an async goal that needs refinement", + goal=True, + ) + await asyncio.to_thread(harness.responses.wait_for_requests, 2) + steer = await turn.steer("Prioritize async edge-case coverage.") + result = await turn.run() + requests = await asyncio.to_thread(harness.responses.wait_for_requests, 4) + + assert { + "steer_id": steer.turn_id, + "result": (result.id, result.status, result.final_response), + "messages": agent_message_texts_from_items(result.items), + "steering_input": requests[2].message_input_texts("user")[-1:], + } == { + "steer_id": turn.id, + "result": (turn.id, TurnStatus.completed, "Async steered goal complete."), + "messages": [ + "Async initial work complete.", + "async continuation work", + "Async steered goal complete.", + ], + "steering_input": ["Prioritize async edge-case coverage."], + } + + asyncio.run(scenario()) + + +def test_async_goal_interrupts_an_active_continuation(tmp_path) -> None: + """Async interruption should stop the goal and leave the thread usable.""" + + async def scenario() -> None: + with AppServerHarness(tmp_path, enable_goals=True) as harness: + _enqueue_interruptible_goal( + harness, + "async-interrupt", + initial_text="Async initial work complete.", + continuation_chunks=["async still ", "working"], + follow_up_text="Async ordinary follow-up complete.", + ) + + async with AsyncCodex(config=_source_config(harness)) as codex: + thread = await codex.thread_start() + turn = await thread.turn("Start async interruptible goal work", goal=True) + await asyncio.to_thread(harness.responses.wait_for_requests, 2) + interrupt = await turn.interrupt() + interrupted = await turn.run() + follow_up = await thread.run("Continue with an async ordinary turn") + await asyncio.to_thread(harness.responses.wait_for_requests, 3) + requests = harness.responses.requests() + + assert { + "interrupt": interrupt.model_dump(by_alias=True, mode="json"), + "interrupted": (interrupted.id, interrupted.status), + "follow_up": (follow_up.status, follow_up.final_response), + "request_count": len(requests), + "follow_up_input": requests[2].message_input_texts("user")[-1:], + } == { + "interrupt": {}, + "interrupted": (turn.id, TurnStatus.interrupted), + "follow_up": (TurnStatus.completed, "Async ordinary follow-up complete."), + "request_count": 3, + "follow_up_input": ["Continue with an async ordinary turn"], + } + + asyncio.run(scenario()) diff --git a/sdk/python/tests/test_client_rpc_methods.py b/sdk/python/tests/test_client_rpc_methods.py index 406926ca64..89a3e0f463 100644 --- a/sdk/python/tests/test_client_rpc_methods.py +++ b/sdk/python/tests/test_client_rpc_methods.py @@ -14,8 +14,6 @@ from openai_codex.generated.v2_all import ( ThreadResumeResponse, ThreadTokenUsageUpdatedNotification, TurnCompletedNotification, - TurnStartParams, - TurnStartResponse, WarningNotification, ) from openai_codex.models import Notification, UnknownNotification @@ -31,73 +29,11 @@ def test_generated_params_models_are_snake_case_and_dump_by_alias() -> None: assert dumped == {"searchTerm": "needle", "limit": 5} -def test_turn_start_sends_goal_only_when_enabled() -> None: - """The low-level request should preserve false-by-omission wire behavior.""" - client = CodexClient() - requests: list[dict[str, object]] = [] - - def fake_request(method, params, *, response_model): - requests.append({"method": method, "params": params, "response_model": response_model}) - return TurnStartResponse.model_validate( - {"turn": {"id": f"turn-{len(requests)}", "items": [], "status": "inProgress"}} - ) - - client.request = fake_request # type: ignore[method-assign] - - client.turn_start("thread-1", "ordinary") - client.turn_start("thread-1", "goal", goal=True) - client.turn_start( - "thread-1", - "typed goal", - TurnStartParams(threadId="thread-1", input=[], goal=True), - ) - - assert requests == [ - { - "method": "turn/start", - "params": { - "threadId": "thread-1", - "input": [{"type": "text", "text": "ordinary"}], - }, - "response_model": TurnStartResponse, - }, - { - "method": "turn/start", - "params": { - "threadId": "thread-1", - "input": [{"type": "text", "text": "goal"}], - "goal": True, - }, - "response_model": TurnStartResponse, - }, - { - "method": "turn/start", - "params": { - "threadId": "thread-1", - "input": [{"type": "text", "text": "typed goal"}], - "goal": True, - }, - "response_model": TurnStartResponse, - }, - ] - - def test_generated_v2_bundle_has_single_shared_plan_type_definition() -> None: source = (ROOT / "src" / "openai_codex" / "generated" / "v2_all.py").read_text() assert source.count("class PlanType(") == 1 -def test_generated_turn_start_params_include_goal_mode() -> None: - assert ( - TurnStartParams( - threadId="thread-1", - input=[{"type": "text", "text": "goal"}], - goal=True, - ).goal - is True - ) - - def test_thread_resume_response_accepts_auto_review_reviewer() -> None: """Generated response models should keep accepting the auto review enum value.""" response = ThreadResumeResponse.model_validate( diff --git a/sdk/python/tests/test_goal_turns.py b/sdk/python/tests/test_goal_turns.py deleted file mode 100644 index 039d8ffc80..0000000000 --- a/sdk/python/tests/test_goal_turns.py +++ /dev/null @@ -1,598 +0,0 @@ -import asyncio -import threading -from dataclasses import dataclass - -import pytest - -from openai_codex.api import AsyncTurnHandle, TurnHandle -from openai_codex.client import CodexClient -from openai_codex.errors import InvalidRequestError -from openai_codex.generated.notification_registry import notification_turn_id -from openai_codex.generated.v2_all import ( - TurnCompletedNotification, - TurnInterruptResponse, - TurnStatus, - TurnSteerResponse, -) - - -def _route(client: CodexClient, method: str, params: dict[str, object]) -> None: - client._router.route_notification(client._coerce_notification(method, params)) - - -def _turn(turn_id: str, status: str, *, started_at: int, completed_at: int | None = None): - turn: dict[str, object] = { - "id": turn_id, - "items": [], - "startedAt": started_at, - "status": status, - } - if completed_at is not None: - turn["completedAt"] = completed_at - return turn - - -def _goal(thread_id: str, status: str) -> dict[str, object]: - return { - "createdAt": 1, - "objective": "Improve benchmark coverage", - "status": status, - "threadId": thread_id, - "timeUsedSeconds": 0, - "tokensUsed": 0, - "updatedAt": 1, - } - - -def test_goal_turn_collects_continuations_as_one_logical_turn() -> None: - client = CodexClient() - state = client.register_goal_operation("thread-1") - state.bind("turn-1") - handle = TurnHandle(client, "thread-1", "turn-1", _goal=state) - - _route( - client, - "thread/goal/updated", - {"threadId": "thread-1", "turnId": "turn-1", "goal": _goal("thread-1", "active")}, - ) - _route( - client, - "turn/started", - {"threadId": "thread-1", "turn": _turn("turn-1", "inProgress", started_at=10)}, - ) - _route( - client, - "item/completed", - { - "threadId": "thread-1", - "turnId": "turn-1", - "item": {"id": "message-1", "type": "agentMessage", "text": "working"}, - }, - ) - _route( - client, - "turn/completed", - { - "threadId": "thread-1", - "turn": _turn("turn-1", "completed", started_at=10, completed_at=12), - }, - ) - _route( - client, - "turn/started", - {"threadId": "thread-1", "turn": _turn("turn-2", "inProgress", started_at=13)}, - ) - _route( - client, - "item/completed", - { - "threadId": "thread-1", - "turnId": "turn-2", - "item": { - "id": "message-2", - "type": "agentMessage", - "text": "done", - "phase": "final_answer", - }, - }, - ) - _route( - client, - "thread/goal/updated", - {"threadId": "thread-1", "turnId": "turn-2", "goal": _goal("thread-1", "complete")}, - ) - _route( - client, - "thread/tokenUsage/updated", - { - "threadId": "thread-1", - "turnId": "turn-2", - "tokenUsage": { - "last": { - "cachedInputTokens": 1, - "inputTokens": 2, - "outputTokens": 3, - "reasoningOutputTokens": 4, - "totalTokens": 9, - }, - "total": { - "cachedInputTokens": 10, - "inputTokens": 20, - "outputTokens": 30, - "reasoningOutputTokens": 40, - "totalTokens": 90, - }, - }, - }, - ) - _route( - client, - "turn/completed", - { - "threadId": "thread-1", - "turn": _turn("turn-2", "completed", started_at=13, completed_at=15), - }, - ) - - result = handle.run() - - assert { - "id": result.id, - "status": result.status, - "started_at": result.started_at, - "completed_at": result.completed_at, - "duration_ms": result.duration_ms, - "final_response": result.final_response, - "item_count": len(result.items), - "total_tokens": result.usage.total.total_tokens if result.usage is not None else None, - "registered_goals": client._router._goal_operations, - } == { - "id": "turn-1", - "status": TurnStatus.completed, - "started_at": 10, - "completed_at": 15, - "duration_ms": 5000, - "final_response": "done", - "item_count": 2, - "total_tokens": 90, - "registered_goals": {}, - } - - -def test_goal_stream_hides_continuation_boundaries_and_rewrites_ids() -> None: - client = CodexClient() - state = client.register_goal_operation("thread-1") - state.bind("turn-1") - handle = TurnHandle(client, "thread-1", "turn-1", _goal=state) - - for method, params in [ - ( - "thread/goal/updated", - {"threadId": "thread-1", "turnId": "turn-1", "goal": _goal("thread-1", "active")}, - ), - ( - "turn/started", - {"threadId": "thread-1", "turn": _turn("turn-1", "inProgress", started_at=10)}, - ), - ( - "turn/completed", - { - "threadId": "thread-1", - "turn": _turn("turn-1", "completed", started_at=10, completed_at=11), - }, - ), - ( - "turn/started", - {"threadId": "thread-1", "turn": _turn("turn-2", "inProgress", started_at=12)}, - ), - ( - "thread/goal/updated", - {"threadId": "thread-1", "turnId": "turn-2", "goal": _goal("thread-1", "complete")}, - ), - ( - "turn/completed", - { - "threadId": "thread-1", - "turn": _turn("turn-2", "completed", started_at=12, completed_at=13), - }, - ), - ]: - _route(client, method, params) - - events = list(handle.stream()) - lifecycle = [event for event in events if event.method in {"turn/started", "turn/completed"}] - - assert { - "lifecycle": [event.method for event in lifecycle], - "turn_ids": [ - turn_id - for event in events - if (turn_id := notification_turn_id(event.payload)) is not None - ], - "completed": [ - event.payload.turn.status - for event in events - if isinstance(event.payload, TurnCompletedNotification) - ], - } == { - "lifecycle": ["turn/started", "turn/completed"], - "turn_ids": ["turn-1", "turn-1"], - "completed": [TurnStatus.completed], - } - - -def test_goal_router_releases_and_wakes_operations_on_transport_failure() -> None: - client = CodexClient() - state = client.register_goal_operation("thread-1") - failure = RuntimeError("transport closed") - - client._router.fail_all(failure) - - with pytest.raises(RuntimeError, match="transport closed"): - state.next_notification() - assert client._router._goal_operations == {} - - -def test_closing_unstarted_goal_stream_releases_route_and_controls() -> None: - client = CodexClient() - state = client.register_goal_operation("thread-1") - state.bind("turn-1") - handle = TurnHandle(client, "thread-1", "turn-1", _goal=state) - - stream = handle.stream() - stream.close() - - assert { - "active_turn": state.active_turn(), - "registered_goals": client._router._goal_operations, - } == {"active_turn": None, "registered_goals": {}} - - -def test_failed_goal_start_propagates_turn_error() -> None: - client = CodexClient() - state = client.register_goal_operation("thread-1") - state.bind("turn-1") - handle = TurnHandle(client, "thread-1", "turn-1", _goal=state) - failed_turn = _turn("turn-1", "failed", started_at=10, completed_at=11) - failed_turn["error"] = {"message": "failed to start goal"} - - _route( - client, - "turn/started", - {"threadId": "thread-1", "turn": _turn("turn-1", "inProgress", started_at=10)}, - ) - _route( - client, - "turn/completed", - {"threadId": "thread-1", "turn": failed_turn}, - ) - - with pytest.raises(RuntimeError, match="failed to start goal"): - handle.run() - assert client._router._goal_operations == {} - - -class _SyncControlClient: - def __init__(self, *, fail_first: bool = True) -> None: - self.steer_ids: list[str] = [] - self.interrupt_ids: list[str] = [] - self.paused = False - self.fail_first = fail_first - - def turn_steer(self, _thread_id, turn_id, _input): - self.steer_ids.append(turn_id) - if self.fail_first and len(self.steer_ids) == 1: - raise InvalidRequestError( - -32600, - "expected active turn id `turn-1` but found `turn-2`", - ) - return TurnSteerResponse(turn_id=turn_id) - - def pause_goal(self, _thread_id): - self.paused = True - - def turn_interrupt(self, _thread_id, turn_id): - self.interrupt_ids.append(turn_id) - if self.fail_first and len(self.interrupt_ids) == 1: - raise InvalidRequestError( - -32600, - "expected active turn id turn-1 but found turn-2", - ) - return TurnInterruptResponse() - - -def test_goal_controls_hide_rollover_turn_ids() -> None: - state = CodexClient().register_goal_operation("thread-1") - state.bind("turn-1") - client = _SyncControlClient() - handle = TurnHandle(client, "thread-1", "turn-1", _goal=state) # type: ignore[arg-type] - - steered = handle.steer("keep going") - interrupted = handle.interrupt() - - assert { - "steer_ids": client.steer_ids, - "public_steer_id": steered.turn_id, - "paused": client.paused, - "interrupt_ids": client.interrupt_ids, - "interrupt": interrupted.model_dump(mode="json"), - } == { - "steer_ids": ["turn-1", "turn-2"], - "public_steer_id": "turn-1", - "paused": True, - "interrupt_ids": ["turn-1", "turn-2"], - "interrupt": {}, - } - - -def test_finished_goal_controls_match_inactive_turn_errors() -> None: - state = CodexClient().register_goal_operation("thread-1") - state.bind("turn-1") - state.finish() - client = _SyncControlClient(fail_first=False) - handle = TurnHandle(client, "thread-1", "turn-1", _goal=state) # type: ignore[arg-type] - - with pytest.raises(InvalidRequestError, match="no active turn to steer"): - handle.steer("keep going") - with pytest.raises(InvalidRequestError, match="no active turn to interrupt"): - handle.interrupt() - assert client.paused is False - - -def test_goal_steer_waits_for_rollover_turn() -> None: - client = CodexClient() - state = client.register_goal_operation("thread-1") - state.bind("turn-1") - control = _SyncControlClient(fail_first=False) - handle = TurnHandle(control, "thread-1", "turn-1", _goal=state) # type: ignore[arg-type] - results: list[TurnSteerResponse] = [] - - _route( - client, - "thread/goal/updated", - {"threadId": "thread-1", "turnId": "turn-1", "goal": _goal("thread-1", "active")}, - ) - _route( - client, - "turn/started", - {"threadId": "thread-1", "turn": _turn("turn-1", "inProgress", started_at=10)}, - ) - _route( - client, - "turn/completed", - { - "threadId": "thread-1", - "turn": _turn("turn-1", "completed", started_at=10, completed_at=11), - }, - ) - - steering = threading.Thread(target=lambda: results.append(handle.steer("continue"))) - steering.start() - _route( - client, - "turn/started", - {"threadId": "thread-1", "turn": _turn("turn-2", "inProgress", started_at=12)}, - ) - steering.join(timeout=5) - - assert { - "alive": steering.is_alive(), - "steer_ids": control.steer_ids, - "public_turn_id": results[0].turn_id, - } == { - "alive": False, - "steer_ids": ["turn-2"], - "public_turn_id": "turn-1", - } - - -def test_goal_interrupt_succeeds_during_rollover() -> None: - client = CodexClient() - state = client.register_goal_operation("thread-1") - state.bind("turn-1") - control = _SyncControlClient() - handle = TurnHandle(control, "thread-1", "turn-1", _goal=state) # type: ignore[arg-type] - _route( - client, - "thread/goal/updated", - {"threadId": "thread-1", "turnId": "turn-1", "goal": _goal("thread-1", "active")}, - ) - _route( - client, - "turn/started", - {"threadId": "thread-1", "turn": _turn("turn-1", "inProgress", started_at=10)}, - ) - _route( - client, - "turn/completed", - { - "threadId": "thread-1", - "turn": _turn("turn-1", "completed", started_at=10, completed_at=11), - }, - ) - - response = handle.interrupt() - - assert { - "paused": control.paused, - "interrupt_ids": control.interrupt_ids, - "response": response.model_dump(mode="json"), - "interrupted": state.interrupted, - } == { - "paused": True, - "interrupt_ids": [], - "response": {}, - "interrupted": True, - } - - -class _AsyncGoalClient: - def __init__(self, state) -> None: - self.state = state - self.unregistered = False - - async def next_goal_notification(self, state): - return await asyncio.to_thread(state.next_notification) - - def unregister_goal_operation(self, _state) -> None: - self.unregistered = True - - -@dataclass(slots=True) -class _AsyncCodexStub: - _client: object - - async def _ensure_initialized(self) -> None: - return None - - -def test_async_goal_stream_matches_sync_logical_lifecycle() -> None: - async def scenario() -> None: - client = CodexClient() - state = client.register_goal_operation("thread-1") - state.bind("turn-1") - async_client = _AsyncGoalClient(state) - codex = _AsyncCodexStub(async_client) - handle = AsyncTurnHandle(codex, "thread-1", "turn-1", _goal=state) # type: ignore[arg-type] - - for method, params in [ - ( - "thread/goal/updated", - { - "threadId": "thread-1", - "turnId": "turn-1", - "goal": _goal("thread-1", "active"), - }, - ), - ( - "turn/started", - {"threadId": "thread-1", "turn": _turn("turn-1", "inProgress", started_at=10)}, - ), - ( - "thread/goal/updated", - { - "threadId": "thread-1", - "turnId": "turn-1", - "goal": _goal("thread-1", "complete"), - }, - ), - ( - "turn/completed", - { - "threadId": "thread-1", - "turn": _turn("turn-1", "completed", started_at=10, completed_at=11), - }, - ), - ]: - _route(client, method, params) - - events = [event async for event in handle.stream()] - - assert { - "methods": [event.method for event in events], - "ids": [notification_turn_id(event.payload) for event in events], - "unregistered": async_client.unregistered, - } == { - "methods": ["turn/started", "turn/completed"], - "ids": ["turn-1", "turn-1"], - "unregistered": True, - } - - asyncio.run(scenario()) - - -def test_cancelling_async_goal_stream_wakes_notification_reader() -> None: - async def scenario() -> None: - state = CodexClient().register_goal_operation("thread-1") - state.bind("turn-1") - async_client = _AsyncGoalClient(state) - codex = _AsyncCodexStub(async_client) - handle = AsyncTurnHandle(codex, "thread-1", "turn-1", _goal=state) # type: ignore[arg-type] - stream = handle.stream() - reading = asyncio.create_task(anext(stream)) - await asyncio.sleep(0) - - reading.cancel() - with pytest.raises(asyncio.CancelledError): - await reading - - assert {"finished": state.is_finished(), "unregistered": async_client.unregistered} == { - "finished": True, - "unregistered": True, - } - - asyncio.run(scenario()) - - -class _AsyncControlClient: - def __init__(self) -> None: - self.steer_ids: list[str] = [] - self.interrupt_ids: list[str] = [] - self.paused = False - - async def turn_steer(self, _thread_id, turn_id, _input): - self.steer_ids.append(turn_id) - if len(self.steer_ids) == 1: - raise InvalidRequestError( - -32600, - "expected active turn id `turn-1` but found `turn-2`", - ) - return TurnSteerResponse(turn_id=turn_id) - - async def pause_goal(self, _thread_id): - self.paused = True - - async def turn_interrupt(self, _thread_id, turn_id): - self.interrupt_ids.append(turn_id) - if len(self.interrupt_ids) == 1: - raise InvalidRequestError( - -32600, - "expected active turn id turn-1 but found turn-2", - ) - return TurnInterruptResponse() - - -def test_async_goal_controls_hide_rollover_turn_ids() -> None: - async def scenario() -> None: - state = CodexClient().register_goal_operation("thread-1") - state.bind("turn-1") - client = _AsyncControlClient() - codex = _AsyncCodexStub(client) # type: ignore[arg-type] - handle = AsyncTurnHandle(codex, "thread-1", "turn-1", _goal=state) # type: ignore[arg-type] - - steered = await handle.steer("keep going") - interrupted = await handle.interrupt() - - assert { - "steer_ids": client.steer_ids, - "public_steer_id": steered.turn_id, - "paused": client.paused, - "interrupt_ids": client.interrupt_ids, - "interrupt": interrupted.model_dump(mode="json"), - } == { - "steer_ids": ["turn-1", "turn-2"], - "public_steer_id": "turn-1", - "paused": True, - "interrupt_ids": ["turn-1", "turn-2"], - "interrupt": {}, - } - - asyncio.run(scenario()) - - -def test_finished_async_goal_controls_match_inactive_turn_errors() -> None: - async def scenario() -> None: - state = CodexClient().register_goal_operation("thread-1") - state.bind("turn-1") - state.finish() - client = _AsyncControlClient() - codex = _AsyncCodexStub(client) - handle = AsyncTurnHandle(codex, "thread-1", "turn-1", _goal=state) # type: ignore[arg-type] - - with pytest.raises(InvalidRequestError, match="no active turn to steer"): - await handle.steer("keep going") - with pytest.raises(InvalidRequestError, match="no active turn to interrupt"): - await handle.interrupt() - assert client.paused is False - - asyncio.run(scenario())