From b1f4852021fb5c783ec2d1c43e2198d05847895a Mon Sep 17 00:00:00 2001 From: Claudear <262350598+claudear@users.noreply.github.com> Date: Mon, 10 Aug 2026 12:53:12 +0000 Subject: [PATCH 01/18] fix(watcher): reap finished issue-processing task handles MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The watcher pushed a JoinHandle into `spawn_handles` for every issue it dispatched and never removed it outside of tests, so the list was push-only in production. Tokio frees a task's allocation only once both the scheduler and the last JoinHandle are dropped, so each completed `process_issue` task stayed resident for the daemon's lifetime — and that allocation is large, since `process_issue` inlined the whole `IssueProcessor::run` pipeline. RSS therefore grew monotonically with issues processed until the container was OOM-killed and restarted with a fresh (empty) list, matching the reported intermittent OOM kills. - Reap finished handles once per poll cycle and on every dispatch, so the list is bounded by concurrency instead of issues-processed-ever. - Drain the spawned tasks in `stop_and_drain` so shutdown waits for their teardown too. - Box the `processor.run(...)` future so each spawned task allocation carries a pointer instead of the fully inlined pipeline state machine. Co-Authored-By: Claude Opus 5 --- crates/claudear-engine/src/watcher.rs | 93 +++++++++++++++++++++++++-- 1 file changed, 88 insertions(+), 5 deletions(-) diff --git a/crates/claudear-engine/src/watcher.rs b/crates/claudear-engine/src/watcher.rs index b67ee0f..9a04b20 100644 --- a/crates/claudear-engine/src/watcher.rs +++ b/crates/claudear-engine/src/watcher.rs @@ -247,7 +247,10 @@ pub struct Watcher { /// agent-based (Claude Code) by default, local-LLM-based when `qa.use_llm` is /// set. intent_classifier: Option>, - /// Join handles for spawned issue-processing tasks (used by tests to drain). + /// Join handles for in-flight issue-processing tasks. + /// + /// Finished handles are reaped by [`Self::reap_finished_spawn_handles`] so the + /// list stays bounded by concurrency rather than by issues-processed-ever. spawn_handles: tokio::sync::Mutex>>, } @@ -336,10 +339,26 @@ impl Watcher { } } + /// Drop the join handles of issue-processing tasks that have already finished. + /// + /// Tokio keeps a task's allocation alive until both the scheduler and the last + /// `JoinHandle` are dropped. Retaining handles for completed tasks therefore + /// pins one whole `process_issue` state machine per issue for the lifetime of + /// the daemon, which grows RSS without bound until the container is OOM-killed. + /// Called once per poll cycle and on every dispatch. + async fn reap_finished_spawn_handles(&self) { + let mut handles = self.spawn_handles.lock().await; + handles.retain(|handle| !handle.is_finished()); + // Also release the backing capacity of an unusually large burst. + if handles.capacity() > handles.len().saturating_mul(4) + 16 { + handles.shrink_to_fit(); + } + } + /// Wait for all spawned issue-processing tasks to complete. /// - /// Primarily useful in tests that need to assert on processing outcomes - /// after a non-blocking `poll_source` call. + /// Used on graceful shutdown, and by tests that need to assert on processing + /// outcomes after a non-blocking `poll_source` call. pub async fn drain_spawned_tasks(&self) { let handles: Vec<_> = { let mut guard = self.spawn_handles.lock().await; @@ -1163,6 +1182,11 @@ impl Watcher { let _ = tokio::time::timeout(remaining, self.slot_available.notified()).await; } + // Join the spawned tasks themselves so shutdown waits for their teardown too, + // and so their handles are released rather than dropped with the watcher. + let remaining = max_wait.saturating_sub(start.elapsed()); + let _ = tokio::time::timeout(remaining, self.drain_spawned_tasks()).await; + tracing::info!("Claude Watcher stopped gracefully"); } @@ -2985,6 +3009,10 @@ Create a PR with your changes.{custom_instructions}"#, /// Poll a single source. async fn poll_source(self: &Arc, source: &Arc) -> Result<()> { + // Release tasks that finished since the last cycle before doing anything else, + // so an idle or rate-limit-paused watcher still frees their allocations. + self.reap_finished_spawn_handles().await; + if self.is_rate_limit_paused().await { return Ok(()); } @@ -3468,7 +3496,11 @@ Create a PR with your changes.{custom_instructions}"#, .process_issue(source_clone, issue, match_result, None, None, intent) .await; }); - self.spawn_handles.lock().await.push(handle); + { + let mut handles = self.spawn_handles.lock().await; + handles.retain(|h| !h.is_finished()); + handles.push(handle); + } // Add delay between starting new issues (skip trailing delay after the last item). if i + 1 < total && self.config.processing_delay_ms > 0 { @@ -3873,7 +3905,9 @@ Create a PR with your changes.{custom_instructions}"#, }; let context_provider = crate::processing::SourceContext(source.as_ref()); - let outcome = processor.run(input, &context_provider).await; + // Box the pipeline future: `run` inlines the whole processing state machine, + // so awaiting it directly would make every spawned task allocation carry it. + let outcome = Box::pin(processor.run(input, &context_provider)).await; // Watcher-specific: check for rate limit errors and pause if needed if let ProcessingOutcome::Failed { ref error } = outcome { @@ -5025,6 +5059,55 @@ mod tests { })) } + /// Completed issue-processing tasks must not stay pinned in `spawn_handles`. + /// + /// Tokio frees a task's allocation only once the scheduler *and* the last + /// `JoinHandle` are gone, so a push-only handle list leaks one whole + /// `process_issue` state machine per issue, growing the daemon's RSS until + /// the container is OOM-killed. + #[tokio::test] + async fn test_watcher_reaps_finished_spawn_handles() { + let notifier = Arc::new(MockNotifier::new(true)); + let tracker = Arc::new(SqliteTracker::in_memory().unwrap()); + // Source with no issues: this test is about reaping, not dispatching. + let source = Arc::new(MockSource::new("reap")) as Arc; + + let watcher = create_test_watcher(notifier, tracker, vec![source.clone()], false); + watcher.is_running.store(true, Ordering::SeqCst); + + // Stand in for the handles left behind by 64 already-processed issues. + { + let mut handles = watcher.spawn_handles.lock().await; + for _ in 0..64 { + handles.push(tokio::spawn(async {})); + } + } + + // Let every simulated task run to completion. + for _ in 0..1000 { + if watcher + .spawn_handles + .lock() + .await + .iter() + .all(|handle| handle.is_finished()) + { + break; + } + tokio::task::yield_now().await; + } + + // A normal poll cycle must release them; nothing calls drain in production. + watcher.poll_source(&source).await.unwrap(); + + let retained = watcher.spawn_handles.lock().await.len(); + assert_eq!( + retained, 0, + "watcher retained {retained} handles for completed issue-processing tasks; \ + each one pins a full process_issue allocation for the daemon's lifetime" + ); + } + #[test] fn test_watcher_new() { let notifier = Arc::new(MockNotifier::new(true)); From 2ec80ea430ee7222519b90901cecdf8604afcd32 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Sun, 16 Aug 2026 16:46:38 +0530 Subject: [PATCH 02/18] fix(watcher): close shutdown race recording spawned tasks Re-check is_running and spawn+record the handle under the spawn_handles lock that drain_spawned_tasks takes. Since stop() sets is_running=false before draining, a dispatch either records before the drain's take (so shutdown joins it) or sees the stop and never spawns. The top-of-loop check alone left a window between check and push. --- crates/claudear-engine/src/watcher.rs | 58 ++++++++++++++++++++++++--- 1 file changed, 52 insertions(+), 6 deletions(-) diff --git a/crates/claudear-engine/src/watcher.rs b/crates/claudear-engine/src/watcher.rs index 70a3e53..3f36d09 100644 --- a/crates/claudear-engine/src/watcher.rs +++ b/crates/claudear-engine/src/watcher.rs @@ -3574,15 +3574,25 @@ Create a PR with your changes.{custom_instructions}"#, // the housekeeping loop (review checks, auto-close, retries) is not starved. let watcher = Arc::clone(self); let source_clone = Arc::clone(source); - let handle = tokio::spawn(async move { - watcher - .process_issue(source_clone, issue, match_result, None, None, intent) - .await; - }); { let mut handles = self.spawn_handles.lock().await; + // Re-check is_running under the same lock that drain_spawned_tasks + // takes, then spawn+record atomically. This closes the shutdown + // race: once stop() has set is_running=false (before the drain), + // a dispatch either records its handle before the drain's take (so + // shutdown joins it) or sees the stop here and never spawns. The + // top-of-loop check is not enough on its own — it runs before the + // spawn, so a concurrent drain could take the vector between it and + // the push. + if !self.is_running.load(Ordering::SeqCst) { + break; + } handles.retain(|h| !h.is_finished()); - handles.push(handle); + handles.push(tokio::spawn(async move { + watcher + .process_issue(source_clone, issue, match_result, None, None, intent) + .await; + })); } // Add delay between starting new issues (skip trailing delay after the last item). @@ -5193,6 +5203,42 @@ mod tests { ); } + #[tokio::test] + async fn test_stop_and_drain_joins_recorded_task() { + use std::sync::atomic::AtomicBool; + let notifier = Arc::new(MockNotifier::new(true)); + let tracker = Arc::new(SqliteTracker::in_memory().unwrap()); + let source = Arc::new(MockSource::new("drain")) as Arc; + let watcher = create_test_watcher(notifier, tracker, vec![source], false); + watcher.is_running.store(true, Ordering::SeqCst); + + // A recorded task whose teardown completes shortly after it is spawned. + let done = Arc::new(AtomicBool::new(false)); + let done_clone = Arc::clone(&done); + { + let mut handles = watcher.spawn_handles.lock().await; + handles.push(tokio::spawn(async move { + tokio::task::yield_now().await; + done_clone.store(true, Ordering::SeqCst); + })); + } + + // Graceful shutdown must wait for the recorded task's teardown rather than + // reporting a clean stop while it is still mutating shared state. + watcher.stop_and_drain().await; + + assert!( + done.load(Ordering::SeqCst), + "stop_and_drain returned before the recorded task finished; a \ + concurrently processing issue could still be mutating tracker/notifier/agent state" + ); + assert!( + watcher.spawn_handles.lock().await.is_empty(), + "drain must release recorded handles" + ); + assert!(!watcher.is_running.load(Ordering::SeqCst)); + } + #[test] fn test_watcher_new() { let notifier = Arc::new(MockNotifier::new(true)); From e4d295fd6c02085cb4e6827c7ca6177b0b139885 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Sun, 16 Aug 2026 16:47:30 +0530 Subject: [PATCH 03/18] empty From 7b76708a77d24e70ad39721f8e829b0882b02365 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Sun, 16 Aug 2026 16:53:17 +0530 Subject: [PATCH 04/18] fix(watcher): make shutdown drain cancel-safe Wrapping drain_spawned_tasks in a timeout dropped the handles it had already taken from spawn_handles, detaching unfinished tasks so the runtime aborted them mid-operation. drain_spawned_tasks_until pops one handle at a time and joins via &mut handle, putting any handle that exceeds the budget back into spawn_handles instead of dropping it. --- crates/claudear-engine/src/watcher.rs | 87 +++++++++++++++++++++++++-- 1 file changed, 81 insertions(+), 6 deletions(-) diff --git a/crates/claudear-engine/src/watcher.rs b/crates/claudear-engine/src/watcher.rs index 3f36d09..91378fc 100644 --- a/crates/claudear-engine/src/watcher.rs +++ b/crates/claudear-engine/src/watcher.rs @@ -380,6 +380,31 @@ impl Watcher { } } + /// Join spawned tasks until they finish or `deadline` passes. + /// + /// Cancel-safe, unlike wrapping [`Self::drain_spawned_tasks`] in a + /// `timeout`: that takes every handle out of `spawn_handles` first, so a + /// firing timeout drops the taken handles and *detaches* the still-running + /// tasks, which the runtime then aborts mid-operation. Here each handle is + /// popped one at a time and joined via `&mut handle`, so a handle whose join + /// exceeds the deadline is put back into `spawn_handles` rather than dropped. + /// Returns `true` when every task was joined within the budget. + async fn drain_spawned_tasks_until(&self, deadline: std::time::Instant) -> bool { + loop { + let mut handle = match self.spawn_handles.lock().await.pop() { + Some(h) => h, + None => return true, + }; + let remaining = deadline.saturating_duration_since(std::time::Instant::now()); + // `&mut handle` is the future, so a timeout drops only the borrow; + // `handle` survives and goes back into the list, never detached. + if remaining.is_zero() || tokio::time::timeout(remaining, &mut handle).await.is_err() { + self.spawn_handles.lock().await.push(handle); + return false; + } + } + } + /// Get a trait-object reference to the LLM analyzer, if available. fn llm(&self) -> Option<&dyn claudear_analysis::llm::LlmAnalyzer> { self.llm_analyzer @@ -1193,12 +1218,18 @@ impl Watcher { let _ = tokio::time::timeout(remaining, self.slot_available.notified()).await; } - // Join the spawned tasks themselves so shutdown waits for their teardown too, - // and so their handles are released rather than dropped with the watcher. - let remaining = max_wait.saturating_sub(start.elapsed()); - let _ = tokio::time::timeout(remaining, self.drain_spawned_tasks()).await; - - tracing::info!("Claude Watcher stopped gracefully"); + // Join the spawned tasks themselves so shutdown waits for their teardown + // too, and so their handles are released rather than dropped with the + // watcher. Bounded by the same 30s budget, but cancel-safely: an + // unfinished task is left in spawn_handles rather than detached. + if self.drain_spawned_tasks_until(start + max_wait).await { + tracing::info!("Claude Watcher stopped gracefully"); + } else { + tracing::warn!( + "Graceful shutdown budget exhausted while draining spawned tasks; \ + unfinished tasks remain and were not detached" + ); + } } /// Check if the watcher is currently running. @@ -5239,6 +5270,50 @@ mod tests { assert!(!watcher.is_running.load(Ordering::SeqCst)); } + #[tokio::test] + async fn test_drain_spawned_tasks_until_keeps_unfinished_handle() { + use std::sync::atomic::AtomicBool; + use std::time::{Duration, Instant}; + let notifier = Arc::new(MockNotifier::new(true)); + let tracker = Arc::new(SqliteTracker::in_memory().unwrap()); + let source = Arc::new(MockSource::new("drain")) as Arc; + let watcher = create_test_watcher(notifier, tracker, vec![source], false); + + // A task that outlives a short drain budget. + let done = Arc::new(AtomicBool::new(false)); + let done_clone = Arc::clone(&done); + { + let mut handles = watcher.spawn_handles.lock().await; + handles.push(tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(200)).await; + done_clone.store(true, Ordering::SeqCst); + })); + } + + // Budget shorter than the task: reports incomplete and must NOT detach it + // (dropping the handle would let the runtime abort it mid-operation). + let ok = watcher + .drain_spawned_tasks_until(Instant::now() + Duration::from_millis(20)) + .await; + assert!(!ok, "budget was too short, drain should report incomplete"); + assert_eq!( + watcher.spawn_handles.lock().await.len(), + 1, + "the unfinished task's handle must be kept, not dropped/detached" + ); + + // The task is still alive; draining with ample budget joins it to completion. + let ok = watcher + .drain_spawned_tasks_until(Instant::now() + Duration::from_secs(5)) + .await; + assert!(ok); + assert!( + done.load(Ordering::SeqCst), + "task ran to completion — it was never aborted" + ); + assert!(watcher.spawn_handles.lock().await.is_empty()); + } + #[test] fn test_watcher_new() { let notifier = Arc::new(MockNotifier::new(true)); From c2e028c611aac341905efaadf1f97e0c1db4e419 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Sun, 16 Aug 2026 17:03:08 +0530 Subject: [PATCH 05/18] fix(watcher): drain to completion instead of a fixed budget The 30s cap let stop_and_drain return while a task was still running, after which runtime teardown aborted it mid-operation. Drain the spawn_handles to completion instead (active_processing is only bumped inside handle-tracked process_issue, so the handles subsume it). Unbounded on purpose: production races this against an operator force-quit, the intended hard limit. Supersedes the cancel-safe timeout drain, which the fixed budget still undermined. --- crates/claudear-engine/src/watcher.rs | 150 ++++++++------------------ 1 file changed, 46 insertions(+), 104 deletions(-) diff --git a/crates/claudear-engine/src/watcher.rs b/crates/claudear-engine/src/watcher.rs index 91378fc..01786a0 100644 --- a/crates/claudear-engine/src/watcher.rs +++ b/crates/claudear-engine/src/watcher.rs @@ -380,31 +380,6 @@ impl Watcher { } } - /// Join spawned tasks until they finish or `deadline` passes. - /// - /// Cancel-safe, unlike wrapping [`Self::drain_spawned_tasks`] in a - /// `timeout`: that takes every handle out of `spawn_handles` first, so a - /// firing timeout drops the taken handles and *detaches* the still-running - /// tasks, which the runtime then aborts mid-operation. Here each handle is - /// popped one at a time and joined via `&mut handle`, so a handle whose join - /// exceeds the deadline is put back into `spawn_handles` rather than dropped. - /// Returns `true` when every task was joined within the budget. - async fn drain_spawned_tasks_until(&self, deadline: std::time::Instant) -> bool { - loop { - let mut handle = match self.spawn_handles.lock().await.pop() { - Some(h) => h, - None => return true, - }; - let remaining = deadline.saturating_duration_since(std::time::Instant::now()); - // `&mut handle` is the future, so a timeout drops only the borrow; - // `handle` survives and goes back into the list, never detached. - if remaining.is_zero() || tokio::time::timeout(remaining, &mut handle).await.is_err() { - self.spawn_handles.lock().await.push(handle); - return false; - } - } - } - /// Get a trait-object reference to the LLM analyzer, if available. fn llm(&self) -> Option<&dyn claudear_analysis::llm::LlmAnalyzer> { self.llm_analyzer @@ -1194,42 +1169,17 @@ impl Watcher { pub async fn stop_and_drain(&self) { self.stop(); - // Wait for any active processing to complete (up to 30 seconds). - // Uses slot_available to wake immediately when a task finishes rather - // than polling on a fixed interval. - let max_wait = std::time::Duration::from_secs(30); - let start = std::time::Instant::now(); - - while self.active_processing.load(Ordering::SeqCst) > 0 { - if start.elapsed() > max_wait { - tracing::warn!( - remaining = self.active_processing.load(Ordering::SeqCst), - "Graceful shutdown timeout reached, some tasks may not have completed" - ); - break; - } - tracing::info!( - active_count = self.active_processing.load(Ordering::SeqCst), - "Waiting for active tasks to complete..." - ); - // Wait for a task to finish (notifies via slot_available) or fall back - // to a periodic check in case the notification was missed. - let remaining = max_wait.saturating_sub(start.elapsed()); - let _ = tokio::time::timeout(remaining, self.slot_available.notified()).await; - } - - // Join the spawned tasks themselves so shutdown waits for their teardown - // too, and so their handles are released rather than dropped with the - // watcher. Bounded by the same 30s budget, but cancel-safely: an - // unfinished task is left in spawn_handles rather than detached. - if self.drain_spawned_tasks_until(start + max_wait).await { - tracing::info!("Claude Watcher stopped gracefully"); - } else { - tracing::warn!( - "Graceful shutdown budget exhausted while draining spawned tasks; \ - unfinished tasks remain and were not detached" - ); - } + // Join every in-flight issue-processing task to completion before + // returning. This deliberately has no internal deadline: a fixed budget + // let stop_and_drain return while a task was still running, after which + // production tore down the Tokio runtime and aborted that task + // mid-operation, corrupting a partially-applied fix. `stop()` has already + // cleared is_running, so no new tasks are spawned; each running + // process_issue carries its own internal timeouts, so this terminates. + // Production bounds it by racing this against an operator force-quit + // (a second Ctrl+C -> process::exit), which is the intended hard limit. + self.drain_spawned_tasks().await; + tracing::info!("Claude Watcher stopped gracefully"); } /// Check if the watcher is currently running. @@ -5271,47 +5221,36 @@ mod tests { } #[tokio::test] - async fn test_drain_spawned_tasks_until_keeps_unfinished_handle() { + async fn test_stop_and_drain_joins_task_to_completion() { use std::sync::atomic::AtomicBool; - use std::time::{Duration, Instant}; + use std::time::Duration; let notifier = Arc::new(MockNotifier::new(true)); let tracker = Arc::new(SqliteTracker::in_memory().unwrap()); let source = Arc::new(MockSource::new("drain")) as Arc; let watcher = create_test_watcher(notifier, tracker, vec![source], false); + watcher.is_running.store(true, Ordering::SeqCst); - // A task that outlives a short drain budget. + // An in-flight task that only finishes after a delay. Shutdown must wait + // for it to complete rather than returning and letting it be aborted. let done = Arc::new(AtomicBool::new(false)); let done_clone = Arc::clone(&done); { let mut handles = watcher.spawn_handles.lock().await; handles.push(tokio::spawn(async move { - tokio::time::sleep(Duration::from_millis(200)).await; + tokio::time::sleep(Duration::from_millis(150)).await; done_clone.store(true, Ordering::SeqCst); })); } - // Budget shorter than the task: reports incomplete and must NOT detach it - // (dropping the handle would let the runtime abort it mid-operation). - let ok = watcher - .drain_spawned_tasks_until(Instant::now() + Duration::from_millis(20)) - .await; - assert!(!ok, "budget was too short, drain should report incomplete"); - assert_eq!( - watcher.spawn_handles.lock().await.len(), - 1, - "the unfinished task's handle must be kept, not dropped/detached" - ); + watcher.stop_and_drain().await; - // The task is still alive; draining with ample budget joins it to completion. - let ok = watcher - .drain_spawned_tasks_until(Instant::now() + Duration::from_secs(5)) - .await; - assert!(ok); assert!( done.load(Ordering::SeqCst), - "task ran to completion — it was never aborted" + "stop_and_drain returned before the in-flight task completed; runtime \ + teardown would then abort it mid-operation" ); assert!(watcher.spawn_handles.lock().await.is_empty()); + assert!(!watcher.is_running.load(Ordering::SeqCst)); } #[test] @@ -7688,21 +7627,30 @@ mod tests { let watcher = Arc::new(create_test_watcher(notifier, tracker, sources, false)); watcher.is_running.store(true, Ordering::SeqCst); - watcher.active_processing.fetch_add(1, Ordering::SeqCst); - // Simulate task finishing after a short delay + // A handle-tracked in-flight task, mirroring process_issue: it holds an + // active_processing count and clears it only when it finishes. Shutdown + // must wait for the handle, so the count is 0 by the time it returns. + watcher.active_processing.fetch_add(1, Ordering::SeqCst); let release = Arc::clone(&watcher); - tokio::spawn(async move { - tokio::time::sleep(std::time::Duration::from_millis(100)).await; - release.active_processing.fetch_sub(1, Ordering::SeqCst); - release.slot_available.notify_waiters(); - }); + { + let mut handles = watcher.spawn_handles.lock().await; + handles.push(tokio::spawn(async move { + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + release.active_processing.fetch_sub(1, Ordering::SeqCst); + release.slot_available.notify_waiters(); + })); + } let result = tokio::time::timeout(std::time::Duration::from_secs(5), watcher.stop_and_drain()).await; assert!(result.is_ok(), "stop_and_drain timed out"); assert!(!watcher.is_running()); - assert_eq!(watcher.active_count(), 0); + assert_eq!( + watcher.active_count(), + 0, + "shutdown waited for the in-flight task to finish" + ); } #[test] @@ -10100,23 +10048,17 @@ mod tests { let watcher = Arc::new(create_test_watcher(notifier, tracker, vec![], false)); watcher.is_running.store(true, Ordering::SeqCst); - // Simulate a task that never completes (active count stays > 0) - watcher.active_processing.store(1, Ordering::SeqCst); - - // stop_and_drain has a 5-minute internal timeout, but we use an outer timeout - // We just verify it eventually returns (the internal max_wait breaks the loop) + // A stray active_processing count with no recorded handle must not wedge + // shutdown: stop_and_drain waits on the spawn_handles, which are empty + // here, so it returns promptly and clears is_running. let result = tokio::time::timeout(std::time::Duration::from_secs(10), watcher.stop_and_drain()) .await; - // In test the internal max_wait is 300s which we can't wait for, - // so this test verifies the method was called correctly and stop was set - // The timeout will trigger because 300s > 10s, but that's fine - if result.is_err() { - // Timed out externally - that's expected since internal timeout is 300s - assert!(!watcher.is_running()); - } else { - assert!(!watcher.is_running()); - } + assert!( + result.is_ok(), + "stop_and_drain should return promptly with no handles" + ); + assert!(!watcher.is_running()); } #[tokio::test] From daa8c232bd668c26bc9d0c12d1378fa5c562ac36 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Sun, 16 Aug 2026 17:13:53 +0530 Subject: [PATCH 06/18] fix(watcher): bound shutdown drain, abort stragglers Unbounded drain could stall a non-interactive shutdown (redeploy SIGTERM) when a task was wedged in an external git/LLM op with no internal timeout. drain_or_abort waits up to GRACEFUL_DRAIN_BUDGET (30s), then aborts remaining tasks explicitly and logs it, rather than stalling or leaving them for runtime teardown. Cancel-safe: handles are joined via &mut so the timeout drops only the borrow, keeping them for the abort pass. --- crates/claudear-engine/src/watcher.rs | 106 +++++++++++++++++++++++--- 1 file changed, 95 insertions(+), 11 deletions(-) diff --git a/crates/claudear-engine/src/watcher.rs b/crates/claudear-engine/src/watcher.rs index 01786a0..a4af2cd 100644 --- a/crates/claudear-engine/src/watcher.rs +++ b/crates/claudear-engine/src/watcher.rs @@ -44,6 +44,11 @@ type QueuedIssue = (Issue, MatchResult, Option); /// on (marked handled) instead of re-triggering the fix agent every cycle. const MAX_REVIEW_COMMENT_ATTEMPTS: i64 = 5; +/// How long graceful shutdown waits for in-flight issue-processing tasks to +/// finish before aborting the stragglers. Bounds shutdown so a task wedged in an +/// external git/LLM operation with no internal timeout cannot stall a redeploy. +const GRACEFUL_DRAIN_BUDGET: std::time::Duration = std::time::Duration::from_secs(30); + /// Extracts the source name from a processing key of the form "source:issue_id". fn source_from_processing_key(key: &str) -> &str { key.split_once(':').map_or(key, |(source, _)| source) @@ -1168,18 +1173,61 @@ impl Watcher { /// all in-progress work completes before the application exits. pub async fn stop_and_drain(&self) { self.stop(); + self.drain_or_abort(GRACEFUL_DRAIN_BUDGET).await; + } + + /// Drain in-flight issue-processing tasks, giving them `budget` to finish + /// gracefully and then aborting any stragglers. + /// + /// `stop()` has already cleared is_running, so no new tasks are recorded. + /// The two failure modes this balances, both flagged in review: + /// - Returning early while a task still runs let runtime teardown abort it + /// mid-operation and misreport a graceful stop. So we wait for real + /// completion within the budget, and any straggler is aborted *explicitly* + /// and logged here rather than left to implicit teardown. + /// - Waiting with no deadline stalled a non-interactive shutdown (e.g. a + /// redeploy's SIGTERM) when a task was wedged in an external git/LLM + /// operation with no internal timeout. So the wait is bounded. + /// + /// Cancel-safe: handles are joined via `&mut`, so the timeout drops only the + /// borrow — the owned `handles` survive for the abort pass, never detached. + async fn drain_or_abort(&self, budget: std::time::Duration) { + // Safe to take: is_running is already false and dispatch re-checks it + // under this same lock, so nothing new is pushed after this take. + let mut handles: Vec<_> = { + let mut guard = self.spawn_handles.lock().await; + std::mem::take(&mut *guard) + }; + + let joined = tokio::time::timeout(budget, async { + for handle in &mut handles { + let _ = handle.await; + } + }) + .await; + + if joined.is_ok() { + tracing::info!("Claude Watcher stopped gracefully"); + return; + } - // Join every in-flight issue-processing task to completion before - // returning. This deliberately has no internal deadline: a fixed budget - // let stop_and_drain return while a task was still running, after which - // production tore down the Tokio runtime and aborted that task - // mid-operation, corrupting a partially-applied fix. `stop()` has already - // cleared is_running, so no new tasks are spawned; each running - // process_issue carries its own internal timeouts, so this terminates. - // Production bounds it by racing this against an operator force-quit - // (a second Ctrl+C -> process::exit), which is the intended hard limit. - self.drain_spawned_tasks().await; - tracing::info!("Claude Watcher stopped gracefully"); + // Budget exhausted: abort the still-running stragglers deterministically + // instead of stalling shutdown or leaving them for runtime teardown. + let mut aborted = 0usize; + for handle in &mut handles { + if !handle.is_finished() { + handle.abort(); + aborted += 1; + } + } + for handle in handles { + let _ = handle.await; + } + tracing::warn!( + aborted, + budget_secs = budget.as_secs(), + "Graceful shutdown budget exhausted; aborted in-flight tasks to avoid stalling shutdown" + ); } /// Check if the watcher is currently running. @@ -5253,6 +5301,42 @@ mod tests { assert!(!watcher.is_running.load(Ordering::SeqCst)); } + #[tokio::test] + async fn test_drain_or_abort_bounds_wedged_task() { + use std::sync::atomic::AtomicBool; + use std::time::{Duration, Instant}; + let notifier = Arc::new(MockNotifier::new(true)); + let tracker = Arc::new(SqliteTracker::in_memory().unwrap()); + let source = Arc::new(MockSource::new("drain")) as Arc; + let watcher = create_test_watcher(notifier, tracker, vec![source], false); + + // A task wedged far longer than the budget (stands in for an external + // git/LLM op with no internal timeout). + let done = Arc::new(AtomicBool::new(false)); + let done_clone = Arc::clone(&done); + { + let mut handles = watcher.spawn_handles.lock().await; + handles.push(tokio::spawn(async move { + tokio::time::sleep(Duration::from_secs(30)).await; + done_clone.store(true, Ordering::SeqCst); + })); + } + + // Shutdown must not stall: it returns within the (tiny) budget, aborting + // the straggler rather than waiting for it or leaving it detached. + let start = Instant::now(); + watcher.drain_or_abort(Duration::from_millis(50)).await; + assert!( + start.elapsed() < Duration::from_secs(2), + "drain_or_abort stalled past its budget" + ); + assert!( + !done.load(Ordering::SeqCst), + "the wedged task was aborted, not awaited to completion" + ); + assert!(watcher.spawn_handles.lock().await.is_empty()); + } + #[test] fn test_watcher_new() { let notifier = Arc::new(MockNotifier::new(true)); From 1ae5bc7725e94dfb63b23ce08740d0499679f56a Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Wed, 19 Aug 2026 10:22:25 +0530 Subject: [PATCH 07/18] added sentry issue classify as error --- crates/claudear-engine/src/processing.rs | 43 ++++++++++++++++++++++++ 1 file changed, 43 insertions(+) diff --git a/crates/claudear-engine/src/processing.rs b/crates/claudear-engine/src/processing.rs index e855ed8..58c54b2 100644 --- a/crates/claudear-engine/src/processing.rs +++ b/crates/claudear-engine/src/processing.rs @@ -2522,6 +2522,12 @@ impl IssueProcessor { /// else (routes to Reply). Uses the LLM classifier when available, falling /// back to the label/source heuristic (matching `FixAttempt::is_bug`). async fn classify_is_bug_or_security(&self, issue: &Issue) -> bool { + // Sentry issues are genuine errors and always route to the fix pipeline; + // enforce that before the classifier, which could otherwise misroute them + // to the QA/reply path (matches heuristic_is_bug / FixAttempt::is_bug). + if issue.source == "sentry" { + return true; + } if let Some(classifier) = self.intent_classifier.as_ref() { // Classify against the reply thread so a follow-up is judged in context, // but only Claudear's own answers — never untrusted user text — feed the @@ -5927,8 +5933,45 @@ mod tests { .is_none()); } + // Reproduces: a Sentry issue reaching classify_is_bug_or_security with an + // active LLM classifier is routed by the classifier verdict, bypassing the + // "sentry is always a bug" invariant that heuristic_is_bug enforces. If the + // classifier calls it a Question, the genuine Sentry error lands on the QA + // (reply) path instead of the fix pipeline. + #[tokio::test] + async fn test_classify_sentry_never_routed_to_qa() { + let tracker: Arc = + Arc::new(claudear_storage::SqliteTracker::in_memory().unwrap()); + let mut processor = make_reply_chain_processor(tracker); + processor.intent_classifier = Some(Arc::new(StubIntentClassifier(Some(Intent::Question)))); + + // No reply_to_message_id metadata, so assemble_reply_chain short-circuits + // to None (no network) and the classifier verdict alone decides routing. + let issue = Issue::new("id-1", "S-1", "NullPointerException", "https://s/1", "sentry"); + + assert!( + processor.classify_is_bug_or_security(&issue).await, + "sentry errors are genuine bugs and must route to the fix pipeline, \ + never QA, even when the classifier calls them a Question" + ); + } + // --- Dummy test helpers --- + /// Intent classifier stub returning a fixed verdict (for routing tests). + struct StubIntentClassifier(Option); + + #[async_trait] + impl IntentClassifier for StubIntentClassifier { + async fn classify_intent( + &self, + _issue: &Issue, + _conversation: Option<&str>, + ) -> Option { + self.0 + } + } + /// Dummy agent runner that does nothing (for IssueProcessor tests). struct DummyAgent; From 16c801099e0c769c36c3e3a60c81540fcacba95c Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Wed, 19 Aug 2026 10:45:00 +0530 Subject: [PATCH 08/18] fix(watcher): track retry/review processing in spawn_handles Retries and review feedback reached process_issue through trigger_issue_with_feedback, which awaited it inline on the housekeeping task. That execution was never recorded in spawn_handles, so it escaped drain_or_abort: the select! in start() could drop it mid-operation, and graceful shutdown reported a clean stop while it was still mutating tracker, notifier, or agent state. Route that path through the same spawn-and-track mechanism as the poll loop. trigger_issue_with_feedback now spawns process_issue into spawn_handles under the same lock drain takes, with the identical is_running re-check, so the work survives the housekeeping future being dropped and is joined by the graceful drain. A oneshot channel carries the started bool back and fires at the end of process_issue, preserving the synchronous contract: the call still blocks to completion and still returns "already being processed" when it did not start, so retry concurrency and the review-wait loop are unchanged. A trigger arriving after stop() is refused instead of leaking a handle past the drain. Propagate self: &Arc through trigger_issue, trigger_issue_with_feedback, process_review_action, check_reviews, process_ready_retries, and run_housekeeping_cycle. Add test_trigger_records_spawn_handle_for_drain and test_trigger_refuses_once_stopped. --- crates/claudear-engine/src/processing.rs | 8 +- crates/claudear-engine/src/watcher.rs | 129 ++++++++++++++++++++--- 2 files changed, 120 insertions(+), 17 deletions(-) diff --git a/crates/claudear-engine/src/processing.rs b/crates/claudear-engine/src/processing.rs index 58c54b2..c2263a1 100644 --- a/crates/claudear-engine/src/processing.rs +++ b/crates/claudear-engine/src/processing.rs @@ -5947,7 +5947,13 @@ mod tests { // No reply_to_message_id metadata, so assemble_reply_chain short-circuits // to None (no network) and the classifier verdict alone decides routing. - let issue = Issue::new("id-1", "S-1", "NullPointerException", "https://s/1", "sentry"); + let issue = Issue::new( + "id-1", + "S-1", + "NullPointerException", + "https://s/1", + "sentry", + ); assert!( processor.classify_is_bug_or_security(&issue).await, diff --git a/crates/claudear-engine/src/watcher.rs b/crates/claudear-engine/src/watcher.rs index a4af2cd..a31e466 100644 --- a/crates/claudear-engine/src/watcher.rs +++ b/crates/claudear-engine/src/watcher.rs @@ -1290,7 +1290,7 @@ impl Watcher { /// /// This polls the ReviewWatcher for any new CHANGES_REQUESTED or COMMENTED reviews /// and triggers Claude to address the feedback. - pub async fn check_reviews(&self) -> Result<()> { + pub async fn check_reviews(self: &Arc) -> Result<()> { let review_watcher = match &self.review_watcher { Some(rw) => rw, None => return Ok(()), @@ -1433,7 +1433,7 @@ impl Watcher { /// This creates a new Claude session with the original issue context plus /// the review feedback appended to help Claude understand what to fix. async fn process_review_action( - &self, + self: &Arc, attempt: &claudear_core::types::FixAttempt, feedback: &str, ) -> Result<()> { @@ -2466,7 +2466,7 @@ Create a PR with your changes.{custom_instructions}"#, /// Run housekeeping tasks: retries, cascades, and metrics. /// Called on the global timer, separate from per-source polling. - pub async fn run_housekeeping_cycle(&self) -> Result<()> { + pub async fn run_housekeeping_cycle(self: &Arc) -> Result<()> { let housekeeping_started_at = std::time::Instant::now(); // Run retries, PR merge cascades, and release cascades concurrently @@ -2533,7 +2533,7 @@ Create a PR with your changes.{custom_instructions}"#, } /// Process any issues that are ready for retry. - async fn process_ready_retries(&self) -> Result<()> { + async fn process_ready_retries(self: &Arc) -> Result<()> { // Skip retries while paused for rate limits — attempting them would // just burn retry attempts without doing any work. if self.is_rate_limit_paused().await { @@ -4605,7 +4605,7 @@ Create a PR with your changes.{custom_instructions}"#, } /// Manually trigger processing for a specific issue. - pub async fn trigger_issue(&self, source_name: &str, issue_id: &str) -> Result<()> { + pub async fn trigger_issue(self: &Arc, source_name: &str, issue_id: &str) -> Result<()> { self.trigger_issue_with_feedback( source_name, issue_id, @@ -4618,7 +4618,7 @@ Create a PR with your changes.{custom_instructions}"#, /// Manually trigger processing for a specific issue with optional review feedback context. pub async fn trigger_issue_with_feedback( - &self, + self: &Arc, source_name: &str, issue_id: &str, review_feedback: Option, @@ -4645,16 +4645,52 @@ Create a PR with your changes.{custom_instructions}"#, issue.set_metadata("trigger_reason", reason); } - let started = self - .process_issue( - Arc::clone(source), - issue, - match_result, - review_feedback, - existing_pr_branch, - None, - ) - .await; + // Run processing as a tracked background task rather than inline. Retries + // and review-feedback both reach this path from the housekeeping loop; if + // process_issue ran inline it would be part of that loop's future and get + // aborted mid-operation on shutdown (the select! in start() drops the + // losing branch, and drain_or_abort only joins spawn_handles). Spawning + // into spawn_handles makes the work survive the drop and be joined by the + // graceful drain, exactly like the poll-loop dispatch path. + // + // `started_rx` preserves the original synchronous contract: it fires at + // the end of process_issue, so awaiting it here blocks until processing + // completes and yields whether it actually began. If the caller is dropped + // on shutdown, the spawned task keeps running and drain joins it. + let (started_tx, started_rx) = tokio::sync::oneshot::channel(); + let watcher = Arc::clone(self); + let source = Arc::clone(source); + { + let mut handles = self.spawn_handles.lock().await; + // Re-check is_running under the same lock drain_or_abort takes, then + // push atomically. This closes the shutdown race: a trigger either + // records its handle before the drain's take (so shutdown joins it) + // or sees the stop here and never spawns. + if !self.is_running.load(Ordering::SeqCst) { + return Err(claudear_core::error::Error::source( + source_name, + format!("Watcher stopping; trigger for {} not started", issue_id), + )); + } + handles.retain(|h| !h.is_finished()); + handles.push(tokio::spawn(async move { + let started = watcher + .process_issue( + source, + issue, + match_result, + review_feedback, + existing_pr_branch, + None, + ) + .await; + let _ = started_tx.send(started); + })); + } + + // A dropped sender (task aborted by drain before it finished) reports + // not-started; during shutdown the caller is typically already gone. + let started = started_rx.await.unwrap_or(false); if !started { return Err(claudear_core::error::Error::source( source_name, @@ -6801,6 +6837,9 @@ mod tests { let sources = vec![source]; let watcher = create_test_watcher(notifier, tracker, sources, true); // dry run + // Manual triggers only arrive while the daemon is running; the trigger now + // spawns tracked work and refuses once the watcher is stopping. + watcher.is_running.store(true, Ordering::SeqCst); let result = watcher.trigger_issue("mock", "123").await; // Should succeed in dry run (doesn't actually process) @@ -6823,6 +6862,7 @@ mod tests { let sources = vec![source]; let watcher = create_test_watcher(notifier, tracker, sources, true); + watcher.is_running.store(true, Ordering::SeqCst); { let mut processing = watcher.processing.write().await; processing.insert("mock:123".to_string()); @@ -6836,6 +6876,63 @@ mod tests { .contains("already being processed")); } + /// Retries and review feedback trigger processing through + /// [`Watcher::trigger_issue_with_feedback`]. That processing must run as a + /// tracked task in `spawn_handles` so graceful shutdown drains it, rather + /// than inline where the housekeeping loop's cancellation would abort it + /// mid-operation. Regression for the inline-processing-escapes-drain bug. + #[tokio::test] + async fn test_trigger_records_spawn_handle_for_drain() { + let notifier = Arc::new(MockNotifier::new(true)); + let tracker = Arc::new(SqliteTracker::in_memory().unwrap()); + + let issues = vec![Issue::new( + "123", + "T-123", + "Test Issue", + "http://example.com/123", + "mock", + )]; + let source = Arc::new(MockSource::with_issues("mock", issues)) as Arc; + + let watcher = create_test_watcher(notifier, tracker, vec![source], true); // dry run + watcher.is_running.store(true, Ordering::SeqCst); + + assert!(watcher.spawn_handles.lock().await.is_empty()); + + watcher.trigger_issue("mock", "123").await.unwrap(); + + // The processing task was recorded in spawn_handles (retained until the + // next reap), so drain_or_abort will join it on shutdown. + assert_eq!(watcher.spawn_handles.lock().await.len(), 1); + } + + /// A trigger that arrives after the watcher has stopped must not spawn new + /// tracked work, or it could push a handle into spawn_handles after + /// drain_or_abort already took the vector and be missed by the drain. + #[tokio::test] + async fn test_trigger_refuses_once_stopped() { + let notifier = Arc::new(MockNotifier::new(true)); + let tracker = Arc::new(SqliteTracker::in_memory().unwrap()); + + let issues = vec![Issue::new( + "123", + "T-123", + "Test Issue", + "http://example.com/123", + "mock", + )]; + let source = Arc::new(MockSource::with_issues("mock", issues)) as Arc; + + let watcher = create_test_watcher(notifier, tracker, vec![source], true); + // is_running left false: watcher is stopping / not started. + + let result = watcher.trigger_issue("mock", "123").await; + assert!(result.is_err()); + assert!(result.unwrap_err().to_string().contains("stopping")); + assert!(watcher.spawn_handles.lock().await.is_empty()); + } + #[test] fn test_watcher_processing_set() { let notifier = Arc::new(MockNotifier::new(true)); From 6b310e3e024721a3d3b8ae867f7678922aa1484f Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Wed, 19 Aug 2026 10:53:15 +0530 Subject: [PATCH 09/18] fix(cli): mark one-shot trigger watchers running trigger_issue now spawns tracked processing and refuses once the watcher is stopping, so the one-shot retry and Trigger CLI paths must mark their never-started watchers running or every trigger returns a stopping error. Also Arc-wrap the retry-path watcher for the &Arc receiver. --- src/main.rs | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/src/main.rs b/src/main.rs index e1b99ea..1c59c35 100644 --- a/src/main.rs +++ b/src/main.rs @@ -3810,7 +3810,7 @@ async fn async_main(cli: Cli) -> anyhow::Result<()> { tracker.clone(), ))); - let watcher = Watcher::new(WatcherOptions { + let watcher = Arc::new(Watcher::new(WatcherOptions { config: config.clone(), sources, notifier, @@ -3832,7 +3832,10 @@ async fn async_main(cli: Cli) -> anyhow::Result<()> { qa_agent: None, dry_run: false, llm_engine: None, - }); + })); + // One-shot retry: trigger_issue spawns tracked processing and refuses + // once the watcher is stopping, so mark it running for the duration. + watcher.set_running(true); for attempt in ready { println!("\n Retrying [{}] {}...", attempt.source, attempt.short_id); @@ -4362,6 +4365,9 @@ async fn async_main(cli: Cli) -> anyhow::Result<()> { } Commands::Trigger { source, issue_id } => { + // trigger_issue spawns tracked processing and refuses once the + // watcher is stopping, so mark it running for this one-shot. + watcher.set_running(true); watcher.trigger_issue(&source, &issue_id).await?; } From 1f317803b4c8dd3501e35224980fa7c3a21fdfae Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Wed, 19 Aug 2026 11:01:48 +0530 Subject: [PATCH 10/18] fix(watcher): bound the post-abort join in shutdown drain After the graceful budget is exhausted the drain aborted stragglers and then awaited every handle to completion, unbounded. abort() only takes effect at a task's next await, so a task parked in synchronous blocking work (block_in_place bridging an LLM call) never observes it until that call returns, and the join blocked until then, re-introducing the unbounded shutdown wait the budget exists to prevent. Cap the post-abort join with a short grace joined via &mut so the timeout drops only the borrow. Handles still unfinished are dropped and left to the imminent runtime teardown, and the straggler is logged as detached. Add test_drain_or_abort_bounds_post_abort_join_for_blocking_task. --- crates/claudear-engine/src/watcher.rs | 75 +++++++++++++++++++++++++-- 1 file changed, 72 insertions(+), 3 deletions(-) diff --git a/crates/claudear-engine/src/watcher.rs b/crates/claudear-engine/src/watcher.rs index a31e466..f04827e 100644 --- a/crates/claudear-engine/src/watcher.rs +++ b/crates/claudear-engine/src/watcher.rs @@ -49,6 +49,15 @@ const MAX_REVIEW_COMMENT_ATTEMPTS: i64 = 5; /// external git/LLM operation with no internal timeout cannot stall a redeploy. const GRACEFUL_DRAIN_BUDGET: std::time::Duration = std::time::Duration::from_secs(30); +/// Grace given to aborted stragglers to actually terminate before the drain +/// stops waiting on them. `abort()` only takes effect at a task's next await, so +/// a task parked in synchronous blocking work (`block_in_place` bridging an LLM +/// call) won't observe it until that call returns. Capping the post-abort join +/// keeps shutdown bounded instead of re-introducing the unbounded wait the +/// budget exists to prevent; any handle still unfinished is dropped and left to +/// runtime teardown, which is imminent on the shutdown path anyway. +const ABORT_JOIN_GRACE: std::time::Duration = std::time::Duration::from_secs(2); + /// Extracts the source name from a processing key of the form "source:issue_id". fn source_from_processing_key(key: &str) -> &str { key.split_once(':').map_or(key, |(source, _)| source) @@ -1220,11 +1229,26 @@ impl Watcher { aborted += 1; } } - for handle in handles { - let _ = handle.await; - } + // Bound the post-abort join too. Aborting a task parked in synchronous + // blocking work (block_in_place around an LLM call) does not preempt it, + // so an unbounded await here would let a wedged operation stall shutdown + // despite the budget. Joined via &mut so the timeout drops only the + // borrow; handles still unfinished are dropped when the vec goes out of + // scope, detached for imminent runtime teardown. + let reaped = tokio::time::timeout(ABORT_JOIN_GRACE, async { + for handle in &mut handles { + let _ = handle.await; + } + }) + .await; + let detached = if reaped.is_ok() { + 0 + } else { + handles.iter().filter(|h| !h.is_finished()).count() + }; tracing::warn!( aborted, + detached, budget_secs = budget.as_secs(), "Graceful shutdown budget exhausted; aborted in-flight tasks to avoid stalling shutdown" ); @@ -5373,6 +5397,51 @@ mod tests { assert!(watcher.spawn_handles.lock().await.is_empty()); } + /// A task parked in synchronous blocking work does not observe `abort()` + /// until it returns, so the post-abort join must be bounded too or a wedged + /// LLM call stalls shutdown despite the budget. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn test_drain_or_abort_bounds_post_abort_join_for_blocking_task() { + use std::time::{Duration, Instant}; + let notifier = Arc::new(MockNotifier::new(true)); + let tracker = Arc::new(SqliteTracker::in_memory().unwrap()); + let source = Arc::new(MockSource::new("drain")) as Arc; + let watcher = create_test_watcher(notifier, tracker, vec![source], false); + + // Blocking recv() never yields to the scheduler, so aborting the task has + // no effect until it is released — a stand-in for block_in_place bridging + // a wedged LLM call. + let (release_tx, release_rx) = std::sync::mpsc::channel::<()>(); + { + let mut handles = watcher.spawn_handles.lock().await; + handles.push(tokio::spawn(async move { + let _ = release_rx.recv(); + })); + } + + // With an unbounded post-abort join this only returns once the task is + // released; the abort-join grace must cap it instead. + let start = Instant::now(); + let drained = tokio::time::timeout( + Duration::from_secs(20), + watcher.drain_or_abort(Duration::from_millis(50)), + ) + .await; + let elapsed = start.elapsed(); + // Release the wedged worker so runtime teardown is prompt regardless. + let _ = release_tx.send(()); + + assert!( + drained.is_ok(), + "drain_or_abort never returned; post-abort join is unbounded for a \ + task wedged in non-preemptible blocking work" + ); + assert!( + elapsed < ABORT_JOIN_GRACE + Duration::from_secs(3), + "post-abort join stalled shutdown past the abort grace" + ); + } + #[test] fn test_watcher_new() { let notifier = Arc::new(MockNotifier::new(true)); From 4937e762b952365197a734d26050b2946c1677ba Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Wed, 19 Aug 2026 11:18:39 +0530 Subject: [PATCH 11/18] fix(cli): bound tokio runtime teardown on shutdown The watcher drain aborts stragglers and detaches any parked in non-preemptible blocking work, but dropping the multi-thread runtime by default waits indefinitely for that work to return, so an orphaned block_in_place LLM call could still stall a redeploy's SIGTERM during runtime teardown. Own the runtime explicitly and shut it down with a 5s timeout after block_on returns, detaching whatever blocking work remains for process exit to reclaim. Keeps the sentry and logging flush guards intact. --- crates/claudear-engine/src/watcher.rs | 3 ++- src/main.rs | 16 +++++++++++++--- 2 files changed, 15 insertions(+), 4 deletions(-) diff --git a/crates/claudear-engine/src/watcher.rs b/crates/claudear-engine/src/watcher.rs index f04827e..6881829 100644 --- a/crates/claudear-engine/src/watcher.rs +++ b/crates/claudear-engine/src/watcher.rs @@ -55,7 +55,8 @@ const GRACEFUL_DRAIN_BUDGET: std::time::Duration = std::time::Duration::from_sec /// call) won't observe it until that call returns. Capping the post-abort join /// keeps shutdown bounded instead of re-introducing the unbounded wait the /// budget exists to prevent; any handle still unfinished is dropped and left to -/// runtime teardown, which is imminent on the shutdown path anyway. +/// runtime teardown, which `main` bounds via `Runtime::shutdown_timeout` so an +/// orphaned blocking op cannot stall process exit either. const ABORT_JOIN_GRACE: std::time::Duration = std::time::Duration::from_secs(2); /// Extracts the source name from a processing key of the form "source:issue_id". diff --git a/src/main.rs b/src/main.rs index 1c59c35..82d2b39 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1695,10 +1695,20 @@ fn main() -> anyhow::Result<()> { scope.set_tag("app.component", "claudear-backend"); }); - tokio::runtime::Builder::new_multi_thread() + let runtime = tokio::runtime::Builder::new_multi_thread() .enable_all() - .build()? - .block_on(async_main(cli)) + .build()?; + let result = runtime.block_on(async_main(cli)); + + // Bound runtime teardown. The watcher's shutdown drain aborts stragglers and + // detaches any still parked in non-preemptible blocking work (block_in_place + // bridging a wedged LLM call), but dropping the runtime by default waits + // indefinitely for such work to return, so an orphaned operation could still + // stall a redeploy's SIGTERM. shutdown_timeout caps that wait and detaches + // whatever remains, which the process exit then reclaims. + runtime.shutdown_timeout(std::time::Duration::from_secs(5)); + + result } async fn async_main(cli: Cli) -> anyhow::Result<()> { From 764b8c2305db53b063d17abbe67a16424c0a5ed5 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Wed, 19 Aug 2026 11:21:53 +0530 Subject: [PATCH 12/18] test(e2e): make harness watcher Arc and mark it running trigger_issue now takes &Arc and refuses once the watcher is stopping, so the e2e harness stores an Arc and marks it running for its one-shot trigger calls. Fixes the integration-test compile break caught by tarpaulin. --- tests/e2e_real_repo.rs | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/tests/e2e_real_repo.rs b/tests/e2e_real_repo.rs index 9743497..748d91c 100644 --- a/tests/e2e_real_repo.rs +++ b/tests/e2e_real_repo.rs @@ -162,7 +162,7 @@ struct E2eHarness { source: Arc, notifier: Arc, tracker: Arc, - watcher: Watcher, + watcher: Arc, } fn run_git(cwd: &Path, args: &[&str]) { @@ -359,7 +359,7 @@ fn create_harness(tasks: Vec) -> E2eHarness { tracker.clone(), )); - let watcher = Watcher::new(WatcherOptions { + let watcher = Arc::new(Watcher::new(WatcherOptions { config: build_config(&temp_dir), sources: vec![source.clone() as Arc], notifier: notifier.clone() as Arc, @@ -381,7 +381,10 @@ fn create_harness(tasks: Vec) -> E2eHarness { qa_agent: None, dry_run: false, llm_engine: None, - }); + })); + // trigger_issue spawns tracked processing and refuses once the watcher is + // stopping, so mark it running for these one-shot triggers. + watcher.set_running(true); E2eHarness { _temp_dir: temp_dir, From fa8f7244d0bfd85a51828023b2603be737ad098e Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Wed, 19 Aug 2026 11:30:17 +0530 Subject: [PATCH 13/18] fix(watcher): skip finished handles in post-abort drain join The graceful drain pass polls handles via &mut and can complete an early handle before timing out on a later wedged one. The post-abort pass then iterated all handles again and awaited the already-completed one, and awaiting a JoinHandle after it returned Ready panics with "JoinHandle polled after completion", crashing shutdown. Skip finished handles in the post-abort join so only still-running tasks are awaited. Add test_drain_or_abort_no_double_poll_of_completed_handle, which panics without the guard. --- crates/claudear-engine/src/watcher.rs | 50 +++++++++++++++++++++++++++ 1 file changed, 50 insertions(+) diff --git a/crates/claudear-engine/src/watcher.rs b/crates/claudear-engine/src/watcher.rs index 6881829..a54e411 100644 --- a/crates/claudear-engine/src/watcher.rs +++ b/crates/claudear-engine/src/watcher.rs @@ -1236,8 +1236,15 @@ impl Watcher { // despite the budget. Joined via &mut so the timeout drops only the // borrow; handles still unfinished are dropped when the vec goes out of // scope, detached for imminent runtime teardown. + // + // Skip already-finished handles: the graceful pass may have polled some + // to completion before it timed out on a later one, and awaiting a + // JoinHandle again after it returned `Ready` panics. let reaped = tokio::time::timeout(ABORT_JOIN_GRACE, async { for handle in &mut handles { + if handle.is_finished() { + continue; + } let _ = handle.await; } }) @@ -5443,6 +5450,49 @@ mod tests { ); } + /// The graceful pass may poll some handles to completion before it times out + /// on a later one; the post-abort pass must not re-poll those, as awaiting a + /// `JoinHandle` again after it returned `Ready` panics. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn test_drain_or_abort_no_double_poll_of_completed_handle() { + use std::time::Duration; + let notifier = Arc::new(MockNotifier::new(true)); + let tracker = Arc::new(SqliteTracker::in_memory().unwrap()); + let source = Arc::new(MockSource::new("drain")) as Arc; + let watcher = create_test_watcher(notifier, tracker, vec![source], false); + + // Ordering is the point: a completed handle precedes a wedged one, so the + // graceful pass polls the first to completion and then times out on the + // second. + let done = tokio::spawn(async {}); + while !done.is_finished() { + tokio::time::sleep(Duration::from_millis(5)).await; + } + let (release_tx, release_rx) = std::sync::mpsc::channel::<()>(); + let wedged = tokio::spawn(async move { + let _ = release_rx.recv(); + }); + { + let mut handles = watcher.spawn_handles.lock().await; + handles.push(done); + handles.push(wedged); + } + + // Must return (not panic) despite the completed-then-wedged ordering. + let drained = tokio::time::timeout( + Duration::from_secs(20), + watcher.drain_or_abort(Duration::from_millis(50)), + ) + .await; + // Release the wedged worker so runtime teardown is prompt. + let _ = release_tx.send(()); + + assert!( + drained.is_ok(), + "drain_or_abort stalled or panicked re-polling a completed handle" + ); + } + #[test] fn test_watcher_new() { let notifier = Arc::new(MockNotifier::new(true)); From ca2acba0c92ff08f949a3fa80de6a0d737ad2d68 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Thu, 20 Aug 2026 15:06:37 +0530 Subject: [PATCH 14/18] chore(scripts): add claudear resource footprint monitor --- scripts/monitor_claudear.sh | 184 ++++++++++++++++++++++++++++++++++++ 1 file changed, 184 insertions(+) create mode 100755 scripts/monitor_claudear.sh diff --git a/scripts/monitor_claudear.sh b/scripts/monitor_claudear.sh new file mode 100755 index 0000000..3a617a7 --- /dev/null +++ b/scripts/monitor_claudear.sh @@ -0,0 +1,184 @@ +#!/usr/bin/env bash +# +# monitor_claudear.sh - sample the claudear daemon's resource footprint over +# time so a slow leak / runaway spawn can be traced. +# +# Runs on Linux (reads /proc). Meant for the droplet where claudear runs as: +# /usr/bin/claudear --config claudear.toml start --poll +# +# It appends one CSV row per interval to a central log (default ./memory.log): +# timestamp,pid,rss_mb,peak_rss_mb,vsz_mb,threads,fds,cpu_pct, +# descendants,claude_procs,git_procs,sys_mem_used_pct,sys_mem_avail_mb,load1 +# +# What each column tells you when it climbs monotonically: +# rss_mb / peak_rss_mb -> memory leak (the retained-task / handle growth) +# threads -> tokio blocking-pool growth (spawn_blocking piling up) +# fds -> leaked file descriptors / sockets / worktrees +# descendants/claude/git-> child processes not being reaped +# +# Usage: +# ./monitor_claudear.sh start # loop in foreground (nohup it for a droplet) +# ./monitor_claudear.sh once # take a single sample and exit +# ./monitor_claudear.sh status # show whether a monitor is running +# ./monitor_claudear.sh stop # stop a running monitor (via pidfile) +# +# Config via env vars: +# INTERVAL seconds between samples (default 60) +# LOG output CSV path (default ./memory.log) +# PIDFILE monitor pidfile (default ./claudear-monitor.pid) +# MATCH pgrep pattern for the daemon (default the start --poll cmd) +# THRESHOLD_MB if set, log a WARN line when rss exceeds this +# +set -u + +INTERVAL="${INTERVAL:-60}" +LOG="${LOG:-./memory.log}" +PIDFILE="${PIDFILE:-./claudear-monitor.pid}" +MATCH="${MATCH:-claudear.*start}" +THRESHOLD_MB="${THRESHOLD_MB:-}" + +HEADER="timestamp,pid,rss_mb,peak_rss_mb,vsz_mb,threads,fds,cpu_pct,descendants,claude_procs,git_procs,sys_mem_used_pct,sys_mem_avail_mb,load1" + +now() { date '+%Y-%m-%dT%H:%M:%S%z'; } + +# Find the claudear daemon pid. Prefer an exact process-name match, fall back to +# the command-line pattern. Returns empty string if not running. +find_pid() { + local pid + pid="$(pgrep -x claudear 2>/dev/null | head -n1)" + if [ -z "$pid" ]; then + pid="$(pgrep -f "$MATCH" 2>/dev/null | grep -v "$$" | head -n1)" + fi + printf '%s' "$pid" +} + +# Recursively list all descendant pids of $1, one per line. +list_descendants() { + local pid="$1" child + for child in $(pgrep -P "$pid" 2>/dev/null); do + printf '%s\n' "$child" + list_descendants "$child" + done +} + +kb_to_mb() { awk -v k="${1:-0}" 'BEGIN{ printf "%.1f", (k+0)/1024 }'; } + +sample() { + local pid ts + ts="$(now)" + pid="$(find_pid)" + + if [ -z "$pid" ] || [ ! -d "/proc/$pid" ]; then + # Daemon not running right now - record the gap so a crash/restart is visible. + printf '%s,,,,,,,,,,,,,\n' "$ts" >>"$LOG" + echo "$ts claudear not running" + return + fi + + local vmrss vmhwm vmsize threads fds cpu + vmrss="$(awk '/^VmRSS:/{print $2}' "/proc/$pid/status" 2>/dev/null)" + vmhwm="$(awk '/^VmHWM:/{print $2}' "/proc/$pid/status" 2>/dev/null)" + vmsize="$(awk '/^VmSize:/{print $2}' "/proc/$pid/status" 2>/dev/null)" + threads="$(awk '/^Threads:/{print $2}' "/proc/$pid/status" 2>/dev/null)" + fds="$(ls -1 "/proc/$pid/fd" 2>/dev/null | wc -l | tr -d ' ')" + cpu="$(ps -o %cpu= -p "$pid" 2>/dev/null | tr -d ' ')" + + local rss_mb peak_mb vsz_mb + rss_mb="$(kb_to_mb "$vmrss")" + peak_mb="$(kb_to_mb "$vmhwm")" + vsz_mb="$(kb_to_mb "$vmsize")" + + # Descendant process accounting. + local desc descendants=0 claude_procs=0 git_procs=0 k comm + desc="$(list_descendants "$pid")" + for k in $desc; do + descendants=$((descendants + 1)) + comm="$(cat "/proc/$k/comm" 2>/dev/null)" + case "$comm" in + claude*) claude_procs=$((claude_procs + 1)) ;; + git*) git_procs=$((git_procs + 1)) ;; + esac + done + + # System memory + load. + local memtotal memavail used_pct avail_mb load1 + memtotal="$(awk '/^MemTotal:/{print $2}' /proc/meminfo 2>/dev/null)" + memavail="$(awk '/^MemAvailable:/{print $2}' /proc/meminfo 2>/dev/null)" + avail_mb="$(kb_to_mb "$memavail")" + used_pct="$(awk -v t="${memtotal:-0}" -v a="${memavail:-0}" \ + 'BEGIN{ if (t>0) printf "%.1f", (t-a)/t*100; else printf "" }')" + load1="$(awk '{print $1}' /proc/loadavg 2>/dev/null)" + + printf '%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s\n' \ + "$ts" "$pid" "$rss_mb" "$peak_mb" "$vsz_mb" "$threads" "$fds" "$cpu" \ + "$descendants" "$claude_procs" "$git_procs" "$used_pct" "$avail_mb" "$load1" \ + >>"$LOG" + + echo "$ts pid=$pid rss=${rss_mb}MB peak=${peak_mb}MB threads=$threads fds=$fds desc=$descendants (claude=$claude_procs git=$git_procs) cpu=${cpu}%" + + if [ -n "$THRESHOLD_MB" ]; then + awk -v r="$rss_mb" -v t="$THRESHOLD_MB" 'BEGIN{ exit !(r+0 > t+0) }' && { + echo "$ts,WARN,rss ${rss_mb}MB exceeded threshold ${THRESHOLD_MB}MB,pid=$pid" >>"$LOG" + echo "$ts WARN rss ${rss_mb}MB > ${THRESHOLD_MB}MB" + } + fi +} + +ensure_header() { + if [ ! -s "$LOG" ]; then + echo "$HEADER" >>"$LOG" + fi +} + +cmd_start() { + ensure_header + echo "$$" >"$PIDFILE" + trap 'rm -f "$PIDFILE"; echo "monitor stopped"; exit 0' INT TERM + echo "monitoring claudear every ${INTERVAL}s -> $LOG (monitor pid $$)" + while true; do + sample + sleep "$INTERVAL" + done +} + +cmd_once() { + ensure_header + sample +} + +cmd_status() { + if [ -f "$PIDFILE" ] && kill -0 "$(cat "$PIDFILE" 2>/dev/null)" 2>/dev/null; then + echo "monitor running (pid $(cat "$PIDFILE"))" + else + echo "monitor not running" + fi + local dpid + dpid="$(find_pid)" + if [ -n "$dpid" ]; then + echo "claudear daemon pid: $dpid" + else + echo "claudear daemon: not running" + fi +} + +cmd_stop() { + if [ -f "$PIDFILE" ]; then + local p; p="$(cat "$PIDFILE" 2>/dev/null)" + if [ -n "$p" ] && kill -0 "$p" 2>/dev/null; then + kill "$p" && echo "stopped monitor (pid $p)" + else + echo "no live monitor for pidfile; cleaning up" + fi + rm -f "$PIDFILE" + else + echo "no pidfile at $PIDFILE" + fi +} + +case "${1:-start}" in + start) cmd_start ;; + once) cmd_once ;; + status) cmd_status ;; + stop) cmd_stop ;; + *) echo "usage: $0 {start|once|status|stop}"; exit 2 ;; +esac From 8ba53a008edbd997c9b72083f089c4a0bf6a51f9 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Thu, 20 Aug 2026 17:45:51 +0530 Subject: [PATCH 15/18] feat(scripts): auto-restart claudear from the monitor when it dies --- scripts/monitor_claudear.sh | 72 +++++++++++++++++++++++++++++++++++++ 1 file changed, 72 insertions(+) diff --git a/scripts/monitor_claudear.sh b/scripts/monitor_claudear.sh index 3a617a7..bd7a1cf 100755 --- a/scripts/monitor_claudear.sh +++ b/scripts/monitor_claudear.sh @@ -22,6 +22,14 @@ # ./monitor_claudear.sh status # show whether a monitor is running # ./monitor_claudear.sh stop # stop a running monitor (via pidfile) # +# Monitor + auto-restart (relaunches the daemon whenever it is found dead): +# RESTART=1 WORKDIR="$HOME/workspace" LOG=/var/log/claudear-memory.log \ +# nohup ./monitor_claudear.sh start >/var/log/claudear-monitor.out 2>&1 & +# # equivalent daemon launch: cd $WORKDIR && \ +# # /usr/bin/claudear --config claudear.toml start --poll +# # NOTE: with RESTART=1 the monitor supervises the daemon, so stop the monitor +# # (./monitor_claudear.sh stop) BEFORE stopping claudear, or it will respawn. +# # Config via env vars: # INTERVAL seconds between samples (default 60) # LOG output CSV path (default ./memory.log) @@ -29,6 +37,16 @@ # MATCH pgrep pattern for the daemon (default the start --poll cmd) # THRESHOLD_MB if set, log a WARN line when rss exceeds this # +# Supervision (opt-in) - restart the daemon when it is found not running: +# RESTART=1 enable auto-restart (default off) +# WORKDIR cwd for the launch (default $HOME/workspace) +# CLAUDEAR_BIN binary path (default /usr/bin/claudear) +# CONFIG --config value (default claudear.toml) +# START_ARGS args after --config (default "start --poll") +# DAEMON_LOG daemon stdout/stderr log (default $WORKDIR/claudear-daemon.out) +# RUN_AS run the daemon as this user (default: current user) +# RESTART_COOLDOWN min seconds between restarts (default 30) +# set -u INTERVAL="${INTERVAL:-60}" @@ -37,6 +55,18 @@ PIDFILE="${PIDFILE:-./claudear-monitor.pid}" MATCH="${MATCH:-claudear.*start}" THRESHOLD_MB="${THRESHOLD_MB:-}" +RESTART="${RESTART:-0}" +WORKDIR="${WORKDIR:-$HOME/workspace}" +CLAUDEAR_BIN="${CLAUDEAR_BIN:-/usr/bin/claudear}" +CONFIG="${CONFIG:-claudear.toml}" +START_ARGS="${START_ARGS:-start --poll}" +DAEMON_LOG="${DAEMON_LOG:-$WORKDIR/claudear-daemon.out}" +RUN_AS="${RUN_AS:-}" +RESTART_COOLDOWN="${RESTART_COOLDOWN:-30}" + +LAST_RESTART=0 +RESTART_COUNT=0 + HEADER="timestamp,pid,rss_mb,peak_rss_mb,vsz_mb,threads,fds,cpu_pct,descendants,claude_procs,git_procs,sys_mem_used_pct,sys_mem_avail_mb,load1" now() { date '+%Y-%m-%dT%H:%M:%S%z'; } @@ -63,6 +93,45 @@ list_descendants() { kb_to_mb() { awk -v k="${1:-0}" 'BEGIN{ printf "%.1f", (k+0)/1024 }'; } +# Relaunch the daemon. `claudear start` (without --foreground) self-daemonizes, +# and startup cleans up stale pid/socket files from the crashed run, so this is +# safe to call whenever no live daemon is found. Cooldown-guarded to avoid a +# tight restart loop when the daemon dies immediately on boot. +restart_daemon() { + local ts now + ts="$(now)" + now="$(date +%s)" + if [ $((now - LAST_RESTART)) -lt "$RESTART_COOLDOWN" ]; then + echo "$ts restart suppressed (within ${RESTART_COOLDOWN}s cooldown)" + return + fi + if [ ! -x "$CLAUDEAR_BIN" ]; then + echo "$ts,ERROR,cannot restart: $CLAUDEAR_BIN not executable,," >>"$LOG" + echo "$ts ERROR: $CLAUDEAR_BIN not executable" + return + fi + if [ ! -d "$WORKDIR" ]; then + echo "$ts,ERROR,cannot restart: WORKDIR $WORKDIR missing,," >>"$LOG" + echo "$ts ERROR: WORKDIR $WORKDIR missing" + return + fi + + LAST_RESTART="$now" + RESTART_COUNT=$((RESTART_COUNT + 1)) + echo "$ts,RESTART,#$RESTART_COUNT launching $CLAUDEAR_BIN --config $CONFIG $START_ARGS,workdir=$WORKDIR," >>"$LOG" + echo "$ts RESTART #$RESTART_COUNT: $CLAUDEAR_BIN --config $CONFIG $START_ARGS (cwd $WORKDIR)" + + # Launch from WORKDIR so the relative --config path resolves, detached from the + # monitor so it survives the monitor stopping. + if [ -n "$RUN_AS" ]; then + ( cd "$WORKDIR" && setsid runuser -u "$RUN_AS" -- \ + "$CLAUDEAR_BIN" --config "$CONFIG" $START_ARGS >>"$DAEMON_LOG" 2>&1 & ) + else + ( cd "$WORKDIR" && setsid \ + "$CLAUDEAR_BIN" --config "$CONFIG" $START_ARGS >>"$DAEMON_LOG" 2>&1 & ) + fi +} + sample() { local pid ts ts="$(now)" @@ -72,6 +141,9 @@ sample() { # Daemon not running right now - record the gap so a crash/restart is visible. printf '%s,,,,,,,,,,,,,\n' "$ts" >>"$LOG" echo "$ts claudear not running" + if [ "$RESTART" = "1" ]; then + restart_daemon + fi return fi From 9e48e1dc4690167d6e31461ef7595733298efe41 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Thu, 20 Aug 2026 17:48:21 +0530 Subject: [PATCH 16/18] revert(scripts): keep monitor pure-observation, decoupled from daemon lifecycle --- scripts/monitor_claudear.sh | 72 ------------------------------------- 1 file changed, 72 deletions(-) diff --git a/scripts/monitor_claudear.sh b/scripts/monitor_claudear.sh index bd7a1cf..3a617a7 100755 --- a/scripts/monitor_claudear.sh +++ b/scripts/monitor_claudear.sh @@ -22,14 +22,6 @@ # ./monitor_claudear.sh status # show whether a monitor is running # ./monitor_claudear.sh stop # stop a running monitor (via pidfile) # -# Monitor + auto-restart (relaunches the daemon whenever it is found dead): -# RESTART=1 WORKDIR="$HOME/workspace" LOG=/var/log/claudear-memory.log \ -# nohup ./monitor_claudear.sh start >/var/log/claudear-monitor.out 2>&1 & -# # equivalent daemon launch: cd $WORKDIR && \ -# # /usr/bin/claudear --config claudear.toml start --poll -# # NOTE: with RESTART=1 the monitor supervises the daemon, so stop the monitor -# # (./monitor_claudear.sh stop) BEFORE stopping claudear, or it will respawn. -# # Config via env vars: # INTERVAL seconds between samples (default 60) # LOG output CSV path (default ./memory.log) @@ -37,16 +29,6 @@ # MATCH pgrep pattern for the daemon (default the start --poll cmd) # THRESHOLD_MB if set, log a WARN line when rss exceeds this # -# Supervision (opt-in) - restart the daemon when it is found not running: -# RESTART=1 enable auto-restart (default off) -# WORKDIR cwd for the launch (default $HOME/workspace) -# CLAUDEAR_BIN binary path (default /usr/bin/claudear) -# CONFIG --config value (default claudear.toml) -# START_ARGS args after --config (default "start --poll") -# DAEMON_LOG daemon stdout/stderr log (default $WORKDIR/claudear-daemon.out) -# RUN_AS run the daemon as this user (default: current user) -# RESTART_COOLDOWN min seconds between restarts (default 30) -# set -u INTERVAL="${INTERVAL:-60}" @@ -55,18 +37,6 @@ PIDFILE="${PIDFILE:-./claudear-monitor.pid}" MATCH="${MATCH:-claudear.*start}" THRESHOLD_MB="${THRESHOLD_MB:-}" -RESTART="${RESTART:-0}" -WORKDIR="${WORKDIR:-$HOME/workspace}" -CLAUDEAR_BIN="${CLAUDEAR_BIN:-/usr/bin/claudear}" -CONFIG="${CONFIG:-claudear.toml}" -START_ARGS="${START_ARGS:-start --poll}" -DAEMON_LOG="${DAEMON_LOG:-$WORKDIR/claudear-daemon.out}" -RUN_AS="${RUN_AS:-}" -RESTART_COOLDOWN="${RESTART_COOLDOWN:-30}" - -LAST_RESTART=0 -RESTART_COUNT=0 - HEADER="timestamp,pid,rss_mb,peak_rss_mb,vsz_mb,threads,fds,cpu_pct,descendants,claude_procs,git_procs,sys_mem_used_pct,sys_mem_avail_mb,load1" now() { date '+%Y-%m-%dT%H:%M:%S%z'; } @@ -93,45 +63,6 @@ list_descendants() { kb_to_mb() { awk -v k="${1:-0}" 'BEGIN{ printf "%.1f", (k+0)/1024 }'; } -# Relaunch the daemon. `claudear start` (without --foreground) self-daemonizes, -# and startup cleans up stale pid/socket files from the crashed run, so this is -# safe to call whenever no live daemon is found. Cooldown-guarded to avoid a -# tight restart loop when the daemon dies immediately on boot. -restart_daemon() { - local ts now - ts="$(now)" - now="$(date +%s)" - if [ $((now - LAST_RESTART)) -lt "$RESTART_COOLDOWN" ]; then - echo "$ts restart suppressed (within ${RESTART_COOLDOWN}s cooldown)" - return - fi - if [ ! -x "$CLAUDEAR_BIN" ]; then - echo "$ts,ERROR,cannot restart: $CLAUDEAR_BIN not executable,," >>"$LOG" - echo "$ts ERROR: $CLAUDEAR_BIN not executable" - return - fi - if [ ! -d "$WORKDIR" ]; then - echo "$ts,ERROR,cannot restart: WORKDIR $WORKDIR missing,," >>"$LOG" - echo "$ts ERROR: WORKDIR $WORKDIR missing" - return - fi - - LAST_RESTART="$now" - RESTART_COUNT=$((RESTART_COUNT + 1)) - echo "$ts,RESTART,#$RESTART_COUNT launching $CLAUDEAR_BIN --config $CONFIG $START_ARGS,workdir=$WORKDIR," >>"$LOG" - echo "$ts RESTART #$RESTART_COUNT: $CLAUDEAR_BIN --config $CONFIG $START_ARGS (cwd $WORKDIR)" - - # Launch from WORKDIR so the relative --config path resolves, detached from the - # monitor so it survives the monitor stopping. - if [ -n "$RUN_AS" ]; then - ( cd "$WORKDIR" && setsid runuser -u "$RUN_AS" -- \ - "$CLAUDEAR_BIN" --config "$CONFIG" $START_ARGS >>"$DAEMON_LOG" 2>&1 & ) - else - ( cd "$WORKDIR" && setsid \ - "$CLAUDEAR_BIN" --config "$CONFIG" $START_ARGS >>"$DAEMON_LOG" 2>&1 & ) - fi -} - sample() { local pid ts ts="$(now)" @@ -141,9 +72,6 @@ sample() { # Daemon not running right now - record the gap so a crash/restart is visible. printf '%s,,,,,,,,,,,,,\n' "$ts" >>"$LOG" echo "$ts claudear not running" - if [ "$RESTART" = "1" ]; then - restart_daemon - fi return fi From 5322c85e683d90821f221ecf21cffcfe7977db62 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Thu, 20 Aug 2026 18:04:03 +0530 Subject: [PATCH 17/18] feat(monitor): add auto-restart functionality for claudear daemon with cooldown --- scripts/monitor_claudear.sh | 65 ++++++++++++++++++++++++++++++++++++- 1 file changed, 64 insertions(+), 1 deletion(-) diff --git a/scripts/monitor_claudear.sh b/scripts/monitor_claudear.sh index 3a617a7..cb76e40 100755 --- a/scripts/monitor_claudear.sh +++ b/scripts/monitor_claudear.sh @@ -16,12 +16,18 @@ # fds -> leaked file descriptors / sockets / worktrees # descendants/claude/git-> child processes not being reaped # -# Usage: +# Usage (start/stop the monitor whenever you want): # ./monitor_claudear.sh start # loop in foreground (nohup it for a droplet) # ./monitor_claudear.sh once # take a single sample and exit # ./monitor_claudear.sh status # show whether a monitor is running # ./monitor_claudear.sh stop # stop a running monitor (via pidfile) # +# By default it ONLY observes. Set RESTART=1 to also relaunch the daemon when it +# is found dead (opt-in supervision): +# RESTART=1 nohup ./monitor_claudear.sh start >/dev/null 2>&1 & +# # while RESTART=1 the monitor respawns the daemon, so to stop the daemon for +# # real: stop the monitor first, then `claudear stop`. +# # Config via env vars: # INTERVAL seconds between samples (default 60) # LOG output CSV path (default ./memory.log) @@ -29,6 +35,15 @@ # MATCH pgrep pattern for the daemon (default the start --poll cmd) # THRESHOLD_MB if set, log a WARN line when rss exceeds this # +# Auto-restart (only used when RESTART=1): +# WORKDIR cwd for the launch (default $HOME/workspace) +# CLAUDEAR_BIN binary path (default /usr/bin/claudear) +# CONFIG --config value (default claudear.toml) +# START_ARGS args after --config (default "start --poll") +# DAEMON_LOG daemon stdout/stderr log (default $WORKDIR/claudear-daemon.out) +# RUN_AS run the daemon as this user (default: current user) +# COOLDOWN min seconds between restarts (default 30) +# set -u INTERVAL="${INTERVAL:-60}" @@ -37,6 +52,18 @@ PIDFILE="${PIDFILE:-./claudear-monitor.pid}" MATCH="${MATCH:-claudear.*start}" THRESHOLD_MB="${THRESHOLD_MB:-}" +RESTART="${RESTART:-0}" +WORKDIR="${WORKDIR:-$HOME/workspace}" +CLAUDEAR_BIN="${CLAUDEAR_BIN:-/usr/bin/claudear}" +CONFIG="${CONFIG:-claudear.toml}" +START_ARGS="${START_ARGS:-start --poll}" +DAEMON_LOG="${DAEMON_LOG:-$WORKDIR/claudear-daemon.out}" +RUN_AS="${RUN_AS:-}" +COOLDOWN="${COOLDOWN:-30}" + +LAST_RESTART=0 +RESTART_COUNT=0 + HEADER="timestamp,pid,rss_mb,peak_rss_mb,vsz_mb,threads,fds,cpu_pct,descendants,claude_procs,git_procs,sys_mem_used_pct,sys_mem_avail_mb,load1" now() { date '+%Y-%m-%dT%H:%M:%S%z'; } @@ -63,6 +90,41 @@ list_descendants() { kb_to_mb() { awk -v k="${1:-0}" 'BEGIN{ printf "%.1f", (k+0)/1024 }'; } +# Relaunch the daemon (only called when RESTART=1). `claudear start` (without +# --foreground) self-daemonizes and cleans up stale pid/socket files on boot, so +# this is safe whenever no live daemon is found. Cooldown-guarded so a daemon +# that dies on boot doesn't spin-restart. +restart_daemon() { + local ts n; ts="$(now)"; n="$(date +%s)" + if [ $((n - LAST_RESTART)) -lt "$COOLDOWN" ]; then + echo "$ts restart suppressed (within ${COOLDOWN}s cooldown)" + return + fi + if [ ! -x "$CLAUDEAR_BIN" ]; then + echo "$ts,ERROR,$CLAUDEAR_BIN not executable,," >>"$LOG" + echo "$ts ERROR: $CLAUDEAR_BIN not executable"; return + fi + if [ ! -d "$WORKDIR" ]; then + echo "$ts,ERROR,WORKDIR $WORKDIR missing,," >>"$LOG" + echo "$ts ERROR: WORKDIR $WORKDIR missing"; return + fi + + LAST_RESTART="$n" + RESTART_COUNT=$((RESTART_COUNT + 1)) + echo "$ts,RESTART,#$RESTART_COUNT (cd $WORKDIR && $CLAUDEAR_BIN --config $CONFIG $START_ARGS),," >>"$LOG" + echo "$ts RESTART #$RESTART_COUNT: (cd $WORKDIR && $CLAUDEAR_BIN --config $CONFIG $START_ARGS)" + + # Launch from WORKDIR so the relative --config resolves; setsid-detached so the + # daemon survives the monitor stopping. + if [ -n "$RUN_AS" ]; then + ( cd "$WORKDIR" && setsid runuser -u "$RUN_AS" -- \ + "$CLAUDEAR_BIN" --config "$CONFIG" $START_ARGS >>"$DAEMON_LOG" 2>&1 & ) + else + ( cd "$WORKDIR" && setsid \ + "$CLAUDEAR_BIN" --config "$CONFIG" $START_ARGS >>"$DAEMON_LOG" 2>&1 & ) + fi +} + sample() { local pid ts ts="$(now)" @@ -72,6 +134,7 @@ sample() { # Daemon not running right now - record the gap so a crash/restart is visible. printf '%s,,,,,,,,,,,,,\n' "$ts" >>"$LOG" echo "$ts claudear not running" + [ "$RESTART" = "1" ] && restart_daemon return fi From 1ba31d3dca930095a66362637baceb23e00bfadd Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Thu, 20 Aug 2026 18:12:59 +0530 Subject: [PATCH 18/18] added comments for running --- scripts/monitor_claudear.sh | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/scripts/monitor_claudear.sh b/scripts/monitor_claudear.sh index cb76e40..3d8a9c8 100755 --- a/scripts/monitor_claudear.sh +++ b/scripts/monitor_claudear.sh @@ -28,6 +28,14 @@ # # while RESTART=1 the monitor respawns the daemon, so to stop the daemon for # # real: stop the monitor first, then `claudear stop`. # +# Typical droplet run (script copied to ~/monitor, daemon in ~/workspace): +# cd ~/monitor +# RESTART=1 nohup ./monitor.sh start >>monitor.out 2>&1 & +# disown +# ./monitor.sh status # shows monitor + daemon pids +# ./monitor.sh once # take one sample right now and print it +# tail -f memory.log # watch samples accrue +# # Config via env vars: # INTERVAL seconds between samples (default 60) # LOG output CSV path (default ./memory.log)