From bd334bab4a525ef24c16846189bb95c82a757c0b Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 20 May 2026 05:25:39 +0000 Subject: [PATCH] fix code queue mgr trace summary contract --- .../code-queue-mgr/src-rs/main.rs | 280 +++++++++++++++++- 1 file changed, 267 insertions(+), 13 deletions(-) diff --git a/src/components/microservices/code-queue-mgr/src-rs/main.rs b/src/components/microservices/code-queue-mgr/src-rs/main.rs index ce52aac9..dbf24581 100644 --- a/src/components/microservices/code-queue-mgr/src-rs/main.rs +++ b/src/components/microservices/code-queue-mgr/src-rs/main.rs @@ -3,7 +3,7 @@ use chrono::{DateTime, SecondsFormat, Utc}; use postgres::{Client, NoTls}; use serde_json::{json, Map, Value}; -use std::collections::{HashSet, VecDeque}; +use std::collections::{BTreeSet, HashSet, VecDeque}; use std::env; use std::error::Error; use std::fs::{create_dir_all, OpenOptions}; @@ -641,6 +641,139 @@ fn output_max_seq(task: &TaskMeta) -> i64 { task.last_output_seq.max(from_output) } +fn trace_countable_outputs(task: &TaskMeta) -> Vec { + output_array(task) + .into_iter() + .filter(|item| !(item.get("channel").and_then(Value::as_str) == Some("system") && item.get("method").and_then(Value::as_str) == Some("enqueue"))) + .collect() +} + +fn configured_step_count(task: &TaskMeta) -> Option { + task.task_json.as_ref().and_then(|value| value.get("stepCount")).and_then(Value::as_i64) +} + +fn raw_trace_step_count(task: &TaskMeta) -> i64 { + let output_count = trace_countable_outputs(task).len() as i64; + let output_count = if output_count > 0 { output_count } else { task.output_count }; + configured_step_count(task).unwrap_or(output_count).max(output_count) +} + +fn trace_stats_contract(scope_id: &str, step_count: i64, output_max_seq: i64) -> Value { + json!({ + "scopeId": scope_id, + "source": "oa-event-flow", + "sourceHint": "raw-trace-fallback", + "stepCount": step_count, + "llmStepCount": 0, + "toolCallCount": step_count, + "editedFileCount": 0, + "commandCount": 0, + "firstSeq": if step_count > 0 { json!(1) } else { Value::Null }, + "lastSeq": if output_max_seq > 0 { json!(output_max_seq) } else { Value::Null }, + "computedAt": now_iso(), + "incomplete": true + }) +} + +fn trace_execution_contract(scope_id: &str, step_count: i64, output_max_seq: i64) -> Value { + let trace_stats = if step_count > 0 || output_max_seq > 0 { + trace_stats_contract(scope_id, step_count, output_max_seq) + } else { + Value::Null + }; + json!({ + "stepCount": step_count, + "llmStepCount": 0, + "toolCallCount": step_count, + "editedFileCount": 0, + "commandCount": 0, + "traceStats": trace_stats, + "statsSource": if step_count > 0 || output_max_seq > 0 { "raw-trace-fallback" } else { "raw-trace-empty" }, + "traceStatsState": if step_count > 0 || output_max_seq > 0 { "degraded" } else { "empty" }, + "traceStatsReason": if step_count > 0 || output_max_seq > 0 { "oa-event-flow-stats-unavailable-raw-trace-present" } else { "no-trace-steps-yet" }, + "statsUnavailable": true, + "statsSyncing": step_count > 0 || output_max_seq > 0, + "rawTraceStepCount": step_count + }) +} + +fn attempt_index_from_value(value: &Value) -> Option { + value + .get("index") + .or_else(|| value.get("attempt")) + .or_else(|| value.get("attemptIndex")) + .and_then(Value::as_i64) + .filter(|index| *index > 0) +} + +fn output_attempt_index(task: &TaskMeta, item: &Value) -> i64 { + if let Some(index) = item + .get("attempt") + .or_else(|| item.get("attemptIndex")) + .and_then(Value::as_i64) + .filter(|index| *index > 0) + { + return index; + } + let seq = seq_from_output(item); + for attempt in attempts_array(task) { + let Some(index) = attempt_index_from_value(&attempt) else { continue }; + let start = attempt.get("outputStartSeq").and_then(Value::as_i64).unwrap_or(i64::MIN); + let end = attempt.get("outputEndSeq").and_then(Value::as_i64).unwrap_or(i64::MAX); + if seq >= start && seq <= end { + return index; + } + } + task.current_attempt.max(1) +} + +fn attempt_indexes_for_trace(task: &TaskMeta, outputs: &[Value]) -> Vec { + let mut indexes = BTreeSet::new(); + for attempt in attempts_array(task) { + if let Some(index) = attempt_index_from_value(&attempt) { + indexes.insert(index); + } + } + if task.current_attempt > 0 { + indexes.insert(task.current_attempt); + } + for item in outputs { + indexes.insert(output_attempt_index(task, item)); + } + indexes.into_iter().collect() +} + +fn attempt_trace_window(task: &TaskMeta, outputs: &[Value], index: i64) -> Value { + let existing = attempts_array(task) + .into_iter() + .find(|attempt| attempt_index_from_value(attempt) == Some(index)); + let attempt_outputs: Vec<&Value> = outputs.iter().filter(|item| output_attempt_index(task, item) == index).collect(); + let step_count = attempt_outputs.len() as i64; + let output_max = attempt_outputs.iter().map(|item| seq_from_output(item)).max().unwrap_or(0); + let scope_id = format!("task:{}:attempt:{}", task.id, index); + let mut record = existing.and_then(|value| value.as_object().cloned()).unwrap_or_else(Map::new); + record.insert("index".to_string(), json!(index)); + record.entry("mode".to_string()).or_insert_with(|| { + if index == task.current_attempt { + task.current_mode.as_ref().map(|value| json!(value)).unwrap_or(Value::Null) + } else { + Value::Null + } + }); + record.insert("stepCount".to_string(), json!(step_count)); + record.insert("retainedStepCount".to_string(), json!(step_count)); + record.insert("outputMaxSeq".to_string(), json!(output_max)); + record.insert("traceStats".to_string(), if step_count > 0 { trace_stats_contract(&scope_id, step_count, output_max) } else { Value::Null }); + record.insert("statsSource".to_string(), json!(if step_count > 0 { "raw-trace-fallback" } else { "raw-trace-empty" })); + record.insert("traceStatsState".to_string(), json!(if step_count > 0 { "degraded" } else { "empty" })); + record.insert("traceStatsReason".to_string(), json!(if step_count > 0 { "oa-event-flow-stats-unavailable-raw-trace-present" } else { "no-trace-steps-yet" })); + record.insert("statsUnavailable".to_string(), json!(true)); + record.insert("statsSyncing".to_string(), json!(step_count > 0)); + record.insert("rawTraceStepCount".to_string(), json!(step_count)); + record.insert("execution".to_string(), trace_execution_contract(&scope_id, step_count, output_max)); + Value::Object(record) +} + fn task_scheduler_heartbeat(task: &TaskMeta) -> Option { task.task_json .as_ref() @@ -912,6 +1045,9 @@ fn task_list_response(_state: &AppState, task: &TaskMeta, lite: bool) -> Value { let final_text = final_response(task); let agent_port = code_agent_port(&task.model); let preview_limit = if lite { 360 } else { 2000 }; + let raw_step_count = raw_trace_step_count(task); + let execution = trace_execution_contract(&format!("task:{}", task.id), raw_step_count, output_max_seq(task)); + let trace_stats = execution.get("traceStats").cloned().unwrap_or(Value::Null); json!({ "id": task.id, "queueId": task.queue_id, @@ -927,10 +1063,16 @@ fn task_list_response(_state: &AppState, task: &TaskMeta, lite: bool) -> Value { "displayPromptChars": display_prompt.chars().count(), "promptEditable": queued_prompt_editable(task), "finalResponseChars": final_text.chars().count(), - "stepCount": task.task_json.as_ref().and_then(|value| value.get("stepCount")).and_then(Value::as_i64).unwrap_or(task.last_output_seq), + "stepCount": raw_step_count, "llmStepCount": task.task_json.as_ref().and_then(|value| value.get("llmStepCount")).and_then(Value::as_i64).unwrap_or(task.last_output_seq), - "traceStats": Value::Null, - "statsSource": "code-queue-mgr-rust", + "traceStats": trace_stats, + "statsSource": execution.get("statsSource").cloned().unwrap_or_else(|| json!("raw-trace-empty")), + "traceStatsState": execution.get("traceStatsState").cloned().unwrap_or_else(|| json!("empty")), + "traceStatsReason": execution.get("traceStatsReason").cloned().unwrap_or_else(|| json!("no-trace-steps-yet")), + "statsUnavailable": true, + "statsSyncing": raw_step_count > 0, + "rawTraceStepCount": raw_step_count, + "execution": execution, "summaryOnly": true, "referenceTaskIds": task.reference_task_ids, "referenceInjection": task.reference_injection, @@ -986,6 +1128,14 @@ fn task_meta_response(state: &AppState, task: &TaskMeta) -> Value { } else { task.base_prompt.clone() }; + let trace_outputs = trace_countable_outputs(task); + let raw_step_count = raw_trace_step_count(task); + let execution = trace_execution_contract(&format!("task:{}", task.id), raw_step_count, output_max_seq(task)); + let trace_stats = execution.get("traceStats").cloned().unwrap_or(Value::Null); + let attempts = attempt_indexes_for_trace(task, &trace_outputs) + .into_iter() + .map(|index| attempt_trace_window(task, &trace_outputs, index)) + .collect::>(); map.insert("prompt".to_string(), json!(task.prompt)); map.insert("basePrompt".to_string(), json!(task.base_prompt)); map.insert("displayPrompt".to_string(), json!(display_prompt)); @@ -995,7 +1145,7 @@ fn task_meta_response(state: &AppState, task: &TaskMeta) -> Value { "promptHistory".to_string(), task.task_json.as_ref().and_then(|value| value.get("promptHistory")).cloned().unwrap_or_else(|| json!([])), ); - map.insert("attempts".to_string(), json!(attempts_array(task))); + map.insert("attempts".to_string(), json!(attempts)); map.insert( "nextMode".to_string(), task.task_json.as_ref().and_then(|value| value.get("nextMode")).cloned().unwrap_or(Value::Null), @@ -1009,6 +1159,16 @@ fn task_meta_response(state: &AppState, task: &TaskMeta) -> Value { map.insert("output".to_string(), json!([])); map.insert("events".to_string(), json!([])); map.insert("summaryOnly".to_string(), json!(false)); + map.insert("stepCount".to_string(), json!(raw_step_count)); + map.insert("llmStepCount".to_string(), json!(task.task_json.as_ref().and_then(|value| value.get("llmStepCount")).and_then(Value::as_i64).unwrap_or(task.last_output_seq))); + map.insert("traceStats".to_string(), trace_stats); + map.insert("statsSource".to_string(), execution.get("statsSource").cloned().unwrap_or_else(|| json!("raw-trace-empty"))); + map.insert("traceStatsState".to_string(), execution.get("traceStatsState").cloned().unwrap_or_else(|| json!("empty"))); + map.insert("traceStatsReason".to_string(), execution.get("traceStatsReason").cloned().unwrap_or_else(|| json!("no-trace-steps-yet"))); + map.insert("statsUnavailable".to_string(), json!(true)); + map.insert("statsSyncing".to_string(), json!(raw_step_count > 0)); + map.insert("rawTraceStepCount".to_string(), json!(raw_step_count)); + map.insert("execution".to_string(), execution); base } @@ -2010,6 +2170,7 @@ fn trace_step_detail(task: &TaskMeta, url: &RequestUrl) -> Value { fn task_summary(task: &TaskMeta) -> Value { let outputs = output_array(task); + let trace_outputs = trace_countable_outputs(task); let tool_outputs: Vec = outputs .iter() .filter(|item| matches!(item.get("channel").and_then(Value::as_str), Some("command" | "tool" | "diff" | "error"))) @@ -2035,6 +2196,13 @@ fn task_summary(task: &TaskMeta) -> Value { }) .collect::>(); let agent_port = code_agent_port(&task.model); + let raw_step_count = raw_trace_step_count(task); + let execution = trace_execution_contract(&format!("task:{}", task.id), raw_step_count, output_max_seq(task)); + let trace_stats = execution.get("traceStats").cloned().unwrap_or(Value::Null); + let attempts = attempt_indexes_for_trace(task, &trace_outputs) + .into_iter() + .map(|index| attempt_trace_window(task, &trace_outputs, index)) + .collect::>(); json!({ "id": task.id, "queueId": task.queue_id, @@ -2068,7 +2236,7 @@ fn task_summary(task: &TaskMeta) -> Value { "referenceInjection": task.reference_injection, "lastAssistantMessage": last_assistant_message(task), "toolSummary": { "count": tool_outputs.len(), "returned": tool_outputs.len().saturating_sub(start), "limit": 160, "truncated": start > 0, "items": tool_items }, - "attempts": attempts_array(task), + "attempts": attempts, "lastJudge": task.last_judge, "lastError": task.last_error, "cancelRequested": task.task_json.as_ref().and_then(|value| value.get("cancelRequested")).and_then(Value::as_bool).unwrap_or(false), @@ -2077,34 +2245,66 @@ fn task_summary(task: &TaskMeta) -> Value { "transcriptCount": outputs.len() + json_array_len(task.task_json.as_ref(), "promptHistory", 0) as usize, "transcriptMaxSeq": output_max_seq(task), "finalResponse": final_response(task), - "stepCount": task.task_json.as_ref().and_then(|value| value.get("stepCount")).and_then(Value::as_i64).unwrap_or(task.last_output_seq), + "stepCount": raw_step_count, "llmStepCount": task.task_json.as_ref().and_then(|value| value.get("llmStepCount")).and_then(Value::as_i64).unwrap_or(task.last_output_seq), - "statsSource": "code-queue-mgr-rust-postgres" + "traceStats": trace_stats, + "statsSource": execution.get("statsSource").cloned().unwrap_or_else(|| json!("raw-trace-empty")), + "traceStatsState": execution.get("traceStatsState").cloned().unwrap_or_else(|| json!("empty")), + "traceStatsReason": execution.get("traceStatsReason").cloned().unwrap_or_else(|| json!("no-trace-steps-yet")), + "statsUnavailable": true, + "statsSyncing": raw_step_count > 0, + "rawTraceStepCount": raw_step_count, + "execution": execution }) } fn trace_summary(task: &TaskMeta) -> Value { - let outputs = output_array(task); + let outputs = trace_countable_outputs(task); + let step_count = raw_trace_step_count(task); + let output_max_seq = output_max_seq(task); + let execution = trace_execution_contract(&format!("task:{}", task.id), step_count, output_max_seq); + let trace_stats = execution.get("traceStats").cloned().unwrap_or(Value::Null); + let attempts = attempt_indexes_for_trace(task, &outputs) + .into_iter() + .map(|index| attempt_trace_window(task, &outputs, index)) + .collect::>(); json!({ + "id": task.id, "taskId": task.id, "queueId": task.queue_id, "status": task.status, "providerId": task.provider_id, "executionMode": task.execution_mode, + "executionModeInfo": execution_mode_info(&task.execution_mode), "model": task.model, - "stepCount": task.task_json.as_ref().and_then(|value| value.get("stepCount")).and_then(Value::as_i64).unwrap_or(outputs.len() as i64), + "cwd": task.cwd, + "reasoningEffort": task.reasoning_effort, + "stepCount": step_count, + "llmStepCount": 0, + "toolCallCount": step_count, + "editedFileCount": 0, + "commandCount": 0, "retainedStepCount": outputs.len(), - "outputMaxSeq": output_max_seq(task), + "outputMaxSeq": output_max_seq, "schedulerHeartbeat": task_scheduler_heartbeat(task).unwrap_or(Value::Null), - "statsSource": "code-queue-mgr-rust-postgres", - "attempts": attempts_array(task), + "traceStats": trace_stats, + "statsSource": execution.get("statsSource").cloned().unwrap_or_else(|| json!("raw-trace-empty")), + "traceStatsState": execution.get("traceStatsState").cloned().unwrap_or_else(|| json!("empty")), + "traceStatsReason": execution.get("traceStatsReason").cloned().unwrap_or_else(|| json!("no-trace-steps-yet")), + "statsUnavailable": true, + "statsSyncing": step_count > 0, + "rawTraceStepCount": step_count, + "execution": execution, + "attempts": attempts, "prompt": { "initialPreview": preview(&task.prompt, 1600), "basePreview": preview(&task.base_prompt, 1600), "chars": task.prompt.chars().count(), "lines": task.prompt.lines().count() }, "lastAssistantMessage": last_assistant_message(task), + "createdAt": task.created_at, "updatedAt": task.updated_at, "finalResponse": final_response(task), "lastJudge": task.last_judge, "lastError": task.last_error, "currentAttempt": task.current_attempt, + "currentMode": task.current_mode, "maxAttempts": task.max_attempts, "startedAt": task.started_at, "finishedAt": task.finished_at, @@ -3038,4 +3238,58 @@ mod tests { Some("mgr-visible") ); } + + #[test] + fn trace_summary_contract_exposes_retry_window_and_raw_fallback() { + let mut task = test_task("trace summary retry contract".to_string()); + task.id = "codex_trace_contract_rust".to_string(); + task.status = "running".to_string(); + task.current_attempt = 2; + task.current_mode = Some("retry".to_string()); + task.max_attempts = 99; + task.attempt_count = 1; + task.last_output_seq = 21; + task.task_json = Some(json!({ + "status": "running", + "stepCount": Value::Null, + "llmStepCount": Value::Null, + "attempts": [ + { + "index": 1, + "mode": "initial", + "startedAt": "2026-05-19T00:00:10.000Z", + "finishedAt": "2026-05-19T00:02:00.000Z", + "judge": { "decision": "retry", "reason": "try again" } + } + ], + "output": [ + { "seq": 1, "at": "2026-05-19T00:00:00.000Z", "channel": "system", "method": "enqueue", "text": "queued" }, + { "seq": 20, "attempt": 2, "at": "2026-05-19T00:06:00.000Z", "channel": "command", "method": "item/started", "text": "attempt 2 / 99\npnpm test" }, + { "seq": 21, "attempt": 2, "at": "2026-05-19T00:06:20.000Z", "channel": "tool", "method": "item/completed", "text": "src/components/microservices/code-queue-mgr/src-rs/main.rs" } + ], + "events": [], + "promptHistory": [] + })); + + let summary = trace_summary(&task); + assert_eq!(summary.get("currentAttempt").and_then(Value::as_i64), Some(2)); + assert_eq!(summary.get("statsSource").and_then(Value::as_str), Some("raw-trace-fallback")); + assert_eq!(summary.get("traceStatsState").and_then(Value::as_str), Some("degraded")); + assert_eq!( + summary.get("traceStatsReason").and_then(Value::as_str), + Some("oa-event-flow-stats-unavailable-raw-trace-present") + ); + assert_eq!(summary.get("statsUnavailable").and_then(Value::as_bool), Some(true)); + assert_eq!(summary.get("statsSyncing").and_then(Value::as_bool), Some(true)); + assert!(summary.get("stepCount").and_then(Value::as_i64).unwrap_or(0) > 0); + + let attempts = summary.get("attempts").and_then(Value::as_array).expect("attempts array"); + let attempt2 = attempts + .iter() + .find(|attempt| attempt.get("index").and_then(Value::as_i64) == Some(2)) + .expect("attempt 2 visible"); + assert_eq!(attempt2.get("stepCount").and_then(Value::as_i64), Some(2)); + assert_eq!(attempt2.get("statsSource").and_then(Value::as_str), Some("raw-trace-fallback")); + assert_eq!(attempt2.pointer("/execution/traceStatsState").and_then(Value::as_str), Some("degraded")); + } }