diff --git a/Cargo.lock b/Cargo.lock index 174a0230e..77ae61b4a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1873,6 +1873,7 @@ dependencies = [ "fabro-proc", "fabro-retro", "fabro-sandbox", + "fabro-slack", "fabro-store", "fabro-types", "fabro-util", diff --git a/docs-internal/run-directory-keys.md b/docs-internal/run-directory-keys.md index 043bbbe7d..89568554f 100644 --- a/docs-internal/run-directory-keys.md +++ b/docs-internal/run-directory-keys.md @@ -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 | diff --git a/docs/api-reference/fabro-api.yaml b/docs/api-reference/fabro-api.yaml index 805c5ff55..2c4d6ccd2 100644 --- a/docs/api-reference/fabro-api.yaml +++ b/docs/api-reference/fabro-api.yaml @@ -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. diff --git a/docs/reference/run-directory.mdx b/docs/reference/run-directory.mdx index 722de705a..91da2b393 100644 --- a/docs/reference/run-directory.mdx +++ b/docs/reference/run-directory.mdx @@ -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/ diff --git a/lib/crates/fabro-cli/src/commands/run/attach.rs b/lib/crates/fabro-cli/src/commands/run/attach.rs index 51a7e02b8..e682bb519 100644 --- a/lib/crates/fabro-cli/src/commands/run/attach.rs +++ b/lib/crates/fabro-cli/src/commands/run/attach.rs @@ -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 } diff --git a/lib/crates/fabro-cli/src/commands/run/runner.rs b/lib/crates/fabro-cli/src/commands/run/runner.rs index 04e8a3e5f..f6d39c36b 100644 --- a/lib/crates/fabro-cli/src/commands/run/runner.rs +++ b/lib/crates/fabro-cli/src/commands/run/runner.rs @@ -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(reader: R, broker: Arc) +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::(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); + } } diff --git a/lib/crates/fabro-cli/tests/it/cmd/runner.rs b/lib/crates/fabro-cli/tests/it/cmd/runner.rs index f4c11d718..47fc6780d 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/runner.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/runner.rs @@ -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" + ))); +} diff --git a/lib/crates/fabro-config/src/config.rs b/lib/crates/fabro-config/src/config.rs index fd3a44c34..6e87eaee7 100644 --- a/lib/crates/fabro-config/src/config.rs +++ b/lib/crates/fabro-config/src/config.rs @@ -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, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub slack: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] pub api: Option, @@ -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), diff --git a/lib/crates/fabro-config/src/server.rs b/lib/crates/fabro-config/src/server.rs index 4c8b9e563..e038b5ab7 100644 --- a/lib/crates/fabro-config/src/server.rs +++ b/lib/crates/fabro-config/src/server.rs @@ -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 for WebSettings { } } +#[derive(Debug, Clone, Default, Deserialize, PartialEq, Serialize, crate::Combine)] +pub struct SlackConfig { + pub default_channel: Option, +} + +impl From 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, diff --git a/lib/crates/fabro-config/src/settings.rs b/lib/crates/fabro-config/src/settings.rs index 7643e6993..90369ef45 100644 --- a/lib/crates/fabro-config/src/settings.rs +++ b/lib/crates/fabro-config/src/settings.rs @@ -35,6 +35,7 @@ impl TryFrom 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), diff --git a/lib/crates/fabro-config/src/storage.rs b/lib/crates/fabro-config/src/storage.rs index 4fa0c9ce3..24e1f6b69 100644 --- a/lib/crates/fabro-config/src/storage.rs +++ b/lib/crates/fabro-config/src/storage.rs @@ -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 diff --git a/lib/crates/fabro-interview/src/control.rs b/lib/crates/fabro-interview/src/control.rs new file mode 100644 index 000000000..2e48912e0 --- /dev/null +++ b/lib/crates/fabro-interview/src/control.rs @@ -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>, + queued: HashMap, +} + +#[derive(Default)] +pub struct InterviewBroker { + state: Mutex, +} + +impl InterviewBroker { + #[must_use] + pub fn new() -> Self { + Self::default() + } + + pub async fn register(&self, question_id: String) -> oneshot::Receiver { + 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::>(); + 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, +} + +impl ControlInterviewer { + #[must_use] + pub fn new(broker: Arc) -> Self { + Self { broker } + } +} + +#[async_trait] +impl Interviewer for ControlInterviewer { + async fn ask(&self, question: Question) -> Answer { + let receiver = self.broker.register(question.id.clone()).await; + 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) + ); + } +} diff --git a/lib/crates/fabro-interview/src/control_protocol.rs b/lib/crates/fabro-interview/src/control_protocol.rs new file mode 100644 index 000000000..5fa8105b5 --- /dev/null +++ b/lib/crates/fabro-interview/src/control_protocol.rs @@ -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, 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 }, + Text { text: String }, +} + +impl From 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 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); + } +} diff --git a/lib/crates/fabro-interview/src/file.rs b/lib/crates/fabro-interview/src/file.rs deleted file mode 100644 index ec9e63793..000000000 --- a/lib/crates/fabro-interview/src/file.rs +++ /dev/null @@ -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 = None; - loop { - match fs::read_to_string(&response_path).await { - Ok(data) => match serde_json::from_str::(&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); - } -} diff --git a/lib/crates/fabro-interview/src/lib.rs b/lib/crates/fabro-interview/src/lib.rs index 513bc6ce9..fc5bdee3c 100644 --- a/lib/crates/fabro-interview/src/lib.rs +++ b/lib/crates/fabro-interview/src/lib.rs @@ -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, @@ -60,6 +62,7 @@ pub struct Question { impl Question { pub fn new(text: impl Into, 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); + } } diff --git a/lib/crates/fabro-interview/src/web.rs b/lib/crates/fabro-interview/src/web.rs deleted file mode 100644 index ad04730c7..000000000 --- a/lib/crates/fabro-interview/src/web.rs +++ /dev/null @@ -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>, - questions: Vec, - 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>, -} - -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 { - 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::>() - }; - - 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); - } -} diff --git a/lib/crates/fabro-server/Cargo.toml b/lib/crates/fabro-server/Cargo.toml index db4d1149e..83f1395a3 100644 --- a/lib/crates/fabro-server/Cargo.toml +++ b/lib/crates/fabro-server/Cargo.toml @@ -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"] } diff --git a/lib/crates/fabro-server/src/demo/mod.rs b/lib/crates/fabro-server/src/demo/mod.rs index b67d9c08f..8472463e0 100644 --- a/lib/crates/fabro-server/src/demo/mod.rs +++ b/lib/crates/fabro-server/src/demo/mod.rs @@ -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, }, ] } diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index e84c3a1d6..22802ce93 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -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, enqueued_at: Instant, // Populated when running: - interviewer: Option>, + answer_transport: Option, + accepted_questions: HashSet, event_tx: Option>, checkpoint: Option, cancel_tx: Option>, @@ -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) -> HandlerRegistry + Send + Sync; +#[derive(Clone)] +enum RunAnswerTransport { + Subprocess { + control_tx: mpsc::Sender, + }, + InProcess { + broker: Arc, + }, +} + +#[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>>, + thread_registry: Arc, +} + +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, 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>, @@ -315,6 +527,8 @@ pub struct AppState { pub(crate) local_daemon_mode: bool, shutting_down: AtomicBool, registry_factory_override: Option>, + slack_service: Option>, + slack_started: AtomicBool, } fn nonzero_i64(value: i64) -> Option { @@ -526,8 +740,55 @@ fn decode_secret_pem(name: &str, raw: &str) -> Result { .map_err(|err| format!("{name} base64 decoded to invalid UTF-8: {err}")) } +fn start_optional_slack_service(state: &Arc) { + 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 = + 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, 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> { 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, 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 { + 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 { + 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, + run_id: RunId, mut run_events: broadcast::Receiver, - global_event_tx: broadcast::Sender, ) { 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, +) -> 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 { + 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> { - 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>, @@ -2937,7 +3420,8 @@ async fn execute_run_in_process(state: Arc, run_id: RunId) { let _ = queued_for; // Create interviewer and event plumbing (this is the "provisioning" phase) - let interviewer = Arc::new(WebInterviewer::new()); + let broker = Arc::new(InterviewBroker::new()); + let interviewer: Arc = 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, run_id: RunId) { let registry_override = state .registry_factory_override .as_ref() - .map(|factory| Arc::new(factory(Arc::clone(&interviewer) as Arc))); + .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, 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, 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, run_id: RunId) { run_id, cancel_token: Some(Arc::clone(&cancel_token)), emitter: Arc::clone(&emitter), - interviewer: Arc::clone(&interviewer) as Arc, + 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, 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, 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, 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, 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::>(); + (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 = 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::::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::::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(()); diff --git a/lib/crates/fabro-server/tests/it/scenario/lifecycle.rs b/lib/crates/fabro-server/tests/it/scenario/lifecycle.rs index 3ba6d6e37..69d694d69 100644 --- a/lib/crates/fabro-server/tests/it/scenario/lifecycle.rs +++ b/lib/crates/fabro-server/tests/it/scenario/lifecycle.rs @@ -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() diff --git a/lib/crates/fabro-slack/Cargo.toml b/lib/crates/fabro-slack/Cargo.toml index 97c48e23a..681a683f5 100644 --- a/lib/crates/fabro-slack/Cargo.toml +++ b/lib/crates/fabro-slack/Cargo.toml @@ -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 diff --git a/lib/crates/fabro-slack/examples/slack_e2e.rs b/lib/crates/fabro-slack/examples/slack_e2e.rs deleted file mode 100644 index 532ddb162..000000000 --- a/lib/crates/fabro-slack/examples/slack_e2e.rs +++ /dev/null @@ -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 { - 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, - 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!"); -} diff --git a/lib/crates/fabro-slack/src/blocks.rs b/lib/crates/fabro-slack/src/blocks.rs index 904085a9d..c3869865c 100644 --- a/lib/crates/fabro-slack/src/blocks.rs +++ b/lib/crates/fabro-slack/src/blocks.rs @@ -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 { ))] } -pub fn question_to_blocks(question_id: &str, question: &Question) -> Vec { +pub fn question_to_blocks(run_id: &str, question_id: &str, question: &Question) -> Vec { 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 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 let elements: Vec = 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 .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\"") ); } } diff --git a/lib/crates/fabro-slack/src/connection.rs b/lib/crates/fabro-slack/src/connection.rs index eaf922da9..9a13301f9 100644 --- a/lib/crates/fabro-slack/src/connection.rs +++ b/lib/crates/fabro-slack/src/connection.rs @@ -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, thread_registry: &ThreadRegistry, + on_submit: &Arc, ) -> 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, thread_registry: &ThreadRegistry, + on_submit: Arc, ) { 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 = + 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, ®istry()); - 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, ®istry, &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); } } diff --git a/lib/crates/fabro-slack/src/dispatch.rs b/lib/crates/fabro-slack/src/dispatch.rs index 7afe56c27..d3002459d 100644 --- a/lib/crates/fabro-slack/src/dispatch.rs +++ b/lib/crates/fabro-slack/src/dispatch.rs @@ -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, ®istry); 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, ®istry); 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()) ); } diff --git a/lib/crates/fabro-slack/src/interaction.rs b/lib/crates/fabro-slack/src/interaction.rs index 1884fc54c..6d8acdc1c 100644 --- a/lib/crates/fabro-slack/src/interaction.rs +++ b/lib/crates/fabro-slack/src/interaction.rs @@ -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 { 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] diff --git a/lib/crates/fabro-slack/src/lib.rs b/lib/crates/fabro-slack/src/lib.rs index 5fa7bb737..b5c2c8a5c 100644 --- a/lib/crates/fabro-slack/src/lib.rs +++ b/lib/crates/fabro-slack/src/lib.rs @@ -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; diff --git a/lib/crates/fabro-slack/src/payload.rs b/lib/crates/fabro-slack/src/payload.rs new file mode 100644 index 000000000..c8cf2035b --- /dev/null +++ b/lib/crates/fabro-slack/src/payload.rs @@ -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"}"# + ); + } +} diff --git a/lib/crates/fabro-slack/src/threads.rs b/lib/crates/fabro-slack/src/threads.rs index 0fb1a02c7..3fc9870bc 100644 --- a/lib/crates/fabro-slack/src/threads.rs +++ b/lib/crates/fabro-slack/src/threads.rs @@ -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>, + ts_to_question: Mutex>, } 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 { + pub fn resolve(&self, thread_ts: &str) -> Option { 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); } diff --git a/lib/crates/fabro-store/src/lib.rs b/lib/crates/fabro-store/src/lib.rs index ff63fe0d4..c9e52f739 100644 --- a/lib/crates/fabro-store/src/lib.rs +++ b/lib/crates/fabro-store/src/lib.rs @@ -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}; diff --git a/lib/crates/fabro-store/src/run_state.rs b/lib/crates/fabro-store/src/run_state.rs index 0402db3af..b4f8df870 100644 --- a/lib/crates/fabro-store/src/run_state.rs +++ b/lib/crates/fabro-store/src/run_state.rs @@ -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, pub final_patch: Option, pub pull_request: Option, + pub pending_interviews: BTreeMap, nodes: HashMap, } +#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)] +pub struct PendingInterviewRecord { + pub question_id: String, + pub question: String, + pub stage: String, + pub question_type: String, + pub options: Vec, + pub allow_freeform: bool, + pub timeout_seconds: Option, + pub context_display: Option, + pub started_at: Option>, +} + #[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)] pub struct NodeState { pub prompt: Option, @@ -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" + ); + } } diff --git a/lib/crates/fabro-types/src/run_event/misc.rs b/lib/crates/fabro-types/src/run_event/misc.rs index d4ab4d9e5..f1ee37455 100644 --- a/lib/crates/fabro-types/src/run_event/misc.rs +++ b/lib/crates/fabro-types/src/run_event/misc.rs @@ -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, + #[serde(default)] + pub allow_freeform: bool, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub timeout_seconds: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub context_display: Option, } #[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, } diff --git a/lib/crates/fabro-types/src/run_event/mod.rs b/lib/crates/fabro-types/src/run_event/mod.rs index eafaeffc3..ed6663073 100644 --- a/lib/crates/fabro-types/src/run_event/mod.rs +++ b/lib/crates/fabro-types/src/run_event/mod.rs @@ -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", diff --git a/lib/crates/fabro-types/src/settings/mod.rs b/lib/crates/fabro-types/src/settings/mod.rs index 83e6473c1..229b65b36 100644 --- a/lib/crates/fabro-types/src/settings/mod.rs +++ b/lib/crates/fabro-types/src/settings/mod.rs @@ -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, #[serde(default, skip_serializing_if = "Option::is_none")] + pub slack: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] pub api: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub features: Option, @@ -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)] diff --git a/lib/crates/fabro-types/src/settings/server.rs b/lib/crates/fabro-types/src/settings/server.rs index 2ad6437a1..83e943b87 100644 --- a/lib/crates/fabro-types/src/settings/server.rs +++ b/lib/crates/fabro-types/src/settings/server.rs @@ -116,6 +116,11 @@ impl Default for WebSettings { } } +#[derive(Debug, Clone, Default, Deserialize, PartialEq, Serialize, crate::Combine)] +pub struct SlackSettings { + pub default_channel: Option, +} + #[derive(Debug, Clone, Default, Deserialize, PartialEq, Serialize)] pub struct FeaturesSettings { #[serde(default)] diff --git a/lib/crates/fabro-workflow/src/event.rs b/lib/crates/fabro-workflow/src/event.rs index 437869fed..518b052b1 100644 --- a/lib/crates/fabro-workflow/src/event.rs +++ b/lib/crates/fabro-workflow/src/event.rs @@ -212,20 +212,38 @@ pub enum Event { results: Vec, }, InterviewStarted { + question_id: String, question: String, stage: String, question_type: String, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + options: Vec, + #[serde(default)] + allow_freeform: bool, + #[serde(default, skip_serializing_if = "Option::is_none")] + timeout_seconds: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + context_display: Option, }, 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 { diff --git a/lib/crates/fabro-workflow/src/handler/human.rs b/lib/crates/fabro-workflow/src/handler/human.rs index 695c86d3a..68845b159 100644 --- a/lib/crates/fabro-workflow/src/handler/human.rs +++ b/lib/crates/fabro-workflow/src/handler/human.rs @@ -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, 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 { // 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) { diff --git a/lib/crates/fabro-workflow/src/operations/resume.rs b/lib/crates/fabro-workflow/src/operations/resume.rs index e3aa31a8c..cbbfc2032 100644 --- a/lib/crates/fabro-workflow/src/operations/resume.rs +++ b/lib/crates/fabro-workflow/src/operations/resume.rs @@ -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