use std::collections::HashSet;
use anyhow::Result;
use rusqlite::Connection;
use super::normalize::{compact_line, relevant_to_request, search_query};
use super::types::{
ClaimCandidate, NormalizedRequest, RecallCandidate, RecallState, UserRecallDroppedItem,
MAX_CLAIM_SCAN, MAX_MEMORY_SCAN, MAX_SESSION_SCAN,
};
use crate::user_context::claims::{self, DEFAULT_OWNER_KEY, DEFAULT_OWNER_SCOPE};
pub(super) fn collect_summary(
conn: &Connection,
req: &NormalizedRequest,
state: &mut RecallState,
) -> Result<()> {
let summary_req = crate::user_context::summary::SummaryRequest {
owner_scope: Some(&req.owner_scope),
owner_key: Some(&req.owner_key),
project: &req.project,
};
if let Some(summary) = crate::user_context::summary::load_active_summary(conn, &summary_req)? {
state.counts.summaries += 1;
if !relevant_to_request(&summary.summary_text, req) {
return Ok(());
}
state.candidates.push(RecallCandidate {
source_type: "profile_summary".to_string(),
source_id: Some(summary.id),
title: Some(format!("profile summary v{}", summary.version)),
text: compact_line(&summary.summary_text, 900),
reason_codes: vec![
"profile_summary".to_string(),
"query_match".to_string(),
"safe_sources".to_string(),
],
source_refs: Some(serde_json::json!({
"claim_ids": summary.source_claim_ids,
"memory_ids": summary.source_memory_ids,
"activity_refs": summary.source_activity_refs,
})),
priority: 80,
});
}
Ok(())
}
pub(super) fn collect_claims(
conn: &Connection,
req: &NormalizedRequest,
state: &mut RecallState,
) -> Result<()> {
let claims = load_claim_candidates(conn, req)?;
state.counts.claims += claims.len();
let now = chrono::Utc::now().timestamp();
for claim in claims {
let label = Some(format!("{}:{}", claim.claim_type, claim.claim_key));
if let Some(reason) = recall_claim_drop_reason(conn, &claim, req, now)? {
state.dropped.push(UserRecallDroppedItem {
source_type: "user_claim".to_string(),
source_id: Some(claim.id),
label,
reason_code: reason,
});
continue;
}
if !relevant_to_request(&claim.claim_text, req)
&& !relevant_to_request(&claim.claim_key, req)
&& !relevant_to_request(&claim.claim_type, req)
{
state.dropped.push(UserRecallDroppedItem {
source_type: "user_claim".to_string(),
source_id: Some(claim.id),
label,
reason_code: "not_relevant".to_string(),
});
continue;
}
let source_refs = serde_json::from_str::<serde_json::Value>(&claim.source_refs_json).ok();
state.candidates.push(RecallCandidate {
source_type: "user_claim".to_string(),
source_id: Some(claim.id),
title: Some(format!("{}:{}", claim.claim_type, claim.claim_key)),
text: compact_line(&claim.claim_text, 500),
reason_codes: vec![
"active_user_claim".to_string(),
"query_match".to_string(),
format!("owner:{}:{}", claim.owner_scope, claim.owner_key),
],
source_refs,
priority: 100,
});
}
Ok(())
}
pub(super) fn collect_memories(
conn: &Connection,
req: &NormalizedRequest,
state: &mut RecallState,
) -> Result<()> {
let query = search_query(req);
let mut offset = 0_i64;
let mut scanned = Vec::new();
let mut visibility = std::collections::HashMap::new();
let mut explicitly_suppressed = HashSet::new();
let page_size = 100_i64;
let max_pages = MAX_MEMORY_SCAN / page_size;
for page_index in 0..max_pages {
let result = crate::memory::service::search_memories(
conn,
&crate::memory::service::SearchRequest {
query: Some(query.clone()),
project: Some(req.project.clone()),
memory_type: None,
limit: page_size,
offset,
include_stale: false,
include_suppressed: req.include_suppressed,
branch: None,
multi_hop: false,
explain: false,
},
)?;
let page_len = result.memories.len();
let has_more = result.has_more;
let ids = result
.memories
.iter()
.map(|memory| memory.id)
.collect::<Vec<_>>();
if req.include_suppressed {
explicitly_suppressed.extend(crate::memory::suppression::active_suppressed_memory_ids(
conn, &ids,
)?);
}
visibility.extend(crate::truth::admit_many_for_current_context(
conn,
&ids,
chrono::Utc::now().timestamp(),
)?);
scanned.extend(result.memories);
let mut eligible = 0usize;
for memory in &scanned {
if (visibility
.get(&memory.id)
.is_some_and(|row| row.current_context_eligible)
|| explicitly_suppressed.contains(&memory.id))
&& !claims::active_preference_backfill_covers_user_preference_memory(
conn, memory.id,
)?
{
eligible += 1;
}
}
if eligible >= 5 || !has_more || page_len < page_size as usize {
break;
}
if page_index + 1 == max_pages {
anyhow::bail!(
"user-context memory G2 scan budget exhausted after {MAX_MEMORY_SCAN} rows"
);
}
offset = offset
.checked_add(page_size)
.ok_or_else(|| anyhow::anyhow!("recall G2 offset overflow"))?;
}
state.counts.memories += scanned.len();
let mut admitted = 0usize;
for memory in scanned {
if claims::active_preference_backfill_covers_user_preference_memory(conn, memory.id)? {
state.dropped.push(UserRecallDroppedItem {
source_type: "memory".to_string(),
source_id: Some(memory.id),
label: Some(memory.title),
reason_code: "backfilled_as_user_claim".to_string(),
});
continue;
}
let classification = visibility.get(&memory.id).copied().ok_or_else(|| {
anyhow::anyhow!("missing visibility for recalled memory {}", memory.id)
})?;
let suppression_override =
req.include_suppressed && explicitly_suppressed.contains(&memory.id);
if !classification.current_context_eligible && !suppression_override {
state.dropped.push(UserRecallDroppedItem {
source_type: "memory".to_string(),
source_id: Some(memory.id),
label: Some(memory.title),
reason_code: classification.reason.as_str().to_string(),
});
continue;
}
if admitted == 5 {
break;
}
let mut reason_codes = vec![
"repo_memory_match".to_string(),
"search_result".to_string(),
format!("type:{}", memory.memory_type),
];
if suppression_override {
reason_codes.push("explicit_suppression_override".to_string());
}
state.candidates.push(RecallCandidate {
source_type: "memory".to_string(),
source_id: Some(memory.id),
title: Some(memory.title),
text: compact_line(&memory.text, 650),
reason_codes,
source_refs: Some(serde_json::json!({
"topic_key": memory.topic_key,
"project": memory.project,
"status": memory.status,
})),
priority: 70,
});
admitted += 1;
}
Ok(())
}
pub(super) fn collect_current_state(
conn: &Connection,
req: &NormalizedRequest,
state: &mut RecallState,
) -> Result<()> {
for state_key in &req.state_keys {
let result = crate::memory::current_state::current_state(
conn,
&crate::memory::current_state::CurrentStateRequest {
state_key: state_key.clone(),
project: Some(req.project.clone()),
owner_scope: None,
owner_key: None,
memory_type: None,
as_of_epoch: None,
include_history: false,
},
)?;
if let Some(current) = result.current {
state.counts.current_state += 1;
state.candidates.push(RecallCandidate {
source_type: "current_state".to_string(),
source_id: Some(current.id),
title: Some(format!("current state: {state_key}")),
text: compact_line(¤t.text, 650),
reason_codes: vec![
"current_state_answer".to_string(),
format!("state_key:{state_key}"),
],
source_refs: Some(serde_json::json!({
"state_key": state_key,
"memory_id": current.id,
"topic_key": current.topic_key,
"status": current.status,
})),
priority: 90,
});
} else {
state.dropped.push(UserRecallDroppedItem {
source_type: "current_state".to_string(),
source_id: None,
label: Some(state_key.clone()),
reason_code: format!("current_state_{}", result.status),
});
}
}
Ok(())
}
pub(super) fn collect_workstreams(
conn: &Connection,
req: &NormalizedRequest,
state: &mut RecallState,
) -> Result<()> {
let workstreams = crate::workstream::query_active_workstreams(conn, &req.project)?;
state.counts.workstreams += workstreams.len();
for workstream in workstreams {
let text = [
workstream.title.as_str(),
workstream.progress.as_deref().unwrap_or_default(),
workstream.next_action.as_deref().unwrap_or_default(),
workstream.blockers.as_deref().unwrap_or_default(),
]
.join(" ");
if !relevant_to_request(&text, req) {
continue;
}
state.candidates.push(RecallCandidate {
source_type: "workstream".to_string(),
source_id: Some(workstream.id),
title: Some(workstream.title),
text: compact_line(&text, 500),
reason_codes: vec!["active_workstream".to_string(), "query_match".to_string()],
source_refs: Some(serde_json::json!({
"project": workstream.project,
"status": workstream.status.as_str(),
"updated_at_epoch": workstream.updated_at_epoch,
})),
priority: 60,
});
}
Ok(())
}
pub(super) fn collect_recent_sessions(
conn: &Connection,
req: &NormalizedRequest,
state: &mut RecallState,
) -> Result<()> {
let mut params_vec: Vec<Box<dyn rusqlite::types::ToSql>> = Vec::new();
let mut idx = 1;
let (owner_clause, next) = crate::project_alias::push_project_value_filter(
conn,
"owner_key",
&req.project,
idx,
&mut params_vec,
)?;
idx = next;
let (target_clause, next) = crate::project_alias::push_project_value_filter(
conn,
"target_project",
&req.project,
idx,
&mut params_vec,
)?;
idx = next;
let (legacy_clause, next) = crate::project_alias::push_project_value_filter(
conn,
"project",
&req.project,
idx,
&mut params_vec,
)?;
idx = next;
let sql = format!(
"SELECT id,
CASE
WHEN request LIKE 'Captured event range %..%' THEN
COALESCE(NULLIF(decisions, ''), NULLIF(learned, ''),
NULLIF(next_steps, ''), NULLIF(preferences, ''),
NULLIF(completed, ''), '')
ELSE COALESCE(request, '')
END AS display_request,
COALESCE(completed, ''),
COALESCE(decisions, ''), COALESCE(learned, ''),
COALESCE(next_steps, ''), COALESCE(preferences, ''),
created_at_epoch
FROM session_summaries
WHERE COALESCE(poisoning_status, 'legacy_unscanned') != 'quarantined'
AND (session_row_id IS NULL
OR request NOT LIKE 'Captured event range %..%'
OR COALESCE(decisions, '') != ''
OR COALESCE(learned, '') != ''
OR COALESCE(next_steps, '') != ''
OR COALESCE(preferences, '') != '')
AND ((owner_scope = 'repo' AND {owner_clause})
OR (owner_scope = 'repo' AND {target_clause})
OR (owner_scope IS NULL AND {legacy_clause}))
ORDER BY created_at_epoch DESC, id DESC
LIMIT ?{idx}"
);
params_vec.push(Box::new(MAX_SESSION_SCAN));
let mut stmt = conn.prepare(&sql)?;
let refs = crate::db::to_sql_refs(¶ms_vec);
let rows = stmt.query_map(refs.as_slice(), |row| {
Ok(SessionCandidate {
id: row.get(0)?,
request: row.get(1)?,
completed: row.get(2)?,
decisions: row.get(3)?,
learned: row.get(4)?,
next_steps: row.get(5)?,
preferences: row.get(6)?,
created_at_epoch: row.get(7)?,
})
})?;
let mut sessions = crate::db::query::collect_rows(rows)?;
sessions.retain(|session| {
crate::db::summary_poisoning::summary_injectable(
conn,
session.id,
&[
("request", Some(session.request.as_str())),
("completed", Some(session.completed.as_str())),
("decisions", Some(session.decisions.as_str())),
("learned", Some(session.learned.as_str())),
("next_steps", Some(session.next_steps.as_str())),
("preferences", Some(session.preferences.as_str())),
],
"user_context_recall",
)
});
state.counts.sessions += sessions.len();
let mut seen_session_text = HashSet::new();
for session in sessions {
let text = session.text();
let dedupe_key = text.to_ascii_lowercase();
if !seen_session_text.insert(dedupe_key) {
continue;
}
if !relevant_to_request(&text, req) {
continue;
}
state.candidates.push(RecallCandidate {
source_type: "session_summary".to_string(),
source_id: Some(session.id),
title: Some(compact_line(&session.request, 120)),
text: compact_line(&text, 550),
reason_codes: vec!["recent_session".to_string(), "query_match".to_string()],
source_refs: Some(serde_json::json!({
"project": req.project,
"created_at_epoch": session.created_at_epoch,
})),
priority: 40,
});
}
Ok(())
}
fn load_claim_candidates(
conn: &Connection,
req: &NormalizedRequest,
) -> Result<Vec<ClaimCandidate>> {
let mut values: Vec<Box<dyn rusqlite::types::ToSql>> = Vec::new();
let mut idx = 1;
let owner_filter =
if req.owner_scope == DEFAULT_OWNER_SCOPE && req.owner_key == DEFAULT_OWNER_KEY {
values.push(Box::new(req.owner_scope.clone()));
values.push(Box::new(req.owner_key.clone()));
idx += 2;
let (repo_clause, next) = crate::project_alias::push_project_value_filter(
conn,
"owner_key",
&req.project,
idx,
&mut values,
)?;
idx = next;
format!(
"((owner_scope = ?1 AND owner_key = ?2) \
OR (owner_scope = 'repo' AND {repo_clause}))"
)
} else {
values.push(Box::new(req.owner_scope.clone()));
values.push(Box::new(req.owner_key.clone()));
idx += 2;
"owner_scope = ?1 AND owner_key = ?2".to_string()
};
let sql = format!(
"SELECT id, claim_type, claim_key, claim_text, owner_scope, owner_key,
sensitivity, source_refs_json, status, valid_from_epoch, valid_to_epoch
FROM user_context_claims
WHERE {owner_filter}
ORDER BY updated_at_epoch DESC, id DESC
LIMIT ?{idx}",
);
values.push(Box::new(MAX_CLAIM_SCAN));
let mut stmt = conn.prepare(&sql)?;
let refs = crate::db::to_sql_refs(&values);
let rows = stmt.query_map(refs.as_slice(), |row| {
Ok(ClaimCandidate {
id: row.get(0)?,
claim_type: row.get(1)?,
claim_key: row.get(2)?,
claim_text: row.get(3)?,
owner_scope: row.get(4)?,
owner_key: row.get(5)?,
sensitivity: row.get(6)?,
source_refs_json: row.get(7)?,
status: row.get(8)?,
valid_from_epoch: row.get(9)?,
valid_to_epoch: row.get(10)?,
})
})?;
crate::db::query::collect_rows(rows)
}
fn recall_claim_drop_reason(
conn: &Connection,
claim: &ClaimCandidate,
req: &NormalizedRequest,
now: i64,
) -> Result<Option<String>> {
if claim.status == "suppressed" && !req.include_suppressed {
return Ok(Some("status:suppressed".to_string()));
}
if claim.status != "active" && claim.status != "suppressed" {
return Ok(Some(format!("status:{}", claim.status)));
}
if !req.include_sensitive
&& matches!(
claim.sensitivity.as_str(),
"personal" | "sensitive" | "restricted"
)
{
return Ok(Some(format!("sensitivity:{}", claim.sensitivity)));
}
if claim
.valid_from_epoch
.is_some_and(|valid_from| valid_from > now)
{
return Ok(Some("not_yet_valid".to_string()));
}
if claim.valid_to_epoch.is_some_and(|valid_to| valid_to <= now) {
return Ok(Some("expired".to_string()));
}
if !req.include_suppressed
&& crate::memory::suppression::user_claim_is_policy_suppressed(conn, claim.id)?
{
return Ok(Some("policy_suppressed".to_string()));
}
Ok(None)
}
struct SessionCandidate {
id: i64,
request: String,
completed: String,
decisions: String,
learned: String,
next_steps: String,
preferences: String,
created_at_epoch: i64,
}
impl SessionCandidate {
fn text(&self) -> String {
[
self.request.as_str(),
self.completed.as_str(),
self.decisions.as_str(),
self.learned.as_str(),
self.next_steps.as_str(),
self.preferences.as_str(),
]
.join(" ")
}
}