Coverage for src/qdrant_loader_mcp_server/mcp/intelligence_handler.py: 67%
303 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-20 10:15 +0000
1"""Cross-document intelligence operations handler for MCP server."""
3import asyncio
4import time
5import uuid
6from pathlib import Path
7from typing import Any
9from qdrant_loader.config import get_settings, initialize_config
10from qdrant_loader_core.graph import get_graph_store
12from ..search.engine import SearchEngine
13from ..utils import LoggingConfig
14from .formatters import MCPFormatters
15from .graph_handler import handle_find_ticket_dependencies
16from .handlers.intelligence import (
17 get_or_create_document_id as _get_or_create_document_id_fn,
18)
19from .handlers.intelligence import process_analysis_results
20from .protocol import MCPProtocol
22# Get logger for this module
23logger = LoggingConfig.get_logger("src.mcp.intelligence_handler")
26class IntelligenceHandler:
27 """Handler for cross-document intelligence operations."""
29 def __init__(
30 self,
31 search_engine: SearchEngine,
32 protocol: MCPProtocol,
33 config_path: Path | None = None,
34 ):
35 """Initialize intelligence handler."""
36 self.search_engine = search_engine
37 self.protocol = protocol
38 self.formatters = MCPFormatters()
39 self._cluster_store = {}
40 self._ttl = 300
41 self._max_sessions = 500
42 self._lock = asyncio.Lock()
43 self._graph_store = None
44 self._graph_store_lock = asyncio.Lock()
45 # Resolved by fastmcp_app._lifespan (MCP_CONFIG / --config) so the graph store loads the same config as the search engine.
46 self._config_path = config_path
48 async def _get_graph_store(self):
49 """Lazy-initialize graph store with proper locking."""
50 async with self._graph_store_lock:
51 if self._graph_store is None:
52 config_path = self._config_path or (Path.cwd() / "config.yaml")
53 project_root = config_path.parent
54 initialize_config(
55 yaml_path=config_path,
56 env_path=project_root / ".env",
57 skip_validation=True,
58 )
59 settings = get_settings()
60 graph_cfg = getattr(settings.global_config, "graph", None)
61 self._graph_store = await get_graph_store(
62 **(graph_cfg.store_kwargs() if graph_cfg else {})
63 )
64 return self._graph_store
66 async def _run_graph_query(
67 self,
68 cypher: str,
69 params: dict | None = None,
70 ):
72 store = await self._get_graph_store()
73 return await store.query_cypher(cypher, params or {})
75 def _get_or_create_document_id(self, doc: Any) -> str:
76 return _get_or_create_document_id_fn(doc)
78 def _expand_cluster_docs_to_schema(
79 self, docs: list[Any], include_metadata: bool
80 ) -> list[dict[str, Any]]:
81 """Build documents array to match expand_cluster outputSchema (id, text, metadata)."""
82 result = []
83 for doc in docs:
84 doc_id = getattr(doc, "document_id", None) or getattr(doc, "id", None) or ""
85 item = {"id": str(doc_id), "text": getattr(doc, "text", "") or ""}
86 if include_metadata:
87 item["metadata"] = {
88 "title": getattr(doc, "source_title", ""),
89 "source_type": getattr(doc, "source_type", ""),
90 "source_url": getattr(doc, "source_url", None),
91 "file_path": getattr(doc, "file_path", None),
92 }
93 result.append(item)
94 return result
96 async def handle_analyze_document_relationships(
97 self, request_id: str | int | None, params: dict[str, Any]
98 ) -> dict[str, Any]:
99 """Handle document relationship analysis request."""
100 logger.debug(
101 "Handling document relationship analysis with params", params=params
102 )
104 if "query" not in params:
105 logger.error("Missing required parameter: query")
106 return self.protocol.create_response(
107 request_id,
108 error={
109 "code": -32602,
110 "message": "Invalid params",
111 "data": "Missing required parameter: query",
112 },
113 )
115 try:
116 logger.info(
117 "Performing document relationship analysis using SearchEngine..."
118 )
120 # Use the sophisticated SearchEngine method
121 analysis_results = await self.search_engine.analyze_document_relationships(
122 query=params["query"],
123 limit=params.get("limit", 20),
124 source_types=params.get("source_types"),
125 project_ids=params.get("project_ids"),
126 )
128 logger.info("Analysis completed successfully")
130 # Transform complex analysis to MCP schema-compliant format
131 raw_result = process_analysis_results(analysis_results, params)
133 # Map to output schema: relationships items only allow specific keys
134 relationships = []
135 for rel in raw_result.get("relationships", []) or []:
136 relationships.append(
137 {
138 "document_1": str(
139 rel.get("document_1") or rel.get("document_1_id") or ""
140 ),
141 "document_2": str(
142 rel.get("document_2") or rel.get("document_2_id") or ""
143 ),
144 "relationship_type": rel.get("relationship_type", ""),
145 "score": float(
146 rel.get("score", rel.get("confidence_score", 0.0))
147 ),
148 "description": rel.get(
149 "description", rel.get("relationship_summary", "")
150 ),
151 }
152 )
154 mcp_result = {
155 "relationships": relationships,
156 "total_analyzed": int(raw_result.get("total_analyzed", 0)),
157 # summary is optional in the schema but useful if present
158 "summary": raw_result.get("summary", ""),
159 }
161 return self.protocol.create_response(
162 request_id,
163 result={
164 "content": [
165 {
166 "type": "text",
167 "text": self.formatters.format_relationship_analysis(
168 analysis_results
169 ),
170 }
171 ],
172 "structuredContent": mcp_result,
173 "isError": False,
174 },
175 )
177 except Exception:
178 logger.error("Error during document relationship analysis", exc_info=True)
179 return self.protocol.create_response(
180 request_id,
181 error={"code": -32603, "message": "Internal server error"},
182 )
184 async def handle_find_similar_documents(
185 self, request_id: str | int | None, params: dict[str, Any]
186 ) -> dict[str, Any]:
187 """
188 Handle a "find similar documents" request and return MCP-formatted results.
190 Parameters:
191 request_id (str | int | None): The request identifier to include in the MCP response.
192 params (dict[str, Any]): Request parameters. Required keys:
193 - target_query: The primary query or document to compare against.
194 - comparison_query: The query or document set to compare with the target.
195 Optional keys:
196 - similarity_metrics: Metrics or configuration used to compute similarity.
197 - max_similar (int): Maximum number of similar documents to return (default 5).
198 - source_types: Restrict search to specific source types.
199 - project_ids: Restrict search to specific project identifiers.
200 - similarity_threshold (float): Minimum similarity score to consider (default 0.7).
202 Returns:
203 dict[str, Any]: An MCP protocol response dictionary. On success the response's `result` contains:
204 - content: a list with a single text block (human-readable summary).
205 - structuredContent: a dict with
206 - similar_documents: list of similar document entries, each containing
207 `document_id`, `title`, `similarity_score`, `similarity_metrics`,
208 `similarity_reason`, and `content_preview`.
209 - similarity_summary: metadata including `total_compared`, `similar_found`,
210 `highest_similarity`, and `metrics_used`.
211 - isError: False
212 On invalid parameters the function returns an MCP error response with code -32602.
213 On internal failures the function returns an MCP error response with code -32603.
214 """
215 logger.debug("Handling find similar documents with params", params=params)
217 # Validate required parameters
218 if "target_query" not in params or "comparison_query" not in params:
219 logger.error(
220 "Missing required parameters: target_query and comparison_query"
221 )
222 return self.protocol.create_response(
223 request_id,
224 error={
225 "code": -32602,
226 "message": "Invalid params",
227 "data": "Missing required parameters: target_query and comparison_query",
228 },
229 )
231 try:
232 logger.info(
233 "Performing find similar documents using SearchEngine...",
234 target_query=params["target_query"],
235 comparison_query=params["comparison_query"],
236 )
238 # Use the sophisticated SearchEngine method
239 similar_docs_raw = await self.search_engine.find_similar_documents(
240 target_query=params["target_query"],
241 comparison_query=params["comparison_query"],
242 similarity_metrics=params.get("similarity_metrics"),
243 max_similar=params.get("max_similar", 5),
244 source_types=params.get("source_types"),
245 project_ids=params.get("project_ids"),
246 similarity_threshold=params.get(
247 "similarity_threshold", 0.7
248 ), # Default 0.7
249 )
251 # Normalize result: engine may return list, but can return {} on empty
252 if isinstance(similar_docs_raw, list):
253 similar_docs = similar_docs_raw
254 elif isinstance(similar_docs_raw, dict):
255 similar_docs = (
256 similar_docs_raw.get("similar_documents", [])
257 or similar_docs_raw.get("results", [])
258 or []
259 )
260 else:
261 similar_docs = []
263 logger.info(f"Got {len(similar_docs)} similar documents from SearchEngine")
265 # ✅ Add response validation
266 expected_count = params.get("max_similar", 5)
267 if len(similar_docs) < expected_count:
268 logger.warning(
269 f"Expected up to {expected_count} similar documents, but only got {len(similar_docs)}. "
270 f"This may indicate similarity threshold issues or insufficient comparison documents."
271 )
273 # ✅ Log document IDs for debugging
274 doc_ids = [doc.get("document_id") for doc in similar_docs]
275 logger.debug(f"Similar document IDs: {doc_ids}")
277 # ✅ Validate that document_id is present in responses
278 missing_ids = [
279 i for i, doc in enumerate(similar_docs) if not doc.get("document_id")
280 ]
281 if missing_ids:
282 logger.error(
283 f"Missing document_id in similar documents at indices: {missing_ids}"
284 )
286 # ✅ Also create lightweight content for back-compat (unit tests expect this call)
287 _legacy_lightweight = (
288 self.formatters.create_lightweight_similar_documents_results(
289 similar_docs, params["target_query"], params["comparison_query"]
290 )
291 )
293 # ✅ Build schema-compliant structured content for find_similar_documents
294 similar_documents = []
295 metrics_used_set: set[str] = set()
296 highest_similarity = 0.0
298 for item in similar_docs:
299 # Normalize access to document fields
300 document = item.get("document") if isinstance(item, dict) else None
302 # Extract document_id - try both dict and object attribute access
303 document_id = (
304 item.get("document_id", "") if isinstance(item, dict) else ""
305 )
306 if not document_id and document:
307 document_id = (
308 document.get("document_id")
309 if isinstance(document, dict)
310 else getattr(document, "document_id", "")
311 )
313 # Extract title - try both dict and object attribute access
314 title = "Untitled"
315 if document:
316 if isinstance(document, dict):
317 title = document.get("source_title", "Untitled")
318 else:
319 title = getattr(document, "source_title", "Untitled")
320 if not title or title == "Untitled":
321 title = (
322 item.get("source_title", "Untitled")
323 if isinstance(item, dict)
324 else "Untitled"
325 )
327 # Extract text content - try both dict and object attribute access
328 content_text = ""
329 if document:
330 if isinstance(document, dict):
331 content_text = document.get("text", "")
332 else:
333 content_text = getattr(document, "text", "")
335 # Create content preview
336 content_preview = ""
337 if content_text and isinstance(content_text, str):
338 content_preview = (
339 content_text[:200] + "..."
340 if len(content_text) > 200
341 else content_text
342 )
344 similarity_score = float(item.get("similarity_score", 0.0))
345 highest_similarity = max(highest_similarity, similarity_score)
347 metric_scores = item.get("metric_scores", {})
348 if isinstance(metric_scores, dict):
349 # Normalize metric keys to strings (Enums -> value) to avoid sort/type errors
350 normalized_metric_keys = [
351 (getattr(k, "value", None) or str(k))
352 for k in metric_scores.keys()
353 ]
354 metrics_used_set.update(normalized_metric_keys)
356 similar_documents.append(
357 {
358 "document_id": str(document_id),
359 "title": title,
360 "similarity_score": similarity_score,
361 "similarity_metrics": {
362 (getattr(k, "value", None) or str(k)): float(v)
363 for k, v in metric_scores.items()
364 if isinstance(v, int | float)
365 },
366 "similarity_reason": (
367 ", ".join(reasons)
368 if isinstance(
369 reasons := item.get("similarity_reasons"), list
370 )
371 else (
372 item.get("similarity_reason", "") or str(reasons or "")
373 )
374 ),
375 "content_preview": content_preview,
376 }
377 )
379 structured_content = {
380 "similar_documents": similar_documents,
381 # target_document is optional; omitted when unknown
382 "similarity_summary": {
383 "total_compared": len(similar_docs),
384 "similar_found": len(similar_documents),
385 "highest_similarity": highest_similarity,
386 # Ensure metrics are strings for deterministic sorting
387 "metrics_used": (
388 sorted(metrics_used_set) if metrics_used_set else []
389 ),
390 },
391 }
393 return self.protocol.create_response(
394 request_id,
395 result={
396 "content": [
397 {
398 "type": "text",
399 "text": self.formatters.format_similar_documents(
400 similar_docs
401 ),
402 }
403 ],
404 "structuredContent": structured_content,
405 "isError": False,
406 },
407 )
409 except Exception:
410 logger.error("Error finding similar documents", exc_info=True)
411 return self.protocol.create_response(
412 request_id,
413 error={
414 "code": -32603,
415 "message": "Internal server error",
416 },
417 )
419 async def handle_detect_document_conflicts(
420 self, request_id: str | int | None, params: dict[str, Any]
421 ) -> dict[str, Any]:
422 """Handle conflict detection request."""
423 logger.debug("Handling conflict detection with params", params=params)
425 if "query" not in params:
426 logger.error("Missing required parameter: query")
427 return self.protocol.create_response(
428 request_id,
429 error={
430 "code": -32602,
431 "message": "Invalid params",
432 "data": "Missing required parameter: query",
433 },
434 )
436 try:
437 logger.info("Performing conflict detection using SearchEngine...")
439 # Use the sophisticated SearchEngine method
440 # Build kwargs, include overrides only if explicitly provided
441 conflict_kwargs: dict[str, Any] = {
442 "query": params["query"],
443 "limit": params.get("limit"),
444 "source_types": params.get("source_types"),
445 "project_ids": params.get("project_ids"),
446 }
447 for opt in (
448 "use_llm",
449 "max_llm_pairs",
450 "overall_timeout_s",
451 "max_pairs_total",
452 "text_window_chars",
453 ):
454 if opt in params and params[opt] is not None:
455 conflict_kwargs[opt] = params[opt]
457 conflict_results = await self.search_engine.detect_document_conflicts(
458 **conflict_kwargs
459 )
461 logger.info("Conflict detection completed successfully")
463 # Create lightweight structured content for MCP compliance
464 structured_content = self.formatters.create_lightweight_conflict_results(
465 conflict_results, params["query"]
466 )
468 return self.protocol.create_response(
469 request_id,
470 result={
471 "content": [
472 {
473 "type": "text",
474 "text": self.formatters.format_conflict_analysis(
475 conflict_results
476 ),
477 }
478 ],
479 "structuredContent": structured_content,
480 "isError": False,
481 },
482 )
484 except Exception:
485 logger.error("Error detecting conflicts", exc_info=True)
486 return self.protocol.create_response(
487 request_id,
488 error={"code": -32603, "message": "Internal server error"},
489 )
491 async def handle_find_complementary_content(
492 self, request_id: str | int | None, params: dict[str, Any]
493 ) -> dict[str, Any]:
494 """Handle complementary content request."""
495 logger.debug("Handling complementary content with params", params=params)
497 required_params = ["target_query", "context_query"]
498 for param in required_params:
499 if param not in params:
500 logger.error(f"Missing required parameter: {param}")
501 return self.protocol.create_response(
502 request_id,
503 error={
504 "code": -32602,
505 "message": "Invalid params",
506 "data": f"Missing required parameter: {param}",
507 },
508 )
510 try:
511 logger.debug(
512 "Calling search_engine.find_complementary_content (%s)",
513 type(self.search_engine).__name__,
514 )
516 result = await self.search_engine.find_complementary_content(
517 target_query=params["target_query"],
518 context_query=params["context_query"],
519 max_recommendations=params.get("max_recommendations", 5),
520 source_types=params.get("source_types"),
521 project_ids=params.get("project_ids"),
522 )
524 # Defensive check to ensure we received the expected result type
525 if not isinstance(result, dict):
526 logger.error(
527 "Unexpected complementary content result type",
528 got_type=str(type(result)),
529 )
530 return self.protocol.create_response(
531 request_id,
532 error={"code": -32603, "message": "Internal server error"},
533 )
535 complementary_recommendations = result.get(
536 "complementary_recommendations", []
537 )
538 target_document = result.get("target_document")
539 context_documents_analyzed = result.get("context_documents_analyzed", 0)
541 logger.debug(
542 "find_complementary_content completed, got %s results",
543 len(complementary_recommendations),
544 )
546 # Create lightweight structured content using the new formatter
547 structured_content = (
548 self.formatters.create_lightweight_complementary_results(
549 complementary_recommendations=complementary_recommendations,
550 target_document=target_document,
551 context_documents_analyzed=context_documents_analyzed,
552 target_query=params["target_query"],
553 )
554 )
556 return self.protocol.create_response(
557 request_id,
558 result={
559 "content": [
560 {
561 "type": "text",
562 "text": self.formatters.format_complementary_content(
563 complementary_recommendations
564 ),
565 }
566 ],
567 "structuredContent": structured_content,
568 "isError": False,
569 },
570 )
572 except Exception:
573 logger.error("Error finding complementary content", exc_info=True)
574 return self.protocol.create_response(
575 request_id,
576 error={"code": -32603, "message": "Internal server error"},
577 )
579 async def handle_cluster_documents(
580 self, request_id: str | int | None, params: dict[str, Any]
581 ) -> dict[str, Any]:
582 """Handle document clustering request."""
583 logger.debug("Handling document clustering with params", params=params)
585 if "query" not in params:
586 logger.error("Missing required parameter: query")
587 return self.protocol.create_response(
588 request_id,
589 error={
590 "code": -32602,
591 "message": "Invalid params",
592 "data": "Missing required parameter: query",
593 },
594 )
596 try:
597 logger.info("Performing document clustering using SearchEngine...")
599 # Use the sophisticated SearchEngine method
600 clustering_results = await self.search_engine.cluster_documents(
601 query=params["query"],
602 limit=params.get("limit", 25),
603 max_clusters=params.get("max_clusters", 10),
604 min_cluster_size=params.get("min_cluster_size", 2),
605 strategy=params.get("strategy", "mixed_features"),
606 source_types=params.get("source_types"),
607 project_ids=params.get("project_ids"),
608 )
610 logger.info("Document clustering completed successfully")
612 # Also produce lightweight clusters for back-compat (unit tests expect this call)
613 _legacy_lightweight_clusters = (
614 self.formatters.create_lightweight_cluster_results(
615 clustering_results, params.get("query", "")
616 )
617 )
619 # Store for expand_cluster call (keep full document object)
620 cluster_session_id = str(uuid.uuid4())
621 async with self._lock:
622 self._cleanup_sessions_locked()
623 self._cluster_store[cluster_session_id] = {
624 "data": {
625 "clusters": clustering_results.get("clusters", []),
626 "clustering_metadata": clustering_results.get(
627 "clustering_metadata"
628 ),
629 },
630 "expires_at": time.time() + self._ttl,
631 }
633 # Build schema-compliant clustering response
634 schema_clusters: list[dict[str, Any]] = []
635 for idx, cluster in enumerate(clustering_results.get("clusters", []) or []):
636 # Documents within cluster
637 docs_schema: list[dict[str, Any]] = []
638 for d in cluster.get("documents", []) or []:
639 try:
640 score = float(getattr(d, "score", 0.0))
641 except Exception:
642 score = 0.0
643 # Clamp to [0,1]
644 if score < 0:
645 score = 0.0
646 if score > 1:
647 score = 1.0
648 text_val = getattr(d, "text", "")
649 content_preview = (
650 text_val[:200] + "..."
651 if isinstance(text_val, str) and len(text_val) > 200
652 else (text_val if isinstance(text_val, str) else "")
653 )
654 docs_schema.append(
655 {
656 "document_id": str(getattr(d, "document_id", "")),
657 "title": getattr(d, "source_title", "Untitled"),
658 "content_preview": content_preview,
659 "source_type": getattr(d, "source_type", "unknown"),
660 "cluster_relevance": score,
661 }
662 )
664 # Derive theme and keywords
665 centroid_topics = cluster.get("centroid_topics") or []
666 shared_entities = cluster.get("shared_entities") or []
667 theme_str = (
668 ", ".join(centroid_topics[:3])
669 if centroid_topics
670 else (
671 ", ".join(shared_entities[:3])
672 if shared_entities
673 else (cluster.get("cluster_summary") or "")
674 )
675 )
677 # Clamp cohesion_score to [0,1] as required by schema
678 try:
679 cohesion = float(cluster.get("coherence_score", 0.0))
680 except Exception:
681 cohesion = 0.0
682 if cohesion < 0:
683 cohesion = 0.0
684 if cohesion > 1:
685 cohesion = 1.0
687 schema_clusters.append(
688 {
689 "cluster_id": str(cluster.get("id", f"cluster_{idx + 1}")),
690 "cluster_name": cluster.get("name") or f"Cluster {idx + 1}",
691 "cluster_theme": theme_str,
692 "document_count": int(
693 cluster.get(
694 "document_count",
695 len(cluster.get("documents", []) or []),
696 )
697 ),
698 "cohesion_score": cohesion,
699 "documents": docs_schema,
700 "cluster_keywords": shared_entities or centroid_topics,
701 "cluster_summary": cluster.get("cluster_summary", ""),
702 }
703 )
705 meta_src = clustering_results.get("clustering_metadata", {}) or {}
706 clustering_metadata = {
707 "total_documents": int(meta_src.get("total_documents", 0)),
708 "clusters_created": int(
709 meta_src.get("clusters_created", len(schema_clusters))
710 ),
711 "strategy": str(meta_src.get("strategy", "unknown")),
712 }
713 # Optional metadata
714 if "unclustered_documents" in meta_src:
715 clustering_metadata["unclustered_documents"] = int(
716 meta_src.get("unclustered_documents", 0)
717 )
718 if "clustering_quality" in meta_src:
719 try:
720 clustering_metadata["clustering_quality"] = float(
721 meta_src.get("clustering_quality", 0.0)
722 )
723 except Exception:
724 pass
725 if "processing_time_ms" in meta_src:
726 clustering_metadata["processing_time_ms"] = int(
727 meta_src.get("processing_time_ms", 0)
728 )
730 # Normalize cluster relationships to schema
731 normalized_relationships: list[dict[str, Any]] = []
732 for rel in clustering_results.get("cluster_relationships", []) or []:
733 cluster_1 = (
734 rel.get("cluster_1")
735 or rel.get("source_cluster")
736 or rel.get("a")
737 or rel.get("from")
738 or rel.get("cluster_a")
739 or rel.get("id1")
740 or ""
741 )
742 cluster_2 = (
743 rel.get("cluster_2")
744 or rel.get("target_cluster")
745 or rel.get("b")
746 or rel.get("to")
747 or rel.get("cluster_b")
748 or rel.get("id2")
749 or ""
750 )
751 relationship_type = (
752 rel.get("relationship_type") or rel.get("type") or "related"
753 )
754 try:
755 relationship_strength = float(
756 rel.get("relationship_strength")
757 or rel.get("score")
758 or rel.get("overlap_score")
759 or 0.0
760 )
761 except Exception:
762 relationship_strength = 0.0
764 normalized_relationships.append(
765 {
766 "cluster_1": str(cluster_1),
767 "cluster_2": str(cluster_2),
768 "relationship_type": relationship_type,
769 "relationship_strength": relationship_strength,
770 }
771 )
773 mcp_clustering_results = {
774 "clusters": schema_clusters,
775 "clustering_metadata": clustering_metadata,
776 "cluster_relationships": normalized_relationships,
777 }
779 return self.protocol.create_response(
780 request_id,
781 result={
782 "content": [
783 {
784 "type": "text",
785 "text": self.formatters.format_document_clusters(
786 clustering_results
787 ),
788 }
789 ],
790 "structuredContent": {
791 **mcp_clustering_results,
792 "cluster_session_id": cluster_session_id,
793 },
794 "isError": False,
795 },
796 )
798 except Exception:
799 logger.error("Error clustering documents", exc_info=True)
800 return self.protocol.create_response(
801 request_id,
802 error={"code": -32603, "message": "Internal server error"},
803 )
805 async def handle_expand_cluster(
806 self, request_id: str | int | None, params: dict[str, Any]
807 ) -> dict[str, Any]:
808 """Handle cluster expansion request for lazy loading."""
809 logger.debug("Handling expand cluster with params", params=params)
811 # 1. Validate cluster_session_id
812 cluster_session_id = params.get("cluster_session_id")
813 if not cluster_session_id:
814 logger.error("Missing required parameter: cluster_session_id")
815 return self.protocol.create_response(
816 request_id,
817 error={
818 "code": -32602,
819 "message": "Invalid params",
820 "data": "Missing required parameter: cluster_session_id",
821 },
822 )
824 # 2. Validate cluster_id
825 cluster_id = params.get("cluster_id")
826 if not cluster_id:
827 logger.error("Missing required parameter: cluster_id")
828 return self.protocol.create_response(
829 request_id,
830 error={
831 "code": -32602,
832 "message": "Invalid params",
833 "data": "Missing required parameter: cluster_id",
834 },
835 )
837 cluster_id = str(cluster_id).strip()
839 # 3. Pagination params
840 try:
841 limit = max(1, min(100, int(params.get("limit", 20))))
842 except Exception:
843 limit = 20
845 try:
846 offset = max(0, int(params.get("offset", 0)))
847 except Exception:
848 offset = 0
850 include_metadata = params.get("include_metadata", True)
852 # 4. Get cache (LOCK)
853 now = time.time()
855 # Note: Cache is in-memory and per-process. In multi-worker deployments,
856 # cluster_session_id created on one worker won't be available on others
857 # unless using shared storage (e.g., Redis) or sticky routing.
858 async with self._lock:
859 entry = self._cluster_store.get(cluster_session_id)
861 if entry and entry.get("expires_at", 0) < now:
862 self._cluster_store.pop(cluster_session_id, None)
863 entry = None
865 if entry is None:
866 return self.protocol.create_response(
867 request_id,
868 error={
869 "code": -32001,
870 "message": "Session not found or expired",
871 "data": f"Cluster session '{cluster_session_id}' not found or expired",
872 },
873 )
875 cache = entry.get("data") or {}
876 clusters = cache.get("clusters") or []
878 # 5. Find cluster
879 cluster = next(
880 (
881 c
882 for idx, c in enumerate(clusters)
883 if str(c.get("id", f"cluster_{idx + 1}")) == cluster_id
884 ),
885 None,
886 )
888 if not cluster:
889 return self.protocol.create_response(
890 request_id,
891 error={
892 "code": -32002,
893 "message": "Cluster not found",
894 "data": f"No cluster with id '{cluster_id}' found",
895 },
896 )
898 # 6. Pagination
899 all_docs = cluster.get("documents") or []
900 total = len(all_docs)
902 slice_docs = all_docs[offset : offset + limit]
903 has_more = offset + len(slice_docs) < total
904 page = (offset // limit) + 1 if limit > 0 else 1
905 # 7. Transform documents
906 doc_schema_list = self._expand_cluster_docs_to_schema(
907 slice_docs, include_metadata
908 )
910 # 8. Extract theme
911 theme = (
912 cluster.get("cluster_summary")
913 or ", ".join(
914 (
915 cluster.get("shared_entities")
916 or cluster.get("centroid_topics")
917 or []
918 )[:3]
919 )
920 or "N/A"
921 )
923 # 9. Build result
924 result = {
925 "cluster_id": cluster_id,
926 "cluster_info": {
927 "cluster_name": cluster.get("name") or f"Cluster {cluster_id}",
928 "cluster_theme": theme,
929 "document_count": total,
930 },
931 "documents": doc_schema_list,
932 "pagination": {
933 "page": page,
934 "page_size": limit,
935 "total": total,
936 "has_more": has_more,
937 },
938 }
940 # 10. Return response
941 return self.protocol.create_response(
942 request_id,
943 result={
944 "content": [
945 {
946 "type": "text",
947 "text": self._format_text_block(result),
948 }
949 ],
950 "structuredContent": result,
951 "isError": False,
952 },
953 )
955 def _format_text_block(self, result: dict) -> str:
956 info = result.get("cluster_info", {})
957 docs = result.get("documents", [])
958 total = info.get("document_count", 0)
960 text = (
961 f"**Cluster: {info.get('cluster_name', 'Unknown')}**\n"
962 f"Theme: {info.get('cluster_theme', 'N/A')}\n"
963 f"Documents: {total}\n\n"
964 )
966 for i, d in enumerate(docs[:5], 1):
967 title = d.get("metadata", {}).get("title", d.get("id", "Unknown"))
968 text += f"{i}. {title}\n"
970 if total > 5:
971 text += f"... and {total - 5} more.\n"
973 return text
975 def _cleanup_sessions_locked(self):
976 now = time.time()
978 expired_keys = [
979 k for k, v in self._cluster_store.items() if v.get("expires_at", 0) < now
980 ]
982 for k in expired_keys:
983 self._cluster_store.pop(k, None)
985 if len(self._cluster_store) > self._max_sessions:
986 sorted_items = sorted(
987 self._cluster_store.items(), key=lambda x: x[1].get("expires_at", 0)
988 )
989 overflow = len(self._cluster_store) - self._max_sessions
990 for k, _ in sorted_items[:overflow]:
991 self._cluster_store.pop(k, None)
993 @staticmethod
994 def _validate_depth(depth: int) -> int:
995 if depth < 1 or depth > 10:
996 raise ValueError("depth must be between 1 and 10")
997 return depth
1000IntelligenceHandler.handle_find_ticket_dependencies = handle_find_ticket_dependencies