feat(watcher): per-source workers + global Claude session ceiling

claudear.example.toml

@@ -23,9 +23,18 @@ poll_interval_ms = 300000
 # Max issues to process per poll cycle (default: 5)
 max_issues_per_cycle = 5
 
-# Max concurrent issue processing (default: 1)
+# Max concurrent issue processing PER SOURCE for sources without an explicit
+# override (default: 1). Each source runs its own worker that polls and fixes
+# independently; this caps how many fixes a single source runs at once.
 max_concurrent = 1
 
+# Global ceiling on concurrent Claude sessions across EVERY source and path
+# (fix, QA, retry, deploy-QA), regardless of how high per-source max_concurrent
+# limits sum (default: 12). This is the machine-wide safety bound. Clamped to
+# >= 1. The built-in Sentry and GitHub sources default to a per-source
+# max_concurrent of 6; all other sources default to the global max_concurrent.
+max_concurrent_sessions = 12
+
 # Delay between processing issues in ms (default: 5000)
 processing_delay_ms = 5000
 
@@ -301,6 +310,10 @@ poll_interval_ms = 60000
 # Auto-resolve issues on Linear/Sentry when PRs merge (default: false)
 auto_resolve_on_merge = false
 
+# Max concurrent fixes for the GitHub source worker (default: 6). GitHub issues
+# frequently retry, so it ships with a wider worker than the global default.
+# max_concurrent = 6
+
 # Optional: Webhook secret for verifying GitHub webhook signatures
 # Set via GITHUB_WEBHOOK_SECRET env var for security
 webhook_secret = ""
@@ -475,7 +488,9 @@ escalation_threshold_percent = 50
 # Set via SENTRY_CLIENT_SECRET env var for security
 client_secret = ""
 
-# Per-source rate limiting (overrides global values if set)
+# Per-source rate limiting (overrides global values if set).
+# The Sentry worker defaults to max_concurrent = 6 when omitted (wider than the
+# global default since Sentry is high-volume and retry-heavy).
 max_issues_per_cycle = 2
 max_concurrent = 4
 
@@ -1065,8 +1080,9 @@ use_llm = false
 # Max questions answered concurrently per source, independent of the fix-lane
 # concurrency budget (max_concurrent / [sources.<name>].max_concurrent). Increase
 # this to answer more questions in parallel without affecting fix throughput.
-# Clamped to 1 if set to 0 to avoid deadlock. (default: 1)
-max_concurrent = 1
+# Still bounded by the global max_concurrent_sessions ceiling.
+# Clamped to 1 if set to 0 to avoid deadlock. (default: 6)
+max_concurrent = 6
 
 # ============================================
 # Reply Action

crates/claudear-config/src/config.rs

@@ -447,8 +447,14 @@ pub struct Config {
     pub db_path: PathBuf,
     /// Maximum issues to process per poll cycle.
     pub max_issues_per_cycle: usize,
-    /// Maximum concurrent issue processing.
+    /// Maximum concurrent issue processing (per source, default for sources
+    /// without an explicit override).
     pub max_concurrent: usize,
+    /// Global ceiling on concurrent Claude sessions across every source and
+    /// path (fix, QA, retry, deploy-QA). Per-source `max_concurrent` bounds how
+    /// many a single source runs; this bounds the machine-wide total regardless
+    /// of how high the per-source limits are summed. Clamped to >= 1.
+    pub max_concurrent_sessions: usize,
     /// Delay between processing issues (ms).
     pub processing_delay_ms: u64,
     /// Maximum number of activity entries to keep in the IPC server (default: 10,000).
@@ -793,6 +799,7 @@ impl Default for Config {
             db_path: PathBuf::from("claudear.db"),
             max_issues_per_cycle: 5,
             max_concurrent: 1,
+            max_concurrent_sessions: 12,
             processing_delay_ms: 5000,
             max_activity_entries: 10_000,
             ipc_timeout_secs: 30,
@@ -900,7 +907,7 @@ impl Default for QaConfig {
             answer_timeout_secs: 600,
             max_qa_per_cycle: 20,
             use_llm: false,
-            max_concurrent: 1,
+            max_concurrent: 6,
         }
     }
 }
@@ -1740,6 +1747,10 @@ pub struct GitHubConfig {
     /// GitHub App configuration (nested under [scm.github.app]).
     #[serde(default)]
     pub app: GitHubAppConfig,
+    /// Maximum concurrent issue processing for the GitHub source (overrides
+    /// global `max_concurrent`). Defaults to a wider worker since GitHub issues
+    /// frequently retry.
+    pub max_concurrent: Option<usize>,
 }
 
 impl Default for GitHubConfig {
@@ -1756,6 +1767,7 @@ impl Default for GitHubConfig {
             trigger_labels: Vec::new(),
             trigger_states: Vec::new(),
             app: GitHubAppConfig::default(),
+            max_concurrent: Some(6),
         }
     }
 }
@@ -1775,6 +1787,7 @@ impl GitHubConfig {
             trigger_labels: vec!["auto-implement".to_string(), "claude".to_string()],
             trigger_states: vec!["open".to_string()],
             app: GitHubAppConfig::default(),
+            max_concurrent: Some(6),
         }
     }
 }
@@ -2053,7 +2066,9 @@ impl Default for SentryConfig {
             escalation_threshold_percent: 50,
             client_secret: None,
             max_issues_per_cycle: None,
-            max_concurrent: None,
+            // Sentry issues are high-volume and often need retries, so run a
+            // wider worker by default. Override in [issues.sentry] if needed.
+            max_concurrent: Some(6),
             poll_interval_ms: None,
         }
     }
@@ -3808,6 +3823,17 @@ impl Config {
                 .as_ref()
                 .and_then(|c| c.max_concurrent)
                 .unwrap_or(self.max_concurrent),
+            "github_issues" => self
+                .scm
+                .github
+                .max_concurrent
+                .unwrap_or(self.max_concurrent),
+            "helpscout" => self
+                .issues
+                .helpscout
+                .as_ref()
+                .and_then(|c| c.max_concurrent)
+                .unwrap_or(self.max_concurrent),
             _ => self.max_concurrent,
         }
     }
@@ -4852,6 +4878,8 @@ api_key = "key"
                 sentry: Some(SentryConfig {
                     auth_token: "tok".into(),
                     org_slug: "org".into(),
+                    // Opt out of the Sentry default (6) to exercise fallback.
+                    max_concurrent: None,
                     ..Default::default()
                 }),
                 ..Default::default()
@@ -4863,6 +4891,30 @@ api_key = "key"
         assert_eq!(config.max_concurrent_for("unknown"), 4);
     }
 
+    #[test]
+    fn test_default_worker_concurrency_for_heavy_sources() {
+        // Sentry and GitHub ship with a wider default worker; everything else
+        // falls back to the global max_concurrent (1 by default).
+        let config = Config {
+            issues: IssuesConfig {
+                sentry: Some(SentryConfig {
+                    auth_token: "tok".into(),
+                    org_slug: "org".into(),
+                    ..Default::default()
+                }),
+                ..Default::default()
+            },
+            ..Default::default()
+        };
+        assert_eq!(config.max_concurrent, 1);
+        assert_eq!(config.max_concurrent_sessions, 12);
+        assert_eq!(config.max_concurrent_for("sentry"), 6);
+        // GitHub config is always present (not optional) and defaults to 6.
+        assert_eq!(config.max_concurrent_for("github_issues"), 6);
+        assert_eq!(config.max_concurrent_for("linear"), 1);
+        assert_eq!(config.max_concurrent_for("unknown"), 1);
+    }
+
     #[test]
     fn test_per_source_max_concurrent_overrides_global() {
         let config = Config {

crates/claudear-engine/src/api/auth.rs

@@ -922,6 +922,7 @@ mod tests {
             db_path: ":memory:".into(),
             max_issues_per_cycle: 5,
             max_concurrent: 1,
+            max_concurrent_sessions: 12,
             processing_delay_ms: 5000,
             max_activity_entries: 100,
             ipc_timeout_secs: 30,

crates/claudear-engine/src/api/routes.rs

@@ -3055,6 +3055,7 @@ mod tests {
             db_path: ":memory:".into(),
             max_issues_per_cycle: 5,
             max_concurrent: 1,
+            max_concurrent_sessions: 12,
             processing_delay_ms: 5000,
             max_activity_entries: 100,
             ipc_timeout_secs: 30,

crates/claudear-engine/src/processing.rs

@@ -726,6 +726,11 @@ pub struct IssueProcessor {
     /// provider, local-LLM-based when `agent.use_llm` is set. `None` falls back
     /// to the label/source heuristic.
     pub intent_classifier: Option<Arc<dyn IntentClassifier>>,
+    /// Shared global Claude-session ceiling. When set, detached agent work the
+    /// processor spawns (the retrieval-quality judge) acquires a permit per
+    /// session so it counts against the same limit as the main run. `None`
+    /// leaves that work ungated (callers without a watcher-level ceiling).
+    pub session_limiter: Option<Arc<tokio::sync::Semaphore>>,
 }
 
 /// Everything the caller provides to `IssueProcessor::run()`.
@@ -3643,6 +3648,9 @@ impl IssueProcessor {
         let tracker = self.tracker.clone();
         let analyzer = self.llm_analyzer.clone();
         let agent = self.qa_agent.clone().unwrap_or_else(|| self.agent.clone());
+        // Each agent-backed score is a Claude session, so bound it by the same
+        // global ceiling as the main run (no-op for the local-LLM backend).
+        let session_limiter = self.session_limiter.clone();
         // Issue identity captured for the detached task's timeline events.
         let issue_id = issue.id.clone();
         let short_id = issue.short_id.clone();
@@ -3712,8 +3720,15 @@ impl IssueProcessor {
                     let tracker = &tracker;
                     let agent = agent.as_ref();
                     let scored = &scored;
+                    let session_limiter = session_limiter.as_ref();
                     futures::stream::iter(items.iter())
                         .for_each_concurrent(RETRIEVAL_JUDGE_CONCURRENCY, |item| async move {
+                            // Hold a global session permit for the scoring call so
+                            // these detached sessions never exceed the ceiling.
+                            let _session_permit = match session_limiter {
+                                Some(limiter) => limiter.acquire().await.ok(),
+                                None => None,
+                            };
                             if let Some(score) =
                                 crate::agent_classifier::score_chunk_relevance_via_agent(
                                     agent,
@@ -5242,6 +5257,7 @@ mod tests {
             github_client: None,
             llm_analyzer: None,
             intent_classifier: None,
+            session_limiter: None,
         };
 
         let input = ProcessingInput {
@@ -6356,6 +6372,7 @@ mod tests {
             github_client: None,
             llm_analyzer: None,
             intent_classifier: None,
+            session_limiter: None,
         }
     }