From 0c5a8561e68560f8c16018c7cf2e3218ba732438 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Fri, 7 Aug 2026 18:48:41 +0530 Subject: [PATCH 01/16] empty commit From 6adb1b0b92f1030ac927016704df3add160f332f Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 11 Aug 2026 12:55:29 +0530 Subject: [PATCH 02/16] feat(storage): track PR conversation comments cursor Add last_issue_comment_id/last_issue_comment_time to PrReviewState and the pr_review_states table (migration V9). PR conversation comments come from the issues/{n}/comments endpoint, a distinct GitHub comment id space from inline review comments, so they need their own polling cursor. --- crates/claudear-core/src/types.rs | 7 ++++++ crates/claudear-storage/src/lib.rs | 2 ++ crates/claudear-storage/src/migrator.rs | 19 ++++++++++++++-- crates/claudear-storage/src/sqlite.rs | 22 ++++++++++++++++--- .../V9__pr_review_states_issue_comments.sql | 6 +++++ 5 files changed, 51 insertions(+), 5 deletions(-) create mode 100644 migrations/V9__pr_review_states_issue_comments.sql diff --git a/crates/claudear-core/src/types.rs b/crates/claudear-core/src/types.rs index 6ba512f..51fc4b0 100644 --- a/crates/claudear-core/src/types.rs +++ b/crates/claudear-core/src/types.rs @@ -2971,6 +2971,11 @@ pub struct PrReviewState { pub last_review_time: Option, pub last_comment_id: Option, pub last_comment_time: Option, + /// Cursor for PR *conversation* comments (the `issues/{n}/comments` timeline), + /// tracked separately from inline review comments because the two live in + /// distinct GitHub comment id spaces and are fetched from different endpoints. + pub last_issue_comment_id: Option, + pub last_issue_comment_time: Option, pub is_active: bool, } @@ -2992,6 +2997,8 @@ impl PrReviewState { last_review_time: None, last_comment_id: None, last_comment_time: None, + last_issue_comment_id: None, + last_issue_comment_time: None, is_active: true, } } diff --git a/crates/claudear-storage/src/lib.rs b/crates/claudear-storage/src/lib.rs index c96a9df..db4002b 100644 --- a/crates/claudear-storage/src/lib.rs +++ b/crates/claudear-storage/src/lib.rs @@ -3384,6 +3384,8 @@ mod tests { last_review_time: None, last_comment_id: None, last_comment_time: None, + last_issue_comment_id: None, + last_issue_comment_time: None, is_active: true, }; assert!(t.save_pr_review_state(&state).is_ok()); diff --git a/crates/claudear-storage/src/migrator.rs b/crates/claudear-storage/src/migrator.rs index 8ebffbb..b022a52 100644 --- a/crates/claudear-storage/src/migrator.rs +++ b/crates/claudear-storage/src/migrator.rs @@ -56,6 +56,11 @@ const MIGRATIONS: &[Migration] = &[ name: "answer_message_ids", sql: include_str!("../../../migrations/V8__answer_message_ids.sql"), }, + Migration { + version: 9, + name: "pr_review_states_issue_comments", + sql: include_str!("../../../migrations/V9__pr_review_states_issue_comments.sql"), + }, ]; /// Run all pending migrations against the given connection. @@ -116,7 +121,7 @@ mod tests { row.get(0) }) .unwrap(); - assert_eq!(version, 8); + assert_eq!(version, 9); // Verify a table from V1 exists let count: u32 = conn @@ -137,6 +142,16 @@ mod tests { ) .unwrap(); assert_eq!(has_col, 1); + + // Verify the V9 column exists on pr_review_states. + let has_issue_comment_col: u32 = conn + .query_row( + "SELECT COUNT(*) FROM pragma_table_info('pr_review_states') WHERE name = 'last_issue_comment_id'", + [], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(has_issue_comment_col, 1); } #[test] @@ -151,7 +166,7 @@ mod tests { row.get(0) }) .unwrap(); - assert_eq!(version, 8); + assert_eq!(version, 9); } #[test] diff --git a/crates/claudear-storage/src/sqlite.rs b/crates/claudear-storage/src/sqlite.rs index 15f0739..bd8c1d7 100644 --- a/crates/claudear-storage/src/sqlite.rs +++ b/crates/claudear-storage/src/sqlite.rs @@ -1814,9 +1814,10 @@ impl ActivityStore for SqliteTracker { INSERT INTO pr_review_states ( pr_url, repo, pr_number, issue_id, source, last_review_id, last_review_time, last_comment_id, last_comment_time, + last_issue_comment_id, last_issue_comment_time, is_active, created_at ) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, datetime('now')) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, datetime('now')) ON CONFLICT(pr_url) DO UPDATE SET repo = excluded.repo, pr_number = excluded.pr_number, @@ -1826,6 +1827,8 @@ impl ActivityStore for SqliteTracker { last_review_time = excluded.last_review_time, last_comment_id = excluded.last_comment_id, last_comment_time = excluded.last_comment_time, + last_issue_comment_id = excluded.last_issue_comment_id, + last_issue_comment_time = excluded.last_issue_comment_time, is_active = excluded.is_active "#, params![ @@ -1838,6 +1841,8 @@ impl ActivityStore for SqliteTracker { state.last_review_time, state.last_comment_id, state.last_comment_time, + state.last_issue_comment_id, + state.last_issue_comment_time, state.is_active as i32, ], )?; @@ -1858,6 +1863,7 @@ impl ActivityStore for SqliteTracker { r#" SELECT pr_url, repo, pr_number, issue_id, source, last_review_id, last_review_time, last_comment_id, last_comment_time, + last_issue_comment_id, last_issue_comment_time, is_active FROM pr_review_states WHERE is_active = 1 @@ -5050,7 +5056,8 @@ impl SqliteTracker { /// Convert a database row to a PrReviewState. /// Expects columns: pr_url, repo, pr_number, issue_id, source, - /// last_review_id, last_review_time, last_comment_id, last_comment_time, is_active + /// last_review_id, last_review_time, last_comment_id, last_comment_time, + /// last_issue_comment_id, last_issue_comment_time, is_active fn row_to_pr_review_state( row: &rusqlite::Row<'_>, ) -> rusqlite::Result { @@ -5064,7 +5071,9 @@ impl SqliteTracker { last_review_time: row.get(6)?, last_comment_id: row.get(7)?, last_comment_time: row.get(8)?, - is_active: row.get::<_, i32>(9)? != 0, + last_issue_comment_id: row.get(9)?, + last_issue_comment_time: row.get(10)?, + is_active: row.get::<_, i32>(11)? != 0, }) } @@ -11106,6 +11115,8 @@ mod tests { state.last_review_time = Some("2024-01-15T10:00:00Z".to_string()); state.last_comment_id = Some(888); state.last_comment_time = Some("2024-01-15T11:00:00Z".to_string()); + state.last_issue_comment_id = Some(777); + state.last_issue_comment_time = Some("2024-01-15T12:00:00Z".to_string()); tracker.save_pr_review_state(&state).unwrap(); // Verify the update @@ -11121,6 +11132,11 @@ mod tests { states[0].last_comment_time, Some("2024-01-15T11:00:00Z".to_string()) ); + assert_eq!(states[0].last_issue_comment_id, Some(777)); + assert_eq!( + states[0].last_issue_comment_time, + Some("2024-01-15T12:00:00Z".to_string()) + ); } #[test] diff --git a/migrations/V9__pr_review_states_issue_comments.sql b/migrations/V9__pr_review_states_issue_comments.sql new file mode 100644 index 0000000..54d1ea7 --- /dev/null +++ b/migrations/V9__pr_review_states_issue_comments.sql @@ -0,0 +1,6 @@ +-- Track PR conversation (issue) comments separately from inline review comments. +-- GitHub delivers plain "@claudear ..." PR comments via the issues/{n}/comments +-- endpoint, which uses a distinct comment id space from inline review comments, +-- so the review watcher advances an independent cursor for them. +ALTER TABLE pr_review_states ADD COLUMN last_issue_comment_id INTEGER; +ALTER TABLE pr_review_states ADD COLUMN last_issue_comment_time TEXT; From 1e81585bba91a082c469a4145754436ca1bf8038 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 11 Aug 2026 12:55:41 +0530 Subject: [PATCH 03/16] feat(review): ingest PR conversation comments into the review loop Fetch a PR's issues/{n}/comments timeline so a plain "@claudear fix this" left on the PR conversation (not an inline review comment or formal review) is picked up. Trigger-matched comments emit CommentsAdded and flow through the existing feedback path, which pushes to the same PR branch. Advances a separate issue-comment cursor and forwards the new ScmProvider methods through InstrumentedScm. --- crates/claudear-integrations/src/github.rs | 99 ++++++ crates/claudear-integrations/src/scm.rs | 295 +++++++++++++++++- crates/claudear-integrations/src/telemetry.rs | 19 ++ 3 files changed, 405 insertions(+), 8 deletions(-) diff --git a/crates/claudear-integrations/src/github.rs b/crates/claudear-integrations/src/github.rs index f8c79bd..f90727a 100644 --- a/crates/claudear-integrations/src/github.rs +++ b/crates/claudear-integrations/src/github.rs @@ -290,6 +290,97 @@ impl GitHubClient { Ok(all_comments) } + /// Get a PR's *conversation* comments (the issue-comment timeline), as opposed + /// to inline review-thread comments. GitHub models a PR as an issue, so these + /// come from `/repos/{repo}/issues/{n}/comments` — a plain "@claudear fix this" + /// left on the PR conversation lands here, not in `/pulls/{n}/comments`. + /// + /// They are mapped into [`ReviewComment`] with an empty `path`/absent `line` + /// and no `pull_request_review_id`, so they flow through the same standalone + /// review-feedback pipeline as inline comments. + pub async fn get_pr_issue_comments( + &self, + repo: &str, + pr_number: i64, + ) -> Result> { + // A PR conversation comment as returned by the issues-comments endpoint. + // Only the fields we carry forward are deserialized. + #[derive(serde::Deserialize)] + struct GitHubIssueComment { + id: i64, + #[serde(default)] + body: Option, + user: ReviewUser, + created_at: String, + updated_at: String, + html_url: String, + } + + let token = self + .config + .token + .as_ref() + .ok_or_else(|| Error::config("GitHub token not configured"))? + .expose(); + + let base_url = format!( + "https://api.github.com/repos/{}/issues/{}/comments", + repo, pr_number + ); + let headers = self.build_headers(token); + + let mut all_comments = Vec::new(); + let mut page = 1usize; + const DEFAULT_PAGE_SIZE: usize = 30; + const MAX_PAGES: usize = 100; + + loop { + let url = if page == 1 { + base_url.clone() + } else { + format!("{}?page={}", base_url, page) + }; + let response = self.http.get(&url, headers.clone()).await?; + + if !response.is_success() { + return Err(Error::Other(format!( + "GitHub API error ({}): {}", + response.status, response.body + ))); + } + + let comments: Vec = response.json()?; + let count = comments.len(); + all_comments.extend(comments.into_iter().map(|c| ReviewComment { + id: c.id, + path: String::new(), + position: None, + original_position: None, + body: c.body.unwrap_or_default(), + user: c.user, + created_at: c.created_at, + updated_at: c.updated_at, + html_url: c.html_url, + pull_request_review_id: None, + line: None, + start_line: None, + side: None, + })); + + if count < DEFAULT_PAGE_SIZE { + break; + } + + page += 1; + if page > MAX_PAGES { + tracing::warn!(repo = %repo, pr_number, "Hit pagination limit for PR issue comments"); + break; + } + } + + Ok(all_comments) + } + /// Get the GitHub token (if configured). pub fn token(&self) -> Option<&str> { self.config.token.expose_as_deref() @@ -952,6 +1043,14 @@ impl ScmProvider for GitHubClient { self.get_pr_review_comments(project, number).await } + async fn get_pr_conversation_comments( + &self, + project: &str, + number: i64, + ) -> Result> { + self.get_pr_issue_comments(project, number).await + } + async fn list_repos(&self, org_or_group: &str) -> Result> { self.list_org_repos(org_or_group).await } diff --git a/crates/claudear-integrations/src/scm.rs b/crates/claudear-integrations/src/scm.rs index 9321970..86eca92 100644 --- a/crates/claudear-integrations/src/scm.rs +++ b/crates/claudear-integrations/src/scm.rs @@ -113,6 +113,39 @@ pub trait ScmProvider: Send + Sync { } } + /// Get a PR/MR's *conversation* comments (non-inline), e.g. GitHub's + /// `issues/{n}/comments` timeline. These are plain comments left on the PR + /// outside a formal review. Default: none, since not every provider exposes a + /// separate conversation timeline. + async fn get_pr_conversation_comments( + &self, + _project: &str, + _number: i64, + ) -> Result> { + Ok(Vec::new()) + } + + /// Get conversation comments updated at or after `since` (RFC 3339 timestamp). + /// + /// Default implementation fetches all conversation comments and filters by + /// timestamp, mirroring [`get_new_review_comments`](Self::get_new_review_comments). + async fn get_new_conversation_comments( + &self, + project: &str, + number: i64, + since: Option<&str>, + ) -> Result> { + let comments = self.get_pr_conversation_comments(project, number).await?; + if let Some(since_time) = since { + Ok(comments + .into_iter() + .filter(|c| timestamp_at_or_after(&c.updated_at, since_time)) + .collect()) + } else { + Ok(comments) + } + } + /// List repositories for an organization / group. async fn list_repos(&self, org_or_group: &str) -> Result>; @@ -407,12 +440,19 @@ impl ReviewEvent { ReviewEvent::CommentsAdded { comments, .. } => { let mut summary = String::new(); for comment in comments { - summary.push_str(&format!( - "Comment from @{} on `{}`", - comment.user.login, comment.path - )); - if let Some(line) = comment.line { - summary.push_str(&format!(" (line {})", line)); + // Inline comments carry a file path (and often a line); PR + // conversation comments have neither, so omit the "on `path`" + // clause for them rather than printing an empty backtick pair. + if comment.path.is_empty() { + summary.push_str(&format!("Comment from @{}", comment.user.login)); + } else { + summary.push_str(&format!( + "Comment from @{} on `{}`", + comment.user.login, comment.path + )); + if let Some(line) = comment.line { + summary.push_str(&format!(" (line {})", line)); + } } summary.push_str(&format!(":\n{}\n\n", comment.body)); } @@ -1233,12 +1273,119 @@ impl ReviewWatcher { } } - if !standalone_comments.is_empty() { + // PR conversation comments (the issues-comments timeline). A reviewer who + // leaves a plain "@claudear fix this" on the PR conversation — rather than + // an inline review-thread comment or a formal review — lands here. These + // use a distinct GitHub comment id space and endpoint, so they carry their + // own cursor (`last_issue_comment_*`). + let new_conversation_comments: Vec = match self + .provider + .get_new_conversation_comments( + &state.repo, + state.pr_number, + state.last_issue_comment_time.as_deref(), + ) + .await + { + Ok(comments) => comments + .into_iter() + .filter(|c| !is_skippable_bot(&c.user, allowed_bots)) + .filter(|c| { + Self::comment_is_after_cursor( + c, + state.last_issue_comment_time.as_deref(), + state.last_issue_comment_id, + ) + }) + .collect(), + Err(e) => { + // A transient failure here must not drop the review/comment events + // already collected this cycle. + tracing::warn!( + component = "review_watcher", + pr_url = %state.pr_url, + error = %e, + "Failed to fetch PR conversation comments; continuing without them" + ); + Vec::new() + } + }; + + if !new_conversation_comments.is_empty() { + // Advance the issue-comment cursor over ALL new conversation comments + // (including non-trigger ones) so unchanged comments aren't rescanned. + let mut latest_id = state.last_issue_comment_id; + let mut latest_time = state.last_issue_comment_time.clone(); + for comment in &new_conversation_comments { + let replace = latest_time + .as_deref() + .map(|existing_time| { + let cmp = compare_timestamps(&comment.updated_at, existing_time); + cmp == std::cmp::Ordering::Greater + || (cmp == std::cmp::Ordering::Equal + && comment.id > latest_id.unwrap_or(i64::MIN)) + }) + .unwrap_or(true); + if replace { + latest_id = Some(comment.id); + latest_time = Some(comment.updated_at.clone()); + } + } + + let mut states = self.states.write().unwrap_or_else(|poisoned| { + tracing::warn!(component = "review_watcher", "RwLock poisoned, recovering"); + poisoned.into_inner() + }); + if let Some(s) = states.get_mut(&state.pr_url) { + s.last_issue_comment_id = latest_id; + if let Some(t) = latest_time { + s.last_issue_comment_time = Some(t); + } + if let Some(ref tracker) = self.tracker { + if let Err(e) = tracker.save_pr_review_state(s) { + tracing::warn!( + component = "review_watcher", + pr_url = %s.pr_url, + error = %e, + "Failed to persist PR review state update" + ); + } + } + } + } + + // Only trigger-matched conversation comments are actionable feedback. + let conversation_feedback: Vec = new_conversation_comments + .into_iter() + .filter(|c| trigger.is_empty() || c.body.to_lowercase().contains(&trigger.to_lowercase())) + .collect(); + + if !conversation_feedback.is_empty() { + if let Some(ref tracker) = self.tracker { + for comment in &conversation_feedback { + if let Err(e) = tracker.record_pr_review_comment(&state.pr_url, comment) { + tracing::warn!( + component = "review_watcher", + pr_url = %state.pr_url, + comment_id = comment.id, + error = %e, + "Failed to record PR conversation comment" + ); + } + } + } + } + + // Emit inline standalone comments and conversation comments together so the + // downstream fix loop sees a single feedback batch per PR per cycle. + let mut added_comments = standalone_comments; + added_comments.extend(conversation_feedback); + if !added_comments.is_empty() { events.push(ReviewEvent::CommentsAdded { pr_url: state.pr_url.clone(), repo: state.repo.clone(), pr_number: state.pr_number, - comments: standalone_comments, + comments: added_comments, }); } @@ -1688,6 +1835,8 @@ mod tests { last_review_time: Some("2024-03-15T10:30:00Z".to_string()), last_comment_id: Some(888), last_comment_time: Some("2024-03-15T11:00:00Z".to_string()), + last_issue_comment_id: Some(777), + last_issue_comment_time: Some("2024-03-15T11:30:00Z".to_string()), is_active: true, }; let json = serde_json::to_string(&state).expect("serialize"); @@ -2793,6 +2942,7 @@ mod tests { allowed_bots: Vec, reviews: Arc>>, comments: Arc>>, + conversation: Arc>>, /// When true, get_review_comments returns an error. comments_error: Arc>, } @@ -2806,6 +2956,7 @@ mod tests { allowed_bots: Vec::new(), reviews: Arc::new(Mutex::new(Vec::new())), comments: Arc::new(Mutex::new(Vec::new())), + conversation: Arc::new(Mutex::new(Vec::new())), comments_error: Arc::new(Mutex::new(false)), } } @@ -2826,6 +2977,10 @@ mod tests { fn set_comments_error(&self, should_error: bool) { *self.comments_error.lock().unwrap() = should_error; } + + fn set_conversation_comments(&self, comments: Vec) { + *self.conversation.lock().unwrap() = comments; + } } #[async_trait] @@ -2881,11 +3036,43 @@ mod tests { Ok(self.comments.lock().unwrap().clone()) } + async fn get_pr_conversation_comments( + &self, + _project: &str, + _number: i64, + ) -> Result> { + Ok(self.conversation.lock().unwrap().clone()) + } + async fn list_repos(&self, _org_or_group: &str) -> Result> { Ok(vec![]) } } + /// A PR conversation (issue-timeline) comment: no file path, no line, + /// no parent review id — the shape produced by `get_pr_issue_comments`. + fn make_conversation_comment(id: i64, body: &str, updated_at: &str) -> ReviewComment { + ReviewComment { + id, + path: String::new(), + position: None, + original_position: None, + body: body.to_string(), + user: ReviewUser { + id: id + 3000, + login: "reviewer".to_string(), + user_type: Some("User".to_string()), + }, + created_at: updated_at.to_string(), + updated_at: updated_at.to_string(), + html_url: format!("https://github.com/org/repo/pull/1#issuecomment-{}", id), + pull_request_review_id: None, + line: None, + start_line: None, + side: None, + } + } + fn make_review(id: i64, state: &str, user: &str, submitted_at: &str) -> CodeReview { CodeReview { id, @@ -3313,6 +3500,98 @@ mod tests { assert!(comments_events.is_empty()); } + #[tokio::test] + async fn test_conversation_comment_with_trigger_is_actionable() { + // A plain "@claudear ..." left on the PR conversation timeline (not an + // inline review comment) must surface as actionable review feedback. + let mock = MockScmProvider::new("github", true, "@claudear"); + mock.set_conversation_comments(vec![make_conversation_comment( + 501, + "@claudear this shutdown path misses a task, please fix it", + "2025-01-02T00:00:00Z", + )]); + let provider: Arc = Arc::new(mock); + let watcher = ReviewWatcher::new(provider); + watcher.watch_pr(make_state( + "https://github.com/org/repo/pull/1", + "org/repo", + 1, + )); + + let events = watcher.check_for_reviews().await.unwrap(); + let comment_events: Vec<_> = events + .iter() + .filter(|e| matches!(e, ReviewEvent::CommentsAdded { .. })) + .collect(); + assert_eq!(comment_events.len(), 1, "conversation comment not surfaced"); + match comment_events[0] { + ReviewEvent::CommentsAdded { comments, .. } => { + assert_eq!(comments.len(), 1); + assert_eq!(comments[0].id, 501); + assert!(comments[0].path.is_empty()); + } + _ => unreachable!(), + } + } + + #[tokio::test] + async fn test_conversation_comment_without_trigger_is_ignored() { + let mock = MockScmProvider::new("github", true, "@claudear"); + mock.set_conversation_comments(vec![make_conversation_comment( + 502, + "nice work, LGTM", + "2025-01-02T00:00:00Z", + )]); + let provider: Arc = Arc::new(mock); + let watcher = ReviewWatcher::new(provider); + watcher.watch_pr(make_state( + "https://github.com/org/repo/pull/1", + "org/repo", + 1, + )); + + let events = watcher.check_for_reviews().await.unwrap(); + assert!(events + .iter() + .all(|e| !matches!(e, ReviewEvent::CommentsAdded { .. }))); + } + + #[tokio::test] + async fn test_conversation_comment_cursor_prevents_reprocessing() { + // The issue-comment cursor must advance so the same conversation + // comment is not re-emitted on the next poll cycle. + let mock = MockScmProvider::new("github", true, "@claudear"); + mock.set_conversation_comments(vec![make_conversation_comment( + 503, + "@claudear please address this", + "2025-01-02T00:00:00Z", + )]); + let provider: Arc = Arc::new(mock); + let watcher = ReviewWatcher::new(provider); + watcher.watch_pr(make_state( + "https://github.com/org/repo/pull/1", + "org/repo", + 1, + )); + + let first = watcher.check_for_reviews().await.unwrap(); + assert_eq!( + first + .iter() + .filter(|e| matches!(e, ReviewEvent::CommentsAdded { .. })) + .count(), + 1 + ); + + let second = watcher.check_for_reviews().await.unwrap(); + assert!( + second + .iter() + .all(|e| !matches!(e, ReviewEvent::CommentsAdded { .. })), + "conversation comment re-emitted after cursor should have advanced" + ); + } + #[tokio::test] async fn test_check_for_reviews_skips_pending_reviews() { let mock = MockScmProvider::new("github", true, "@claudear"); diff --git a/crates/claudear-integrations/src/telemetry.rs b/crates/claudear-integrations/src/telemetry.rs index b17b4c5..90c1cfe 100644 --- a/crates/claudear-integrations/src/telemetry.rs +++ b/crates/claudear-integrations/src/telemetry.rs @@ -356,6 +356,25 @@ impl ScmProvider for InstrumentedScm { .get_new_review_comments(project, number, since) .await } + async fn get_pr_conversation_comments( + &self, + project: &str, + number: i64, + ) -> Result> { + self.inner + .get_pr_conversation_comments(project, number) + .await + } + async fn get_new_conversation_comments( + &self, + project: &str, + number: i64, + since: Option<&str>, + ) -> Result> { + self.inner + .get_new_conversation_comments(project, number, since) + .await + } async fn list_repos(&self, org_or_group: &str) -> Result> { self.inner.list_repos(org_or_group).await } From fdc88b2ee82935ba1b1327b1c4ed3ff06abcc729 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 11 Aug 2026 12:55:49 +0530 Subject: [PATCH 04/16] feat(webhook): handle issue_comment events on watched PRs Dispatch GitHub issue_comment events to a new handler that ignores non-PR issues, requires the review trigger, and re-polls via check_for_pr so PR conversation comments are addressed in real time instead of waiting for the next poll cycle. --- .../src/webhook/github.rs | 231 +++++++++++++++++- 1 file changed, 229 insertions(+), 2 deletions(-) diff --git a/crates/claudear-integrations/src/webhook/github.rs b/crates/claudear-integrations/src/webhook/github.rs index 0f9603e..9b17a30 100644 --- a/crates/claudear-integrations/src/webhook/github.rs +++ b/crates/claudear-integrations/src/webhook/github.rs @@ -1,8 +1,9 @@ //! GitHub webhook handler for PR review and pull request events. //! //! This handler processes `pull_request_review`, `pull_request_review_comment`, -//! and `pull_request` events from GitHub webhooks to trigger review processing -//! and detect PR merges/closes in real-time instead of relying solely on polling. +//! `issue_comment` (PR conversation comments), and `pull_request` events from +//! GitHub webhooks to trigger review processing and detect PR merges/closes in +//! real-time instead of relying solely on polling. use crate::scm::{is_skippable_bot, CodeReview, ReviewComment, ReviewUser, ReviewWatcher}; use claudear_config::config::GitHubConfig; @@ -170,6 +171,7 @@ impl GitHubWebhookHandler { match event_type { "pull_request_review" => self.handle_review_submitted(payload).await, "pull_request_review_comment" => self.handle_review_comment(payload).await, + "issue_comment" => self.handle_issue_comment(payload).await, "pull_request" => self.handle_pull_request(payload).await, _ => { tracing::debug!( @@ -378,6 +380,149 @@ impl GitHubWebhookHandler { Ok(WebhookAction::Processed) } + /// Handle an `issue_comment.created` event. + /// + /// GitHub fires this for comments on the PR *conversation* timeline (a plain + /// "@claudear fix this" left outside a formal review or inline thread). Only + /// comments on pull requests are relevant, so non-PR issue comments are + /// ignored. When the comment mentions the review trigger on a watched PR, we + /// re-poll via `check_for_pr`, which now picks up conversation comments. + async fn handle_issue_comment(&self, payload: &serde_json::Value) -> Result { + let action = payload.get("action").and_then(|v| v.as_str()).unwrap_or(""); + + if action != "created" { + tracing::debug!( + source = "github", + action = %action, + "Ignoring non-created issue comment action" + ); + return Ok(WebhookAction::Ignored); + } + + let issue = match payload.get("issue") { + Some(i) => i, + None => { + tracing::warn!(source = "github", "Missing issue in payload"); + return Ok(WebhookAction::Ignored); + } + }; + + // Only PR conversation comments matter; a plain issue has no `pull_request`. + let pr = match issue.get("pull_request") { + Some(p) => p, + None => { + tracing::debug!( + source = "github", + "Issue comment is not on a pull request, ignoring" + ); + return Ok(WebhookAction::Ignored); + } + }; + + let pr_url = pr + .get("html_url") + .and_then(|v| v.as_str()) + .or_else(|| issue.get("html_url").and_then(|v| v.as_str())) + .unwrap_or_default(); + + let comment = match payload.get("comment") { + Some(c) => c, + None => { + tracing::warn!(source = "github", "Missing comment in payload"); + return Ok(WebhookAction::Ignored); + } + }; + + let review_watcher = match &self.review_watcher { + Some(rw) => rw, + None => { + tracing::debug!( + source = "github", + pr_url = %pr_url, + "ReviewWatcher not available, ignoring event" + ); + return Ok(WebhookAction::Ignored); + } + }; + + // Only act on PRs we're actively watching. + let state = match review_watcher.get_state(pr_url) { + Some(s) if s.is_active => s, + _ => { + tracing::debug!( + source = "github", + pr_url = %pr_url, + "PR not being watched, ignoring issue comment" + ); + return Ok(WebhookAction::Ignored); + } + }; + + // Skip bot comments (unless the bot is in the allowed list). + let user = ReviewUser { + id: comment + .get("user") + .and_then(|u| u.get("id")) + .and_then(|v| v.as_i64()) + .unwrap_or_default(), + login: comment + .get("user") + .and_then(|u| u.get("login")) + .and_then(|v| v.as_str()) + .unwrap_or_default() + .to_string(), + user_type: comment + .get("user") + .and_then(|u| u.get("type")) + .and_then(|v| v.as_str()) + .map(|s| s.to_string()), + }; + if is_skippable_bot(&user, &self.config.allowed_bots) { + tracing::debug!( + source = "github", + pr_url = %pr_url, + author = %user.login, + "Skipping bot issue comment" + ); + return Ok(WebhookAction::Ignored); + } + + // Require the review trigger so unrelated PR chatter doesn't spin up a + // re-poll. The polling path applies the same filter, so this only avoids + // needless work; correctness does not depend on it. + let trigger = self.config.review_trigger.to_lowercase(); + let body = comment + .get("body") + .and_then(|v| v.as_str()) + .unwrap_or_default(); + if !trigger.is_empty() && !body.to_lowercase().contains(&trigger) { + tracing::debug!( + source = "github", + pr_url = %pr_url, + "Issue comment does not mention review trigger, ignoring" + ); + return Ok(WebhookAction::Ignored); + } + + tracing::info!( + source = "github", + pr_url = %pr_url, + author = %user.login, + issue_id = %state.issue_id, + "Received PR conversation comment via webhook" + ); + + let processed_events = review_watcher.check_for_pr(pr_url).await?; + tracing::info!( + source = "github", + pr_url = %pr_url, + events = processed_events.len(), + "Processed issue comment webhook through ReviewWatcher" + ); + + Ok(WebhookAction::Processed) + } + /// Handle a pull_request.closed event (merged or closed without merge). async fn handle_pull_request(&self, payload: &serde_json::Value) -> Result { let action = payload.get("action").and_then(|v| v.as_str()).unwrap_or(""); @@ -1106,6 +1251,88 @@ mod tests { ); } + /// Helper: an `issue_comment.created` payload on a pull request. + fn issue_comment_payload(pr_url: &str, body: &str, is_pr: bool) -> serde_json::Value { + let mut issue = serde_json::json!({ + "number": 1, + "html_url": pr_url + }); + if is_pr { + issue["pull_request"] = serde_json::json!({ "html_url": pr_url }); + } + serde_json::json!({ + "action": "created", + "comment": { + "id": 300, + "user": { "id": 3, "login": "reviewer", "type": "User" }, + "body": body, + "created_at": "2024-01-15T12:00:00Z", + "updated_at": "2024-01-15T12:00:00Z", + "html_url": "https://github.com/owner/repo/pull/1#issuecomment-300" + }, + "issue": issue + }) + } + + #[tokio::test] + async fn test_issue_comment_on_watched_pr_with_trigger_is_processed() { + let pr_url = "https://github.com/owner/repo/pull/1"; + let handler = make_handler_watching_pr(pr_url, Arc::new(MockScmProvider::new())); + + let payload_value = + issue_comment_payload(pr_url, "@claudear please fix the shutdown race", true); + let payload_bytes = serde_json::to_vec(&payload_value).unwrap(); + let sig = make_valid_signature("test_secret", &payload_bytes); + let headers = make_headers("issue_comment", &sig); + + let result = handler + .process_webhook(&payload_bytes, &payload_value, &headers) + .await; + assert!( + result.unwrap().is_processed(), + "Triggered PR conversation comment should be processed" + ); + } + + #[tokio::test] + async fn test_issue_comment_without_trigger_is_ignored() { + let pr_url = "https://github.com/owner/repo/pull/1"; + let handler = make_handler_watching_pr(pr_url, Arc::new(MockScmProvider::new())); + + let payload_value = issue_comment_payload(pr_url, "LGTM, nice", true); + let payload_bytes = serde_json::to_vec(&payload_value).unwrap(); + let sig = make_valid_signature("test_secret", &payload_bytes); + let headers = make_headers("issue_comment", &sig); + + let result = handler + .process_webhook(&payload_bytes, &payload_value, &headers) + .await; + assert!( + !result.unwrap().is_processed(), + "Untriggered PR conversation comment should be ignored" + ); + } + + #[tokio::test] + async fn test_issue_comment_on_plain_issue_is_ignored() { + let pr_url = "https://github.com/owner/repo/pull/1"; + let handler = make_handler_watching_pr(pr_url, Arc::new(MockScmProvider::new())); + + // No `pull_request` field => a regular issue, not a PR. + let payload_value = issue_comment_payload(pr_url, "@claudear fix this", false); + let payload_bytes = serde_json::to_vec(&payload_value).unwrap(); + let sig = make_valid_signature("test_secret", &payload_bytes); + let headers = make_headers("issue_comment", &sig); + + let result = handler + .process_webhook(&payload_bytes, &payload_value, &headers) + .await; + assert!( + !result.unwrap().is_processed(), + "Comment on a non-PR issue should be ignored" + ); + } + // handle_review_submitted tests (5 tests) #[tokio::test] From a56f00ff5009f8477ef28c8590442c3d7156052e Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 11 Aug 2026 14:47:28 +0530 Subject: [PATCH 05/16] feat(storage): add handled ledger for PR review comments Extend V9 with handled_at/attempts on pr_review_comments and add the tracker methods that make the ledger the authority for outstanding review feedback: get_unhandled_pr_review_comments, mark_pr_review_comments_handled, and note_pr_review_comment_failure (bumps attempts, gives up at the cap). --- crates/claudear-storage/src/lib.rs | 25 ++++ crates/claudear-storage/src/migrator.rs | 12 +- crates/claudear-storage/src/sqlite.rs | 138 ++++++++++++++++++ .../V9__pr_review_states_issue_comments.sql | 11 ++ 4 files changed, 185 insertions(+), 1 deletion(-) diff --git a/crates/claudear-storage/src/lib.rs b/crates/claudear-storage/src/lib.rs index db4002b..94440bf 100644 --- a/crates/claudear-storage/src/lib.rs +++ b/crates/claudear-storage/src/lib.rs @@ -653,6 +653,31 @@ pub trait ActivityStore: Send + Sync { Ok(Vec::new()) } + /// Get review comments recorded for a PR that have not yet been durably + /// handled (their feedback has not been acted upon). Re-surfaced each poll so + /// review-comment processing is at-least-once: a crash or downstream failure + /// between detecting a comment and acting on it does not drop it. + fn get_unhandled_pr_review_comments( + &self, + _pr_url: &str, + ) -> Result> { + Ok(Vec::new()) + } + + /// Mark every not-yet-handled review comment on a PR as durably handled. + /// Called once the PR's review feedback has been successfully acted upon. + fn mark_pr_review_comments_handled(&self, _pr_url: &str) -> Result<()> { + Ok(()) + } + + /// Record a failed attempt to act on a PR's unhandled review comments: bumps + /// each unhandled comment's attempt count and gives up (marks it handled) once + /// the count reaches `max_attempts`, so a poison comment does not re-run the + /// fix agent forever. + fn note_pr_review_comment_failure(&self, _pr_url: &str, _max_attempts: i64) -> Result<()> { + Ok(()) + } + /// Get fix attempts for a batch of (source, issue_id) keys. fn get_attempts_batch(&self, _keys: &[(&str, &str)]) -> Result>> { Ok(Vec::new()) diff --git a/crates/claudear-storage/src/migrator.rs b/crates/claudear-storage/src/migrator.rs index b022a52..10ff542 100644 --- a/crates/claudear-storage/src/migrator.rs +++ b/crates/claudear-storage/src/migrator.rs @@ -143,7 +143,8 @@ mod tests { .unwrap(); assert_eq!(has_col, 1); - // Verify the V9 column exists on pr_review_states. + // Verify the V9 columns exist: the issue-comment cursor on pr_review_states + // and the handled ledger on pr_review_comments. let has_issue_comment_col: u32 = conn .query_row( "SELECT COUNT(*) FROM pragma_table_info('pr_review_states') WHERE name = 'last_issue_comment_id'", @@ -152,6 +153,15 @@ mod tests { ) .unwrap(); assert_eq!(has_issue_comment_col, 1); + + let has_handled_col: u32 = conn + .query_row( + "SELECT COUNT(*) FROM pragma_table_info('pr_review_comments') WHERE name = 'handled_at'", + [], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(has_handled_col, 1); } #[test] diff --git a/crates/claudear-storage/src/sqlite.rs b/crates/claudear-storage/src/sqlite.rs index bd8c1d7..515b2bd 100644 --- a/crates/claudear-storage/src/sqlite.rs +++ b/crates/claudear-storage/src/sqlite.rs @@ -1960,6 +1960,78 @@ impl ActivityStore for SqliteTracker { Ok(results) } + fn get_unhandled_pr_review_comments( + &self, + pr_url: &str, + ) -> Result> { + let conn = self.acquire_lock()?; + let mut stmt = conn.prepare( + r#" + SELECT scm_comment_id, review_id, path, position, line, + body, author, created_at, updated_at, html_url + FROM pr_review_comments + WHERE pr_url = ? AND handled_at IS NULL + ORDER BY scm_comment_id ASC + "#, + )?; + + let rows = stmt.query_map(params![pr_url], |row| { + Ok(claudear_core::types::ReviewComment { + id: row.get(0)?, + pull_request_review_id: row.get(1)?, + path: row.get(2)?, + position: row.get(3)?, + line: row.get(4)?, + body: row.get(5)?, + user: claudear_core::types::ReviewUser { + id: 0, + login: row.get(6)?, + user_type: None, + }, + created_at: row.get(7)?, + updated_at: row.get(8)?, + html_url: row.get(9)?, + original_position: None, + start_line: None, + side: None, + }) + })?; + + let mut results = Vec::new(); + for row in rows.flatten() { + results.push(row); + } + Ok(results) + } + + fn mark_pr_review_comments_handled(&self, pr_url: &str) -> Result<()> { + let conn = self.acquire_lock()?; + conn.execute( + "UPDATE pr_review_comments SET handled_at = datetime('now') \ + WHERE pr_url = ? AND handled_at IS NULL", + params![pr_url], + )?; + Ok(()) + } + + fn note_pr_review_comment_failure(&self, pr_url: &str, max_attempts: i64) -> Result<()> { + let conn = self.acquire_lock()?; + // Count this failed cycle against every still-unhandled comment on the PR. + conn.execute( + "UPDATE pr_review_comments SET attempts = attempts + 1 \ + WHERE pr_url = ? AND handled_at IS NULL", + params![pr_url], + )?; + // Give up on comments that have failed too many times so a poison comment + // does not re-trigger the fix agent forever. + conn.execute( + "UPDATE pr_review_comments SET handled_at = datetime('now') \ + WHERE pr_url = ? AND handled_at IS NULL AND attempts >= ?", + params![pr_url, max_attempts], + )?; + Ok(()) + } + /// Get metric row counts grouped by name since a timestamp. fn get_metric_counts_since( &self, @@ -11211,6 +11283,72 @@ mod tests { assert_eq!(comments[0].line, Some(42)); } + #[test] + fn test_pr_review_comment_handled_ledger() { + let tracker = SqliteTracker::in_memory().unwrap(); + let pr_url = "https://github.com/owner/repo/pull/7"; + + let mk = |id: i64| claudear_core::types::ReviewComment { + id, + path: String::new(), + position: None, + original_position: None, + body: "@claudear please fix".to_string(), + user: claudear_core::types::ReviewUser { + id: 1, + login: "reviewer".to_string(), + user_type: Some("User".to_string()), + }, + created_at: "2024-01-15T10:00:00Z".to_string(), + updated_at: "2024-01-15T10:00:00Z".to_string(), + html_url: format!("https://github.com/owner/repo/pull/7#c{}", id), + pull_request_review_id: None, + start_line: None, + line: None, + side: None, + }; + + // A freshly recorded comment is unhandled. + tracker.record_pr_review_comment(pr_url, &mk(1)).unwrap(); + tracker.record_pr_review_comment(pr_url, &mk(2)).unwrap(); + let unhandled = tracker.get_unhandled_pr_review_comments(pr_url).unwrap(); + assert_eq!(unhandled.len(), 2); + assert_eq!(unhandled[0].id, 1); + + // Marking handled clears them. + tracker.mark_pr_review_comments_handled(pr_url).unwrap(); + assert!(tracker + .get_unhandled_pr_review_comments(pr_url) + .unwrap() + .is_empty()); + + // Re-recording a handled comment does not resurrect it (handled_at kept). + tracker.record_pr_review_comment(pr_url, &mk(1)).unwrap(); + assert!(tracker + .get_unhandled_pr_review_comments(pr_url) + .unwrap() + .is_empty()); + + // A new comment fails repeatedly and is given up on at the cap. + tracker.record_pr_review_comment(pr_url, &mk(3)).unwrap(); + assert_eq!(tracker.get_unhandled_pr_review_comments(pr_url).unwrap().len(), 1); + tracker.note_pr_review_comment_failure(pr_url, 3).unwrap(); // attempts=1 + tracker.note_pr_review_comment_failure(pr_url, 3).unwrap(); // attempts=2 + assert_eq!( + tracker.get_unhandled_pr_review_comments(pr_url).unwrap().len(), + 1, + "should still retry below the cap" + ); + tracker.note_pr_review_comment_failure(pr_url, 3).unwrap(); // attempts=3 -> give up + assert!( + tracker + .get_unhandled_pr_review_comments(pr_url) + .unwrap() + .is_empty(), + "should stop retrying once the attempt cap is reached" + ); + } + #[test] fn test_get_comments_for_pr() { let tracker = SqliteTracker::in_memory().unwrap(); diff --git a/migrations/V9__pr_review_states_issue_comments.sql b/migrations/V9__pr_review_states_issue_comments.sql index 54d1ea7..d008f84 100644 --- a/migrations/V9__pr_review_states_issue_comments.sql +++ b/migrations/V9__pr_review_states_issue_comments.sql @@ -4,3 +4,14 @@ -- so the review watcher advances an independent cursor for them. ALTER TABLE pr_review_states ADD COLUMN last_issue_comment_id INTEGER; ALTER TABLE pr_review_states ADD COLUMN last_issue_comment_time TEXT; + +-- Make PR review-comment processing at-least-once. +-- The pr_review_comments ledger (keyed by scm_comment_id UNIQUE) becomes the +-- authority for "what still needs action", so a crash or downstream failure +-- between detecting a comment and acting on it no longer drops it: unhandled +-- rows are re-surfaced on the next poll regardless of the polling cursor. +-- handled_at: set once the comment's feedback has been durably acted upon. +-- attempts: consecutive processing failures; used to give up on a poison +-- comment instead of retrying the fix agent forever. +ALTER TABLE pr_review_comments ADD COLUMN handled_at TEXT; +ALTER TABLE pr_review_comments ADD COLUMN attempts INTEGER NOT NULL DEFAULT 0; From d349167e097c0909717928f2c97f02d4b7190296 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 11 Aug 2026 14:47:40 +0530 Subject: [PATCH 06/16] feat(review): process review comments at-least-once Re-surface any recorded-but-unhandled comments each poll so a crash or a failed fix run between detecting a comment and acting on it no longer drops it; the ledger, not the polling cursor, is the authority. The engine marks a PR's comments handled only after process_review_action succeeds, counts failures toward the give-up cap, and closes the ledger for terminal PRs. --- crates/claudear-engine/src/watcher.rs | 50 +++++++++-- crates/claudear-integrations/src/scm.rs | 105 ++++++++++++++++++++++++ 2 files changed, 146 insertions(+), 9 deletions(-) diff --git a/crates/claudear-engine/src/watcher.rs b/crates/claudear-engine/src/watcher.rs index e36fccc..331d37f 100644 --- a/crates/claudear-engine/src/watcher.rs +++ b/crates/claudear-engine/src/watcher.rs @@ -40,6 +40,10 @@ use tokio::time::{interval, Duration}; /// decided routing `Intent` (`None` for non-QA-eligible / QA-disabled sources). type QueuedIssue = (Issue, MatchResult, Option); +/// How many times a PR review comment may fail processing before it is given up +/// on (marked handled) instead of re-triggering the fix agent every cycle. +const MAX_REVIEW_COMMENT_ATTEMPTS: i64 = 5; + /// 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) @@ -1252,24 +1256,52 @@ impl Watcher { status = %attempt.status, "Skipping review feedback for terminal attempt status" ); + // The PR is merged/closed/cannot-fix: close out the ledger so + // its comments stop being re-surfaced, then stop watching. + if let Err(e) = self.tracker.mark_pr_review_comments_handled(&pr_url) { + tracing::warn!(pr_url = %pr_url, error = %e, "Failed to close review-comment ledger for terminal PR"); + } review_watcher.unwatch_pr(&pr_url); continue; } - if let Err(e) = self - .process_review_action(&attempt, &feedback_summary) - .await - { - tracing::error!( - pr_url = %pr_url, - error = %e, - "Failed to process review feedback" - ); + match self.process_review_action(&attempt, &feedback_summary).await { + Ok(()) => { + // Durably handled: mark the PR's comments so they aren't + // re-surfaced next cycle (at-least-once, exactly the success + // path marks completion). + if let Err(e) = self.tracker.mark_pr_review_comments_handled(&pr_url) { + tracing::warn!(pr_url = %pr_url, error = %e, "Failed to mark review comments handled"); + } + } + Err(e) => { + tracing::error!( + pr_url = %pr_url, + error = %e, + "Failed to process review feedback; will retry next cycle" + ); + // Leave comments unhandled so they retry, but count the + // failure so a poison comment eventually gives up. + if let Err(e) = self + .tracker + .note_pr_review_comment_failure(&pr_url, MAX_REVIEW_COMMENT_ATTEMPTS) + { + tracing::warn!(pr_url = %pr_url, error = %e, "Failed to record review-comment failure"); + } + } } } else { tracing::warn!( pr_url = %pr_url, "Received review for unknown PR, skipping" ); + // No attempt to act on; count it as a failure so unhandled ledger + // comments for an uncorrelated PR don't re-surface forever. + if let Err(e) = self + .tracker + .note_pr_review_comment_failure(&pr_url, MAX_REVIEW_COMMENT_ATTEMPTS) + { + tracing::warn!(pr_url = %pr_url, error = %e, "Failed to record review-comment failure"); + } } } diff --git a/crates/claudear-integrations/src/scm.rs b/crates/claudear-integrations/src/scm.rs index 86eca92..c061fb5 100644 --- a/crates/claudear-integrations/src/scm.rs +++ b/crates/claudear-integrations/src/scm.rs @@ -1380,6 +1380,45 @@ impl ReviewWatcher { // downstream fix loop sees a single feedback batch per PR per cycle. let mut added_comments = standalone_comments; added_comments.extend(conversation_feedback); + + // Re-surface any recorded comments not yet durably handled so a crash or + // downstream failure between detecting a comment and acting on it doesn't + // drop it. The pr_review_comments ledger, not the polling cursor, is the + // authority for outstanding work; the cursor is only a fetch-window + // optimization. Newly detected comments were already recorded above, so + // they come back through this query too — dedup against what we're already + // emitting this cycle (the batch plus inline comments on review events). + if let Some(ref tracker) = self.tracker { + let mut seen: std::collections::HashSet = + added_comments.iter().map(|c| c.id).collect(); + for event in &events { + if let ReviewEvent::ReviewSubmitted { + inline_comments, .. + } = event + { + seen.extend(inline_comments.iter().map(|c| c.id)); + } + } + match tracker.get_unhandled_pr_review_comments(&state.pr_url) { + Ok(pending) => { + for comment in pending { + if seen.insert(comment.id) { + added_comments.push(comment); + } + } + } + Err(e) => { + tracing::warn!( + component = "review_watcher", + pr_url = %state.pr_url, + error = %e, + "Failed to load unhandled PR review comments; \ + relying on freshly detected ones this cycle" + ); + } + } + } + if !added_comments.is_empty() { events.push(ReviewEvent::CommentsAdded { pr_url: state.pr_url.clone(), @@ -5634,6 +5673,7 @@ mod tests { use claudear_core::types::{ ActivityLogEntry, FixAttempt, FixAttemptStats, FixAttemptStatus, PrReviewRecord, }; + use crate::scm::ReviewEvent; use claudear_storage::{ ActivityStore, AttemptTracker, ChatStore, DiscordStore, EmbeddingStore, EvaluationStore, ExperimentStore, FixAttemptTracker, KnowledgeStore, RegressionStore, @@ -5713,6 +5753,8 @@ mod tests { struct MockTrackerWithRecording { review_calls: Arc>>, activity_calls: Arc>>, + unhandled: Arc>>, + handled_calls: Arc>>, } impl MockTrackerWithRecording { @@ -5720,8 +5762,14 @@ mod tests { Self { review_calls: Arc::new(Mutex::new(Vec::new())), activity_calls: Arc::new(Mutex::new(Vec::new())), + unhandled: Arc::new(Mutex::new(Vec::new())), + handled_calls: Arc::new(Mutex::new(Vec::new())), } } + + fn set_unhandled(&self, comments: Vec) { + *self.unhandled.lock().unwrap() = comments; + } } impl AttemptTracker for MockTrackerWithRecording { @@ -5824,6 +5872,16 @@ mod tests { .push(entry.activity_type.clone()); Ok(1) } + fn get_unhandled_pr_review_comments( + &self, + _pr_url: &str, + ) -> Result> { + Ok(self.unhandled.lock().unwrap().clone()) + } + fn mark_pr_review_comments_handled(&self, pr_url: &str) -> Result<()> { + self.handled_calls.lock().unwrap().push(pr_url.to_string()); + Ok(()) + } } impl KnowledgeStore for MockTrackerWithRecording {} @@ -5921,6 +5979,53 @@ mod tests { assert_eq!(activity_calls[0], "pr_review_received"); } + #[tokio::test] + async fn test_unhandled_comments_are_resurfaced() { + // No new reviews/comments from GitHub this cycle, but the ledger still + // holds an unhandled comment (e.g. a prior cycle detected it but the + // fix run failed). It must be re-emitted so processing is at-least-once. + let provider: Arc = + Arc::new(MockScmProvider::new("github", true, "@claudear")); + let tracker = Arc::new(MockTrackerWithRecording::new()); + tracker.set_unhandled(vec![ReviewComment { + id: 4242, + path: String::new(), + position: None, + original_position: None, + body: "@claudear please address the shutdown race".to_string(), + user: ReviewUser { + id: 0, + login: "reviewer".to_string(), + user_type: None, + }, + created_at: "2025-01-01T00:00:00Z".to_string(), + updated_at: "2025-01-01T00:00:00Z".to_string(), + html_url: "https://github.com/org/repo/pull/1#issuecomment-4242".to_string(), + pull_request_review_id: None, + start_line: None, + line: None, + side: None, + }]); + let watcher = ReviewWatcher::with_tracker(provider, tracker.clone()); + watcher.watch_pr(make_state( + "https://github.com/org/repo/pull/1", + "org/repo", + 1, + )); + + let events = watcher.check_for_reviews().await.unwrap(); + let comment_events: Vec<_> = events + .iter() + .filter_map(|e| match e { + ReviewEvent::CommentsAdded { comments, .. } => Some(comments), + _ => None, + }) + .collect(); + assert_eq!(comment_events.len(), 1, "unhandled comment not re-surfaced"); + assert_eq!(comment_events[0].len(), 1); + assert_eq!(comment_events[0][0].id, 4242); + } + #[tokio::test] async fn test_record_review_to_db_no_submitted_at() { let mock = MockScmProvider::new("github", true, "@claudear"); From e1307c0b2d28030958b05f86387bc59843942e07 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 11 Aug 2026 14:49:06 +0530 Subject: [PATCH 07/16] empty to trigger tests From faa1298525b4d3ac53ac2249a0f1331244a9b0a5 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 11 Aug 2026 14:57:23 +0530 Subject: [PATCH 08/16] fix(review): record conversation comments before advancing cursor The issue-comment cursor was saved before the comment was written to the recovery ledger, so a crash or a failed record in that window advanced the cursor past a comment the ledger never captured, permanently dropping it. Record actionable comments first and hold the cursor back when any record fails, so the comment is re-fetched next poll. Matches the inline path. --- crates/claudear-integrations/src/scm.rs | 61 +++++++++++++++---------- 1 file changed, 36 insertions(+), 25 deletions(-) diff --git a/crates/claudear-integrations/src/scm.rs b/crates/claudear-integrations/src/scm.rs index c061fb5..cb64afb 100644 --- a/crates/claudear-integrations/src/scm.rs +++ b/crates/claudear-integrations/src/scm.rs @@ -1311,9 +1311,42 @@ impl ReviewWatcher { } }; - if !new_conversation_comments.is_empty() { - // Advance the issue-comment cursor over ALL new conversation comments - // (including non-trigger ones) so unchanged comments aren't rescanned. + // Only trigger-matched conversation comments are actionable feedback. + let conversation_feedback: Vec = new_conversation_comments + .iter() + .filter(|c| trigger.is_empty() || c.body.to_lowercase().contains(&trigger.to_lowercase())) + .cloned() + .collect(); + + // Record actionable comments into the durable ledger BEFORE advancing the + // cursor. The ledger is the recovery authority: if the cursor moved past a + // comment the ledger never captured, that comment would be neither + // re-fetched (cursor advanced) nor re-surfaced (no ledger row) — a + // permanent drop. If any record fails, hold the cursor back so the comment + // is re-fetched next poll rather than lost. + let mut all_recorded = true; + if !conversation_feedback.is_empty() { + if let Some(ref tracker) = self.tracker { + for comment in &conversation_feedback { + if let Err(e) = tracker.record_pr_review_comment(&state.pr_url, comment) { + all_recorded = false; + tracing::warn!( + component = "review_watcher", + pr_url = %state.pr_url, + comment_id = comment.id, + error = %e, + "Failed to record PR conversation comment; holding issue cursor" + ); + } + } + } + } + + // Advance the issue-comment cursor over all fetched comments (including + // non-trigger ones, so unchanged comments aren't rescanned), but only once + // every actionable comment is durably recorded. The cursor is a scan-window + // optimization; the ledger, not the cursor, guarantees no comment is lost. + if all_recorded && !new_conversation_comments.is_empty() { let mut latest_id = state.last_issue_comment_id; let mut latest_time = state.last_issue_comment_time.clone(); for comment in &new_conversation_comments { @@ -1354,28 +1387,6 @@ impl ReviewWatcher { } } - // Only trigger-matched conversation comments are actionable feedback. - let conversation_feedback: Vec = new_conversation_comments - .into_iter() - .filter(|c| trigger.is_empty() || c.body.to_lowercase().contains(&trigger.to_lowercase())) - .collect(); - - if !conversation_feedback.is_empty() { - if let Some(ref tracker) = self.tracker { - for comment in &conversation_feedback { - if let Err(e) = tracker.record_pr_review_comment(&state.pr_url, comment) { - tracing::warn!( - component = "review_watcher", - pr_url = %state.pr_url, - comment_id = comment.id, - error = %e, - "Failed to record PR conversation comment" - ); - } - } - } - } - // Emit inline standalone comments and conversation comments together so the // downstream fix loop sees a single feedback batch per PR per cycle. let mut added_comments = standalone_comments; From 4ee6936df1b43b8615c99f555e483a120e23190a Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 11 Aug 2026 15:14:09 +0530 Subject: [PATCH 09/16] fix(review): acknowledge review comments by id, not whole PR Marking every unhandled comment on a PR handled after a batch succeeded could clear a comment recorded concurrently (e.g. a webhook check_for_pr) while process_review_action was awaiting, dropping its work. Carry the batch's comment ids through group_review_feedback_by_pr and acknowledge (or count failures against) exactly those ids. PR-wide handling is kept only for terminal PRs. --- crates/claudear-engine/src/watcher.rs | 104 +++++++++++++++---- crates/claudear-integrations/src/scm.rs | 16 +++ crates/claudear-storage/src/lib.rs | 28 ++++-- crates/claudear-storage/src/sqlite.rs | 128 +++++++++++++++++++++--- 4 files changed, 239 insertions(+), 37 deletions(-) diff --git a/crates/claudear-engine/src/watcher.rs b/crates/claudear-engine/src/watcher.rs index 331d37f..d73b2c6 100644 --- a/crates/claudear-engine/src/watcher.rs +++ b/crates/claudear-engine/src/watcher.rs @@ -1238,7 +1238,8 @@ impl Watcher { // Check for new reviews let events = review_watcher.check_for_reviews().await?; - for (pr_url, feedback_summary, feedback_count) in Self::group_review_feedback_by_pr(events) + for (pr_url, feedback_summary, feedback_count, comment_ids) in + Self::group_review_feedback_by_pr(events) { tracing::info!( pr_url = %pr_url, @@ -1266,10 +1267,13 @@ impl Watcher { } match self.process_review_action(&attempt, &feedback_summary).await { Ok(()) => { - // Durably handled: mark the PR's comments so they aren't - // re-surfaced next cycle (at-least-once, exactly the success - // path marks completion). - if let Err(e) = self.tracker.mark_pr_review_comments_handled(&pr_url) { + // Durably handled: acknowledge exactly the comments in this + // batch so they aren't re-surfaced, without touching any + // comment recorded concurrently while we were processing. + if let Err(e) = self + .tracker + .mark_pr_review_comments_handled_by_ids(&pr_url, &comment_ids) + { tracing::warn!(pr_url = %pr_url, error = %e, "Failed to mark review comments handled"); } } @@ -1279,12 +1283,13 @@ impl Watcher { error = %e, "Failed to process review feedback; will retry next cycle" ); - // Leave comments unhandled so they retry, but count the - // failure so a poison comment eventually gives up. - if let Err(e) = self - .tracker - .note_pr_review_comment_failure(&pr_url, MAX_REVIEW_COMMENT_ATTEMPTS) - { + // Leave the batch's comments unhandled so they retry, but + // count the failure so a poison comment eventually gives up. + if let Err(e) = self.tracker.note_pr_review_comment_failure_by_ids( + &pr_url, + &comment_ids, + MAX_REVIEW_COMMENT_ATTEMPTS, + ) { tracing::warn!(pr_url = %pr_url, error = %e, "Failed to record review-comment failure"); } } @@ -1294,12 +1299,13 @@ impl Watcher { pr_url = %pr_url, "Received review for unknown PR, skipping" ); - // No attempt to act on; count it as a failure so unhandled ledger - // comments for an uncorrelated PR don't re-surface forever. - if let Err(e) = self - .tracker - .note_pr_review_comment_failure(&pr_url, MAX_REVIEW_COMMENT_ATTEMPTS) - { + // No attempt to act on; count it as a failure so this batch's + // uncorrelated comments don't re-surface forever. + if let Err(e) = self.tracker.note_pr_review_comment_failure_by_ids( + &pr_url, + &comment_ids, + MAX_REVIEW_COMMENT_ATTEMPTS, + ) { tracing::warn!(pr_url = %pr_url, error = %e, "Failed to record review-comment failure"); } } @@ -1315,9 +1321,17 @@ impl Watcher { ) } - fn group_review_feedback_by_pr(events: Vec) -> Vec<(String, String, usize)> { + /// Group actionable review events per PR into (pr_url, feedback_summary, + /// feedback_count, comment_ids). `comment_ids` are exactly the ledger comments + /// carried by the batch, so acknowledgement can target them rather than the + /// whole PR (which would wrongly mark a concurrently-recorded comment handled). + fn group_review_feedback_by_pr( + events: Vec, + ) -> Vec<(String, String, usize, Vec)> { let mut feedback_by_pr: std::collections::HashMap> = std::collections::HashMap::new(); + let mut ids_by_pr: std::collections::HashMap> = + std::collections::HashMap::new(); let mut pr_order: Vec = Vec::new(); for event in events { @@ -1329,6 +1343,10 @@ impl Watcher { if !feedback_by_pr.contains_key(&pr_url) { pr_order.push(pr_url.clone()); } + ids_by_pr + .entry(pr_url.clone()) + .or_default() + .extend(event.comment_ids()); feedback_by_pr .entry(pr_url) .or_default() @@ -1338,9 +1356,10 @@ impl Watcher { pr_order .into_iter() .filter_map(|pr_url| { + let ids = ids_by_pr.remove(&pr_url).unwrap_or_default(); feedback_by_pr.remove(&pr_url).map(|feedbacks| { let count = feedbacks.len(); - (pr_url, feedbacks.join("\n\n---\n\n"), count) + (pr_url, feedbacks.join("\n\n---\n\n"), count, ids) }) }) .collect() @@ -6323,6 +6342,53 @@ mod tests { assert!(grouped[0].1.contains("first")); assert!(grouped[0].1.contains("second")); assert!(grouped[0].1.contains("---")); + // Reviews carry no ledger comment ids. + assert!(grouped[0].3.is_empty()); + } + + #[test] + fn test_group_review_feedback_collects_comment_ids() { + // CommentsAdded events contribute their comment ids so acknowledgement can + // target exactly the batch, not the whole PR. + let mk = |id: i64| claudear_integrations::scm::ReviewComment { + id, + path: String::new(), + position: None, + original_position: None, + body: "@claudear fix".to_string(), + user: claudear_integrations::scm::ReviewUser { + id: 1, + login: "reviewer".to_string(), + user_type: None, + }, + created_at: "2024-01-01T00:00:00Z".to_string(), + updated_at: "2024-01-01T00:00:00Z".to_string(), + html_url: format!("h{}", id), + pull_request_review_id: None, + start_line: None, + line: None, + side: None, + }; + let events = vec![ + claudear_integrations::scm::ReviewEvent::CommentsAdded { + pr_url: "https://github.com/org/repo/pull/1".to_string(), + repo: "org/repo".to_string(), + pr_number: 1, + comments: vec![mk(101), mk(102)], + }, + claudear_integrations::scm::ReviewEvent::CommentsAdded { + pr_url: "https://github.com/org/repo/pull/1".to_string(), + repo: "org/repo".to_string(), + pr_number: 1, + comments: vec![mk(103)], + }, + ]; + + let grouped = Watcher::group_review_feedback_by_pr(events); + assert_eq!(grouped.len(), 1); + let mut ids = grouped[0].3.clone(); + ids.sort(); + assert_eq!(ids, vec![101, 102, 103]); } #[tokio::test] diff --git a/crates/claudear-integrations/src/scm.rs b/crates/claudear-integrations/src/scm.rs index cb64afb..e707a58 100644 --- a/crates/claudear-integrations/src/scm.rs +++ b/crates/claudear-integrations/src/scm.rs @@ -393,6 +393,22 @@ impl ReviewEvent { } } + /// The ledger comment ids carried by this event: standalone/conversation + /// comments for `CommentsAdded`, inline comments for `ReviewSubmitted`. Used to + /// acknowledge exactly the comments in a processed batch, so a comment recorded + /// concurrently (e.g. a webhook `check_for_pr`) isn't marked handled without + /// being processed. A formal review has no ledger row and contributes none. + pub fn comment_ids(&self) -> Vec { + match self { + ReviewEvent::ReviewSubmitted { + inline_comments, .. + } => inline_comments.iter().map(|c| c.id).collect(), + ReviewEvent::CommentsAdded { comments, .. } => { + comments.iter().map(|c| c.id).collect() + } + } + } + /// Check if this event requires agent action. pub fn requires_action(&self) -> bool { match self { diff --git a/crates/claudear-storage/src/lib.rs b/crates/claudear-storage/src/lib.rs index 94440bf..0e6df2c 100644 --- a/crates/claudear-storage/src/lib.rs +++ b/crates/claudear-storage/src/lib.rs @@ -665,16 +665,32 @@ pub trait ActivityStore: Send + Sync { } /// Mark every not-yet-handled review comment on a PR as durably handled. - /// Called once the PR's review feedback has been successfully acted upon. + /// Use only when the whole PR is done (merged/closed); to acknowledge a + /// processed batch, prefer [`mark_pr_review_comments_handled_by_ids`] so a + /// concurrently-recorded comment isn't marked without being processed. fn mark_pr_review_comments_handled(&self, _pr_url: &str) -> Result<()> { Ok(()) } - /// Record a failed attempt to act on a PR's unhandled review comments: bumps - /// each unhandled comment's attempt count and gives up (marks it handled) once - /// the count reaches `max_attempts`, so a poison comment does not re-run the - /// fix agent forever. - fn note_pr_review_comment_failure(&self, _pr_url: &str, _max_attempts: i64) -> Result<()> { + /// Mark only the given comment ids on a PR as durably handled. Called once the + /// batch that carried exactly these comments has been successfully acted upon. + fn mark_pr_review_comments_handled_by_ids( + &self, + _pr_url: &str, + _comment_ids: &[i64], + ) -> Result<()> { + Ok(()) + } + + /// Record a failed attempt to act on the given comment ids: bumps each one's + /// attempt count and gives up (marks it handled) once the count reaches + /// `max_attempts`, so a poison comment does not re-run the fix agent forever. + fn note_pr_review_comment_failure_by_ids( + &self, + _pr_url: &str, + _comment_ids: &[i64], + _max_attempts: i64, + ) -> Result<()> { Ok(()) } diff --git a/crates/claudear-storage/src/sqlite.rs b/crates/claudear-storage/src/sqlite.rs index 515b2bd..106e41b 100644 --- a/crates/claudear-storage/src/sqlite.rs +++ b/crates/claudear-storage/src/sqlite.rs @@ -2014,20 +2014,82 @@ impl ActivityStore for SqliteTracker { Ok(()) } - fn note_pr_review_comment_failure(&self, pr_url: &str, max_attempts: i64) -> Result<()> { + fn mark_pr_review_comments_handled_by_ids( + &self, + pr_url: &str, + comment_ids: &[i64], + ) -> Result<()> { + if comment_ids.is_empty() { + return Ok(()); + } let conn = self.acquire_lock()?; - // Count this failed cycle against every still-unhandled comment on the PR. - conn.execute( + let placeholders = vec!["?"; comment_ids.len()].join(", "); + let sql = format!( + "UPDATE pr_review_comments SET handled_at = datetime('now') \ + WHERE pr_url = ?1 AND handled_at IS NULL AND scm_comment_id IN ({})", + placeholders + ); + let mut binds: Vec> = Vec::with_capacity(comment_ids.len() + 1); + binds.push(Box::new(pr_url.to_string())); + for id in comment_ids { + binds.push(Box::new(*id)); + } + conn.execute(&sql, rusqlite::params_from_iter(binds.iter().map(|b| &**b)))?; + Ok(()) + } + + fn note_pr_review_comment_failure_by_ids( + &self, + pr_url: &str, + comment_ids: &[i64], + max_attempts: i64, + ) -> Result<()> { + if comment_ids.is_empty() { + return Ok(()); + } + let conn = self.acquire_lock()?; + let placeholders = vec!["?"; comment_ids.len()].join(", "); + + // Count this failed cycle against exactly the comments we tried. + let bump_sql = format!( "UPDATE pr_review_comments SET attempts = attempts + 1 \ - WHERE pr_url = ? AND handled_at IS NULL", - params![pr_url], - )?; + WHERE pr_url = ?1 AND handled_at IS NULL AND scm_comment_id IN ({})", + placeholders + ); // Give up on comments that have failed too many times so a poison comment // does not re-trigger the fix agent forever. - conn.execute( + let giveup_sql = format!( "UPDATE pr_review_comments SET handled_at = datetime('now') \ - WHERE pr_url = ? AND handled_at IS NULL AND attempts >= ?", - params![pr_url, max_attempts], + WHERE pr_url = ?1 AND handled_at IS NULL AND attempts >= ?2 \ + AND scm_comment_id IN ({})", + // ids start at bind position 3 for the give-up query + (0..comment_ids.len()) + .map(|i| format!("?{}", i + 3)) + .collect::>() + .join(", ") + ); + + let mut bump_binds: Vec> = + Vec::with_capacity(comment_ids.len() + 1); + bump_binds.push(Box::new(pr_url.to_string())); + for id in comment_ids { + bump_binds.push(Box::new(*id)); + } + conn.execute( + &bump_sql, + rusqlite::params_from_iter(bump_binds.iter().map(|b| &**b)), + )?; + + let mut giveup_binds: Vec> = + Vec::with_capacity(comment_ids.len() + 2); + giveup_binds.push(Box::new(pr_url.to_string())); + giveup_binds.push(Box::new(max_attempts)); + for id in comment_ids { + giveup_binds.push(Box::new(*id)); + } + conn.execute( + &giveup_sql, + rusqlite::params_from_iter(giveup_binds.iter().map(|b| &**b)), )?; Ok(()) } @@ -11332,14 +11394,20 @@ mod tests { // A new comment fails repeatedly and is given up on at the cap. tracker.record_pr_review_comment(pr_url, &mk(3)).unwrap(); assert_eq!(tracker.get_unhandled_pr_review_comments(pr_url).unwrap().len(), 1); - tracker.note_pr_review_comment_failure(pr_url, 3).unwrap(); // attempts=1 - tracker.note_pr_review_comment_failure(pr_url, 3).unwrap(); // attempts=2 + tracker + .note_pr_review_comment_failure_by_ids(pr_url, &[3], 3) + .unwrap(); // attempts=1 + tracker + .note_pr_review_comment_failure_by_ids(pr_url, &[3], 3) + .unwrap(); // attempts=2 assert_eq!( tracker.get_unhandled_pr_review_comments(pr_url).unwrap().len(), 1, "should still retry below the cap" ); - tracker.note_pr_review_comment_failure(pr_url, 3).unwrap(); // attempts=3 -> give up + tracker + .note_pr_review_comment_failure_by_ids(pr_url, &[3], 3) + .unwrap(); // attempts=3 -> give up assert!( tracker .get_unhandled_pr_review_comments(pr_url) @@ -11349,6 +11417,42 @@ mod tests { ); } + #[test] + fn test_mark_handled_by_ids_targets_only_given_comments() { + let tracker = SqliteTracker::in_memory().unwrap(); + let pr_url = "https://github.com/owner/repo/pull/9"; + + let mk = |id: i64| claudear_core::types::ReviewComment { + id, + path: String::new(), + position: None, + original_position: None, + body: "@claudear fix".to_string(), + user: claudear_core::types::ReviewUser { + id: 1, + login: "reviewer".to_string(), + user_type: None, + }, + created_at: "2024-01-15T10:00:00Z".to_string(), + updated_at: "2024-01-15T10:00:00Z".to_string(), + html_url: format!("h{}", id), + pull_request_review_id: None, + start_line: None, + line: None, + side: None, + }; + tracker.record_pr_review_comment(pr_url, &mk(1)).unwrap(); + tracker.record_pr_review_comment(pr_url, &mk(2)).unwrap(); + + // Acknowledge only comment 1; comment 2 (recorded concurrently) survives. + tracker + .mark_pr_review_comments_handled_by_ids(pr_url, &[1]) + .unwrap(); + let unhandled = tracker.get_unhandled_pr_review_comments(pr_url).unwrap(); + assert_eq!(unhandled.len(), 1); + assert_eq!(unhandled[0].id, 2); + } + #[test] fn test_get_comments_for_pr() { let tracker = SqliteTracker::in_memory().unwrap(); From 1f835c77f44cc374e7f2ffdc58cb6cd27db47083 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 11 Aug 2026 16:34:07 +0530 Subject: [PATCH 10/16] fix(red-green): fail closed when reproduction can't be confirmed require_red_green is an enforcement gate, but two paths fell through into the mutation pipeline unguarded: no test tool detected, and a red-phase agent error. Both now abort the attempt (record, mark failed, clean up the worktree, return Failed), matching the existing not-reproduced path, so a fix never proceeds without a confirmed failing reproduction. --- crates/claudear-engine/src/processing.rs | 57 +++++++++++++++++++++--- 1 file changed, 52 insertions(+), 5 deletions(-) diff --git a/crates/claudear-engine/src/processing.rs b/crates/claudear-engine/src/processing.rs index 2e0c4d5..6632073 100644 --- a/crates/claudear-engine/src/processing.rs +++ b/crates/claudear-engine/src/processing.rs @@ -538,10 +538,31 @@ impl IssueProcessor { .iter() .any(|s| s.category == claudear_core::types::EvalCategory::Test); if !has_test_baseline { - tracing::warn!( - short_id = %issue.short_id, - "require_red_green set but no test tool detected; skipping red-green" + // Fail closed: require_red_green demands a confirmed failing + // reproduction before any mutation. With no test tool we cannot + // author or run a failing test, so abort rather than fall through + // into the fix pipeline unguarded. + let error = "Red-green: require_red_green is set but no test tool was detected; cannot confirm a failing reproduction".to_string(); + tracing::warn!(short_id = %issue.short_id, "{}", error); + let _ = self.tracker.record_action_run( + source_name, + &issue.id, + &issue.short_id, + "red_green", + "no_test_tool", + &error, ); + self.record_issue_decision( + issue, + "red_green_no_test_tool", + error.clone(), + json!({}), + ); + self.tracker + .mark_failed(source_name, &issue.id, &error) + .ok(); + self.cleanup_worktree(resolution, issue, &project_dir).await; + return Ok(ProcessingOutcome::Failed { error }); } else { self.record_timeline_event( issue, @@ -627,8 +648,34 @@ impl IssueProcessor { ); } Err(e) => { - // Agent infra error: skip the red gate rather than fail the attempt. - tracing::warn!(short_id = %issue.short_id, error = %e, "Red phase agent run failed; skipping red-green"); + // Fail closed: a red-phase agent error means we never + // confirmed a failing reproduction. Under require_red_green + // that gate is mandatory, so abort rather than proceed into + // the mutation pipeline unguarded. + let error = format!( + "Red-green: red-phase agent run failed; cannot confirm a failing reproduction: {}", + e + ); + tracing::warn!(short_id = %issue.short_id, error = %e, "Red phase agent run failed; failing closed under require_red_green"); + let _ = self.tracker.record_action_run( + source_name, + &issue.id, + &issue.short_id, + "red_green", + "agent_error", + &error, + ); + self.record_issue_decision( + issue, + "red_green_agent_error", + error.clone(), + json!({}), + ); + self.tracker + .mark_failed(source_name, &issue.id, &error) + .ok(); + self.cleanup_worktree(resolution, issue, &project_dir).await; + return Ok(ProcessingOutcome::Failed { error }); } } } From a5244e67c1d4e44bd9fce7448ff25c6fc5ebaef1 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 11 Aug 2026 16:42:25 +0530 Subject: [PATCH 11/16] fix(review): hold inline cursor when a comment record fails MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The inline/standalone path recorded before saving its cursor, but ignored record errors and advanced the cursor anyway — a failed write left the comment neither in the ledger nor re-fetchable, dropping it. Track record success and hold the inline cursor when any write fails, so the comment is re-fetched next poll. Matches the conversation path. --- crates/claudear-integrations/src/scm.rs | 63 +++++++++++++++++++++++-- 1 file changed, 59 insertions(+), 4 deletions(-) diff --git a/crates/claudear-integrations/src/scm.rs b/crates/claudear-integrations/src/scm.rs index e707a58..0bf03d5 100644 --- a/crates/claudear-integrations/src/scm.rs +++ b/crates/claudear-integrations/src/scm.rs @@ -1207,8 +1207,12 @@ impl ReviewWatcher { .cloned() .collect(); + // Record new comments to the durable ledger. Track whether every write + // succeeded: if any failed, the inline cursor is held back below so the + // affected comment is re-fetched next poll rather than lost (a comment that + // is neither recorded nor re-fetchable would be dropped). + let mut inline_records_ok = true; if !attached_comment_ids.is_empty() || !standalone_comments.is_empty() { - // Record all new comments to database if let Some(ref tracker) = self.tracker { for event in &events { if let ReviewEvent::ReviewSubmitted { @@ -1218,12 +1222,13 @@ impl ReviewWatcher { for comment in inline_comments { if let Err(e) = tracker.record_pr_review_comment(&state.pr_url, comment) { + inline_records_ok = false; tracing::warn!( component = "review_watcher", pr_url = %state.pr_url, comment_id = comment.id, error = %e, - "Failed to record PR review comment" + "Failed to record PR review comment; holding cursor" ); } } @@ -1231,19 +1236,22 @@ impl ReviewWatcher { } for comment in &standalone_comments { if let Err(e) = tracker.record_pr_review_comment(&state.pr_url, comment) { + inline_records_ok = false; tracing::warn!( component = "review_watcher", pr_url = %state.pr_url, comment_id = comment.id, error = %e, - "Failed to record PR review comment" + "Failed to record PR review comment; holding cursor" ); } } } } - if !cursor_comments.is_empty() { + // Advance the inline-comment cursor only once every comment is durably + // recorded; the ledger, not the cursor, guarantees no comment is lost. + if inline_records_ok && !cursor_comments.is_empty() { // Update state cursor using all processed comments (including non-trigger comments) // to prevent repeatedly scanning unchanged comments every poll cycle. let mut latest_comment_id = state.last_comment_id; @@ -5782,6 +5790,7 @@ mod tests { activity_calls: Arc>>, unhandled: Arc>>, handled_calls: Arc>>, + fail_record: Arc>, } impl MockTrackerWithRecording { @@ -5791,12 +5800,17 @@ mod tests { activity_calls: Arc::new(Mutex::new(Vec::new())), unhandled: Arc::new(Mutex::new(Vec::new())), handled_calls: Arc::new(Mutex::new(Vec::new())), + fail_record: Arc::new(Mutex::new(false)), } } fn set_unhandled(&self, comments: Vec) { *self.unhandled.lock().unwrap() = comments; } + + fn set_fail_record(&self, fail: bool) { + *self.fail_record.lock().unwrap() = fail; + } } impl AttemptTracker for MockTrackerWithRecording { @@ -5899,6 +5913,18 @@ mod tests { .push(entry.activity_type.clone()); Ok(1) } + fn record_pr_review_comment( + &self, + _pr_url: &str, + _comment: &ReviewComment, + ) -> Result { + if *self.fail_record.lock().unwrap() { + return Err(claudear_core::error::Error::Other( + "simulated record failure".to_string(), + )); + } + Ok(1) + } fn get_unhandled_pr_review_comments( &self, _pr_url: &str, @@ -6006,6 +6032,35 @@ mod tests { assert_eq!(activity_calls[0], "pr_review_received"); } + #[tokio::test] + async fn test_failed_comment_record_holds_cursor() { + // If recording a standalone comment fails, the inline cursor must NOT + // advance past it — otherwise it is neither in the ledger nor re-fetched, + // and its feedback is lost. + let mock = MockScmProvider::new("github", true, "@claudear"); + mock.set_comments(vec![make_comment( + 55, + "@claudear please fix", + "2025-01-02T00:00:00Z", + None, + )]); + let provider: Arc = Arc::new(mock); + let tracker = Arc::new(MockTrackerWithRecording::new()); + tracker.set_fail_record(true); + let watcher = ReviewWatcher::with_tracker(provider, tracker.clone()); + let pr_url = "https://github.com/org/repo/pull/1"; + watcher.watch_pr(make_state(pr_url, "org/repo", 1)); + + let _ = watcher.check_for_reviews().await.unwrap(); + + // Cursor held at its initial (unset) value so the comment is re-fetched. + let state = watcher.get_state(pr_url).unwrap(); + assert_eq!( + state.last_comment_id, None, + "cursor advanced past a comment that failed to record" + ); + } + #[tokio::test] async fn test_unhandled_comments_are_resurfaced() { // No new reviews/comments from GitHub this cycle, but the ledger still From 4f9d96ce6ddfcb273b45f7dce84f0b9224208456 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 11 Aug 2026 17:57:44 +0530 Subject: [PATCH 12/16] fix(review): namespace comment ledger and retire worst offender per cycle P1-a: inline and conversation comments come from distinct GitHub id sequences, but pr_review_comments keyed uniqueness on scm_comment_id alone, so a colliding id let one overwrite the other (inheriting handled state). Rebuild the table in V9 (shadow-copy the rows, recreate with comment_kind and UNIQUE(comment_kind, scm_comment_id), copy back) and set comment_kind on record. A column-level UNIQUE can't be dropped in place, hence the rebuild. P1-b: a batch failure can't be blamed on one comment, so note_failure now retires only the single worst offender (most attempts, at cap) per cycle instead of the whole over-cap set. A valid comment that co-occurs with a poison one drains out on its own merits rather than being dropped alongside. --- crates/claudear-storage/src/migrator.rs | 9 ++ crates/claudear-storage/src/sqlite.rs | 122 +++++++++++++++++- .../V9__pr_review_states_issue_comments.sql | 63 +++++++-- 3 files changed, 177 insertions(+), 17 deletions(-) diff --git a/crates/claudear-storage/src/migrator.rs b/crates/claudear-storage/src/migrator.rs index 10ff542..cc3c728 100644 --- a/crates/claudear-storage/src/migrator.rs +++ b/crates/claudear-storage/src/migrator.rs @@ -162,6 +162,15 @@ mod tests { ) .unwrap(); assert_eq!(has_handled_col, 1); + + let has_kind_col: u32 = conn + .query_row( + "SELECT COUNT(*) FROM pragma_table_info('pr_review_comments') WHERE name = 'comment_kind'", + [], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(has_kind_col, 1); } #[test] diff --git a/crates/claudear-storage/src/sqlite.rs b/crates/claudear-storage/src/sqlite.rs index 106e41b..7ca1cd1 100644 --- a/crates/claudear-storage/src/sqlite.rs +++ b/crates/claudear-storage/src/sqlite.rs @@ -1908,19 +1908,30 @@ impl ActivityStore for SqliteTracker { ) -> Result { let conn = self.acquire_lock()?; + // Conversation comments (issues/{n}/comments) carry no file path; inline + // review comments (pulls/{n}/comments) always do. The two are separate + // GitHub id sequences, so uniqueness is keyed on (comment_kind, id) to keep + // a colliding id in one namespace from overwriting the other's row. + let comment_kind = if comment.path.is_empty() { + "conversation" + } else { + "inline" + }; + conn.execute( r#" INSERT INTO pr_review_comments ( - scm_comment_id, pr_url, review_id, path, position, line, + scm_comment_id, comment_kind, pr_url, review_id, path, position, line, body, author, created_at, updated_at, html_url ) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11) - ON CONFLICT(scm_comment_id) DO UPDATE SET + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12) + ON CONFLICT(comment_kind, scm_comment_id) DO UPDATE SET body = excluded.body, updated_at = excluded.updated_at "#, params![ comment.id, + comment_kind, pr_url, comment.pull_request_review_id, comment.path, @@ -2056,12 +2067,21 @@ impl ActivityStore for SqliteTracker { WHERE pr_url = ?1 AND handled_at IS NULL AND scm_comment_id IN ({})", placeholders ); - // Give up on comments that have failed too many times so a poison comment - // does not re-trigger the fix agent forever. + // Give up on at most ONE comment per failed cycle: the single worst + // offender (most attempts) that has hit the cap. A batch failure can't be + // attributed to a specific comment, so retiring the whole over-cap set at + // once would drop valid comments that merely co-occur with a poison one. + // Retiring one per cycle lets a poison comment drain out first, after which + // the remaining comments get a fresh batch that can succeed. let giveup_sql = format!( "UPDATE pr_review_comments SET handled_at = datetime('now') \ - WHERE pr_url = ?1 AND handled_at IS NULL AND attempts >= ?2 \ - AND scm_comment_id IN ({})", + WHERE id = ( \ + SELECT id FROM pr_review_comments \ + WHERE pr_url = ?1 AND handled_at IS NULL AND attempts >= ?2 \ + AND scm_comment_id IN ({}) \ + ORDER BY attempts DESC, scm_comment_id ASC \ + LIMIT 1 \ + )", // ids start at bind position 3 for the give-up query (0..comment_ids.len()) .map(|i| format!("?{}", i + 3)) @@ -11453,6 +11473,94 @@ mod tests { assert_eq!(unhandled[0].id, 2); } + #[test] + fn test_inline_and_conversation_same_id_coexist() { + // The same numeric GitHub id can appear as both an inline review comment + // and a PR conversation comment (distinct id sequences). The composite + // (comment_kind, scm_comment_id) key must let both rows exist rather than + // one overwriting the other. + let tracker = SqliteTracker::in_memory().unwrap(); + let pr_url = "https://github.com/owner/repo/pull/1"; + + let base = |id: i64, path: &str| claudear_core::types::ReviewComment { + id, + path: path.to_string(), + position: None, + original_position: None, + body: format!("body for {}", path), + user: claudear_core::types::ReviewUser { + id: 1, + login: "reviewer".to_string(), + user_type: None, + }, + created_at: "2024-01-15T10:00:00Z".to_string(), + updated_at: "2024-01-15T10:00:00Z".to_string(), + html_url: "h".to_string(), + pull_request_review_id: None, + start_line: None, + line: None, + side: None, + }; + // Inline (non-empty path) and conversation (empty path) share id 42. + tracker + .record_pr_review_comment(pr_url, &base(42, "src/main.rs")) + .unwrap(); + tracker + .record_pr_review_comment(pr_url, &base(42, "")) + .unwrap(); + + let comments = tracker.get_comments_for_pr(pr_url).unwrap(); + assert_eq!(comments.len(), 2, "colliding id overwrote a row"); + assert_eq!( + tracker.get_unhandled_pr_review_comments(pr_url).unwrap().len(), + 2 + ); + } + + #[test] + fn test_note_failure_retires_one_worst_offender() { + // A batch failure can't be blamed on a specific comment, so only the single + // worst offender (most attempts, at cap) is retired per cycle — a valid + // comment that merely co-occurs with a poison one isn't dropped alongside it. + let tracker = SqliteTracker::in_memory().unwrap(); + let pr_url = "https://github.com/owner/repo/pull/3"; + let mk = |id: i64| claudear_core::types::ReviewComment { + id, + path: String::new(), + position: None, + original_position: None, + body: "@claudear fix".to_string(), + user: claudear_core::types::ReviewUser { + id: 1, + login: "reviewer".to_string(), + user_type: None, + }, + created_at: "2024-01-15T10:00:00Z".to_string(), + updated_at: "2024-01-15T10:00:00Z".to_string(), + html_url: format!("h{}", id), + pull_request_review_id: None, + start_line: None, + line: None, + side: None, + }; + tracker.record_pr_review_comment(pr_url, &mk(1)).unwrap(); // poison + tracker.record_pr_review_comment(pr_url, &mk(2)).unwrap(); // valid, arrives later + + // Comment 1 fails once alone (attempts=1), then both fail together with + // cap=2: comment 1 reaches the cap, comment 2 is only at 1. + tracker + .note_pr_review_comment_failure_by_ids(pr_url, &[1], 2) + .unwrap(); + tracker + .note_pr_review_comment_failure_by_ids(pr_url, &[1, 2], 2) + .unwrap(); + + // Only the worst offender (id 1) is retired; the valid comment survives. + let unhandled = tracker.get_unhandled_pr_review_comments(pr_url).unwrap(); + assert_eq!(unhandled.len(), 1); + assert_eq!(unhandled[0].id, 2); + } + #[test] fn test_get_comments_for_pr() { let tracker = SqliteTracker::in_memory().unwrap(); diff --git a/migrations/V9__pr_review_states_issue_comments.sql b/migrations/V9__pr_review_states_issue_comments.sql index d008f84..59d974d 100644 --- a/migrations/V9__pr_review_states_issue_comments.sql +++ b/migrations/V9__pr_review_states_issue_comments.sql @@ -5,13 +5,56 @@ ALTER TABLE pr_review_states ADD COLUMN last_issue_comment_id INTEGER; ALTER TABLE pr_review_states ADD COLUMN last_issue_comment_time TEXT; --- Make PR review-comment processing at-least-once. --- The pr_review_comments ledger (keyed by scm_comment_id UNIQUE) becomes the --- authority for "what still needs action", so a crash or downstream failure --- between detecting a comment and acting on it no longer drops it: unhandled --- rows are re-surfaced on the next poll regardless of the polling cursor. --- handled_at: set once the comment's feedback has been durably acted upon. --- attempts: consecutive processing failures; used to give up on a poison --- comment instead of retrying the fix agent forever. -ALTER TABLE pr_review_comments ADD COLUMN handled_at TEXT; -ALTER TABLE pr_review_comments ADD COLUMN attempts INTEGER NOT NULL DEFAULT 0; +-- Make PR review-comment processing at-least-once, and namespace the comment id. +-- +-- The pr_review_comments ledger becomes the authority for "what still needs +-- action", so a crash or downstream failure between detecting a comment and +-- acting on it no longer drops it: unhandled rows are re-surfaced on the next +-- poll regardless of the polling cursor. +-- handled_at: set once the comment's feedback has been durably acted upon. +-- attempts: consecutive processing failures; used to give up on a poison +-- comment instead of retrying the fix agent forever. +-- comment_kind: 'inline' (/pulls/{n}/comments) vs 'conversation' +-- (/issues/{n}/comments). These are distinct GitHub id sequences, +-- so a global UNIQUE(scm_comment_id) could let one silently +-- overwrite the other. Uniqueness is keyed on +-- (comment_kind, scm_comment_id) instead. +-- +-- A column-level UNIQUE can't be dropped in place (SQLite forbids DROP INDEX on a +-- constraint's auto-index), so pr_review_comments is rebuilt: stash the rows in a +-- transient shadow table, recreate the real table with the namespaced schema, +-- copy the rows back, drop the shadow. Nothing references pr_review_comments, so +-- this is FK-safe; every pre-existing row is an inline review comment. +CREATE TABLE pr_review_comments_shadow AS SELECT * FROM pr_review_comments; + +DROP TABLE pr_review_comments; + +CREATE TABLE pr_review_comments ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + scm_comment_id INTEGER NOT NULL, + comment_kind TEXT NOT NULL DEFAULT 'inline', + pr_url TEXT NOT NULL, + review_id INTEGER REFERENCES pr_reviews(id), + path TEXT NOT NULL, + position INTEGER, + line INTEGER, + body TEXT NOT NULL, + author TEXT NOT NULL, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + html_url TEXT, + handled_at TEXT, + attempts INTEGER NOT NULL DEFAULT 0, + UNIQUE(comment_kind, scm_comment_id) +); + +INSERT INTO pr_review_comments + (id, scm_comment_id, comment_kind, pr_url, review_id, path, position, line, + body, author, created_at, updated_at, html_url, handled_at, attempts) +SELECT id, scm_comment_id, 'inline', pr_url, review_id, path, position, line, + body, author, created_at, updated_at, html_url, NULL, 0 +FROM pr_review_comments_shadow; + +DROP TABLE pr_review_comments_shadow; + +CREATE INDEX IF NOT EXISTS idx_pr_review_comments_pr ON pr_review_comments(pr_url); From c269165f20ed091aaa79f8809893cd7f6d9b433a Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 11 Aug 2026 17:58:10 +0530 Subject: [PATCH 13/16] linting --- crates/claudear-engine/src/watcher.rs | 5 ++++- crates/claudear-integrations/src/scm.rs | 10 +++++----- crates/claudear-storage/src/sqlite.rs | 18 +++++++++++++++--- 3 files changed, 24 insertions(+), 9 deletions(-) diff --git a/crates/claudear-engine/src/watcher.rs b/crates/claudear-engine/src/watcher.rs index d73b2c6..93b2b0e 100644 --- a/crates/claudear-engine/src/watcher.rs +++ b/crates/claudear-engine/src/watcher.rs @@ -1265,7 +1265,10 @@ impl Watcher { review_watcher.unwatch_pr(&pr_url); continue; } - match self.process_review_action(&attempt, &feedback_summary).await { + match self + .process_review_action(&attempt, &feedback_summary) + .await + { Ok(()) => { // Durably handled: acknowledge exactly the comments in this // batch so they aren't re-surfaced, without touching any diff --git a/crates/claudear-integrations/src/scm.rs b/crates/claudear-integrations/src/scm.rs index 0bf03d5..351b6e4 100644 --- a/crates/claudear-integrations/src/scm.rs +++ b/crates/claudear-integrations/src/scm.rs @@ -403,9 +403,7 @@ impl ReviewEvent { ReviewEvent::ReviewSubmitted { inline_comments, .. } => inline_comments.iter().map(|c| c.id).collect(), - ReviewEvent::CommentsAdded { comments, .. } => { - comments.iter().map(|c| c.id).collect() - } + ReviewEvent::CommentsAdded { comments, .. } => comments.iter().map(|c| c.id).collect(), } } @@ -1338,7 +1336,9 @@ impl ReviewWatcher { // Only trigger-matched conversation comments are actionable feedback. let conversation_feedback: Vec = new_conversation_comments .iter() - .filter(|c| trigger.is_empty() || c.body.to_lowercase().contains(&trigger.to_lowercase())) + .filter(|c| { + trigger.is_empty() || c.body.to_lowercase().contains(&trigger.to_lowercase()) + }) .cloned() .collect(); @@ -5699,6 +5699,7 @@ mod tests { #[cfg(feature = "sqlite")] mod sqlite_persistence_tests { + use crate::scm::ReviewEvent; use crate::scm::{ CodeReview, PrInfo, PrReviewState, PrStatus, RemoteRepo, ReviewComment, ReviewUser, ReviewWatcher, ScmProvider, @@ -5708,7 +5709,6 @@ mod tests { use claudear_core::types::{ ActivityLogEntry, FixAttempt, FixAttemptStats, FixAttemptStatus, PrReviewRecord, }; - use crate::scm::ReviewEvent; use claudear_storage::{ ActivityStore, AttemptTracker, ChatStore, DiscordStore, EmbeddingStore, EvaluationStore, ExperimentStore, FixAttemptTracker, KnowledgeStore, RegressionStore, diff --git a/crates/claudear-storage/src/sqlite.rs b/crates/claudear-storage/src/sqlite.rs index 7ca1cd1..df9b902 100644 --- a/crates/claudear-storage/src/sqlite.rs +++ b/crates/claudear-storage/src/sqlite.rs @@ -11413,7 +11413,13 @@ mod tests { // A new comment fails repeatedly and is given up on at the cap. tracker.record_pr_review_comment(pr_url, &mk(3)).unwrap(); - assert_eq!(tracker.get_unhandled_pr_review_comments(pr_url).unwrap().len(), 1); + assert_eq!( + tracker + .get_unhandled_pr_review_comments(pr_url) + .unwrap() + .len(), + 1 + ); tracker .note_pr_review_comment_failure_by_ids(pr_url, &[3], 3) .unwrap(); // attempts=1 @@ -11421,7 +11427,10 @@ mod tests { .note_pr_review_comment_failure_by_ids(pr_url, &[3], 3) .unwrap(); // attempts=2 assert_eq!( - tracker.get_unhandled_pr_review_comments(pr_url).unwrap().len(), + tracker + .get_unhandled_pr_review_comments(pr_url) + .unwrap() + .len(), 1, "should still retry below the cap" ); @@ -11512,7 +11521,10 @@ mod tests { let comments = tracker.get_comments_for_pr(pr_url).unwrap(); assert_eq!(comments.len(), 2, "colliding id overwrote a row"); assert_eq!( - tracker.get_unhandled_pr_review_comments(pr_url).unwrap().len(), + tracker + .get_unhandled_pr_review_comments(pr_url) + .unwrap() + .len(), 2 ); } From e1d24e7e2c6d6e250d92fc4258dd8236234dd972 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 11 Aug 2026 18:10:18 +0530 Subject: [PATCH 14/16] fix(review): make comment acknowledgement namespace-aware After namespacing the ledger by (comment_kind, scm_comment_id), the by-ids acknowledge/failure updates still matched on id alone, so on a collision they hit both the inline and conversation rows: processing one marked the other handled, and a failure charged both. Carry the kind alongside each id (ReviewEvent::comment_refs) and match (comment_kind, scm_comment_id) pairs in mark/note-failure so each op touches only its own namespace row. --- crates/claudear-engine/src/watcher.rs | 42 ++++++++----- crates/claudear-integrations/src/scm.rs | 26 +++++++-- crates/claudear-storage/src/lib.rs | 16 +++-- crates/claudear-storage/src/sqlite.rs | 78 ++++++++++++++++--------- 4 files changed, 105 insertions(+), 57 deletions(-) diff --git a/crates/claudear-engine/src/watcher.rs b/crates/claudear-engine/src/watcher.rs index 93b2b0e..cd5faff 100644 --- a/crates/claudear-engine/src/watcher.rs +++ b/crates/claudear-engine/src/watcher.rs @@ -1238,7 +1238,7 @@ impl Watcher { // Check for new reviews let events = review_watcher.check_for_reviews().await?; - for (pr_url, feedback_summary, feedback_count, comment_ids) in + for (pr_url, feedback_summary, feedback_count, comment_refs) in Self::group_review_feedback_by_pr(events) { tracing::info!( @@ -1275,7 +1275,7 @@ impl Watcher { // comment recorded concurrently while we were processing. if let Err(e) = self .tracker - .mark_pr_review_comments_handled_by_ids(&pr_url, &comment_ids) + .mark_pr_review_comments_handled_by_ids(&pr_url, &comment_refs) { tracing::warn!(pr_url = %pr_url, error = %e, "Failed to mark review comments handled"); } @@ -1290,7 +1290,7 @@ impl Watcher { // count the failure so a poison comment eventually gives up. if let Err(e) = self.tracker.note_pr_review_comment_failure_by_ids( &pr_url, - &comment_ids, + &comment_refs, MAX_REVIEW_COMMENT_ATTEMPTS, ) { tracing::warn!(pr_url = %pr_url, error = %e, "Failed to record review-comment failure"); @@ -1306,7 +1306,7 @@ impl Watcher { // uncorrelated comments don't re-surface forever. if let Err(e) = self.tracker.note_pr_review_comment_failure_by_ids( &pr_url, - &comment_ids, + &comment_refs, MAX_REVIEW_COMMENT_ATTEMPTS, ) { tracing::warn!(pr_url = %pr_url, error = %e, "Failed to record review-comment failure"); @@ -1325,15 +1325,17 @@ impl Watcher { } /// Group actionable review events per PR into (pr_url, feedback_summary, - /// feedback_count, comment_ids). `comment_ids` are exactly the ledger comments - /// carried by the batch, so acknowledgement can target them rather than the - /// whole PR (which would wrongly mark a concurrently-recorded comment handled). + /// feedback_count, comment_refs). `comment_refs` are exactly the ledger comments + /// carried by the batch as `(scm_comment_id, comment_kind)`, so acknowledgement + /// targets those specific rows rather than the whole PR (which would wrongly + /// mark a concurrently-recorded comment handled) or a colliding id in the other + /// namespace. fn group_review_feedback_by_pr( events: Vec, - ) -> Vec<(String, String, usize, Vec)> { + ) -> Vec<(String, String, usize, Vec<(i64, &'static str)>)> { let mut feedback_by_pr: std::collections::HashMap> = std::collections::HashMap::new(); - let mut ids_by_pr: std::collections::HashMap> = + let mut refs_by_pr: std::collections::HashMap> = std::collections::HashMap::new(); let mut pr_order: Vec = Vec::new(); @@ -1346,10 +1348,10 @@ impl Watcher { if !feedback_by_pr.contains_key(&pr_url) { pr_order.push(pr_url.clone()); } - ids_by_pr + refs_by_pr .entry(pr_url.clone()) .or_default() - .extend(event.comment_ids()); + .extend(event.comment_refs()); feedback_by_pr .entry(pr_url) .or_default() @@ -1359,10 +1361,10 @@ impl Watcher { pr_order .into_iter() .filter_map(|pr_url| { - let ids = ids_by_pr.remove(&pr_url).unwrap_or_default(); + let refs = refs_by_pr.remove(&pr_url).unwrap_or_default(); feedback_by_pr.remove(&pr_url).map(|feedbacks| { let count = feedbacks.len(); - (pr_url, feedbacks.join("\n\n---\n\n"), count, ids) + (pr_url, feedbacks.join("\n\n---\n\n"), count, refs) }) }) .collect() @@ -6389,9 +6391,17 @@ mod tests { let grouped = Watcher::group_review_feedback_by_pr(events); assert_eq!(grouped.len(), 1); - let mut ids = grouped[0].3.clone(); - ids.sort(); - assert_eq!(ids, vec![101, 102, 103]); + let mut refs = grouped[0].3.clone(); + refs.sort(); + // Empty-path comments are conversation-kind; each ref namespaces its id. + assert_eq!( + refs, + vec![ + (101, "conversation"), + (102, "conversation"), + (103, "conversation") + ] + ); } #[tokio::test] diff --git a/crates/claudear-integrations/src/scm.rs b/crates/claudear-integrations/src/scm.rs index 351b6e4..ee8f492 100644 --- a/crates/claudear-integrations/src/scm.rs +++ b/crates/claudear-integrations/src/scm.rs @@ -393,17 +393,31 @@ impl ReviewEvent { } } - /// The ledger comment ids carried by this event: standalone/conversation - /// comments for `CommentsAdded`, inline comments for `ReviewSubmitted`. Used to - /// acknowledge exactly the comments in a processed batch, so a comment recorded + /// The ledger comment references carried by this event as + /// `(scm_comment_id, comment_kind)`: standalone/conversation comments for + /// `CommentsAdded`, inline comments for `ReviewSubmitted`. The kind namespaces + /// the id so acknowledgement targets exactly the right ledger row even when an + /// inline and a conversation comment share a numeric id. Used to acknowledge + /// exactly the comments in a processed batch, so a comment recorded /// concurrently (e.g. a webhook `check_for_pr`) isn't marked handled without /// being processed. A formal review has no ledger row and contributes none. - pub fn comment_ids(&self) -> Vec { + pub fn comment_refs(&self) -> Vec<(i64, &'static str)> { + // Conversation comments carry no file path; inline comments always do — + // the same discriminator the storage layer uses when recording. + let kind = |c: &ReviewComment| { + if c.path.is_empty() { + "conversation" + } else { + "inline" + } + }; match self { ReviewEvent::ReviewSubmitted { inline_comments, .. - } => inline_comments.iter().map(|c| c.id).collect(), - ReviewEvent::CommentsAdded { comments, .. } => comments.iter().map(|c| c.id).collect(), + } => inline_comments.iter().map(|c| (c.id, kind(c))).collect(), + ReviewEvent::CommentsAdded { comments, .. } => { + comments.iter().map(|c| (c.id, kind(c))).collect() + } } } diff --git a/crates/claudear-storage/src/lib.rs b/crates/claudear-storage/src/lib.rs index 0e6df2c..2337980 100644 --- a/crates/claudear-storage/src/lib.rs +++ b/crates/claudear-storage/src/lib.rs @@ -672,23 +672,27 @@ pub trait ActivityStore: Send + Sync { Ok(()) } - /// Mark only the given comment ids on a PR as durably handled. Called once the + /// Mark only the given comments on a PR as durably handled. Called once the /// batch that carried exactly these comments has been successfully acted upon. + /// Each comment is `(scm_comment_id, comment_kind)` so a colliding id in the + /// other namespace is not acknowledged by mistake. fn mark_pr_review_comments_handled_by_ids( &self, _pr_url: &str, - _comment_ids: &[i64], + _comments: &[(i64, &str)], ) -> Result<()> { Ok(()) } - /// Record a failed attempt to act on the given comment ids: bumps each one's - /// attempt count and gives up (marks it handled) once the count reaches - /// `max_attempts`, so a poison comment does not re-run the fix agent forever. + /// Record a failed attempt to act on the given comments: bumps each one's + /// attempt count and gives up (marks the single worst offender handled) once + /// it reaches `max_attempts`, so a poison comment does not re-run the fix agent + /// forever. Each comment is `(scm_comment_id, comment_kind)` so a colliding id + /// in the other namespace is not charged by mistake. fn note_pr_review_comment_failure_by_ids( &self, _pr_url: &str, - _comment_ids: &[i64], + _comments: &[(i64, &str)], _max_attempts: i64, ) -> Result<()> { Ok(()) diff --git a/crates/claudear-storage/src/sqlite.rs b/crates/claudear-storage/src/sqlite.rs index df9b902..0c48e2b 100644 --- a/crates/claudear-storage/src/sqlite.rs +++ b/crates/claudear-storage/src/sqlite.rs @@ -2028,21 +2028,25 @@ impl ActivityStore for SqliteTracker { fn mark_pr_review_comments_handled_by_ids( &self, pr_url: &str, - comment_ids: &[i64], + comments: &[(i64, &str)], ) -> Result<()> { - if comment_ids.is_empty() { + if comments.is_empty() { return Ok(()); } let conn = self.acquire_lock()?; - let placeholders = vec!["?"; comment_ids.len()].join(", "); + // Match (comment_kind, scm_comment_id) pairs so a colliding id in the other + // namespace is not acknowledged by mistake. + let pair_clause = + vec!["(comment_kind = ? AND scm_comment_id = ?)"; comments.len()].join(" OR "); let sql = format!( "UPDATE pr_review_comments SET handled_at = datetime('now') \ - WHERE pr_url = ?1 AND handled_at IS NULL AND scm_comment_id IN ({})", - placeholders + WHERE pr_url = ? AND handled_at IS NULL AND ({})", + pair_clause ); - let mut binds: Vec> = Vec::with_capacity(comment_ids.len() + 1); + let mut binds: Vec> = Vec::with_capacity(comments.len() * 2 + 1); binds.push(Box::new(pr_url.to_string())); - for id in comment_ids { + for (id, kind) in comments { + binds.push(Box::new(kind.to_string())); binds.push(Box::new(*id)); } conn.execute(&sql, rusqlite::params_from_iter(binds.iter().map(|b| &**b)))?; @@ -2052,20 +2056,23 @@ impl ActivityStore for SqliteTracker { fn note_pr_review_comment_failure_by_ids( &self, pr_url: &str, - comment_ids: &[i64], + comments: &[(i64, &str)], max_attempts: i64, ) -> Result<()> { - if comment_ids.is_empty() { + if comments.is_empty() { return Ok(()); } let conn = self.acquire_lock()?; - let placeholders = vec!["?"; comment_ids.len()].join(", "); + // Match (comment_kind, scm_comment_id) pairs so a colliding id in the other + // namespace is not charged by mistake. + let pair_clause = + vec!["(comment_kind = ? AND scm_comment_id = ?)"; comments.len()].join(" OR "); // Count this failed cycle against exactly the comments we tried. let bump_sql = format!( "UPDATE pr_review_comments SET attempts = attempts + 1 \ - WHERE pr_url = ?1 AND handled_at IS NULL AND scm_comment_id IN ({})", - placeholders + WHERE pr_url = ? AND handled_at IS NULL AND ({})", + pair_clause ); // Give up on at most ONE comment per failed cycle: the single worst // offender (most attempts) that has hit the cap. A batch failure can't be @@ -2077,22 +2084,18 @@ impl ActivityStore for SqliteTracker { "UPDATE pr_review_comments SET handled_at = datetime('now') \ WHERE id = ( \ SELECT id FROM pr_review_comments \ - WHERE pr_url = ?1 AND handled_at IS NULL AND attempts >= ?2 \ - AND scm_comment_id IN ({}) \ + WHERE pr_url = ? AND handled_at IS NULL AND attempts >= ? AND ({}) \ ORDER BY attempts DESC, scm_comment_id ASC \ LIMIT 1 \ )", - // ids start at bind position 3 for the give-up query - (0..comment_ids.len()) - .map(|i| format!("?{}", i + 3)) - .collect::>() - .join(", ") + pair_clause ); let mut bump_binds: Vec> = - Vec::with_capacity(comment_ids.len() + 1); + Vec::with_capacity(comments.len() * 2 + 1); bump_binds.push(Box::new(pr_url.to_string())); - for id in comment_ids { + for (id, kind) in comments { + bump_binds.push(Box::new(kind.to_string())); bump_binds.push(Box::new(*id)); } conn.execute( @@ -2101,10 +2104,11 @@ impl ActivityStore for SqliteTracker { )?; let mut giveup_binds: Vec> = - Vec::with_capacity(comment_ids.len() + 2); + Vec::with_capacity(comments.len() * 2 + 2); giveup_binds.push(Box::new(pr_url.to_string())); giveup_binds.push(Box::new(max_attempts)); - for id in comment_ids { + for (id, kind) in comments { + giveup_binds.push(Box::new(kind.to_string())); giveup_binds.push(Box::new(*id)); } conn.execute( @@ -11421,10 +11425,10 @@ mod tests { 1 ); tracker - .note_pr_review_comment_failure_by_ids(pr_url, &[3], 3) + .note_pr_review_comment_failure_by_ids(pr_url, &[(3, "conversation")], 3) .unwrap(); // attempts=1 tracker - .note_pr_review_comment_failure_by_ids(pr_url, &[3], 3) + .note_pr_review_comment_failure_by_ids(pr_url, &[(3, "conversation")], 3) .unwrap(); // attempts=2 assert_eq!( tracker @@ -11435,7 +11439,7 @@ mod tests { "should still retry below the cap" ); tracker - .note_pr_review_comment_failure_by_ids(pr_url, &[3], 3) + .note_pr_review_comment_failure_by_ids(pr_url, &[(3, "conversation")], 3) .unwrap(); // attempts=3 -> give up assert!( tracker @@ -11475,7 +11479,7 @@ mod tests { // Acknowledge only comment 1; comment 2 (recorded concurrently) survives. tracker - .mark_pr_review_comments_handled_by_ids(pr_url, &[1]) + .mark_pr_review_comments_handled_by_ids(pr_url, &[(1, "conversation")]) .unwrap(); let unhandled = tracker.get_unhandled_pr_review_comments(pr_url).unwrap(); assert_eq!(unhandled.len(), 1); @@ -11527,6 +11531,18 @@ mod tests { .len(), 2 ); + + // Acknowledging only the inline row must not touch the conversation row + // that shares the id. + tracker + .mark_pr_review_comments_handled_by_ids(pr_url, &[(42, "inline")]) + .unwrap(); + let unhandled = tracker.get_unhandled_pr_review_comments(pr_url).unwrap(); + assert_eq!(unhandled.len(), 1, "colliding id acknowledged wrong row"); + assert!( + unhandled[0].path.is_empty(), + "the surviving row should be the conversation comment" + ); } #[test] @@ -11561,10 +11577,14 @@ mod tests { // Comment 1 fails once alone (attempts=1), then both fail together with // cap=2: comment 1 reaches the cap, comment 2 is only at 1. tracker - .note_pr_review_comment_failure_by_ids(pr_url, &[1], 2) + .note_pr_review_comment_failure_by_ids(pr_url, &[(1, "conversation")], 2) .unwrap(); tracker - .note_pr_review_comment_failure_by_ids(pr_url, &[1, 2], 2) + .note_pr_review_comment_failure_by_ids( + pr_url, + &[(1, "conversation"), (2, "conversation")], + 2, + ) .unwrap(); // Only the worst offender (id 1) is retired; the valid comment survives. From 9f8ccf4f602cc80ac9b3a29ca0fac590d346f6b1 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 11 Aug 2026 18:17:20 +0530 Subject: [PATCH 15/16] fix(review): key re-surface dedup on (id, kind) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The re-surface dedup used an id-only seen set, so an inline and a conversation comment sharing a numeric id collapsed — whichever ledger row was seen second was never re-emitted, leaving it permanently unhandled. Key the dedup on (id, comment_kind), matching the namespaced ledger. --- crates/claudear-integrations/src/scm.rs | 65 +++++++++++++++++++++++-- 1 file changed, 61 insertions(+), 4 deletions(-) diff --git a/crates/claudear-integrations/src/scm.rs b/crates/claudear-integrations/src/scm.rs index ee8f492..536123e 100644 --- a/crates/claudear-integrations/src/scm.rs +++ b/crates/claudear-integrations/src/scm.rs @@ -1438,20 +1438,31 @@ impl ReviewWatcher { // they come back through this query too — dedup against what we're already // emitting this cycle (the batch plus inline comments on review events). if let Some(ref tracker) = self.tracker { - let mut seen: std::collections::HashSet = - added_comments.iter().map(|c| c.id).collect(); + // Dedup keyed on (id, kind): an inline and a conversation comment can + // share a numeric id, so an id-only key would collapse them and drop + // whichever is seen second. Conversation comments carry no path. + let key = |c: &ReviewComment| { + let kind = if c.path.is_empty() { + "conversation" + } else { + "inline" + }; + (c.id, kind) + }; + let mut seen: std::collections::HashSet<(i64, &'static str)> = + added_comments.iter().map(&key).collect(); for event in &events { if let ReviewEvent::ReviewSubmitted { inline_comments, .. } = event { - seen.extend(inline_comments.iter().map(|c| c.id)); + seen.extend(inline_comments.iter().map(&key)); } } match tracker.get_unhandled_pr_review_comments(&state.pr_url) { Ok(pending) => { for comment in pending { - if seen.insert(comment.id) { + if seen.insert(key(&comment)) { added_comments.push(comment); } } @@ -6122,6 +6133,52 @@ mod tests { assert_eq!(comment_events[0][0].id, 4242); } + #[tokio::test] + async fn test_resurface_keeps_colliding_ids_distinct() { + // An inline (non-empty path) and a conversation (empty path) comment + // share id 7. The re-surface dedup keys on (id, kind), so both must be + // emitted rather than collapsed to one. + let base = |path: &str| ReviewComment { + id: 7, + path: path.to_string(), + position: None, + original_position: None, + body: "@claudear fix".to_string(), + user: ReviewUser { + id: 0, + login: "reviewer".to_string(), + user_type: None, + }, + created_at: "2025-01-01T00:00:00Z".to_string(), + updated_at: "2025-01-01T00:00:00Z".to_string(), + html_url: "h".to_string(), + pull_request_review_id: None, + start_line: None, + line: None, + side: None, + }; + let provider: Arc = + Arc::new(MockScmProvider::new("github", true, "@claudear")); + let tracker = Arc::new(MockTrackerWithRecording::new()); + tracker.set_unhandled(vec![base("src/main.rs"), base("")]); + let watcher = ReviewWatcher::with_tracker(provider, tracker.clone()); + watcher.watch_pr(make_state( + "https://github.com/org/repo/pull/1", + "org/repo", + 1, + )); + + let events = watcher.check_for_reviews().await.unwrap(); + let emitted: usize = events + .iter() + .filter_map(|e| match e { + ReviewEvent::CommentsAdded { comments, .. } => Some(comments.len()), + _ => None, + }) + .sum(); + assert_eq!(emitted, 2, "colliding-id comments collapsed on re-surface"); + } + #[tokio::test] async fn test_record_review_to_db_no_submitted_at() { let mock = MockScmProvider::new("github", true, "@claudear"); From 007a85b25badde65cdf5bd2a212432c3f1c387d0 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Sun, 16 Aug 2026 15:28:28 +0530 Subject: [PATCH 16/16] linting --- crates/claudear-engine/src/watcher.rs | 13 +++++++++---- repos/database | 1 + repos/sdk-for-node | 1 + 3 files changed, 11 insertions(+), 4 deletions(-) create mode 160000 repos/database create mode 160000 repos/sdk-for-node diff --git a/crates/claudear-engine/src/watcher.rs b/crates/claudear-engine/src/watcher.rs index cd5faff..824333f 100644 --- a/crates/claudear-engine/src/watcher.rs +++ b/crates/claudear-engine/src/watcher.rs @@ -255,6 +255,13 @@ pub struct Watcher { spawn_handles: tokio::sync::Mutex>>, } +/// A ledger comment carried by a review batch: `(scm_comment_id, comment_kind)`. +type CommentRef = (i64, &'static str); + +/// Actionable review feedback grouped for one PR: +/// `(pr_url, feedback_summary, feedback_count, comment_refs)`. +type PrReviewFeedback = (String, String, usize, Vec); + impl Watcher { /// Create a new watcher. pub fn new(options: WatcherOptions) -> Self { @@ -1330,12 +1337,10 @@ impl Watcher { /// targets those specific rows rather than the whole PR (which would wrongly /// mark a concurrently-recorded comment handled) or a colliding id in the other /// namespace. - fn group_review_feedback_by_pr( - events: Vec, - ) -> Vec<(String, String, usize, Vec<(i64, &'static str)>)> { + fn group_review_feedback_by_pr(events: Vec) -> Vec { let mut feedback_by_pr: std::collections::HashMap> = std::collections::HashMap::new(); - let mut refs_by_pr: std::collections::HashMap> = + let mut refs_by_pr: std::collections::HashMap> = std::collections::HashMap::new(); let mut pr_order: Vec = Vec::new(); diff --git a/repos/database b/repos/database new file mode 160000 index 0000000..5679019 --- /dev/null +++ b/repos/database @@ -0,0 +1 @@ +Subproject commit 567901977c0128a3bd1cb9917553ebb4c1055582 diff --git a/repos/sdk-for-node b/repos/sdk-for-node new file mode 160000 index 0000000..56349bf --- /dev/null +++ b/repos/sdk-for-node @@ -0,0 +1 @@ +Subproject commit 56349bfe41d4c92ce78be1613708192ed4e98dc2