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

fix: clear cancel flag on code queue retry

Browse Source
This commit is contained in:
Codex
2026-05-19 16:48:50 +00:00
parent a769242207
commit 24c590f551
1 changed files with 159 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/components/microservices/code-queue-mgr/src-rs/main.rs
+159 -18
View File
@@ -2186,6 +2186,50 @@ fn mark_all_read(state: &AppState, url: &RequestUrl) -> Result<(Value, u16), (Va
Ok((json!({ "ok": true, "count": rows.len(), "readAt": read_at, "queue": summary }), 200))
}
fn manual_retry_status_sql_condition() -> &'static str {
"started_at IS NULL AND current_attempt = 0 AND codex_thread_id IS NULL AND active_turn_id IS NULL AND attempt_count = 0"
}
fn manual_retry_status_sql_value() -> String {
format!(
"CASE WHEN {} THEN 'queued'::text ELSE 'retry_wait'::text END",
manual_retry_status_sql_condition()
)
}
fn manual_retry_task_json_sql_expression() -> String {
let next_status = manual_retry_status_sql_value();
format!(
"
jsonb_set(
jsonb_set(
jsonb_set(
jsonb_set(
jsonb_set(
jsonb_set(
jsonb_set(
jsonb_set(
jsonb_set(task_json, '{{status}}', to_jsonb({next_status}), true),
'{{finishedAt}}', 'null'::jsonb, true
),
'{{readAt}}', 'null'::jsonb, true
),
'{{cancelRequested}}', 'false'::jsonb, true
),
'{{lastError}}', 'null'::jsonb, true
),
'{{maxAttempts}}', to_jsonb(GREATEST(max_attempts, current_attempt + 1, attempt_count + 1)), true
),
'{{updatedAt}}', to_jsonb($2::text), true
),
'{{queueEnteredAt}}', to_jsonb($2::text), true
),
'{{nextMode}}', to_jsonb('retry'::text), true
)
"
)
}
fn move_task(state: &AppState, task_id: &str, request: &mut Request) -> Result<(Value, u16), (Value, u16)> {
let body = read_json_body(request)?;
let queue_id = validate_queue_id(body.get("queueId").or_else(|| body.get("id")).and_then(Value::as_str).unwrap_or(DEFAULT_QUEUE_ID))
@@ -2244,35 +2288,28 @@ fn retry_task(state: &AppState, task_id: &str, request: &mut Request) -> Result<
.to_string();
let queued_at = now_iso();
let mut client = db_client(state, "unidesk-code-queue-mgr-rust-retry").map_err(|error| (json!({ "ok": false, "error": error }), 500))?;
let rows = client
.query(
"
let retry_status = manual_retry_status_sql_value();
let task_json_update = manual_retry_task_json_sql_expression();
let sql = format!(
"
UPDATE unidesk_code_queue_tasks
SET status = CASE WHEN started_at IS NULL AND current_attempt = 0 AND codex_thread_id IS NULL THEN 'queued' ELSE 'retry_wait' END,
SET status = {retry_status},
finished_at = NULL,
read_at = NULL,
last_error = NULL,
max_attempts = GREATEST(max_attempts, current_attempt + 1, attempt_count + 1),
updated_at = $2::text::timestamptz,
task_json = jsonb_set(
jsonb_set(
jsonb_set(
jsonb_set(
jsonb_set(task_json, '{status}', to_jsonb(CASE WHEN started_at IS NULL AND current_attempt = 0 AND codex_thread_id IS NULL THEN 'queued'::text ELSE 'retry_wait'::text END), true),
'{finishedAt}', 'null'::jsonb, true
),
'{readAt}', 'null'::jsonb, true
),
'{nextMode}', to_jsonb('retry'::text), true
),
'{nextPrompt}', to_jsonb($3::text), true
{task_json_update},
'{{nextPrompt}}', to_jsonb($3::text), true
)
WHERE id = $1
AND (status IN ('succeeded', 'failed', 'canceled') OR (status IN ('queued', 'retry_wait') AND active_turn_id IS NULL AND (started_at IS NOT NULL OR current_attempt > 0 OR codex_thread_id IS NOT NULL)))
RETURNING id
",
&[&task_id, &queued_at, &prompt],
)
"
);
let rows = client
.query(sql.as_str(), &[&task_id, &queued_at, &prompt])
.map_err(|error| (json!({ "ok": false, "error": error.to_string() }), 500))?;
if rows.is_empty() {
let Some(task) = load_task(state, task_id, false).map_err(|error| (json!({ "ok": false, "error": error }), 500))? else {
@@ -2789,6 +2826,39 @@ mod tests {
value.get(key).and_then(Value::as_str).unwrap_or("")
}
fn json_bool(task: &TaskMeta, key: &str) -> Option<bool> {
task.task_json.as_ref().and_then(|value| value.get(key)).and_then(Value::as_bool)
}
fn apply_manual_retry_contract_for_test(task: &TaskMeta, queued_at: &str, prompt: &str) -> TaskMeta {
let mut next = task.clone();
let claimed = next.started_at.is_some()
|| next.current_attempt > 0
|| next.codex_thread_id.is_some()
|| next.active_turn_id.is_some()
|| next.attempt_count > 0;
next.status = if claimed { "retry_wait" } else { "queued" }.to_string();
next.finished_at = None;
next.read_at = None;
next.last_error = None;
next.max_attempts = next.max_attempts.max(next.current_attempt + 1).max(next.attempt_count + 1);
next.updated_at = queued_at.to_string();
let mut task_json = next.task_json.take().unwrap_or_else(|| json!({}));
let json_object = task_json.as_object_mut().expect("test task json object");
json_object.insert("status".to_string(), json!(next.status));
json_object.insert("finishedAt".to_string(), Value::Null);
json_object.insert("readAt".to_string(), Value::Null);
json_object.insert("cancelRequested".to_string(), Value::Bool(false));
json_object.insert("lastError".to_string(), Value::Null);
json_object.insert("maxAttempts".to_string(), json!(next.max_attempts));
json_object.insert("updatedAt".to_string(), json!(queued_at));
json_object.insert("queueEnteredAt".to_string(), json!(queued_at));
json_object.insert("nextMode".to_string(), json!("retry"));
json_object.insert("nextPrompt".to_string(), json!(prompt));
next.task_json = Some(task_json);
next
}
#[test]
fn prompt_observation_preserves_raw_fields_and_explicit_previews() {
let state = test_state();
@@ -2829,4 +2899,75 @@ mod tests {
assert_eq!(record.get("executionMode").and_then(Value::as_str), Some("default"));
assert_eq!(record.get("path").and_then(Value::as_str), Some("/workspace"));
}
#[test]
fn manual_retry_sql_clears_cancel_requested_contract() {
let sql = manual_retry_task_json_sql_expression();
assert!(sql.contains("'{cancelRequested}', 'false'::jsonb"), "{sql}");
assert!(sql.contains("'{maxAttempts}', to_jsonb(GREATEST(max_attempts, current_attempt + 1, attempt_count + 1))"), "{sql}");
assert!(sql.contains("'{queueEnteredAt}', to_jsonb($2::text)"), "{sql}");
assert!(sql.contains("'{nextMode}', to_jsonb('retry'::text)"), "{sql}");
}
#[test]
fn manual_retry_canceled_task_becomes_scheduler_runnable() {
let state = test_state();
let canceled_at = "2026-05-19T10:00:00.000Z".to_string();
let mut task = test_task("interrupted task".to_string());
task.status = "canceled".to_string();
task.started_at = Some(canceled_at.clone());
task.finished_at = Some(canceled_at.clone());
task.read_at = Some(canceled_at.clone());
task.current_attempt = 1;
task.current_mode = Some("initial".to_string());
task.max_attempts = 1;
task.last_error = Some("Task canceled by request.".to_string());
task.attempt_count = 1;
task.task_json = Some(json!({
"status": "canceled",
"queueEnteredAt": "2026-05-19T09:59:00.000Z",
"finishedAt": canceled_at,
"readAt": canceled_at,
"updatedAt": canceled_at,
"lastError": "Task canceled by request.",
"finalResponse": "",
"stepCount": 0,
"llmStepCount": 0,
"judgeFailCount": 0,
"maxAttempts": 1,
"cancelRequested": true,
"nextPrompt": Value::Null,
"nextMode": Value::Null,
"attempts": [{ "index": 1, "mode": "initial" }],
"output": [],
"events": [],
"promptHistory": []
}));
assert_eq!(json_bool(&task, "cancelRequested"), Some(true));
let retry_prompt = "Manual retry requested from master code-queue-mgr.";
let retried = apply_manual_retry_contract_for_test(&task, "2026-05-19T10:01:00.000Z", retry_prompt);
assert_eq!(retried.status, "retry_wait");
assert_eq!(retried.finished_at, None);
assert_eq!(retried.read_at, None);
assert_eq!(retried.last_error, None);
assert_eq!(retried.max_attempts, 2);
assert_eq!(json_bool(&retried, "cancelRequested"), Some(false));
assert_eq!(
retried.task_json.as_ref().and_then(|value| value.get("nextMode")).and_then(Value::as_str),
Some("retry")
);
assert_eq!(
retried.task_json.as_ref().and_then(|value| value.get("nextPrompt")).and_then(Value::as_str),
Some(retry_prompt)
);
let scheduler_visible = task_list_response(&state, &retried, false);
assert_eq!(scheduler_visible.get("status").and_then(Value::as_str), Some("retry_wait"));
assert_eq!(scheduler_visible.get("cancelRequested").and_then(Value::as_bool), Some(false));
assert_eq!(
scheduler_visible.pointer("/queuedReason/code").and_then(Value::as_str),
Some("mgr-visible")
);
}
}