scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-10-05 02:41:45 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

refactor(interview): simplify pending question handling

Browse source 
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.
This commit is contained in:
Bryan Helmkamp 2026-04-07 19:44:32 -04:00
parent 326e0c27fa
commit 2dd58237b4
No known key found for this signature in database
8 changed files with 352 additions and 187 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

46
lib/crates/fabro-cli/src/commands/run/runner.rs
View file

@ -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<R>(reader: R, broker: Arc<InterviewBroker>)
async fn read_worker_control_stream<R>(reader: R, interviewer: Arc<ControlInterviewer>)
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);
}
}

96
lib/crates/fabro-interview/src/control.rs
View file

@ -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<String, oneshot::Sender<Answer>>,
queued: HashMap<String, Answer>,
closed: bool,
}
#[derive(Default)]
pub struct InterviewBroker {
state: Mutex<InterviewBrokerState>,
pub struct ControlInterviewer {
state: Mutex<ControlInterviewerState>,
}
impl InterviewBroker {
impl ControlInterviewer {
#[must_use]
pub fn new() -> Self {
Self::default()
}
pub async fn register(&self, question_id: String) -> oneshot::Receiver<Answer> {
async fn register(&self, question_id: String) -> oneshot::Receiver<Answer> {
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<InterviewBroker>,
}
impl ControlInterviewer {
#[must_use]
pub fn new(broker: Arc<InterviewBroker>) -> 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);
}
}

10
lib/crates/fabro-interview/src/lib.rs
View file

@ -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);

258
lib/crates/fabro-server/src/server.rs
View file

@ -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<WorkerControlEnvelope>,
},
InProcess {
broker: Arc<InterviewBroker>,
interviewer: Arc<ControlInterviewer>,
},
}
@ -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<Question, Response> {
) -> Result<LoadedPendingInterview, Response> {
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<Answer, Response> {
fn answer_from_request(
req: SubmitAnswerRequest,
question: &InterviewQuestionRecord,
) -> Result<Answer, Response> {
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<AppState>, 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<dyn Interviewer> = Arc::new(ControlInterviewer::new(Arc::clone(&broker)));
let interviewer = Arc::new(ControlInterviewer::new());
let interview_runtime: Arc<dyn Interviewer> = 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<AppState>, 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<AppState>, 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<AppState>, 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();

2
lib/crates/fabro-server/tests/it/scenario/lifecycle.rs
View file

@ -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();

63
lib/crates/fabro-store/src/run_state.rs
View file

@ -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<InterviewOption>,
pub allow_freeform: bool,
pub timeout_seconds: Option<f64>,
pub context_display: Option<String>,
pub question: InterviewQuestionRecord,
pub started_at: Option<DateTime<Utc>>,
}
@ -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(

62
lib/crates/fabro-types/src/interview.rs Normal file
View file

@ -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<InterviewOption>,
#[serde(default)]
pub allow_freeform: bool,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub timeout_seconds: Option<f64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub context_display: Option<String>,
}

2
lib/crates/fabro-types/src/lib.rs
View file

@ -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;