Mnemex

Instructions

Promote high-value memories to long-term storage.

Memories with high scores or frequent usage are promoted to the Obsidian
vault (or other long-term storage) where they become permanent.

Args:
    memory_id: Specific memory ID to promote.
    auto_detect: Automatically detect promotion candidates.
    dry_run: Preview what would be promoted without promoting.
    target: Target for promotion (default: "obsidian").
    force: Force promotion even if criteria not met.

Returns:
    List of promoted memories and promotion statistics.

Implementation Reference

• src/cortexgraph/tools/promote.py:16-158 (handler)

  The promote_memory tool handler function, decorated with @mcp.tool() for registration. Handles promotion of high-value memories to long-term storage like Obsidian vault, with options for specific memory ID, auto-detection, dry-run, and force promotion.

  @mcp.tool()
  def promote_memory(
      memory_id: str | None = None,
      auto_detect: bool = False,
      dry_run: bool = False,
      target: str = "obsidian",
      force: bool = False,
  ) -> dict[str, Any]:
      """
      Promote high-value memories to long-term storage.
  
      Memories with high scores or frequent usage are promoted to the Obsidian
      vault (or other long-term storage) where they become permanent.
  
      Args:
          memory_id: Specific memory ID to promote (valid UUID).
          auto_detect: Automatically detect promotion candidates.
          dry_run: Preview what would be promoted without promoting.
          target: Storage backend for promotion. Default: "obsidian" (Obsidian-compatible markdown).
                  Note: This is a storage format, not a file path. Path configured via LTM_VAULT_PATH.
          force: Force promotion even if criteria not met.
  
      Returns:
          List of promoted memories and promotion statistics.
  
      Raises:
          ValueError: If memory_id is invalid or target is not supported.
      """
      # Input validation
      if memory_id is not None:
          memory_id = validate_uuid(memory_id, "memory_id")
  
      target = validate_target(target, "target")
  
      now = int(time.time())
      promoted_ids = []
      candidates = []
  
      if memory_id:
          memory = db.get_memory(memory_id)
          if memory is None:
              return {"success": False, "message": f"Memory not found: {memory_id}"}
          if memory.status == MemoryStatus.PROMOTED:
              return {
                  "success": False,
                  "message": f"Memory already promoted: {memory_id}",
                  "promoted_to": memory.promoted_to,
              }
  
          promote_it, reason, score = should_promote(memory, now)
          if not promote_it and not force:
              return {
                  "success": False,
                  "message": f"Memory does not meet promotion criteria: {reason}",
                  "score": round(score, 4),
              }
  
          candidates = [
              PromotionCandidate(
                  memory=memory,
                  reason=reason,
                  score=score,
                  use_count=memory.use_count,
                  age_days=calculate_memory_age(memory, now),
              )
          ]
      elif auto_detect:
          memories = db.list_memories(status=MemoryStatus.ACTIVE)
          for memory in memories:
              promote_it, reason, score = should_promote(memory, now)
              if promote_it:
                  candidates.append(
                      PromotionCandidate(
                          memory=memory,
                          reason=reason,
                          score=score,
                          use_count=memory.use_count,
                          age_days=calculate_memory_age(memory, now),
                      )
                  )
          candidates.sort(key=lambda c: c.score, reverse=True)
      else:
          return {
              "success": False,
              "message": "Must specify memory_id or set auto_detect=true",
          }
  
      if not dry_run and candidates:
          integration = BasicMemoryIntegration()
          config = get_config()
  
          # Initialize LTM index if vault is configured
          ltm_index = None
          if config.ltm_vault_path and config.ltm_vault_path.exists():
              ltm_index = LTMIndex(vault_path=config.ltm_vault_path)
              # Load existing index if it exists
              if ltm_index.index_path.exists():
                  ltm_index.load_index()
  
          for candidate in candidates:
              if target == "obsidian":
                  result = integration.promote_to_obsidian(candidate.memory)
              else:
                  return {"success": False, "message": f"Unknown target: {target}"}
  
              if result["success"]:
                  db.update_memory(
                      memory_id=candidate.memory.id,
                      status=MemoryStatus.PROMOTED,
                      promoted_at=now,
                      promoted_to=result.get("path"),
                  )
                  promoted_ids.append(candidate.memory.id)
  
                  # Incrementally update LTM index with newly promoted file
                  if ltm_index and result.get("full_path"):
                      try:
                          file_path = Path(result["full_path"])
                          ltm_index.add_document(file_path)
                      except Exception as e:
                          print(f"Warning: Failed to update LTM index for {result['path']}: {e}")
  
      return {
          "success": True,
          "dry_run": dry_run,
          "candidates_found": len(candidates),
          "promoted_count": len(promoted_ids),
          "promoted_ids": promoted_ids,
          "candidates": [
              {
                  "id": c.memory.id,
                  "content_preview": c.memory.content[:100],
                  "reason": c.reason,
                  "score": round(c.score, 4),
                  "use_count": c.use_count,
                  "age_days": round(c.age_days, 1),
              }
              for c in candidates[:10]
          ],
          "message": (
              f"{'Would promote' if dry_run else 'Promoted'} {len(promoted_ids)} memories to {target}"
          ),
      }