fix: stabilize storage, memory, LSP, and Hermes LCM

src/agents/hermes/templates/plugin_init.py

@@ -2243,6 +2243,116 @@ def _replay_message_list(value):
             return None
     return list(value)
 
+def _normalize_replay_tool_pairs(messages):
+    """Return a replay containing only structurally paired tool events.
+
+    The returned issue list is intentionally separate from the normalized
+    candidate: callers must reject any replay that required repair instead of
+    silently persisting a transcript different from the one TraceDecay built.
+    """
+    issues = []
+    normalized = []
+    index = 0
+    messages = list(messages or [])
+    while index < len(messages):
+        message = messages[index]
+        if not isinstance(message, dict):
+            index += 1
+            continue
+        if message.get("role") == "tool":
+            issues.append({
+                "code": "orphan_tool_result",
+                "tool_call_id": str(message.get("tool_call_id") or "").strip(),
+                "message_index": index,
+            })
+            index += 1
+            continue
+        tool_calls = message.get("tool_calls")
+        if tool_calls is None:
+            normalized.append(message)
+            index += 1
+            continue
+        if message.get("role") != "assistant" or not isinstance(tool_calls, list) or not tool_calls:
+            issues.append({"code": "invalid_tool_calls", "message_index": index})
+            cleaned = dict(message)
+            cleaned.pop("tool_calls", None)
+            if str(cleaned.get("content") or "").strip():
+                normalized.append(cleaned)
+            index += 1
+            continue
+
+        call_ids = [_summary_tool_call_id(call) for call in tool_calls]
+        result_end = index + 1
+        while result_end < len(messages) and isinstance(messages[result_end], dict) \
+                and messages[result_end].get("role") == "tool":
+            result_end += 1
+        results = messages[index + 1:result_end]
+        result_ids = [str(result.get("tool_call_id") or "").strip() for result in results]
+        trailing_open = (
+            result_end == len(messages)
+            and not results
+            and all(call_ids)
+            and len(set(call_ids)) == len(call_ids)
+        )
+        if trailing_open:
+            normalized.append(message)
+            index = result_end
+            continue
+        complete = (
+            all(call_ids)
+            and len(set(call_ids)) == len(call_ids)
+            and all(result_ids)
+            and len(set(result_ids)) == len(result_ids)
+            and set(result_ids) == set(call_ids)
+            and len(results) == len(tool_calls)
+        )
+        if complete:
+            normalized.append(message)
+            normalized.extend(results)
+        else:
+            for tool_call_id in sorted(set(call_ids) | set(result_ids)):
+                issues.append({
+                    "code": "unmatched_tool_pair",
+                    "tool_call_id": tool_call_id,
+                    "call_count": call_ids.count(tool_call_id),
+                    "result_count": result_ids.count(tool_call_id),
+                    "message_index": index,
+                })
+            cleaned = dict(message)
+            cleaned.pop("tool_calls", None)
+            if str(cleaned.get("content") or "").strip():
+                normalized.append(cleaned)
+        index = result_end
+    return normalized, issues
+
+def _compression_boundary_abort(
+    result,
+    code,
+    *,
+    source_token_estimate,
+    replay_token_estimate,
+    reported_source_tokens=None,
+    reported_replay_tokens=None,
+    issues=None,
+):
+    payload = dict(result) if isinstance(result, dict) else {}
+    raw_replay = payload.pop("replay_messages", None)
+    payload["status"] = "aborted"
+    payload["reason"] = code
+    payload["replay_messages"] = []
+    payload["compression_diagnostic"] = {
+        "type": "replay_boundary_rejection",
+        "code": code,
+        "defer_preflight_to_real_usage": True,
+        "source_token_estimate": source_token_estimate,
+        "replay_token_estimate": replay_token_estimate,
+        "reported_source_tokens": reported_source_tokens,
+        "reported_replay_tokens": reported_replay_tokens,
+        "rejected_replay_message_count": len(raw_replay) if isinstance(raw_replay, list) else 0,
+        "issues": list(issues or []),
+    }
+    return payload
+
 def _compression_replay_is_compacted(result):
     if not isinstance(result, dict):
         return False
@@ -2911,7 +3021,17 @@ def has_content_to_compress(self, messages, current_tokens=None, **kwargs):
         return len(non_empty) >= 2
 
     def should_defer_preflight_to_real_usage(self, rough_tokens=None):
-        if not (
+        diagnostic = (
+            self.last_compress_result.get("compression_diagnostic")
+            if isinstance(self.last_compress_result, dict)
+            else None
+        )
+        replay_boundary_deferred = bool(
+            isinstance(diagnostic, dict)
+            and diagnostic.get("type") == "replay_boundary_rejection"
+            and diagnostic.get("defer_preflight_to_real_usage")
+        )
+        if not replay_boundary_deferred and not (
             isinstance(self.last_compress_result, dict)
             and self.last_compress_result.get("status") == "compressed"
         ):
@@ -3488,6 +3608,51 @@ def compress(self, messages, current_tokens=None, focus_topic=None, **kwargs):
             return original
         if replay == original:
             return original
+        normalized_replay, replay_issues = _normalize_replay_tool_pairs(replay)
+        source_token_estimate = _count_messages_tokens(original)
+        replay_token_estimate = _count_messages_tokens(normalized_replay)
+        try:
+            reported_source_tokens = int(current_tokens) if current_tokens is not None else None
+        except (TypeError, ValueError):
+            reported_source_tokens = None
+        try:
+            reported_replay_tokens = int(result.get("replay_token_estimate"))
+        except (TypeError, ValueError):
+            reported_replay_tokens = None
+        has_reported_estimates = (
+            reported_source_tokens is not None
+            and reported_source_tokens > 0
+            and reported_replay_tokens is not None
+        )
+        boundary_code = None
+        if replay_issues:
+            boundary_code = "invalid_replay_tool_pairs"
+        elif (
+            has_reported_estimates
+            and reported_replay_tokens >= reported_source_tokens
+        ):
+            boundary_code = "non_shrinking_reported_replay"
+        elif (
+            not has_reported_estimates
+            and source_token_estimate > 0
+            and replay_token_estimate >= source_token_estimate
+        ):
+            boundary_code = "non_shrinking_replay"
+        if boundary_code is not None:
+            result = _compression_boundary_abort(
+                result,
+                boundary_code,
+                source_token_estimate=source_token_estimate,
+                replay_token_estimate=replay_token_estimate,
+                reported_source_tokens=reported_source_tokens,
+                reported_replay_tokens=reported_replay_tokens,
+                issues=replay_issues,
+            )
+            self.last_compress_result = result
+            self._last_compress_aborted = True
+            self._last_summary_error = boundary_code
+            return original
+        replay = normalized_replay
         if replay != original:
             self.compression_count += 1
         return replay

src/mcp/tools/handlers/analytics.rs

@@ -442,7 +442,6 @@ async fn facts_section(
             }
         };
     let query = target
-        .db
         .conn()
         .query(
             "SELECT COUNT(*),

src/mcp/tools/handlers/memory.rs

@@ -28,19 +28,33 @@ use super::support::{
 const DEFAULT_FACT_LIMIT: usize = 20;
 const MAX_FACT_LIMIT: usize = 200;
 
-pub(super) struct TargetMemoryDb {
-    pub(super) db: Database,
+enum TargetMemoryDbHandle<'a> {
+    Active(&'a Database),
+    Owned(Box<Database>),
+}
+
+pub(super) struct TargetMemoryDb<'a> {
+    db: TargetMemoryDbHandle<'a>,
     pub(super) project_root: PathBuf,
     pub(super) user_scope: bool,
 }
 
+impl TargetMemoryDb<'_> {
+    pub(super) fn conn(&self) -> &libsql::Connection {
+        match &self.db {
+            TargetMemoryDbHandle::Active(db) => db.conn(),
+            TargetMemoryDbHandle::Owned(db) => db.conn(),
+        }
+    }
+}
+
 fn requests_user_memory(args: &Value) -> bool {
     args.get("memory_scope").and_then(Value::as_str) == Some("user")
 }
 
-async fn open_user_memory_target(profile_root: &Path) -> Result<TargetMemoryDb> {
+async fn open_user_memory_target(profile_root: &Path) -> Result<TargetMemoryDb<'static>> {
     Ok(TargetMemoryDb {
-        db: open_user_memory_db(profile_root).await?,
+        db: TargetMemoryDbHandle::Owned(Box::new(open_user_memory_db(profile_root).await?)),
         project_root: profile_root.to_path_buf(),
         user_scope: true,
     })
@@ -65,12 +79,12 @@ fn rendered_fact_store(project_root: Option<&Path>, args: &Value, value: &Value)
     text_tool_result(&text)
 }
 
-pub(super) async fn open_target_memory_db(
-    cg: &TraceDecay,
+pub(super) async fn open_target_memory_db<'a>(
+    cg: &'a TraceDecay,
     args: &Value,
     global_db: Option<&GlobalDb>,
     allow_default_registry_fallback: bool,
-) -> Result<TargetMemoryDb> {
+) -> Result<TargetMemoryDb<'a>> {
     if requests_user_memory(args) {
         if project_selector_present(args, &["project_path"]) {
             return Err(config_error(
@@ -88,8 +102,13 @@ pub(super) async fn open_target_memory_db(
     )
     .await?
     else {
+        let db = if cg.db_path() == cg.store_layout().graph_db_path {
+            TargetMemoryDbHandle::Active(cg.db())
+        } else {
+            TargetMemoryDbHandle::Owned(Box::new(cg.open_project_store_db().await?))
+        };
         return Ok(TargetMemoryDb {
-            db: cg.open_project_store_db().await?,
+            db,
             project_root: cg.project_root().to_path_buf(),
             user_scope: false,
         });
@@ -116,7 +135,7 @@ pub(super) async fn open_target_memory_db(
     }
     let (db, _) = Database::open(&db_path).await?;
     Ok(TargetMemoryDb {
-        db,
+        db: TargetMemoryDbHandle::Owned(Box::new(db)),
         project_root: PathBuf::from(context.project.display_root),
         user_scope: false,
     })
@@ -317,10 +336,10 @@ pub(super) async fn handle_fact_store(
 async fn handle_fact_store_for_target(
     args: Value,
     cross_project_selector: bool,
-    target_memory: TargetMemoryDb,
+    target_memory: TargetMemoryDb<'_>,
 ) -> Result<ToolResult> {
     let action = required_str(&args, "action")?;
-    let conn = target_memory.db.conn();
+    let conn = target_memory.conn();
     let store = MemoryStore::new(conn);
     let mut refresh_digest = false;
     let out = match action {
@@ -548,7 +567,7 @@ pub(super) async fn handle_fact_feedback(
         .map(ToOwned::to_owned);
     let target_memory =
         open_target_memory_db(cg, &args, global_db, allow_default_registry_fallback).await?;
-    let result = MemoryStore::new(target_memory.db.conn())
+    let result = MemoryStore::new(target_memory.conn())
         .record_feedback_event(FeedbackRequest {
             fact_id: fact_id(&args)?,
             action: feedback_action(&args)?,
@@ -561,7 +580,7 @@ pub(super) async fn handle_fact_feedback(
         .await?;
     if !target_memory.user_scope {
         refresh_memory_digest_after_memory_change(
-            target_memory.db.conn(),
+            target_memory.conn(),
             &target_memory.project_root,
         )
         .await;
@@ -582,7 +601,7 @@ pub(super) async fn handle_memory_status(
 ) -> Result<ToolResult> {
     let target_memory =
         open_target_memory_db(cg, &args, global_db, allow_default_registry_fallback).await?;
-    let status = TraceDecay::memory_status_for_conn(target_memory.db.conn()).await?;
+    let status = TraceDecay::memory_status_for_conn(target_memory.conn()).await?;
     let value = json!({ "status": "ok", "memory": status });
     Ok(rendered_tool_json(
         (!target_memory.user_scope).then_some(target_memory.project_root.as_path()),
@@ -618,7 +637,7 @@ pub async fn handle_user_memory_tool(
                 .or_else(|| args.get("reason"))
                 .and_then(Value::as_str)
                 .map(ToOwned::to_owned);
-            let result = MemoryStore::new(target_memory.db.conn())
+            let result = MemoryStore::new(target_memory.conn())
                 .record_feedback_event(FeedbackRequest {
                     fact_id: fact_id(&args)?,
                     action: feedback_action(&args)?,
@@ -636,7 +655,7 @@ pub async fn handle_user_memory_tool(
             ))
         }
         "tracedecay_memory_status" => {
-            let status = TraceDecay::memory_status_for_conn(target_memory.db.conn()).await?;
+            let status = TraceDecay::memory_status_for_conn(target_memory.conn()).await?;
             Ok(rendered_tool_json(
                 None,
                 &args,
@@ -646,3 +665,32 @@ pub async fn handle_user_memory_tool(
         other => Err(config_error(format!("{other} is not a user-memory tool"))),
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn active_project_memory_uses_the_served_database_handle() {
+        let tmp = tempfile::tempdir().unwrap();
+        let project_root = tmp.path().join("project");
+        let profile_root = tmp.path().join("profile");
+        std::fs::create_dir_all(&project_root).unwrap();
+        let cg = TraceDecay::init_with_options(
+            &project_root,
+            crate::tracedecay::TraceDecayOpenOptions {
+                global_db_path: Some(profile_root.join("global.db")),
+                profile_root: Some(profile_root),
+            },
+        )
+        .await
+        .unwrap();
+
+        let target = open_target_memory_db(&cg, &json!({}), None, true)
+            .await
+            .unwrap();
+
+        assert!(matches!(target.db, TargetMemoryDbHandle::Active(_)));
+        assert!(std::ptr::eq(target.conn(), cg.db().conn()));
+    }
+}