From 326e0c27fa9dd6cd14f59ef9906f59fa97803260 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Tue, 7 Apr 2026 19:23:36 -0400 Subject: [PATCH] refactor(interview): move run answers onto control channels Persist pending interviews in run state, deliver accepted answers to workers through the server-owned control path, and remove the old scratch-file and WebInterviewer transports. This also moves Slack onto the canonical server answer flow, adds richer question metadata to the API and run events, and covers the subprocess question lifecycle with end-to-end tests. --- Cargo.lock | 1 + docs-internal/run-directory-keys.md | 4 +- docs/api-reference/fabro-api.yaml | 16 + docs/reference/run-directory.mdx | 7 +- .../fabro-cli/src/commands/run/attach.rs | 4 + .../fabro-cli/src/commands/run/runner.rs | 89 ++- lib/crates/fabro-cli/tests/it/cmd/runner.rs | 154 ++++ lib/crates/fabro-config/src/config.rs | 6 +- lib/crates/fabro-config/src/server.rs | 15 +- lib/crates/fabro-config/src/settings.rs | 1 + lib/crates/fabro-config/src/storage.rs | 36 - lib/crates/fabro-interview/src/control.rs | 168 ++++ .../fabro-interview/src/control_protocol.rs | 100 +++ lib/crates/fabro-interview/src/file.rs | 393 ---------- lib/crates/fabro-interview/src/lib.rs | 34 +- lib/crates/fabro-interview/src/web.rs | 324 -------- lib/crates/fabro-server/Cargo.toml | 1 + lib/crates/fabro-server/src/demo/mod.rs | 6 + lib/crates/fabro-server/src/server.rs | 715 ++++++++++++++---- .../tests/it/scenario/lifecycle.rs | 24 +- lib/crates/fabro-slack/Cargo.toml | 4 - lib/crates/fabro-slack/examples/slack_e2e.rs | 204 ----- lib/crates/fabro-slack/src/blocks.rs | 78 +- lib/crates/fabro-slack/src/connection.rs | 144 ++-- lib/crates/fabro-slack/src/dispatch.rs | 47 +- lib/crates/fabro-slack/src/interaction.rs | 126 +-- lib/crates/fabro-slack/src/lib.rs | 1 + lib/crates/fabro-slack/src/payload.rs | 76 ++ lib/crates/fabro-slack/src/threads.rs | 28 +- lib/crates/fabro-store/src/lib.rs | 2 +- lib/crates/fabro-store/src/run_state.rs | 140 +++- lib/crates/fabro-types/src/run_event/misc.rs | 35 + lib/crates/fabro-types/src/run_event/mod.rs | 3 + lib/crates/fabro-types/src/settings/mod.rs | 8 +- lib/crates/fabro-types/src/settings/server.rs | 5 + lib/crates/fabro-workflow/src/event.rs | 61 +- .../fabro-workflow/src/handler/human.rs | 93 ++- .../fabro-workflow/src/operations/resume.rs | 11 - 38 files changed, 1830 insertions(+), 1334 deletions(-) create mode 100644 lib/crates/fabro-interview/src/control.rs create mode 100644 lib/crates/fabro-interview/src/control_protocol.rs delete mode 100644 lib/crates/fabro-interview/src/file.rs delete mode 100644 lib/crates/fabro-interview/src/web.rs delete mode 100644 lib/crates/fabro-slack/examples/slack_e2e.rs create mode 100644 lib/crates/fabro-slack/src/payload.rs 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