diff --git a/claudear.example.toml b/claudear.example.toml index 392272ec..e8e1b001 100644 --- a/claudear.example.toml +++ b/claudear.example.toml @@ -500,6 +500,16 @@ bot_role_id = "" # Polling interval in milliseconds (overrides global) poll_interval_ms = 300000 +# Reply-chain conversation context. When a user replies to a message, resolve +# the reply thread and feed it to the agent as grounding, so follow-ups continue +# the conversation without re-quoting. Claudear's own prior answers are read from +# the DB; other users' messages are fetched from Discord. Defaults to true. +reply_chain_enabled = true + +# Maximum number of ancestor messages to walk when building reply-chain context. +# Bounds Discord fetches and guards against very long threads. Defaults to 15. +reply_chain_max_depth = 15 + # Slack as an issue source (messages become issues) # Shared credentials (bot_token, channel_id) are inherited from notifiers.slack if not set here. diff --git a/crates/claudear-config/src/config.rs b/crates/claudear-config/src/config.rs index da20c450..4cd7869f 100644 --- a/crates/claudear-config/src/config.rs +++ b/crates/claudear-config/src/config.rs @@ -280,6 +280,15 @@ pub struct DiscordSourceConfig { /// alongside `bot_id` so the source ingests the message regardless of which /// form the sender picked from autocomplete. pub bot_role_id: Option, + /// When a message is a reply, resolve its reply thread into conversation + /// context and feed it to the agent (Claudear's own answers come from the + /// DB; other users' messages are fetched from Discord). `None` (omitted) + /// uses the default: enabled. + pub reply_chain_enabled: Option, + /// Maximum number of ancestor messages to walk when building reply-chain + /// context. Bounds fetches and guards against long threads. `None` (omitted) + /// uses the default: 15. + pub reply_chain_max_depth: Option, } /// Discord notifier-only configuration (for outbound notifications). @@ -1269,6 +1278,11 @@ pub struct DiscordConfig { pub bot_role_id: Option, /// Verbose per-message polling diagnostics for the Discord source. pub debug_logging: bool, + /// Resolve reply threads into conversation context (see + /// `DiscordSourceConfig::reply_chain_enabled`). + pub reply_chain_enabled: bool, + /// Maximum ancestor messages to walk when building reply-chain context. + pub reply_chain_max_depth: usize, } /// Email (SMTP) notification configuration. @@ -3266,6 +3280,8 @@ impl Config { bot_role_id: src.and_then(|s| s.bot_role_id.clone()), // Sourced from the global `debug_logging` flag, not per-source. debug_logging: self.debug_logging, + reply_chain_enabled: src.and_then(|s| s.reply_chain_enabled).unwrap_or(true), + reply_chain_max_depth: src.and_then(|s| s.reply_chain_max_depth).unwrap_or(15), } } diff --git a/crates/claudear-engine/src/processing.rs b/crates/claudear-engine/src/processing.rs index bba958ab..ead047cb 100644 --- a/crates/claudear-engine/src/processing.rs +++ b/crates/claudear-engine/src/processing.rs @@ -20,12 +20,14 @@ use claudear_core::types::{ ActionKind, ActivityLogEntry, AskRequest, ErrorPattern, Issue, ProcessingMetric, QaKnowledgeEntry, ReplyKind, RetrievalUsageRecord, TimelineEventStatus, VerifyResult, }; +use claudear_integrations::discord::DiscordClient; use claudear_integrations::github::GitHubClient; use claudear_integrations::notifier::{send_to_all_and_wait_first_reply, Notifier}; use claudear_integrations::runner::{self, AgentRunner}; use claudear_integrations::scm::{PrReviewState, ReviewWatcher}; use claudear_storage::{classify_error, compute_error_hash, FixAttemptTracker}; use serde_json::json; +use std::collections::HashSet; use std::sync::Arc; use crate::intent::{Intent, IntentClassifier}; @@ -34,6 +36,20 @@ use crate::intent::{Intent, IntentClassifier}; /// relevance judge — each call spawns an agent process, so keep the fan-out small. const RETRIEVAL_JUDGE_CONCURRENCY: usize = 4; +/// Strip Discord mention tokens (`<@ID>`, `<@!ID>`, `<@&ID>`) from text and trim. +/// Used when rendering reply-chain turns so mention noise doesn't reach the agent. +fn strip_discord_mentions(content: &str) -> String { + let mut result = content.to_string(); + while let Some(start) = result.find("<@") { + if let Some(end) = result[start..].find('>') { + result.replace_range(start..start + end + 1, ""); + } else { + break; + } + } + result.trim().to_string() +} + #[derive(Debug, Clone)] struct RetrievedItem { source_kind: String, @@ -1000,6 +1016,9 @@ impl IssueProcessor { json!({}), ); + // Ground the fix in the reply thread when this issue is a reply. + context = self.with_reply_chain(issue, context).await; + // Claude execution + ask loop let mut rounds: u8 = 0; let claude_result = loop { @@ -1659,6 +1678,121 @@ impl IssueProcessor { } } + /// Resolve a Discord reply thread into a labelled conversation transcript to + /// ground the agent, or `None` when the issue is not a reply, the feature is + /// disabled, or nothing could be resolved. + /// + /// Walks parents newest-first: when a parent is one of Claudear's own answers + /// (recognised via `lookup_answer_issue`), the original question and answer + /// are pulled from the DB and the walk stops (that answer already + /// encapsulates everything upstream). Other users' messages aren't in our DB, + /// so they're fetched from Discord and the walk continues to their parent. + async fn assemble_reply_chain(&self, issue: &Issue) -> Option { + let parent_id = issue.get_metadata::("reply_to_message_id")?; + let discord_cfg = self.config.discord_merged(); + if !discord_cfg.reply_chain_enabled { + return None; + } + let max_depth = discord_cfg.reply_chain_max_depth.max(1); + let mut parent_channel = issue + .get_metadata::("reply_to_channel_id") + .or_else(|| issue.get_metadata::("channel_id")) + .unwrap_or_default(); + + // Newest-first accumulation; reversed to chronological order at the end. + let mut lines_rev: Vec = Vec::new(); + let mut client: Option = None; + let mut seen: HashSet = HashSet::new(); + let mut current_id = Some(parent_id); + let mut depth = 0usize; + + while let Some(pid) = current_id.take() { + if depth >= max_depth || !seen.insert(pid.clone()) { + break; + } + depth += 1; + + // Claudear's own answer? Pull question + answer from the DB and stop. + if let Ok(Some((src, answered_issue_id))) = self.tracker.lookup_answer_issue(&pid) { + if let Ok(Some(att)) = self.tracker.get_attempt(&src, &answered_issue_id) { + if let Some(ans) = att.error_message.filter(|a| !a.trim().is_empty()) { + lines_rev.push(format!("[Claudear]: {}", ans.trim())); + } + } + if let Ok(Some(emb)) = self.tracker.get_embedding(&src, &answered_issue_id) { + let question = emb + .description + .filter(|d| !d.trim().is_empty()) + .or(emb.title) + .unwrap_or_default(); + let question = strip_discord_mentions(question.trim()); + if !question.is_empty() { + lines_rev.push(format!("[User]: {}", question)); + } + } + break; + } + + // Otherwise it's another user's message not in our DB: fetch it. + let fetch_client = match client.as_ref() { + Some(c) => c, + None => { + let token = discord_cfg.bot_token.as_ref().map(|s| s.expose())?; + match DiscordClient::new(token) { + Ok(c) => { + client = Some(c); + client.as_ref().unwrap() + } + Err(_) => break, + } + } + }; + + match fetch_client.get_message(&parent_channel, &pid).await { + Ok(msg) => { + let name = msg + .author + .as_ref() + .map(|a| a.username.clone()) + .unwrap_or_else(|| "user".to_string()); + let content = strip_discord_mentions(msg.content.trim()); + if !content.is_empty() { + lines_rev.push(format!("[User {}]: {}", name, content)); + } + // Continue to this message's own parent, if it is a reply. + match msg.message_reference { + Some(reference) => { + if let Some(ch) = reference.channel_id { + parent_channel = ch; + } + current_id = reference.message_id; + } + None => current_id = None, + } + } + // Deleted / no permission: stop gracefully with what we have. + Err(_) => break, + } + } + + if lines_rev.is_empty() { + return None; + } + lines_rev.reverse(); + let mut out = String::from("## Prior conversation (Discord reply thread)\n"); + out.push_str(&lines_rev.join("\n")); + Some(out) + } + + /// Prepend the resolved reply-chain transcript (if any) to `context`. + async fn with_reply_chain(&self, issue: &Issue, context: String) -> String { + match self.assemble_reply_chain(issue).await { + Some(chain) if context.is_empty() => chain, + Some(chain) => format!("{}\n\n{}", chain, context), + None => context, + } + } + /// Answer a pure question with RAG-grounded context, read-only (no PR). async fn answer_question_issue( &self, @@ -1749,6 +1883,9 @@ impl IssueProcessor { self.spawn_retrieval_judge(id, issue, &retrieved_items); } + // Ground the answer in the reply thread when this question is a reply. + let context = self.with_reply_chain(issue, context).await; + self.record_issue_decision( issue, "question_detected", @@ -1767,11 +1904,25 @@ impl IssueProcessor { match answer_result { Ok(Ok(answer)) => { - if let Err(e) = self.notifier.notify_answer(issue, &answer).await { - tracing::warn!(short_id = %issue.short_id, error = %e, "Failed to deliver answer"); + match self.notifier.notify_answer(issue, &answer).await { + Ok(sent_ids) => { + if !sent_ids.is_empty() { + if let Err(e) = self.tracker.record_answer_message_ids( + source_name, + &issue.id, + &sent_ids, + ) { + tracing::warn!(short_id = %issue.short_id, error = %e, "Failed to record answer message ids"); + } + } + } + Err(e) => { + tracing::warn!(short_id = %issue.short_id, error = %e, "Failed to deliver answer"); + } } - let summary: String = answer.chars().take(500).collect(); - if let Err(e) = self.tracker.mark_answered(source_name, &issue.id, &summary) { + // Store the FULL answer so it can ground a later reply-chain + // continuation; truncation is a display/send concern only. + if let Err(e) = self.tracker.mark_answered(source_name, &issue.id, &answer) { tracing::warn!(short_id = %issue.short_id, error = %e, "Failed to mark answered"); } self.record_issue_decision( @@ -1834,6 +1985,7 @@ impl IssueProcessor { &input.resolution, &input.source_name, context_provider, + input.attempt_id, ) .await; ProcessingOutcome::CompletedNoPr { @@ -1855,6 +2007,7 @@ impl IssueProcessor { &input.resolution, &input.source_name, context_provider, + input.attempt_id, ) .await } @@ -1900,6 +2053,7 @@ impl IssueProcessor { &input.resolution, &input.source_name, context_provider, + input.attempt_id, ) .await; if verdict.reproduced { @@ -1926,6 +2080,7 @@ impl IssueProcessor { &input.resolution, &input.source_name, context_provider, + input.attempt_id, ) .await } else { @@ -1935,6 +2090,7 @@ impl IssueProcessor { &input.resolution, &input.source_name, context_provider, + input.attempt_id, ) .await } @@ -2011,9 +2167,10 @@ impl IssueProcessor { resolution: &RepoResolution, source_name: &str, context_provider: &dyn ContextProvider, + attempt_id: Option, ) -> VerifyResult { let project_dir = self.action_project_dir(resolution); - let context = self.build_rag_context(issue).await; + let context = self.build_rag_context(issue, attempt_id).await; self.record_issue_decision( issue, @@ -2129,9 +2286,10 @@ impl IssueProcessor { resolution: &RepoResolution, source_name: &str, context_provider: &dyn ContextProvider, + attempt_id: Option, ) -> ProcessingOutcome { let project_dir = self.action_project_dir(resolution); - let context = self.build_rag_context(issue).await; + let context = self.build_rag_context(issue, attempt_id).await; // The inbox key is the HelpScout mailbox id when present, else the source. let inbox_key = issue @@ -2139,6 +2297,9 @@ impl IssueProcessor { .unwrap_or_else(|| source_name.to_string()); let guideline = self.config.reply().template_for(Some(&inbox_key)); + // Ground the reply in the reply thread when this issue is a reply. + let context = self.with_reply_chain(issue, context).await; + self.record_issue_decision( issue, "reply_started", @@ -2164,23 +2325,47 @@ impl IssueProcessor { Ok(Ok(reply)) => { // Deliver: conversational sources go via the notifier; tracker // sources post a comment on the ticket (falling back to notifier). - let delivered = if qa_eligible_source(source_name) { - self.notifier.notify_answer(issue, &reply).await + // Capture any sent message ids so a later reply maps back here. + let mut answer_ids: Vec = Vec::new(); + let delivered: Result<()> = if qa_eligible_source(source_name) { + match self.notifier.notify_answer(issue, &reply).await { + Ok(ids) => { + answer_ids = ids; + Ok(()) + } + Err(e) => Err(e), + } } else { match context_provider.post_reply(&issue.id, &reply).await { Ok(()) => Ok(()), Err(e) => { tracing::warn!(short_id = %issue.short_id, error = %e, "post_reply failed; falling back to notifier"); - self.notifier.notify_answer(issue, &reply).await + match self.notifier.notify_answer(issue, &reply).await { + Ok(ids) => { + answer_ids = ids; + Ok(()) + } + Err(e) => Err(e), + } } } }; if let Err(e) = delivered { tracing::warn!(short_id = %issue.short_id, error = %e, "Failed to deliver reply"); } + if !answer_ids.is_empty() { + if let Err(e) = + self.tracker + .record_answer_message_ids(source_name, &issue.id, &answer_ids) + { + tracing::warn!(short_id = %issue.short_id, error = %e, "Failed to record answer message ids"); + } + } + // Store the FULL reply for reply-chain grounding; the truncated + // `summary` is only for the action-run preview below. let summary: String = reply.chars().take(500).collect(); - let _ = self.tracker.mark_answered(source_name, &issue.id, &summary); + let _ = self.tracker.mark_answered(source_name, &issue.id, &reply); let _ = self.tracker.record_action_run( source_name, &issue.id, @@ -2236,7 +2421,9 @@ impl IssueProcessor { /// Retrieve RAG grounding context for an issue from the code index, plus any /// indexed Discord discussions. - async fn build_rag_context(&self, issue: &Issue) -> String { + /// Build the RAG grounding context for the action pipeline (verify/reply). + async fn build_rag_context(&self, issue: &Issue, attempt_id: Option) -> String { + let mut retrieved_items: Vec = Vec::new(); let mut context = String::new(); if let Some(ref code_search) = self.code_search_service { let query = claudear_analysis::repo::code_index::build_code_search_query(issue); @@ -2245,13 +2432,39 @@ impl IssueProcessor { .await { if !results.is_empty() { + if let Some(id) = attempt_id { + let mut rows: Vec = Vec::with_capacity(results.len()); + for (rank, r) in results.iter().enumerate() { + let chunk_ref = r + .chunk + .id + .map(|i| i.to_string()) + .unwrap_or_else(|| r.chunk.file_path.clone()); + retrieved_items.push(RetrievedItem { + source_kind: "code_chunk".to_string(), + chunk_ref: chunk_ref.clone(), + text: r.chunk.chunk_text.clone(), + }); + rows.push(RetrievalUsageRecord::new( + id, + "code_chunk", + chunk_ref, + Some(r.chunk.file_path.clone()), + rank as i64, + r.score, + Some(r.chunk.chunk_text.chars().count() as i64), + )); + } + self.tracker.record_retrieval_usage(&rows).ok(); + self.log_retrieval_recorded(&rows); + } context = claudear_analysis::repo::code_index::format_code_search_context(&results); } } } - let (discord_ctx, _) = self - .discord_grounding_context(issue, self.config.qa.max_context_chunks, None) + let (discord_ctx, discord_items) = self + .discord_grounding_context(issue, self.config.qa.max_context_chunks, attempt_id) .await; if !discord_ctx.is_empty() { context = if context.is_empty() { @@ -2259,7 +2472,17 @@ impl IssueProcessor { } else { format!("{}\n{}", context, discord_ctx) }; + if attempt_id.is_some() { + retrieved_items.extend(discord_items); + } } + + // Score retrieval quality for this action run (opt-in, detached — never + // blocks the reply/verify). Mirrors the Q&A pipeline. + if let Some(id) = attempt_id { + self.spawn_retrieval_judge(id, issue, &retrieved_items); + } + context } @@ -5047,6 +5270,117 @@ mod tests { assert!(confidence_comment.contains("85/100")); } + // --- Reply-chain assembly --- + + #[test] + fn test_strip_discord_mentions() { + assert_eq!( + strip_discord_mentions("<@123> hey <@!456> there <@&789>!"), + "hey there !" + ); + assert_eq!(strip_discord_mentions(" no mentions "), "no mentions"); + // Unterminated mention is left as-is (no infinite loop). + assert_eq!(strip_discord_mentions("<@123 broken"), "<@123 broken"); + } + + fn make_reply_chain_processor(tracker: Arc) -> IssueProcessor { + let notifier: Arc = + Arc::new(claudear_integrations::notifier::ConsoleNotifier::new()); + let agent: Arc = Arc::new(DummyAgent); + let feedback_analyzer = Arc::new(tokio::sync::Mutex::new( + claudear_analysis::feedback::FeedbackAnalyzer::new(), + )); + IssueProcessor { + config: Config::default(), + tracker, + notifier, + agent, + qa_agent: None, + inferrer: None, + embedding_client: None, + issue_embedding_service: None, + code_search_service: None, + discord_search_service: None, + feedback_analyzer, + review_watcher: None, + user_registry: claudear_config::users::UserRegistry::new( + std::collections::HashMap::new(), + ), + github_client: None, + llm_analyzer: None, + intent_classifier: None, + } + } + + #[tokio::test] + async fn test_assemble_reply_chain_claudear_branch_from_db() { + use claudear_storage::EmbeddingStore; + let tracker = claudear_storage::SqliteTracker::in_memory().unwrap(); + // Seed the original question issue and Claudear's stored answer, mapped to + // the answer message id the follow-up will reply to. + let mut q_issue = Issue::new( + "QID", + "DISCORD-QID", + "how does X work?", + "https://d/x", + "discord", + ); + q_issue.description = Some("how does X work in detail?".to_string()); + tracker + .store_issue(&claudear_core::types::IssueEmbedding::from_issue(&q_issue)) + .unwrap(); + tracker + .record_attempt("discord", "QID", "DISCORD-QID") + .unwrap(); + tracker + .mark_answered( + "discord", + "QID", + "X works via the frobnicator; long answer.", + ) + .unwrap(); + tracker + .record_answer_message_ids("discord", "QID", &["ANSMSG1".to_string()]) + .unwrap(); + + let tracker: Arc = Arc::new(tracker); + let processor = make_reply_chain_processor(tracker); + + // Follow-up replying to Claudear's answer (ANSMSG1): resolves purely from + // the DB, no Discord fetch. + let mut follow = Issue::new( + "FID", + "DISCORD-FID", + "what about Y?", + "https://d/y", + "discord", + ); + follow.set_metadata("reply_to_message_id", "ANSMSG1"); + follow.set_metadata("reply_to_channel_id", "chan"); + + let chain = processor + .assemble_reply_chain(&follow) + .await + .expect("reply chain resolved from DB"); + assert!(chain.contains("## Prior conversation")); + assert!(chain.contains("[Claudear]: X works via the frobnicator")); + assert!(chain.contains("[User]: how does X work in detail?")); + // Chronological order: the question (older) precedes the answer (newer). + assert!(chain.find("[User]:").unwrap() < chain.find("[Claudear]:").unwrap()); + } + + #[tokio::test] + async fn test_assemble_reply_chain_non_reply_returns_none() { + let tracker: Arc = + Arc::new(claudear_storage::SqliteTracker::in_memory().unwrap()); + let processor = make_reply_chain_processor(tracker); + // make_test_issue carries no reply_to_message_id metadata. + assert!(processor + .assemble_reply_chain(&make_test_issue()) + .await + .is_none()); + } + // --- Dummy test helpers --- /// Dummy agent runner that does nothing (for IssueProcessor tests). diff --git a/crates/claudear-integrations/src/notifier/discord.rs b/crates/claudear-integrations/src/notifier/discord.rs index 5c5268b2..5ea3a480 100644 --- a/crates/claudear-integrations/src/notifier/discord.rs +++ b/crates/claudear-integrations/src/notifier/discord.rs @@ -1362,7 +1362,7 @@ impl Notifier for DiscordNotifier { Ok(()) } - async fn notify_answer(&self, issue: &Issue, answer: &str) -> Result<()> { + async fn notify_answer(&self, issue: &Issue, answer: &str) -> Result> { if !self.has_delivery_path() { return Err(Error::notifier( "discord", @@ -1371,14 +1371,19 @@ impl Notifier for DiscordNotifier { } let mention = self.get_user_mention_for_issue(issue); // Reply to the original question with the first message so the answer is - // threaded to it; any continuation chunks follow as normal messages. + // threaded to it; any continuation chunks follow as normal messages. We + // return the ids of the sent messages so a later reply to any chunk maps + // back to the issue for reply-chain context. + let mut ids = Vec::new(); for (i, message) in build_answer_messages(issue, answer, mention) .into_iter() .enumerate() { - let _ = self.send_to_issue_channel(issue, message, i == 0).await?; + if let Some(info) = self.send_to_issue_channel(issue, message, i == 0).await? { + ids.push(info.message_id); + } } - Ok(()) + Ok(ids) } async fn notify_urgent_issues(&self, issues: &[Issue]) -> Result<()> { diff --git a/crates/claudear-integrations/src/notifier/mod.rs b/crates/claudear-integrations/src/notifier/mod.rs index a6ab6f05..43826039 100644 --- a/crates/claudear-integrations/src/notifier/mod.rs +++ b/crates/claudear-integrations/src/notifier/mod.rs @@ -172,11 +172,14 @@ pub trait Notifier: Send + Sync { /// Send a RAG-grounded answer to a question back to the originating channel. /// - /// Default implementation falls back to a plain status message; channels that - /// can reply to a specific conversation (e.g. Discord) should override this. - async fn notify_answer(&self, issue: &Issue, answer: &str) -> Result<()> { + /// Returns the ids of any messages the channel sent for this answer (used to + /// map a later reply back to the issue). The default falls back to a plain + /// status message and returns no ids; channels that can reply to a specific + /// conversation (e.g. Discord) should override this and return their ids. + async fn notify_answer(&self, issue: &Issue, answer: &str) -> Result> { self.notify_status(&format!("Answer for {}:\n{}", issue.short_id, answer)) - .await + .await?; + Ok(Vec::new()) } /// Notify that a repo swap is happening for an issue. @@ -314,16 +317,26 @@ impl Notifier for CompositeNotifier { Ok(()) } - async fn notify_answer(&self, issue: &Issue, answer: &str) -> Result<()> { - let issue = issue.clone(); - let answer = answer.to_string(); - self.broadcast(|n| { + async fn notify_answer(&self, issue: &Issue, answer: &str) -> Result> { + // Deliver to every notifier concurrently (like `broadcast`), tolerating + // per-notifier failures, and collect the sent message ids so callers can + // map a later reply back to the issue (only reply-capable channels like + // Discord contribute ids). `broadcast` can't be reused here: it requires + // `Result<()>` futures and discards their return values. + let futures = self.notifiers.iter().map(|n| { + let n = Arc::clone(n); let issue = issue.clone(); - let answer = answer.clone(); + let answer = answer.to_string(); async move { n.notify_answer(&issue, &answer).await } - }) - .await; - Ok(()) + }); + let mut ids = Vec::new(); + for result in futures::future::join_all(futures).await { + match result { + Ok(mut sent) => ids.append(&mut sent), + Err(e) => tracing::warn!(error = %e, "notify_answer failed"), + } + } + Ok(ids) } async fn notify_merged(&self, issue: &Issue, pr_url: &str) -> Result<()> { diff --git a/crates/claudear-integrations/src/source/discord.rs b/crates/claudear-integrations/src/source/discord.rs index 16dcb0a6..436df3a2 100644 --- a/crates/claudear-integrations/src/source/discord.rs +++ b/crates/claudear-integrations/src/source/discord.rs @@ -164,6 +164,22 @@ impl DiscordSource { } issue.set_metadata("channel_id", &msg.channel_id); + // When this message is a reply, record the parent it points at so the + // engine can reconstruct the reply-chain conversation as grounding + // context. Only the parent id is available on the payload; the parent's + // content is resolved later (from our DB for Claudear's own answers, or + // by fetching from Discord for other users' messages). + if let Some(ref reference) = msg.message_reference { + if let Some(ref parent_id) = reference.message_id { + issue.set_metadata("reply_to_message_id", parent_id); + let parent_channel = reference + .channel_id + .clone() + .unwrap_or_else(|| msg.channel_id.clone()); + issue.set_metadata("reply_to_channel_id", &parent_channel); + } + } + issue } } @@ -518,6 +534,55 @@ mod tests { ); } + #[test] + fn test_message_to_issue_captures_reply_pointer() { + let source = DiscordSource::new(make_config()); + let mut msg = make_message("999", "what about Y?", false); + msg.message_reference = Some(crate::discord::DiscordMessageReference { + message_id: Some("parent-42".to_string()), + channel_id: Some("thread-7".to_string()), + guild_id: None, + fail_if_not_exists: None, + }); + let issue = source.message_to_issue(&msg); + assert_eq!( + issue.get_metadata::("reply_to_message_id"), + Some("parent-42".to_string()) + ); + assert_eq!( + issue.get_metadata::("reply_to_channel_id"), + Some("thread-7".to_string()) + ); + } + + #[test] + fn test_message_to_issue_reply_channel_falls_back_to_message_channel() { + let source = DiscordSource::new(make_config()); + let mut msg = make_message("999", "follow up", false); + msg.message_reference = Some(crate::discord::DiscordMessageReference { + message_id: Some("parent-42".to_string()), + channel_id: None, + guild_id: None, + fail_if_not_exists: None, + }); + let issue = source.message_to_issue(&msg); + // No reference channel: falls back to the message's own channel. + assert_eq!( + issue.get_metadata::("reply_to_channel_id"), + Some("chan-123".to_string()) + ); + } + + #[test] + fn test_message_to_issue_no_reply_pointer_when_not_reply() { + let source = DiscordSource::new(make_config()); + let msg = make_message("999", "not a reply", false); + let issue = source.message_to_issue(&msg); + assert!(issue + .get_metadata::("reply_to_message_id") + .is_none()); + } + #[test] fn test_listen_channel_id_fallback() { let source = DiscordSource::new(make_config()); diff --git a/crates/claudear-integrations/src/telemetry.rs b/crates/claudear-integrations/src/telemetry.rs index fc58a835..b17b4c51 100644 --- a/crates/claudear-integrations/src/telemetry.rs +++ b/crates/claudear-integrations/src/telemetry.rs @@ -212,6 +212,22 @@ impl Notifier for InstrumentedNotifier { } } + async fn notify_answer(&self, issue: &Issue, answer: &str) -> Result> { + // Forward to the inner notifier so reply-capable channels (e.g. Discord) + // actually run and return their sent message ids. Without this the trait + // default would run here and the ids would be lost. + match self.inner.notify_answer(issue, answer).await { + Ok(ids) => { + tracing::info!(component = self.inner.name(), issue = %issue.short_id, sent = ids.len(), "Notified answer"); + Ok(ids) + } + Err(e) => { + tracing::warn!(component = self.inner.name(), issue = %issue.short_id, error = %e, "Failed to notify answer"); + Err(e) + } + } + } + async fn notify_urgent_issues(&self, issues: &[Issue]) -> Result<()> { match self.inner.notify_urgent_issues(issues).await { Ok(v) => { diff --git a/crates/claudear-storage/src/lib.rs b/crates/claudear-storage/src/lib.rs index f05fec3d..c96a9dfa 100644 --- a/crates/claudear-storage/src/lib.rs +++ b/crates/claudear-storage/src/lib.rs @@ -197,6 +197,30 @@ pub trait AttemptTracker: Send + Sync { Ok(()) } + /// Record the Discord message ids of the answer chunks Claudear sent for an + /// issue, so a user's reply to any chunk can be mapped back to the issue. + /// + /// Default no-op; persistent trackers should store the ids for later lookup + /// via [`FixAttemptTracker::lookup_answer_issue`]. + fn record_answer_message_ids( + &self, + source: &str, + issue_id: &str, + message_ids: &[String], + ) -> Result<()> { + let _ = (source, issue_id, message_ids); + Ok(()) + } + + /// Reverse lookup: given a Discord message id the bot sent as (part of) an + /// answer, return the `(source, issue_id)` it belongs to, if known. + /// + /// Default returns `None`. + fn lookup_answer_issue(&self, message_id: &str) -> Result> { + let _ = message_id; + Ok(None) + } + /// Get issues that are eligible for retry (failed/closed with retry_count < max_retries). fn get_retryable_issues(&self, max_retries: u32) -> Result>; diff --git a/crates/claudear-storage/src/migrator.rs b/crates/claudear-storage/src/migrator.rs index 922ce251..8ebffbb8 100644 --- a/crates/claudear-storage/src/migrator.rs +++ b/crates/claudear-storage/src/migrator.rs @@ -51,6 +51,11 @@ const MIGRATIONS: &[Migration] = &[ name: "retrieval_usage", sql: include_str!("../../../migrations/V7__retrieval_usage.sql"), }, + Migration { + version: 8, + name: "answer_message_ids", + sql: include_str!("../../../migrations/V8__answer_message_ids.sql"), + }, ]; /// Run all pending migrations against the given connection. @@ -111,7 +116,7 @@ mod tests { row.get(0) }) .unwrap(); - assert_eq!(version, 7); + assert_eq!(version, 8); // Verify a table from V1 exists let count: u32 = conn @@ -122,6 +127,16 @@ mod tests { ) .unwrap(); assert_eq!(count, 1); + + // Verify the V8 column exists on fix_attempts. + let has_col: u32 = conn + .query_row( + "SELECT COUNT(*) FROM pragma_table_info('fix_attempts') WHERE name = 'answer_message_ids'", + [], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(has_col, 1); } #[test] @@ -136,7 +151,7 @@ mod tests { row.get(0) }) .unwrap(); - assert_eq!(version, 7); + assert_eq!(version, 8); } #[test] diff --git a/crates/claudear-storage/src/sqlite.rs b/crates/claudear-storage/src/sqlite.rs index 27b65206..15f0739f 100644 --- a/crates/claudear-storage/src/sqlite.rs +++ b/crates/claudear-storage/src/sqlite.rs @@ -1342,6 +1342,74 @@ impl AttemptTracker for SqliteTracker { Ok(()) } + fn record_answer_message_ids( + &self, + source: &str, + issue_id: &str, + message_ids: &[String], + ) -> Result<()> { + if message_ids.is_empty() { + return Ok(()); + } + let conn = self.acquire_lock()?; + let existing: Option = conn + .query_row( + "SELECT answer_message_ids FROM fix_attempts WHERE source = ?1 AND issue_id = ?2", + params![source, issue_id], + |row| row.get::<_, Option>(0), + ) + .ok() + .flatten(); + let mut ids: Vec = existing + .as_deref() + .unwrap_or("") + .split(',') + .filter(|s| !s.is_empty()) + .map(str::to_string) + .collect(); + for id in message_ids { + if !ids.iter().any(|existing| existing == id) { + ids.push(id.clone()); + } + } + let packed = format!(",{},", ids.join(",")); + conn.execute( + r#" + UPDATE fix_attempts + SET answer_message_ids = ? + WHERE source = ? AND issue_id = ? + "#, + params![packed, source, issue_id], + )?; + Ok(()) + } + + fn lookup_answer_issue(&self, message_id: &str) -> Result> { + if message_id.is_empty() { + return Ok(None); + } + let escaped = message_id + .replace('\\', "\\\\") + .replace('%', "\\%") + .replace('_', "\\_"); + let pattern = format!("%,{},%", escaped); + let conn = self.acquire_lock()?; + let mut stmt = conn.prepare_cached( + r#" + SELECT source, issue_id + FROM fix_attempts + WHERE answer_message_ids LIKE ? ESCAPE '\' + LIMIT 1 + "#, + )?; + let result = stmt + .query_row(params![pattern], |row| { + Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)) + }) + .ok(); + Ok(result) + } + fn get_retryable_issues(&self, max_retries: u32) -> Result> { let conn = self.acquire_lock()?; let mut stmt = conn.prepare_cached( @@ -9580,6 +9648,58 @@ mod tests { assert_eq!(attempt.status, FixAttemptStatus::Pending); } + #[test] + fn test_answer_message_ids_round_trip() { + let tracker = SqliteTracker::in_memory().unwrap(); + tracker + .record_attempt("discord", "999000111", "DISCORD-99900011") + .unwrap(); + + let chunk_ids = vec!["555111".to_string(), "555222".to_string()]; + tracker + .record_answer_message_ids("discord", "999000111", &chunk_ids) + .unwrap(); + + // Any chunk id resolves back to the (source, issue_id). + assert_eq!( + tracker.lookup_answer_issue("555111").unwrap(), + Some(("discord".to_string(), "999000111".to_string())) + ); + assert_eq!( + tracker.lookup_answer_issue("555222").unwrap(), + Some(("discord".to_string(), "999000111".to_string())) + ); + + // A partial id must NOT match (delimiter guard). + assert_eq!(tracker.lookup_answer_issue("5551").unwrap(), None); + // An unknown id resolves to nothing. + assert_eq!(tracker.lookup_answer_issue("777").unwrap(), None); + // Empty input is safe. + assert_eq!(tracker.lookup_answer_issue("").unwrap(), None); + tracker + .record_answer_message_ids("discord", "999000111", &[]) + .unwrap(); + + // A LIKE wildcard input must not match (wildcards are escaped). + assert_eq!(tracker.lookup_answer_issue("%").unwrap(), None); + + // Recording again appends without dropping earlier ids, and re-recording + // an existing id does not duplicate it. + tracker + .record_answer_message_ids("discord", "999000111", &["555333".to_string()]) + .unwrap(); + tracker + .record_answer_message_ids("discord", "999000111", &["555111".to_string()]) + .unwrap(); + for id in ["555111", "555222", "555333"] { + assert_eq!( + tracker.lookup_answer_issue(id).unwrap(), + Some(("discord".to_string(), "999000111".to_string())), + "id {id} should still resolve after append" + ); + } + } + #[test] fn test_mark_success() { let tracker = SqliteTracker::in_memory().unwrap(); diff --git a/migrations/V8__answer_message_ids.sql b/migrations/V8__answer_message_ids.sql new file mode 100644 index 00000000..f54e6ba3 --- /dev/null +++ b/migrations/V8__answer_message_ids.sql @@ -0,0 +1,9 @@ +-- V8: Track the Discord message ids of the answers Claudear sends, so a user's +-- reply to any answer chunk can be mapped back to the issue it belongs to +-- (reply-chain context) without fetching the message from Discord. +-- +-- Stored as a comma-delimited list with leading/trailing commas, e.g. +-- ",123,456,789," +-- so reverse lookups can match a whole id with LIKE '%,,%' and never match +-- a partial id. +ALTER TABLE fix_attempts ADD COLUMN answer_message_ids TEXT;