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 b5f407b5..868416bb 100644 --- a/src/components/microservices/code-queue-mgr/src-rs/main.rs +++ b/src/components/microservices/code-queue-mgr/src-rs/main.rs @@ -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 { + 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") + ); + } }