Run detached reviews as review-agent turns (#33156)

## Why

Detached reviews should behave like ordinary forked turns so clients get the normal steering, tool, permission, and item-stream behavior.

## What changed

- Add a bundled `$review-agent` skill with read-only, defect-first review guidance.
- Start detached reviews through `AgentRunner` with a target-specific prompt that explicitly references the bundled skill. Keep inline reviews on the existing review-mode flow.
- Continue emitting the detached review thread and turn identifiers through the app-server API.

## Testing

- Update the detached review integration test to verify the forked turn prompt, ordinary turn completion, and selection of the bundled skill when a user skill has the same name.

GitOrigin-RevId: 979664cd80800fb0d51eabdc4e3211da90c12db9
This commit is contained in:
jif
2026-07-14 20:56:43 +00:00
committed by copyberry
parent 2012d87211
commit 83a4187837
8 changed files with 214 additions and 101 deletions
+2
View File
@@ -1987,6 +1987,7 @@ dependencies = [
"base64 0.22.1",
"chrono",
"clap",
"codex-agent-extension",
"codex-analytics",
"codex-app-server-protocol",
"codex-app-server-transport",
@@ -2027,6 +2028,7 @@ dependencies = [
"codex-rollout",
"codex-sandboxing",
"codex-shell-command",
"codex-skills",
"codex-skills-extension",
"codex-state",
"codex-thread-store",
+1
View File
@@ -170,6 +170,7 @@ codex-http-client = { path = "http-client" }
codex-websocket-client = { path = "websocket-client" }
codex-config = { path = "config" }
codex-connectors = { path = "connectors" }
codex-agent-extension = { path = "ext/agent" }
codex-connectors-extension = { path = "ext/connectors" }
codex-context-fragments = { path = "context-fragments" }
codex-core = { path = "core" }
+2
View File
@@ -34,6 +34,7 @@ axum = { workspace = true, default-features = false, features = [
"ws",
] }
codex-analytics = { workspace = true }
codex-agent-extension = { workspace = true }
codex-arg0 = { workspace = true }
codex-cloud-config = { workspace = true }
codex-config = { workspace = true }
@@ -54,6 +55,7 @@ codex-http-client = { workspace = true }
codex-otel = { workspace = true }
codex-plugin = { workspace = true }
codex-shell-command = { workspace = true }
codex-skills = { workspace = true }
codex-skills-extension = { workspace = true }
codex-utils-cli = { workspace = true }
codex-utils-pty = { workspace = true }
+3 -3
View File
@@ -177,7 +177,7 @@ Example with notification opt-out:
- `thread/realtime/appendText` — append text input to the active realtime session with a required `role` of `user`, `developer`, or `assistant` (experimental); returns `{}`. Older clients that omit `role` default to `user`.
- `thread/realtime/appendSpeech` — append text that the realtime model should speak to the user (experimental); returns `{}`.
- `thread/realtime/stop` — stop the active realtime session for the thread (experimental); returns `{}`.
- `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.
- `review/start` — kick off Codex’s automated reviewer for a thread; responds like `turn/start`. Inline reviews emit `item/started`/`item/completed` notifications with `enteredReviewMode` and `exitedReviewMode` items, plus a final assistant `agentMessage` containing the review. Detached reviews stream ordinary turn items on the new review thread.
- `command/exec` — run a single command under the server sandbox without starting a thread/turn (handy for utilities and validation).
- `command/exec/write` — write base64-decoded stdin bytes to a running `command/exec` session or close stdin; returns `{}`.
- `command/exec/resize` — resize a running PTY-backed `command/exec` session by `processId`; returns `{}`.
@@ -1029,9 +1029,9 @@ Example request/response:
} }
```
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.
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. Internally, this is a normal forked thread and turn whose prompt mentions the bundled `$review-agent` skill, so normal turn steering, tool, permission, and item-stream behavior applies.
Codex streams the usual `turn/started` notification followed by an `item/started`
For an inline review, Codex streams the usual `turn/started` notification followed by an `item/started`
with an `enteredReviewMode` item so clients can show progress:
```json
@@ -1,4 +1,7 @@
use super::*;
use codex_agent_extension::AgentInvocation;
use codex_agent_extension::AgentRun;
use codex_agent_extension::AgentRunner;
use codex_protocol::models::ContentItem;
use codex_protocol::models::FunctionCallOutputContentItem;
use codex_protocol::models::PermissionProfile;
@@ -7,6 +10,7 @@ use codex_protocol::protocol::AdditionalContextKind as CoreAdditionalContextKind
use codex_protocol::protocol::MultiAgentVersion;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
use codex_skills::system_cache_root_dir;
use crate::image_url::REMOTE_IMAGE_URL_ERROR;
use crate::image_url::is_remote_image_url;
@@ -68,6 +72,7 @@ fn validate_response_item_image_urls(items: &[ResponseItem]) -> Result<(), JSONR
#[derive(Clone)]
pub(crate) struct TurnRequestProcessor {
agent_runner: AgentRunner,
auth_manager: Arc<AuthManager>,
thread_manager: Arc<ThreadManager>,
outgoing: Arc<OutgoingMessageSender>,
@@ -136,7 +141,9 @@ impl TurnRequestProcessor {
thread_list_state_permit: Arc<Semaphore>,
skills_watcher: Arc<SkillsWatcher>,
) -> Self {
let agent_runner = AgentRunner::new(Arc::downgrade(&thread_manager));
Self {
agent_runner,
auth_manager,
thread_manager,
outgoing,
@@ -352,7 +359,7 @@ impl TurnRequestProcessor {
fn review_request_from_target(
target: ApiReviewTarget,
) -> Result<(ReviewRequest, String), JSONRPCErrorError> {
) -> Result<(ReviewRequest, String, String), JSONRPCErrorError> {
let cleaned_target = match target {
ApiReviewTarget::UncommittedChanges => ApiReviewTarget::UncommittedChanges,
ApiReviewTarget::BaseBranch { branch } => {
@@ -391,6 +398,19 @@ impl TurnRequestProcessor {
ApiReviewTarget::Commit { sha, title } => CoreReviewTarget::Commit { sha, title },
ApiReviewTarget::Custom { instructions } => CoreReviewTarget::Custom { instructions },
};
let target_prompt = match &core_target {
CoreReviewTarget::UncommittedChanges => {
"Review the current code changes (staged, unstaged, and untracked files)."
.to_string()
}
CoreReviewTarget::BaseBranch { branch } => {
format!("Review the code changes against the base branch {branch:?}.")
}
CoreReviewTarget::Commit { sha, .. } => {
format!("Review the changes introduced by commit {sha:?}.")
}
CoreReviewTarget::Custom { instructions } => instructions.clone(),
};
let hint = codex_core::review_prompts::user_facing_hint(&core_target);
let review_request = ReviewRequest {
@@ -398,7 +418,7 @@ impl TurnRequestProcessor {
user_facing_hint: Some(hint.clone()),
};
Ok((review_request, hint))
Ok((review_request, hint, target_prompt))
}
async fn request_trace_context(
@@ -1235,53 +1255,65 @@ impl TurnRequestProcessor {
async fn start_detached_review(
&self,
request_id: &ConnectionRequestId,
parent_thread_id: ThreadId,
parent_thread: Arc<CodexThread>,
review_request: ReviewRequest,
display_text: &str,
prompt: &str,
) -> std::result::Result<(), JSONRPCErrorError> {
parent_thread.ensure_rollout_materialized().await;
parent_thread.flush_rollout().await.map_err(|err| {
internal_error(format!(
"failed to flush parent thread {parent_thread_id}: {err}"
))
})?;
let parent_history = parent_thread
.load_history(/*include_archived*/ true)
.await
.map_err(|err| {
internal_error(format!(
"failed to load parent thread {parent_thread_id}: {err}"
))
})?;
let mut config = self.config.as_ref().clone();
if let Some(review_model) = &config.review_model {
config.model = Some(review_model.clone());
}
let NewThread {
let AgentRun {
thread_id,
thread: review_thread,
..
turn_id,
} = self
.thread_manager
.fork_thread_from_history(
ForkSnapshot::Interrupted,
config.clone(),
InitialHistory::Resumed(ResumedHistory {
conversation_id: parent_thread_id,
history: Arc::new(parent_history.items),
rollout_path: parent_thread.rollout_path(),
}),
/*thread_source*/ None,
self.request_trace_context(request_id).await,
/*supports_openai_form_elicitation*/ false,
.agent_runner
.start(
parent_thread.session_configured().thread_id,
AgentInvocation {
config,
prompt: prompt.to_string(),
parent_trace: self.request_trace_context(request_id).await,
},
)
.await
.map_err(|err| {
internal_error(format!("error creating detached review thread: {err}"))
})?;
.map_err(|err| internal_error(format!("failed to start detached review: {err}")))?;
let fallback_provider = self.config.model_provider_id.as_str();
let stored_thread = match review_thread
.read_thread(
/*include_archived*/ true, /*include_history*/ false,
)
.await
{
Ok(stored_thread) => {
let (thread, _) =
thread_from_stored_thread(stored_thread, fallback_provider, &self.config.cwd);
Some(thread)
}
Err(err) => {
tracing::warn!("failed to load summary for review thread {thread_id}: {err}");
None
}
};
if let Some(mut thread) = stored_thread {
thread.session_id = review_thread.session_configured().session_id.to_string();
self.thread_watch_manager
.upsert_thread_silently(thread.clone())
.await;
thread.status = resolve_thread_status(
self.thread_watch_manager
.loaded_status_for_thread(&thread.id)
.await,
/*has_in_progress_turn*/ false,
);
let notif = thread_started_notification(thread);
self.outgoing
.send_server_notification(ServerNotification::ThreadStarted(notif))
.await;
}
log_listener_attach_result(
self.ensure_conversation_listener(
@@ -1295,48 +1327,7 @@ impl TurnRequestProcessor {
"review thread",
);
let fallback_provider = self.config.model_provider_id.as_str();
match review_thread
.read_thread(
/*include_archived*/ true, /*include_history*/ false,
)
.await
{
Ok(stored_thread) => {
let (mut thread, _) =
thread_from_stored_thread(stored_thread, fallback_provider, &self.config.cwd);
thread.session_id = review_thread.session_configured().session_id.to_string();
self.thread_watch_manager
.upsert_thread_silently(thread.clone())
.await;
thread.status = resolve_thread_status(
self.thread_watch_manager
.loaded_status_for_thread(&thread.id)
.await,
/*has_in_progress_turn*/ false,
);
let notif = thread_started_notification(thread);
self.outgoing
.send_server_notification(ServerNotification::ThreadStarted(notif))
.await;
}
Err(err) => {
tracing::warn!("failed to load summary for review thread {thread_id}: {err}");
}
}
let turn_id = self
.submit_core_op(
request_id,
review_thread.as_ref(),
Op::Review { review_request },
)
.await
.map_err(|err| {
internal_error(format!("failed to start detached review turn: {err}"))
})?;
let turn = Self::build_review_turn(turn_id, display_text);
let turn = Self::build_review_turn(turn_id, prompt);
let review_thread_id = thread_id.to_string();
self.emit_review_started(request_id, turn, review_thread_id)
.await;
@@ -1355,8 +1346,9 @@ impl TurnRequestProcessor {
delivery,
} = params;
let (parent_thread_id, parent_thread) = self.load_thread(&thread_id).await?;
let (review_request, display_text) = Self::review_request_from_target(target)?;
let (_, parent_thread) = self.load_thread(&thread_id).await?;
let (review_request, display_text, target_prompt) =
Self::review_request_from_target(target)?;
match delivery.unwrap_or(ApiReviewDelivery::Inline).to_core() {
CoreReviewDelivery::Inline => {
self.start_inline_review(
@@ -1369,14 +1361,19 @@ impl TurnRequestProcessor {
.await?;
}
CoreReviewDelivery::Detached => {
self.start_detached_review(
request_id,
parent_thread_id,
parent_thread,
review_request,
&display_text,
)
.await?;
let review_skill_path = system_cache_root_dir(&self.config.codex_home)
.join("review-agent")
.join("SKILL.md");
let prompt = format!(
"Use [$review-agent]({}) for this review.\n\n{target_prompt}",
review_skill_path.display()
);
let actual_chars = prompt.chars().count();
if actual_chars > MAX_USER_INPUT_TEXT_CHARS {
return Err(Self::input_too_large_error(actual_chars));
}
self.start_detached_review(request_id, parent_thread, &prompt)
.await?;
}
}
Ok(())
+57 -9
View File
@@ -26,6 +26,8 @@ use codex_app_server_protocol::TurnItemsView;
use codex_app_server_protocol::TurnStartParams;
use codex_app_server_protocol::TurnStatus;
use codex_app_server_protocol::UserInput as V2UserInput;
use codex_skills::system_cache_root_dir;
use core_test_support::responses;
use pretty_assertions::assert_eq;
use serde_json::json;
use tempfile::TempDir;
@@ -33,6 +35,7 @@ use tokio::time::timeout;
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
const INVALID_REQUEST_ERROR_CODE: i64 = -32600;
const COLLIDING_REVIEW_SKILL_MARKER: &str = "COLLIDING_REVIEW_SKILL_MARKER";
#[tokio::test]
async fn review_start_runs_review_turn_and_emits_code_review_item() -> Result<()> {
@@ -296,17 +299,42 @@ async fn review_start_rejects_empty_base_branch() -> Result<()> {
#[cfg_attr(target_os = "windows", ignore = "flaky on windows CI")]
#[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 server = create_mock_responses_server_repeating_assistant(&review_payload).await;
let server = responses::start_mock_server().await;
let response_mock = responses::mount_sse_sequence(
&server,
vec![
responses::sse(vec![
responses::ev_response_created("materialize-response"),
responses::ev_assistant_message("materialize-message", "materialized"),
responses::ev_completed("materialize-response"),
]),
responses::sse(vec![
responses::ev_response_created("review-response"),
responses::ev_assistant_message("review-message", "No findings."),
responses::ev_completed("review-response"),
]),
],
)
.await;
let codex_home = TempDir::new()?;
create_config_toml(codex_home.path(), &server.uri())?;
let colliding_skill_dir = codex_home.path().join("skills/review-agent-collision");
std::fs::create_dir_all(&colliding_skill_dir)?;
std::fs::write(
colliding_skill_dir.join("SKILL.md"),
format!(
"---\nname: review-agent\ndescription: Colliding user review skill.\n---\n\n{COLLIDING_REVIEW_SKILL_MARKER}\n"
),
)?;
let canonical_codex_home = std::fs::canonicalize(codex_home.path())?.try_into()?;
let review_skill_path = system_cache_root_dir(&canonical_codex_home)
.join("review-agent")
.join("SKILL.md");
let expected_prompt = format!(
"Use [$review-agent]({}) for this review.\n\ndetached review",
review_skill_path.display()
);
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
@@ -344,7 +372,7 @@ async fn review_start_with_detached_delivery_returns_new_thread_id() -> Result<(
id: turn.id.clone(),
client_id: None,
content: vec![V2UserInput::Text {
text: "detached review".to_string(),
text: expected_prompt.clone(),
text_elements: Vec::new(),
}],
}]
@@ -380,6 +408,26 @@ async fn review_start_with_detached_delivery_returns_new_thread_id() -> Result<(
assert_eq!(started.thread.id, review_thread_id);
assert_eq!(started.thread.session_id, review_thread_id);
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("turn/completed"),
)
.await??;
let requests = response_mock.requests();
assert_eq!(requests.len(), 2);
let review_request = &requests[1];
assert_eq!(review_request.header("x-openai-subagent"), None);
assert!(review_request.body_contains_text("Colliding user review skill."));
let user_messages = review_request.message_input_texts("user");
assert!(user_messages.iter().any(|text| text == &expected_prompt));
assert!(user_messages.iter().any(|text| {
text.starts_with("<skill>")
&& text.contains("<name>review-agent</name>")
&& text.contains("Do not modify files")
}));
assert!(!review_request.body_contains_text(COLLIDING_REVIEW_SKILL_MARKER));
Ok(())
}
@@ -0,0 +1,57 @@
---
name: review-agent
description: Perform a read-only, defect-first review of a specified code change and return every actionable finding. Use when another agent delegates review of uncommitted changes, a base-branch diff, a commit, or custom review instructions.
---
# Review Agent
Inspect the requested target directly and return every finding that the author would likely fix.
Do not modify files, create commits, push branches, post review comments, or delegate the review
to another agent.
## Review the change
1. Read the applicable `AGENTS.md` instructions.
2. Inspect the complete diff for the requested target and enough surrounding code to understand
each changed path.
3. Identify concrete regressions introduced by the change. Continue through the whole diff after
finding the first issue.
4. Check the relevant tests and call sites to confirm that each finding is real and actionable.
For a base-branch review, compare the changes that would actually merge rather than diffing
directly against the branch tip. Resolve the comparison ref to the branch's upstream when that
upstream exists and is ahead of the local branch; otherwise use the local branch. Run
`git merge-base HEAD <comparison-ref>`, then inspect `git diff <merge-base-sha>`. If the local
branch cannot be resolved, try its configured upstream explicitly before reporting that the target
is unavailable.
Flag an issue only when all of these are true:
- It affects correctness, security, performance, or maintainability in a meaningful way.
- It is discrete and actionable.
- It was introduced by the reviewed change.
- The affected scenario or call path can be demonstrated from the code.
- The author would probably fix it if they knew about it.
Do not flag speculative concerns, pre-existing problems, intentional behavior changes, or style
nits that do not obscure the code.
## Write the result
Present findings first, ordered by severity. Use one entry per issue in this form:
`[P1] Imperative finding title — path/to/file.rs:line`
Follow the title with one short paragraph explaining the affected scenario and why the behavior is
wrong. Keep the cited range as small as possible and make sure it overlaps the reviewed diff.
Use these priorities:
- `P0`: universal release blocker or critical failure.
- `P1`: urgent defect that should be fixed next.
- `P2`: ordinary defect that should be fixed.
- `P3`: low-impact issue that is still worth fixing.
If there are no qualifying findings, say `No findings.` Do not invent a finding to fill the result.
After the findings, add a brief overall assessment and mention any material test gaps or residual
risks.
@@ -0,0 +1,6 @@
interface:
display_name: "Review Agent"
short_description: "Find actionable bugs in code changes"
default_prompt: "Use $review-agent to review the requested code changes and return actionable findings."
policy:
allow_implicit_invocation: false