mirror of
https://github.com/tinyhumansai/openhuman.git
synced 2026-07-27 21:08:00 +00:00
Steven Enamakel
Generated
+1
-1
@@ -4520,7 +4520,7 @@ checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381"
|
||||
|
||||
[[package]]
|
||||
name = "openhuman"
|
||||
version = "0.53.20"
|
||||
version = "0.53.22"
|
||||
dependencies = [
|
||||
"aes-gcm",
|
||||
"anyhow",
|
||||
|
||||
@@ -809,6 +809,12 @@ impl Agent {
|
||||
// librarian task that's idempotent across reruns.
|
||||
if result.is_ok() && self.context.should_extract_session_memory() {
|
||||
self.spawn_session_memory_extraction();
|
||||
// Sibling pipeline (#1399): heuristic transcript ingestion
|
||||
// turns the just-written transcript into durable
|
||||
// conversational memory + reflections so a brand-new chat
|
||||
// can recover continuity. Background-only, never blocks the
|
||||
// user-facing turn return.
|
||||
self.spawn_transcript_ingestion();
|
||||
}
|
||||
|
||||
result
|
||||
@@ -1475,6 +1481,49 @@ impl Agent {
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/// Spawn a background task that ingests the current session
|
||||
/// transcript into the conversational-memory store.
|
||||
///
|
||||
/// Issue #1399: complements `spawn_session_memory_extraction`. The
|
||||
/// archivist path writes dense bullets into `MEMORY.md`; this path
|
||||
/// extracts importance-tagged, provenance-bearing memories via the
|
||||
/// heuristic [`crate::openhuman::learning::transcript_ingest`]
|
||||
/// pipeline. The two are deliberately independent so the prompt
|
||||
/// retrieval layer can pull from `conversation_memory` without
|
||||
/// needing the archivist's extraction to have fired this session.
|
||||
///
|
||||
/// Fire-and-forget: failures are logged, never propagated.
|
||||
pub(super) fn spawn_transcript_ingestion(&self) {
|
||||
let Some(path) = self.session_transcript_path.clone() else {
|
||||
log::debug!("[transcript_ingest] no session transcript path yet — skipping spawn");
|
||||
return;
|
||||
};
|
||||
let memory = std::sync::Arc::clone(&self.memory);
|
||||
|
||||
tokio::spawn(async move {
|
||||
match crate::openhuman::learning::transcript_ingest::ingest_transcript_path(
|
||||
memory.as_ref(),
|
||||
&path,
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(report) => tracing::info!(
|
||||
transcript = %path.display(),
|
||||
extracted = report.extracted,
|
||||
stored = report.stored,
|
||||
deduped = report.deduped,
|
||||
reflections_stored = report.reflections_stored,
|
||||
"[transcript_ingest] background ingest complete"
|
||||
),
|
||||
Err(err) => tracing::warn!(
|
||||
transcript = %path.display(),
|
||||
error = %err,
|
||||
"[transcript_ingest] background ingest failed — will retry next threshold window"
|
||||
),
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
/// Wrapper around
|
||||
|
||||
@@ -3,6 +3,17 @@ use async_trait::async_trait;
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
use super::harness::memory_context::{WORKING_MEMORY_KEY_PREFIX, WORKING_MEMORY_LIMIT};
|
||||
use crate::openhuman::learning::transcript_ingest::CONVERSATION_MEMORY_NAMESPACE;
|
||||
|
||||
/// Maximum number of `[Prior conversations]` lines surfaced into the prompt
|
||||
/// at the start of a fresh chat. Tight cap on purpose: this block is meant
|
||||
/// to recover continuity for high-importance facts, not to dump session
|
||||
/// history into context. See issue #1399.
|
||||
const PRIOR_CONVERSATION_LIMIT: usize = 3;
|
||||
/// Only the importance prefix `high.` survives into the prompt block.
|
||||
/// Medium/low entries stay queryable via the on-demand memory tool but
|
||||
/// do not auto-pollute every fresh chat.
|
||||
const PRIOR_CONVERSATION_KEY_PREFIX: &str = "high.";
|
||||
|
||||
#[async_trait]
|
||||
pub trait MemoryLoader: Send + Sync {
|
||||
@@ -165,6 +176,76 @@ impl MemoryLoader for DefaultMemoryLoader {
|
||||
context.push_str(&line);
|
||||
}
|
||||
|
||||
// ── Prior conversations (issue #1399) ─────────────────────────
|
||||
// High-importance, transcript-derived facts from earlier chats.
|
||||
// Namespace-scoped recall keeps this block small and tightly
|
||||
// bounded — only entries the heuristic extractor flagged as
|
||||
// `high.*` are eligible, and only the first short snippet of
|
||||
// each is included so the block never crowds out the user's
|
||||
// actual message.
|
||||
let prior_query = format!("{} {}", CONVERSATION_MEMORY_NAMESPACE, user_message);
|
||||
let prior_entries = memory
|
||||
.recall(
|
||||
&prior_query,
|
||||
PRIOR_CONVERSATION_LIMIT * 4,
|
||||
crate::openhuman::memory::RecallOpts {
|
||||
namespace: Some(CONVERSATION_MEMORY_NAMESPACE),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.unwrap_or_default();
|
||||
|
||||
let mut appended_prior_header = false;
|
||||
let mut prior_added = 0usize;
|
||||
for entry in prior_entries
|
||||
.into_iter()
|
||||
.filter(|e| e.key.starts_with(PRIOR_CONVERSATION_KEY_PREFIX))
|
||||
.filter(|e| match e.score {
|
||||
Some(score) => score >= self.min_relevance_score,
|
||||
None => true,
|
||||
})
|
||||
{
|
||||
if prior_added >= PRIOR_CONVERSATION_LIMIT {
|
||||
break;
|
||||
}
|
||||
// The stored content is two lines:
|
||||
// [high preference] I prefer Postgres ...
|
||||
// [provenance] {"thread_id":"thr_…", ...}
|
||||
// For the prompt we keep only the first line so the block
|
||||
// stays compact. Provenance survives in the underlying
|
||||
// memory entry and is queryable through the memory tool.
|
||||
let primary = entry
|
||||
.content
|
||||
.lines()
|
||||
.find(|l| !l.trim_start().starts_with("[provenance]"))
|
||||
.unwrap_or(&entry.content)
|
||||
.trim();
|
||||
if primary.is_empty() {
|
||||
continue;
|
||||
}
|
||||
if !appended_prior_header {
|
||||
let section = "[Prior conversations]\n";
|
||||
if context.len() + section.len() > budget {
|
||||
break;
|
||||
}
|
||||
context.push_str(section);
|
||||
appended_prior_header = true;
|
||||
}
|
||||
let line = format!("- {primary}\n");
|
||||
if context.len() + line.len() > budget {
|
||||
tracing::debug!(
|
||||
budget,
|
||||
current_len = context.len(),
|
||||
skipped_line_len = line.len(),
|
||||
"[memory_loader] context budget reached while appending prior conversations"
|
||||
);
|
||||
break;
|
||||
}
|
||||
context.push_str(&line);
|
||||
prior_added += 1;
|
||||
}
|
||||
|
||||
if context.is_empty() {
|
||||
return Ok(String::new());
|
||||
}
|
||||
@@ -253,6 +334,57 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn loader_surfaces_prior_conversation_high_importance_only() {
|
||||
// Prior chat extracted two memories: one high-importance preference
|
||||
// and one medium-importance unresolved task. Only the high one
|
||||
// should make it into the loader's prompt block (#1399).
|
||||
let mem = MockMemory {
|
||||
entries: vec![
|
||||
MemoryEntry {
|
||||
id: "id-1".into(),
|
||||
key: "high.preference.aaaaaaaaaaaa".into(),
|
||||
content: "[high preference] I prefer Postgres for new services.\n[provenance] {\"thread_id\":\"thr_old\"}".into(),
|
||||
namespace: Some(super::CONVERSATION_MEMORY_NAMESPACE.to_string()),
|
||||
category: MemoryCategory::Conversation,
|
||||
timestamp: "2026-04-22T00:00:00Z".into(),
|
||||
session_id: Some("thr_old".into()),
|
||||
score: Some(0.9),
|
||||
},
|
||||
MemoryEntry {
|
||||
id: "id-2".into(),
|
||||
key: "med.unresolved_task.bbbbbbbbbbbb".into(),
|
||||
content: "[med unresolved_task] still need to migrate auth.".into(),
|
||||
namespace: Some(super::CONVERSATION_MEMORY_NAMESPACE.to_string()),
|
||||
category: MemoryCategory::Conversation,
|
||||
timestamp: "2026-04-22T00:00:00Z".into(),
|
||||
session_id: None,
|
||||
score: Some(0.9),
|
||||
},
|
||||
],
|
||||
};
|
||||
|
||||
let loader = DefaultMemoryLoader::default();
|
||||
let out = loader
|
||||
.load_context(&mem, "what should I default to for storage?")
|
||||
.await
|
||||
.expect("loader must succeed");
|
||||
|
||||
assert!(
|
||||
out.contains("[Prior conversations]"),
|
||||
"expected prior conversations block, got:\n{out}"
|
||||
);
|
||||
assert!(out.contains("Postgres"));
|
||||
assert!(
|
||||
!out.contains("migrate auth"),
|
||||
"med-importance entries must not auto-surface, got:\n{out}"
|
||||
);
|
||||
assert!(
|
||||
!out.contains("[provenance]"),
|
||||
"provenance is not rendered into the prompt block, got:\n{out}"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn collect_recall_citations_filters_and_truncates_entries() {
|
||||
let mem = MockMemory {
|
||||
|
||||
@@ -8,6 +8,7 @@ pub mod prompt_sections;
|
||||
pub mod reflection;
|
||||
pub mod schemas;
|
||||
pub mod tool_tracker;
|
||||
pub mod transcript_ingest;
|
||||
pub mod user_profile;
|
||||
|
||||
pub use prompt_sections::{LearnedContextSection, UserProfileSection};
|
||||
|
||||
@@ -0,0 +1,125 @@
|
||||
//! Dedupe transcript-derived candidates against what's already stored.
|
||||
//!
|
||||
//! Strategy: hash the normalised candidate content and embed the hash in
|
||||
//! the storage key (`<importance>.<kind>.<hash>`). Before persisting, we
|
||||
//! list the existing entries in the target namespace and skip any
|
||||
//! candidate whose key already exists. This is intentionally cheap — we
|
||||
//! do not call `recall` (semantic) for dedupe because a fresh chat's
|
||||
//! semantic recall would mask updates to the same fact.
|
||||
|
||||
use crate::openhuman::memory::Memory;
|
||||
|
||||
use super::persist;
|
||||
use super::types::{ConversationReflection, MemoryCandidate};
|
||||
|
||||
/// Stable, deterministic content fingerprint used for dedupe.
|
||||
///
|
||||
/// Lower-cased, whitespace-collapsed, then hashed via FxHash. We expose
|
||||
/// it as a hex string truncated to 12 chars — collisions on 48 bits are
|
||||
/// astronomically unlikely for a single workspace's transcript volume,
|
||||
/// and the short suffix keeps storage keys readable.
|
||||
pub fn content_hash(content: &str) -> String {
|
||||
let mut normalised = String::with_capacity(content.len());
|
||||
let mut last_was_space = false;
|
||||
for ch in content.trim().chars() {
|
||||
if ch.is_whitespace() {
|
||||
if !last_was_space {
|
||||
normalised.push(' ');
|
||||
last_was_space = true;
|
||||
}
|
||||
} else {
|
||||
for lower in ch.to_lowercase() {
|
||||
normalised.push(lower);
|
||||
}
|
||||
last_was_space = false;
|
||||
}
|
||||
}
|
||||
|
||||
// FNV-1a 64-bit. Tiny, deterministic, no extra dependency.
|
||||
let mut hash: u64 = 0xcbf2_9ce4_8422_2325;
|
||||
for byte in normalised.as_bytes() {
|
||||
hash ^= *byte as u64;
|
||||
hash = hash.wrapping_mul(0x100_0000_01b3);
|
||||
}
|
||||
format!("{:012x}", hash & 0x0000_ffff_ffff_ffff)
|
||||
}
|
||||
|
||||
/// Filter out candidates that already exist in the conversation-memory
|
||||
/// namespace. Returns `(kept, deduped_count)`.
|
||||
pub async fn filter_new(
|
||||
memory: &dyn Memory,
|
||||
candidates: Vec<MemoryCandidate>,
|
||||
) -> anyhow::Result<(Vec<MemoryCandidate>, usize)> {
|
||||
let existing = memory
|
||||
.list(
|
||||
Some(super::types::CONVERSATION_MEMORY_NAMESPACE),
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.unwrap_or_default();
|
||||
let existing_keys: std::collections::HashSet<String> =
|
||||
existing.into_iter().map(|e| e.key).collect();
|
||||
|
||||
let mut kept = Vec::with_capacity(candidates.len());
|
||||
let mut deduped = 0usize;
|
||||
let mut seen_in_batch: std::collections::HashSet<String> = std::collections::HashSet::new();
|
||||
for c in candidates {
|
||||
let key = persist::candidate_key(&c);
|
||||
if existing_keys.contains(&key) || !seen_in_batch.insert(key) {
|
||||
deduped += 1;
|
||||
continue;
|
||||
}
|
||||
kept.push(c);
|
||||
}
|
||||
Ok((kept, deduped))
|
||||
}
|
||||
|
||||
/// Filter out reflections that already exist.
|
||||
pub async fn filter_new_reflections(
|
||||
memory: &dyn Memory,
|
||||
reflections: Vec<ConversationReflection>,
|
||||
) -> anyhow::Result<(Vec<ConversationReflection>, usize)> {
|
||||
let existing = memory
|
||||
.list(
|
||||
Some(super::types::CONVERSATION_REFLECTIONS_NAMESPACE),
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.unwrap_or_default();
|
||||
let existing_keys: std::collections::HashSet<String> =
|
||||
existing.into_iter().map(|e| e.key).collect();
|
||||
|
||||
let mut kept = Vec::with_capacity(reflections.len());
|
||||
let mut deduped = 0usize;
|
||||
let mut seen_in_batch: std::collections::HashSet<String> = std::collections::HashSet::new();
|
||||
for r in reflections {
|
||||
let key = persist::reflection_key(&r);
|
||||
if existing_keys.contains(&key) || !seen_in_batch.insert(key) {
|
||||
deduped += 1;
|
||||
continue;
|
||||
}
|
||||
kept.push(r);
|
||||
}
|
||||
Ok((kept, deduped))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn content_hash_is_stable_under_whitespace_and_case() {
|
||||
let a = content_hash("I prefer Postgres for new services.");
|
||||
let b = content_hash(" i PREFER postgres for new services. ");
|
||||
assert_eq!(a, b);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn content_hash_differs_for_different_text() {
|
||||
let a = content_hash("I prefer Postgres for new services.");
|
||||
let b = content_hash("I prefer SQLite for new services.");
|
||||
assert_ne!(a, b);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,442 @@
|
||||
//! Heuristic extractor: scans the user/assistant messages of a session
|
||||
//! transcript and pulls out durable memory candidates plus higher-level
|
||||
//! reflections.
|
||||
//!
|
||||
//! Heuristic-only on purpose — see the module doc for [`super`]. The goal
|
||||
//! is high-precision extraction of *unmistakable* user statements
|
||||
//! (preferences, decisions, commitments, unresolved work, explicit
|
||||
//! self-reflections) so a fresh chat regains continuity without the
|
||||
//! pipeline ever calling out to a model.
|
||||
//!
|
||||
//! ## Filtering rules
|
||||
//!
|
||||
//! - User messages only for preferences/commitments — assistant text
|
||||
//! *can* echo a preference but is not authoritative.
|
||||
//! - Decisions and unresolved tasks may come from either side.
|
||||
//! - Filler messages (under [`MIN_USEFUL_CHARS`] chars after trimming, or
|
||||
//! matching [`is_filler`]) are skipped entirely.
|
||||
//! - Tool messages are never mined — they're high-noise and fully
|
||||
//! reconstructable from the transcript itself.
|
||||
|
||||
use crate::openhuman::providers::ChatMessage;
|
||||
|
||||
use super::types::{CandidateKind, ConversationReflection, Importance, MemoryCandidate};
|
||||
|
||||
/// Internal-to-the-module mirror of [`super::types::Provenance`] without
|
||||
/// `message_indices` — the per-candidate indices are filled in as we
|
||||
/// match each line.
|
||||
#[derive(Debug, Clone)]
|
||||
pub(super) struct Provenance {
|
||||
pub thread_id: Option<String>,
|
||||
pub transcript_path: String,
|
||||
pub transcript_basename: String,
|
||||
pub extracted_at: String,
|
||||
}
|
||||
|
||||
/// Below this length a message is treated as filler regardless of its
|
||||
/// content. Tuned empirically against short acks ("ok", "thanks!", "yes
|
||||
/// please") that otherwise survive the keyword filters.
|
||||
pub const MIN_USEFUL_CHARS: usize = 20;
|
||||
|
||||
/// Cap individual candidate snippets so a single rambling user turn
|
||||
/// can't dominate the prompt block on retrieval.
|
||||
pub const MAX_CANDIDATE_CHARS: usize = 400;
|
||||
|
||||
/// User-text patterns that indicate an explicit, durable preference.
|
||||
/// Case-insensitive substring match; ordering is informational only —
|
||||
/// the first match wins.
|
||||
const PREFERENCE_PHRASES: &[&str] = &[
|
||||
"i prefer",
|
||||
"i'd prefer",
|
||||
"i would prefer",
|
||||
"i like",
|
||||
"i don't like",
|
||||
"i hate",
|
||||
"i always",
|
||||
"i never",
|
||||
"please always",
|
||||
"please don't",
|
||||
"please do not",
|
||||
"from now on",
|
||||
"going forward",
|
||||
"i'd rather",
|
||||
"i would rather",
|
||||
"i want you to",
|
||||
];
|
||||
|
||||
/// Phrases that indicate a decision (either side may state these).
|
||||
const DECISION_PHRASES: &[&str] = &[
|
||||
"let's go with",
|
||||
"let's use",
|
||||
"we'll use",
|
||||
"we will use",
|
||||
"i'll use",
|
||||
"i will use",
|
||||
"decided to",
|
||||
"going with",
|
||||
"we're going to use",
|
||||
"we picked",
|
||||
"we chose",
|
||||
];
|
||||
|
||||
/// Phrases that indicate a commitment by the user (something they
|
||||
/// promised or planned to do).
|
||||
const COMMITMENT_PHRASES: &[&str] = &[
|
||||
"i'll ",
|
||||
"i will ",
|
||||
"i'm going to ",
|
||||
"i am going to ",
|
||||
"i plan to ",
|
||||
"i need to ",
|
||||
];
|
||||
|
||||
/// Phrases that indicate an open / unresolved task.
|
||||
const UNRESOLVED_PHRASES: &[&str] = &[
|
||||
"todo",
|
||||
"still need to",
|
||||
"haven't done",
|
||||
"have not done",
|
||||
"not done yet",
|
||||
"still pending",
|
||||
"blocked on",
|
||||
"waiting on",
|
||||
"follow up on",
|
||||
"needs follow-up",
|
||||
"next step",
|
||||
];
|
||||
|
||||
/// Phrases that indicate an explicit reflection / improvement signal.
|
||||
const REFLECTION_PHRASES: &[&str] = &[
|
||||
"i realized",
|
||||
"i realised",
|
||||
"lesson learned",
|
||||
"in hindsight",
|
||||
"next time",
|
||||
"remember that i",
|
||||
"remember that we",
|
||||
"we keep ",
|
||||
"we always end up ",
|
||||
"this is the second time",
|
||||
"this keeps happening",
|
||||
];
|
||||
|
||||
/// Generic filler patterns that should always be skipped even if a
|
||||
/// keyword matched — protects against false positives on reactions
|
||||
/// like "I like that, thanks!".
|
||||
const FILLER_PATTERNS: &[&str] = &[
|
||||
"thanks",
|
||||
"thank you",
|
||||
"thx",
|
||||
"ok cool",
|
||||
"sounds good",
|
||||
"got it",
|
||||
];
|
||||
|
||||
/// True when `msg` is too short or matches a known filler pattern.
|
||||
fn is_filler(msg: &str) -> bool {
|
||||
let trimmed = msg.trim();
|
||||
if trimmed.chars().count() < MIN_USEFUL_CHARS {
|
||||
return true;
|
||||
}
|
||||
let lower = trimmed.to_ascii_lowercase();
|
||||
// Pure-filler short messages: the whole message is essentially one
|
||||
// of the filler patterns.
|
||||
for pat in FILLER_PATTERNS {
|
||||
if lower == *pat || lower.trim_end_matches(['.', '!', '?']) == *pat {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
false
|
||||
}
|
||||
|
||||
/// Find the first matching phrase from `phrases` in `lower` (already
|
||||
/// lowercased) and return the substring of `original` starting at that
|
||||
/// match, truncated to [`MAX_CANDIDATE_CHARS`] and trimmed at the end of
|
||||
/// the sentence (`.`, `!`, `?`, or newline) where possible.
|
||||
fn find_phrase_snippet(original: &str, lower: &str, phrases: &[&str]) -> Option<String> {
|
||||
let mut best: Option<usize> = None;
|
||||
for phrase in phrases {
|
||||
if let Some(idx) = lower.find(phrase) {
|
||||
best = Some(best.map_or(idx, |b| b.min(idx)));
|
||||
}
|
||||
}
|
||||
let start = best?;
|
||||
// Walk back to the start of the containing sentence so the snippet
|
||||
// reads naturally (e.g. "I think I prefer X" rather than
|
||||
// "I prefer X").
|
||||
let prefix = &original[..start];
|
||||
let sentence_start = prefix
|
||||
.rfind(|c: char| matches!(c, '.' | '!' | '?' | '\n'))
|
||||
.map(|i| i + 1)
|
||||
.unwrap_or(0);
|
||||
|
||||
let tail = &original[sentence_start..];
|
||||
let mut end = tail.len();
|
||||
if let Some(rel) = tail.find(|c: char| matches!(c, '\n')) {
|
||||
end = end.min(rel);
|
||||
}
|
||||
if let Some(rel) = tail.find(['.', '!', '?']) {
|
||||
// Include the punctuation itself.
|
||||
end = end.min(rel + 1);
|
||||
}
|
||||
let snippet = tail[..end].trim();
|
||||
if snippet.is_empty() {
|
||||
return None;
|
||||
}
|
||||
let truncated: String = snippet.chars().take(MAX_CANDIDATE_CHARS).collect();
|
||||
Some(truncated)
|
||||
}
|
||||
|
||||
fn make_candidate(
|
||||
kind: CandidateKind,
|
||||
importance: Importance,
|
||||
content: String,
|
||||
idx: usize,
|
||||
prov: &Provenance,
|
||||
) -> MemoryCandidate {
|
||||
MemoryCandidate {
|
||||
kind,
|
||||
importance,
|
||||
content,
|
||||
provenance: super::types::Provenance {
|
||||
thread_id: prov.thread_id.clone(),
|
||||
transcript_path: prov.transcript_path.clone(),
|
||||
transcript_basename: prov.transcript_basename.clone(),
|
||||
message_indices: vec![idx],
|
||||
extracted_at: prov.extracted_at.clone(),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
/// Extract durable-fact candidates from a transcript.
|
||||
pub(super) fn extract_candidates(
|
||||
messages: &[ChatMessage],
|
||||
prov: &Provenance,
|
||||
) -> Vec<MemoryCandidate> {
|
||||
let mut out: Vec<MemoryCandidate> = Vec::new();
|
||||
|
||||
for (idx, msg) in messages.iter().enumerate() {
|
||||
if msg.role == "tool" || msg.role == "system" {
|
||||
continue;
|
||||
}
|
||||
if is_filler(&msg.content) {
|
||||
continue;
|
||||
}
|
||||
|
||||
let lower = msg.content.to_ascii_lowercase();
|
||||
let is_user = msg.role == "user";
|
||||
|
||||
// Preference / commitment: user-only. High importance — these
|
||||
// steer future agent behaviour.
|
||||
if is_user {
|
||||
if let Some(snippet) = find_phrase_snippet(&msg.content, &lower, PREFERENCE_PHRASES) {
|
||||
out.push(make_candidate(
|
||||
CandidateKind::Preference,
|
||||
Importance::High,
|
||||
snippet,
|
||||
idx,
|
||||
prov,
|
||||
));
|
||||
continue;
|
||||
}
|
||||
if let Some(snippet) = find_phrase_snippet(&msg.content, &lower, COMMITMENT_PHRASES) {
|
||||
out.push(make_candidate(
|
||||
CandidateKind::Commitment,
|
||||
Importance::Medium,
|
||||
snippet,
|
||||
idx,
|
||||
prov,
|
||||
));
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
// Decisions and unresolved tasks: either side may state these.
|
||||
if let Some(snippet) = find_phrase_snippet(&msg.content, &lower, DECISION_PHRASES) {
|
||||
out.push(make_candidate(
|
||||
CandidateKind::Decision,
|
||||
Importance::High,
|
||||
snippet,
|
||||
idx,
|
||||
prov,
|
||||
));
|
||||
continue;
|
||||
}
|
||||
if let Some(snippet) = find_phrase_snippet(&msg.content, &lower, UNRESOLVED_PHRASES) {
|
||||
out.push(make_candidate(
|
||||
CandidateKind::UnresolvedTask,
|
||||
Importance::Medium,
|
||||
snippet,
|
||||
idx,
|
||||
prov,
|
||||
));
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
out
|
||||
}
|
||||
|
||||
/// Extract higher-level reflections from a transcript.
|
||||
///
|
||||
/// Two sources today:
|
||||
///
|
||||
/// 1. **Explicit user reflections** — sentences containing one of the
|
||||
/// [`REFLECTION_PHRASES`]. Tagged `Importance::High` because the user
|
||||
/// has signalled they want this remembered.
|
||||
/// 2. **Repeated-pattern signal** — when the same preference / commitment
|
||||
/// phrase appears in three or more user messages across the transcript
|
||||
/// we surface it as a `recurring` reflection so the next session
|
||||
/// knows this is a stable pattern rather than a one-off remark.
|
||||
pub(super) fn extract_reflections(
|
||||
messages: &[ChatMessage],
|
||||
prov: &Provenance,
|
||||
) -> Vec<ConversationReflection> {
|
||||
let mut out: Vec<ConversationReflection> = Vec::new();
|
||||
|
||||
// Explicit reflections from the user.
|
||||
for (idx, msg) in messages.iter().enumerate() {
|
||||
if msg.role != "user" || is_filler(&msg.content) {
|
||||
continue;
|
||||
}
|
||||
let lower = msg.content.to_ascii_lowercase();
|
||||
if let Some(snippet) = find_phrase_snippet(&msg.content, &lower, REFLECTION_PHRASES) {
|
||||
out.push(ConversationReflection {
|
||||
importance: Importance::High,
|
||||
theme: "user_reflection".into(),
|
||||
detail: snippet,
|
||||
provenance: super::types::Provenance {
|
||||
thread_id: prov.thread_id.clone(),
|
||||
transcript_path: prov.transcript_path.clone(),
|
||||
transcript_basename: prov.transcript_basename.clone(),
|
||||
message_indices: vec![idx],
|
||||
extracted_at: prov.extracted_at.clone(),
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
// Recurring-preference detection: count how many user turns mention
|
||||
// any preference phrase. If ≥3, emit one recurring reflection
|
||||
// citing all matching message indices.
|
||||
let mut recurring_indices: Vec<usize> = Vec::new();
|
||||
for (idx, msg) in messages.iter().enumerate() {
|
||||
if msg.role != "user" || is_filler(&msg.content) {
|
||||
continue;
|
||||
}
|
||||
let lower = msg.content.to_ascii_lowercase();
|
||||
if PREFERENCE_PHRASES.iter().any(|p| lower.contains(p)) {
|
||||
recurring_indices.push(idx);
|
||||
}
|
||||
}
|
||||
if recurring_indices.len() >= 3 {
|
||||
out.push(ConversationReflection {
|
||||
importance: Importance::Medium,
|
||||
theme: "recurring_preferences".into(),
|
||||
detail: format!(
|
||||
"User stated personal preferences in {} messages this session — treat as a stable pattern, not a one-off.",
|
||||
recurring_indices.len()
|
||||
),
|
||||
provenance: super::types::Provenance {
|
||||
thread_id: prov.thread_id.clone(),
|
||||
transcript_path: prov.transcript_path.clone(),
|
||||
transcript_basename: prov.transcript_basename.clone(),
|
||||
message_indices: recurring_indices,
|
||||
extracted_at: prov.extracted_at.clone(),
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
out
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod inline_tests {
|
||||
use super::*;
|
||||
|
||||
fn prov() -> Provenance {
|
||||
Provenance {
|
||||
thread_id: Some("thr_abc".into()),
|
||||
transcript_path: "/tmp/session_raw/123_main.jsonl".into(),
|
||||
transcript_basename: "123_main.jsonl".into(),
|
||||
extracted_at: "2026-05-09T12:00:00Z".into(),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn skips_short_filler() {
|
||||
assert!(is_filler("ok"));
|
||||
assert!(is_filler("thanks!"));
|
||||
assert!(is_filler("hi"));
|
||||
assert!(!is_filler("I prefer Postgres for this kind of thing."));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn extracts_user_preference_as_high() {
|
||||
let msgs = vec![ChatMessage::user(
|
||||
"I prefer Postgres over MySQL for new services.",
|
||||
)];
|
||||
let cands = extract_candidates(&msgs, &prov());
|
||||
assert_eq!(cands.len(), 1);
|
||||
assert_eq!(cands[0].kind, CandidateKind::Preference);
|
||||
assert_eq!(cands[0].importance, Importance::High);
|
||||
assert!(cands[0].content.contains("Postgres"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn does_not_extract_preference_from_assistant() {
|
||||
let msgs = vec![ChatMessage::assistant(
|
||||
"You said earlier that I prefer Postgres for these.",
|
||||
)];
|
||||
let cands = extract_candidates(&msgs, &prov());
|
||||
assert!(
|
||||
cands.iter().all(|c| c.kind != CandidateKind::Preference),
|
||||
"assistant text must not produce Preference: {:?}",
|
||||
cands,
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn extracts_decision_from_either_side() {
|
||||
let msgs = vec![
|
||||
ChatMessage::user("Let's go with Postgres for the metadata store."),
|
||||
ChatMessage::assistant("Sure, going with Postgres."),
|
||||
];
|
||||
let cands = extract_candidates(&msgs, &prov());
|
||||
let decisions: Vec<_> = cands
|
||||
.iter()
|
||||
.filter(|c| c.kind == CandidateKind::Decision)
|
||||
.collect();
|
||||
assert!(
|
||||
!decisions.is_empty(),
|
||||
"should extract at least one decision"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn extracts_unresolved_task() {
|
||||
let msgs = vec![ChatMessage::user(
|
||||
"Still need to migrate the old auth service before Friday.",
|
||||
)];
|
||||
let cands = extract_candidates(&msgs, &prov());
|
||||
assert!(cands
|
||||
.iter()
|
||||
.any(|c| c.kind == CandidateKind::UnresolvedTask));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn captures_reflection_with_provenance_indices() {
|
||||
let msgs = vec![
|
||||
ChatMessage::user("Hello, can you help with the deploy?"),
|
||||
ChatMessage::assistant("Sure, what's broken?"),
|
||||
ChatMessage::user(
|
||||
"I realized our staging cluster is the bottleneck — \
|
||||
next time let's pre-warm it.",
|
||||
),
|
||||
];
|
||||
let refls = extract_reflections(&msgs, &prov());
|
||||
assert_eq!(refls.len(), 1);
|
||||
assert_eq!(refls[0].theme, "user_reflection");
|
||||
assert_eq!(refls[0].provenance.message_indices, vec![2]);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,157 @@
|
||||
//! Transcript-to-memory ingestion pipeline.
|
||||
//!
|
||||
//! Reads completed session transcripts (`session_raw/*.jsonl`) and extracts
|
||||
//! durable conversational memory plus higher-level reflections so that fresh
|
||||
//! chats can recover continuity from prior conversations. See issue #1399.
|
||||
//!
|
||||
//! ## Outputs
|
||||
//!
|
||||
//! Two distinct memory streams, each persisted via [`crate::openhuman::memory::Memory`]:
|
||||
//!
|
||||
//! - **Conversational memory** (`conversation_memory` namespace) — durable
|
||||
//! facts (preferences, decisions, commitments, unresolved tasks) tagged with
|
||||
//! importance + provenance pointing back at the source transcript.
|
||||
//! - **Conversational reflections** (`conversation_reflections` namespace) —
|
||||
//! higher-level patterns, recurring themes, or improvement signals.
|
||||
//!
|
||||
//! ## Pipeline
|
||||
//!
|
||||
//! ```text
|
||||
//! SessionTranscript → extract → dedupe → persist → IngestionReport
|
||||
//! ```
|
||||
//!
|
||||
//! Heuristic-only by design: the goal of the first pass is to make the
|
||||
//! pipeline available to the rest of the system *without* a hard LLM
|
||||
//! dependency, so it can run as a background task on session close, in tests,
|
||||
//! and on machines without provider credentials. A subsequent iteration can
|
||||
//! layer an LLM-driven extractor on the same trait surface.
|
||||
//!
|
||||
//! ## Provenance
|
||||
//!
|
||||
//! Every persisted entry carries enough metadata (`thread_id`, transcript
|
||||
//! basename, source message indices, RFC-3339 timestamp) to trace the memory
|
||||
//! back to the conversation it came from and to deduplicate repeats.
|
||||
|
||||
mod dedupe;
|
||||
mod extract;
|
||||
mod persist;
|
||||
pub mod types;
|
||||
|
||||
pub use types::{
|
||||
CandidateKind, ConversationReflection, Importance, IngestionReport, MemoryCandidate,
|
||||
Provenance, CONVERSATION_MEMORY_NAMESPACE, CONVERSATION_REFLECTIONS_NAMESPACE,
|
||||
};
|
||||
|
||||
use crate::openhuman::agent::harness::session::transcript::{self, SessionTranscript};
|
||||
use crate::openhuman::memory::Memory;
|
||||
use std::path::Path;
|
||||
|
||||
/// Ingest a single session transcript file: extract memory candidates,
|
||||
/// dedupe against what's already stored, and persist new entries.
|
||||
///
|
||||
/// Background-first: callers should invoke this from a `tokio::spawn` so
|
||||
/// chat latency is unaffected (see
|
||||
/// `Agent::spawn_transcript_ingestion`). Failures are returned but the
|
||||
/// caller should generally just log them — ingestion is best-effort and
|
||||
/// retried on the next transcript write.
|
||||
pub async fn ingest_transcript_path(
|
||||
memory: &dyn Memory,
|
||||
path: &Path,
|
||||
) -> anyhow::Result<IngestionReport> {
|
||||
log::debug!("[transcript_ingest] starting ingest for {}", path.display());
|
||||
let parsed = transcript::read_transcript(path)?;
|
||||
ingest_session_transcript(memory, &parsed, path).await
|
||||
}
|
||||
|
||||
/// Ingest an already-parsed [`SessionTranscript`].
|
||||
///
|
||||
/// Exposed separately from `ingest_transcript_path` so tests can drive the
|
||||
/// pipeline without touching the filesystem.
|
||||
pub async fn ingest_session_transcript(
|
||||
memory: &dyn Memory,
|
||||
transcript: &SessionTranscript,
|
||||
path: &Path,
|
||||
) -> anyhow::Result<IngestionReport> {
|
||||
let basename = path
|
||||
.file_name()
|
||||
.and_then(|s| s.to_str())
|
||||
.unwrap_or("unknown")
|
||||
.to_string();
|
||||
let path_display = path.display().to_string();
|
||||
let thread_id = transcript.meta.thread_id.clone();
|
||||
let now = chrono::Utc::now().to_rfc3339();
|
||||
|
||||
let extracted = extract::extract_candidates(
|
||||
&transcript.messages,
|
||||
&extract::Provenance {
|
||||
thread_id: thread_id.clone(),
|
||||
transcript_path: path_display.clone(),
|
||||
transcript_basename: basename.clone(),
|
||||
extracted_at: now.clone(),
|
||||
},
|
||||
);
|
||||
|
||||
let reflections = extract::extract_reflections(
|
||||
&transcript.messages,
|
||||
&extract::Provenance {
|
||||
thread_id: thread_id.clone(),
|
||||
transcript_path: path_display.clone(),
|
||||
transcript_basename: basename.clone(),
|
||||
extracted_at: now,
|
||||
},
|
||||
);
|
||||
|
||||
let extracted_total = extracted.len();
|
||||
let reflection_total = reflections.len();
|
||||
|
||||
let (kept, deduped) = dedupe::filter_new(memory, extracted).await?;
|
||||
let (kept_reflections, deduped_reflections) =
|
||||
dedupe::filter_new_reflections(memory, reflections).await?;
|
||||
|
||||
let mut stored = 0usize;
|
||||
for candidate in &kept {
|
||||
match persist::store_candidate(memory, candidate).await {
|
||||
Ok(()) => stored += 1,
|
||||
Err(err) => log::warn!(
|
||||
"[transcript_ingest] failed to persist candidate kind={:?} importance={:?}: {err}",
|
||||
candidate.kind,
|
||||
candidate.importance
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
let mut stored_reflections = 0usize;
|
||||
for reflection in &kept_reflections {
|
||||
match persist::store_reflection(memory, reflection).await {
|
||||
Ok(()) => stored_reflections += 1,
|
||||
Err(err) => log::warn!("[transcript_ingest] failed to persist reflection: {err}"),
|
||||
}
|
||||
}
|
||||
|
||||
log::info!(
|
||||
"[transcript_ingest] ingested {}: extracted={} stored={} deduped={} reflections={}/{} (deduped={}) thread={}",
|
||||
path.display(),
|
||||
extracted_total,
|
||||
stored,
|
||||
deduped,
|
||||
stored_reflections,
|
||||
reflection_total,
|
||||
deduped_reflections,
|
||||
thread_id.as_deref().unwrap_or("-"),
|
||||
);
|
||||
|
||||
Ok(IngestionReport {
|
||||
processed_messages: transcript.messages.len(),
|
||||
extracted: extracted_total,
|
||||
stored,
|
||||
deduped,
|
||||
reflections_extracted: reflection_total,
|
||||
reflections_stored: stored_reflections,
|
||||
candidates: kept,
|
||||
reflections: kept_reflections,
|
||||
})
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "tests.rs"]
|
||||
mod tests;
|
||||
@@ -0,0 +1,107 @@
|
||||
//! Persist memory candidates and reflections via the [`Memory`] trait.
|
||||
//!
|
||||
//! Storage format
|
||||
//! --------------
|
||||
//!
|
||||
//! Conversation memory entries:
|
||||
//! - **namespace**: [`super::types::CONVERSATION_MEMORY_NAMESPACE`]
|
||||
//! - **key**: `<importance>.<kind>.<hash12>` — the importance prefix lets
|
||||
//! the retrieval side prune to `high.*` cheaply, and the hash dedupes.
|
||||
//! - **content**: human-readable line followed by a single
|
||||
//! `[provenance] {…}` JSON line so retrievers can cite source.
|
||||
//!
|
||||
//! Reflections follow the same shape under
|
||||
//! [`super::types::CONVERSATION_REFLECTIONS_NAMESPACE`].
|
||||
|
||||
use crate::openhuman::memory::{Memory, MemoryCategory};
|
||||
|
||||
use super::dedupe::content_hash;
|
||||
use super::types::{
|
||||
ConversationReflection, MemoryCandidate, CONVERSATION_MEMORY_NAMESPACE,
|
||||
CONVERSATION_REFLECTIONS_NAMESPACE,
|
||||
};
|
||||
|
||||
/// Compute the storage key for a candidate. Public to the module so
|
||||
/// `dedupe` can reuse the exact same scheme.
|
||||
pub fn candidate_key(candidate: &MemoryCandidate) -> String {
|
||||
let hash = content_hash(&candidate.content);
|
||||
format!(
|
||||
"{}.{}.{}",
|
||||
candidate.importance.as_str(),
|
||||
candidate.kind.as_str(),
|
||||
hash
|
||||
)
|
||||
}
|
||||
|
||||
/// Compute the storage key for a reflection.
|
||||
pub fn reflection_key(reflection: &ConversationReflection) -> String {
|
||||
let hash = content_hash(&format!("{}::{}", reflection.theme, reflection.detail));
|
||||
format!(
|
||||
"{}.{}.{}",
|
||||
reflection.importance.as_str(),
|
||||
reflection.theme,
|
||||
hash
|
||||
)
|
||||
}
|
||||
|
||||
/// Render the human-readable + provenance content payload for a
|
||||
/// candidate.
|
||||
fn render_candidate_content(candidate: &MemoryCandidate) -> String {
|
||||
let prov_json =
|
||||
serde_json::to_string(&candidate.provenance).unwrap_or_else(|_| "{}".to_string());
|
||||
format!(
|
||||
"[{} {}] {}\n[provenance] {}",
|
||||
candidate.importance.as_str(),
|
||||
candidate.kind.as_str(),
|
||||
candidate.content,
|
||||
prov_json
|
||||
)
|
||||
}
|
||||
|
||||
fn render_reflection_content(reflection: &ConversationReflection) -> String {
|
||||
let prov_json =
|
||||
serde_json::to_string(&reflection.provenance).unwrap_or_else(|_| "{}".to_string());
|
||||
format!(
|
||||
"[{} {}] {}\n[provenance] {}",
|
||||
reflection.importance.as_str(),
|
||||
reflection.theme,
|
||||
reflection.detail,
|
||||
prov_json
|
||||
)
|
||||
}
|
||||
|
||||
pub async fn store_candidate(
|
||||
memory: &dyn Memory,
|
||||
candidate: &MemoryCandidate,
|
||||
) -> anyhow::Result<()> {
|
||||
let key = candidate_key(candidate);
|
||||
let content = render_candidate_content(candidate);
|
||||
let session_id = candidate.provenance.thread_id.as_deref();
|
||||
memory
|
||||
.store(
|
||||
CONVERSATION_MEMORY_NAMESPACE,
|
||||
&key,
|
||||
&content,
|
||||
MemoryCategory::Conversation,
|
||||
session_id,
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
pub async fn store_reflection(
|
||||
memory: &dyn Memory,
|
||||
reflection: &ConversationReflection,
|
||||
) -> anyhow::Result<()> {
|
||||
let key = reflection_key(reflection);
|
||||
let content = render_reflection_content(reflection);
|
||||
let session_id = reflection.provenance.thread_id.as_deref();
|
||||
memory
|
||||
.store(
|
||||
CONVERSATION_REFLECTIONS_NAMESPACE,
|
||||
&key,
|
||||
&content,
|
||||
MemoryCategory::Conversation,
|
||||
session_id,
|
||||
)
|
||||
.await
|
||||
}
|
||||
@@ -0,0 +1,271 @@
|
||||
//! Integration-style unit tests for the transcript ingestion pipeline.
|
||||
//!
|
||||
//! Uses an in-memory [`Memory`] mock so the pipeline can be exercised
|
||||
//! end-to-end without a SQLite/vector backend.
|
||||
|
||||
use super::*;
|
||||
use crate::openhuman::agent::harness::session::transcript::{SessionTranscript, TranscriptMeta};
|
||||
use crate::openhuman::memory::{Memory, MemoryCategory, MemoryEntry, NamespaceSummary, RecallOpts};
|
||||
use crate::openhuman::providers::ChatMessage;
|
||||
use async_trait::async_trait;
|
||||
use std::path::PathBuf;
|
||||
use std::sync::Mutex;
|
||||
|
||||
/// Tiny in-memory `Memory` implementation good enough to drive the
|
||||
/// transcript-ingest pipeline. Not exposed outside tests.
|
||||
struct InMemory {
|
||||
entries: Mutex<Vec<MemoryEntry>>,
|
||||
}
|
||||
|
||||
impl InMemory {
|
||||
fn new() -> Self {
|
||||
Self {
|
||||
entries: Mutex::new(Vec::new()),
|
||||
}
|
||||
}
|
||||
|
||||
fn snapshot(&self) -> Vec<MemoryEntry> {
|
||||
self.entries.lock().unwrap().clone()
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl Memory for InMemory {
|
||||
fn name(&self) -> &str {
|
||||
"in_memory_test"
|
||||
}
|
||||
|
||||
async fn store(
|
||||
&self,
|
||||
namespace: &str,
|
||||
key: &str,
|
||||
content: &str,
|
||||
category: MemoryCategory,
|
||||
session_id: Option<&str>,
|
||||
) -> anyhow::Result<()> {
|
||||
let mut e = self.entries.lock().unwrap();
|
||||
// Replace-on-collision so re-ingest is idempotent.
|
||||
if let Some(existing) = e
|
||||
.iter_mut()
|
||||
.find(|e| e.namespace.as_deref() == Some(namespace) && e.key == key)
|
||||
{
|
||||
existing.content = content.to_string();
|
||||
existing.timestamp = "2026-05-09T12:00:00Z".to_string();
|
||||
return Ok(());
|
||||
}
|
||||
e.push(MemoryEntry {
|
||||
id: format!("id-{}-{}", namespace, key),
|
||||
key: key.to_string(),
|
||||
content: content.to_string(),
|
||||
namespace: Some(namespace.to_string()),
|
||||
category,
|
||||
timestamp: "2026-05-09T12:00:00Z".to_string(),
|
||||
session_id: session_id.map(|s| s.to_string()),
|
||||
score: None,
|
||||
});
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn recall(
|
||||
&self,
|
||||
query: &str,
|
||||
limit: usize,
|
||||
opts: RecallOpts<'_>,
|
||||
) -> anyhow::Result<Vec<MemoryEntry>> {
|
||||
let q = query.to_ascii_lowercase();
|
||||
let entries = self.entries.lock().unwrap().clone();
|
||||
let mut hits: Vec<MemoryEntry> = entries
|
||||
.into_iter()
|
||||
.filter(|e| {
|
||||
opts.namespace
|
||||
.map(|n| e.namespace.as_deref() == Some(n))
|
||||
.unwrap_or(true)
|
||||
})
|
||||
.filter(|e| e.content.to_ascii_lowercase().contains(&q) || q.is_empty())
|
||||
.map(|mut e| {
|
||||
e.score = Some(1.0);
|
||||
e
|
||||
})
|
||||
.collect();
|
||||
hits.truncate(limit);
|
||||
Ok(hits)
|
||||
}
|
||||
|
||||
async fn get(&self, namespace: &str, key: &str) -> anyhow::Result<Option<MemoryEntry>> {
|
||||
Ok(self
|
||||
.entries
|
||||
.lock()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.find(|e| e.namespace.as_deref() == Some(namespace) && e.key == key)
|
||||
.cloned())
|
||||
}
|
||||
|
||||
async fn list(
|
||||
&self,
|
||||
namespace: Option<&str>,
|
||||
_category: Option<&MemoryCategory>,
|
||||
_session_id: Option<&str>,
|
||||
) -> anyhow::Result<Vec<MemoryEntry>> {
|
||||
Ok(self
|
||||
.entries
|
||||
.lock()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.filter(|e| {
|
||||
namespace
|
||||
.map(|n| e.namespace.as_deref() == Some(n))
|
||||
.unwrap_or(true)
|
||||
})
|
||||
.cloned()
|
||||
.collect())
|
||||
}
|
||||
|
||||
async fn forget(&self, _namespace: &str, _key: &str) -> anyhow::Result<bool> {
|
||||
Ok(false)
|
||||
}
|
||||
|
||||
async fn namespace_summaries(&self) -> anyhow::Result<Vec<NamespaceSummary>> {
|
||||
Ok(Vec::new())
|
||||
}
|
||||
|
||||
async fn count(&self) -> anyhow::Result<usize> {
|
||||
Ok(self.entries.lock().unwrap().len())
|
||||
}
|
||||
|
||||
async fn health_check(&self) -> bool {
|
||||
true
|
||||
}
|
||||
}
|
||||
|
||||
fn fake_meta(thread_id: Option<&str>) -> TranscriptMeta {
|
||||
TranscriptMeta {
|
||||
agent_name: "main".into(),
|
||||
dispatcher: "native".into(),
|
||||
created: "2026-05-09T11:00:00Z".into(),
|
||||
updated: "2026-05-09T12:00:00Z".into(),
|
||||
turn_count: 4,
|
||||
input_tokens: 0,
|
||||
output_tokens: 0,
|
||||
cached_input_tokens: 0,
|
||||
charged_amount_usd: 0.0,
|
||||
thread_id: thread_id.map(|s| s.into()),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn ingest_extracts_high_importance_preference_with_provenance() {
|
||||
let mem = InMemory::new();
|
||||
let transcript = SessionTranscript {
|
||||
meta: fake_meta(Some("thr_alpha")),
|
||||
messages: vec![
|
||||
ChatMessage::user("hi"),
|
||||
ChatMessage::assistant("hello"),
|
||||
ChatMessage::user("I prefer Postgres over MySQL for any new metadata service we ship."),
|
||||
ChatMessage::user("Still need to migrate the auth service before Friday."),
|
||||
],
|
||||
};
|
||||
|
||||
let report =
|
||||
ingest_session_transcript(&mem, &transcript, &PathBuf::from("/tmp/123_main.jsonl"))
|
||||
.await
|
||||
.expect("ingest must succeed");
|
||||
|
||||
assert!(report.extracted >= 2, "report: {:?}", report);
|
||||
assert!(report.stored >= 2);
|
||||
|
||||
let stored = mem.snapshot();
|
||||
assert!(stored.iter().any(
|
||||
|e| e.namespace.as_deref() == Some(CONVERSATION_MEMORY_NAMESPACE)
|
||||
&& e.key.starts_with("high.preference.")
|
||||
&& e.content.contains("Postgres")
|
||||
&& e.content.contains("[provenance]")
|
||||
&& e.content.contains("thr_alpha")
|
||||
));
|
||||
assert!(stored
|
||||
.iter()
|
||||
.any(|e| e.key.starts_with("med.unresolved_task.") && e.content.contains("Friday")));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn re_ingest_is_idempotent() {
|
||||
let mem = InMemory::new();
|
||||
let transcript = SessionTranscript {
|
||||
meta: fake_meta(Some("thr_beta")),
|
||||
messages: vec![ChatMessage::user(
|
||||
"I prefer Postgres for everything new — please default to it.",
|
||||
)],
|
||||
};
|
||||
let path = PathBuf::from("/tmp/200_main.jsonl");
|
||||
|
||||
let r1 = ingest_session_transcript(&mem, &transcript, &path)
|
||||
.await
|
||||
.unwrap();
|
||||
let r2 = ingest_session_transcript(&mem, &transcript, &path)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(r1.stored, 1);
|
||||
assert_eq!(r2.stored, 0, "second pass must dedupe everything");
|
||||
assert!(r2.deduped >= 1);
|
||||
assert_eq!(mem.snapshot().len(), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn ingest_captures_user_reflection_and_recurring_pattern() {
|
||||
let mem = InMemory::new();
|
||||
let transcript = SessionTranscript {
|
||||
meta: fake_meta(Some("thr_gamma")),
|
||||
messages: vec![
|
||||
ChatMessage::user("I prefer terse responses with no preamble."),
|
||||
ChatMessage::user("Going forward I want code-first answers."),
|
||||
ChatMessage::user("I always want bullet points when listing options."),
|
||||
ChatMessage::user(
|
||||
"I realized we keep reintroducing the same schema bug — \
|
||||
next time write a regression test first.",
|
||||
),
|
||||
],
|
||||
};
|
||||
|
||||
let report =
|
||||
ingest_session_transcript(&mem, &transcript, &PathBuf::from("/tmp/300_main.jsonl"))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert!(
|
||||
report.reflections_extracted >= 2,
|
||||
"expected at least one explicit + one recurring reflection: {:?}",
|
||||
report
|
||||
);
|
||||
assert!(report.reflections_stored >= 2);
|
||||
let stored = mem.snapshot();
|
||||
assert!(stored.iter().any(|e| e.namespace.as_deref()
|
||||
== Some(CONVERSATION_REFLECTIONS_NAMESPACE)
|
||||
&& e.key.contains("user_reflection")));
|
||||
assert!(stored.iter().any(|e| e.namespace.as_deref()
|
||||
== Some(CONVERSATION_REFLECTIONS_NAMESPACE)
|
||||
&& e.key.contains("recurring_preferences")));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn ingest_filters_low_signal_chatter() {
|
||||
let mem = InMemory::new();
|
||||
let transcript = SessionTranscript {
|
||||
meta: fake_meta(None),
|
||||
messages: vec![
|
||||
ChatMessage::user("ok"),
|
||||
ChatMessage::user("thanks!"),
|
||||
ChatMessage::assistant("👍"),
|
||||
ChatMessage::user("hi there"),
|
||||
],
|
||||
};
|
||||
|
||||
let report =
|
||||
ingest_session_transcript(&mem, &transcript, &PathBuf::from("/tmp/400_main.jsonl"))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(report.extracted, 0);
|
||||
assert_eq!(report.stored, 0);
|
||||
assert!(mem.snapshot().is_empty());
|
||||
}
|
||||
@@ -0,0 +1,110 @@
|
||||
//! Public types for the transcript-to-memory ingestion pipeline.
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
/// Memory namespace where transcript-derived durable facts live.
|
||||
///
|
||||
/// Kept distinct from `learning_observations` (turn-level reflection),
|
||||
/// `learning_reflections` (LLM-extracted user reflections) and
|
||||
/// `working.user.*` (sync-derived profile facts) so retrieval can target
|
||||
/// transcript-only memory without polluting other sources.
|
||||
pub const CONVERSATION_MEMORY_NAMESPACE: &str = "conversation_memory";
|
||||
|
||||
/// Memory namespace for transcript-derived higher-level reflections —
|
||||
/// patterns, repeated mistakes, opportunities. Surfaced through the
|
||||
/// subconscious / Intelligence UI rather than the prompt context block.
|
||||
pub const CONVERSATION_REFLECTIONS_NAMESPACE: &str = "conversation_reflections";
|
||||
|
||||
/// Importance tier — controls which memories are surfaced into a fresh
|
||||
/// chat by default. Only `High` candidates make it into the prompt block;
|
||||
/// `Medium` is retrievable on demand; `Low` is stored but never auto-
|
||||
/// surfaced (kept for audit / debugging).
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub enum Importance {
|
||||
Low,
|
||||
Medium,
|
||||
High,
|
||||
}
|
||||
|
||||
impl Importance {
|
||||
pub fn as_str(self) -> &'static str {
|
||||
match self {
|
||||
Self::Low => "low",
|
||||
Self::Medium => "med",
|
||||
Self::High => "high",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Discriminator for what a memory candidate represents. Drives the
|
||||
/// human-readable prefix on the stored content and downstream filtering.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub enum CandidateKind {
|
||||
Preference,
|
||||
Decision,
|
||||
Commitment,
|
||||
UnresolvedTask,
|
||||
Fact,
|
||||
}
|
||||
|
||||
impl CandidateKind {
|
||||
pub fn as_str(self) -> &'static str {
|
||||
match self {
|
||||
Self::Preference => "preference",
|
||||
Self::Decision => "decision",
|
||||
Self::Commitment => "commitment",
|
||||
Self::UnresolvedTask => "unresolved_task",
|
||||
Self::Fact => "fact",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Provenance metadata attached to every persisted memory.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct Provenance {
|
||||
/// Backend `thread_id` from the transcript meta header, if known.
|
||||
pub thread_id: Option<String>,
|
||||
/// Full transcript path (display form) — useful for debugging.
|
||||
pub transcript_path: String,
|
||||
/// Just the file basename (e.g. `1714000000_main.jsonl`) — included
|
||||
/// in the rendered content so readers don't see absolute paths.
|
||||
pub transcript_basename: String,
|
||||
/// Indices of the source messages within the transcript message
|
||||
/// array. A reflection or merged fact may cite multiple indices.
|
||||
pub message_indices: Vec<usize>,
|
||||
/// RFC-3339 timestamp of when the candidate was extracted.
|
||||
pub extracted_at: String,
|
||||
}
|
||||
|
||||
/// A memory candidate ready to persist.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct MemoryCandidate {
|
||||
pub kind: CandidateKind,
|
||||
pub importance: Importance,
|
||||
pub content: String,
|
||||
pub provenance: Provenance,
|
||||
}
|
||||
|
||||
/// A higher-level reflection extracted from a transcript window —
|
||||
/// patterns, recurring themes, repeated failures, improvement signals.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct ConversationReflection {
|
||||
pub importance: Importance,
|
||||
pub theme: String,
|
||||
pub detail: String,
|
||||
pub provenance: Provenance,
|
||||
}
|
||||
|
||||
/// Summary of one ingestion pass — surfaced in logs and returned to
|
||||
/// callers (mainly tests) for assertion.
|
||||
#[derive(Debug, Clone, Default)]
|
||||
pub struct IngestionReport {
|
||||
pub processed_messages: usize,
|
||||
pub extracted: usize,
|
||||
pub stored: usize,
|
||||
pub deduped: usize,
|
||||
pub reflections_extracted: usize,
|
||||
pub reflections_stored: usize,
|
||||
pub candidates: Vec<MemoryCandidate>,
|
||||
pub reflections: Vec<ConversationReflection>,
|
||||
}
|
||||
Reference in New Issue
Block a user