diff --git a/src/basic_memory/repository/repository.py b/src/basic_memory/repository/repository.py
index 6c1904fc..211fbebe 100644
--- a/src/basic_memory/repository/repository.py
+++ b/src/basic_memory/repository/repository.py
@@ -1,5 +1,6 @@
 """Base repository implementation."""
 
+from datetime import datetime
 from typing import Type, Optional, Any, Sequence, TypeVar, List
 
 from loguru import logger
@@ -148,6 +149,35 @@ class Repository[T: Base]:
             logger.debug(f"No {self.Model.__name__} found")
         return entity
 
+    async def find_modified_since(self, since: datetime) -> Sequence[T]:
+        """Find all records modified since the given timestamp.
+        
+        This method assumes the model has an updated_at column. Override
+        in subclasses if a different column should be used.
+        
+        Args:
+            since: Datetime to search from
+            
+        Returns:
+            Sequence of records modified since the timestamp
+        """
+        logger.debug(f"Finding {self.Model.__name__} modified since: {since}")
+        
+        if not hasattr(self.Model, 'updated_at'):
+            raise AttributeError(f"{self.Model.__name__} does not have updated_at column")
+            
+        query = (
+            select(self.Model)
+            .filter(self.Model.updated_at >= since)
+            .options(*self.get_load_options())
+        )
+        
+        async with db.scoped_session(self.session_maker) as session:
+            result = await session.execute(query)
+            items = result.scalars().all()
+            logger.debug(f"Found {len(items)} modified {self.Model.__name__} records")
+            return items
+
     async def create(self, data: dict) -> T:
         """Create a new record from a model instance."""
         logger.debug(f"Creating {self.Model.__name__} from entity_data: {data}")
@@ -260,4 +290,4 @@ class Repository[T: Base]:
     def get_load_options(self) -> List[LoaderOption]:
         """Get list of loader options for eager loading relationships.
         Override in subclasses to specify what to load."""
-        return []
+        return []
\ No newline at end of file
diff --git a/src/basic_memory/schemas/activity.py b/src/basic_memory/schemas/activity.py
new file mode 100644
index 00000000..89349161
--- /dev/null
+++ b/src/basic_memory/schemas/activity.py
@@ -0,0 +1,83 @@
+from datetime import datetime, timedelta
+from enum import Enum
+from typing import List, Optional
+
+from pydantic import BaseModel, Field
+
+
+class TimeFrame:
+    """Represents a time period for querying activity."""
+    
+    def __init__(self, timeframe_str: str):
+        """Parse timeframe string (e.g., '1d', '2w', '1m')"""
+        if not timeframe_str or len(timeframe_str) < 2:
+            raise ValueError("Invalid timeframe format")
+            
+        try:
+            self.value = int(timeframe_str[:-1])
+            self.unit = timeframe_str[-1]
+        except ValueError:
+            raise ValueError("Invalid timeframe format")
+            
+        if self.unit not in ['h', 'd', 'w', 'm']:
+            raise ValueError("Invalid timeframe unit")
+            
+        if self.value < 1:
+            raise ValueError("Timeframe value must be positive")
+            
+    @property
+    def to_timedelta(self) -> timedelta:
+        """Convert to Python timedelta."""
+        if self.unit == 'h':
+            return timedelta(hours=self.value)
+        elif self.unit == 'd':
+            return timedelta(days=self.value)
+        elif self.unit == 'w':
+            return timedelta(weeks=self.value)
+        elif self.unit == 'm':
+            # Approximate month as 30 days
+            return timedelta(days=self.value * 30)
+        else:
+            raise ValueError(f"Invalid unit: {self.unit}")
+
+
+class ActivityType(str, Enum):
+    """Types of activities that can be tracked."""
+    DOCUMENT = "document"
+    ENTITY = "entity"
+    RELATION = "relation"
+
+
+class ChangeType(str, Enum):
+    """Types of changes that can occur."""
+    CREATED = "created"
+    UPDATED = "updated"
+    DELETED = "deleted"
+
+
+class ActivityChange(BaseModel):
+    """Represents a single change in the system."""
+    activity_type: ActivityType
+    change_type: ChangeType
+    timestamp: datetime
+    path_id: str
+    summary: str
+    content: Optional[str] = None
+
+
+class ActivitySummary(BaseModel):
+    """Summary statistics about recent activity."""
+    document_changes: int = Field(default=0, description="Number of document changes")
+    entity_changes: int = Field(default=0, description="Number of entity changes")
+    relation_changes: int = Field(default=0, description="Number of relation changes")
+    most_active_paths: List[str] = Field(
+        default_factory=list,
+        description="List of most frequently changed paths"
+    )
+
+
+class RecentActivity(BaseModel):
+    """Complete activity report."""
+    timeframe: str
+    changes: List[ActivityChange] = Field(default_factory=list)
+    summary: ActivitySummary
diff --git a/src/basic_memory/services/activity_service.py b/src/basic_memory/services/activity_service.py
new file mode 100644
index 00000000..5b320f57
--- /dev/null
+++ b/src/basic_memory/services/activity_service.py
@@ -0,0 +1,181 @@
+from datetime import datetime
+from typing import List, Optional, Sequence
+
+from . import EntityService, DocumentService, RelationService
+from ..models import Document
+from ..schemas.activity import (
+    ActivityChange,
+    ActivitySummary,
+    ActivityType,
+    ChangeType,
+    RecentActivity,
+    TimeFrame,
+)
+
+
+class ActivityService:
+    """Service for tracking and querying activity across the knowledge base."""
+
+    def __init__(
+        self,
+        entity_service: EntityService,
+        document_service: DocumentService,
+        relation_service: RelationService,
+    ):
+        """Initialize with required services."""
+        self.entity_service = entity_service
+        self.document_service = document_service
+        self.relation_service = relation_service
+
+    async def get_recent_activity(
+        self,
+        timeframe: str = "1d",
+        activity_types: Optional[List[str]] = None,
+        include_content: bool = True
+    ) -> RecentActivity:
+        """Get all recent activity in the knowledge base.
+        
+        Args:
+            timeframe: Time window to look back (1h, 1d, 1w, 1m)
+            activity_types: Optional list of types to include
+            include_content: Whether to include full content
+            
+        Returns:
+            RecentActivity object containing changes and summary
+        """
+        # Parse timeframe and get cutoff date
+        tf = TimeFrame(timeframe)
+        since = datetime.utcnow() - tf.to_timedelta
+
+        # Get changes based on requested types
+        changes = []
+        types_to_fetch = (
+            [ActivityType(t) for t in activity_types]
+            if activity_types
+            else list(ActivityType)
+        )
+
+        for activity_type in types_to_fetch:
+            if activity_type == ActivityType.ENTITY:
+                changes.extend(await self._get_entity_changes(since))
+            elif activity_type == ActivityType.DOCUMENT:
+                changes.extend(await self._get_document_changes(since))
+            elif activity_type == ActivityType.RELATION:
+                changes.extend(await self._get_relation_changes(since))
+
+        # Sort all changes by timestamp
+        changes.sort(key=lambda x: x.timestamp, reverse=True)
+
+        # Remove content if not requested
+        if not include_content:
+            for change in changes:
+                change.content = None
+
+        # Generate summary
+        summary = ActivitySummary(
+            document_changes=len([c for c in changes if c.activity_type == ActivityType.DOCUMENT]),
+            entity_changes=len([c for c in changes if c.activity_type == ActivityType.ENTITY]),
+            relation_changes=len([c for c in changes if c.activity_type == ActivityType.RELATION]),
+            most_active_paths=self._get_most_active_paths(changes)
+        )
+
+        return RecentActivity(
+            timeframe=timeframe,
+            changes=changes,
+            summary=summary
+        )
+
+    async def _get_entity_changes(self, since: datetime) -> List[ActivityChange]:
+        """Get recent entity changes."""
+        # Query entities updated since the cutoff
+        entities = await self.entity_service.get_modified_since(since)
+        
+        changes = []
+        for entity in entities:
+            change_type = (
+                ChangeType.CREATED 
+                if entity.created_at >= since 
+                else ChangeType.UPDATED
+            )
+            
+            changes.append(
+                ActivityChange(
+                    activity_type=ActivityType.ENTITY,
+                    change_type=change_type,
+                    timestamp=entity.updated_at,
+                    path_id=entity.path_id,
+                    summary=f"{change_type.value.title()} entity: {entity.name}",
+                    content=entity.description
+                )
+            )
+            
+        return changes
+
+    async def _get_document_changes(self, since: datetime) -> List[ActivityChange]:
+        """Get recent document changes."""
+        # Query documents updated since the cutoff
+        documents: Sequence[Document] = await self.document_service.get_modified_since(since)
+        
+        changes = []
+        for doc in documents:
+            change_type = (
+                ChangeType.CREATED 
+                if doc.created_at >= since 
+                else ChangeType.UPDATED
+            )
+            
+            changes.append(
+                ActivityChange(
+                    activity_type=ActivityType.DOCUMENT,
+                    change_type=change_type,
+                    timestamp=doc.updated_at,
+                    path_id=doc.path_id,
+                    summary=f"{change_type.value.title()} document: {doc.path_id}",
+                    #content=doc.content[:500] if doc.content else None  # First 500 chars
+                )
+            )
+            
+        return changes
+
+    async def _get_relation_changes(self, since: datetime) -> List[ActivityChange]:
+        """Get recent relation changes."""
+        # Query relations updated since the cutoff
+        relations = await self.relation_service.get_modified_since(since)
+        
+        changes = []
+        for relation in relations:
+            change_type = (
+                ChangeType.CREATED 
+                if relation.created_at >= since 
+                else ChangeType.UPDATED
+            )
+            
+            changes.append(
+                ActivityChange(
+                    activity_type=ActivityType.RELATION,
+                    change_type=change_type,
+                    timestamp=relation.updated_at,
+                    path_id=f"{relation.from_id}->{relation.to_id}",
+                    summary=(
+                        f"{change_type.value.title()} relation: "
+                        f"{relation.from_id} {relation.relation_type} {relation.to_id}"
+                    ),
+                    content=relation.context
+                )
+            )
+            
+        return changes
+
+    def _get_most_active_paths(self, changes: List[ActivityChange], limit: int = 5) -> List[str]:
+        """Get the most frequently changed paths."""
+        path_counts = {}
+        for change in changes:
+            path_counts[change.path_id] = path_counts.get(change.path_id, 0) + 1
+            
+        # Sort by count descending and take top paths
+        sorted_paths = sorted(
+            path_counts.items(),
+            key=lambda x: (-x[1], x[0])  # Sort by count desc, then path asc
+        )
+        
+        return [path for path, _ in sorted_paths[:limit]]
diff --git a/src/basic_memory/services/service.py b/src/basic_memory/services/service.py
index 763db145..db6ccf8c 100644
--- a/src/basic_memory/services/service.py
+++ b/src/basic_memory/services/service.py
@@ -1,10 +1,12 @@
 """Base service class."""
 
+from datetime import datetime
 from typing import TypeVar, Generic, List, Sequence
 
+from basic_memory.models import Base
 from basic_memory.repository.repository import Repository
 
-T = TypeVar("T", bound=Repository)
+T = TypeVar("T", bound=Base)
 
 
 class BaseService(Generic[T]):
@@ -21,3 +23,14 @@ class BaseService(Generic[T]):
     async def add_all(self, models: List[T]) -> Sequence[T]:
         """Add a List of models to repository."""
         return await self.repository.add_all(models)
+
+    async def get_modified_since(self, since: datetime) -> Sequence[T]:
+        """Get all items modified since the given timestamp.
+        
+        Args:
+            since: Datetime to search from
+            
+        Returns:
+            Sequence of items modified since the timestamp
+        """
+        return await self.repository.find_modified_since(since)
\ No newline at end of file
diff --git a/tests/conftest.py b/tests/conftest.py
index 7d62cb65..7b7f9704 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -25,6 +25,7 @@ from basic_memory.services import (
     DocumentService,
     FileChangeScanner,
 )
+from basic_memory.services.activity_service import ActivityService
 from basic_memory.services.file_service import FileService
 from basic_memory.services import KnowledgeService
 
@@ -171,6 +172,12 @@ async def knowledge_service(
         base_path=test_config.knowledge_dir,
     )
 
+@pytest_asyncio.fixture
+async def activity_service(document_service, entity_service, relation_service):
+    """Create activity service with real dependencies."""
+    return ActivityService(entity_service, document_service, relation_service)
+
+
 
 @pytest_asyncio.fixture(scope="function")
 async def sample_entity(entity_repository: EntityRepository) -> Entity:
diff --git a/tests/repository/test_repository.py b/tests/repository/test_repository.py
index e9f7fad6..e1a56fb5 100644
--- a/tests/repository/test_repository.py
+++ b/tests/repository/test_repository.py
@@ -1,7 +1,8 @@
 """Test repository implementation."""
 
+from datetime import datetime, timedelta
 import pytest
-from sqlalchemy import String
+from sqlalchemy import String, DateTime
 from sqlalchemy.orm import Mapped, mapped_column
 
 from basic_memory.models import Base
@@ -16,6 +17,12 @@ class TestModel(Base):
     id: Mapped[str] = mapped_column(String(255), primary_key=True)
     name: Mapped[str] = mapped_column(String(255))
     description: Mapped[str | None] = mapped_column(String(255), nullable=True)
+    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime, 
+        default=datetime.utcnow,
+        onupdate=datetime.utcnow
+    )
 
 
 @pytest.fixture
@@ -114,3 +121,57 @@ async def test_delete_by_ids(repository):
     assert await repository.find_by_id(ids_to_delete[0]) is None
     assert await repository.find_by_id(ids_to_delete[1]) is None
     assert await repository.find_by_id(ids_to_delete[2]) is None
+
+
+@pytest.mark.asyncio
+async def test_find_modified_since(repository):
+    """Test finding entities modified since a timestamp."""
+    # Create initial test data
+    now = datetime.utcnow()
+    base_instances = [
+        TestModel(
+            id=f"test_{i}", 
+            name=f"Test {i}",
+            created_at=now - timedelta(days=2),
+            updated_at=now - timedelta(days=2)
+        ) for i in range(5)
+    ]
+    await repository.create_all([instance.__dict__ for instance in base_instances])
+
+    # Update some instances to have recent changes
+    cutoff_time = now - timedelta(hours=1)
+    recent_updates = ["test_1", "test_3"]
+    
+    for id in recent_updates:
+        await repository.update(id, {"name": f"Updated {id}"})
+
+    # Find recently modified
+    modified = await repository.find_modified_since(cutoff_time)
+    assert len(modified) == 2
+    assert sorted([e.id for e in modified]) == sorted(recent_updates)
+    for entity in modified:
+        assert entity.updated_at >= cutoff_time
+        assert entity.name.startswith("Updated")
+
+    # Test with older cutoff
+    all_modified = await repository.find_modified_since(now - timedelta(days=3))
+    assert len(all_modified) == 5  # Should find all instances
+
+    # Test with future cutoff
+    future_modified = await repository.find_modified_since(now + timedelta(hours=1))
+    assert len(future_modified) == 0  # Should find no instances
+
+
+@pytest.mark.asyncio
+async def test_find_modified_since_invalid_model():
+    """Test finding modified entities on model without updated_at."""
+    class InvalidModel(Base):
+        __tablename__ = "invalid_model"
+        id: Mapped[str] = mapped_column(String(255), primary_key=True)
+        name: Mapped[str] = mapped_column(String(255))
+
+    repository = Repository(None, InvalidModel)  # type: ignore
+    
+    with pytest.raises(AttributeError) as exc:
+        await repository.find_modified_since(datetime.utcnow())
+    assert "does not have updated_at column" in str(exc.value)
\ No newline at end of file
diff --git a/tests/test_activity_schemas.py b/tests/test_activity_schemas.py
new file mode 100644
index 00000000..7a5a0eb1
--- /dev/null
+++ b/tests/test_activity_schemas.py
@@ -0,0 +1,105 @@
+from datetime import datetime, timedelta
+import pytest
+
+from basic_memory.schemas.activity import (
+    TimeFrame,
+    ActivityType,
+    ChangeType,
+    ActivityChange,
+    ActivitySummary,
+    RecentActivity,
+)
+
+
+def test_timeframe_parsing():
+    """Test TimeFrame string parsing."""
+    # Valid timeframes
+    assert TimeFrame("1h").value == 1
+    assert TimeFrame("1h").unit == "h"
+    assert TimeFrame("24h").value == 24
+    assert TimeFrame("7d").unit == "d"
+    assert TimeFrame("4w").unit == "w"
+    assert TimeFrame("2m").unit == "m"
+
+    # Invalid timeframes
+    with pytest.raises(ValueError):
+        TimeFrame("")
+    with pytest.raises(ValueError):
+        TimeFrame("h")
+    with pytest.raises(ValueError):
+        TimeFrame("abc")
+    with pytest.raises(ValueError):
+        TimeFrame("1x")  # Invalid unit
+    with pytest.raises(ValueError):
+        TimeFrame("-1h")  # Negative value
+
+
+def test_timeframe_to_timedelta():
+    """Test conversion to timedelta."""
+    assert TimeFrame("1h").to_timedelta == timedelta(hours=1)
+    assert TimeFrame("24h").to_timedelta == timedelta(hours=24)
+    assert TimeFrame("1d").to_timedelta == timedelta(days=1)
+    assert TimeFrame("1w").to_timedelta == timedelta(weeks=1)
+    assert TimeFrame("1m").to_timedelta == timedelta(days=30)  # Approximate
+
+    
+def test_activity_change_model():
+    """Test ActivityChange model."""
+    now = datetime.utcnow()
+    change = ActivityChange(
+        activity_type=ActivityType.DOCUMENT,
+        change_type=ChangeType.CREATED,
+        timestamp=now,
+        path_id="test/path",
+        summary="Created test document",
+        content="Test content"
+    )
+    
+    assert change.activity_type == ActivityType.DOCUMENT
+    assert change.change_type == ChangeType.CREATED
+    assert change.timestamp == now
+    assert change.path_id == "test/path"
+    assert change.summary == "Created test document"
+    assert change.content == "Test content"
+
+
+def test_activity_summary_model():
+    """Test ActivitySummary model."""
+    summary = ActivitySummary(
+        document_changes=5,
+        entity_changes=3,
+        relation_changes=2,
+        most_active_paths=["path1", "path2"]
+    )
+    
+    assert summary.document_changes == 5
+    assert summary.entity_changes == 3
+    assert summary.relation_changes == 2
+    assert summary.most_active_paths == ["path1", "path2"]
+
+
+def test_recent_activity_model():
+    """Test RecentActivity model."""
+    now = datetime.utcnow()
+    change = ActivityChange(
+        activity_type=ActivityType.DOCUMENT,
+        change_type=ChangeType.CREATED,
+        timestamp=now,
+        path_id="test/path",
+        summary="Created test document"
+    )
+    
+    summary = ActivitySummary(
+        document_changes=1
+    )
+    
+    activity = RecentActivity(
+        timeframe="1d",
+        changes=[change],
+        summary=summary
+    )
+    
+    assert activity.timeframe == "1d"
+    assert len(activity.changes) == 1
+    assert activity.changes[0].path_id == "test/path"
+    assert activity.summary.document_changes == 1
\ No newline at end of file
diff --git a/tests/test_activity_service.py b/tests/test_activity_service.py
new file mode 100644
index 00000000..dda52b00
--- /dev/null
+++ b/tests/test_activity_service.py
@@ -0,0 +1,122 @@
+"""Test activity service."""
+
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+from basic_memory.services.document_service import DocumentService
+
+
+async def create_test_document(
+    service: DocumentService, name: str, created_delta: timedelta, updated_delta: timedelta
+):
+    """Helper to create document with specific timestamps."""
+    now = datetime.now(timezone.utc)
+    doc = await service.create_document(
+        path_id=f"test/{name}.md",
+        content=f"Content for {name}",
+        metadata={
+            "created": (now - created_delta).isoformat(),
+            "updated": (now - updated_delta).isoformat(),
+        },
+    )
+    return doc
+
+
+@pytest.mark.asyncio
+async def test_get_recent_activity_all_types(activity_service, document_service):
+    """Test getting recent activity for all types."""
+    # Create test documents with various timestamps
+    test_docs = [
+        await create_test_document(
+            document_service, "doc1", timedelta(hours=2), timedelta(hours=2)
+        ),
+        await create_test_document(document_service, "doc2", timedelta(days=2), timedelta(hours=1)),
+        await create_test_document(
+            document_service,
+            "doc3",
+            timedelta(days=3),
+            timedelta(days=3),  # This one should be too old
+        ),
+    ]
+
+    # Get activity from last day
+    result = await activity_service.get_recent_activity(timeframe="1d")
+
+    # Verify results
+    assert len(result.changes) == 2  # Should only find 2 recent docs
+    assert result.summary.document_changes == 2
+
+    # Verify changes are sorted by timestamp (most recent first)
+    timestamps = [change.timestamp for change in result.changes]
+    assert timestamps == sorted(timestamps, reverse=True)
+
+    # Check specific document paths
+    paths = {change.path_id for change in result.changes}
+    assert "test/doc1.md" in paths
+    assert "test/doc2.md" in paths
+    assert "test/doc3.md" not in paths  # Too old to be included
+
+
+@pytest.mark.asyncio
+async def test_get_recent_activity_filtered_types(activity_service, document_service):
+    """Test getting recent activity with type filtering."""
+    # Create some test documents
+    await create_test_document(
+        document_service, "filtered_doc", timedelta(hours=1), timedelta(hours=1)
+    )
+
+    # Get activity filtered to only documents
+    result = await activity_service.get_recent_activity(timeframe="1d", activity_types=["document"])
+
+    # Verify results
+    assert len(result.changes) == 1
+    assert result.summary.document_changes == 1
+    assert result.summary.entity_changes == 0
+    assert result.changes[0].activity_type == "document"
+
+
+@pytest.mark.asyncio
+async def test_get_recent_activity_without_content(activity_service, document_service):
+    """Test getting activity without content."""
+    # Create test document
+    await create_test_document(
+        document_service, "no_content_doc", timedelta(hours=1), timedelta(hours=1)
+    )
+
+    # Get activity without content
+    result = await activity_service.get_recent_activity(timeframe="1d", include_content=False)
+
+    # Verify results
+    assert len(result.changes) == 1
+    assert result.changes[0].content is None
+
+
+@pytest.mark.asyncio
+async def test_change_type_detection(activity_service, document_service):
+    """Test correct detection of created vs updated changes."""
+    now = datetime.now(timezone.utc)
+
+    # Create documents with different creation/update patterns
+    new_doc = await create_test_document(
+        document_service,
+        "new_doc",
+        timedelta(hours=1),  # Created recently
+        timedelta(hours=1),  # Updated recently
+    )
+
+    updated_doc = await create_test_document(
+        document_service,
+        "updated_doc",
+        timedelta(days=2),  # Created a while ago
+        timedelta(hours=2),  # But updated recently
+    )
+
+    # Get activity
+    result = await activity_service.get_recent_activity(timeframe="1d")
+
+    # Verify change types
+    changes_by_path = {change.path_id: change for change in result.changes}
+
+    assert changes_by_path["test/new_doc.md"].change_type == "created"
+    assert changes_by_path["test/updated_doc.md"].change_type == "updated"