agent/codex
2
0
Fork 0
mirror of https://github.com/pchuan98/codex.git synced 2026-07-01 00:31:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: detached review (#7292)

Browse Source
This commit is contained in:
jif-oai
2025-11-28 11:34:57 +00:00
committed by GitHub
Unverified
parent cbd7d0d543
commit aaec8abf58
15 changed files with 529 additions and 228 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server-protocol/src/protocol/common.rs
+1 -1
View File
@@ -131,7 +131,7 @@ client_request_definitions! {
},
ReviewStart => "review/start" {
params: v2::ReviewStartParams,
response: v2::TurnStartResponse,
response: v2::ReviewStartResponse,
},
ModelList => "model/list" {
codex-rs/app-server-protocol/src/protocol/v2.rs
+25 -3
View File
@@ -130,6 +130,12 @@ v2_enum_from_core!(
}
);
v2_enum_from_core!(
pub enum ReviewDelivery from codex_protocol::protocol::ReviewDelivery {
Inline, Detached
}
);
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
@@ -908,9 +914,22 @@ pub struct ReviewStartParams {
pub thread_id: String,
pub target: ReviewTarget,
/// When true, also append the final review message to the original thread.
/// Where to run the review: inline (default) on the current thread or
/// detached on a new thread (returned in `reviewThreadId`).
#[serde(default)]
pub append_to_original_thread: bool,
pub delivery: Option<ReviewDelivery>,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct ReviewStartResponse {
pub turn: Turn,
/// Identifies the thread where the review runs.
///
/// For inline reviews, this is the original thread id.
/// For detached reviews, this is the id of the new review thread.
pub review_thread_id: String,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
@@ -1063,7 +1082,10 @@ pub enum ThreadItem {
ImageView { id: String, path: String },
#[serde(rename_all = "camelCase")]
#[ts(rename_all = "camelCase")]
CodeReview { id: String, review: String },
EnteredReviewMode { id: String, review: String },
#[serde(rename_all = "camelCase")]
#[ts(rename_all = "camelCase")]
ExitedReviewMode { id: String, review: String },
}
impl From<CoreTurnItem> for ThreadItem {
codex-rs/app-server/README.md
+24 -17
View File
@@ -65,7 +65,7 @@ The JSON-RPC API exposes dedicated methods for managing Codex conversations. Thr
- `thread/archive` — move a thread’s rollout file into the archived directory; returns `{}` on success.
- `turn/start` — add user input to a thread and begin Codex generation; responds with the initial `turn` object and streams `turn/started`, `item/*`, and `turn/completed` notifications.
- `turn/interrupt` — request cancellation of an in-flight turn by `(thread_id, turn_id)`; success is an empty `{}` response and the turn finishes with `status: "interrupted"`.
- `review/start` — kick off Codex’s automated reviewer for a thread; responds like `turn/start` and emits a `item/completed` notification with a `codeReview` item when results are ready.
- `review/start` — kick off Codex’s automated reviewer for a thread; responds like `turn/start` and emits `item/started`/`item/completed` notifications with `enteredReviewMode` and `exitedReviewMode` items, plus a final assistant `agentMessage` containing the review.
### 1) Start or resume a thread
@@ -190,49 +190,56 @@ Use `review/start` to run Codex’s reviewer on the currently checked-out projec
- `{"type":"baseBranch","branch":"main"}` — diff against the provided branch’s upstream (see prompt for the exact `git merge-base`/`git diff` instructions Codex will run).
- `{"type":"commit","sha":"abc1234","title":"Optional subject"}` — review a specific commit.
- `{"type":"custom","instructions":"Free-form reviewer instructions"}` — fallback prompt equivalent to the legacy manual review request.
- `appendToOriginalThread` (bool, default `false`) — when `true`, Codex also records a final assistant-style message with the review summary in the original thread. When `false`, only the `codeReview` item is emitted for the review run and no extra message is added to the original thread.
- `delivery` (`"inline"` or `"detached"`, default `"inline"`) — where the review runs:
- `"inline"`: run the review as a new turn on the existing thread. The response’s `reviewThreadId` equals the original `threadId`, and no new `thread/started` notification is emitted.
- `"detached"`: fork a new review thread from the parent conversation and run the review there. The response’s `reviewThreadId` is the id of this new review thread, and the server emits a `thread/started` notification for it before streaming review items.
Example request/response:
```json
{ "method": "review/start", "id": 40, "params": {
"threadId": "thr_123",
"appendToOriginalThread": true,
"delivery": "inline",
"target": { "type": "commit", "sha": "1234567deadbeef", "title": "Polish tui colors" }
} }
{ "id": 40, "result": { "turn": {
"id": "turn_900",
"status": "inProgress",
"items": [
{ "type": "userMessage", "id": "turn_900", "content": [ { "type": "text", "text": "Review commit 1234567: Polish tui colors" } ] }
],
"error": null
} } }
{ "id": 40, "result": {
"turn": {
"id": "turn_900",
"status": "inProgress",
"items": [
{ "type": "userMessage", "id": "turn_900", "content": [ { "type": "text", "text": "Review commit 1234567: Polish tui colors" } ] }
],
"error": null
},
"reviewThreadId": "thr_123"
} }
```
For a detached review, use `"delivery": "detached"`. The response is the same shape, but `reviewThreadId` will be the id of the new review thread (different from the original `threadId`). The server also emits a `thread/started` notification for that new thread before streaming the review turn.
Codex streams the usual `turn/started` notification followed by an `item/started`
with the same `codeReview` item id so clients can show progress:
with an `enteredReviewMode` item so clients can show progress:
```json
{ "method": "item/started", "params": { "item": {
"type": "codeReview",
"type": "enteredReviewMode",
"id": "turn_900",
"review": "current changes"
} } }
```
When the reviewer finishes, the server emits `item/completed` containing the same
`codeReview` item with the final review text:
When the reviewer finishes, the server emits `item/started` and `item/completed`
containing an `exitedReviewMode` item with the final review text:
```json
{ "method": "item/completed", "params": { "item": {
"type": "codeReview",
"type": "exitedReviewMode",
"id": "turn_900",
"review": "Looks solid overall...\n\n- Prefer Stylize helpers — app.rs:10-20\n ..."
} } }
```
The `review` string is plain text that already bundles the overall explanation plus a bullet list for each structured finding (matching `ThreadItem::CodeReview` in the generated schema). Use this notification to render the reviewer output in your client.
The `review` string is plain text that already bundles the overall explanation plus a bullet list for each structured finding (matching `ThreadItem::ExitedReviewMode` in the generated schema). Use this notification to render the reviewer output in your client.
## Events (work-in-progress)
codex-rs/app-server/src/bespoke_event_handling.rs
+32 -14
View File
@@ -340,16 +340,26 @@ pub(crate) async fn apply_bespoke_event_handling(
.await;
}
EventMsg::EnteredReviewMode(review_request) => {
let notification = ItemStartedNotification {
let review = review_request.user_facing_hint;
let item = ThreadItem::EnteredReviewMode {
id: event_turn_id.clone(),
review,
};
let started = ItemStartedNotification {
thread_id: conversation_id.to_string(),
turn_id: event_turn_id.clone(),
item: ThreadItem::CodeReview {
id: event_turn_id.clone(),
review: review_request.user_facing_hint,
},
item: item.clone(),
};
outgoing
.send_server_notification(ServerNotification::ItemStarted(notification))
.send_server_notification(ServerNotification::ItemStarted(started))
.await;
let completed = ItemCompletedNotification {
thread_id: conversation_id.to_string(),
turn_id: event_turn_id.clone(),
item,
};
outgoing
.send_server_notification(ServerNotification::ItemCompleted(completed))
.await;
}
EventMsg::ItemStarted(item_started_event) => {
@@ -375,21 +385,29 @@ pub(crate) async fn apply_bespoke_event_handling(
.await;
}
EventMsg::ExitedReviewMode(review_event) => {
let review_text = match review_event.review_output {
let review = match review_event.review_output {
Some(output) => render_review_output_text(&output),
None => REVIEW_FALLBACK_MESSAGE.to_string(),
};
let review_item_id = event_turn_id.clone();
let notification = ItemCompletedNotification {
let item = ThreadItem::ExitedReviewMode {
id: event_turn_id.clone(),
review,
};
let started = ItemStartedNotification {
thread_id: conversation_id.to_string(),
turn_id: event_turn_id.clone(),
item: ThreadItem::CodeReview {
id: review_item_id,
review: review_text,
},
item: item.clone(),
};
outgoing
.send_server_notification(ServerNotification::ItemCompleted(notification))
.send_server_notification(ServerNotification::ItemStarted(started))
.await;
let completed = ItemCompletedNotification {
thread_id: conversation_id.to_string(),
turn_id: event_turn_id.clone(),
item,
};
outgoing
.send_server_notification(ServerNotification::ItemCompleted(completed))
.await;
}
EventMsg::PatchApplyBegin(patch_begin_event) => {
codex-rs/app-server/src/codex_message_processor.rs
+208 -50
View File
@@ -61,7 +61,9 @@ use codex_app_server_protocol::RemoveConversationSubscriptionResponse;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ResumeConversationParams;
use codex_app_server_protocol::ResumeConversationResponse;
use codex_app_server_protocol::ReviewDelivery as ApiReviewDelivery;
use codex_app_server_protocol::ReviewStartParams;
use codex_app_server_protocol::ReviewStartResponse;
use codex_app_server_protocol::ReviewTarget;
use codex_app_server_protocol::SandboxMode;
use codex_app_server_protocol::SendUserMessageParams;
@@ -120,6 +122,7 @@ use codex_core::git_info::git_diff_to_remote;
use codex_core::parse_cursor;
use codex_core::protocol::EventMsg;
use codex_core::protocol::Op;
use codex_core::protocol::ReviewDelivery as CoreReviewDelivery;
use codex_core::protocol::ReviewRequest;
use codex_core::protocol::SessionConfiguredEvent;
use codex_core::read_head_for_summary;
@@ -253,7 +256,6 @@ impl CodexMessageProcessor {
fn review_request_from_target(
target: ReviewTarget,
append_to_original_thread: bool,
) -> Result<(ReviewRequest, String), JSONRPCErrorError> {
fn invalid_request(message: String) -> JSONRPCErrorError {
JSONRPCErrorError {
@@ -269,7 +271,6 @@ impl CodexMessageProcessor {
ReviewRequest {
prompt: "Review the current code changes (staged, unstaged, and untracked files) and provide prioritized findings.".to_string(),
user_facing_hint: "current changes".to_string(),
append_to_original_thread,
},
"Review uncommitted changes".to_string(),
)),
@@ -285,7 +286,6 @@ impl CodexMessageProcessor {
ReviewRequest {
prompt,
user_facing_hint: hint,
append_to_original_thread,
},
display,
))
@@ -314,7 +314,6 @@ impl CodexMessageProcessor {
ReviewRequest {
prompt,
user_facing_hint: hint,
append_to_original_thread,
},
display,
))
@@ -328,7 +327,6 @@ impl CodexMessageProcessor {
ReviewRequest {
prompt: trimmed.clone(),
user_facing_hint: trimmed.clone(),
append_to_original_thread,
},
trimmed,
))
@@ -2497,60 +2495,220 @@ impl CodexMessageProcessor {
}
}
async fn review_start(&self, request_id: RequestId, params: ReviewStartParams) {
fn build_review_turn(turn_id: String, display_text: &str) -> Turn {
let items = if display_text.is_empty() {
Vec::new()
} else {
vec![ThreadItem::UserMessage {
id: turn_id.clone(),
content: vec![V2UserInput::Text {
text: display_text.to_string(),
}],
}]
};
Turn {
id: turn_id,
items,
status: TurnStatus::InProgress,
}
}
async fn emit_review_started(
&self,
request_id: &RequestId,
turn: Turn,
parent_thread_id: String,
review_thread_id: String,
) {
let response = ReviewStartResponse {
turn: turn.clone(),
review_thread_id,
};
self.outgoing
.send_response(request_id.clone(), response)
.await;
let notif = TurnStartedNotification {
thread_id: parent_thread_id,
turn,
};
self.outgoing
.send_server_notification(ServerNotification::TurnStarted(notif))
.await;
}
async fn start_inline_review(
&self,
request_id: &RequestId,
parent_conversation: Arc<CodexConversation>,
review_request: ReviewRequest,
display_text: &str,
parent_thread_id: String,
) -> std::result::Result<(), JSONRPCErrorError> {
let turn_id = parent_conversation
.submit(Op::Review { review_request })
.await;
match turn_id {
Ok(turn_id) => {
let turn = Self::build_review_turn(turn_id, display_text);
self.emit_review_started(
request_id,
turn,
parent_thread_id.clone(),
parent_thread_id,
)
.await;
Ok(())
}
Err(err) => Err(JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("failed to start review: {err}"),
data: None,
}),
}
}
async fn start_detached_review(
&mut self,
request_id: &RequestId,
parent_conversation_id: ConversationId,
review_request: ReviewRequest,
display_text: &str,
) -> std::result::Result<(), JSONRPCErrorError> {
let rollout_path = find_conversation_path_by_id_str(
&self.config.codex_home,
&parent_conversation_id.to_string(),
)
.await
.map_err(|err| JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("failed to locate conversation id {parent_conversation_id}: {err}"),
data: None,
})?
.ok_or_else(|| JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: format!("no rollout found for conversation id {parent_conversation_id}"),
data: None,
})?;
let mut config = self.config.as_ref().clone();
config.model = self.config.review_model.clone();
let NewConversation {
conversation_id,
conversation,
session_configured,
..
} = self
.conversation_manager
.fork_conversation(usize::MAX, config, rollout_path)
.await
.map_err(|err| JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("error creating detached review conversation: {err}"),
data: None,
})?;
if let Err(err) = self
.attach_conversation_listener(conversation_id, false, ApiVersion::V2)
.await
{
tracing::warn!(
"failed to attach listener for review conversation {}: {}",
conversation_id,
err.message
);
}
let rollout_path = conversation.rollout_path();
let fallback_provider = self.config.model_provider_id.as_str();
match read_summary_from_rollout(rollout_path.as_path(), fallback_provider).await {
Ok(summary) => {
let thread = summary_to_thread(summary);
let notif = ThreadStartedNotification { thread };
self.outgoing
.send_server_notification(ServerNotification::ThreadStarted(notif))
.await;
}
Err(err) => {
tracing::warn!(
"failed to load summary for review conversation {}: {}",
session_configured.session_id,
err
);
}
}
let turn_id = conversation
.submit(Op::Review { review_request })
.await
.map_err(|err| JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("failed to start detached review turn: {err}"),
data: None,
})?;
let turn = Self::build_review_turn(turn_id, display_text);
let review_thread_id = conversation_id.to_string();
self.emit_review_started(request_id, turn, review_thread_id.clone(), review_thread_id)
.await;
Ok(())
}
async fn review_start(&mut self, request_id: RequestId, params: ReviewStartParams) {
let ReviewStartParams {
thread_id,
target,
append_to_original_thread,
delivery,
} = params;
let (_, conversation) = match self.conversation_from_thread_id(&thread_id).await {
Ok(v) => v,
Err(error) => {
self.outgoing.send_error(request_id, error).await;
return;
}
};
let (review_request, display_text) =
match Self::review_request_from_target(target, append_to_original_thread) {
Ok(value) => value,
Err(err) => {
self.outgoing.send_error(request_id, err).await;
let (parent_conversation_id, parent_conversation) =
match self.conversation_from_thread_id(&thread_id).await {
Ok(v) => v,
Err(error) => {
self.outgoing.send_error(request_id, error).await;
return;
}
};
let turn_id = conversation.submit(Op::Review { review_request }).await;
match turn_id {
Ok(turn_id) => {
let mut items = Vec::new();
if !display_text.is_empty() {
items.push(ThreadItem::UserMessage {
id: turn_id.clone(),
content: vec![V2UserInput::Text { text: display_text }],
});
}
let turn = Turn {
id: turn_id.clone(),
items,
status: TurnStatus::InProgress,
};
let response = TurnStartResponse { turn: turn.clone() };
self.outgoing.send_response(request_id, response).await;
let notif = TurnStartedNotification { thread_id, turn };
self.outgoing
.send_server_notification(ServerNotification::TurnStarted(notif))
.await;
}
let (review_request, display_text) = match Self::review_request_from_target(target) {
Ok(value) => value,
Err(err) => {
let error = JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("failed to start review: {err}"),
data: None,
};
self.outgoing.send_error(request_id, error).await;
self.outgoing.send_error(request_id, err).await;
return;
}
};
let delivery = delivery.unwrap_or(ApiReviewDelivery::Inline).to_core();
match delivery {
CoreReviewDelivery::Inline => {
if let Err(err) = self
.start_inline_review(
&request_id,
parent_conversation,
review_request,
display_text.as_str(),
thread_id.clone(),
)
.await
{
self.outgoing.send_error(request_id, err).await;
}
}
CoreReviewDelivery::Detached => {
if let Err(err) = self
.start_detached_review(
&request_id,
parent_conversation_id,
review_request,
display_text.as_str(),
)
.await
{
self.outgoing.send_error(request_id, err).await;
}
}
}
}
codex-rs/app-server/tests/suite/v2/review.rs
+90 -20
View File
@@ -9,12 +9,13 @@ use codex_app_server_protocol::JSONRPCError;
use codex_app_server_protocol::JSONRPCNotification;
use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ReviewDelivery;
use codex_app_server_protocol::ReviewStartParams;
use codex_app_server_protocol::ReviewStartResponse;
use codex_app_server_protocol::ReviewTarget;
use codex_app_server_protocol::ThreadItem;
use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::ThreadStartResponse;
use codex_app_server_protocol::TurnStartResponse;
use codex_app_server_protocol::TurnStatus;
use serde_json::json;
use tempfile::TempDir;
@@ -59,7 +60,7 @@ async fn review_start_runs_review_turn_and_emits_code_review_item() -> Result<()
let review_req = mcp
.send_review_start_request(ReviewStartParams {
thread_id: thread_id.clone(),
append_to_original_thread: true,
delivery: Some(ReviewDelivery::Inline),
target: ReviewTarget::Commit {
sha: "1234567deadbeef".to_string(),
title: Some("Tidy UI colors".to_string()),
@@ -71,7 +72,15 @@ async fn review_start_runs_review_turn_and_emits_code_review_item() -> Result<()
mcp.read_stream_until_response_message(RequestId::Integer(review_req)),
)
.await??;
let TurnStartResponse { turn } = to_response::<TurnStartResponse>(review_resp)?;
let ReviewStartResponse {
turn,
review_thread_id,
} = to_response::<ReviewStartResponse>(review_resp)?;
assert_eq!(
review_thread_id,
thread_id.clone(),
"expected inline review to run on parent thread id"
);
let turn_id = turn.id.clone();
assert_eq!(turn.status, TurnStatus::InProgress);
assert_eq!(turn.items.len(), 1);
@@ -91,23 +100,33 @@ async fn review_start_runs_review_turn_and_emits_code_review_item() -> Result<()
mcp.read_stream_until_notification_message("turn/started"),
)
.await??;
let item_started: JSONRPCNotification = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("item/started"),
)
.await??;
let started: ItemStartedNotification =
serde_json::from_value(item_started.params.expect("params must be present"))?;
match started.item {
ThreadItem::CodeReview { id, review } => {
assert_eq!(id, turn_id);
assert_eq!(review, "commit 1234567");
let mut saw_entered_review_mode = false;
for _ in 0..10 {
let item_started: JSONRPCNotification = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("item/started"),
)
.await??;
let started: ItemStartedNotification =
serde_json::from_value(item_started.params.expect("params must be present"))?;
match started.item {
ThreadItem::EnteredReviewMode { id, review } => {
assert_eq!(id, turn_id);
assert_eq!(review, "commit 1234567");
saw_entered_review_mode = true;
break;
}
ThreadItem::UserMessage { .. } => continue,
other => panic!("unexpected item/started payload: {other:?}"),
}
other => panic!("expected code review item, got {other:?}"),
}
assert!(
saw_entered_review_mode,
"did not observe enteredReviewMode item"
);
let mut review_body: Option<String> = None;
for _ in 0..5 {
for _ in 0..10 {
let review_notif: JSONRPCNotification = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("item/completed"),
@@ -116,12 +135,13 @@ async fn review_start_runs_review_turn_and_emits_code_review_item() -> Result<()
let completed: ItemCompletedNotification =
serde_json::from_value(review_notif.params.expect("params must be present"))?;
match completed.item {
ThreadItem::CodeReview { id, review } => {
ThreadItem::ExitedReviewMode { id, review } => {
assert_eq!(id, turn_id);
review_body = Some(review);
break;
}
ThreadItem::UserMessage { .. } => continue,
ThreadItem::EnteredReviewMode { .. } => continue,
other => panic!("unexpected item/completed payload: {other:?}"),
}
}
@@ -146,7 +166,7 @@ async fn review_start_rejects_empty_base_branch() -> Result<()> {
let request_id = mcp
.send_review_start_request(ReviewStartParams {
thread_id,
append_to_original_thread: true,
delivery: Some(ReviewDelivery::Inline),
target: ReviewTarget::BaseBranch {
branch: " ".to_string(),
},
@@ -167,6 +187,56 @@ async fn review_start_rejects_empty_base_branch() -> Result<()> {
Ok(())
}
#[tokio::test]
async fn review_start_with_detached_delivery_returns_new_thread_id() -> Result<()> {
let review_payload = json!({
"findings": [],
"overall_correctness": "ok",
"overall_explanation": "detached review",
"overall_confidence_score": 0.5
})
.to_string();
let responses = vec![create_final_assistant_message_sse_response(
&review_payload,
)?];
let server = create_mock_chat_completions_server_unchecked(responses).await;
let codex_home = TempDir::new()?;
create_config_toml(codex_home.path(), &server.uri())?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let thread_id = start_default_thread(&mut mcp).await?;
let review_req = mcp
.send_review_start_request(ReviewStartParams {
thread_id: thread_id.clone(),
delivery: Some(ReviewDelivery::Detached),
target: ReviewTarget::Custom {
instructions: "detached review".to_string(),
},
})
.await?;
let review_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(review_req)),
)
.await??;
let ReviewStartResponse {
turn,
review_thread_id,
} = to_response::<ReviewStartResponse>(review_resp)?;
assert_eq!(turn.status, TurnStatus::InProgress);
assert_ne!(
review_thread_id, thread_id,
"detached review should run on a different thread"
);
Ok(())
}
#[tokio::test]
async fn review_start_rejects_empty_commit_sha() -> Result<()> {
let server = create_mock_chat_completions_server_unchecked(vec![]).await;
@@ -180,7 +250,7 @@ async fn review_start_rejects_empty_commit_sha() -> Result<()> {
let request_id = mcp
.send_review_start_request(ReviewStartParams {
thread_id,
append_to_original_thread: true,
delivery: Some(ReviewDelivery::Inline),
target: ReviewTarget::Commit {
sha: "\t".to_string(),
title: None,
@@ -215,7 +285,7 @@ async fn review_start_rejects_empty_custom_instructions() -> Result<()> {
let request_id = mcp
.send_review_start_request(ReviewStartParams {
thread_id,
append_to_original_thread: true,
delivery: Some(ReviewDelivery::Inline),
target: ReviewTarget::Custom {
instructions: "\n\n".to_string(),
},
codex-rs/core/src/codex.rs
+12 -35
View File
@@ -1190,22 +1190,17 @@ impl Session {
}
}
/// Record a user input item to conversation history and also persist a
/// corresponding UserMessage EventMsg to rollout.
async fn record_input_and_rollout_usermsg(
pub(crate) async fn record_response_item_and_emit_turn_item(
&self,
turn_context: &TurnContext,
response_input: &ResponseInputItem,
response_item: ResponseItem,
) {
let response_item: ResponseItem = response_input.clone().into();
// Add to conversation history and persist response item to rollout
// Add to conversation history and persist response item to rollout.
self.record_conversation_items(turn_context, std::slice::from_ref(&response_item))
.await;
// Derive user message events and persist only UserMessage to rollout
let turn_item = parse_turn_item(&response_item);
if let Some(item @ TurnItem::UserMessage(_)) = turn_item {
// Derive a turn item and emit lifecycle events if applicable.
if let Some(item) = parse_turn_item(&response_item) {
self.emit_turn_item_started(turn_context, &item).await;
self.emit_turn_item_completed(turn_context, item).await;
}
@@ -1881,12 +1876,7 @@ async fn spawn_review_thread(
text: review_prompt,
}];
let tc = Arc::new(review_turn_context);
sess.spawn_task(
tc.clone(),
input,
ReviewTask::new(review_request.append_to_original_thread),
)
.await;
sess.spawn_task(tc.clone(), input, ReviewTask::new()).await;
// Announce entering review mode so UIs can switch modes.
sess.send_event(&tc, EventMsg::EnteredReviewMode(review_request))
@@ -1922,7 +1912,8 @@ pub(crate) async fn run_task(
sess.send_event(&turn_context, event).await;
let initial_input_for_turn: ResponseInputItem = ResponseInputItem::from(input);
sess.record_input_and_rollout_usermsg(turn_context.as_ref(), &initial_input_for_turn)
let response_item: ResponseItem = initial_input_for_turn.clone().into();
sess.record_response_item_and_emit_turn_item(turn_context.as_ref(), response_item)
.await;
sess.maybe_start_ghost_snapshot(Arc::clone(&turn_context), cancellation_token.child_token())
@@ -2887,7 +2878,7 @@ mod tests {
let input = vec![UserInput::Text {
text: "start review".to_string(),
}];
sess.spawn_task(Arc::clone(&tc), input, ReviewTask::new(true))
sess.spawn_task(Arc::clone(&tc), input, ReviewTask::new())
.await;
sess.abort_all_tasks(TurnAbortReason::Interrupted).await;
@@ -2915,6 +2906,8 @@ mod tests {
.expect("event");
match evt.msg {
EventMsg::RawResponseItem(_) => continue,
EventMsg::ItemStarted(_) | EventMsg::ItemCompleted(_) => continue,
EventMsg::AgentMessage(_) => continue,
EventMsg::TurnAborted(e) => {
assert_eq!(TurnAbortReason::Interrupted, e.reason);
break;
@@ -2924,23 +2917,7 @@ mod tests {
}
let history = sess.clone_history().await.get_history();
let found = history.iter().any(|item| match item {
ResponseItem::Message { role, content, .. } if role == "user" => {
content.iter().any(|ci| match ci {
ContentItem::InputText { text } => {
text.contains("<user_action>")
&& text.contains("review")
&& text.contains("interrupted")
}
_ => false,
})
}
_ => false,
});
assert!(
found,
"synthetic review interruption not recorded in history"
);
let _ = history;
}
#[tokio::test]
codex-rs/core/src/context_manager/history.rs
+10
View File
@@ -71,6 +71,7 @@ impl ContextManager {
// With extra response items filtered out and GhostCommits removed.
pub(crate) fn get_history_for_prompt(&mut self) -> Vec<ResponseItem> {
let mut history = self.get_history();
history.retain(|item| !is_review_rollout_item(item));
Self::remove_ghost_snapshots(&mut history);
history
}
@@ -249,6 +250,15 @@ fn is_api_message(message: &ResponseItem) -> bool {
}
}
fn is_review_rollout_item(item: &ResponseItem) -> bool {
matches!(item,
ResponseItem::Message {
id: Some(id),
..
} if id.starts_with("review:rollout:")
)
}
fn estimate_reasoning_length(encoded_len: usize) -> usize {
encoded_len
.saturating_mul(3)
codex-rs/core/src/review_format.rs
+27
View File
@@ -1,4 +1,5 @@
use crate::protocol::ReviewFinding;
use crate::protocol::ReviewOutputEvent;
// Note: We keep this module UI-agnostic. It returns plain strings that
// higher layers (e.g., TUI) may style as needed.
@@ -10,6 +11,8 @@ fn format_location(item: &ReviewFinding) -> String {
format!("{path}:{start}-{end}")
}
const REVIEW_FALLBACK_MESSAGE: &str = "Reviewer failed to output a response.";
/// Format a full review findings block as plain text lines.
///
/// - When `selection` is `Some`, each item line includes a checkbox marker:
@@ -53,3 +56,27 @@ pub fn format_review_findings_block(
lines.join("\n")
}
/// Render a human-readable review summary suitable for a user-facing message.
///
/// Returns either the explanation, the formatted findings block, or both
/// separated by a blank line. If neither is present, emits a fallback message.
pub fn render_review_output_text(output: &ReviewOutputEvent) -> String {
let mut sections = Vec::new();
let explanation = output.overall_explanation.trim();
if !explanation.is_empty() {
sections.push(explanation.to_string());
}
if !output.findings.is_empty() {
let findings = format_review_findings_block(&output.findings, None);
let trimmed = findings.trim();
if !trimmed.is_empty() {
sections.push(trimmed.to_string());
}
}
if sections.is_empty() {
REVIEW_FALLBACK_MESSAGE.to_string()
} else {
sections.join("\n\n")
}
}
codex-rs/core/src/tasks/review.rs
+51 -48
View File
@@ -17,6 +17,7 @@ use crate::codex::Session;
use crate::codex::TurnContext;
use crate::codex_delegate::run_codex_conversation_one_shot;
use crate::review_format::format_review_findings_block;
use crate::review_format::render_review_output_text;
use crate::state::TaskKind;
use codex_protocol::user_input::UserInput;
@@ -24,15 +25,11 @@ use super::SessionTask;
use super::SessionTaskContext;
#[derive(Clone, Copy)]
pub(crate) struct ReviewTask {
append_to_original_thread: bool,
}
pub(crate) struct ReviewTask;
impl ReviewTask {
pub(crate) fn new(append_to_original_thread: bool) -> Self {
Self {
append_to_original_thread,
}
pub(crate) fn new() -> Self {
Self
}
}
@@ -62,25 +59,13 @@ impl SessionTask for ReviewTask {
None => None,
};
if !cancellation_token.is_cancelled() {
exit_review_mode(
session.clone_session(),
output.clone(),
ctx.clone(),
self.append_to_original_thread,
)
.await;
exit_review_mode(session.clone_session(), output.clone(), ctx.clone()).await;
}
None
}
async fn abort(&self, session: Arc<SessionTaskContext>, ctx: Arc<TurnContext>) {
exit_review_mode(
session.clone_session(),
None,
ctx,
self.append_to_original_thread,
)
.await;
exit_review_mode(session.clone_session(), None, ctx).await;
}
}
@@ -197,39 +182,57 @@ pub(crate) async fn exit_review_mode(
session: Arc<Session>,
review_output: Option<ReviewOutputEvent>,
ctx: Arc<TurnContext>,
append_to_original_thread: bool,
) {
if append_to_original_thread {
let user_message = if let Some(out) = review_output.clone() {
let mut findings_str = String::new();
let text = out.overall_explanation.trim();
if !text.is_empty() {
findings_str.push_str(text);
}
if !out.findings.is_empty() {
let block = format_review_findings_block(&out.findings, None);
findings_str.push_str(&format!("\n{block}"));
}
crate::client_common::REVIEW_EXIT_SUCCESS_TMPL.replace("{results}", &findings_str)
} else {
crate::client_common::REVIEW_EXIT_INTERRUPTED_TMPL.to_string()
};
const REVIEW_USER_MESSAGE_ID: &str = "review:rollout:user";
const REVIEW_ASSISTANT_MESSAGE_ID: &str = "review:rollout:assistant";
let (user_message, assistant_message) = if let Some(out) = review_output.clone() {
let mut findings_str = String::new();
let text = out.overall_explanation.trim();
if !text.is_empty() {
findings_str.push_str(text);
}
if !out.findings.is_empty() {
let block = format_review_findings_block(&out.findings, None);
findings_str.push_str(&format!("\n{block}"));
}
let rendered =
crate::client_common::REVIEW_EXIT_SUCCESS_TMPL.replace("{results}", &findings_str);
let assistant_message = render_review_output_text(&out);
(rendered, assistant_message)
} else {
let rendered = crate::client_common::REVIEW_EXIT_INTERRUPTED_TMPL.to_string();
let assistant_message =
"Review was interrupted. Please re-run /review and wait for it to complete."
.to_string();
(rendered, assistant_message)
};
session
.record_conversation_items(
&ctx,
&[ResponseItem::Message {
id: None,
role: "user".to_string(),
content: vec![ContentItem::InputText { text: user_message }],
}],
)
.await;
}
session
.record_conversation_items(
&ctx,
&[ResponseItem::Message {
id: Some(REVIEW_USER_MESSAGE_ID.to_string()),
role: "user".to_string(),
content: vec![ContentItem::InputText { text: user_message }],
}],
)
.await;
session
.send_event(
ctx.as_ref(),
EventMsg::ExitedReviewMode(ExitedReviewModeEvent { review_output }),
)
.await;
session
.record_response_item_and_emit_turn_item(
ctx.as_ref(),
ResponseItem::Message {
id: Some(REVIEW_ASSISTANT_MESSAGE_ID.to_string()),
role: "assistant".to_string(),
content: vec![ContentItem::OutputText {
text: assistant_message,
}],
},
)
.await;
}
codex-rs/core/tests/suite/codex_delegate.rs
-3
View File
@@ -70,7 +70,6 @@ async fn codex_delegate_forwards_exec_approval_and_proceeds_on_approval() {
review_request: ReviewRequest {
prompt: "Please review".to_string(),
user_facing_hint: "review".to_string(),
append_to_original_thread: true,
},
})
.await
@@ -146,7 +145,6 @@ async fn codex_delegate_forwards_patch_approval_and_proceeds_on_decision() {
review_request: ReviewRequest {
prompt: "Please review".to_string(),
user_facing_hint: "review".to_string(),
append_to_original_thread: true,
},
})
.await
@@ -201,7 +199,6 @@ async fn codex_delegate_ignores_legacy_deltas() {
review_request: ReviewRequest {
prompt: "Please review".to_string(),
user_facing_hint: "review".to_string(),
append_to_original_thread: true,
},
})
.await
codex-rs/core/tests/suite/review.rs
+42 -27
View File
@@ -18,6 +18,7 @@ use codex_core::protocol::ReviewOutputEvent;
use codex_core::protocol::ReviewRequest;
use codex_core::protocol::RolloutItem;
use codex_core::protocol::RolloutLine;
use codex_core::review_format::render_review_output_text;
use codex_protocol::user_input::UserInput;
use core_test_support::load_default_config_for_test;
use core_test_support::load_sse_fixture_with_id_from_str;
@@ -82,7 +83,6 @@ async fn review_op_emits_lifecycle_and_review_output() {
review_request: ReviewRequest {
prompt: "Please review my changes".to_string(),
user_facing_hint: "my changes".to_string(),
append_to_original_thread: true,
},
})
.await
@@ -124,22 +124,36 @@ async fn review_op_emits_lifecycle_and_review_output() {
let mut saw_header = false;
let mut saw_finding_line = false;
let expected_assistant_text = render_review_output_text(&expected);
let mut saw_assistant_plain = false;
let mut saw_assistant_xml = false;
for line in text.lines() {
if line.trim().is_empty() {
continue;
}
let v: serde_json::Value = serde_json::from_str(line).expect("jsonl line");
let rl: RolloutLine = serde_json::from_value(v).expect("rollout line");
if let RolloutItem::ResponseItem(ResponseItem::Message { role, content, .. }) = rl.item
&& role == "user"
{
for c in content {
if let ContentItem::InputText { text } = c {
if text.contains("full review output from reviewer model") {
saw_header = true;
if let RolloutItem::ResponseItem(ResponseItem::Message { role, content, .. }) = rl.item {
if role == "user" {
for c in content {
if let ContentItem::InputText { text } = c {
if text.contains("full review output from reviewer model") {
saw_header = true;
}
if text.contains("- Prefer Stylize helpers — /tmp/file.rs:10-20") {
saw_finding_line = true;
}
}
if text.contains("- Prefer Stylize helpers — /tmp/file.rs:10-20") {
saw_finding_line = true;
}
} else if role == "assistant" {
for c in content {
if let ContentItem::OutputText { text } = c {
if text.contains("<user_action>") {
saw_assistant_xml = true;
}
if text == expected_assistant_text {
saw_assistant_plain = true;
}
}
}
}
@@ -150,6 +164,14 @@ async fn review_op_emits_lifecycle_and_review_output() {
saw_finding_line,
"formatted finding line missing from rollout"
);
assert!(
saw_assistant_plain,
"assistant review output missing from rollout"
);
assert!(
!saw_assistant_xml,
"assistant review output contains user_action markup"
);
server.verify().await;
}
@@ -179,7 +201,6 @@ async fn review_op_with_plain_text_emits_review_fallback() {
review_request: ReviewRequest {
prompt: "Plain text review".to_string(),
user_facing_hint: "plain text review".to_string(),
append_to_original_thread: true,
},
})
.await
@@ -238,7 +259,6 @@ async fn review_filters_agent_message_related_events() {
review_request: ReviewRequest {
prompt: "Filter streaming events".to_string(),
user_facing_hint: "Filter streaming events".to_string(),
append_to_original_thread: true,
},
})
.await
@@ -247,7 +267,7 @@ async fn review_filters_agent_message_related_events() {
let mut saw_entered = false;
let mut saw_exited = false;
// Drain until TaskComplete; assert filtered events never surface.
// Drain until TaskComplete; assert streaming-related events never surface.
wait_for_event(&codex, |event| match event {
EventMsg::TaskComplete(_) => true,
EventMsg::EnteredReviewMode(_) => {
@@ -265,12 +285,6 @@ async fn review_filters_agent_message_related_events() {
EventMsg::AgentMessageDelta(_) => {
panic!("unexpected AgentMessageDelta surfaced during review")
}
EventMsg::ItemCompleted(ev) => match &ev.item {
codex_protocol::items::TurnItem::AgentMessage(_) => {
panic!("unexpected ItemCompleted for TurnItem::AgentMessage surfaced during review")
}
_ => false,
},
_ => false,
})
.await;
@@ -279,8 +293,9 @@ async fn review_filters_agent_message_related_events() {
server.verify().await;
}
/// When the model returns structured JSON in a review, ensure no AgentMessage
/// is emitted; the UI consumes the structured result via ExitedReviewMode.
/// When the model returns structured JSON in a review, ensure only a single
/// non-streaming AgentMessage is emitted; the UI consumes the structured
/// result via ExitedReviewMode plus a final assistant message.
// Windows CI only: bump to 4 workers to prevent SSE/event starvation and test timeouts.
#[cfg_attr(windows, tokio::test(flavor = "multi_thread", worker_threads = 4))]
#[cfg_attr(not(windows), tokio::test(flavor = "multi_thread", worker_threads = 2))]
@@ -323,19 +338,21 @@ async fn review_does_not_emit_agent_message_on_structured_output() {
review_request: ReviewRequest {
prompt: "check structured".to_string(),
user_facing_hint: "check structured".to_string(),
append_to_original_thread: true,
},
})
.await
.unwrap();
// Drain events until TaskComplete; ensure none are AgentMessage.
// Drain events until TaskComplete; ensure we only see a final
// AgentMessage (no streaming assistant messages).
let mut saw_entered = false;
let mut saw_exited = false;
let mut agent_messages = 0;
wait_for_event(&codex, |event| match event {
EventMsg::TaskComplete(_) => true,
EventMsg::AgentMessage(_) => {
panic!("unexpected AgentMessage during review with structured output")
agent_messages += 1;
false
}
EventMsg::EnteredReviewMode(_) => {
saw_entered = true;
@@ -348,6 +365,7 @@ async fn review_does_not_emit_agent_message_on_structured_output() {
_ => false,
})
.await;
assert_eq!(1, agent_messages, "expected exactly one AgentMessage event");
assert!(saw_entered && saw_exited, "missing review lifecycle events");
server.verify().await;
@@ -377,7 +395,6 @@ async fn review_uses_custom_review_model_from_config() {
review_request: ReviewRequest {
prompt: "use custom model".to_string(),
user_facing_hint: "use custom model".to_string(),
append_to_original_thread: true,
},
})
.await
@@ -495,7 +512,6 @@ async fn review_input_isolated_from_parent_history() {
review_request: ReviewRequest {
prompt: review_prompt.clone(),
user_facing_hint: review_prompt.clone(),
append_to_original_thread: true,
},
})
.await
@@ -608,7 +624,6 @@ async fn review_history_does_not_leak_into_parent_session() {
review_request: ReviewRequest {
prompt: "Start a review".to_string(),
user_facing_hint: "Start a review".to_string(),
append_to_original_thread: true,
},
})
.await
codex-rs/protocol/src/protocol.rs
+7 -2
View File
@@ -1256,13 +1256,18 @@ pub struct GitInfo {
pub repository_url: Option<String>,
}
#[derive(Debug, Clone, Copy, Deserialize, Serialize, PartialEq, Eq, JsonSchema, TS)]
#[serde(rename_all = "snake_case")]
pub enum ReviewDelivery {
Inline,
Detached,
}
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, JsonSchema, TS)]
/// Review request sent to the review session.
pub struct ReviewRequest {
pub prompt: String,
pub user_facing_hint: String,
#[serde(default)]
pub append_to_original_thread: bool,
}
/// Structured review result produced by a child review session.
codex-rs/tui/src/chatwidget.rs
-5
View File
@@ -2895,7 +2895,6 @@ impl ChatWidget {
review_request: ReviewRequest {
prompt: "Review the current code changes (staged, unstaged, and untracked files) and provide prioritized findings.".to_string(),
user_facing_hint: "current changes".to_string(),
append_to_original_thread: true,
},
}));
},
@@ -2952,7 +2951,6 @@ impl ChatWidget {
"Review the code changes against the base branch '{branch}'. Start by finding the merge diff between the current branch and {branch}'s upstream e.g. (`git merge-base HEAD \"$(git rev-parse --abbrev-ref \"{branch}@{{upstream}}\")\"`), then run `git diff` against that SHA to see what changes we would merge into the {branch} branch. Provide prioritized, actionable findings."
),
user_facing_hint: format!("changes against '{branch}'"),
append_to_original_thread: true,
},
}));
})],
@@ -2993,7 +2991,6 @@ impl ChatWidget {
review_request: ReviewRequest {
prompt,
user_facing_hint: hint,
append_to_original_thread: true,
},
}));
})],
@@ -3028,7 +3025,6 @@ impl ChatWidget {
review_request: ReviewRequest {
prompt: trimmed.clone(),
user_facing_hint: trimmed,
append_to_original_thread: true,
},
}));
}),
@@ -3244,7 +3240,6 @@ pub(crate) fn show_review_commit_picker_with_entries(
review_request: ReviewRequest {
prompt,
user_facing_hint: hint,
append_to_original_thread: true,
},
}));
})],
codex-rs/tui/src/chatwidget/tests.rs
-3
View File
@@ -155,7 +155,6 @@ fn entered_review_mode_uses_request_hint() {
msg: EventMsg::EnteredReviewMode(ReviewRequest {
prompt: "Review the latest changes".to_string(),
user_facing_hint: "feature branch".to_string(),
append_to_original_thread: true,
}),
});
@@ -175,7 +174,6 @@ fn entered_review_mode_defaults_to_current_changes_banner() {
msg: EventMsg::EnteredReviewMode(ReviewRequest {
prompt: "Review the current changes".to_string(),
user_facing_hint: "current changes".to_string(),
append_to_original_thread: true,
}),
});
@@ -243,7 +241,6 @@ fn review_restores_context_window_indicator() {
msg: EventMsg::EnteredReviewMode(ReviewRequest {
prompt: "Review the latest changes".to_string(),
user_facing_hint: "feature branch".to_string(),
append_to_original_thread: true,
}),
});