From ac4c67e536425a3f5e5c2a1424ca5e6deeba1f33 Mon Sep 17 00:00:00 2001 From: Edwin Date: Sun, 2 Aug 2026 07:49:51 -0700 Subject: [PATCH] Show that a Slack turn is still running MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A service turn that takes a while left its Slack thread completely silent until the answer arrived. From the thread that is indistinguishable from a delivery that was dropped, and the reasonable response β€” send it again β€” starts a second turn in the same thread. Slack channels now show that a turn is in flight, configured per channel: [channels.my-bot] progress = "placeholder" # off | placeholder | reaction | both - placeholder (default) posts a thread message that later becomes the answer - reaction marks the triggering message πŸ‘€, then βœ… (⚠️ on failure) Nothing appears for a turn that answers promptly. The affordance waits 8s first, so the common case stays exactly as clean as it is today; it exists for a wait that has already become long enough to look like a failure. Three things make it honest rather than decorative: - A turn stopped at a tool approval says so and names the tool. That turn will not resume until an operator acts in the TUI, and the person waiting in Slack cannot see that prompt β€” so "working on it" would be a lie. The ingress publishes the turn's phase on a watch channel; the channel renders it. Publishing is advisory and nothing waits on a reader. - Whatever the affordance posted is replaced by the outcome, including when the turn fails. Failures previously went only to the daemon log, leaving the asker with silence; now the thread says the turn ended without an answer. - Anything cosmetic that Slack refuses is logged, not fatal. reactions:write is a scope an existing app install will not have, so a missing scope must not turn a delivered answer into a failed delivery. The channel put path cannot express `progress` yet, so it carries the existing value forward β€” otherwise saving an unrelated field would silently reset an operator's choice. Also fixes a latent conflation: reactions target the message that triggered the turn, which is not the thread root for a reply inside a thread. --- crates/daemon/src/service.rs | 103 ++++ crates/daemon/src/service/ingress.rs | 113 +++- crates/daemon/src/service/slack.rs | 564 +++++++++++++++++- crates/daemon/src/service_supervisor.rs | 2 + docs/services.md | 38 ++ ...nnel-shows-that-a-turn-is-still-running.md | 75 +++ 6 files changed, 864 insertions(+), 31 deletions(-) create mode 100644 specs/0178-a-channel-shows-that-a-turn-is-still-running.md diff --git a/crates/daemon/src/service.rs b/crates/daemon/src/service.rs index 147b33c7..10a2bea6 100644 --- a/crates/daemon/src/service.rs +++ b/crates/daemon/src/service.rs @@ -131,6 +131,36 @@ pub enum ServiceRouting { Single, } +/// What a Slack channel shows while a turn it accepted is still running. +/// +/// A long turn is indistinguishable from a dropped one when the channel stays +/// silent, so the operator picks how visible the wait should be. `Reaction` +/// and `Both` call `reactions.add`, which needs the `reactions:write` scope β€” +/// an app the operator has not reinstalled since granting it will log the +/// refusal and keep answering normally. +#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "kebab-case")] +pub enum SlackProgress { + /// Say nothing until the answer is ready. + Off, + /// A thread message that later becomes the answer itself. + #[default] + Placeholder, + /// An emoji reaction on the message that triggered the turn. + Reaction, + Both, +} + +impl SlackProgress { + pub(crate) fn posts_placeholder(self) -> bool { + matches!(self, Self::Placeholder | Self::Both) + } + + pub(crate) fn reacts(self) -> bool { + matches!(self, Self::Reaction | Self::Both) + } +} + #[derive(Debug, Clone, Serialize, Deserialize)] pub struct ServiceChannelConfig { #[serde(default)] @@ -147,6 +177,9 @@ pub struct ServiceChannelConfig { pub allowed_workspaces: Vec, #[serde(default, skip_serializing_if = "Vec::is_empty")] pub allowed_channels: Vec, + /// Slack only. Omitted definitions keep the default affordance. + #[serde(default)] + pub progress: SlackProgress, } fn default_channel_enabled() -> bool { @@ -469,6 +502,13 @@ pub fn put_channel( bot_token, allowed_workspaces: normalize_allowlist(params.channel.allowed_workspaces), allowed_channels: normalize_allowlist(params.channel.allowed_channels), + // Not editable through this path yet, so carry the operator's choice + // forward. Defaulting here would silently reset a definition every + // time an unrelated field (a rotated token, an allowlist) was saved. + progress: existing + .as_ref() + .map(|channel| channel.progress) + .unwrap_or_default(), }; service .channels @@ -881,6 +921,7 @@ pub(crate) fn slack_config( bot_token, allowed_workspaces: channel.allowed_workspaces.clone(), allowed_channels: channel.allowed_channels.clone(), + progress: channel.progress, }) } @@ -944,6 +985,7 @@ mod tests { bot_token: None, allowed_workspaces: Vec::new(), allowed_channels: Vec::new(), + progress: Default::default(), }, )]), } @@ -1408,6 +1450,26 @@ mod tests { assert_eq!(services["alerts"].channels["http"].port, Some(8787)); } + #[test] + fn a_slack_channel_chooses_its_progress_affordance() { + let dir = tempfile::tempdir().unwrap(); + std::fs::write( + dir.path().join("bot.toml"), + "harness = \"codex\"\n\ + [channels.a]\nkind = \"slack\"\nprogress = \"reaction\"\n\ + [channels.b]\nkind = \"slack\"\nprogress = \"off\"\n\ + [channels.c]\nkind = \"slack\"\n", + ) + .unwrap(); + + let channels = &load_definitions(dir.path()).unwrap()["bot"].channels; + assert_eq!(channels["a"].progress, SlackProgress::Reaction); + assert_eq!(channels["b"].progress, SlackProgress::Off); + // A definition written before this option existed keeps working and + // gets the default rather than an unset/failed parse. + assert_eq!(channels["c"].progress, SlackProgress::Placeholder); + } + #[test] fn service_put_preserves_channels_and_channel_crud_rotates_credentials() { let config = tempfile::tempdir().unwrap(); @@ -1761,6 +1823,47 @@ mod tests { assert!(list_channel_catalog(&services).unwrap()[0].has_credential); } + #[test] + fn editing_a_channel_keeps_the_progress_affordance_it_was_given() { + // The channel put path cannot express `progress` yet, so a save that + // touched anything else β€” rotating a token, editing an allowlist β€” + // would reset an operator's choice back to the default without ever + // saying so. + let config = tempfile::tempdir().unwrap(); + let services = config.path().join("services"); + std::fs::create_dir_all(&services).unwrap(); + std::fs::write( + services.join("chat.toml"), + "harness = \"codex\"\n\ + [channels.bot]\nkind = \"slack\"\nprogress = \"reaction\"\n\ + app_token = \"xapp-1\"\nbot_token = \"xoxb-1\"\n", + ) + .unwrap(); + + put_channel( + &services, + construct_protocol::ServiceChannelPutParams { + service_name: "chat".into(), + channel: construct_protocol::ServiceChannelPut { + id: "bot".into(), + kind: "slack".into(), + enabled: true, + port: None, + app_token: None, + bot_token: None, + allowed_workspaces: vec!["T9".into()], + allowed_channels: Vec::new(), + }, + rotate_secret: false, + }, + ) + .unwrap(); + + let stored = &load_definitions(&services).unwrap()["chat"].channels["bot"]; + assert_eq!(stored.progress, SlackProgress::Reaction); + assert_eq!(stored.allowed_workspaces, vec!["T9".to_string()]); + } + #[test] fn slack_credentials_are_persisted_but_never_returned_in_summaries() { let config = tempfile::tempdir().unwrap(); diff --git a/crates/daemon/src/service/ingress.rs b/crates/daemon/src/service/ingress.rs index fdf03536..c4bcc9a7 100644 --- a/crates/daemon/src/service/ingress.rs +++ b/crates/daemon/src/service/ingress.rs @@ -15,7 +15,7 @@ use serde::{Deserialize, Serialize}; use std::collections::{HashMap, HashSet, VecDeque}; use std::path::PathBuf; use std::sync::Arc; -use tokio::sync::Mutex; +use tokio::sync::{watch, Mutex}; use tokio_util::sync::CancellationToken; use uuid::Uuid; @@ -320,14 +320,20 @@ impl ServiceIngress { /// Wait for the final assistant answer belonging to one submitted turn. /// The transport is cancelled on configuration reload or daemon shutdown. + /// + /// `progress` carries what the turn is doing while it runs, so a channel + /// can tell its user rather than going silent for the whole turn. It is + /// advisory: nothing here waits on a reader, and a channel that does not + /// render progress simply ignores it. pub(super) async fn wait_for_final( &self, receipt: &IngressReceipt, cancel: &CancellationToken, + progress: &watch::Sender, ) -> Result { if let Some(delivery_id) = receipt.delivery_id.as_deref() { return self - .wait_for_explicit_reply(receipt, delivery_id, cancel) + .wait_for_explicit_reply(receipt, delivery_id, cancel, progress) .await; } let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(30 * 60); @@ -342,6 +348,7 @@ impl ServiceIngress { let Ok(detail) = self.shared.manager.detail(&receipt.session).await else { continue; }; + publish_progress(progress, &detail.events); let events = detail.events.get(receipt.event_cursor..).unwrap_or(&[]); let saw_user = events.iter().any(|event| { matches!( @@ -373,6 +380,7 @@ impl ServiceIngress { receipt: &IngressReceipt, delivery_id: &str, cancel: &CancellationToken, + progress: &watch::Sender, ) -> Result { let deadline = tokio::time::Instant::now() + PENDING_DELIVERY_TTL; loop { @@ -390,6 +398,7 @@ impl ServiceIngress { let Ok(detail) = self.shared.manager.detail(&receipt.session).await else { continue; }; + publish_progress(progress, &detail.events); let events = detail.events.get(receipt.event_cursor..).unwrap_or(&[]); if let Some(reply) = explicit_service_reply(events.iter().map(|event| &event.event), delivery_id) @@ -704,6 +713,40 @@ fn explicit_service_reply<'a>( None } +/// What a turn that has not answered yet is currently doing. +/// +/// This is deliberately coarse. It exists so a channel can distinguish "still +/// thinking" from "stopped, and only a human at the TUI can unstick it" β€” a +/// difference the person who is waiting cares about a great deal, and which is +/// invisible from outside otherwise. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub(super) enum IngressProgress { + #[default] + Working, + AwaitingApproval { + tool: String, + summary: String, + }, +} + +/// Publish the turn's current phase, if it changed. Only the waiting task +/// writes here, so a plain compare-then-send cannot race itself. +fn publish_progress( + progress: &watch::Sender, + events: &[construct_protocol::TimestampedEvent], +) { + let phase = match pending_approval(events) { + Some(pending) => IngressProgress::AwaitingApproval { + tool: pending.tool, + summary: pending.summary, + }, + None => IngressProgress::Working, + }; + if *progress.borrow() != phase { + let _ = progress.send(phase); + } +} + /// A tool call this session is stopped at, waiting for the operator. pub(super) struct PendingApproval { pub(super) call_id: String, @@ -893,6 +936,72 @@ mod tests { ); } + #[test] + fn progress_reports_the_approval_a_turn_is_stopped_at() { + let at = chrono::Utc::now(); + let stamped = |event| construct_protocol::TimestampedEvent { + at, + seq: 0, + event, + }; + let (progress, mut rx) = watch::channel(IngressProgress::default()); + + // A turn mid-tool is still just working β€” the tool call was answered. + publish_progress( + &progress, + &[ + stamped(SessionEvent::ToolApprovalRequest { + call_id: "c1".into(), + tool: "bash".into(), + args_summary: "cargo test".into(), + risk: construct_protocol::ToolRisk::Risky, + allow_auto_review: true, + }), + stamped(SessionEvent::ToolUse { + tool: "bash".into(), + args: serde_json::Value::Null, + call_id: Some("c1".into()), + }), + ], + ); + assert_eq!(*rx.borrow_and_update(), IngressProgress::Working); + assert!(!rx.has_changed().unwrap()); + + // A trailing request means nobody has answered it yet. + publish_progress( + &progress, + &[stamped(SessionEvent::ToolApprovalRequest { + call_id: "c2".into(), + tool: "bash".into(), + args_summary: "rm -rf build".into(), + risk: construct_protocol::ToolRisk::Risky, + allow_auto_review: true, + })], + ); + assert!(rx.has_changed().unwrap()); + assert_eq!( + *rx.borrow_and_update(), + IngressProgress::AwaitingApproval { + tool: "bash".into(), + summary: "rm -rf build".into(), + } + ); + + // Unchanged phase must not wake a reader; a channel re-renders on + // every notification, and this loop polls four times a second. + publish_progress( + &progress, + &[stamped(SessionEvent::ToolApprovalRequest { + call_id: "c2".into(), + tool: "bash".into(), + args_summary: "rm -rf build".into(), + risk: construct_protocol::ToolRisk::Risky, + allow_auto_review: true, + })], + ); + assert!(!rx.has_changed().unwrap()); + } + #[test] fn explicit_reply_requires_the_matching_delivery_tool_call() { let events = [ diff --git a/crates/daemon/src/service/slack.rs b/crates/daemon/src/service/slack.rs index 27bf190c..54a48e05 100644 --- a/crates/daemon/src/service/slack.rs +++ b/crates/daemon/src/service/slack.rs @@ -4,11 +4,13 @@ //! token for a short-lived WebSocket URL, acknowledges each envelope before //! doing work, then posts the completed answer with the bot token. -use super::ingress::{IngressRequest, ServiceIngress}; +use super::ingress::{IngressProgress, IngressRequest, ServiceIngress}; +use super::SlackProgress; use anyhow::{anyhow, Context, Result}; use futures::{SinkExt, StreamExt}; use serde::Deserialize; use std::sync::Arc; +use tokio::sync::watch; use tokio_tungstenite::tungstenite::Message; use tokio_util::sync::CancellationToken; @@ -20,6 +22,7 @@ pub(crate) struct SlackConfig { pub(super) bot_token: String, pub(super) allowed_workspaces: Vec, pub(super) allowed_channels: Vec, + pub(super) progress: SlackProgress, } #[derive(Clone)] @@ -46,6 +49,8 @@ struct SlackResponse { #[serde(default)] url: Option, #[serde(default)] + ts: Option, + #[serde(default)] error: Option, } @@ -76,41 +81,85 @@ impl SlackApi { .ok_or_else(|| anyhow!("Slack response omitted WebSocket URL")) } - async fn post_message( - &self, - token: &str, - channel: &str, - thread_ts: &str, - text: &str, - ) -> Result<()> { + /// One Slack Web API call, returning the `ts` of whatever it addressed. + async fn call(&self, token: &str, method: &str, body: serde_json::Value) -> Result> { let response = self .client - .post(format!("{}/chat.postMessage", self.base_url)) + .post(format!("{}/{method}", self.base_url)) .bearer_auth(token) - .json(&serde_json::json!({ - "channel": channel, - "thread_ts": thread_ts, - "text": text, - })) + .json(&body) .send() .await - .context("post Slack reply")? + .with_context(|| format!("call Slack {method}"))? .error_for_status() - .context("Slack reply HTTP response")? + .with_context(|| format!("Slack {method} HTTP response"))? .json::() .await - .context("decode Slack reply response")?; + .with_context(|| format!("decode Slack {method} response"))?; if response.ok { - Ok(()) + Ok(response.ts) } else { Err(anyhow!( - "Slack rejected reply: {}", + "Slack rejected {method}: {}", response .error .unwrap_or_else(|| "unknown error".to_string()) )) } } + + async fn post_message( + &self, + token: &str, + channel: &str, + thread_ts: &str, + text: &str, + ) -> Result> { + self.call( + token, + "chat.postMessage", + serde_json::json!({ + "channel": channel, + "thread_ts": thread_ts, + "text": text, + }), + ) + .await + } + + async fn update_message( + &self, + token: &str, + channel: &str, + ts: &str, + text: &str, + ) -> Result> { + self.call( + token, + "chat.update", + serde_json::json!({ "channel": channel, "ts": ts, "text": text }), + ) + .await + } + + /// Reactions need the `reactions:write` scope. An app that was installed + /// before the operator selected the progress affordance will not have it, + /// so callers treat a failure here as cosmetic and answer anyway. + async fn set_reaction( + &self, + token: &str, + method: &str, + channel: &str, + ts: &str, + name: &str, + ) -> Result> { + self.call( + token, + method, + serde_json::json!({ "channel": channel, "timestamp": ts, "name": name }), + ) + .await + } } pub(super) async fn serve( @@ -238,6 +287,112 @@ where .context("acknowledge Slack envelope") } +/// How long a turn may run before the channel admits it is still working. +/// +/// Most turns answer well inside this, and announcing those would put a +/// placeholder on every message for no benefit. The affordance is for the wait +/// that has already become long enough to look like a dropped request. +const PROGRESS_AFTER: std::time::Duration = std::time::Duration::from_secs(8); + +const WORKING_EMOJI: &str = "eyes"; +const ANSWERED_EMOJI: &str = "white_check_mark"; +const FAILED_EMOJI: &str = "warning"; + +fn progress_text(progress: &IngressProgress) -> String { + match progress { + IngressProgress::Working => "_Working on it…_".to_string(), + // Say who has to act. A turn stopped here will not move on its own, + // and the person waiting in Slack cannot see the approval prompt. + IngressProgress::AwaitingApproval { tool, summary } if summary.is_empty() => { + format!("_Waiting for an operator to approve `{tool}`._") + } + IngressProgress::AwaitingApproval { tool, summary } => { + format!("_Waiting for an operator to approve `{tool}`: {summary}_") + } + } +} + +/// What the affordance left behind in Slack, so the answer can replace it. +#[derive(Default)] +struct Affordance { + placeholder_ts: Option, + reacted: bool, +} + +/// Show, and keep current, the "still working" affordance for one delivery. +/// +/// Returns once cancelled β€” which the caller does as soon as the turn +/// resolves β€” handing back whatever it put in the channel. +async fn run_affordance( + api: SlackApi, + config: SlackConfig, + channel: String, + thread_ts: String, + message_ts: String, + after: std::time::Duration, + mut progress: watch::Receiver, + cancel: CancellationToken, +) -> Affordance { + let mut state = Affordance::default(); + if config.progress == SlackProgress::Off { + return state; + } + tokio::select! { + _ = cancel.cancelled() => return state, + _ = tokio::time::sleep(after) => {} + } + if config.progress.reacts() { + match api + .set_reaction( + &config.bot_token, + "reactions.add", + &channel, + &message_ts, + WORKING_EMOJI, + ) + .await + { + Ok(_) => state.reacted = true, + Err(error) => tracing::warn!( + %error, + "Slack progress reaction failed; the answer is unaffected \ + (does the app have the reactions:write scope?)" + ), + } + } + if config.progress.posts_placeholder() { + let text = progress_text(&progress.borrow_and_update().clone()); + match api + .post_message(&config.bot_token, &channel, &thread_ts, &text) + .await + { + Ok(ts) => state.placeholder_ts = ts, + Err(error) => tracing::warn!(%error, "Slack progress placeholder failed"), + } + } + // Keep the placeholder honest: a turn that stops at an approval must stop + // claiming it is working. + loop { + tokio::select! { + _ = cancel.cancelled() => return state, + changed = progress.changed() => { + if changed.is_err() { + return state; + } + let text = progress_text(&progress.borrow_and_update().clone()); + let Some(ts) = state.placeholder_ts.as_deref() else { + continue; + }; + if let Err(error) = + api.update_message(&config.bot_token, &channel, ts, &text).await + { + tracing::warn!(%error, "Slack progress update failed"); + } + } + } + } +} + async fn process_delivery( ingress: &ServiceIngress, config: &SlackConfig, @@ -247,23 +402,94 @@ async fn process_delivery( ) -> Result<()> { let receipt = ingress .submit_tracked(IngressRequest { - message: delivery.text, + message: delivery.text.clone(), session_key: Some(format!( "{}:{}:{}", delivery.team_id, delivery.channel, delivery.thread_ts )), - request_id: Some(delivery.event_id), + request_id: Some(delivery.event_id.clone()), }) .await?; - let reply = ingress.wait_for_final(&receipt, cancel).await?; - tokio::select! { - _ = cancel.cancelled() => Err(anyhow!("channel stopped")), - result = api.post_message( + let (progress_tx, progress_rx) = watch::channel(IngressProgress::default()); + let affordance_cancel = CancellationToken::new(); + let affordance = tokio::spawn(run_affordance( + api.clone(), + config.clone(), + delivery.channel.clone(), + delivery.thread_ts.clone(), + delivery.message_ts.clone(), + PROGRESS_AFTER, + progress_rx, + affordance_cancel.clone(), + )); + + let reply = ingress.wait_for_final(&receipt, cancel, &progress_tx).await; + affordance_cancel.cancel(); + let affordance = affordance.await.unwrap_or_default(); + + // A turn that failed used to leave the thread silent forever. Now that + // something in the channel says "working", it must not keep saying that. + let text = match &reply { + Ok(reply) => reply.clone(), + Err(error) => format!("_The turn ended without an answer: {error}_"), + }; + if cancel.is_cancelled() { + return Err(anyhow!("channel stopped")); + } + match affordance.placeholder_ts.as_deref() { + Some(ts) => { + api.update_message(&config.bot_token, &delivery.channel, ts, &text) + .await?; + } + None => { + api.post_message( + &config.bot_token, + &delivery.channel, + &delivery.thread_ts, + &text, + ) + .await?; + } + } + if affordance.reacted { + settle_reaction(api, config, &delivery, reply.is_ok()).await; + } + reply.map(|_| ()) +} + +/// Swap the working reaction for the outcome. Cosmetic: a workspace that +/// denies the scope mid-turn should not turn a delivered answer into a failure. +async fn settle_reaction( + api: &SlackApi, + config: &SlackConfig, + delivery: &SlackDelivery, + answered: bool, +) { + let _ = api + .set_reaction( + &config.bot_token, + "reactions.remove", + &delivery.channel, + &delivery.message_ts, + WORKING_EMOJI, + ) + .await; + let settled = if answered { + ANSWERED_EMOJI + } else { + FAILED_EMOJI + }; + if let Err(error) = api + .set_reaction( &config.bot_token, + "reactions.add", &delivery.channel, - &delivery.thread_ts, - &reply, - ) => result, + &delivery.message_ts, + settled, + ) + .await + { + tracing::warn!(%error, "Slack outcome reaction failed"); } } @@ -313,6 +539,9 @@ struct SlackDelivery { team_id: String, channel: String, thread_ts: String, + /// The message that triggered this turn. Distinct from `thread_ts` for a + /// reply inside a thread, and it is this one a reaction belongs on. + message_ts: String, text: String, } @@ -345,7 +574,8 @@ fn delivery_from_envelope(envelope: SocketEnvelope, config: &SlackConfig) -> Opt event_id: payload.event_id?, team_id, channel, - thread_ts: event.thread_ts.unwrap_or(ts), + thread_ts: event.thread_ts.unwrap_or_else(|| ts.clone()), + message_ts: ts, text, }) } @@ -372,6 +602,7 @@ mod tests { bot_token: "xoxb-test".into(), allowed_workspaces: vec!["T1".into()], allowed_channels: vec!["C1".into()], + progress: SlackProgress::default(), } } @@ -399,11 +630,286 @@ mod tests { team_id: "T1".into(), channel: "C1".into(), thread_ts: "123.45".into(), + message_ts: "123.45".into(), text: "deploy status".into(), }) ); } + #[test] + fn a_thread_reply_reacts_on_itself_not_on_the_thread_root() { + // thread_ts routes the conversation; message_ts is the message a + // reaction belongs on. Conflating them would stack every reaction of a + // long thread onto its first message. + let envelope = serde_json::from_value::(serde_json::json!({ + "payload": { + "team_id": "T1", "event_id": "Ev2", + "event": { + "type": "app_mention", "channel": "C1", "user": "U1", + "text": "<@UBOT> and now?", "ts": "222.22", "thread_ts": "111.11" + } + } + })) + .unwrap(); + let delivery = delivery_from_envelope(envelope, &config()).unwrap(); + assert_eq!(delivery.thread_ts, "111.11"); + assert_eq!(delivery.message_ts, "222.22"); + } + + #[test] + fn progress_says_who_has_to_act_when_a_turn_stops_at_an_approval() { + // "Working on it" is a lie once the turn is parked on an approval: + // nothing moves until a human at the TUI acts, and the person waiting + // in Slack cannot see that prompt. + assert_eq!( + progress_text(&IngressProgress::Working), + "_Working on it…_" + ); + assert_eq!( + progress_text(&IngressProgress::AwaitingApproval { + tool: "bash".into(), + summary: "cargo test".into(), + }), + "_Waiting for an operator to approve `bash`: cargo test_" + ); + assert_eq!( + progress_text(&IngressProgress::AwaitingApproval { + tool: "bash".into(), + summary: String::new(), + }), + "_Waiting for an operator to approve `bash`._" + ); + } + + #[test] + fn progress_modes_select_their_affordances() { + assert!(!SlackProgress::Off.posts_placeholder() && !SlackProgress::Off.reacts()); + assert!(SlackProgress::Placeholder.posts_placeholder()); + assert!(!SlackProgress::Placeholder.reacts()); + assert!(SlackProgress::Reaction.reacts()); + assert!(!SlackProgress::Reaction.posts_placeholder()); + assert!(SlackProgress::Both.posts_placeholder() && SlackProgress::Both.reacts()); + } + + #[tokio::test] + async fn a_turn_that_answers_quickly_leaves_no_progress_message() { + // The affordance is for a wait long enough to look dropped. Cancelling + // before the delay elapses β€” which is what a fast turn does β€” must + // leave the thread untouched, with no Slack call attempted at all. + let api = SlackApi { + client: reqwest::Client::new(), + // Any call would try to reach this and fail the test by erroring. + base_url: "http://127.0.0.1:1".to_string(), + }; + let (_tx, rx) = watch::channel(IngressProgress::default()); + let cancel = CancellationToken::new(); + cancel.cancel(); + + let state = run_affordance( + api, + config(), + "C1".into(), + "1.1".into(), + "1.1".into(), + PROGRESS_AFTER, + rx, + cancel, + ) + .await; + + assert!(state.placeholder_ts.is_none()); + assert!(!state.reacted); + } + + #[tokio::test] + async fn the_off_mode_never_touches_the_channel() { + let api = SlackApi { + client: reqwest::Client::new(), + base_url: "http://127.0.0.1:1".to_string(), + }; + let mut config = config(); + config.progress = SlackProgress::Off; + let (_tx, rx) = watch::channel(IngressProgress::default()); + + // Not cancelled: Off must return immediately on its own, without even + // waiting out the delay. + let state = run_affordance( + api, + config, + "C1".into(), + "1.1".into(), + "1.1".into(), + PROGRESS_AFTER, + rx, + CancellationToken::new(), + ) + .await; + + assert!(state.placeholder_ts.is_none()); + assert!(!state.reacted); + } + + /// Stub Slack Web API: answers every method `ok` and records the calls. + async fn stub_slack() -> (SlackApi, Arc>>) { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let calls = Arc::new(std::sync::Mutex::new(Vec::new())); + let recorder = calls.clone(); + tokio::spawn(async move { + loop { + let Ok((mut stream, _)) = listener.accept().await else { + return; + }; + let recorder = recorder.clone(); + tokio::spawn(async move { + let mut request = Vec::new(); + let mut chunk = [0_u8; 4096]; + loop { + let Ok(read) = stream.read(&mut chunk).await else { + return; + }; + if read == 0 { + return; + } + request.extend_from_slice(&chunk[..read]); + let Some(end) = request + .windows(4) + .position(|window| window == b"\r\n\r\n") + .map(|index| index + 4) + else { + continue; + }; + let head = String::from_utf8_lossy(&request[..end]).to_string(); + let length = head + .lines() + .find_map(|line| { + line.split_once(':') + .filter(|(name, _)| name.eq_ignore_ascii_case("content-length")) + .and_then(|(_, value)| value.trim().parse::().ok()) + }) + .unwrap_or(0); + if request.len() < end + length { + continue; + } + let method = head + .split_whitespace() + .nth(1) + .unwrap_or("/") + .trim_start_matches('/') + .to_string(); + let body: serde_json::Value = + serde_json::from_slice(&request[end..end + length]) + .unwrap_or(serde_json::Value::Null); + recorder.lock().unwrap().push((method, body)); + let payload = br#"{"ok":true,"ts":"P1"}"#; + let _ = stream + .write_all( + format!( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n", + payload.len() + ) + .as_bytes(), + ) + .await; + let _ = stream.write_all(payload).await; + return; + } + }); + } + }); + ( + SlackApi { + client: reqwest::Client::new(), + base_url: format!("http://{address}"), + }, + calls, + ) + } + + #[tokio::test] + async fn a_slow_turn_announces_itself_then_says_what_it_is_blocked_on() { + let (api, calls) = stub_slack().await; + let mut config = config(); + config.progress = SlackProgress::Both; + let (tx, rx) = watch::channel(IngressProgress::default()); + let cancel = CancellationToken::new(); + + let affordance = tokio::spawn(run_affordance( + api, + config, + "C1".into(), + "111.11".into(), + "222.22".into(), + std::time::Duration::from_millis(10), + rx, + cancel.clone(), + )); + + // Wait for the placeholder, then park the turn on an approval. + let posted = tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + if calls + .lock() + .unwrap() + .iter() + .any(|(method, _)| method == "chat.postMessage") + { + return; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }) + .await; + assert!(posted.is_ok(), "the placeholder was never posted"); + + tx.send(IngressProgress::AwaitingApproval { + tool: "bash".into(), + summary: "rm -rf build".into(), + }) + .unwrap(); + let updated = tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + if calls + .lock() + .unwrap() + .iter() + .any(|(method, _)| method == "chat.update") + { + return; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }) + .await; + assert!(updated.is_ok(), "the approval was never surfaced"); + + cancel.cancel(); + let state = affordance.await.unwrap(); + assert_eq!(state.placeholder_ts.as_deref(), Some("P1")); + assert!(state.reacted); + + let calls = calls.lock().unwrap(); + let by = |name: &str| { + calls + .iter() + .find(|(method, _)| method == name) + .map(|(_, body)| body.clone()) + .unwrap_or(serde_json::Value::Null) + }; + // The reaction goes on the triggering message, the placeholder into + // the thread β€” two different timestamps. + assert_eq!(by("reactions.add")["timestamp"], "222.22"); + assert_eq!(by("reactions.add")["name"], WORKING_EMOJI); + assert_eq!(by("chat.postMessage")["thread_ts"], "111.11"); + assert_eq!(by("chat.postMessage")["text"], "_Working on it…_"); + // And the placeholder stops claiming to be working once it isn't. + assert_eq!(by("chat.update")["ts"], "P1"); + assert_eq!( + by("chat.update")["text"], + "_Waiting for an operator to approve `bash`: rm -rf build_" + ); + } + #[test] fn direct_messages_are_accepted_but_bots_and_unlisted_channels_are_not() { let event = |channel: &str, bot: bool| { diff --git a/crates/daemon/src/service_supervisor.rs b/crates/daemon/src/service_supervisor.rs index 72092b77..c05cb961 100644 --- a/crates/daemon/src/service_supervisor.rs +++ b/crates/daemon/src/service_supervisor.rs @@ -763,6 +763,7 @@ mod tests { bot_token: None, allowed_workspaces: Vec::new(), allowed_channels: Vec::new(), + progress: Default::default(), }, ) }) @@ -814,6 +815,7 @@ mod tests { bot_token: Some("xoxb-secret".into()), allowed_workspaces: vec!["T1".into()], allowed_channels: vec!["C1".into()], + progress: Default::default(), }, )]), }; diff --git a/docs/services.md b/docs/services.md index d10f27a7..c0b0d29d 100644 --- a/docs/services.md +++ b/docs/services.md @@ -26,6 +26,44 @@ curl http://127.0.0.1:8787/svc/alerts/sessions/ \ HTTP channels start loopback-only. Enabling or attaching one never exposes it to the network. +## Slack channels + +A Slack channel connects over Socket Mode and needs no inbound port. It routes +each thread to its own session, so a conversation in Slack is a conversation in +Construct. + +```toml +[channels.my-bot] +kind = "slack" +enabled = true +app_token = "xapp-…" # Socket Mode +bot_token = "xoxb-…" # posting +progress = "placeholder" # off | placeholder | reaction | both +``` + +### Showing that a turn is still running + +A turn that takes a while would otherwise leave the thread silent, which is +indistinguishable from a delivery that was dropped. `progress` chooses what the +channel shows while it works. Nothing appears for a turn that answers promptly +β€” the affordance is only for a wait long enough to look like a failure. + +| value | behavior | +| --- | --- | +| `placeholder` (default) | Posts a message in the thread that later becomes the answer itself. | +| `reaction` | Reacts πŸ‘€ to the message that triggered the turn, then βœ… when it answers (⚠️ if it fails). | +| `both` | Both of the above. | +| `off` | Says nothing until the answer is ready. | + +If the turn stops at a tool approval, the affordance says so and names the +tool β€” that turn will not resume until an operator acts in the TUI, and the +person waiting in Slack cannot see that prompt. A turn that ends without an +answer now reports that in the thread instead of only in the daemon log. + +`reaction` and `both` call `reactions.add`, which needs the **`reactions:write`** +scope. If the app was installed without it, Construct logs the refusal and still +delivers the answer β€” reinstall the app to your workspace to enable it. + ## Publishing a channel Select an attached ingress channel. Its TUI action bar shows **Publish**; press diff --git a/specs/0178-a-channel-shows-that-a-turn-is-still-running.md b/specs/0178-a-channel-shows-that-a-turn-is-still-running.md new file mode 100644 index 00000000..a2449e05 --- /dev/null +++ b/specs/0178-a-channel-shows-that-a-turn-is-still-running.md @@ -0,0 +1,75 @@ +# 0178-a-channel-shows-that-a-turn-is-still-running + +Status: accepted +Date: 2026-08-02 +Area: ux +Scope: What a service channel tells the person waiting while a turn it accepted has not answered yet. + +## Decision + +A channel that accepts a delivery and then goes silent is indistinguishable +from one that dropped it. A channel may therefore show that a turn is still +running, and the operator chooses how visible that is per channel β€” including +turning it off. + +Three rules constrain it. + +**Silence is correct for a turn that answers promptly.** The affordance exists +for a wait that has already become long enough to look like a failure, so it +appears only after such a wait. A quick turn leaves no trace of one. + +**It must not claim to be working when it is not.** A turn stopped at a tool +approval is not making progress and will not resume until a human acts at +another surface entirely. Whatever the channel is showing has to change to say +so, naming what is being approved. + +**It must resolve.** Whatever the affordance put in the channel is replaced by +the outcome β€” the answer, or a statement that the turn ended without one. A +turn that fails must not leave "working on it" standing forever, which means a +failure is now reported to the channel rather than only to the daemon's log. + +The affordance is presentation, so it belongs to the channel. The channel does +not inspect sessions to build it: the ingress publishes what the turn is doing, +and publishing is advisory β€” nothing about a turn's progress or outcome depends +on anyone rendering it. + +## Reason + +The person waiting is not the operator. They cannot see the session, the +harness, or the daemon log, so from their side a slow turn and a lost one look +the same, and the reasonable response to both is to send the message again β€” +which starts a second turn in the same thread and makes things worse. + +The approval case is the sharpest version. The turn is stopped, and only a +human at the TUI can unstick it. Reporting that plainly turns an unexplained +silence into an action someone can take. + +Making it configurable acknowledges that channels differ: a busy shared channel +may want nothing, a quiet one may want the acknowledgement, and some +affordances cost permissions the operator's workspace may not have granted. + +## Consequences + +- A new affordance must degrade to silence, not to an error. Anything cosmetic + that the workspace refuses is logged and the answer still gets delivered. +- The channel is the only place that renders progress. New progress states are + added to what the ingress publishes; a channel that does not know a state + keeps working. +- Turn failures are now visible to the channel's users, not only in the log. +- The delay before showing anything is a product decision about what counts as + a long wait, not a tuning knob for load. + +## Non-Goals + +- Streaming a turn's partial output, tool calls, or reasoning to the channel. +- Making progress delivery reliable or ordered; it is a best-effort hint. +- Giving the channel a way to answer an approval. Approvals stay with the + operator. + +## Examples + +A Slack thread asks a question that takes two minutes. After a few seconds the +channel says it is working; the turn then stops at an approval and the same +message changes to name the tool awaiting sign-off; when the operator approves +and the turn finishes, that message becomes the answer. A question answered in +three seconds produces just the answer, with nothing before it.