File size: 2,988 Bytes
f15fb1d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 | from __future__ import annotations
from app_server_harness import AppServerHarness
from app_server_helpers import agent_message_texts, streaming_response
from openai_codex import Codex
from openai_codex.generated.v2_all import TurnStatus
def test_turn_steer_adds_follow_up_input(tmp_path) -> None:
"""Steering an active turn should create a follow-up Responses request."""
with AppServerHarness(tmp_path) as harness:
harness.responses.enqueue_sse(
streaming_response("steer-first", "msg-steer-first", ["before steer"]),
delay_between_events_s=0.2,
)
harness.responses.enqueue_assistant_message(
"after steer",
response_id="steer-second",
)
with Codex(config=harness.app_server_config()) as codex:
thread = codex.thread_start()
turn = thread.turn("Start a steerable turn.")
harness.responses.wait_for_requests(1)
steer = turn.steer("Use this steering input.")
events = list(turn.stream())
requests = harness.responses.wait_for_requests(2)
assert {
"steered_turn_id": steer.turn_id,
"turn_id": turn.id,
"agent_messages": agent_message_texts(events),
"last_user_texts": [request.message_input_texts("user")[-1] for request in requests],
} == {
"steered_turn_id": turn.id,
"turn_id": turn.id,
"agent_messages": ["before steer", "after steer"],
"last_user_texts": [
"Start a steerable turn.",
"Use this steering input.",
],
}
def test_turn_interrupt_stops_active_turn_and_follow_up_runs(tmp_path) -> None:
"""Interrupting an active turn should complete it and leave the thread usable."""
with AppServerHarness(tmp_path) as harness:
harness.responses.enqueue_sse(
streaming_response(
"interrupt-first",
"msg-interrupt-first",
["still ", "running"],
),
delay_between_events_s=0.2,
)
harness.responses.enqueue_assistant_message(
"after interrupt",
response_id="interrupt-follow-up",
)
with Codex(config=harness.app_server_config()) as codex:
thread = codex.thread_start()
interrupted_turn = thread.turn("Start a long turn.")
harness.responses.wait_for_requests(1)
interrupt_response = interrupted_turn.interrupt()
completed = interrupted_turn.run()
follow_up = thread.run("Continue after the interrupt.")
assert {
"interrupt_response": interrupt_response.model_dump(
by_alias=True,
mode="json",
),
"interrupted_status": completed.status,
"follow_up": follow_up.final_response,
} == {
"interrupt_response": {},
"interrupted_status": TurnStatus.interrupted,
"follow_up": "after interrupt",
}
|