use anyhow::{Context, Result};
use rusqlite::{params, OptionalExtension};
use crate::db;
use crate::memory::format;
use crate::workstream::ParsedWorkStream;
use super::super::parse::ParsedSummary;
pub(super) fn build_existing_summary_context(
conn: &rusqlite::Connection,
memory_sid: &str,
project: &str,
) -> Result<String> {
let prev = db::get_summary_by_session(conn, memory_sid, project)?;
let prev_workstream = get_linked_workstream_context(conn, memory_sid, project)?;
if prev.is_none() && prev_workstream.is_none() {
return Ok(String::new());
}
let mut parts = Vec::new();
if let Some(prev) = prev {
push_summary_tag(&mut parts, "request", prev.request.as_deref());
push_summary_tag(&mut parts, "completed", prev.completed.as_deref());
push_summary_tag(&mut parts, "decisions", prev.decisions.as_deref());
push_summary_tag(&mut parts, "learned", prev.learned.as_deref());
push_summary_tag(&mut parts, "next_steps", prev.next_steps.as_deref());
push_summary_tag(&mut parts, "preferences", prev.preferences.as_deref());
}
if let Some(prev_workstream) = prev_workstream {
push_summary_tag(&mut parts, "workstream", Some(&prev_workstream.title));
push_summary_tag(
&mut parts,
"workstream_progress",
prev_workstream.progress.as_deref(),
);
push_summary_tag(
&mut parts,
"workstream_next",
prev_workstream.next_action.as_deref(),
);
push_summary_tag(
&mut parts,
"workstream_blockers",
prev_workstream.blockers.as_deref(),
);
}
Ok(format!(
"<existing_summary>\n{}\n</existing_summary>\n\n",
parts.join("\n")
))
}
pub(super) fn finalize_summary(
conn: &mut rusqlite::Connection,
session_id: &str,
memory_sid: &str,
project: &str,
msg_hash: &str,
summary: ParsedSummary,
) -> Result<()> {
let usage = summary_text_usage(&summary);
let _deleted = match db::finalize_summarize(
conn,
memory_sid,
project,
msg_hash,
summary.request.as_deref(),
summary.completed.as_deref(),
summary.decisions.as_deref(),
summary.learned.as_deref(),
summary.next_steps.as_deref(),
summary.preferences.as_deref(),
None,
usage,
) {
Ok(deleted) => deleted,
Err(err) => {
release_lock_after_error(conn, project, "finalize-failure");
return Err(err);
}
};
db::release_summarize_lock(conn, project)?;
crate::log::info(
"summary-job",
&format!("saved summary project={} session={}", project, session_id),
);
if let Some(workstream) = parsed_workstream_from_summary(&summary) {
match crate::workstream::upsert_workstream(conn, project, memory_sid, &workstream) {
Ok(workstream_id) => crate::log::info(
"summary-job",
&format!(
"upserted workstream id={} project={} session={}",
workstream_id, project, session_id
),
),
Err(err) => crate::log::warn(
"summary-job",
&format!("workstream persistence failed: {}", err),
),
}
}
if let Err(err) = crate::memory::promote_summary_to_memories(
conn,
session_id,
project,
summary.request.as_deref(),
summary.decisions.as_deref(),
summary.learned.as_deref(),
summary.preferences.as_deref(),
) {
crate::log::warn("summary-job", &format!("memory promotion failed: {}", err));
db::clear_summarize_cooldown_for_message(conn, project, msg_hash)
.context("failed to clear summary retry marker after memory promotion failure")?;
return Err(err).context("memory promotion failed");
}
Ok(())
}
pub(super) fn sync_native_memory(cwd: &str, project: &str) {
if let Err(err) = crate::context::claude_memory::sync_to_claude_memory(cwd, project) {
crate::log::warn(
"summary-job",
&format!("claude memory sync failed: {}", err),
);
}
}
fn push_summary_tag(parts: &mut Vec<String>, tag: &str, value: Option<&str>) {
if let Some(value) = value {
parts.push(format!("<{tag}>{}</{tag}>", format::xml_escape_text(value)));
}
}
fn summary_text_usage(summary: &ParsedSummary) -> i64 {
let total_len = [
summary.request.as_deref(),
summary.completed.as_deref(),
summary.decisions.as_deref(),
summary.learned.as_deref(),
summary.next_steps.as_deref(),
summary.preferences.as_deref(),
summary.workstream.as_deref(),
summary.workstream_progress.as_deref(),
summary.workstream_next.as_deref(),
summary.workstream_blockers.as_deref(),
]
.into_iter()
.flatten()
.map(str::len)
.sum::<usize>();
(total_len / 4) as i64
}
fn release_lock_after_error(conn: &rusqlite::Connection, project: &str, reason: &str) {
if let Err(e) = db::release_summarize_lock(conn, project) {
crate::log::error(
"summary-job",
&format!(
"[LOCK LEAK] failed to release summarize lock for {project} after {reason}: {e}"
),
);
}
}
#[derive(Debug)]
struct ExistingWorkStreamContext {
title: String,
progress: Option<String>,
next_action: Option<String>,
blockers: Option<String>,
}
fn get_linked_workstream_context(
conn: &rusqlite::Connection,
memory_sid: &str,
project: &str,
) -> Result<Option<ExistingWorkStreamContext>> {
conn.query_row(
"SELECT ws.title, ws.progress, ws.next_action, ws.blockers
FROM workstreams ws
JOIN workstream_sessions wss ON wss.workstream_id = ws.id
WHERE wss.memory_session_id = ?1 AND ws.project = ?2
ORDER BY wss.linked_at_epoch DESC, ws.updated_at_epoch DESC
LIMIT 1",
params![memory_sid, project],
|row| {
Ok(ExistingWorkStreamContext {
title: row.get(0)?,
progress: row.get(1)?,
next_action: row.get(2)?,
blockers: row.get(3)?,
})
},
)
.optional()
.map_err(Into::into)
}
fn parsed_workstream_from_summary(summary: &ParsedSummary) -> Option<ParsedWorkStream> {
Some(ParsedWorkStream {
title: clean_field(summary.workstream.as_deref()),
progress: clean_field(summary.workstream_progress.as_deref()),
next_action: clean_field(summary.workstream_next.as_deref()),
blockers: clean_field(summary.workstream_blockers.as_deref()),
is_completed: false,
})
.filter(|workstream| workstream.title.is_some())
}
fn clean_field(value: Option<&str>) -> Option<String> {
value
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
}
#[cfg(test)]
mod tests {
use anyhow::Result;
use rusqlite::{params, Connection};
use crate::{db, summarize::ParsedSummary};
use super::{build_existing_summary_context, finalize_summary};
fn setup_conn() -> Result<Connection> {
let conn = Connection::open_in_memory()?;
crate::migrate::run_migrations(&conn)?;
Ok(conn)
}
#[test]
fn promotion_failure_releases_lock_and_clears_retry_marker() -> Result<()> {
let mut conn = setup_conn()?;
let project = "proj/promotion-failure";
let session_id = "content-session-1";
let memory_sid = "memory-session-1";
let msg_hash = "message-hash-1";
assert!(db::try_acquire_summarize_lock(&mut conn, project, 60)?);
conn.execute_batch("DROP TABLE memories;")?;
let err = finalize_summary(
&mut conn,
session_id,
memory_sid,
project,
msg_hash,
ParsedSummary {
request: Some("Capture decisions from a summary".to_string()),
completed: Some("Saved session summary".to_string()),
decisions: Some(
"Use a retryable worker failure when summary promotion cannot persist"
.to_string(),
),
learned: None,
next_steps: None,
preferences: None,
workstream: None,
workstream_progress: None,
workstream_next: None,
workstream_blockers: None,
},
)
.expect_err("promotion failure should surface to the worker");
assert!(
err.to_string().contains("memory promotion failed"),
"unexpected error: {err:#}"
);
let locks: i64 = conn.query_row(
"SELECT COUNT(*) FROM summarize_locks WHERE project = ?1",
params![project],
|row| row.get(0),
)?;
assert_eq!(locks, 0, "summarize lock should be released");
assert!(
!db::is_summarize_on_cooldown(&conn, project, 60 * 60)?,
"cooldown should not suppress retry after promotion failure"
);
assert!(
!db::is_duplicate_message(&conn, project, msg_hash)?,
"duplicate marker should not suppress retry after promotion failure"
);
Ok(())
}
#[test]
fn build_existing_summary_context_includes_linked_workstream_fields() -> Result<()> {
let conn = setup_conn()?;
conn.execute(
"INSERT INTO session_summaries
(memory_session_id, project, request, completed, decisions, learned,
next_steps, preferences, created_at, created_at_epoch)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
params![
"mem-ctx",
"test/proj",
"Repair summary capture",
"Saved summary row",
"Use WorkStream table as source of truth",
"Incremental summaries need linked task state",
"Run persistence test",
"Prefer evidence",
"2026-05-13T00:00:00Z",
1_768_000_000_i64,
],
)?;
conn.execute(
"INSERT INTO workstreams
(project, title, status, progress, next_action, blockers,
created_at_epoch, updated_at_epoch)
VALUES (?1, ?2, 'active', ?3, ?4, ?5, ?6, ?7)",
params![
"test/proj",
"Summary WorkStream Persistence",
"Parsed fields wired",
"Finalize summary test",
"Need clippy",
1_768_000_000_i64,
1_768_000_010_i64,
],
)?;
let workstream_id = conn.last_insert_rowid();
conn.execute(
"INSERT INTO workstream_sessions
(workstream_id, memory_session_id, linked_at_epoch)
VALUES (?1, ?2, ?3)",
params![workstream_id, "mem-ctx", 1_768_000_011_i64],
)?;
let context = build_existing_summary_context(&conn, "mem-ctx", "test/proj")?;
assert!(context.contains("<request>Repair summary capture</request>"));
assert!(context.contains("<workstream>Summary WorkStream Persistence</workstream>"));
assert!(context.contains("<workstream_progress>Parsed fields wired</workstream_progress>"));
assert!(context.contains("<workstream_next>Finalize summary test</workstream_next>"));
assert!(context.contains("<workstream_blockers>Need clippy</workstream_blockers>"));
Ok(())
}
#[test]
fn finalize_summary_persists_non_empty_workstream() -> Result<()> {
let mut conn = setup_conn()?;
let summary = ParsedSummary {
request: Some("Wire summary workstream fields".to_string()),
completed: Some("Added parser and persistence wiring".to_string()),
decisions: None,
learned: None,
next_steps: Some("Run full validation".to_string()),
preferences: None,
workstream: Some("Summary WorkStream Persistence".to_string()),
workstream_progress: Some("ParsedSummary now carries workstream fields".to_string()),
workstream_next: Some("Open PR for issue 136".to_string()),
workstream_blockers: Some("Waiting for review".to_string()),
};
finalize_summary(
&mut conn,
"content-session-1",
"mem-finalize",
"test/proj",
"hash-finalize",
summary,
)?;
let row: (String, Option<String>, Option<String>, Option<String>) = conn.query_row(
"SELECT title, progress, next_action, blockers
FROM workstreams
WHERE project = ?1",
params!["test/proj"],
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
)?;
assert_eq!(row.0, "Summary WorkStream Persistence");
assert_eq!(
row.1.as_deref(),
Some("ParsedSummary now carries workstream fields")
);
assert_eq!(row.2.as_deref(), Some("Open PR for issue 136"));
assert_eq!(row.3.as_deref(), Some("Waiting for review"));
let linked_count: i64 = conn.query_row(
"SELECT COUNT(*)
FROM workstream_sessions wss
JOIN workstreams ws ON ws.id = wss.workstream_id
WHERE wss.memory_session_id = ?1 AND ws.project = ?2",
params!["mem-finalize", "test/proj"],
|row| row.get(0),
)?;
assert_eq!(linked_count, 1);
let request: String = conn.query_row(
"SELECT request FROM session_summaries
WHERE memory_session_id = ?1 AND project = ?2",
params!["mem-finalize", "test/proj"],
|row| row.get(0),
)?;
assert_eq!(request, "Wire summary workstream fields");
Ok(())
}
}