refactor(interview): move run answers onto control channels
Some checks are pending
Rust / Format (push) Waiting to run
Rust / Clippy (push) Waiting to run
Rust / Test (Linux) (push) Waiting to run
Rust / Test (macOS) (push) Waiting to run
TypeScript / Typecheck (push) Waiting to run
TypeScript / Test (push) Waiting to run
TypeScript / Build (push) Waiting to run

Persist pending interviews in run state, deliver accepted answers to workers
through the server-owned control path, and remove the old scratch-file and
WebInterviewer transports.

This also moves Slack onto the canonical server answer flow, adds richer
question metadata to the API and run events, and covers the subprocess
question lifecycle with end-to-end tests.
This commit is contained in:
Bryan Helmkamp 2026-04-07 19:23:36 -04:00
parent 6839ba9e91
commit 326e0c27fa
No known key found for this signature in database
38 changed files with 1830 additions and 1334 deletions

1
Cargo.lock generated
View file

@ -1873,6 +1873,7 @@ dependencies = [
"fabro-proc",
"fabro-retro",
"fabro-sandbox",
"fabro-slack",
"fabro-store",
"fabro-types",
"fabro-util",

View file

@ -34,9 +34,7 @@ These paths are local runtime state, not canonical event projections.
| Path | Purpose |
|---|---|
| `worktree/` | Git worktree used by checkpointed runs |
| `runtime/interview_request.json` | Detached-run interview request IPC file |
| `runtime/interview_response.json` | Detached-run interview response IPC file |
| `runtime/interview_request.claim` | Detached-run interview claim lock |
| `runtime/blobs/` | Materialized local blob payloads for file-backed `fabro+blob://` references |
| `cache/artifacts/values/` | Large context values spilled to the filesystem |
| `cache/artifacts/files/` | Captured artifact files organized by node and retry |

View file

@ -2516,6 +2516,7 @@ components:
required:
- id
- text
- stage
- question_type
- options
- allow_freeform
@ -2528,6 +2529,10 @@ components:
type: string
description: The question text displayed to the user.
example: Should we proceed with the proposed changes?
stage:
type: string
description: Workflow stage identifier that produced the question.
example: gate
question_type:
$ref: "#/components/schemas/QuestionType"
options:
@ -2539,6 +2544,17 @@ components:
type: boolean
description: Whether the user may provide freeform text in addition to selecting options.
example: true
timeout_seconds:
type: number
format: double
nullable: true
description: Timeout for the question when configured by the workflow.
example: 30
context_display:
type: string
nullable: true
description: Optional contextual text shown alongside the question.
example: Latest draft
QuestionType:
description: The interaction type of a human-in-the-loop question.

View file

@ -69,7 +69,7 @@ Manager nodes that run sub-workflows write a nested `child/` directory containin
**`worktree/`** — When running in git checkpoint mode, Fabro creates a Git worktree here as the working directory for agents and commands.
**`runtime/`** — Local-only runtime files, including interview IPC files used by detached runs and `fabro attach`.
**`runtime/`** — Local-only runtime files such as materialized blob payloads under `runtime/blobs/`.
**`cache/`** — Local filesystem cache for file-backed artifacts and captured test artifacts:
@ -104,9 +104,8 @@ fabro ps --filter workflow=my-workflow
│ ├── retro.json
│ ├── cli.log
│ ├── runtime/
│ │ ├── interview_request.json
│ │ ├── interview_response.json
│ │ └── interview_request.claim
│ │ └── blobs/
│ │ └── 01JT5Y3KJ0N5S9E1Y7YFBR2G4D.json
│ ├── cache/
│ │ └── artifacts/
│ │ ├── values/

View file

@ -245,6 +245,7 @@ fn api_question_to_question(question: &types::ApiQuestion) -> Question {
types::QuestionType::Confirmation => QuestionType::Confirmation,
};
let mut converted = Question::new(question.text.clone(), question_type);
converted.id = question.id.clone();
converted.options = question
.options
.iter()
@ -254,6 +255,9 @@ fn api_question_to_question(question: &types::ApiQuestion) -> Question {
})
.collect();
converted.allow_freeform = question.allow_freeform;
converted.stage = question.stage.clone();
converted.timeout_seconds = question.timeout_seconds;
converted.context_display = question.context_display.clone();
converted
}

View file

@ -5,8 +5,9 @@ use std::time::Duration;
use anyhow::{Context, Result, anyhow};
use async_trait::async_trait;
use fabro_config::RunScratch;
use fabro_interview::FileInterviewer;
use fabro_interview::{
ControlInterviewer, InterviewBroker, WorkerControlEnvelope, WorkerControlMessage,
};
use fabro_store::{EventEnvelope, EventPayload, RunProjection};
use fabro_types::{EventBody, RunBlobId, RunEvent, RunId, Settings, StatusReason};
use fabro_workflow::artifact_snapshot::CapturedArtifactInfo;
@ -15,6 +16,7 @@ use fabro_workflow::event::{Emitter, RunEventSink};
use fabro_workflow::operations::{self, StartServices};
use fabro_workflow::run_control::RunControlState;
use fabro_workflow::runtime_store::{RunStoreBackend, RunStoreHandle};
use tokio::io::{self, AsyncBufReadExt, AsyncRead, BufReader};
#[cfg(unix)]
use tokio::signal::unix::{SignalKind, signal};
use tokio::sync::Mutex;
@ -69,12 +71,9 @@ pub(crate) async fn execute(
client.clone_for_reuse(),
artifact_upload_token,
);
let scratch = RunScratch::new(&run_dir);
let interviewer = Arc::new(FileInterviewer::new(
scratch.interview_request_path(),
scratch.interview_response_path(),
scratch.interview_claim_path(),
));
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 run_control = RunControlState::new();
let cancel_token = Arc::new(AtomicBool::new(false));
install_signal_handlers(Arc::clone(&run_control), Arc::clone(&cancel_token))?;
@ -111,6 +110,40 @@ pub(crate) async fn execute(
Ok(())
}
async fn read_worker_control_stream<R>(reader: R, broker: Arc<InterviewBroker>)
where
R: AsyncRead + Unpin,
{
let mut lines = BufReader::new(reader).lines();
loop {
match lines.next_line().await {
Ok(Some(line)) => {
apply_worker_control_line(&broker, &line).await;
}
Ok(None) | Err(_) => {
broker.abort_all().await;
break;
}
}
}
}
async fn apply_worker_control_line(broker: &InterviewBroker, line: &str) {
if line.trim().is_empty() {
return;
}
let Ok(message) = serde_json::from_str::<WorkerControlEnvelope>(line) else {
return;
};
match message.message {
WorkerControlMessage::InterviewAnswer { qid, answer } => {
let _ = broker.submit(&qid, answer.into()).await;
}
}
}
fn build_artifact_uploader(
run_id: RunId,
run_record: &fabro_types::RunRecord,
@ -445,14 +478,17 @@ fn install_signal_handlers(
#[cfg(test)]
mod tests {
use std::sync::Arc;
use httpmock::MockServer;
use serde_json::json;
use super::{
WorkerTitlePhase, execute, initial_worker_title_phase, worker_title,
worker_title_phase_for_event,
WorkerTitlePhase, apply_worker_control_line, execute, initial_worker_title_phase,
read_worker_control_stream, worker_title, worker_title_phase_for_event,
};
use crate::args::RunWorkerMode;
use fabro_interview::{AnswerValue, InterviewBroker};
use fabro_types::fixtures;
use fabro_types::run_event::{
InterviewCompletedProps, InterviewStartedProps, RunCompletedProps, RunControlEffectProps,
@ -499,13 +535,20 @@ mod tests {
);
assert_eq!(
worker_title_phase_for_event(&EventBody::InterviewStarted(InterviewStartedProps {
question_id: "q-1".to_string(),
question: "Approve?".to_string(),
stage: "gate".to_string(),
question_type: "yes_no".to_string(),
options: Vec::new(),
allow_freeform: false,
timeout_seconds: None,
context_display: None,
})),
Some(WorkerTitlePhase::Waiting)
);
assert_eq!(
worker_title_phase_for_event(&EventBody::InterviewCompleted(InterviewCompletedProps {
question_id: "q-1".to_string(),
question: "Approve?".to_string(),
answer: "yes".to_string(),
duration_ms: 10,
@ -602,4 +645,30 @@ mod tests {
state_mock.assert_async().await;
assert_eq!(events_mock.calls_async().await, 0);
}
#[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;
apply_worker_control_line(
&broker,
r#"{"v":1,"type":"interview.answer","qid":"q-1","answer":{"kind":"yes"}}"#,
)
.await;
let answer: fabro_interview::Answer = receiver.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;
read_worker_control_stream(tokio::io::empty(), Arc::clone(&broker)).await;
let answer: fabro_interview::Answer = receiver.await.unwrap();
assert_eq!(answer.value, AnswerValue::Aborted);
}
}

View file

@ -4,6 +4,7 @@ use fabro_types::{EventBody, RunEvent};
use super::support::{run_events, run_state, server_target};
use crate::support::{fabro_json_snapshot, unique_run_id};
use fabro_config::RunScratch;
const SHARED_DAEMON_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
@ -27,6 +28,61 @@ fn assert_worker_succeeded(run_dir: &std::path::Path, stdout: &[u8]) {
)));
}
fn server_endpoint(storage_dir: &std::path::Path) -> (reqwest::Client, String) {
let target = server_target(storage_dir);
if target.starts_with('/') {
(
reqwest::ClientBuilder::new()
.unix_socket(target)
.no_proxy()
.build()
.expect("test Unix-socket HTTP client should build"),
"http://fabro".to_string(),
)
} else {
(
reqwest::ClientBuilder::new()
.no_proxy()
.build()
.expect("test TCP HTTP client should build"),
target,
)
}
}
async fn wait_for_server_question(
client: &reqwest::Client,
base_url: &str,
run_id: &str,
) -> serde_json::Value {
let deadline = std::time::Instant::now() + SHARED_DAEMON_TIMEOUT;
loop {
let response = client
.get(format!("{base_url}/api/v1/runs/{run_id}/questions"))
.query(&[("page[limit]", "100"), ("page[offset]", "0")])
.send()
.await
.expect("question request should succeed");
assert!(
response.status().is_success(),
"question request failed: {}",
response.status()
);
let body: serde_json::Value = response
.json()
.await
.expect("question response should parse");
if let Some(question) = body["data"].as_array().and_then(|items| items.first()) {
return question.clone();
}
assert!(
std::time::Instant::now() < deadline,
"timed out waiting for a pending question"
);
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
}
}
#[test]
fn help() {
let context = test_context!();
@ -341,3 +397,101 @@ digraph Test {
assert_eq!(after_summary, before_summary);
}
#[test]
fn detached_run_answers_pending_question_without_interview_scratch_files() {
let context = test_context!();
let run_id = unique_run_id();
let workflow_path = context.temp_dir.join("human-gate.fabro");
context.write_temp(
"human-gate.fabro",
r#"digraph HumanGate {
graph [goal="Approve the release"]
start [shape=Mdiamond, label="Start"]
exit [shape=Msquare, label="Exit"]
work [shape=parallelogram, script="echo ready"]
approve [shape=hexagon, label="Approve?"]
ship [shape=parallelogram, script="echo shipped"]
revise [shape=parallelogram, script="echo revised"]
start -> work -> approve
approve -> ship [label="[A] Approve"]
approve -> revise [label="[R] Revise"]
ship -> exit
revise -> exit
}
"#,
);
let output = context
.command()
.args([
"run",
"--detach",
"--run-id",
run_id.as_str(),
"--no-retro",
"--sandbox",
"local",
workflow_path.to_str().unwrap(),
])
.timeout(SHARED_DAEMON_TIMEOUT)
.output()
.expect("detached run should execute");
assert!(
output.status.success(),
"detached run failed:\nstdout:\n{}\nstderr:\n{}",
String::from_utf8_lossy(&output.stdout),
String::from_utf8_lossy(&output.stderr)
);
let run_dir = context.find_run_dir(&run_id);
let scratch = RunScratch::new(&run_dir);
let legacy_interview_request = scratch.runtime_dir().join("interview_request.json");
let legacy_interview_response = scratch.runtime_dir().join("interview_response.json");
let runtime = tokio::runtime::Runtime::new().expect("test runtime should build");
let question_id = runtime.block_on(async {
let (client, base_url) = server_endpoint(&context.storage_dir);
let question = wait_for_server_question(&client, &base_url, &run_id).await;
let question_id = question["id"]
.as_str()
.expect("question id should be present")
.to_string();
assert_eq!(question["stage"], "approve");
assert!(
!legacy_interview_request.exists(),
"worker should not create interview_request.json"
);
assert!(
!legacy_interview_response.exists(),
"worker should not create interview_response.json"
);
let response = client
.post(format!(
"{base_url}/api/v1/runs/{run_id}/questions/{question_id}/answer"
))
.json(&serde_json::json!({ "selected_option_key": "A" }))
.send()
.await
.expect("answer submission should succeed");
assert_eq!(response.status(), reqwest::StatusCode::NO_CONTENT);
question_id
});
context
.command()
.args(["wait", &run_id])
.timeout(SHARED_DAEMON_TIMEOUT)
.assert()
.success();
let events = stored_worker_events(&run_dir);
assert!(events.iter().any(|event| matches!(
&event.body,
EventBody::InterviewCompleted(props)
if props.question_id == question_id && props.answer == "A"
)));
}

View file

@ -11,7 +11,7 @@ use crate::run::{
ArtifactsConfig, CheckpointConfig, GitHubConfig, LlmConfig, PullRequestConfig, SetupConfig,
};
use crate::sandbox::SandboxConfig;
use crate::server::{ApiConfig, FeaturesConfig, GitConfig, LogConfig, WebConfig};
use crate::server::{ApiConfig, FeaturesConfig, GitConfig, LogConfig, SlackConfig, WebConfig};
use crate::user::{self, ExecConfig, ServerConfig};
use fabro_types::Settings;
@ -114,6 +114,9 @@ pub struct ConfigLayer {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub web: Option<WebConfig>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub slack: Option<SlackConfig>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub api: Option<ApiConfig>,
@ -173,6 +176,7 @@ impl Combine for ConfigLayer {
max_concurrent_runs: self.max_concurrent_runs.combine(other.max_concurrent_runs),
artifact_storage: self.artifact_storage.combine(other.artifact_storage),
web: self.web.combine(other.web),
slack: self.slack.combine(other.slack),
api: self.api.combine(other.api),
features: self.features.combine(other.features),
log: self.log.combine(other.log),

View file

@ -7,7 +7,7 @@ use fabro_types::Settings;
pub use fabro_types::settings::server::{
ApiAuthStrategy, ApiSettings, ArtifactStorageBackend, ArtifactStorageSettings, AuthProvider,
AuthSettings, FeaturesSettings, GitAuthorSettings, GitProvider, GitSettings, LogSettings,
TlsSettings, WebSettings, WebhookSettings, WebhookStrategy,
SlackSettings, TlsSettings, WebSettings, WebhookSettings, WebhookStrategy,
};
#[derive(Debug, Clone, Default, Deserialize, PartialEq, Serialize, crate::Combine)]
@ -151,6 +151,19 @@ impl From<WebConfig> for WebSettings {
}
}
#[derive(Debug, Clone, Default, Deserialize, PartialEq, Serialize, crate::Combine)]
pub struct SlackConfig {
pub default_channel: Option<String>,
}
impl From<SlackConfig> for SlackSettings {
fn from(value: SlackConfig) -> Self {
Self {
default_channel: value.default_channel,
}
}
}
#[derive(Debug, Clone, Default, Deserialize, PartialEq, Serialize, crate::Combine)]
pub struct FeaturesConfig {
pub session_sandboxes: Option<bool>,

View file

@ -35,6 +35,7 @@ impl TryFrom<ConfigLayer> for Settings {
max_concurrent_runs: value.max_concurrent_runs,
artifact_storage: value.artifact_storage,
web: value.web.map(Into::into),
slack: value.slack.map(Into::into),
api: value.api.map(TryInto::try_into).transpose()?,
features: value.features.map(Into::into),
log: value.log.map(Into::into),

View file

@ -130,21 +130,6 @@ impl RunScratch {
self.artifact_cache_dir().join("files")
}
#[must_use]
pub fn interview_request_path(&self) -> PathBuf {
self.runtime_dir().join("interview_request.json")
}
#[must_use]
pub fn interview_response_path(&self) -> PathBuf {
self.runtime_dir().join("interview_response.json")
}
#[must_use]
pub fn interview_claim_path(&self) -> PathBuf {
self.runtime_dir().join("interview_request.claim")
}
#[must_use]
pub fn artifact_stage_dir(&self, node_slug: &str, attempt: u32) -> PathBuf {
self.artifact_files_dir()
@ -257,27 +242,6 @@ mod tests {
scratch.artifact_files_dir(),
scratch.root().join("cache").join("artifacts").join("files")
);
assert_eq!(
scratch.interview_request_path(),
scratch
.root()
.join("runtime")
.join("interview_request.json")
);
assert_eq!(
scratch.interview_response_path(),
scratch
.root()
.join("runtime")
.join("interview_response.json")
);
assert_eq!(
scratch.interview_claim_path(),
scratch
.root()
.join("runtime")
.join("interview_request.claim")
);
assert_eq!(
scratch.artifact_stage_dir("plan", 2),
scratch

View file

@ -0,0 +1,168 @@
use std::collections::HashMap;
use std::sync::Arc;
use async_trait::async_trait;
use tokio::sync::{Mutex, oneshot};
use crate::{Answer, Interviewer, Question};
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SubmitError {
UnknownQuestion,
AlreadyResolved,
}
#[derive(Default)]
struct InterviewBrokerState {
pending: HashMap<String, oneshot::Sender<Answer>>,
queued: HashMap<String, Answer>,
}
#[derive(Default)]
pub struct InterviewBroker {
state: Mutex<InterviewBrokerState>,
}
impl InterviewBroker {
#[must_use]
pub fn new() -> Self {
Self::default()
}
pub async fn register(&self, question_id: String) -> oneshot::Receiver<Answer> {
let mut state = self.state.lock().await;
if let Some(answer) = state.queued.remove(&question_id) {
let (tx, rx) = oneshot::channel();
let _ = tx.send(answer);
return rx;
}
let (tx, rx) = oneshot::channel();
state.pending.insert(question_id, tx);
rx
}
pub async fn submit(&self, question_id: &str, answer: Answer) -> Result<(), SubmitError> {
let pending_sender = {
let mut state = self.state.lock().await;
if let Some(sender) = state.pending.remove(question_id) {
Some(sender)
} else if state.queued.contains_key(question_id) {
return Err(SubmitError::AlreadyResolved);
} else {
state.queued.insert(question_id.to_string(), answer);
return Ok(());
}
};
match pending_sender {
Some(sender) => sender
.send(answer)
.map_err(|_| SubmitError::AlreadyResolved),
None => Err(SubmitError::UnknownQuestion),
}
}
pub async fn abort_all(&self) {
let (pending, queued) = {
let mut state = self.state.lock().await;
let pending = state
.pending
.drain()
.map(|(_, sender)| sender)
.collect::<Vec<_>>();
let queued = state.queued.len();
state.queued.clear();
(pending, queued)
};
for sender in pending {
let _ = sender.send(Answer::aborted());
}
if queued > 0 {
tracing::debug!(
count = queued,
"Dropped queued interview answers while aborting broker"
);
}
}
}
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;
match receiver.await {
Ok(answer) => answer,
Err(_) => Answer::aborted(),
}
}
async fn inform(&self, _message: &str, _stage: &str) {
// No-op: progress rendering happens via run events.
}
}
#[cfg(test)]
mod tests {
use std::sync::Arc;
use crate::{AnswerValue, QuestionType};
use super::*;
#[tokio::test]
async fn submit_unknown_question_returns_error() {
let broker = InterviewBroker::new();
let result = broker.submit("missing", 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 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;
assert_eq!(submit_result, Ok(()));
let answer = ask.await.unwrap();
assert_eq!(answer.value, AnswerValue::Yes);
}
#[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 receiver = broker.register("q-1".to_string()).await;
let answer = receiver.await.unwrap();
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(()));
assert_eq!(
broker.submit("q-1", Answer::no()).await,
Err(SubmitError::AlreadyResolved)
);
}
}

View file

@ -0,0 +1,100 @@
use serde::{Deserialize, Serialize};
use crate::{Answer, AnswerValue};
pub const WORKER_CONTROL_PROTOCOL_VERSION: u8 = 1;
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct WorkerControlEnvelope {
pub v: u8,
#[serde(flatten)]
pub message: WorkerControlMessage,
}
impl WorkerControlEnvelope {
#[must_use]
pub fn interview_answer(qid: impl Into<String>, answer: Answer) -> Self {
Self {
v: WORKER_CONTROL_PROTOCOL_VERSION,
message: WorkerControlMessage::InterviewAnswer {
qid: qid.into(),
answer: answer.into(),
},
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "type")]
pub enum WorkerControlMessage {
#[serde(rename = "interview.answer")]
InterviewAnswer {
qid: String,
answer: WorkerControlAnswer,
},
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum WorkerControlAnswer {
Yes,
No,
Aborted,
Skipped,
Timeout,
Selected { key: String },
MultiSelected { keys: Vec<String> },
Text { text: String },
}
impl From<Answer> for WorkerControlAnswer {
fn from(answer: Answer) -> Self {
match answer.value {
AnswerValue::Yes => Self::Yes,
AnswerValue::No => Self::No,
AnswerValue::Aborted => Self::Aborted,
AnswerValue::Skipped => Self::Skipped,
AnswerValue::Timeout => Self::Timeout,
AnswerValue::Selected(key) => Self::Selected { key },
AnswerValue::MultiSelected(keys) => Self::MultiSelected { keys },
AnswerValue::Text(text) => Self::Text { text },
}
}
}
impl From<WorkerControlAnswer> for Answer {
fn from(answer: WorkerControlAnswer) -> Self {
match answer {
WorkerControlAnswer::Yes => Self::yes(),
WorkerControlAnswer::No => Self::no(),
WorkerControlAnswer::Aborted => Self::aborted(),
WorkerControlAnswer::Skipped => Self::skipped(),
WorkerControlAnswer::Timeout => Self::timeout(),
WorkerControlAnswer::Selected { key } => Self {
value: AnswerValue::Selected(key),
selected_option: None,
text: None,
},
WorkerControlAnswer::MultiSelected { keys } => Self::multi_selected(keys),
WorkerControlAnswer::Text { text } => Self::text(text),
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn interview_answer_round_trips_through_json() {
let envelope = WorkerControlEnvelope::interview_answer("q-1", Answer::text("ship it"));
let json = serde_json::to_string(&envelope).unwrap();
assert_eq!(
json,
r#"{"v":1,"type":"interview.answer","qid":"q-1","answer":{"kind":"text","text":"ship it"}}"#
);
let parsed: WorkerControlEnvelope = serde_json::from_str(&json).unwrap();
assert_eq!(parsed, envelope);
}
}

View file

@ -1,393 +0,0 @@
use std::path::PathBuf;
use std::time::Duration;
use async_trait::async_trait;
use tokio::fs;
use tokio::time;
use crate::{Answer, Interviewer, Question};
#[cfg(test)]
use std::path::Path;
/// An interviewer that communicates via JSON files in the runtime directory.
///
/// The engine process writes `interview_request.json` and polls for
/// `interview_response.json`. The attach process watches for the request
/// file, prompts the user, and writes the response file.
#[allow(clippy::struct_field_names)]
pub struct FileInterviewer {
request_path: PathBuf,
response_path: PathBuf,
claim_path: PathBuf,
poll_interval: Duration,
reattach_window: Duration,
}
#[cfg(test)]
const DEFAULT_REATTACH_WINDOW: Duration = Duration::from_millis(300);
#[cfg(not(test))]
const DEFAULT_REATTACH_WINDOW: Duration = Duration::from_secs(30);
const DEFAULT_POLL_INTERVAL: Duration = Duration::from_millis(100);
#[cfg(test)]
const TEST_POLL_INTERVAL: Duration = Duration::from_millis(1);
#[cfg(test)]
const TEST_REATTACH_WINDOW: Duration = Duration::from_millis(5);
fn default_reattach_window() -> Duration {
DEFAULT_REATTACH_WINDOW
}
fn default_poll_interval() -> Duration {
DEFAULT_POLL_INTERVAL
}
impl FileInterviewer {
pub fn new(request_path: PathBuf, response_path: PathBuf, claim_path: PathBuf) -> Self {
Self {
request_path,
response_path,
claim_path,
poll_interval: default_poll_interval(),
reattach_window: default_reattach_window(),
}
}
#[cfg(test)]
fn with_timing(
request_path: PathBuf,
response_path: PathBuf,
claim_path: PathBuf,
poll_interval: Duration,
reattach_window: Duration,
) -> Self {
Self {
request_path,
response_path,
claim_path,
poll_interval,
reattach_window,
}
}
fn request_path(&self) -> PathBuf {
self.request_path.clone()
}
fn response_path(&self) -> PathBuf {
self.response_path.clone()
}
fn claim_path(&self) -> PathBuf {
self.claim_path.clone()
}
async fn write_request_atomically(&self, question: &Question) -> std::io::Result<()> {
let json = serde_json::to_string_pretty(question).expect("Question serialization failed");
let request_path = self.request_path();
if let Some(parent) = request_path.parent() {
fs::create_dir_all(parent).await?;
}
let temp_path = request_path.with_extension("json.tmp");
fs::write(&temp_path, json).await?;
fs::rename(temp_path, request_path).await
}
async fn cleanup_ipc_files(&self) {
let _ = fs::remove_file(self.request_path()).await;
let _ = fs::remove_file(self.response_path()).await;
let _ = fs::remove_file(self.claim_path()).await;
}
}
#[async_trait]
impl Interviewer for FileInterviewer {
async fn ask(&self, question: Question) -> Answer {
let timeout_secs = question.timeout_seconds;
let default_answer = question.default.clone();
// Write the request file
if let Err(e) = self.write_request_atomically(&question).await {
tracing::warn!(error = %e, "Failed to write interview request");
return default_answer.unwrap_or_else(Answer::timeout);
}
// Poll for response with optional timeout
let default_for_claim_timeout = default_answer.clone();
let poll = async {
let response_path = self.response_path();
let claim_path = self.claim_path();
let mut claim_was_seen = false;
let mut reattach_deadline: Option<time::Instant> = None;
loop {
match fs::read_to_string(&response_path).await {
Ok(data) => match serde_json::from_str::<Answer>(&data) {
Ok(answer) => {
self.cleanup_ipc_files().await;
return answer;
}
Err(e) => {
tracing::warn!(error = %e, "Failed to parse interview response, retrying");
// File might be partially written, wait and retry
}
},
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
// Not written yet — check claim state below
}
Err(e) => {
tracing::warn!(error = %e, "Failed to read interview response, retrying");
}
}
// Monitor claim file to detect attacher departure
if claim_path.exists() {
claim_was_seen = true;
reattach_deadline = None;
} else if claim_was_seen && reattach_deadline.is_none() {
reattach_deadline = Some(time::Instant::now() + self.reattach_window);
}
if let Some(deadline) = reattach_deadline {
if time::Instant::now() >= deadline {
self.cleanup_ipc_files().await;
return default_for_claim_timeout.unwrap_or_else(Answer::timeout);
}
}
time::sleep(self.poll_interval).await;
}
};
if let Some(secs) = timeout_secs {
let duration = std::time::Duration::from_secs_f64(secs);
if let Ok(answer) = time::timeout(duration, poll).await {
answer
} else {
self.cleanup_ipc_files().await;
default_answer.unwrap_or_else(Answer::timeout)
}
} else {
poll.await
}
}
async fn inform(&self, _message: &str, _stage: &str) {
// No-op: inform messages are rendered by the attach process via progress.jsonl
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::{AnswerValue, QuestionType};
fn test_interviewer(
request_path: PathBuf,
response_path: PathBuf,
claim_path: PathBuf,
) -> FileInterviewer {
FileInterviewer::with_timing(
request_path,
response_path,
claim_path,
TEST_POLL_INTERVAL,
TEST_REATTACH_WINDOW,
)
}
fn interviewer_paths(run_dir: &Path) -> (PathBuf, PathBuf, PathBuf) {
let runtime_dir = run_dir.join("runtime");
(
runtime_dir.join("interview_request.json"),
runtime_dir.join("interview_response.json"),
runtime_dir.join("interview_request.claim"),
)
}
async fn wait_for_exists(path: &Path) {
for _ in 0..200 {
if path.exists() {
return;
}
time::sleep(TEST_POLL_INTERVAL).await;
}
panic!("{} should exist", path.display());
}
async fn wait_for_claim_observation() {
// Full-workspace test load can delay the poller enough that a 2ms wait is flaky.
time::sleep(Duration::from_millis(25)).await;
}
#[tokio::test]
async fn write_request_poll_response() {
let dir = tempfile::tempdir().unwrap();
let run_dir = dir.path().to_path_buf();
let (request_path, response_path, claim_path) = interviewer_paths(&run_dir);
let interviewer = test_interviewer(
request_path.clone(),
response_path.clone(),
claim_path.clone(),
);
let question = Question::new("approve?", QuestionType::YesNo);
// Spawn the ask in a background task
let ask_handle = tokio::spawn(async move { interviewer.ask(question).await });
// Wait for the request file to appear
wait_for_exists(&request_path).await;
// Verify the request contains valid Question JSON
let request_data = fs::read_to_string(&request_path).await.unwrap();
let parsed: Question = serde_json::from_str(&request_data).unwrap();
assert_eq!(parsed.text, "approve?");
// Write a response
let answer = Answer::yes();
let response_json = serde_json::to_string_pretty(&answer).unwrap();
fs::write(&response_path, response_json).await.unwrap();
// Wait for the ask to complete
let result = ask_handle.await.unwrap();
assert_eq!(result.value, AnswerValue::Yes);
// Both files should be cleaned up
assert!(!request_path.exists());
assert!(!response_path.exists());
assert!(!claim_path.exists());
}
#[tokio::test]
async fn timeout_returns_default() {
let dir = tempfile::tempdir().unwrap();
let (request_path, response_path, claim_path) = interviewer_paths(dir.path());
let interviewer = test_interviewer(request_path, response_path, claim_path);
let mut question = Question::new("approve?", QuestionType::YesNo);
question.timeout_seconds = Some(0.02);
question.default = Some(Answer::no());
let answer = interviewer.ask(question).await;
assert_eq!(answer.value, AnswerValue::No);
}
#[tokio::test]
async fn claim_released_without_response_returns_timeout() {
let dir = tempfile::tempdir().unwrap();
let run_dir = dir.path().to_path_buf();
let (request_path, response_path, claim_path) = interviewer_paths(&run_dir);
let interviewer = test_interviewer(request_path.clone(), response_path, claim_path.clone());
let question = Question::new("approve?", QuestionType::YesNo);
let ask_handle = tokio::spawn(async move { interviewer.ask(question).await });
// Wait for request file to appear
wait_for_exists(&request_path).await;
// Simulate attacher creating claim file
std::fs::write(&claim_path, "12345\n").unwrap();
// Let the poll loop see the claim
wait_for_claim_observation().await;
// Simulate attacher departing (deletes claim without writing response)
std::fs::remove_file(&claim_path).unwrap();
// Should return timeout within REATTACH_WINDOW
let started = time::Instant::now();
let answer = time::timeout(Duration::from_millis(250), ask_handle)
.await
.expect("should complete quickly")
.unwrap();
assert_eq!(answer.value, AnswerValue::Timeout);
assert!(
started.elapsed() <= Duration::from_millis(100),
"should resolve well within the reattach window"
);
}
#[tokio::test]
async fn claim_released_without_response_returns_default() {
let dir = tempfile::tempdir().unwrap();
let run_dir = dir.path().to_path_buf();
let (request_path, response_path, claim_path) = interviewer_paths(&run_dir);
let interviewer = test_interviewer(request_path.clone(), response_path, claim_path.clone());
let mut question = Question::new("approve?", QuestionType::YesNo);
question.default = Some(Answer::no());
let ask_handle = tokio::spawn(async move { interviewer.ask(question).await });
// Wait for request file
wait_for_exists(&request_path).await;
// Simulate attacher creating then deleting claim
std::fs::write(&claim_path, "12345\n").unwrap();
wait_for_claim_observation().await;
std::fs::remove_file(&claim_path).unwrap();
let answer = time::timeout(Duration::from_millis(250), ask_handle)
.await
.expect("should complete quickly")
.unwrap();
assert_eq!(answer.value, AnswerValue::No);
}
#[tokio::test]
async fn claim_released_then_new_attacher_answers() {
let dir = tempfile::tempdir().unwrap();
let run_dir = dir.path().to_path_buf();
let (request_path, response_path, claim_path) = interviewer_paths(&run_dir);
let interviewer = test_interviewer(
request_path.clone(),
response_path.clone(),
claim_path.clone(),
);
let question = Question::new("approve?", QuestionType::YesNo);
let ask_handle = tokio::spawn(async move { interviewer.ask(question).await });
// Wait for request file
wait_for_exists(&request_path).await;
// First attacher creates then releases claim
std::fs::write(&claim_path, "12345\n").unwrap();
wait_for_claim_observation().await;
std::fs::remove_file(&claim_path).unwrap();
// Second attacher picks up and answers before reattach window expires
time::sleep(TEST_POLL_INTERVAL * 2).await;
std::fs::write(&claim_path, "12346\n").unwrap();
let answer = Answer::yes();
let response_json = serde_json::to_string_pretty(&answer).unwrap();
fs::write(response_path, response_json).await.unwrap();
let result = time::timeout(Duration::from_millis(250), ask_handle)
.await
.expect("should complete quickly")
.unwrap();
assert_eq!(result.value, AnswerValue::Yes);
}
#[tokio::test]
async fn timeout_without_default_returns_timeout() {
let dir = tempfile::tempdir().unwrap();
let (request_path, response_path, claim_path) = interviewer_paths(dir.path());
let interviewer = test_interviewer(request_path, response_path, claim_path);
let mut question = Question::new("approve?", QuestionType::YesNo);
question.timeout_seconds = Some(0.02);
let answer = interviewer.ask(question).await;
assert_eq!(answer.value, AnswerValue::Timeout);
}
}

View file

@ -1,11 +1,11 @@
mod auto_approve;
mod callback;
mod console;
pub mod file;
mod control;
mod control_protocol;
mod queue;
mod recording;
mod replay;
mod web;
use std::collections::HashMap;
@ -45,6 +45,8 @@ pub struct QuestionOption {
/// A question presented to the user.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Question {
#[serde(default)]
pub id: String,
pub text: String,
pub question_type: QuestionType,
pub options: Vec<QuestionOption>,
@ -60,6 +62,7 @@ pub struct Question {
impl Question {
pub fn new(text: impl Into<String>, question_type: QuestionType) -> Self {
Self {
id: String::new(),
text: text.into(),
question_type,
options: Vec::new(),
@ -206,14 +209,19 @@ pub trait Interviewer: Send + Sync {
pub use auto_approve::AutoApproveInterviewer;
pub use callback::CallbackInterviewer;
pub use console::ConsoleInterviewer;
pub use file::FileInterviewer;
pub use control::{ControlInterviewer, InterviewBroker, SubmitError};
pub use control_protocol::{
WORKER_CONTROL_PROTOCOL_VERSION, WorkerControlAnswer, WorkerControlEnvelope,
WorkerControlMessage,
};
pub use queue::QueueInterviewer;
pub use recording::RecordingInterviewer;
pub use replay::ReplayInterviewer;
pub use web::{PendingQuestion, WebInterviewer};
#[cfg(test)]
mod tests {
use std::sync::Arc;
use super::*;
use tokio::time;
@ -229,6 +237,7 @@ mod tests {
#[test]
fn question_new() {
let q = Question::new("Do you approve?", QuestionType::YesNo);
assert!(q.id.is_empty());
assert_eq!(q.text, "Do you approve?");
assert_eq!(q.question_type, QuestionType::YesNo);
assert!(q.options.is_empty());
@ -363,4 +372,21 @@ mod tests {
let answer = ask_with_timeout(&interviewer, q).await;
assert_eq!(answer.value, AnswerValue::Yes);
}
#[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 mut question = Question::new("Approve?", QuestionType::YesNo);
question.id = "q-1".to_string();
let ask = tokio::spawn(async move { interviewer.ask(question).await });
time::sleep(std::time::Duration::from_millis(10)).await;
broker.submit("q-1", Answer::yes()).await.unwrap();
let answer = ask.await.unwrap();
assert_eq!(answer.value, AnswerValue::Yes);
}
}

View file

@ -1,324 +0,0 @@
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use async_trait::async_trait;
use tokio::sync::oneshot;
use crate::{Answer, Interviewer, Question};
/// A pending question waiting for an answer from an external source (e.g., HTTP endpoint).
#[derive(Debug)]
pub struct PendingQuestion {
pub id: String,
pub question: Question,
}
/// Internal state: maps question ID to its oneshot sender.
struct WebInterviewerInner {
pending: HashMap<String, oneshot::Sender<Answer>>,
questions: Vec<PendingQuestion>,
next_id: u64,
}
/// An interviewer that holds questions until answers are submitted externally.
///
/// When `ask()` is called, the question is enqueued with a unique ID and the call
/// blocks until `submit_answer()` is called with the matching ID.
pub struct WebInterviewer {
inner: Arc<Mutex<WebInterviewerInner>>,
}
impl WebInterviewer {
#[must_use]
pub fn new() -> Self {
Self {
inner: Arc::new(Mutex::new(WebInterviewerInner {
pending: HashMap::new(),
questions: Vec::new(),
next_id: 1,
})),
}
}
/// Returns a snapshot of currently pending questions.
///
/// # Panics
///
/// Panics if the internal lock is poisoned.
#[must_use]
pub fn pending_questions(&self) -> Vec<PendingQuestion> {
let inner = self.inner.lock().expect("web interviewer lock poisoned");
inner
.questions
.iter()
.map(|pq| PendingQuestion {
id: pq.id.clone(),
question: pq.question.clone(),
})
.collect()
}
/// Submit an answer for a pending question by ID.
/// Returns `true` if the question was found and the answer was delivered,
/// `false` if no such question was pending.
///
/// # Panics
///
/// Panics if the internal lock is poisoned.
#[must_use]
pub fn submit_answer(&self, question_id: &str, answer: Answer) -> bool {
let sender = {
let mut inner = self.inner.lock().expect("web interviewer lock poisoned");
let sender = inner.pending.remove(question_id);
if sender.is_some() {
inner.questions.retain(|pq| pq.id != question_id);
}
sender
};
sender.is_some_and(|tx| tx.send(answer).is_ok())
}
/// Abort all pending questions, unblocking any waiting `ask()` calls.
pub fn abort_pending(&self) {
let pending = {
let mut inner = self.inner.lock().expect("web interviewer lock poisoned");
inner.questions.clear();
inner.pending.drain().map(|(_, tx)| tx).collect::<Vec<_>>()
};
for sender in pending {
let _ = sender.send(Answer::aborted());
}
}
}
impl Default for WebInterviewer {
fn default() -> Self {
Self::new()
}
}
#[async_trait]
impl Interviewer for WebInterviewer {
async fn ask(&self, question: Question) -> Answer {
let (tx, rx) = oneshot::channel();
{
let mut inner = self.inner.lock().expect("web interviewer lock poisoned");
let id = format!("q-{}", inner.next_id);
inner.next_id += 1;
inner.pending.insert(id.clone(), tx);
inner.questions.push(PendingQuestion {
id,
question: question.clone(),
});
}
// Block until answer arrives or sender is dropped
rx.await.unwrap_or_else(|_| Answer::aborted())
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::{AnswerValue, QuestionType};
use std::sync::Arc;
use std::time::Duration;
use tokio::time::sleep;
async fn wait_for_pending_count(interviewer: &WebInterviewer, expected: usize) {
for _ in 0..200 {
if interviewer.pending_questions().len() == expected {
return;
}
sleep(Duration::from_millis(1)).await;
}
panic!("pending question count did not reach {expected}");
}
#[tokio::test]
async fn ask_blocks_until_answer_submitted() {
let interviewer = Arc::new(WebInterviewer::new());
let interviewer_clone = Arc::clone(&interviewer);
let ask_handle = tokio::spawn(async move {
let q = Question::new("approve?", QuestionType::YesNo);
interviewer_clone.ask(q).await
});
wait_for_pending_count(interviewer.as_ref(), 1).await;
let pending = interviewer.pending_questions();
assert_eq!(pending.len(), 1);
assert_eq!(pending[0].question.text, "approve?");
// Submit answer
let submitted = interviewer.submit_answer(&pending[0].id, Answer::yes());
assert!(submitted);
// ask() should now return
let answer = ask_handle.await.expect("task should complete");
assert_eq!(answer.value, AnswerValue::Yes);
}
#[tokio::test]
async fn submit_answer_unblocks_ask() {
let interviewer = Arc::new(WebInterviewer::new());
let interviewer_clone = Arc::clone(&interviewer);
let ask_handle = tokio::spawn(async move {
let q = Question::new("name?", QuestionType::Freeform);
interviewer_clone.ask(q).await
});
wait_for_pending_count(interviewer.as_ref(), 1).await;
let pending = interviewer.pending_questions();
assert_eq!(pending.len(), 1);
let _ = interviewer.submit_answer(&pending[0].id, Answer::text("Alice"));
let answer = ask_handle.await.expect("task should complete");
assert_eq!(answer.value, AnswerValue::Text("Alice".to_string()));
assert_eq!(answer.text, Some("Alice".to_string()));
}
#[tokio::test]
async fn timeout_returns_default_or_timeout_answer() {
let interviewer = Arc::new(WebInterviewer::new());
let mut q = Question::new("approve?", QuestionType::YesNo);
q.timeout_seconds = Some(0.05);
// Use ask_with_timeout from the parent module
let answer = crate::ask_with_timeout(interviewer.as_ref(), q).await;
assert_eq!(answer.value, AnswerValue::Timeout);
}
#[tokio::test]
async fn question_id_correlation() {
let interviewer = Arc::new(WebInterviewer::new());
let i1 = Arc::clone(&interviewer);
let i2 = Arc::clone(&interviewer);
// Spawn two concurrent asks
let handle1 = tokio::spawn(async move {
let q = Question::new("first?", QuestionType::YesNo);
i1.ask(q).await
});
let handle2 = tokio::spawn(async move {
let q = Question::new("second?", QuestionType::YesNo);
i2.ask(q).await
});
wait_for_pending_count(interviewer.as_ref(), 2).await;
let pending = interviewer.pending_questions();
assert_eq!(pending.len(), 2);
// Find which ID corresponds to which question
let first_id = pending
.iter()
.find(|pq| pq.question.text == "first?")
.expect("first question should be pending")
.id
.clone();
let second_id = pending
.iter()
.find(|pq| pq.question.text == "second?")
.expect("second question should be pending")
.id
.clone();
// Answer them in reverse order
let _ = interviewer.submit_answer(&second_id, Answer::no());
let _ = interviewer.submit_answer(&first_id, Answer::yes());
let answer1 = handle1.await.expect("task should complete");
let answer2 = handle2.await.expect("task should complete");
assert_eq!(answer1.value, AnswerValue::Yes);
assert_eq!(answer2.value, AnswerValue::No);
}
#[test]
fn submit_answer_for_unknown_id_returns_false() {
let interviewer = WebInterviewer::new();
let result = interviewer.submit_answer("nonexistent", Answer::yes());
assert!(!result);
}
#[tokio::test]
async fn pending_questions_empty_initially() {
let interviewer = WebInterviewer::new();
assert!(interviewer.pending_questions().is_empty());
}
#[tokio::test]
async fn pending_questions_cleared_after_answer() {
let interviewer = Arc::new(WebInterviewer::new());
let i_clone = Arc::clone(&interviewer);
let handle = tokio::spawn(async move {
let q = Question::new("q?", QuestionType::YesNo);
i_clone.ask(q).await
});
wait_for_pending_count(interviewer.as_ref(), 1).await;
let pending = interviewer.pending_questions();
assert_eq!(pending.len(), 1);
let _ = interviewer.submit_answer(&pending[0].id, Answer::yes());
handle.await.expect("task should complete");
assert!(interviewer.pending_questions().is_empty());
}
#[tokio::test]
async fn abort_pending_unblocks_ask_and_clears_questions() {
let interviewer = Arc::new(WebInterviewer::new());
let interviewer_clone = Arc::clone(&interviewer);
let ask_handle = tokio::spawn(async move {
let q = Question::new("approve?", QuestionType::YesNo);
interviewer_clone.ask(q).await
});
wait_for_pending_count(interviewer.as_ref(), 1).await;
interviewer.abort_pending();
let answer = ask_handle.await.expect("task should complete");
assert_eq!(answer.value, AnswerValue::Aborted);
assert!(interviewer.pending_questions().is_empty());
}
#[tokio::test]
async fn ask_returns_aborted_when_pending_sender_is_dropped() {
let interviewer = Arc::new(WebInterviewer::new());
let interviewer_clone = Arc::clone(&interviewer);
let ask_handle = tokio::spawn(async move {
let q = Question::new("approve?", QuestionType::YesNo);
interviewer_clone.ask(q).await
});
wait_for_pending_count(interviewer.as_ref(), 1).await;
{
let mut inner = interviewer
.inner
.lock()
.expect("web interviewer lock poisoned");
let pending_id = inner
.questions
.first()
.expect("question should be pending")
.id
.clone();
inner.pending.remove(&pending_id);
inner.questions.retain(|pq| pq.id != pending_id);
}
let answer = ask_handle.await.expect("task should complete");
assert_eq!(answer.value, AnswerValue::Aborted);
}
}

View file

@ -17,6 +17,7 @@ fabro-config = { path = "../fabro-config" }
fabro-graphviz = { path = "../fabro-graphviz" }
fabro-hooks = { path = "../fabro-hooks" }
fabro-interview = { path = "../fabro-interview" }
fabro-slack = { path = "../fabro-slack" }
fabro-workflow = { path = "../fabro-workflow" }
fabro-validate = { path = "../fabro-validate" }
fabro-sandbox = { path = "../fabro-sandbox", features = ["daytona"] }

View file

@ -1265,6 +1265,7 @@ mod runs {
ApiQuestion {
id: "q-001".into(),
text: "Should we proceed with the proposed changes?".into(),
stage: "review".into(),
question_type: QuestionType::YesNo,
options: vec![
ApiQuestionOption {
@ -1277,10 +1278,13 @@ mod runs {
},
],
allow_freeform: false,
timeout_seconds: None,
context_display: None,
},
ApiQuestion {
id: "q-002".into(),
text: "Which approach do you prefer for the migration?".into(),
stage: "migration".into(),
question_type: QuestionType::MultipleChoice,
options: vec![
ApiQuestionOption {
@ -1293,6 +1297,8 @@ mod runs {
},
],
allow_freeform: true,
timeout_seconds: None,
context_display: None,
},
]
}

View file

@ -30,7 +30,9 @@ use fabro_llm::types::{
Response as LlmResponse, Role, StreamEvent, TokenCounts, ToolChoice, ToolDefinition,
};
use fabro_model::{BilledModelUsage, BilledTokenCounts};
use fabro_store::{ArtifactStore, Database, EventEnvelope, EventPayload, StageId};
use fabro_store::{
ArtifactStore, Database, EventEnvelope, EventPayload, PendingInterviewRecord, StageId,
};
use fabro_types::{
EventBody, RunArtifactStorage, RunBlobId, RunClientProvenance, RunControlAction, RunEvent,
RunId, RunProvenance, RunServerProvenance, RunSubjectProvenance, Settings,
@ -48,7 +50,7 @@ use sha2::{Digest, Sha256};
use tempfile::NamedTempFile;
use tokio::fs;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::process::{ChildStderr, Command};
use tokio::process::{ChildStderr, ChildStdin, Command};
use tokio::sync::Notify;
use tokio::sync::RwLock as AsyncRwLock;
use tokio::sync::broadcast;
@ -74,10 +76,19 @@ use crate::run_manifest;
use crate::secret_store::{SecretStore, SecretStoreError};
use crate::static_files;
use crate::web_auth;
use fabro_interview::{Answer, Interviewer, Question, QuestionType, WebInterviewer};
use fabro_interview::{
Answer, ControlInterviewer, InterviewBroker, Interviewer, Question, QuestionType,
WorkerControlEnvelope,
};
use fabro_sandbox::daytona::DaytonaSandbox;
use fabro_sandbox::reconnect::reconnect;
use fabro_sandbox::{Sandbox, SandboxProvider};
use fabro_slack::blocks as slack_blocks;
use fabro_slack::client::{PostedMessage as SlackPostedMessage, SlackClient};
use fabro_slack::config::resolve_credentials as resolve_slack_credentials;
use fabro_slack::connection as slack_connection;
use fabro_slack::payload::SlackAnswerSubmission;
use fabro_slack::threads::ThreadRegistry;
use fabro_workflow::event::{self as workflow_event, Emitter};
use fabro_workflow::operations::{self};
use fabro_workflow::pipeline::Persisted;
@ -214,7 +225,8 @@ struct ManagedRun {
created_at: chrono::DateTime<chrono::Utc>,
enqueued_at: Instant,
// Populated when running:
interviewer: Option<Arc<WebInterviewer>>,
answer_transport: Option<RunAnswerTransport>,
accepted_questions: HashSet<String>,
event_tx: Option<broadcast::Sender<RunEvent>>,
checkpoint: Option<Checkpoint>,
cancel_tx: Option<oneshot::Sender<()>>,
@ -236,9 +248,10 @@ enum ExecutionResult {
CancelledBySignal,
}
const FILE_INTERVIEW_QUESTION_ID: &str = "q-file";
const WORKER_STDERR_LOG: &str = "worker.stderr.log";
const WORKER_CANCEL_GRACE: Duration = Duration::from_secs(5);
const WORKER_CONTROL_QUEUE_CAPACITY: usize = 8;
const WORKER_CONTROL_ENQUEUE_TIMEOUT: Duration = Duration::from_secs(1);
const ARTIFACT_UPLOAD_TOKEN_ISSUER: &str = "fabro-server-artifact-upload";
const ARTIFACT_UPLOAD_TOKEN_SCOPE: &str = "stage_artifacts:upload";
const ARTIFACT_UPLOAD_TOKEN_TTL_SECS: u64 = 24 * 60 * 60;
@ -297,6 +310,205 @@ struct BillingAccumulator {
type RegistryFactoryOverride = dyn Fn(Arc<dyn Interviewer>) -> HandlerRegistry + Send + Sync;
#[derive(Clone)]
enum RunAnswerTransport {
Subprocess {
control_tx: mpsc::Sender<WorkerControlEnvelope>,
},
InProcess {
broker: Arc<InterviewBroker>,
},
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum AnswerTransportError {
Closed,
Timeout,
}
impl RunAnswerTransport {
async fn submit(&self, qid: &str, answer: Answer) -> Result<(), AnswerTransportError> {
match self {
Self::Subprocess { control_tx } => {
let message = WorkerControlEnvelope::interview_answer(qid.to_string(), answer);
tokio::time::timeout(WORKER_CONTROL_ENQUEUE_TIMEOUT, control_tx.send(message))
.await
.map_err(|_| AnswerTransportError::Timeout)?
.map_err(|_| AnswerTransportError::Closed)
}
Self::InProcess { broker } => broker
.submit(qid, answer)
.await
.map_err(|_| AnswerTransportError::Closed),
}
}
async fn abort_pending(&self) {
if let Self::InProcess { broker } = self {
broker.abort_all().await;
}
}
}
#[derive(Clone)]
struct SlackService {
client: SlackClient,
app_token: String,
default_channel: String,
posted_messages: Arc<Mutex<HashMap<(RunId, String), SlackPostedMessage>>>,
thread_registry: Arc<ThreadRegistry>,
}
impl SlackService {
fn new(bot_token: String, app_token: String, default_channel: String) -> Self {
Self {
client: SlackClient::new(bot_token),
app_token,
default_channel,
posted_messages: Arc::new(Mutex::new(HashMap::new())),
thread_registry: Arc::new(ThreadRegistry::new()),
}
}
async fn handle_event(&self, event: &RunEvent) {
match &event.body {
EventBody::InterviewStarted(props) => {
if props.question_id.is_empty() {
return;
}
let key = (event.run_id, props.question_id.clone());
if self
.posted_messages
.lock()
.expect("slack posted messages lock poisoned")
.contains_key(&key)
{
return;
}
let question = Question {
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(),
context_display: props.context_display.clone(),
};
let blocks = slack_blocks::question_to_blocks(
&event.run_id.to_string(),
&props.question_id,
&question,
);
if let Ok(posted) = self
.client
.post_message(&self.default_channel, &blocks, None)
.await
{
if question.allow_freeform || question.question_type == QuestionType::Freeform {
self.thread_registry.register(
&posted.ts,
&event.run_id.to_string(),
&props.question_id,
);
}
self.posted_messages
.lock()
.expect("slack posted messages lock poisoned")
.insert(key, posted);
}
}
EventBody::InterviewCompleted(props) => {
self.finish_interview(
event.run_id,
&props.question_id,
&props.question,
&props.answer,
)
.await;
}
EventBody::InterviewTimeout(props) => {
self.finish_interview(
event.run_id,
&props.question_id,
&props.question,
"Timed out",
)
.await;
}
EventBody::InterviewAborted(props) => {
let answer_text = if props.reason == "skipped" {
"Skipped"
} else {
"Aborted"
};
self.finish_interview(
event.run_id,
&props.question_id,
&props.question,
answer_text,
)
.await;
}
_ => {}
}
}
async fn finish_interview(
&self,
run_id: RunId,
qid: &str,
question_text: &str,
answer_text: &str,
) {
let key = (run_id, qid.to_string());
let posted = self
.posted_messages
.lock()
.expect("slack posted messages lock poisoned")
.remove(&key);
let Some(posted) = posted else {
return;
};
self.thread_registry.remove(&posted.ts);
let blocks = slack_blocks::answered_blocks(question_text, answer_text);
let _ = self
.client
.update_message(&posted.channel_id, &posted.ts, &blocks)
.await;
}
async fn submit_answer(&self, state: Arc<AppState>, submission: SlackAnswerSubmission) {
let run_id = match RunId::from_str(&submission.run_id) {
Ok(run_id) => run_id,
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() {
return;
}
let _ =
deliver_answer_to_run(state.as_ref(), run_id, &submission.qid, submission.answer).await;
}
}
/// Shared application state for the server.
pub struct AppState {
runs: Mutex<HashMap<RunId, ManagedRun>>,
@ -315,6 +527,8 @@ pub struct AppState {
pub(crate) local_daemon_mode: bool,
shutting_down: AtomicBool,
registry_factory_override: Option<Box<RegistryFactoryOverride>>,
slack_service: Option<Arc<SlackService>>,
slack_started: AtomicBool,
}
fn nonzero_i64(value: i64) -> Option<i64> {
@ -526,8 +740,55 @@ fn decode_secret_pem(name: &str, raw: &str) -> Result<String, String> {
.map_err(|err| format!("{name} base64 decoded to invalid UTF-8: {err}"))
}
fn start_optional_slack_service(state: &Arc<AppState>) {
let Some(service) = state.slack_service.clone() else {
return;
};
if state.slack_started.swap(true, Ordering::SeqCst) {
return;
}
let event_state = Arc::clone(state);
let event_service = Arc::clone(&service);
tokio::spawn(async move {
let mut rx = event_state.global_event_tx.subscribe();
loop {
match rx.recv().await {
Ok(envelope) => {
if let Ok(event) = RunEvent::try_from(&envelope.payload) {
event_service.handle_event(&event).await;
}
}
Err(RecvError::Lagged(_)) => {}
Err(RecvError::Closed) => break,
}
}
});
let socket_state = Arc::clone(state);
tokio::spawn(async move {
let submit_service = Arc::clone(&service);
let on_submit: Arc<dyn Fn(SlackAnswerSubmission) + Send + Sync> =
Arc::new(move |submission| {
let state = Arc::clone(&socket_state);
let service = Arc::clone(&submit_service);
tokio::spawn(async move {
service.submit_answer(state, submission).await;
});
});
slack_connection::run(
&service.client,
&service.app_token,
&service.thread_registry,
on_submit,
)
.await;
});
}
/// Build the axum Router with all run endpoints and embedded static assets.
pub fn build_router(state: Arc<AppState>, auth_mode: AuthMode) -> Router {
start_optional_slack_service(&state);
let middleware_state = Arc::clone(&state);
let api_common = Router::new()
.route("/openapi.json", get(openapi_spec))
@ -1703,6 +1964,21 @@ pub(crate) fn build_app_state_with_path(
) -> anyhow::Result<Arc<AppState>> {
let secret_store = SecretStore::load(secret_store_path)?;
let (global_event_tx, _) = broadcast::channel(4096);
let slack_service = {
let settings = settings.read().expect("settings lock poisoned");
settings
.slack_settings()
.and_then(|slack| slack.default_channel.clone())
.and_then(|default_channel| {
resolve_slack_credentials().map(|credentials| {
Arc::new(SlackService::new(
credentials.bot_token,
credentials.app_token,
default_channel,
))
})
})
};
Ok(Arc::new(AppState {
runs: Mutex::new(HashMap::new()),
aggregate_billing: Mutex::new(BillingAccumulator::default()),
@ -1719,6 +1995,8 @@ pub(crate) fn build_app_state_with_path(
local_daemon_mode,
shutting_down: AtomicBool::new(false),
registry_factory_override,
slack_service,
slack_started: AtomicBool::new(false),
}))
}
@ -1836,10 +2114,10 @@ async fn delete_run_internal(state: &Arc<AppState>, id: RunId) -> Result<(), Res
if let Some(mut managed_run) = managed_run {
if let Some(token) = &managed_run.cancel_token {
token.store(true, Ordering::Relaxed);
token.store(true, Ordering::SeqCst);
}
if let Some(interviewer) = &managed_run.interviewer {
interviewer.abort_pending();
if let Some(answer_transport) = managed_run.answer_transport.clone() {
answer_transport.abort_pending().await;
}
if let Some(cancel_tx) = managed_run.cancel_tx.take() {
let _ = cancel_tx.send(());
@ -2049,7 +2327,8 @@ fn api_event_envelope_from_store(event: &EventEnvelope) -> Result<ApiEventEnvelo
}
fn clear_live_run_state(run: &mut ManagedRun) {
run.interviewer = None;
run.answer_transport = None;
run.accepted_questions.clear();
run.event_tx = None;
run.cancel_tx = None;
run.cancel_token = None;
@ -2057,6 +2336,50 @@ fn clear_live_run_state(run: &mut ManagedRun) {
run.worker_pgid = None;
}
fn reconcile_live_interview_state_for_event(run: &mut ManagedRun, event: &RunEvent) {
match &event.body {
EventBody::InterviewCompleted(props) => {
run.accepted_questions.remove(&props.question_id);
}
EventBody::InterviewTimeout(props) => {
run.accepted_questions.remove(&props.question_id);
}
EventBody::InterviewAborted(props) => {
run.accepted_questions.remove(&props.question_id);
}
EventBody::RunCompleted(_) | EventBody::RunFailed(_) | EventBody::RunRewound(_) => {
run.accepted_questions.clear();
}
_ => {}
}
}
fn claim_run_answer_transport(
state: &AppState,
run_id: RunId,
qid: &str,
) -> Result<RunAnswerTransport, StatusCode> {
let mut runs = state.runs.lock().expect("runs lock poisoned");
let managed_run = runs.get_mut(&run_id).ok_or(StatusCode::NOT_FOUND)?;
let transport = managed_run
.answer_transport
.clone()
.ok_or(StatusCode::CONFLICT)?;
if !managed_run.accepted_questions.insert(qid.to_string()) {
return Err(StatusCode::CONFLICT);
}
Ok(transport)
}
fn release_run_answer_claim(state: &AppState, run_id: RunId, qid: &str) {
let mut runs = state.runs.lock().expect("runs lock poisoned");
if let Some(managed_run) = runs.get_mut(&run_id) {
managed_run.accepted_questions.remove(qid);
}
}
#[derive(Clone, Copy)]
struct LiveWorkerProcess {
run_id: RunId,
@ -2252,13 +2575,20 @@ async fn persist_cancelled_run_status(state: &AppState, run_id: RunId) -> anyhow
}
async fn forward_run_events_to_global(
state: Arc<AppState>,
run_id: RunId,
mut run_events: broadcast::Receiver<EventEnvelope>,
global_event_tx: broadcast::Sender<EventEnvelope>,
) {
loop {
match run_events.recv().await {
Ok(event) => {
let _ = global_event_tx.send(event);
if let Ok(run_event) = RunEvent::try_from(&event.payload) {
let mut runs = state.runs.lock().expect("runs lock poisoned");
if let Some(managed_run) = runs.get_mut(&run_id) {
reconcile_live_interview_state_for_event(managed_run, &run_event);
}
}
let _ = state.global_event_tx.send(event);
}
Err(RecvError::Lagged(_)) => {}
Err(RecvError::Closed) => break,
@ -2279,7 +2609,8 @@ fn managed_run(
error: None,
created_at,
enqueued_at: Instant::now(),
interviewer: None,
answer_transport: None,
accepted_questions: HashSet::new(),
event_tx: None,
checkpoint: None,
cancel_tx: None,
@ -2429,6 +2760,20 @@ async fn drain_worker_stderr(
Ok(())
}
async fn pump_worker_control_jsonl(
mut stdin: ChildStdin,
mut control_rx: mpsc::Receiver<WorkerControlEnvelope>,
) -> anyhow::Result<()> {
while let Some(message) = control_rx.recv().await {
let mut line = serde_json::to_vec(&message)?;
line.push(b'\n');
stdin.write_all(&line).await?;
stdin.flush().await?;
}
Ok(())
}
async fn append_worker_exit_failure(
run_store: &fabro_store::RunDatabase,
run_id: RunId,
@ -2522,7 +2867,7 @@ fn worker_command(
.arg(run_id.to_string())
.arg("--mode")
.arg(worker_mode_arg(mode))
.stdin(Stdio::null())
.stdin(Stdio::piped())
.stdout(Stdio::null())
.stderr(Stdio::piped());
@ -2538,6 +2883,7 @@ fn api_question_from_interview_question(id: &str, question: &Question) -> ApiQue
ApiQuestion {
id: id.to_string(),
text: question.text.clone(),
stage: question.stage.clone(),
question_type: match question.question_type {
QuestionType::YesNo => ApiQuestionType::YesNo,
QuestionType::MultipleChoice => ApiQuestionType::MultipleChoice,
@ -2554,6 +2900,162 @@ fn api_question_from_interview_question(id: &str, question: &Question) -> ApiQue
})
.collect(),
allow_freeform: question.allow_freeform,
timeout_seconds: question.timeout_seconds,
context_display: question.context_display.clone(),
}
}
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),
)
}
#[allow(clippy::result_large_err)] // Axum handlers naturally propagate full `Response` errors.
async fn load_pending_interview_question(
state: &AppState,
run_id: RunId,
qid: &str,
) -> Result<Question, Response> {
let run_store = match state.store.open_run_reader(&run_id).await {
Ok(run_store) => run_store,
Err(fabro_store::StoreError::RunNotFound(_)) => {
return Err(ApiError::not_found("Run not found.").into_response());
}
Err(err) => {
return Err(
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response(),
);
}
};
let run_state = match run_store.state().await {
Ok(run_state) => run_state,
Err(err) => {
return Err(
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response(),
);
}
};
let Some(record) = run_state.pending_interviews.get(qid) else {
return Err(ApiError::new(
StatusCode::CONFLICT,
"Question no longer exists or was already answered.",
)
.into_response());
};
Ok(question_from_pending_interview(record))
}
#[allow(clippy::result_large_err)] // Axum handlers naturally propagate full `Response` errors.
fn validate_answer_for_question(question: &Question, 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)
| (_, fabro_interview::AnswerValue::Aborted)
| (_, fabro_interview::AnswerValue::Skipped)
| (_, fabro_interview::AnswerValue::Timeout) => Ok(()),
(QuestionType::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)) => {
if keys
.iter()
.all(|key| question.options.iter().any(|option| option.key == *key))
{
Ok(())
} else {
Err(ApiError::bad_request("Invalid option key.").into_response())
}
}
(QuestionType::Freeform, fabro_interview::AnswerValue::Text(text))
if !text.trim().is_empty() =>
{
Ok(())
}
(_, fabro_interview::AnswerValue::Text(text))
if question.allow_freeform && !text.trim().is_empty() =>
{
Ok(())
}
_ => Err(ApiError::bad_request("Answer does not match question type.").into_response()),
}
}
#[allow(clippy::result_large_err)] // Axum handlers naturally propagate full `Response` errors.
async fn deliver_answer_to_run(
state: &AppState,
run_id: RunId,
qid: &str,
answer: Answer,
) -> Result<(), Response> {
let transport = match claim_run_answer_transport(state, run_id, qid) {
Ok(transport) => transport,
Err(StatusCode::NOT_FOUND) => {
return Err(ApiError::not_found("Run not found.").into_response());
}
Err(StatusCode::CONFLICT) => {
return Err(ApiError::new(
StatusCode::CONFLICT,
"Question no longer exists or was already answered.",
)
.into_response());
}
Err(status) => {
return Err(
ApiError::new(status, "Run is not ready to accept answers.").into_response()
);
}
};
match transport.submit(qid, answer).await {
Ok(()) => Ok(()),
Err(_) => {
release_run_answer_claim(state, run_id, qid);
Err(ApiError::new(
StatusCode::SERVICE_UNAVAILABLE,
"Failed to deliver answer to the active run.",
)
.into_response())
}
}
}
@ -2587,25 +3089,6 @@ fn answer_from_request(req: SubmitAnswerRequest, question: &Question) -> Result<
}
}
async fn load_file_question(run_dir: &std::path::Path) -> anyhow::Result<Option<Question>> {
let request_path = fabro_config::RunScratch::new(run_dir).interview_request_path();
match fs::read_to_string(&request_path).await {
Ok(data) => Ok(Some(serde_json::from_str(&data)?)),
Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(None),
Err(err) => Err(err.into()),
}
}
async fn write_file_answer(run_dir: &std::path::Path, answer: &Answer) -> anyhow::Result<()> {
let response_path = fabro_config::RunScratch::new(run_dir).interview_response_path();
if let Some(parent) = response_path.parent() {
fs::create_dir_all(parent).await?;
}
let data = serde_json::to_string_pretty(answer)?;
fs::write(response_path, data).await?;
Ok(())
}
async fn create_run(
subject: AuthenticatedSubject,
State(state): State<Arc<AppState>>,
@ -2937,7 +3420,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 interviewer = Arc::new(WebInterviewer::new());
let broker = Arc::new(InterviewBroker::new());
let interviewer: Arc<dyn Interviewer> = Arc::new(ControlInterviewer::new(Arc::clone(&broker)));
let emitter = Emitter::new(run_id);
if let Some(tx_clone) = event_tx {
emitter.on_event(move |event| {
@ -2947,7 +3431,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) as Arc<dyn Interviewer>)));
.map(|factory| Arc::new(factory(Arc::clone(&interviewer))));
let emitter = Arc::new(emitter);
// Transition to Running, populate interviewer
@ -2956,7 +3440,9 @@ async fn execute_run_in_process(state: Arc<AppState>, run_id: RunId) {
if let Some(managed_run) = runs.get_mut(&run_id) {
if managed_run.status == RunStatus::Starting {
managed_run.status = RunStatus::Running;
managed_run.interviewer = Some(Arc::clone(&interviewer));
managed_run.answer_transport = Some(RunAnswerTransport::InProcess {
broker: Arc::clone(&broker),
});
false
} else {
// Was cancelled during setup
@ -2990,8 +3476,9 @@ async fn execute_run_in_process(state: Arc<AppState>, run_id: RunId) {
}
};
tokio::spawn(forward_run_events_to_global(
Arc::clone(&state),
run_id,
run_store.subscribe(),
state.global_event_tx.clone(),
));
let persisted = match Persisted::load_from_store(&run_store.clone().into(), &run_dir).await {
Ok(persisted) => persisted,
@ -3028,7 +3515,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) as Arc<dyn Interviewer>,
interviewer: Arc::clone(&interviewer),
run_store: run_store.clone().into(),
event_sink: workflow_event::RunEventSink::store(run_store.clone()),
artifact_uploader: None,
@ -3166,8 +3653,9 @@ async fn execute_run_subprocess(state: Arc<AppState>, run_id: RunId) {
}
};
tokio::spawn(forward_run_events_to_global(
Arc::clone(&state),
run_id,
run_store.subscribe(),
state.global_event_tx.clone(),
));
let mut child = match worker_command(state.as_ref(), run_id, execution_mode, &run_dir)
@ -3222,6 +3710,26 @@ async fn execute_run_subprocess(state: Arc<AppState>, run_id: RunId) {
}
}
let Some(stdin) = child.stdin.take() else {
let message = "Worker stdin pipe was unavailable".to_string();
tracing::error!(run_id = %run_id, "{message}");
let _ = child.start_kill();
let _ = workflow_event::append_event(
&run_store,
&run_id,
&workflow_event::Event::WorkflowRunFailed {
error: FabroError::engine(message.clone()),
duration_ms: 0,
reason: Some(WorkflowStatusReason::LaunchFailed),
git_commit_sha: None,
},
)
.await;
fail_managed_run(&state, run_id, message);
state.scheduler_notify.notify_one();
return;
};
let Some(stderr) = child.stderr.take() else {
let message = "Worker stderr pipe was unavailable".to_string();
tracing::error!(run_id = %run_id, "{message}");
@ -3242,6 +3750,15 @@ async fn execute_run_subprocess(state: Arc<AppState>, run_id: RunId) {
return;
};
let (control_tx, control_rx) = mpsc::channel(WORKER_CONTROL_QUEUE_CAPACITY);
{
let mut runs = state.runs.lock().expect("runs lock poisoned");
if let Some(managed_run) = runs.get_mut(&run_id) {
managed_run.answer_transport = Some(RunAnswerTransport::Subprocess { control_tx });
}
}
let control_task = tokio::spawn(pump_worker_control_jsonl(stdin, control_rx));
let stderr_task = tokio::spawn(drain_worker_stderr(run_id, run_dir.clone(), stderr));
let wait_status = match child.wait().await {
@ -3266,6 +3783,9 @@ async fn execute_run_subprocess(state: Arc<AppState>, run_id: RunId) {
}
};
control_task.abort();
let _ = control_task.await;
match stderr_task.await {
Ok(Ok(())) => {}
Ok(Err(err)) => {
@ -3417,44 +3937,23 @@ async fn get_questions(
Ok(id) => id,
Err(response) => return response,
};
let (interviewer, run_dir) = {
let runs = state.runs.lock().expect("runs lock poisoned");
match runs.get(&id) {
Some(managed_run) => (managed_run.interviewer.clone(), managed_run.run_dir.clone()),
None => return ApiError::not_found("Run not found.").into_response(),
match state.store.open_run_reader(&id).await {
Ok(run_store) => match run_store.state().await {
Ok(run_state) => {
let questions = run_state
.pending_interviews
.values()
.map(api_question_from_pending_interview)
.collect::<Vec<_>>();
(StatusCode::OK, Json(ListResponse::new(questions))).into_response()
}
Err(err) => {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
}
},
Err(fabro_store::StoreError::RunNotFound(_)) => {
ApiError::not_found("Run not found.").into_response()
}
};
if let Some(interviewer) = interviewer {
let questions: Vec<ApiQuestion> = interviewer
.pending_questions()
.into_iter()
.map(|pending| api_question_from_interview_question(&pending.id, &pending.question))
.collect();
return (StatusCode::OK, Json(ListResponse::new(questions))).into_response();
}
let Some(run_dir) = run_dir else {
return (
StatusCode::OK,
Json(ListResponse::new(Vec::<ApiQuestion>::new())),
)
.into_response();
};
match load_file_question(&run_dir).await {
Ok(Some(question)) => (
StatusCode::OK,
Json(ListResponse::new(vec![
api_question_from_interview_question(FILE_INTERVIEW_QUESTION_ID, &question),
])),
)
.into_response(),
Ok(None) => (
StatusCode::OK,
Json(ListResponse::new(Vec::<ApiQuestion>::new())),
)
.into_response(),
Err(err) => {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
}
@ -3471,66 +3970,17 @@ async fn submit_answer(
Ok(id) => id,
Err(response) => return response,
};
let (interviewer, run_dir) = {
let runs = state.runs.lock().expect("runs lock poisoned");
match runs.get(&id) {
Some(managed_run) => (managed_run.interviewer.clone(), managed_run.run_dir.clone()),
None => return ApiError::not_found("Run not found.").into_response(),
}
};
if let Some(interviewer) = interviewer {
let pending = interviewer.pending_questions();
let Some(question) = pending.iter().find(|pending| pending.id == qid) else {
return ApiError::new(
StatusCode::CONFLICT,
"Question no longer exists or was already answered.",
)
.into_response();
};
let answer = match answer_from_request(req, &question.question) {
Ok(answer) => answer,
Err(response) => return response,
};
if interviewer.submit_answer(&qid, answer) {
return StatusCode::NO_CONTENT.into_response();
}
return ApiError::new(
StatusCode::CONFLICT,
"Question no longer exists or was already answered.",
)
.into_response();
}
let Some(run_dir) = run_dir else {
return ApiError::new(StatusCode::CONFLICT, "Run is not yet running.").into_response();
};
if qid != FILE_INTERVIEW_QUESTION_ID {
return ApiError::new(
StatusCode::CONFLICT,
"Question no longer exists or was already answered.",
)
.into_response();
}
let question = match load_file_question(&run_dir).await {
Ok(Some(question)) => question,
Ok(None) => {
return ApiError::new(StatusCode::CONFLICT, "Run is not yet running.").into_response();
}
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
let question = match load_pending_interview_question(state.as_ref(), id, &qid).await {
Ok(question) => question,
Err(response) => return response,
};
let answer = match answer_from_request(req, &question) {
Ok(answer) => answer,
Err(response) => return response,
};
match write_file_answer(&run_dir, &answer).await {
match deliver_answer_to_run(state.as_ref(), id, &qid, answer).await {
Ok(()) => StatusCode::NO_CONTENT.into_response(),
Err(err) => {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
}
Err(response) => response,
}
}
@ -4696,7 +5146,6 @@ async fn cancel_run(
persist_cancelled_status,
cancel_token,
cancel_tx,
interviewer,
worker_pid,
) = {
let mut runs = state.runs.lock().expect("runs lock poisoned");
@ -4721,7 +5170,6 @@ async fn cancel_run(
persist_cancelled_status,
managed_run.cancel_token.clone(),
managed_run.cancel_tx.take(),
managed_run.interviewer.clone(),
managed_run.worker_pid,
)
}
@ -4743,10 +5191,7 @@ async fn cancel_run(
}
if let Some(token) = &cancel_token {
token.store(true, Ordering::Relaxed);
}
if let Some(interviewer) = &interviewer {
interviewer.abort_pending();
token.store(true, Ordering::SeqCst);
}
if let Some(cancel_tx) = cancel_tx {
let _ = cancel_tx.send(());

View file

@ -51,6 +51,24 @@ async fn wait_for_question_id(app: &axum::Router, run_id: &str) -> String {
panic!("question should have appeared");
}
async fn wait_for_question(app: &axum::Router, run_id: &str) -> serde_json::Value {
for _ in 0..POLL_ATTEMPTS {
let req = Request::builder()
.method("GET")
.uri(api(&format!("/runs/{run_id}/questions")))
.body(Body::empty())
.unwrap();
let response = app.clone().oneshot(req).await.unwrap();
let body = body_json(response.into_body()).await;
let arr = body["data"].as_array().unwrap();
if let Some(question) = arr.first() {
return question.clone();
}
sleep(POLL_INTERVAL).await;
}
panic!("question should have appeared");
}
const GATE_DOT: &str = r#"digraph GateTest {
graph [goal="Test gate"]
start [shape=Mdiamond]
@ -98,7 +116,11 @@ async fn full_http_lifecycle_approve_and_complete() {
assert_eq!(response.status(), StatusCode::OK);
// 2. Poll for question to appear (run goes start -> work -> gate, then blocks)
let question_id = wait_for_question_id(&app, &run_id).await;
let question = wait_for_question(&app, &run_id).await;
let question_id = question["id"].as_str().unwrap().to_string();
assert_eq!(question["stage"], "gate");
assert!(question["timeout_seconds"].is_null());
assert!(question["context_display"].is_null() || question["context_display"].is_string());
// 3. Submit answer selecting first option (Approve)
let req = Request::builder()

View file

@ -24,10 +24,6 @@ tokio-tungstenite.workspace = true
reqwest.workspace = true
tracing.workspace = true
[[example]]
name = "slack-e2e"
path = "examples/slack_e2e.rs"
[dev-dependencies]
tokio = { workspace = true, features = ["test-util", "macros"] }
toml.workspace = true

View file

@ -1,204 +0,0 @@
#![allow(clippy::print_stderr, clippy::absolute_paths, clippy::exit)]
use std::sync::Arc;
use fabro_interview::{
Answer, AnswerValue, Interviewer, Question, QuestionOption, QuestionType, WebInterviewer,
};
use fabro_slack::blocks::{answered_blocks, question_to_blocks};
use fabro_slack::client::{PostedMessage, SlackClient};
use fabro_slack::connection;
use fabro_slack::threads::ThreadRegistry;
struct TestCase {
label: &'static str,
question: Question,
}
fn test_cases() -> Vec<TestCase> {
let mut mc = Question::new("Pick a language:", QuestionType::MultipleChoice);
mc.options = vec![
QuestionOption {
key: "rs".to_string(),
label: "Rust".to_string(),
},
QuestionOption {
key: "ts".to_string(),
label: "TypeScript".to_string(),
},
QuestionOption {
key: "py".to_string(),
label: "Python".to_string(),
},
];
let mut ms = Question::new("Select features to enable:", QuestionType::MultiSelect);
ms.options = vec![
QuestionOption {
key: "auth".to_string(),
label: "Authentication".to_string(),
},
QuestionOption {
key: "billing".to_string(),
label: "Billing".to_string(),
},
QuestionOption {
key: "notifications".to_string(),
label: "Notifications".to_string(),
},
];
vec![
TestCase {
label: "YesNo",
question: Question::new("Do you approve this deployment?", QuestionType::YesNo),
},
TestCase {
label: "Confirmation",
question: Question::new(
"This will delete all staging data. Continue?",
QuestionType::Confirmation,
),
},
TestCase {
label: "MultipleChoice",
question: mc,
},
TestCase {
label: "MultiSelect",
question: ms,
},
TestCase {
label: "Freeform",
question: Question::new("What is the repository URL?", QuestionType::Freeform),
},
]
}
fn format_answer(answer: &Answer) -> String {
match &answer.value {
AnswerValue::Yes => "Yes".to_string(),
AnswerValue::No => "No".to_string(),
AnswerValue::Aborted => "Aborted".to_string(),
AnswerValue::Text(t) => t.clone(),
AnswerValue::Selected(k) => {
if let Some(opt) = &answer.selected_option {
format!("{} ({})", opt.label, k)
} else {
k.clone()
}
}
AnswerValue::MultiSelected(keys) => keys.join(", "),
AnswerValue::Skipped => "Skipped".to_string(),
AnswerValue::Timeout => "Timed out".to_string(),
}
}
async fn ask_question(
test_case: TestCase,
interviewer: &Arc<WebInterviewer>,
thread_registry: &ThreadRegistry,
slack_client: &SlackClient,
channel: &str,
) {
eprintln!("\n--- {} ---", test_case.label);
let question_text = test_case.question.text.clone();
let is_freeform = test_case.question.question_type == QuestionType::Freeform;
let interviewer_clone = Arc::clone(interviewer);
let ask_handle = tokio::spawn(async move { interviewer_clone.ask(test_case.question).await });
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
let pending = interviewer.pending_questions();
let pq = pending
.iter()
.find(|pq| pq.question.text == question_text)
.expect("Question should be pending");
let question_id = pq.id.clone();
let blocks = question_to_blocks(&question_id, &pq.question);
let posted: PostedMessage = slack_client
.post_message(channel, &blocks, None)
.await
.unwrap_or_else(|e| {
eprintln!("Failed to post message: {e}");
std::process::exit(1);
});
// For freeform questions, register the message ts so thread replies get routed
if is_freeform {
thread_registry.register(&posted.ts, &question_id);
eprintln!("Posted. Reply in thread in Slack...");
} else {
eprintln!("Posted. Respond in Slack...");
}
let answer = ask_handle.await.expect("ask task panicked");
let answer_text = format_answer(&answer);
eprintln!("Got answer: {answer_text}");
// Clean up thread registration
if is_freeform {
thread_registry.remove(&posted.ts);
}
let updated = answered_blocks(&question_text, &answer_text);
if let Err(e) = slack_client
.update_message(&posted.channel_id, &posted.ts, &updated)
.await
{
eprintln!("Failed to update message: {e}");
}
}
#[tokio::main]
async fn main() {
let _ = rustls::crypto::ring::default_provider().install_default();
tracing_subscriber::fmt()
.with_env_filter("fabro_slack=debug,info")
.init();
let bot_token = std::env::var("FABRO_SLACK_BOT_TOKEN").expect("FABRO_SLACK_BOT_TOKEN required");
let app_token = std::env::var("FABRO_SLACK_APP_TOKEN").expect("FABRO_SLACK_APP_TOKEN required");
let channel = std::env::var("FABRO_SLACK_CHANNEL").unwrap_or_else(|_| "#arc-test".to_string());
eprintln!("Connecting to Slack Socket Mode...");
let slack_client = SlackClient::new(bot_token);
let wss_url = connection::open_socket_url(slack_client.http(), &app_token)
.await
.expect("Failed to open socket URL");
let interviewer = Arc::new(WebInterviewer::new());
let thread_registry = Arc::new(ThreadRegistry::new());
// Start the event loop in the background
let interviewer_for_loop = Arc::clone(&interviewer);
let thread_registry_for_loop = Arc::clone(&thread_registry);
tokio::spawn(async move {
connection::run_event_loop(&wss_url, &interviewer_for_loop, &thread_registry_for_loop)
.await
.ok();
});
// Wait for the socket to connect
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
eprintln!("Connected. Running all question types...\n");
let cases = test_cases();
for case in cases {
ask_question(
case,
&interviewer,
&thread_registry,
&slack_client,
&channel,
)
.await;
}
eprintln!("\nAll question types tested!");
}

View file

@ -1,6 +1,13 @@
use fabro_interview::{Question, QuestionType};
use serde_json::{Value, json};
use crate::payload::{SlackActionPayload, encode_action_value};
const ANSWER_ACTION_ID: &str = "interview.answer";
const MULTI_SELECT_BLOCK_ID: &str = "interview.checkboxes";
const MULTI_SELECT_ACTION_ID: &str = "interview.select";
const MULTI_SELECT_SUBMIT_ACTION_ID: &str = "interview.submit";
fn text_block(text: &str) -> Value {
json!({
"type": "section",
@ -29,7 +36,7 @@ pub fn answered_blocks(question_text: &str, answer_text: &str) -> Vec<Value> {
))]
}
pub fn question_to_blocks(question_id: &str, question: &Question) -> Vec<Value> {
pub fn question_to_blocks(run_id: &str, question_id: &str, question: &Question) -> Vec<Value> {
let section = text_block(&question.text);
match question.question_type {
@ -37,8 +44,14 @@ pub fn question_to_blocks(question_id: &str, question: &Question) -> Vec<Value>
let actions = json!({
"type": "actions",
"elements": [
button("Yes", "yes", &format!("{question_id}:yes")),
button("No", "no", &format!("{question_id}:no")),
button("Yes", &encode_action_value(&SlackActionPayload::Yes {
run_id: run_id.to_string(),
qid: question_id.to_string(),
}), ANSWER_ACTION_ID),
button("No", &encode_action_value(&SlackActionPayload::No {
run_id: run_id.to_string(),
qid: question_id.to_string(),
}), ANSWER_ACTION_ID),
]
});
vec![section, actions]
@ -47,7 +60,17 @@ pub fn question_to_blocks(question_id: &str, question: &Question) -> Vec<Value>
let elements: Vec<Value> = question
.options
.iter()
.map(|opt| button(&opt.label, &opt.key, &format!("{question_id}:{}", opt.key)))
.map(|opt| {
button(
&opt.label,
&encode_action_value(&SlackActionPayload::Selected {
run_id: run_id.to_string(),
qid: question_id.to_string(),
key: opt.key.clone(),
}),
ANSWER_ACTION_ID,
)
})
.collect();
let actions = json!({
"type": "actions",
@ -68,17 +91,20 @@ pub fn question_to_blocks(question_id: &str, question: &Question) -> Vec<Value>
.collect();
let checkboxes = json!({
"type": "actions",
"block_id": format!("{question_id}:checkboxes"),
"block_id": MULTI_SELECT_BLOCK_ID,
"elements": [{
"type": "checkboxes",
"action_id": format!("{question_id}:select"),
"action_id": MULTI_SELECT_ACTION_ID,
"options": options
}]
});
let submit = json!({
"type": "actions",
"elements": [
button("Submit", "submit", &format!("{question_id}:submit")),
button("Submit", &encode_action_value(&SlackActionPayload::SubmitMulti {
run_id: run_id.to_string(),
qid: question_id.to_string(),
}), MULTI_SELECT_SUBMIT_ACTION_ID),
]
});
vec![section, checkboxes, submit]
@ -100,7 +126,7 @@ mod tests {
#[test]
fn yes_no_produces_two_buttons() {
let q = Question::new("Approve this PR?", QuestionType::YesNo);
let blocks = question_to_blocks("q-1", &q);
let blocks = question_to_blocks("run-1", "q-1", &q);
let blocks_json: Value = serde_json::to_value(&blocks).unwrap();
let section = &blocks_json[0];
@ -123,7 +149,7 @@ mod tests {
#[test]
fn confirmation_produces_two_buttons() {
let q = Question::new("Continue?", QuestionType::Confirmation);
let blocks = question_to_blocks("q-2", &q);
let blocks = question_to_blocks("run-1", "q-2", &q);
let blocks_json: Value = serde_json::to_value(&blocks).unwrap();
let actions = &blocks_json[1];
@ -150,14 +176,20 @@ mod tests {
label: "Python".to_string(),
},
];
let blocks = question_to_blocks("q-3", &q);
let blocks = question_to_blocks("run-1", "q-3", &q);
let blocks_json: Value = serde_json::to_value(&blocks).unwrap();
let actions = &blocks_json[1];
let elements = actions["elements"].as_array().unwrap();
assert_eq!(elements.len(), 3);
assert_eq!(elements[0]["text"]["text"], "Rust");
assert_eq!(elements[0]["value"], "rs");
assert_eq!(elements[0]["action_id"], ANSWER_ACTION_ID);
assert!(
elements[0]["value"]
.as_str()
.unwrap()
.contains("\"run_id\":\"run-1\"")
);
assert_eq!(elements[1]["text"]["text"], "TypeScript");
assert_eq!(elements[2]["text"]["text"], "Python");
}
@ -165,7 +197,7 @@ mod tests {
#[test]
fn freeform_produces_section_prompting_thread_reply() {
let q = Question::new("What's the repo URL?", QuestionType::Freeform);
let blocks = question_to_blocks("q-4", &q);
let blocks = question_to_blocks("run-1", "q-4", &q);
let blocks_json: Value = serde_json::to_value(&blocks).unwrap();
assert_eq!(blocks_json.as_array().unwrap().len(), 1);
@ -176,14 +208,17 @@ mod tests {
}
#[test]
fn question_id_embedded_in_action_ids() {
fn action_values_include_run_id_and_question_id() {
let q = Question::new("Approve?", QuestionType::YesNo);
let blocks = question_to_blocks("q-7", &q);
let blocks = question_to_blocks("run-7", "q-7", &q);
let blocks_json: Value = serde_json::to_value(&blocks).unwrap();
let actions = &blocks_json[1];
let elements = actions["elements"].as_array().unwrap();
assert!(elements[0]["action_id"].as_str().unwrap().contains("q-7"));
assert_eq!(elements[0]["action_id"], ANSWER_ACTION_ID);
let value = elements[0]["value"].as_str().unwrap();
assert!(value.contains("\"run_id\":\"run-7\""));
assert!(value.contains("\"qid\":\"q-7\""));
}
#[test]
@ -223,15 +258,16 @@ mod tests {
label: "Billing".to_string(),
},
];
let blocks = question_to_blocks("q-5", &q);
let blocks = question_to_blocks("run-1", "q-5", &q);
let blocks_json: Value = serde_json::to_value(&blocks).unwrap();
// Checkboxes in their own block with a block_id
let checkbox_block = &blocks_json[1];
assert_eq!(checkbox_block["type"], "actions");
assert!(checkbox_block["block_id"].as_str().unwrap().contains("q-5"));
assert_eq!(checkbox_block["block_id"], MULTI_SELECT_BLOCK_ID);
let cb_elements = checkbox_block["elements"].as_array().unwrap();
assert_eq!(cb_elements[0]["type"], "checkboxes");
assert_eq!(cb_elements[0]["action_id"], MULTI_SELECT_ACTION_ID);
// Submit button in a separate actions block
let submit_block = &blocks_json[2];
@ -239,11 +275,15 @@ mod tests {
let submit_elements = submit_block["elements"].as_array().unwrap();
assert_eq!(submit_elements[0]["type"], "button");
assert_eq!(submit_elements[0]["text"]["text"], "Submit");
assert_eq!(
submit_elements[0]["action_id"],
MULTI_SELECT_SUBMIT_ACTION_ID
);
assert!(
submit_elements[0]["action_id"]
submit_elements[0]["value"]
.as_str()
.unwrap()
.contains("q-5")
.contains("\"qid\":\"q-5\"")
);
}
}

View file

@ -1,6 +1,5 @@
use std::sync::Arc;
use fabro_interview::WebInterviewer;
use futures_util::{SinkExt, StreamExt};
use tokio::time::sleep;
use tokio_tungstenite::tungstenite::Message;
@ -8,6 +7,7 @@ use tracing::{debug, error, info, warn};
use crate::client::{SlackApiError, SlackClient, parse_wss_url};
use crate::dispatch::{DispatchAction, dispatch};
use crate::payload::SlackAnswerSubmission;
use crate::socket::{SocketAck, SocketEnvelope};
use crate::threads::ThreadRegistry;
@ -83,8 +83,8 @@ pub async fn open_socket_url(
/// On disconnect, returns so the caller can reconnect.
pub async fn run_event_loop(
wss_url: &str,
interviewer: &Arc<WebInterviewer>,
thread_registry: &ThreadRegistry,
on_submit: &Arc<dyn Fn(SlackAnswerSubmission) + Send + Sync>,
) -> Result<(), ConnectionError> {
let (ws_stream, _) = tokio_tungstenite::connect_async(wss_url)
.await
@ -117,21 +117,20 @@ pub async fn run_event_loop(
let (ack_json, outcome, action) = process_message(&text, thread_registry);
// Send ack immediately (Slack requires within 3 seconds)
if let Some(ack) = ack_json {
if let Err(e) = write.send(Message::Text(ack.into())).await {
error!("Failed to send ack: {e}");
}
}
// Handle dispatch action
match action {
DispatchAction::SubmitAnswer {
question_id,
answer,
} => {
debug!(question_id, "Submitting answer from Slack");
let _ = interviewer.submit_answer(&question_id, answer);
DispatchAction::SubmitAnswer(submission) => {
debug!(
run_id = submission.run_id,
qid = submission.qid,
"Submitting answer from Slack"
);
on_submit(submission);
}
DispatchAction::Connected => {
info!("Socket Mode handshake complete");
@ -153,8 +152,8 @@ pub async fn run_event_loop(
pub async fn run(
slack_client: &SlackClient,
app_token: &str,
interviewer: Arc<WebInterviewer>,
thread_registry: &ThreadRegistry,
on_submit: Arc<dyn Fn(SlackAnswerSubmission) + Send + Sync>,
) {
let mut backoff = std::time::Duration::from_secs(1);
let max_backoff = std::time::Duration::from_secs(30);
@ -173,7 +172,7 @@ pub async fn run(
}
};
match run_event_loop(&wss_url, &interviewer, thread_registry).await {
match run_event_loop(&wss_url, thread_registry, &on_submit).await {
Ok(()) => {
info!("Event loop ended, reconnecting...");
backoff = std::time::Duration::from_secs(1);
@ -189,6 +188,8 @@ pub async fn run(
#[cfg(test)]
mod tests {
use std::sync::{Arc, Mutex};
use super::*;
use fabro_interview::AnswerValue;
@ -213,9 +214,9 @@ mod tests {
"payload": {
"type": "block_actions",
"actions": [{
"action_id": "q-1:yes",
"action_id": "interview.answer",
"type": "button",
"value": "yes"
"value": "{\"kind\":\"yes\",\"run_id\":\"run-1\",\"qid\":\"q-1\"}"
}]
}
}"#;
@ -224,12 +225,10 @@ mod tests {
assert!(ack.unwrap().contains("env-1"));
assert_eq!(outcome, ProcessOutcome::Continue);
match action {
DispatchAction::SubmitAnswer {
question_id,
answer,
} => {
assert_eq!(question_id, "q-1");
assert_eq!(answer.value, AnswerValue::Yes);
DispatchAction::SubmitAnswer(submission) => {
assert_eq!(submission.run_id, "run-1");
assert_eq!(submission.qid, "q-1");
assert_eq!(submission.answer.value, AnswerValue::Yes);
}
other => panic!("expected SubmitAnswer, got {other:?}"),
}
@ -272,7 +271,7 @@ mod tests {
#[test]
fn process_thread_reply_with_registered_question() {
let reg = registry();
reg.register("1234.5678", "q-10");
reg.register("1234.5678", "run-10", "q-10");
let text = serde_json::json!({
"type": "events_api",
"envelope_id": "env-50",
@ -290,59 +289,78 @@ mod tests {
assert!(ack.is_some());
assert_eq!(outcome, ProcessOutcome::Continue);
match action {
DispatchAction::SubmitAnswer {
question_id,
answer,
} => {
assert_eq!(question_id, "q-10");
assert_eq!(answer.value, AnswerValue::Text("my answer".to_string()));
DispatchAction::SubmitAnswer(submission) => {
assert_eq!(submission.run_id, "run-10");
assert_eq!(submission.qid, "q-10");
assert_eq!(
submission.answer.value,
AnswerValue::Text("my answer".to_string())
);
}
other => panic!("expected SubmitAnswer, got {other:?}"),
}
}
#[tokio::test]
async fn submit_answer_reaches_web_interviewer() {
let interviewer = Arc::new(WebInterviewer::new());
let i_clone = Arc::clone(&interviewer);
async fn run_event_loop_submits_answers_via_callback() {
use tokio::net::TcpListener;
let handle = tokio::spawn(async move {
use fabro_interview::{Interviewer, Question, QuestionType};
let q = Question::new("approve?", QuestionType::YesNo);
i_clone.ask(q).await
});
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let url = format!("ws://{}", addr);
sleep(std::time::Duration::from_millis(50)).await;
let registry = registry();
let submissions = Arc::new(Mutex::new(Vec::new()));
let callback_submissions = Arc::clone(&submissions);
let on_submit: Arc<dyn Fn(SlackAnswerSubmission) + Send + Sync> =
Arc::new(move |submission| {
callback_submissions.lock().unwrap().push(submission);
});
let pending = interviewer.pending_questions();
assert_eq!(pending.len(), 1);
let server = async move {
let (stream, _) = listener.accept().await.unwrap();
let mut ws = tokio_tungstenite::accept_async(stream).await.unwrap();
let question_id = pending[0].id.clone();
let text = serde_json::json!({
"type": "interactive",
"envelope_id": "e1",
"payload": {
"type": "block_actions",
"actions": [{
"action_id": format!("{question_id}:yes"),
"type": "button",
"value": "yes"
}]
ws.send(Message::Text(r#"{"type":"hello"}"#.into()))
.await
.unwrap();
ws.send(Message::Text(
r#"{
"type": "interactive",
"envelope_id": "env-1",
"payload": {
"type": "block_actions",
"actions": [{
"action_id": "interview.answer",
"type": "button",
"value": "{\"kind\":\"yes\",\"run_id\":\"run-1\",\"qid\":\"q-1\"}"
}]
}
}"#
.into(),
))
.await
.unwrap();
while let Some(msg) = ws.next().await {
match msg.unwrap() {
Message::Text(text) if text.contains("\"envelope_id\":\"env-1\"") => {
let _ = ws.send(Message::Close(None)).await;
break;
}
_ => {}
}
}
})
.to_string();
let (_, _, action) = process_message(&text, &registry());
match action {
DispatchAction::SubmitAnswer {
question_id: qid,
answer,
} => {
assert!(interviewer.submit_answer(&qid, answer));
}
other => panic!("expected SubmitAnswer, got {other:?}"),
}
};
let answer = handle.await.unwrap();
assert_eq!(answer.value, AnswerValue::Yes);
let _server_task = tokio::spawn(server);
let loop_result = run_event_loop(&url, &registry, &on_submit).await;
assert!(loop_result.is_ok());
let submissions = submissions.lock().unwrap();
assert_eq!(submissions.len(), 1);
assert_eq!(submissions[0].run_id, "run-1");
assert_eq!(submissions[0].qid, "q-1");
assert_eq!(submissions[0].answer.value, AnswerValue::Yes);
}
}

View file

@ -1,13 +1,12 @@
use fabro_interview::Answer;
use crate::interaction;
use crate::payload::SlackAnswerSubmission;
use crate::socket::{SocketEnvelope, SocketEventKind, classify_envelope};
use crate::threads::{self, ThreadRegistry};
#[derive(Debug)]
pub enum DispatchAction {
Connected,
SubmitAnswer { question_id: String, answer: Answer },
SubmitAnswer(SlackAnswerSubmission),
Reconnect,
Ignored,
}
@ -20,10 +19,7 @@ pub fn dispatch(envelope: &SocketEnvelope, thread_registry: &ThreadRegistry) ->
return DispatchAction::Ignored;
};
match interaction::parse_interaction(payload) {
Some((question_id, answer)) => DispatchAction::SubmitAnswer {
question_id,
answer,
},
Some(submission) => DispatchAction::SubmitAnswer(submission),
None => DispatchAction::Ignored,
}
}
@ -34,13 +30,14 @@ pub fn dispatch(envelope: &SocketEnvelope, thread_registry: &ThreadRegistry) ->
let Some((thread_ts, text)) = threads::parse_thread_reply(payload) else {
return DispatchAction::Ignored;
};
let Some(question_id) = thread_registry.resolve(&thread_ts) else {
let Some(question_ref) = thread_registry.resolve(&thread_ts) else {
return DispatchAction::Ignored;
};
DispatchAction::SubmitAnswer {
question_id,
answer: Answer::text(text),
}
DispatchAction::SubmitAnswer(SlackAnswerSubmission {
run_id: question_ref.run_id,
qid: question_ref.qid,
answer: fabro_interview::Answer::text(text),
})
}
SocketEventKind::Disconnect => DispatchAction::Reconnect,
SocketEventKind::Unknown => DispatchAction::Ignored,
@ -73,20 +70,18 @@ mod tests {
payload: Some(serde_json::json!({
"type": "block_actions",
"actions": [{
"action_id": "q-1:yes",
"action_id": "interview.answer",
"type": "button",
"value": "yes"
"value": "{\"kind\":\"yes\",\"run_id\":\"run-1\",\"qid\":\"q-1\"}"
}]
})),
};
let action = dispatch(&envelope, &registry);
match action {
DispatchAction::SubmitAnswer {
question_id,
answer,
} => {
assert_eq!(question_id, "q-1");
assert_eq!(answer.value, AnswerValue::Yes);
DispatchAction::SubmitAnswer(submission) => {
assert_eq!(submission.run_id, "run-1");
assert_eq!(submission.qid, "q-1");
assert_eq!(submission.answer.value, AnswerValue::Yes);
}
other => panic!("expected SubmitAnswer, got {other:?}"),
}
@ -147,7 +142,7 @@ mod tests {
#[test]
fn events_api_thread_reply_to_registered_question() {
let registry = ThreadRegistry::new();
registry.register("1234.5678", "q-10");
registry.register("1234.5678", "run-10", "q-10");
let envelope = SocketEnvelope {
envelope_type: "events_api".to_string(),
envelope_id: Some("env-5".to_string()),
@ -162,13 +157,11 @@ mod tests {
};
let action = dispatch(&envelope, &registry);
match action {
DispatchAction::SubmitAnswer {
question_id,
answer,
} => {
assert_eq!(question_id, "q-10");
DispatchAction::SubmitAnswer(submission) => {
assert_eq!(submission.run_id, "run-10");
assert_eq!(submission.qid, "q-10");
assert_eq!(
answer.value,
submission.answer.value,
AnswerValue::Text("https://github.com/org/repo".to_string())
);
}

View file

@ -1,52 +1,61 @@
use fabro_interview::Answer;
use serde_json::Value;
/// Parses a Slack interaction payload and returns (question_id, Answer).
///
/// Action IDs follow the format `{question_id}:{action}` as set by `blocks::question_to_blocks`.
pub fn parse_interaction(payload: &Value) -> Option<(String, Answer)> {
use crate::payload::{SlackActionPayload, SlackAnswerSubmission};
const MULTI_SELECT_BLOCK_ID: &str = "interview.checkboxes";
const MULTI_SELECT_ACTION_ID: &str = "interview.select";
const ANSWER_ACTION_ID: &str = "interview.answer";
const MULTI_SELECT_SUBMIT_ACTION_ID: &str = "interview.submit";
/// Parses a Slack interaction payload and returns a server-routable answer submission.
pub fn parse_interaction(payload: &Value) -> Option<SlackAnswerSubmission> {
if payload["type"].as_str()? != "block_actions" {
return None;
}
let action = payload["actions"].as_array()?.first()?;
let action_id = action["action_id"].as_str()?;
let (question_id, action_key) = action_id.split_once(':')?;
let value = action["value"].as_str()?;
let routed: SlackActionPayload = serde_json::from_str(value).ok()?;
let question_ref = routed.question_ref();
let action_type = action["type"].as_str().unwrap_or("button");
let answer = match action_type {
"button" => match action_key {
"yes" => Answer::yes(),
"no" => Answer::no(),
"submit" => extract_checkbox_selections(question_id, payload),
key => {
let value = action["value"].as_str().unwrap_or(key);
Answer::text(value.to_string())
}
"button" if action_id == ANSWER_ACTION_ID => match routed {
SlackActionPayload::Yes { .. } => Answer::yes(),
SlackActionPayload::No { .. } => Answer::no(),
SlackActionPayload::Selected { key, .. } => Answer {
value: fabro_interview::AnswerValue::Selected(key),
selected_option: None,
text: None,
},
SlackActionPayload::SubmitMulti { .. } => return None,
},
"button" if action_id == MULTI_SELECT_SUBMIT_ACTION_ID => {
extract_checkbox_selections(payload)
}
"checkboxes" => {
// Ignore checkbox toggle events — wait for Submit button
return None;
}
"plain_text_input" => {
let value = action["value"].as_str()?;
Answer::text(value.to_string())
}
"plain_text_input" => return None,
_ => return None,
};
Some((question_id.to_string(), answer))
Some(SlackAnswerSubmission {
run_id: question_ref.run_id,
qid: question_ref.qid,
answer,
})
}
/// Extract selected checkbox values from `payload.state.values`.
/// The checkbox block has block_id `{question_id}:checkboxes` and
/// action_id `{question_id}:select`.
fn extract_checkbox_selections(question_id: &str, payload: &Value) -> Answer {
let block_id = format!("{question_id}:checkboxes");
let action_id = format!("{question_id}:select");
let selected = payload["state"]["values"][&block_id][&action_id]["selected_options"].as_array();
fn extract_checkbox_selections(payload: &Value) -> Answer {
let selected =
payload["state"]["values"][MULTI_SELECT_BLOCK_ID][MULTI_SELECT_ACTION_ID]["selected_options"]
.as_array();
match selected {
Some(options) if !options.is_empty() => {
@ -54,7 +63,7 @@ fn extract_checkbox_selections(question_id: &str, payload: &Value) -> Answer {
.iter()
.filter_map(|opt| opt["value"].as_str().map(String::from))
.collect();
Answer::text(values.join(", "))
Answer::multi_selected(values)
}
_ => Answer::skipped(),
}
@ -70,14 +79,15 @@ mod tests {
let payload = serde_json::json!({
"type": "block_actions",
"actions": [{
"action_id": "q-1:yes",
"action_id": "interview.answer",
"type": "button",
"value": "yes"
"value": "{\"kind\":\"yes\",\"run_id\":\"run-1\",\"qid\":\"q-1\"}"
}]
});
let result = parse_interaction(&payload).unwrap();
assert_eq!(result.0, "q-1");
assert_eq!(result.1.value, AnswerValue::Yes);
assert_eq!(result.run_id, "run-1");
assert_eq!(result.qid, "q-1");
assert_eq!(result.answer.value, AnswerValue::Yes);
}
#[test]
@ -85,14 +95,15 @@ mod tests {
let payload = serde_json::json!({
"type": "block_actions",
"actions": [{
"action_id": "q-2:no",
"action_id": "interview.answer",
"type": "button",
"value": "no"
"value": "{\"kind\":\"no\",\"run_id\":\"run-1\",\"qid\":\"q-2\"}"
}]
});
let result = parse_interaction(&payload).unwrap();
assert_eq!(result.0, "q-2");
assert_eq!(result.1.value, AnswerValue::No);
assert_eq!(result.run_id, "run-1");
assert_eq!(result.qid, "q-2");
assert_eq!(result.answer.value, AnswerValue::No);
}
#[test]
@ -100,14 +111,14 @@ mod tests {
let payload = serde_json::json!({
"type": "block_actions",
"actions": [{
"action_id": "q-3:rs",
"action_id": "interview.answer",
"type": "button",
"value": "rs"
"value": "{\"kind\":\"selected\",\"run_id\":\"run-1\",\"qid\":\"q-3\",\"key\":\"rs\"}"
}]
});
let result = parse_interaction(&payload).unwrap();
assert_eq!(result.0, "q-3");
assert_eq!(result.1.value, AnswerValue::Text("rs".to_string()));
assert_eq!(result.qid, "q-3");
assert_eq!(result.answer.value, AnswerValue::Selected("rs".to_string()));
}
#[test]
@ -115,7 +126,7 @@ mod tests {
let payload = serde_json::json!({
"type": "block_actions",
"actions": [{
"action_id": "q-5:select",
"action_id": "interview.select",
"type": "checkboxes",
"selected_options": [
{ "value": "a" },
@ -131,14 +142,14 @@ mod tests {
let payload = serde_json::json!({
"type": "block_actions",
"actions": [{
"action_id": "q-5:submit",
"action_id": "interview.submit",
"type": "button",
"value": "submit"
"value": "{\"kind\":\"submit_multi\",\"run_id\":\"run-1\",\"qid\":\"q-5\"}"
}],
"state": {
"values": {
"q-5:checkboxes": {
"q-5:select": {
"interview.checkboxes": {
"interview.select": {
"type": "checkboxes",
"selected_options": [
{ "value": "auth" },
@ -150,10 +161,10 @@ mod tests {
}
});
let result = parse_interaction(&payload).unwrap();
assert_eq!(result.0, "q-5");
assert_eq!(result.qid, "q-5");
assert_eq!(
result.1.value,
AnswerValue::Text("auth, billing".to_string())
result.answer.value,
AnswerValue::MultiSelected(vec!["auth".to_string(), "billing".to_string()])
);
}
@ -162,14 +173,14 @@ mod tests {
let payload = serde_json::json!({
"type": "block_actions",
"actions": [{
"action_id": "q-5:submit",
"action_id": "interview.submit",
"type": "button",
"value": "submit"
"value": "{\"kind\":\"submit_multi\",\"run_id\":\"run-1\",\"qid\":\"q-5\"}"
}],
"state": {
"values": {
"q-5:checkboxes": {
"q-5:select": {
"interview.checkboxes": {
"interview.select": {
"type": "checkboxes",
"selected_options": []
}
@ -178,8 +189,8 @@ mod tests {
}
});
let result = parse_interaction(&payload).unwrap();
assert_eq!(result.0, "q-5");
assert_eq!(result.1.value, AnswerValue::Skipped);
assert_eq!(result.qid, "q-5");
assert_eq!(result.answer.value, AnswerValue::Skipped);
}
#[test]
@ -187,17 +198,12 @@ mod tests {
let payload = serde_json::json!({
"type": "block_actions",
"actions": [{
"action_id": "q-6:input",
"action_id": "interview.answer",
"type": "plain_text_input",
"value": "https://github.com/org/repo"
"value": "{\"kind\":\"selected\",\"run_id\":\"run-1\",\"qid\":\"q-6\",\"key\":\"input\"}"
}]
});
let result = parse_interaction(&payload).unwrap();
assert_eq!(result.0, "q-6");
assert_eq!(
result.1.value,
AnswerValue::Text("https://github.com/org/repo".to_string())
);
assert!(parse_interaction(&payload).is_none());
}
#[test]

View file

@ -4,5 +4,6 @@ pub mod config;
pub mod connection;
pub mod dispatch;
pub mod interaction;
pub mod payload;
pub mod socket;
pub mod threads;

View file

@ -0,0 +1,76 @@
use fabro_interview::Answer;
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SlackQuestionRef {
pub run_id: String,
pub qid: String,
}
#[derive(Debug, Clone)]
pub struct SlackAnswerSubmission {
pub run_id: String,
pub qid: String,
pub answer: Answer,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum SlackActionPayload {
Yes {
run_id: String,
qid: String,
},
No {
run_id: String,
qid: String,
},
Selected {
run_id: String,
qid: String,
key: String,
},
SubmitMulti {
run_id: String,
qid: String,
},
}
impl SlackActionPayload {
#[must_use]
pub fn question_ref(&self) -> SlackQuestionRef {
match self {
Self::Yes { run_id, qid }
| Self::No { run_id, qid }
| Self::Selected { run_id, qid, .. }
| Self::SubmitMulti { run_id, qid } => SlackQuestionRef {
run_id: run_id.clone(),
qid: qid.clone(),
},
}
}
}
#[must_use]
pub fn encode_action_value(payload: &SlackActionPayload) -> String {
serde_json::to_string(payload).expect("Slack action payload serialization should succeed")
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn action_payload_serializes_run_id_and_qid() {
let payload = SlackActionPayload::Selected {
run_id: "run_123".to_string(),
qid: "q_123".to_string(),
key: "approve".to_string(),
};
let json = encode_action_value(&payload);
assert_eq!(
json,
r#"{"kind":"selected","run_id":"run_123","qid":"q_123","key":"approve"}"#
);
}
}

View file

@ -3,9 +3,11 @@ use std::sync::Mutex;
use serde_json::Value;
use crate::payload::SlackQuestionRef;
#[derive(Default)]
pub struct ThreadRegistry {
ts_to_question: Mutex<HashMap<String, String>>,
ts_to_question: Mutex<HashMap<String, SlackQuestionRef>>,
}
impl ThreadRegistry {
@ -13,14 +15,20 @@ impl ThreadRegistry {
Self::default()
}
pub fn register(&self, message_ts: &str, question_id: &str) {
pub fn register(&self, message_ts: &str, run_id: &str, question_id: &str) {
self.ts_to_question
.lock()
.expect("thread registry lock poisoned")
.insert(message_ts.to_string(), question_id.to_string());
.insert(
message_ts.to_string(),
SlackQuestionRef {
run_id: run_id.to_string(),
qid: question_id.to_string(),
},
);
}
pub fn resolve(&self, thread_ts: &str) -> Option<String> {
pub fn resolve(&self, thread_ts: &str) -> Option<SlackQuestionRef> {
self.ts_to_question
.lock()
.expect("thread registry lock poisoned")
@ -73,8 +81,14 @@ mod tests {
#[test]
fn register_and_resolve() {
let registry = ThreadRegistry::new();
registry.register("1234.5678", "q-1");
assert_eq!(registry.resolve("1234.5678"), Some("q-1".to_string()));
registry.register("1234.5678", "run-1", "q-1");
assert_eq!(
registry.resolve("1234.5678"),
Some(SlackQuestionRef {
run_id: "run-1".to_string(),
qid: "q-1".to_string(),
})
);
}
#[test]
@ -86,7 +100,7 @@ mod tests {
#[test]
fn remove_clears_mapping() {
let registry = ThreadRegistry::new();
registry.register("1234.5678", "q-1");
registry.register("1234.5678", "run-1", "q-1");
registry.remove("1234.5678");
assert_eq!(registry.resolve("1234.5678"), None);
}

View file

@ -10,7 +10,7 @@ mod types;
pub use artifact_store::{ArtifactStore, NodeArtifact};
pub use error::{Result, StoreError};
pub use fabro_types::{RunBlobId, StageId};
pub use run_state::{NodeState, RunProjection};
pub use run_state::{NodeState, PendingInterviewRecord, RunProjection};
pub use slate::{Database, RunDatabase, Runs};
pub use types::{EventEnvelope, EventPayload, RunSummary};

View file

@ -7,8 +7,8 @@ use serde_json::Value;
use crate::{EventEnvelope, Result, RunSummary, StageId, StoreError};
use fabro_types::run_event::{
AgentCliStartedProps, AgentSessionStartedProps, CheckpointCompletedProps, RunCompletedProps,
RunFailedProps, StageCompletedProps, StagePromptProps,
AgentCliStartedProps, AgentSessionStartedProps, CheckpointCompletedProps, InterviewOption,
RunCompletedProps, RunFailedProps, StageCompletedProps, StagePromptProps,
};
use fabro_types::{
BilledModelUsage, Checkpoint, Conclusion, EventBody, FailureSignature, NodeStatusRecord,
@ -33,9 +33,23 @@ pub struct RunProjection {
pub sandbox: Option<SandboxRecord>,
pub final_patch: Option<String>,
pub pull_request: Option<PullRequestRecord>,
pub pending_interviews: BTreeMap<String, PendingInterviewRecord>,
nodes: HashMap<StageId, NodeState>,
}
#[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 started_at: Option<DateTime<Utc>>,
}
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
pub struct NodeState {
pub prompt: Option<String>,
@ -132,11 +146,13 @@ impl RunProjection {
self.pending_control = None;
self.conclusion = Some(conclusion_from_completed(props, ts)?);
self.final_patch.clone_from(&props.final_patch);
self.pending_interviews.clear();
}
EventBody::RunFailed(props) => {
self.status = Some(run_status_record(RunStatus::Failed, props.reason, ts));
self.pending_control = None;
self.conclusion = Some(conclusion_from_failed(props, ts));
self.pending_interviews.clear();
}
EventBody::RunRewound(_) => {
self.reset_for_rewind();
@ -190,6 +206,40 @@ impl RunProjection {
title: props.title.clone(),
});
}
EventBody::InterviewStarted(props) => {
if props.question_id.is_empty() {
return Ok(());
}
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(),
started_at: Some(ts),
},
);
}
EventBody::InterviewCompleted(props) => {
if !props.question_id.is_empty() {
self.pending_interviews.remove(&props.question_id);
}
}
EventBody::InterviewTimeout(props) => {
if !props.question_id.is_empty() {
self.pending_interviews.remove(&props.question_id);
}
}
EventBody::InterviewAborted(props) => {
if !props.question_id.is_empty() {
self.pending_interviews.remove(&props.question_id);
}
}
EventBody::StagePrompt(props) => {
let Some(node_id) = stored.node_id.as_deref() else {
return Ok(());
@ -376,6 +426,7 @@ impl RunProjection {
self.sandbox = None;
self.final_patch = None;
self.pull_request = None;
self.pending_interviews.clear();
self.nodes.clear();
}
}
@ -540,9 +591,31 @@ fn provider_used_from_agent_cli_started(props: &AgentCliStartedProps) -> Value {
mod tests {
use std::collections::HashMap;
use chrono::Utc;
use super::{NodeState, RunProjection};
use crate::StageId;
use fabro_types::{Checkpoint, RunControlAction};
use crate::{EventEnvelope, EventPayload, StageId};
use fabro_types::run_event::{InterviewCompletedProps, InterviewOption, InterviewStartedProps};
use fabro_types::{Checkpoint, EventBody, RunControlAction, RunEvent, fixtures};
fn test_event(seq: u32, body: EventBody, node_id: Option<&str>) -> EventEnvelope {
let event = RunEvent {
id: format!("evt-{seq}"),
ts: Utc::now(),
run_id: fixtures::RUN_1,
node_id: node_id.map(ToOwned::to_owned),
node_label: None,
session_id: None,
parent_session_id: None,
body,
};
EventEnvelope {
seq,
payload: EventPayload::new(serde_json::to_value(event).unwrap(), &fixtures::RUN_1)
.unwrap(),
}
}
#[test]
fn deserialize_projection_defaults_missing_nodes_and_checkpoints() {
@ -647,4 +720,63 @@ mod tests {
Some(RunControlAction::Unpause)
);
}
#[test]
fn interview_events_populate_and_clear_pending_interviews() {
let mut state = RunProjection::default();
state
.apply_event(&test_event(
1,
EventBody::InterviewStarted(InterviewStartedProps {
question_id: "q-1".to_string(),
question: "Approve deploy?".to_string(),
stage: "gate".to_string(),
question_type: "multiple_choice".to_string(),
options: vec![
InterviewOption {
key: "approve".to_string(),
label: "Approve".to_string(),
},
InterviewOption {
key: "revise".to_string(),
label: "Revise".to_string(),
},
],
allow_freeform: true,
timeout_seconds: Some(30.0),
context_display: Some("Latest draft".to_string()),
}),
Some("gate"),
))
.unwrap();
let pending = state
.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"));
state
.apply_event(&test_event(
2,
EventBody::InterviewCompleted(InterviewCompletedProps {
question_id: "q-1".to_string(),
question: "Approve deploy?".to_string(),
answer: "approve".to_string(),
duration_ms: 42,
}),
Some("gate"),
))
.unwrap();
assert!(
state.pending_interviews.is_empty(),
"completed interview should clear pending state"
);
}
}

View file

@ -1,6 +1,12 @@
use serde::{Deserialize, Serialize};
use serde_json::Value;
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct InterviewOption {
pub key: String,
pub label: String,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ParallelStartedProps {
pub visit: u32,
@ -34,12 +40,26 @@ pub struct ParallelCompletedProps {
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct InterviewStartedProps {
#[serde(default)]
pub question_id: String,
pub question: String,
#[serde(default)]
pub stage: String,
pub question_type: String,
#[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>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct InterviewCompletedProps {
#[serde(default)]
pub question_id: String,
pub question: String,
pub answer: String,
pub duration_ms: u64,
@ -47,7 +67,22 @@ pub struct InterviewCompletedProps {
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct InterviewTimeoutProps {
#[serde(default)]
pub question_id: String,
pub question: String,
#[serde(default)]
pub stage: String,
pub duration_ms: u64,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct InterviewAbortedProps {
#[serde(default)]
pub question_id: String,
pub question: String,
#[serde(default)]
pub stage: String,
pub reason: String,
pub duration_ms: u64,
}

View file

@ -95,6 +95,8 @@ pub enum EventBody {
InterviewCompleted(InterviewCompletedProps),
#[serde(rename = "interview.timeout")]
InterviewTimeout(InterviewTimeoutProps),
#[serde(rename = "interview.aborted")]
InterviewAborted(InterviewAbortedProps),
#[serde(rename = "checkpoint.completed")]
CheckpointCompleted(CheckpointCompletedProps),
#[serde(rename = "checkpoint.failed")]
@ -310,6 +312,7 @@ impl EventBody {
Self::InterviewStarted(_) => "interview.started",
Self::InterviewCompleted(_) => "interview.completed",
Self::InterviewTimeout(_) => "interview.timeout",
Self::InterviewAborted(_) => "interview.aborted",
Self::CheckpointCompleted(_) => "checkpoint.completed",
Self::CheckpointFailed(_) => "checkpoint.failed",
Self::GitCommit(_) => "git.commit",

View file

@ -28,7 +28,7 @@ pub use sandbox::{
pub use server::{
ApiAuthStrategy, ApiSettings, ArtifactStorageBackend, ArtifactStorageSettings, AuthProvider,
AuthSettings, FeaturesSettings, GitAuthorSettings, GitProvider, GitSettings, LogSettings,
TlsSettings, WebSettings, WebhookSettings, WebhookStrategy,
SlackSettings, TlsSettings, WebSettings, WebhookSettings, WebhookStrategy,
};
pub use user::{ClientTlsSettings, ExecSettings, OutputFormat, PermissionLevel, ServerSettings};
@ -95,6 +95,8 @@ pub struct Settings {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub web: Option<WebSettings>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub slack: Option<SlackSettings>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub api: Option<ApiSettings>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub features: Option<FeaturesSettings>,
@ -187,6 +189,10 @@ impl Settings {
.clone()
.unwrap_or_else(|| fabro_util::Home::from_env().storage_dir())
}
pub fn slack_settings(&self) -> Option<&SlackSettings> {
self.slack.as_ref()
}
}
#[cfg(test)]

View file

@ -116,6 +116,11 @@ impl Default for WebSettings {
}
}
#[derive(Debug, Clone, Default, Deserialize, PartialEq, Serialize, crate::Combine)]
pub struct SlackSettings {
pub default_channel: Option<String>,
}
#[derive(Debug, Clone, Default, Deserialize, PartialEq, Serialize)]
pub struct FeaturesSettings {
#[serde(default)]

View file

@ -212,20 +212,38 @@ pub enum Event {
results: Vec<serde_json::Value>,
},
InterviewStarted {
question_id: String,
question: String,
stage: String,
question_type: String,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
options: Vec<fabro_types::InterviewOption>,
#[serde(default)]
allow_freeform: bool,
#[serde(default, skip_serializing_if = "Option::is_none")]
timeout_seconds: Option<f64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
context_display: Option<String>,
},
InterviewCompleted {
question_id: String,
question: String,
answer: String,
duration_ms: u64,
},
InterviewTimeout {
question_id: String,
question: String,
stage: String,
duration_ms: u64,
},
InterviewAborted {
question_id: String,
question: String,
stage: String,
reason: String,
duration_ms: u64,
},
CheckpointCompleted {
node_id: String,
status: String,
@ -745,6 +763,14 @@ impl Event {
} => {
warn!(stage, duration_ms, "Interview timeout");
}
Self::InterviewAborted {
stage,
reason,
duration_ms,
..
} => {
warn!(stage, reason, duration_ms, "Interview aborted");
}
Self::CheckpointCompleted {
node_id,
status,
@ -1120,6 +1146,7 @@ pub fn event_name(event: &Event) -> &'static str {
Event::InterviewStarted { .. } => "interview.started",
Event::InterviewCompleted { .. } => "interview.completed",
Event::InterviewTimeout { .. } => "interview.timeout",
Event::InterviewAborted { .. } => "interview.aborted",
Event::CheckpointCompleted { .. } => "checkpoint.completed",
Event::CheckpointFailed { .. } => "checkpoint.failed",
Event::GitCommit { .. } => "git.commit",
@ -1318,6 +1345,7 @@ fn stored_event_fields(event: &Event) -> StoredEventFields {
Event::Prompt { stage, .. }
| Event::InterviewStarted { stage, .. }
| Event::InterviewTimeout { stage, .. }
| Event::InterviewAborted { stage, .. }
| Event::Failover { stage, .. } => {
let node_id = Some(stage.clone());
let node_label = default_node_label(node_id.as_ref(), None);
@ -1594,28 +1622,57 @@ fn event_body_from_event(event: &Event) -> EventBody {
results: results.clone(),
}),
Event::InterviewStarted {
question_id,
question,
stage,
question_type,
..
options,
allow_freeform,
timeout_seconds,
context_display,
} => EventBody::InterviewStarted(fabro_types::InterviewStartedProps {
question_id: question_id.clone(),
question: question.clone(),
stage: stage.clone(),
question_type: question_type.clone(),
options: options.clone(),
allow_freeform: *allow_freeform,
timeout_seconds: *timeout_seconds,
context_display: context_display.clone(),
}),
Event::InterviewCompleted {
question_id,
question,
answer,
duration_ms,
} => EventBody::InterviewCompleted(fabro_types::InterviewCompletedProps {
question_id: question_id.clone(),
question: question.clone(),
answer: answer.clone(),
duration_ms: *duration_ms,
}),
Event::InterviewTimeout {
question_id,
question,
stage,
duration_ms,
..
} => EventBody::InterviewTimeout(fabro_types::InterviewTimeoutProps {
question_id: question_id.clone(),
question: question.clone(),
stage: stage.clone(),
duration_ms: *duration_ms,
}),
Event::InterviewAborted {
question_id,
question,
stage,
reason,
duration_ms,
} => EventBody::InterviewAborted(fabro_types::InterviewAbortedProps {
question_id: question_id.clone(),
question: question.clone(),
stage: stage.clone(),
reason: reason.clone(),
duration_ms: *duration_ms,
}),
Event::CheckpointCompleted {

View file

@ -1,5 +1,6 @@
use std::path::Path;
use std::sync::Arc;
use std::sync::atomic::Ordering;
use std::time::Instant;
use async_trait::async_trait;
@ -12,6 +13,7 @@ use crate::millis_u64;
use crate::outcome::{Outcome, OutcomeExt};
use fabro_graphviz::graph::{Graph, Node};
use fabro_interview::{Answer, AnswerValue, Interviewer, Question, QuestionOption, QuestionType};
use ulid::Ulid;
use super::{EngineServices, Handler};
@ -85,9 +87,10 @@ impl HumanHandler {
self
}
fn emit(&self, event: &Event) {
if let Some(emitter) = &self.emitter {
emitter.emit(event);
fn emit(&self, default_emitter: &Arc<Emitter>, event: &Event) {
match &self.emitter {
Some(emitter) => emitter.emit(event),
None => default_emitter.emit(event),
}
}
}
@ -143,7 +146,7 @@ impl Handler for HumanHandler {
context: &Context,
graph: &Graph,
_run_dir: &Path,
_services: &EngineServices,
services: &EngineServices,
) -> Result<Outcome, FabroError> {
// 1. Derive choices from outgoing edges
let edges = graph.outgoing_edges(&node.id);
@ -185,6 +188,7 @@ impl Handler for HumanHandler {
QuestionType::MultipleChoice
};
let mut question = Question::new(node.label(), question_type);
question.id = Ulid::new().to_string();
question.options = options;
question.allow_freeform = freeform_target.is_some();
question.stage.clone_from(&node.id);
@ -203,21 +207,41 @@ impl Handler for HumanHandler {
// 3. Present to interviewer
let question_text = node.label().to_string();
self.emit(&Event::InterviewStarted {
question: question_text.clone(),
stage: node.id.clone(),
question_type: question.question_type.to_string(),
});
let question_id = question.id.clone();
self.emit(
&services.emitter,
&Event::InterviewStarted {
question_id: question_id.clone(),
question: question_text.clone(),
stage: node.id.clone(),
question_type: question.question_type.to_string(),
options: question
.options
.iter()
.map(|option| fabro_types::run_event::InterviewOption {
key: option.key.clone(),
label: option.label.clone(),
})
.collect(),
allow_freeform: question.allow_freeform,
timeout_seconds: question.timeout_seconds,
context_display: question.context_display.clone(),
},
);
let interview_start = Instant::now();
let answer = self.interviewer.ask(question).await;
// 4. Handle timeout
if answer.value == AnswerValue::Timeout {
self.emit(&Event::InterviewTimeout {
question: question_text,
stage: node.id.clone(),
duration_ms: millis_u64(interview_start.elapsed()),
});
self.emit(
&services.emitter,
&Event::InterviewTimeout {
question_id: question_id.clone(),
question: question_text,
stage: node.id.clone(),
duration_ms: millis_u64(interview_start.elapsed()),
},
);
let default_choice = node
.attrs
.get("human.default_choice")
@ -234,20 +258,51 @@ impl Handler for HumanHandler {
// 5. Handle unanswered / aborted interview sessions.
if answer.value == AnswerValue::Aborted {
if services
.cancel_requested
.as_ref()
.is_some_and(|flag| flag.load(Ordering::SeqCst))
{
return Err(FabroError::Cancelled);
}
self.emit(
&services.emitter,
&Event::InterviewAborted {
question_id: question_id.clone(),
question: question_text,
stage: node.id.clone(),
reason: "aborted".to_string(),
duration_ms: millis_u64(interview_start.elapsed()),
},
);
return Ok(unanswered_human_gate(
"human interaction aborted before an answer was provided",
));
}
if answer.value == AnswerValue::Skipped {
self.emit(
&services.emitter,
&Event::InterviewAborted {
question_id: question_id.clone(),
question: question_text,
stage: node.id.clone(),
reason: "skipped".to_string(),
duration_ms: millis_u64(interview_start.elapsed()),
},
);
return Ok(unanswered_human_gate("human skipped interaction"));
}
// Emit interview completed for successful interactions
self.emit(&Event::InterviewCompleted {
question: question_text,
answer: answer_text(&answer),
duration_ms: millis_u64(interview_start.elapsed()),
});
self.emit(
&services.emitter,
&Event::InterviewCompleted {
question_id,
question: question_text,
answer: answer_text(&answer),
duration_ms: millis_u64(interview_start.elapsed()),
},
);
// 6. Try fixed-choice match
if let Some(selected) = find_choice_match(&answer, &choices) {

View file

@ -1,7 +1,5 @@
use std::path::Path;
use fabro_config::RunScratch;
use crate::error::FabroError;
use crate::event::{Event, append_event_to_sink};
use crate::outcome::StageStatus;
@ -52,14 +50,5 @@ pub async fn resume(run_dir: &Path, services: StartServices) -> Result<Started,
}
fn cleanup_resume_artifacts(run_dir: &Path) {
let run_scratch = RunScratch::new(run_dir);
for path in [
run_scratch.interview_request_path(),
run_scratch.interview_response_path(),
run_scratch.interview_claim_path(),
] {
let _ = std::fs::remove_file(path);
}
let _ = std::fs::remove_file(run_dir.join("detached_failure.json"));
}