use anyhow::{anyhow, bail, Context, Result};
use rusqlite::{params, Connection, OptionalExtension};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use std::collections::HashSet;
use crate::memory::lifecycle::MemoryLifecycleOp;
use crate::memory::operation::{insert_operation_log, MemoryOperationInput, MemoryOperationPlan};
use super::audit::load_memory_audit_rows;
use super::mutate::{insert_scope_cleanup_event, load_target, ObjectMutation};
use super::preference_cluster::preference_clusters;
use super::ObjectRef;
pub const CLEANUP_PLANNER_VERSION: &str = "memory-cleanup-v1";
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct MemoryCleanupPlan {
pub project: String,
pub created_at_epoch: i64,
pub planner_version: String,
pub groups: Vec<MemoryCleanupGroup>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct MemoryCleanupGroup {
pub cluster_key: String,
pub owner_scope: Option<String>,
pub owner_key: Option<String>,
pub memory_type: String,
pub state_key: Option<String>,
pub current_id: i64,
pub stale_ids: Vec<i64>,
pub reason: String,
pub confidence: f64,
pub preview: Vec<String>,
pub merged_content: Option<String>,
pub row_snapshots: Vec<MemoryCleanupRowSnapshot>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct MemoryCleanupRowSnapshot {
pub id: i64,
pub project: String,
pub scope: Option<String>,
pub source_project: Option<String>,
pub target_project: Option<String>,
pub status: String,
pub content_sha256: String,
pub updated_at_epoch: i64,
pub owner_scope: Option<String>,
pub owner_key: Option<String>,
pub memory_type: String,
pub topic_key: Option<String>,
pub state_key_id: Option<i64>,
pub state_key: Option<String>,
pub current_memory_id: Option<i64>,
}
#[derive(Debug, Clone, Serialize)]
pub struct MemoryCleanupApplyResult {
pub project: String,
pub planner_version: String,
pub groups_applied: usize,
pub current_ids: Vec<i64>,
pub stale_ids: Vec<i64>,
pub operation_ids: Vec<i64>,
pub edge_count: usize,
pub affected: Vec<ObjectMutation>,
}
pub fn build_preference_cleanup_plan(
conn: &Connection,
project: &str,
) -> Result<MemoryCleanupPlan> {
let memories = load_memory_audit_rows(conn, project)?;
let clusters = preference_clusters(&memories, project);
let mut groups = Vec::with_capacity(clusters.len());
for cluster in clusters {
let current_ref = ObjectRef::parse(&cluster.canonical_ref)?;
let stale_ids = cluster
.refs
.iter()
.filter(|object_ref| *object_ref != &cluster.canonical_ref)
.map(|object_ref| ObjectRef::parse(object_ref).map(|parsed| parsed.id))
.collect::<Result<Vec<_>>>()?;
if stale_ids.is_empty() {
continue;
}
let mut ids = Vec::with_capacity(stale_ids.len() + 1);
ids.push(current_ref.id);
ids.extend(stale_ids.iter().copied());
let row_snapshots = load_row_snapshots(conn, &ids)?;
let current = snapshot_for(&row_snapshots, current_ref.id)?;
let preview = row_snapshots
.iter()
.take(4)
.map(|row| format!("memory:{} {}", row.id, row.status))
.collect();
groups.push(MemoryCleanupGroup {
cluster_key: cluster.cluster_key,
owner_scope: current.owner_scope.clone(),
owner_key: current.owner_key.clone(),
memory_type: current.memory_type.clone(),
state_key: current.state_key.clone(),
current_id: current_ref.id,
stale_ids,
reason: cluster.reason,
confidence: 1.0,
preview,
merged_content: cluster.merged_content,
row_snapshots,
});
}
Ok(MemoryCleanupPlan {
project: project.to_string(),
created_at_epoch: chrono::Utc::now().timestamp(),
planner_version: CLEANUP_PLANNER_VERSION.to_string(),
groups,
})
}
pub fn apply_memory_cleanup_plan(
conn: &Connection,
plan: &MemoryCleanupPlan,
) -> Result<MemoryCleanupApplyResult> {
if plan.planner_version != CLEANUP_PLANNER_VERSION {
bail!(
"unsupported cleanup planner version: {}",
plan.planner_version
);
}
let tx = rusqlite::Transaction::new_unchecked(conn, rusqlite::TransactionBehavior::Immediate)?;
validate_plan_shape(plan)?;
let now = chrono::Utc::now().timestamp();
let mut affected = Vec::new();
let mut current_ids = Vec::new();
let mut stale_ids = Vec::new();
let mut operation_ids = Vec::new();
let mut edge_count = 0usize;
for group in &plan.groups {
validate_group_shape(plan, group)?;
let group_json = serde_json::to_string(group)?;
let payload_sha256 = crate::memory::activation::payload_sha256(&[
&plan.project,
&plan.planner_version,
&plan.created_at_epoch.to_string(),
&group_json,
]);
let activation_id =
crate::memory::activation::activation_id_from_key("scope-cleanup", &payload_sha256);
let provenance_ref = format!(
"{}:{}:{}",
plan.planner_version, plan.created_at_epoch, group.cluster_key
);
if let Some(replayed) = crate::memory::activation::replay_scope_cleanup_if_present(
&tx,
&activation_id,
&payload_sha256,
&provenance_ref,
&group.stale_ids,
)? {
let applied = super::receipt::load(&tx, &activation_id, replayed.memory_id)?;
current_ids.push(applied.current_id);
stale_ids.extend(applied.stale_ids);
operation_ids.push(applied.operation_id);
edge_count += applied.edge_count;
affected.extend(applied.affected);
continue;
}
let current_snapshot = snapshot_for(&group.row_snapshots, group.current_id)?;
let (branch, source_trust_class): (Option<String>, String) = tx.query_row(
"SELECT branch, source_trust_class FROM memories WHERE id = ?1",
[group.current_id],
|row| Ok((row.get(0)?, row.get(1)?)),
)?;
let source_trust = crate::memory::poisoning::SourceTrustClass::parse(&source_trust_class)
.context("cleanup current memory has unknown source trust class")?;
let scope = current_snapshot
.scope
.clone()
.unwrap_or_else(|| "project".to_string());
let owner_scope = current_snapshot.owner_scope.clone().unwrap_or_else(|| {
if scope == "global" {
"user".to_string()
} else {
"repo".to_string()
}
});
let owner_key = current_snapshot.owner_key.clone().unwrap_or_else(|| {
if scope == "global" {
"user:default".to_string()
} else {
current_snapshot.project.clone()
}
});
let source_project = current_snapshot
.source_project
.clone()
.unwrap_or_else(|| current_snapshot.project.clone());
let target_project = if owner_scope == "repo" {
Some(
current_snapshot
.target_project
.clone()
.unwrap_or_else(|| current_snapshot.project.clone()),
)
} else {
current_snapshot.target_project.clone()
};
let mut expected_memory =
crate::memory::activation::ExpectedActiveMemory::from_existing(&tx, group.current_id)?;
let reviewed_title = expected_memory.title.clone();
let reviewed_content = expected_memory.content.clone();
let preserves_current_provenance = group.merged_content.as_deref().is_none_or(|content| {
crate::memory::preference::reinforcement::cleanup_preserves_candidate_provenance(
&expected_memory.content,
content,
)
});
let expected_memory = match group.merged_content.as_deref() {
Some(content) => {
if !preserves_current_provenance {
expected_memory.source_candidate_id = None;
expected_memory.evidence_event_ids = None;
}
expected_memory.with_content(content)
}
None => expected_memory,
};
let reviewed_payload_unchanged =
expected_memory.title == reviewed_title && expected_memory.content == reviewed_content;
let poisoning_verdict = cleanup_poisoning_verdict(
&tx,
group.current_id,
&expected_memory,
reviewed_payload_unchanged,
)?;
let request = crate::memory::activation::ActiveMemoryWriteRequest {
activation_id: activation_id.clone(),
route_kind: crate::memory::activation::ActivationRouteKind::ScopeCleanup,
actor_kind: crate::memory::activation::ActivationActorKind::Operator,
source_operation: "memory_cleanup".to_string(),
source_trust,
result_source_trust: if preserves_current_provenance {
source_trust
} else {
crate::memory::poisoning::SourceTrustClass::ExternalContent
},
source_project,
route: crate::memory::activation::ActiveMemoryRoute {
project: current_snapshot.project.clone(),
branch,
scope,
owner_scope,
owner_key,
target_project,
},
provenance_kind: crate::memory::activation::ActivationProvenanceKind::ScopePlan,
provenance_ref,
payload_sha256,
expected_memory,
poisoning_verdict,
superseded_ids: group.stale_ids.clone(),
};
let mut group_result = None;
let activation_result = crate::memory::activation::execute_one(&tx, &request, |_permit| {
validate_group(&tx, plan, group)?;
let applied = apply_cleanup_group(&tx, plan, group, now, preserves_current_provenance)?;
group_result = Some(applied);
Ok(group.current_id)
})?;
let applied = if activation_result.replayed {
super::receipt::load(&tx, &activation_id, activation_result.memory_id)?
} else {
let applied = group_result.context("cleanup activation produced no group result")?;
super::receipt::insert(&tx, &activation_id, &applied)?;
let bound = tx
.execute(
"UPDATE memory_operation_log SET activation_id = ?1 WHERE id = ?2",
params![activation_id, applied.operation_id],
)
.context("bind cleanup operation log to activation")?;
if bound != 1 {
bail!(
"failed to bind cleanup operation {} to activation",
applied.operation_id
);
}
applied
};
current_ids.push(applied.current_id);
stale_ids.extend(applied.stale_ids);
operation_ids.push(applied.operation_id);
edge_count += applied.edge_count;
affected.extend(applied.affected);
}
tx.commit()?;
Ok(MemoryCleanupApplyResult {
project: plan.project.clone(),
planner_version: plan.planner_version.clone(),
groups_applied: plan.groups.len(),
current_ids,
stale_ids,
operation_ids,
edge_count,
affected,
})
}
fn cleanup_poisoning_verdict(
conn: &Connection,
memory_id: i64,
expected: &crate::memory::activation::ExpectedActiveMemory,
reviewed_payload_unchanged: bool,
) -> Result<crate::memory::activation::ActivationPoisoningVerdict> {
let acknowledgement: (Option<String>, Option<i64>, Option<i64>) = conn.query_row(
"SELECT acknowledged_pattern_id, acknowledged_pattern_version, acknowledged_at_epoch
FROM memories WHERE id = ?1",
[memory_id],
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
)?;
let acknowledgement_absent =
acknowledgement.0.is_none() && acknowledgement.1.is_none() && acknowledgement.2.is_none();
let acknowledgement_complete = acknowledgement
.0
.as_deref()
.is_some_and(|pattern_id| !pattern_id.is_empty())
&& acknowledgement.1.is_some_and(|version| version > 0)
&& acknowledgement.2.is_some_and(|epoch| epoch > 0);
if !acknowledgement_absent && !acknowledgement_complete {
bail!("cleanup current memory has incomplete acknowledgement evidence");
}
let Some(matched) = crate::memory::poisoning::scan_instruction_pattern(&format!(
"{}\n{}",
expected.title, expected.content
)) else {
return Ok(crate::memory::activation::ActivationPoisoningVerdict::UpstreamValidated);
};
if reviewed_payload_unchanged
&& acknowledgement.0.as_deref() == Some(matched.pattern_id)
&& acknowledgement.1 == Some(matched.pattern_set_version)
&& acknowledgement.2.is_some_and(|epoch| epoch > 0)
{
Ok(crate::memory::activation::ActivationPoisoningVerdict::Acknowledged)
} else {
Ok(crate::memory::activation::ActivationPoisoningVerdict::UpstreamValidated)
}
}
#[derive(Debug, Serialize, Deserialize)]
pub(super) struct CleanupGroupApplyResult {
pub(super) current_id: i64,
pub(super) stale_ids: Vec<i64>,
pub(super) operation_id: i64,
pub(super) edge_count: usize,
pub(super) affected: Vec<ObjectMutation>,
}
fn apply_cleanup_group(
conn: &Connection,
plan: &MemoryCleanupPlan,
group: &MemoryCleanupGroup,
now: i64,
preserves_current_provenance: bool,
) -> Result<CleanupGroupApplyResult> {
let current_ref = ObjectRef::memory(group.current_id);
let canonical = load_target(conn, current_ref)?;
let current_snapshot = snapshot_for(&group.row_snapshots, group.current_id)?;
let merged = group.merged_content.as_deref();
let final_text = if let Some(merged) = merged {
merged.to_string()
} else {
conn.query_row(
"SELECT content FROM memories WHERE id = ?1",
[group.current_id],
|row| row.get::<_, String>(0),
)?
};
let affected_ids = std::iter::once(group.current_id)
.chain(group.stale_ids.iter().copied())
.collect::<Vec<_>>();
crate::memory::preference::compilation::enqueue_for_memory_ids(conn, &affected_ids)?;
crate::memory::preference::reinforcement::reconcile_cleanup_preference(
conn,
group.current_id,
&group.stale_ids,
&final_text,
now,
)?;
let updated = conn.execute(
"UPDATE memories
SET content = COALESCE(?1, content), status = 'active', updated_at_epoch = ?2
WHERE id = ?3",
params![merged, now, group.current_id],
)?;
if updated != 1 {
bail!(
"failed to update cleanup current memory {}",
group.current_id
);
}
if !preserves_current_provenance {
conn.execute(
"UPDATE memories
SET evidence_event_ids = NULL, source_candidate_id = NULL,
confidence = NULL, valid_from_epoch = NULL,
source_trust_class = 'external_content'
WHERE id = ?1",
[group.current_id],
)?;
}
let mut affected = vec![ObjectMutation {
object_ref: current_ref.to_string(),
title: canonical.title.clone(),
previous_status: canonical.status.clone(),
new_status: "active".to_string(),
previous_owner: canonical.owner.clone(),
new_owner: canonical.owner.clone(),
}];
insert_scope_cleanup_event(
conn,
"memory-cleanup",
&canonical,
"active",
&canonical.owner,
Some(group.reason.as_str()),
now,
)?;
if let Some(state_key_id) = current_snapshot.state_key_id {
conn.execute(
"UPDATE memory_state_keys SET current_memory_id = ?1, updated_at_epoch = ?2 WHERE id = ?3",
params![group.current_id, now, state_key_id],
)?;
}
for stale_id in &group.stale_ids {
let stale_ref = ObjectRef::memory(*stale_id);
let target = load_target(conn, stale_ref)?;
let updated = conn.execute(
"UPDATE memories SET status = 'stale', updated_at_epoch = ?1 WHERE id = ?2",
params![now, stale_id],
)?;
if updated != 1 {
bail!("failed to stale cleanup memory {stale_id}");
}
affected.push(ObjectMutation {
object_ref: stale_ref.to_string(),
title: target.title.clone(),
previous_status: target.status.clone(),
new_status: "stale".to_string(),
previous_owner: target.owner.clone(),
new_owner: target.owner.clone(),
});
insert_scope_cleanup_event(
conn,
"memory-cleanup",
&target,
"stale",
&target.owner,
Some("duplicate preference superseded by cleanup plan"),
now,
)?;
}
let operation_id = insert_cleanup_operation_log(conn, plan, group)?;
let edge_count = crate::memory::edge::insert_replacement_edges(
conn,
crate::memory::edge::MemoryEdgeType::Duplicates,
&group.stale_ids,
group.current_id,
crate::memory::edge::MemoryEdgeWriteContext {
state_key_id: current_snapshot.state_key_id,
source_operation_id: Some(operation_id),
confidence: Some(group.confidence),
reason: Some(group.reason.as_str()),
..Default::default()
},
)?;
Ok(CleanupGroupApplyResult {
current_id: group.current_id,
stale_ids: group.stale_ids.clone(),
operation_id,
edge_count,
affected,
})
}
fn validate_plan_shape(plan: &MemoryCleanupPlan) -> Result<()> {
let mut ids = HashSet::new();
for group in &plan.groups {
for id in std::iter::once(group.current_id).chain(group.stale_ids.iter().copied()) {
if !ids.insert(id) {
bail!("cleanup plan lists memory:{id} in more than one action");
}
}
}
Ok(())
}
fn validate_group(
conn: &Connection,
plan: &MemoryCleanupPlan,
group: &MemoryCleanupGroup,
) -> Result<()> {
validate_group_shape(plan, group)?;
for snapshot in &group.row_snapshots {
let current = load_row_snapshot(conn, snapshot.id)?
.ok_or_else(|| anyhow!("cleanup plan row {} no longer exists", snapshot.id))?;
if ¤t != snapshot {
bail!(
"cleanup plan row {} changed since dry-run; refresh the plan before applying",
snapshot.id
);
}
}
Ok(())
}
fn validate_group_shape(plan: &MemoryCleanupPlan, group: &MemoryCleanupGroup) -> Result<()> {
let mut canonical_stale_ids = group.stale_ids.clone();
canonical_stale_ids.sort_unstable();
canonical_stale_ids.dedup();
if canonical_stale_ids != group.stale_ids {
bail!(
"cleanup group {} stale ids must be sorted unique positive integers",
group.cluster_key
);
}
if group.stale_ids.iter().any(|id| *id <= 0) {
bail!(
"cleanup group {} stale ids must be sorted unique positive integers",
group.cluster_key
);
}
if group.stale_ids.contains(&group.current_id) {
bail!(
"cleanup group {} lists current id {} as stale",
group.cluster_key,
group.current_id
);
}
if group.memory_type != "preference" {
bail!(
"unsupported cleanup group memory type {}",
group.memory_type
);
}
let mut expected_ids = group.stale_ids.clone();
expected_ids.push(group.current_id);
expected_ids.sort_unstable();
expected_ids.dedup();
let mut snapshot_ids = group
.row_snapshots
.iter()
.map(|snapshot| snapshot.id)
.collect::<Vec<_>>();
snapshot_ids.sort_unstable();
snapshot_ids.dedup();
if snapshot_ids != expected_ids {
bail!(
"cleanup group {} row snapshots do not match current/stale ids",
group.cluster_key
);
}
let current_snapshot = snapshot_for(&group.row_snapshots, group.current_id)?;
if group.owner_scope != current_snapshot.owner_scope
|| group.owner_key != current_snapshot.owner_key
{
bail!(
"cleanup group {} owner does not match current row owner",
group.cluster_key
);
}
if group.state_key != current_snapshot.state_key {
bail!(
"cleanup group {} state key does not match current row",
group.cluster_key
);
}
let current_owner = current_snapshot.owner_namespace(&plan.project);
let current_state_key_id = current_snapshot.state_key_id;
let current_state_key = current_snapshot.state_key.as_deref();
let topic_group = group.cluster_key.starts_with("topic:");
for snapshot in &group.row_snapshots {
if snapshot.status != "active" {
bail!("cleanup plan row {} is no longer active", snapshot.id);
}
if snapshot.memory_type != group.memory_type {
bail!(
"cleanup plan row {} type {} does not match group type {}",
snapshot.id,
snapshot.memory_type,
group.memory_type
);
}
if !snapshot.belongs_to_project(&plan.project) {
bail!(
"cleanup plan row {} does not belong to project {}",
snapshot.id,
plan.project
);
}
if snapshot.owner_namespace(&plan.project) != current_owner {
bail!(
"cleanup plan row {} owner does not match current row owner",
snapshot.id
);
}
match (current_state_key_id, current_state_key) {
(Some(state_key_id), _) if snapshot.state_key_id != Some(state_key_id) => {
bail!(
"cleanup plan row {} state key does not match current row",
snapshot.id
);
}
(None, Some(state_key)) if snapshot.state_key.as_deref() != Some(state_key) => {
bail!(
"cleanup plan row {} state key does not match current row",
snapshot.id
);
}
_ => {}
}
if topic_group && snapshot.topic_key != current_snapshot.topic_key {
bail!(
"cleanup plan row {} topic key does not match current row",
snapshot.id
);
}
}
Ok(())
}
fn insert_cleanup_operation_log(
conn: &Connection,
plan: &MemoryCleanupPlan,
group: &MemoryCleanupGroup,
) -> Result<i64> {
let current = snapshot_for(&group.row_snapshots, group.current_id)?;
let mut operation_plan = MemoryOperationPlan::new(
MemoryLifecycleOp::Update,
group.state_key.clone(),
group.reason.clone(),
)
.with_target_memory_id(Some(group.current_id))
.with_superseded_ids(group.stale_ids.clone());
operation_plan.planner_version = CLEANUP_PLANNER_VERSION;
let input = MemoryOperationInput {
source: "memory_cleanup".to_string(),
actor: "memory_cleanup".to_string(),
source_project: plan.project.clone(),
owner_scope: group
.owner_scope
.clone()
.unwrap_or_else(|| "repo".to_string()),
owner_key: group
.owner_key
.clone()
.unwrap_or_else(|| plan.project.clone()),
memory_type: group.memory_type.clone(),
topic_key: current.topic_key.clone(),
state_key: group.state_key.clone(),
source_candidate_id: None,
confidence: Some(group.confidence),
};
insert_operation_log(conn, &input, &operation_plan, Some(group.current_id))
}
fn load_row_snapshots(conn: &Connection, ids: &[i64]) -> Result<Vec<MemoryCleanupRowSnapshot>> {
ids.iter()
.copied()
.map(|id| {
load_row_snapshot(conn, id)?
.ok_or_else(|| anyhow!("cleanup plan target memory:{id} not found"))
})
.collect()
}
fn load_row_snapshot(conn: &Connection, id: i64) -> Result<Option<MemoryCleanupRowSnapshot>> {
conn.query_row(
"SELECT m.id, m.status, m.content, m.updated_at_epoch, m.owner_scope,
m.owner_key, m.memory_type, m.topic_key, m.state_key_id, sk.state_key,
sk.current_memory_id, m.project, m.scope, m.source_project, m.target_project
FROM memories m
LEFT JOIN memory_state_keys sk ON sk.id = m.state_key_id
WHERE m.id = ?1",
params![id],
|row| {
let content: String = row.get(2)?;
Ok(MemoryCleanupRowSnapshot {
id: row.get(0)?,
status: row.get(1)?,
content_sha256: content_sha256(&content),
updated_at_epoch: row.get(3)?,
owner_scope: row.get(4)?,
owner_key: row.get(5)?,
memory_type: row.get(6)?,
topic_key: row.get(7)?,
state_key_id: row.get(8)?,
state_key: row.get(9)?,
current_memory_id: row.get(10)?,
project: row.get(11)?,
scope: row.get(12)?,
source_project: row.get(13)?,
target_project: row.get(14)?,
})
},
)
.optional()
.with_context(|| format!("load cleanup plan row snapshot for memory:{id}"))
}
impl MemoryCleanupRowSnapshot {
fn owner_namespace(&self, project: &str) -> (String, String) {
match (self.owner_scope.as_deref(), self.owner_key.as_deref()) {
(Some(scope), Some(key)) => (scope.to_string(), key.to_string()),
_ if self.project == project
&& self.scope.as_deref().unwrap_or("project") != "global" =>
{
("legacy_repo".to_string(), project.to_string())
}
_ => ("legacy_other".to_string(), self.project.clone()),
}
}
fn belongs_to_project(&self, project: &str) -> bool {
self.source_project.as_deref() == Some(project)
|| self.target_project.as_deref() == Some(project)
|| (self.owner_scope.as_deref() == Some("repo")
&& self.owner_key.as_deref() == Some(project))
|| (self.owner_scope.is_none()
&& self.project == project
&& self.scope.as_deref().unwrap_or("project") != "global")
}
}
fn snapshot_for(
snapshots: &[MemoryCleanupRowSnapshot],
id: i64,
) -> Result<&MemoryCleanupRowSnapshot> {
snapshots
.iter()
.find(|snapshot| snapshot.id == id)
.ok_or_else(|| anyhow!("cleanup plan missing snapshot for memory:{id}"))
}
fn content_sha256(content: &str) -> String {
let mut hasher = Sha256::new();
hasher.update(content.as_bytes());
format!("{:x}", hasher.finalize())
}