//! Question + order backchannel for directive-spawned tasks.
//!
//! Originally a much larger handler that orchestrated contract-supervisor
//! task trees (spawn / wait / merge / PR / etc.). Legacy contracts and
//! supervisor tasks have been removed; what remains is the in-memory
//! question machinery (`makima directive ask`) and order creation
//! (`makima directive create-order`).
//!
//! Module name is kept as `mesh_supervisor` for route-path stability —
//! the CLI client still hits `/api/v1/mesh/supervisor/...` endpoints.
use axum::{
extract::{Path, State},
http::{HeaderMap, StatusCode},
response::IntoResponse,
Json,
};
use serde::{Deserialize, Serialize};
use utoipa::ToSchema;
use uuid::Uuid;
use crate::db::models::CreateOrderRequest;
use crate::db::repository;
use crate::server::auth::Authenticated;
use crate::server::handlers::mesh::{extract_auth, AuthSource};
use crate::server::messages::ApiError;
use crate::server::state::SharedState;
// =============================================================================
// Auth helper
// =============================================================================
/// Verify the request comes from a directive task (tool-key auth) and
/// return the calling task id + owner id.
async fn verify_task_auth(
state: &SharedState,
headers: &HeaderMap,
) -> Result<(Uuid, Uuid), (StatusCode, Json<ApiError>)> {
let auth = extract_auth(state, headers);
let task_id = match auth {
AuthSource::ToolKey(task_id) => task_id,
_ => {
return Err((
StatusCode::UNAUTHORIZED,
Json(ApiError::new("UNAUTHORIZED", "These endpoints require tool key auth")),
));
}
};
let pool = state.db_pool.as_ref().ok_or_else(|| {
(
StatusCode::SERVICE_UNAVAILABLE,
Json(ApiError::new("DB_UNAVAILABLE", "Database not configured")),
)
})?;
let task = repository::get_task(pool, task_id)
.await
.map_err(|e| {
tracing::error!(error = %e, "Failed to load task");
(
StatusCode::INTERNAL_SERVER_ERROR,
Json(ApiError::new("DB_ERROR", "Failed to load task")),
)
})?
.ok_or_else(|| {
(
StatusCode::NOT_FOUND,
Json(ApiError::new("NOT_FOUND", "Task not found")),
)
})?;
// Only directive-attached tasks may use this backchannel.
if task.directive_id.is_none() {
return Err((
StatusCode::FORBIDDEN,
Json(ApiError::new(
"NOT_DIRECTIVE_TASK",
"Only directive-attached tasks can use these endpoints",
)),
));
}
Ok((task_id, task.owner_id))
}
// =============================================================================
// Question types
// =============================================================================
#[derive(Debug, Deserialize, ToSchema)]
#[serde(rename_all = "camelCase")]
pub struct AskQuestionRequest {
pub question: String,
#[serde(default)]
pub choices: Vec<String>,
pub context: Option<String>,
#[serde(default = "default_question_timeout")]
pub timeout_seconds: i32,
/// When true the request blocks until the user responds (no
/// timeout) — the CLI reconnects via the poll endpoint if the
/// server-side timeout is reached.
#[serde(default)]
pub phaseguard: bool,
#[serde(default)]
pub multi_select: bool,
/// Return immediately without waiting for a response.
#[serde(default)]
pub non_blocking: bool,
/// Question type: general, phase_confirmation, contract_complete.
#[serde(default = "default_question_type")]
pub question_type: String,
}
fn default_question_type() -> String {
"general".to_string()
}
fn default_question_timeout() -> i32 {
3600
}
#[derive(Debug, Serialize, Deserialize, ToSchema)]
#[serde(rename_all = "camelCase")]
pub struct AskQuestionResponse {
pub question_id: Uuid,
pub response: Option<String>,
pub timed_out: bool,
/// Server-side timeout was reached but the question is still
/// pending. CLI should re-poll via `/poll`.
#[serde(default)]
pub still_pending: bool,
}
#[derive(Debug, Deserialize, ToSchema)]
#[serde(rename_all = "camelCase")]
pub struct AnswerQuestionRequest {
pub response: String,
}
#[derive(Debug, Serialize, ToSchema)]
#[serde(rename_all = "camelCase")]
pub struct AnswerQuestionResponse {
pub success: bool,
}
#[derive(Debug, Serialize, ToSchema)]
#[serde(rename_all = "camelCase")]
pub struct PendingQuestionSummary {
pub question_id: Uuid,
pub task_id: Uuid,
#[serde(skip_serializing_if = "Option::is_none")]
pub directive_id: Option<Uuid>,
pub question: String,
pub choices: Vec<String>,
pub context: Option<String>,
pub created_at: chrono::DateTime<chrono::Utc>,
#[serde(default)]
pub multi_select: bool,
#[serde(default)]
pub question_type: String,
}
// =============================================================================
// Question handlers
// =============================================================================
/// Ask the user a question from a directive task. Blocks until the user
/// answers, the timeout fires, or `non_blocking` returns immediately.
#[utoipa::path(
post,
path = "/api/v1/mesh/supervisor/questions",
request_body = AskQuestionRequest,
responses(
(status = 200, description = "Question asked", body = AskQuestionResponse),
(status = 401, description = "Unauthorized"),
(status = 403, description = "Not a directive task"),
),
security(("tool_key" = [])),
tag = "Mesh Supervisor"
)]
pub async fn ask_question(
State(state): State<SharedState>,
headers: HeaderMap,
Json(request): Json<AskQuestionRequest>,
) -> impl IntoResponse {
let (task_id, owner_id) = match verify_task_auth(&state, &headers).await {
Ok(ids) => ids,
Err(e) => return e.into_response(),
};
let pool = state.db_pool.as_ref().unwrap();
// Pull the directive_id off the calling task so subscribers can
// route the question to the right directive view.
let task = match repository::get_task_for_owner(pool, task_id, owner_id).await {
Ok(Some(t)) => t,
Ok(None) => {
return (
StatusCode::NOT_FOUND,
Json(ApiError::new("NOT_FOUND", "Task not found")),
)
.into_response();
}
Err(e) => {
tracing::error!(error = %e, "Failed to fetch task");
return (
StatusCode::INTERNAL_SERVER_ERROR,
Json(ApiError::new("DB_ERROR", "Failed to fetch task")),
)
.into_response();
}
};
let directive_id = task.directive_id;
// Reconcile mode controls block-vs-timeout behaviour on directive
// tasks: semi-auto / manual block indefinitely (effectively
// phaseguard); auto times out after 30s.
let reconcile_mode: String = match directive_id {
Some(did) => match repository::get_directive_for_owner(pool, owner_id, did).await {
Ok(Some(d)) => d.reconcile_mode.clone(),
_ => "auto".to_string(),
},
None => "auto".to_string(),
};
let question_id = state.add_supervisor_question(
task_id,
directive_id,
owner_id,
request.question.clone(),
request.choices.clone(),
request.context.clone(),
request.multi_select,
request.question_type.clone(),
);
if request.non_blocking {
return (
StatusCode::OK,
Json(AskQuestionResponse {
question_id,
response: None,
timed_out: false,
still_pending: false,
}),
)
.into_response();
}
// Determine block behaviour.
let use_phaseguard =
request.phaseguard || reconcile_mode == "semi-auto" || reconcile_mode == "manual";
let timeout_secs = if use_phaseguard {
300
} else if reconcile_mode == "auto" {
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);
loop {
if let Some(response) = state.get_question_response(question_id) {
state.cleanup_question_response(question_id);
return (
StatusCode::OK,
Json(AskQuestionResponse {
question_id,
response: Some(response.response),
timed_out: false,
still_pending: false,
}),
)
.into_response();
}
if start.elapsed() >= timeout_duration {
if use_phaseguard {
return (
StatusCode::OK,
Json(AskQuestionResponse {
question_id,
response: None,
timed_out: false,
still_pending: true,
}),
)
.into_response();
}
state.remove_pending_question(question_id);
return (
StatusCode::REQUEST_TIMEOUT,
Json(AskQuestionResponse {
question_id,
response: None,
timed_out: true,
still_pending: false,
}),
)
.into_response();
}
tokio::time::sleep(poll_interval).await;
}
}
/// Re-poll a question by id. Used by the CLI to reconnect after
/// `still_pending` from `ask_question`. Blocks up to 5 minutes.
#[utoipa::path(
get,
path = "/api/v1/mesh/supervisor/questions/{question_id}/poll",
params(("question_id" = Uuid, Path, description = "Question id")),
responses(
(status = 200, description = "Answered or still pending", body = AskQuestionResponse),
(status = 404, description = "Not found"),
),
security(("tool_key" = [])),
tag = "Mesh Supervisor"
)]
pub async fn poll_question(
State(state): State<SharedState>,
headers: HeaderMap,
Path(question_id): Path<Uuid>,
) -> impl IntoResponse {
if verify_task_auth(&state, &headers).await.is_err() {
return (
StatusCode::UNAUTHORIZED,
Json(ApiError::new("UNAUTHORIZED", "Tool key required")),
)
.into_response();
}
if let Some(response) = state.get_question_response(question_id) {
state.cleanup_question_response(question_id);
return (
StatusCode::OK,
Json(AskQuestionResponse {
question_id,
response: Some(response.response),
timed_out: false,
still_pending: false,
}),
)
.into_response();
}
if state.get_pending_question(question_id).is_none() {
return (
StatusCode::NOT_FOUND,
Json(ApiError::new("NOT_FOUND", "Question not found")),
)
.into_response();
}
let timeout = std::time::Duration::from_secs(300);
let start = std::time::Instant::now();
let poll_interval = std::time::Duration::from_millis(500);
loop {
if let Some(response) = state.get_question_response(question_id) {
state.cleanup_question_response(question_id);
return (
StatusCode::OK,
Json(AskQuestionResponse {
question_id,
response: Some(response.response),
timed_out: false,
still_pending: false,
}),
)
.into_response();
}
if start.elapsed() >= timeout {
return (
StatusCode::OK,
Json(AskQuestionResponse {
question_id,
response: None,
timed_out: false,
still_pending: true,
}),
)
.into_response();
}
tokio::time::sleep(poll_interval).await;
}
}
/// List currently-pending questions for the caller.
#[utoipa::path(
get,
path = "/api/v1/mesh/questions",
responses(
(status = 200, description = "Pending questions", body = Vec<PendingQuestionSummary>),
),
security(("bearer_auth" = []), ("api_key" = [])),
tag = "Mesh"
)]
pub async fn list_pending_questions(
State(state): State<SharedState>,
Authenticated(auth): Authenticated,
) -> impl IntoResponse {
let questions: Vec<PendingQuestionSummary> = state
.get_pending_questions_for_owner(auth.owner_id)
.into_iter()
.map(|q| PendingQuestionSummary {
question_id: q.question_id,
task_id: q.task_id,
directive_id: q.directive_id,
question: q.question,
choices: q.choices,
context: q.context,
created_at: q.created_at,
multi_select: q.multi_select,
question_type: q.question_type,
})
.collect();
Json(questions).into_response()
}
/// Answer a pending question.
#[utoipa::path(
post,
path = "/api/v1/mesh/questions/{question_id}/answer",
params(("question_id" = Uuid, Path, description = "Question id")),
request_body = AnswerQuestionRequest,
responses(
(status = 200, description = "Answered", body = AnswerQuestionResponse),
(status = 404, description = "Not found"),
),
security(("bearer_auth" = []), ("api_key" = [])),
tag = "Mesh"
)]
pub async fn answer_question(
State(state): State<SharedState>,
Authenticated(auth): Authenticated,
Path(question_id): Path<Uuid>,
Json(req): Json<AnswerQuestionRequest>,
) -> impl IntoResponse {
// Ownership check: only the owner of the question can answer it.
let question = match state.get_pending_question(question_id) {
Some(q) => q,
None => {
return (
StatusCode::NOT_FOUND,
Json(ApiError::new("NOT_FOUND", "Question not found")),
)
.into_response();
}
};
if question.owner_id != auth.owner_id {
return (
StatusCode::FORBIDDEN,
Json(ApiError::new("FORBIDDEN", "Not your question")),
)
.into_response();
}
if state.submit_question_response(question_id, req.response) {
Json(AnswerQuestionResponse { success: true }).into_response()
} else {
(
StatusCode::NOT_FOUND,
Json(ApiError::new("NOT_FOUND", "Question not found")),
)
.into_response()
}
}
// =============================================================================
// Order creation (from directive tasks)
// =============================================================================
#[derive(Debug, Deserialize, ToSchema)]
#[serde(rename_all = "camelCase")]
pub struct CreateOrderForTaskRequest {
pub title: String,
#[serde(default)]
pub description: Option<String>,
#[serde(default = "default_order_priority")]
pub priority: String,
#[serde(default = "default_order_type")]
pub order_type: String,
#[serde(default = "default_order_labels")]
pub labels: serde_json::Value,
#[serde(default)]
pub repository_url: Option<String>,
}
fn default_order_priority() -> String {
"medium".to_string()
}
fn default_order_type() -> String {
"spike".to_string()
}
fn default_order_labels() -> serde_json::Value {
serde_json::json!([])
}
/// Create a follow-up order from a directive task (spike/chore only).
#[utoipa::path(
post,
path = "/api/v1/mesh/supervisor/orders",
request_body = CreateOrderForTaskRequest,
responses(
(status = 201, description = "Order created"),
(status = 400, description = "Invalid order type or no directive context"),
(status = 401, description = "Unauthorized"),
(status = 403, description = "Not a directive task"),
),
security(("tool_key" = [])),
tag = "Mesh Supervisor"
)]
pub async fn create_order_for_task(
State(state): State<SharedState>,
headers: HeaderMap,
Json(request): Json<CreateOrderForTaskRequest>,
) -> impl IntoResponse {
let (task_id, owner_id) = match verify_task_auth(&state, &headers).await {
Ok(ids) => ids,
Err(e) => return e.into_response(),
};
if request.order_type != "spike" && request.order_type != "chore" {
return (
StatusCode::BAD_REQUEST,
Json(ApiError::new(
"INVALID_ORDER_TYPE",
"Only spike and chore order types are allowed from directive tasks",
)),
)
.into_response();
}
let pool = state.db_pool.as_ref().unwrap();
let task = match repository::get_task(pool, task_id).await {
Ok(Some(t)) => t,
Ok(None) => {
return (
StatusCode::NOT_FOUND,
Json(ApiError::new("NOT_FOUND", "Task not found")),
)
.into_response();
}
Err(e) => {
tracing::error!(error = %e, "Failed to fetch task");
return (
StatusCode::INTERNAL_SERVER_ERROR,
Json(ApiError::new("DB_ERROR", "Failed to fetch task")),
)
.into_response();
}
};
let directive_id = match task.directive_id {
Some(id) => id,
None => {
return (
StatusCode::BAD_REQUEST,
Json(ApiError::new("NO_DIRECTIVE", "Task is not directive-attached")),
)
.into_response();
}
};
let repository_url = if request.repository_url.is_some() {
request.repository_url
} else {
match repository::get_directive_for_owner(pool, owner_id, directive_id).await {
Ok(Some(d)) => d.repository_url,
_ => None,
}
};
let order_req = CreateOrderRequest {
title: request.title,
description: request.description,
priority: Some(request.priority),
status: Some("open".to_string()),
order_type: Some(request.order_type),
labels: request.labels,
directive_id,
repository_url,
dog_id: None,
};
match repository::create_order(pool, owner_id, order_req).await {
Ok(order) => (
StatusCode::CREATED,
Json(serde_json::json!({
"id": order.id,
"title": order.title,
"description": order.description,
"priority": order.priority,
"status": order.status,
"orderType": order.order_type,
"directiveId": order.directive_id,
"labels": order.labels,
"repositoryUrl": order.repository_url,
"createdAt": order.created_at,
})),
)
.into_response(),
Err(e) => {
tracing::error!(error = %e, "Failed to create order");
(
StatusCode::INTERNAL_SERVER_ERROR,
Json(ApiError::new("DB_ERROR", "Failed to create order")),
)
.into_response()
}
}
}
