mirror of
https://github.com/openai/codex.git
synced 2026-10-28 18:58:46 +00:00
Merge af5615b4b9 into sapling-pr-archive-rhan-oai
This commit is contained in:
@@ -1,5 +1,6 @@
|
||||
use crate::events::AppServerRpcTransport;
|
||||
use crate::events::GuardianReviewEventParams;
|
||||
use crate::events::GuardianReviewAnalyticsResult;
|
||||
use crate::events::GuardianReviewTrackContext;
|
||||
use crate::events::TrackEventRequest;
|
||||
use crate::events::TrackEventsRequest;
|
||||
use crate::events::current_runtime_metadata;
|
||||
@@ -166,9 +167,13 @@ impl AnalyticsEventsClient {
|
||||
));
|
||||
}
|
||||
|
||||
pub fn track_guardian_review(&self, input: GuardianReviewEventParams) {
|
||||
pub fn track_guardian_review(
|
||||
&self,
|
||||
tracking: &GuardianReviewTrackContext,
|
||||
result: GuardianReviewAnalyticsResult,
|
||||
) {
|
||||
self.record_fact(AnalyticsFact::Custom(CustomAnalyticsFact::GuardianReview(
|
||||
Box::new(input),
|
||||
Box::new(tracking.event_params(result)),
|
||||
)));
|
||||
}
|
||||
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
use std::time::Instant;
|
||||
|
||||
use crate::facts::AppInvocation;
|
||||
use crate::facts::CodexCompactionEvent;
|
||||
use crate::facts::CodexResponsesApiCallStatus;
|
||||
@@ -18,6 +20,7 @@ use crate::facts::TurnStatus;
|
||||
use crate::facts::TurnSteerRejectionReason;
|
||||
use crate::facts::TurnSteerResult;
|
||||
use crate::facts::TurnSubmissionType;
|
||||
use crate::now_unix_seconds;
|
||||
use codex_app_server_protocol::CodexErrorInfo;
|
||||
use codex_login::default_client::originator;
|
||||
use codex_plugin::PluginTelemetryMetadata;
|
||||
@@ -32,6 +35,7 @@ use codex_protocol::protocol::HookEventName;
|
||||
use codex_protocol::protocol::HookRunStatus;
|
||||
use codex_protocol::protocol::HookSource;
|
||||
use codex_protocol::protocol::SubAgentSource;
|
||||
use codex_protocol::protocol::TokenUsage;
|
||||
use serde::Serialize;
|
||||
|
||||
#[derive(Clone, Copy, Debug, Serialize)]
|
||||
@@ -203,6 +207,7 @@ pub enum GuardianReviewedAction {
|
||||
connector_name: Option<String>,
|
||||
tool_title: Option<String>,
|
||||
},
|
||||
RequestPermissions {},
|
||||
}
|
||||
|
||||
#[derive(Clone, Serialize)]
|
||||
@@ -238,6 +243,142 @@ pub struct GuardianReviewEventParams {
|
||||
pub total_tokens: Option<i64>,
|
||||
}
|
||||
|
||||
pub struct GuardianReviewTrackContext {
|
||||
thread_id: String,
|
||||
turn_id: String,
|
||||
review_id: String,
|
||||
target_item_id: Option<String>,
|
||||
approval_request_source: GuardianApprovalRequestSource,
|
||||
reviewed_action: GuardianReviewedAction,
|
||||
review_timeout_ms: u64,
|
||||
started_at: u64,
|
||||
started_instant: Instant,
|
||||
}
|
||||
|
||||
impl GuardianReviewTrackContext {
|
||||
pub fn new(
|
||||
thread_id: String,
|
||||
turn_id: String,
|
||||
review_id: String,
|
||||
target_item_id: Option<String>,
|
||||
approval_request_source: GuardianApprovalRequestSource,
|
||||
reviewed_action: GuardianReviewedAction,
|
||||
review_timeout_ms: u64,
|
||||
) -> Self {
|
||||
Self {
|
||||
thread_id,
|
||||
turn_id,
|
||||
review_id,
|
||||
target_item_id,
|
||||
approval_request_source,
|
||||
reviewed_action,
|
||||
review_timeout_ms,
|
||||
started_at: now_unix_seconds(),
|
||||
started_instant: Instant::now(),
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn event_params(
|
||||
&self,
|
||||
result: GuardianReviewAnalyticsResult,
|
||||
) -> GuardianReviewEventParams {
|
||||
GuardianReviewEventParams {
|
||||
thread_id: self.thread_id.clone(),
|
||||
turn_id: self.turn_id.clone(),
|
||||
review_id: self.review_id.clone(),
|
||||
target_item_id: self.target_item_id.clone(),
|
||||
approval_request_source: self.approval_request_source,
|
||||
reviewed_action: self.reviewed_action.clone(),
|
||||
reviewed_action_truncated: result.reviewed_action_truncated,
|
||||
decision: result.decision,
|
||||
terminal_status: result.terminal_status,
|
||||
failure_reason: result.failure_reason,
|
||||
risk_level: result.risk_level,
|
||||
user_authorization: result.user_authorization,
|
||||
outcome: result.outcome,
|
||||
guardian_thread_id: result.guardian_thread_id,
|
||||
guardian_session_kind: result.guardian_session_kind,
|
||||
guardian_model: result.guardian_model,
|
||||
guardian_reasoning_effort: result.guardian_reasoning_effort,
|
||||
had_prior_review_context: result.had_prior_review_context,
|
||||
review_timeout_ms: self.review_timeout_ms,
|
||||
// TODO(rhan-oai): plumb nested Guardian review session tool-call counts.
|
||||
tool_call_count: None,
|
||||
time_to_first_token_ms: result.time_to_first_token_ms,
|
||||
completion_latency_ms: Some(self.started_instant.elapsed().as_millis() as u64),
|
||||
started_at: self.started_at,
|
||||
completed_at: Some(now_unix_seconds()),
|
||||
input_tokens: result.token_usage.as_ref().map(|usage| usage.input_tokens),
|
||||
cached_input_tokens: result
|
||||
.token_usage
|
||||
.as_ref()
|
||||
.map(|usage| usage.cached_input_tokens),
|
||||
output_tokens: result.token_usage.as_ref().map(|usage| usage.output_tokens),
|
||||
reasoning_output_tokens: result
|
||||
.token_usage
|
||||
.as_ref()
|
||||
.map(|usage| usage.reasoning_output_tokens),
|
||||
total_tokens: result.token_usage.as_ref().map(|usage| usage.total_tokens),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub struct GuardianReviewAnalyticsResult {
|
||||
pub decision: GuardianReviewDecision,
|
||||
pub terminal_status: GuardianReviewTerminalStatus,
|
||||
pub failure_reason: Option<GuardianReviewFailureReason>,
|
||||
pub risk_level: Option<GuardianRiskLevel>,
|
||||
pub user_authorization: Option<GuardianUserAuthorization>,
|
||||
pub outcome: Option<GuardianAssessmentOutcome>,
|
||||
pub guardian_thread_id: Option<String>,
|
||||
pub guardian_session_kind: Option<GuardianReviewSessionKind>,
|
||||
pub guardian_model: Option<String>,
|
||||
pub guardian_reasoning_effort: Option<String>,
|
||||
pub had_prior_review_context: Option<bool>,
|
||||
pub reviewed_action_truncated: bool,
|
||||
pub token_usage: Option<TokenUsage>,
|
||||
pub time_to_first_token_ms: Option<u64>,
|
||||
}
|
||||
|
||||
impl GuardianReviewAnalyticsResult {
|
||||
pub fn without_session() -> Self {
|
||||
Self {
|
||||
decision: GuardianReviewDecision::Denied,
|
||||
terminal_status: GuardianReviewTerminalStatus::FailedClosed,
|
||||
failure_reason: None,
|
||||
risk_level: None,
|
||||
user_authorization: None,
|
||||
outcome: None,
|
||||
guardian_thread_id: None,
|
||||
guardian_session_kind: None,
|
||||
guardian_model: None,
|
||||
guardian_reasoning_effort: None,
|
||||
had_prior_review_context: None,
|
||||
reviewed_action_truncated: false,
|
||||
token_usage: None,
|
||||
time_to_first_token_ms: None,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn from_session(
|
||||
guardian_thread_id: String,
|
||||
guardian_session_kind: GuardianReviewSessionKind,
|
||||
guardian_model: String,
|
||||
guardian_reasoning_effort: Option<String>,
|
||||
had_prior_review_context: bool,
|
||||
) -> Self {
|
||||
Self {
|
||||
guardian_thread_id: Some(guardian_thread_id),
|
||||
guardian_session_kind: Some(guardian_session_kind),
|
||||
guardian_model: Some(guardian_model),
|
||||
guardian_reasoning_effort,
|
||||
had_prior_review_context: Some(had_prior_review_context),
|
||||
..Self::without_session()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
pub(crate) struct GuardianReviewEventPayload {
|
||||
pub(crate) app_server_client: CodexAppServerClientMetadata,
|
||||
|
||||
@@ -11,11 +11,13 @@ use std::time::UNIX_EPOCH;
|
||||
pub use client::AnalyticsEventsClient;
|
||||
pub use events::AppServerRpcTransport;
|
||||
pub use events::GuardianApprovalRequestSource;
|
||||
pub use events::GuardianReviewAnalyticsResult;
|
||||
pub use events::GuardianReviewDecision;
|
||||
pub use events::GuardianReviewEventParams;
|
||||
pub use events::GuardianReviewFailureReason;
|
||||
pub use events::GuardianReviewSessionKind;
|
||||
pub use events::GuardianReviewTerminalStatus;
|
||||
pub use events::GuardianReviewTrackContext;
|
||||
pub use events::GuardianReviewedAction;
|
||||
pub use facts::AnalyticsJsonRpcError;
|
||||
pub use facts::AppInvocation;
|
||||
|
||||
@@ -1359,6 +1359,7 @@ mod tests {
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
}),
|
||||
];
|
||||
|
||||
@@ -1433,6 +1434,7 @@ mod tests {
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
})),
|
||||
];
|
||||
|
||||
@@ -1756,6 +1758,7 @@ mod tests {
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
}),
|
||||
];
|
||||
|
||||
@@ -2269,6 +2272,7 @@ mod tests {
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
}),
|
||||
EventMsg::TurnStarted(TurnStartedEvent {
|
||||
turn_id: "turn-b".into(),
|
||||
@@ -2306,6 +2310,7 @@ mod tests {
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
}),
|
||||
];
|
||||
|
||||
@@ -2358,6 +2363,7 @@ mod tests {
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
}),
|
||||
EventMsg::TurnStarted(TurnStartedEvent {
|
||||
turn_id: "turn-b".into(),
|
||||
@@ -2395,6 +2401,7 @@ mod tests {
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
}),
|
||||
];
|
||||
|
||||
@@ -2569,6 +2576,7 @@ mod tests {
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
}),
|
||||
EventMsg::TurnStarted(TurnStartedEvent {
|
||||
turn_id: "turn-b".into(),
|
||||
@@ -2587,6 +2595,7 @@ mod tests {
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
}),
|
||||
EventMsg::AgentMessage(AgentMessageEvent {
|
||||
message: "still in b".into(),
|
||||
@@ -2598,6 +2607,7 @@ mod tests {
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
}),
|
||||
];
|
||||
|
||||
@@ -2632,6 +2642,7 @@ mod tests {
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
}),
|
||||
EventMsg::TurnStarted(TurnStartedEvent {
|
||||
turn_id: "turn-b".into(),
|
||||
@@ -2688,6 +2699,7 @@ mod tests {
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
})),
|
||||
];
|
||||
|
||||
@@ -2933,6 +2945,7 @@ mod tests {
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
}),
|
||||
EventMsg::Error(ErrorEvent {
|
||||
message: "request-level failure".into(),
|
||||
@@ -2992,6 +3005,7 @@ mod tests {
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
}),
|
||||
];
|
||||
|
||||
@@ -3043,6 +3057,7 @@ mod tests {
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
})),
|
||||
];
|
||||
|
||||
@@ -3091,6 +3106,7 @@ mod tests {
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
})),
|
||||
];
|
||||
|
||||
|
||||
@@ -3079,6 +3079,7 @@ mod tests {
|
||||
last_agent_message: None,
|
||||
completed_at: Some(TEST_TURN_COMPLETED_AT),
|
||||
duration_ms: Some(TEST_TURN_DURATION_MS),
|
||||
time_to_first_token_ms: None,
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -280,6 +280,7 @@ async fn on_event_updates_status_from_task_complete() {
|
||||
last_agent_message: Some("done".to_string()),
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
}));
|
||||
let expected = AgentStatus::Completed(Some("done".to_string()));
|
||||
assert_eq!(status, Some(expected));
|
||||
@@ -1230,6 +1231,7 @@ async fn multi_agent_v2_completion_ignores_dead_direct_parent() {
|
||||
last_agent_message: Some("done".to_string()),
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
@@ -1316,6 +1318,7 @@ async fn multi_agent_v2_completion_queues_message_for_direct_parent() {
|
||||
last_agent_message: Some("done".to_string()),
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
use std::path::Path;
|
||||
|
||||
use codex_analytics::GuardianReviewedAction;
|
||||
use codex_protocol::approvals::GuardianAssessmentAction;
|
||||
use codex_protocol::approvals::GuardianCommandSource;
|
||||
use codex_protocol::approvals::NetworkApprovalProtocol;
|
||||
@@ -183,32 +184,49 @@ fn guardian_command_source_tool_name(source: GuardianCommandSource) -> &'static
|
||||
}
|
||||
}
|
||||
|
||||
fn truncate_guardian_action_value(value: Value) -> Value {
|
||||
fn truncate_guardian_action_value(value: Value) -> (Value, bool) {
|
||||
match value {
|
||||
Value::String(text) => Value::String(guardian_truncate_text(
|
||||
&text,
|
||||
GUARDIAN_MAX_ACTION_STRING_TOKENS,
|
||||
)),
|
||||
Value::Array(values) => Value::Array(
|
||||
values
|
||||
Value::String(text) => {
|
||||
let (text, truncated) =
|
||||
guardian_truncate_text(&text, GUARDIAN_MAX_ACTION_STRING_TOKENS);
|
||||
(Value::String(text), truncated)
|
||||
}
|
||||
Value::Array(values) => {
|
||||
let mut truncated = false;
|
||||
let values = values
|
||||
.into_iter()
|
||||
.map(truncate_guardian_action_value)
|
||||
.collect::<Vec<_>>(),
|
||||
),
|
||||
.map(|value| {
|
||||
let (value, value_truncated) = truncate_guardian_action_value(value);
|
||||
truncated |= value_truncated;
|
||||
value
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
(Value::Array(values), truncated)
|
||||
}
|
||||
Value::Object(values) => {
|
||||
let mut entries = values.into_iter().collect::<Vec<_>>();
|
||||
entries.sort_by(|(left, _), (right, _)| left.cmp(right));
|
||||
Value::Object(
|
||||
entries
|
||||
.into_iter()
|
||||
.map(|(key, value)| (key, truncate_guardian_action_value(value)))
|
||||
.collect(),
|
||||
)
|
||||
let mut truncated = false;
|
||||
let values = entries
|
||||
.into_iter()
|
||||
.map(|(key, value)| {
|
||||
let (value, value_truncated) = truncate_guardian_action_value(value);
|
||||
truncated |= value_truncated;
|
||||
(key, value)
|
||||
})
|
||||
.collect();
|
||||
(Value::Object(values), truncated)
|
||||
}
|
||||
other => other,
|
||||
other => (other, false),
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub(crate) struct FormattedGuardianAction {
|
||||
pub(crate) text: String,
|
||||
pub(crate) truncated: bool,
|
||||
}
|
||||
|
||||
pub(crate) fn guardian_approval_request_to_json(
|
||||
action: &GuardianApprovalRequest,
|
||||
) -> serde_json::Result<Value> {
|
||||
@@ -390,6 +408,66 @@ pub(crate) fn guardian_assessment_action(
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn guardian_reviewed_action(
|
||||
request: &GuardianApprovalRequest,
|
||||
) -> GuardianReviewedAction {
|
||||
match request {
|
||||
GuardianApprovalRequest::Shell {
|
||||
sandbox_permissions,
|
||||
additional_permissions,
|
||||
..
|
||||
} => GuardianReviewedAction::Shell {
|
||||
sandbox_permissions: *sandbox_permissions,
|
||||
additional_permissions: additional_permissions.clone(),
|
||||
},
|
||||
GuardianApprovalRequest::ExecCommand {
|
||||
sandbox_permissions,
|
||||
additional_permissions,
|
||||
tty,
|
||||
..
|
||||
} => GuardianReviewedAction::UnifiedExec {
|
||||
sandbox_permissions: *sandbox_permissions,
|
||||
additional_permissions: additional_permissions.clone(),
|
||||
tty: *tty,
|
||||
},
|
||||
#[cfg(unix)]
|
||||
GuardianApprovalRequest::Execve {
|
||||
source,
|
||||
program,
|
||||
additional_permissions,
|
||||
..
|
||||
} => GuardianReviewedAction::Execve {
|
||||
source: *source,
|
||||
program: program.clone(),
|
||||
additional_permissions: additional_permissions.clone(),
|
||||
},
|
||||
GuardianApprovalRequest::ApplyPatch { .. } => GuardianReviewedAction::ApplyPatch {},
|
||||
GuardianApprovalRequest::NetworkAccess { protocol, port, .. } => {
|
||||
GuardianReviewedAction::NetworkAccess {
|
||||
protocol: *protocol,
|
||||
port: *port,
|
||||
}
|
||||
}
|
||||
GuardianApprovalRequest::McpToolCall {
|
||||
server,
|
||||
tool_name,
|
||||
connector_id,
|
||||
connector_name,
|
||||
tool_title,
|
||||
..
|
||||
} => GuardianReviewedAction::McpToolCall {
|
||||
server: server.clone(),
|
||||
tool_name: tool_name.clone(),
|
||||
connector_id: connector_id.clone(),
|
||||
connector_name: connector_name.clone(),
|
||||
tool_title: tool_title.clone(),
|
||||
},
|
||||
GuardianApprovalRequest::RequestPermissions { .. } => {
|
||||
GuardianReviewedAction::RequestPermissions {}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn guardian_request_target_item_id(request: &GuardianApprovalRequest) -> Option<&str> {
|
||||
match request {
|
||||
GuardianApprovalRequest::Shell { id, .. }
|
||||
@@ -421,8 +499,11 @@ pub(crate) fn guardian_request_turn_id<'a>(
|
||||
|
||||
pub(crate) fn format_guardian_action_pretty(
|
||||
action: &GuardianApprovalRequest,
|
||||
) -> serde_json::Result<String> {
|
||||
let mut value = guardian_approval_request_to_json(action)?;
|
||||
value = truncate_guardian_action_value(value);
|
||||
serde_json::to_string_pretty(&value)
|
||||
) -> serde_json::Result<FormattedGuardianAction> {
|
||||
let value = guardian_approval_request_to_json(action)?;
|
||||
let (value, truncated) = truncate_guardian_action_value(value);
|
||||
Ok(FormattedGuardianAction {
|
||||
text: serde_json::to_string_pretty(&value)?,
|
||||
truncated,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -56,6 +56,7 @@ impl GuardianTranscriptEntryKind {
|
||||
pub(crate) struct GuardianPromptItems {
|
||||
pub(crate) items: Vec<UserInput>,
|
||||
pub(crate) transcript_cursor: GuardianTranscriptCursor,
|
||||
pub(crate) reviewed_action_truncated: bool,
|
||||
}
|
||||
|
||||
/// Points to the end of the transcript that the guardian has already reviewed.
|
||||
@@ -176,11 +177,12 @@ pub(crate) async fn build_guardian_prompt_items(
|
||||
.to_string(),
|
||||
);
|
||||
push_text("Planned action JSON:\n".to_string());
|
||||
push_text(format!("{planned_action_json}\n"));
|
||||
push_text(format!("{}\n", planned_action_json.text));
|
||||
push_text(">>> APPROVAL REQUEST END\n".to_string());
|
||||
Ok(GuardianPromptItems {
|
||||
items,
|
||||
transcript_cursor,
|
||||
reviewed_action_truncated: planned_action_json.truncated,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -240,7 +242,7 @@ fn render_guardian_transcript_entries_with_offset(
|
||||
} else {
|
||||
GUARDIAN_MAX_MESSAGE_ENTRY_TOKENS
|
||||
};
|
||||
let text = guardian_truncate_text(&entry.text, token_cap);
|
||||
let (text, _) = guardian_truncate_text(&entry.text, token_cap);
|
||||
let rendered = format!(
|
||||
"[{}] {}: {}",
|
||||
index + entry_number_offset + 1,
|
||||
@@ -420,20 +422,20 @@ pub(crate) fn collect_guardian_transcript_entries(
|
||||
entries
|
||||
}
|
||||
|
||||
pub(crate) fn guardian_truncate_text(content: &str, token_cap: usize) -> String {
|
||||
pub(crate) fn guardian_truncate_text(content: &str, token_cap: usize) -> (String, bool) {
|
||||
if content.is_empty() {
|
||||
return String::new();
|
||||
return (String::new(), false);
|
||||
}
|
||||
|
||||
let max_bytes = approx_bytes_for_tokens(token_cap);
|
||||
if content.len() <= max_bytes {
|
||||
return content.to_string();
|
||||
return (content.to_string(), false);
|
||||
}
|
||||
|
||||
let omitted_tokens = approx_tokens_from_byte_count(content.len().saturating_sub(max_bytes));
|
||||
let marker = format!("<{TRUNCATION_TAG} omitted_approx_tokens=\"{omitted_tokens}\" />");
|
||||
if max_bytes <= marker.len() {
|
||||
return marker;
|
||||
return (marker, true);
|
||||
}
|
||||
|
||||
let available_bytes = max_bytes.saturating_sub(marker.len());
|
||||
@@ -441,7 +443,7 @@ pub(crate) fn guardian_truncate_text(content: &str, token_cap: usize) -> String
|
||||
let suffix_budget = available_bytes.saturating_sub(prefix_budget);
|
||||
let (prefix, suffix) = split_guardian_truncation_bounds(content, prefix_budget, suffix_budget);
|
||||
|
||||
format!("{prefix}{marker}{suffix}")
|
||||
(format!("{prefix}{marker}{suffix}"), true)
|
||||
}
|
||||
|
||||
fn split_guardian_truncation_bounds(
|
||||
|
||||
@@ -1,5 +1,12 @@
|
||||
use std::sync::Arc;
|
||||
|
||||
use codex_analytics::GuardianApprovalRequestSource;
|
||||
use codex_analytics::GuardianReviewAnalyticsResult;
|
||||
use codex_analytics::GuardianReviewDecision;
|
||||
use codex_analytics::GuardianReviewFailureReason;
|
||||
use codex_analytics::GuardianReviewTerminalStatus;
|
||||
use codex_analytics::GuardianReviewTrackContext;
|
||||
use codex_features::Feature;
|
||||
use codex_protocol::config_types::ApprovalsReviewer;
|
||||
use codex_protocol::protocol::AskForApproval;
|
||||
use codex_protocol::protocol::EventMsg;
|
||||
@@ -17,6 +24,7 @@ use tokio_util::sync::CancellationToken;
|
||||
use crate::session::session::Session;
|
||||
use crate::session::turn_context::TurnContext;
|
||||
|
||||
use super::GUARDIAN_REVIEW_TIMEOUT;
|
||||
use super::GUARDIAN_REVIEWER_NAME;
|
||||
use super::GuardianApprovalRequest;
|
||||
use super::GuardianAssessment;
|
||||
@@ -25,6 +33,7 @@ use super::GuardianRejection;
|
||||
use super::approval_request::guardian_assessment_action;
|
||||
use super::approval_request::guardian_request_target_item_id;
|
||||
use super::approval_request::guardian_request_turn_id;
|
||||
use super::approval_request::guardian_reviewed_action;
|
||||
use super::prompt::guardian_output_schema;
|
||||
use super::prompt::parse_guardian_assessment;
|
||||
use super::review_session::GuardianReviewSessionOutcome;
|
||||
@@ -76,11 +85,55 @@ pub(crate) fn guardian_timeout_message() -> String {
|
||||
|
||||
#[derive(Debug)]
|
||||
pub(super) enum GuardianReviewOutcome {
|
||||
Completed(anyhow::Result<GuardianAssessment>),
|
||||
Completed(GuardianAssessment),
|
||||
Failed(GuardianReviewError),
|
||||
TimedOut,
|
||||
Aborted,
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub(super) enum GuardianReviewError {
|
||||
PromptBuild { message: String },
|
||||
Session { message: String },
|
||||
Parse { message: String },
|
||||
}
|
||||
|
||||
impl GuardianReviewError {
|
||||
fn prompt_build(err: anyhow::Error) -> Self {
|
||||
Self::PromptBuild {
|
||||
message: err.to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
fn session(err: anyhow::Error) -> Self {
|
||||
Self::Session {
|
||||
message: err.to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
fn parse(err: anyhow::Error) -> Self {
|
||||
Self::Parse {
|
||||
message: err.to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
fn reason(&self) -> GuardianReviewFailureReason {
|
||||
match self {
|
||||
Self::PromptBuild { .. } => GuardianReviewFailureReason::PromptBuildError,
|
||||
Self::Session { .. } => GuardianReviewFailureReason::SessionError,
|
||||
Self::Parse { .. } => GuardianReviewFailureReason::ParseError,
|
||||
}
|
||||
}
|
||||
|
||||
fn message(&self) -> &str {
|
||||
match self {
|
||||
Self::PromptBuild { message } | Self::Session { message } | Self::Parse { message } => {
|
||||
message
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn guardian_risk_level_str(level: GuardianRiskLevel) -> &'static str {
|
||||
match level {
|
||||
GuardianRiskLevel::Low => "low",
|
||||
@@ -110,6 +163,21 @@ pub(crate) fn is_guardian_reviewer_source(
|
||||
)
|
||||
}
|
||||
|
||||
fn track_guardian_review(
|
||||
session: &Session,
|
||||
turn: &TurnContext,
|
||||
tracking: &GuardianReviewTrackContext,
|
||||
result: GuardianReviewAnalyticsResult,
|
||||
) {
|
||||
if !turn.config.features.enabled(Feature::GeneralAnalytics) {
|
||||
return;
|
||||
}
|
||||
session
|
||||
.services
|
||||
.analytics_events_client
|
||||
.track_guardian_review(tracking, result);
|
||||
}
|
||||
|
||||
/// This function always fails closed: timeouts, review-session failures, and
|
||||
/// parse failures all block execution, but timeouts are still surfaced to the
|
||||
/// caller as distinct from explicit guardian denials.
|
||||
@@ -119,11 +187,21 @@ async fn run_guardian_review(
|
||||
review_id: String,
|
||||
request: GuardianApprovalRequest,
|
||||
retry_reason: Option<String>,
|
||||
approval_request_source: GuardianApprovalRequestSource,
|
||||
external_cancel: Option<CancellationToken>,
|
||||
) -> ReviewDecision {
|
||||
let target_item_id = guardian_request_target_item_id(&request).map(str::to_string);
|
||||
let assessment_turn_id = guardian_request_turn_id(&request, &turn.sub_id).to_string();
|
||||
let action_summary = guardian_assessment_action(&request);
|
||||
let review_tracking = GuardianReviewTrackContext::new(
|
||||
session.conversation_id.to_string(),
|
||||
assessment_turn_id.clone(),
|
||||
review_id.clone(),
|
||||
target_item_id.clone(),
|
||||
approval_request_source,
|
||||
guardian_reviewed_action(&request),
|
||||
GUARDIAN_REVIEW_TIMEOUT.as_millis() as u64,
|
||||
);
|
||||
session
|
||||
.send_event(
|
||||
turn.as_ref(),
|
||||
@@ -145,6 +223,17 @@ async fn run_guardian_review(
|
||||
.as_ref()
|
||||
.is_some_and(CancellationToken::is_cancelled)
|
||||
{
|
||||
track_guardian_review(
|
||||
session.as_ref(),
|
||||
turn.as_ref(),
|
||||
&review_tracking,
|
||||
GuardianReviewAnalyticsResult {
|
||||
decision: GuardianReviewDecision::Aborted,
|
||||
terminal_status: GuardianReviewTerminalStatus::Aborted,
|
||||
failure_reason: Some(GuardianReviewFailureReason::Cancelled),
|
||||
..GuardianReviewAnalyticsResult::without_session()
|
||||
},
|
||||
);
|
||||
session
|
||||
.send_event(
|
||||
turn.as_ref(),
|
||||
@@ -166,28 +255,78 @@ async fn run_guardian_review(
|
||||
|
||||
let schema = guardian_output_schema();
|
||||
let terminal_action = action_summary.clone();
|
||||
let outcome = Box::pin(run_guardian_review_session(
|
||||
let (outcome, analytics_result) = Box::pin(run_guardian_review_session(
|
||||
session.clone(),
|
||||
turn.clone(),
|
||||
request,
|
||||
retry_reason,
|
||||
retry_reason.clone(),
|
||||
schema,
|
||||
external_cancel,
|
||||
))
|
||||
.await;
|
||||
|
||||
let assessment = match outcome {
|
||||
GuardianReviewOutcome::Completed(Ok(assessment)) => assessment,
|
||||
GuardianReviewOutcome::Completed(Err(err)) => GuardianAssessment {
|
||||
risk_level: GuardianRiskLevel::High,
|
||||
user_authorization: GuardianUserAuthorization::Unknown,
|
||||
outcome: GuardianAssessmentOutcome::Deny,
|
||||
rationale: format!("Automatic approval review failed: {err}"),
|
||||
},
|
||||
GuardianReviewOutcome::Completed(assessment) => {
|
||||
let approved = matches!(assessment.outcome, GuardianAssessmentOutcome::Allow);
|
||||
track_guardian_review(
|
||||
session.as_ref(),
|
||||
turn.as_ref(),
|
||||
&review_tracking,
|
||||
GuardianReviewAnalyticsResult {
|
||||
decision: if approved {
|
||||
GuardianReviewDecision::Approved
|
||||
} else {
|
||||
GuardianReviewDecision::Denied
|
||||
},
|
||||
terminal_status: if approved {
|
||||
GuardianReviewTerminalStatus::Approved
|
||||
} else {
|
||||
GuardianReviewTerminalStatus::Denied
|
||||
},
|
||||
failure_reason: None,
|
||||
risk_level: Some(assessment.risk_level),
|
||||
user_authorization: Some(assessment.user_authorization),
|
||||
outcome: Some(assessment.outcome),
|
||||
..analytics_result
|
||||
},
|
||||
);
|
||||
assessment
|
||||
}
|
||||
GuardianReviewOutcome::Failed(failure) => {
|
||||
let rationale = format!("Automatic approval review failed: {}", failure.message());
|
||||
track_guardian_review(
|
||||
session.as_ref(),
|
||||
turn.as_ref(),
|
||||
&review_tracking,
|
||||
GuardianReviewAnalyticsResult {
|
||||
decision: GuardianReviewDecision::Denied,
|
||||
terminal_status: GuardianReviewTerminalStatus::FailedClosed,
|
||||
failure_reason: Some(failure.reason()),
|
||||
..analytics_result
|
||||
},
|
||||
);
|
||||
GuardianAssessment {
|
||||
risk_level: GuardianRiskLevel::High,
|
||||
user_authorization: GuardianUserAuthorization::Unknown,
|
||||
outcome: GuardianAssessmentOutcome::Deny,
|
||||
rationale,
|
||||
}
|
||||
}
|
||||
GuardianReviewOutcome::TimedOut => {
|
||||
let rationale =
|
||||
"Automatic approval review timed out while evaluating the requested approval."
|
||||
.to_string();
|
||||
track_guardian_review(
|
||||
session.as_ref(),
|
||||
turn.as_ref(),
|
||||
&review_tracking,
|
||||
GuardianReviewAnalyticsResult {
|
||||
decision: GuardianReviewDecision::Denied,
|
||||
terminal_status: GuardianReviewTerminalStatus::TimedOut,
|
||||
failure_reason: Some(GuardianReviewFailureReason::Timeout),
|
||||
..analytics_result
|
||||
},
|
||||
);
|
||||
session
|
||||
.send_event(
|
||||
turn.as_ref(),
|
||||
@@ -215,6 +354,17 @@ async fn run_guardian_review(
|
||||
return ReviewDecision::TimedOut;
|
||||
}
|
||||
GuardianReviewOutcome::Aborted => {
|
||||
track_guardian_review(
|
||||
session.as_ref(),
|
||||
turn.as_ref(),
|
||||
&review_tracking,
|
||||
GuardianReviewAnalyticsResult {
|
||||
decision: GuardianReviewDecision::Aborted,
|
||||
terminal_status: GuardianReviewTerminalStatus::Aborted,
|
||||
failure_reason: Some(GuardianReviewFailureReason::Cancelled),
|
||||
..analytics_result
|
||||
},
|
||||
);
|
||||
session
|
||||
.send_event(
|
||||
turn.as_ref(),
|
||||
@@ -314,6 +464,7 @@ pub(crate) async fn review_approval_request(
|
||||
review_id,
|
||||
request,
|
||||
retry_reason,
|
||||
GuardianApprovalRequestSource::MainTurn,
|
||||
/*external_cancel*/ None,
|
||||
))
|
||||
.await
|
||||
@@ -325,16 +476,18 @@ pub(crate) async fn review_approval_request_with_cancel(
|
||||
review_id: String,
|
||||
request: GuardianApprovalRequest,
|
||||
retry_reason: Option<String>,
|
||||
approval_request_source: GuardianApprovalRequestSource,
|
||||
cancel_token: CancellationToken,
|
||||
) -> ReviewDecision {
|
||||
Box::pin(run_guardian_review(
|
||||
run_guardian_review(
|
||||
Arc::clone(session),
|
||||
Arc::clone(turn),
|
||||
review_id,
|
||||
request,
|
||||
retry_reason,
|
||||
approval_request_source,
|
||||
Some(cancel_token),
|
||||
))
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
@@ -361,6 +514,7 @@ pub(crate) fn spawn_approval_request_review(
|
||||
review_id,
|
||||
request,
|
||||
retry_reason,
|
||||
GuardianApprovalRequestSource::DelegatedSubagent,
|
||||
cancel_token,
|
||||
));
|
||||
let _ = tx.send(decision);
|
||||
@@ -389,11 +543,16 @@ pub(super) async fn run_guardian_review_session(
|
||||
retry_reason: Option<String>,
|
||||
schema: serde_json::Value,
|
||||
external_cancel: Option<CancellationToken>,
|
||||
) -> GuardianReviewOutcome {
|
||||
) -> (GuardianReviewOutcome, GuardianReviewAnalyticsResult) {
|
||||
let live_network_config = match session.services.network_proxy.as_ref() {
|
||||
Some(network_proxy) => match network_proxy.proxy().current_cfg().await {
|
||||
Ok(config) => Some(config),
|
||||
Err(err) => return GuardianReviewOutcome::Completed(Err(err)),
|
||||
Err(err) => {
|
||||
return (
|
||||
GuardianReviewOutcome::Failed(GuardianReviewError::prompt_build(err)),
|
||||
GuardianReviewAnalyticsResult::without_session(),
|
||||
);
|
||||
}
|
||||
},
|
||||
None => None,
|
||||
};
|
||||
@@ -443,10 +602,15 @@ pub(super) async fn run_guardian_review_session(
|
||||
);
|
||||
let guardian_config = match guardian_config {
|
||||
Ok(config) => config,
|
||||
Err(err) => return GuardianReviewOutcome::Completed(Err(err)),
|
||||
Err(err) => {
|
||||
return (
|
||||
GuardianReviewOutcome::Failed(GuardianReviewError::prompt_build(err)),
|
||||
GuardianReviewAnalyticsResult::without_session(),
|
||||
);
|
||||
}
|
||||
};
|
||||
|
||||
match Box::pin(
|
||||
let (session_outcome, session_analytics_result) = Box::pin(
|
||||
session
|
||||
.guardian_review_session
|
||||
.run_review(GuardianReviewSessionParams {
|
||||
@@ -463,17 +627,69 @@ pub(super) async fn run_guardian_review_session(
|
||||
external_cancel,
|
||||
}),
|
||||
)
|
||||
.await
|
||||
{
|
||||
GuardianReviewSessionOutcome::Completed(Ok(last_agent_message)) => {
|
||||
GuardianReviewOutcome::Completed(parse_guardian_assessment(
|
||||
last_agent_message.as_deref(),
|
||||
))
|
||||
.await;
|
||||
|
||||
match session_outcome {
|
||||
GuardianReviewSessionOutcome::Completed(Ok(last_agent_message)) => match last_agent_message
|
||||
{
|
||||
Some(last_agent_message) => {
|
||||
match parse_guardian_assessment(Some(&last_agent_message)) {
|
||||
Ok(assessment) => (
|
||||
GuardianReviewOutcome::Completed(assessment),
|
||||
session_analytics_result,
|
||||
),
|
||||
Err(err) => (
|
||||
GuardianReviewOutcome::Failed(GuardianReviewError::parse(err)),
|
||||
session_analytics_result,
|
||||
),
|
||||
}
|
||||
}
|
||||
None => (
|
||||
GuardianReviewOutcome::Failed(GuardianReviewError::session(anyhow::anyhow!(
|
||||
"guardian review completed without an assessment payload"
|
||||
))),
|
||||
session_analytics_result,
|
||||
),
|
||||
},
|
||||
GuardianReviewSessionOutcome::Completed(Err(err)) => (
|
||||
GuardianReviewOutcome::Failed(GuardianReviewError::session(err)),
|
||||
session_analytics_result,
|
||||
),
|
||||
GuardianReviewSessionOutcome::PromptBuildFailed(err) => (
|
||||
GuardianReviewOutcome::Failed(GuardianReviewError::prompt_build(err)),
|
||||
session_analytics_result,
|
||||
),
|
||||
GuardianReviewSessionOutcome::TimedOut => {
|
||||
(GuardianReviewOutcome::TimedOut, session_analytics_result)
|
||||
}
|
||||
GuardianReviewSessionOutcome::Completed(Err(err)) => {
|
||||
GuardianReviewOutcome::Completed(Err(err))
|
||||
GuardianReviewSessionOutcome::Aborted => {
|
||||
(GuardianReviewOutcome::Aborted, session_analytics_result)
|
||||
}
|
||||
GuardianReviewSessionOutcome::TimedOut => GuardianReviewOutcome::TimedOut,
|
||||
GuardianReviewSessionOutcome::Aborted => GuardianReviewOutcome::Aborted,
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod review_tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn guardian_review_error_reason_distinguishes_error_kinds() {
|
||||
let parse_error = GuardianReviewError::parse(anyhow::anyhow!("bad guardian JSON"));
|
||||
let prompt_error = GuardianReviewError::prompt_build(anyhow::anyhow!("bad prompt/config"));
|
||||
let session_error =
|
||||
GuardianReviewError::session(anyhow::anyhow!("guardian runtime failed"));
|
||||
|
||||
assert!(matches!(
|
||||
parse_error.reason(),
|
||||
GuardianReviewFailureReason::ParseError
|
||||
));
|
||||
assert!(matches!(
|
||||
prompt_error.reason(),
|
||||
GuardianReviewFailureReason::PromptBuildError
|
||||
));
|
||||
assert!(matches!(
|
||||
session_error.reason(),
|
||||
GuardianReviewFailureReason::SessionError
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -5,6 +5,8 @@ use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use anyhow::anyhow;
|
||||
use codex_analytics::GuardianReviewAnalyticsResult;
|
||||
use codex_analytics::GuardianReviewSessionKind;
|
||||
use codex_protocol::config_types::Personality;
|
||||
use codex_protocol::config_types::ReasoningSummary as ReasoningSummaryConfig;
|
||||
use codex_protocol::models::DeveloperInstructions;
|
||||
@@ -17,6 +19,7 @@ use codex_protocol::protocol::Op;
|
||||
use codex_protocol::protocol::RolloutItem;
|
||||
use codex_protocol::protocol::SandboxPolicy;
|
||||
use codex_protocol::protocol::SubAgentSource;
|
||||
use codex_protocol::protocol::TokenUsage;
|
||||
use serde_json::Value;
|
||||
use tokio::sync::Mutex;
|
||||
use tokio::sync::Semaphore;
|
||||
@@ -59,6 +62,7 @@ const GUARDIAN_FOLLOWUP_REVIEW_REMINDER: &str = concat!(
|
||||
#[derive(Debug)]
|
||||
pub(crate) enum GuardianReviewSessionOutcome {
|
||||
Completed(anyhow::Result<Option<String>>),
|
||||
PromptBuildFailed(anyhow::Error),
|
||||
TimedOut,
|
||||
Aborted,
|
||||
}
|
||||
@@ -102,6 +106,21 @@ struct GuardianReviewState {
|
||||
last_committed_fork_snapshot: Option<GuardianReviewForkSnapshot>,
|
||||
}
|
||||
|
||||
fn had_prior_review_context(prompt_mode: &GuardianPromptMode) -> bool {
|
||||
matches!(prompt_mode, GuardianPromptMode::Delta { .. })
|
||||
}
|
||||
|
||||
fn token_usage_delta(start: &TokenUsage, end: &TokenUsage) -> TokenUsage {
|
||||
TokenUsage {
|
||||
input_tokens: (end.input_tokens - start.input_tokens).max(0),
|
||||
cached_input_tokens: (end.cached_input_tokens - start.cached_input_tokens).max(0),
|
||||
output_tokens: (end.output_tokens - start.output_tokens).max(0),
|
||||
reasoning_output_tokens: (end.reasoning_output_tokens - start.reasoning_output_tokens)
|
||||
.max(0),
|
||||
total_tokens: (end.total_tokens - start.total_tokens).max(0),
|
||||
}
|
||||
}
|
||||
|
||||
struct EphemeralReviewCleanup {
|
||||
state: Arc<Mutex<GuardianReviewSessionState>>,
|
||||
review_session: Option<Arc<GuardianReviewSession>>,
|
||||
@@ -269,13 +288,14 @@ impl GuardianReviewSessionManager {
|
||||
clippy::await_holding_invalid_type,
|
||||
reason = "review session selection and trunk spawning must stay serialized"
|
||||
)]
|
||||
pub(crate) async fn run_review(
|
||||
pub(super) async fn run_review(
|
||||
&self,
|
||||
params: GuardianReviewSessionParams,
|
||||
) -> GuardianReviewSessionOutcome {
|
||||
) -> (GuardianReviewSessionOutcome, GuardianReviewAnalyticsResult) {
|
||||
let deadline = tokio::time::Instant::now() + GUARDIAN_REVIEW_TIMEOUT;
|
||||
let next_reuse_key = GuardianReviewSessionReuseKey::from_spawn_config(¶ms.spawn_config);
|
||||
let mut stale_trunk_to_shutdown = None;
|
||||
let mut spawned_trunk = false;
|
||||
let trunk_candidate = match run_before_review_deadline(
|
||||
deadline,
|
||||
params.external_cancel.as_ref(),
|
||||
@@ -309,16 +329,22 @@ impl GuardianReviewSessionManager {
|
||||
{
|
||||
Ok(Ok(review_session)) => Arc::new(review_session),
|
||||
Ok(Err(err)) => {
|
||||
return GuardianReviewSessionOutcome::Completed(Err(err));
|
||||
return (
|
||||
GuardianReviewSessionOutcome::PromptBuildFailed(err),
|
||||
GuardianReviewAnalyticsResult::without_session(),
|
||||
);
|
||||
}
|
||||
Err(outcome) => {
|
||||
return (outcome, GuardianReviewAnalyticsResult::without_session());
|
||||
}
|
||||
Err(outcome) => return outcome,
|
||||
};
|
||||
state.trunk = Some(Arc::clone(&review_session));
|
||||
spawned_trunk = true;
|
||||
}
|
||||
|
||||
state.trunk.as_ref().cloned()
|
||||
}
|
||||
Err(outcome) => return outcome,
|
||||
Err(outcome) => return (outcome, GuardianReviewAnalyticsResult::without_session()),
|
||||
};
|
||||
|
||||
if let Some(review_session) = stale_trunk_to_shutdown {
|
||||
@@ -326,9 +352,12 @@ impl GuardianReviewSessionManager {
|
||||
}
|
||||
|
||||
let Some(trunk) = trunk_candidate else {
|
||||
return GuardianReviewSessionOutcome::Completed(Err(anyhow!(
|
||||
"guardian review session was not available after spawn"
|
||||
)));
|
||||
return (
|
||||
GuardianReviewSessionOutcome::Completed(Err(anyhow!(
|
||||
"guardian review session was not available after spawn"
|
||||
))),
|
||||
GuardianReviewAnalyticsResult::without_session(),
|
||||
);
|
||||
};
|
||||
|
||||
if trunk.reuse_key != next_reuse_key {
|
||||
@@ -354,20 +383,30 @@ impl GuardianReviewSessionManager {
|
||||
}
|
||||
};
|
||||
|
||||
let (outcome, keep_review_session) =
|
||||
Box::pin(run_review_on_session(trunk.as_ref(), ¶ms, deadline)).await;
|
||||
let guardian_session_kind = if spawned_trunk {
|
||||
GuardianReviewSessionKind::TrunkNew
|
||||
} else {
|
||||
GuardianReviewSessionKind::TrunkReused
|
||||
};
|
||||
let (outcome, keep_review_session, analytics_result) = Box::pin(run_review_on_session(
|
||||
trunk.as_ref(),
|
||||
¶ms,
|
||||
guardian_session_kind,
|
||||
deadline,
|
||||
))
|
||||
.await;
|
||||
if keep_review_session && matches!(outcome, GuardianReviewSessionOutcome::Completed(_)) {
|
||||
trunk.refresh_last_committed_fork_snapshot().await;
|
||||
}
|
||||
drop(trunk_guard);
|
||||
|
||||
if keep_review_session {
|
||||
outcome
|
||||
(outcome, analytics_result)
|
||||
} else {
|
||||
if let Some(review_session) = self.remove_trunk_if_current(&trunk).await {
|
||||
review_session.shutdown_in_background();
|
||||
}
|
||||
outcome
|
||||
(outcome, analytics_result)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -464,7 +503,7 @@ impl GuardianReviewSessionManager {
|
||||
reuse_key: GuardianReviewSessionReuseKey,
|
||||
deadline: tokio::time::Instant,
|
||||
fork_snapshot: Option<GuardianReviewForkSnapshot>,
|
||||
) -> GuardianReviewSessionOutcome {
|
||||
) -> (GuardianReviewSessionOutcome, GuardianReviewAnalyticsResult) {
|
||||
let spawn_cancel_token = CancellationToken::new();
|
||||
let mut fork_config = params.spawn_config.clone();
|
||||
fork_config.ephemeral = true;
|
||||
@@ -483,17 +522,23 @@ impl GuardianReviewSessionManager {
|
||||
.await
|
||||
{
|
||||
Ok(Ok(review_session)) => Arc::new(review_session),
|
||||
Ok(Err(err)) => return GuardianReviewSessionOutcome::Completed(Err(err)),
|
||||
Err(outcome) => return outcome,
|
||||
Ok(Err(err)) => {
|
||||
return (
|
||||
GuardianReviewSessionOutcome::PromptBuildFailed(err),
|
||||
GuardianReviewAnalyticsResult::without_session(),
|
||||
);
|
||||
}
|
||||
Err(outcome) => return (outcome, GuardianReviewAnalyticsResult::without_session()),
|
||||
};
|
||||
self.register_active_ephemeral(Arc::clone(&review_session))
|
||||
.await;
|
||||
let mut cleanup =
|
||||
EphemeralReviewCleanup::new(Arc::clone(&self.state), Arc::clone(&review_session));
|
||||
|
||||
let (outcome, _) = Box::pin(run_review_on_session(
|
||||
let (outcome, _, analytics_result) = Box::pin(run_review_on_session(
|
||||
review_session.as_ref(),
|
||||
¶ms,
|
||||
GuardianReviewSessionKind::EphemeralForked,
|
||||
deadline,
|
||||
))
|
||||
.await;
|
||||
@@ -501,7 +546,7 @@ impl GuardianReviewSessionManager {
|
||||
cleanup.disarm();
|
||||
review_session.shutdown_in_background();
|
||||
}
|
||||
outcome
|
||||
(outcome, analytics_result)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -548,8 +593,13 @@ async fn spawn_guardian_review_session(
|
||||
async fn run_review_on_session(
|
||||
review_session: &GuardianReviewSession,
|
||||
params: &GuardianReviewSessionParams,
|
||||
guardian_session_kind: GuardianReviewSessionKind,
|
||||
deadline: tokio::time::Instant,
|
||||
) -> (GuardianReviewSessionOutcome, bool) {
|
||||
) -> (
|
||||
GuardianReviewSessionOutcome,
|
||||
bool,
|
||||
GuardianReviewAnalyticsResult,
|
||||
) {
|
||||
let (send_followup_reminder, prompt_mode) = {
|
||||
let state = review_session.state.lock().await;
|
||||
|
||||
@@ -564,6 +614,29 @@ async fn run_review_on_session(
|
||||
|
||||
(send_followup_reminder, prompt_mode)
|
||||
};
|
||||
let model_info = params
|
||||
.parent_session
|
||||
.services
|
||||
.models_manager
|
||||
.get_model_info(
|
||||
params.model.as_str(),
|
||||
¶ms.spawn_config.to_models_manager_config(),
|
||||
)
|
||||
.await;
|
||||
let guardian_reasoning_effort = if model_info.supports_reasoning_summaries {
|
||||
params
|
||||
.reasoning_effort
|
||||
.or(model_info.default_reasoning_level)
|
||||
} else {
|
||||
None
|
||||
};
|
||||
let mut analytics_result = GuardianReviewAnalyticsResult::from_session(
|
||||
review_session.codex.session.conversation_id.to_string(),
|
||||
guardian_session_kind,
|
||||
params.model.clone(),
|
||||
guardian_reasoning_effort.map(|effort| effort.to_string()),
|
||||
had_prior_review_context(&prompt_mode),
|
||||
);
|
||||
if send_followup_reminder {
|
||||
append_guardian_followup_reminder(review_session).await;
|
||||
}
|
||||
@@ -588,6 +661,13 @@ async fn run_review_on_session(
|
||||
prompt_mode,
|
||||
)
|
||||
.await?;
|
||||
let reviewed_action_truncated = prompt_items.reviewed_action_truncated;
|
||||
let token_usage_at_review_start = review_session
|
||||
.codex
|
||||
.session
|
||||
.total_token_usage()
|
||||
.await
|
||||
.unwrap_or_default();
|
||||
|
||||
review_session
|
||||
.codex
|
||||
@@ -607,29 +687,52 @@ async fn run_review_on_session(
|
||||
})
|
||||
.await?;
|
||||
|
||||
Ok::<GuardianTranscriptCursor, anyhow::Error>(prompt_items.transcript_cursor)
|
||||
Ok::<(GuardianTranscriptCursor, bool, TokenUsage), anyhow::Error>((
|
||||
prompt_items.transcript_cursor,
|
||||
reviewed_action_truncated,
|
||||
token_usage_at_review_start,
|
||||
))
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
let submit_result = match submit_result {
|
||||
Ok(submit_result) => submit_result,
|
||||
Err(outcome) => return (outcome, false),
|
||||
};
|
||||
let transcript_cursor = match submit_result {
|
||||
Ok(transcript_cursor) => transcript_cursor,
|
||||
Err(err) => {
|
||||
return (GuardianReviewSessionOutcome::Completed(Err(err)), false);
|
||||
}
|
||||
Err(outcome) => return (outcome, false, analytics_result),
|
||||
};
|
||||
let (transcript_cursor, reviewed_action_truncated, token_usage_at_review_start) =
|
||||
match submit_result {
|
||||
Ok(submit_result) => submit_result,
|
||||
Err(err) => {
|
||||
return (
|
||||
GuardianReviewSessionOutcome::PromptBuildFailed(err),
|
||||
false,
|
||||
analytics_result,
|
||||
);
|
||||
}
|
||||
};
|
||||
analytics_result.reviewed_action_truncated = reviewed_action_truncated;
|
||||
|
||||
let outcome =
|
||||
wait_for_guardian_review(review_session, deadline, params.external_cancel.as_ref()).await;
|
||||
let outcome = wait_for_guardian_review(
|
||||
review_session,
|
||||
deadline,
|
||||
params.external_cancel.as_ref(),
|
||||
&mut analytics_result,
|
||||
)
|
||||
.await;
|
||||
if matches!(outcome.0, GuardianReviewSessionOutcome::Completed(_)) {
|
||||
if outcome.2
|
||||
&& let Some(total_token_usage) = review_session.codex.session.total_token_usage().await
|
||||
{
|
||||
analytics_result.token_usage = Some(token_usage_delta(
|
||||
&token_usage_at_review_start,
|
||||
&total_token_usage,
|
||||
));
|
||||
}
|
||||
let mut state = review_session.state.lock().await;
|
||||
state.prior_review_count = state.prior_review_count.saturating_add(1);
|
||||
state.last_reviewed_transcript_cursor = Some(transcript_cursor);
|
||||
}
|
||||
outcome
|
||||
(outcome.0, outcome.1, analytics_result)
|
||||
}
|
||||
|
||||
async fn append_guardian_followup_reminder(review_session: &GuardianReviewSession) {
|
||||
@@ -658,7 +761,8 @@ async fn wait_for_guardian_review(
|
||||
review_session: &GuardianReviewSession,
|
||||
deadline: tokio::time::Instant,
|
||||
external_cancel: Option<&CancellationToken>,
|
||||
) -> (GuardianReviewSessionOutcome, bool) {
|
||||
analytics_result: &mut GuardianReviewAnalyticsResult,
|
||||
) -> (GuardianReviewSessionOutcome, bool, bool) {
|
||||
let timeout = tokio::time::sleep_until(deadline);
|
||||
tokio::pin!(timeout);
|
||||
let mut last_error_message: Option<String> = None;
|
||||
@@ -667,7 +771,7 @@ async fn wait_for_guardian_review(
|
||||
tokio::select! {
|
||||
_ = &mut timeout => {
|
||||
let keep_review_session = interrupt_and_drain_turn(&review_session.codex).await.is_ok();
|
||||
return (GuardianReviewSessionOutcome::TimedOut, keep_review_session);
|
||||
return (GuardianReviewSessionOutcome::TimedOut, keep_review_session, false);
|
||||
}
|
||||
_ = async {
|
||||
if let Some(cancel_token) = external_cancel {
|
||||
@@ -677,30 +781,35 @@ async fn wait_for_guardian_review(
|
||||
}
|
||||
} => {
|
||||
let keep_review_session = interrupt_and_drain_turn(&review_session.codex).await.is_ok();
|
||||
return (GuardianReviewSessionOutcome::Aborted, keep_review_session);
|
||||
return (GuardianReviewSessionOutcome::Aborted, keep_review_session, false);
|
||||
}
|
||||
event = review_session.codex.next_event() => {
|
||||
match event {
|
||||
Ok(event) => match event.msg {
|
||||
EventMsg::TurnComplete(turn_complete) => {
|
||||
analytics_result.time_to_first_token_ms = turn_complete
|
||||
.time_to_first_token_ms
|
||||
.and_then(|ms| u64::try_from(ms).ok());
|
||||
if turn_complete.last_agent_message.is_none()
|
||||
&& let Some(error_message) = last_error_message
|
||||
{
|
||||
return (
|
||||
GuardianReviewSessionOutcome::Completed(Err(anyhow!(error_message))),
|
||||
true,
|
||||
true,
|
||||
);
|
||||
}
|
||||
return (
|
||||
GuardianReviewSessionOutcome::Completed(Ok(turn_complete.last_agent_message)),
|
||||
true,
|
||||
true,
|
||||
);
|
||||
}
|
||||
EventMsg::Error(error) => {
|
||||
last_error_message = Some(error.message);
|
||||
}
|
||||
EventMsg::TurnAborted(_) => {
|
||||
return (GuardianReviewSessionOutcome::Aborted, true);
|
||||
return (GuardianReviewSessionOutcome::Aborted, true, false);
|
||||
}
|
||||
_ => {}
|
||||
},
|
||||
@@ -708,6 +817,7 @@ async fn wait_for_guardian_review(
|
||||
return (
|
||||
GuardianReviewSessionOutcome::Completed(Err(err.into())),
|
||||
false,
|
||||
false,
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -1005,4 +1115,44 @@ mod tests {
|
||||
assert_eq!(outcome.unwrap(), 42);
|
||||
assert!(!cancel_token.is_cancelled());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn had_prior_review_context_tracks_prompt_mode() {
|
||||
assert!(!had_prior_review_context(&GuardianPromptMode::Full));
|
||||
assert!(had_prior_review_context(&GuardianPromptMode::Delta {
|
||||
cursor: GuardianTranscriptCursor {
|
||||
parent_history_version: 7,
|
||||
transcript_entry_count: 42,
|
||||
}
|
||||
}));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn token_usage_delta_never_reports_negative_usage() {
|
||||
let start = TokenUsage {
|
||||
input_tokens: 10,
|
||||
cached_input_tokens: 8,
|
||||
output_tokens: 6,
|
||||
reasoning_output_tokens: 4,
|
||||
total_tokens: 28,
|
||||
};
|
||||
let end = TokenUsage {
|
||||
input_tokens: 15,
|
||||
cached_input_tokens: 7,
|
||||
output_tokens: 10,
|
||||
reasoning_output_tokens: 2,
|
||||
total_tokens: 34,
|
||||
};
|
||||
|
||||
assert_eq!(
|
||||
token_usage_delta(&start, &end),
|
||||
TokenUsage {
|
||||
input_tokens: 5,
|
||||
cached_input_tokens: 0,
|
||||
output_tokens: 4,
|
||||
reasoning_output_tokens: 0,
|
||||
total_tokens: 6,
|
||||
}
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -15,6 +15,7 @@ use crate::config_loader::Sourced;
|
||||
use crate::session::session::Session;
|
||||
use crate::session::turn_context::TurnContext;
|
||||
use crate::test_support;
|
||||
use codex_analytics::GuardianApprovalRequestSource;
|
||||
use codex_config::config_toml::ConfigToml;
|
||||
use codex_config::types::McpServerConfig;
|
||||
use codex_exec_server::LOCAL_FS;
|
||||
@@ -573,11 +574,12 @@ fn collect_guardian_transcript_entries_includes_recent_tool_calls_and_output() {
|
||||
fn guardian_truncate_text_keeps_prefix_suffix_and_xml_marker() {
|
||||
let content = "prefix ".repeat(200) + &" suffix".repeat(200);
|
||||
|
||||
let truncated = guardian_truncate_text(&content, /*token_cap*/ 20);
|
||||
let (truncated, was_truncated) = guardian_truncate_text(&content, /*token_cap*/ 20);
|
||||
|
||||
assert!(truncated.starts_with("prefix"));
|
||||
assert!(truncated.contains("<truncated omitted_approx_tokens=\""));
|
||||
assert!(truncated.ends_with("suffix"));
|
||||
assert!(was_truncated);
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -592,9 +594,27 @@ fn format_guardian_action_pretty_truncates_large_string_fields() -> serde_json::
|
||||
|
||||
let rendered = format_guardian_action_pretty(&action)?;
|
||||
|
||||
assert!(rendered.contains("\"tool\": \"apply_patch\""));
|
||||
assert!(rendered.contains("<truncated omitted_approx_tokens="));
|
||||
assert!(rendered.len() < patch.len());
|
||||
assert!(rendered.text.contains("\"tool\": \"apply_patch\""));
|
||||
assert!(rendered.text.contains("<truncated omitted_approx_tokens="));
|
||||
assert!(rendered.text.len() < patch.len());
|
||||
assert!(rendered.truncated);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn format_guardian_action_pretty_reports_no_truncation_for_small_payload() -> serde_json::Result<()>
|
||||
{
|
||||
let action = GuardianApprovalRequest::ApplyPatch {
|
||||
id: "patch-1".to_string(),
|
||||
cwd: test_path_buf("/tmp").abs(),
|
||||
files: Vec::new(),
|
||||
patch: "line\n".to_string(),
|
||||
};
|
||||
|
||||
let rendered = format_guardian_action_pretty(&action)?;
|
||||
|
||||
assert!(rendered.text.contains("\"tool\": \"apply_patch\""));
|
||||
assert!(!rendered.truncated);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -707,6 +727,7 @@ async fn cancelled_guardian_review_emits_terminal_abort_without_warning() {
|
||||
.to_string(),
|
||||
},
|
||||
/*retry_reason*/ None,
|
||||
GuardianApprovalRequestSource::MainTurn,
|
||||
cancel_token,
|
||||
)
|
||||
.await;
|
||||
@@ -941,12 +962,41 @@ async fn guardian_review_request_layout_matches_model_visible_request_snapshot()
|
||||
/*external_cancel*/ None,
|
||||
)
|
||||
.await;
|
||||
let GuardianReviewOutcome::Completed(Ok(assessment)) = outcome else {
|
||||
let (GuardianReviewOutcome::Completed(assessment), metadata) = outcome else {
|
||||
panic!("expected guardian assessment");
|
||||
};
|
||||
let guardian_thread_id = metadata
|
||||
.guardian_thread_id
|
||||
.as_deref()
|
||||
.expect("guardian thread id");
|
||||
assert_eq!(assessment.outcome, GuardianAssessmentOutcome::Allow);
|
||||
|
||||
assert_ne!(guardian_thread_id, session.conversation_id.to_string());
|
||||
ThreadId::from_string(guardian_thread_id).expect("guardian thread id should be a valid UUID");
|
||||
assert!(matches!(
|
||||
metadata.guardian_session_kind,
|
||||
Some(codex_analytics::GuardianReviewSessionKind::TrunkNew)
|
||||
));
|
||||
let request = request_log.single_request();
|
||||
let request_body = request.body_json();
|
||||
let request_model = request_body
|
||||
.get("model")
|
||||
.and_then(|value| value.as_str())
|
||||
.expect("guardian request should include a model");
|
||||
let request_reasoning_effort = request_body
|
||||
.get("reasoning")
|
||||
.and_then(|reasoning| reasoning.get("effort"))
|
||||
.and_then(|value| value.as_str());
|
||||
assert_eq!(metadata.guardian_model.as_deref(), Some(request_model));
|
||||
assert_eq!(
|
||||
metadata.guardian_reasoning_effort.as_deref(),
|
||||
request_reasoning_effort
|
||||
);
|
||||
assert_eq!(metadata.had_prior_review_context, Some(false));
|
||||
assert!(
|
||||
metadata.time_to_first_token_ms.is_some(),
|
||||
"guardian review metadata should capture TTFT when the nested turn completes"
|
||||
);
|
||||
|
||||
let mut settings = Settings::clone_current();
|
||||
settings.set_snapshot_path("snapshots");
|
||||
settings.set_prepend_module_to_snapshot(false);
|
||||
@@ -1150,18 +1200,63 @@ async fn guardian_reuses_prompt_cache_key_and_appends_prior_reviews() -> anyhow:
|
||||
)
|
||||
.await;
|
||||
|
||||
let GuardianReviewOutcome::Completed(Ok(first_assessment)) = first_outcome else {
|
||||
let (GuardianReviewOutcome::Completed(first_assessment), first_metadata) = first_outcome else {
|
||||
panic!("expected first guardian assessment");
|
||||
};
|
||||
let GuardianReviewOutcome::Completed(Ok(second_assessment)) = second_outcome else {
|
||||
let (GuardianReviewOutcome::Completed(second_assessment), second_metadata) = second_outcome
|
||||
else {
|
||||
panic!("expected second guardian assessment");
|
||||
};
|
||||
let GuardianReviewOutcome::Completed(Ok(third_assessment)) = third_outcome else {
|
||||
let (GuardianReviewOutcome::Completed(third_assessment), third_metadata) = third_outcome else {
|
||||
panic!("expected third guardian assessment");
|
||||
};
|
||||
assert_eq!(first_assessment.outcome, GuardianAssessmentOutcome::Allow);
|
||||
assert_eq!(second_assessment.outcome, GuardianAssessmentOutcome::Allow);
|
||||
assert_eq!(third_assessment.outcome, GuardianAssessmentOutcome::Allow);
|
||||
assert!(matches!(
|
||||
first_metadata.guardian_session_kind,
|
||||
Some(codex_analytics::GuardianReviewSessionKind::TrunkNew)
|
||||
));
|
||||
assert!(matches!(
|
||||
second_metadata.guardian_session_kind,
|
||||
Some(codex_analytics::GuardianReviewSessionKind::TrunkReused)
|
||||
));
|
||||
assert!(matches!(
|
||||
third_metadata.guardian_session_kind,
|
||||
Some(codex_analytics::GuardianReviewSessionKind::TrunkReused)
|
||||
));
|
||||
ThreadId::from_string(
|
||||
first_metadata
|
||||
.guardian_thread_id
|
||||
.as_deref()
|
||||
.expect("first guardian thread id"),
|
||||
)
|
||||
.expect("first guardian thread id should be a valid UUID");
|
||||
ThreadId::from_string(
|
||||
second_metadata
|
||||
.guardian_thread_id
|
||||
.as_deref()
|
||||
.expect("second guardian thread id"),
|
||||
)
|
||||
.expect("second guardian thread id should be a valid UUID");
|
||||
ThreadId::from_string(
|
||||
third_metadata
|
||||
.guardian_thread_id
|
||||
.as_deref()
|
||||
.expect("third guardian thread id"),
|
||||
)
|
||||
.expect("third guardian thread id should be a valid UUID");
|
||||
assert_eq!(first_metadata.had_prior_review_context, Some(false));
|
||||
assert_eq!(second_metadata.had_prior_review_context, Some(true));
|
||||
assert_eq!(third_metadata.had_prior_review_context, Some(true));
|
||||
assert_eq!(
|
||||
first_metadata.guardian_thread_id,
|
||||
second_metadata.guardian_thread_id
|
||||
);
|
||||
assert_eq!(
|
||||
second_metadata.guardian_thread_id,
|
||||
third_metadata.guardian_thread_id
|
||||
);
|
||||
|
||||
let requests = request_log.requests();
|
||||
assert_eq!(requests.len(), 3);
|
||||
|
||||
@@ -151,6 +151,7 @@ async fn record_initial_history_resumed_hydrates_previous_turn_settings_from_lif
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
},
|
||||
)),
|
||||
];
|
||||
@@ -218,6 +219,7 @@ async fn reconstruct_history_rollback_keeps_history_and_metadata_in_sync_for_com
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
},
|
||||
)),
|
||||
RolloutItem::EventMsg(EventMsg::TurnStarted(
|
||||
@@ -245,6 +247,7 @@ async fn reconstruct_history_rollback_keeps_history_and_metadata_in_sync_for_com
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
},
|
||||
)),
|
||||
RolloutItem::EventMsg(EventMsg::ThreadRolledBack(
|
||||
@@ -314,6 +317,7 @@ async fn reconstruct_history_rollback_keeps_history_and_metadata_in_sync_for_inc
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
},
|
||||
)),
|
||||
RolloutItem::EventMsg(EventMsg::TurnStarted(
|
||||
@@ -403,6 +407,7 @@ async fn reconstruct_history_rollback_skips_non_user_turns_for_history_and_metad
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
},
|
||||
)),
|
||||
RolloutItem::EventMsg(EventMsg::TurnStarted(
|
||||
@@ -429,6 +434,7 @@ async fn reconstruct_history_rollback_skips_non_user_turns_for_history_and_metad
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
},
|
||||
)),
|
||||
RolloutItem::EventMsg(EventMsg::TurnStarted(
|
||||
@@ -446,6 +452,7 @@ async fn reconstruct_history_rollback_skips_non_user_turns_for_history_and_metad
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
},
|
||||
)),
|
||||
RolloutItem::EventMsg(EventMsg::ThreadRolledBack(
|
||||
@@ -518,6 +525,7 @@ async fn reconstruct_history_rollback_counts_inter_agent_assistant_turns() {
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
},
|
||||
)),
|
||||
RolloutItem::EventMsg(EventMsg::TurnStarted(
|
||||
@@ -537,6 +545,7 @@ async fn reconstruct_history_rollback_counts_inter_agent_assistant_turns() {
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
},
|
||||
)),
|
||||
RolloutItem::EventMsg(EventMsg::ThreadRolledBack(
|
||||
@@ -604,6 +613,7 @@ async fn reconstruct_history_rollback_clears_history_and_metadata_when_exceeding
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
},
|
||||
)),
|
||||
RolloutItem::EventMsg(EventMsg::ThreadRolledBack(
|
||||
@@ -653,6 +663,7 @@ async fn record_initial_history_resumed_rollback_skips_only_user_turns() {
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
},
|
||||
)),
|
||||
// Standalone task turn (no UserMessage) should not consume rollback skips.
|
||||
@@ -670,6 +681,7 @@ async fn record_initial_history_resumed_rollback_skips_only_user_turns() {
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
},
|
||||
)),
|
||||
RolloutItem::EventMsg(EventMsg::ThreadRolledBack(
|
||||
@@ -723,6 +735,7 @@ async fn record_initial_history_resumed_rollback_drops_incomplete_user_turn_comp
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
},
|
||||
)),
|
||||
RolloutItem::EventMsg(EventMsg::TurnStarted(
|
||||
@@ -879,6 +892,7 @@ async fn reconstruct_history_legacy_compaction_without_replacement_history_clear
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
},
|
||||
)),
|
||||
];
|
||||
@@ -949,6 +963,7 @@ async fn record_initial_history_resumed_turn_context_after_compaction_reestablis
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
},
|
||||
)),
|
||||
];
|
||||
@@ -1052,6 +1067,7 @@ async fn record_initial_history_resumed_aborted_turn_without_id_clears_active_tu
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
},
|
||||
)),
|
||||
RolloutItem::EventMsg(EventMsg::TurnStarted(
|
||||
@@ -1160,6 +1176,7 @@ async fn record_initial_history_resumed_unmatched_abort_preserves_active_turn_fo
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
},
|
||||
)),
|
||||
RolloutItem::EventMsg(EventMsg::TurnStarted(
|
||||
@@ -1193,6 +1210,7 @@ async fn record_initial_history_resumed_unmatched_abort_preserves_active_turn_fo
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
},
|
||||
)),
|
||||
];
|
||||
@@ -1276,6 +1294,7 @@ async fn record_initial_history_resumed_trailing_incomplete_turn_compaction_clea
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
},
|
||||
)),
|
||||
RolloutItem::EventMsg(EventMsg::TurnStarted(
|
||||
@@ -1427,6 +1446,7 @@ async fn record_initial_history_resumed_replaced_incomplete_compacted_turn_clear
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
},
|
||||
)),
|
||||
RolloutItem::EventMsg(EventMsg::TurnStarted(
|
||||
|
||||
@@ -1778,6 +1778,7 @@ async fn record_initial_history_forked_hydrates_previous_turn_settings() {
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
},
|
||||
)),
|
||||
];
|
||||
@@ -1963,6 +1964,7 @@ async fn thread_rollback_recomputes_previous_turn_settings_and_reference_context
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
})),
|
||||
RolloutItem::EventMsg(EventMsg::TurnStarted(
|
||||
codex_protocol::protocol::TurnStartedEvent {
|
||||
@@ -1988,6 +1990,7 @@ async fn thread_rollback_recomputes_previous_turn_settings_and_reference_context
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
})),
|
||||
])
|
||||
.await;
|
||||
@@ -2065,6 +2068,7 @@ async fn thread_rollback_restores_cleared_reference_context_item_after_compactio
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
})),
|
||||
RolloutItem::EventMsg(EventMsg::TurnStarted(
|
||||
codex_protocol::protocol::TurnStartedEvent {
|
||||
@@ -2083,6 +2087,7 @@ async fn thread_rollback_restores_cleared_reference_context_item_after_compactio
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
})),
|
||||
RolloutItem::EventMsg(EventMsg::TurnStarted(
|
||||
codex_protocol::protocol::TurnStartedEvent {
|
||||
@@ -2110,6 +2115,7 @@ async fn thread_rollback_restores_cleared_reference_context_item_after_compactio
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
})),
|
||||
])
|
||||
.await;
|
||||
@@ -2156,6 +2162,7 @@ async fn thread_rollback_persists_marker_and_replays_cumulatively() {
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
})),
|
||||
RolloutItem::EventMsg(EventMsg::TurnStarted(
|
||||
codex_protocol::protocol::TurnStartedEvent {
|
||||
@@ -2179,6 +2186,7 @@ async fn thread_rollback_persists_marker_and_replays_cumulatively() {
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
})),
|
||||
RolloutItem::EventMsg(EventMsg::TurnStarted(
|
||||
codex_protocol::protocol::TurnStartedEvent {
|
||||
@@ -2202,6 +2210,7 @@ async fn thread_rollback_persists_marker_and_replays_cumulatively() {
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
})),
|
||||
])
|
||||
.await;
|
||||
@@ -5852,6 +5861,7 @@ async fn task_finish_emits_turn_item_lifecycle_for_leftover_pending_user_input()
|
||||
EventMsg::TurnComplete(TurnCompleteEvent {
|
||||
turn_id,
|
||||
last_agent_message: None,
|
||||
time_to_first_token_ms: None,
|
||||
..
|
||||
}) if turn_id == tc.sub_id
|
||||
));
|
||||
|
||||
@@ -556,11 +556,16 @@ impl Session {
|
||||
.turn_timing_state
|
||||
.completed_at_and_duration_ms()
|
||||
.await;
|
||||
let time_to_first_token_ms = turn_context
|
||||
.turn_timing_state
|
||||
.time_to_first_token_ms()
|
||||
.await;
|
||||
let event = EventMsg::TurnComplete(TurnCompleteEvent {
|
||||
turn_id: turn_context.sub_id.clone(),
|
||||
last_agent_message,
|
||||
completed_at,
|
||||
duration_ms,
|
||||
time_to_first_token_ms,
|
||||
});
|
||||
self.send_event(turn_context.as_ref(), event).await;
|
||||
|
||||
|
||||
@@ -1156,6 +1156,7 @@ async fn multi_agent_v2_list_agents_returns_completed_status_and_last_task_messa
|
||||
last_agent_message: Some("done".to_string()),
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
@@ -1634,6 +1635,7 @@ async fn multi_agent_v2_followup_task_completion_notifies_parent_on_every_turn()
|
||||
last_agent_message: Some("first done".to_string()),
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
@@ -1662,6 +1664,7 @@ async fn multi_agent_v2_followup_task_completion_notifies_parent_on_every_turn()
|
||||
last_agent_message: Some("second done".to_string()),
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
|
||||
@@ -74,6 +74,13 @@ impl TurnTimingState {
|
||||
(completed_at, duration_ms)
|
||||
}
|
||||
|
||||
pub(crate) async fn time_to_first_token_ms(&self) -> Option<i64> {
|
||||
let state = self.state.lock().await;
|
||||
state
|
||||
.time_to_first_token()
|
||||
.map(|duration| i64::try_from(duration.as_millis()).unwrap_or(i64::MAX))
|
||||
}
|
||||
|
||||
pub(crate) async fn record_ttft_for_response_event(
|
||||
&self,
|
||||
event: &ResponseEvent,
|
||||
@@ -102,14 +109,17 @@ fn now_unix_timestamp_secs() -> i64 {
|
||||
}
|
||||
|
||||
impl TurnTimingStateInner {
|
||||
fn time_to_first_token(&self) -> Option<Duration> {
|
||||
Some(self.first_token_at?.duration_since(self.started_at?))
|
||||
}
|
||||
|
||||
fn record_turn_ttft(&mut self) -> Option<Duration> {
|
||||
if self.first_token_at.is_some() {
|
||||
return None;
|
||||
}
|
||||
let started_at = self.started_at?;
|
||||
let first_token_at = Instant::now();
|
||||
self.first_token_at = Some(first_token_at);
|
||||
Some(first_token_at.duration_since(started_at))
|
||||
self.started_at?;
|
||||
self.first_token_at = Some(Instant::now());
|
||||
self.time_to_first_token()
|
||||
}
|
||||
|
||||
fn record_turn_ttfm(&mut self) -> Option<Duration> {
|
||||
|
||||
@@ -70,6 +70,7 @@ fn resume_history(
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
})),
|
||||
],
|
||||
rollout_path: rollout_path.to_path_buf(),
|
||||
|
||||
@@ -2098,6 +2098,10 @@ pub struct TurnCompleteEvent {
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
#[ts(type = "number | null", optional)]
|
||||
pub duration_ms: Option<i64>,
|
||||
/// Duration between turn start and the first model token in milliseconds, if known.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
#[ts(type = "number | null", optional)]
|
||||
pub time_to_first_token_ms: Option<i64>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS)]
|
||||
|
||||
@@ -785,6 +785,7 @@ fn append_terminal_turn_events(events: &mut Vec<Event>, turn: &Turn, include_fai
|
||||
last_agent_message: None,
|
||||
completed_at: turn.completed_at,
|
||||
duration_ms: turn.duration_ms,
|
||||
time_to_first_token_ms: None,
|
||||
}),
|
||||
}),
|
||||
TurnStatus::Interrupted => events.push(Event {
|
||||
@@ -816,6 +817,7 @@ fn append_terminal_turn_events(events: &mut Vec<Event>, turn: &Turn, include_fai
|
||||
last_agent_message: None,
|
||||
completed_at: turn.completed_at,
|
||||
duration_ms: turn.duration_ms,
|
||||
time_to_first_token_ms: None,
|
||||
}),
|
||||
});
|
||||
}
|
||||
|
||||
@@ -654,6 +654,7 @@ async fn unified_exec_wait_after_final_agent_message_snapshot() {
|
||||
last_agent_message: Some("Final response.".into()),
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
}),
|
||||
});
|
||||
|
||||
@@ -699,6 +700,7 @@ async fn unified_exec_wait_before_streamed_agent_message_snapshot() {
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
}),
|
||||
});
|
||||
|
||||
@@ -766,6 +768,7 @@ async fn unified_exec_waiting_multiple_empty_snapshots() {
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
}),
|
||||
});
|
||||
|
||||
@@ -846,6 +849,7 @@ async fn unified_exec_non_empty_then_empty_snapshots() {
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
}),
|
||||
});
|
||||
|
||||
@@ -1359,6 +1363,7 @@ async fn turn_complete_keeps_unified_exec_processes() {
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
}),
|
||||
});
|
||||
|
||||
|
||||
@@ -639,6 +639,7 @@ async fn plan_implementation_popup_skips_replayed_turn_complete() {
|
||||
last_agent_message: Some("Plan details".to_string()),
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
})]);
|
||||
|
||||
let popup = render_bottom_popup(&chat, /*width*/ 80);
|
||||
@@ -665,6 +666,7 @@ async fn plan_implementation_popup_shows_once_when_replay_precedes_live_turn_com
|
||||
last_agent_message: Some("Plan details".to_string()),
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
})]);
|
||||
let replay_popup = render_bottom_popup(&chat, /*width*/ 80);
|
||||
assert!(
|
||||
@@ -679,6 +681,7 @@ async fn plan_implementation_popup_shows_once_when_replay_precedes_live_turn_com
|
||||
last_agent_message: Some("Plan details".to_string()),
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
}),
|
||||
});
|
||||
|
||||
@@ -702,6 +705,7 @@ async fn plan_implementation_popup_shows_once_when_replay_precedes_live_turn_com
|
||||
last_agent_message: Some("Plan details".to_string()),
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
}),
|
||||
});
|
||||
let duplicate_popup = render_bottom_popup(&chat, /*width*/ 80);
|
||||
|
||||
@@ -314,6 +314,7 @@ async fn steer_rejection_queues_review_follow_up_before_existing_queued_messages
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
}),
|
||||
});
|
||||
|
||||
@@ -335,6 +336,7 @@ async fn steer_rejection_queues_review_follow_up_before_existing_queued_messages
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
}),
|
||||
});
|
||||
|
||||
|
||||
@@ -807,6 +807,7 @@ async fn slash_copy_state_tracks_turn_complete_final_reply() {
|
||||
last_agent_message: Some("Final reply **markdown**".to_string()),
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
}),
|
||||
});
|
||||
|
||||
@@ -839,6 +840,7 @@ async fn slash_copy_state_tracks_plan_item_completion() {
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
}),
|
||||
});
|
||||
|
||||
@@ -925,6 +927,7 @@ async fn slash_copy_state_is_preserved_during_running_task() {
|
||||
last_agent_message: Some("Previous completed reply".to_string()),
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
}),
|
||||
});
|
||||
chat.on_task_started();
|
||||
@@ -955,6 +958,7 @@ async fn slash_copy_tracks_replayed_legacy_agent_message_when_turn_complete_omit
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
}),
|
||||
});
|
||||
let _ = drain_insert_history(&mut rx);
|
||||
@@ -992,6 +996,7 @@ async fn slash_copy_uses_agent_message_item_when_turn_complete_omits_final_text(
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
}),
|
||||
});
|
||||
let _ = drain_insert_history(&mut rx);
|
||||
|
||||
@@ -1315,6 +1315,7 @@ async fn status_line_branch_refreshes_after_turn_complete() {
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
}),
|
||||
});
|
||||
|
||||
@@ -1665,6 +1666,7 @@ async fn multiple_agent_messages_in_single_turn_emit_multiple_headers() {
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
}),
|
||||
});
|
||||
|
||||
@@ -2610,6 +2612,7 @@ printf 'fenced within fenced\n'
|
||||
last_agent_message: None,
|
||||
completed_at: None,
|
||||
duration_ms: None,
|
||||
time_to_first_token_ms: None,
|
||||
}),
|
||||
});
|
||||
for lines in drain_insert_history(&mut rx) {
|
||||
|
||||
Reference in New Issue
Block a user