profit/lakehouse
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

pathway_memory: Mem0 versioning + deletion (upsert/revise/retire/history)

Browse Source 
Per J 2026-04-25: pathway_memory was append-only — every agent run added
a new trace, bad/failed runs polluted the matrix forever, no notion of
"this is the canonical evolved playbook." Ported playbook_memory's
Phase 25/27 patterns into pathway_memory so the agent loop's matrix
converges on best-known approaches per task class instead of bloating.

Fields added to PathwayTrace (all #[serde(default)] for back-compat):
- trace_uid: stable UUID per individual trace within a bucket
- version: u32 default 1
- parent_trace_uid, superseded_at, superseded_by_trace_uid
- retirement_reason (paired with existing retired:bool)

Methods added to PathwayMemory:
- upsert(trace) → PathwayUpsertOutcome {Added|Updated|Noop}
  Workflow-fingerprint dedup: ladder_attempts + final_verdict hash.
  Identical workflow → bumps existing replay_count instead of duplicating.
- revise(parent_uid, new_trace) → PathwayReviseOutcome
  Chains versions; rejects retired or already-superseded parents.
- retire(trace_uid, reason) → bool
  Marks specific trace retired with reason. Idempotent.
- history(trace_uid) → Vec<PathwayTrace>
  Walks parent_trace_uid back to root, then superseded_by forward to tip.
  Cycle-safe via visited set.

Retrieval gates updated:
- query_hot_swap skips superseded_at.is_some()
- bug_fingerprints_for skips both retired AND superseded

HTTP endpoints in service.rs:
- POST /vectors/pathway/upsert
- POST /vectors/pathway/retire
- POST /vectors/pathway/revise
- GET  /vectors/pathway/history/{trace_uid}

scripts/seal_agent_playbook.ts switched insert→upsert + accepts SESSION_DIR
arg so it can seal any archived session, not just iter4.

Verified live (4/4 ops):
- UPSERT first run: Added trace_uid 542ae53f
- UPSERT identical: Updated, replay_count bumped 0→1 (no duplicate)
- REVISE 542ae53f→87a70a61: parent stamped superseded_at, v2 created
- HISTORY of v2: chain_len=2, v1 superseded, v2 tip
- RETIRE iter-6 broken trace: retired=true, retirement_reason preserved
- pathway_memory.stats: total=79, retired=1, reuse_rate=0.0127

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
root
2026-04-25 19:31:44 -05:00
co-authored by Claude Opus 4.7
parent ed83754f20
commit 6ac7f61819
4 changed files with 890 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/vectord/src/pathway_memory.rs
+303
View File
@@ -217,8 +217,39 @@ pub struct PathwayTrace {
/// such that a retired pathway would work again, a new PathwayTrace
/// with a fresh id will be inserted — retirement is per-id.)
pub retired: bool,
// ─── Mem0 versioning + deletion (J 2026-04-25 directive, mirrors
// playbook_memory's Phase 25/27 patterns) ───
/// UUID for THIS specific trace. pathway_id is the bucket key
/// (shared by traces of the same task/file_prefix/signal); trace_uid
/// addresses an individual trace within that bucket so retire/revise
/// can target it precisely. Empty on legacy traces; populated by
/// upsert/insert callers (or filled with a generated UUID on insert).
#[serde(default)]
pub trace_uid: String,
/// Mem0-style version chain. v1 for original traces; bumped on
/// `revise()`. Legacy traces deserialize as version=1 via default.
#[serde(default = "default_version")]
pub version: u32,
/// trace_uid of the trace this one supersedes (None = root version).
#[serde(default)]
pub parent_trace_uid: Option<String>,
/// Set when a newer version supersedes this trace. Excluded from
/// retrieval (hot-swap, bug_fingerprints_for) once set.
#[serde(default)]
pub superseded_at: Option<String>,
/// trace_uid of the new version. Pairs with superseded_at.
#[serde(default)]
pub superseded_by_trace_uid: Option<String>,
/// Human-readable reason recorded with retire(). Pairs with
/// `retired: true`. Empty on probation-driven retirements (those
/// just set retired=true without a textual reason).
#[serde(default)]
pub retirement_reason: Option<String>,
}
fn default_version() -> u32 { 1 }
impl PathwayTrace {
/// Compute the narrow fingerprint id from task_class + file_prefix
/// + signal_class. `file_prefix` is the first path segment
@@ -349,6 +380,48 @@ pub struct HotSwapCandidate {
pub recommended_model: String,
}
/// Mem0-style outcome of an upsert. Mirrors playbook_memory::UpsertOutcome
/// but adapts the UPDATE semantic to PathwayTrace's bucket model: there
/// is no notion of merging endorsed_names — each trace is an immutable
/// run record. UPDATE here means "we found a non-retired non-superseded
/// trace with the same workflow shape; bumped its replay_count instead
/// of appending a duplicate." NOOP is reserved for the case where the
/// caller asked for an upsert that would change nothing observable.
#[derive(Debug, Serialize)]
pub enum PathwayUpsertOutcome {
Added { pathway_id: String, trace_uid: String },
Updated { pathway_id: String, trace_uid: String, replay_count: u32 },
Noop { pathway_id: String, trace_uid: String },
}
/// Mem0-style outcome of revise — chains versions across traces.
#[derive(Debug, Serialize)]
pub struct PathwayReviseOutcome {
pub parent_trace_uid: String,
pub parent_version: u32,
pub new_trace_uid: String,
pub new_version: u32,
pub superseded_at: String,
}
/// Compute a stable fingerprint for upsert dedup. Captures the
/// workflow shape: the sequence of (rung, model) pairs from
/// ladder_attempts, plus the final_verdict. Two traces with the same
/// fingerprint represent the same proven approach on the same task —
/// don't store duplicates.
fn workflow_fingerprint(trace: &PathwayTrace) -> String {
let mut h = Sha256::new();
h.update(trace.final_verdict.as_bytes());
h.update(b"|");
for a in &trace.ladder_attempts {
h.update(a.model.as_bytes());
h.update(b":");
h.update(a.rung.to_string().as_bytes());
h.update(b";");
}
format!("{:x}", h.finalize())
}
impl PathwayMemory {
pub fn new(store: Arc<dyn ObjectStore>) -> Self {
Self {
@@ -385,6 +458,10 @@ impl PathwayMemory {
if trace.pathway_vec.is_empty() {
trace.pathway_vec = build_pathway_vec(&trace);
}
if trace.trace_uid.is_empty() {
trace.trace_uid = uuid::Uuid::new_v4().to_string();
}
if trace.version == 0 { trace.version = 1; }
let mut s = self.state.write().await;
s.pathways
.entry(trace.pathway_id.clone())
@@ -395,6 +472,222 @@ impl PathwayMemory {
self.persist().await
}
/// Mem0-style upsert. ADD if no existing live trace in the bucket
/// matches this trace's workflow fingerprint. UPDATE (bump
/// replay_count) if a match exists. NOOP semantically equivalent
/// to UPDATE here — kept for symmetry with playbook_memory and
/// future-proofing if we add merge logic.
///
/// "Live" means: not retired, not superseded.
///
/// Replaces raw `insert` for callers that want dedup. Existing
/// `insert` callers (scrum_master) keep raw-append semantics so
/// behavior is back-compat.
pub async fn upsert(&self, mut trace: PathwayTrace) -> Result<PathwayUpsertOutcome, String> {
if trace.pathway_vec.is_empty() {
trace.pathway_vec = build_pathway_vec(&trace);
}
if trace.trace_uid.is_empty() {
trace.trace_uid = uuid::Uuid::new_v4().to_string();
}
if trace.version == 0 { trace.version = 1; }
let new_fp = workflow_fingerprint(&trace);
let mut s = self.state.write().await;
let bucket = s.pathways.entry(trace.pathway_id.clone()).or_default();
// Find a live trace (not retired, not superseded) with same workflow.
let mut existing_idx: Option<usize> = None;
for (i, t) in bucket.iter().enumerate() {
if t.retired { continue; }
if t.superseded_at.is_some() { continue; }
if workflow_fingerprint(t) == new_fp {
existing_idx = Some(i);
break;
}
}
let pathway_id = trace.pathway_id.clone();
let outcome = match existing_idx {
None => {
let trace_uid = trace.trace_uid.clone();
bucket.push(trace);
PathwayUpsertOutcome::Added { pathway_id, trace_uid }
}
Some(i) => {
// UPDATE: bump replay counters on the existing trace
// instead of duplicating. Replays_succeeded only bumps
// on accepted final_verdict (mirror record_replay logic).
let existing = &mut bucket[i];
existing.replay_count = existing.replay_count.saturating_add(1);
if trace.final_verdict == "accepted" {
existing.replays_succeeded = existing.replays_succeeded.saturating_add(1);
}
PathwayUpsertOutcome::Updated {
pathway_id,
trace_uid: existing.trace_uid.clone(),
replay_count: existing.replay_count,
}
}
};
s.last_updated_at = Utc::now().timestamp_millis();
drop(s);
self.persist().await?;
Ok(outcome)
}
/// Mem0-style retire. Marks a specific trace (by trace_uid) retired
/// with a human-readable reason. Retired traces are excluded from
/// hot-swap and bug_fingerprints retrieval. Idempotent: retiring an
/// already-retired trace returns Ok(false) without modification.
pub async fn retire(&self, trace_uid: &str, reason: &str) -> Result<bool, String> {
let mut touched = false;
{
let mut s = self.state.write().await;
'outer: for traces in s.pathways.values_mut() {
for t in traces.iter_mut() {
if t.trace_uid == trace_uid && !t.retired {
t.retired = true;
t.retirement_reason = Some(reason.to_string());
touched = true;
break 'outer;
}
}
}
if touched {
s.last_updated_at = Utc::now().timestamp_millis();
}
}
if touched { self.persist().await?; }
Ok(touched)
}
/// Mem0-style revise. Supersedes parent trace, chains the new
/// version. New version inherits parent_trace_uid; parent gets
/// superseded_at + superseded_by_trace_uid stamped. Rejects if
/// parent is retired or already superseded (revise the tip, not
/// the middle of the chain).
pub async fn revise(
&self,
parent_trace_uid: &str,
mut new_trace: PathwayTrace,
) -> Result<PathwayReviseOutcome, String> {
let now = Utc::now().to_rfc3339();
if new_trace.pathway_vec.is_empty() {
new_trace.pathway_vec = build_pathway_vec(&new_trace);
}
if new_trace.trace_uid.is_empty() {
new_trace.trace_uid = uuid::Uuid::new_v4().to_string();
}
let mut s = self.state.write().await;
// Locate parent across all buckets
let mut parent_loc: Option<(String, usize)> = None;
for (bucket_key, traces) in s.pathways.iter() {
for (i, t) in traces.iter().enumerate() {
if t.trace_uid == parent_trace_uid {
parent_loc = Some((bucket_key.clone(), i));
break;
}
}
if parent_loc.is_some() { break; }
}
let (parent_bucket, parent_idx) = parent_loc
.ok_or_else(|| format!("parent trace_uid '{parent_trace_uid}' not found"))?;
// Validate parent state
{
let parent = &s.pathways[&parent_bucket][parent_idx];
if parent.retired {
return Err(format!(
"cannot revise retired trace '{parent_trace_uid}' — retirement is terminal"
));
}
if parent.superseded_at.is_some() {
return Err(format!(
"trace '{parent_trace_uid}' already superseded; revise the tip of the chain"
));
}
}
let parent_version = s.pathways[&parent_bucket][parent_idx].version;
let new_version = parent_version.saturating_add(1);
let new_uid = new_trace.trace_uid.clone();
new_trace.version = new_version;
new_trace.parent_trace_uid = Some(parent_trace_uid.to_string());
new_trace.superseded_at = None;
new_trace.superseded_by_trace_uid = None;
// Stamp parent
{
let parent_mut = &mut s.pathways.get_mut(&parent_bucket).unwrap()[parent_idx];
parent_mut.superseded_at = Some(now.clone());
parent_mut.superseded_by_trace_uid = Some(new_uid.clone());
}
// Append new version (same bucket if same pathway_id)
s.pathways
.entry(new_trace.pathway_id.clone())
.or_default()
.push(new_trace);
s.last_updated_at = Utc::now().timestamp_millis();
drop(s);
self.persist().await?;
Ok(PathwayReviseOutcome {
parent_trace_uid: parent_trace_uid.to_string(),
parent_version,
new_trace_uid: new_uid,
new_version,
superseded_at: now,
})
}
/// Walk the version chain containing trace_uid. Returns root→tip.
/// Empty if trace_uid not found. Cycle-safe.
pub async fn history(&self, trace_uid: &str) -> Vec<PathwayTrace> {
let s = self.state.read().await;
// Build trace_uid → trace map across all buckets
let mut by_uid: HashMap<String, PathwayTrace> = HashMap::new();
for traces in s.pathways.values() {
for t in traces {
if !t.trace_uid.is_empty() {
by_uid.insert(t.trace_uid.clone(), t.clone());
}
}
}
let Some(seed) = by_uid.get(trace_uid).cloned() else {
return Vec::new();
};
// Walk back to root
let mut visited: std::collections::HashSet<String> = std::collections::HashSet::new();
let mut root = seed.clone();
while let Some(parent_id) = root.parent_trace_uid.clone() {
if !visited.insert(parent_id.clone()) { break; }
match by_uid.get(&parent_id) {
Some(p) => root = p.clone(),
None => break,
}
}
// Walk forward to tip
let mut chain = vec![root.clone()];
let mut visited_fwd: std::collections::HashSet<String> = std::collections::HashSet::new();
visited_fwd.insert(root.trace_uid.clone());
let mut current = root;
while let Some(succ) = current.superseded_by_trace_uid.clone() {
if !visited_fwd.insert(succ.clone()) { break; }
match by_uid.get(&succ) {
Some(n) => {
chain.push(n.clone());
current = n.clone();
}
None => break,
}
}
chain
}
/// Query for a hot-swap candidate. Returns `None` if no eligible
/// pathway exists — caller should run the full ladder. Returns
/// `Some(cand)` if all gates pass — caller can short-circuit to
@@ -422,6 +715,11 @@ impl PathwayMemory {
if p.retired {
continue;
}
// Mem0 versioning: superseded traces are excluded from
// retrieval — only the tip of each version chain counts.
if p.superseded_at.is_some() {
continue;
}
// audit_consensus gate: explicit FAIL blocks hot-swap. A null
// audit_consensus (auditor hasn't seen this pathway yet) is
// NOT a block — the success_rate gate below still requires
@@ -523,6 +821,11 @@ impl PathwayMemory {
// are semantically equivalent within a pattern_key by design).
let mut agg: HashMap<(String, String), (SemanticFlag, String, u32)> = HashMap::new();
for t in traces {
// Mem0 versioning: skip retired + superseded traces so
// their bug patterns don't leak into future retrievals.
if t.retired || t.superseded_at.is_some() {
continue;
}
for bp in &t.bug_fingerprints {
let key = (format!("{:?}", bp.flag), bp.pattern_key.clone());
let entry = agg.entry(key).or_insert_with(|| {
crates/vectord/src/service.rs
+57
View File
@@ -157,6 +157,11 @@ pub fn router(state: VectorState) -> Router {
.route("/pathway/stats", get(pathway_stats))
// ADR-021 Phase C: pre-review bug-fingerprint retrieval.
.route("/pathway/bug_fingerprints", post(pathway_bug_fingerprints))
// Mem0 ops (J 2026-04-25): upsert/retire/revise/history.
.route("/pathway/upsert", post(pathway_upsert))
.route("/pathway/retire", post(pathway_retire))
.route("/pathway/revise", post(pathway_revise))
.route("/pathway/history/{trace_uid}", get(pathway_history))
.with_state(state)
}
@@ -2904,6 +2909,58 @@ async fn pathway_bug_fingerprints(
Json(json!({ "fingerprints": fps }))
}
// ─── Mem0 ops endpoints (J 2026-04-25) ───
async fn pathway_upsert(
State(state): State<VectorState>,
Json(trace): Json<pathway_memory::PathwayTrace>,
) -> impl IntoResponse {
match state.pathway_memory.upsert(trace).await {
Ok(outcome) => Ok(Json(json!({"ok": true, "outcome": outcome}))),
Err(e) => Err((StatusCode::INTERNAL_SERVER_ERROR, e)),
}
}
#[derive(Deserialize)]
struct PathwayRetireRequest {
trace_uid: String,
reason: String,
}
async fn pathway_retire(
State(state): State<VectorState>,
Json(req): Json<PathwayRetireRequest>,
) -> impl IntoResponse {
match state.pathway_memory.retire(&req.trace_uid, &req.reason).await {
Ok(touched) => Ok(Json(json!({"ok": true, "retired": touched}))),
Err(e) => Err((StatusCode::INTERNAL_SERVER_ERROR, e)),
}
}
#[derive(Deserialize)]
struct PathwayReviseRequest {
parent_trace_uid: String,
new_trace: pathway_memory::PathwayTrace,
}
async fn pathway_revise(
State(state): State<VectorState>,
Json(req): Json<PathwayReviseRequest>,
) -> impl IntoResponse {
match state.pathway_memory.revise(&req.parent_trace_uid, req.new_trace).await {
Ok(outcome) => Ok(Json(json!({"ok": true, "outcome": outcome}))),
Err(e) => Err((StatusCode::INTERNAL_SERVER_ERROR, e)),
}
}
async fn pathway_history(
State(state): State<VectorState>,
axum::extract::Path(trace_uid): axum::extract::Path<String>,
) -> impl IntoResponse {
let chain = state.pathway_memory.history(&trace_uid).await;
Json(json!({"trace_uid": trace_uid, "chain_len": chain.len(), "chain": chain}))
}
#[cfg(test)]
mod extractor_tests {
use super::*;