Show that a Slack turn is still running

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<String>,
     #[serde(default, skip_serializing_if = "Vec::is_empty")]
     pub allowed_channels: Vec<String>,
+    /// 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();

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<IngressProgress>,
     ) -> Result<String> {
         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<IngressProgress>,
     ) -> Result<String> {
         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<IngressProgress>,
+    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 = [