use anyhow::Result;
use rusqlite::{params, Connection};
use crate::db;
use super::tests::{setup_conn, setup_task};
use super::{process_with_generator, MemoryCandidateResult};
pub(super) fn insert_source_observation_typed(
conn: &Connection,
task: &db::ExtractionTask,
observation_type: &str,
text: &str,
) -> Result<()> {
let obs_id = db::insert_observation_with_branch(
conn,
"capture-observation-test",
&task.project,
observation_type,
Some("Typed observation"),
None,
Some(text),
None,
None,
None,
None,
None,
12,
None,
None,
)?;
let event_id = task.high_watermark_event_id.unwrap_or(1);
conn.execute(
"UPDATE observations
SET host_id = ?1,
project_id = ?2,
session_row_id = ?3,
observation_type = ?4,
text = ?5,
evidence_event_ids = ?6,
confidence = ?7
WHERE id = ?8",
params![
task.host_id,
task.project_id,
task.session_row_id,
observation_type,
text,
serde_json::to_string(&vec![event_id])?,
0.91,
obs_id
],
)?;
Ok(())
}
#[tokio::test]
async fn memory_candidate_auto_promotes_architecture_from_discovery_observation() -> Result<()> {
let mut conn = setup_conn();
let task = setup_task(&mut conn, "sess-candidate-architecture")?;
insert_source_observation_typed(
&conn,
&task,
"discovery",
"The extraction worker uses a single-writer SQLite connection per process.",
)?;
let result = process_with_generator(&mut conn, &task, |_prompt| async {
Ok("<memory_candidate><scope>project</scope><type>architecture</type><topic_key>architecture-worker-db</topic_key><risk_class>low</risk_class><confidence>0.92</confidence><text>The extraction worker uses a single-writer SQLite connection per process.</text></memory_candidate>".to_string())
})
.await?;
assert_eq!(
result,
MemoryCandidateResult::Written {
candidates: 1,
promoted: 1,
pending_review: 0,
to_event_id: task.high_watermark_event_id.expect("task watermark")
}
);
let (memory_type, review_status): (String, String) = conn.query_row(
"SELECT memory_type, review_status FROM memory_candidates",
[],
|row| Ok((row.get(0)?, row.get(1)?)),
)?;
assert_eq!(memory_type, "architecture");
assert_eq!(review_status, "auto_promoted");
let memory_count: i64 =
conn.query_row("SELECT COUNT(*) FROM memories", [], |row| row.get(0))?;
assert_eq!(memory_count, 1);
let embedding_count: i64 = conn.query_row(
"SELECT COUNT(*)
FROM memory_embeddings e
JOIN memories m ON m.id = e.memory_id
WHERE m.topic_key = 'architecture-worker-db'",
[],
|row| row.get(0),
)?;
assert_eq!(embedding_count, 1);
Ok(())
}
#[tokio::test]
async fn memory_candidate_auto_promotes_discovery_from_feature_observation() -> Result<()> {
let mut conn = setup_conn();
let task = setup_task(&mut conn, "sess-candidate-feature")?;
insert_source_observation_typed(
&conn,
&task,
"feature",
"Added a structured retrieval gate that scores candidates before promotion.",
)?;
let result = process_with_generator(&mut conn, &task, |_prompt| async {
Ok("<memory_candidate><scope>project</scope><type>discovery</type><topic_key>discovery-retrieval-gate</topic_key><risk_class>low</risk_class><confidence>0.92</confidence><text>Added a structured retrieval gate that scores candidates before promotion.</text></memory_candidate>".to_string())
})
.await?;
assert_eq!(
result,
MemoryCandidateResult::Written {
candidates: 1,
promoted: 1,
pending_review: 0,
to_event_id: task.high_watermark_event_id.expect("task watermark")
}
);
let review_status: String =
conn.query_row("SELECT review_status FROM memory_candidates", [], |row| {
row.get(0)
})?;
assert_eq!(review_status, "auto_promoted");
Ok(())
}
#[tokio::test]
async fn memory_candidate_auto_promotes_condensed_candidate_from_token_overlap() -> Result<()> {
let mut conn = setup_conn();
let task = setup_task(&mut conn, "sess-candidate-condensed-overlap")?;
insert_source_observation_typed(
&conn,
&task,
"feature",
"The worker lifecycle now records auto promote block reasons in memory candidates, keeping pending review diagnostics visible for operator status screens.",
)?;
let result = process_with_generator(&mut conn, &task, |_prompt| async {
Ok("<memory_candidate><scope>project</scope><type>discovery</type><topic_key>discovery-auto-promote-block-reasons</topic_key><risk_class>low</risk_class><confidence>0.92</confidence><text>Worker lifecycle records auto promote block reasons for memory candidate diagnostics.</text></memory_candidate>".to_string())
})
.await?;
assert_eq!(
result,
MemoryCandidateResult::Written {
candidates: 1,
promoted: 1,
pending_review: 0,
to_event_id: task
.high_watermark_event_id
.ok_or_else(|| anyhow::anyhow!("task watermark"))?
}
);
let (review_status, block_reason): (String, Option<String>) = conn.query_row(
"SELECT review_status, auto_promote_block_reason FROM memory_candidates",
[],
|row| Ok((row.get(0)?, row.get(1)?)),
)?;
assert_eq!(review_status, "auto_promoted");
assert_eq!(block_reason, None);
Ok(())
}
#[tokio::test]
async fn memory_candidate_auto_promotes_partial_ordered_support_window() -> Result<()> {
let mut conn = setup_conn();
let task = setup_task(&mut conn, "sess-candidate-partial-overlap")?;
insert_source_observation_typed(
&conn,
&task,
"feature",
"The worker lifecycle records auto promote block reasons in memory candidates for operator diagnostics.",
)?;
let result = process_with_generator(&mut conn, &task, |_prompt| async {
Ok("<memory_candidate><scope>project</scope><type>discovery</type><topic_key>discovery-auto-promote-partial-window</topic_key><risk_class>low</risk_class><confidence>0.92</confidence><text>Worker lifecycle records auto promote block reasons for memory candidate review diagnostics.</text></memory_candidate>".to_string())
})
.await?;
assert_eq!(
result,
MemoryCandidateResult::Written {
candidates: 1,
promoted: 1,
pending_review: 0,
to_event_id: task
.high_watermark_event_id
.ok_or_else(|| anyhow::anyhow!("task watermark"))?
}
);
let review_status: String =
conn.query_row("SELECT review_status FROM memory_candidates", [], |row| {
row.get(0)
})?;
assert_eq!(review_status, "auto_promoted");
Ok(())
}
#[tokio::test]
async fn memory_candidate_auto_promotes_later_duplicate_support_window() -> Result<()> {
let mut conn = setup_conn();
let task = setup_task(&mut conn, "sess-candidate-later-duplicate-overlap")?;
insert_source_observation_typed(
&conn,
&task,
"feature",
"The worker lifecycle initializes queues and refreshes indexes before the worker lifecycle records auto promote block reasons in memory candidate diagnostics.",
)?;
let result = process_with_generator(&mut conn, &task, |_prompt| async {
Ok("<memory_candidate><scope>project</scope><type>discovery</type><topic_key>discovery-auto-promote-later-window</topic_key><risk_class>low</risk_class><confidence>0.92</confidence><text>Worker lifecycle records auto promote block reasons for memory candidate diagnostics.</text></memory_candidate>".to_string())
})
.await?;
assert_eq!(
result,
MemoryCandidateResult::Written {
candidates: 1,
promoted: 1,
pending_review: 0,
to_event_id: task
.high_watermark_event_id
.ok_or_else(|| anyhow::anyhow!("task watermark"))?
}
);
let review_status: String =
conn.query_row("SELECT review_status FROM memory_candidates", [], |row| {
row.get(0)
})?;
assert_eq!(review_status, "auto_promoted");
Ok(())
}
#[tokio::test]
async fn memory_candidate_keeps_weak_overlap_pending_with_block_reason() -> Result<()> {
let mut conn = setup_conn();
let task = setup_task(&mut conn, "sess-candidate-weak-overlap")?;
insert_source_observation_typed(
&conn,
&task,
"feature",
"The review dashboard shows memory candidate counts and pending review diagnostics.",
)?;
let result = process_with_generator(&mut conn, &task, |_prompt| async {
Ok("<memory_candidate><scope>project</scope><type>discovery</type><topic_key>discovery-auto-promote-support</topic_key><risk_class>low</risk_class><confidence>0.92</confidence><text>Auto promotion support validates rewritten durable memory facts against source observations.</text></memory_candidate>".to_string())
})
.await?;
assert_eq!(
result,
MemoryCandidateResult::Written {
candidates: 1,
promoted: 0,
pending_review: 1,
to_event_id: task
.high_watermark_event_id
.ok_or_else(|| anyhow::anyhow!("task watermark"))?
}
);
let (review_status, block_reason): (String, Option<String>) = conn.query_row(
"SELECT review_status, auto_promote_block_reason FROM memory_candidates",
[],
|row| Ok((row.get(0)?, row.get(1)?)),
)?;
assert_eq!(review_status, "pending_review");
assert_eq!(
block_reason.as_deref(),
Some("no_supporting_source_observation")
);
Ok(())
}
#[tokio::test]
async fn memory_candidate_keeps_inverted_action_overlap_pending() -> Result<()> {
let mut conn = setup_conn();
let task = setup_task(&mut conn, "sess-candidate-inverted-action")?;
insert_source_observation_typed(
&conn,
&task,
"feature",
"The worker lifecycle disables auto promote block reasons for memory candidate diagnostics.",
)?;
let result = process_with_generator(&mut conn, &task, |_prompt| async {
Ok("<memory_candidate><scope>project</scope><type>discovery</type><topic_key>discovery-auto-promote-support</topic_key><risk_class>low</risk_class><confidence>0.92</confidence><text>Worker lifecycle records auto promote block reasons for memory candidate diagnostics.</text></memory_candidate>".to_string())
})
.await?;
assert_eq!(
result,
MemoryCandidateResult::Written {
candidates: 1,
promoted: 0,
pending_review: 1,
to_event_id: task
.high_watermark_event_id
.ok_or_else(|| anyhow::anyhow!("task watermark"))?
}
);
let (review_status, block_reason): (String, Option<String>) = conn.query_row(
"SELECT review_status, auto_promote_block_reason FROM memory_candidates",
[],
|row| Ok((row.get(0)?, row.get(1)?)),
)?;
assert_eq!(review_status, "pending_review");
assert_eq!(
block_reason.as_deref(),
Some("no_supporting_source_observation")
);
Ok(())
}
#[tokio::test]
async fn memory_candidate_keeps_negated_support_overlap_pending() -> Result<()> {
let mut conn = setup_conn();
let task = setup_task(&mut conn, "sess-candidate-negated-support")?;
insert_source_observation_typed(
&conn,
&task,
"bugfix",
"The ingestion worker cannot promote condensed memory candidates because source evidence support validation fails.",
)?;
let result = process_with_generator(&mut conn, &task, |_prompt| async {
Ok("<memory_candidate><scope>project</scope><type>bugfix</type><topic_key>bugfix-auto-promote-support</topic_key><risk_class>low</risk_class><confidence>0.92</confidence><text>Ingestion worker promotes condensed memory candidates because source evidence support validation succeeds.</text></memory_candidate>".to_string())
})
.await?;
assert_eq!(
result,
MemoryCandidateResult::Written {
candidates: 1,
promoted: 0,
pending_review: 1,
to_event_id: task
.high_watermark_event_id
.ok_or_else(|| anyhow::anyhow!("task watermark"))?
}
);
let (review_status, block_reason): (String, Option<String>) = conn.query_row(
"SELECT review_status, auto_promote_block_reason FROM memory_candidates",
[],
|row| Ok((row.get(0)?, row.get(1)?)),
)?;
assert_eq!(review_status, "pending_review");
assert_eq!(
block_reason.as_deref(),
Some("no_supporting_source_observation")
);
Ok(())
}
#[tokio::test]
async fn memory_candidate_keeps_contracted_negation_overlap_pending() -> Result<()> {
let mut conn = setup_conn();
let task = setup_task(&mut conn, "sess-candidate-contracted-negation")?;
insert_source_observation_typed(
&conn,
&task,
"feature",
"The ingestion worker doesn't persist durable memory candidates from summarized observations.",
)?;
let result = process_with_generator(&mut conn, &task, |_prompt| async {
Ok("<memory_candidate><scope>project</scope><type>discovery</type><topic_key>discovery-contracted-negation</topic_key><risk_class>low</risk_class><confidence>0.92</confidence><text>Ingestion worker persists durable memory candidates from summarized observations.</text></memory_candidate>".to_string())
})
.await?;
assert_eq!(
result,
MemoryCandidateResult::Written {
candidates: 1,
promoted: 0,
pending_review: 1,
to_event_id: task
.high_watermark_event_id
.ok_or_else(|| anyhow::anyhow!("task watermark"))?
}
);
let review_status: String =
conn.query_row("SELECT review_status FROM memory_candidates", [], |row| {
row.get(0)
})?;
assert_eq!(review_status, "pending_review");
Ok(())
}
#[tokio::test]
async fn memory_candidate_keeps_uncertain_overlap_pending() -> Result<()> {
let mut conn = setup_conn();
let task = setup_task(&mut conn, "sess-candidate-uncertain-overlap")?;
insert_source_observation_typed(
&conn,
&task,
"feature",
"The ingestion worker may promote summarized memory candidates because source evidence support validation passes.",
)?;
let result = process_with_generator(&mut conn, &task, |_prompt| async {
Ok("<memory_candidate><scope>project</scope><type>discovery</type><topic_key>discovery-uncertain-overlap</topic_key><risk_class>low</risk_class><confidence>0.92</confidence><text>Ingestion worker promotes summarized memory candidates because source evidence support validation passes.</text></memory_candidate>".to_string())
})
.await?;
assert_eq!(
result,
MemoryCandidateResult::Written {
candidates: 1,
promoted: 0,
pending_review: 1,
to_event_id: task
.high_watermark_event_id
.ok_or_else(|| anyhow::anyhow!("task watermark"))?
}
);
let review_status: String =
conn.query_row("SELECT review_status FROM memory_candidates", [], |row| {
row.get(0)
})?;
assert_eq!(review_status, "pending_review");
Ok(())
}
#[tokio::test]
async fn memory_candidate_keeps_prescriptive_modal_overlap_pending() -> Result<()> {
let mut conn = setup_conn();
let task = setup_task(&mut conn, "sess-candidate-prescriptive-overlap")?;
insert_source_observation_typed(
&conn,
&task,
"feature",
"The ingestion worker must persist durable memory candidates from summarized observations.",
)?;
let result = process_with_generator(&mut conn, &task, |_prompt| async {
Ok("<memory_candidate><scope>project</scope><type>discovery</type><topic_key>discovery-prescriptive-overlap</topic_key><risk_class>low</risk_class><confidence>0.92</confidence><text>Ingestion worker persists durable memory candidates from summarized observations.</text></memory_candidate>".to_string())
})
.await?;
assert_eq!(
result,
MemoryCandidateResult::Written {
candidates: 1,
promoted: 0,
pending_review: 1,
to_event_id: task
.high_watermark_event_id
.ok_or_else(|| anyhow::anyhow!("task watermark"))?
}
);
let review_status: String =
conn.query_row("SELECT review_status FROM memory_candidates", [], |row| {
row.get(0)
})?;
assert_eq!(review_status, "pending_review");
Ok(())
}
#[tokio::test]
async fn memory_candidate_keeps_conditional_overlap_pending() -> Result<()> {
let mut conn = setup_conn();
let task = setup_task(&mut conn, "sess-candidate-conditional-overlap")?;
insert_source_observation_typed(
&conn,
&task,
"feature",
"If the ingestion worker persists durable memory candidates from summarized observations, the dashboard reports them.",
)?;
let result = process_with_generator(&mut conn, &task, |_prompt| async {
Ok("<memory_candidate><scope>project</scope><type>discovery</type><topic_key>discovery-conditional-overlap</topic_key><risk_class>low</risk_class><confidence>0.92</confidence><text>Ingestion worker persists durable memory candidates from summarized observations.</text></memory_candidate>".to_string())
})
.await?;
assert_eq!(
result,
MemoryCandidateResult::Written {
candidates: 1,
promoted: 0,
pending_review: 1,
to_event_id: task
.high_watermark_event_id
.ok_or_else(|| anyhow::anyhow!("task watermark"))?
}
);
let review_status: String =
conn.query_row("SELECT review_status FROM memory_candidates", [], |row| {
row.get(0)
})?;
assert_eq!(review_status, "pending_review");
Ok(())
}
#[tokio::test]
async fn memory_candidate_keeps_actor_swap_overlap_pending() -> Result<()> {
let mut conn = setup_conn();
let task = setup_task(&mut conn, "sess-candidate-actor-swap")?;
insert_source_observation_typed(
&conn,
&task,
"feature",
"The review dashboard shows pending memory candidate diagnostics while the ingestion worker validates rewritten durable memory facts against source observations.",
)?;
let result = process_with_generator(&mut conn, &task, |_prompt| async {
Ok("<memory_candidate><scope>project</scope><type>discovery</type><topic_key>discovery-actor-swap</topic_key><risk_class>low</risk_class><confidence>0.92</confidence><text>Review dashboard validates rewritten durable memory facts against source observations.</text></memory_candidate>".to_string())
})
.await?;
assert_eq!(
result,
MemoryCandidateResult::Written {
candidates: 1,
promoted: 0,
pending_review: 1,
to_event_id: task
.high_watermark_event_id
.ok_or_else(|| anyhow::anyhow!("task watermark"))?
}
);
let review_status: String =
conn.query_row("SELECT review_status FROM memory_candidates", [], |row| {
row.get(0)
})?;
assert_eq!(review_status, "pending_review");
Ok(())
}
#[tokio::test]
async fn memory_candidate_keeps_clause_spanning_actor_swap_pending() -> Result<()> {
let mut conn = setup_conn();
let task = setup_task(&mut conn, "sess-candidate-clause-actor-swap")?;
insert_source_observation_typed(
&conn,
&task,
"feature",
"The dashboard status panel validates data while the worker records durable memory candidate errors.",
)?;
let result = process_with_generator(&mut conn, &task, |_prompt| async {
Ok("<memory_candidate><scope>project</scope><type>discovery</type><topic_key>discovery-clause-actor-swap</topic_key><risk_class>low</risk_class><confidence>0.92</confidence><text>Dashboard status panel records durable memory candidate errors.</text></memory_candidate>".to_string())
})
.await?;
assert_eq!(
result,
MemoryCandidateResult::Written {
candidates: 1,
promoted: 0,
pending_review: 1,
to_event_id: task
.high_watermark_event_id
.ok_or_else(|| anyhow::anyhow!("task watermark"))?
}
);
let (review_status, block_reason): (String, Option<String>) = conn.query_row(
"SELECT review_status, auto_promote_block_reason FROM memory_candidates",
[],
|row| Ok((row.get(0)?, row.get(1)?)),
)?;
assert_eq!(review_status, "pending_review");
assert_eq!(
block_reason.as_deref(),
Some("no_supporting_source_observation")
);
Ok(())
}
#[tokio::test]
async fn memory_candidate_keeps_temporal_boundary_actor_swap_pending() -> Result<()> {
let mut conn = setup_conn();
let task = setup_task(&mut conn, "sess-candidate-temporal-actor-swap")?;
insert_source_observation_typed(
&conn,
&task,
"feature",
"The dashboard status panel displays metrics before the worker records durable memory candidate errors.",
)?;
let result = process_with_generator(&mut conn, &task, |_prompt| async {
Ok("<memory_candidate><scope>project</scope><type>discovery</type><topic_key>discovery-temporal-actor-swap</topic_key><risk_class>low</risk_class><confidence>0.92</confidence><text>Dashboard status panel records durable memory candidate errors.</text></memory_candidate>".to_string())
})
.await?;
assert_eq!(
result,
MemoryCandidateResult::Written {
candidates: 1,
promoted: 0,
pending_review: 1,
to_event_id: task
.high_watermark_event_id
.ok_or_else(|| anyhow::anyhow!("task watermark"))?
}
);
let (review_status, block_reason): (String, Option<String>) = conn.query_row(
"SELECT review_status, auto_promote_block_reason FROM memory_candidates",
[],
|row| Ok((row.get(0)?, row.get(1)?)),
)?;
assert_eq!(review_status, "pending_review");
assert_eq!(
block_reason.as_deref(),
Some("no_supporting_source_observation")
);
Ok(())
}
#[tokio::test]
async fn memory_candidate_keeps_mismatched_short_identifier_overlap_pending() -> Result<()> {
let mut conn = setup_conn();
let task = setup_task(&mut conn, "sess-candidate-short-id-mismatch")?;
insert_source_observation_typed(
&conn,
&task,
"feature",
"CLI gateway stores SSH auth keys in Redis cache layer.",
)?;
let result = process_with_generator(&mut conn, &task, |_prompt| async {
Ok("<memory_candidate><scope>project</scope><type>discovery</type><topic_key>discovery-short-id-mismatch</topic_key><risk_class>low</risk_class><confidence>0.92</confidence><text>API gateway stores JWT auth keys in Redis cache layer.</text></memory_candidate>".to_string())
})
.await?;
assert_eq!(
result,
MemoryCandidateResult::Written {
candidates: 1,
promoted: 0,
pending_review: 1,
to_event_id: task
.high_watermark_event_id
.ok_or_else(|| anyhow::anyhow!("task watermark"))?
}
);
let (review_status, block_reason): (String, Option<String>) = conn.query_row(
"SELECT review_status, auto_promote_block_reason FROM memory_candidates",
[],
|row| Ok((row.get(0)?, row.get(1)?)),
)?;
assert_eq!(review_status, "pending_review");
assert_eq!(
block_reason.as_deref(),
Some("no_supporting_source_observation")
);
Ok(())
}
#[tokio::test]
async fn memory_candidate_auto_promotes_supported_short_identifiers() -> Result<()> {
let mut conn = setup_conn();
let task = setup_task(&mut conn, "sess-candidate-short-id-supported")?;
insert_source_observation_typed(
&conn,
&task,
"feature",
"The API gateway stores JWT auth keys inside the Redis cache layer for operator login flows.",
)?;
let result = process_with_generator(&mut conn, &task, |_prompt| async {
Ok("<memory_candidate><scope>project</scope><type>discovery</type><topic_key>discovery-short-id-supported</topic_key><risk_class>low</risk_class><confidence>0.92</confidence><text>API gateway stores JWT auth key in Redis cache layer.</text></memory_candidate>".to_string())
})
.await?;
assert_eq!(
result,
MemoryCandidateResult::Written {
candidates: 1,
promoted: 1,
pending_review: 0,
to_event_id: task
.high_watermark_event_id
.ok_or_else(|| anyhow::anyhow!("task watermark"))?
}
);
let review_status: String =
conn.query_row("SELECT review_status FROM memory_candidates", [], |row| {
row.get(0)
})?;
assert_eq!(review_status, "auto_promoted");
Ok(())
}
#[tokio::test]
async fn memory_candidate_keeps_architecture_unsupported_by_bugfix_pending() -> Result<()> {
let mut conn = setup_conn();
let task = setup_task(&mut conn, "sess-candidate-architecture-bugfix")?;
insert_source_observation_typed(
&conn,
&task,
"bugfix",
"The extraction worker uses a single-writer SQLite connection per process.",
)?;
let result = process_with_generator(&mut conn, &task, |_prompt| async {
Ok("<memory_candidate><scope>project</scope><type>architecture</type><topic_key>architecture-worker-db</topic_key><risk_class>low</risk_class><confidence>0.92</confidence><text>The extraction worker uses a single-writer SQLite connection per process.</text></memory_candidate>".to_string())
})
.await?;
assert_eq!(
result,
MemoryCandidateResult::Written {
candidates: 1,
promoted: 0,
pending_review: 1,
to_event_id: task.high_watermark_event_id.expect("task watermark")
}
);
let (review_status, block_reason): (String, Option<String>) = conn.query_row(
"SELECT review_status, auto_promote_block_reason FROM memory_candidates",
[],
|row| Ok((row.get(0)?, row.get(1)?)),
)?;
assert_eq!(review_status, "pending_review");
assert_eq!(
block_reason.as_deref(),
Some("no_supporting_source_observation")
);
Ok(())
}