Feat/pr conversation comments

crates/claudear-core/src/types.rs

@@ -2971,6 +2971,11 @@ pub struct PrReviewState {
     pub last_review_time: Option<String>,
     pub last_comment_id: Option<i64>,
     pub last_comment_time: Option<String>,
+    /// 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<i64>,
+    pub last_issue_comment_time: Option<String>,
     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,
         }
     }

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 });
                     }
                 }
             }

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<Intent>);
 
+/// 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)
@@ -251,6 +255,13 @@ pub struct Watcher {
     spawn_handles: tokio::sync::Mutex<Vec<tokio::task::JoinHandle<()>>>,
 }
 
+/// 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<CommentRef>);
+
 impl Watcher {
     /// Create a new watcher.
     pub fn new(options: WatcherOptions) -> Self {
@@ -1234,7 +1245,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_refs) in
+            Self::group_review_feedback_by_pr(events)
         {
             tracing::info!(
                 pr_url = %pr_url,
@@ -1252,24 +1264,60 @@ 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
+                match self
                     .process_review_action(&attempt, &feedback_summary)
                     .await
                 {
-                    tracing::error!(
-                        pr_url = %pr_url,
-                        error = %e,
-                        "Failed to process review feedback"
-                    );
+                    Ok(()) => {
+                        // 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_refs)
+                        {
+                            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 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_refs,
+                            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 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_refs,
+                    MAX_REVIEW_COMMENT_ATTEMPTS,
+                ) {
+                    tracing::warn!(pr_url = %pr_url, error = %e, "Failed to record review-comment failure");
+                }
             }
         }
 
@@ -1283,9 +1331,17 @@ impl Watcher {
         )
     }
 
-    fn group_review_feedback_by_pr(events: Vec<ReviewEvent>) -> Vec<(String, String, usize)> {
+    /// Group actionable review events per PR into (pr_url, feedback_summary,
+    /// 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<ReviewEvent>) -> Vec<PrReviewFeedback> {
         let mut feedback_by_pr: std::collections::HashMap<String, Vec<String>> =
             std::collections::HashMap::new();
+        let mut refs_by_pr: std::collections::HashMap<String, Vec<CommentRef>> =
+            std::collections::HashMap::new();
         let mut pr_order: Vec<String> = Vec::new();
 
         for event in events {
@@ -1297,6 +1353,10 @@ impl Watcher {
             if !feedback_by_pr.contains_key(&pr_url) {
                 pr_order.push(pr_url.clone());
             }
+            refs_by_pr
+                .entry(pr_url.clone())
+                .or_default()
+                .extend(event.comment_refs());
             feedback_by_pr
                 .entry(pr_url)
                 .or_default()
@@ -1306,9 +1366,10 @@ impl Watcher {
         pr_order
             .into_iter()
             .filter_map(|pr_url| {
+                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)
+                    (pr_url, feedbacks.join("\n\n---\n\n"), count, refs)
                 })
             })
             .collect()
@@ -6291,6 +6352,61 @@ 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 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]