feat: bind loguru context to telemetry scopes

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

Author: phernandez

docs/logfire-instrumentation-strategy.md

@@ -349,6 +349,7 @@ Avoid full automatic SQL span firehose by default.
 - [x] Phase 2: Root spans for entrypoints and primary operations
 - [x] Phase 3: Child spans for sync, search, and routing
 - [x] Phase 4: Failure-focused detail and final verification
+- [x] Phase 5: Loguru context binding and scoped context inheritance
 
 ## Recommended Rollout Phases
 
@@ -392,6 +393,16 @@ Add selective deeper spans/log enrichment for:
 
 This keeps normal traces clean while improving debuggability.
 
+### Phase 5: Loguru context binding and scoped context inheritance
+
+Add:
+
+- context-local telemetry state in `basic_memory.telemetry`
+- a shared `scope(...)` helper that opens a span and binds stable logger context together
+- context inheritance for routing, sync, and search so downstream `loguru` logs carry the active operation fields
+
+This makes the trace view and the log stream tell the same story without forcing logger rewrites across the codebase.
+
 ## Validation Checklist
 
 We should consider the integration successful when the following are true:

src/basic_memory/mcp/project_context.py

@@ -205,7 +205,7 @@ async def resolve_workspace_parameter(
     context: Optional[Context] = None,
 ) -> WorkspaceInfo:
     """Resolve workspace using explicit input, session cache, and cloud discovery."""
-    with telemetry.span(
+    with telemetry.scope(
         "routing.resolve_workspace",
         workspace_requested=workspace is not None,
         has_context=context is not None,
@@ -281,7 +281,7 @@ async def get_active_project(
         ValueError: If no project can be resolved
         HTTPError: If project doesn't exist or is inaccessible
     """
-    with telemetry.span(
+    with telemetry.scope(
         "routing.validate_project",
         requested_project=project,
         has_context=context is not None,
@@ -364,7 +364,7 @@ async def resolve_project_and_path(
     """
     is_memory_url = identifier.strip().startswith("memory://")
     include_project = ConfigManager().config.permalinks_include_project if is_memory_url else None
-    with telemetry.span(
+    with telemetry.scope(
         "routing.resolve_memory_url",
         is_memory_url=is_memory_url,
         requested_project=project,
@@ -548,14 +548,13 @@ async def get_project_client(
     # Outcome: use the factory client directly, skip workspace resolution
     if is_factory_mode():
         route_mode = "factory"
-        with telemetry.contextualize(route_mode=route_mode, workspace_id=workspace):
-            with telemetry.span(
-                "routing.resolve_client",
-                project_name=resolved_project,
-                route_mode=route_mode,
-                workspace_id=workspace,
-            ):
-                logger.debug("Using injected client factory for project routing")
+        with telemetry.scope(
+            "routing.resolve_client",
+            project_name=resolved_project,
+            route_mode=route_mode,
+            workspace_id=workspace,
+        ):
+            logger.debug("Using injected client factory for project routing")
             async with get_client() as client:
                 active_project = await get_active_project(client, resolved_project, context)
                 yield client, active_project
@@ -567,13 +566,12 @@ async def get_project_client(
     # Outcome: route strictly based on explicit flag, no workspace network calls
     if _explicit_routing() and _force_local_mode():
         route_mode = "explicit_local"
-        with telemetry.contextualize(route_mode=route_mode, workspace_id=None):
-            with telemetry.span(
-                "routing.resolve_client",
-                project_name=resolved_project,
-                route_mode=route_mode,
-            ):
-                logger.debug("Explicit local routing selected for project client")
+        with telemetry.scope(
+            "routing.resolve_client",
+            project_name=resolved_project,
+            route_mode=route_mode,
+        ):
+            logger.debug("Explicit local routing selected for project client")
             async with get_client(project_name=resolved_project) as client:
                 active_project = await get_active_project(client, resolved_project, context)
                 yield client, active_project
@@ -611,17 +609,13 @@ async def get_project_client(
         # which checks context cache, auto-selects single workspace, or errors
         if effective_workspace is not None:
             # Config-resolved workspace — pass directly to get_client, skip network lookup
-            with telemetry.contextualize(
+            with telemetry.scope(
+                "routing.resolve_client",
+                project_name=resolved_project,
                 route_mode=route_mode,
                 workspace_id=effective_workspace,
             ):
-                with telemetry.span(
-                    "routing.resolve_client",
-                    project_name=resolved_project,
-                    route_mode=route_mode,
-                    workspace_id=effective_workspace,
-                ):
-                    logger.debug("Using configured workspace for cloud project routing")
+                logger.debug("Using configured workspace for cloud project routing")
                 async with get_client(
                     project_name=resolved_project,
                     workspace=effective_workspace,
@@ -631,17 +625,13 @@ async def get_project_client(
         else:
             # No config-based workspace — use resolve_workspace_parameter for discovery
             active_ws = await resolve_workspace_parameter(workspace=None, context=context)
-            with telemetry.contextualize(
+            with telemetry.scope(
+                "routing.resolve_client",
+                project_name=resolved_project,
                 route_mode=route_mode,
                 workspace_id=active_ws.tenant_id,
             ):
-                with telemetry.span(
-                    "routing.resolve_client",
-                    project_name=resolved_project,
-                    route_mode=route_mode,
-                    workspace_id=active_ws.tenant_id,
-                ):
-                    logger.debug("Resolved workspace dynamically for cloud project routing")
+                logger.debug("Resolved workspace dynamically for cloud project routing")
                 async with get_client(
                     project_name=resolved_project,
                     workspace=active_ws.tenant_id,
@@ -652,13 +642,12 @@ async def get_project_client(
 
     # Step 4: Local routing (default)
     route_mode = "local_asgi"
-    with telemetry.contextualize(route_mode=route_mode, workspace_id=None):
-        with telemetry.span(
-            "routing.resolve_client",
-            project_name=resolved_project,
-            route_mode=route_mode,
-        ):
-            logger.debug("Using default local ASGI routing for project client")
+    with telemetry.scope(
+        "routing.resolve_client",
+        project_name=resolved_project,
+        route_mode=route_mode,
+    ):
+        logger.debug("Using default local ASGI routing for project client")
         async with get_client(project_name=resolved_project) as client:
             active_project = await get_active_project(client, resolved_project, context)
             yield client, active_project

src/basic_memory/services/search_service.py

@@ -152,8 +152,6 @@ async def search(self, query: SearchQuery, limit=10, offset=0) -> List[SearchInd
             logger.debug("no criteria passed to query")
             return []
 
-        logger.trace(f"Searching with query: {query}")
-
         after_date = (
             (
                 query.after_date
@@ -176,7 +174,7 @@ async def search(self, query: SearchQuery, limit=10, offset=0) -> List[SearchInd
         retrieval_mode = query.retrieval_mode or SearchRetrievalMode.FTS
         strict_search_text = query.text
 
-        with telemetry.span(
+        with telemetry.scope(
             "search.execute",
             retrieval_mode=retrieval_mode.value,
             has_text_query=bool(strict_search_text),
@@ -186,6 +184,7 @@ async def search(self, query: SearchQuery, limit=10, offset=0) -> List[SearchInd
             limit=limit,
             offset=offset,
         ):
+            logger.trace(f"Searching with query: {query}")
             # First pass: preserve existing strict search behavior.
             results = await self.repository.search(
                 search_text=strict_search_text,
@@ -219,7 +218,7 @@ async def search(self, query: SearchQuery, limit=10, offset=0) -> List[SearchInd
             "Strict FTS returned 0 results; retrying relaxed FTS query "
             f"strict='{strict_search_text}' relaxed='{relaxed_search_text}'"
         )
-        with telemetry.span(
+        with telemetry.scope(
             "search.relaxed_fts_retry",
             retrieval_mode=retrieval_mode.value,
             token_count=len(self._tokenize_fts_text(strict_search_text)),

src/basic_memory/sync/sync_service.py

@@ -278,7 +278,7 @@ async def sync(
 
             # initial paths from db to sync
             # path -> checksum
-            with telemetry.span("sync.project.scan", force_full=force_full):
+            with telemetry.scope("sync.project.scan", force_full=force_full):
                 report = await self.scan(directory, force_full=force_full)
 
             # order of sync matters to resolve relations effectively
@@ -287,7 +287,7 @@ async def sync(
                 + f"deleted_files={len(report.deleted)}, moved_files={len(report.moves)}"
             )
 
-            with telemetry.span(
+            with telemetry.scope(
                 "sync.project.apply_changes",
                 new_count=len(report.new),
                 modified_count=len(report.modified),
@@ -350,15 +350,15 @@ async def sync(
             # Only resolve relations if there were actual changes
             # If no files changed, no new unresolved relations could have been created
             if report.total > 0:
-                with telemetry.span("sync.project.resolve_relations", relation_scope="all_pending"):
+                with telemetry.scope("sync.project.resolve_relations", relation_scope="all_pending"):
                     await self.resolve_relations()
             else:
                 logger.info("Skipping relation resolution - no file changes detected")
 
             # Batch-generate vector embeddings for all synced entities
             if synced_entity_ids and self.app_config.semantic_search_enabled:
                 try:
-                    with telemetry.span(
+                    with telemetry.scope(
                         "sync.project.sync_embeddings",
                         entity_count=len(synced_entity_ids),
                     ):
@@ -382,7 +382,7 @@ async def sync(
             # Update scan watermark after successful sync
             # Use the timestamp from sync start (not end) to ensure we catch files
             # created during the sync on the next iteration
-            with telemetry.span("sync.project.update_watermark"):
+            with telemetry.scope("sync.project.update_watermark"):
                 current_file_count = await self._quick_count_files(directory)
                 if self.entity_repository.project_id is not None:
                     project = await self.project_repository.find_by_id(
@@ -458,7 +458,7 @@ async def scan(self, directory, force_full: bool = False):
         if project is None:
             raise ValueError(f"Project not found: {self.entity_repository.project_id}")
 
-        with telemetry.span("sync.project.select_scan_strategy", force_full=force_full):
+        with telemetry.scope("sync.project.select_scan_strategy", force_full=force_full):
             # Step 1: Quick file count
             logger.debug("Counting files in directory")
             current_count = await self._quick_count_files(directory)
@@ -502,13 +502,12 @@ async def scan(self, directory, force_full: bool = False):
                 logger.warning("No scan watermark available, falling back to full scan")
                 scan_coro = self._scan_directory_full(directory)
 
-        with telemetry.contextualize(scan_type=scan_type):
-            with telemetry.span("sync.project.filesystem_scan", scan_type=scan_type):
-                file_paths_to_scan = await scan_coro
-                if scan_type == "incremental":
-                    logger.debug(
-                        f"Incremental scan found {len(file_paths_to_scan)} potentially changed files"
-                    )
+        with telemetry.scope("sync.project.filesystem_scan", scan_type=scan_type):
+            file_paths_to_scan = await scan_coro
+            if scan_type == "incremental":
+                logger.debug(
+                    f"Incremental scan found {len(file_paths_to_scan)} potentially changed files"
+                )
 
             # Step 3: Process each file with mtime-based comparison
             scanned_paths: Set[str] = set()
@@ -569,7 +568,7 @@ async def scan(self, directory, force_full: bool = False):
 
             # Step 4: Detect moves (for both full and incremental scans)
             # Check if any "new" files are actually moves by matching checksums
-            with telemetry.span("sync.project.detect_moves", new_count=len(report.new)):
+            with telemetry.scope("sync.project.detect_moves", new_count=len(report.new)):
                 for new_path in list(
                     report.new
                 ):  # Use list() to allow modification during iteration
@@ -604,7 +603,7 @@ async def scan(self, directory, force_full: bool = False):
             # Step 5: Detect deletions (only for full scans)
             # Incremental scans can't reliably detect deletions since they only see modified files
             if scan_type in ("full_initial", "full_deletions", "full_fallback", "full_forced"):
-                with telemetry.span("sync.project.detect_deletions", scan_type=scan_type):
+                with telemetry.scope("sync.project.detect_deletions", scan_type=scan_type):
                     # Use optimized query for just file paths (not full entities)
                     db_file_paths = await self.entity_repository.get_all_file_paths()
                     logger.debug(f"Found {len(db_file_paths)} db paths for deletion detection")
@@ -693,7 +692,7 @@ async def sync_file(
         except FileNotFoundError:
             # File exists in database but not on filesystem
             # This indicates a database/filesystem inconsistency - treat as deletion
-            with telemetry.span(
+            with telemetry.scope(
                 "sync.file.failure",
                 failure_type="file_not_found",
                 path=path,
@@ -715,7 +714,7 @@ async def sync_file(
             if isinstance(e, SyncFatalError) or isinstance(
                 e.__cause__, SyncFatalError
             ):  # pragma: no cover
-                with telemetry.span(
+                with telemetry.scope(
                     "sync.file.failure",
                     failure_type=failure_type,
                     path=path,
@@ -728,7 +727,7 @@ async def sync_file(
 
             # Otherwise treat as recoverable file-level error
             error_msg = str(e)
-            with telemetry.span(
+            with telemetry.scope(
                 "sync.file.failure",
                 failure_type=failure_type,
                 path=path,
@@ -1133,7 +1132,7 @@ async def resolve_relations(self, entity_id: int | None = None):
                     # update search index only on successful resolution
                     await self.search_service.index_entity(resolved_entity)
                 except IntegrityError:
-                    with telemetry.span(
+                    with telemetry.scope(
                         "sync.relation.resolve_conflict",
                         relation_id=relation.id,
                         relation_type=relation.relation_type,
@@ -1154,7 +1153,7 @@ async def resolve_relations(self, entity_id: int | None = None):
                         try:
                             await self.relation_repository.delete(relation.id)
                         except Exception as e:
-                            with telemetry.span(
+                            with telemetry.scope(
                                 "sync.relation.cleanup_failure",
                                 relation_id=relation.id,
                                 relation_type=relation.relation_type,