mirrors/pikasTech-unidesk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix code queue mgr trace summary contract

Browse Source
This commit is contained in:
Codex
2026-05-20 05:25:39 +00:00
parent 941fa85390
commit bd334bab4a
1 changed files with 267 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/components/microservices/code-queue-mgr/src-rs/main.rs
+267 -13
View File
@@ -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<Value> {
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<i64> {
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<i64> {
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<i64> {
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<Value> {
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::<Vec<_>>();
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<Value> = 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::<Vec<_>>();
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::<Vec<_>>();
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::<Vec<_>>();
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"));
}
}