convert recent_activity output to text, add code review feedback

Signed-off-by: phernandez <paul@basicmachines.co>

Author: phernandez

src/basic_memory/cli/commands/tool.py

@@ -210,30 +210,21 @@ def recent_activity(
     type: Annotated[Optional[List[SearchItemType]], typer.Option()] = None,
     depth: Optional[int] = 1,
     timeframe: Optional[TimeFrame] = "7d",
-    page: int = 1,
-    page_size: int = 10,
-    max_related: int = 10,
 ):
     """Get recent activity across the knowledge base."""
     try:
-        context = asyncio.run(
+        result = asyncio.run(
             mcp_recent_activity.fn(
                 type=type,  # pyright: ignore [reportArgumentType]
                 depth=depth,
                 timeframe=timeframe,
-                page=page,
-                page_size=page_size,
-                max_related=max_related,
             )
         )
-        # Use json module for more controlled serialization
-        import json
-
-        context_dict = context.model_dump(exclude_none=True)
-        print(json.dumps(context_dict, indent=2, ensure_ascii=True, default=str))
+        # The tool now returns a formatted string directly
+        print(result)
     except Exception as e:  # pragma: no cover
         if not isinstance(e, typer.Exit):
-            typer.echo(f"Error during build_context: {e}", err=True)
+            typer.echo(f"Error during recent_activity: {e}", err=True)
             raise typer.Exit(1)
         raise
 

src/basic_memory/mcp/tools/recent_activity.py

@@ -1,6 +1,5 @@
 """Recent activity tool for Basic Memory MCP server."""
 
-from datetime import datetime, timezone
 from typing import List, Union, Optional
 
 from loguru import logger
@@ -13,7 +12,6 @@
 from basic_memory.schemas.base import TimeFrame
 from basic_memory.schemas.memory import (
     GraphContext,
-    ProjectActivitySummary,
     ProjectActivity,
     ActivityStats,
 )
@@ -37,12 +35,9 @@ async def recent_activity(
     type: Union[str, List[str]] = "",
     depth: int = 1,
     timeframe: TimeFrame = "7d",
-    page: int = 1,
-    page_size: int = 10,
-    max_related: int = 10,
     project: Optional[str] = None,
     context: Context | None = None,
-) -> Union[GraphContext, ProjectActivitySummary]:
+) -> str:
     """Get recent activity for a specific project or across all projects.
 
     This tool works in two modes based on the project parameter:
@@ -67,9 +62,6 @@ async def recent_activity(
             - Relative: "2 days ago", "last week", "yesterday"
             - Points in time: "2024-01-01", "January 1st"
             - Standard format: "7d", "24h"
-        page: Page number of results to return (default: 1)
-        page_size: Number of results to return per page (default: 10)
-        max_related: Maximum number of related results to return (default: 10)
         project: Optional project name. If not provided, uses default_project (if default_project_mode=true)
                 or returns activity across all projects for discovery.
                 If unknown, use list_memory_projects() to discover available projects.
@@ -104,9 +96,9 @@ async def recent_activity(
     """
     # Build common parameters for API calls
     params = {
-        "page": page,
-        "page_size": page_size,
-        "max_related": max_related,
+        "page": 1,
+        "page_size": 10,
+        "max_related": 10,
     }
     if depth:
         params["depth"] = depth
@@ -237,18 +229,13 @@ async def recent_activity(
 
         guidance = "\n".join(guidance_lines)
 
-        return ProjectActivitySummary(
-            projects=projects_activity,
-            summary=summary,
-            timeframe=str(timeframe),
-            generated_at=datetime.now(timezone.utc),
-            guidance=guidance,
-        )
+        # Format discovery mode output
+        return _format_discovery_output(projects_activity, summary, timeframe, guidance)
 
     else:
         # Project-Specific Mode: Get activity for specific project
         logger.info(
-            f"Getting recent activity from project {resolved_project}: type={type}, depth={depth}, timeframe={timeframe}, page={page}, page_size={page_size}, max_related={max_related}"
+            f"Getting recent activity from project {resolved_project}: type={type}, depth={depth}, timeframe={timeframe}"
         )
 
         active_project = await get_active_project(client, resolved_project, context)
@@ -259,7 +246,10 @@ async def recent_activity(
             f"{project_url}/memory/recent",
             params=params,
         )
-        return GraphContext.model_validate(response.json())
+        activity_data = GraphContext.model_validate(response.json())
+
+        # Format project-specific mode output
+        return _format_project_output(resolved_project, activity_data, timeframe, type)
 
 
 async def _get_project_activity(
@@ -313,3 +303,224 @@ async def _get_project_activity(
         last_activity=last_activity,
         active_folders=list(active_folders)[:5],  # Limit to top 5 folders
     )
+
+
+def _format_discovery_output(
+    projects_activity: dict, summary: ActivityStats, timeframe: str, guidance: str
+) -> str:
+    """Format discovery mode output as human-readable text."""
+    lines = [f"## Recent Activity Summary ({timeframe})"]
+
+    # Most active project section
+    if summary.most_active_project and summary.total_items > 0:
+        most_active = projects_activity[summary.most_active_project]
+        lines.append(
+            f"\n**Most Active Project:** {summary.most_active_project} ({most_active.item_count} items)"
+        )
+
+        # Get latest activity from most active project
+        if most_active.activity.results:
+            latest = most_active.activity.results[0].primary_result
+            title = latest.title if hasattr(latest, "title") and latest.title else "Recent activity"
+            # Format relative time
+            time_str = (
+                _format_relative_time(latest.created_at) if latest.created_at else "unknown time"
+            )
+            lines.append(f"- 🔧 **Latest:** {title} ({time_str})")
+
+        # Active folders
+        if most_active.active_folders:
+            folders = ", ".join(most_active.active_folders[:3])
+            lines.append(f"- 📋 **Focus areas:** {folders}")
+
+    # Other active projects
+    other_active = [
+        (name, activity)
+        for name, activity in projects_activity.items()
+        if activity.item_count > 0 and name != summary.most_active_project
+    ]
+
+    if other_active:
+        lines.append("\n**Other Active Projects:**")
+        for name, activity in sorted(other_active, key=lambda x: x[1].item_count, reverse=True)[:4]:
+            lines.append(f"- **{name}** ({activity.item_count} items)")
+
+    # Key developments - extract from recent entities
+    key_items = []
+    for name, activity in projects_activity.items():
+        if activity.item_count > 0:
+            for result in activity.activity.results[:3]:  # Top 3 from each active project
+                if result.primary_result.type == "entity" and hasattr(
+                    result.primary_result, "title"
+                ):
+                    title = result.primary_result.title
+                    # Look for status indicators in titles
+                    if any(word in title.lower() for word in ["complete", "fix", "test", "spec"]):
+                        key_items.append(title)
+
+    if key_items:
+        lines.append("\n**Key Developments:**")
+        for item in key_items[:5]:  # Show top 5
+            status = "✅" if any(word in item.lower() for word in ["complete", "fix"]) else "🧪"
+            lines.append(f"- {status} **{item}**")
+
+    # Add summary stats
+    lines.append(
+        f"\n**Summary:** {summary.active_projects} active projects, {summary.total_items} recent items"
+    )
+
+    # Add guidance
+    lines.append(guidance)
+
+    return "\n".join(lines)
+
+
+def _format_project_output(
+    project_name: str,
+    activity_data: GraphContext,
+    timeframe: str,
+    type_filter: Union[str, List[str]],
+) -> str:
+    """Format project-specific mode output as human-readable text."""
+    lines = [f"## Recent Activity: {project_name} ({timeframe})"]
+
+    if not activity_data.results:
+        lines.append(f"\nNo recent activity found in '{project_name}' project.")
+        return "\n".join(lines)
+
+    # Group results by type
+    entities = []
+    relations = []
+    observations = []
+
+    for result in activity_data.results:
+        if result.primary_result.type == "entity":
+            entities.append(result.primary_result)
+        elif result.primary_result.type == "relation":
+            relations.append(result.primary_result)
+        elif result.primary_result.type == "observation":
+            observations.append(result.primary_result)
+
+    # Show entities (notes/documents)
+    if entities:
+        lines.append(f"\n**📄 Recent Notes & Documents ({len(entities)}):**")
+        for entity in entities[:5]:  # Show top 5
+            title = entity.title if hasattr(entity, "title") and entity.title else "Untitled"
+            # Get folder from file_path if available
+            folder = ""
+            if hasattr(entity, "file_path") and entity.file_path:
+                folder_path = "/".join(entity.file_path.split("/")[:-1])
+                if folder_path:
+                    folder = f" ({folder_path})"
+            lines.append(f"  • {title}{folder}")
+
+    # Show observations (categorized insights)
+    if observations:
+        lines.append(f"\n**🔍 Recent Observations ({len(observations)}):**")
+        # Group by category
+        by_category = {}
+        for obs in observations[:10]:  # Limit to recent ones
+            category = (
+                getattr(obs, "category", "general") if hasattr(obs, "category") else "general"
+            )
+            if category not in by_category:
+                by_category[category] = []
+            by_category[category].append(obs)
+
+        for category, obs_list in list(by_category.items())[:5]:  # Show top 5 categories
+            lines.append(f"  **{category}:** {len(obs_list)} items")
+            for obs in obs_list[:2]:  # Show 2 examples per category
+                content = (
+                    getattr(obs, "content", "No content")
+                    if hasattr(obs, "content")
+                    else "No content"
+                )
+                # Truncate at word boundary
+                if len(content) > 80:
+                    content = _truncate_at_word(content, 80)
+                lines.append(f"    - {content}")
+
+    # Show relations (connections)
+    if relations:
+        lines.append(f"\n**🔗 Recent Connections ({len(relations)}):**")
+        for rel in relations[:5]:  # Show top 5
+            rel_type = (
+                getattr(rel, "relation_type", "relates_to")
+                if hasattr(rel, "relation_type")
+                else "relates_to"
+            )
+            from_entity = (
+                getattr(rel, "from_entity", "Unknown") if hasattr(rel, "from_entity") else "Unknown"
+            )
+            to_entity = getattr(rel, "to_entity", None) if hasattr(rel, "to_entity") else None
+
+            # Format as WikiLinks to show they're readable notes
+            from_link = f"[[{from_entity}]]" if from_entity != "Unknown" else from_entity
+            to_link = f"[[{to_entity}]]" if to_entity else "[Missing Link]"
+
+            lines.append(f"  • {from_link} → {rel_type} → {to_link}")
+
+    # Activity summary
+    total = len(activity_data.results)
+    lines.append(f"\n**Activity Summary:** {total} items found")
+    if hasattr(activity_data, "metadata") and activity_data.metadata:
+        if hasattr(activity_data.metadata, "total_results"):
+            lines.append(f"Total available: {activity_data.metadata.total_results}")
+
+    return "\n".join(lines)
+
+
+def _format_relative_time(timestamp) -> str:
+    """Format timestamp as relative time like '2 hours ago'."""
+    try:
+        from datetime import datetime, timezone
+        from dateutil.relativedelta import relativedelta
+
+        if isinstance(timestamp, str):
+            # Parse ISO format timestamp
+            dt = datetime.fromisoformat(timestamp.replace("Z", "+00:00"))
+        else:
+            dt = timestamp
+
+        now = datetime.now(timezone.utc)
+        if dt.tzinfo is None:
+            dt = dt.replace(tzinfo=timezone.utc)
+
+        # Use relativedelta for accurate time differences
+        diff = relativedelta(now, dt)
+
+        if diff.years > 0:
+            return f"{diff.years} year{'s' if diff.years > 1 else ''} ago"
+        elif diff.months > 0:
+            return f"{diff.months} month{'s' if diff.months > 1 else ''} ago"
+        elif diff.days > 0:
+            if diff.days == 1:
+                return "yesterday"
+            elif diff.days < 7:
+                return f"{diff.days} days ago"
+            else:
+                weeks = diff.days // 7
+                return f"{weeks} week{'s' if weeks > 1 else ''} ago"
+        elif diff.hours > 0:
+            return f"{diff.hours} hour{'s' if diff.hours > 1 else ''} ago"
+        elif diff.minutes > 0:
+            return f"{diff.minutes} minute{'s' if diff.minutes > 1 else ''} ago"
+        else:
+            return "just now"
+    except Exception:
+        return "recently"
+
+
+def _truncate_at_word(text: str, max_length: int) -> str:
+    """Truncate text at word boundary."""
+    if len(text) <= max_length:
+        return text
+
+    # Find last space before max_length
+    truncated = text[:max_length]
+    last_space = truncated.rfind(" ")
+
+    if last_space > max_length * 0.7:  # Only truncate at word if we're not losing too much
+        return text[:last_space] + "..."
+    else:
+        return text[: max_length - 3] + "..."