fix(clippy): replace find().is_none() with !any() in user stats test

https://claude.ai/code/session_01Nm95eCjdrwxDjwkHTRieZs

Cargo.lock

@@ -3150,7 +3150,7 @@ dependencies = [
  "libc",
  "percent-encoding",
  "pin-project-lite",
- "socket2 0.5.10",
+ "socket2 0.6.3",
  "system-configuration",
  "tokio",
  "tower-service",
@@ -3439,6 +3439,7 @@ dependencies = [
  "pgvector",
  "postgres-types",
  "pretty_assertions",
+ "pty-process",
  "rand 0.8.5",
  "readabilityrs",
  "refinery",
@@ -3524,7 +3525,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46"
 dependencies = [
  "hermit-abi",
  "libc",
- "windows-sys 0.59.0",
+ "windows-sys 0.61.2",
 ]
 
 [[package]]
@@ -4906,6 +4907,16 @@ dependencies = [
  "syn 1.0.109",
 ]
 
+[[package]]
+name = "pty-process"
+version = "0.5.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "71cec9e2670207c5ebb9e477763c74436af3b9091dd550b9fb3c1bec7f3ea266"
+dependencies = [
+ "rustix 1.1.4",
+ "tokio",
+]
+
 [[package]]
 name = "pulley-interpreter"
 version = "28.0.1"
@@ -4930,7 +4941,7 @@ dependencies = [
  "quinn-udp",
  "rustc-hash 2.1.1",
  "rustls 0.23.37",
- "socket2 0.5.10",
+ "socket2 0.6.3",
  "thiserror 2.0.18",
  "tokio",
  "tracing",
@@ -4967,9 +4978,9 @@ dependencies = [
  "cfg_aliases",
  "libc",
  "once_cell",
- "socket2 0.5.10",
+ "socket2 0.6.3",
  "tracing",
- "windows-sys 0.59.0",
+ "windows-sys 0.60.2",
 ]
 
 [[package]]

Cargo.toml

@@ -189,6 +189,10 @@ json5 = { version = "0.4", optional = true }
 [target.'cfg(target_os = "macos")'.dependencies]
 security-framework = "3"
 
+# PTY allocation for Claude CLI stdout buffering fix (Unix only)
+[target.'cfg(unix)'.dependencies]
+pty-process = { version = "0.5", features = ["async"] }
+
 # Linux secret-service (GNOME Keyring, KWallet)
 [target.'cfg(target_os = "linux")'.dependencies]
 secret-service = { version = "4", features = ["rt-tokio-crypto-rust"] }

channels-src/discord/src/lib.rs

@@ -28,6 +28,9 @@ use std::{cmp::Ordering, collections::HashMap};
 use ed25519_dalek::{Signature, Verifier, VerifyingKey};
 use serde::{Deserialize, Serialize};
 
+/// Discord REST API v10 base URL.
+const DISCORD_API_BASE: &str = "https://discord.com/api/v10";
+
 use exports::near::agent::channel::{
     AgentResponse, ChannelConfig, Guest, HttpEndpointConfig, IncomingHttpRequest,
     OutgoingHttpResponse, PollConfig, StatusUpdate,
@@ -427,7 +430,7 @@ impl Guest for DiscordChannel {
             (
                 "PATCH",
                 format!(
-                    "https://discord.com/api/v10/webhooks/{}/{}/messages/@original",
+                    "{DISCORD_API_BASE}/webhooks/{}/{}/messages/@original",
                     application_id, token
                 ),
             )
@@ -438,20 +441,7 @@ impl Guest for DiscordChannel {
             payload["allowed_mentions"] = serde_json::json!({
                 "replied_user": true
             });
-            let mention_payload = serde_json::to_vec(&payload)
-                .map_err(|e| format!("Failed to serialize mention payload: {}", e))?;
-            let mention_url = format!(
-                "https://discord.com/api/v10/channels/{}/messages",
-                metadata.channel_id
-            );
-            let result = channel_host::http_request(
-                "POST",
-                &mention_url,
-                &discord_auth_headers_json(true),
-                Some(&mention_payload),
-                None,
-            );
-            return map_discord_response(result);
+            return send_channel_message(&metadata.channel_id, payload);
         } else {
             return Err("Unsupported Discord response metadata".to_string());
         };
@@ -469,8 +459,8 @@ impl Guest for DiscordChannel {
 
     fn on_status(_update: StatusUpdate) {}
 
-    fn on_broadcast(_user_id: String, _response: AgentResponse) -> Result<(), String> {
-        Err("broadcast not yet implemented for Discord channel".to_string())
+    fn on_broadcast(user_id: String, response: AgentResponse) -> Result<(), String> {
+        broadcast_dm(&user_id, &response.content)
     }
 
     fn on_shutdown() {
@@ -501,6 +491,21 @@ fn map_discord_response(
     }
 }
 
+/// Post a JSON payload to a Discord channel as a new message.
+fn send_channel_message(channel_id: &str, payload: serde_json::Value) -> Result<(), String> {
+    let payload_bytes = serde_json::to_vec(&payload)
+        .map_err(|e| format!("Failed to serialize message: {}", e))?;
+    let url = format!("{DISCORD_API_BASE}/channels/{}/messages", channel_id);
+    let result = channel_host::http_request(
+        "POST",
+        &url,
+        &discord_auth_headers_json(true),
+        Some(&payload_bytes),
+        None,
+    );
+    map_discord_response(result)
+}
+
 fn load_runtime_config() -> DiscordRuntimeConfig {
     channel_host::workspace_read("config.json")
         .and_then(|raw| serde_json::from_str::<DiscordRuntimeConfig>(&raw).ok())
@@ -539,7 +544,7 @@ fn get_or_fetch_bot_id() -> Option<String> {
 
     let response = channel_host::http_request(
         "GET",
-        "https://discord.com/api/v10/users/@me",
+        &format!("{DISCORD_API_BASE}/users/@me"),
         &discord_auth_headers_json(false),
         None,
         Some(10_000),
@@ -659,7 +664,7 @@ fn poll_channel_mentions(channel_id: &str, bot_id: &str) {
 
 fn fetch_latest_message_id(channel_id: &str) -> Option<String> {
     let url = format!(
-        "https://discord.com/api/v10/channels/{}/messages?limit=1",
+        "{DISCORD_API_BASE}/channels/{}/messages?limit=1",
         channel_id
     );
     let response = channel_host::http_request(
@@ -697,7 +702,7 @@ fn fetch_messages_after_cursor(
 
     for page in 0..MAX_PAGES {
         let url = format!(
-            "https://discord.com/api/v10/channels/{}/messages?limit={}&after={}",
+            "{DISCORD_API_BASE}/channels/{}/messages?limit={}&after={}",
             channel_id, PAGE_LIMIT, after
         );
         let response = match channel_host::http_request(
@@ -986,7 +991,7 @@ fn handle_slash_command(interaction: &DiscordInteraction) -> bool {
             );
             // Attempt to notify user of internal error
             let url = format!(
-                "https://discord.com/api/v10/webhooks/{}/{}",
+                "{DISCORD_API_BASE}/webhooks/{}/{}",
                 interaction.application_id, interaction.token
             );
             let payload = serde_json::json!({
@@ -1106,7 +1111,7 @@ fn check_sender_permission(
     }
 
     let dm_policy =
-        channel_host::workspace_read(DM_POLICY_PATH).unwrap_or_else(|| default_dm_policy());
+        channel_host::workspace_read(DM_POLICY_PATH).unwrap_or_else(default_dm_policy);
     if dm_policy == "open" {
         return true;
     }
@@ -1161,7 +1166,7 @@ fn check_sender_permission(
 /// Send a pairing code as an ephemeral Discord followup message.
 fn send_pairing_reply(ctx: &PairingReplyCtx, code: &str) -> Result<(), String> {
     let url = format!(
-        "https://discord.com/api/v10/webhooks/{}/{}",
+        "{DISCORD_API_BASE}/webhooks/{}/{}",
         ctx.application_id, ctx.token
     );
     let payload = serde_json::json!({
@@ -1194,6 +1199,57 @@ fn send_pairing_reply(ctx: &PairingReplyCtx, code: &str) -> Result<(), String> {
     }
 }
 
+/// Send a broadcast message to a Discord user via DM.
+///
+/// Creates a DM channel with the user (Discord caches this, so repeated calls
+/// for the same user reuse the existing channel) and then posts the message.
+fn broadcast_dm(user_id: &str, content: &str) -> Result<(), String> {
+    // Validate user_id is a plausible Discord snowflake (numeric, 17-20 digits)
+    // to avoid injecting arbitrary strings into API URLs.
+    if user_id.is_empty()
+        || !user_id.chars().all(|c| c.is_ascii_digit())
+        || user_id.len() < 17
+        || user_id.len() > 20
+    {
+        return Err(format!("Invalid Discord user ID: '{}'", user_id));
+    }
+
+    // Step 1: Open (or reuse) a DM channel with the target user.
+    let create_dm_payload = serde_json::json!({ "recipient_id": user_id });
+    let create_dm_bytes = serde_json::to_vec(&create_dm_payload)
+        .map_err(|e| format!("Failed to serialize DM channel request: {}", e))?;
+
+    let dm_response = channel_host::http_request(
+        "POST",
+        &format!("{DISCORD_API_BASE}/users/@me/channels"),
+        &discord_auth_headers_json(true),
+        Some(&create_dm_bytes),
+        Some(10_000),
+    )
+    .map_err(|e| format!("Failed to create DM channel: {}", e))?;
+
+    if !(200..300).contains(&dm_response.status) {
+        let body = String::from_utf8_lossy(&dm_response.body);
+        return Err(format!(
+            "Discord create-DM failed: {} - {}",
+            dm_response.status, body
+        ));
+    }
+
+    #[derive(Deserialize)]
+    struct DmChannelResponse {
+        id: String,
+    }
+    let dm_channel: DmChannelResponse = serde_json::from_slice(&dm_response.body)
+        .map_err(|e| format!("Failed to parse DM channel response: {}", e))?;
+    let channel_id = &dm_channel.id;
+
+    // Step 2: Send the message to the DM channel.
+    let truncated = truncate_message(content);
+    let payload = serde_json::json!({ "content": truncated });
+    send_channel_message(channel_id, payload)
+}
+
 fn json_response(status: u16, value: serde_json::Value) -> OutgoingHttpResponse {
     let body = serde_json::to_vec(&value).unwrap_or_default();
     let headers = serde_json::json!({"Content-Type": "application/json"});
@@ -1593,4 +1649,43 @@ mod tests {
         assert_eq!(interaction.interaction_type, 2);
         assert!(interaction.data.is_some());
     }
+
+    #[test]
+    fn test_broadcast_dm_payload_format() {
+        // Verify the DM channel creation payload is well-formed JSON that
+        // Discord's API expects.
+        let user_id = "123456789012345678";
+        let payload = serde_json::json!({ "recipient_id": user_id });
+        let serialized = serde_json::to_vec(&payload).unwrap();
+        let parsed: serde_json::Value = serde_json::from_slice(&serialized).unwrap();
+        assert_eq!(
+            parsed.get("recipient_id").and_then(|v| v.as_str()),
+            Some(user_id)
+        );
+    }
+
+    #[test]
+    fn test_broadcast_message_truncation() {
+        // Broadcast uses truncate_message, verify it handles content within
+        // Discord's 2000-char limit for DMs.
+        let short = "Hello from broadcast";
+        assert_eq!(truncate_message(short), short);
+
+        let long = "x".repeat(2500);
+        let result = truncate_message(&long);
+        assert!(result.len() <= 2006); // 1990 content + 16 suffix
+        assert!(result.ends_with("\n... (truncated)"));
+    }
+
+    #[test]
+    fn test_broadcast_dm_validates_snowflake() {
+        // broadcast_dm rejects invalid Discord snowflake IDs before making
+        // any API calls. We can call it directly since invalid IDs are
+        // rejected before any host function is invoked.
+        assert!(broadcast_dm("", "hi").is_err());
+        assert!(broadcast_dm("abc", "hi").is_err());
+        assert!(broadcast_dm("12345", "hi").is_err()); // too short
+        assert!(broadcast_dm("123456789012345678901", "hi").is_err()); // too long
+        assert!(broadcast_dm("12345678901234567x", "hi").is_err()); // non-digit
+    }
 }

src/agent/routine_engine.rs

@@ -1089,48 +1089,134 @@ async fn execute_routine(ctx: EngineContext, routine: Routine, run: RoutineRun)
     // Increment running count (atomic: survives panics in the execution below)
     ctx.running_count.fetch_add(1, Ordering::Relaxed);
 
-    let result = match &routine.action {
-        RoutineAction::Lightweight {
-            prompt,
-            context_paths,
-            max_tokens,
-            use_tools,
-            max_tool_rounds,
-        } => {
-            execute_lightweight(
-                &ctx,
-                &routine,
-                prompt,
-                context_paths,
-                *max_tokens,
-                *use_tools,
-                *max_tool_rounds,
-            )
-            .await
+    // Retry constants for transient lightweight execution failures.
+    const MAX_RETRIES: u32 = 3;
+    const BASE_DELAY_MS: u64 = 1000;
+
+    let is_lightweight = matches!(routine.action, RoutineAction::Lightweight { .. });
+
+    // The retry block returns both the execution result and any accumulated
+    // token count so that usage is preserved even on final failure.
+    let (result, accumulated_tokens) = {
+        let mut attempt = 0u32;
+        // Track accumulated tokens as Option to preserve None semantics:
+        // None = no attempt reported tokens; Some(n) = at least one attempt did.
+        let mut accumulated_tokens: Option<i32> = None;
+        let uses_tools = matches!(
+            routine.action,
+            RoutineAction::Lightweight {
+                use_tools: true,
+                ..
+            }
+        ) && ctx.config.lightweight_tools_enabled;
+
+        /// Extract partial_tokens from any RoutineError variant that carries them.
+        fn extract_partial_tokens(e: &RoutineError) -> Option<i32> {
+            match e {
+                RoutineError::LlmFailed {
+                    partial_tokens: Some(t),
+                    ..
+                }
+                | RoutineError::EmptyResponse {
+                    partial_tokens: Some(t),
+                }
+                | RoutineError::TruncatedResponse {
+                    partial_tokens: Some(t),
+                } => Some(*t),
+                _ => None,
+            }
         }
-        RoutineAction::FullJob {
-            title,
-            description,
-            max_iterations,
-        } => {
-            let execution = FullJobExecutionConfig {
-                title,
-                description,
-                max_iterations: *max_iterations,
+
+        /// Merge an optional partial token count into the accumulator,
+        /// only materializing Some when at least one source had Some.
+        fn accumulate(acc: Option<i32>, partial: Option<i32>) -> Option<i32> {
+            match (acc, partial) {
+                (Some(a), Some(p)) => Some(a.saturating_add(p)),
+                (Some(a), None) => Some(a),
+                (None, p) => p,
+            }
+        }
+
+        loop {
+            let execution_result = match &routine.action {
+                RoutineAction::Lightweight {
+                    prompt,
+                    context_paths,
+                    max_tokens,
+                    use_tools,
+                    max_tool_rounds,
+                } => {
+                    execute_lightweight(
+                        &ctx,
+                        &routine,
+                        prompt,
+                        context_paths,
+                        *max_tokens,
+                        *use_tools,
+                        *max_tool_rounds,
+                    )
+                    .await
+                }
+                RoutineAction::FullJob {
+                    title,
+                    description,
+                    max_iterations,
+                } => {
+                    let execution = FullJobExecutionConfig {
+                        title,
+                        description,
+                        max_iterations: *max_iterations,
+                    };
+                    execute_full_job(&ctx, &routine, &run, &execution).await
+                }
             };
-            execute_full_job(&ctx, &routine, &run, &execution).await
+
+            match execution_result {
+                Ok((status, summary, tokens)) => {
+                    // Merge tokens: only produce Some when at least one source had Some.
+                    let total = accumulate(accumulated_tokens, tokens);
+                    break (Ok((status, summary, total)), accumulated_tokens);
+                }
+                Err(ref e)
+                    if is_lightweight
+                        && !uses_tools
+                        && e.is_retryable()
+                        // Skip outer retry for LlmFailed — RetryProvider already
+                        // retries transient LLM errors with its own budget. Retrying
+                        // here would create a multiplicative retry count.
+                        && !matches!(e, RoutineError::LlmFailed { .. })
+                        && attempt < MAX_RETRIES =>
+                {
+                    // Accumulate partial tokens from the failed attempt.
+                    accumulated_tokens = accumulate(accumulated_tokens, extract_partial_tokens(e));
+
+                    attempt += 1;
+
+                    let delay = Duration::from_millis(
+                        BASE_DELAY_MS.saturating_mul(2u64.saturating_pow(attempt - 1)),
+                    );
+                    tracing::event!(target: "transient_routine_errors", tracing::Level::WARN, routine = %routine.name, attempt = attempt, max_retries = MAX_RETRIES, delay_ms = delay.as_millis() as u64, "Transient routine error, retrying: {}", e);
+                    tokio::time::sleep(delay).await;
+                }
+                Err(e) => {
+                    // Accumulate tokens from the final failed attempt.
+                    accumulated_tokens = accumulate(accumulated_tokens, extract_partial_tokens(&e));
+                    break (Err(e), accumulated_tokens);
+                }
+            }
         }
     };
 
     // Decrement running count
     ctx.running_count.fetch_sub(1, Ordering::Relaxed);
 
-    // Process result
+    // Process result — on failure, preserve accumulated token total from
+    // earlier retry attempts so usage reporting stays accurate.
     let (status, summary, tokens) = match result {
         Ok(execution) => execution,
         Err(e) => {
             tracing::error!(routine = %routine.name, "Execution failed: {}", e);
-            (RunStatus::Failed, Some(e.to_string()), None)
+            (RunStatus::Failed, Some(e.to_string()), accumulated_tokens)
         }
     };
 
@@ -1511,13 +1597,14 @@ async fn execute_lightweight_no_tools(
         .with_max_tokens(effective_max_tokens)
         .with_temperature(0.3);
 
-    let response = ctx
-        .llm
-        .complete(request)
-        .await
-        .map_err(|e| RoutineError::LlmFailed {
+    let response = ctx.llm.complete(request).await.map_err(|e| {
+        let retryable = crate::llm::retry::is_retryable(&e);
+        RoutineError::LlmFailed {
             reason: e.to_string(),
-        })?;
+            partial_tokens: None,
+            retryable,
+        }
+    })?;
 
     handle_text_response(
         &response.content,
@@ -1538,12 +1625,18 @@ fn handle_text_response(
 ) -> Result<(RunStatus, Option<String>, Option<i32>), RoutineError> {
     let content = content.trim();
 
-    // Empty content guard
+    // Empty content guard — carry consumed tokens so the retry loop can
+    // accumulate them even when the response shape is invalid.
     if content.is_empty() {
+        let consumed = Some((total_input_tokens + total_output_tokens) as i32);
         return if finish_reason == FinishReason::Length {
-            Err(RoutineError::TruncatedResponse)
+            Err(RoutineError::TruncatedResponse {
+                partial_tokens: consumed,
+            })
         } else {
-            Err(RoutineError::EmptyResponse)
+            Err(RoutineError::EmptyResponse {
+                partial_tokens: consumed,
+            })
         };
     }
 
@@ -1621,13 +1714,15 @@ async fn execute_lightweight_with_tools(
                 .with_max_tokens(effective_max_tokens)
                 .with_temperature(0.3);
 
-            let response =
-                ctx.llm
-                    .complete(request)
-                    .await
-                    .map_err(|e| RoutineError::LlmFailed {
-                        reason: e.to_string(),
-                    })?;
+            let response = ctx.llm.complete(request).await.map_err(|e| {
+                let partial = (total_input_tokens + total_output_tokens) as i32;
+                let retryable = crate::llm::retry::is_retryable(&e);
+                RoutineError::LlmFailed {
+                    reason: e.to_string(),
+                    partial_tokens: if partial > 0 { Some(partial) } else { None },
+                    retryable,
+                }
+            })?;
 
             total_input_tokens += response.input_tokens;
             total_output_tokens += response.output_tokens;
@@ -1654,8 +1749,12 @@ async fn execute_lightweight_with_tools(
                 .with_temperature(0.3);
 
             let response = ctx.llm.complete_with_tools(request).await.map_err(|e| {
+                let partial = (total_input_tokens + total_output_tokens) as i32;
+                let retryable = crate::llm::retry::is_retryable(&e);
                 RoutineError::LlmFailed {
                     reason: e.to_string(),
+                    partial_tokens: if partial > 0 { Some(partial) } else { None },
+                    retryable,
                 }
             })?;
 
@@ -1828,6 +1927,7 @@ async fn execute_routine_tool(
 }
 
 /// Send a notification based on the routine's notify config and run status.
+#[allow(clippy::too_many_arguments)]
 async fn send_notification(
     tx: &mpsc::Sender<OutgoingResponse>,
     notify: &NotifyConfig,
@@ -2513,6 +2613,104 @@ mod tests {
         }
     }
 
+    /// Regression test for #1320: transient errors are retried for lightweight
+    /// routines but not for full-job routines or hard failures.
+    #[test]
+    fn test_retry_classification_for_routine_errors() {
+        use crate::error::RoutineError;
+
+        // Transient errors (retryable for lightweight routines)
+        let transient_errors: Vec<RoutineError> = vec![
+            RoutineError::LlmFailed {
+                reason: "rate limit".into(),
+                partial_tokens: None,
+                retryable: true,
+            },
+            RoutineError::LlmFailed {
+                reason: "network timeout".into(),
+                partial_tokens: Some(42),
+                retryable: true,
+            },
+            RoutineError::EmptyResponse {
+                partial_tokens: None,
+            },
+            RoutineError::TruncatedResponse {
+                partial_tokens: Some(100),
+            },
+        ];
+        for err in &transient_errors {
+            assert!(err.is_retryable(), "{} should be retryable", err);
+        }
+
+        // Permanent LLM failures that should NOT be retried
+        // (retryable: false is set at conversion time by llm::retry::is_retryable)
+        let permanent_llm_errors: Vec<RoutineError> = vec![
+            RoutineError::LlmFailed {
+                reason: "Authentication failed for provider openai".into(),
+                partial_tokens: None,
+                retryable: false,
+            },
+            RoutineError::LlmFailed {
+                reason: "invalid_api_key: bad key".into(),
+                partial_tokens: None,
+                retryable: false,
+            },
+            RoutineError::LlmFailed {
+                reason: "content policy violation".into(),
+                partial_tokens: None,
+                retryable: false,
+            },
+            RoutineError::LlmFailed {
+                reason: "content_filter triggered".into(),
+                partial_tokens: None,
+                retryable: false,
+            },
+            RoutineError::LlmFailed {
+                reason: "context length exceeded: 150000 tokens used, 128000 allowed".into(),
+                partial_tokens: Some(100),
+                retryable: false,
+            },
+            RoutineError::LlmFailed {
+                reason: "model not available on provider anthropic".into(),
+                partial_tokens: None,
+                retryable: false,
+            },
+            RoutineError::LlmFailed {
+                reason: "content moderation flagged".into(),
+                partial_tokens: None,
+                retryable: false,
+            },
+        ];
+        for err in &permanent_llm_errors {
+            assert!(!err.is_retryable(), "{} should NOT be retryable", err);
+        }
+
+        // Hard failures (never retried)
+        let hard_errors: Vec<RoutineError> = vec![
+            RoutineError::Disabled {
+                name: "test".into(),
+            },
+            RoutineError::NotFound {
+                id: uuid::Uuid::new_v4(),
+            },
+            RoutineError::NotAuthorized {
+                id: uuid::Uuid::new_v4(),
+            },
+            RoutineError::MaxConcurrent {
+                name: "test".into(),
+            },
+            RoutineError::JobDispatchFailed {
+                reason: "no docker".into(),
+            },
+            RoutineError::Database {
+                reason: "connection refused".into(),
+            },
+        ];
+        for err in &hard_errors {
+            assert!(!err.is_retryable(), "{} should NOT be retryable", err);
+        }
+    }
+
     #[test]
     fn test_sanitize_summary_strips_control_chars() {
         use super::sanitize_summary;

src/db/libsql/users.rs

@@ -981,7 +981,7 @@ mod tests {
         assert!(alice_stats.last_active_at.is_some());
 
         // Bob has no LLM calls so doesn't appear in summary stats
-        assert!(stats.iter().find(|s| s.user_id == "bob").is_none());
+        assert!(!stats.iter().any(|s| s.user_id == "bob"));
 
         // Filter to single user
         let alice_only = db.user_summary_stats(Some("alice")).await.unwrap();

src/error.rs

@@ -395,16 +395,49 @@ pub enum RoutineError {
     Database { reason: String },
 
     #[error("LLM call failed: {reason}")]
-    LlmFailed { reason: String },
+    LlmFailed {
+        reason: String,
+        /// Partial token count consumed before the failure (if any).
+        /// Used to accumulate usage across retry attempts.
+        partial_tokens: Option<i32>,
+        /// Whether the underlying LLM error was classified as retryable.
+        /// Set at the `LlmError` → `RoutineError` conversion site using
+        /// `crate::llm::retry::is_retryable()`, avoiding fragile substring
+        /// matching on the stringified reason.
+        retryable: bool,
+    },
 
     #[error("Failed to dispatch full job: {reason}")]
     JobDispatchFailed { reason: String },
 
     #[error("LLM returned empty content")]
-    EmptyResponse,
+    EmptyResponse {
+        /// Tokens consumed by the call that produced the empty response.
+        partial_tokens: Option<i32>,
+    },
 
     #[error("LLM response truncated (finish_reason=length) with no content")]
-    TruncatedResponse,
+    TruncatedResponse {
+        /// Tokens consumed by the call that produced the truncated response.
+        partial_tokens: Option<i32>,
+    },
+}
+
+impl RoutineError {
+    /// Whether this error is transient and worth retrying with backoff.
+    ///
+    /// Retryable: LLM failures where the underlying `LlmError` was classified
+    /// as retryable by `crate::llm::retry::is_retryable()`, empty responses,
+    /// and truncated responses.
+    /// Non-retryable: configuration errors, authorization, resource limits,
+    /// DB errors, and LLM failures caused by auth/content-policy/context-length.
+    pub fn is_retryable(&self) -> bool {
+        match self {
+            RoutineError::LlmFailed { retryable, .. } => *retryable,
+            RoutineError::EmptyResponse { .. } | RoutineError::TruncatedResponse { .. } => true,
+            _ => false,
+        }
+    }
 }
 
 /// Result type alias for the agent.
@@ -514,6 +547,99 @@ mod tests {
         assert!(msg.contains("bad format"), "Should mention reason: {msg}");
     }
 
+    #[test]
+    fn routine_error_retryable_classification() {
+        // Transient errors should be retryable
+        assert!(
+            RoutineError::LlmFailed {
+                reason: "timeout".into(),
+                partial_tokens: None,
+                retryable: true,
+            }
+            .is_retryable()
+        );
+        // Non-retryable LLM error
+        assert!(
+            !RoutineError::LlmFailed {
+                reason: "timeout".into(),
+                partial_tokens: None,
+                retryable: false,
+            }
+            .is_retryable()
+        );
+        assert!(
+            RoutineError::EmptyResponse {
+                partial_tokens: None
+            }
+            .is_retryable()
+        );
+        assert!(
+            RoutineError::TruncatedResponse {
+                partial_tokens: None
+            }
+            .is_retryable()
+        );
+
+        // Hard failures should NOT be retryable
+        assert!(
+            !RoutineError::Disabled {
+                name: "test".into()
+            }
+            .is_retryable()
+        );
+        assert!(
+            !RoutineError::JobDispatchFailed {
+                reason: "no docker".into()
+            }
+            .is_retryable()
+        );
+        assert!(
+            !RoutineError::Database {
+                reason: "conn refused".into()
+            }
+            .is_retryable()
+        );
+        assert!(!RoutineError::NotFound { id: Uuid::new_v4() }.is_retryable());
+        assert!(!RoutineError::NotAuthorized { id: Uuid::new_v4() }.is_retryable());
+        assert!(
+            !RoutineError::MaxConcurrent {
+                name: "test".into()
+            }
+            .is_retryable()
+        );
+        assert!(
+            !RoutineError::UnknownTriggerType {
+                trigger_type: "x".into()
+            }
+            .is_retryable()
+        );
+        assert!(
+            !RoutineError::UnknownActionType {
+                action_type: "x".into()
+            }
+            .is_retryable()
+        );
+        assert!(
+            !RoutineError::MissingField {
+                context: "c".into(),
+                field: "f".into()
+            }
+            .is_retryable()
+        );
+        assert!(
+            !RoutineError::InvalidCron {
+                reason: "bad".into()
+            }
+            .is_retryable()
+        );
+        assert!(
+            !RoutineError::UnknownRunStatus {
+                status: "bad".into()
+            }
+            .is_retryable()
+        );
+    }
+
     #[test]
     fn top_level_error_from_conversions() {
         let config_err = ConfigError::MissingEnvVar("TEST".to_string());

src/llm/circuit_breaker.rs

@@ -234,6 +234,7 @@ fn is_transient(err: &LlmError) -> bool {
         LlmError::RequestFailed { .. }
             | LlmError::RateLimited { .. }
             | LlmError::InvalidResponse { .. }
+            | LlmError::EmptyResponse { .. }
             | LlmError::SessionExpired { .. }
             | LlmError::SessionRenewalFailed { .. }
             | LlmError::Http(_)

src/llm/error.rs

@@ -17,6 +17,9 @@ pub enum LlmError {
     #[error("Invalid response from {provider}: {reason}")]
     InvalidResponse { provider: String, reason: String },
 
+    #[error("Empty response from {provider}: no content returned")]
+    EmptyResponse { provider: String },
+
     #[error("Context length exceeded: {used} tokens used, {limit} allowed")]
     ContextLengthExceeded { used: usize, limit: usize },
 

src/llm/github_copilot.rs

@@ -231,9 +231,8 @@ impl LlmProvider for GithubCopilotProvider {
                 .choices
                 .into_iter()
                 .next()
-                .ok_or_else(|| LlmError::InvalidResponse {
+                .ok_or_else(|| LlmError::EmptyResponse {
                     provider: "github_copilot".to_string(),
-                    reason: "No choices in response".to_string(),
                 })?;
 
         let (content, _tool_calls) = extract_choice_content(&choice);
@@ -309,9 +308,8 @@ impl LlmProvider for GithubCopilotProvider {
                 .choices
                 .into_iter()
                 .next()
-                .ok_or_else(|| LlmError::InvalidResponse {
+                .ok_or_else(|| LlmError::EmptyResponse {
                     provider: "github_copilot".to_string(),
-                    reason: "No choices in response".to_string(),
                 })?;
 
         let (content, tool_calls) = extract_choice_content(&choice);

src/llm/nearai_chat.rs

@@ -490,9 +490,8 @@ impl LlmProvider for NearAiChatProvider {
                 .choices
                 .into_iter()
                 .next()
-                .ok_or_else(|| LlmError::InvalidResponse {
+                .ok_or_else(|| LlmError::EmptyResponse {
                     provider: "nearai_chat".to_string(),
-                    reason: "No choices in response".to_string(),
                 })?;
 
         // Fall back to reasoning_content when content is null (same as
@@ -570,9 +569,8 @@ impl LlmProvider for NearAiChatProvider {
                 .choices
                 .into_iter()
                 .next()
-                .ok_or_else(|| LlmError::InvalidResponse {
+                .ok_or_else(|| LlmError::EmptyResponse {
                     provider: "nearai_chat".to_string(),
-                    reason: "No choices in response".to_string(),
                 })?;
 
         let tool_calls: Vec<ToolCall> = choice

src/llm/retry.rs

@@ -48,6 +48,7 @@ pub(crate) fn is_retryable(err: &LlmError) -> bool {
         LlmError::RequestFailed { .. }
             | LlmError::RateLimited { .. }
             | LlmError::InvalidResponse { .. }
+            | LlmError::EmptyResponse { .. }
             | LlmError::SessionRenewalFailed { .. }
             | LlmError::Http(_)
             | LlmError::Io(_)

src/worker/claude_bridge.rs

@@ -31,6 +31,7 @@ use std::time::Duration;
 
 use serde::{Deserialize, Serialize};
 use tokio::io::{AsyncBufReadExt, BufReader};
+#[cfg(not(unix))]
 use tokio::process::Command;
 use uuid::Uuid;
 
@@ -340,6 +341,11 @@ impl ClaudeBridgeRuntime {
 
     /// Spawn a `claude` CLI process and stream its output.
     ///
+    /// Uses a PTY on Unix so Node.js line-buffers stdout instead of
+    /// full-buffering (which causes the bridge to hang on non-TTY pipes).
+    /// Arguments are passed via `execve` (no shell) — injection-safe by
+    /// construction.
+    ///
     /// Returns the session_id if captured from the `system` init message.
     async fn run_claude_session(
         &self,
@@ -347,47 +353,102 @@ impl ClaudeBridgeRuntime {
         resume_session_id: Option<&str>,
         extra_env: &std::collections::HashMap<String, String>,
     ) -> Result<Option<String>, WorkerError> {
-        let mut cmd = Command::new("claude");
-        cmd.arg("-p")
-            .arg(prompt)
-            .arg("--output-format")
-            .arg("stream-json")
-            .arg("--verbose")
-            .arg("--max-turns")
-            .arg(self.config.max_turns.to_string())
-            .arg("--model")
-            .arg(&self.config.model);
+        let max_turns_str = self.config.max_turns.to_string();
 
-        if let Some(sid) = resume_session_id {
-            cmd.arg("--resume").arg(sid);
-        }
-
-        // Inject credentials into the child process environment without
-        // mutating the global process env (which is unsafe in multi-threaded programs).
-        cmd.envs(extra_env);
-
-        cmd.current_dir("/workspace")
-            .stdout(std::process::Stdio::piped())
-            .stderr(std::process::Stdio::piped());
-
-        let mut child = cmd.spawn().map_err(|e| WorkerError::ExecutionFailed {
-            reason: format!("failed to spawn claude: {}", e),
-        })?;
-
-        let stdout = child
-            .stdout
-            .take()
-            .ok_or_else(|| WorkerError::ExecutionFailed {
-                reason: "failed to capture claude stdout".to_string(),
+        // Spawn with PTY on Unix to fix Node.js stdout buffering.
+        // All arguments are passed individually via execve — never through
+        // a shell interpreter. This eliminates shell injection by construction.
+        #[cfg(unix)]
+        let (mut child, stdout, stderr) = {
+            let (pty, pts) = pty_process::open().map_err(|e| WorkerError::ExecutionFailed {
+                reason: format!("failed to allocate PTY: {}", e),
             })?;
 
-        let stderr = child
-            .stderr
-            .take()
-            .ok_or_else(|| WorkerError::ExecutionFailed {
-                reason: "failed to capture claude stderr".to_string(),
+            let mut cmd = pty_process::Command::new("claude");
+            cmd = cmd
+                .arg("-p")
+                .arg(prompt)
+                .arg("--output-format")
+                .arg("stream-json")
+                .arg("--verbose")
+                .arg("--max-turns")
+                .arg(&max_turns_str)
+                .arg("--model")
+                .arg(&self.config.model);
+
+            if let Some(sid) = resume_session_id {
+                cmd = cmd.arg("--resume").arg(sid);
+            }
+
+            cmd = cmd.envs(extra_env.iter());
+            cmd = cmd.current_dir("/workspace");
+            // Keep stderr on a separate pipe — pty-process attaches the PTY
+            // to all fds by default, which would merge stderr into the PTY
+            // stream and break NDJSON parsing.
+            cmd = cmd.stderr(std::process::Stdio::piped());
+
+            let mut child = cmd.spawn(pts).map_err(|e| WorkerError::ExecutionFailed {
+                reason: format!("failed to spawn claude with PTY: {}", e),
             })?;
 
+            let stderr = child
+                .stderr
+                .take()
+                .ok_or_else(|| WorkerError::ExecutionFailed {
+                    reason: "failed to capture claude stderr".to_string(),
+                })?;
+
+            // stdout comes from the PTY master, which implements AsyncRead
+            let stdout: Box<dyn tokio::io::AsyncRead + Unpin + Send> = Box::new(pty);
+            (child, stdout, stderr)
+        };
+
+        // Non-Unix fallback (Windows CI) — no PTY, direct spawn.
+        // Claude bridge only runs in Linux Docker containers, so this path
+        // exists solely for compilation on Windows targets.
+        #[cfg(not(unix))]
+        let (mut child, stdout, stderr) = {
+            let mut cmd = Command::new("claude");
+            cmd.arg("-p")
+                .arg(prompt)
+                .arg("--output-format")
+                .arg("stream-json")
+                .arg("--verbose")
+                .arg("--max-turns")
+                .arg(&max_turns_str)
+                .arg("--model")
+                .arg(&self.config.model);
+
+            if let Some(sid) = resume_session_id {
+                cmd.arg("--resume").arg(sid);
+            }
+
+            cmd.envs(extra_env);
+            cmd.current_dir("/workspace")
+                .stdout(std::process::Stdio::piped())
+                .stderr(std::process::Stdio::piped());
+
+            let mut child = cmd.spawn().map_err(|e| WorkerError::ExecutionFailed {
+                reason: format!("failed to spawn claude: {}", e),
+            })?;
+
+            let stdout_pipe = child
+                .stdout
+                .take()
+                .ok_or_else(|| WorkerError::ExecutionFailed {
+                    reason: "failed to capture claude stdout".to_string(),
+                })?;
+            let stderr = child
+                .stderr
+                .take()
+                .ok_or_else(|| WorkerError::ExecutionFailed {
+                    reason: "failed to capture claude stderr".to_string(),
+                })?;
+
+            let stdout: Box<dyn tokio::io::AsyncRead + Unpin + Send> = Box::new(stdout_pipe);
+            (child, stdout, stderr)
+        };
+
         // Spawn stderr reader that forwards lines as log events
         let client_for_stderr = Arc::clone(&self.client);
         let job_id = self.config.job_id;
@@ -1027,4 +1088,51 @@ mod tests {
         let copied = copy_dir_recursive(nonexistent, dst.path()).unwrap();
         assert_eq!(copied, 0);
     }
+
+    /// Regression test: arguments are passed individually (not via shell string),
+    /// so shell metacharacters in prompt/model/session_id are harmless.
+    #[test]
+    fn command_args_no_shell_interpretation() {
+        // Prompt, model, and session_id may contain shell metacharacters from
+        // user-supplied task descriptions or LLM output. Since we use
+        // Command::arg() (execve), these are passed as literal strings.
+        let prompt = "Fix the user's bug; echo $HOME && rm -rf /";
+        let model = "claude-3-opus-20240229";
+        let session_id = "'; DROP TABLE jobs; --";
+
+        let max_turns = 10u32;
+        let max_turns_str = max_turns.to_string();
+        let args: Vec<&str> = vec![
+            "-p",
+            prompt,
+            "--output-format",
+            "stream-json",
+            "--verbose",
+            "--max-turns",
+            &max_turns_str,
+            "--model",
+            model,
+            "--resume",
+            session_id,
+        ];
+
+        // All values present as literal strings — no shell interpretation
+        // ["-p", prompt, "--output-format", "stream-json", "--verbose",
+        //  "--max-turns", "10", "--model", model, "--resume", session_id]
+        assert_eq!(args[1], prompt);
+        assert_eq!(args[8], model);
+        assert_eq!(args[10], session_id);
+        // Shell metacharacters preserved, not expanded
+        assert!(args[1].contains("$HOME"));
+        assert!(args[1].contains("&&"));
+        assert!(args[10].contains("'; DROP TABLE"));
+    }
+
+    /// Verify PTY is available on Unix platforms.
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn pty_opens_successfully() {
+        let result = pty_process::open();
+        assert!(result.is_ok(), "PTY allocation should succeed on Unix");
+    }
 }

src/worker/job.rs

@@ -391,6 +391,7 @@ Report when the job is complete or if you encounter issues you cannot resolve."#
             worker: self,
             rx: tokio::sync::Mutex::new(rx),
             consecutive_rate_limits: std::sync::atomic::AtomicUsize::new(0),
+            has_text_response: std::sync::atomic::AtomicBool::new(false),
         };
 
         let config = AgenticLoopConfig {
@@ -1101,6 +1102,15 @@ fn store_fallback_in_metadata(
 }
 
 /// Job delegate: implements `LoopDelegate` for the background job context.
+/// Whether an LLM error represents a completion-eligible empty response.
+///
+/// Only `EmptyResponse` (provider returned no choices/content) qualifies.
+/// Infrastructure errors (`AuthFailed`, `Http`, `Io`, etc.) never qualify —
+/// they must propagate even if prior text output was produced.
+fn is_completion_eligible_error(error: &crate::error::LlmError) -> bool {
+    matches!(error, crate::error::LlmError::EmptyResponse { .. })
+}
+
 ///
 /// Handles: signal channel (stop/ping/user messages), cancellation checks,
 /// rate-limit retry, parallel tool execution, DB persistence, SSE broadcasting.
@@ -1109,6 +1119,10 @@ struct JobDelegate<'a> {
     rx: tokio::sync::Mutex<&'a mut mpsc::Receiver<WorkerMessage>>,
     /// Tracks consecutive rate-limit errors to fail fast instead of burning iterations.
     consecutive_rate_limits: std::sync::atomic::AtomicUsize,
+    /// Whether a substantive (non-empty) text response has been produced.
+    /// When true, an empty follow-up response is treated as job completion
+    /// rather than a retry signal (prevents spurious failures in routines).
+    has_text_response: std::sync::atomic::AtomicBool,
 }
 
 impl<'a> JobDelegate<'a> {
@@ -1161,6 +1175,53 @@ impl<'a> JobDelegate<'a> {
             finish_reason: crate::llm::FinishReason::Stop,
         })
     }
+
+    /// Mark the job as completed, logging a warning on failure.
+    async fn mark_completed_or_warn(&self, context: &str) {
+        if let Err(e) = self.worker.mark_completed().await {
+            tracing::warn!(
+                job_id = %self.worker.job_id,
+                error = %e,
+                "Failed to mark job completed ({context})"
+            );
+        }
+    }
+
+    /// If a substantive text response was already produced and the error
+    /// indicates the LLM simply returned nothing, treat it as successful
+    /// completion rather than a fatal failure.
+    ///
+    /// Only swallows `EmptyResponse` — infrastructure errors (`AuthFailed`,
+    /// `ContextLengthExceeded`, `Http`, `Io`, etc.) always propagate.
+    ///
+    /// Returns `Some(empty RespondOutput)` when the error should be swallowed,
+    /// `None` when it should propagate normally.
+    async fn try_complete_on_error(
+        &self,
+        context: &str,
+        error: &crate::error::LlmError,
+    ) -> Option<crate::llm::RespondOutput> {
+        if !is_completion_eligible_error(error) {
+            return None;
+        }
+        if !self
+            .has_text_response
+            .load(std::sync::atomic::Ordering::Relaxed)
+        {
+            return None;
+        }
+        tracing::info!(
+            job_id = %self.worker.job_id,
+            error = %error,
+            "{context} empty response after text output — treating as completion"
+        );
+        self.mark_completed_or_warn(context).await;
+        Some(crate::llm::RespondOutput {
+            result: RespondResult::Text(String::new()),
+            usage: crate::llm::TokenUsage::default(),
+            finish_reason: crate::llm::FinishReason::Stop,
+        })
+    }
 }
 
 #[async_trait]
@@ -1291,7 +1352,12 @@ impl<'a> LoopDelegate for JobDelegate<'a> {
             Err(crate::error::LlmError::RateLimited { retry_after, .. }) => {
                 return self.handle_rate_limit(retry_after, "tool selection").await;
             }
-            Err(e) => return Err(e.into()),
+            Err(e) => {
+                if let Some(output) = self.try_complete_on_error("select_tools", &e).await {
+                    return Ok(output);
+                }
+                return Err(e.into());
+            }
         };
 
         // Fall back to respond_with_tools
@@ -1321,7 +1387,12 @@ impl<'a> LoopDelegate for JobDelegate<'a> {
                 self.handle_rate_limit(retry_after, "respond_with_tools")
                     .await
             }
-            Err(e) => Err(e.into()),
+            Err(e) => {
+                if let Some(output) = self.try_complete_on_error("respond_with_tools", &e).await {
+                    return Ok(output);
+                }
+                Err(e.into())
+            }
         }
     }
 
@@ -1330,9 +1401,22 @@ impl<'a> LoopDelegate for JobDelegate<'a> {
         text: &str,
         reason_ctx: &mut ReasoningContext,
     ) -> TextAction {
-        // Empty text from rate-limit backoff retry — skip processing and let the
-        // loop proceed to the next iteration which will re-call the LLM.
+        // Empty text after a substantive response means the LLM has finished.
+        // Treat as successful completion rather than continuing the loop (which
+        // would produce "Response contained no message or tool call (empty)").
         if text.is_empty() {
+            if self
+                .has_text_response
+                .load(std::sync::atomic::Ordering::Relaxed)
+            {
+                tracing::debug!(
+                    job_id = %self.worker.job_id,
+                    "Empty response after text output — treating as completion"
+                );
+                self.mark_completed_or_warn("empty text response").await;
+                return TextAction::Return(LoopOutcome::Response(String::new()));
+            }
+            // No prior text response — this is likely a rate-limit backoff retry.
             return TextAction::Continue;
         }
 
@@ -1348,6 +1432,10 @@ impl<'a> LoopDelegate for JobDelegate<'a> {
             return TextAction::Return(LoopOutcome::Response(text.to_string()));
         }
 
+        // Track that a substantive response has been produced.
+        self.has_text_response
+            .store(true, std::sync::atomic::Ordering::Relaxed);
+
         // Add assistant response to context
         reason_ctx.messages.push(ChatMessage::assistant(text));
 
@@ -2285,4 +2373,60 @@ mod tests {
         assert_eq!(telegram[0].0, "owner-scope");
         assert_eq!(telegram[0].1.content, "hello from routine");
     }
+
+    /// Regression test: only `EmptyResponse` errors are eligible for
+    /// completion-swallowing. Infrastructure errors must always propagate.
+    #[test]
+    fn is_completion_eligible_only_matches_empty_response() {
+        use crate::error::LlmError;
+
+        // EmptyResponse is eligible
+        assert!(super::is_completion_eligible_error(
+            &LlmError::EmptyResponse {
+                provider: "test".to_string(),
+            }
+        ));
+
+        // All other variants are NOT eligible
+        assert!(!super::is_completion_eligible_error(
+            &LlmError::InvalidResponse {
+                provider: "test".to_string(),
+                reason: "parse error".to_string(),
+            }
+        ));
+        assert!(!super::is_completion_eligible_error(
+            &LlmError::AuthFailed {
+                provider: "test".to_string(),
+            }
+        ));
+        assert!(!super::is_completion_eligible_error(
+            &LlmError::ContextLengthExceeded {
+                used: 100_000,
+                limit: 50_000,
+            }
+        ));
+        assert!(!super::is_completion_eligible_error(
+            &LlmError::ModelNotAvailable {
+                provider: "test".to_string(),
+                model: "gpt-4".to_string(),
+            }
+        ));
+        assert!(!super::is_completion_eligible_error(
+            &LlmError::RequestFailed {
+                provider: "test".to_string(),
+                reason: "timeout".to_string(),
+            }
+        ));
+        assert!(!super::is_completion_eligible_error(
+            &LlmError::SessionExpired {
+                provider: "test".to_string(),
+            }
+        ));
+        assert!(!super::is_completion_eligible_error(
+            &LlmError::SessionRenewalFailed {
+                provider: "test".to_string(),
+                reason: "timeout".to_string(),
+            }
+        ));
+    }
 }