1use anyhow::{anyhow, bail, Context, Result};
2use rusqlite::{params, Connection, OptionalExtension};
3use serde::{Deserialize, Serialize};
4use sha2::{Digest, Sha256};
5use std::collections::HashSet;
6
7use crate::memory::lifecycle::MemoryLifecycleOp;
8use crate::memory::operation::{insert_operation_log, MemoryOperationInput, MemoryOperationPlan};
9
10use super::audit::load_memory_audit_rows;
11use super::mutate::{insert_scope_cleanup_event, load_target, ObjectMutation};
12use super::preference_cluster::preference_clusters;
13use super::ObjectRef;
14
15pub const CLEANUP_PLANNER_VERSION: &str = "memory-cleanup-v1";
16
17#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
18pub struct MemoryCleanupPlan {
19 pub project: String,
20 pub created_at_epoch: i64,
21 pub planner_version: String,
22 pub groups: Vec<MemoryCleanupGroup>,
23}
24
25#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
26pub struct MemoryCleanupGroup {
27 pub cluster_key: String,
28 pub owner_scope: Option<String>,
29 pub owner_key: Option<String>,
30 pub memory_type: String,
31 pub state_key: Option<String>,
32 pub current_id: i64,
33 pub stale_ids: Vec<i64>,
34 pub reason: String,
35 pub confidence: f64,
36 pub preview: Vec<String>,
37 pub merged_content: Option<String>,
38 pub row_snapshots: Vec<MemoryCleanupRowSnapshot>,
39}
40
41#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
42pub struct MemoryCleanupRowSnapshot {
43 pub id: i64,
44 pub project: String,
45 pub scope: Option<String>,
46 pub source_project: Option<String>,
47 pub target_project: Option<String>,
48 pub status: String,
49 pub content_sha256: String,
50 pub updated_at_epoch: i64,
51 pub owner_scope: Option<String>,
52 pub owner_key: Option<String>,
53 pub memory_type: String,
54 pub topic_key: Option<String>,
55 pub state_key_id: Option<i64>,
56 pub state_key: Option<String>,
57 pub current_memory_id: Option<i64>,
58}
59
60#[derive(Debug, Clone, Serialize)]
61pub struct MemoryCleanupApplyResult {
62 pub project: String,
63 pub planner_version: String,
64 pub groups_applied: usize,
65 pub current_ids: Vec<i64>,
66 pub stale_ids: Vec<i64>,
67 pub operation_ids: Vec<i64>,
68 pub edge_count: usize,
69 pub affected: Vec<ObjectMutation>,
70}
71
72pub fn build_preference_cleanup_plan(
73 conn: &Connection,
74 project: &str,
75) -> Result<MemoryCleanupPlan> {
76 let memories = load_memory_audit_rows(conn, project)?;
77 let clusters = preference_clusters(&memories, project);
78 let mut groups = Vec::with_capacity(clusters.len());
79
80 for cluster in clusters {
81 let current_ref = ObjectRef::parse(&cluster.canonical_ref)?;
82 let stale_ids = cluster
83 .refs
84 .iter()
85 .filter(|object_ref| *object_ref != &cluster.canonical_ref)
86 .map(|object_ref| ObjectRef::parse(object_ref).map(|parsed| parsed.id))
87 .collect::<Result<Vec<_>>>()?;
88 if stale_ids.is_empty() {
89 continue;
90 }
91 let mut ids = Vec::with_capacity(stale_ids.len() + 1);
92 ids.push(current_ref.id);
93 ids.extend(stale_ids.iter().copied());
94 let row_snapshots = load_row_snapshots(conn, &ids)?;
95 let current = snapshot_for(&row_snapshots, current_ref.id)?;
96 let preview = row_snapshots
97 .iter()
98 .take(4)
99 .map(|row| format!("memory:{} {}", row.id, row.status))
100 .collect();
101 groups.push(MemoryCleanupGroup {
102 cluster_key: cluster.cluster_key,
103 owner_scope: current.owner_scope.clone(),
104 owner_key: current.owner_key.clone(),
105 memory_type: current.memory_type.clone(),
106 state_key: current.state_key.clone(),
107 current_id: current_ref.id,
108 stale_ids,
109 reason: cluster.reason,
110 confidence: 1.0,
111 preview,
112 merged_content: cluster.merged_content,
113 row_snapshots,
114 });
115 }
116
117 Ok(MemoryCleanupPlan {
118 project: project.to_string(),
119 created_at_epoch: chrono::Utc::now().timestamp(),
120 planner_version: CLEANUP_PLANNER_VERSION.to_string(),
121 groups,
122 })
123}
124
125pub fn apply_memory_cleanup_plan(
126 conn: &Connection,
127 plan: &MemoryCleanupPlan,
128) -> Result<MemoryCleanupApplyResult> {
129 if plan.planner_version != CLEANUP_PLANNER_VERSION {
130 bail!(
131 "unsupported cleanup planner version: {}",
132 plan.planner_version
133 );
134 }
135
136 let tx = rusqlite::Transaction::new_unchecked(conn, rusqlite::TransactionBehavior::Immediate)?;
137 validate_plan_shape(plan)?;
138
139 let now = chrono::Utc::now().timestamp();
140 let mut affected = Vec::new();
141 let mut current_ids = Vec::new();
142 let mut stale_ids = Vec::new();
143 let mut operation_ids = Vec::new();
144 let mut edge_count = 0usize;
145
146 for group in &plan.groups {
147 validate_group_shape(plan, group)?;
148 let group_json = serde_json::to_string(group)?;
149 let payload_sha256 = crate::memory::activation::payload_sha256(&[
150 &plan.project,
151 &plan.planner_version,
152 &plan.created_at_epoch.to_string(),
153 &group_json,
154 ]);
155 let activation_id =
156 crate::memory::activation::activation_id_from_key("scope-cleanup", &payload_sha256);
157 let provenance_ref = format!(
158 "{}:{}:{}",
159 plan.planner_version, plan.created_at_epoch, group.cluster_key
160 );
161 if let Some(replayed) = crate::memory::activation::replay_scope_cleanup_if_present(
162 &tx,
163 &activation_id,
164 &payload_sha256,
165 &provenance_ref,
166 &group.stale_ids,
167 )? {
168 let applied = super::receipt::load(&tx, &activation_id, replayed.memory_id)?;
169 current_ids.push(applied.current_id);
170 stale_ids.extend(applied.stale_ids);
171 operation_ids.push(applied.operation_id);
172 edge_count += applied.edge_count;
173 affected.extend(applied.affected);
174 continue;
175 }
176 let current_snapshot = snapshot_for(&group.row_snapshots, group.current_id)?;
177 let (branch, source_trust_class): (Option<String>, String) = tx.query_row(
178 "SELECT branch, source_trust_class FROM memories WHERE id = ?1",
179 [group.current_id],
180 |row| Ok((row.get(0)?, row.get(1)?)),
181 )?;
182 let source_trust = crate::memory::poisoning::SourceTrustClass::parse(&source_trust_class)
183 .context("cleanup current memory has unknown source trust class")?;
184 let scope = current_snapshot
185 .scope
186 .clone()
187 .unwrap_or_else(|| "project".to_string());
188 let owner_scope = current_snapshot.owner_scope.clone().unwrap_or_else(|| {
189 if scope == "global" {
190 "user".to_string()
191 } else {
192 "repo".to_string()
193 }
194 });
195 let owner_key = current_snapshot.owner_key.clone().unwrap_or_else(|| {
196 if scope == "global" {
197 "user:default".to_string()
198 } else {
199 current_snapshot.project.clone()
200 }
201 });
202 let source_project = current_snapshot
203 .source_project
204 .clone()
205 .unwrap_or_else(|| current_snapshot.project.clone());
206 let target_project = if owner_scope == "repo" {
207 Some(
208 current_snapshot
209 .target_project
210 .clone()
211 .unwrap_or_else(|| current_snapshot.project.clone()),
212 )
213 } else {
214 current_snapshot.target_project.clone()
215 };
216 let mut expected_memory =
217 crate::memory::activation::ExpectedActiveMemory::from_existing(&tx, group.current_id)?;
218 let reviewed_title = expected_memory.title.clone();
219 let reviewed_content = expected_memory.content.clone();
220 let preserves_current_provenance = group.merged_content.as_deref().is_none_or(|content| {
221 crate::memory::preference::reinforcement::cleanup_preserves_candidate_provenance(
222 &expected_memory.content,
223 content,
224 )
225 });
226 let expected_memory = match group.merged_content.as_deref() {
227 Some(content) => {
228 if !preserves_current_provenance {
229 expected_memory.source_candidate_id = None;
230 expected_memory.evidence_event_ids = None;
231 }
232 expected_memory.with_content(content)
233 }
234 None => expected_memory,
235 };
236 let reviewed_payload_unchanged =
237 expected_memory.title == reviewed_title && expected_memory.content == reviewed_content;
238 let poisoning_verdict = cleanup_poisoning_verdict(
239 &tx,
240 group.current_id,
241 &expected_memory,
242 reviewed_payload_unchanged,
243 )?;
244 let request = crate::memory::activation::ActiveMemoryWriteRequest {
245 activation_id: activation_id.clone(),
246 route_kind: crate::memory::activation::ActivationRouteKind::ScopeCleanup,
247 actor_kind: crate::memory::activation::ActivationActorKind::Operator,
248 source_operation: "memory_cleanup".to_string(),
249 source_trust,
250 result_source_trust: if preserves_current_provenance {
251 source_trust
252 } else {
253 crate::memory::poisoning::SourceTrustClass::ExternalContent
254 },
255 source_project,
256 route: crate::memory::activation::ActiveMemoryRoute {
257 project: current_snapshot.project.clone(),
258 branch,
259 scope,
260 owner_scope,
261 owner_key,
262 target_project,
263 },
264 provenance_kind: crate::memory::activation::ActivationProvenanceKind::ScopePlan,
265 provenance_ref,
266 payload_sha256,
267 expected_memory,
268 poisoning_verdict,
269 superseded_ids: group.stale_ids.clone(),
270 };
271 let mut group_result = None;
272 let activation_result = crate::memory::activation::execute_one(&tx, &request, |_permit| {
273 validate_group(&tx, plan, group)?;
274 let applied = apply_cleanup_group(&tx, plan, group, now, preserves_current_provenance)?;
275 group_result = Some(applied);
276 Ok(group.current_id)
277 })?;
278 let applied = if activation_result.replayed {
279 super::receipt::load(&tx, &activation_id, activation_result.memory_id)?
280 } else {
281 let applied = group_result.context("cleanup activation produced no group result")?;
282 super::receipt::insert(&tx, &activation_id, &applied)?;
283 let bound = tx
284 .execute(
285 "UPDATE memory_operation_log SET activation_id = ?1 WHERE id = ?2",
286 params![activation_id, applied.operation_id],
287 )
288 .context("bind cleanup operation log to activation")?;
289 if bound != 1 {
290 bail!(
291 "failed to bind cleanup operation {} to activation",
292 applied.operation_id
293 );
294 }
295 applied
296 };
297 current_ids.push(applied.current_id);
298 stale_ids.extend(applied.stale_ids);
299 operation_ids.push(applied.operation_id);
300 edge_count += applied.edge_count;
301 affected.extend(applied.affected);
302 }
303
304 tx.commit()?;
305 Ok(MemoryCleanupApplyResult {
306 project: plan.project.clone(),
307 planner_version: plan.planner_version.clone(),
308 groups_applied: plan.groups.len(),
309 current_ids,
310 stale_ids,
311 operation_ids,
312 edge_count,
313 affected,
314 })
315}
316
317fn cleanup_poisoning_verdict(
318 conn: &Connection,
319 memory_id: i64,
320 expected: &crate::memory::activation::ExpectedActiveMemory,
321 reviewed_payload_unchanged: bool,
322) -> Result<crate::memory::activation::ActivationPoisoningVerdict> {
323 let acknowledgement: (Option<String>, Option<i64>, Option<i64>) = conn.query_row(
324 "SELECT acknowledged_pattern_id, acknowledged_pattern_version, acknowledged_at_epoch
325 FROM memories WHERE id = ?1",
326 [memory_id],
327 |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
328 )?;
329 let acknowledgement_absent =
330 acknowledgement.0.is_none() && acknowledgement.1.is_none() && acknowledgement.2.is_none();
331 let acknowledgement_complete = acknowledgement
332 .0
333 .as_deref()
334 .is_some_and(|pattern_id| !pattern_id.is_empty())
335 && acknowledgement.1.is_some_and(|version| version > 0)
336 && acknowledgement.2.is_some_and(|epoch| epoch > 0);
337 if !acknowledgement_absent && !acknowledgement_complete {
338 bail!("cleanup current memory has incomplete acknowledgement evidence");
339 }
340 let Some(matched) = crate::memory::poisoning::scan_instruction_pattern(&format!(
341 "{}\n{}",
342 expected.title, expected.content
343 )) else {
344 return Ok(crate::memory::activation::ActivationPoisoningVerdict::UpstreamValidated);
345 };
346 if reviewed_payload_unchanged
347 && acknowledgement.0.as_deref() == Some(matched.pattern_id)
348 && acknowledgement.1 == Some(matched.pattern_set_version)
349 && acknowledgement.2.is_some_and(|epoch| epoch > 0)
350 {
351 Ok(crate::memory::activation::ActivationPoisoningVerdict::Acknowledged)
352 } else {
353 Ok(crate::memory::activation::ActivationPoisoningVerdict::UpstreamValidated)
354 }
355}
356
357#[derive(Debug, Serialize, Deserialize)]
358pub(super) struct CleanupGroupApplyResult {
359 pub(super) current_id: i64,
360 pub(super) stale_ids: Vec<i64>,
361 pub(super) operation_id: i64,
362 pub(super) edge_count: usize,
363 pub(super) affected: Vec<ObjectMutation>,
364}
365
366fn apply_cleanup_group(
367 conn: &Connection,
368 plan: &MemoryCleanupPlan,
369 group: &MemoryCleanupGroup,
370 now: i64,
371 preserves_current_provenance: bool,
372) -> Result<CleanupGroupApplyResult> {
373 let current_ref = ObjectRef::memory(group.current_id);
374 let canonical = load_target(conn, current_ref)?;
375 let current_snapshot = snapshot_for(&group.row_snapshots, group.current_id)?;
376 let merged = group.merged_content.as_deref();
377 let final_text = if let Some(merged) = merged {
378 merged.to_string()
379 } else {
380 conn.query_row(
381 "SELECT content FROM memories WHERE id = ?1",
382 [group.current_id],
383 |row| row.get::<_, String>(0),
384 )?
385 };
386 let affected_ids = std::iter::once(group.current_id)
387 .chain(group.stale_ids.iter().copied())
388 .collect::<Vec<_>>();
389 crate::memory::preference::compilation::enqueue_for_memory_ids(conn, &affected_ids)?;
390 crate::memory::preference::reinforcement::reconcile_cleanup_preference(
391 conn,
392 group.current_id,
393 &group.stale_ids,
394 &final_text,
395 now,
396 )?;
397 let updated = conn.execute(
398 "UPDATE memories
399 SET content = COALESCE(?1, content), status = 'active', updated_at_epoch = ?2
400 WHERE id = ?3",
401 params![merged, now, group.current_id],
402 )?;
403 if updated != 1 {
404 bail!(
405 "failed to update cleanup current memory {}",
406 group.current_id
407 );
408 }
409 if !preserves_current_provenance {
410 conn.execute(
411 "UPDATE memories
412 SET evidence_event_ids = NULL, source_candidate_id = NULL,
413 confidence = NULL, valid_from_epoch = NULL,
414 source_trust_class = 'external_content'
415 WHERE id = ?1",
416 [group.current_id],
417 )?;
418 }
419 let mut affected = vec![ObjectMutation {
420 object_ref: current_ref.to_string(),
421 title: canonical.title.clone(),
422 previous_status: canonical.status.clone(),
423 new_status: "active".to_string(),
424 previous_owner: canonical.owner.clone(),
425 new_owner: canonical.owner.clone(),
426 }];
427 insert_scope_cleanup_event(
428 conn,
429 "memory-cleanup",
430 &canonical,
431 "active",
432 &canonical.owner,
433 Some(group.reason.as_str()),
434 now,
435 )?;
436 if let Some(state_key_id) = current_snapshot.state_key_id {
437 conn.execute(
438 "UPDATE memory_state_keys SET current_memory_id = ?1, updated_at_epoch = ?2 WHERE id = ?3",
439 params![group.current_id, now, state_key_id],
440 )?;
441 }
442 for stale_id in &group.stale_ids {
443 let stale_ref = ObjectRef::memory(*stale_id);
444 let target = load_target(conn, stale_ref)?;
445 let updated = conn.execute(
446 "UPDATE memories SET status = 'stale', updated_at_epoch = ?1 WHERE id = ?2",
447 params![now, stale_id],
448 )?;
449 if updated != 1 {
450 bail!("failed to stale cleanup memory {stale_id}");
451 }
452 affected.push(ObjectMutation {
453 object_ref: stale_ref.to_string(),
454 title: target.title.clone(),
455 previous_status: target.status.clone(),
456 new_status: "stale".to_string(),
457 previous_owner: target.owner.clone(),
458 new_owner: target.owner.clone(),
459 });
460 insert_scope_cleanup_event(
461 conn,
462 "memory-cleanup",
463 &target,
464 "stale",
465 &target.owner,
466 Some("duplicate preference superseded by cleanup plan"),
467 now,
468 )?;
469 }
470 let operation_id = insert_cleanup_operation_log(conn, plan, group)?;
471 let edge_count = crate::memory::edge::insert_replacement_edges(
472 conn,
473 crate::memory::edge::MemoryEdgeType::Duplicates,
474 &group.stale_ids,
475 group.current_id,
476 crate::memory::edge::MemoryEdgeWriteContext {
477 state_key_id: current_snapshot.state_key_id,
478 source_operation_id: Some(operation_id),
479 confidence: Some(group.confidence),
480 reason: Some(group.reason.as_str()),
481 ..Default::default()
482 },
483 )?;
484 Ok(CleanupGroupApplyResult {
485 current_id: group.current_id,
486 stale_ids: group.stale_ids.clone(),
487 operation_id,
488 edge_count,
489 affected,
490 })
491}
492
493fn validate_plan_shape(plan: &MemoryCleanupPlan) -> Result<()> {
494 let mut ids = HashSet::new();
495 for group in &plan.groups {
496 for id in std::iter::once(group.current_id).chain(group.stale_ids.iter().copied()) {
497 if !ids.insert(id) {
498 bail!("cleanup plan lists memory:{id} in more than one action");
499 }
500 }
501 }
502 Ok(())
503}
504
505fn validate_group(
506 conn: &Connection,
507 plan: &MemoryCleanupPlan,
508 group: &MemoryCleanupGroup,
509) -> Result<()> {
510 validate_group_shape(plan, group)?;
511 for snapshot in &group.row_snapshots {
512 let current = load_row_snapshot(conn, snapshot.id)?
513 .ok_or_else(|| anyhow!("cleanup plan row {} no longer exists", snapshot.id))?;
514 if ¤t != snapshot {
515 bail!(
516 "cleanup plan row {} changed since dry-run; refresh the plan before applying",
517 snapshot.id
518 );
519 }
520 }
521 Ok(())
522}
523
524fn validate_group_shape(plan: &MemoryCleanupPlan, group: &MemoryCleanupGroup) -> Result<()> {
525 let mut canonical_stale_ids = group.stale_ids.clone();
526 canonical_stale_ids.sort_unstable();
527 canonical_stale_ids.dedup();
528 if canonical_stale_ids != group.stale_ids {
529 bail!(
530 "cleanup group {} stale ids must be sorted unique positive integers",
531 group.cluster_key
532 );
533 }
534 if group.stale_ids.iter().any(|id| *id <= 0) {
535 bail!(
536 "cleanup group {} stale ids must be sorted unique positive integers",
537 group.cluster_key
538 );
539 }
540 if group.stale_ids.contains(&group.current_id) {
541 bail!(
542 "cleanup group {} lists current id {} as stale",
543 group.cluster_key,
544 group.current_id
545 );
546 }
547 if group.memory_type != "preference" {
548 bail!(
549 "unsupported cleanup group memory type {}",
550 group.memory_type
551 );
552 }
553 let mut expected_ids = group.stale_ids.clone();
554 expected_ids.push(group.current_id);
555 expected_ids.sort_unstable();
556 expected_ids.dedup();
557 let mut snapshot_ids = group
558 .row_snapshots
559 .iter()
560 .map(|snapshot| snapshot.id)
561 .collect::<Vec<_>>();
562 snapshot_ids.sort_unstable();
563 snapshot_ids.dedup();
564 if snapshot_ids != expected_ids {
565 bail!(
566 "cleanup group {} row snapshots do not match current/stale ids",
567 group.cluster_key
568 );
569 }
570
571 let current_snapshot = snapshot_for(&group.row_snapshots, group.current_id)?;
572 if group.owner_scope != current_snapshot.owner_scope
573 || group.owner_key != current_snapshot.owner_key
574 {
575 bail!(
576 "cleanup group {} owner does not match current row owner",
577 group.cluster_key
578 );
579 }
580 if group.state_key != current_snapshot.state_key {
581 bail!(
582 "cleanup group {} state key does not match current row",
583 group.cluster_key
584 );
585 }
586 let current_owner = current_snapshot.owner_namespace(&plan.project);
587 let current_state_key_id = current_snapshot.state_key_id;
588 let current_state_key = current_snapshot.state_key.as_deref();
589 let topic_group = group.cluster_key.starts_with("topic:");
590
591 for snapshot in &group.row_snapshots {
592 if snapshot.status != "active" {
593 bail!("cleanup plan row {} is no longer active", snapshot.id);
594 }
595 if snapshot.memory_type != group.memory_type {
596 bail!(
597 "cleanup plan row {} type {} does not match group type {}",
598 snapshot.id,
599 snapshot.memory_type,
600 group.memory_type
601 );
602 }
603 if !snapshot.belongs_to_project(&plan.project) {
604 bail!(
605 "cleanup plan row {} does not belong to project {}",
606 snapshot.id,
607 plan.project
608 );
609 }
610 if snapshot.owner_namespace(&plan.project) != current_owner {
611 bail!(
612 "cleanup plan row {} owner does not match current row owner",
613 snapshot.id
614 );
615 }
616 match (current_state_key_id, current_state_key) {
617 (Some(state_key_id), _) if snapshot.state_key_id != Some(state_key_id) => {
618 bail!(
619 "cleanup plan row {} state key does not match current row",
620 snapshot.id
621 );
622 }
623 (None, Some(state_key)) if snapshot.state_key.as_deref() != Some(state_key) => {
624 bail!(
625 "cleanup plan row {} state key does not match current row",
626 snapshot.id
627 );
628 }
629 _ => {}
630 }
631 if topic_group && snapshot.topic_key != current_snapshot.topic_key {
632 bail!(
633 "cleanup plan row {} topic key does not match current row",
634 snapshot.id
635 );
636 }
637 }
638 Ok(())
639}
640
641fn insert_cleanup_operation_log(
642 conn: &Connection,
643 plan: &MemoryCleanupPlan,
644 group: &MemoryCleanupGroup,
645) -> Result<i64> {
646 let current = snapshot_for(&group.row_snapshots, group.current_id)?;
647 let mut operation_plan = MemoryOperationPlan::new(
648 MemoryLifecycleOp::Update,
649 group.state_key.clone(),
650 group.reason.clone(),
651 )
652 .with_target_memory_id(Some(group.current_id))
653 .with_superseded_ids(group.stale_ids.clone());
654 operation_plan.planner_version = CLEANUP_PLANNER_VERSION;
655 let input = MemoryOperationInput {
656 source: "memory_cleanup".to_string(),
657 actor: "memory_cleanup".to_string(),
658 source_project: plan.project.clone(),
659 owner_scope: group
660 .owner_scope
661 .clone()
662 .unwrap_or_else(|| "repo".to_string()),
663 owner_key: group
664 .owner_key
665 .clone()
666 .unwrap_or_else(|| plan.project.clone()),
667 memory_type: group.memory_type.clone(),
668 topic_key: current.topic_key.clone(),
669 state_key: group.state_key.clone(),
670 source_candidate_id: None,
671 confidence: Some(group.confidence),
672 };
673 insert_operation_log(conn, &input, &operation_plan, Some(group.current_id))
674}
675
676fn load_row_snapshots(conn: &Connection, ids: &[i64]) -> Result<Vec<MemoryCleanupRowSnapshot>> {
677 ids.iter()
678 .copied()
679 .map(|id| {
680 load_row_snapshot(conn, id)?
681 .ok_or_else(|| anyhow!("cleanup plan target memory:{id} not found"))
682 })
683 .collect()
684}
685
686fn load_row_snapshot(conn: &Connection, id: i64) -> Result<Option<MemoryCleanupRowSnapshot>> {
687 conn.query_row(
688 "SELECT m.id, m.status, m.content, m.updated_at_epoch, m.owner_scope,
689 m.owner_key, m.memory_type, m.topic_key, m.state_key_id, sk.state_key,
690 sk.current_memory_id, m.project, m.scope, m.source_project, m.target_project
691 FROM memories m
692 LEFT JOIN memory_state_keys sk ON sk.id = m.state_key_id
693 WHERE m.id = ?1",
694 params![id],
695 |row| {
696 let content: String = row.get(2)?;
697 Ok(MemoryCleanupRowSnapshot {
698 id: row.get(0)?,
699 status: row.get(1)?,
700 content_sha256: content_sha256(&content),
701 updated_at_epoch: row.get(3)?,
702 owner_scope: row.get(4)?,
703 owner_key: row.get(5)?,
704 memory_type: row.get(6)?,
705 topic_key: row.get(7)?,
706 state_key_id: row.get(8)?,
707 state_key: row.get(9)?,
708 current_memory_id: row.get(10)?,
709 project: row.get(11)?,
710 scope: row.get(12)?,
711 source_project: row.get(13)?,
712 target_project: row.get(14)?,
713 })
714 },
715 )
716 .optional()
717 .with_context(|| format!("load cleanup plan row snapshot for memory:{id}"))
718}
719
720impl MemoryCleanupRowSnapshot {
721 fn owner_namespace(&self, project: &str) -> (String, String) {
722 match (self.owner_scope.as_deref(), self.owner_key.as_deref()) {
723 (Some(scope), Some(key)) => (scope.to_string(), key.to_string()),
724 _ if self.project == project
725 && self.scope.as_deref().unwrap_or("project") != "global" =>
726 {
727 ("legacy_repo".to_string(), project.to_string())
728 }
729 _ => ("legacy_other".to_string(), self.project.clone()),
730 }
731 }
732
733 fn belongs_to_project(&self, project: &str) -> bool {
734 self.source_project.as_deref() == Some(project)
735 || self.target_project.as_deref() == Some(project)
736 || (self.owner_scope.as_deref() == Some("repo")
737 && self.owner_key.as_deref() == Some(project))
738 || (self.owner_scope.is_none()
739 && self.project == project
740 && self.scope.as_deref().unwrap_or("project") != "global")
741 }
742}
743
744fn snapshot_for(
745 snapshots: &[MemoryCleanupRowSnapshot],
746 id: i64,
747) -> Result<&MemoryCleanupRowSnapshot> {
748 snapshots
749 .iter()
750 .find(|snapshot| snapshot.id == id)
751 .ok_or_else(|| anyhow!("cleanup plan missing snapshot for memory:{id}"))
752}
753
754fn content_sha256(content: &str) -> String {
755 let mut hasher = Sha256::new();
756 hasher.update(content.as_bytes());
757 format!("{:x}", hasher.finalize())
758}