diff options
Diffstat (limited to 'makima/src/server/handlers/mesh_supervisor.rs')
| -rw-r--r-- | makima/src/server/handlers/mesh_supervisor.rs | 90 |
1 files changed, 68 insertions, 22 deletions
diff --git a/makima/src/server/handlers/mesh_supervisor.rs b/makima/src/server/handlers/mesh_supervisor.rs index c9cb849..90c6dc7 100644 --- a/makima/src/server/handlers/mesh_supervisor.rs +++ b/makima/src/server/handlers/mesh_supervisor.rs @@ -129,6 +129,9 @@ pub struct PendingQuestionSummary { pub question_id: Uuid, pub task_id: Uuid, pub contract_id: Uuid, + /// Directive this question relates to (if from a directive task) + #[serde(skip_serializing_if = "Option::is_none")] + pub directive_id: Option<Uuid>, pub question: String, pub choices: Vec<String>, pub context: Option<String>, @@ -257,11 +260,11 @@ async fn verify_supervisor_auth( ) })?; - // Verify task is a supervisor - if !task.is_supervisor { + // Verify task is a supervisor or a directive task + if !task.is_supervisor && task.directive_id.is_none() { return Err(( StatusCode::FORBIDDEN, - Json(ApiError::new("NOT_SUPERVISOR", "Only supervisor tasks can use these endpoints")), + Json(ApiError::new("NOT_SUPERVISOR", "Only supervisor or directive tasks can use these endpoints")), )); } @@ -1694,17 +1697,43 @@ pub async fn ask_question( } }; - let Some(contract_id) = supervisor.contract_id else { + // Determine context: contract or directive + let contract_id = supervisor.contract_id; + let directive_id = supervisor.directive_id; + + if contract_id.is_none() && directive_id.is_none() { return ( StatusCode::BAD_REQUEST, - Json(ApiError::new("NO_CONTRACT", "Supervisor has no associated contract")), + Json(ApiError::new("NO_CONTEXT", "Supervisor has no associated contract or directive")), ).into_response(); + } + + let is_directive_context = directive_id.is_some() && contract_id.is_none(); + + // For directive context, check reconcile_mode to determine behavior + let directive_reconcile_mode = if let Some(did) = directive_id { + if is_directive_context { + match repository::get_directive_for_owner(pool, owner_id, did).await { + Ok(Some(d)) => d.reconcile_mode, + Ok(None) => false, + Err(e) => { + tracing::warn!(error = %e, "Failed to get directive for reconcile_mode check"); + false + } + } + } else { + false + } + } else { + false }; - // Add the question - let question_id = state.add_supervisor_question( + // Add the question (use Uuid::nil() for contract_id in directive-only context) + let effective_contract_id = contract_id.unwrap_or(Uuid::nil()); + let question_id = state.add_supervisor_question_with_directive( supervisor_id, - contract_id, + effective_contract_id, + directive_id, owner_id, request.question.clone(), request.choices.clone(), @@ -1714,15 +1743,18 @@ pub async fn ask_question( ); // Save state: question asked is a key save point (Task 3.3) - let pending_question = PendingQuestion { - id: question_id, - question: request.question.clone(), - choices: request.choices.clone(), - context: request.context.clone(), - question_type: request.question_type.clone(), - asked_at: chrono::Utc::now(), - }; - save_state_on_question_asked(pool, contract_id, pending_question).await; + // Only for contract context — directive tasks don't use supervisor_states table + if let Some(cid) = contract_id { + let pending_question = PendingQuestion { + id: question_id, + question: request.question.clone(), + choices: request.choices.clone(), + context: request.context.clone(), + question_type: request.question_type.clone(), + asked_at: chrono::Utc::now(), + }; + save_state_on_question_asked(pool, cid, pending_question).await; + } // Broadcast question as task output entry for the task's chat let question_data = serde_json::json!({ @@ -1775,9 +1807,10 @@ pub async fn ask_question( ).into_response(); } - // If phaseguard is enabled, pause the supervisor task and return + // If phaseguard is enabled (or directive reconcile mode), pause the supervisor task and return // The task will be auto-resumed when a message is sent to it (e.g., when user answers) - if request.phaseguard { + let use_phaseguard = request.phaseguard || (is_directive_context && directive_reconcile_mode); + if use_phaseguard { // Pause the supervisor task if let Some(daemon_id) = supervisor.daemon_id { let cmd = DaemonCommand::PauseTask { task_id: supervisor_id }; @@ -1808,7 +1841,13 @@ pub async fn ask_question( } // Poll for response with timeout - let timeout_duration = std::time::Duration::from_secs(request.timeout_seconds.max(1) as u64); + // For directive tasks without reconcile mode, use 30s default timeout + let timeout_secs = if is_directive_context && !directive_reconcile_mode { + 30 + } else { + request.timeout_seconds.max(1) as u64 + }; + let timeout_duration = std::time::Duration::from_secs(timeout_secs); let start = std::time::Instant::now(); let poll_interval = std::time::Duration::from_millis(500); @@ -1819,7 +1858,10 @@ pub async fn ask_question( state.cleanup_question_response(question_id); // Clear pending question from supervisor state (Task 3.3) - clear_pending_question(pool, contract_id, question_id).await; + // Skip for directive context — no supervisor_states for directives + if let Some(cid) = contract_id { + clear_pending_question(pool, cid, question_id).await; + } return ( StatusCode::OK, @@ -1837,7 +1879,10 @@ pub async fn ask_question( state.remove_pending_question(question_id); // Clear pending question from supervisor state on timeout (Task 3.3) - clear_pending_question(pool, contract_id, question_id).await; + // Skip for directive context — no supervisor_states for directives + if let Some(cid) = contract_id { + clear_pending_question(pool, cid, question_id).await; + } return ( StatusCode::REQUEST_TIMEOUT, @@ -1880,6 +1925,7 @@ pub async fn list_pending_questions( question_id: q.question_id, task_id: q.task_id, contract_id: q.contract_id, + directive_id: q.directive_id, question: q.question, choices: q.choices, context: q.context, |