mirror of
https://github.com/openai/codex.git
synced 2026-10-21 05:22:21 +00:00
Attach failed Guardian reviews to diagnostic reports (#41936)
## What changed - Retain bounded, process-local records for failed Guardian reviews, including the reviewed action, decision, status, model, instructions, and reviewer history. - Add recent records from the reported thread and its descendants to log-enabled report uploads as `auto-review-failures.jsonl`. - Preserve the action and decision while omitting oversized reviewer context, and exclude successful, ephemeral, and capture-disabled reviews. ## Testing - Cover denied and invalid decisions, cleaned-up ephemeral reviewers, descendant-thread selection, record count and byte limits, and oversized-context fallback. GitOrigin-RevId: 5f12f7605041c3c84e826e6bac1cf06b1518ab24
This commit is contained in:
@@ -296,7 +296,7 @@ Example with notification opt-out:
|
||||
- `mcpServer/event/stream/stop` (experimental) — stop the caller's event subscription by `subscriptionId`.
|
||||
- `mcpServer/tool/call` — call a tool on a thread's configured MCP server by `threadId`, `server`, `tool`, optional `arguments`, and optional `_meta`, returning the MCP tool result. Parent-owned Multi-Agent V2 subagents reject direct tool calls.
|
||||
- `windowsSandbox/setupStart` — start Windows sandbox setup for the selected mode (`elevated` or `unelevated`); accepts an optional absolute `cwd` to target setup for a specific workspace, returns `{ started: true }` immediately, and later emits `windowsSandbox/setupCompleted`.
|
||||
- `feedback/upload` — submit a feedback report (classification + optional reason/logs, conversation_id, and optional `extraLogFiles` attachments array); returns the tracking thread id.
|
||||
- `feedback/upload` — submit a feedback report (classification + optional reason/logs, conversation_id, and optional `extraLogFiles` attachments array); returns the tracking thread id. With logs enabled, includes bounded recent failed Guardian review actions, decisions, and reviewer history from the reported thread and its descendants.
|
||||
- `config/read` — fetch the runtime-effective config after resolving config layering and managed requirements, including opaque `desktop` values stored in `config.toml`. When configured, the `packagedDefaults` layer has the lowest precedence.
|
||||
- `externalAgentConfig/detect` — detect migratable external-agent artifacts with `includeHome`, optional `cwds`, and an optional `migrationSource` selector. Omitted, `null`, or unrecognized migration-source values retain the default behavior. The deprecated optional `source` field remains accepted for compatibility but does not select the migration source. Each detected item includes `cwd` (`null` for home), and multi-item migrations may additionally include structured `details` with plugin ids, skill names, memory, session metadata, or other artifact names. The response also includes connector candidates inferred from detected source sessions, with a normalized display `name`, the number of detected sessions that used the connector, and the source metadata field used for detection.
|
||||
- `externalAgentConfig/import` — apply selected external-agent migration items by passing explicit `migrationItems` with `cwd` (`null` for home) and any `details` returned by detect. Pass the same optional `migrationSource` used for detection so the server reads from the matching source; omitted, `null`, or unrecognized values retain the default behavior. The optional `source` identifies the product that initiated the import, while the optional opaque `providerId` attributes analytics to the provider selected by that product without affecting migration-source selection. The response acknowledges the synchronous import phase with an `importId`. Expected migration failures are reported as per-item failures rather than JSON-RPC errors, so the server still returns that `importId` and emits `externalAgentConfig/import/completed` with the same ID once all synchronous and background work finishes. The completion notification contains type-level `itemTypeResults` with successes and failures, including raw failure messages for the client to report separately.
|
||||
|
||||
@@ -6,6 +6,7 @@ use codex_feedback::CODEX_APP_DIRECTORY_CACHE_ATTACHMENT_FILENAME;
|
||||
use codex_feedback::CODEX_APPS_TOOLS_CACHE_ATTACHMENT_FILENAME;
|
||||
#[cfg(target_os = "windows")]
|
||||
use codex_feedback::WINDOWS_SANDBOX_LOG_ATTACHMENT_FILENAME;
|
||||
use codex_feedback::guardian_review_failures_attachment;
|
||||
use codex_rollout::RolloutRecorder;
|
||||
use sha2::Digest;
|
||||
use sha2::Sha256;
|
||||
@@ -108,6 +109,7 @@ impl FeedbackRequestProcessor {
|
||||
}
|
||||
let snapshot = self.feedback.snapshot(conversation_id);
|
||||
let thread_id = snapshot.thread_id.clone();
|
||||
let mut extra_attachments = Vec::new();
|
||||
let (feedback_thread_ids, sqlite_feedback_logs, state_db_ctx) = if include_logs {
|
||||
if let Some(log_db) = self.log_db.as_ref() {
|
||||
log_db.flush().await;
|
||||
@@ -129,6 +131,7 @@ impl FeedbackRequestProcessor {
|
||||
},
|
||||
None => Vec::new(),
|
||||
};
|
||||
extra_attachments.extend(guardian_review_failures_attachment(&feedback_thread_ids));
|
||||
let mut feedback_thread_ids = feedback_thread_ids;
|
||||
let original_len = feedback_thread_ids.len();
|
||||
if let Some(conversation_id) = conversation_id {
|
||||
@@ -243,7 +246,6 @@ impl FeedbackRequestProcessor {
|
||||
}
|
||||
}
|
||||
|
||||
let mut extra_attachments = Vec::new();
|
||||
if include_logs {
|
||||
let doctor_cwd = feedback_cwd(
|
||||
&self.thread_manager,
|
||||
|
||||
@@ -0,0 +1,119 @@
|
||||
//! Captures bounded failed-review evidence for opt-in feedback uploads.
|
||||
//! Records outlive replaced and short-lived reviewers without writing extra history files.
|
||||
|
||||
use super::GuardianAssessmentOutcome;
|
||||
use super::approval_request::format_guardian_action_pretty;
|
||||
use super::prompt::parse_guardian_assessment;
|
||||
use super::review_session::GuardianReviewSessionOutcome;
|
||||
use super::review_session::GuardianReviewSessionParams;
|
||||
use crate::session::session::Session;
|
||||
use codex_feedback::record_guardian_review_failure;
|
||||
use codex_protocol::ThreadId;
|
||||
use codex_protocol::models::ResponseItem;
|
||||
use serde::Serialize;
|
||||
use std::io;
|
||||
use std::io::Write;
|
||||
|
||||
const MAX_RECORD_BYTES: usize = 4 * 1024 * 1024;
|
||||
|
||||
#[derive(Serialize)]
|
||||
struct ReviewFeedbackRecord<'a> {
|
||||
reviewed_thread_id: ThreadId,
|
||||
reviewed_turn_id: &'a str,
|
||||
reviewer_thread_id: ThreadId,
|
||||
model: &'a str,
|
||||
status: &'a str,
|
||||
decision: Option<&'a str>,
|
||||
action: &'a str,
|
||||
action_truncated: bool,
|
||||
instructions: Option<&'a str>,
|
||||
history: Vec<&'a ResponseItem>,
|
||||
context_omitted: bool,
|
||||
}
|
||||
|
||||
pub(super) async fn record_failed_review(
|
||||
reviewer: &Session,
|
||||
params: &GuardianReviewSessionParams,
|
||||
outcome: &GuardianReviewSessionOutcome,
|
||||
) {
|
||||
if !params.spawn_config.feedback_enabled || params.spawn_config.ephemeral {
|
||||
return;
|
||||
}
|
||||
let (status, decision) = match outcome {
|
||||
GuardianReviewSessionOutcome::Completed(Ok(Some(decision))) => {
|
||||
match parse_guardian_assessment(Some(decision)) {
|
||||
Ok(assessment) if assessment.outcome == GuardianAssessmentOutcome::Allow => {
|
||||
return;
|
||||
}
|
||||
Ok(_) => ("denied", Some(decision.as_str())),
|
||||
Err(_) => ("invalid_decision", Some(decision.as_str())),
|
||||
}
|
||||
}
|
||||
GuardianReviewSessionOutcome::Completed(Ok(None)) => ("missing_decision", None),
|
||||
GuardianReviewSessionOutcome::Completed(Err(_))
|
||||
| GuardianReviewSessionOutcome::PromptBuildFailed(_)
|
||||
| GuardianReviewSessionOutcome::SessionFailed { .. } => ("failed", None),
|
||||
GuardianReviewSessionOutcome::TimedOut => ("timed_out", None),
|
||||
GuardianReviewSessionOutcome::Aborted => ("aborted", None),
|
||||
};
|
||||
let Ok(action) = format_guardian_action_pretty(¶ms.request) else {
|
||||
tracing::warn!("Could not serialize Guardian feedback action");
|
||||
return;
|
||||
};
|
||||
let instructions = reviewer.get_prompt_base_instructions().await;
|
||||
let history = reviewer.clone_history().await;
|
||||
ReviewFeedbackRecord {
|
||||
reviewed_thread_id: params.parent_session.thread_id(),
|
||||
reviewed_turn_id: ¶ms.parent_context.turn().sub_id,
|
||||
reviewer_thread_id: reviewer.thread_id(),
|
||||
model: ¶ms.model,
|
||||
status,
|
||||
decision,
|
||||
action: &action.text,
|
||||
action_truncated: action.truncated,
|
||||
instructions: Some(&instructions.text),
|
||||
history: history.raw_items().collect(),
|
||||
context_omitted: false,
|
||||
}
|
||||
.store();
|
||||
}
|
||||
|
||||
impl ReviewFeedbackRecord<'_> {
|
||||
fn store(mut self) {
|
||||
let mut buffer = BoundedBuffer(Vec::new());
|
||||
if serde_json::to_writer(&mut buffer, &self).is_err() {
|
||||
// Preserve the action and decision even when optional reviewer history is too large.
|
||||
self.instructions = None;
|
||||
self.history.clear();
|
||||
self.context_omitted = true;
|
||||
buffer = BoundedBuffer(Vec::new());
|
||||
if serde_json::to_writer(&mut buffer, &self).is_err() {
|
||||
tracing::warn!("Guardian feedback action and decision exceed the record limit");
|
||||
return;
|
||||
}
|
||||
}
|
||||
record_guardian_review_failure(self.reviewed_thread_id, buffer.0);
|
||||
}
|
||||
}
|
||||
|
||||
struct BoundedBuffer(Vec<u8>);
|
||||
|
||||
impl Write for BoundedBuffer {
|
||||
fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
|
||||
if bytes.len() > MAX_RECORD_BYTES.saturating_sub(self.0.len()) {
|
||||
return Err(io::Error::other(
|
||||
"Guardian feedback record exceeds its size limit",
|
||||
));
|
||||
}
|
||||
self.0.extend_from_slice(bytes);
|
||||
Ok(bytes.len())
|
||||
}
|
||||
|
||||
fn flush(&mut self) -> io::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "feedback_tests.rs"]
|
||||
mod tests;
|
||||
@@ -0,0 +1,45 @@
|
||||
use super::*;
|
||||
use pretty_assertions::assert_eq;
|
||||
use serde_json::json;
|
||||
|
||||
#[test]
|
||||
fn oversized_context_keeps_the_action_and_decision() -> anyhow::Result<()> {
|
||||
let thread_id = ThreadId::new();
|
||||
let reviewer_thread_id = ThreadId::new();
|
||||
let decision = r#"{"outcome":"deny","rationale":"Missing approval."}"#;
|
||||
let action = r#"{"command":"git push"}"#;
|
||||
ReviewFeedbackRecord {
|
||||
reviewed_thread_id: thread_id,
|
||||
reviewed_turn_id: "parent-turn",
|
||||
reviewer_thread_id,
|
||||
model: "review-model",
|
||||
status: "denied",
|
||||
decision: Some(decision),
|
||||
action,
|
||||
action_truncated: false,
|
||||
instructions: Some(&"x".repeat(MAX_RECORD_BYTES)),
|
||||
history: Vec::new(),
|
||||
context_omitted: false,
|
||||
}
|
||||
.store();
|
||||
let contents = codex_feedback::guardian_review_failures_attachment(&[thread_id])
|
||||
.expect("failed-review record")
|
||||
.buffer;
|
||||
assert_eq!(
|
||||
serde_json::from_slice::<serde_json::Value>(&contents)?,
|
||||
json!({
|
||||
"reviewed_thread_id": thread_id,
|
||||
"reviewed_turn_id": "parent-turn",
|
||||
"reviewer_thread_id": reviewer_thread_id,
|
||||
"model": "review-model",
|
||||
"status": "denied",
|
||||
"decision": decision,
|
||||
"action": action,
|
||||
"action_truncated": false,
|
||||
"instructions": null,
|
||||
"history": [],
|
||||
"context_omitted": true,
|
||||
})
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
@@ -12,6 +12,7 @@
|
||||
//! 4. Apply the guardian's explicit allow/deny outcome.
|
||||
|
||||
mod approval_request;
|
||||
mod feedback;
|
||||
mod metrics;
|
||||
mod prompt;
|
||||
mod review;
|
||||
|
||||
@@ -79,6 +79,7 @@ use super::ApprovalRequestReasons;
|
||||
use super::GUARDIAN_REVIEWER_NAME;
|
||||
use super::GuardianApprovalRequest;
|
||||
use super::GuardianReviewContext;
|
||||
use super::feedback::record_failed_review;
|
||||
#[cfg(test)]
|
||||
use super::prompt::BUNDLED_GUARDIAN_POLICY;
|
||||
use super::prompt::BUNDLED_GUARDIAN_POLICY_TEMPLATE;
|
||||
@@ -652,6 +653,7 @@ impl GuardianReviewSessionManager {
|
||||
deadline,
|
||||
))
|
||||
.await;
|
||||
record_failed_review(&trunk.session, ¶ms, &outcome).await;
|
||||
if keep_review_session && matches!(outcome, GuardianReviewSessionOutcome::Completed(_)) {
|
||||
trunk.refresh_last_committed_fork_snapshot().await;
|
||||
}
|
||||
@@ -826,6 +828,7 @@ impl GuardianReviewSessionManager {
|
||||
deadline,
|
||||
))
|
||||
.await;
|
||||
record_failed_review(&review_session.session, ¶ms, &outcome).await;
|
||||
if let Some(review_session) = self.take_active_ephemeral(&review_session).await {
|
||||
cleanup.disarm();
|
||||
review_session.shutdown_in_background();
|
||||
|
||||
@@ -3349,15 +3349,14 @@ async fn guardian_ephemeral_retry_preserves_parallel_trunk_and_fork_history() ->
|
||||
{
|
||||
const TEST_STACK_SIZE_BYTES: usize = 4 * 1024 * 1024;
|
||||
|
||||
let handle =
|
||||
std::thread::Builder::new()
|
||||
.name("guardian_ephemeral_retry_preserves_parallel_trunk_and_fork_history".to_string())
|
||||
.stack_size(TEST_STACK_SIZE_BYTES)
|
||||
.spawn(|| -> anyhow::Result<()> {
|
||||
let runtime = tokio::runtime::Builder::new_current_thread()
|
||||
.enable_all()
|
||||
.build()?;
|
||||
runtime.block_on(Box::pin(async {
|
||||
let handle = std::thread::Builder::new()
|
||||
.name("guardian_ephemeral_retry_preserves_parallel_trunk_and_fork_history".to_string())
|
||||
.stack_size(TEST_STACK_SIZE_BYTES)
|
||||
.spawn(|| -> anyhow::Result<()> {
|
||||
let runtime = tokio::runtime::Builder::new_current_thread()
|
||||
.enable_all()
|
||||
.build()?;
|
||||
runtime.block_on(Box::pin(async {
|
||||
let first_assessment = serde_json::json!({
|
||||
"risk_level": "low",
|
||||
"user_authorization": "high",
|
||||
@@ -3421,7 +3420,11 @@ async fn guardian_ephemeral_retry_preserves_parallel_trunk_and_fork_history() ->
|
||||
])
|
||||
.await;
|
||||
|
||||
let (session, turn) = guardian_test_session_and_turn_with_base_url(server.uri()).await;
|
||||
let (mut session, turn) = guardian_test_session_and_turn_with_base_url(server.uri()).await;
|
||||
// Isolate feedback from other tests using the fixed parent session ID.
|
||||
Arc::get_mut(&mut session)
|
||||
.expect("session should be uniquely owned")
|
||||
.thread_id = ThreadId::new();
|
||||
turn.turn_metadata_state
|
||||
.set_parent_turn_id("upstream-parent-turn".to_string());
|
||||
turn.turn_metadata_state
|
||||
@@ -3616,11 +3619,30 @@ async fn guardian_ephemeral_retry_preserves_parallel_trunk_and_fork_history() ->
|
||||
.send(())
|
||||
.expect("second guardian review gate should still be open");
|
||||
assert_eq!(second_review.await?, ReviewDecision::Approved);
|
||||
let feedback = codex_feedback::guardian_review_failures_attachment(&[session.thread_id()])
|
||||
.expect("failed ephemeral review survives cleanup and subsequent allowed reviews");
|
||||
let record: serde_json::Value = serde_json::from_slice(&feedback.buffer)?;
|
||||
assert_eq!(
|
||||
serde_json::json!({
|
||||
"reviewer_thread_id": record["reviewer_thread_id"],
|
||||
"status": record["status"],
|
||||
"decision": record["decision"],
|
||||
"command": serde_json::from_str::<serde_json::Value>(
|
||||
record["action"].as_str().expect("reviewed action")
|
||||
)?["command"],
|
||||
}),
|
||||
serde_json::json!({
|
||||
"reviewer_thread_id": failed_ephemeral_request_body["client_metadata"]["thread_id"],
|
||||
"status": "invalid_decision",
|
||||
"decision": "not valid guardian json",
|
||||
"command": ["git", "push"],
|
||||
})
|
||||
);
|
||||
server.shutdown().await;
|
||||
|
||||
Ok(())
|
||||
}))
|
||||
})?;
|
||||
})?;
|
||||
|
||||
match handle.join() {
|
||||
Ok(result) => result,
|
||||
|
||||
@@ -27,8 +27,10 @@ use codex_login::CodexAuth;
|
||||
use codex_protocol::ThreadId;
|
||||
use codex_protocol::config_types::ApprovalsReviewer;
|
||||
use codex_protocol::config_types::WindowsSandboxLevel;
|
||||
use codex_protocol::models::ContentItem;
|
||||
use codex_protocol::models::PermissionProfile;
|
||||
use codex_protocol::models::PermissionProfileSnapshot;
|
||||
use codex_protocol::models::ResponseItem;
|
||||
use codex_protocol::openai_models::AutoReviewMessages;
|
||||
use codex_protocol::openai_models::MODEL_SPECIALTY_CYBER;
|
||||
use codex_protocol::openai_models::ModelsResponse;
|
||||
@@ -1324,6 +1326,15 @@ async fn guardian_denial_rejects_tool_call_with_rationale(
|
||||
),
|
||||
ev_completed("resp-parent-tool-denied"),
|
||||
]),
|
||||
sse(vec![
|
||||
ev_response_created("resp-guardian-inspect"),
|
||||
ev_function_call(
|
||||
"exec-guardian-inspect",
|
||||
"exec_command",
|
||||
&json!({"cmd": "printf guardian-inspection-evidence"}).to_string(),
|
||||
),
|
||||
ev_completed("resp-guardian-inspect"),
|
||||
]),
|
||||
sse(vec![
|
||||
ev_response_created("resp-guardian-denied"),
|
||||
ev_assistant_message(
|
||||
@@ -1374,6 +1385,67 @@ async fn guardian_denial_rejects_tool_call_with_rationale(
|
||||
assert!(guardian_request.body_contains_text(&command));
|
||||
assert_eq!(guardian_request.body_json()["model"], "gpt-5.6-luna");
|
||||
|
||||
let feedback =
|
||||
codex_feedback::guardian_review_failures_attachment(&[test.session_configured.thread_id])
|
||||
.expect("failed Guardian review");
|
||||
let record: serde_json::Value = serde_json::from_slice(&feedback.buffer)?;
|
||||
assert!(
|
||||
guardian_request.body_contains_text(record["action"].as_str().expect("reviewed action"))
|
||||
);
|
||||
let recorded_history: Vec<ResponseItem> = serde_json::from_value(record["history"].clone())?;
|
||||
let inspection_request = requests
|
||||
.iter()
|
||||
.find(|request| {
|
||||
request
|
||||
.function_call_output_text("exec-guardian-inspect")
|
||||
.is_some()
|
||||
})
|
||||
.expect("Guardian request following its inspection tool");
|
||||
assert!(
|
||||
inspection_request
|
||||
.function_call_output_text("exec-guardian-inspect")
|
||||
.expect("inspection output")
|
||||
.contains("guardian-inspection-evidence")
|
||||
);
|
||||
let inspection_output = inspection_request
|
||||
.input()
|
||||
.into_iter()
|
||||
.find(|item| {
|
||||
item["type"] == "function_call_output" && item["call_id"] == "exec-guardian-inspect"
|
||||
})
|
||||
.expect("inspection response item");
|
||||
assert!(recorded_history.contains(&serde_json::from_value(inspection_output)?));
|
||||
assert!(recorded_history.iter().any(|item| {
|
||||
matches!(item, ResponseItem::Message { role, content, .. }
|
||||
if role == "assistant" && content.iter().any(|part| {
|
||||
matches!(part, ContentItem::OutputText { text }
|
||||
if Some(text.as_str()) == record["decision"].as_str())
|
||||
}))
|
||||
}));
|
||||
assert_eq!(
|
||||
json!({
|
||||
"reviewed_thread_id": record["reviewed_thread_id"],
|
||||
"reviewer_thread_id": record["reviewer_thread_id"],
|
||||
"status": record["status"],
|
||||
"decision": serde_json::from_str::<serde_json::Value>(
|
||||
record["decision"].as_str().expect("raw Guardian decision"),
|
||||
)?,
|
||||
"context_omitted": record["context_omitted"],
|
||||
}),
|
||||
json!({
|
||||
"reviewed_thread_id": test.session_configured.thread_id,
|
||||
"reviewer_thread_id": guardian_request.body_json()["client_metadata"]["thread_id"],
|
||||
"status": "denied",
|
||||
"decision": {
|
||||
"risk_level": "high",
|
||||
"user_authorization": "low",
|
||||
"outcome": "deny",
|
||||
"rationale": "The requested write has unacceptable test risk.",
|
||||
},
|
||||
"context_omitted": false,
|
||||
})
|
||||
);
|
||||
|
||||
let tool_output = requests
|
||||
.iter()
|
||||
.find_map(|request| request.function_call_output_text("exec-call-denied"))
|
||||
|
||||
@@ -323,5 +323,34 @@ async fn guardian_subagent_review_preserves_late_root_user_authorization() -> Re
|
||||
assert!(guardian_transcript.contains("assistant: Agent message from /root"));
|
||||
assert!(guardian_transcript.contains(FORWARDED_AGENT_MESSAGE));
|
||||
|
||||
let feedback_thread_ids = test
|
||||
.thread_manager
|
||||
.list_agent_subtree_thread_ids(root_thread_id)
|
||||
.await?;
|
||||
let feedback = codex_feedback::guardian_review_failures_attachment(&feedback_thread_ids)
|
||||
.expect("failed worker review");
|
||||
let record: Value = serde_json::from_slice(&feedback.buffer)?;
|
||||
assert_eq!(
|
||||
json!({
|
||||
"reviewed_thread_id": record["reviewed_thread_id"],
|
||||
"reviewer_thread_id": record["reviewer_thread_id"],
|
||||
"status": record["status"],
|
||||
"decision": serde_json::from_str::<Value>(
|
||||
record["decision"].as_str().expect("raw Guardian decision"),
|
||||
)?,
|
||||
}),
|
||||
json!({
|
||||
"reviewed_thread_id": worker_thread_id,
|
||||
"reviewer_thread_id": guardian_review.single_request().body_json()["client_metadata"]["thread_id"],
|
||||
"status": "denied",
|
||||
"decision": {
|
||||
"risk_level": "high",
|
||||
"user_authorization": "high",
|
||||
"outcome": "deny",
|
||||
"rationale": "The agent message requests a different action.",
|
||||
},
|
||||
})
|
||||
);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -0,0 +1,84 @@
|
||||
//! Bounded process-local failed-review evidence, shared by root and child reviewers.
|
||||
//! Captures are exported only by a feedback request that includes logs.
|
||||
|
||||
use crate::FeedbackAttachment;
|
||||
use codex_protocol::ThreadId;
|
||||
use std::collections::HashSet;
|
||||
use std::collections::VecDeque;
|
||||
use std::sync::Mutex;
|
||||
|
||||
const MAX_RECORDS: usize = 64;
|
||||
const MAX_RECORDS_PER_THREAD: usize = 8;
|
||||
const MAX_BYTES: usize = 8 * 1024 * 1024;
|
||||
static RECORDS: Mutex<ReviewRecords> = Mutex::new(ReviewRecords {
|
||||
records: VecDeque::new(),
|
||||
bytes: 0,
|
||||
});
|
||||
|
||||
#[derive(Default)]
|
||||
struct ReviewRecords {
|
||||
records: VecDeque<(ThreadId, Vec<u8>)>,
|
||||
bytes: usize,
|
||||
}
|
||||
|
||||
impl ReviewRecords {
|
||||
fn push(&mut self, thread_id: ThreadId, record: Vec<u8>) {
|
||||
if record.len() + 1 > MAX_BYTES {
|
||||
return;
|
||||
}
|
||||
if self
|
||||
.records
|
||||
.iter()
|
||||
.filter(|(id, _)| *id == thread_id)
|
||||
.count()
|
||||
>= MAX_RECORDS_PER_THREAD
|
||||
&& let Some(index) = self.records.iter().position(|(id, _)| *id == thread_id)
|
||||
&& let Some((_, removed)) = self.records.remove(index)
|
||||
{
|
||||
self.bytes -= removed.len() + 1;
|
||||
}
|
||||
while self.records.len() >= MAX_RECORDS || self.bytes + record.len() + 1 > MAX_BYTES {
|
||||
if let Some((_, removed)) = self.records.pop_front() {
|
||||
self.bytes -= removed.len() + 1;
|
||||
}
|
||||
}
|
||||
self.bytes += record.len() + 1;
|
||||
self.records.push_back((thread_id, record));
|
||||
}
|
||||
|
||||
fn attachment(&self, thread_ids: &[ThreadId]) -> Option<FeedbackAttachment> {
|
||||
let thread_ids = thread_ids.iter().copied().collect::<HashSet<_>>();
|
||||
let mut buffer = Vec::new();
|
||||
for (thread_id, record) in &self.records {
|
||||
if thread_ids.contains(thread_id) {
|
||||
buffer.extend_from_slice(record);
|
||||
buffer.push(b'\n');
|
||||
}
|
||||
}
|
||||
(!buffer.is_empty()).then_some(FeedbackAttachment {
|
||||
filename: "auto-review-failures.jsonl".to_string(),
|
||||
buffer,
|
||||
content_type: Some("application/x-ndjson".to_string()),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
/// Retain one complete, serialized failed-review JSON record for later opt-in feedback.
|
||||
pub fn record_guardian_review_failure(thread_id: ThreadId, record: Vec<u8>) {
|
||||
RECORDS
|
||||
.lock()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner)
|
||||
.push(thread_id, record);
|
||||
}
|
||||
|
||||
/// Snapshot failures for the reported task tree without including other tasks' records.
|
||||
pub fn guardian_review_failures_attachment(thread_ids: &[ThreadId]) -> Option<FeedbackAttachment> {
|
||||
RECORDS
|
||||
.lock()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner)
|
||||
.attachment(thread_ids)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "guardian_tests.rs"]
|
||||
mod tests;
|
||||
@@ -0,0 +1,46 @@
|
||||
use super::*;
|
||||
use pretty_assertions::assert_eq;
|
||||
|
||||
#[test]
|
||||
fn records_keep_recent_failures_and_select_the_requested_tree() {
|
||||
let mut records = ReviewRecords::default();
|
||||
let root = ThreadId::new();
|
||||
let child = ThreadId::new();
|
||||
let unrelated = ThreadId::new();
|
||||
records.push(child, br#"{"child":"denied"}"#.to_vec());
|
||||
records.push(unrelated, br#"{"unrelated":"denied"}"#.to_vec());
|
||||
for index in 0..12 {
|
||||
records.push(root, format!(r#"{{"root":{index}}}"#).into_bytes());
|
||||
}
|
||||
let attachment = records
|
||||
.attachment(&[root, child])
|
||||
.expect("task-tree records");
|
||||
let expected = std::iter::once("{\"child\":\"denied\"}\n".to_string())
|
||||
.chain((4..12).map(|index| format!("{{\"root\":{index}}}\n")))
|
||||
.collect::<String>();
|
||||
assert_eq!(attachment.buffer, expected.into_bytes());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn record_count_and_bytes_are_bounded_across_threads() {
|
||||
let mut records = ReviewRecords::default();
|
||||
for _ in 0..MAX_RECORDS + 1 {
|
||||
records.push(ThreadId::new(), b"{}".to_vec());
|
||||
}
|
||||
assert_eq!(records.records.len(), MAX_RECORDS);
|
||||
let payload = vec![b' '; MAX_BYTES / 3];
|
||||
let thread_id = ThreadId::new();
|
||||
for _ in 0..4 {
|
||||
records.push(thread_id, payload.clone());
|
||||
}
|
||||
assert!(records.bytes <= MAX_BYTES);
|
||||
assert_eq!(records.records.len(), 2);
|
||||
assert_eq!(
|
||||
records
|
||||
.attachment(&[thread_id])
|
||||
.expect("bounded records")
|
||||
.buffer
|
||||
.len(),
|
||||
2 * (payload.len() + 1)
|
||||
);
|
||||
}
|
||||
@@ -32,11 +32,14 @@ use tracing_subscriber::registry::LookupSpan;
|
||||
|
||||
mod attachment_truncation;
|
||||
pub(crate) mod feedback_diagnostics;
|
||||
mod guardian;
|
||||
mod report_upload;
|
||||
mod upload;
|
||||
pub use feedback_diagnostics::FEEDBACK_DIAGNOSTICS_ATTACHMENT_FILENAME;
|
||||
pub use feedback_diagnostics::FeedbackDiagnostic;
|
||||
pub use feedback_diagnostics::FeedbackDiagnostics;
|
||||
pub use guardian::guardian_review_failures_attachment;
|
||||
pub use guardian::record_guardian_review_failure;
|
||||
pub use report_upload::FeedbackDelivery;
|
||||
pub use report_upload::FeedbackTransport;
|
||||
pub use report_upload::prepare_report_attachment;
|
||||
|
||||
Reference in New Issue
Block a user