From 2dd58237b4406af144a5213c9dc44f38b1195407 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Tue, 7 Apr 2026 19:44:32 -0400 Subject: [PATCH] refactor(interview): simplify pending question handling Collapse the live answer rendezvous into ControlInterviewer, move pending question storage onto a shared typed record, and route HTTP and Slack answer submission through one server-side flow. --- .../fabro-cli/src/commands/run/runner.rs | 46 ++-- lib/crates/fabro-interview/src/control.rs | 96 ++++--- lib/crates/fabro-interview/src/lib.rs | 10 +- lib/crates/fabro-server/src/server.rs | 258 +++++++++++------- .../tests/it/scenario/lifecycle.rs | 2 +- lib/crates/fabro-store/src/run_state.rs | 63 +++-- lib/crates/fabro-types/src/interview.rs | 62 +++++ lib/crates/fabro-types/src/lib.rs | 2 + 8 files changed, 352 insertions(+), 187 deletions(-) create mode 100644 lib/crates/fabro-types/src/interview.rs diff --git a/lib/crates/fabro-cli/src/commands/run/runner.rs b/lib/crates/fabro-cli/src/commands/run/runner.rs index f6d39c36b..ebb1bfbfd 100644 --- a/lib/crates/fabro-cli/src/commands/run/runner.rs +++ b/lib/crates/fabro-cli/src/commands/run/runner.rs @@ -5,9 +5,7 @@ use std::time::Duration; use anyhow::{Context, Result, anyhow}; use async_trait::async_trait; -use fabro_interview::{ - ControlInterviewer, InterviewBroker, WorkerControlEnvelope, WorkerControlMessage, -}; +use fabro_interview::{ControlInterviewer, WorkerControlEnvelope, WorkerControlMessage}; use fabro_store::{EventEnvelope, EventPayload, RunProjection}; use fabro_types::{EventBody, RunBlobId, RunEvent, RunId, Settings, StatusReason}; use fabro_workflow::artifact_snapshot::CapturedArtifactInfo; @@ -71,9 +69,11 @@ pub(crate) async fn execute( client.clone_for_reuse(), artifact_upload_token, ); - let broker = Arc::new(InterviewBroker::new()); - let interviewer = Arc::new(ControlInterviewer::new(Arc::clone(&broker))); - tokio::spawn(read_worker_control_stream(io::stdin(), broker)); + let interviewer = Arc::new(ControlInterviewer::new()); + tokio::spawn(read_worker_control_stream( + io::stdin(), + Arc::clone(&interviewer), + )); let run_control = RunControlState::new(); let cancel_token = Arc::new(AtomicBool::new(false)); install_signal_handlers(Arc::clone(&run_control), Arc::clone(&cancel_token))?; @@ -110,7 +110,7 @@ pub(crate) async fn execute( Ok(()) } -async fn read_worker_control_stream(reader: R, broker: Arc) +async fn read_worker_control_stream(reader: R, interviewer: Arc) where R: AsyncRead + Unpin, { @@ -118,17 +118,17 @@ where loop { match lines.next_line().await { Ok(Some(line)) => { - apply_worker_control_line(&broker, &line).await; + apply_worker_control_line(&interviewer, &line).await; } Ok(None) | Err(_) => { - broker.abort_all().await; + interviewer.abort_all().await; break; } } } } -async fn apply_worker_control_line(broker: &InterviewBroker, line: &str) { +async fn apply_worker_control_line(interviewer: &ControlInterviewer, line: &str) { if line.trim().is_empty() { return; } @@ -139,7 +139,7 @@ async fn apply_worker_control_line(broker: &InterviewBroker, line: &str) { match message.message { WorkerControlMessage::InterviewAnswer { qid, answer } => { - let _ = broker.submit(&qid, answer.into()).await; + let _ = interviewer.submit(&qid, answer.into()).await; } } } @@ -488,7 +488,7 @@ mod tests { read_worker_control_stream, worker_title, worker_title_phase_for_event, }; use crate::args::RunWorkerMode; - use fabro_interview::{AnswerValue, InterviewBroker}; + use fabro_interview::{AnswerValue, ControlInterviewer, Interviewer, Question, QuestionType}; use fabro_types::fixtures; use fabro_types::run_event::{ InterviewCompletedProps, InterviewStartedProps, RunCompletedProps, RunControlEffectProps, @@ -648,27 +648,33 @@ mod tests { #[tokio::test] async fn worker_control_line_routes_answer_by_question_id() { - let broker = Arc::new(InterviewBroker::new()); - let receiver = broker.register("q-1".to_string()).await; + let interviewer = Arc::new(ControlInterviewer::new()); + let mut question = Question::new("Approve?", QuestionType::YesNo); + question.id = "q-1".to_string(); + let ask_interviewer = Arc::clone(&interviewer); + let answer_task = tokio::spawn(async move { ask_interviewer.ask(question).await }); apply_worker_control_line( - &broker, + &interviewer, r#"{"v":1,"type":"interview.answer","qid":"q-1","answer":{"kind":"yes"}}"#, ) .await; - let answer: fabro_interview::Answer = receiver.await.unwrap(); + let answer: fabro_interview::Answer = answer_task.await.unwrap(); assert_eq!(answer.value, AnswerValue::Yes); } #[tokio::test] async fn worker_control_stream_eof_aborts_pending_interviews() { - let broker = Arc::new(InterviewBroker::new()); - let receiver = broker.register("q-1".to_string()).await; + let interviewer = Arc::new(ControlInterviewer::new()); + let mut question = Question::new("Approve?", QuestionType::YesNo); + question.id = "q-1".to_string(); + let ask_interviewer = Arc::clone(&interviewer); + let answer_task = tokio::spawn(async move { ask_interviewer.ask(question).await }); - read_worker_control_stream(tokio::io::empty(), Arc::clone(&broker)).await; + read_worker_control_stream(tokio::io::empty(), Arc::clone(&interviewer)).await; - let answer: fabro_interview::Answer = receiver.await.unwrap(); + let answer: fabro_interview::Answer = answer_task.await.unwrap(); assert_eq!(answer.value, AnswerValue::Aborted); } } diff --git a/lib/crates/fabro-interview/src/control.rs b/lib/crates/fabro-interview/src/control.rs index 2e48912e0..24af80cd0 100644 --- a/lib/crates/fabro-interview/src/control.rs +++ b/lib/crates/fabro-interview/src/control.rs @@ -1,5 +1,4 @@ use std::collections::HashMap; -use std::sync::Arc; use async_trait::async_trait; use tokio::sync::{Mutex, oneshot}; @@ -8,29 +7,34 @@ use crate::{Answer, Interviewer, Question}; #[derive(Debug, Clone, PartialEq, Eq)] pub enum SubmitError { - UnknownQuestion, AlreadyResolved, } #[derive(Default)] -struct InterviewBrokerState { +struct ControlInterviewerState { pending: HashMap>, queued: HashMap, + closed: bool, } #[derive(Default)] -pub struct InterviewBroker { - state: Mutex, +pub struct ControlInterviewer { + state: Mutex, } -impl InterviewBroker { +impl ControlInterviewer { #[must_use] pub fn new() -> Self { Self::default() } - pub async fn register(&self, question_id: String) -> oneshot::Receiver { + async fn register(&self, question_id: String) -> oneshot::Receiver { let mut state = self.state.lock().await; + if state.closed { + let (tx, rx) = oneshot::channel(); + let _ = tx.send(Answer::aborted()); + return rx; + } if let Some(answer) = state.queued.remove(&question_id) { let (tx, rx) = oneshot::channel(); let _ = tx.send(answer); @@ -45,6 +49,9 @@ impl InterviewBroker { pub async fn submit(&self, question_id: &str, answer: Answer) -> Result<(), SubmitError> { let pending_sender = { let mut state = self.state.lock().await; + if state.closed { + return Err(SubmitError::AlreadyResolved); + } if let Some(sender) = state.pending.remove(question_id) { Some(sender) } else if state.queued.contains_key(question_id) { @@ -59,13 +66,14 @@ impl InterviewBroker { Some(sender) => sender .send(answer) .map_err(|_| SubmitError::AlreadyResolved), - None => Err(SubmitError::UnknownQuestion), + None => Err(SubmitError::AlreadyResolved), } } pub async fn abort_all(&self) { let (pending, queued) = { let mut state = self.state.lock().await; + state.closed = true; let pending = state .pending .drain() @@ -83,27 +91,16 @@ impl InterviewBroker { if queued > 0 { tracing::debug!( count = queued, - "Dropped queued interview answers while aborting broker" + "Dropped queued interview answers while aborting control interviewer" ); } } } -pub struct ControlInterviewer { - broker: Arc, -} - -impl ControlInterviewer { - #[must_use] - pub fn new(broker: Arc) -> Self { - Self { broker } - } -} - #[async_trait] impl Interviewer for ControlInterviewer { async fn ask(&self, question: Question) -> Answer { - let receiver = self.broker.register(question.id.clone()).await; + let receiver = self.register(question.id.clone()).await; match receiver.await { Ok(answer) => answer, Err(_) => Answer::aborted(), @@ -124,22 +121,22 @@ mod tests { use super::*; #[tokio::test] - async fn submit_unknown_question_returns_error() { - let broker = InterviewBroker::new(); - let result = broker.submit("missing", Answer::yes()).await; + async fn submit_before_ask_buffers_answer() { + let interviewer = ControlInterviewer::new(); + let result = interviewer.submit("q-1", Answer::yes()).await; assert_eq!(result, Ok(())); } #[tokio::test] async fn register_then_submit_delivers_answer() { - let broker = Arc::new(InterviewBroker::new()); - let interviewer = ControlInterviewer::new(Arc::clone(&broker)); + let interviewer = Arc::new(ControlInterviewer::new()); let mut question = Question::new("approve?", QuestionType::YesNo); question.id = "q-1".to_string(); - let ask = tokio::spawn(async move { interviewer.ask(question).await }); - let submit_result = broker.submit("q-1", Answer::yes()).await; + let ask_interviewer = Arc::clone(&interviewer); + let ask = tokio::spawn(async move { ask_interviewer.ask(question).await }); + let submit_result = interviewer.submit("q-1", Answer::yes()).await; assert_eq!(submit_result, Ok(())); let answer = ask.await.unwrap(); @@ -148,21 +145,50 @@ mod tests { #[tokio::test] async fn submit_before_register_buffers_answer() { - let broker = Arc::new(InterviewBroker::new()); - assert_eq!(broker.submit("q-1", Answer::no()).await, Ok(())); + let interviewer = Arc::new(ControlInterviewer::new()); + assert_eq!(interviewer.submit("q-1", Answer::no()).await, Ok(())); - let receiver = broker.register("q-1".to_string()).await; - let answer = receiver.await.unwrap(); + let mut question = Question::new("approve?", QuestionType::YesNo); + question.id = "q-1".to_string(); + let answer = interviewer.ask(question).await; assert_eq!(answer.value, AnswerValue::No); } #[tokio::test] async fn duplicate_buffered_answer_is_rejected() { - let broker = InterviewBroker::new(); - assert_eq!(broker.submit("q-1", Answer::yes()).await, Ok(())); + let interviewer = ControlInterviewer::new(); + assert_eq!(interviewer.submit("q-1", Answer::yes()).await, Ok(())); assert_eq!( - broker.submit("q-1", Answer::no()).await, + interviewer.submit("q-1", Answer::no()).await, Err(SubmitError::AlreadyResolved) ); } + + #[tokio::test] + async fn abort_all_aborts_pending_questions() { + let interviewer = Arc::new(ControlInterviewer::new()); + let mut question = Question::new("approve?", QuestionType::YesNo); + question.id = "q-1".to_string(); + + let ask_interviewer = Arc::clone(&interviewer); + let ask = tokio::spawn(async move { ask_interviewer.ask(question).await }); + tokio::task::yield_now().await; + + interviewer.abort_all().await; + + let answer = ask.await.unwrap(); + assert_eq!(answer.value, AnswerValue::Aborted); + } + + #[tokio::test] + async fn ask_after_abort_all_returns_aborted() { + let interviewer = ControlInterviewer::new(); + interviewer.abort_all().await; + + let mut question = Question::new("approve?", QuestionType::YesNo); + question.id = "q-1".to_string(); + + let answer = interviewer.ask(question).await; + assert_eq!(answer.value, AnswerValue::Aborted); + } } diff --git a/lib/crates/fabro-interview/src/lib.rs b/lib/crates/fabro-interview/src/lib.rs index fc5bdee3c..d390e1d8f 100644 --- a/lib/crates/fabro-interview/src/lib.rs +++ b/lib/crates/fabro-interview/src/lib.rs @@ -209,7 +209,7 @@ pub trait Interviewer: Send + Sync { pub use auto_approve::AutoApproveInterviewer; pub use callback::CallbackInterviewer; pub use console::ConsoleInterviewer; -pub use control::{ControlInterviewer, InterviewBroker, SubmitError}; +pub use control::{ControlInterviewer, SubmitError}; pub use control_protocol::{ WORKER_CONTROL_PROTOCOL_VERSION, WorkerControlAnswer, WorkerControlEnvelope, WorkerControlMessage, @@ -375,16 +375,16 @@ mod tests { #[tokio::test] async fn control_interviewer_routes_answers_by_question_id() { - let broker = Arc::new(InterviewBroker::new()); - let interviewer = ControlInterviewer::new(Arc::clone(&broker)); + let interviewer = Arc::new(ControlInterviewer::new()); let mut question = Question::new("Approve?", QuestionType::YesNo); question.id = "q-1".to_string(); - let ask = tokio::spawn(async move { interviewer.ask(question).await }); + let ask_interviewer = Arc::clone(&interviewer); + let ask = tokio::spawn(async move { ask_interviewer.ask(question).await }); time::sleep(std::time::Duration::from_millis(10)).await; - broker.submit("q-1", Answer::yes()).await.unwrap(); + interviewer.submit("q-1", Answer::yes()).await.unwrap(); let answer = ask.await.unwrap(); assert_eq!(answer.value, AnswerValue::Yes); diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index 22802ce93..064eb956c 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -34,8 +34,9 @@ use fabro_store::{ ArtifactStore, Database, EventEnvelope, EventPayload, PendingInterviewRecord, StageId, }; use fabro_types::{ - EventBody, RunArtifactStorage, RunBlobId, RunClientProvenance, RunControlAction, RunEvent, - RunId, RunProvenance, RunServerProvenance, RunSubjectProvenance, Settings, + EventBody, InterviewQuestionRecord, InterviewQuestionType, RunArtifactStorage, RunBlobId, + RunClientProvenance, RunControlAction, RunEvent, RunId, RunProvenance, RunServerProvenance, + RunSubjectProvenance, Settings, }; use fabro_util::redact::redact_jsonl_line; use fabro_util::version::FABRO_VERSION; @@ -77,8 +78,7 @@ use crate::secret_store::{SecretStore, SecretStoreError}; use crate::static_files; use crate::web_auth; use fabro_interview::{ - Answer, ControlInterviewer, InterviewBroker, Interviewer, Question, QuestionType, - WorkerControlEnvelope, + Answer, ControlInterviewer, Interviewer, Question, QuestionType, WorkerControlEnvelope, }; use fabro_sandbox::daytona::DaytonaSandbox; use fabro_sandbox::reconnect::reconnect; @@ -316,7 +316,7 @@ enum RunAnswerTransport { control_tx: mpsc::Sender, }, InProcess { - broker: Arc, + interviewer: Arc, }, } @@ -336,7 +336,7 @@ impl RunAnswerTransport { .map_err(|_| AnswerTransportError::Timeout)? .map_err(|_| AnswerTransportError::Closed) } - Self::InProcess { broker } => broker + Self::InProcess { interviewer } => interviewer .submit(qid, answer) .await .map_err(|_| AnswerTransportError::Closed), @@ -344,12 +344,19 @@ impl RunAnswerTransport { } async fn abort_pending(&self) { - if let Self::InProcess { broker } = self { - broker.abort_all().await; + if let Self::InProcess { interviewer } = self { + interviewer.abort_all().await; } } } +#[derive(Debug, Clone)] +struct LoadedPendingInterview { + run_id: RunId, + qid: String, + question: InterviewQuestionRecord, +} + #[derive(Clone)] struct SlackService { client: SlackClient, @@ -386,25 +393,16 @@ impl SlackService { return; } - let question = Question { + let question = runtime_question_from_interview_record(&InterviewQuestionRecord { id: props.question_id.clone(), text: props.question.clone(), - question_type: parse_question_type(&props.question_type), - options: props - .options - .iter() - .map(|option| fabro_interview::QuestionOption { - key: option.key.clone(), - label: option.label.clone(), - }) - .collect(), - allow_freeform: props.allow_freeform, - default: None, - timeout_seconds: props.timeout_seconds, stage: props.stage.clone(), - metadata: HashMap::new(), + question_type: InterviewQuestionType::from_wire_name(&props.question_type), + options: props.options.clone(), + allow_freeform: props.allow_freeform, + timeout_seconds: props.timeout_seconds, context_display: props.context_display.clone(), - }; + }); let blocks = slack_blocks::question_to_blocks( &event.run_id.to_string(), &props.question_id, @@ -496,16 +494,16 @@ impl SlackService { Err(_) => return, }; - let question = - match load_pending_interview_question(state.as_ref(), run_id, &submission.qid).await { - Ok(question) => question, - Err(_) => return, - }; - if validate_answer_for_question(&question, &submission.answer).is_err() { + let pending = match load_pending_interview(state.as_ref(), run_id, &submission.qid).await { + Ok(pending) => pending, + Err(_) => return, + }; + if submit_pending_interview_answer(state.as_ref(), &pending, submission.answer) + .await + .is_err() + { return; } - let _ = - deliver_answer_to_run(state.as_ref(), run_id, &submission.qid, submission.answer).await; } } @@ -2879,18 +2877,54 @@ fn worker_command( Ok(cmd) } -fn api_question_from_interview_question(id: &str, question: &Question) -> ApiQuestion { +fn api_question_type(question_type: InterviewQuestionType) -> ApiQuestionType { + match question_type { + InterviewQuestionType::YesNo => ApiQuestionType::YesNo, + InterviewQuestionType::MultipleChoice => ApiQuestionType::MultipleChoice, + InterviewQuestionType::MultiSelect => ApiQuestionType::MultiSelect, + InterviewQuestionType::Freeform => ApiQuestionType::Freeform, + InterviewQuestionType::Confirmation => ApiQuestionType::Confirmation, + } +} + +fn runtime_question_type(question_type: InterviewQuestionType) -> QuestionType { + match question_type { + InterviewQuestionType::YesNo => QuestionType::YesNo, + InterviewQuestionType::MultipleChoice => QuestionType::MultipleChoice, + InterviewQuestionType::MultiSelect => QuestionType::MultiSelect, + InterviewQuestionType::Freeform => QuestionType::Freeform, + InterviewQuestionType::Confirmation => QuestionType::Confirmation, + } +} + +fn runtime_question_from_interview_record(question: &InterviewQuestionRecord) -> Question { + Question { + id: question.id.clone(), + text: question.text.clone(), + question_type: runtime_question_type(question.question_type), + options: question + .options + .iter() + .map(|option| fabro_interview::QuestionOption { + key: option.key.clone(), + label: option.label.clone(), + }) + .collect(), + allow_freeform: question.allow_freeform, + default: None, + timeout_seconds: question.timeout_seconds, + stage: question.stage.clone(), + metadata: HashMap::new(), + context_display: question.context_display.clone(), + } +} + +fn api_question_from_interview_record(question: &InterviewQuestionRecord) -> ApiQuestion { ApiQuestion { - id: id.to_string(), + id: question.id.clone(), text: question.text.clone(), stage: question.stage.clone(), - question_type: match question.question_type { - QuestionType::YesNo => ApiQuestionType::YesNo, - QuestionType::MultipleChoice => ApiQuestionType::MultipleChoice, - QuestionType::MultiSelect => ApiQuestionType::MultiSelect, - QuestionType::Freeform => ApiQuestionType::Freeform, - QuestionType::Confirmation => ApiQuestionType::Confirmation, - }, + question_type: api_question_type(question.question_type), options: question .options .iter() @@ -2905,52 +2939,16 @@ fn api_question_from_interview_question(id: &str, question: &Question) -> ApiQue } } -fn parse_question_type(question_type: &str) -> QuestionType { - match question_type { - "yes_no" => QuestionType::YesNo, - "multiple_choice" => QuestionType::MultipleChoice, - "multi_select" => QuestionType::MultiSelect, - "freeform" => QuestionType::Freeform, - "confirmation" => QuestionType::Confirmation, - _ => QuestionType::Freeform, - } -} - -fn question_from_pending_interview(record: &PendingInterviewRecord) -> Question { - Question { - id: record.question_id.clone(), - text: record.question.clone(), - question_type: parse_question_type(&record.question_type), - options: record - .options - .iter() - .map(|option| fabro_interview::QuestionOption { - key: option.key.clone(), - label: option.label.clone(), - }) - .collect(), - allow_freeform: record.allow_freeform, - default: None, - timeout_seconds: record.timeout_seconds, - stage: record.stage.clone(), - metadata: HashMap::new(), - context_display: record.context_display.clone(), - } -} - fn api_question_from_pending_interview(record: &PendingInterviewRecord) -> ApiQuestion { - api_question_from_interview_question( - &record.question_id, - &question_from_pending_interview(record), - ) + api_question_from_interview_record(&record.question) } #[allow(clippy::result_large_err)] // Axum handlers naturally propagate full `Response` errors. -async fn load_pending_interview_question( +async fn load_pending_interview( state: &AppState, run_id: RunId, qid: &str, -) -> Result { +) -> Result { let run_store = match state.store.open_run_reader(&run_id).await { Ok(run_store) => run_store, Err(fabro_store::StoreError::RunNotFound(_)) => { @@ -2978,25 +2976,38 @@ async fn load_pending_interview_question( .into_response()); }; - Ok(question_from_pending_interview(record)) + Ok(LoadedPendingInterview { + run_id, + qid: qid.to_string(), + question: record.question.clone(), + }) } #[allow(clippy::result_large_err)] // Axum handlers naturally propagate full `Response` errors. -fn validate_answer_for_question(question: &Question, answer: &Answer) -> Result<(), Response> { +fn validate_answer_for_question( + question: &InterviewQuestionRecord, + answer: &Answer, +) -> Result<(), Response> { match (&question.question_type, &answer.value) { - (QuestionType::YesNo | QuestionType::Confirmation, fabro_interview::AnswerValue::Yes) - | (QuestionType::YesNo | QuestionType::Confirmation, fabro_interview::AnswerValue::No) + ( + InterviewQuestionType::YesNo | InterviewQuestionType::Confirmation, + fabro_interview::AnswerValue::Yes, + ) + | ( + InterviewQuestionType::YesNo | InterviewQuestionType::Confirmation, + fabro_interview::AnswerValue::No, + ) | (_, fabro_interview::AnswerValue::Aborted) | (_, fabro_interview::AnswerValue::Skipped) | (_, fabro_interview::AnswerValue::Timeout) => Ok(()), - (QuestionType::MultipleChoice, fabro_interview::AnswerValue::Selected(key)) => { + (InterviewQuestionType::MultipleChoice, fabro_interview::AnswerValue::Selected(key)) => { if question.options.iter().any(|option| option.key == *key) { Ok(()) } else { Err(ApiError::bad_request("Invalid option key.").into_response()) } } - (QuestionType::MultiSelect, fabro_interview::AnswerValue::MultiSelected(keys)) => { + (InterviewQuestionType::MultiSelect, fabro_interview::AnswerValue::MultiSelected(keys)) => { if keys .iter() .all(|key| question.options.iter().any(|option| option.key == *key)) @@ -3006,7 +3017,7 @@ fn validate_answer_for_question(question: &Question, answer: &Answer) -> Result< Err(ApiError::bad_request("Invalid option key.").into_response()) } } - (QuestionType::Freeform, fabro_interview::AnswerValue::Text(text)) + (InterviewQuestionType::Freeform, fabro_interview::AnswerValue::Text(text)) if !text.trim().is_empty() => { Ok(()) @@ -3020,6 +3031,16 @@ fn validate_answer_for_question(question: &Question, answer: &Answer) -> Result< } } +#[allow(clippy::result_large_err)] // Axum handlers naturally propagate full `Response` errors. +async fn submit_pending_interview_answer( + state: &AppState, + pending: &LoadedPendingInterview, + answer: Answer, +) -> Result<(), Response> { + validate_answer_for_question(&pending.question, &answer)?; + deliver_answer_to_run(state, pending.run_id, &pending.qid, answer).await +} + #[allow(clippy::result_large_err)] // Axum handlers naturally propagate full `Response` errors. async fn deliver_answer_to_run( state: &AppState, @@ -3060,7 +3081,10 @@ async fn deliver_answer_to_run( } #[allow(clippy::result_large_err)] // Axum handlers naturally propagate full `Response` errors. -fn answer_from_request(req: SubmitAnswerRequest, question: &Question) -> Result { +fn answer_from_request( + req: SubmitAnswerRequest, + question: &InterviewQuestionRecord, +) -> Result { if let Some(key) = req.selected_option_key { let option = question .options @@ -3068,7 +3092,13 @@ fn answer_from_request(req: SubmitAnswerRequest, question: &Question) -> Result< .find(|option| option.key == key) .cloned(); match option { - Some(option) => Ok(Answer::selected(key, option)), + Some(option) => Ok(Answer::selected( + key, + fabro_interview::QuestionOption { + key: option.key, + label: option.label, + }, + )), None => Err(ApiError::bad_request("Invalid option key.").into_response()), } } else if !req.selected_option_keys.is_empty() { @@ -3420,8 +3450,8 @@ async fn execute_run_in_process(state: Arc, run_id: RunId) { let _ = queued_for; // Create interviewer and event plumbing (this is the "provisioning" phase) - let broker = Arc::new(InterviewBroker::new()); - let interviewer: Arc = Arc::new(ControlInterviewer::new(Arc::clone(&broker))); + let interviewer = Arc::new(ControlInterviewer::new()); + let interview_runtime: Arc = interviewer.clone(); let emitter = Emitter::new(run_id); if let Some(tx_clone) = event_tx { emitter.on_event(move |event| { @@ -3431,7 +3461,7 @@ async fn execute_run_in_process(state: Arc, run_id: RunId) { let registry_override = state .registry_factory_override .as_ref() - .map(|factory| Arc::new(factory(Arc::clone(&interviewer)))); + .map(|factory| Arc::new(factory(Arc::clone(&interview_runtime)))); let emitter = Arc::new(emitter); // Transition to Running, populate interviewer @@ -3441,7 +3471,7 @@ async fn execute_run_in_process(state: Arc, run_id: RunId) { if managed_run.status == RunStatus::Starting { managed_run.status = RunStatus::Running; managed_run.answer_transport = Some(RunAnswerTransport::InProcess { - broker: Arc::clone(&broker), + interviewer: Arc::clone(&interviewer), }); false } else { @@ -3515,7 +3545,7 @@ async fn execute_run_in_process(state: Arc, run_id: RunId) { run_id, cancel_token: Some(Arc::clone(&cancel_token)), emitter: Arc::clone(&emitter), - interviewer: Arc::clone(&interviewer), + interviewer: Arc::clone(&interview_runtime), run_store: run_store.clone().into(), event_sink: workflow_event::RunEventSink::store(run_store.clone()), artifact_uploader: None, @@ -3970,15 +4000,15 @@ async fn submit_answer( Ok(id) => id, Err(response) => return response, }; - let question = match load_pending_interview_question(state.as_ref(), id, &qid).await { - Ok(question) => question, + let pending = match load_pending_interview(state.as_ref(), id, &qid).await { + Ok(pending) => pending, Err(response) => return response, }; - let answer = match answer_from_request(req, &question) { + let answer = match answer_from_request(req, &pending.question) { Ok(answer) => answer, Err(response) => return response, }; - match deliver_answer_to_run(state.as_ref(), id, &qid, answer).await { + match submit_pending_interview_answer(state.as_ref(), &pending, answer).await { Ok(()) => StatusCode::NO_CONTENT.into_response(), Err(response) => response, } @@ -5792,7 +5822,7 @@ mod tests { use fabro_config::server::{ AuthProvider, AuthSettings, GitAuthorSettings, GitProvider, GitSettings, WebSettings, }; - use fabro_types::fixtures; + use fabro_types::{InterviewQuestionRecord, InterviewQuestionType, fixtures}; #[cfg(unix)] use std::process::Stdio; use tower::ServiceExt; @@ -6335,6 +6365,38 @@ mod tests { assert_eq!(response.status(), StatusCode::NOT_FOUND); } + #[tokio::test] + async fn submit_pending_interview_answer_rejects_invalid_answer_shape() { + let state = create_app_state(); + let pending = LoadedPendingInterview { + run_id: fixtures::RUN_1, + qid: "q-1".to_string(), + question: InterviewQuestionRecord { + id: "q-1".to_string(), + text: "Approve deploy?".to_string(), + stage: "gate".to_string(), + question_type: InterviewQuestionType::MultipleChoice, + options: vec![fabro_types::run_event::InterviewOption { + key: "approve".to_string(), + label: "Approve".to_string(), + }], + allow_freeform: false, + timeout_seconds: None, + context_display: None, + }, + }; + + let response = submit_pending_interview_answer( + state.as_ref(), + &pending, + Answer::text("not a valid multiple choice answer"), + ) + .await + .unwrap_err(); + + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + } + #[tokio::test] async fn get_events_not_found() { let app = test_app_with(); diff --git a/lib/crates/fabro-server/tests/it/scenario/lifecycle.rs b/lib/crates/fabro-server/tests/it/scenario/lifecycle.rs index 69d694d69..a1286253b 100644 --- a/lib/crates/fabro-server/tests/it/scenario/lifecycle.rs +++ b/lib/crates/fabro-server/tests/it/scenario/lifecycle.rs @@ -130,7 +130,7 @@ async fn full_http_lifecycle_approve_and_complete() { ))) .header("content-type", "application/json") .body(Body::from( - serde_json::to_string(&serde_json::json!({"value": "A"})).unwrap(), + serde_json::to_string(&serde_json::json!({"selected_option_key": "A"})).unwrap(), )) .unwrap(); let response = app.clone().oneshot(req).await.unwrap(); diff --git a/lib/crates/fabro-store/src/run_state.rs b/lib/crates/fabro-store/src/run_state.rs index b4f8df870..4d793e18e 100644 --- a/lib/crates/fabro-store/src/run_state.rs +++ b/lib/crates/fabro-store/src/run_state.rs @@ -7,13 +7,14 @@ use serde_json::Value; use crate::{EventEnvelope, Result, RunSummary, StageId, StoreError}; use fabro_types::run_event::{ - AgentCliStartedProps, AgentSessionStartedProps, CheckpointCompletedProps, InterviewOption, - RunCompletedProps, RunFailedProps, StageCompletedProps, StagePromptProps, + AgentCliStartedProps, AgentSessionStartedProps, CheckpointCompletedProps, RunCompletedProps, + RunFailedProps, StageCompletedProps, StagePromptProps, }; use fabro_types::{ - BilledModelUsage, Checkpoint, Conclusion, EventBody, FailureSignature, NodeStatusRecord, - Outcome, PullRequestRecord, Retro, RunControlAction, RunEvent, RunId, RunRecord, RunStatus, - RunStatusRecord, SandboxRecord, StageStatus, StartRecord, StatusReason, + BilledModelUsage, Checkpoint, Conclusion, EventBody, FailureSignature, InterviewQuestionRecord, + InterviewQuestionType, NodeStatusRecord, Outcome, PullRequestRecord, Retro, RunControlAction, + RunEvent, RunId, RunRecord, RunStatus, RunStatusRecord, SandboxRecord, StageStatus, + StartRecord, StatusReason, }; #[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)] @@ -39,14 +40,7 @@ pub struct RunProjection { #[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)] pub struct PendingInterviewRecord { - pub question_id: String, - pub question: String, - pub stage: String, - pub question_type: String, - pub options: Vec, - pub allow_freeform: bool, - pub timeout_seconds: Option, - pub context_display: Option, + pub question: InterviewQuestionRecord, pub started_at: Option>, } @@ -213,14 +207,18 @@ impl RunProjection { self.pending_interviews.insert( props.question_id.clone(), PendingInterviewRecord { - question_id: props.question_id.clone(), - question: props.question.clone(), - stage: props.stage.clone(), - question_type: props.question_type.clone(), - options: props.options.clone(), - allow_freeform: props.allow_freeform, - timeout_seconds: props.timeout_seconds, - context_display: props.context_display.clone(), + question: InterviewQuestionRecord { + id: props.question_id.clone(), + text: props.question.clone(), + stage: props.stage.clone(), + question_type: InterviewQuestionType::from_wire_name( + &props.question_type, + ), + options: props.options.clone(), + allow_freeform: props.allow_freeform, + timeout_seconds: props.timeout_seconds, + context_display: props.context_display.clone(), + }, started_at: Some(ts), }, ); @@ -596,7 +594,9 @@ mod tests { use super::{NodeState, RunProjection}; use crate::{EventEnvelope, EventPayload, StageId}; use fabro_types::run_event::{InterviewCompletedProps, InterviewOption, InterviewStartedProps}; - use fabro_types::{Checkpoint, EventBody, RunControlAction, RunEvent, fixtures}; + use fabro_types::{ + Checkpoint, EventBody, InterviewQuestionType, RunControlAction, RunEvent, fixtures, + }; fn test_event(seq: u32, body: EventBody, node_id: Option<&str>) -> EventEnvelope { let event = RunEvent { @@ -754,12 +754,19 @@ mod tests { .pending_interviews .get("q-1") .expect("pending interview should be present"); - assert_eq!(pending.question_id, "q-1"); - assert_eq!(pending.stage, "gate"); - assert_eq!(pending.options.len(), 2); - assert!(pending.allow_freeform); - assert_eq!(pending.timeout_seconds, Some(30.0)); - assert_eq!(pending.context_display.as_deref(), Some("Latest draft")); + assert_eq!(pending.question.id, "q-1"); + assert_eq!(pending.question.stage, "gate"); + assert_eq!( + pending.question.question_type, + InterviewQuestionType::MultipleChoice + ); + assert_eq!(pending.question.options.len(), 2); + assert!(pending.question.allow_freeform); + assert_eq!(pending.question.timeout_seconds, Some(30.0)); + assert_eq!( + pending.question.context_display.as_deref(), + Some("Latest draft") + ); state .apply_event(&test_event( diff --git a/lib/crates/fabro-types/src/interview.rs b/lib/crates/fabro-types/src/interview.rs new file mode 100644 index 000000000..a32e62d3a --- /dev/null +++ b/lib/crates/fabro-types/src/interview.rs @@ -0,0 +1,62 @@ +use std::fmt; + +use serde::{Deserialize, Serialize}; + +use crate::run_event::InterviewOption; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)] +#[serde(rename_all = "snake_case")] +pub enum InterviewQuestionType { + YesNo, + MultipleChoice, + MultiSelect, + #[default] + Freeform, + Confirmation, +} + +impl InterviewQuestionType { + #[must_use] + pub fn from_wire_name(value: &str) -> Self { + match value { + "yes_no" => Self::YesNo, + "multiple_choice" => Self::MultipleChoice, + "multi_select" => Self::MultiSelect, + "freeform" => Self::Freeform, + "confirmation" => Self::Confirmation, + _ => Self::Freeform, + } + } +} + +impl fmt::Display for InterviewQuestionType { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::YesNo => write!(f, "yes_no"), + Self::MultipleChoice => write!(f, "multiple_choice"), + Self::MultiSelect => write!(f, "multi_select"), + Self::Freeform => write!(f, "freeform"), + Self::Confirmation => write!(f, "confirmation"), + } + } +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Default)] +pub struct InterviewQuestionRecord { + #[serde(default)] + pub id: String, + #[serde(default)] + pub text: String, + #[serde(default)] + pub stage: String, + #[serde(default)] + pub question_type: InterviewQuestionType, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub options: Vec, + #[serde(default)] + pub allow_freeform: bool, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub timeout_seconds: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub context_display: Option, +} diff --git a/lib/crates/fabro-types/src/lib.rs b/lib/crates/fabro-types/src/lib.rs index 0084d3a48..0331da290 100644 --- a/lib/crates/fabro-types/src/lib.rs +++ b/lib/crates/fabro-types/src/lib.rs @@ -7,6 +7,7 @@ pub mod combine; pub mod conclusion; pub mod failure_signature; pub mod graph; +pub mod interview; pub mod node_status; pub mod outcome; pub mod pull_request; @@ -35,6 +36,7 @@ pub use conclusion::{Conclusion, StageSummary}; pub use fabro_macros::Combine; pub use failure_signature::FailureSignature; pub use graph::{AttrValue, Edge, Graph, Node, is_llm_handler_type, shape_to_handler_type}; +pub use interview::{InterviewQuestionRecord, InterviewQuestionType}; pub use node_status::NodeStatusRecord; pub use outcome::{FailureCategory, FailureDetail, NodeResult, Outcome, OutcomeMeta, StageStatus}; pub use pull_request::PullRequestRecord;